From a950fa27418c2f125fd20f28894c0ccbb9ff0e4f Mon Sep 17 00:00:00 2001 From: Aiden Cline <63023139+rekram1-node@users.noreply.github.com> Date: Thu, 5 Nov 2026 17:58:28 -0600 Subject: [PATCH] fix(cli): rewrite noninteractive run and remove the V1 bridge (#61911) --- packages/cli/package.json | 1 - packages/cli/src/commands/handlers/run.ts | 46 +- packages/cli/src/index.ts | 22 +- packages/cli/src/run/index.ts | 2 - packages/cli/src/run/noninteractive.ts | 1409 +++++++---------- packages/cli/src/run/run.ts | 105 +- packages/cli/src/run/v1.ts | 86 - .../cli/src/services/server-connection.ts | 6 +- packages/cli/src/util/io.ts | 19 + packages/cli/test/import-boundaries.test.ts | 2 - packages/cli/test/run/noninteractive.test.ts | 385 +++-- packages/cli/test/run/process.test.ts | 141 ++ 12 files changed, 1077 insertions(+), 1147 deletions(-) delete mode 100644 packages/cli/src/run/index.ts delete mode 100644 packages/cli/src/run/v1.ts create mode 100644 packages/cli/test/run/process.test.ts diff --git a/packages/cli/package.json b/packages/cli/package.json index ecc5abf437b5..79857a0fa898 100644 --- a/packages/cli/package.json +++ b/packages/cli/package.json @@ -13,7 +13,6 @@ ], "exports": { "./vite-host": "./dev/host.js", - "./run": "./src/run/index.ts", "./server-process": "./src/server-process.ts" }, "scripts": { diff --git a/packages/cli/src/commands/handlers/run.ts b/packages/cli/src/commands/handlers/run.ts index 91f22df7765a..a018101648f3 100644 --- a/packages/cli/src/commands/handlers/run.ts +++ b/packages/cli/src/commands/handlers/run.ts @@ -12,23 +12,33 @@ export default Runtime.handler(Commands.commands.run, (input) => Effect.catch((error) => Effect.sync(() => reportRunError(input, error.message))), ) if (!server) return - yield* Effect.promise(() => - runNonInteractive({ - server, - message: [...input.message], - continue: input.continue, - session: Option.getOrUndefined(input.session), - fork: input.fork, - worktree: Option.getOrUndefined(input.worktree), - model: Option.getOrUndefined(input.model), - agent: Option.getOrUndefined(input.agent), - format: input.format, - file: [...input.file], - title: Option.getOrUndefined(input.title), - archived: input.archived, - thinking: input.thinking, - auto: input.auto || input.yolo || input.dangerouslySkipPermissions, - }), - ) + // SIGINT and SIGTERM interrupt this fiber. Interruption waits for the run to cancel its Session and + // jobs, so the private server and the process shut down only after that. + yield* Effect.callback((resume) => { + const controller = new AbortController() + const done = runNonInteractive( + { + server, + message: [...input.message], + continue: input.continue, + session: Option.getOrUndefined(input.session), + fork: input.fork, + worktree: Option.getOrUndefined(input.worktree), + model: Option.getOrUndefined(input.model), + agent: Option.getOrUndefined(input.agent), + format: input.format, + file: [...input.file], + title: Option.getOrUndefined(input.title), + archived: input.archived, + thinking: input.thinking, + auto: input.auto || input.yolo || input.dangerouslySkipPermissions, + }, + controller.signal, + ).then(() => resume(Effect.void)) + return Effect.promise(() => { + controller.abort() + return done + }) + }) }), ) diff --git a/packages/cli/src/index.ts b/packages/cli/src/index.ts index 26784ac56660..267bd24d9dbf 100755 --- a/packages/cli/src/index.ts +++ b/packages/cli/src/index.ts @@ -3,7 +3,7 @@ import { ensurePluginRuntime } from "@opencode/plugin/runtime" import { NodeRuntime, NodeServices } from "@effect/platform-node" import { Cause, Effect } from "effect" -import { getErrorReported } from "effect/Runtime" +import { defaultTeardown, getErrorReported } from "effect/Runtime" import { Commands } from "./commands/commands" import { Runtime } from "./framework/runtime" import { Observability } from "@opencode/util/observability" @@ -39,6 +39,8 @@ if (process.env.OPENCODE_SSH_ASKPASS_PORT) { process.exit(await Effect.runPromise(askpass.pipe(Effect.provide(NodeServices.layer)))) } +let terminated = false + const Handlers = Runtime.handlers(Commands, { $: () => import("./commands/handlers/default"), upgrade: () => import("./commands/handlers/upgrade"), @@ -100,6 +102,10 @@ Effect.gen(function* () { yield* Heap.listen yield* CpuProfile.listen yield* ShutdownDeadline.install() + // runMain reports every signal interruption as 130; remember a SIGTERM so the exit code is 143. + process.once("SIGTERM", () => { + terminated = true + }) const runFork = Effect.runForkWith(yield* Effect.context()) const uncaughtException = (cause: Error, origin: "uncaughtException" | "unhandledRejection") => { runFork(Effect.logError("uncaught exception", { cause, origin })) @@ -152,7 +158,6 @@ Effect.gen(function* () { ), Effect.provide(NodeServices.layer), Effect.scoped, - Effect.tap(() => Effect.sync(() => process.exit(process.exitCode ?? 0))), // runMain's default reporter logs the fatal cause to stdout. Write it to stderr instead: the // desktop and `Service.ensure` only capture stderr from `serve --service`, so this is the only // channel through which a startup failure's reason reaches the user. @@ -163,5 +168,16 @@ Effect.gen(function* () { process.stderr.write(Cause.pretty(cause) + "\n") }), ), - NodeRuntime.runMain({ disableErrorReporting: true }), + NodeRuntime.runMain({ + disableErrorReporting: true, + teardown: (exit) => + defaultTeardown(exit, (code) => { + // Commands report failures through process.exitCode. + const status = code === 0 ? (process.exitCode ?? 0) : code === 130 && terminated ? 143 : code + // process.exit drops output still queued for a pipe; ending the streams waits for it. + void Promise.all( + [process.stdout, process.stderr].map((stream) => new Promise((resolve) => stream.end(() => resolve()))), + ).then(() => process.exit(status)) + }), + }), ) diff --git a/packages/cli/src/run/index.ts b/packages/cli/src/run/index.ts deleted file mode 100644 index 68057239de80..000000000000 --- a/packages/cli/src/run/index.ts +++ /dev/null @@ -1,2 +0,0 @@ -export { runNonInteractive, type RunCommandInput } from "./run" -export { runV1Bridge, type V1RunCommandInput } from "./v1" diff --git a/packages/cli/src/run/noninteractive.ts b/packages/cli/src/run/noninteractive.ts index bf00071c7b6e..ee8db3ee71cb 100644 --- a/packages/cli/src/run/noninteractive.ts +++ b/packages/cli/src/run/noninteractive.ts @@ -1,104 +1,364 @@ -import type { - EventSubscribeOutput, - JsonValue, - LocationRef, - OpenCodeClient, - SessionMessageAssistantTool, - SessionMessageInfo, - ToolContent, +import { + ClientError, + type EventSubscribeOutput, + type ExperimentalJobListOutput, + type FormInfo, + type JsonValue, + type LocationRef, + type ModelRef, + type OpenCodeClient, + type PermissionRequest, + type SessionMessageAssistantTool, + type SessionMessageIdle, + type SessionMessageInfo, } from "@opencode/client/promise" import { SessionMessage } from "@opencode/schema/session-message" import { EOL } from "node:os" -import { readFile } from "node:fs/promises" -import { nonEmptyToolContent, toolOutputText, type MiniToolPart } from "@opencode/tui/mini/tool" +import { toolOutputText, type MiniToolPart } from "@opencode/tui/mini/tool" import { UI } from "./ui" -type Model = { - providerID: string - modelID: string -} - -type File = { - url: string - filename: string - mime: string -} - type Input = { client: OpenCodeClient sessionID: string location: LocationRef message: string - files: File[] + files: { uri: string; name: string }[] agent?: string - model?: Model - variant?: string + model?: ModelRef thinking: boolean format: "default" | "json" auto: boolean - /** True when the client is attached to a shared server rather than an exclusive in-process one. */ - attached: boolean - compatibility?: "v1" + /** The server is private to this run, so MCP input requests in its location are this run's too. */ + standalone: boolean + /** Aborted on SIGINT or SIGTERM: the run cancels its Session and background jobs, then returns. */ + signal: AbortSignal renderTool: (part: SessionMessageAssistantTool) => Promise renderToolError: (part: SessionMessageAssistantTool) => Promise } -type StartedPart = { +type Event = EventSubscribeOutput +type Job = ExperimentalJobListOutput[number] +type RunError = { type: string; message: string } +type Outcome = Pick + +type Tool = { id: string timestamp: number -} - -type ToolState = StartedPart & { - assistantMessageID: string - tool: string + name: string input: Record raw?: string - provider?: unknown - native?: SessionMessageAssistantTool["native"] + call?: { executed: boolean; state: SessionMessageAssistantTool["native"] } metadata: Record - content: ToolContent[] } -type V2Event = EventSubscribeOutput -type FormRequest = Extract["data"]["form"] - -// MCP elicitations are temporarily owned by the "global" sentinel instead of a real -// session. An exclusive local process may treat them as this run's blockers; an -// attached client must not cancel input that may belong to another session. -const GLOBAL_FORM_SESSION_ID = "global" - const QUESTION_CANCELLED_FEEDBACK = "This non-interactive run cannot ask the user questions, so the question was cancelled. Continue without an answer; make reasonable assumptions and state them." const PERMISSION_REJECTED_FEEDBACK = "This non-interactive run cannot ask the user for permission, so the request was rejected. Continue without this action." +/** + * Runs one prompt to completion: + * + * admitting → running → settling → waiting for this prompt's jobs → done(exit code) + * any state + abort (SIGINT/SIGTERM) → cancelling → done + */ export async function runNonInteractivePrompt(input: Input) { const controller = new AbortController() const stream = input.client.event.subscribe({ signal: controller.signal })[Symbol.asyncIterator]() - const connected = await stream.next() - if (connected.done) throw new Error("Event stream disconnected before prompt admission") + if ((await stream.next()).done) throw new Error("Event stream disconnected before prompt admission") const messageID = SessionMessage.ID.create() - const starts = new Map() - const tools = new Map() - const renderedText = new Map() - const renderedReasoning = new Map() - const renderedTools = new Set() - // Live step events stop being read once the session goes idle, and the last step usually ends after - // that. Reconcile uses this to emit only the step_start/step_finish lines the live stream missed. + const render = renderer(input) + const state = { + phase: "running" as "running" | "settled" | "cancelling", + promoted: false, + admission: undefined as Promise<{ time: { created: number } }> | undefined, + failure: undefined as RunError | undefined, + outcomes: [] as Outcome[], + // Sessions in this run's tree that own jobs; an execution ending in one can change background state. + jobSessions: new Set([input.sessionID]), + ended: deferred(), + } + + // Subagents' asks belong to this run too; other sessions on a shared server must be left alone. + const owned = new Map([[input.sessionID, Promise.resolve(true)]]) + const ownsSession = (sessionID: string): Promise => { + const known = owned.get(sessionID) + if (known) return known + const result = input.client.session.get({ sessionID }).then( + (session) => (session.parentID ? ownsSession(session.parentID) : false), + () => false, + ) + owned.set(sessionID, result) + return result + } + + const replyPermission = async (request: PermissionRequest) => { + if (!input.auto) + UI.println( + UI.Style.TEXT_WARNING_BOLD + "!", + UI.Style.TEXT_NORMAL + + `permission requested: ${request.action} (${request.resources.join(", ")}); auto-rejecting`, + ) + await input.client.permission + .reply({ + sessionID: request.sessionID, + requestID: request.id, + decision: input.auto ? "once" : "reject", + ...(input.auto ? {} : { message: PERMISSION_REJECTED_FEEDBACK }), + }) + .catch(() => {}) + } + + // A cancelled question lets the model continue; any other input request fails the run. + // MCP elicitations belong to the "global" sentinel rather than a session; on a shared server they + // may belong to another client. + const ownsForm = (form: Pick, location: LocationRef | undefined) => + form.sessionID === "global" + ? input.standalone && location?.directory === input.location.directory + : ownsSession(form.sessionID) + + const cancelForm = async (form: Pick) => { + const question = form.metadata?.kind === "question" + await input.client.session.form + .cancel( + { sessionID: form.sessionID, formID: form.id, ...(question ? { message: QUESTION_CANCELLED_FEEDBACK } : {}) }, + // A global form is found through its location. + ...(form.sessionID === "global" + ? [{ headers: { "x-opencode-directory": encodeURIComponent(input.location.directory) } }] + : []), + ) + .catch((error) => { + if (!formAlreadySettled(error)) throw error + }) + if (!question) + state.failure ??= { + type: "unknown", + message: `Cancelled a request for input this run cannot answer: ${form.title}`, + } + } + + const consume = async () => { + while (true) { + const next = await stream.next().catch((error) => { + if (controller.signal.aborted) return { done: true as const, value: undefined } + throw error + }) + if (next.done) { + if (controller.signal.aborted) return + throw new Error("Event stream disconnected during prompt execution") + } + const event = next.value + if (event.type === "permission.asked") { + if (await ownsSession(event.data.sessionID)) await replyPermission(event.data) + continue + } + if (event.type === "form.created") { + if (await ownsForm(event.data.form, event.location)) await cancelForm(event.data.form) + continue + } + if (!("sessionID" in event.data)) continue + const sessionID = event.data.sessionID + const terminal = terminalOutcome(event) + if (state.jobSessions.has(sessionID)) { + if (event.type === "session.execution.started" && state.phase === "cancelling") + void input.client.session.interrupt({ sessionID }).catch(() => {}) + if (terminal) { + state.ended.resolve() + state.ended = deferred() + } + } + if (sessionID !== input.sessionID) continue + if (event.type === "session.inbox.delivered" && event.data.inboxID === messageID) state.promoted = true + // A busy period that ends before this prompt is delivered leaves the prompt queued for good. + if (terminal && !state.promoted && terminal.error) state.failure ??= terminal.error + if (terminal && state.promoted) state.outcomes.push(terminal) + if (state.promoted && state.phase === "running") await render.event(event) + } + } + + // The jobs this prompt started (older ones belong to someone else), or `[]` while one of their wakes + // is queued or running. Work moves job → inbox → active, so reading in that order can't miss it. + const background = async () => { + const since = (await state.admission)?.time.created + if (since === undefined) return undefined + const jobs = await input.client.experimental.job + .list({ sessionID: input.sessionID, descendants: true }) + .catch((error) => { + // Servers older than the job list endpoint answer 404 with no JSON body; they have no jobs to report. + if (error instanceof ClientError && error.reason === "UnsupportedContentType") return [] + throw error + }) + const running = jobs.filter((job) => job.time.started >= since) + running.forEach((job) => { + state.jobSessions.add(job.sessionID) + if (job.kind === "subagent") state.jobSessions.add(job.childSessionID) + }) + if (running.length > 0) return running + const inboxes = await Promise.all( + [...state.jobSessions].map((sessionID) => input.client.session.inbox.list({ sessionID })), + ) + if (inboxes.some((inbox) => inbox.length > 0)) return running + const active = await input.client.session.active() + for (const sessionID of Object.keys(active)) { + if (await ownsSession(sessionID)) return running + } + return undefined + } + + const lifecycle = async () => { + // admitting + if (input.agent) await input.client.session.switchAgent({ sessionID: input.sessionID, agent: input.agent }) + if (input.model) await input.client.session.switchModel({ sessionID: input.sessionID, model: input.model }) + if (input.signal.aborted) return + // A failed stream rejects whatever wait is pending, since no further end could be seen. + consume().catch((error) => state.ended.reject(error)) + state.admission = input.client.session.prompt({ + sessionID: input.sessionID, + id: messageID, + text: input.message, + files: input.files, + delivery: "steer", + }) + await state.admission + const [permissions, forms, globals] = await Promise.all([ + input.client.permission.list({ sessionID: input.sessionID }).catch(() => []), + input.client.session.form.list({ sessionID: input.sessionID }).catch(() => []), + input.standalone + ? input.client.form.list({ location: { directory: input.location.directory } }).catch(() => undefined) + : undefined, + ]) + await Promise.all([ + ...permissions.map(replyPermission), + ...forms.map(cancelForm), + ...(globals?.data.filter((form) => ownsForm(form, globals.location) === true) ?? []).map(cancelForm), + ]) + + // running → settling → waiting for this prompt's jobs + let waited = false + let shown = 0 + while (true) { + const ended = state.ended.promise + await input.client.experimental.session.wait({ sessionID: input.sessionID }) + if (input.signal.aborted) return + const projected = await projectedMessages(input.client, input.sessionID, messageID) + if (state.failure || state.outcomes.length > 0 || projected.messages.some((item) => item.type === "idle")) { + const pending = state.failure ? undefined : await background() + if (!pending) return { ...projected, background: waited } + if (pending.length > 0 && pending.length !== shown) render.waiting(pending) + shown = pending.length || shown + waited = true + } + await ended + } + } + + // cancelling: a wake can still start after the interrupt (the notice each cancelled job queues), + // so the consumer interrupts every execution that starts until the tree is quiet. + const cancel = async () => { + state.phase = "cancelling" + if (!(await state.admission?.catch(() => undefined))) return + if (!state.promoted) + await input.client.session.inbox.cancel({ sessionID: input.sessionID, inboxID: messageID }).catch(() => {}) + const running = (await background().catch(() => undefined)) ?? [] + await Promise.all([ + input.client.session.interrupt({ sessionID: input.sessionID }).catch(() => {}), + ...running.map((job) => + (job.kind === "shell" + ? input.client.shell.remove({ id: job.shellID, location: { directory: input.location.directory } }) + : job.kind === "subagent" + ? input.client.session.interrupt({ sessionID: job.childSessionID }) + : Promise.resolve() + ).catch(() => {}), + ), + ]) + while (true) { + const ended = state.ended.promise + await input.client.experimental.session.wait({ sessionID: input.sessionID }) + const pending = await background() + // Image jobs can't be cancelled through the API, so waiting on them would only hold the exit. + if (!pending || (pending.length > 0 && pending.every((job) => job.kind === "image"))) return + await ended + } + } + + try { + const running = lifecycle() + const aborted = new Promise((resolve) => + input.signal.addEventListener("abort", () => resolve(undefined), { once: true }), + ) + const settled = await Promise.race([running, aborted]) + // Requests failing during or after cancellation (e.g. the private server shutting down) are ignored. + if (!settled) { + running.catch(() => {}) + return await cancel().catch(() => {}) + } + state.phase = "settled" + await render.reconcile(settled.messages) + const error = finalError(settled, state) + if (!error) return + process.exitCode = 1 + render.error(error) + } finally { + controller.abort() + void stream.return?.(undefined).catch(() => {}) + } +} + +/** The run's exit code comes from here alone: the first stored idle marker after this run's prompt. */ +function finalError( + run: { found: boolean; messages: SessionMessageInfo[]; background: boolean }, + state: { failure?: RunError; outcomes: Outcome[] }, +) { + if (state.failure) return state.failure + if (!run.found) return { type: "unknown", message: "Prompt was not promoted" } + const idles = run.messages.filter((message) => message.type === "idle") + const recorded = idles.length > 0 ? idles : state.outcomes + // The first marker closes this prompt's turn. Later ones count only as replies woken by this + // prompt's background jobs; otherwise they belong to other prompts, possibly from another client. + const turns = run.background ? recorded : recorded.slice(0, 1) + const final = turns.find((turn) => turn.outcome !== "succeeded") ?? turns[0] + if (!final) return { type: "unknown", message: "Session ended without a recorded outcome" } + if (final.outcome === "succeeded") return + const fallback = { type: final.outcome === "failed" ? "unknown" : "aborted", message: `Session ${final.outcome}` } + return final.error ?? state.outcomes.find((turn) => turn.id === final.id)?.error ?? fallback +} + +function terminalOutcome(event: Event): Outcome | undefined { + const id = SessionMessage.ID.fromEvent(event.id) + if (event.type === "session.execution.succeeded") return { id, outcome: "succeeded" } + if (event.type === "session.execution.failed") return { id, outcome: "failed", error: event.data.error } + if (event.type !== "session.execution.interrupted") return undefined + return { + id, + outcome: "interrupted", + error: { type: "aborted", message: `Session interrupted: ${event.data.reason}` }, + } +} + +async function projectedMessages(client: OpenCodeClient, sessionID: string, messageID: string) { + const messages: SessionMessageInfo[] = [] + let cursor: string | undefined + while (true) { + const page = await client.message.list( + cursor ? { sessionID, limit: 200, cursor } : { sessionID, limit: 200, order: "desc" }, + ) + for (const message of page.data) { + if (message.id === messageID) return { found: true, messages: messages.toReversed() } + messages.push(message) + } + cursor = page.cursor.next ?? undefined + if (!cursor) return { found: false, messages: [] } + } +} + +function renderer(input: Input) { + const starts = new Map() + const tools = new Map() const steps = new Map() - let submitted = false - let promoted = false - let emittedError = false - let permissionRejected = false - let formCancelled = false - let interrupted = false - let v1InvalidOutput = false - let prePromotionError: { message: string; [key: string]: unknown } | undefined - let finalizing = false - let admission: AbortController | undefined - let pendingStep: { timestamp: number; part: Record; label: string } | undefined + // The session can go idle before the last live events arrive; reconcile prints what they missed. + const printed = new Map() const emit = (type: string, timestamp: number, data: Record) => { if (input.format !== "json") return false @@ -106,848 +366,297 @@ export async function runNonInteractivePrompt(input: Input) { return true } - const writeText = (part: { text: string; [key: string]: unknown }, timestamp: number) => { - if (emit("text", timestamp, { part })) return - const text = part.text.trim() - if (!text) return - if (!process.stdout.isTTY) { - process.stdout.write(text + EOL) - return - } + const block = (line: string) => { UI.empty() - UI.println(text) + UI.println(line) UI.empty() } - const writeReasoning = (part: { text: string; [key: string]: unknown }, timestamp: number) => { - if (emit("reasoning", timestamp, { part })) return - const text = part.text.trim() - if (!text) return - const line = `Thinking: ${text}` + const text = ( + part: { + kind: "text" | "reasoning" + id: string + messageID: string + ordinal: number + text: string + metadata?: unknown + }, + start: number, + end: number, + ) => { + if (part.kind === "reasoning" && !input.thinking) return + const id = key(part.kind, part.messageID, part.ordinal) + const previous = printed.get(id) ?? "" + if (printed.has(id) && (previous === part.text || !part.text.startsWith(previous))) return + printed.set(id, part.text) + const value = part.text.slice(previous.length) + const json = { + id: part.id, + sessionID: input.sessionID, + messageID: part.messageID, + type: part.kind, + text: value, + ...(part.kind === "reasoning" ? { metadata: part.metadata } : {}), + time: { start, end }, + } + if (emit(part.kind, end, { part: json })) return + const trimmed = value.trim() + if (!trimmed) return + const line = part.kind === "text" ? trimmed : `Thinking: ${trimmed}` if (!process.stdout.isTTY) return void process.stdout.write(line + EOL) - UI.empty() - UI.println(`${UI.Style.TEXT_DIM}\u001b[3m${line}\u001b[0m${UI.Style.TEXT_NORMAL}`) - UI.empty() + block(part.kind === "text" ? line : `${UI.Style.TEXT_DIM}\u001b[3m${line}\u001b[0m${UI.Style.TEXT_NORMAL}`) } - const flushStep = () => { - if (!pendingStep) return - const value = pendingStep - pendingStep = undefined - if (!emit("step_start", value.timestamp, { part: value.part }) && input.format !== "json") { - UI.empty() - UI.println(value.label) - UI.empty() - } - } - - // Subagents run in child sessions; their asks and questions belong to this run too. Other - // sessions on a shared server (e.g. the TUI's) must be left alone. - const owned = new Map>([[input.sessionID, Promise.resolve(true)]]) - const ownsSession = (sessionID: string): Promise => { - const known = owned.get(sessionID) - if (known) return known - const result = - sessionID === GLOBAL_FORM_SESSION_ID - ? Promise.resolve(false) - : input.client.session - .get({ sessionID }) - .then((session) => (session.parentID ? ownsSession(session.parentID) : false)) - .catch(() => false) - owned.set(sessionID, result) - return result - } - - // Sessions in this run's tree that have owned a job, so a wake still queued for one of them is seen. - const jobSessions = new Set([input.sessionID]) - let waitingForBackground = false - - // Any job in the tree holds the run open: this session's, its subagents', and theirs. - const pendingBackground = async () => { - if (input.compatibility === "v1" || interrupted || permissionRejected || formCancelled || emittedError) return false - const jobs = await input.client.experimental.job - .list({ sessionID: input.sessionID, descendants: true }) - .catch(() => []) - jobs.forEach((job) => { - jobSessions.add(job.sessionID) - if (job.kind === "subagent") jobSessions.add(job.childSessionID) - }) - if (jobs.length > 0) return true - // A finished job wakes its session before its row is removed, so a wake that raced the - // list above shows up as a queued inbox item or an active session in the tree. - const [active, inboxes] = await Promise.all([ - input.client.session.active().catch(() => ({})), - Promise.all([...jobSessions].map((sessionID) => input.client.session.inbox.list({ sessionID }).catch(() => []))), - ]) - if (inboxes.some((inbox) => inbox.length > 0)) return true - for (const sessionID of Object.keys(active)) { - if (await ownsSession(sessionID)) return true - } - return false - } - - const replyPermission = async (request: { - id: string - sessionID: string - action: string - resources: ReadonlyArray - }) => { - // Nobody can approve here. Outside V1 compatibility, reject with feedback so the tool fails - // as ordinary model-visible output and the model continues without the action. - const continuing = !input.auto && input.compatibility !== "v1" - if (!input.auto) { - if (!continuing) permissionRejected = true - UI.println( - UI.Style.TEXT_WARNING_BOLD + "!", - UI.Style.TEXT_NORMAL + - `permission requested: ${request.action} (${request.resources.join(", ")}); auto-rejecting`, - ) - } - await input.client.permission - .reply({ - sessionID: request.sessionID, - requestID: request.id, - decision: input.auto ? "once" : "reject", - ...(continuing ? { message: PERMISSION_REJECTED_FEEDBACK } : {}), - }) - .catch(() => {}) - if (!input.auto && !continuing) { - await input.client.session.interrupt({ sessionID: input.sessionID }).catch(() => {}) - } - } - - const cancelForm = async (request: Pick) => { - // Nobody can answer a question here. Outside V1 compatibility, cancel it with a message so the - // question tool fails as ordinary model-visible output and the model continues without an answer. - const continuing = request.metadata?.kind === "question" && input.compatibility !== "v1" - try { - await input.client.session.form.cancel( - { - sessionID: request.sessionID, - formID: request.id, - ...(continuing ? { message: QUESTION_CANCELLED_FEEDBACK } : {}), - }, - ...formRequestOptions(request.sessionID === GLOBAL_FORM_SESSION_ID ? input.location : undefined), - ) - } catch (error) { - if (!formAlreadySettled(error)) throw error - } - if (continuing) return - formCancelled = true - if (input.compatibility !== "v1") process.exitCode = 1 - } - - const consume = async () => { - const aborted = new Promise<{ done: true; value: undefined }>((resolve) => { - if (controller.signal.aborted) resolve({ done: true, value: undefined }) - else controller.signal.addEventListener("abort", () => resolve({ done: true, value: undefined }), { once: true }) - }) - while (!controller.signal.aborted) { - const next = await Promise.race([stream.next(), aborted]).catch((error) => { - if (!emittedError && !interrupted) throw error - return { done: true as const, value: undefined } - }) - if (next.done) { - if (emittedError || interrupted) return - throw new Error("Event stream disconnected during prompt execution") - } - const event = next.value - - if (event.type === "permission.asked" && submitted && (await ownsSession(event.data.sessionID))) { - await replyPermission(event.data) - continue - } - if ( - event.type === "form.created" && - submitted && - ((await ownsSession(event.data.form.sessionID)) || - (!input.attached && - event.data.form.sessionID === GLOBAL_FORM_SESSION_ID && - sameLocation(event.location, input.location))) - ) { - await cancelForm(event.data.form) - continue - } - if (!("sessionID" in event.data)) continue - if (event.data.sessionID !== input.sessionID) { - // While waiting, a reply ending anywhere in the tree may have finished the last of the work. - if ( - waitingForBackground && - (event.type === "session.execution.succeeded" || - event.type === "session.execution.failed" || - event.type === "session.execution.interrupted") && - (await ownsSession(event.data.sessionID)) && - !(await pendingBackground()) - ) - return - continue - } - const time = toMillis("created" in event ? event.created : undefined) - - if (event.type === "session.inbox.delivered") { - if (event.data.inboxID === messageID) { - promoted = true - prePromotionError = undefined - continue - } - } - if ( - event.type === "session.execution.interrupted" && - event.data.reason === "user" && - (interrupted || permissionRejected || formCancelled) - ) { - return - } - if (!promoted && event.type === "session.execution.failed") { - prePromotionError = event.data.error - if (finalizing) return - continue - } - if ( - !promoted && - finalizing && - (event.type === "session.execution.succeeded" || event.type === "session.execution.interrupted") - ) - return - if (!promoted) continue - if (finalizing && !event.type.startsWith("session.execution.")) continue - - if (event.type === "session.step.started") { - const part = { - id: partID(event.id), - sessionID: input.sessionID, - messageID: event.data.assistantMessageID, - type: "step-start", - snapshot: event.data.snapshot, - } - if (input.compatibility === "v1") { - pendingStep = { - timestamp: time, - part, - label: `> ${event.data.agent} · ${event.data.model.id}`, + const tool = async ( + partID: string, + messageID: string, + item: SessionMessageAssistantTool & { state: { status: "completed" | "error" } }, + metadata: Record, + timestamp: number, + ) => { + const id = key(messageID, item.id) + if (printed.has(id)) return + printed.set(id, "") + const time = { start: item.time.ran ?? item.time.created, end: item.time.completed ?? timestamp } + const state = + item.state.status === "completed" + ? { + status: "completed" as const, + input: item.state.input, + output: toolOutputText(item.name, item.state.content), + title: item.name, + metadata, + time, } - continue - } - steps.set(event.data.assistantMessageID, "started") - if (!emit("step_start", time, { part }) && input.format !== "json") { - UI.empty() - UI.println(`> ${event.data.agent} · ${event.data.model.id}`) - UI.empty() - } - continue - } + : { status: "error" as const, input: item.state.input, error: item.state.error.message, metadata, time } + const part = { + partID, + sessionID: input.sessionID, + messageID, + type: "tool" as const, + id: item.id, + tool: item.name, + state, + } + if (emit("tool_use", timestamp, { part: part satisfies MiniToolPart })) return + if (item.state.status === "completed") return input.renderTool(item) + if (item.state.content && toolOutputText(item.name, item.state.content).trim()) + await input.renderTool({ + ...item, + state: { + status: "completed", + input: item.state.input, + metadata: item.state.metadata, + content: item.state.content, + }, + }) + await input.renderToolError(item) + UI.error(item.state.error.message) + } - if (event.type === "session.text.started") { - flushStep() - starts.set(`text\u0000${contentKey(event.data.assistantMessageID, event.data.ordinal)}`, { - id: partID(event.id), - timestamp: time, - }) - continue - } - if (event.type === "session.text.ended") { - const key = contentKey(event.data.assistantMessageID, event.data.ordinal) - const started = starts.get(`text\u0000${key}`) - starts.delete(`text\u0000${key}`) - const part = { - id: started?.id ?? partID(event.id), - sessionID: input.sessionID, + const step = (type: "step-start" | "step-finish", timestamp: number, id: string, messageID: string, rest: object) => { + steps.set(messageID, type === "step-start" ? "started" : "ended") + const part = { id, sessionID: input.sessionID, messageID, type, ...rest } + return emit(type === "step-start" ? "step_start" : "step_finish", timestamp, { part }) + } + + return { + error: (error: RunError) => { + if (!emit("error", Date.now(), { error })) UI.error(error.message) + }, + + waiting: (jobs: Job[]) => { + const list = jobs.map((job) => ({ id: job.id, kind: job.kind, title: job.title })) + if (emit("background_wait", Date.now(), { jobs: list })) return + UI.println( + UI.Style.TEXT_DIM + + `waiting for ${jobs.length} background job${jobs.length === 1 ? "" : "s"}` + + UI.Style.TEXT_NORMAL, + ) + }, + + event: async (event: Event) => { + const time = "created" in event ? event.created : Date.now() + if (event.type === "session.retry.scheduled") { + const retry = { messageID: event.data.assistantMessageID, - type: "text", - text: event.data.text, - time: { start: started?.timestamp ?? time, end: time }, + attempt: event.data.attempt, + nextRetryAt: event.data.at, + error: event.data.error, } - renderedText.set(key, event.data.text) - writeText(part, time) - continue + if (emit("retry", time, retry)) return + UI.println( + UI.Style.TEXT_WARNING_BOLD + "!", + UI.Style.TEXT_NORMAL + `retrying (attempt ${retry.attempt}): ${retry.error.message}`, + ) + return } - - if (event.type === "session.reasoning.started") { - flushStep() - starts.set(`reasoning\u0000${contentKey(event.data.assistantMessageID, event.data.ordinal)}`, { + // Each retried attempt of a step starts it again. + if (event.type === "session.step.started" && !steps.has(event.data.assistantMessageID)) { + const rest = { snapshot: event.data.snapshot } + if (!step("step-start", time, partID(event.id), event.data.assistantMessageID, rest)) + block(`> ${event.data.agent} · ${event.data.model.id}`) + return + } + if (event.type === "session.step.ended") { + const data = event.data + const rest = { reason: data.finish, snapshot: data.snapshot, cost: data.cost, tokens: data.tokens } + step("step-finish", time, partID(event.id), data.assistantMessageID, rest) + return + } + if (event.type === "session.text.started" || event.type === "session.reasoning.started") { + const kind = event.type === "session.text.started" ? "text" : "reasoning" + starts.set(key(kind, event.data.assistantMessageID, event.data.ordinal), { id: partID(event.id), timestamp: time, }) - continue + return } - if (event.type === "session.reasoning.ended" && input.thinking) { - const key = contentKey(event.data.assistantMessageID, event.data.ordinal) - const started = starts.get(`reasoning\u0000${key}`) - starts.delete(`reasoning\u0000${key}`) + if (event.type === "session.text.ended" || event.type === "session.reasoning.ended") { + const kind = event.type === "session.text.ended" ? ("text" as const) : ("reasoning" as const) + const started = starts.get(key(kind, event.data.assistantMessageID, event.data.ordinal)) + const metadata = event.type === "session.reasoning.ended" ? event.data.native : undefined const part = { - id: started?.id ?? partID(event.id), - sessionID: input.sessionID, + kind, messageID: event.data.assistantMessageID, - type: "reasoning", + ordinal: event.data.ordinal, text: event.data.text, - metadata: event.data.native, - time: { start: started?.timestamp ?? time, end: time }, } - renderedReasoning.set(key, event.data.text) - writeReasoning(part, time) - continue + return text({ ...part, id: started?.id ?? partID(event.id), metadata }, started?.timestamp ?? time, time) } - - if (event.type === "session.tool.input.started") { - flushStep() - tools.set(toolKey(event.data.assistantMessageID, event.data.id), { - id: partID(event.id), - timestamp: time, - assistantMessageID: event.data.assistantMessageID, - tool: event.data.name, - input: {}, - metadata: {}, - content: [], - }) - continue - } - if (event.type === "session.tool.input.ended") { - const current = tools.get(toolKey(event.data.assistantMessageID, event.data.id)) - if (current) current.raw = event.data.text - continue - } - if (event.type === "session.tool.input.delta") { - const current = tools.get(toolKey(event.data.assistantMessageID, event.data.id)) - if (current) current.raw = (current.raw ?? "") + event.data.delta - continue - } - if (event.type === "session.tool.called") { - flushStep() - const key = toolKey(event.data.assistantMessageID, event.data.id) - const current = tools.get(key) - tools.set(key, { + if (event.type === "session.tool.input.started" || event.type === "session.tool.called") { + const id = key(event.data.assistantMessageID, event.data.id) + const current = tools.get(id) + tools.set(id, { id: current?.id ?? partID(event.id), timestamp: current?.timestamp ?? time, - assistantMessageID: event.data.assistantMessageID, - tool: current?.tool ?? "tool", - input: event.data.input, + name: event.type === "session.tool.input.started" ? event.data.name : (current?.name ?? "tool"), + input: event.type === "session.tool.called" ? event.data.input : {}, raw: current?.raw, - provider: { executed: event.data.executed, state: event.data.native }, - native: event.data.native, + call: + event.type === "session.tool.called" + ? { executed: event.data.executed, state: event.data.native } + : undefined, metadata: {}, - content: [], }) - continue + return } - if (event.type === "session.tool.progress") { - const current = tools.get(toolKey(event.data.assistantMessageID, event.data.id)) - if (current) { - current.metadata = event.data.metadata - } - continue + if ( + event.type === "session.tool.input.delta" || + event.type === "session.tool.input.ended" || + event.type === "session.tool.progress" + ) { + const current = tools.get(key(event.data.assistantMessageID, event.data.id)) + if (!current) return + if (event.type === "session.tool.input.delta") current.raw = (current.raw ?? "") + event.data.delta + if (event.type === "session.tool.input.ended") current.raw = event.data.text + if (event.type === "session.tool.progress") current.metadata = event.data.metadata + return } - if (event.type === "session.tool.success") { - const key = toolKey(event.data.assistantMessageID, event.data.id) - const current = tools.get(key) ?? fallbackTool(event) - const tool: SessionMessageAssistantTool = { - type: "tool", - id: event.data.id, - name: current.tool, - executed: event.data.executed, - native: event.data.native ? { ...current.native, ...event.data.native } : current.native, - state: { - status: "completed", + if (event.type !== "session.tool.success" && event.type !== "session.tool.failed") return + const current = tools.get(key(event.data.assistantMessageID, event.data.id)) ?? { + id: partID(event.id), + timestamp: time, + name: "tool", + input: {}, + metadata: {}, + } + const call = { + providerCall: current.call, + providerResult: { executed: event.data.executed, state: event.data.native }, + rawInput: current.raw, + } + const item = { + type: "tool" as const, + id: event.data.id, + name: current.name, + executed: event.data.executed, + native: event.data.native ? { ...current.call?.state, ...event.data.native } : current.call?.state, + time: { created: current.timestamp, ran: current.timestamp, completed: time }, + } + const success = event.type === "session.tool.success" + const metadata = success ? { metadata: event.data.metadata, content: event.data.content, ...call } : call + const state = success + ? { + status: "completed" as const, input: current.input, metadata: event.data.metadata, content: event.data.content, - }, - time: { created: current.timestamp, ran: current.timestamp, completed: time }, - } - const part: MiniToolPart = { - partID: current.id, - sessionID: input.sessionID, - messageID: event.data.assistantMessageID, - type: "tool", - id: event.data.id, - tool: current.tool, - state: { - status: "completed", + } + : { + status: "error" as const, input: current.input, - output: toolOutputText(current.tool, event.data.content), - title: current.tool, - metadata: { - metadata: event.data.metadata, - content: event.data.content, - providerCall: current.provider, - providerResult: { executed: event.data.executed, state: event.data.native }, - rawInput: current.raw, - }, - time: { start: current.timestamp, end: time }, - }, - } - tools.delete(key) - renderedTools.add(key) - if (!emit("tool_use", time, { part })) await input.renderTool(tool) - continue - } - if (event.type === "session.tool.failed") { - const key = toolKey(event.data.assistantMessageID, event.data.id) - const current = tools.get(key) ?? fallbackTool(event) - const error = event.data.error.message - const metadata = event.data.metadata ?? current.metadata - const content = event.data.content ?? nonEmptyToolContent(current.content) - const tool: SessionMessageAssistantTool = { - type: "tool", - id: event.data.id, - name: current.tool, - executed: event.data.executed, - native: event.data.native ? { ...current.native, ...event.data.native } : current.native, - state: { - status: "error", - input: current.input, - metadata, - content, + metadata: event.data.metadata ?? current.metadata, + content: event.data.content, error: event.data.error, - }, - time: { created: current.timestamp, ran: current.timestamp, completed: time }, - } - const part: MiniToolPart = { - partID: current.id, - sessionID: input.sessionID, - messageID: event.data.assistantMessageID, - type: "tool", - id: event.data.id, - tool: current.tool, - state: { - status: "error", - input: current.input, - error, - metadata: { - providerCall: current.provider, - providerResult: { executed: event.data.executed, state: event.data.native }, - rawInput: current.raw, - }, - time: { start: current.timestamp, end: time }, - }, - } - tools.delete(key) - renderedTools.add(key) - if (input.compatibility === "v1" && (permissionRejected || formCancelled)) continue - if (!emit("tool_use", time, { part })) { - if (content && toolOutputText(current.tool, content).trim()) - await input.renderTool({ - ...tool, - state: { - status: "completed", - input: current.input, - metadata, - content, - }, - }) - await input.renderToolError(tool) - UI.error(error) - } - continue - } + } + return tool(current.id, event.data.assistantMessageID, { ...item, state }, metadata, time) + }, - if (event.type === "session.step.ended") { - flushStep() - steps.set(event.data.assistantMessageID, "ended") - const part = { - id: partID(event.id), - sessionID: input.sessionID, - messageID: event.data.assistantMessageID, - type: "step-finish", - reason: event.data.finish, - snapshot: event.data.snapshot, - cost: event.data.cost, - tokens: event.data.tokens, - } - emit("step_finish", time, { part }) - continue - } - if (event.type === "session.step.failed") { - if (input.compatibility === "v1" && event.data.error.message === "The provider response ended unexpectedly.") { - pendingStep = undefined - v1InvalidOutput = true - continue - } - if (interrupted || permissionRejected || formCancelled) continue - flushStep() - emittedError = true - process.exitCode = 1 - if (!emit("error", time, { error: event.data.error })) UI.error(event.data.error.message) - continue - } - if (event.type === "session.execution.failed") { - if (input.compatibility === "v1" && (v1InvalidOutput || permissionRejected || formCancelled)) return - flushStep() - if (!emittedError && !formCancelled) { - emittedError = true - process.exitCode = 1 - if (!emit("error", time, { error: event.data.error })) UI.error(event.data.error.message) - } - return - } - if (event.type === "session.execution.interrupted") { - if (input.compatibility === "v1" && (permissionRejected || formCancelled)) return - if (event.data.reason === "user" && interrupted) process.exitCode = 130 - // A declined tool call ends the step with an interruption; it was already reported above. - if (event.data.reason !== "user" && !emittedError && !permissionRejected && !formCancelled) { - emittedError = true - process.exitCode = 1 - const error = { type: "aborted" as const, message: `Session interrupted: ${event.data.reason}` } - if (!emit("error", time, { error })) UI.error(error.message) - } - return - } - if (event.type === "session.execution.succeeded") { - waitingForBackground = await pendingBackground() - if (waitingForBackground) continue - return - } - } - } - - const projectedMessages = async () => { - const messages: SessionMessageInfo[] = [] - let cursor: string | undefined - while (true) { - const page = await input.client.message.list( - cursor - ? { sessionID: input.sessionID, limit: 200, cursor } - : { sessionID: input.sessionID, limit: 200, order: "desc" }, - ) - for (const message of page.data) { - if (message.id === messageID) return { found: true, messages: messages.toReversed() } - messages.push(message) - } - cursor = page.cursor.next ?? undefined - if (!cursor) return { found: false, messages: [] } - } - } - - const reconcile = async () => { - const projected = await projectedMessages() - for (const message of projected.messages) { - if (message.type !== "assistant") continue - const timestamp = message.time.completed ?? message.time.created - const step = steps.get(message.id) - if (!step) { - emit("step_start", message.time.created, { - part: { - id: projectedPartID(message.id, "step-start"), - sessionID: input.sessionID, - messageID: message.id, - type: "step-start", - snapshot: message.snapshot?.start, - }, - }) - } - let textOrdinal = 0 - let reasoningOrdinal = 0 - for (const item of message.content) { - if (item.type === "text") { - const ordinal = textOrdinal++ - const key = contentKey(message.id, ordinal) - const rendered = renderedText.get(key) ?? "" - if (rendered === item.text || !item.text.startsWith(rendered)) continue - const text = item.text.slice(rendered.length) - writeText( + reconcile: async (messages: SessionMessageInfo[]) => { + for (const message of messages) { + if (message.type !== "assistant") continue + const timestamp = message.time.completed ?? message.time.created + const seen = steps.get(message.id) + const id = (part: string) => projectedPartID(message.id, part) + if (!seen) + step("step-start", message.time.created, id("step-start"), message.id, { snapshot: message.snapshot?.start }) + const ordinals = { text: 0, reasoning: 0 } + for (const item of message.content) { + if (item.type === "tool") { + if (item.state.status === "streaming" || item.state.status === "running") continue + const metadata = { metadata: item.state.metadata, content: item.state.content } + const done = { ...item, state: item.state } + await tool(id(`tool-${item.id}`), message.id, done, metadata, timestamp) + continue + } + const ordinal = ordinals[item.type]++ + // A part never printed live and still empty has nothing to show. + if (!item.text && !printed.has(key(item.type, message.id, ordinal))) continue + const metadata = item.type === "reasoning" ? item.native : undefined + text( { - id: projectedPartID(message.id, `text-${ordinal}`), - sessionID: input.sessionID, + kind: item.type, + id: id(`${item.type}-${ordinal}`), messageID: message.id, - type: "text", - text, - time: { start: message.time.created, end: timestamp }, + ordinal, + text: item.text, + metadata, }, + message.time.created, timestamp, ) - renderedText.set(key, item.text) - continue } - if (item.type === "reasoning") { - const ordinal = reasoningOrdinal++ - if (!input.thinking) continue - const key = contentKey(message.id, ordinal) - const rendered = renderedReasoning.get(key) ?? "" - if (rendered === item.text || !item.text.startsWith(rendered)) continue - const text = item.text.slice(rendered.length) - const part = { - id: projectedPartID(message.id, `reasoning-${ordinal}`), - sessionID: input.sessionID, - messageID: message.id, - type: "reasoning", - text, - metadata: item.native, - time: { start: message.time.created, end: timestamp }, - } - renderedReasoning.set(key, item.text) - writeReasoning(part, timestamp) - continue - } - - const key = toolKey(message.id, item.id) - if (renderedTools.has(key) || item.state.status === "streaming" || item.state.status === "running") continue - const part: MiniToolPart = { - partID: projectedPartID(message.id, `tool-${item.id}`), - sessionID: input.sessionID, - messageID: message.id, - type: "tool", - id: item.id, - tool: item.name, - state: - item.state.status === "completed" - ? { - status: "completed", - input: item.state.input, - output: toolOutputText(item.name, item.state.content), - title: item.name, - metadata: { metadata: item.state.metadata, content: item.state.content }, - time: { start: item.time.ran ?? item.time.created, end: item.time.completed ?? timestamp }, - } - : { - status: "error", - input: item.state.input, - error: item.state.error.message, - metadata: { metadata: item.state.metadata, content: item.state.content }, - time: { start: item.time.ran ?? item.time.created, end: item.time.completed ?? timestamp }, - }, - } - renderedTools.add(key) - if (emit("tool_use", timestamp, { part })) continue - if (item.state.status === "completed") { - await input.renderTool(item) - continue - } - if (item.state.content && toolOutputText(item.name, item.state.content).trim()) { - await input.renderTool({ - ...item, - state: { - status: "completed", - input: item.state.input, - metadata: item.state.metadata, - content: item.state.content, - }, - }) - } - await input.renderToolError(item) - UI.error(item.state.error.message) - } - - if (step !== "ended" && message.finish && message.tokens) { - emit("step_finish", timestamp, { - part: { - id: projectedPartID(message.id, "step-finish"), - sessionID: input.sessionID, - messageID: message.id, - type: "step-finish", + if (seen !== "ended" && message.finish && message.tokens) + step("step-finish", timestamp, id("step-finish"), message.id, { reason: message.finish, snapshot: message.snapshot?.end, cost: message.cost ?? 0, tokens: message.tokens, - }, - }) + }) } - - // A declined tool call ends its step with an interrupted-step error that is - // only a consequence of our own rejection; it was already reported above. - if (message.error && !emittedError && !permissionRejected && !formCancelled) { - emittedError = true - process.exitCode = 1 - if (!emit("error", timestamp, { error: message.error })) UI.error(message.error.message) - } - } - return { - found: projected.found, - responded: projected.messages.some((message) => message.type === "assistant"), - } - } - - const interrupt = () => { - if (interrupted) process.exit(130) - interrupted = true - process.exitCode = 130 - admission?.abort() - void input.client.session - .interrupt({ sessionID: input.sessionID }) - .then((result) => { - if (!result.interrupted) controller.abort() - }) - .catch(() => { - controller.abort() - }) - } - process.on("SIGINT", interrupt) - - let completed: Promise | undefined - try { - if (input.agent) { - await input.client.session.switchAgent({ sessionID: input.sessionID, agent: input.agent }) - } - const selected = input.model - ? { providerID: input.model.providerID, id: input.model.modelID, variant: input.variant } - : input.variant - ? await input.client.session - .get({ sessionID: input.sessionID }) - .then((result) => result.model) - .then(async (model) => { - if (model) return { ...model, variant: input.variant } - const result = await input.client.model.default() - const fallback = result.data - return fallback ? { providerID: fallback.providerID, id: fallback.id, variant: input.variant } : undefined - }) - : undefined - if (input.variant && !selected) throw new Error("Cannot select a variant before selecting a model") - if (selected) { - await input.client.session.switchModel({ sessionID: input.sessionID, model: selected }) - } - - const prepared = await Promise.all(input.files.map(prepareFile)) - if (interrupted) return - submitted = true - completed = consume() - admission = new AbortController() - const response = await input.client.session - .prompt( - { - sessionID: input.sessionID, - id: messageID, - text: [input.message, ...prepared.flatMap((file) => (file.text ? [file.text] : []))].join("\n\n"), - files: prepared.flatMap((file) => (file.attachment ? [file.attachment] : [])), - delivery: "steer", - }, - { signal: admission.signal }, - ) - .catch(async (error) => { - if (interrupted) { - await input.client.session.interrupt({ sessionID: input.sessionID }).catch(() => {}) - } - controller.abort() - await completed?.catch(() => {}) - if (interrupted || emittedError) return undefined - throw error - }) - admission = undefined - if (!response) return - if (interrupted) await input.client.session.interrupt({ sessionID: input.sessionID }).catch(() => {}) - - const [permissions, forms, globals] = await Promise.all([ - input.client.permission.list({ sessionID: input.sessionID }).catch(() => undefined), - input.client.session.form.list({ sessionID: input.sessionID }).catch(() => undefined), - input.attached - ? Promise.resolve(undefined) - : input.client.form - .list({ - location: { directory: input.location.directory }, - }) - .catch(() => undefined), - ]) - await Promise.all([ - ...(permissions ?? []).map(replyPermission), - ...(forms ?? []).map(cancelForm), - ...(globals && sameLocation(globals.location, input.location) - ? globals.data.filter((form) => form.sessionID === GLOBAL_FORM_SESSION_ID).map(cancelForm) - : []), - ]) - if (input.compatibility === "v1") { - await completed - return - } - - const waitUntilIdle = async () => { - await input.client.experimental.session.wait({ sessionID: input.sessionID }) - if (!(await pendingBackground())) return - await completed - if (!interrupted && !permissionRejected && !formCancelled && !emittedError) { - await input.client.experimental.session.wait({ sessionID: input.sessionID }) - } - } - const waiting = waitUntilIdle() - await Promise.race([waiting, completed.then(() => waiting)]) - finalizing = true - const projected = await reconcile() - if ( - !projected.responded && - !interrupted && - !permissionRejected && - !formCancelled && - !emittedError && - !prePromotionError - ) { - await completed - } - if (!projected.found && !interrupted && !permissionRejected && !formCancelled && !emittedError) { - const error = prePromotionError ?? { type: "unknown", message: "Prompt was not promoted" } - emittedError = true - process.exitCode = 1 - if (!emit("error", Date.now(), { error })) UI.error(error.message) - } - } finally { - process.off("SIGINT", interrupt) - controller.abort() - if (input.compatibility === "v1") await stream.return?.(undefined).catch(() => {}) - else void stream.return?.(undefined).catch(() => {}) - } -} - -function sameLocation(left: LocationRef | undefined, right: LocationRef) { - return !!left && left.directory === right.directory -} - -function formRequestOptions(location: LocationRef | undefined): [] | [{ headers: Record }] { - if (!location) return [] - return [ - { - headers: { - "x-opencode-directory": encodeURIComponent(location.directory), - }, }, - ] + } +} + +function deferred() { + const result = Promise.withResolvers() + result.promise.catch(() => {}) + return result } function formAlreadySettled(error: unknown) { return !!error && typeof error === "object" && "_tag" in error && error._tag === "FormAlreadySettledError" } +function key(...parts: Array) { + return parts.join("\u0000") +} + function partID(eventID: string) { return `prt_${eventID.replace(/^evt_/, "")}` } -function toolKey(messageID: string, id: string) { - return `${messageID}\u0000${id}` -} - -function contentKey(messageID: string, ordinal: number) { - return `${messageID}\u0000${ordinal}` -} - function projectedPartID(messageID: string, part: string) { return `prt_${messageID.replace(/^msg_/, "")}_${part}` } - -function fallbackTool(event: { - id: string - created: number - data: { assistantMessageID: string; id: string } -}): ToolState { - return { - id: partID(event.id), - timestamp: toMillis(event.created), - assistantMessageID: event.data.assistantMessageID, - tool: "tool", - input: {}, - metadata: {}, - content: [], - } -} - -function toMillis(value: unknown) { - if (typeof value === "number") return value - if (typeof value === "string") return new Date(value).getTime() - return Date.now() -} - -async function prepareFile(file: File) { - if (file.mime !== "text/plain") { - const uri = file.url.startsWith("data:") - ? file.url - : `data:${file.mime};base64,${(await readFile(new URL(file.url))).toString("base64")}` - return { attachment: { uri, name: file.filename } } - } - const content = file.url.startsWith("data:") - ? Buffer.from(file.url.slice(file.url.indexOf(",") + 1), "base64").toString("utf8") - : await readFile(new URL(file.url), "utf8") - return { text: `\n${content}\n` } -} diff --git a/packages/cli/src/run/run.ts b/packages/cli/src/run/run.ts index 24e20a6b339f..5c2fc11137ad 100644 --- a/packages/cli/src/run/run.ts +++ b/packages/cli/src/run/run.ts @@ -4,7 +4,7 @@ import { FSUtil } from "@opencode/util/fs-util" import { noTimeoutFetch } from "@opencode/util/no-timeout-fetch" import { open } from "node:fs/promises" import path from "node:path" -import { readStdin } from "../util/io" +import { readStdin, readStdinIfReady } from "../util/io" import { ServerConnection } from "../services/server-connection" import { createWorktreeLocation, parseSessionTargetModel, resolveSessionTarget } from "../session-target" import { toolInlineInfo } from "@opencode/tui/mini/tool" @@ -30,27 +30,14 @@ export type RunCommandInput = { auto?: boolean } -type FilePart = { - url: string - filename: string - mime: string -} +type FilePart = { text: string } | { attachment: { uri: string; name: string } } type Prepared = { - directory?: string + directory: string message: string files: FilePart[] } -type ExecutionOptions = { - root?: string - directory?: string - useServerDirectory?: boolean - variant?: string - attached?: boolean - compatibility?: "v1" -} - class RunTargetError extends Error { constructor( message: string, @@ -61,31 +48,34 @@ class RunTargetError extends Error { } const ATTACH_FILE_MAX_BYTES = 10 * 1024 * 1024 +const PIPED_INPUT_GRACE_MS = 1000 -export function runNonInteractive(input: RunCommandInput) { - return runNonInteractiveWithOptions(input, {}) +/** `signal` aborts on SIGINT or SIGTERM; the run then cancels its Session and background jobs and returns. */ +export function runNonInteractive(input: RunCommandInput, signal: AbortSignal) { + return run(input, signal).catch((error) => reportRunError(input, errorMessage(error))) } -/** @internal Used only by the V1 command boundary. */ -export function runNonInteractiveWithOptions(input: RunCommandInput, options: ExecutionOptions) { - return run(input, options).catch((error) => reportRunError(input, errorMessage(error))) -} - -async function run(input: RunCommandInput, options: ExecutionOptions) { +async function run(input: RunCommandInput, signal: AbortSignal) { if (input.fork && !input.continue && !input.session) fail("--fork requires --continue or --session") if (input.worktree !== undefined && (input.continue || input.session)) fail("--worktree starts a new session and cannot be combined with --continue or --session") - const root = options.root ?? process.env.PWD ?? process.cwd() - const local = localDirectory(root) - const directory = options.useServerDirectory ? undefined : (options.directory ?? local) - const message = mergeInput(formatMessage(input.message), process.stdin.isTTY ? undefined : await readStdin()) + const root = process.env.PWD ?? process.cwd() + const directory = localDirectory(root) + const argument = input.message.join(" ") || undefined + // With a message argument, stdin is extra context: a shell pipe is read in full, but the open, silent + // stdin that spawn's default stdio gives a child must not hold the run. + const piped = process.stdin.isTTY + ? undefined + : argument + ? await readStdinIfReady(PIPED_INPUT_GRACE_MS) + : await readStdin() + const message = mergeInput(argument, piped) if (!message?.trim()) fail("You must provide a message") - const files = await Promise.all(input.file.map((file) => prepareFile(file, root, options))) - const prepared = { directory, message, files } - return execute(input, prepared, input.server.endpoint, options) + const files = await Promise.all(input.file.map((file) => prepareFile(file, root))) + return execute(input, { directory, message, files }, input.server.endpoint, signal) } -async function execute(input: RunCommandInput, prepared: Prepared, endpoint: Endpoint, options: ExecutionOptions) { +async function execute(input: RunCommandInput, prepared: Prepared, endpoint: Endpoint, signal: AbortSignal) { const client = OpenCode.make({ baseUrl: endpoint.url, headers: Service.headers(endpoint), @@ -93,7 +83,7 @@ async function execute(input: RunCommandInput, prepared: Prepared, endpoint: End fetch: noTimeoutFetch, }) const explicit = parseRunModel(input.model) - const location = prepared.directory ? { directory: prepared.directory } : undefined + const location = { directory: prepared.directory } const worktree = input.worktree === undefined ? undefined @@ -115,23 +105,9 @@ async function execute(input: RunCommandInput, prepared: Prepared, endpoint: End agent: input.agent, environment: input.server.service ? Env.session() : undefined, prepare: async (next) => { - const selected = - next.model ?? - (options.variant - ? await client.model - .default({ location: { directory: next.location.directory } }) - .then((result) => result.data) - : undefined) - const model = selected - ? { - providerID: selected.providerID, - id: selected.id, - variant: options.variant ?? ("variant" in selected ? selected.variant : undefined), - } - : undefined - if ((options.variant ?? explicit?.variant) && !model) + if (explicit?.variant && !next.model) throw new RunTargetError("Cannot select a variant before selecting a model", next.session?.id) - return { model, agent: next.agent } + return { model: next.model, agent: next.agent } }, }).catch((error) => { if (!(error instanceof RunTargetError)) throw error @@ -139,8 +115,6 @@ async function execute(input: RunCommandInput, prepared: Prepared, endpoint: End return undefined }) if (!target) return - const model = target.model ? { providerID: target.model.providerID, modelID: target.model.id } : undefined - const variant = target.model?.variant const title = !target.resume && input.title !== undefined ? input.title || prepared.message.slice(0, 50) + (prepared.message.length > 50 ? "..." : "") @@ -155,16 +129,15 @@ async function execute(input: RunCommandInput, prepared: Prepared, endpoint: End client, sessionID: target.session.id, location: target.location, - message: prepared.message, - files: prepared.files, + message: [prepared.message, ...prepared.files.flatMap((file) => ("text" in file ? [file.text] : []))].join("\n\n"), + files: prepared.files.flatMap((file) => ("attachment" in file ? [file.attachment] : [])), agent: target.agent, - model, - variant, + model: target.model, thinking: input.thinking ?? false, format: input.format, auto: input.auto ?? false, - attached: options.attached ?? true, - compatibility: options.compatibility, + standalone: input.server.standalone ?? false, + signal, renderTool: (part) => renderTool(part, target.location.directory), renderToolError: (part) => renderToolError(part, target.location.directory), }).catch((error) => reportRunError(input, errorMessage(error), target.session.id)) @@ -176,11 +149,6 @@ export function mergeInput(message: string | undefined, piped: string | undefine return message + "\n" + piped } -function formatMessage(message: string[]) { - const value = message.map((part) => (part.includes(" ") ? `"${part.replace(/"/g, '\\"')}"` : part)).join(" ") - return value || undefined -} - function localDirectory(root: string) { try { process.chdir(root) @@ -199,13 +167,11 @@ export function parseRunModel(value?: string) { } } -async function prepareFile(input: string, directory: string, options: ExecutionOptions): Promise { +async function prepareFile(input: string, directory: string): Promise { const file = path.resolve(directory, input) const handle = await open(file, "r").catch(() => fail(`File not found: ${input}`)) try { const stat = await handle.stat() - if (options.compatibility === "v1" && options.attached && stat.isDirectory()) - fail(`Cannot attach local directory without a shared filesystem: ${input}`) if (!stat.isFile() || stat.size > ATTACH_FILE_MAX_BYTES) fail(`Cannot attach a directory, special file, or file larger than 10 MiB: ${input}`) const content = Buffer.alloc(Number(stat.size)) @@ -224,11 +190,9 @@ async function prepareFile(input: string, directory: string, options: ExecutionO : !isBinaryContent(bytes) && Buffer.from(text, "utf8").equals(bytes) ? "text/plain" : detected - return { - url: `data:${mime};base64,${bytes.toString("base64")}`, - filename: path.basename(file), - mime, - } + const name = path.basename(file) + if (mime === "text/plain") return { text: `\n${text}\n` } + return { attachment: { uri: `data:${mime};base64,${bytes.toString("base64")}`, name } } } finally { await handle.close() } @@ -261,7 +225,6 @@ async function renderToolError(part: SessionMessageAssistantTool, directory: str UI.println(UI.Style.TEXT_NORMAL + "✗", UI.Style.TEXT_NORMAL + `${info.title} failed`) } -/** @internal Used by the V1 command boundary before a Session exists. */ export function reportRunError(input: Pick, message: string, sessionID?: string) { process.exitCode = 1 if (input.format === "json") { diff --git a/packages/cli/src/run/v1.ts b/packages/cli/src/run/v1.ts deleted file mode 100644 index b5bd6022aadd..000000000000 --- a/packages/cli/src/run/v1.ts +++ /dev/null @@ -1,86 +0,0 @@ -import type { Endpoint } from "@opencode/client/effect/service" -import { Effect } from "effect" -import path from "node:path" -import { Standalone } from "../services/standalone" -import { errorMessage } from "../util/error" -import { reportRunError, runNonInteractiveWithOptions, type RunCommandInput } from "./run" - -export type V1RunCommandInput = { - message: string[] - continue?: boolean - session?: string - fork?: boolean - model?: string - agent?: string - format: "default" | "json" - file: string[] - title?: string - server?: string - password?: string - username?: string - directory?: string - variant?: string - thinking?: boolean - dangerouslySkipPermissions?: boolean - standaloneCommand?: ReadonlyArray -} - -export function runV1Bridge(input: V1RunCommandInput) { - const root = process.env.PWD ?? process.cwd() - const attached = input.server !== undefined - const local = !attached && input.directory ? path.resolve(root, input.directory) : root - try { - process.chdir(local) - } catch { - reportRunError(input, `Failed to change directory to ${local}`) - return Promise.resolve() - } - - return Effect.runPromise( - Effect.scoped( - Effect.gen(function* () { - const endpoint = attached - ? explicitEndpoint(input) - : yield* Standalone.start({ command: input.standaloneCommand }) - yield* Effect.promise(() => - runNonInteractiveWithOptions(nativeInput(input, endpoint), { - root: local, - directory: attached ? input.directory : local, - useServerDirectory: attached && input.directory === undefined, - variant: input.variant, - attached, - compatibility: "v1", - }), - ) - }), - ), - ).catch((error) => reportRunError(input, errorMessage(error))) -} - -function nativeInput(input: V1RunCommandInput, endpoint: Endpoint): RunCommandInput { - return { - server: { endpoint }, - message: input.message, - continue: input.continue, - session: input.session, - fork: input.fork, - model: input.model, - agent: input.agent, - format: input.format, - file: input.file, - title: input.title, - thinking: input.thinking, - auto: input.dangerouslySkipPermissions, - } -} - -function explicitEndpoint(input: V1RunCommandInput): Endpoint { - const url = input.server - if (!url) throw new Error("Missing V1 server URL") - return { - url, - auth: input.password - ? { type: "basic", username: input.username ?? "opencode", password: input.password } - : undefined, - } -} diff --git a/packages/cli/src/services/server-connection.ts b/packages/cli/src/services/server-connection.ts index 9bc4f154d49d..48af79aac52e 100644 --- a/packages/cli/src/services/server-connection.ts +++ b/packages/cli/src/services/server-connection.ts @@ -20,6 +20,8 @@ export type Args = { export type Resolved = { readonly endpoint: Endpoint readonly service?: ReturnType + /** A private server started for this command alone. */ + readonly standalone?: boolean } export const resolve = Effect.fn("cli.server-connection.resolve")(function* (args: Args = {}) { @@ -47,14 +49,14 @@ export const resolve = Effect.fn("cli.server-connection.resolve")(function* (arg return { endpoint } satisfies Resolved } if (standalone || (yield* ServiceConfig.read()).disabled === true) { - return { endpoint: yield* Standalone.start() } satisfies Resolved + return { endpoint: yield* Standalone.start(), standalone: true } satisfies Resolved } const mismatch = args.mismatch ?? "ignore" const options = yield* ServiceConfig.options({ checkVersion: mismatch !== "ignore" }) if (args.autostart === false) { const endpoint = yield* Service.discover({ ...options, version: undefined }) - if (!endpoint) return { endpoint: yield* Standalone.start() } satisfies Resolved + if (!endpoint) return { endpoint: yield* Standalone.start(), standalone: true } satisfies Resolved return { endpoint, service: managedService(options) } satisfies Resolved } return { diff --git a/packages/cli/src/util/io.ts b/packages/cli/src/util/io.ts index e1995779a73e..1f5a7b5573ba 100644 --- a/packages/cli/src/util/io.ts +++ b/packages/cli/src/util/io.ts @@ -1,5 +1,24 @@ +import { once } from "node:events" +import { fstatSync } from "node:fs" import { text } from "node:stream/consumers" export function readStdin() { return text(process.stdin) } + +/** + * Reads a shell pipe or redirected file in full, however slow its producer. Any other stdin (the socket + * that spawn's default stdio gives a child) is read only if it has data or ends within `grace` + * milliseconds; otherwise it is closed unread. + */ +export async function readStdinIfReady(grace: number) { + const stat = fstatSync(0) + if (stat.isFIFO() || stat.isFile()) return readStdin() + const ready = await once(process.stdin, "readable", { signal: AbortSignal.timeout(grace) }).then( + () => true, + () => false, + ) + if (ready) return readStdin() + process.stdin.destroy() + return undefined +} diff --git a/packages/cli/test/import-boundaries.test.ts b/packages/cli/test/import-boundaries.test.ts index 7071610ce97e..5d259acd9428 100644 --- a/packages/cli/test/import-boundaries.test.ts +++ b/packages/cli/test/import-boundaries.test.ts @@ -26,12 +26,10 @@ describe("CLI frontend import boundaries", () => { }) test("exposes only the intentional package entrypoints", async () => { - const run = await import("@opencode/cli/run") const mini = await import("@opencode/tui/mini") const tool = await import("@opencode/tui/mini/tool") const cli = await Bun.file(path.join(root, "packages/cli/package.json")).json() - expect(Object.keys(run).sort()).toEqual(["runNonInteractive", "runV1Bridge"]) expect(Object.keys(mini).sort()).toEqual(["runMiniFrontend"]) expect(Object.keys(tool).sort()).toEqual([ "nonEmptyToolContent", diff --git a/packages/cli/test/run/noninteractive.test.ts b/packages/cli/test/run/noninteractive.test.ts index 0130ac5497e9..f6ee2eb9ca91 100644 --- a/packages/cli/test/run/noninteractive.test.ts +++ b/packages/cli/test/run/noninteractive.test.ts @@ -1,8 +1,11 @@ import { afterEach, describe, expect, mock, spyOn, test } from "bun:test" import { + ClientError, OpenCode, type EventSubscribeOutput, type ExperimentalJobListOutput, + type SessionActiveOutput, + type SessionInboxListOutput, type SessionMessageAssistantTool, type SessionMessageInfo, } from "@opencode/client/promise" @@ -100,6 +103,50 @@ function executionFailed(message: string): V2Event { } } +function idleMessage(outcome: "succeeded" | "failed" | "interrupted", id = "msg_idle") { + return { + id, + type: "idle", + outcome, + ...(outcome === "failed" ? { error: { type: "provider.transport", message: "Execution failed" } } : {}), + time: { created: 5 }, + } satisfies SessionMessageInfo +} + +// A step that failed and was retried, followed by the step that recovered. +function recoveredMessages(outcome: "succeeded" | "failed" | "interrupted") { + return (messageID: string): SessionMessageInfo[] => [ + idleMessage(outcome), + { + id: "msg_recovered", + type: "assistant", + agent: "build", + model: { providerID: "test", id: "test-model" }, + content: [{ type: "text", text: "Recovered answer" }], + finish: "stop", + time: { created: 3, completed: 4 }, + }, + { + id: "msg_assistant", + type: "assistant", + agent: "build", + model: { providerID: "test", id: "test-model" }, + content: [], + finish: "error", + error: { type: "provider.transport", message: "Recovered transport error" }, + time: { created: 2, completed: 3 }, + }, + { id: messageID, type: "user", text: "hello", time: { created: 1 } }, + ] +} + +function jsonLines(stdout: string) { + return stdout + .split("\n") + .filter(Boolean) + .map((line) => JSON.parse(line)) +} + function failedTool(inboxID: string): V2Event[] { return [ prompted(inboxID), @@ -210,16 +257,18 @@ function successfulGrep(inboxID: string): V2Event[] { async function run(input: { turn: (inboxID: string) => V2Event[] pendingForms?: FormInfo[] - attached?: boolean + standalone?: boolean format?: "default" | "json" - compatibility?: "v1" cancel?: (input: { sessionID: string; formID: string }) => Promise renderTool?: (part: SessionMessageAssistantTool) => Promise renderToolError?: (part: SessionMessageAssistantTool) => Promise messages?: (inboxID: string) => SessionMessageInfo[] wait?: () => Promise jobs?: () => ExperimentalJobListOutput + inbox?: (sessionID: string) => SessionInboxListOutput + active?: () => SessionActiveOutput terminalDelay?: number + signal?: AbortSignal }) { const sdk = OpenCode.make({ baseUrl: "https://opencode.test" }) const values: V2Event[] = [{ id: "evt_connected", type: "server.connected", data: {} }] @@ -256,9 +305,14 @@ async function run(input: { spyOn(sdk.session.form, "cancel").mockImplementation((request) => (input.cancel?.(request) ?? ok(undefined)) as never) let promptID = "msg_prompt" spyOn(sdk.experimental.session, "wait").mockImplementation(() => input.wait?.() ?? wait.promise) - spyOn(sdk.experimental.job, "list").mockImplementation(() => ok(input.jobs?.() ?? []) as never) - spyOn(sdk.session, "active").mockImplementation(() => ok({}) as never) - spyOn(sdk.session.inbox, "list").mockImplementation(() => ok([]) as never) + spyOn(sdk.experimental.job, "list").mockImplementation((async () => input.jobs?.() ?? []) as never) + spyOn(sdk.session, "active").mockImplementation((async () => input.active?.() ?? {}) as never) + spyOn(sdk.session.inbox, "list").mockImplementation( + (async (request: { sessionID: string }) => input.inbox?.(request.sessionID) ?? []) as never, + ) + spyOn(sdk.session.inbox, "cancel").mockImplementation(() => ok(undefined)) + spyOn(sdk.session, "interrupt").mockImplementation(() => ok({ interrupted: true }) as never) + spyOn(sdk.shell, "remove").mockImplementation(() => ok(undefined) as never) spyOn(sdk.session, "get").mockImplementation( ({ sessionID }) => ok({ id: sessionID, parentID: sessionID === "ses_child" ? "ses_1" : undefined }) as never, ) @@ -285,8 +339,8 @@ async function run(input: { thinking: false, format: input.format ?? "default", auto: false, - attached: input.attached ?? false, - compatibility: input.compatibility, + standalone: input.standalone ?? false, + signal: input.signal ?? new AbortController().signal, renderTool: input.renderTool ?? (() => Promise.resolve()), renderToolError: input.renderToolError ?? (() => Promise.resolve()), }) @@ -370,6 +424,7 @@ describe("runNonInteractivePrompt", () => { turn: (messageID) => [prompted(messageID)], wait: () => idle.promise, messages: (messageID) => [ + idleMessage("succeeded"), { id: "msg_assistant", type: "assistant", @@ -405,6 +460,7 @@ describe("runNonInteractivePrompt", () => { test("reports usage for a step that ends after the Session goes idle", async () => { const tokens = { input: 3, output: 5, reasoning: 7, cache: { read: 11, write: 13 } } const assistant = (messageID: string): SessionMessageInfo[] => [ + idleMessage("succeeded"), { id: "msg_assistant", type: "assistant", @@ -483,105 +539,30 @@ describe("runNonInteractivePrompt", () => { } }) - test("cancels session and global form blockers and exits on pre-promotion interrupt", async () => { - const sdk = await run({ - pendingForms: [form("frm_pending", "ses_1"), form("frm_pending_global", "global")], - // No prompted event: the execution settles interrupted before promotion, - // which must not leave the consume loop waiting forever. - turn: () => [formCreated(form("frm_live", "global")), settled("interrupted")], + for (const standalone of [true, false]) { + test(`${standalone ? "a standalone" : "an attached"} run ${standalone ? "cancels" : "leaves"} MCP input requests in its location`, async () => { + const cancelled: { sessionID: string; formID: string }[] = [] + const output = await capture({ + format: "json", + standalone, + pendingForms: [form("frm_pending", "ses_1"), form("frm_pending_global", "global")], + turn: (messageID) => [ + formCreated(form("frm_live", "global")), + formCreated(form("frm_elsewhere", "global"), { ...location, directory: "/elsewhere" }), + prompted(messageID), + settled(), + ], + cancel: (request) => { + cancelled.push(request) + return Promise.resolve() + }, + }) + const global = standalone ? ["frm_live", "frm_pending_global"] : [] + expect(cancelled.map((request) => request.formID).sort()).toEqual(["frm_pending", ...global].sort()) + expect(output.exitCode).toBe(1) + expect(jsonLines(output.stdout).filter((event) => event.type === "error")).toHaveLength(1) }) - const globalOptions = { - headers: { - "x-opencode-directory": "%2Fwork%20tree", - }, - } - expect(sdk.session.form.cancel).toHaveBeenCalledWith({ sessionID: "global", formID: "frm_live" }, globalOptions) - expect(sdk.session.form.cancel).toHaveBeenCalledWith({ sessionID: "ses_1", formID: "frm_pending" }) - expect(sdk.session.form.cancel).toHaveBeenCalledWith( - { sessionID: "global", formID: "frm_pending_global" }, - globalOptions, - ) - expect(sdk.form.list).toHaveBeenCalledWith({ - location: { directory: "/work tree" }, - }) - expect(process.exitCode).toBe(1) - }) - - test("attach mode cancels only session-owned forms", async () => { - const sdk = await run({ - attached: true, - pendingForms: [form("frm_pending", "ses_1"), form("frm_pending_global", "global")], - turn: (messageID) => [formCreated(form("frm_live", "global")), prompted(messageID), settled()], - }) - expect(sdk.session.form.cancel).toHaveBeenCalledWith({ sessionID: "ses_1", formID: "frm_pending" }) - expect(sdk.form.list).not.toHaveBeenCalled() - expect(sdk.session.form.cancel).not.toHaveBeenCalledWith( - { sessionID: "global", formID: "frm_live" }, - expect.anything(), - ) - expect(sdk.session.form.cancel).not.toHaveBeenCalledWith( - { sessionID: "global", formID: "frm_pending_global" }, - expect.anything(), - ) - expect(process.exitCode).toBe(1) - }) - - test("V1 JSON output flushes step_start before an unrelated step failure", async () => { - const output = await capture({ - compatibility: "v1", - format: "json", - turn: (messageID) => [ - prompted(messageID), - stepStarted(), - stepFailed("Provider request failed"), - executionFailed("Provider request failed"), - ], - }) - - expect( - output.stdout - .split("\n") - .filter(Boolean) - .map((line) => JSON.parse(line)), - ).toEqual([ - expect.objectContaining({ type: "step_start", part: expect.objectContaining({ type: "step-start" }) }), - expect.objectContaining({ - type: "error", - error: { type: "provider.transport", message: "Provider request failed" }, - }), - ]) - expect(output.stderr).toBe("") - const sdk = await run({ compatibility: "v1", turn: (messageID) => [prompted(messageID), settled()] }) - expect(sdk.experimental.session.wait).not.toHaveBeenCalled() - expect(sdk.message.list).not.toHaveBeenCalled() - }) - - test("V1 default output flushes step_start before an unrelated execution failure", async () => { - const output = await capture({ - compatibility: "v1", - turn: (messageID) => [prompted(messageID), stepStarted(), executionFailed("Execution failed")], - }) - - expect(output.stdout).toBe("") - expect(output.stderr).toContain("> build · test-model") - expect(output.stderr).toContain("Error: \u001b[0mExecution failed") - expect(output.stderr.indexOf("> build · test-model")).toBeLessThan(output.stderr.indexOf("Execution failed")) - }) - - test("V1 preserves terminal-finish failure suppression before content", async () => { - const output = await capture({ - compatibility: "v1", - format: "json", - turn: (messageID) => [ - prompted(messageID), - stepStarted(), - stepFailed("The provider response ended unexpectedly."), - executionFailed("The provider response ended unexpectedly."), - ], - }) - - expect(output).toEqual({ stdout: "", stderr: "", exitCode: 0 }) - }) + } test("renders a native terminal failure snapshot when live progress was missed", async () => { const rendered: SessionMessageAssistantTool[] = [] @@ -621,8 +602,8 @@ describe("runNonInteractivePrompt", () => { ]) }) - test("keeps failed tool partial output out of the explicit V1 JSON bridge shape", async () => { - const output = await capture({ compatibility: "v1", format: "json", turn: failedTool }) + test("keeps failed tool partial output out of the JSON tool_use shape", async () => { + const output = await capture({ format: "json", turn: failedTool }) const events = output.stdout .split("\n") .filter(Boolean) @@ -689,11 +670,191 @@ describe("runNonInteractivePrompt", () => { expect(lists).toEqual([]) }) - test("Ctrl-C while waiting on a background subagent exits with code 130", async () => { - const running = run({ turn: (id) => [prompted(id), settled()], jobs: () => [subagentJob] }) + test("a recovered step failure exits 0, reports its retry once, and still waits for background work", async () => { + const lists: ExperimentalJobListOutput[] = [[shellJob("sh_train", "python train.py")], []] + const retry = { attempt: 1, error: { type: "provider.transport", message: "Recovered transport error" } } + const output = await capture({ + format: "json", + jobs: () => lists.shift() ?? [], + messages: recoveredMessages("succeeded"), + turn: (id) => [ + prompted(id), + stepStarted(), + stepFailed("Recovered transport error"), + { + id: "evt_retry", + created: 3, + type: "session.retry.scheduled", + durable: { aggregateID: "ses_1", seq: 3, version: 1 }, + data: { sessionID: "ses_1", assistantMessageID: "msg_assistant", at: 2003, ...retry }, + }, + stepStarted(), + settled(), + { + id: "evt_text", + created: 4, + type: "session.text.ended", + durable: { aggregateID: "ses_1", seq: 4, version: 1 }, + data: { sessionID: "ses_1", assistantMessageID: "msg_2", ordinal: 0, text: "Training finished." }, + }, + settled(), + ], + }) + const events = jsonLines(output.stdout) + expect(output.exitCode).toBe(0) + expect(events.filter((event) => event.type === "error" || event.type === "retry")).toEqual([ + { type: "retry", timestamp: 3, sessionID: "ses_1", messageID: "msg_assistant", nextRetryAt: 2003, ...retry }, + ]) + expect( + events.filter((event) => event.part?.messageID === "msg_assistant" && event.type === "step_start"), + ).toHaveLength(1) + expect(events.find((event) => event.type === "background_wait")).toMatchObject({ + jobs: [{ id: "sh_train", kind: "shell" }], + }) + expect(output.stdout).toContain("Training finished.") + expect(lists).toEqual([]) + }) + + for (const outcome of ["failed", "interrupted"] as const) { + test(`a final ${outcome} outcome exits 1 with one error when the live event is dropped`, async () => { + const output = await capture({ + format: "json", + turn: (id) => [prompted(id)], + // Without the failed step, so only the idle marker can report the outcome. + messages: (id) => recoveredMessages(outcome)(id).filter((message) => message.id !== "msg_assistant"), + wait: () => Promise.resolve(), + }) + expect(output.exitCode).toBe(1) + expect(jsonLines(output.stdout).filter((event) => event.type === "error")).toHaveLength(1) + }) + } + + test("only the first idle marker after the prompt decides the outcome", async () => { + const output = await capture({ + format: "json", + turn: (id) => [prompted(id)], + wait: () => Promise.resolve(), + messages: (messageID) => [ + idleMessage("failed", "msg_later_idle"), + { id: "msg_other_prompt", type: "user", text: "another client", time: { created: 6 } }, + ...recoveredMessages("succeeded")(messageID), + idleMessage("failed", "msg_previous_idle"), + ], + }) + expect(output.exitCode).toBe(0) + expect(output.stdout).not.toContain('"type":"error"') + }) + + test("a prompt left in the inbox by a failure before promotion does not hang", async () => { + const output = await capture({ + format: "json", + turn: () => [executionFailed("instructions unavailable")], + messages: () => [], + inbox: (sessionID) => (sessionID === "ses_1" ? [{ id: "msg_prompt" } as SessionInboxListOutput[number]] : []), + }) + expect(output.exitCode).toBe(1) + expect(jsonLines(output.stdout)).toEqual([ + expect.objectContaining({ + type: "error", + error: expect.objectContaining({ message: "instructions unavailable" }), + }), + ]) + }) + + test("keeps waiting when a woken reply moves from the inbox to running between reads", async () => { + // The wake leaves the inbox as it is read, and its reply runs until the next check. + let wake: "queued" | "running" | "done" = "queued" + const output = await capture({ + format: "json", + inbox: () => { + if (wake === "queued") wake = "running" + return [] + }, + active: (): SessionActiveOutput => { + if (wake !== "running") return {} + wake = "done" + return { ses_1: { type: "running" } } + }, + turn: (id) => [ + prompted(id), + settled(), + { + id: "evt_text", + created: 2, + type: "session.text.ended", + durable: { aggregateID: "ses_1", seq: 2, version: 1 }, + data: { sessionID: "ses_1", assistantMessageID: "msg_2", ordinal: 0, text: "Shell finished." }, + }, + settled(), + ], + }) + expect(output.stdout).toContain("Shell finished.") + }) + + test("fails the run when a background state read fails instead of treating it as done", async () => { + await expect( + run({ + turn: (id) => [prompted(id), settled()], + active: () => { + throw new Error("state unavailable") + }, + }), + ).rejects.toThrow("state unavailable") + }) + + test("treats a server without the job list endpoint as having no jobs", async () => { + const output = await capture({ + turn: (id) => [prompted(id), settled()], + jobs: () => { + throw new ClientError("UnsupportedContentType", { detail: "text/plain" }) + }, + }) + expect(output.exitCode).toBe(0) + expect(output.stderr).toBe("") + }) + + test("does not wait on a job started before this prompt", async () => { + // A dev server left running by an earlier turn, as with --continue. + const output = await capture({ + turn: (id) => [prompted(id), settled()], + jobs: () => [{ ...shellJob("sh_dev", "bun dev"), time: { started: 0 } }], + }) + expect(output.exitCode).toBe(0) + expect(output.stderr).not.toContain("waiting for") + }) + + test("cancelling while waiting on background work stops the Session and this prompt's jobs", async () => { + const controller = new AbortController() + const old = { ...shellJob("sh_dev", "bun dev"), time: { started: 0 } } + const all = [old, shellJob("sh_train", "python train.py"), subagentJob] + // After the cancel reads and stops this prompt's jobs, only the older dev server is left. + let reads = 0 + const running = run({ + signal: controller.signal, + turn: (id) => [prompted(id), settled()], + jobs: () => (controller.signal.aborted && reads++ > 0 ? [old] : all), + }) expect(await Promise.race([running.then(() => "exited"), Bun.sleep(20).then(() => "waiting")])).toBe("waiting") - process.emit("SIGINT") - await running - expect(process.exitCode).toBe(130) + controller.abort() + const sdk = await running + expect(sdk.session.interrupt).toHaveBeenCalledWith({ sessionID: "ses_1" }) + expect(sdk.session.interrupt).toHaveBeenCalledWith({ sessionID: "ses_child" }) + expect(sdk.shell.remove).toHaveBeenCalledWith({ id: "sh_train", location: { directory: "/work tree" } }) + expect(sdk.shell.remove).not.toHaveBeenCalledWith(expect.objectContaining({ id: "sh_dev" })) + }) + + test("cancelling mid-turn interrupts the Session and ignores the transport errors that follow", async () => { + const controller = new AbortController() + const closed = Promise.withResolvers() + const output = capture({ + signal: controller.signal, + turn: (id) => [prompted(id)], + // The private server shuts down while the idle wait is open. + wait: () => closed.promise, + }) + await Bun.sleep(10) + controller.abort() + closed.reject(new ClientError("Transport", { detail: "ECONNRESET" })) + expect(await output).toMatchObject({ stdout: "", stderr: "", exitCode: 0 }) }) }) diff --git a/packages/cli/test/run/process.test.ts b/packages/cli/test/run/process.test.ts new file mode 100644 index 000000000000..76c8d12fea6e --- /dev/null +++ b/packages/cli/test/run/process.test.ts @@ -0,0 +1,141 @@ +import { expect, test } from "bun:test" +import path from "node:path" +import { isolatedEnv } from "../fixture/environment" +import { tmpdir } from "../fixture/tmpdir" + +// A server that admits the prompt and then keeps the Session running until interrupted. +async function serve(directory: string) { + const requests: { method: string; path: string; body?: unknown }[] = [] + const prompted = Promise.withResolvers() + const interrupted = Promise.withResolvers() + const location = { directory, project: { id: "project", directory, canonical: directory } } + const server = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + idleTimeout: 0, + async fetch(request) { + const url = new URL(request.url) + const text = await request.text() + requests.push({ method: request.method, path: url.pathname, body: text ? JSON.parse(text) : undefined }) + if (url.pathname === "/api/info") + return Response.json({ version: "test", pid: 0, urls: [], paths: { tmp: directory } }) + if (request.method === "GET" && url.pathname === "/api/location") return Response.json(location) + if (request.method === "POST" && url.pathname === "/api/session") + return Response.json({ + data: { + id: "ses_1", + projectID: "project", + location: { directory }, + cost: 0, + tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + time: { created: 1, updated: 1 }, + }, + }) + if (url.pathname === "/api/event") + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue( + `data: ${JSON.stringify({ id: "evt_connected", type: "server.connected", data: {} })}\n\n`, + ) + }, + }), + { headers: { "content-type": "text/event-stream" } }, + ) + if (url.pathname === "/api/session/ses_1/prompt") { + prompted.resolve() + const body = JSON.parse(text) + return Response.json({ + data: { id: body.id, sessionID: "ses_1", type: "user", time: { created: 1 }, payload: {}, delivery: "steer" }, + }) + } + if (url.pathname === "/api/session/ses_1/interrupt") { + interrupted.resolve() + return Response.json({ interrupted: true }) + } + if (url.pathname.endsWith("/wait")) return interrupted.promise.then(() => new Response(null, { status: 204 })) + if (request.method === "GET") return Response.json({ data: [] }) + return new Response(null, { status: 404 }) + }, + }) + return { server, requests, prompted: prompted.promise } +} + +function spawnRun(url: string, directory: string, message: string[], stdin: "pipe" | "ignore") { + return Bun.spawn([process.execPath, "run", "src/index.ts", "run", "--server", url, ...message], { + cwd: path.join(import.meta.dir, "../.."), + env: isolatedEnv(directory, { PWD: directory }), + signal: AbortSignal.timeout(25_000), + stdin, + stdout: "pipe", + stderr: "pipe", + }) +} + +test("run sends the message as typed", async () => { + await using directory = await tmpdir() + const fake = await serve(directory.path) + const child = spawnRun(fake.server.url.toString(), directory.path, ["say hello world", `it's "quoted"`], "ignore") + try { + await fake.prompted + expect(fake.requests.find((item) => item.path.endsWith("/prompt"))?.body).toMatchObject({ + text: `say hello world it's "quoted"`, + }) + } finally { + child.kill("SIGKILL") + await child.exited + await fake.server.stop(true) + } +}, 30_000) + +const posix = test.skipIf(process.platform === "win32") + +// Windows has no SIGTERM delivery: kill() terminates the process outright. +posix( + "run does not wait on spawn's open stdin, and SIGTERM interrupts the Session with 143", + async () => { + await using directory = await tmpdir() + const fake = await serve(directory.path) + // A stdin pipe that stays open and silent, as with spawn's default stdio. + const child = spawnRun(fake.server.url.toString(), directory.path, ["hello"], "pipe") + try { + await fake.prompted + await Bun.sleep(200) + child.kill("SIGTERM") + expect(await child.exited).toBe(143) + expect(fake.requests.some((item) => item.path === "/api/session/ses_1/interrupt")).toBe(true) + } finally { + child.kill("SIGKILL") + await child.exited + await fake.server.stop(true) + } + }, + 30_000, +) + +posix( + "run reads a slow shell pipe in full alongside a message argument", + async () => { + await using directory = await tmpdir() + const fake = await serve(directory.path) + const command = `(sleep 4; echo slow context) | "${process.execPath}" run src/index.ts run --server ${fake.server.url} review` + const child = Bun.spawn(["sh", "-c", command], { + cwd: path.join(import.meta.dir, "../.."), + env: isolatedEnv(directory.path, { PWD: directory.path }), + signal: AbortSignal.timeout(25_000), + stdout: "pipe", + stderr: "pipe", + }) + try { + await fake.prompted + expect(fake.requests.find((item) => item.path.endsWith("/prompt"))?.body).toMatchObject({ + text: "review\nslow context\n", + }) + } finally { + child.kill("SIGKILL") + await child.exited + await fake.server.stop(true) + } + }, + 30_000, +)