diff --git a/packages/server/src/server/agent/providers/jsonl-rpc-process.test.ts b/packages/server/src/server/agent/providers/jsonl-rpc-process.test.ts index 2b3e88659a..62b6c9ead9 100644 --- a/packages/server/src/server/agent/providers/jsonl-rpc-process.test.ts +++ b/packages/server/src/server/agent/providers/jsonl-rpc-process.test.ts @@ -268,6 +268,41 @@ describe("JsonlRpcProcess", () => { }, ); + test("requestStopWork resolves once the child has exited", async () => { + const child = createInMemoryChildProcess(); + const transport = startProcess({ child }); + + child.emit("exit", 1, null); + + await expect(transport.requestStopWork({ type: "abort" })).resolves.toBeUndefined(); + }); + + // A dying child's stdin pipe breaks a fraction before its exit event, so deciding at the + // moment of the write failure would refuse a stop the runtime did honor. + test("requestStopWork resolves when stdin fails just before the child exits", async () => { + const child = createInMemoryChildProcess(); + const transport = startProcess({ child }); + + const stopped = transport.requestStopWork({ type: "abort" }); + child.stdin.emit("error", Object.assign(new Error("write EPIPE"), { code: "EPIPE" })); + + await expect(stopped).resolves.toBeUndefined(); + }); + + // `close()` and a spawn error dispose the transport before the child is known to be + // gone, and a child can outlive SIGKILL, so a disposed transport is not proof that the + // requested work stopped. + test("requestStopWork keeps failing while the transport is disposed but the child has not exited", async () => { + const child = createInMemoryChildProcess(); + const transport = startProcess({ child }); + + child.emit("error", new Error("spawn failed")); + + await expect(transport.requestStopWork({ type: "abort" })).rejects.toThrow( + "JSONL RPC process is closed", + ); + }); + test("stdin error events close the transport instead of becoming uncaught exceptions", async () => { const child = createInMemoryChildProcess(); const transport = startProcess({ child }); diff --git a/packages/server/src/server/agent/providers/jsonl-rpc-process.ts b/packages/server/src/server/agent/providers/jsonl-rpc-process.ts index 791df2500a..d58e2235ca 100644 --- a/packages/server/src/server/agent/providers/jsonl-rpc-process.ts +++ b/packages/server/src/server/agent/providers/jsonl-rpc-process.ts @@ -81,6 +81,8 @@ export class JsonlRpcProcess { private stderrBuffer = ""; private nextRequestId = 1; private disposed = false; + private exited = false; + private closing: Promise | null = null; private readonly frameDecoder: JsonlFrameDecoder; constructor(private readonly options: JsonlRpcProcessOptions) { @@ -111,6 +113,7 @@ export class JsonlRpcProcess { this.failAll(error instanceof Error ? error : new Error(String(error))); }); this.child.on("exit", (code, signal) => { + this.exited = true; const error = new Error( `${this.diagnosticName} process exited with code ${code ?? "null"} and signal ${signal ?? "null"}\n${this.stderrBuffer}`.trim(), ); @@ -175,6 +178,36 @@ export class JsonlRpcProcess { return this.startRequest(command, timeoutMs).promise; } + /** + * Send a command whose goal an exited process has already met — stopping a turn, + * clearing a queue. Resolves once the child has exited instead of rejecting, + * because nothing is queued and nothing is running, which is what the caller + * asked for. Use `request` when the caller needs an answer from a live process. + * + * Keyed on an observed exit rather than on `disposed`: `close()` disposes the + * transport before termination completes, and a child that outlives SIGKILL can + * still be working, so a disposed transport is not proof the work stopped. + */ + async requestStopWork( + command: { type: string; [key: string]: unknown }, + timeoutMs?: number | null, + ): Promise { + if (this.exited) { + return; + } + try { + await this.request(command, timeoutMs); + } catch (error) { + // A dying child breaks its stdin pipe a fraction before it emits `exit`, which + // fails this request and starts a termination. Wait for that termination to reach + // an answer rather than refusing a stop the runtime went on to honor. + await this.closing?.catch(() => undefined); + if (!this.exited) { + throw error; + } + } + } + send(message: Record): void { if (this.disposed) { return; @@ -193,6 +226,11 @@ export class JsonlRpcProcess { async close(error = new Error(`${this.diagnosticName} process is closed`)): Promise { if (this.disposed) return; this.failAll(error); + this.closing = this.terminate(); + await this.closing; + } + + private async terminate(): Promise { try { this.child.stdin.end(); } catch { diff --git a/packages/server/src/server/agent/providers/omp/cli-runtime.test.ts b/packages/server/src/server/agent/providers/omp/cli-runtime.test.ts index 9df9da68dd..cb8bd13b51 100644 --- a/packages/server/src/server/agent/providers/omp/cli-runtime.test.ts +++ b/packages/server/src/server/agent/providers/omp/cli-runtime.test.ts @@ -66,10 +66,7 @@ function createRuntime( }); } -function replyToCommands( - child: OmpChild, - handler: (command: Record) => unknown, -): void { +function onOmpCommand(child: OmpChild, handler: (command: Record) => void): void { let buffer = ""; child.stdin.on("data", (chunk) => { buffer += chunk.toString(); @@ -78,17 +75,34 @@ function replyToCommands( if (newlineIndex === -1) break; const line = buffer.slice(0, newlineIndex); buffer = buffer.slice(newlineIndex + 1); - const command = JSON.parse(line) as Record; - const result = handler(command); - child.stdout.write( - `${JSON.stringify({ - id: command.id, - type: "response", - command: command.type, - success: true, - data: result, - })}\n`, - ); + handler(JSON.parse(line) as Record); + } + }); +} + +function replyToCommands( + child: OmpChild, + handler: (command: Record) => unknown, +): void { + onOmpCommand(child, (command) => { + const result = handler(command); + child.stdout.write( + `${JSON.stringify({ + id: command.id, + type: "response", + command: command.type, + success: true, + data: result, + })}\n`, + ); + }); +} + +/** Kill the child the moment it receives `type`, so the request is in flight when it dies. */ +function exitOnCommand(child: OmpChild, type: string): void { + onOmpCommand(child, (command) => { + if (command.type === type) { + child.emit("exit", 1, null); } }); } @@ -418,4 +432,24 @@ describe("OMP CLI runtime", () => { expect(result).toHaveLength(2_000); expect(result[0]).toEqual(expect.objectContaining({ id: "model-0" })); }); + + // A dead runtime owns no turn, so aborting it is already satisfied. Rejecting here + // makes AgentManager treat the interrupt as unacknowledged and refuse the stop, which + // pins the agent at `running` until the daemon restarts. See issue #3749. + test("abort resolves once the OMP process has exited", async () => { + const child = createOmpChild(); + const session = await createRuntime(child).startSession({ cwd: "/workspace/project" }); + + child.emit("exit", 1, null); + + await expect(session.abort()).resolves.toBeUndefined(); + }); + + test("abort resolves when the OMP process exits while the abort is in flight", async () => { + const child = createOmpChild(); + const session = await createRuntime(child).startSession({ cwd: "/workspace/project" }); + exitOnCommand(child, "abort"); + + await expect(session.abort()).resolves.toBeUndefined(); + }); }); diff --git a/packages/server/src/server/agent/providers/omp/cli-runtime.ts b/packages/server/src/server/agent/providers/omp/cli-runtime.ts index 2ec98fb9b6..06e5158a03 100644 --- a/packages/server/src/server/agent/providers/omp/cli-runtime.ts +++ b/packages/server/src/server/agent/providers/omp/cli-runtime.ts @@ -155,7 +155,7 @@ class OmpCliRuntimeSession implements OmpRuntimeSession { } async abort(): Promise { - await this.request({ type: "abort" }); + await this.requestStopWork({ type: "abort" }); } async getState(): Promise { @@ -292,6 +292,10 @@ class OmpCliRuntimeSession implements OmpRuntimeSession { return this.process.request(OmpRpcCommandSchema.parse(command), timeoutMs); } + private requestStopWork(command: OmpRpcCommand): Promise { + return this.process.requestStopWork(OmpRpcCommandSchema.parse(command)); + } + private emit(event: OmpRuntimeEvent): void { for (const subscriber of this.subscribers) { subscriber(event); diff --git a/packages/server/src/server/agent/providers/pi/cli-runtime.test.ts b/packages/server/src/server/agent/providers/pi/cli-runtime.test.ts index 94517bde0d..c178bb5405 100644 --- a/packages/server/src/server/agent/providers/pi/cli-runtime.test.ts +++ b/packages/server/src/server/agent/providers/pi/cli-runtime.test.ts @@ -62,6 +62,15 @@ function onPiCommand(child: PiChild, handler: (command: Record) }); } +/** Kill the child the moment it receives `type`, so the request is in flight when it dies. */ +function exitOnCommand(child: PiChild, type: string): void { + onPiCommand(child, (command) => { + if (command.type === type) { + child.emit("exit", 1, null); + } + }); +} + function replyToCommands( child: PiChild, handler: (command: Record) => unknown, @@ -506,4 +515,25 @@ describe("PiCliRuntime", () => { // Neither RPC returned usable data — should resolve with empty object expect(stats).toEqual({}); }); + + // A dead runtime owns no turn, so interrupting it is already satisfied. Rejecting here + // makes AgentManager treat the interrupt as unacknowledged and refuse the stop, which + // pins the agent at `running` until the daemon restarts. See issue #3749. + test("abort and clearQueue resolve once the Pi process has exited", async () => { + const child = createPiChild(); + const session = await createRuntime(child).startSession({ cwd: "/workspace/project" }); + + child.emit("exit", 1, null); + + await expect(session.clearQueue()).resolves.toBeUndefined(); + await expect(session.abort()).resolves.toBeUndefined(); + }); + + test("abort resolves when the Pi process exits while the abort is in flight", async () => { + const child = createPiChild(); + const session = await createRuntime(child).startSession({ cwd: "/workspace/project" }); + exitOnCommand(child, "abort"); + + await expect(session.abort()).resolves.toBeUndefined(); + }); }); diff --git a/packages/server/src/server/agent/providers/pi/cli-runtime.ts b/packages/server/src/server/agent/providers/pi/cli-runtime.ts index 783be30249..0262ac13a2 100644 --- a/packages/server/src/server/agent/providers/pi/cli-runtime.ts +++ b/packages/server/src/server/agent/providers/pi/cli-runtime.ts @@ -133,7 +133,7 @@ class PiCliRuntimeSession implements PiRuntimeSession { } async clearQueue(): Promise { - await this.request({ type: "clear_queue" }); + await this.requestStopWork({ type: "clear_queue" }); } async compact(customInstructions?: string): Promise { @@ -148,7 +148,7 @@ class PiCliRuntimeSession implements PiRuntimeSession { } async abort(): Promise { - await this.request({ type: "abort" }); + await this.requestStopWork({ type: "abort" }); } async getState(): Promise { @@ -237,6 +237,10 @@ class PiCliRuntimeSession implements PiRuntimeSession { return this.process.request(command, timeoutMs); } + private requestStopWork(command: PiRpcCommand): Promise { + return this.process.requestStopWork(command); + } + private async waitForCompletion(command: PiRpcCommand): Promise { // Pi only replies after its compaction work is durable. Its child process and // session close paths already reject pending RPCs, so no elapsed-time failure