diff --git a/packages/cli/src/run/noninteractive.ts b/packages/cli/src/run/noninteractive.ts index 22900f3e116..9062df1be6c 100644 --- a/packages/cli/src/run/noninteractive.ts +++ b/packages/cli/src/run/noninteractive.ts @@ -85,6 +85,9 @@ export async function runNonInteractivePrompt(input: Input) { 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 steps = new Map() let submitted = false let promoted = false let emittedError = false @@ -280,6 +283,7 @@ export async function runNonInteractivePrompt(input: Input) { } 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}`) @@ -493,6 +497,7 @@ export async function runNonInteractivePrompt(input: Input) { if (event.type === "session.step.ended") { flushStep() + steps.set(event.data.assistantMessageID, "ended") const part = { id: partID(event.id), sessionID: input.sessionID, @@ -568,6 +573,18 @@ export async function runNonInteractivePrompt(input: Input) { 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) { @@ -660,6 +677,21 @@ export async function runNonInteractivePrompt(input: Input) { 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", + 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) { diff --git a/packages/cli/test/run/noninteractive.test.ts b/packages/cli/test/run/noninteractive.test.ts index 27779c82a68..0add848a59f 100644 --- a/packages/cli/test/run/noninteractive.test.ts +++ b/packages/cli/test/run/noninteractive.test.ts @@ -373,7 +373,51 @@ describe("runNonInteractivePrompt", () => { .split("\n") .filter(Boolean) .map((line) => JSON.parse(line)), - ).toEqual([expect.objectContaining({ type: "text", part: expect.objectContaining({ text: "projected answer" }) })]) + ).toEqual([ + expect.objectContaining({ type: "step_start", part: expect.objectContaining({ messageID: "msg_assistant" }) }), + expect.objectContaining({ type: "text", part: expect.objectContaining({ text: "projected answer" }) }), + ]) + }) + + 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[] => [ + { + id: "msg_assistant", + type: "assistant", + agent: "build", + model: { providerID: "test", id: "test-model" }, + content: [{ type: "text", text: "done" }], + finish: "stop", + cost: 0.25, + tokens, + time: { created: 2, completed: 3 }, + }, + { id: messageID, type: "user", text: "hello", time: { created: 1 } }, + ] + const lines = async (turn: (messageID: string) => V2Event[]) => + (await capture({ format: "json", turn, messages: assistant, wait: () => Promise.resolve() })).stdout + .split("\n") + .filter(Boolean) + .map((line) => JSON.parse(line)) + + const missed = await lines((messageID) => [prompted(messageID), stepStarted()]) + expect(missed.map((event) => event.type)).toEqual(["step_start", "text", "step_finish"]) + expect(missed[2].part).toMatchObject({ messageID: "msg_assistant", reason: "stop", cost: 0.25, tokens }) + + const seen = await lines((messageID) => [ + prompted(messageID), + stepStarted(), + { + id: "evt_step_ended", + created: 2, + type: "session.step.ended", + durable: { aggregateID: "ses_1", seq: 2, version: 1 }, + data: { sessionID: "ses_1", assistantMessageID: "msg_assistant", finish: "stop", cost: 0.25, tokens }, + }, + settled(), + ]) + expect(seen.filter((event) => event.type === "step_finish")).toHaveLength(1) }) test("reports an observed execution failure before prompt promotion", async () => {