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
This commit is contained in:
Lucas Meijer 2026-07-27 11:56:15 +02:00 • committed by GitHub
parent 60f6a8034d
commit f9a49869c2
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
27 changed files with 237 additions and 45 deletions

View file

@ -120,7 +120,7 @@ export function streamProxy(model: Model<any>, 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,

View file

@ -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

View file

@ -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");
}

View file

@ -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");
}

View file

@ -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");
}

View file

@ -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");
}

View file

@ -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");
}

View file

@ -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(),
};
}

View file

@ -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) {

View file

@ -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");
}

View file

@ -423,6 +423,11 @@ export async function processResponsesStream<TApi extends Api>(
let sawTerminalResponseEvent = false;
const outputSlots = new Map<number, ResponsesOutputSlot>();
const reasoningBlocksById = new Map<string, ThinkingContent>();
const applyMessagePhaseStopReason = (item: ResponseOutputItem): void => {
if (item.type === "message" && item.phase === "final_answer") {
output.stopReason = "stop";
}
};
const getSlot = <TType extends ResponsesOutputSlot["type"]>(
outputIndex: number,
type: TType,
@ -453,6 +458,7 @@ export async function processResponsesStream<TApi extends Api>(
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<TApi extends Api>(
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") {

View file

@ -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");
}

View file

@ -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<number, string>();

View file

@ -313,7 +313,7 @@ async function streamWithDeltas(
tokensPerSecond: number | undefined,
signal: AbortSignal | undefined,
): Promise<void> {
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);

View file

@ -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";

View file

@ -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(),
};
}

View file

@ -44,7 +44,7 @@ function makeOutput(): AssistantMessage {
provider: "openai",
model: "gpt-test",
usage: makeUsage(),
stopReason: "stop",
stopReason: "pending",
timestamp: Date.now(),
};
}

View file

@ -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",

View file

@ -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(),
};
}

View file

@ -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<ResponseStreamEvent> {
} as ResponseStreamEvent;
}
async function* createPhasedMessageEvents(
phases: readonly ["commentary" | "final_answer", "commentary" | "final_answer"],
terminalStatus: "completed" | "incomplete" = "completed",
): AsyncIterable<ResponseStreamEvent> {
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);

View file

@ -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");

View file

@ -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();

View file

@ -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

View file

@ -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;

View file

@ -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;

View file

@ -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"],

View file

@ -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(