From cfd7d8d0bcab29f41215fd43720e4671ea59667a Mon Sep 17 00:00:00 2001 From: Bulat Yapparov Date: Sun, 19 Jul 2026 19:29:05 +0100 Subject: [PATCH 1/5] fix(cli): cancel headless runs gracefully on signals --- EVENTS.md | 6 +- packages/cli/src/cli/cmd/run.errors.ts | 10 +- packages/cli/src/cli/cmd/run.ts | 38 ++++++ packages/cli/src/cli/signals.ts | 66 +++++++++ .../cli/test/cli/graceful-signals.test.ts | 112 +++++++++++++++ .../test/cli/run-signal-cancellation.test.ts | 127 ++++++++++++++++++ 6 files changed, 355 insertions(+), 4 deletions(-) create mode 100644 packages/cli/src/cli/signals.ts create mode 100644 packages/cli/test/cli/graceful-signals.test.ts create mode 100644 packages/cli/test/cli/run-signal-cancellation.test.ts diff --git a/EVENTS.md b/EVENTS.md index 2b41488..7c2fa6e 100644 --- a/EVENTS.md +++ b/EVENTS.md @@ -89,7 +89,7 @@ Emitted once when the session ends (success or failure). > **Deprecated in v1.** Prefer the structured `session_error` event for new consumers. `session_complete.error` remains populated for back-compat and will be removed in a future schema version. > -> Note: `session_error` is only emitted when the session terminates abnormally (provider/auth/rate-limit/timeout/OOM). Non-fatal errors that accumulate during an otherwise-successful run surface as individual `error` events and may also appear concatenated in `session_complete.error` without a preceding `session_error`. Alerting logic should key on `session_error`, not on `session_complete.error`. +> Note: `session_error` is only emitted when the session terminates abnormally (provider/auth/rate-limit/timeout/OOM/interruption/termination). Non-fatal errors that accumulate during an otherwise-successful run surface as individual `error` events and may also appear concatenated in `session_complete.error` without a preceding `session_error`. Alerting logic should key on `session_error`, not on `session_complete.error`. ### `session_error` @@ -104,8 +104,8 @@ Emitted immediately before `session_complete` when the session terminates abnorm } ``` -- `reason` (string, **required**) — one of `rate_limit`, `auth`, `timeout`, `oom`, `provider`, `unknown`. -- `code` (string, optional) — provider HTTP status code or error code when available. +- `reason` (string, **required**) — one of `rate_limit`, `auth`, `timeout`, `oom`, `provider`, `interrupted`, `terminated`, `unknown`. `SIGINT` produces `interrupted`; `SIGTERM` produces `terminated`. Signals are not inferred to be timeouts. +- `code` (string, optional) — provider HTTP status code, error code, or conventional signal-derived exit code (`130` for `SIGINT`, `143` for `SIGTERM`) when available. - `message` (string, **required**) — human-readable error message. ## Message Events diff --git a/packages/cli/src/cli/cmd/run.errors.ts b/packages/cli/src/cli/cmd/run.errors.ts index 229c061..1b86c54 100644 --- a/packages/cli/src/cli/cmd/run.errors.ts +++ b/packages/cli/src/cli/cmd/run.errors.ts @@ -1,6 +1,14 @@ export const SCHEMA_VERSION = "1" -export type SessionErrorReason = "rate_limit" | "auth" | "timeout" | "oom" | "provider" | "unknown" +export type SessionErrorReason = + | "rate_limit" + | "auth" + | "timeout" + | "oom" + | "provider" + | "interrupted" + | "terminated" + | "unknown" export type ClassifiedSessionError = { reason: SessionErrorReason diff --git a/packages/cli/src/cli/cmd/run.ts b/packages/cli/src/cli/cmd/run.ts index 02bd79a..4215b91 100644 --- a/packages/cli/src/cli/cmd/run.ts +++ b/packages/cli/src/cli/cmd/run.ts @@ -38,6 +38,7 @@ import { BashTool } from "../../tool/bash" import { TodoWriteTool } from "../../tool/todo" import { Locale } from "../../util/locale" import { Stdout } from "../stdout" +import { signals } from "../signals" type ToolProps = { input: Tool.InferParameters @@ -825,6 +826,32 @@ export const RunCommand = cmd({ permissions: PermissionNext.merge(agentInfo.permission, rules), }) + function cancel() { + Promise.resolve(sdk.session.abort({ sessionID })).catch((e) => { + console.error(e) + }) + } + + using control = signals((signal) => { + error = error ? error + EOL + signal.message : signal.message + if (!sessionErrorEmitted) { + sessionErrorEmitted = true + emit("session_error", { + reason: signal.reason, + code: String(signal.code), + message: signal.message, + }) + } + cancel() + }) + + function flush() { + if (!control.current) return Promise.resolve() + return new Promise((resolve) => { + process.stdout.write("", () => resolve()) + }) + } + // Emit the resolved tool catalog (builtin + MCP tools) and available // skills before the first model turn. Enables structural detection of // "tool was not exposed" failure modes (issue #85). @@ -878,6 +905,12 @@ export const RunCommand = cmd({ process.exitCode = 1 }) + if (control.current) { + await loopDone + await flush() + return + } + if (args.command) { await sdk.session.command({ sessionID, @@ -897,6 +930,7 @@ export const RunCommand = cmd({ parts: [...files, { type: "text", text: message }], }) } + if (control.current) cancel() // Race loopDone against promptResult to handle early prompt failures. // If SessionPrompt.prompt() rejects BEFORE it enters its internal loop() @@ -929,6 +963,7 @@ export const RunCommand = cmd({ }, ), ]) + await flush() } if (args.attach) { @@ -961,6 +996,9 @@ export const RunCommand = cmd({ const share = await Session.share(opts.sessionID) return { data: { share } } }, + abort(opts: any) { + SessionPrompt.cancel(opts.sessionID) + }, async prompt(opts: any) { promptResult = SessionPrompt.prompt({ sessionID: opts.sessionID, diff --git a/packages/cli/src/cli/signals.ts b/packages/cli/src/cli/signals.ts new file mode 100644 index 0000000..4f041e8 --- /dev/null +++ b/packages/cli/src/cli/signals.ts @@ -0,0 +1,66 @@ +export namespace Signals { + export type Info = { + name: "SIGINT" | "SIGTERM" + reason: "interrupted" | "terminated" + code: 130 | 143 + message: string + } +} + +const info = { + SIGINT: { + name: "SIGINT", + reason: "interrupted", + code: 130, + message: "Session interrupted by SIGINT", + }, + SIGTERM: { + name: "SIGTERM", + reason: "terminated", + code: 143, + message: "Session terminated by SIGTERM", + }, +} as const satisfies Record + +export function signals(cancel: (info: Signals.Info) => void, grace = 5_000) { + const state: { + current?: Signals.Info + timer?: ReturnType + resolve?: (info: Signals.Info) => void + } = {} + const received = new Promise((resolve) => { + state.resolve = resolve + }) + + function handle(name: Signals.Info["name"]) { + const signal = info[name] + if (state.current) { + process.exit(state.current.code) + } + state.current = signal + process.exitCode = signal.code + state.timer = setTimeout(() => process.exit(signal.code), grace) + cancel(signal) + state.resolve?.(signal) + } + + const sigint = () => handle("SIGINT") + const sigterm = () => handle("SIGTERM") + process.on("SIGINT", sigint) + process.on("SIGTERM", sigterm) + + const dispose = () => { + if (state.timer) clearTimeout(state.timer) + process.off("SIGINT", sigint) + process.off("SIGTERM", sigterm) + } + + return { + received, + get current() { + return state.current + }, + dispose, + [Symbol.dispose]: dispose, + } +} diff --git a/packages/cli/test/cli/graceful-signals.test.ts b/packages/cli/test/cli/graceful-signals.test.ts new file mode 100644 index 0000000..67a6f87 --- /dev/null +++ b/packages/cli/test/cli/graceful-signals.test.ts @@ -0,0 +1,112 @@ +import { describe, expect, test } from "bun:test" +import path from "path" + +const signal = path.resolve(import.meta.dir, "../../src/cli/signals.ts") + +function child(grace = 1_000, complete = true) { + return Bun.spawn( + [ + "bun", + "--eval", + ` + import { signals } from ${JSON.stringify(signal)} + + const emit = (type, data = {}) => + process.stdout.write(JSON.stringify({ type, ...data }) + "\\n") + const control = signals((info) => { + emit("session_error", { + reason: info.reason, + code: String(info.code), + message: info.message, + }) + }, ${grace}) + + emit("ready") + await control.received + ${ + complete + ? ` + emit("session_complete") + control.dispose() + emit("disposed", { + sigint: process.listenerCount("SIGINT"), + sigterm: process.listenerCount("SIGTERM"), + }) + ` + : "await Bun.sleep(30_000)" + } + `, + ], + { + stdout: "pipe", + stderr: "pipe", + }, + ) +} + +async function ready(proc: ReturnType) { + const reader = proc.stdout.getReader() + const first = await reader.read() + expect(new TextDecoder().decode(first.value)).toBe('{"type":"ready"}\n') + return reader +} + +async function output(reader: ReadableStreamDefaultReader) { + const chunks: Uint8Array[] = [] + while (true) { + const chunk = await reader.read() + if (chunk.done) break + chunks.push(chunk.value) + } + return new TextDecoder() + .decode(Buffer.concat(chunks)) + .trim() + .split("\n") + .filter(Boolean) + .map((line) => JSON.parse(line) as Record) +} + +describe("graceful headless signals", () => { + test.each([ + ["SIGINT", "interrupted", 130], + ["SIGTERM", "terminated", 143], + ] as const)("%s emits a structured terminal sequence and exits %i", async (name, reason, code) => { + const proc = child() + const reader = await ready(proc) + + proc.kill(name) + + expect(await proc.exited).toBe(code) + const events = await output(reader) + expect(events.map((event) => event.type)).toEqual(["session_error", "session_complete", "disposed"]) + expect(events[0]).toMatchObject({ + reason, + code: String(code), + message: name === "SIGINT" ? "Session interrupted by SIGINT" : "Session terminated by SIGTERM", + }) + expect(events[0].reason).not.toBe("timeout") + expect(events[2]).toMatchObject({ sigint: 0, sigterm: 0 }) + }) + + test("a second signal causes immediate hard termination", async () => { + const proc = child(10_000, false) + const reader = await ready(proc) + + proc.kill("SIGTERM") + await Bun.sleep(50) + proc.kill("SIGINT") + + expect(await proc.exited).toBe(143) + expect((await output(reader)).map((event) => event.type)).toEqual(["session_error"]) + }) + + test("an expired grace period causes hard termination", async () => { + const proc = child(50, false) + const reader = await ready(proc) + + proc.kill("SIGINT") + + expect(await proc.exited).toBe(130) + expect((await output(reader)).map((event) => event.type)).toEqual(["session_error"]) + }) +}) diff --git a/packages/cli/test/cli/run-signal-cancellation.test.ts b/packages/cli/test/cli/run-signal-cancellation.test.ts new file mode 100644 index 0000000..4899ab2 --- /dev/null +++ b/packages/cli/test/cli/run-signal-cancellation.test.ts @@ -0,0 +1,127 @@ +import { describe, expect, test } from "bun:test" +import path from "path" + +const entry = path.resolve(import.meta.dir, "../../src/index.ts") +const sessionID = "ses_signal_test" + +async function server() { + const state: { + stream?: ReadableStreamDefaultController + prompted: boolean + } = { prompted: false } + const encoder = new TextEncoder() + const app = Bun.serve({ + port: 0, + fetch(request) { + const url = new URL(request.url) + if (request.method === "POST" && url.pathname === "/session") { + return Response.json({ id: sessionID }) + } + if (request.method === "GET" && url.pathname === "/config") { + return Response.json({}) + } + if (request.method === "GET" && url.pathname === "/event") { + return new Response( + new ReadableStream({ + start(controller) { + state.stream = controller + }, + }), + { headers: { "content-type": "text/event-stream" } }, + ) + } + if (request.method === "POST" && url.pathname.endsWith("/message")) { + state.prompted = true + return Response.json({}) + } + if (request.method === "POST" && url.pathname.endsWith("/abort")) { + state.stream?.enqueue( + encoder.encode( + `data: ${JSON.stringify({ + type: "session.status", + properties: { + sessionID, + status: { type: "idle" }, + }, + })}\n\n`, + ), + ) + return Response.json(true) + } + return Response.json({ error: "not found" }, { status: 404 }) + }, + }) + return { + app, + state, + url: `http://localhost:${app.port}`, + } +} + +function events(stdout: string) { + return stdout + .split("\n") + .filter(Boolean) + .flatMap((line) => { + try { + return [JSON.parse(line) as Record] + } catch { + return [] + } + }) +} + +describe("run --format json graceful signal cancellation", () => { + test.each([ + ["SIGINT", "interrupted", 130], + ["SIGTERM", "terminated", 143], + ] as const)( + "%s cancels the active session before completing", + async (name, reason, code) => { + const mock = await server() + const proc = Bun.spawn( + ["bun", "run", "--conditions=browser", entry, "run", "--format", "json", "--attach", mock.url, "test"], + { + cwd: process.cwd(), + env: { + ...process.env, + ANTHROPIC_API_KEY: "", + OPENAI_API_KEY: "", + }, + stdout: "pipe", + stderr: "pipe", + }, + ) + const timeout = setTimeout(() => proc.kill("SIGKILL"), 15_000) + + try { + for (let tries = 0; !mock.state.prompted && tries < 300; tries++) { + await Bun.sleep(25) + } + expect(mock.state.prompted).toBe(true) + + proc.kill(name) + + const [stdout, exit] = await Promise.all([new Response(proc.stdout).text(), proc.exited]) + expect(exit).toBe(code) + const output = events(stdout) + const error = output.findIndex((event) => event.type === "session_error") + const complete = output.findIndex((event) => event.type === "session_complete") + expect(output.filter((event) => event.type === "session_error")).toHaveLength(1) + expect(output.filter((event) => event.type === "session_complete")).toHaveLength(1) + expect(error).toBeGreaterThan(-1) + expect(complete).toBeGreaterThan(error) + expect(output[error]).toMatchObject({ + reason, + code: String(code), + }) + expect(output[error].reason).not.toBe("timeout") + } finally { + clearTimeout(timeout) + proc.kill("SIGKILL") + mock.app.stop(true) + } + }, + 20_000, + ) +}) From f3d44117bbcf87444039f4ea5341ce20bcb98ade Mon Sep 17 00:00:00 2001 From: Bulat Yapparov Date: Sun, 19 Jul 2026 19:49:49 +0100 Subject: [PATCH 2/5] fix(cli): harden signal shutdown review findings --- EVENTS.md | 2 +- packages/cli/src/cli/cmd/run.ts | 78 +++++++++++-------- packages/cli/src/cli/signals.ts | 23 +++++- .../cli/test/cli/graceful-signals.test.ts | 37 +++++++-- packages/cli/test/cli/run-schema-v1.test.ts | 3 +- .../test/cli/run-signal-cancellation.test.ts | 62 ++++++++++++++- 6 files changed, 157 insertions(+), 48 deletions(-) diff --git a/EVENTS.md b/EVENTS.md index 7c2fa6e..c8250c1 100644 --- a/EVENTS.md +++ b/EVENTS.md @@ -10,7 +10,7 @@ When running `aictrl run --format json`, the CLI emits newline-delimited JSON (N } ``` -The schema is versioned via `session_start.schemaVersion`. This document describes **schema version `"1"`**. Consumers should pin to this version and treat unknown fields as forward-compatible additions. +The schema is versioned via `session_start.schemaVersion`. This document describes **schema version `"1"`**. Consumers should pin to this version and treat unknown event types, fields, and enum-like string values as forward-compatible additions. ## Lifecycle Events diff --git a/packages/cli/src/cli/cmd/run.ts b/packages/cli/src/cli/cmd/run.ts index 4215b91..f606e46 100644 --- a/packages/cli/src/cli/cmd/run.ts +++ b/packages/cli/src/cli/cmd/run.ts @@ -37,8 +37,9 @@ import { SkillTool } from "../../tool/skill" import { BashTool } from "../../tool/bash" import { TodoWriteTool } from "../../tool/todo" import { Locale } from "../../util/locale" +import { Log } from "../../util/log" import { Stdout } from "../stdout" -import { signals } from "../signals" +import { cancel, signals } from "../signals" type ToolProps = { input: Tool.InferParameters @@ -486,6 +487,8 @@ export const RunCommand = cmd({ let promptResult: Promise = Promise.resolve() async function execute(sdk: any) { + let output = Promise.resolve() + function tool(part: ToolPart) { try { if (part.tool === "bash") return bash(props(part)) @@ -522,6 +525,7 @@ export const RunCommand = cmd({ // or promptResult rejection). Without this a mid-run session.error event // can fire site 1 while promptResult/loop() also rejects and fires site 2/3. let sessionErrorEmitted = false + let sessionCompleteEmitted = false const startTime = Date.now() const childSessions = new Set() const emitted = new Set() @@ -532,6 +536,15 @@ export const RunCommand = cmd({ return n } + function complete(message = error ?? null) { + if (sessionCompleteEmitted) return + sessionCompleteEmitted = true + emit("session_complete", { + durationMs: Date.now() - startTime, + error: message, + }) + } + async function loop() { const toggles = new Map() @@ -826,30 +839,36 @@ export const RunCommand = cmd({ permissions: PermissionNext.merge(agentInfo.permission, rules), }) - function cancel() { - Promise.resolve(sdk.session.abort({ sessionID })).catch((e) => { - console.error(e) - }) + function abort() { + cancel( + () => sdk.session.abort({ sessionID }), + () => Log.Default.error("session abort failed"), + ) } - using control = signals((signal) => { - error = error ? error + EOL + signal.message : signal.message - if (!sessionErrorEmitted) { - sessionErrorEmitted = true - emit("session_error", { - reason: signal.reason, - code: String(signal.code), - message: signal.message, - }) - } - cancel() - }) + using control = signals( + (signal) => { + if (!sessionErrorEmitted) { + error = signal.message + sessionErrorEmitted = true + emit("session_error", { + reason: signal.reason, + code: String(signal.code), + message: signal.message, + }) + } + abort() + }, + 5_000, + async (signal) => { + complete(error ?? signal.message) + await flush() + }, + ) function flush() { if (!control.current) return Promise.resolve() - return new Promise((resolve) => { - process.stdout.write("", () => resolve()) - }) + return output } // Emit the resolved tool catalog (builtin + MCP tools) and available @@ -882,10 +901,7 @@ export const RunCommand = cmd({ const loopDone = loop() .then(() => { - emit("session_complete", { - durationMs: Date.now() - startTime, - error: error ?? null, - }) + complete() }) .catch((e) => { const classified = classifySessionError(e) @@ -897,10 +913,7 @@ export const RunCommand = cmd({ message: classified.message, }) } - emit("session_complete", { - durationMs: Date.now() - startTime, - error: classified.message, - }) + complete(error ?? classified.message) console.error(e) process.exitCode = 1 }) @@ -930,7 +943,7 @@ export const RunCommand = cmd({ parts: [...files, { type: "text", text: message }], }) } - if (control.current) cancel() + if (control.current) abort() // Race loopDone against promptResult to handle early prompt failures. // If SessionPrompt.prompt() rejects BEFORE it enters its internal loop() @@ -954,10 +967,7 @@ export const RunCommand = cmd({ message: classified.message, }) } - emit("session_complete", { - durationMs: Date.now() - startTime, - error: classified.message, - }) + complete(error ?? classified.message) console.error(e) process.exitCode = 1 }, @@ -997,7 +1007,7 @@ export const RunCommand = cmd({ return { data: { share } } }, abort(opts: any) { - SessionPrompt.cancel(opts.sessionID) + return SessionPrompt.cancel(opts.sessionID) }, async prompt(opts: any) { promptResult = SessionPrompt.prompt({ diff --git a/packages/cli/src/cli/signals.ts b/packages/cli/src/cli/signals.ts index 4f041e8..d41020e 100644 --- a/packages/cli/src/cli/signals.ts +++ b/packages/cli/src/cli/signals.ts @@ -22,10 +22,15 @@ const info = { }, } as const satisfies Record -export function signals(cancel: (info: Signals.Info) => void, grace = 5_000) { +export function cancel(run: () => unknown, fail: () => void) { + Promise.resolve().then(run).catch(fail) +} + +export function signals(stop: (info: Signals.Info) => void, grace = 5_000, expire?: (info: Signals.Info) => unknown) { const state: { current?: Signals.Info timer?: ReturnType + hard?: ReturnType resolve?: (info: Signals.Info) => void } = {} const received = new Promise((resolve) => { @@ -39,8 +44,19 @@ export function signals(cancel: (info: Signals.Info) => void, grace = 5_000) { } state.current = signal process.exitCode = signal.code - state.timer = setTimeout(() => process.exit(signal.code), grace) - cancel(signal) + state.timer = setTimeout(() => { + state.hard = setTimeout(() => process.exit(signal.code), 250) + Promise.resolve() + .then(() => expire?.(signal)) + .then( + () => { + if (state.hard) clearTimeout(state.hard) + process.exit(signal.code) + }, + () => process.exit(signal.code), + ) + }, grace) + stop(signal) state.resolve?.(signal) } @@ -51,6 +67,7 @@ export function signals(cancel: (info: Signals.Info) => void, grace = 5_000) { const dispose = () => { if (state.timer) clearTimeout(state.timer) + if (state.hard) clearTimeout(state.hard) process.off("SIGINT", sigint) process.off("SIGTERM", sigterm) } diff --git a/packages/cli/test/cli/graceful-signals.test.ts b/packages/cli/test/cli/graceful-signals.test.ts index 67a6f87..24da8ba 100644 --- a/packages/cli/test/cli/graceful-signals.test.ts +++ b/packages/cli/test/cli/graceful-signals.test.ts @@ -1,5 +1,6 @@ import { describe, expect, test } from "bun:test" import path from "path" +import { cancel } from "../../src/cli/signals" const signal = path.resolve(import.meta.dir, "../../src/cli/signals.ts") @@ -13,13 +14,19 @@ function child(grace = 1_000, complete = true) { const emit = (type, data = {}) => process.stdout.write(JSON.stringify({ type, ...data }) + "\\n") - const control = signals((info) => { - emit("session_error", { - reason: info.reason, - code: String(info.code), - message: info.message, - }) - }, ${grace}) + const control = signals( + (info) => { + emit("session_error", { + reason: info.reason, + code: String(info.code), + message: info.message, + }) + }, + ${grace}, + () => new Promise((resolve) => { + process.stdout.write(JSON.stringify({ type: "session_complete" }) + "\\n", resolve) + }), + ) emit("ready") await control.received @@ -67,6 +74,20 @@ async function output(reader: ReadableStreamDefaultReader) { } describe("graceful headless signals", () => { + test.each(["sync", "async"])("cancel contains %s failures", async (mode) => { + const failure = Promise.withResolvers() + + cancel( + () => { + if (mode === "sync") throw new Error("private abort detail") + return Promise.reject(new Error("private abort detail")) + }, + () => failure.resolve("abort failed"), + ) + + expect(await failure.promise).toBe("abort failed") + }) + test.each([ ["SIGINT", "interrupted", 130], ["SIGTERM", "terminated", 143], @@ -107,6 +128,6 @@ describe("graceful headless signals", () => { proc.kill("SIGINT") expect(await proc.exited).toBe(130) - expect((await output(reader)).map((event) => event.type)).toEqual(["session_error"]) + expect((await output(reader)).map((event) => event.type)).toEqual(["session_error", "session_complete"]) }) }) diff --git a/packages/cli/test/cli/run-schema-v1.test.ts b/packages/cli/test/cli/run-schema-v1.test.ts index a26252f..ab261f8 100644 --- a/packages/cli/test/cli/run-schema-v1.test.ts +++ b/packages/cli/test/cli/run-schema-v1.test.ts @@ -39,7 +39,8 @@ describe("run.ts v1 schema emissions (#63)", () => { const source = await Bun.file(RUN_SRC).text() const errIdx = source.indexOf('emit("session_error"') expect(errIdx).toBeGreaterThan(-1) - expect(errIdx).toBeLessThan(source.lastIndexOf('emit("session_complete"')) + expect(source.slice(errIdx).indexOf("complete(")).toBeGreaterThan(-1) + expect(source).toContain("if (sessionCompleteEmitted) return") }) test("primary session.error events are surfaced as session_error", async () => { diff --git a/packages/cli/test/cli/run-signal-cancellation.test.ts b/packages/cli/test/cli/run-signal-cancellation.test.ts index 4899ab2..3a3b986 100644 --- a/packages/cli/test/cli/run-signal-cancellation.test.ts +++ b/packages/cli/test/cli/run-signal-cancellation.test.ts @@ -4,7 +4,7 @@ import path from "path" const entry = path.resolve(import.meta.dir, "../../src/index.ts") const sessionID = "ses_signal_test" -async function server() { +async function server(provider = false) { const state: { stream?: ReadableStreamDefaultController prompted: boolean @@ -32,6 +32,22 @@ async function server() { } if (request.method === "POST" && url.pathname.endsWith("/message")) { state.prompted = true + if (provider) { + state.stream?.enqueue( + encoder.encode( + `data: ${JSON.stringify({ + type: "session.error", + properties: { + sessionID, + error: { + name: "ProviderError", + data: { message: "provider failed first" }, + }, + }, + })}\n\n`, + ), + ) + } return Response.json({}) } if (request.method === "POST" && url.pathname.endsWith("/abort")) { @@ -124,4 +140,48 @@ describe("run --format json graceful signal cancellation", () => { }, 20_000, ) + + test("a provider error remains the canonical terminal error when a signal follows", async () => { + const mock = await server(true) + const proc = Bun.spawn( + ["bun", "run", "--conditions=browser", entry, "run", "--format", "json", "--attach", mock.url, "test"], + { + cwd: process.cwd(), + env: { + ...process.env, + ANTHROPIC_API_KEY: "", + OPENAI_API_KEY: "", + }, + stdout: "pipe", + stderr: "pipe", + }, + ) + const timeout = setTimeout(() => proc.kill("SIGKILL"), 15_000) + + try { + for (let tries = 0; !mock.state.prompted && tries < 300; tries++) { + await Bun.sleep(25) + } + expect(mock.state.prompted).toBe(true) + await Bun.sleep(100) + + proc.kill("SIGTERM") + + const [stdout, exit] = await Promise.all([new Response(proc.stdout).text(), proc.exited]) + expect(exit).toBe(143) + const output = events(stdout) + expect(output.filter((event) => event.type === "session_error")).toHaveLength(1) + expect(output.find((event) => event.type === "session_error")).toMatchObject({ + reason: "unknown", + message: "provider failed first", + }) + expect(output.find((event) => event.type === "session_complete")).toMatchObject({ + error: "provider failed first", + }) + } finally { + clearTimeout(timeout) + proc.kill("SIGKILL") + mock.app.stop(true) + } + }, 20_000) }) From b848c8c8319092433d5d712b65d29f357b2198b4 Mon Sep 17 00:00:00 2001 From: Bulat Yapparov Date: Sun, 19 Jul 2026 20:09:56 +0100 Subject: [PATCH 3/5] fix(cli): close final signal review gaps --- CHANGELOG.md | 6 ++ packages/cli/src/cli/cmd/run.ts | 48 ++++++++--- packages/cli/src/cli/signals.ts | 2 +- .../cli/test/cli/graceful-signals.test.ts | 6 +- packages/cli/test/cli/run-schema-v1.test.ts | 7 ++ .../test/cli/run-signal-cancellation.test.ts | 84 ++++++++++++++++++- 6 files changed, 133 insertions(+), 20 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9ba205c..fd05e13 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog +## Unreleased + +### Compatibility + +- **NDJSON v1 terminal reasons are an open set** — `session_error.reason` now includes `interrupted` for `SIGINT` and `terminated` for `SIGTERM`, and `code` may contain the conventional signal-derived exit code (`130` or `143`). Schema v1 consumers should treat unknown event types, fields, and enum-like string values as forward-compatible additions. + ## 0.3.2 (2026-04-11) ### Fixes diff --git a/packages/cli/src/cli/cmd/run.ts b/packages/cli/src/cli/cmd/run.ts index f606e46..2e6a7f3 100644 --- a/packages/cli/src/cli/cmd/run.ts +++ b/packages/cli/src/cli/cmd/run.ts @@ -39,7 +39,7 @@ import { TodoWriteTool } from "../../tool/todo" import { Locale } from "../../util/locale" import { Log } from "../../util/log" import { Stdout } from "../stdout" -import { cancel, signals } from "../signals" +import { attempt, signals } from "../signals" type ToolProps = { input: Tool.InferParameters @@ -487,7 +487,22 @@ export const RunCommand = cmd({ let promptResult: Promise = Promise.resolve() async function execute(sdk: any) { - let output = Promise.resolve() + // Signal-local delivery barrier. #91 owns the shared stdout writer that + // will replace this state when its PR is integrated. + const output = { + failed: false, + pending: new Set>(), + } + const fail = (error: unknown) => { + if (error instanceof Error && "code" in error && error.code === "EPIPE") return + output.failed = true + } + if (args.format === "json") process.stdout.on("error", fail) + using _stdout = { + [Symbol.dispose]() { + if (args.format === "json") process.stdout.off("error", fail) + }, + } function tool(part: ToolPart) { try { @@ -691,7 +706,7 @@ export const RunCommand = cmd({ // to a non-zero exit code so CI wrappers see the failure instead of a // spuriously-green job. process.exitCode (not process.exit) lets the // loop drain to session.status idle and emit session_complete first. - process.exitCode = 1 + if (!control.current) process.exitCode = 1 if (!sessionErrorEmitted) { sessionErrorEmitted = true const classified = classifySessionError(props.error) @@ -839,8 +854,11 @@ export const RunCommand = cmd({ permissions: PermissionNext.merge(agentInfo.permission, rules), }) + let aborted = false function abort() { - cancel( + if (aborted) return + aborted = true + attempt( () => sdk.session.abort({ sessionID }), () => Log.Default.error("session abort failed"), ) @@ -866,9 +884,9 @@ export const RunCommand = cmd({ }, ) - function flush() { - if (!control.current) return Promise.resolve() - return output + async function flush() { + while (output.pending.size) await Promise.allSettled(output.pending) + if (output.failed) throw new Error("stdout write failed") } // Emit the resolved tool catalog (builtin + MCP tools) and available @@ -914,8 +932,10 @@ export const RunCommand = cmd({ }) } complete(error ?? classified.message) - console.error(e) - process.exitCode = 1 + if (!control.current) { + console.error(e) + process.exitCode = 1 + } }) if (control.current) { @@ -958,7 +978,9 @@ export const RunCommand = cmd({ // If prompt rejects, surface the error immediately (e) => { const classified = classifySessionError(e) - error = error ? error + EOL + classified.message : classified.message + if (!control.current) { + error = error ? error + EOL + classified.message : classified.message + } if (!sessionErrorEmitted) { sessionErrorEmitted = true emit("session_error", { @@ -968,8 +990,10 @@ export const RunCommand = cmd({ }) } complete(error ?? classified.message) - console.error(e) - process.exitCode = 1 + if (!control.current) { + console.error(e) + process.exitCode = 1 + } }, ), ]) diff --git a/packages/cli/src/cli/signals.ts b/packages/cli/src/cli/signals.ts index d41020e..b999640 100644 --- a/packages/cli/src/cli/signals.ts +++ b/packages/cli/src/cli/signals.ts @@ -22,7 +22,7 @@ const info = { }, } as const satisfies Record -export function cancel(run: () => unknown, fail: () => void) { +export function attempt(run: () => unknown, fail: () => void) { Promise.resolve().then(run).catch(fail) } diff --git a/packages/cli/test/cli/graceful-signals.test.ts b/packages/cli/test/cli/graceful-signals.test.ts index 24da8ba..a4d50ea 100644 --- a/packages/cli/test/cli/graceful-signals.test.ts +++ b/packages/cli/test/cli/graceful-signals.test.ts @@ -1,6 +1,6 @@ import { describe, expect, test } from "bun:test" import path from "path" -import { cancel } from "../../src/cli/signals" +import { attempt } from "../../src/cli/signals" const signal = path.resolve(import.meta.dir, "../../src/cli/signals.ts") @@ -74,10 +74,10 @@ async function output(reader: ReadableStreamDefaultReader) { } describe("graceful headless signals", () => { - test.each(["sync", "async"])("cancel contains %s failures", async (mode) => { + test.each(["sync", "async"])("attempt contains %s failures", async (mode) => { const failure = Promise.withResolvers() - cancel( + attempt( () => { if (mode === "sync") throw new Error("private abort detail") return Promise.reject(new Error("private abort detail")) diff --git a/packages/cli/test/cli/run-schema-v1.test.ts b/packages/cli/test/cli/run-schema-v1.test.ts index ab261f8..9956804 100644 --- a/packages/cli/test/cli/run-schema-v1.test.ts +++ b/packages/cli/test/cli/run-schema-v1.test.ts @@ -57,6 +57,13 @@ describe("run.ts v1 schema emissions (#63)", () => { expect(block).toContain('emit("error"') }) + test("signal cancellation errors preserve the signal exit code", async () => { + const source = await Bun.file(RUN_SRC).text() + const caught = source.match(/if \(!control\.current\) \{\s+console\.error\(e\)\s+process\.exitCode = 1\s+\}/g) + expect(caught).toHaveLength(2) + expect(source).toContain("if (!control.current) process.exitCode = 1") + }) + test("text / reasoning / tool_use include sequenceNum", async () => { const source = await Bun.file(RUN_SRC).text() expect(source).toContain("sequenceNum") diff --git a/packages/cli/test/cli/run-signal-cancellation.test.ts b/packages/cli/test/cli/run-signal-cancellation.test.ts index 3a3b986..701bfbb 100644 --- a/packages/cli/test/cli/run-signal-cancellation.test.ts +++ b/packages/cli/test/cli/run-signal-cancellation.test.ts @@ -4,11 +4,13 @@ import path from "path" const entry = path.resolve(import.meta.dir, "../../src/index.ts") const sessionID = "ses_signal_test" -async function server(provider = false) { +async function server(options: { provider?: boolean; hold?: boolean } = {}) { const state: { + aborts: number + message?: ReturnType> stream?: ReadableStreamDefaultController prompted: boolean - } = { prompted: false } + } = { aborts: 0, prompted: false } const encoder = new TextEncoder() const app = Bun.serve({ port: 0, @@ -32,7 +34,7 @@ async function server(provider = false) { } if (request.method === "POST" && url.pathname.endsWith("/message")) { state.prompted = true - if (provider) { + if (options.provider) { state.stream?.enqueue( encoder.encode( `data: ${JSON.stringify({ @@ -48,9 +50,14 @@ async function server(provider = false) { ), ) } + if (options.hold) { + state.message = Promise.withResolvers() + return state.message.promise + } return Response.json({}) } if (request.method === "POST" && url.pathname.endsWith("/abort")) { + state.aborts++ state.stream?.enqueue( encoder.encode( `data: ${JSON.stringify({ @@ -62,6 +69,7 @@ async function server(provider = false) { })}\n\n`, ), ) + state.message?.resolve(Response.json({})) return Response.json(true) } return Response.json({ error: "not found" }, { status: 404 }) @@ -142,7 +150,7 @@ describe("run --format json graceful signal cancellation", () => { ) test("a provider error remains the canonical terminal error when a signal follows", async () => { - const mock = await server(true) + const mock = await server({ provider: true }) const proc = Bun.spawn( ["bun", "run", "--conditions=browser", entry, "run", "--format", "json", "--attach", mock.url, "test"], { @@ -184,4 +192,72 @@ describe("run --format json graceful signal cancellation", () => { mock.app.stop(true) } }, 20_000) + + test("a signal aborts an in-flight prompt exactly once", async () => { + const mock = await server({ hold: true }) + const proc = Bun.spawn( + ["bun", "run", "--conditions=browser", entry, "run", "--format", "json", "--attach", mock.url, "test"], + { + cwd: process.cwd(), + env: { + ...process.env, + ANTHROPIC_API_KEY: "", + OPENAI_API_KEY: "", + }, + stdout: "pipe", + stderr: "pipe", + }, + ) + const timeout = setTimeout(() => proc.kill("SIGKILL"), 15_000) + + try { + for (let tries = 0; !mock.state.prompted && tries < 300; tries++) { + await Bun.sleep(25) + } + expect(mock.state.prompted).toBe(true) + + proc.kill("SIGTERM") + + expect(await proc.exited).toBe(143) + expect(mock.state.aborts).toBe(1) + } finally { + clearTimeout(timeout) + proc.kill("SIGKILL") + mock.app.stop(true) + } + }, 20_000) + + test("a closed stdout consumer does not replace the signal exit code", async () => { + const mock = await server() + const proc = Bun.spawn( + ["bun", "run", "--conditions=browser", entry, "run", "--format", "json", "--attach", mock.url, "test"], + { + cwd: process.cwd(), + env: { + ...process.env, + ANTHROPIC_API_KEY: "", + OPENAI_API_KEY: "", + }, + stdout: "pipe", + stderr: "pipe", + }, + ) + const timeout = setTimeout(() => proc.kill("SIGKILL"), 15_000) + + try { + for (let tries = 0; !mock.state.prompted && tries < 300; tries++) { + await Bun.sleep(25) + } + expect(mock.state.prompted).toBe(true) + await proc.stdout.cancel() + + proc.kill("SIGINT") + + expect(await proc.exited).toBe(130) + } finally { + clearTimeout(timeout) + proc.kill("SIGKILL") + mock.app.stop(true) + } + }, 20_000) }) From e97241e3a96cda25e0fb7fc8b6990bbae2a806c1 Mon Sep 17 00:00:00 2001 From: Bulat Yapparov Date: Mon, 20 Jul 2026 19:36:27 +0100 Subject: [PATCH 4/5] refactor(run): clarify signal shutdown flow --- packages/cli/src/cli/cmd/run.ts | 141 +++++--------------- packages/cli/src/cli/shutdown.ts | 2 +- packages/cli/test/cli/fixture/stdout.ts | 1 + packages/cli/test/cli/run-schema-v1.test.ts | 9 +- packages/cli/test/cli/stdout.test.ts | 23 ++++ 5 files changed, 63 insertions(+), 113 deletions(-) diff --git a/packages/cli/src/cli/cmd/run.ts b/packages/cli/src/cli/cmd/run.ts index 2e6a7f3..7c7190d 100644 --- a/packages/cli/src/cli/cmd/run.ts +++ b/packages/cli/src/cli/cmd/run.ts @@ -38,8 +38,9 @@ import { BashTool } from "../../tool/bash" import { TodoWriteTool } from "../../tool/todo" import { Locale } from "../../util/locale" import { Log } from "../../util/log" +import { Shutdown } from "../shutdown" import { Stdout } from "../stdout" -import { attempt, signals } from "../signals" +import { attempt, signals, type Signals } from "../signals" type ToolProps = { input: Tool.InferParameters @@ -487,23 +488,6 @@ export const RunCommand = cmd({ let promptResult: Promise = Promise.resolve() async function execute(sdk: any) { - // Signal-local delivery barrier. #91 owns the shared stdout writer that - // will replace this state when its PR is integrated. - const output = { - failed: false, - pending: new Set>(), - } - const fail = (error: unknown) => { - if (error instanceof Error && "code" in error && error.code === "EPIPE") return - output.failed = true - } - if (args.format === "json") process.stdout.on("error", fail) - using _stdout = { - [Symbol.dispose]() { - if (args.format === "json") process.stdout.off("error", fail) - }, - } - function tool(part: ToolPart) { try { if (part.tool === "bash") return bash(props(part)) @@ -560,6 +544,12 @@ export const RunCommand = cmd({ }) } + function report(reason: string, code: string | undefined, message: string) { + if (sessionErrorEmitted) return + sessionErrorEmitted = true + emit("session_error", { reason, code, message }) + } + async function loop() { const toggles = new Map() @@ -707,20 +697,11 @@ export const RunCommand = cmd({ // spuriously-green job. process.exitCode (not process.exit) lets the // loop drain to session.status idle and emit session_complete first. if (!control.current) process.exitCode = 1 - if (!sessionErrorEmitted) { - sessionErrorEmitted = true - const classified = classifySessionError(props.error) - // Structured session_error (classified reason/code/message) is emitted for the - // primary session only. The generic "error" event below fires for both primary - // and child-session failures and carries the raw error object — these are - // intentionally distinct channels: session_error is for structured telemetry/CI - // consumers, "error" is the legacy observable for raw error pass-through. - emit("session_error", { - reason: classified.reason, - code: classified.code, - message: classified.message, - }) - } + const classified = classifySessionError(props.error) + // Structured session_error is emitted for the primary session only. + // The legacy "error" event below remains the raw pass-through for + // both primary and child-session failures. + report(classified.reason, classified.code, classified.message) } if (emit("error", { error: props.error, sourceSessionID: props.sessionID })) continue UI.error(err) @@ -864,29 +845,27 @@ export const RunCommand = cmd({ ) } - using control = signals( - (signal) => { - if (!sessionErrorEmitted) { - error = signal.message - sessionErrorEmitted = true - emit("session_error", { - reason: signal.reason, - code: String(signal.code), - message: signal.message, - }) - } - abort() - }, - 5_000, - async (signal) => { - complete(error ?? signal.message) - await flush() - }, - ) + function interrupt(signal: Signals.Info) { + if (!sessionErrorEmitted) error = signal.message + report(signal.reason, String(signal.code), signal.message) + abort() + } - async function flush() { - while (output.pending.size) await Promise.allSettled(output.pending) - if (output.failed) throw new Error("stdout write failed") + async function expire(signal: Signals.Info) { + complete(error ?? signal.message) + await Shutdown.flush() + } + + using control = signals(interrupt, 5_000, expire) + + function fail(cause: unknown) { + const classified = classifySessionError(cause) + error ??= classified.message + report(classified.reason, classified.code, classified.message) + complete(error) + if (control.current) return + console.error(cause) + process.exitCode = 1 } // Emit the resolved tool catalog (builtin + MCP tools) and available @@ -917,30 +896,11 @@ export const RunCommand = cmd({ }) } - const loopDone = loop() - .then(() => { - complete() - }) - .catch((e) => { - const classified = classifySessionError(e) - if (!sessionErrorEmitted) { - sessionErrorEmitted = true - emit("session_error", { - reason: classified.reason, - code: classified.code, - message: classified.message, - }) - } - complete(error ?? classified.message) - if (!control.current) { - console.error(e) - process.exitCode = 1 - } - }) + const loopDone = loop().then(() => complete(), fail) if (control.current) { await loopDone - await flush() + await Shutdown.flush() return } @@ -963,41 +923,14 @@ export const RunCommand = cmd({ parts: [...files, { type: "text", text: message }], }) } - if (control.current) abort() // Race loopDone against promptResult to handle early prompt failures. // If SessionPrompt.prompt() rejects BEFORE it enters its internal loop() // (e.g., Session.get() fails, model not found), no session.status idle event // is emitted, so loopDone would hang forever. Racing ensures we surface // the error and exit. - await Promise.race([ - loopDone, - promptResult.then( - // If prompt resolves normally, wait for the event loop to finish - () => loopDone, - // If prompt rejects, surface the error immediately - (e) => { - const classified = classifySessionError(e) - if (!control.current) { - error = error ? error + EOL + classified.message : classified.message - } - if (!sessionErrorEmitted) { - sessionErrorEmitted = true - emit("session_error", { - reason: classified.reason, - code: classified.code, - message: classified.message, - }) - } - complete(error ?? classified.message) - if (!control.current) { - console.error(e) - process.exitCode = 1 - } - }, - ), - ]) - await flush() + await Promise.race([loopDone, promptResult.then(() => loopDone, fail)]) + await Shutdown.flush() } if (args.attach) { diff --git a/packages/cli/src/cli/shutdown.ts b/packages/cli/src/cli/shutdown.ts index bef93d3..6196d24 100644 --- a/packages/cli/src/cli/shutdown.ts +++ b/packages/cli/src/cli/shutdown.ts @@ -7,7 +7,7 @@ export namespace Shutdown { Log.Default.error("stdout flush failed", { error: error instanceof Error ? error.message : error, }) - process.exitCode = 1 + process.exitCode ||= 1 }) } } diff --git a/packages/cli/test/cli/fixture/stdout.ts b/packages/cli/test/cli/fixture/stdout.ts index 9bc7760..65d2c02 100644 --- a/packages/cli/test/cli/fixture/stdout.ts +++ b/packages/cli/test/cli/fixture/stdout.ts @@ -6,6 +6,7 @@ if (process.argv.includes("--shutdown-error")) { Stdout.write("") await Stdout.flush() process.stdout.emit("error", Object.assign(new Error("broken stdout"), { code: "EIO" })) + if (process.argv.includes("--signal-exit")) process.exitCode = 143 await Shutdown.flush() const status = process.argv.find((arg) => arg.startsWith("--status="))?.slice("--status=".length) if (status) await Bun.write(status, String(process.exitCode)) diff --git a/packages/cli/test/cli/run-schema-v1.test.ts b/packages/cli/test/cli/run-schema-v1.test.ts index 9956804..f920d06 100644 --- a/packages/cli/test/cli/run-schema-v1.test.ts +++ b/packages/cli/test/cli/run-schema-v1.test.ts @@ -53,17 +53,10 @@ describe("run.ts v1 schema emissions (#63)", () => { const nextHandlerOffset = after.indexOf('if (event.type === "') const block = nextHandlerOffset === -1 ? after : after.slice(0, nextHandlerOffset) expect(block).toContain("classifySessionError(props.error)") - expect(block).toContain('emit("session_error"') + expect(block).toContain("report(classified.reason, classified.code, classified.message)") expect(block).toContain('emit("error"') }) - test("signal cancellation errors preserve the signal exit code", async () => { - const source = await Bun.file(RUN_SRC).text() - const caught = source.match(/if \(!control\.current\) \{\s+console\.error\(e\)\s+process\.exitCode = 1\s+\}/g) - expect(caught).toHaveLength(2) - expect(source).toContain("if (!control.current) process.exitCode = 1") - }) - test("text / reasoning / tool_use include sequenceNum", async () => { const source = await Bun.file(RUN_SRC).text() expect(source).toContain("sequenceNum") diff --git a/packages/cli/test/cli/stdout.test.ts b/packages/cli/test/cli/stdout.test.ts index afeffde..bb1c94b 100644 --- a/packages/cli/test/cli/stdout.test.ts +++ b/packages/cli/test/cli/stdout.test.ts @@ -79,4 +79,27 @@ describe("stdout", () => { expect(await Bun.file(status).text()).toBe("1") expect(await new Response(child.stderr).text()).toBe("") }) + + test("flush failure preserves an existing signal exit status", async () => { + await using tmp = await tmpdir() + const status = path.join(tmp.path, "status") + const child = Bun.spawn( + [ + process.execPath, + path.join(import.meta.dir, "fixture", "stdout.ts"), + "--shutdown-error", + "--signal-exit", + `--status=${status}`, + ], + { + cwd: path.join(import.meta.dir, "../.."), + stdout: "pipe", + stderr: "pipe", + }, + ) + + expect(await child.exited).toBe(143) + expect(await Bun.file(status).text()).toBe("143") + expect(await new Response(child.stderr).text()).toBe("") + }) }) From ab016707025548a9407c949bccef961bc7242da6 Mon Sep 17 00:00:00 2001 From: Bulat Yapparov Date: Tue, 21 Jul 2026 08:14:18 +0100 Subject: [PATCH 5/5] fix(run): preserve terminal event ordering --- packages/cli/src/cli/cmd/run.terminal.ts | 21 +++++++++++++ packages/cli/src/cli/cmd/run.ts | 34 ++++++++++----------- packages/cli/test/cli/run-schema-v1.test.ts | 8 ++--- packages/cli/test/cli/run-terminal.test.ts | 26 ++++++++++++++++ 4 files changed, 67 insertions(+), 22 deletions(-) create mode 100644 packages/cli/src/cli/cmd/run.terminal.ts create mode 100644 packages/cli/test/cli/run-terminal.test.ts diff --git a/packages/cli/src/cli/cmd/run.terminal.ts b/packages/cli/src/cli/cmd/run.terminal.ts new file mode 100644 index 0000000..7c095f6 --- /dev/null +++ b/packages/cli/src/cli/cmd/run.terminal.ts @@ -0,0 +1,21 @@ +type Emit = (type: string, data: Record) => unknown + +export function terminal(emit: Emit) { + const state = { + complete: false, + error: false, + } + + return { + complete(data: Record) { + if (state.complete) return + state.complete = true + emit("session_complete", data) + }, + error(data: Record) { + if (state.complete || state.error) return + state.error = true + emit("session_error", data) + }, + } +} diff --git a/packages/cli/src/cli/cmd/run.ts b/packages/cli/src/cli/cmd/run.ts index 7c7190d..62f0e0d 100644 --- a/packages/cli/src/cli/cmd/run.ts +++ b/packages/cli/src/cli/cmd/run.ts @@ -4,6 +4,7 @@ import { pathToFileURL } from "bun" import { UI } from "../ui" import { cmd } from "./cmd" import { classifySessionError, SCHEMA_VERSION } from "./run.errors" +import { terminal } from "./run.terminal" import { buildToolCatalogItems } from "./tool-catalog" import { withTimeout } from "../../util/timeout" import { Flag } from "../../flag/flag" @@ -519,12 +520,6 @@ export const RunCommand = cmd({ const events = await sdk.event.subscribe() let error: string | undefined - // Guard: at most one session_error is emitted per run regardless of which - // failure path fires first (in-loop session.error handler, loopDone.catch, - // or promptResult rejection). Without this a mid-run session.error event - // can fire site 1 while promptResult/loop() also rejects and fires site 2/3. - let sessionErrorEmitted = false - let sessionCompleteEmitted = false const startTime = Date.now() const childSessions = new Set() const emitted = new Set() @@ -535,19 +530,18 @@ export const RunCommand = cmd({ return n } + // Keep every failure path on one ordered, idempotent terminal lifecycle. + const output = terminal(emit) + function complete(message = error ?? null) { - if (sessionCompleteEmitted) return - sessionCompleteEmitted = true - emit("session_complete", { + output.complete({ durationMs: Date.now() - startTime, error: message, }) } function report(reason: string, code: string | undefined, message: string) { - if (sessionErrorEmitted) return - sessionErrorEmitted = true - emit("session_error", { reason, code, message }) + output.error({ reason, code, message }) } async function loop() { @@ -698,9 +692,9 @@ export const RunCommand = cmd({ // loop drain to session.status idle and emit session_complete first. if (!control.current) process.exitCode = 1 const classified = classifySessionError(props.error) - // Structured session_error is emitted for the primary session only. - // The legacy "error" event below remains the raw pass-through for - // both primary and child-session failures. + // Structured session_error is the telemetry/CI channel for the + // primary session. The legacy "error" event below is the raw + // pass-through for both primary and child-session failures. report(classified.reason, classified.code, classified.message) } if (emit("error", { error: props.error, sourceSessionID: props.sessionID })) continue @@ -846,7 +840,7 @@ export const RunCommand = cmd({ } function interrupt(signal: Signals.Info) { - if (!sessionErrorEmitted) error = signal.message + error ??= signal.message report(signal.reason, String(signal.code), signal.message) abort() } @@ -863,7 +857,13 @@ export const RunCommand = cmd({ error ??= classified.message report(classified.reason, classified.code, classified.message) complete(error) - if (control.current) return + if (control.current) { + Log.Default.error("run failed after signal", { + reason: classified.reason, + code: classified.code, + }) + return + } console.error(cause) process.exitCode = 1 } diff --git a/packages/cli/test/cli/run-schema-v1.test.ts b/packages/cli/test/cli/run-schema-v1.test.ts index f920d06..5079c6b 100644 --- a/packages/cli/test/cli/run-schema-v1.test.ts +++ b/packages/cli/test/cli/run-schema-v1.test.ts @@ -35,12 +35,10 @@ describe("run.ts v1 schema emissions (#63)", () => { expect(block).toContain("permissions:") }) - test("session_error is emitted before session_complete on failure", async () => { + test("session terminal events use the ordered lifecycle", async () => { const source = await Bun.file(RUN_SRC).text() - const errIdx = source.indexOf('emit("session_error"') - expect(errIdx).toBeGreaterThan(-1) - expect(source.slice(errIdx).indexOf("complete(")).toBeGreaterThan(-1) - expect(source).toContain("if (sessionCompleteEmitted) return") + expect(source).toContain("output.error({ reason, code, message })") + expect(source).toContain("output.complete({") }) test("primary session.error events are surfaced as session_error", async () => { diff --git a/packages/cli/test/cli/run-terminal.test.ts b/packages/cli/test/cli/run-terminal.test.ts new file mode 100644 index 0000000..4f05c5d --- /dev/null +++ b/packages/cli/test/cli/run-terminal.test.ts @@ -0,0 +1,26 @@ +import { describe, expect, test } from "bun:test" +import { terminal } from "../../src/cli/cmd/run.terminal" + +describe("run terminal events", () => { + test("error precedes completion and each event is emitted once", () => { + const events: string[] = [] + const output = terminal((type) => events.push(type)) + + output.error({}) + output.error({}) + output.complete({}) + output.complete({}) + + expect(events).toEqual(["session_error", "session_complete"]) + }) + + test("error cannot be emitted after completion", () => { + const events: string[] = [] + const output = terminal((type) => events.push(type)) + + output.complete({}) + output.error({}) + + expect(events).toEqual(["session_complete"]) + }) +})