From f9a49869c20c5dd1eae2e15370d6dfdc8e63dcb8 Mon Sep 17 00:00:00 2001 From: Lucas Meijer Date: Mon, 27 Jul 2026 11:56:15 +0200 Subject: [PATCH] feat(ai): expose pending stop reason while streaming (#7151) * feat(ai): expose pending streaming stop reason Closes #7142 * manual tidy up of human facing copy --- packages/agent/src/proxy.ts | 2 +- packages/ai/README.md | 3 +- packages/ai/src/api/anthropic-messages.ts | 5 +- packages/ai/src/api/azure-openai-responses.ts | 5 +- .../ai/src/api/bedrock-converse-stream.ts | 5 +- packages/ai/src/api/google-generative-ai.ts | 5 +- packages/ai/src/api/google-vertex.ts | 5 +- packages/ai/src/api/mistral-conversations.ts | 5 +- packages/ai/src/api/openai-codex-responses.ts | 19 +++- packages/ai/src/api/openai-completions.ts | 4 +- .../ai/src/api/openai-responses-shared.ts | 7 ++ packages/ai/src/api/openai-responses.ts | 5 +- packages/ai/src/api/pi-messages.ts | 2 +- packages/ai/src/providers/faux.ts | 5 +- packages/ai/src/types.ts | 2 +- ...-openai-responses-reasoning-replay.test.ts | 2 +- packages/ai/test/constrained-sampling.test.ts | 2 +- packages/ai/test/faux-provider.test.ts | 19 ++++ ...nai-responses-partial-json-cleanup.test.ts | 2 +- .../openai-responses-terminal-event.test.ts | 93 ++++++++++++++++++- packages/ai/test/pi-messages.test.ts | 7 +- packages/coding-agent/docs/custom-provider.md | 12 ++- packages/coding-agent/docs/session-format.md | 2 + .../custom-provider-anthropic/index.ts | 10 +- ...tion-retries-transient-stream-drop.test.ts | 22 ++--- .../coding-agent/test/test-harness.test.ts | 11 +++ packages/coding-agent/test/test-harness.ts | 21 +++-- 27 files changed, 237 insertions(+), 45 deletions(-) diff --git a/packages/agent/src/proxy.ts b/packages/agent/src/proxy.ts index 5f0925c91..4169a332d 100644 --- a/packages/agent/src/proxy.ts +++ b/packages/agent/src/proxy.ts @@ -120,7 +120,7 @@ export function streamProxy(model: Model, context: Context, options: ProxyS // Initialize the partial message that we'll build up from events const partial: AssistantMessage = { role: "assistant", - stopReason: "stop", + stopReason: "pending", content: [], api: model.api, provider: model.provider, diff --git a/packages/ai/README.md b/packages/ai/README.md index 7ff6a0917..ed4e4a545 100644 --- a/packages/ai/README.md +++ b/packages/ai/README.md @@ -872,7 +872,8 @@ for await (const event of s) { Every `AssistantMessage` includes a `stopReason` field that indicates how the generation ended: -- `"stop"` - Normal completion, the model finished its response +- `"pending"` - Only present in partial messages when we do not know what the stop reason will be +- `"stop"` - This is the final message the model will produce this turn - `"length"` - Output hit the maximum token limit - `"toolUse"` - Model is calling tools and expects tool results - `"error"` - An error occurred during generation diff --git a/packages/ai/src/api/anthropic-messages.ts b/packages/ai/src/api/anthropic-messages.ts index 99f698b40..385ea3c60 100644 --- a/packages/ai/src/api/anthropic-messages.ts +++ b/packages/ai/src/api/anthropic-messages.ts @@ -506,7 +506,7 @@ export const stream: StreamFunction<"anthropic-messages", AnthropicOptions> = ( totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, }, - stopReason: "stop", + stopReason: "pending", timestamp: Date.now(), }; @@ -746,6 +746,9 @@ export const stream: StreamFunction<"anthropic-messages", AnthropicOptions> = ( throw new Error("Request was aborted"); } + if (output.stopReason === "pending") { + throw new Error("Anthropic stream ended without a stop reason"); + } if (output.stopReason === "aborted" || output.stopReason === "error") { throw new Error(output.errorMessage || "An unknown error occurred"); } diff --git a/packages/ai/src/api/azure-openai-responses.ts b/packages/ai/src/api/azure-openai-responses.ts index 1e409650a..79b59563b 100644 --- a/packages/ai/src/api/azure-openai-responses.ts +++ b/packages/ai/src/api/azure-openai-responses.ts @@ -90,7 +90,7 @@ export const stream: StreamFunction<"azure-openai-responses", AzureOpenAIRespons totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, }, - stopReason: "stop", + stopReason: "pending", timestamp: Date.now(), }; @@ -132,6 +132,9 @@ export const stream: StreamFunction<"azure-openai-responses", AzureOpenAIRespons throw new Error("Request was aborted"); } + if (output.stopReason === "pending") { + throw new Error("Azure OpenAI Responses stream ended without a stop reason"); + } if (output.stopReason === "aborted" || output.stopReason === "error") { throw new Error("An unknown error occurred"); } diff --git a/packages/ai/src/api/bedrock-converse-stream.ts b/packages/ai/src/api/bedrock-converse-stream.ts index 4c8a42051..9b5df97ea 100644 --- a/packages/ai/src/api/bedrock-converse-stream.ts +++ b/packages/ai/src/api/bedrock-converse-stream.ts @@ -125,7 +125,7 @@ export const stream: StreamFunction<"bedrock-converse-stream", BedrockOptions> = totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, }, - stopReason: "stop", + stopReason: "pending", timestamp: Date.now(), }; @@ -285,6 +285,9 @@ export const stream: StreamFunction<"bedrock-converse-stream", BedrockOptions> = throw new Error("Request was aborted"); } + if (output.stopReason === "pending") { + throw new Error("Bedrock stream ended without a stop reason"); + } if (output.stopReason === "error" || output.stopReason === "aborted") { throw new Error(output.errorMessage || "An unknown error occurred"); } diff --git a/packages/ai/src/api/google-generative-ai.ts b/packages/ai/src/api/google-generative-ai.ts index c925cf885..33d1f2227 100644 --- a/packages/ai/src/api/google-generative-ai.ts +++ b/packages/ai/src/api/google-generative-ai.ts @@ -70,7 +70,7 @@ export const stream: StreamFunction<"google-generative-ai", GoogleOptions> = ( totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, }, - stopReason: "stop", + stopReason: "pending", timestamp: Date.now(), }; @@ -259,6 +259,9 @@ export const stream: StreamFunction<"google-generative-ai", GoogleOptions> = ( throw new Error("Request was aborted"); } + if (output.stopReason === "pending") { + throw new Error("Google stream ended without a finish reason"); + } if (output.stopReason === "aborted" || output.stopReason === "error") { throw new Error("An unknown error occurred"); } diff --git a/packages/ai/src/api/google-vertex.ts b/packages/ai/src/api/google-vertex.ts index 24ee3c95f..83d885bc8 100644 --- a/packages/ai/src/api/google-vertex.ts +++ b/packages/ai/src/api/google-vertex.ts @@ -88,7 +88,7 @@ export const stream: StreamFunction<"google-vertex", GoogleVertexOptions> = ( totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, }, - stopReason: "stop", + stopReason: "pending", timestamp: Date.now(), }; @@ -276,6 +276,9 @@ export const stream: StreamFunction<"google-vertex", GoogleVertexOptions> = ( throw new Error("Request was aborted"); } + if (output.stopReason === "pending") { + throw new Error("Google Vertex stream ended without a finish reason"); + } if (output.stopReason === "aborted" || output.stopReason === "error") { throw new Error("An unknown error occurred"); } diff --git a/packages/ai/src/api/mistral-conversations.ts b/packages/ai/src/api/mistral-conversations.ts index 3a5e59dd7..44ab40699 100644 --- a/packages/ai/src/api/mistral-conversations.ts +++ b/packages/ai/src/api/mistral-conversations.ts @@ -84,6 +84,9 @@ export const stream: StreamFunction<"mistral-conversations", MistralOptions> = ( throw new Error("Request was aborted"); } + if (output.stopReason === "pending") { + throw new Error("Mistral stream ended without a finish reason"); + } if (output.stopReason === "aborted" || output.stopReason === "error") { throw new Error("An unknown error occurred"); } @@ -146,7 +149,7 @@ function createOutput(model: Model<"mistral-conversations">): AssistantMessage { totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, }, - stopReason: "stop", + stopReason: "pending", timestamp: Date.now(), }; } diff --git a/packages/ai/src/api/openai-codex-responses.ts b/packages/ai/src/api/openai-codex-responses.ts index cf89ee034..842dcc175 100644 --- a/packages/ai/src/api/openai-codex-responses.ts +++ b/packages/ai/src/api/openai-codex-responses.ts @@ -112,6 +112,17 @@ interface RequestBody { [key: string]: unknown; } +type SuccessfulAssistantMessage = AssistantMessage & { stopReason: "stop" | "length" | "toolUse" }; + +function assertSuccessfulOutput(output: AssistantMessage): asserts output is SuccessfulAssistantMessage { + if (output.stopReason === "pending") { + throw new Error("Codex stream ended without a stop reason"); + } + if (output.stopReason === "error" || output.stopReason === "aborted") { + throw new Error(output.errorMessage || "An unknown error occurred"); + } +} + // ============================================================================ // Retry Helpers // ============================================================================ @@ -252,7 +263,7 @@ export const stream: StreamFunction<"openai-codex-responses", OpenAICodexRespons totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, }, - stopReason: "stop", + stopReason: "pending", timestamp: Date.now(), }; @@ -324,9 +335,10 @@ export const stream: StreamFunction<"openai-codex-responses", OpenAICodexRespons if (options?.signal?.aborted) { throw new Error("Request was aborted"); } + assertSuccessfulOutput(output); stream.push({ type: "done", - reason: output.stopReason as "stop" | "length" | "toolUse", + reason: output.stopReason, message: output, }); stream.end(); @@ -471,7 +483,8 @@ export const stream: StreamFunction<"openai-codex-responses", OpenAICodexRespons throw new Error("Request was aborted"); } - stream.push({ type: "done", reason: output.stopReason as "stop" | "length" | "toolUse", message: output }); + assertSuccessfulOutput(output); + stream.push({ type: "done", reason: output.stopReason, message: output }); stream.end(); } catch (error) { for (const block of output.content) { diff --git a/packages/ai/src/api/openai-completions.ts b/packages/ai/src/api/openai-completions.ts index 29bb064f9..888defe2f 100644 --- a/packages/ai/src/api/openai-completions.ts +++ b/packages/ai/src/api/openai-completions.ts @@ -215,7 +215,7 @@ export const stream: StreamFunction<"openai-completions", OpenAICompletionsOptio totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, }, - stopReason: "stop", + stopReason: "pending", timestamp: Date.now(), }; @@ -574,7 +574,7 @@ export const stream: StreamFunction<"openai-completions", OpenAICompletionsOptio if (output.stopReason === "error") { throw new Error(output.errorMessage || "Provider returned an error stop reason"); } - if (!hasFinishReason) { + if (!hasFinishReason || output.stopReason === "pending") { throw new Error("Stream ended without finish_reason"); } diff --git a/packages/ai/src/api/openai-responses-shared.ts b/packages/ai/src/api/openai-responses-shared.ts index de98cd48d..1b161578a 100644 --- a/packages/ai/src/api/openai-responses-shared.ts +++ b/packages/ai/src/api/openai-responses-shared.ts @@ -423,6 +423,11 @@ export async function processResponsesStream( let sawTerminalResponseEvent = false; const outputSlots = new Map(); const reasoningBlocksById = new Map(); + const applyMessagePhaseStopReason = (item: ResponseOutputItem): void => { + if (item.type === "message" && item.phase === "final_answer") { + output.stopReason = "stop"; + } + }; const getSlot = ( outputIndex: number, type: TType, @@ -453,6 +458,7 @@ export async function processResponsesStream( return slot; } if (item.type === "message") { + applyMessagePhaseStopReason(item); const block: TextContent = { type: "text", text: "" }; output.content.push(block); const slot = { type: "text", block, contentIndex: output.content.length - 1 } satisfies ResponsesOutputSlot; @@ -648,6 +654,7 @@ export async function processResponsesStream( pushToolCallDelta(slot, appendCustomToolCallInput(slot.block, event.input, true)); } else if (event.type === "response.output_item.done") { const item = event.item; + applyMessagePhaseStopReason(item); const slot = getOrCreateSlot(event.output_index, item); if (item.type === "reasoning" && slot?.type === "thinking") { diff --git a/packages/ai/src/api/openai-responses.ts b/packages/ai/src/api/openai-responses.ts index 59f92a25c..a34f89912 100644 --- a/packages/ai/src/api/openai-responses.ts +++ b/packages/ai/src/api/openai-responses.ts @@ -121,7 +121,7 @@ export const stream: StreamFunction<"openai-responses", OpenAIResponsesOptions> totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, }, - stopReason: "stop", + stopReason: "pending", timestamp: Date.now(), }; @@ -167,6 +167,9 @@ export const stream: StreamFunction<"openai-responses", OpenAIResponsesOptions> throw new Error("Request was aborted"); } + if (output.stopReason === "pending") { + throw new Error("OpenAI Responses stream ended without a stop reason"); + } if (output.stopReason === "aborted" || output.stopReason === "error") { throw new Error("An unknown error occurred"); } diff --git a/packages/ai/src/api/pi-messages.ts b/packages/ai/src/api/pi-messages.ts index ee83f4ec6..38c2eac42 100644 --- a/packages/ai/src/api/pi-messages.ts +++ b/packages/ai/src/api/pi-messages.ts @@ -181,7 +181,7 @@ function createEventConverter(model: Model<"pi-messages">) { provider: model.provider, model: model.id, usage: createEmptyUsage(), - stopReason: "stop", + stopReason: "pending", timestamp: Date.now(), }; const toolJson = new Map(); diff --git a/packages/ai/src/providers/faux.ts b/packages/ai/src/providers/faux.ts index 4a26f1ad6..84bdd2a95 100644 --- a/packages/ai/src/providers/faux.ts +++ b/packages/ai/src/providers/faux.ts @@ -313,7 +313,7 @@ async function streamWithDeltas( tokensPerSecond: number | undefined, signal: AbortSignal | undefined, ): Promise { - const partial: AssistantMessage = { ...message, content: [] }; + const partial: AssistantMessage = { ...message, content: [], stopReason: "pending" }; if (signal?.aborted) { const aborted = createAbortedMessage(partial); stream.push({ type: "error", reason: "aborted", error: aborted }); @@ -390,6 +390,9 @@ async function streamWithDeltas( stream.push({ type: "toolcall_end", contentIndex: index, toolCall: block, partial: { ...partial } }); } + if (message.stopReason === "pending") { + throw new Error("Faux response ended without a stop reason"); + } if (message.stopReason === "error" || message.stopReason === "aborted") { stream.push({ type: "error", reason: message.stopReason, error: message }); stream.end(message); diff --git a/packages/ai/src/types.ts b/packages/ai/src/types.ts index 1d8fcfd3a..ed7b61823 100644 --- a/packages/ai/src/types.ts +++ b/packages/ai/src/types.ts @@ -379,7 +379,7 @@ export interface Usage { }; } -export type StopReason = "stop" | "length" | "toolUse" | "error" | "aborted"; +export type StopReason = "pending" | "stop" | "length" | "toolUse" | "error" | "aborted"; export interface UserMessage { role: "user"; diff --git a/packages/ai/test/azure-openai-responses-reasoning-replay.test.ts b/packages/ai/test/azure-openai-responses-reasoning-replay.test.ts index 12c0fb624..762638a32 100644 --- a/packages/ai/test/azure-openai-responses-reasoning-replay.test.ts +++ b/packages/ai/test/azure-openai-responses-reasoning-replay.test.ts @@ -34,7 +34,7 @@ function createOutput(model: Model<"azure-openai-responses">): AssistantMessage totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, }, - stopReason: "stop", + stopReason: "pending", timestamp: Date.now(), }; } diff --git a/packages/ai/test/constrained-sampling.test.ts b/packages/ai/test/constrained-sampling.test.ts index 244bcbcb7..9cf962e70 100644 --- a/packages/ai/test/constrained-sampling.test.ts +++ b/packages/ai/test/constrained-sampling.test.ts @@ -44,7 +44,7 @@ function makeOutput(): AssistantMessage { provider: "openai", model: "gpt-test", usage: makeUsage(), - stopReason: "stop", + stopReason: "pending", timestamp: Date.now(), }; } diff --git a/packages/ai/test/faux-provider.test.ts b/packages/ai/test/faux-provider.test.ts index 3d8a190f6..557885f23 100644 --- a/packages/ai/test/faux-provider.test.ts +++ b/packages/ai/test/faux-provider.test.ts @@ -193,6 +193,24 @@ describe("faux provider", () => { } }); + it("rejects a queued response without a terminal stop reason", async () => { + const registration = registerFauxProvider(); + registrations.push(registration); + registration.setResponses([fauxAssistantMessage("partial", { stopReason: "pending" })]); + + const events = await collectEvents( + stream(registration.getModel(), { messages: [{ role: "user", content: "hi", timestamp: Date.now() }] }), + ); + + expect(events.some((event) => event.type === "done")).toBe(false); + const terminal = events.at(-1); + expect(terminal?.type).toBe("error"); + if (terminal?.type === "error") { + expect(terminal.error.stopReason).toBe("error"); + expect(terminal.error.errorMessage).toBe("Faux response ended without a stop reason"); + } + }); + it("estimates prompt and output tokens from serialized context", async () => { const registration = registerFauxProvider(); registrations.push(registration); @@ -374,6 +392,7 @@ describe("faux provider", () => { stream(registration.getModel(), { messages: [{ role: "user", content: "hi", timestamp: Date.now() }] }), ); + expect(events[0]).toMatchObject({ type: "start", partial: { stopReason: "pending" } }); expect(events.map((event) => event.type)).toEqual([ "start", "thinking_start", diff --git a/packages/ai/test/openai-responses-partial-json-cleanup.test.ts b/packages/ai/test/openai-responses-partial-json-cleanup.test.ts index e4f4de4df..d01800c48 100644 --- a/packages/ai/test/openai-responses-partial-json-cleanup.test.ts +++ b/packages/ai/test/openai-responses-partial-json-cleanup.test.ts @@ -19,7 +19,7 @@ function createOutput(model: Model<"openai-responses">): AssistantMessage { totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, }, - stopReason: "stop", + stopReason: "pending", timestamp: Date.now(), }; } diff --git a/packages/ai/test/openai-responses-terminal-event.test.ts b/packages/ai/test/openai-responses-terminal-event.test.ts index f39366c1c..fcc3cc522 100644 --- a/packages/ai/test/openai-responses-terminal-event.test.ts +++ b/packages/ai/test/openai-responses-terminal-event.test.ts @@ -80,7 +80,7 @@ function createOutput(model: Model<"openai-responses">): AssistantMessage { totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, }, - stopReason: "stop", + stopReason: "pending", timestamp: Date.now(), }; } @@ -153,6 +153,51 @@ async function* createFailedEvents(): AsyncIterable { } as ResponseStreamEvent; } +async function* createPhasedMessageEvents( + phases: readonly ["commentary" | "final_answer", "commentary" | "final_answer"], + terminalStatus: "completed" | "incomplete" = "completed", +): AsyncIterable { + yield { + type: "response.output_item.added", + sequence_number: 0, + output_index: 0, + item: { + type: "message", + id: "msg_phase", + role: "assistant", + status: "in_progress", + content: [], + phase: phases[0], + }, + } as ResponseStreamEvent; + yield { + type: "response.output_item.done", + sequence_number: 1, + output_index: 0, + item: { + type: "message", + id: "msg_phase", + role: "assistant", + status: "completed", + content: [{ type: "output_text", text: "answer", annotations: [] }], + phase: phases[1], + }, + } as ResponseStreamEvent; + if (terminalStatus === "incomplete") { + yield { + type: "response.incomplete", + sequence_number: 2, + response: { id: "resp_phase", status: "incomplete" }, + } as ResponseStreamEvent; + return; + } + yield { + type: "response.completed", + sequence_number: 2, + response: { id: "resp_phase", status: "completed" }, + } as ResponseStreamEvent; +} + describe("OpenAI Responses terminal event handling", () => { it("rejects streams that end before a terminal response event", async () => { const model = createModel(); @@ -173,18 +218,64 @@ describe("OpenAI Responses terminal event handling", () => { }; const stream = streamOpenAIResponses(model, context, { apiKey: "test" }); const events: AssistantMessageEvent[] = []; + let initialStopReason: AssistantMessage["stopReason"] | undefined; for await (const event of stream) { + if (event.type === "start") initialStopReason = event.partial.stopReason; events.push(event); } const result = await stream.result(); const lastEvent = events.at(-1); + expect(initialStopReason).toBe("pending"); expect(lastEvent?.type).toBe("error"); expect(result.stopReason).toBe("error"); expect(result.errorMessage).toBe("OpenAI Responses stream ended before a terminal response event"); }); + it.each([ + { phases: ["commentary", "commentary"], expected: ["pending", "pending"] }, + { phases: ["final_answer", "final_answer"], expected: ["stop", "stop"] }, + { phases: ["commentary", "final_answer"], expected: ["pending", "stop"] }, + ] as const)("tracks message phases $phases", async ({ phases, expected }) => { + const model = createModel(); + const output = createOutput(model); + const stream = new AssistantMessageEventStream(); + const observedStopReasons: AssistantMessage["stopReason"][] = []; + const push = stream.push.bind(stream); + stream.push = (event) => { + if ("partial" in event) observedStopReasons.push(event.partial.stopReason); + push(event); + }; + + await processResponsesStream(createPhasedMessageEvents(phases), output, stream, model); + + expect(observedStopReasons).toEqual(expected); + expect(output.stopReason).toBe("stop"); + }); + + it("replaces a provisional final-answer stop with an incomplete terminal reason", async () => { + const model = createModel(); + const output = createOutput(model); + const stream = new AssistantMessageEventStream(); + const observedStopReasons: AssistantMessage["stopReason"][] = []; + const push = stream.push.bind(stream); + stream.push = (event) => { + if ("partial" in event) observedStopReasons.push(event.partial.stopReason); + push(event); + }; + + await processResponsesStream( + createPhasedMessageEvents(["final_answer", "final_answer"], "incomplete"), + output, + stream, + model, + ); + + expect(observedStopReasons).toEqual(["stop", "stop"]); + expect(output.stopReason).toBe("length"); + }); + it("finalizes completed terminal events as stop", async () => { const model = createModel(); const output = createOutput(model); diff --git a/packages/ai/test/pi-messages.test.ts b/packages/ai/test/pi-messages.test.ts index a689c9f27..f0c49a962 100644 --- a/packages/ai/test/pi-messages.test.ts +++ b/packages/ai/test/pi-messages.test.ts @@ -2,7 +2,7 @@ import { createServer, type IncomingMessage, type Server, type ServerResponse } import type { AddressInfo } from "node:net"; import { afterEach, describe, expect, it } from "vitest"; import { type PiMessagesOptions, stream, streamSimple } from "../src/api/pi-messages.ts"; -import type { Api, AssistantMessageEvent, Context, Model } from "../src/types.ts"; +import type { Api, AssistantMessageEvent, Context, Model, StopReason } from "../src/types.ts"; type RecordedRequest = { url: string; @@ -116,6 +116,7 @@ describe("pi-messages", () => { const model = createModel(baseUrl); const events: AssistantMessageEvent[] = []; + const partialStopReasons: StopReason[] = []; const eventStream = stream(model, context, { apiKey: "test-key", sessionId: "session-1", @@ -124,10 +125,14 @@ describe("pi-messages", () => { headers: { "x-custom": "1" }, }); for await (const event of eventStream) { + if ("partial" in event) { + partialStopReasons.push(event.partial.stopReason); + } events.push(event); } const message = await eventStream.result(); + expect(partialStopReasons[0]).toBe("pending"); expect(message.stopReason).toBe("toolUse"); expect(message.usage).toEqual(usage); expect(message.responseId).toBe("resp_1"); diff --git a/packages/coding-agent/docs/custom-provider.md b/packages/coding-agent/docs/custom-provider.md index 83b9cff66..652e2346b 100644 --- a/packages/coding-agent/docs/custom-provider.md +++ b/packages/coding-agent/docs/custom-provider.md @@ -442,7 +442,7 @@ function streamMyProvider( totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, }, - stopReason: "stop", + stopReason: "pending", timestamp: Date.now(), }; @@ -451,12 +451,18 @@ function streamMyProvider( stream.push({ type: "start", partial: output }); // Make API request and process response... - // Push content events as they arrive... + // Push content events as they arrive and set stopReason from the terminal event. + if (output.stopReason === "pending") { + throw new Error("Provider stream ended without a stop reason"); + } + if (output.stopReason === "error" || output.stopReason === "aborted") { + throw new Error(output.errorMessage || "An unknown error occurred"); + } // Push done event stream.push({ type: "done", - reason: output.stopReason as "stop" | "length" | "toolUse", + reason: output.stopReason, message: output }); stream.end(); diff --git a/packages/coding-agent/docs/session-format.md b/packages/coding-agent/docs/session-format.md index e2c49f1d0..c5956ed3b 100644 --- a/packages/coding-agent/docs/session-format.md +++ b/packages/coding-agent/docs/session-format.md @@ -117,6 +117,8 @@ interface Usage { } ``` +The exported pi-ai `StopReason` type also includes `"pending"`, but that value is reserved for partial messages in streaming events. Terminal `done`/`error` messages replace it with a completion reason before pi persists the assistant message, so `"pending"` should never appear in session JSONL. + ### Extended Message Types (from pi-coding-agent) ```typescript diff --git a/packages/coding-agent/examples/extensions/custom-provider-anthropic/index.ts b/packages/coding-agent/examples/extensions/custom-provider-anthropic/index.ts index c4bebc329..45304be42 100644 --- a/packages/coding-agent/examples/extensions/custom-provider-anthropic/index.ts +++ b/packages/coding-agent/examples/extensions/custom-provider-anthropic/index.ts @@ -353,7 +353,7 @@ function streamCustomAnthropic( totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, }, - stopReason: "stop", + stopReason: "pending", timestamp: Date.now(), }; @@ -546,8 +546,14 @@ function streamCustomAnthropic( if (options?.signal?.aborted) { throw new Error("Request was aborted"); } + if (output.stopReason === "pending") { + throw new Error("Anthropic stream ended without a stop reason"); + } + if (output.stopReason === "error" || output.stopReason === "aborted") { + throw new Error(output.errorMessage || "An unknown error occurred"); + } - stream.push({ type: "done", reason: output.stopReason as "stop" | "length" | "toolUse", message: output }); + stream.push({ type: "done", reason: output.stopReason, message: output }); stream.end(); } catch (error) { for (const block of output.content) delete (block as any).index; diff --git a/packages/coding-agent/test/suite/regressions/6647-compaction-retries-transient-stream-drop.test.ts b/packages/coding-agent/test/suite/regressions/6647-compaction-retries-transient-stream-drop.test.ts index 883836877..d11216cb9 100644 --- a/packages/coding-agent/test/suite/regressions/6647-compaction-retries-transient-stream-drop.test.ts +++ b/packages/coding-agent/test/suite/regressions/6647-compaction-retries-transient-stream-drop.test.ts @@ -59,18 +59,18 @@ describe("#6647 compaction retries transient summarization failures", () => { callCount++; const stream = createAssistantMessageEventStream(); queueMicrotask(() => { - if (message.stopReason === "error" || message.stopReason === "aborted") { - stream.push({ - type: "error", - reason: message.stopReason, - error: { ...message, api: model.api, provider: model.provider, model: model.id }, - }); + const response = { ...message, api: model.api, provider: model.provider, model: model.id }; + if (response.stopReason === "pending") { + const error: AssistantMessage = { + ...response, + stopReason: "error", + errorMessage: "Scripted response ended without a stop reason", + }; + stream.push({ type: "error", reason: "error", error }); + } else if (response.stopReason === "error" || response.stopReason === "aborted") { + stream.push({ type: "error", reason: response.stopReason, error: response }); } else { - stream.push({ - type: "done", - reason: message.stopReason, - message: { ...message, api: model.api, provider: model.provider, model: model.id }, - }); + stream.push({ type: "done", reason: response.stopReason, message: response }); } }); return stream; diff --git a/packages/coding-agent/test/test-harness.test.ts b/packages/coding-agent/test/test-harness.test.ts index 0c4a151fd..3ac716226 100644 --- a/packages/coding-agent/test/test-harness.test.ts +++ b/packages/coding-agent/test/test-harness.test.ts @@ -88,6 +88,17 @@ describe("test harness", () => { expect(assistantMessages[0].errorMessage).toBe("something broke"); }); + it("turns a pending terminal response into an error", async () => { + harness = await createHarness({ responses: [{ text: "partial", stopReason: "pending" }] }); + + await harness.session.prompt("hi"); + + const assistantMessages = harness.session.messages.filter((m): m is AssistantMessage => m.role === "assistant"); + expect(assistantMessages).toHaveLength(1); + expect(assistantMessages[0].stopReason).toBe("error"); + expect(assistantMessages[0].errorMessage).toBe("Faux response ended without a stop reason"); + }); + it("retry on transient error", async () => { harness = await createHarness({ responses: [{ error: "overloaded_error" }, "recovered"], diff --git a/packages/coding-agent/test/test-harness.ts b/packages/coding-agent/test/test-harness.ts index 89a944118..9b14ed402 100644 --- a/packages/coding-agent/test/test-harness.ts +++ b/packages/coding-agent/test/test-harness.ts @@ -184,10 +184,8 @@ function chunkString(text: string): string[] { * intermediate delta events for each content block. */ function streamWithDeltas(stream: AssistantMessageEventStream, message: AssistantMessage): void { - const isError = message.stopReason === "error" || message.stopReason === "aborted"; - // Build partial progressively as we stream content blocks - const partial: AssistantMessage = { ...message, content: [] }; + const partial: AssistantMessage = { ...message, content: [], stopReason: "pending" }; stream.push({ type: "start", partial: { ...partial } }); for (let i = 0; i < message.content.length; i++) { @@ -243,11 +241,20 @@ function streamWithDeltas(stream: AssistantMessageEventStream, message: Assistan } } - if (isError) { - stream.push({ type: "error", reason: message.stopReason as "error" | "aborted", error: message }); - } else { - stream.push({ type: "done", reason: message.stopReason as "stop" | "length" | "toolUse", message }); + if (message.stopReason === "pending") { + const error: AssistantMessage = { + ...message, + stopReason: "error", + errorMessage: "Faux response ended without a stop reason", + }; + stream.push({ type: "error", reason: "error", error }); + return; } + if (message.stopReason === "error" || message.stopReason === "aborted") { + stream.push({ type: "error", reason: message.stopReason, error: message }); + return; + } + stream.push({ type: "done", reason: message.stopReason, message }); } function makeEvent(