diff --git a/EVENTS.md b/EVENTS.md index 2b41488..f3faeb6 100644 --- a/EVENTS.md +++ b/EVENTS.md @@ -1,19 +1,85 @@ # 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 { "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. + +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` + +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": "Invocation failed during validation" +} +``` + +- `phase` (string, **required**) — one of `validation`, `stdin`, `bootstrap`, or `session`. +- `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 +`invocation_complete.status` to `error`, so the invocation result always agrees with the process result. + +### `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.invocation.ts b/packages/cli/src/cli/cmd/run.invocation.ts new file mode 100644 index 0000000..b2df3cb --- /dev/null +++ b/packages/cli/src/cli/cmd/run.invocation.ts @@ -0,0 +1,88 @@ +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 + writes = writes.then(() => + Stdout.json({ + type, + timestamp: Date.now(), + schemaVersion: SCHEMA_VERSION, + invocationID: id, + ...data, + }), + ) + } + + 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 } : {}), + }) + } + + async function abort(cause: unknown, code?: string) { + error(cause, code) + complete() + await writes + } + + return { + id, + phase(next: RunInvocationPhase) { + phase = next + }, + link(session: string) { + sessionID = session + }, + error, + abort, + async guard(task: () => T | Promise) { + try { + return await task() + } catch (cause) { + await abort(cause) + throw cause + } + }, + async run(task: Promise | (() => T | Promise)) { + try { + return await (typeof task === "function" ? task() : task) + } 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 02bd79a..528b5b5 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 { createRunInvocation } from "./run.invocation" type ToolProps = { input: Tool.InferParameters @@ -375,19 +376,26 @@ export const RunCommand = cmd({ }) }, handler: async (args) => { + const invocation = createRunInvocation(args.format === "json") + + async function fail(message: string, code: string) { + UI.error(message) + await invocation.abort(message, code) + process.exit(1) + } + let message = [...args.message, ...(args["--"] || [])] .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 { - UI.error("Failed to change directory to " + args.dir) - process.exit(1) + await fail("Failed to change directory to " + args.dir, "INVOCATION_INVALID_DIRECTORY") } })() @@ -397,12 +405,13 @@ 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) + 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", @@ -413,16 +422,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 invocation.guard(() => 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) + await 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) + await fail("--fork requires --continue or --session", "INVOCATION_INVALID_ARGUMENTS") } const rules: PermissionNext.Ruleset = [ @@ -508,7 +519,14 @@ 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.json({ + type, + timestamp: Date.now(), + schemaVersion: SCHEMA_VERSION, + invocationID: invocation.id, + sessionID, + ...data, + }) return true } return false @@ -678,6 +696,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) @@ -811,11 +830,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) + await fail("Session not found", "INVOCATION_SESSION_CREATE_FAILED") } + invocation.link(sessionID) await share(sdk, sessionID) emit("session_start", { @@ -876,6 +896,7 @@ export const RunCommand = cmd({ }) console.error(e) process.exitCode = 1 + invocation.error(e) }) if (args.command) { @@ -926,17 +947,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 }))) } - await bootstrap(process.cwd(), async () => { + const execution = bootstrap(process.cwd(), async () => { const sdk = { session: { async list() { @@ -1016,5 +1038,6 @@ export const RunCommand = cmd({ } await execute(sdk) }) + await invocation.run(execution) }, }) 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 new file mode 100644 index 0000000..3ae752c --- /dev/null +++ b/packages/cli/test/cli/invocation-lifecycle.test.ts @@ -0,0 +1,203 @@ +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(error.message).toBe(`Invocation failed during ${phase}`) + 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("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"])) + expectFailure(result, "validation") + expect(JSON.stringify(result.events)).not.toContain(missing) + }) + + test("reports empty input as a validation error", async () => { + expectFailure(await output(spawn(["run", "--format", "json"])), "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 () => { + 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) + expect(result.events.every((event) => event.schemaVersion === "1")).toBe(true) + } finally { + server.stop(true) + } + }) +})