Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 32 additions & 0 deletions packages/cli/src/run/noninteractive.ts
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,9 @@ export async function runNonInteractivePrompt(input: Input) {
const renderedText = new Map<string, string>()
const renderedReasoning = new Map<string, string>()
const renderedTools = new Set<string>()
// 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<string, "started" | "ended">()
let submitted = false
let promoted = false
let emittedError = false
Expand Down Expand Up @@ -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}`)
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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) {
Expand Down
46 changes: 45 additions & 1 deletion packages/cli/test/run/noninteractive.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 () => {
Expand Down
Loading