From 4130037d72b4913c8d526992254ceb1818830ca6 Mon Sep 17 00:00:00 2001 From: Sebastian Date: Thu, 27 Aug 2026 14:04:46 +0200 Subject: [PATCH 1/8] fix(coding-agent): remove blanket RPC timeouts, reject on real boundaries --- .../.changes/remove-rpc-blanket-timeouts.md | 1 + .../coding-agent/src/modes/rpc/rpc-client.ts | 185 +++++++++++------- .../fixtures/rpc-client-hanging-fixture.mjs | 1 + .../test/rpc-client-refine.test.ts | 33 ---- .../test/rpc-client-timeout.test.ts | 107 ++++++++++ 5 files changed, 223 insertions(+), 104 deletions(-) create mode 100644 packages/coding-agent/.changes/remove-rpc-blanket-timeouts.md create mode 100644 packages/coding-agent/test/fixtures/rpc-client-hanging-fixture.mjs delete mode 100644 packages/coding-agent/test/rpc-client-refine.test.ts create mode 100644 packages/coding-agent/test/rpc-client-timeout.test.ts diff --git a/packages/coding-agent/.changes/remove-rpc-blanket-timeouts.md b/packages/coding-agent/.changes/remove-rpc-blanket-timeouts.md new file mode 100644 index 0000000000..b9e15b8d33 --- /dev/null +++ b/packages/coding-agent/.changes/remove-rpc-blanket-timeouts.md @@ -0,0 +1 @@ +- Allowed long-running RPC commands and agent turns to complete without fixed client timeouts. diff --git a/packages/coding-agent/src/modes/rpc/rpc-client.ts b/packages/coding-agent/src/modes/rpc/rpc-client.ts index 7c32788164..fa3ea7de1a 100644 --- a/packages/coding-agent/src/modes/rpc/rpc-client.ts +++ b/packages/coding-agent/src/modes/rpc/rpc-client.ts @@ -32,9 +32,6 @@ import type { // Types // ============================================================================ -/** Extended response timeout for refine requests, which run an LLM pass. */ -export const REFINE_REQUEST_TIMEOUT_MS = 10 * 60 * 1000; - /** Distributive Omit that works with union types */ type DistributiveOmit = T extends unknown ? Omit : never; @@ -66,6 +63,11 @@ export interface ModelInfo { export type RpcEventListener = (event: AgentEvent) => void; export type RpcObservedSessionListener = (event: RpcObservedSessionEvent) => void; +interface RpcEventCollection { + promise: Promise; + cancel(): void; +} + // ============================================================================ // RPC Client // ============================================================================ @@ -79,6 +81,8 @@ export class RpcClient { new Map(); private requestId = 0; private stderr = ""; + private transportError: Error | null = null; + private pendingEventWaiters = new Set<(error: Error) => void>(); constructor(private options: RpcClientOptions = {}) {} @@ -103,28 +107,40 @@ export class RpcClient { args.push(...this.options.args); } - this.process = spawn("node", [cliPath, ...args], { + this.transportError = null; + const child = spawn("node", [cliPath, ...args], { cwd: this.options.cwd, env: { ...process.env, ...this.options.env }, stdio: ["pipe", "pipe", "pipe"], }); + this.process = child; + child.on("error", (error) => { + this.failPendingOperations(new Error(`RPC process error: ${error.message}. Stderr: ${this.stderr}`)); + }); + child.stdout?.on("close", () => { + this.failPendingOperations(new Error(`RPC process output closed. Stderr: ${this.stderr}`)); + }); + child.on("close", () => { + this.process = null; + }); // Collect stderr for debugging - this.process.stderr?.on("data", (data) => { + child.stderr?.on("data", (data) => { this.stderr += data.toString(); process.stderr.write(data); }); // Set up strict JSONL reader for stdout. - this.stopReadingStdout = attachJsonlLineReader(this.process.stdout!, (line) => { + this.stopReadingStdout = attachJsonlLineReader(child.stdout!, (line) => { this.handleLine(line); }); // Wait a moment for process to initialize await new Promise((resolve) => setTimeout(resolve, 100)); - if (this.process.exitCode !== null) { - throw new Error(`Agent process exited immediately with code ${this.process.exitCode}. Stderr: ${this.stderr}`); + if (this.transportError) throw this.transportError; + if (child.exitCode !== null) { + throw new Error(`Agent process exited immediately with code ${child.exitCode}. Stderr: ${this.stderr}`); } } @@ -132,27 +148,20 @@ export class RpcClient { * Stop the RPC agent process. */ async stop(): Promise { - if (!this.process) return; + const child = this.process; + if (!child) return; this.stopReadingStdout?.(); this.stopReadingStdout = null; - this.process.kill("SIGTERM"); - - // Wait for process to exit + this.failPendingOperations(new Error(`RPC client stopped. Stderr: ${this.stderr}`)); await new Promise((resolve) => { - const timeout = setTimeout(() => { - this.process?.kill("SIGKILL"); - resolve(); - }, 1000); - - this.process?.on("exit", () => { + const timeout = setTimeout(() => child.kill("SIGKILL"), 1000); + child.once("close", () => { clearTimeout(timeout); resolve(); }); + child.kill("SIGTERM"); }); - - this.process = null; - this.pendingRequests.clear(); } /** @@ -195,7 +204,8 @@ export class RpcClient { * Use waitForIdle() to wait for completion. */ async prompt(message: string, images?: ImageContent[]): Promise { - await this.send({ type: "prompt", message, images }); + const response = await this.send({ type: "prompt", message, images }); + this.getData(response); } /** @@ -308,8 +318,6 @@ export class RpcClient { async refine( options: { instructions?: string; rollbackId?: string; global?: boolean } = {}, ): Promise { - // Refinement runs an LLM pass that routinely exceeds the default 30s response - // timeout, so use the same extended window as the daemon refine path. const command = { type: "refine", instructions: options.instructions, rollbackId: options.rollbackId } as { type: "refine"; instructions?: string; @@ -319,7 +327,7 @@ export class RpcClient { if (options.global !== undefined) { command.global = options.global; } - const response = await this.send(command, REFINE_REQUEST_TIMEOUT_MS); + const response = await this.send(command); return this.getData(response); } @@ -540,58 +548,98 @@ export class RpcClient { * Wait for agent to become idle (no streaming). * Resolves when agent_end event is received. */ - waitForIdle(timeout = 60000): Promise { + waitForIdle(timeout?: number): Promise { + if (this.transportError) return Promise.reject(this.transportError); return new Promise((resolve, reject) => { - const timer = setTimeout(() => { + let timer: ReturnType | undefined; + const cleanup = () => { + if (timer) clearTimeout(timer); unsubscribe(); - reject(new Error(`Timeout waiting for agent to become idle. Stderr: ${this.stderr}`)); - }, timeout); - + this.pendingEventWaiters.delete(onFailure); + }; + const onFailure = (error: Error) => { + cleanup(); + reject(error); + }; const unsubscribe = this.onEvent((event) => { if (event.type === "agent_end") { - clearTimeout(timer); - unsubscribe(); + cleanup(); resolve(); } }); + this.pendingEventWaiters.add(onFailure); + if (timeout !== undefined) { + timer = setTimeout(() => { + cleanup(); + reject(new Error(`Timeout waiting for agent to become idle. Stderr: ${this.stderr}`)); + }, timeout); + } }); } /** * Collect events until agent becomes idle. */ - collectEvents(timeout = 60000): Promise { - return new Promise((resolve, reject) => { - const events: AgentEvent[] = []; - const timer = setTimeout(() => { - unsubscribe(); - reject(new Error(`Timeout collecting events. Stderr: ${this.stderr}`)); - }, timeout); - - const unsubscribe = this.onEvent((event) => { - events.push(event); - if (event.type === "agent_end") { - clearTimeout(timer); - unsubscribe(); - resolve(events); - } - }); - }); + collectEvents(timeout?: number): Promise { + return this.startEventCollection(timeout).promise; } /** * Send prompt and wait for completion, returning all events. */ - async promptAndWait(message: string, images?: ImageContent[], timeout = 60000): Promise { - const eventsPromise = this.collectEvents(timeout); - await this.prompt(message, images); - return eventsPromise; + async promptAndWait(message: string, images?: ImageContent[], timeout?: number): Promise { + const collection = this.startEventCollection(timeout); + try { + const [events] = await Promise.all([collection.promise, this.prompt(message, images)]); + return events; + } finally { + collection.cancel(); + } } // ========================================================================= // Internal // ========================================================================= + private startEventCollection(timeout?: number): RpcEventCollection { + if (this.transportError) { + return { promise: Promise.reject(this.transportError), cancel: () => undefined }; + } + let cancel = () => undefined; + const promise = new Promise((resolve, reject) => { + const events: AgentEvent[] = []; + let timer: ReturnType | undefined; + const cleanup = () => { + if (timer) clearTimeout(timer); + unsubscribe(); + this.pendingEventWaiters.delete(onFailure); + }; + const onFailure = (error: Error) => { + cleanup(); + reject(error); + }; + const unsubscribe = this.onEvent((event) => { + events.push(event); + if (event.type === "agent_end") { + cleanup(); + resolve(events); + } + }); + cancel = () => { + cleanup(); + resolve(events); + }; + this.pendingEventWaiters.add(onFailure); + if (timeout !== undefined) { + timer = setTimeout(() => { + cleanup(); + reject(new Error(`Timeout collecting events. Stderr: ${this.stderr}`)); + }, timeout); + } + }); + return { promise, cancel }; + } + private handleLine(line: string): void { try { const data = JSON.parse(line); @@ -627,7 +675,8 @@ export class RpcClient { } } - private async send(command: RpcCommandBody, timeoutMs = 30000): Promise { + private async send(command: RpcCommandBody): Promise { + if (this.transportError) throw this.transportError; if (!this.process?.stdin) { throw new Error("Client not started"); } @@ -637,27 +686,21 @@ export class RpcClient { return new Promise((resolve, reject) => { this.pendingRequests.set(id, { resolve, reject }); - - const timeout = setTimeout(() => { - this.pendingRequests.delete(id); - reject(new Error(`Timeout waiting for response to ${command.type}. Stderr: ${this.stderr}`)); - }, timeoutMs); - - this.pendingRequests.set(id, { - resolve: (response) => { - clearTimeout(timeout); - resolve(response); - }, - reject: (error) => { - clearTimeout(timeout); - reject(error); - }, - }); - this.process!.stdin!.write(serializeJsonLine(fullCommand)); }); } + private failPendingOperations(error: Error): void { + this.transportError ??= error; + for (const [id, pending] of this.pendingRequests) { + pending.reject(this.transportError); + this.pendingRequests.delete(id); + } + for (const reject of [...this.pendingEventWaiters]) { + reject(this.transportError); + } + } + private getData(response: RpcResponse): T { if (!response.success) { const errorResponse = response as Extract; diff --git a/packages/coding-agent/test/fixtures/rpc-client-hanging-fixture.mjs b/packages/coding-agent/test/fixtures/rpc-client-hanging-fixture.mjs new file mode 100644 index 0000000000..72ca140967 --- /dev/null +++ b/packages/coding-agent/test/fixtures/rpc-client-hanging-fixture.mjs @@ -0,0 +1 @@ +process.stdin.resume(); diff --git a/packages/coding-agent/test/rpc-client-refine.test.ts b/packages/coding-agent/test/rpc-client-refine.test.ts deleted file mode 100644 index 098293bef8..0000000000 --- a/packages/coding-agent/test/rpc-client-refine.test.ts +++ /dev/null @@ -1,33 +0,0 @@ -import { describe, expect, it, vi } from "vitest"; -import { REFINE_REQUEST_TIMEOUT_MS, RpcClient } from "../src/modes/rpc/rpc-client.js"; - -type RpcClientPrivate = { - send: (command: { type: string }, timeoutMs?: number) => Promise; - getData: (response: unknown) => T; -}; - -describe("RpcClient refine", () => { - it("sends the refine command with the extended timeout", async () => { - const client = new RpcClient(); - const privateClient = client as unknown as RpcClientPrivate; - const send = vi.fn(async () => ({ - type: "response", - command: "refine", - success: true, - data: { id: "refine_1", appliedEdits: [], harnessStatePath: "/tmp/harness_state.json" }, - })); - privateClient.send = send; - privateClient.getData = (response: unknown): T => { - return (response as { data: T }).data; - }; - - const result = await client.refine({ instructions: "tighten validation" }); - - expect(send).toHaveBeenCalledWith( - { type: "refine", instructions: "tighten validation" }, - REFINE_REQUEST_TIMEOUT_MS, - ); - expect(REFINE_REQUEST_TIMEOUT_MS).toBeGreaterThan(30000); - expect(result).toMatchObject({ id: "refine_1" }); - }); -}); diff --git a/packages/coding-agent/test/rpc-client-timeout.test.ts b/packages/coding-agent/test/rpc-client-timeout.test.ts new file mode 100644 index 0000000000..f505a53b2c --- /dev/null +++ b/packages/coding-agent/test/rpc-client-timeout.test.ts @@ -0,0 +1,107 @@ +import { fileURLToPath } from "node:url"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { RpcClient } from "../src/modes/rpc/rpc-client.js"; + +const fixturePath = fileURLToPath(new URL("./fixtures/rpc-client-hanging-fixture.mjs", import.meta.url)); +const clients = new Set(); + +async function createClient(): Promise { + const client = new RpcClient({ cliPath: fixturePath }); + await client.start(); + clients.add(client); + return client; +} + +describe("RpcClient operation completion", () => { + afterEach(async () => { + vi.useRealTimers(); + await Promise.all([...clients].map((client) => client.stop())); + clients.clear(); + }); + + it("does not time out a long RPC command by default", async () => { + const client = await createClient(); + vi.useFakeTimers(); + const result = client.bash("sleep 120"); + + await vi.advanceTimersByTimeAsync(120_000); + expect(await Promise.race([result, Promise.resolve("pending")])).toBe("pending"); + + client["handleLine"]( + JSON.stringify({ + id: "req_1", + type: "response", + command: "bash", + success: true, + data: { output: "done", exitCode: 0, cancelled: false, truncated: false }, + }), + ); + await expect(result).resolves.toMatchObject({ output: "done", exitCode: 0 }); + }); + + it("rejects promptAndWait when the prompt response fails", async () => { + const client = await createClient(); + const result = expect(client.promptAndWait("rejected prompt")).rejects.toThrow("prompt rejected"); + + client["handleLine"]( + JSON.stringify({ + id: "req_1", + type: "response", + command: "prompt", + success: false, + error: "prompt rejected", + }), + ); + + await result; + expect(client["pendingEventWaiters"].size).toBe(0); + }); + + it("does not time out agent completion by default", async () => { + const client = await createClient(); + vi.useFakeTimers(); + const idle = client.waitForIdle(); + const events = client.collectEvents(); + + await vi.advanceTimersByTimeAsync(120_000); + expect(await Promise.race([idle, Promise.resolve("pending")])).toBe("pending"); + expect(await Promise.race([events, Promise.resolve("pending")])).toBe("pending"); + + client["handleLine"](JSON.stringify({ type: "agent_end" })); + await expect(idle).resolves.toBeUndefined(); + await expect(events).resolves.toEqual([{ type: "agent_end" }]); + }); + + it("waits for child close before restarting", async () => { + const client = await createClient(); + + await client.stop(); + await client.start(); + const state = client.getState(); + client["handleLine"]( + JSON.stringify({ id: "req_1", type: "response", command: "get_state", success: true, data: {} }), + ); + + await expect(state).resolves.toEqual({}); + }); + + it("rejects start when the child cannot spawn", async () => { + const client = new RpcClient({ cliPath: fixturePath, env: { PATH: "" } }); + + await expect(client.start()).rejects.toThrow("RPC process error"); + }); + + it("rejects pending commands and completion waits when the child output closes", async () => { + const client = await createClient(); + const child = client["process"]; + if (!child) throw new Error("RPC child did not start"); + const command = expect(client.getState()).rejects.toThrow("RPC process output closed"); + const idle = expect(client.waitForIdle()).rejects.toThrow("RPC process output closed"); + const events = expect(client.collectEvents()).rejects.toThrow("RPC process output closed"); + + child.kill("SIGTERM"); + + await Promise.all([command, idle, events]); + clients.delete(client); + }); +}); From c33e51692060ac1855b884d2d5da650e9bc3f4d8 Mon Sep 17 00:00:00 2001 From: Sebastian Date: Thu, 27 Aug 2026 14:05:03 +0200 Subject: [PATCH 2/8] refactor(coding-agent): await the spawn event instead of a startup sleep --- packages/coding-agent/.changes/rpc-start-readiness.md | 1 + packages/coding-agent/src/modes/rpc/rpc-client.ts | 11 +++++------ packages/coding-agent/test/rpc-client-timeout.test.ts | 8 ++++++++ 3 files changed, 14 insertions(+), 6 deletions(-) create mode 100644 packages/coding-agent/.changes/rpc-start-readiness.md diff --git a/packages/coding-agent/.changes/rpc-start-readiness.md b/packages/coding-agent/.changes/rpc-start-readiness.md new file mode 100644 index 0000000000..c2be30c87c --- /dev/null +++ b/packages/coding-agent/.changes/rpc-start-readiness.md @@ -0,0 +1 @@ +- Removed the fixed delay when starting an RPC client. diff --git a/packages/coding-agent/src/modes/rpc/rpc-client.ts b/packages/coding-agent/src/modes/rpc/rpc-client.ts index fa3ea7de1a..007f2ea348 100644 --- a/packages/coding-agent/src/modes/rpc/rpc-client.ts +++ b/packages/coding-agent/src/modes/rpc/rpc-client.ts @@ -5,6 +5,7 @@ */ import { type ChildProcess, spawn } from "node:child_process"; +import { once } from "node:events"; import type { AgentEvent, AgentMessage, ThinkingLevel } from "@earendil-works/pi-agent-core"; import type { ImageContent } from "@earendil-works/pi-ai"; import type { AgentSessionMessageReceipt, AgentSessionMessageSafetyStatus } from "../../core/agent-messages.js"; @@ -135,12 +136,10 @@ export class RpcClient { this.handleLine(line); }); - // Wait a moment for process to initialize - await new Promise((resolve) => setTimeout(resolve, 100)); - - if (this.transportError) throw this.transportError; - if (child.exitCode !== null) { - throw new Error(`Agent process exited immediately with code ${child.exitCode}. Stderr: ${this.stderr}`); + try { + await once(child, "spawn"); + } catch (error) { + throw this.transportError ?? error; } } diff --git a/packages/coding-agent/test/rpc-client-timeout.test.ts b/packages/coding-agent/test/rpc-client-timeout.test.ts index f505a53b2c..b75d30cbc0 100644 --- a/packages/coding-agent/test/rpc-client-timeout.test.ts +++ b/packages/coding-agent/test/rpc-client-timeout.test.ts @@ -85,6 +85,14 @@ describe("RpcClient operation completion", () => { await expect(state).resolves.toEqual({}); }); + it("starts from the child spawn signal without waiting for a timer", async () => { + vi.useFakeTimers(); + const client = new RpcClient({ cliPath: fixturePath }); + clients.add(client); + + await expect(client.start()).resolves.toBeUndefined(); + }); + it("rejects start when the child cannot spawn", async () => { const client = new RpcClient({ cliPath: fixturePath, env: { PATH: "" } }); From 8d416f4bc3793b0f7dc828db8eda4ef838878827 Mon Sep 17 00:00:00 2001 From: Sebastian Date: Thu, 27 Aug 2026 15:05:11 +0200 Subject: [PATCH 3/8] fix(coding-agent): make RpcClient.stop() exit-based so held stdio pipes cannot hang shutdown stop() waited for the child 'close' event with no fallback: a grandchild inheriting the stdio pipes (or a child that never dies) kept 'close' from firing, hanging stop() forever and leaving this.process set so a later start() threw 'Client already started'. Wait on 'exit' instead, resolve after the SIGKILL escalation as a fallback, and clear this.process explicitly (guarded against a restarted child). --- .../coding-agent/src/modes/rpc/rpc-client.ts | 14 ++++++++--- .../fixtures/rpc-client-hanging-fixture.mjs | 13 ++++++++++ .../test/rpc-client-timeout.test.ts | 25 +++++++++++++++++++ 3 files changed, 48 insertions(+), 4 deletions(-) diff --git a/packages/coding-agent/src/modes/rpc/rpc-client.ts b/packages/coding-agent/src/modes/rpc/rpc-client.ts index fa3ea7de1a..e8a56a9452 100644 --- a/packages/coding-agent/src/modes/rpc/rpc-client.ts +++ b/packages/coding-agent/src/modes/rpc/rpc-client.ts @@ -120,8 +120,9 @@ export class RpcClient { child.stdout?.on("close", () => { this.failPendingOperations(new Error(`RPC process output closed. Stderr: ${this.stderr}`)); }); - child.on("close", () => { - this.process = null; + child.on("exit", () => { + // "exit" (not "close"): a grandchild holding the stdio pipes must not block cleanup. + if (this.process === child) this.process = null; }); // Collect stderr for debugging @@ -155,13 +156,18 @@ export class RpcClient { this.stopReadingStdout = null; this.failPendingOperations(new Error(`RPC client stopped. Stderr: ${this.stderr}`)); await new Promise((resolve) => { - const timeout = setTimeout(() => child.kill("SIGKILL"), 1000); - child.once("close", () => { + // Resolve after SIGKILL as a fallback so stop() cannot hang on a child that never exits. + const timeout = setTimeout(() => { + child.kill("SIGKILL"); + resolve(); + }, 1000); + child.once("exit", () => { clearTimeout(timeout); resolve(); }); child.kill("SIGTERM"); }); + this.process = null; } /** diff --git a/packages/coding-agent/test/fixtures/rpc-client-hanging-fixture.mjs b/packages/coding-agent/test/fixtures/rpc-client-hanging-fixture.mjs index 72ca140967..8d71496d6d 100644 --- a/packages/coding-agent/test/fixtures/rpc-client-hanging-fixture.mjs +++ b/packages/coding-agent/test/fixtures/rpc-client-hanging-fixture.mjs @@ -1 +1,14 @@ +import { spawn } from "node:child_process"; + +if (process.env.RPC_FIXTURE_HOLD_STDIO === "1") { + // A grandchild inheriting the stdio pipes keeps them open after this process exits, + // so the parent RpcClient never sees a "close" event for this child. + const grandchild = spawn(process.execPath, ["-e", "setTimeout(() => {}, 30000)"], { + stdio: "inherit", + detached: true, + }); + grandchild.unref(); + process.stdout.write(`${JSON.stringify({ type: "fixture_grandchild", pid: grandchild.pid })}\n`); +} + process.stdin.resume(); diff --git a/packages/coding-agent/test/rpc-client-timeout.test.ts b/packages/coding-agent/test/rpc-client-timeout.test.ts index f505a53b2c..a040b7a151 100644 --- a/packages/coding-agent/test/rpc-client-timeout.test.ts +++ b/packages/coding-agent/test/rpc-client-timeout.test.ts @@ -85,6 +85,31 @@ describe("RpcClient operation completion", () => { await expect(state).resolves.toEqual({}); }); + it("stop resolves and allows restart when a grandchild holds the stdio pipes", async () => { + const client = new RpcClient({ cliPath: fixturePath, env: { RPC_FIXTURE_HOLD_STDIO: "1" } }); + const grandchildPids: number[] = []; + client.onEvent((event) => { + const data = event as unknown as { type: string; pid?: number }; + if (data.type === "fixture_grandchild" && typeof data.pid === "number") grandchildPids.push(data.pid); + }); + try { + await client.start(); + await vi.waitFor(() => expect(grandchildPids).toHaveLength(1)); + await client.stop(); + await client.start(); + await vi.waitFor(() => expect(grandchildPids).toHaveLength(2)); + await client.stop(); + } finally { + for (const pid of grandchildPids) { + try { + process.kill(pid, "SIGKILL"); + } catch { + // Grandchild already exited. + } + } + } + }); + it("rejects start when the child cannot spawn", async () => { const client = new RpcClient({ cliPath: fixturePath, env: { PATH: "" } }); From 7a05456edbce759cb250b1609e1eb462e27c055a Mon Sep 17 00:00:00 2001 From: Sebastian Date: Thu, 27 Aug 2026 15:41:22 +0200 Subject: [PATCH 4/8] fix(coding-agent): scope RPC child lifecycle handlers to the active child Late error/stdout-close events from a replaced child could set transportError on a freshly restarted client, rejecting its pending work. Guard every child handler with this.process === child, fail pending operations on 'exit' so a grandchild holding stdout cannot leave requests and idle waiters hanging, and finalize on 'close' as well so a failed spawn (which emits 'error'/'close' but never 'exit') clears this.process and start() can be retried. --- .../coding-agent/src/modes/rpc/rpc-client.ts | 17 ++++++-- .../test/rpc-client-timeout.test.ts | 41 +++++++++++++++++-- 2 files changed, 51 insertions(+), 7 deletions(-) diff --git a/packages/coding-agent/src/modes/rpc/rpc-client.ts b/packages/coding-agent/src/modes/rpc/rpc-client.ts index e8a56a9452..42e7e24a05 100644 --- a/packages/coding-agent/src/modes/rpc/rpc-client.ts +++ b/packages/coding-agent/src/modes/rpc/rpc-client.ts @@ -114,16 +114,25 @@ export class RpcClient { stdio: ["pipe", "pipe", "pipe"], }); this.process = child; + // All handlers are scoped to this child so late events from a replaced child + // cannot poison a restarted client. child.on("error", (error) => { + if (this.process !== child) return; this.failPendingOperations(new Error(`RPC process error: ${error.message}. Stderr: ${this.stderr}`)); }); child.stdout?.on("close", () => { + if (this.process !== child) return; this.failPendingOperations(new Error(`RPC process output closed. Stderr: ${this.stderr}`)); }); - child.on("exit", () => { - // "exit" (not "close"): a grandchild holding the stdio pipes must not block cleanup. - if (this.process === child) this.process = null; - }); + // "exit" so a grandchild holding the stdio pipes cannot block cleanup; "close" as + // the fallback for failed spawns, where "exit" never fires. + const finalize = () => { + if (this.process !== child) return; + this.failPendingOperations(new Error(`RPC process exited. Stderr: ${this.stderr}`)); + this.process = null; + }; + child.on("exit", finalize); + child.on("close", finalize); // Collect stderr for debugging child.stderr?.on("data", (data) => { diff --git a/packages/coding-agent/test/rpc-client-timeout.test.ts b/packages/coding-agent/test/rpc-client-timeout.test.ts index a040b7a151..2d94ac6292 100644 --- a/packages/coding-agent/test/rpc-client-timeout.test.ts +++ b/packages/coding-agent/test/rpc-client-timeout.test.ts @@ -98,6 +98,16 @@ describe("RpcClient operation completion", () => { await client.stop(); await client.start(); await vi.waitFor(() => expect(grandchildPids).toHaveLength(2)); + // Release the old child's stdio pipes: its late "close" must not poison + // the restarted client. + process.kill(grandchildPids[0], "SIGKILL"); + await vi.waitFor(() => expect(() => process.kill(grandchildPids[0], 0)).toThrow()); + await new Promise((resolve) => setTimeout(resolve, 25)); + const state = client.getState(); + client["handleLine"]( + JSON.stringify({ id: "req_1", type: "response", command: "get_state", success: true, data: {} }), + ); + await expect(state).resolves.toEqual({}); await client.stop(); } finally { for (const pid of grandchildPids) { @@ -114,15 +124,40 @@ describe("RpcClient operation completion", () => { const client = new RpcClient({ cliPath: fixturePath, env: { PATH: "" } }); await expect(client.start()).rejects.toThrow("RPC process error"); + // The failed child is cleaned up, so a retry spawns again instead of + // throwing "Client already started". + await expect(client.start()).rejects.toThrow("RPC process error"); + }); + + it("rejects pending work when the child exits while a grandchild holds stdout", async () => { + const client = new RpcClient({ cliPath: fixturePath, env: { RPC_FIXTURE_HOLD_STDIO: "1" } }); + let grandchildPid: number | undefined; + client.onEvent((event) => { + const data = event as unknown as { type: string; pid?: number }; + if (data.type === "fixture_grandchild") grandchildPid = data.pid; + }); + await client.start(); + await vi.waitFor(() => expect(grandchildPid).toBeDefined()); + const child = client["process"]; + if (!child) throw new Error("RPC child did not start"); + try { + const command = expect(client.getState()).rejects.toThrow("RPC process exited"); + const idle = expect(client.waitForIdle()).rejects.toThrow("RPC process exited"); + child.kill("SIGKILL"); + await Promise.all([command, idle]); + } finally { + if (grandchildPid) process.kill(grandchildPid, "SIGKILL"); + } }); it("rejects pending commands and completion waits when the child output closes", async () => { const client = await createClient(); const child = client["process"]; if (!child) throw new Error("RPC child did not start"); - const command = expect(client.getState()).rejects.toThrow("RPC process output closed"); - const idle = expect(client.waitForIdle()).rejects.toThrow("RPC process output closed"); - const events = expect(client.collectEvents()).rejects.toThrow("RPC process output closed"); + const died = /RPC process (exited|output closed)/; + const command = expect(client.getState()).rejects.toThrow(died); + const idle = expect(client.waitForIdle()).rejects.toThrow(died); + const events = expect(client.collectEvents()).rejects.toThrow(died); child.kill("SIGTERM"); From 887f0de7e1a12b1054ed2a3608bf22ea22e2c174 Mon Sep 17 00:00:00 2001 From: Sebastian Date: Thu, 27 Aug 2026 15:44:12 +0200 Subject: [PATCH 5/8] fix(coding-agent): clear this.process when the spawn await rejects so start() can be retried once(child, 'spawn') rejects on the child 'error' event, which for a failed spawn fires before the 'close' event where the finalizer clears this.process; the rejection therefore won the race and a retried start() threw 'Client already started'. Clear this.process in the rejection path: an error before 'spawn' can only mean the child never came up, so no live process is abandoned, and the finalizer's process-identity guard makes the later 'close' a no-op. --- packages/coding-agent/src/modes/rpc/rpc-client.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/packages/coding-agent/src/modes/rpc/rpc-client.ts b/packages/coding-agent/src/modes/rpc/rpc-client.ts index f3c49a2950..2d91af756b 100644 --- a/packages/coding-agent/src/modes/rpc/rpc-client.ts +++ b/packages/coding-agent/src/modes/rpc/rpc-client.ts @@ -149,6 +149,8 @@ export class RpcClient { try { await once(child, "spawn"); } catch (error) { + // An error before "spawn" means the child never came up; allow retrying start(). + if (this.process === child) this.process = null; throw this.transportError ?? error; } } From f45c05855c7a8e87615637cb0b02a9a69e9e58af Mon Sep 17 00:00:00 2001 From: Sebastian Date: Thu, 27 Aug 2026 16:08:01 +0200 Subject: [PATCH 6/8] fix(coding-agent): drain buffered stdout before failing pending ops on child exit Node may emit 'exit' while a complete response is still buffered in the stdout pipe (docs guarantee only that 'close' runs after the streams drain), so the exit finalizer could reject a request whose answer was already in flight. Defer failPendingOperations until stdout closes, with a 1s unref'd fallback so pipes held open by a grandchild still fail promptly, and skip the deferred failure if the client was restarted. --- .../coding-agent/src/modes/rpc/rpc-client.ts | 19 ++++++++++++++++++- .../fixtures/rpc-client-hanging-fixture.mjs | 10 ++++++++++ .../test/rpc-client-timeout.test.ts | 6 ++++++ 3 files changed, 34 insertions(+), 1 deletion(-) diff --git a/packages/coding-agent/src/modes/rpc/rpc-client.ts b/packages/coding-agent/src/modes/rpc/rpc-client.ts index 42e7e24a05..12d6cc48db 100644 --- a/packages/coding-agent/src/modes/rpc/rpc-client.ts +++ b/packages/coding-agent/src/modes/rpc/rpc-client.ts @@ -128,8 +128,25 @@ export class RpcClient { // the fallback for failed spawns, where "exit" never fires. const finalize = () => { if (this.process !== child) return; - this.failPendingOperations(new Error(`RPC process exited. Stderr: ${this.stderr}`)); this.process = null; + const fail = () => { + // A client restarted meanwhile must not be poisoned by its predecessor. + if (this.process) return; + this.failPendingOperations(new Error(`RPC process exited. Stderr: ${this.stderr}`)); + }; + const stdout = child.stdout; + if (!stdout || stdout.readableEnded || stdout.destroyed) { + fail(); + return; + } + // Let buffered stdout drain so a response already in the pipe resolves instead + // of rejecting; the timer covers pipes a grandchild keeps open past the exit. + const timer = setTimeout(fail, 1000); + timer.unref?.(); + stdout.once("close", () => { + clearTimeout(timer); + fail(); + }); }; child.on("exit", finalize); child.on("close", finalize); diff --git a/packages/coding-agent/test/fixtures/rpc-client-hanging-fixture.mjs b/packages/coding-agent/test/fixtures/rpc-client-hanging-fixture.mjs index 8d71496d6d..5cb2f2dab0 100644 --- a/packages/coding-agent/test/fixtures/rpc-client-hanging-fixture.mjs +++ b/packages/coding-agent/test/fixtures/rpc-client-hanging-fixture.mjs @@ -11,4 +11,14 @@ if (process.env.RPC_FIXTURE_HOLD_STDIO === "1") { process.stdout.write(`${JSON.stringify({ type: "fixture_grandchild", pid: grandchild.pid })}\n`); } +if (process.env.RPC_FIXTURE_REPLY_EXIT === "1") { + // Answer the first command, then die immediately: the response is still in the + // pipe (or draining) when "exit" reaches the parent. + process.stdin.once("data", (chunk) => { + const { id, type } = JSON.parse(chunk.toString()); + process.stdout.write(`${JSON.stringify({ id, type: "response", command: type, success: true, data: {} })}\n`); + process.exit(0); + }); +} + process.stdin.resume(); diff --git a/packages/coding-agent/test/rpc-client-timeout.test.ts b/packages/coding-agent/test/rpc-client-timeout.test.ts index 2d94ac6292..4bbca48586 100644 --- a/packages/coding-agent/test/rpc-client-timeout.test.ts +++ b/packages/coding-agent/test/rpc-client-timeout.test.ts @@ -129,6 +129,12 @@ describe("RpcClient operation completion", () => { await expect(client.start()).rejects.toThrow("RPC process error"); }); + it("resolves a response the child wrote just before exiting", async () => { + const client = new RpcClient({ cliPath: fixturePath, env: { RPC_FIXTURE_REPLY_EXIT: "1" } }); + await client.start(); + await expect(client.getState()).resolves.toEqual({}); + }); + it("rejects pending work when the child exits while a grandchild holds stdout", async () => { const client = new RpcClient({ cliPath: fixturePath, env: { RPC_FIXTURE_HOLD_STDIO: "1" } }); let grandchildPid: number | undefined; From f1a560e48c71fde5182495c0716fa2320cca8188 Mon Sep 17 00:00:00 2001 From: Sebastian Date: Thu, 27 Aug 2026 16:24:10 +0200 Subject: [PATCH 7/8] fix(coding-agent): cut the previous child generation on RPC client restart If start() ran inside a dead child's stdout-drain window, the deferred failure skipped (anti-poisoning guard) and the old generation's pending requests and idle waiters hung forever; the old JSONL reader also stayed attached, so late output from the dead child's pipe could resolve the new session's waiters. start() now detaches the previous reader and fails any leftover pending operations synchronously before spawning. --- .../coding-agent/src/modes/rpc/rpc-client.ts | 5 +++ .../fixtures/rpc-client-hanging-fixture.mjs | 8 +++- .../test/rpc-client-timeout.test.ts | 41 +++++++++++++++++++ 3 files changed, 53 insertions(+), 1 deletion(-) diff --git a/packages/coding-agent/src/modes/rpc/rpc-client.ts b/packages/coding-agent/src/modes/rpc/rpc-client.ts index 12d6cc48db..60e1db2d0c 100644 --- a/packages/coding-agent/src/modes/rpc/rpc-client.ts +++ b/packages/coding-agent/src/modes/rpc/rpc-client.ts @@ -107,6 +107,11 @@ export class RpcClient { args.push(...this.options.args); } + // Cut the previous generation: its reader must not feed this session, and its + // pending work must fail now instead of hanging past the restart. + this.stopReadingStdout?.(); + this.stopReadingStdout = null; + this.failPendingOperations(new Error(`RPC client restarted. Stderr: ${this.stderr}`)); this.transportError = null; const child = spawn("node", [cliPath, ...args], { cwd: this.options.cwd, diff --git a/packages/coding-agent/test/fixtures/rpc-client-hanging-fixture.mjs b/packages/coding-agent/test/fixtures/rpc-client-hanging-fixture.mjs index 5cb2f2dab0..28818ab6b9 100644 --- a/packages/coding-agent/test/fixtures/rpc-client-hanging-fixture.mjs +++ b/packages/coding-agent/test/fixtures/rpc-client-hanging-fixture.mjs @@ -3,7 +3,13 @@ import { spawn } from "node:child_process"; if (process.env.RPC_FIXTURE_HOLD_STDIO === "1") { // A grandchild inheriting the stdio pipes keeps them open after this process exits, // so the parent RpcClient never sees a "close" event for this child. - const grandchild = spawn(process.execPath, ["-e", "setTimeout(() => {}, 30000)"], { + // The ghost variant writes an event into the inherited stdout shortly after this + // process dies, emulating child output that lands after a replacement started. + const script = + process.env.RPC_FIXTURE_GHOST_EVENT === "1" + ? `process.stdin.on("end", () => setTimeout(() => process.stdout.write('{"type":"agent_end"}\\n'), 250)); process.stdin.resume(); setTimeout(() => {}, 30000);` + : "setTimeout(() => {}, 30000)"; + const grandchild = spawn(process.execPath, ["-e", script], { stdio: "inherit", detached: true, }); diff --git a/packages/coding-agent/test/rpc-client-timeout.test.ts b/packages/coding-agent/test/rpc-client-timeout.test.ts index 4bbca48586..484409c75e 100644 --- a/packages/coding-agent/test/rpc-client-timeout.test.ts +++ b/packages/coding-agent/test/rpc-client-timeout.test.ts @@ -156,6 +156,47 @@ describe("RpcClient operation completion", () => { } }); + it("fails a dead generation's pending work on restart and ignores its late output", async () => { + const client = new RpcClient({ + cliPath: fixturePath, + env: { RPC_FIXTURE_HOLD_STDIO: "1", RPC_FIXTURE_GHOST_EVENT: "1" }, + }); + const grandchildPids: number[] = []; + client.onEvent((event) => { + const data = event as unknown as { type: string; pid?: number }; + if (data.type === "fixture_grandchild" && typeof data.pid === "number") grandchildPids.push(data.pid); + }); + try { + await client.start(); + await vi.waitFor(() => expect(grandchildPids).toHaveLength(1)); + const child = client["process"]; + if (!child) throw new Error("RPC child did not start"); + const orphan = expect(client.getState()).rejects.toThrow("RPC client restarted"); + child.kill("SIGKILL"); + await vi.waitFor(() => expect(client["process"]).toBeNull()); + // Restart inside the dead child's stdout-drain window. + await client.start(); + await orphan; + const idle = client.waitForIdle(); + // The old grandchild writes an agent_end into the dead child's stdout ~250ms + // after the kill; it must not resolve the new session's waiter. + await vi.waitFor(() => expect(grandchildPids).toHaveLength(2)); + await new Promise((resolve) => setTimeout(resolve, 500)); + expect(await Promise.race([idle, Promise.resolve("pending")])).toBe("pending"); + client["handleLine"](JSON.stringify({ type: "agent_end" })); + await expect(idle).resolves.toBeUndefined(); + await client.stop(); + } finally { + for (const pid of grandchildPids) { + try { + process.kill(pid, "SIGKILL"); + } catch { + // Grandchild already exited. + } + } + } + }); + it("rejects pending commands and completion waits when the child output closes", async () => { const client = await createClient(); const child = client["process"]; From 6bf726d9a685480c45329b1bf5471be43e6d7928 Mon Sep 17 00:00:00 2001 From: Sebastian Date: Thu, 27 Aug 2026 16:36:09 +0200 Subject: [PATCH 8/8] test(coding-agent): prove ghost-event delivery via a stderr sentinel The ghost grandchild now also writes a sentinel to the inherited stderr, which still feeds the client's shared stderr accumulator after restart. The test awaits the sentinel before asserting the new session's waiter is untouched, so the still-pending assertion cannot pass vacuously if the stdin-EOF trigger ever stops firing. --- .../test/fixtures/rpc-client-hanging-fixture.mjs | 2 +- packages/coding-agent/test/rpc-client-timeout.test.ts | 5 ++++- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/packages/coding-agent/test/fixtures/rpc-client-hanging-fixture.mjs b/packages/coding-agent/test/fixtures/rpc-client-hanging-fixture.mjs index 28818ab6b9..81bf723c8f 100644 --- a/packages/coding-agent/test/fixtures/rpc-client-hanging-fixture.mjs +++ b/packages/coding-agent/test/fixtures/rpc-client-hanging-fixture.mjs @@ -7,7 +7,7 @@ if (process.env.RPC_FIXTURE_HOLD_STDIO === "1") { // process dies, emulating child output that lands after a replacement started. const script = process.env.RPC_FIXTURE_GHOST_EVENT === "1" - ? `process.stdin.on("end", () => setTimeout(() => process.stdout.write('{"type":"agent_end"}\\n'), 250)); process.stdin.resume(); setTimeout(() => {}, 30000);` + ? `process.stdin.on("end", () => setTimeout(() => { process.stdout.write('{"type":"agent_end"}\\n'); process.stderr.write("ghost-event-written\\n"); }, 250)); process.stdin.resume(); setTimeout(() => {}, 30000);` : "setTimeout(() => {}, 30000)"; const grandchild = spawn(process.execPath, ["-e", script], { stdio: "inherit", diff --git a/packages/coding-agent/test/rpc-client-timeout.test.ts b/packages/coding-agent/test/rpc-client-timeout.test.ts index 484409c75e..146ec68f39 100644 --- a/packages/coding-agent/test/rpc-client-timeout.test.ts +++ b/packages/coding-agent/test/rpc-client-timeout.test.ts @@ -181,7 +181,10 @@ describe("RpcClient operation completion", () => { // The old grandchild writes an agent_end into the dead child's stdout ~250ms // after the kill; it must not resolve the new session's waiter. await vi.waitFor(() => expect(grandchildPids).toHaveLength(2)); - await new Promise((resolve) => setTimeout(resolve, 500)); + // The stderr sentinel proves the ghost's stdout write really happened, so the + // still-pending assertion below cannot pass vacuously. + await vi.waitFor(() => expect(client.getStderr()).toContain("ghost-event-written"), { timeout: 3000 }); + await new Promise((resolve) => setTimeout(resolve, 50)); expect(await Promise.race([idle, Promise.resolve("pending")])).toBe("pending"); client["handleLine"](JSON.stringify({ type: "agent_end" })); await expect(idle).resolves.toBeUndefined();