Let Stop settle an agent whose Pi or OMP runtime has exited (#5235)

* Let Stop settle an agent whose Pi or OMP runtime has exited

AgentSession.interrupt() is documented to reject "only when foreground
ownership is still uncertain", and to be a successful no-op when the
session is already idle. The Pi and OMP sessions broke that: interrupt()
issues `abort` (and `clear_queue` for Pi) over the JSONL RPC transport,
and once the child process is gone every request rejects with
"<name> process is closed".

AgentManager reads that rejection as "the provider still owns the turn",
returns `refused`, and leaves the tracked run in place. The user gets
"Cannot stop agent <id> because its active run cancellation was not
acknowledged", the agent stays pinned at `running`, and pressing Stop
again repeats it forever — only a daemon restart clears it.

A dead runtime owns nothing: no turn is running and no prompt is queued,
which is exactly the postcondition abort and clear_queue promise. Route
both through JsonlRpcProcess.requestStopWork, which resolves when the
process is gone and still surfaces real failures from a live process.
The check covers the process dying while the request is in flight, so a
race cannot reintroduce the wedge.

* Require an observed child exit before treating stop work as done

`disposed` marks the transport finished, not the child gone: `close()`
disposes before termination completes, a spawn error disposes without
touching the child, and a child can outlive SIGKILL. Keying
requestStopWork on it could report a turn canceled while the child was
still working.

Track the child's exit event instead, and resolve only on that. A
disposed transport whose child has not exited still rejects, so the
caller keeps refusing a cancellation it cannot prove.

Also move the in-flight tests' command detection into named
`exitOnCommand` helpers, and reuse the extracted `onOmpCommand` for
`replyToCommands` so the OMP file matches the Pi one.

* Let a stop survive an stdin failure that precedes the child's exit

A dying child breaks its stdin pipe a fraction before it emits `exit`.
The broken pipe fails the in-flight abort and starts a termination, and
deciding at that moment saw `exited` still false, so the stop was
refused for a runtime that was already on its way out. That is the
reported wedge again, reached through a narrower race.

close() now records the termination it starts, and requestStopWork waits
for it before deciding. A child that goes on to exit resolves the stop; a
child that outlives SIGKILL still rejects it.

---------

Co-authored-by: paseo-bot[bot] <paseo-bot[bot]@users.noreply.github.com>
This commit is contained in:
paseo-bot[bot]
2026-09-23 04:45:08 +02:00
committed by GitHub
co-authored by paseo-bot[bot]
parent 290306fd16
commit 81ad00790f
6 changed files with 163 additions and 18 deletions
@@ -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 });
@@ -81,6 +81,8 @@ export class JsonlRpcProcess {
private stderrBuffer = "";
private nextRequestId = 1;
private disposed = false;
private exited = false;
private closing: Promise<void> | 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<void> {
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<string, unknown>): void {
if (this.disposed) {
return;
@@ -193,6 +226,11 @@ export class JsonlRpcProcess {
async close(error = new Error(`${this.diagnosticName} process is closed`)): Promise<void> {
if (this.disposed) return;
this.failAll(error);
this.closing = this.terminate();
await this.closing;
}
private async terminate(): Promise<void> {
try {
this.child.stdin.end();
} catch {
@@ -66,10 +66,7 @@ function createRuntime(
});
}
function replyToCommands(
child: OmpChild,
handler: (command: Record<string, unknown>) => unknown,
): void {
function onOmpCommand(child: OmpChild, handler: (command: Record<string, unknown>) => 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<string, unknown>;
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<string, unknown>);
}
});
}
function replyToCommands(
child: OmpChild,
handler: (command: Record<string, unknown>) => 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();
});
});
@@ -155,7 +155,7 @@ class OmpCliRuntimeSession implements OmpRuntimeSession {
}
async abort(): Promise<void> {
await this.request({ type: "abort" });
await this.requestStopWork({ type: "abort" });
}
async getState(): Promise<OmpSessionState> {
@@ -292,6 +292,10 @@ class OmpCliRuntimeSession implements OmpRuntimeSession {
return this.process.request(OmpRpcCommandSchema.parse(command), timeoutMs);
}
private requestStopWork(command: OmpRpcCommand): Promise<void> {
return this.process.requestStopWork(OmpRpcCommandSchema.parse(command));
}
private emit(event: OmpRuntimeEvent): void {
for (const subscriber of this.subscribers) {
subscriber(event);
@@ -62,6 +62,15 @@ function onPiCommand(child: PiChild, handler: (command: Record<string, unknown>)
});
}
/** 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<string, unknown>) => 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();
});
});
@@ -133,7 +133,7 @@ class PiCliRuntimeSession implements PiRuntimeSession {
}
async clearQueue(): Promise<void> {
await this.request({ type: "clear_queue" });
await this.requestStopWork({ type: "clear_queue" });
}
async compact(customInstructions?: string): Promise<void> {
@@ -148,7 +148,7 @@ class PiCliRuntimeSession implements PiRuntimeSession {
}
async abort(): Promise<void> {
await this.request({ type: "abort" });
await this.requestStopWork({ type: "abort" });
}
async getState(): Promise<PiSessionState> {
@@ -237,6 +237,10 @@ class PiCliRuntimeSession implements PiRuntimeSession {
return this.process.request(command, timeoutMs);
}
private requestStopWork(command: PiRpcCommand): Promise<void> {
return this.process.requestStopWork(command);
}
private async waitForCompletion(command: PiRpcCommand): Promise<void> {
// 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