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/EVENTS.md b/EVENTS.md index f3faeb6..911798f 100644 --- a/EVENTS.md +++ b/EVENTS.md @@ -155,7 +155,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` @@ -170,8 +170,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.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 528b5b5..ca65b00 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" @@ -37,7 +38,10 @@ 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 { Shutdown } from "../shutdown" import { Stdout } from "../stdout" +import { attempt, signals, type Signals } from "../signals" import { createRunInvocation } from "./run.invocation" type ToolProps = { @@ -534,11 +538,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 const startTime = Date.now() const childSessions = new Set() const emitted = new Set() @@ -549,6 +548,20 @@ 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) { + output.complete({ + durationMs: Date.now() - startTime, + error: message, + }) + } + + function report(reason: string, code: string | undefined, message: string) { + output.error({ reason, code, message }) + } + async function loop() { const toggles = new Map() @@ -695,22 +708,13 @@ 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 invocation.error(props.error) - 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 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 UI.error(err) @@ -845,6 +849,48 @@ export const RunCommand = cmd({ permissions: PermissionNext.merge(agentInfo.permission, rules), }) + let aborted = false + function abort() { + if (aborted) return + aborted = true + attempt( + () => sdk.session.abort({ sessionID }), + () => Log.Default.error("session abort failed"), + ) + } + + function interrupt(signal: Signals.Info) { + error ??= signal.message + invocation.error(signal.message) + report(signal.reason, String(signal.code), signal.message) + abort() + } + + async function expire(signal: Signals.Info) { + complete(error ?? signal.message) + await invocation.abort(signal.message) + await Shutdown.flush() + } + + using control = signals(interrupt, 5_000, expire) + + function reject(cause: unknown) { + const classified = classifySessionError(cause) + error ??= classified.message + invocation.error(cause) + report(classified.reason, classified.code, classified.message) + complete(error) + if (control.current) { + Log.Default.error("run failed after signal", { + reason: classified.reason, + code: classified.code, + }) + return + } + console.error(cause) + process.exitCode = 1 + } + // 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). @@ -873,31 +919,13 @@ export const RunCommand = cmd({ }) } - const loopDone = loop() - .then(() => { - emit("session_complete", { - durationMs: Date.now() - startTime, - error: error ?? null, - }) - }) - .catch((e) => { - const classified = classifySessionError(e) - if (!sessionErrorEmitted) { - sessionErrorEmitted = true - emit("session_error", { - reason: classified.reason, - code: classified.code, - message: classified.message, - }) - } - emit("session_complete", { - durationMs: Date.now() - startTime, - error: classified.message, - }) - console.error(e) - process.exitCode = 1 - invocation.error(e) - }) + const loopDone = loop().then(() => complete(), reject) + + if (control.current) { + await loopDone + await Shutdown.flush() + return + } if (args.command) { await sdk.session.command({ @@ -924,33 +952,8 @@ export const RunCommand = cmd({ // (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) - error = error ? error + EOL + classified.message : classified.message - if (!sessionErrorEmitted) { - sessionErrorEmitted = true - emit("session_error", { - reason: classified.reason, - code: classified.code, - message: classified.message, - }) - } - emit("session_complete", { - durationMs: Date.now() - startTime, - error: classified.message, - }) - console.error(e) - process.exitCode = 1 - invocation.error(e) - }, - ), - ]) + await Promise.race([loopDone, promptResult.then(() => loopDone, reject)]) + await Shutdown.flush() } invocation.phase("bootstrap") @@ -983,6 +986,9 @@ export const RunCommand = cmd({ const share = await Session.share(opts.sessionID) return { data: { share } } }, + abort(opts: any) { + return SessionPrompt.cancel(opts.sessionID) + }, async prompt(opts: any) { promptResult = SessionPrompt.prompt({ sessionID: opts.sessionID, 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/src/cli/signals.ts b/packages/cli/src/cli/signals.ts new file mode 100644 index 0000000..b999640 --- /dev/null +++ b/packages/cli/src/cli/signals.ts @@ -0,0 +1,83 @@ +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 attempt(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) => { + 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(() => { + 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) + } + + const sigint = () => handle("SIGINT") + const sigterm = () => handle("SIGTERM") + process.on("SIGINT", sigint) + process.on("SIGTERM", sigterm) + + const dispose = () => { + if (state.timer) clearTimeout(state.timer) + if (state.hard) clearTimeout(state.hard) + 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/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/graceful-signals.test.ts b/packages/cli/test/cli/graceful-signals.test.ts new file mode 100644 index 0000000..a4d50ea --- /dev/null +++ b/packages/cli/test/cli/graceful-signals.test.ts @@ -0,0 +1,133 @@ +import { describe, expect, test } from "bun:test" +import path from "path" +import { attempt } from "../../src/cli/signals" + +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}, + () => new Promise((resolve) => { + process.stdout.write(JSON.stringify({ type: "session_complete" }) + "\\n", resolve) + }), + ) + + 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(["sync", "async"])("attempt contains %s failures", async (mode) => { + const failure = Promise.withResolvers() + + attempt( + () => { + 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], + ] 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", "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..5079c6b 100644 --- a/packages/cli/test/cli/run-schema-v1.test.ts +++ b/packages/cli/test/cli/run-schema-v1.test.ts @@ -35,11 +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(errIdx).toBeLessThan(source.lastIndexOf('emit("session_complete"')) + expect(source).toContain("output.error({ reason, code, message })") + expect(source).toContain("output.complete({") }) test("primary session.error events are surfaced as session_error", async () => { @@ -52,7 +51,7 @@ 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"') }) 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..8642b85 --- /dev/null +++ b/packages/cli/test/cli/run-signal-cancellation.test.ts @@ -0,0 +1,317 @@ +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(options: { provider?: boolean; hold?: boolean; expire?: boolean } = {}) { + const state: { + aborts: number + message?: ReturnType> + stream?: ReadableStreamDefaultController + prompted: boolean + } = { aborts: 0, 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 + if (options.provider) { + state.stream?.enqueue( + encoder.encode( + `data: ${JSON.stringify({ + type: "session.error", + properties: { + sessionID, + error: { + name: "ProviderError", + data: { message: "provider failed first" }, + }, + }, + })}\n\n`, + ), + ) + } + if (options.hold) { + state.message = Promise.withResolvers() + return state.message.promise + } + return Response.json({}) + } + if (request.method === "POST" && url.pathname.endsWith("/abort")) { + state.aborts++ + if (!options.expire) { + state.stream?.enqueue( + encoder.encode( + `data: ${JSON.stringify({ + type: "session.status", + properties: { + sessionID, + status: { type: "idle" }, + }, + })}\n\n`, + ), + ) + } + state.message?.resolve(Response.json({})) + 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") + const invocation = output.findIndex((event) => event.type === "invocation_complete") + expect(output.filter((event) => event.type === "session_error")).toHaveLength(1) + expect(output.filter((event) => event.type === "session_complete")).toHaveLength(1) + expect(output.filter((event) => event.type === "invocation_complete")).toHaveLength(1) + expect(error).toBeGreaterThan(-1) + expect(complete).toBeGreaterThan(error) + expect(invocation).toBeGreaterThan(complete) + expect(output[error]).toMatchObject({ + reason, + code: String(code), + }) + expect(output[invocation]).toMatchObject({ + invocationID: output[error].invocationID, + sessionID, + status: "error", + }) + expect(output[error].reason).not.toBe("timeout") + } finally { + clearTimeout(timeout) + proc.kill("SIGKILL") + mock.app.stop(true) + } + }, + 20_000, + ) + + test("a provider error remains the canonical terminal error when a signal follows", async () => { + const mock = await server({ provider: 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) + + 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("grace expiry completes the session and invocation before hard termination", async () => { + const mock = await server({ expire: 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") + + 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) => ["session_error", "session_complete", "invocation_complete"].includes(String(event.type))) + .map((event) => event.type), + ).toEqual(["session_error", "session_complete", "invocation_complete"]) + expect(output.find((event) => event.type === "invocation_complete")).toMatchObject({ + sessionID, + status: "error", + }) + } 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) +}) 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"]) + }) +}) 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("") + }) })