From 0b2d9246d3ba10542d527a98a685796e0b771ced Mon Sep 17 00:00:00 2001 From: Bulat Yapparov Date: Sun, 19 Jul 2026 19:20:55 +0100 Subject: [PATCH 1/6] feat(cli): add invocation lifecycle events --- EVENTS.md | 62 +++++- packages/cli/src/cli/cmd/run.ts | 37 ++-- packages/cli/src/cli/invocation.ts | 98 ++++++++++ packages/cli/src/headless.ts | 10 +- packages/cli/src/index.ts | 10 +- .../cli/test/cli/exception-handler.test.ts | 10 +- .../cli/test/cli/invocation-lifecycle.test.ts | 184 ++++++++++++++++++ 7 files changed, 391 insertions(+), 20 deletions(-) create mode 100644 packages/cli/src/cli/invocation.ts create mode 100644 packages/cli/test/cli/invocation-lifecycle.test.ts diff --git a/EVENTS.md b/EVENTS.md index 2b41488..fad924a 100644 --- a/EVENTS.md +++ b/EVENTS.md @@ -6,14 +6,74 @@ When running `aictrl run --format json`, the CLI emits newline-delimited JSON (N { "type": "", "timestamp": 1741500000000, + "invocationID": "7d142250-8bdc-43df-99af-efa252db62a7", "sessionID": "session_01abc..." } ``` -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. +`invocationID` is present on every event from `run --format json`. `sessionID` is present only after a real session has been created; invocation events never fabricate one. + +The schema is versioned via `invocation_start.schemaVersion` and `session_start.schemaVersion`. This document describes **schema version `"1"`**. Consumers should pin to this version and treat unknown fields as forward-compatible additions. ## Lifecycle Events +### `invocation_start` + +Emitted once, before piped stdin is read and before run validation or bootstrap begins. + +```json +{ + "type": "invocation_start", + "timestamp": 1741500000000, + "schemaVersion": "1", + "invocationID": "7d142250-8bdc-43df-99af-efa252db62a7" +} +``` + +This event intentionally has no `sessionID`, because a session does not exist yet. + +### `invocation_error` + +Emitted for a fatal error before session creation, immediately before `invocation_complete`. + +```json +{ + "type": "invocation_error", + "timestamp": 1741500000001, + "schemaVersion": "1", + "invocationID": "7d142250-8bdc-43df-99af-efa252db62a7", + "phase": "validation", + "code": "INVOCATION_FILE_NOT_FOUND", + "message": "File not found: input.txt" +} +``` + +- `phase` (string, **required**) — one of `parse`, `validation`, `stdin`, `bootstrap`, or `session`. +- `code` (string, **required**) — stable machine-readable error category. +- `message` (string, **required**) — human-readable error message. + +Errors after a real session has been created use `session_error` instead. + +### `invocation_complete` + +Emitted exactly once for every started invocation. + +```json +{ + "type": "invocation_complete", + "timestamp": 1741500000123, + "schemaVersion": "1", + "invocationID": "7d142250-8bdc-43df-99af-efa252db62a7", + "sessionID": "session_01abc...", + "status": "completed", + "durationMs": 123 +} +``` + +- `status` (string, **required**) — `completed` or `error`. +- `durationMs` (number, **required**) — elapsed invocation time. +- `sessionID` (string, optional) — included only when a real session was created. + ### `session_start` Emitted once when the session begins. diff --git a/packages/cli/src/cli/cmd/run.ts b/packages/cli/src/cli/cmd/run.ts index 02bd79a..e8f0e4f 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 { Invocation } from "../invocation" type ToolProps = { input: Tool.InferParameters @@ -375,6 +376,14 @@ export const RunCommand = cmd({ }) }, handler: async (args) => { + Invocation.phase("validation") + + function fail(message: string, code: string) { + Invocation.abort(message, code) + UI.error(message) + process.exit(1) + } + let message = [...args.message, ...(args["--"] || [])] .map((arg) => (arg.includes(" ") ? `"${arg.replace(/"/g, '\\"')}"` : arg)) .join(" ") @@ -386,8 +395,7 @@ export const RunCommand = cmd({ process.chdir(args.dir) return process.cwd() } catch { - UI.error("Failed to change directory to " + args.dir) - process.exit(1) + fail("Failed to change directory to " + args.dir, "INVOCATION_INVALID_DIRECTORY") } })() @@ -398,8 +406,7 @@ export const RunCommand = cmd({ for (const filePath of list) { const resolvedPath = path.resolve(process.cwd(), filePath) if (!(await Filesystem.exists(resolvedPath))) { - UI.error(`File not found: ${filePath}`) - process.exit(1) + fail(`File not found: ${filePath}`, "INVOCATION_FILE_NOT_FOUND") } const mime = (await Filesystem.isDir(resolvedPath)) ? "application/x-directory" : "text/plain" @@ -413,16 +420,18 @@ export const RunCommand = cmd({ } } - if (!process.stdin.isTTY) message += "\n" + (await Bun.stdin.text()) + if (!process.stdin.isTTY) { + Invocation.phase("stdin") + message += "\n" + (await Bun.stdin.text()) + Invocation.phase("validation") + } if (message.trim().length === 0 && !args.command) { - UI.error("You must provide a message or a command") - process.exit(1) + fail("You must provide a message or a command", "INVOCATION_EMPTY_INPUT") } if (args.fork && !args.continue && !args.session) { - UI.error("--fork requires --continue or --session") - process.exit(1) + fail("--fork requires --continue or --session", "INVOCATION_INVALID_ARGUMENTS") } const rules: PermissionNext.Ruleset = [ @@ -508,7 +517,9 @@ export const RunCommand = cmd({ function emit(type: string, data: Record) { if (args.format === "json") { - Stdout.write(JSON.stringify({ type, timestamp: Date.now(), sessionID, ...data }) + EOL) + Stdout.write( + JSON.stringify({ type, timestamp: Date.now(), invocationID: Invocation.id, sessionID, ...data }) + EOL, + ) return true } return false @@ -811,11 +822,12 @@ export const RunCommand = cmd({ })() const agent = agentInfo.name + Invocation.phase("session") const sessionID = await session(sdk) if (!sessionID) { - UI.error("Session not found") - process.exit(1) + fail("Session not found", "INVOCATION_SESSION_CREATE_FAILED") } + Invocation.link(sessionID) await share(sdk, sessionID) emit("session_start", { @@ -936,6 +948,7 @@ export const RunCommand = cmd({ return await execute(sdk) } + Invocation.phase("bootstrap") await bootstrap(process.cwd(), async () => { const sdk = { session: { diff --git a/packages/cli/src/cli/invocation.ts b/packages/cli/src/cli/invocation.ts new file mode 100644 index 0000000..d6a2b6f --- /dev/null +++ b/packages/cli/src/cli/invocation.ts @@ -0,0 +1,98 @@ +import { EOL } from "os" +import { SCHEMA_VERSION } from "./cmd/run.errors" + +export type InvocationPhase = "parse" | "validation" | "stdin" | "bootstrap" | "session" + +function create(argv: string[]) { + const run = argv.indexOf("run") + const json = + run !== -1 && + argv.slice(run + 1).some((arg, index, args) => { + if (arg === "--format=json") return true + return arg === "--format" && args[index + 1] === "json" + }) + const invocationID = json ? crypto.randomUUID() : undefined + const started = Date.now() + let phase: InvocationPhase = "parse" + let sessionID: string | undefined + let failed = false + let completed = false + + function emit(type: string, data: Record = {}) { + if (!invocationID) return + process.stdout.write( + JSON.stringify({ + type, + timestamp: Date.now(), + schemaVersion: SCHEMA_VERSION, + invocationID, + ...data, + }) + EOL, + ) + } + + if (invocationID) emit("invocation_start") + + return { + get id() { + return invocationID + }, + phase(next: InvocationPhase) { + phase = next + }, + link(id: string) { + sessionID = id + }, + error(error: unknown, code = "INVOCATION_FAILED") { + if (!invocationID || sessionID || failed || completed) return + failed = true + emit("invocation_error", { + phase, + code, + message: + error instanceof Error + ? error.message + : error && typeof error === "object" && "message" in error + ? String(error.message) + : String(error), + }) + }, + complete() { + if (!invocationID || completed) return + completed = true + emit("invocation_complete", { + status: failed ? "error" : "completed", + durationMs: Date.now() - started, + ...(sessionID ? { sessionID } : {}), + }) + }, + } +} + +let current = create([]) + +export const Invocation = { + get id() { + return current.id + }, + phase(next: InvocationPhase) { + current.phase(next) + }, + link(id: string) { + current.link(id) + }, + error(error: unknown, code?: string) { + current.error(error, code) + }, + abort(error: unknown, code?: string) { + current.error(error, code) + current.complete() + }, + complete() { + current.complete() + }, +} + +export function startInvocation(argv = process.argv.slice(2)) { + current = create(argv) +} diff --git a/packages/cli/src/headless.ts b/packages/cli/src/headless.ts index f9140aa..4f4346e 100644 --- a/packages/cli/src/headless.ts +++ b/packages/cli/src/headless.ts @@ -20,8 +20,12 @@ import { Global } from "./global" import { JsonMigration } from "./storage/json-migration" import { Database } from "./storage/db" import { Shutdown } from "./cli/shutdown" +import { Invocation, startInvocation } from "./cli/invocation" + +startInvocation() process.on("unhandledRejection", (e) => { + Invocation.abort(e) Log.Default.error("rejection", { e: e instanceof Error ? e.message : e, stack: e instanceof Error ? e.stack : undefined, @@ -30,6 +34,7 @@ process.on("unhandledRejection", (e) => { }) process.on("uncaughtException", (e) => { + Invocation.abort(e) Log.Default.error("exception", { e: e instanceof Error ? e.message : e, stack: e instanceof Error ? e.stack : undefined, @@ -119,9 +124,10 @@ cli = cli msg?.startsWith("Invalid values:") ) { if (err) throw err - cli.showHelp("log") + if (!Invocation.id) cli.showHelp("log") } if (err) throw err + Invocation.abort(msg ?? "Invalid command line arguments", "INVOCATION_PARSE_ERROR") process.exit(1) }) .strict() @@ -129,6 +135,7 @@ cli = cli try { await cli.parse() } catch (e) { + Invocation.error(e) let data: Record = {} if (e instanceof NamedError) { const obj = e.toObject() @@ -155,6 +162,7 @@ try { } process.exitCode = 1 } finally { + Invocation.complete() await Shutdown.flush() process.exit() } diff --git a/packages/cli/src/index.ts b/packages/cli/src/index.ts index e2d4b9c..a0e0910 100644 --- a/packages/cli/src/index.ts +++ b/packages/cli/src/index.ts @@ -30,8 +30,12 @@ import { Global } from "./global" import { JsonMigration } from "./storage/json-migration" import { Database } from "./storage/db" import { Shutdown } from "./cli/shutdown" +import { Invocation, startInvocation } from "./cli/invocation" + +startInvocation() process.on("unhandledRejection", (e) => { + Invocation.abort(e) Log.Default.error("rejection", { e: e instanceof Error ? e.message : e, stack: e instanceof Error ? e.stack : undefined, @@ -40,6 +44,7 @@ process.on("unhandledRejection", (e) => { }) process.on("uncaughtException", (e) => { + Invocation.abort(e) Log.Default.error("exception", { e: e instanceof Error ? e.message : e, stack: e instanceof Error ? e.stack : undefined, @@ -162,9 +167,10 @@ cli = cli msg?.startsWith("Invalid values:") ) { if (err) throw err - cli.showHelp("log") + if (!Invocation.id) cli.showHelp("log") } if (err) throw err + Invocation.abort(msg ?? "Invalid command line arguments", "INVOCATION_PARSE_ERROR") process.exit(1) }) .strict() @@ -172,6 +178,7 @@ cli = cli try { await cli.parse() } catch (e) { + Invocation.error(e) let data: Record = {} if (e instanceof NamedError) { const obj = e.toObject() @@ -209,6 +216,7 @@ try { } process.exitCode = 1 } finally { + Invocation.complete() // Some subprocesses don't react properly to SIGTERM and similar signals. // Most notably, some docker-container-based MCP servers don't handle such signals unless // run using `docker run --init`. diff --git a/packages/cli/test/cli/exception-handler.test.ts b/packages/cli/test/cli/exception-handler.test.ts index 8d78a02..a2b9186 100644 --- a/packages/cli/test/cli/exception-handler.test.ts +++ b/packages/cli/test/cli/exception-handler.test.ts @@ -75,9 +75,9 @@ describe("exception handler exit behavior", () => { // Both should find process.exit(1) within the handler (before the next major block) expect(exitAfterUncaught).toBeGreaterThan(uncaughtIdx) expect(exitAfterRejection).toBeGreaterThan(rejectionIdx) - // Handlers are close together, so exit should be within ~200 chars - expect(exitAfterUncaught - uncaughtIdx).toBeLessThan(200) - expect(exitAfterRejection - rejectionIdx).toBeLessThan(200) + // Handlers are close together, allowing room for invocation lifecycle cleanup. + expect(exitAfterUncaught - uncaughtIdx).toBeLessThan(250) + expect(exitAfterRejection - rejectionIdx).toBeLessThan(250) }) test("headless.ts exception handlers include process.exit(1)", async () => { @@ -94,8 +94,8 @@ describe("exception handler exit behavior", () => { expect(exitAfterUncaught).toBeGreaterThan(uncaughtIdx) expect(exitAfterRejection).toBeGreaterThan(rejectionIdx) - expect(exitAfterUncaught - uncaughtIdx).toBeLessThan(200) - expect(exitAfterRejection - rejectionIdx).toBeLessThan(200) + expect(exitAfterUncaught - uncaughtIdx).toBeLessThan(250) + expect(exitAfterRejection - rejectionIdx).toBeLessThan(250) }) test("exception handlers log stack traces", async () => { diff --git a/packages/cli/test/cli/invocation-lifecycle.test.ts b/packages/cli/test/cli/invocation-lifecycle.test.ts new file mode 100644 index 0000000..6cdf44d --- /dev/null +++ b/packages/cli/test/cli/invocation-lifecycle.test.ts @@ -0,0 +1,184 @@ +import { describe, expect, test } from "bun:test" +import path from "path" +import { tmpdir } from "../fixture/fixture" + +const CLI = path.resolve(import.meta.dir, "../../src/headless.ts") + +function events(stdout: string) { + return stdout + .split("\n") + .filter(Boolean) + .map((line) => JSON.parse(line) as Record) +} + +function spawn(args: string[], cwd = process.cwd()) { + return Bun.spawn(["bun", "run", "--conditions=browser", CLI, ...args], { + cwd, + env: { + ...process.env, + ANTHROPIC_API_KEY: "", + OPENAI_API_KEY: "", + }, + stdin: "ignore", + stdout: "pipe", + stderr: "pipe", + }) +} + +async function output(proc: ReturnType) { + const [stdout, stderr, code] = await Promise.all([ + new Response(proc.stdout).text(), + new Response(proc.stderr).text(), + proc.exited, + ]) + return { events: events(stdout), stderr, code } +} + +function expectFailure(result: Awaited>, phase: string) { + expect(result.code).not.toBe(0) + expect(result.events.map((event) => event.type)).toEqual([ + "invocation_start", + "invocation_error", + "invocation_complete", + ]) + + const [start, error, complete] = result.events + expect(start.schemaVersion).toBe("1") + expect(typeof start.invocationID).toBe("string") + expect(start).not.toHaveProperty("sessionID") + expect(error.invocationID).toBe(start.invocationID) + expect(error.phase).toBe(phase) + expect(error).not.toHaveProperty("sessionID") + expect(typeof error.message).toBe("string") + expect(complete.invocationID).toBe(start.invocationID) + expect(complete.status).toBe("error") + expect(complete).not.toHaveProperty("sessionID") + expect(result.events.filter((event) => event.type === "invocation_complete")).toHaveLength(1) +} + +describe("run --format json invocation lifecycle (#90)", () => { + test("emits invocation_start before waiting for piped stdin", async () => { + const proc = Bun.spawn(["bun", "run", "--conditions=browser", CLI, "run", "--format", "json"], { + cwd: process.cwd(), + env: process.env, + stdin: "pipe", + stdout: "pipe", + stderr: "pipe", + }) + const reader = proc.stdout.getReader() + const timeout = setTimeout(() => proc.kill(), 10_000) + const first = await reader.read() + clearTimeout(timeout) + proc.kill() + await proc.exited + + expect(first.done).toBe(false) + expect(events(new TextDecoder().decode(first.value))[0]).toMatchObject({ + type: "invocation_start", + schemaVersion: "1", + }) + }, 15_000) + + test("reports an invalid directory as a validation error", async () => { + expectFailure( + await output(spawn(["run", "--format", "json", "--dir", "/missing/aictrl-90", "prompt"])), + "validation", + ) + }) + + test("reports a missing file as a validation error", async () => { + expectFailure( + await output(spawn(["run", "--format", "json", "--file", "/missing/aictrl-90.txt", "prompt"])), + "validation", + ) + }) + + test("reports empty input as a validation error", async () => { + expectFailure(await output(spawn(["run", "--format", "json"])), "validation") + }) + + test("reports argument parsing failure", async () => { + expectFailure(await output(spawn(["run", "--format", "json", "--unknown-option"])), "parse") + }) + + test("reports bootstrap failure", async () => { + await using tmp = await tmpdir() + await Bun.write(path.join(tmp.path, "aictrl.json"), "{") + expectFailure(await output(spawn(["run", "--format", "json", "prompt"], tmp.path)), "bootstrap") + }) + + test("reports session creation failure", async () => { + const server = Bun.serve({ + port: 0, + fetch(req) { + const url = new URL(req.url) + if (req.method === "GET" && url.pathname === "/event") { + return new Response("", { headers: { "content-type": "text/event-stream" } }) + } + if (req.method === "POST" && url.pathname === "/session") { + return Response.json({ error: "session create failed" }, { status: 500 }) + } + return Response.json({}) + }, + }) + + try { + expectFailure( + await output(spawn(["run", "--format", "json", "--attach", `http://localhost:${server.port}`, "prompt"])), + "session", + ) + } finally { + server.stop(true) + } + }) + + test("links the invocation to a created session and carries invocationID on session events", async () => { + const id = "sess-invocation-90" + const server = Bun.serve({ + port: 0, + fetch(req) { + const url = new URL(req.url) + if (req.method === "GET" && url.pathname === "/event") { + return new Response( + `data: ${JSON.stringify({ + type: "session.status", + properties: { sessionID: id, status: { type: "idle" } }, + })}\n\n`, + { headers: { "content-type": "text/event-stream" } }, + ) + } + if (req.method === "POST" && url.pathname === "/session") return Response.json({ id }) + if (req.method === "GET" && url.pathname === "/config") return Response.json({}) + if (req.method === "POST" && url.pathname.endsWith("/message")) return Response.json({}) + return Response.json({}) + }, + }) + + try { + const result = await output( + spawn(["run", "--format", "json", "--attach", `http://localhost:${server.port}`, "prompt"]), + ) + expect(result.code).toBe(0) + const start = result.events.find((event) => event.type === "invocation_start") + const session = result.events.find((event) => event.type === "session_start") + const complete = result.events.find((event) => event.type === "invocation_complete") + expect(session).toMatchObject({ + invocationID: start?.invocationID, + sessionID: id, + }) + expect(complete).toMatchObject({ + invocationID: start?.invocationID, + sessionID: id, + status: "completed", + }) + expect(result.events.filter((event) => event.type === "invocation_complete")).toHaveLength(1) + expect( + result.events + .filter((event) => String(event.type).startsWith("session_")) + .every((event) => event.invocationID === start?.invocationID), + ).toBe(true) + } finally { + server.stop(true) + } + }) +}) From 0c8a3108e4520b2123ff913656b24726426e1d99 Mon Sep 17 00:00:00 2001 From: Bulat Yapparov Date: Sun, 19 Jul 2026 19:39:54 +0100 Subject: [PATCH 2/6] fix(cli): harden invocation lifecycle failures --- EVENTS.md | 7 ++- packages/cli/src/cli/cmd/run.ts | 15 ++--- packages/cli/src/cli/invocation.ts | 42 +++++++++---- packages/cli/src/headless.ts | 8 ++- packages/cli/src/index.ts | 8 ++- .../cli/test/cli/invocation-lifecycle.test.ts | 59 +++++++++++++++++-- 6 files changed, 109 insertions(+), 30 deletions(-) diff --git a/EVENTS.md b/EVENTS.md index fad924a..578945e 100644 --- a/EVENTS.md +++ b/EVENTS.md @@ -44,15 +44,16 @@ Emitted for a fatal error before session creation, immediately before `invocatio "invocationID": "7d142250-8bdc-43df-99af-efa252db62a7", "phase": "validation", "code": "INVOCATION_FILE_NOT_FOUND", - "message": "File not found: input.txt" + "message": "Invocation failed during validation" } ``` - `phase` (string, **required**) — one of `parse`, `validation`, `stdin`, `bootstrap`, or `session`. - `code` (string, **required**) — stable machine-readable error category. -- `message` (string, **required**) — human-readable error message. +- `message` (string, **required**) — sanitized human-readable phase summary. Details remain on stderr and in the log. -Errors after a real session has been created use `session_error` instead. +Errors after a real session has been created use `session_error` instead. They also set +`invocation_complete.status` to `error`, so the invocation result always agrees with the process result. ### `invocation_complete` diff --git a/packages/cli/src/cli/cmd/run.ts b/packages/cli/src/cli/cmd/run.ts index e8f0e4f..9eb811b 100644 --- a/packages/cli/src/cli/cmd/run.ts +++ b/packages/cli/src/cli/cmd/run.ts @@ -378,9 +378,10 @@ export const RunCommand = cmd({ handler: async (args) => { Invocation.phase("validation") - function fail(message: string, code: string) { + async function fail(message: string, code: string) { Invocation.abort(message, code) UI.error(message) + await Invocation.flush() process.exit(1) } @@ -388,14 +389,14 @@ export const RunCommand = cmd({ .map((arg) => (arg.includes(" ") ? `"${arg.replace(/"/g, '\\"')}"` : arg)) .join(" ") - const directory = (() => { + const directory = await (async () => { if (!args.dir) return undefined if (args.attach) return args.dir try { process.chdir(args.dir) return process.cwd() } catch { - fail("Failed to change directory to " + args.dir, "INVOCATION_INVALID_DIRECTORY") + await fail("Failed to change directory to " + args.dir, "INVOCATION_INVALID_DIRECTORY") } })() @@ -406,7 +407,7 @@ export const RunCommand = cmd({ for (const filePath of list) { const resolvedPath = path.resolve(process.cwd(), filePath) if (!(await Filesystem.exists(resolvedPath))) { - fail(`File not found: ${filePath}`, "INVOCATION_FILE_NOT_FOUND") + await fail(`File not found: ${filePath}`, "INVOCATION_FILE_NOT_FOUND") } const mime = (await Filesystem.isDir(resolvedPath)) ? "application/x-directory" : "text/plain" @@ -427,11 +428,11 @@ export const RunCommand = cmd({ } if (message.trim().length === 0 && !args.command) { - fail("You must provide a message or a command", "INVOCATION_EMPTY_INPUT") + await fail("You must provide a message or a command", "INVOCATION_EMPTY_INPUT") } if (args.fork && !args.continue && !args.session) { - fail("--fork requires --continue or --session", "INVOCATION_INVALID_ARGUMENTS") + await fail("--fork requires --continue or --session", "INVOCATION_INVALID_ARGUMENTS") } const rules: PermissionNext.Ruleset = [ @@ -825,7 +826,7 @@ export const RunCommand = cmd({ Invocation.phase("session") const sessionID = await session(sdk) if (!sessionID) { - fail("Session not found", "INVOCATION_SESSION_CREATE_FAILED") + await fail("Session not found", "INVOCATION_SESSION_CREATE_FAILED") } Invocation.link(sessionID) await share(sdk, sessionID) diff --git a/packages/cli/src/cli/invocation.ts b/packages/cli/src/cli/invocation.ts index d6a2b6f..4272702 100644 --- a/packages/cli/src/cli/invocation.ts +++ b/packages/cli/src/cli/invocation.ts @@ -4,7 +4,19 @@ import { SCHEMA_VERSION } from "./cmd/run.errors" export type InvocationPhase = "parse" | "validation" | "stdin" | "bootstrap" | "session" function create(argv: string[]) { - const run = argv.indexOf("run") + const run = (() => { + for (let index = 0; index < argv.length; index++) { + const arg = argv[index] + if (arg === "--print-logs") continue + if (arg === "--log-level") { + index++ + continue + } + if (arg.startsWith("--log-level=")) continue + return arg === "run" ? index : -1 + } + return -1 + })() const json = run !== -1 && argv.slice(run + 1).some((arg, index, args) => { @@ -17,17 +29,23 @@ function create(argv: string[]) { let sessionID: string | undefined let failed = false let completed = false + let writes = Promise.resolve() function emit(type: string, data: Record = {}) { if (!invocationID) return - process.stdout.write( + const line = JSON.stringify({ type, timestamp: Date.now(), schemaVersion: SCHEMA_VERSION, invocationID, ...data, - }) + EOL, + }) + EOL + writes = writes.then( + () => + new Promise((resolve) => { + process.stdout.write(line, () => resolve()) + }), ) } @@ -43,18 +61,14 @@ function create(argv: string[]) { link(id: string) { sessionID = id }, - error(error: unknown, code = "INVOCATION_FAILED") { - if (!invocationID || sessionID || failed || completed) return + error(_error: unknown, code = "INVOCATION_FAILED") { + if (!invocationID || failed || completed) return failed = true + if (sessionID) return emit("invocation_error", { phase, code, - message: - error instanceof Error - ? error.message - : error && typeof error === "object" && "message" in error - ? String(error.message) - : String(error), + message: `Invocation failed during ${phase}`, }) }, complete() { @@ -66,6 +80,9 @@ function create(argv: string[]) { ...(sessionID ? { sessionID } : {}), }) }, + flush() { + return writes + }, } } @@ -91,6 +108,9 @@ export const Invocation = { complete() { current.complete() }, + flush() { + return current.flush() + }, } export function startInvocation(argv = process.argv.slice(2)) { diff --git a/packages/cli/src/headless.ts b/packages/cli/src/headless.ts index 4f4346e..789240d 100644 --- a/packages/cli/src/headless.ts +++ b/packages/cli/src/headless.ts @@ -24,21 +24,23 @@ import { Invocation, startInvocation } from "./cli/invocation" startInvocation() -process.on("unhandledRejection", (e) => { +process.on("unhandledRejection", async (e) => { Invocation.abort(e) Log.Default.error("rejection", { e: e instanceof Error ? e.message : e, stack: e instanceof Error ? e.stack : undefined, }) + await Invocation.flush() process.exit(1) }) -process.on("uncaughtException", (e) => { +process.on("uncaughtException", async (e) => { Invocation.abort(e) Log.Default.error("exception", { e: e instanceof Error ? e.message : e, stack: e instanceof Error ? e.stack : undefined, }) + await Invocation.flush() process.exit(1) }) @@ -128,6 +130,7 @@ cli = cli } if (err) throw err Invocation.abort(msg ?? "Invalid command line arguments", "INVOCATION_PARSE_ERROR") + if (Invocation.id) throw new Error(msg ?? "Invalid command line arguments") process.exit(1) }) .strict() @@ -163,6 +166,7 @@ try { process.exitCode = 1 } finally { Invocation.complete() + await Invocation.flush() await Shutdown.flush() process.exit() } diff --git a/packages/cli/src/index.ts b/packages/cli/src/index.ts index a0e0910..abaa7f2 100644 --- a/packages/cli/src/index.ts +++ b/packages/cli/src/index.ts @@ -34,21 +34,23 @@ import { Invocation, startInvocation } from "./cli/invocation" startInvocation() -process.on("unhandledRejection", (e) => { +process.on("unhandledRejection", async (e) => { Invocation.abort(e) Log.Default.error("rejection", { e: e instanceof Error ? e.message : e, stack: e instanceof Error ? e.stack : undefined, }) + await Invocation.flush() process.exit(1) }) -process.on("uncaughtException", (e) => { +process.on("uncaughtException", async (e) => { Invocation.abort(e) Log.Default.error("exception", { e: e instanceof Error ? e.message : e, stack: e instanceof Error ? e.stack : undefined, }) + await Invocation.flush() process.exit(1) }) @@ -171,6 +173,7 @@ cli = cli } if (err) throw err Invocation.abort(msg ?? "Invalid command line arguments", "INVOCATION_PARSE_ERROR") + if (Invocation.id) throw new Error(msg ?? "Invalid command line arguments") process.exit(1) }) .strict() @@ -217,6 +220,7 @@ try { process.exitCode = 1 } finally { Invocation.complete() + await Invocation.flush() // Some subprocesses don't react properly to SIGTERM and similar signals. // Most notably, some docker-container-based MCP servers don't handle such signals unless // run using `docker run --init`. diff --git a/packages/cli/test/cli/invocation-lifecycle.test.ts b/packages/cli/test/cli/invocation-lifecycle.test.ts index 6cdf44d..c228a33 100644 --- a/packages/cli/test/cli/invocation-lifecycle.test.ts +++ b/packages/cli/test/cli/invocation-lifecycle.test.ts @@ -49,7 +49,7 @@ function expectFailure(result: Awaited>, phase: string expect(error.invocationID).toBe(start.invocationID) expect(error.phase).toBe(phase) expect(error).not.toHaveProperty("sessionID") - expect(typeof error.message).toBe("string") + expect(error.message).toBe(`Invocation failed during ${phase}`) expect(complete.invocationID).toBe(start.invocationID) expect(complete.status).toBe("error") expect(complete).not.toHaveProperty("sessionID") @@ -87,10 +87,10 @@ describe("run --format json invocation lifecycle (#90)", () => { }) test("reports a missing file as a validation error", async () => { - expectFailure( - await output(spawn(["run", "--format", "json", "--file", "/missing/aictrl-90.txt", "prompt"])), - "validation", - ) + const missing = "/missing/" + "a".repeat(100_000) + const result = await output(spawn(["run", "--format", "json", "--file", missing, "prompt"])) + expectFailure(result, "validation") + expect(JSON.stringify(result.events)).not.toContain(missing) }) test("reports empty input as a validation error", async () => { @@ -101,6 +101,20 @@ describe("run --format json invocation lifecycle (#90)", () => { expectFailure(await output(spawn(["run", "--format", "json", "--unknown-option"])), "parse") }) + test("does not start a run invocation for a positional token on another command", async () => { + const proc = spawn(["session", "list", "run", "--format", "json"]) + const stdout = await new Response(proc.stdout).text() + await proc.exited + expect(stdout).not.toContain('"type":"invocation_') + }) + + test("recognizes run after global options", async () => { + expectFailure( + await output(spawn(["--log-level", "ERROR", "run", "--format", "json", "--dir", "/missing/aictrl-90", "prompt"])), + "validation", + ) + }) + test("reports bootstrap failure", async () => { await using tmp = await tmpdir() await Bun.write(path.join(tmp.path, "aictrl.json"), "{") @@ -181,4 +195,39 @@ describe("run --format json invocation lifecycle (#90)", () => { server.stop(true) } }) + + test("reports a post-session failure as an invocation error result", async () => { + const id = "sess-invocation-failure-90" + const server = Bun.serve({ + port: 0, + fetch(req) { + const url = new URL(req.url) + if (req.method === "GET" && url.pathname === "/event") { + return new Response("", { headers: { "content-type": "text/event-stream" } }) + } + if (req.method === "POST" && url.pathname === "/session") return Response.json({ id }) + if (req.method === "GET" && url.pathname === "/config") { + return new Response('{"post-session-secret"', { + headers: { "content-type": "application/json" }, + }) + } + if (req.method === "POST" && url.pathname.endsWith("/message")) return Response.json({}) + return Response.json({}) + }, + }) + + try { + const result = await output( + spawn(["run", "--format", "json", "--attach", `http://localhost:${server.port}`, "prompt"]), + ) + expect(result.code).not.toBe(0) + expect(result.events.find((event) => event.type === "invocation_complete")).toMatchObject({ + sessionID: id, + status: "error", + }) + expect(JSON.stringify(result.events)).not.toContain("post-session-secret") + } finally { + server.stop(true) + } + }) }) From 9d64f4747f8b0d008a4fe31e755def854c83ac7d Mon Sep 17 00:00:00 2001 From: Bulat Yapparov Date: Sun, 19 Jul 2026 20:02:25 +0100 Subject: [PATCH 3/6] fix(cli): detect boolean global invocation flags --- packages/cli/src/cli/invocation.ts | 1 + packages/cli/test/cli/invocation-lifecycle.test.ts | 4 ++++ 2 files changed, 5 insertions(+) diff --git a/packages/cli/src/cli/invocation.ts b/packages/cli/src/cli/invocation.ts index 4272702..46d0b95 100644 --- a/packages/cli/src/cli/invocation.ts +++ b/packages/cli/src/cli/invocation.ts @@ -8,6 +8,7 @@ function create(argv: string[]) { for (let index = 0; index < argv.length; index++) { const arg = argv[index] if (arg === "--print-logs") continue + if (arg.startsWith("--print-logs=")) continue if (arg === "--log-level") { index++ continue diff --git a/packages/cli/test/cli/invocation-lifecycle.test.ts b/packages/cli/test/cli/invocation-lifecycle.test.ts index c228a33..025bc51 100644 --- a/packages/cli/test/cli/invocation-lifecycle.test.ts +++ b/packages/cli/test/cli/invocation-lifecycle.test.ts @@ -113,6 +113,10 @@ describe("run --format json invocation lifecycle (#90)", () => { await output(spawn(["--log-level", "ERROR", "run", "--format", "json", "--dir", "/missing/aictrl-90", "prompt"])), "validation", ) + expectFailure( + await output(spawn(["--print-logs=false", "run", "--format", "json", "--dir", "/missing/aictrl-90", "prompt"])), + "validation", + ) }) test("reports bootstrap failure", async () => { From 2835d9a7ee978726cc882593dedf0b806133bd0e Mon Sep 17 00:00:00 2001 From: Bulat Yapparov Date: Sun, 19 Jul 2026 23:06:59 +0100 Subject: [PATCH 4/6] fix(cli): harden invocation finalization --- packages/cli/src/cli/invocation.ts | 14 ++-- packages/cli/src/headless.ts | 7 +- packages/cli/src/index.ts | 7 +- .../cli/test/cli/invocation-lifecycle.test.ts | 80 ++++++++++++++++++- 4 files changed, 92 insertions(+), 16 deletions(-) diff --git a/packages/cli/src/cli/invocation.ts b/packages/cli/src/cli/invocation.ts index 46d0b95..22cf24a 100644 --- a/packages/cli/src/cli/invocation.ts +++ b/packages/cli/src/cli/invocation.ts @@ -1,5 +1,6 @@ import { EOL } from "os" import { SCHEMA_VERSION } from "./cmd/run.errors" +import { Stdout } from "./stdout" export type InvocationPhase = "parse" | "validation" | "stdin" | "bootstrap" | "session" @@ -18,9 +19,10 @@ function create(argv: string[]) { } return -1 })() + const args = argv.slice(run + 1) const json = run !== -1 && - argv.slice(run + 1).some((arg, index, args) => { + args.slice(0, args.indexOf("--") === -1 ? undefined : args.indexOf("--")).some((arg, index, args) => { if (arg === "--format=json") return true return arg === "--format" && args[index + 1] === "json" }) @@ -42,12 +44,7 @@ function create(argv: string[]) { invocationID, ...data, }) + EOL - writes = writes.then( - () => - new Promise((resolve) => { - process.stdout.write(line, () => resolve()) - }), - ) + writes = writes.then(() => Stdout.write(line)).catch(() => {}) } if (invocationID) emit("invocation_start") @@ -112,6 +109,9 @@ export const Invocation = { flush() { return current.flush() }, + drain() { + return current.flush().catch(() => {}) + }, } export function startInvocation(argv = process.argv.slice(2)) { diff --git a/packages/cli/src/headless.ts b/packages/cli/src/headless.ts index 789240d..92162bf 100644 --- a/packages/cli/src/headless.ts +++ b/packages/cli/src/headless.ts @@ -30,7 +30,7 @@ process.on("unhandledRejection", async (e) => { e: e instanceof Error ? e.message : e, stack: e instanceof Error ? e.stack : undefined, }) - await Invocation.flush() + await Invocation.drain() process.exit(1) }) @@ -40,7 +40,7 @@ process.on("uncaughtException", async (e) => { e: e instanceof Error ? e.message : e, stack: e instanceof Error ? e.stack : undefined, }) - await Invocation.flush() + await Invocation.drain() process.exit(1) }) @@ -165,8 +165,9 @@ try { } process.exitCode = 1 } finally { + await Invocation.drain() Invocation.complete() - await Invocation.flush() + await Invocation.drain() await Shutdown.flush() process.exit() } diff --git a/packages/cli/src/index.ts b/packages/cli/src/index.ts index abaa7f2..e6152fa 100644 --- a/packages/cli/src/index.ts +++ b/packages/cli/src/index.ts @@ -40,7 +40,7 @@ process.on("unhandledRejection", async (e) => { e: e instanceof Error ? e.message : e, stack: e instanceof Error ? e.stack : undefined, }) - await Invocation.flush() + await Invocation.drain() process.exit(1) }) @@ -50,7 +50,7 @@ process.on("uncaughtException", async (e) => { e: e instanceof Error ? e.message : e, stack: e instanceof Error ? e.stack : undefined, }) - await Invocation.flush() + await Invocation.drain() process.exit(1) }) @@ -219,8 +219,9 @@ try { } process.exitCode = 1 } finally { + await Invocation.drain() Invocation.complete() - await Invocation.flush() + await Invocation.drain() // Some subprocesses don't react properly to SIGTERM and similar signals. // Most notably, some docker-container-based MCP servers don't handle such signals unless // run using `docker run --init`. diff --git a/packages/cli/test/cli/invocation-lifecycle.test.ts b/packages/cli/test/cli/invocation-lifecycle.test.ts index 025bc51..e9bc59b 100644 --- a/packages/cli/test/cli/invocation-lifecycle.test.ts +++ b/packages/cli/test/cli/invocation-lifecycle.test.ts @@ -1,8 +1,11 @@ import { describe, expect, test } from "bun:test" import path from "path" +import { pathToFileURL } from "url" import { tmpdir } from "../fixture/fixture" const CLI = path.resolve(import.meta.dir, "../../src/headless.ts") +const MAIN = path.resolve(import.meta.dir, "../../src/index.ts") +const INVOCATION = pathToFileURL(path.resolve(import.meta.dir, "../../src/cli/invocation.ts")).href function events(stdout: string) { return stdout @@ -11,8 +14,8 @@ function events(stdout: string) { .map((line) => JSON.parse(line) as Record) } -function spawn(args: string[], cwd = process.cwd()) { - return Bun.spawn(["bun", "run", "--conditions=browser", CLI, ...args], { +function spawn(args: string[], cwd = process.cwd(), cli = CLI) { + return Bun.spawn(["bun", "run", "--conditions=browser", cli, ...args], { cwd, env: { ...process.env, @@ -98,7 +101,18 @@ describe("run --format json invocation lifecycle (#90)", () => { }) test("reports argument parsing failure", async () => { - expectFailure(await output(spawn(["run", "--format", "json", "--unknown-option"])), "parse") + for (const cli of [CLI, MAIN]) { + const result = await output(spawn(["run", "--format", "json", "--unknown-option"], process.cwd(), cli)) + expectFailure(result, "parse") + expect(result.events.find((event) => event.type === "invocation_error")?.code).toBe("INVOCATION_PARSE_ERROR") + } + }) + + test("ignores JSON format tokens after the argument separator", async () => { + const proc = spawn(["run", "--dir", "/missing/aictrl-90", "--", "--format=json", "prompt"]) + const stdout = await new Response(proc.stdout).text() + await proc.exited + expect(stdout).not.toContain('"type":"invocation_') }) test("does not start a run invocation for a positional token on another command", async () => { @@ -234,4 +248,64 @@ describe("run --format json invocation lifecycle (#90)", () => { server.stop(true) } }) + + test("flush resolves when stdout rejects a lifecycle write", async () => { + const proc = Bun.spawn( + [ + "bun", + "-e", + ` + const module = await import(${JSON.stringify(INVOCATION)}) + const write = process.stdout.write.bind(process.stdout) + process.stdout.write = () => { + throw new Error("closed") + } + module.startInvocation(["run", "--format", "json"]) + module.Invocation.abort("failed") + await module.Invocation.flush() + process.stdout.write = write + process.stdout.write("flushed") + `, + ], + { stdout: "pipe", stderr: "pipe" }, + ) + const [stdout, code] = await Promise.all([new Response(proc.stdout).text(), proc.exited]) + expect(code).toBe(0) + expect(stdout).toBe("flushed") + }) + + test("late failure during pre-terminal drain produces one error completion", async () => { + const proc = Bun.spawn( + [ + "bun", + "-e", + ` + const module = await import(${JSON.stringify(INVOCATION)}) + const write = process.stdout.write.bind(process.stdout) + const lines = [] + process.stdout.write = (chunk, callback) => { + lines.push(String(chunk)) + setTimeout(() => callback?.(), 20) + return true + } + module.startInvocation(["run", "--format", "json"]) + const done = (async () => { + await module.Invocation.flush() + module.Invocation.complete() + await module.Invocation.flush() + })() + setTimeout(() => module.Invocation.abort("late"), 5) + await done + process.stdout.write = write + process.stdout.write(JSON.stringify(lines)) + `, + ], + { stdout: "pipe", stderr: "pipe" }, + ) + const [stdout, code] = await Promise.all([new Response(proc.stdout).text(), proc.exited]) + const result = JSON.parse(stdout).flatMap((line: string) => events(line)) + expect(code).toBe(0) + expect(result.filter((event: Record) => event.type === "invocation_complete")).toHaveLength(1) + expect(result.find((event: Record) => event.type === "invocation_complete")?.status).toBe("error") + }) }) From 6ece1b94e1bd3bb09ed6103c1aae3288796696eb Mon Sep 17 00:00:00 2001 From: Bulat Yapparov Date: Tue, 21 Jul 2026 09:52:49 +0100 Subject: [PATCH 5/6] refactor(cli): scope invocation lifecycle to run --- EVENTS.md | 6 +- packages/cli/src/cli/cmd/run.invocation.ts | 79 ++++++++++ packages/cli/src/cli/cmd/run.ts | 28 ++-- packages/cli/src/cli/invocation.ts | 119 --------------- packages/cli/src/headless.ts | 19 +-- packages/cli/src/index.ts | 19 +-- .../cli/test/cli/exception-handler.test.ts | 10 +- .../cli/test/cli/invocation-lifecycle.test.ts | 135 +----------------- 8 files changed, 113 insertions(+), 302 deletions(-) create mode 100644 packages/cli/src/cli/cmd/run.invocation.ts delete mode 100644 packages/cli/src/cli/invocation.ts diff --git a/EVENTS.md b/EVENTS.md index 578945e..cf7d652 100644 --- a/EVENTS.md +++ b/EVENTS.md @@ -1,6 +1,6 @@ # NDJSON Events -When running `aictrl run --format json`, the CLI emits newline-delimited JSON (NDJSON) events to stdout. Each line is a self-contained JSON object with the following base shape: +When the parsed `aictrl run --format json` handler starts, the CLI emits newline-delimited JSON (NDJSON) events to stdout. Each line is a self-contained JSON object with the following base shape: ```json { @@ -15,6 +15,8 @@ When running `aictrl run --format json`, the CLI emits newline-delimited JSON (N The schema is versioned via `invocation_start.schemaVersion` and `session_start.schemaVersion`. This document describes **schema version `"1"`**. Consumers should pin to this version and treat unknown fields as forward-compatible additions. +The invocation envelope covers accepted `run --format json` executions from the first line of the parsed handler through validation, bootstrap, session creation, and execution. Argument-parser failures, other commands, and process-global uncaught exceptions or unhandled rejections are outside this contract. + ## Lifecycle Events ### `invocation_start` @@ -48,7 +50,7 @@ Emitted for a fatal error before session creation, immediately before `invocatio } ``` -- `phase` (string, **required**) — one of `parse`, `validation`, `stdin`, `bootstrap`, or `session`. +- `phase` (string, **required**) — one of `validation`, `stdin`, `bootstrap`, or `session`. - `code` (string, **required**) — stable machine-readable error category. - `message` (string, **required**) — sanitized human-readable phase summary. Details remain on stderr and in the log. diff --git a/packages/cli/src/cli/cmd/run.invocation.ts b/packages/cli/src/cli/cmd/run.invocation.ts new file mode 100644 index 0000000..0db4e70 --- /dev/null +++ b/packages/cli/src/cli/cmd/run.invocation.ts @@ -0,0 +1,79 @@ +import { EOL } from "os" +import { Stdout } from "../stdout" +import { SCHEMA_VERSION } from "./run.errors" + +export type RunInvocationPhase = "validation" | "stdin" | "bootstrap" | "session" + +export function createRunInvocation(enabled: boolean) { + const id = enabled ? crypto.randomUUID() : undefined + const started = Date.now() + let phase: RunInvocationPhase = "validation" + let sessionID: string | undefined + let failed = false + let completed = false + let writes = Promise.resolve() + + function emit(type: string, data: Record = {}) { + if (!id) return + const line = + JSON.stringify({ + type, + timestamp: Date.now(), + schemaVersion: SCHEMA_VERSION, + invocationID: id, + ...data, + }) + EOL + writes = writes.then(() => Stdout.write(line)) + } + + emit("invocation_start") + + function error(_error: unknown, code = `INVOCATION_${phase.toUpperCase()}_FAILED`) { + if (!id || failed || completed) return + failed = true + if (sessionID) return + emit("invocation_error", { + phase, + code, + message: `Invocation failed during ${phase}`, + }) + } + + function complete() { + if (!id || completed) return + completed = true + emit("invocation_complete", { + status: failed ? "error" : "completed", + durationMs: Date.now() - started, + ...(sessionID ? { sessionID } : {}), + }) + } + + return { + id, + phase(next: RunInvocationPhase) { + phase = next + }, + link(id: string) { + sessionID = id + }, + error, + async abort(cause: unknown, code?: string) { + error(cause, code) + complete() + await writes + }, + async run(execution: Promise) { + try { + return await execution + } catch (cause) { + error(cause) + throw cause + } finally { + await writes + complete() + await writes + } + }, + } +} diff --git a/packages/cli/src/cli/cmd/run.ts b/packages/cli/src/cli/cmd/run.ts index 9eb811b..d7aae9b 100644 --- a/packages/cli/src/cli/cmd/run.ts +++ b/packages/cli/src/cli/cmd/run.ts @@ -38,7 +38,7 @@ import { BashTool } from "../../tool/bash" import { TodoWriteTool } from "../../tool/todo" import { Locale } from "../../util/locale" import { Stdout } from "../stdout" -import { Invocation } from "../invocation" +import { createRunInvocation } from "./run.invocation" type ToolProps = { input: Tool.InferParameters @@ -376,12 +376,11 @@ export const RunCommand = cmd({ }) }, handler: async (args) => { - Invocation.phase("validation") + const invocation = createRunInvocation(args.format === "json") async function fail(message: string, code: string) { - Invocation.abort(message, code) UI.error(message) - await Invocation.flush() + await invocation.abort(message, code) process.exit(1) } @@ -422,9 +421,9 @@ export const RunCommand = cmd({ } if (!process.stdin.isTTY) { - Invocation.phase("stdin") + invocation.phase("stdin") message += "\n" + (await Bun.stdin.text()) - Invocation.phase("validation") + invocation.phase("validation") } if (message.trim().length === 0 && !args.command) { @@ -519,7 +518,7 @@ export const RunCommand = cmd({ function emit(type: string, data: Record) { if (args.format === "json") { Stdout.write( - JSON.stringify({ type, timestamp: Date.now(), invocationID: Invocation.id, sessionID, ...data }) + EOL, + JSON.stringify({ type, timestamp: Date.now(), invocationID: invocation.id, sessionID, ...data }) + EOL, ) return true } @@ -690,6 +689,7 @@ 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. process.exitCode = 1 + invocation.error(props.error) if (!sessionErrorEmitted) { sessionErrorEmitted = true const classified = classifySessionError(props.error) @@ -823,12 +823,12 @@ export const RunCommand = cmd({ })() const agent = agentInfo.name - Invocation.phase("session") + invocation.phase("session") const sessionID = await session(sdk) if (!sessionID) { await fail("Session not found", "INVOCATION_SESSION_CREATE_FAILED") } - Invocation.link(sessionID) + invocation.link(sessionID) await share(sdk, sessionID) emit("session_start", { @@ -889,6 +889,7 @@ export const RunCommand = cmd({ }) console.error(e) process.exitCode = 1 + invocation.error(e) }) if (args.command) { @@ -939,18 +940,18 @@ export const RunCommand = cmd({ }) console.error(e) process.exitCode = 1 + invocation.error(e) }, ), ]) } + invocation.phase("bootstrap") if (args.attach) { - const sdk = createAictrlClient({ baseUrl: args.attach, directory }) - return await execute(sdk) + return await invocation.run(execute(createAictrlClient({ baseUrl: args.attach, directory }))) } - Invocation.phase("bootstrap") - await bootstrap(process.cwd(), async () => { + const execution = bootstrap(process.cwd(), async () => { const sdk = { session: { async list() { @@ -1030,5 +1031,6 @@ export const RunCommand = cmd({ } await execute(sdk) }) + await invocation.run(execution) }, }) diff --git a/packages/cli/src/cli/invocation.ts b/packages/cli/src/cli/invocation.ts deleted file mode 100644 index 22cf24a..0000000 --- a/packages/cli/src/cli/invocation.ts +++ /dev/null @@ -1,119 +0,0 @@ -import { EOL } from "os" -import { SCHEMA_VERSION } from "./cmd/run.errors" -import { Stdout } from "./stdout" - -export type InvocationPhase = "parse" | "validation" | "stdin" | "bootstrap" | "session" - -function create(argv: string[]) { - const run = (() => { - for (let index = 0; index < argv.length; index++) { - const arg = argv[index] - if (arg === "--print-logs") continue - if (arg.startsWith("--print-logs=")) continue - if (arg === "--log-level") { - index++ - continue - } - if (arg.startsWith("--log-level=")) continue - return arg === "run" ? index : -1 - } - return -1 - })() - const args = argv.slice(run + 1) - const json = - run !== -1 && - args.slice(0, args.indexOf("--") === -1 ? undefined : args.indexOf("--")).some((arg, index, args) => { - if (arg === "--format=json") return true - return arg === "--format" && args[index + 1] === "json" - }) - const invocationID = json ? crypto.randomUUID() : undefined - const started = Date.now() - let phase: InvocationPhase = "parse" - let sessionID: string | undefined - let failed = false - let completed = false - let writes = Promise.resolve() - - function emit(type: string, data: Record = {}) { - if (!invocationID) return - const line = - JSON.stringify({ - type, - timestamp: Date.now(), - schemaVersion: SCHEMA_VERSION, - invocationID, - ...data, - }) + EOL - writes = writes.then(() => Stdout.write(line)).catch(() => {}) - } - - if (invocationID) emit("invocation_start") - - return { - get id() { - return invocationID - }, - phase(next: InvocationPhase) { - phase = next - }, - link(id: string) { - sessionID = id - }, - error(_error: unknown, code = "INVOCATION_FAILED") { - if (!invocationID || failed || completed) return - failed = true - if (sessionID) return - emit("invocation_error", { - phase, - code, - message: `Invocation failed during ${phase}`, - }) - }, - complete() { - if (!invocationID || completed) return - completed = true - emit("invocation_complete", { - status: failed ? "error" : "completed", - durationMs: Date.now() - started, - ...(sessionID ? { sessionID } : {}), - }) - }, - flush() { - return writes - }, - } -} - -let current = create([]) - -export const Invocation = { - get id() { - return current.id - }, - phase(next: InvocationPhase) { - current.phase(next) - }, - link(id: string) { - current.link(id) - }, - error(error: unknown, code?: string) { - current.error(error, code) - }, - abort(error: unknown, code?: string) { - current.error(error, code) - current.complete() - }, - complete() { - current.complete() - }, - flush() { - return current.flush() - }, - drain() { - return current.flush().catch(() => {}) - }, -} - -export function startInvocation(argv = process.argv.slice(2)) { - current = create(argv) -} diff --git a/packages/cli/src/headless.ts b/packages/cli/src/headless.ts index 92162bf..f9140aa 100644 --- a/packages/cli/src/headless.ts +++ b/packages/cli/src/headless.ts @@ -20,27 +20,20 @@ import { Global } from "./global" import { JsonMigration } from "./storage/json-migration" import { Database } from "./storage/db" import { Shutdown } from "./cli/shutdown" -import { Invocation, startInvocation } from "./cli/invocation" -startInvocation() - -process.on("unhandledRejection", async (e) => { - Invocation.abort(e) +process.on("unhandledRejection", (e) => { Log.Default.error("rejection", { e: e instanceof Error ? e.message : e, stack: e instanceof Error ? e.stack : undefined, }) - await Invocation.drain() process.exit(1) }) -process.on("uncaughtException", async (e) => { - Invocation.abort(e) +process.on("uncaughtException", (e) => { Log.Default.error("exception", { e: e instanceof Error ? e.message : e, stack: e instanceof Error ? e.stack : undefined, }) - await Invocation.drain() process.exit(1) }) @@ -126,11 +119,9 @@ cli = cli msg?.startsWith("Invalid values:") ) { if (err) throw err - if (!Invocation.id) cli.showHelp("log") + cli.showHelp("log") } if (err) throw err - Invocation.abort(msg ?? "Invalid command line arguments", "INVOCATION_PARSE_ERROR") - if (Invocation.id) throw new Error(msg ?? "Invalid command line arguments") process.exit(1) }) .strict() @@ -138,7 +129,6 @@ cli = cli try { await cli.parse() } catch (e) { - Invocation.error(e) let data: Record = {} if (e instanceof NamedError) { const obj = e.toObject() @@ -165,9 +155,6 @@ try { } process.exitCode = 1 } finally { - await Invocation.drain() - Invocation.complete() - await Invocation.drain() await Shutdown.flush() process.exit() } diff --git a/packages/cli/src/index.ts b/packages/cli/src/index.ts index e6152fa..e2d4b9c 100644 --- a/packages/cli/src/index.ts +++ b/packages/cli/src/index.ts @@ -30,27 +30,20 @@ import { Global } from "./global" import { JsonMigration } from "./storage/json-migration" import { Database } from "./storage/db" import { Shutdown } from "./cli/shutdown" -import { Invocation, startInvocation } from "./cli/invocation" -startInvocation() - -process.on("unhandledRejection", async (e) => { - Invocation.abort(e) +process.on("unhandledRejection", (e) => { Log.Default.error("rejection", { e: e instanceof Error ? e.message : e, stack: e instanceof Error ? e.stack : undefined, }) - await Invocation.drain() process.exit(1) }) -process.on("uncaughtException", async (e) => { - Invocation.abort(e) +process.on("uncaughtException", (e) => { Log.Default.error("exception", { e: e instanceof Error ? e.message : e, stack: e instanceof Error ? e.stack : undefined, }) - await Invocation.drain() process.exit(1) }) @@ -169,11 +162,9 @@ cli = cli msg?.startsWith("Invalid values:") ) { if (err) throw err - if (!Invocation.id) cli.showHelp("log") + cli.showHelp("log") } if (err) throw err - Invocation.abort(msg ?? "Invalid command line arguments", "INVOCATION_PARSE_ERROR") - if (Invocation.id) throw new Error(msg ?? "Invalid command line arguments") process.exit(1) }) .strict() @@ -181,7 +172,6 @@ cli = cli try { await cli.parse() } catch (e) { - Invocation.error(e) let data: Record = {} if (e instanceof NamedError) { const obj = e.toObject() @@ -219,9 +209,6 @@ try { } process.exitCode = 1 } finally { - await Invocation.drain() - Invocation.complete() - await Invocation.drain() // Some subprocesses don't react properly to SIGTERM and similar signals. // Most notably, some docker-container-based MCP servers don't handle such signals unless // run using `docker run --init`. diff --git a/packages/cli/test/cli/exception-handler.test.ts b/packages/cli/test/cli/exception-handler.test.ts index a2b9186..8d78a02 100644 --- a/packages/cli/test/cli/exception-handler.test.ts +++ b/packages/cli/test/cli/exception-handler.test.ts @@ -75,9 +75,9 @@ describe("exception handler exit behavior", () => { // Both should find process.exit(1) within the handler (before the next major block) expect(exitAfterUncaught).toBeGreaterThan(uncaughtIdx) expect(exitAfterRejection).toBeGreaterThan(rejectionIdx) - // Handlers are close together, allowing room for invocation lifecycle cleanup. - expect(exitAfterUncaught - uncaughtIdx).toBeLessThan(250) - expect(exitAfterRejection - rejectionIdx).toBeLessThan(250) + // Handlers are close together, so exit should be within ~200 chars + expect(exitAfterUncaught - uncaughtIdx).toBeLessThan(200) + expect(exitAfterRejection - rejectionIdx).toBeLessThan(200) }) test("headless.ts exception handlers include process.exit(1)", async () => { @@ -94,8 +94,8 @@ describe("exception handler exit behavior", () => { expect(exitAfterUncaught).toBeGreaterThan(uncaughtIdx) expect(exitAfterRejection).toBeGreaterThan(rejectionIdx) - expect(exitAfterUncaught - uncaughtIdx).toBeLessThan(250) - expect(exitAfterRejection - rejectionIdx).toBeLessThan(250) + expect(exitAfterUncaught - uncaughtIdx).toBeLessThan(200) + expect(exitAfterRejection - rejectionIdx).toBeLessThan(200) }) test("exception handlers log stack traces", async () => { diff --git a/packages/cli/test/cli/invocation-lifecycle.test.ts b/packages/cli/test/cli/invocation-lifecycle.test.ts index e9bc59b..445ef4d 100644 --- a/packages/cli/test/cli/invocation-lifecycle.test.ts +++ b/packages/cli/test/cli/invocation-lifecycle.test.ts @@ -1,11 +1,8 @@ import { describe, expect, test } from "bun:test" import path from "path" -import { pathToFileURL } from "url" import { tmpdir } from "../fixture/fixture" const CLI = path.resolve(import.meta.dir, "../../src/headless.ts") -const MAIN = path.resolve(import.meta.dir, "../../src/index.ts") -const INVOCATION = pathToFileURL(path.resolve(import.meta.dir, "../../src/cli/invocation.ts")).href function events(stdout: string) { return stdout @@ -14,8 +11,8 @@ function events(stdout: string) { .map((line) => JSON.parse(line) as Record) } -function spawn(args: string[], cwd = process.cwd(), cli = CLI) { - return Bun.spawn(["bun", "run", "--conditions=browser", cli, ...args], { +function spawn(args: string[], cwd = process.cwd()) { + return Bun.spawn(["bun", "run", "--conditions=browser", CLI, ...args], { cwd, env: { ...process.env, @@ -100,37 +97,8 @@ describe("run --format json invocation lifecycle (#90)", () => { expectFailure(await output(spawn(["run", "--format", "json"])), "validation") }) - test("reports argument parsing failure", async () => { - for (const cli of [CLI, MAIN]) { - const result = await output(spawn(["run", "--format", "json", "--unknown-option"], process.cwd(), cli)) - expectFailure(result, "parse") - expect(result.events.find((event) => event.type === "invocation_error")?.code).toBe("INVOCATION_PARSE_ERROR") - } - }) - - test("ignores JSON format tokens after the argument separator", async () => { - const proc = spawn(["run", "--dir", "/missing/aictrl-90", "--", "--format=json", "prompt"]) - const stdout = await new Response(proc.stdout).text() - await proc.exited - expect(stdout).not.toContain('"type":"invocation_') - }) - - test("does not start a run invocation for a positional token on another command", async () => { - const proc = spawn(["session", "list", "run", "--format", "json"]) - const stdout = await new Response(proc.stdout).text() - await proc.exited - expect(stdout).not.toContain('"type":"invocation_') - }) - - test("recognizes run after global options", async () => { - expectFailure( - await output(spawn(["--log-level", "ERROR", "run", "--format", "json", "--dir", "/missing/aictrl-90", "prompt"])), - "validation", - ) - expectFailure( - await output(spawn(["--print-logs=false", "run", "--format", "json", "--dir", "/missing/aictrl-90", "prompt"])), - "validation", - ) + test("reports invalid fork arguments as a validation error", async () => { + expectFailure(await output(spawn(["run", "--format", "json", "--fork", "prompt"])), "validation") }) test("reports bootstrap failure", async () => { @@ -213,99 +181,4 @@ describe("run --format json invocation lifecycle (#90)", () => { server.stop(true) } }) - - test("reports a post-session failure as an invocation error result", async () => { - const id = "sess-invocation-failure-90" - const server = Bun.serve({ - port: 0, - fetch(req) { - const url = new URL(req.url) - if (req.method === "GET" && url.pathname === "/event") { - return new Response("", { headers: { "content-type": "text/event-stream" } }) - } - if (req.method === "POST" && url.pathname === "/session") return Response.json({ id }) - if (req.method === "GET" && url.pathname === "/config") { - return new Response('{"post-session-secret"', { - headers: { "content-type": "application/json" }, - }) - } - if (req.method === "POST" && url.pathname.endsWith("/message")) return Response.json({}) - return Response.json({}) - }, - }) - - try { - const result = await output( - spawn(["run", "--format", "json", "--attach", `http://localhost:${server.port}`, "prompt"]), - ) - expect(result.code).not.toBe(0) - expect(result.events.find((event) => event.type === "invocation_complete")).toMatchObject({ - sessionID: id, - status: "error", - }) - expect(JSON.stringify(result.events)).not.toContain("post-session-secret") - } finally { - server.stop(true) - } - }) - - test("flush resolves when stdout rejects a lifecycle write", async () => { - const proc = Bun.spawn( - [ - "bun", - "-e", - ` - const module = await import(${JSON.stringify(INVOCATION)}) - const write = process.stdout.write.bind(process.stdout) - process.stdout.write = () => { - throw new Error("closed") - } - module.startInvocation(["run", "--format", "json"]) - module.Invocation.abort("failed") - await module.Invocation.flush() - process.stdout.write = write - process.stdout.write("flushed") - `, - ], - { stdout: "pipe", stderr: "pipe" }, - ) - const [stdout, code] = await Promise.all([new Response(proc.stdout).text(), proc.exited]) - expect(code).toBe(0) - expect(stdout).toBe("flushed") - }) - - test("late failure during pre-terminal drain produces one error completion", async () => { - const proc = Bun.spawn( - [ - "bun", - "-e", - ` - const module = await import(${JSON.stringify(INVOCATION)}) - const write = process.stdout.write.bind(process.stdout) - const lines = [] - process.stdout.write = (chunk, callback) => { - lines.push(String(chunk)) - setTimeout(() => callback?.(), 20) - return true - } - module.startInvocation(["run", "--format", "json"]) - const done = (async () => { - await module.Invocation.flush() - module.Invocation.complete() - await module.Invocation.flush() - })() - setTimeout(() => module.Invocation.abort("late"), 5) - await done - process.stdout.write = write - process.stdout.write(JSON.stringify(lines)) - `, - ], - { stdout: "pipe", stderr: "pipe" }, - ) - const [stdout, code] = await Promise.all([new Response(proc.stdout).text(), proc.exited]) - const result = JSON.parse(stdout).flatMap((line: string) => events(line)) - expect(code).toBe(0) - expect(result.filter((event: Record) => event.type === "invocation_complete")).toHaveLength(1) - expect(result.find((event: Record) => event.type === "invocation_complete")?.status).toBe("error") - }) }) From 42424fcaf9259aeb5eadba481b683f3d738b1cd1 Mon Sep 17 00:00:00 2001 From: Bulat Yapparov Date: Tue, 21 Jul 2026 10:37:06 +0100 Subject: [PATCH 6/6] fix(cli): complete guarded invocation failures --- EVENTS.md | 5 ++- packages/cli/src/cli/cmd/run.invocation.ts | 35 ++++++++++++------- packages/cli/src/cli/cmd/run.ts | 21 +++++++---- packages/cli/src/cli/stdout.ts | 6 ++++ .../cli/test/cli/invocation-lifecycle.test.ts | 19 ++++++++++ 5 files changed, 65 insertions(+), 21 deletions(-) diff --git a/EVENTS.md b/EVENTS.md index cf7d652..f3faeb6 100644 --- a/EVENTS.md +++ b/EVENTS.md @@ -51,7 +51,10 @@ Emitted for a fatal error before session creation, immediately before `invocatio ``` - `phase` (string, **required**) — one of `validation`, `stdin`, `bootstrap`, or `session`. -- `code` (string, **required**) — stable machine-readable error category. +- `code` (string, **required**) — stable machine-readable error category. Known failures use + `INVOCATION_INVALID_DIRECTORY`, `INVOCATION_FILE_NOT_FOUND`, `INVOCATION_EMPTY_INPUT`, + `INVOCATION_INVALID_ARGUMENTS`, or `INVOCATION_SESSION_CREATE_FAILED`. Unexpected failures use + `INVOCATION__FAILED`, where `` is `VALIDATION`, `STDIN`, `BOOTSTRAP`, or `SESSION`. - `message` (string, **required**) — sanitized human-readable phase summary. Details remain on stderr and in the log. Errors after a real session has been created use `session_error` instead. They also set diff --git a/packages/cli/src/cli/cmd/run.invocation.ts b/packages/cli/src/cli/cmd/run.invocation.ts index 0db4e70..b2df3cb 100644 --- a/packages/cli/src/cli/cmd/run.invocation.ts +++ b/packages/cli/src/cli/cmd/run.invocation.ts @@ -1,4 +1,3 @@ -import { EOL } from "os" import { Stdout } from "../stdout" import { SCHEMA_VERSION } from "./run.errors" @@ -15,15 +14,15 @@ export function createRunInvocation(enabled: boolean) { function emit(type: string, data: Record = {}) { if (!id) return - const line = - JSON.stringify({ + writes = writes.then(() => + Stdout.json({ type, timestamp: Date.now(), schemaVersion: SCHEMA_VERSION, invocationID: id, ...data, - }) + EOL - writes = writes.then(() => Stdout.write(line)) + }), + ) } emit("invocation_start") @@ -49,23 +48,33 @@ export function createRunInvocation(enabled: boolean) { }) } + async function abort(cause: unknown, code?: string) { + error(cause, code) + complete() + await writes + } + return { id, phase(next: RunInvocationPhase) { phase = next }, - link(id: string) { - sessionID = id + link(session: string) { + sessionID = session }, error, - async abort(cause: unknown, code?: string) { - error(cause, code) - complete() - await writes + abort, + async guard(task: () => T | Promise) { + try { + return await task() + } catch (cause) { + await abort(cause) + throw cause + } }, - async run(execution: Promise) { + async run(task: Promise | (() => T | Promise)) { try { - return await execution + return await (typeof task === "function" ? task() : task) } catch (cause) { error(cause) throw cause diff --git a/packages/cli/src/cli/cmd/run.ts b/packages/cli/src/cli/cmd/run.ts index d7aae9b..528b5b5 100644 --- a/packages/cli/src/cli/cmd/run.ts +++ b/packages/cli/src/cli/cmd/run.ts @@ -405,11 +405,13 @@ export const RunCommand = cmd({ for (const filePath of list) { const resolvedPath = path.resolve(process.cwd(), filePath) - if (!(await Filesystem.exists(resolvedPath))) { + if (!(await invocation.guard(() => Filesystem.exists(resolvedPath)))) { await fail(`File not found: ${filePath}`, "INVOCATION_FILE_NOT_FOUND") } - const mime = (await Filesystem.isDir(resolvedPath)) ? "application/x-directory" : "text/plain" + const mime = (await invocation.guard(() => Filesystem.isDir(resolvedPath))) + ? "application/x-directory" + : "text/plain" files.push({ type: "file", @@ -422,7 +424,7 @@ export const RunCommand = cmd({ if (!process.stdin.isTTY) { invocation.phase("stdin") - message += "\n" + (await Bun.stdin.text()) + message += "\n" + (await invocation.guard(() => Bun.stdin.text())) invocation.phase("validation") } @@ -517,9 +519,14 @@ export const RunCommand = cmd({ function emit(type: string, data: Record) { if (args.format === "json") { - Stdout.write( - JSON.stringify({ type, timestamp: Date.now(), invocationID: invocation.id, sessionID, ...data }) + EOL, - ) + Stdout.json({ + type, + timestamp: Date.now(), + schemaVersion: SCHEMA_VERSION, + invocationID: invocation.id, + sessionID, + ...data, + }) return true } return false @@ -948,7 +955,7 @@ export const RunCommand = cmd({ invocation.phase("bootstrap") if (args.attach) { - return await invocation.run(execute(createAictrlClient({ baseUrl: args.attach, directory }))) + return await invocation.run(() => execute(createAictrlClient({ baseUrl: args.attach, directory }))) } const execution = bootstrap(process.cwd(), async () => { diff --git a/packages/cli/src/cli/stdout.ts b/packages/cli/src/cli/stdout.ts index a59a0af..9a98082 100644 --- a/packages/cli/src/cli/stdout.ts +++ b/packages/cli/src/cli/stdout.ts @@ -1,3 +1,5 @@ +import { EOL } from "os" + const state = { bound: false, closed: false, @@ -56,6 +58,10 @@ export namespace Stdout { if (failure !== undefined) throw failure } + export function json(data: Record) { + return write(JSON.stringify(data) + EOL) + } + export function isClosed() { return state.closed } diff --git a/packages/cli/test/cli/invocation-lifecycle.test.ts b/packages/cli/test/cli/invocation-lifecycle.test.ts index 445ef4d..3ae752c 100644 --- a/packages/cli/test/cli/invocation-lifecycle.test.ts +++ b/packages/cli/test/cli/invocation-lifecycle.test.ts @@ -86,6 +86,24 @@ describe("run --format json invocation lifecycle (#90)", () => { ) }) + test("completes when guarded pre-run I/O rejects", async () => { + const source = path.resolve(import.meta.dir, "../../src/cli/cmd/run.invocation.ts") + const proc = Bun.spawn([ + process.execPath, + "--conditions=browser", + "-e", + `import { createRunInvocation } from ${JSON.stringify(source)} +const invocation = createRunInvocation(true) +invocation.phase("stdin") +await invocation.guard(() => Promise.reject(new Error("stdin failed")))`, + ], { + stdout: "pipe", + stderr: "pipe", + }) + + expectFailure(await output(proc), "stdin") + }) + test("reports a missing file as a validation error", async () => { const missing = "/missing/" + "a".repeat(100_000) const result = await output(spawn(["run", "--format", "json", "--file", missing, "prompt"])) @@ -177,6 +195,7 @@ describe("run --format json invocation lifecycle (#90)", () => { .filter((event) => String(event.type).startsWith("session_")) .every((event) => event.invocationID === start?.invocationID), ).toBe(true) + expect(result.events.every((event) => event.schemaVersion === "1")).toBe(true) } finally { server.stop(true) }