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 00000000000..b9e15b8d33a --- /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 7c327881647..60e1db2d0cb 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,72 @@ export class RpcClient { args.push(...this.options.args); } - this.process = spawn("node", [cliPath, ...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, env: { ...process.env, ...this.options.env }, 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}`)); + }); + // "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.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); // 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 +180,25 @@ 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) => { + // Resolve after SIGKILL as a fallback so stop() cannot hang on a child that never exits. const timeout = setTimeout(() => { - this.process?.kill("SIGKILL"); + child.kill("SIGKILL"); resolve(); }, 1000); - - this.process?.on("exit", () => { + child.once("exit", () => { clearTimeout(timeout); resolve(); }); + child.kill("SIGTERM"); }); - this.process = null; - this.pendingRequests.clear(); } /** @@ -195,7 +241,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 +355,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 +364,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 +585,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 +712,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 +723,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 00000000000..81bf723c8f7 --- /dev/null +++ b/packages/coding-agent/test/fixtures/rpc-client-hanging-fixture.mjs @@ -0,0 +1,30 @@ +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. + // 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'); process.stderr.write("ghost-event-written\\n"); }, 250)); process.stdin.resume(); setTimeout(() => {}, 30000);` + : "setTimeout(() => {}, 30000)"; + const grandchild = spawn(process.execPath, ["-e", script], { + stdio: "inherit", + detached: true, + }); + grandchild.unref(); + 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-refine.test.ts b/packages/coding-agent/test/rpc-client-refine.test.ts deleted file mode 100644 index 098293bef8d..00000000000 --- 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 00000000000..146ec68f397 --- /dev/null +++ b/packages/coding-agent/test/rpc-client-timeout.test.ts @@ -0,0 +1,217 @@ +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("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)); + // 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) { + 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: "" } }); + + 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("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; + 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("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)); + // 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(); + 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"]; + if (!child) throw new Error("RPC child did not start"); + 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"); + + await Promise.all([command, idle, events]); + clients.delete(client); + }); +});