From 2b1a499678745c68eb3b23641822795586369e73 Mon Sep 17 00:00:00 2001 From: Cole Crawford Date: Tue, 16 Jun 2026 09:28:12 -0600 Subject: [PATCH 1/3] fix(plugin-worker-manager): recover when a worker's stdin command channel dies MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A plugin worker communicates with the host over newline-delimited JSON-RPC on the child's stdin/stdout pipes. If the host->worker stdin pipe is destroyed by an EPIPE (or otherwise closes) while the child process keeps running, the worker becomes uncommandable: every `sendMessage` throws `Worker process for plugin "" is not writable`, so all host->worker calls (e.g. http.fetch proxying) fail forever. Crucially `child.on("exit")` never fires, so the existing crash-recovery/backoff path is never reached and the worker silently zombies — observed in a Telegram-bridge plugin whose getUpdates long-poll began failing after ~5 days of uptime and never recovered until a manual disable/enable. Fix (no polling watchdog — supervise the command channel the same way the process is already supervised): 1. `sendMessage` passes a write callback so an async stdin write error is surfaced/logged instead of being swallowed. 2. `attachStdioHandlers` attaches `error`/`close` handlers to `child.stdin`. If the channel dies while the process is still alive (exitCode/signalCode null) and the stop wasn't intentional, it SIGKILLs the child so the normal `handleProcessExit() -> scheduleRestart()` recovery runs. This is the missing third failure mode, mirroring the existing exit/error handlers. Co-Authored-By: Claude Fable 5 --- server/src/services/plugin-worker-manager.ts | 39 +++++++++++++++++++- 1 file changed, 38 insertions(+), 1 deletion(-) diff --git a/server/src/services/plugin-worker-manager.ts b/server/src/services/plugin-worker-manager.ts index e0a5cbd0d8..4b15fac550 100644 --- a/server/src/services/plugin-worker-manager.ts +++ b/server/src/services/plugin-worker-manager.ts @@ -529,7 +529,14 @@ export function createPluginWorkerHandle( throw new Error(`Worker process for plugin "${pluginId}" is not writable`); } const serialized = serializeMessage(message as any); - childProcess.stdin.write(serialized); + // Pass a write callback so an async write error (e.g. EPIPE on the command + // pipe) is surfaced rather than swallowed. The stdin "error" handler wired + // in attachStdioHandlers() drives the actual recovery (forced restart). + childProcess.stdin.write(serialized, (err) => { + if (err) { + log.warn({ err: err.message }, "failed to write message to worker stdin"); + } + }); } function errorCodeForWorkerHostError(err: unknown): number { @@ -928,6 +935,36 @@ export function createPluginWorkerHandle( ); } }); + + // Supervise the command channel (host -> worker stdin), not just the + // process. If stdin errors (EPIPE) or closes while the process is still + // alive, the worker is uncommandable: every host->worker RPC write will + // throw "not writable" forever, yet child.on("exit") never fires, so the + // crash-recovery path is never reached and the worker silently zombies + // (observed after multi-day uptime). Force a real exit so the standard + // handleProcessExit() -> scheduleRestart() recovery runs. This is the + // missing third failure mode, mirroring the exit/error handlers above — + // event-driven, not a polling watchdog. + if (child.stdin) { + const onCommandChannelLost = (err?: Error): void => { + // Ignore during graceful stop, or if this child was already replaced. + if (intentionalStop || childProcess !== child) return; + // Only act while the process is still alive (a real exit is handled by + // handleProcessExit). exitCode/signalCode are null until the child dies. + if (child.exitCode !== null || child.signalCode !== null) return; + log.error( + { err: err?.message }, + "worker stdin (command channel) lost while process alive — forcing restart", + ); + try { + child.kill("SIGKILL"); + } catch { + // Best effort — handleProcessExit still runs on the eventual exit. + } + }; + child.stdin.on("error", onCommandChannelLost); + child.stdin.on("close", onCommandChannelLost); + } } function handleProcessExit( From 505da32def187fd6fc301355dc932f8a9ae8fdc5 Mon Sep 17 00:00:00 2001 From: Tycho Date: Wed, 5 Aug 2026 12:50:11 -0600 Subject: [PATCH 2/3] test(plugin-worker-manager): cover stdin command-channel supervision Adds the missing test for the supervision added in this PR: a worker whose host->worker stdin pipe dies while the process itself stays alive. New suite server/src/__tests__/plugin-worker-stdin-supervision.test.ts, plus a plugin-worker-persistent.cjs fixture that deliberately survives stdin EOF (ref'd keep-alive timer, readline "close" is a no-op) so nothing but the host supervision can end it. It spawns a real child through the existing fixture pattern and captures it by wrapping node:child_process.fork with a pass-through mock; the fork is real. Two cases: 1. stdin destroyed with EPIPE while the process is alive -> the manager SIGKILLs the child and schedules a restart (crash event willRestart=true, status "backoff", diagnostics().nextRestartAt in the future). SIGKILL is the discriminator: this fixture never exits on its own inside the test window and exits 0 when asked politely, so only the supervision path can produce a signalled exit here. 2. Negative control for the risky arm: stdin destroyed *during* an intentional stop must not force-kill or restart. The fixture defers its exit after acking shutdown, which opens the window; the test waits for status "stopping" and for the shutdown to drain out of the host before killing the pipe, so it cannot pass by dropping the shutdown instead. Verified by mutation, not just by going green: - reverting plugin-worker-manager.ts to the pre-fix parent fails case 1 only ("worker was left alive and uncommandable after its command channel died") and leaves case 2 green; - removing only the `intentionalStop ||` guard fails case 2 only ("expected 'SIGKILL' to be null") and leaves case 1 green. Lives in its own file because vi.mock is file-scoped and the sibling plugin-worker-manager.test.ts keeps an unmocked fork; that suite still passes (30 tests across both files). Server typecheck clean. The suite is picked up by the glob-based general-server sharding and lands in exactly one of the four shards. --- .../fixtures/plugin-worker-persistent.cjs | 82 +++++++ .../plugin-worker-stdin-supervision.test.ts | 212 ++++++++++++++++++ 2 files changed, 294 insertions(+) create mode 100644 server/src/__tests__/fixtures/plugin-worker-persistent.cjs create mode 100644 server/src/__tests__/plugin-worker-stdin-supervision.test.ts diff --git a/server/src/__tests__/fixtures/plugin-worker-persistent.cjs b/server/src/__tests__/fixtures/plugin-worker-persistent.cjs new file mode 100644 index 0000000000..21535d908a --- /dev/null +++ b/server/src/__tests__/fixtures/plugin-worker-persistent.cjs @@ -0,0 +1,82 @@ +// Long-lived worker fixture for the stdin command-channel supervision tests. +// +// Unlike the other fixtures, this worker deliberately survives losing its +// stdin: readline "close" does not terminate it, and a ref'd keep-alive timer +// holds the event loop open. That reproduces the field shape the supervision +// fix exists for — the host->worker command pipe dies while the worker process +// itself stays alive and uncommandable. +// +// `shutdown` is acknowledged immediately but the exit is deliberately deferred, +// which leaves a window in which a test can kill the command channel *during* +// an intentional stop (the negative control). + +const readline = require("node:readline"); + +/** How long the worker waits after acking `shutdown` before exiting. */ +const SHUTDOWN_EXIT_DELAY_MS = 300; + +/** Hard ceiling so a fixture never outlives the test run that spawned it. */ +const MAX_LIFETIME_MS = 30_000; + +function send(message) { + process.stdout.write(`${JSON.stringify(message)}\n`); +} + +// Ref'd, so the process stays alive after stdin reaches EOF. +const keepAlive = setInterval(() => {}, 1_000); + +const selfDestruct = setTimeout(() => { + process.exit(0); +}, MAX_LIFETIME_MS); + +function exitNow() { + clearInterval(keepAlive); + clearTimeout(selfDestruct); + process.exit(0); +} + +const rl = readline.createInterface({ + input: process.stdin, + crlfDelay: Infinity, +}); + +// Explicitly do NOT exit here: losing the command channel must leave this +// process alive so the host supervision path is the thing under test. +rl.on("close", () => {}); + +rl.on("line", (line) => { + if (!line.trim()) return; + const message = JSON.parse(line); + const method = message && typeof message.method === "string" ? message.method : null; + + if (method === "initialize") { + send({ + jsonrpc: "2.0", + id: message.id, + result: { + ok: true, + supportedMethods: [], + }, + }); + return; + } + + if (method === "shutdown") { + send({ + jsonrpc: "2.0", + id: message.id, + result: {}, + }); + setTimeout(exitNow, SHUTDOWN_EXIT_DELAY_MS); + return; + } + + send({ + jsonrpc: "2.0", + id: message.id, + error: { + code: -32601, + message: `Unhandled method: ${method}`, + }, + }); +}); diff --git a/server/src/__tests__/plugin-worker-stdin-supervision.test.ts b/server/src/__tests__/plugin-worker-stdin-supervision.test.ts new file mode 100644 index 0000000000..4249054316 --- /dev/null +++ b/server/src/__tests__/plugin-worker-stdin-supervision.test.ts @@ -0,0 +1,212 @@ +/** + * Supervision of the host→worker command channel (child stdin). + * + * A worker can lose its stdin pipe (EPIPE, or the pipe closing) while the + * process itself is still alive. `child.on("exit")` never fires, so the normal + * crash-recovery path is never reached and the worker zombies: alive, holding + * its slot, and rejecting every host→worker RPC with "not writable" forever. + * These tests cover the supervision that turns that into a real exit so the + * existing handleProcessExit() → scheduleRestart() recovery runs. + * + * This lives in its own file rather than in plugin-worker-manager.test.ts + * because it needs to mock `node:child_process` to capture the spawned child, + * and vi.mock is file-scoped — the sibling suite keeps an unmocked fork. + */ + +import path from "node:path"; +import type { ChildProcess } from "node:child_process"; +import { fileURLToPath } from "node:url"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { PaperclipPluginManifestV1 } from "@paperclipai/shared"; +import { createPluginWorkerHandle } from "../services/plugin-worker-manager.js"; + +// Hoisted so the vi.mock factory (which is hoisted above the imports) can see +// it. The fork itself is the real one — only the reference is captured. +const { forkedChildren } = vi.hoisted(() => ({ + forkedChildren: [] as ChildProcess[], +})); + +vi.mock("node:child_process", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + fork: (...args: Parameters): ChildProcess => { + const child = actual.fork(...args); + forkedChildren.push(child); + return child; + }, + }; +}); + +const FIXTURES_DIR = path.join(path.dirname(fileURLToPath(import.meta.url)), "fixtures"); +const PERSISTENT_WORKER_ENTRYPOINT = path.join(FIXTURES_DIR, "plugin-worker-persistent.cjs"); + +const TEST_MANIFEST: PaperclipPluginManifestV1 = { + id: "test.plugin", + apiVersion: 1, + version: "1.0.0", + displayName: "Test plugin", + description: "Test plugin", + author: "Paperclip", + categories: ["automation"], + capabilities: [], + entrypoints: { worker: "dist/worker.js" }, +}; + +/** An EPIPE the way Node surfaces one on a dead pipe. */ +function epipe(): NodeJS.ErrnoException { + const err: NodeJS.ErrnoException = new Error("write EPIPE"); + err.code = "EPIPE"; + err.syscall = "write"; + return err; +} + +type Exit = { code: number | null; signal: NodeJS.Signals | null }; + +function nextExit(child: ChildProcess): Promise { + return new Promise((resolve) => { + if (child.exitCode !== null || child.signalCode !== null) { + resolve({ code: child.exitCode, signal: child.signalCode }); + return; + } + child.once("exit", (code, signal) => resolve({ code, signal })); + }); +} + +/** + * Resolve to the child's exit, or to `null` if it is still alive after + * `timeoutMs`. Reporting "still alive" as a value rather than letting the test + * time out is deliberate: an unsupervised worker zombies forever, and the + * assertion below should name that rather than surface as a bare timeout. + */ +function exitWithin(child: ChildProcess, timeoutMs: number): Promise { + return Promise.race([ + nextExit(child), + new Promise((resolve) => setTimeout(() => resolve(null), timeoutMs)), + ]); +} + +async function startPersistentWorker() { + const before = forkedChildren.length; + const handle = createPluginWorkerHandle("test.plugin", { + entrypointPath: PERSISTENT_WORKER_ENTRYPOINT, + manifest: TEST_MANIFEST, + config: {}, + instanceInfo: { + instanceId: "instance-1", + hostVersion: "1.0.0", + }, + apiVersion: 1, + hostHandlers: {}, + rpcTimeoutMs: 5_000, + }); + + await handle.start(); + + const child = forkedChildren[before]; + expect(child, "expected the handle to have forked exactly one child").toBeDefined(); + expect(handle.status).toBe("running"); + // Precondition for both tests: the process is alive and the command channel + // is usable. Without this the assertions below could pass vacuously. + expect(child.exitCode).toBeNull(); + expect(child.signalCode).toBeNull(); + expect(child.stdin?.destroyed ?? true).toBe(false); + + return { handle, child }; +} + +afterEach(() => { + for (const child of forkedChildren.splice(0)) { + if (child.exitCode === null && child.signalCode === null) { + try { + child.kill("SIGKILL"); + } catch { + // Already gone. + } + } + } +}); + +describe("plugin worker stdin command-channel supervision", () => { + it("kills the worker and schedules a restart when stdin dies while the process is alive", async () => { + const { handle, child } = await startPersistentWorker(); + + try { + const crashes: Array<{ signal: NodeJS.Signals | null; willRestart: boolean }> = []; + handle.on("crash", (payload) => { + crashes.push({ signal: payload.signal, willRestart: payload.willRestart }); + }); + + const exited = exitWithin(child, 2_000); + + // The failure shape from the field: the command pipe dies, the worker + // process does not. This fixture deliberately survives stdin EOF, so + // nothing but the host supervision can end it. + child.stdin?.destroy(epipe()); + + const exit = await exited; + + expect( + exit, + "worker was left alive and uncommandable after its command channel died", + ).not.toBeNull(); + // SIGKILL is the discriminator: the fixture never exits on its own + // within the test window, and it exits 0 when asked politely. Only the + // supervision path produces a signalled exit here. + expect(exit!.signal).toBe("SIGKILL"); + + await vi.waitFor(() => { + expect(crashes).toHaveLength(1); + }); + expect(crashes[0]?.willRestart).toBe(true); + + // The recovery that matters is the restart, not the kill: a worker that + // is killed and not rescheduled is still gone. + expect(handle.status).toBe("backoff"); + const diagnostics = handle.diagnostics(); + expect(diagnostics.consecutiveCrashes).toBe(1); + expect(diagnostics.nextRestartAt).not.toBeNull(); + expect(diagnostics.nextRestartAt!).toBeGreaterThan(Date.now()); + } finally { + // stop() cancels the pending backoff timer, so no restart escapes. + await handle.stop().catch(() => undefined); + } + }); + + it("does not force-kill or restart when stdin dies during an intentional stop", async () => { + // The risky arm of the change: a graceful stop closes the command channel + // as a matter of course, and must not be mistaken for the failure above. + const { handle, child } = await startPersistentWorker(); + + const crashes: unknown[] = []; + handle.on("crash", (payload) => crashes.push(payload)); + + const exited = nextExit(child); + const stopping = handle.stop(); + + // stopInternal() sets intentionalStop before it writes the shutdown RPC. + await vi.waitFor(() => { + expect(handle.status).toBe("stopping"); + // ...and the shutdown must have drained out of the host before the pipe + // is killed, or this would be testing a dropped shutdown instead. + expect(child.stdin?.writableLength ?? 0).toBe(0); + }); + + // Kill the command channel mid-stop, while the fixture is still inside its + // deferred-exit window. Without the intentionalStop guard this SIGKILLs a + // worker that was already shutting down cleanly. + child.stdin?.destroy(epipe()); + + await stopping; + const { code, signal } = await exited; + + expect(signal).toBeNull(); + expect(code).toBe(0); + expect(crashes).toHaveLength(0); + expect(handle.status).toBe("stopped"); + + const diagnostics = handle.diagnostics(); + expect(diagnostics.totalCrashes).toBe(0); + expect(diagnostics.nextRestartAt).toBeNull(); + }); +}); From 1efea22d2f9d0147a0cc2e6d3eca9d17ec828db9 Mon Sep 17 00:00:00 2001 From: Tycho Date: Wed, 5 Aug 2026 13:11:25 -0600 Subject: [PATCH 3/3] test(plugin-worker-manager): close a timing race in the stdin supervision test The negative-control test had a 200ms margin against a real process exit, which is too thin to rely on under a loaded CI runner. stopInternal() races the shutdown RPC against waitForExit(), and the RPC resolves as soon as the worker acks. It then allows only a further 500ms before escalating to SIGTERM. The fixture exited 300ms after acking, so a late timer or a slow exit propagation would have surfaced as SIGTERM and failed the "signal must be null" assertion for a reason unrelated to the guard under test. Two deadlines were squeezing from opposite sides: the destroy had to land before the fixture exited, and the fixture had to exit before the host escalated. Widening one window narrows the other, so the fix is to stop polling rather than to retune the constants. The test now kills the command channel from a hook on the host's own stdin.write, firing the moment the shutdown payload is flushed, instead of waiting for status "stopping" via vi.waitFor. That removes both races: the destroy lands immediately after the shutdown reaches the worker, and the fixture's exit delay drops to 100ms, leaving 400ms of margin. The hook is also a stronger precondition than the status poll it replaces. sendMessage() is only reached for "shutdown" from inside stopInternal(), which sets intentionalStop before it writes -- so firing there proves we are inside the intentional-stop window rather than inferring it from an observable status. The previous writableLength check is subsumed: the pipe is destroyed from the write path itself, after the payload is handed off. Added an explicit aliveAtDestroy assertion so the vacuous case is loud. If the fixture ever exits before the pipe is killed, the guard is never exercised and the old test would still have passed; now it fails with "fixture exited before the command channel was killed". Re-verified by mutation, not just by going green: - manager reverted to the pre-fix parent aa6b6bcb (supervision block absent, grep -c to zero, not a partial mutation) fails case 1 only, with "worker was left alive and uncommandable after its command channel died"; - removing only the `intentionalStop ||` guard fails case 2 only, with "expected 'SIGKILL' to be null". Each mutation still reds exactly one test and leaves the other green. Suite run 5x consecutively, clean. Server typecheck clean. Test-only change: no source file is touched. --- .../fixtures/plugin-worker-persistent.cjs | 12 +++- .../plugin-worker-stdin-supervision.test.ts | 67 ++++++++++++++++--- 2 files changed, 67 insertions(+), 12 deletions(-) diff --git a/server/src/__tests__/fixtures/plugin-worker-persistent.cjs b/server/src/__tests__/fixtures/plugin-worker-persistent.cjs index 21535d908a..65eff19127 100644 --- a/server/src/__tests__/fixtures/plugin-worker-persistent.cjs +++ b/server/src/__tests__/fixtures/plugin-worker-persistent.cjs @@ -12,8 +12,16 @@ const readline = require("node:readline"); -/** How long the worker waits after acking `shutdown` before exiting. */ -const SHUTDOWN_EXIT_DELAY_MS = 300; +// How long the worker waits after acking `shutdown` before exiting. +// +// This has to sit inside the host's post-ack grace period: stopInternal() +// races the shutdown RPC (which resolves as soon as this ack lands) and then +// waits only 500ms more before escalating to SIGTERM. A delay close to that +// ceiling makes the negative-control test a timing race against a real process +// exit on a loaded CI runner, so keep the margin wide. The test does not +// depend on this window being long — it kills the pipe from a write hook the +// moment the shutdown is flushed, not after a poll. +const SHUTDOWN_EXIT_DELAY_MS = 100; /** Hard ceiling so a fixture never outlives the test run that spawned it. */ const MAX_LIFETIME_MS = 30_000; diff --git a/server/src/__tests__/plugin-worker-stdin-supervision.test.ts b/server/src/__tests__/plugin-worker-stdin-supervision.test.ts index 4249054316..9ae2fb50e5 100644 --- a/server/src/__tests__/plugin-worker-stdin-supervision.test.ts +++ b/server/src/__tests__/plugin-worker-stdin-supervision.test.ts @@ -86,6 +86,55 @@ function exitWithin(child: ChildProcess, timeoutMs: number): Promise { + const stdin = child.stdin; + if (!stdin) throw new Error("expected the forked child to have a stdin pipe"); + + return new Promise((resolve) => { + const originalWrite = stdin.write.bind(stdin) as typeof stdin.write; + let fired = false; + + stdin.write = ((chunk: unknown, ...rest: unknown[]) => { + const accepted = (originalWrite as (...a: unknown[]) => boolean)(chunk, ...rest); + + if (!fired && typeof chunk === "string" && chunk.includes('"shutdown"')) { + fired = true; + // Let the manager's own write callback run first, so the shutdown is + // fully handed off before the pipe dies. + setImmediate(() => { + const aliveAtDestroy = child.exitCode === null && child.signalCode === null; + stdin.destroy(epipe()); + resolve({ aliveAtDestroy }); + }); + } + + return accepted; + }) as typeof stdin.write; + }); +} + async function startPersistentWorker() { const before = forkedChildren.length; const handle = createPluginWorkerHandle("test.plugin", { @@ -182,20 +231,18 @@ describe("plugin worker stdin command-channel supervision", () => { handle.on("crash", (payload) => crashes.push(payload)); const exited = nextExit(child); - const stopping = handle.stop(); - - // stopInternal() sets intentionalStop before it writes the shutdown RPC. - await vi.waitFor(() => { - expect(handle.status).toBe("stopping"); - // ...and the shutdown must have drained out of the host before the pipe - // is killed, or this would be testing a dropped shutdown instead. - expect(child.stdin?.writableLength ?? 0).toBe(0); - }); // Kill the command channel mid-stop, while the fixture is still inside its // deferred-exit window. Without the intentionalStop guard this SIGKILLs a // worker that was already shutting down cleanly. - child.stdin?.destroy(epipe()); + const destroyed = destroyStdinOnShutdown(child); + const stopping = handle.stop(); + + const { aliveAtDestroy } = await destroyed; + expect( + aliveAtDestroy, + "fixture exited before the command channel was killed — the guard was never exercised", + ).toBe(true); await stopping; const { code, signal } = await exited;