diff --git a/packages/ai/README.md b/packages/ai/README.md index 050ef00f8..97bf64471 100644 --- a/packages/ai/README.md +++ b/packages/ai/README.md @@ -652,9 +652,9 @@ for await (const event of s) { ### Complete Event Reference -Successful generation follows `start → updates* → done`. A failure after generation starts follows `start → updates* → error`. Request setup may fail before generation starts, in which case the stream contains only `error`; `done` and update events are invalid before `start`. +Successful generation follows `start → updates* → done`. A failure after generation starts follows `start → updates* → error`. Request setup may fail before generation starts, in which case the stream contains only `error`; `done` and update events are invalid before `start`. Direct API `streamSimple()` calls throw synchronously when request auth is missing. -Every non-terminal event's `partial` is the shared live response-so-far helper. It is intentionally not an event-time snapshot: providers may mutate the same message and content blocks as generation advances, including while older events wait in the stream queue. Inspect it when handling an event instead of retaining it as historical state. Text and ordinary thinking blocks are empty when their `*_start` event is emitted and grow only through matching `*_delta` events until the authoritative `*_end`; redacted thinking may be complete at start and emit no deltas. A streaming tool call starts with empty arguments and emits its full raw JSON through `toolcall_delta`. A provider that starts with complete arguments must emit a cumulative delta prefix that parses to those arguments at an event boundary before later argument deltas. +Every non-terminal event's `partial` is the shared live response-so-far helper. It is intentionally not an event-time snapshot: providers may mutate the same message and content blocks as generation advances, including while older events wait in the stream queue. Inspect it when handling an event instead of retaining it as historical state. Text and ordinary thinking blocks are empty when their `*_start` event is emitted and grow only through matching `*_delta` events until the authoritative `*_end`; redacted thinking may be complete at start and emit no deltas. Tool-call arguments at `toolcall_start` are provider-specific; `toolcall_delta` carries subsequent JSON updates. All streaming events emitted during assistant message generation: @@ -927,7 +927,7 @@ Every `AssistantMessage` includes a `stopReason` field that indicates how the ge ## Error Handling -Request failures never throw out of the stream functions: when a request ends with an error (including aborts and tool call validation errors), the streaming API emits an error event and the final message carries the details. Setup failures may emit `error` without `start`; failures after generation begins emit `start`, any observed updates, then `error`: +Request failures after a stream is returned never throw: when a request ends with an error (including aborts and tool call validation errors), the streaming API emits an error event and the final message carries the details. Setup failures may emit `error` without `start`; failures after generation begins emit `start`, any observed updates, then `error`. Direct API `streamSimple()` calls throw synchronously when request auth is missing: ```typescript // In streaming @@ -949,7 +949,7 @@ if (message.stopReason === 'error' || message.stopReason === 'aborted') { } ``` -Auth failures (no key configured, OAuth refresh failed, unknown provider) surface the same way: as a stream error with `stopReason: "error"`. +When using a provider collection, auth failures (OAuth refresh failed, unknown provider) surface as a stream error with `stopReason: "error"`. Direct API `streamSimple()` calls instead throw synchronously when their required auth is absent. ### Aborting Requests diff --git a/packages/ai/src/api/anthropic-messages.ts b/packages/ai/src/api/anthropic-messages.ts index d8d15cabe..de3a51cf4 100644 --- a/packages/ai/src/api/anthropic-messages.ts +++ b/packages/ai/src/api/anthropic-messages.ts @@ -851,6 +851,8 @@ export const streamSimple: StreamFunction<"anthropic-messages", SimpleStreamOpti context: Context, options?: SimpleStreamOptions, ): AssistantMessageEventStream => { + assertRequestAuth(model.provider, options?.apiKey, options?.headers); + const base = { ...buildBaseOptions(model, context, options, options?.apiKey), toolChoice: options?.toolChoice, diff --git a/packages/ai/src/api/azure-openai-responses.ts b/packages/ai/src/api/azure-openai-responses.ts index e0011696a..56578805d 100644 --- a/packages/ai/src/api/azure-openai-responses.ts +++ b/packages/ai/src/api/azure-openai-responses.ts @@ -165,7 +165,10 @@ export const streamSimple: StreamFunction<"azure-openai-responses", SimpleStream context: Context, options?: SimpleStreamOptions, ): AssistantMessageEventStream => { - const apiKey = options?.apiKey ?? ""; + const apiKey = options?.apiKey; + if (!apiKey) { + throw new Error(`No API key for provider: ${model.provider}`); + } const base = { ...buildBaseOptions(model, context, options, apiKey), diff --git a/packages/ai/src/api/google-generative-ai.ts b/packages/ai/src/api/google-generative-ai.ts index 755bbb2d4..a0f9f39c9 100644 --- a/packages/ai/src/api/google-generative-ai.ts +++ b/packages/ai/src/api/google-generative-ai.ts @@ -299,7 +299,10 @@ export const streamSimple: StreamFunction<"google-generative-ai", SimpleStreamOp context: Context, options?: SimpleStreamOptions, ): AssistantMessageEventStream => { - const apiKey = options?.apiKey ?? ""; + const apiKey = options?.apiKey; + if (!apiKey) { + throw new Error(`No API key for provider: ${model.provider}`); + } const base = { ...buildBaseOptions(model, context, options, apiKey), diff --git a/packages/ai/src/api/mistral-conversations.ts b/packages/ai/src/api/mistral-conversations.ts index 821e39b21..979b467ed 100644 --- a/packages/ai/src/api/mistral-conversations.ts +++ b/packages/ai/src/api/mistral-conversations.ts @@ -183,7 +183,10 @@ export const streamSimple: StreamFunction<"mistral-conversations", SimpleStreamO context: Context, options?: SimpleStreamOptions, ): AssistantMessageEventStream => { - const apiKey = options?.apiKey ?? ""; + const apiKey = options?.apiKey; + if (!apiKey) { + throw new Error(`No API key for provider: ${model.provider}`); + } const base = { ...buildBaseOptions(model, context, options, apiKey), diff --git a/packages/ai/src/api/openai-codex-responses.ts b/packages/ai/src/api/openai-codex-responses.ts index 815bba489..d7e3749f5 100644 --- a/packages/ai/src/api/openai-codex-responses.ts +++ b/packages/ai/src/api/openai-codex-responses.ts @@ -494,7 +494,10 @@ export const streamSimple: StreamFunction<"openai-codex-responses", SimpleStream context: Context, options?: SimpleStreamOptions, ): AssistantMessageEventStream => { - const apiKey = options?.apiKey ?? ""; + const apiKey = options?.apiKey; + if (!apiKey) { + throw new Error(`No API key for provider: ${model.provider}`); + } const base = { ...buildBaseOptions(model, context, options, apiKey), diff --git a/packages/ai/src/api/openai-completions.ts b/packages/ai/src/api/openai-completions.ts index 71a541b7b..8ac17dbae 100644 --- a/packages/ai/src/api/openai-completions.ts +++ b/packages/ai/src/api/openai-completions.ts @@ -504,7 +504,7 @@ export const stream: StreamFunction<"openai-completions", OpenAICompletionsOptio type: "toolCall", id: toolCall.id || "", name, - arguments: {}, + arguments: hasCustomInput ? { [customInputProperty]: "" } : {}, partialArgs: hasCustomInput ? undefined : "", customInput: hasCustomInput ? { property: customInputProperty, jsonBuffer: { input: "", started: false, closed: false } } @@ -536,7 +536,7 @@ export const stream: StreamFunction<"openai-completions", OpenAICompletionsOptio } if (toolCall.custom && !toolCall.function && !block.customInput) { const customInputProperty = grammarToolInputProperties.get(block.name) ?? "input"; - block.arguments = {}; + block.arguments = { [customInputProperty]: "" }; block.customInput = { property: customInputProperty, jsonBuffer: { input: "", started: false, closed: false }, @@ -728,6 +728,8 @@ export const streamSimple: StreamFunction<"openai-completions", SimpleStreamOpti context: Context, options?: SimpleStreamOptions, ): AssistantMessageEventStream => { + getClientApiKey(model.provider, options?.apiKey, options?.headers); + const base = { ...buildBaseOptions(model, context, options, options?.apiKey), toolChoice: options?.toolChoice, diff --git a/packages/ai/src/api/openai-responses-shared.ts b/packages/ai/src/api/openai-responses-shared.ts index df3f7a0b6..7ac1add08 100644 --- a/packages/ai/src/api/openai-responses-shared.ts +++ b/packages/ai/src/api/openai-responses-shared.ts @@ -503,11 +503,12 @@ export async function processResponsesStream( } if (item.type === "custom_tool_call") { const inputProperty = options?.grammarToolInputProperties?.get(item.name) ?? "input"; + const input = item.input || ""; const block: StreamingToolCall = { type: "toolCall", id: `${item.call_id}|${item.id}`, name: item.name, - arguments: {}, + arguments: { [inputProperty]: input }, ...(item.namespace !== undefined ? { namespace: item.namespace } : {}), customInput: { property: inputProperty, @@ -522,9 +523,6 @@ export async function processResponsesStream( } satisfies ResponsesOutputSlot; outputSlots.set(outputIndex, slot); stream.push({ type: "toolcall_start", contentIndex: slot.contentIndex, partial: output }); - if (item.input) { - pushToolCallDelta(slot, appendCustomToolCallInput(block, item.input, false)); - } return slot; } return undefined; diff --git a/packages/ai/src/api/openai-responses.ts b/packages/ai/src/api/openai-responses.ts index 99838cfbb..2b0723798 100644 --- a/packages/ai/src/api/openai-responses.ts +++ b/packages/ai/src/api/openai-responses.ts @@ -201,6 +201,8 @@ export const streamSimple: StreamFunction<"openai-responses", SimpleStreamOption context: Context, options?: SimpleStreamOptions, ): AssistantMessageEventStream => { + getClientApiKey(model.provider, options?.apiKey, options?.headers); + const base = { ...buildBaseOptions(model, context, options, options?.apiKey), toolChoice: options?.toolChoice, diff --git a/packages/ai/src/types.ts b/packages/ai/src/types.ts index 152033981..833587b96 100644 --- a/packages/ai/src/types.ts +++ b/packages/ai/src/types.ts @@ -325,8 +325,9 @@ export interface SimpleStreamOptions extends StreamOptions { // // Contract: // - Must return an AssistantMessageEventStream. -// - Once invoked, request/model/runtime failures should be encoded in the -// returned stream, not thrown. +// - Direct streamSimple() calls may throw synchronously when request auth is +// missing. Once a stream is returned, request/model/runtime failures should +// be encoded in that stream. // - Error termination must produce an AssistantMessage with stopReason // "error" or "aborted" and errorMessage, emitted via the stream protocol. export type StreamFunction = ( @@ -532,16 +533,15 @@ export interface Context { * Successful streams emit `start` before partial updates and terminate with * `done`. A stream may terminate directly with `error` when request setup fails * before generation starts; after `start`, failures also terminate with `error`. + * Direct `streamSimple()` calls throw synchronously when request auth is missing. * Updates and `done` must never appear before `start`. * * `partial` is the shared live response-so-far helper, not an event-time * snapshot. Text and thinking blocks are empty when their `*_start` event is * emitted and grow only through their corresponding `*_delta` events until the * authoritative `*_end`. Redacted thinking may be complete at start and emit no - * deltas. A streaming tool call starts with empty arguments and emits its full - * raw JSON through `toolcall_delta`; a provider that starts with complete - * arguments must emit a cumulative delta prefix that parses to those arguments - * before emitting any later argument delta. + * deltas. Tool-call arguments at `toolcall_start` are provider-specific; + * `toolcall_delta` carries subsequent JSON updates. */ export type AssistantMessageEvent = | { type: "start"; partial: AssistantMessage } diff --git a/packages/ai/src/utils/assistant-message-frame.ts b/packages/ai/src/utils/assistant-message-frame.ts index 3c22f78e3..f314a9f1d 100644 --- a/packages/ai/src/utils/assistant-message-frame.ts +++ b/packages/ai/src/utils/assistant-message-frame.ts @@ -113,6 +113,23 @@ function serializedArguments(argumentsValue: ToolCall["arguments"]): string { const EMPTY_PARSED_TOOL_ARGUMENTS = serializedArguments(parseStreamingJson("")); +function isJsonPrefix(snapshot: unknown, current: unknown): boolean { + if (typeof snapshot === "string") return typeof current === "string" && current.startsWith(snapshot); + if (Array.isArray(snapshot)) { + return ( + Array.isArray(current) && + snapshot.length <= current.length && + snapshot.every((value, index) => isJsonPrefix(value, current[index])) + ); + } + if (typeof snapshot !== "object" || snapshot === null) return Object.is(snapshot, current); + if (typeof current !== "object" || current === null || Array.isArray(current)) return false; + const currentRecord = current as Record; + return Object.entries(snapshot).every( + ([key, value]) => Object.hasOwn(currentRecord, key) && isJsonPrefix(value, currentRecord[key]), + ); +} + /** * Encodes one assistant stream. `partial` remains a shared live accumulator; * the encoder uses per-block offsets to avoid replaying deltas already visible @@ -226,11 +243,13 @@ export class AssistantMessageFrameEncoder { : { type: "toolcall_delta", contentIndex: event.contentIndex, delta: event.delta }; } state.catchupJson += event.delta; - if ( - serializedArguments(parseStreamingJson(state.catchupJson)) !== - state.snapshotArguments - ) { - return undefined; + const argumentsValue = parseStreamingJson(state.catchupJson); + if (serializedArguments(argumentsValue) !== state.snapshotArguments) { + // Legacy grammar calls include the initial input in toolcall_start, but their + // JSON delta stream still begins at an empty input. Its parsed arguments can + // therefore extend, rather than exactly reproduce, the start snapshot. + const snapshotArguments = parseStreamingJson(state.snapshotArguments); + if (!isJsonPrefix(snapshotArguments, argumentsValue)) return undefined; } state.caughtUp = true; state.snapshotArguments = ""; diff --git a/packages/ai/test/assistant-message-frame.test.ts b/packages/ai/test/assistant-message-frame.test.ts index c8fa8c306..0c0564e23 100644 --- a/packages/ai/test/assistant-message-frame.test.ts +++ b/packages/ai/test/assistant-message-frame.test.ts @@ -283,7 +283,42 @@ describe("assistant message frames", () => { ]); }); - it("streams grammar-tool JSON compactly from an empty argument start", () => { + it("resumes legacy grammar tool JSON from initial arguments", () => { + const partial = seed(); + const encoder = new AssistantMessageFrameEncoder(); + const frames: AssistantMessageFrame[] = [frame(encoder, { type: "start", partial })]; + const toolCall = { type: "toolCall" as const, id: "call", name: "bash", arguments: { input: "a" } }; + partial.content.push(toolCall); + frames.push(frame(encoder, { type: "toolcall_start", contentIndex: 0, partial })); + toolCall.arguments = { input: "ab" }; + frames.push( + frame(encoder, { + type: "toolcall_delta", + contentIndex: 0, + delta: '{"input":"ab', + partial, + }), + ); + toolCall.arguments = { input: "abc" }; + frames.push( + frame(encoder, { + type: "toolcall_delta", + contentIndex: 0, + delta: 'c"}', + partial, + }), + ); + + expect(frames.slice(2)).toEqual([ + { type: "toolcall_checkpoint", contentIndex: 0, json: '{"input":"ab' }, + { type: "toolcall_delta", contentIndex: 0, delta: 'c"}' }, + ]); + expect(reduceAssistantMessageFrames(frames)?.content).toEqual([ + { type: "toolCall", id: "call", name: "bash", arguments: { input: "abc" } }, + ]); + }); + + it("streams tool JSON compactly from an empty argument start", () => { const partial = seed(); const encoder = new AssistantMessageFrameEncoder(); const frames: AssistantMessageFrame[] = [frame(encoder, { type: "start", partial })]; diff --git a/packages/ai/test/constrained-sampling.test.ts b/packages/ai/test/constrained-sampling.test.ts index f24edd1ff..7f768f69d 100644 --- a/packages/ai/test/constrained-sampling.test.ts +++ b/packages/ai/test/constrained-sampling.test.ts @@ -66,16 +66,23 @@ function makeTool(overrides: Partial = {}): Tool { }; } -function captureToolCallDeltas(stream: AssistantMessageEventStream): string[] { +function captureToolCallEvents(stream: AssistantMessageEventStream): { + starts: ToolCall["arguments"][]; + deltas: string[]; +} { + const starts: ToolCall["arguments"][] = []; const deltas: string[] = []; const originalPush = stream.push.bind(stream); stream.push = (event) => { - if (event.type === "toolcall_delta") { + if (event.type === "toolcall_start") { + const block = event.partial.content[event.contentIndex]; + if (block?.type === "toolCall") starts.push(structuredClone(block.arguments)); + } else if (event.type === "toolcall_delta") { deltas.push(event.delta); } originalPush(event); }; - return deltas; + return { starts, deltas }; } describe("constrained tool sampling", () => { @@ -259,21 +266,21 @@ describe("constrained tool sampling", () => { ); }); - it("streams custom Responses tool calls as string arguments", async () => { + it("starts custom Responses tool calls with their initial input", async () => { const output = makeOutput(); const stream = new AssistantMessageEventStream(); - const deltas = captureToolCallDeltas(stream); + const { starts, deltas } = captureToolCallEvents(stream); const events = [ { type: "response.output_item.added", output_index: 0, - item: { type: "custom_tool_call", call_id: "call_1", id: "ctc_1", name: "sample_tool", input: "" }, + item: { type: "custom_tool_call", call_id: "call_1", id: "ctc_1", name: "sample_tool", input: "a" }, }, { type: "response.custom_tool_call_input.delta", output_index: 0, item_id: "ctc_1", - delta: "ab", + delta: "b", }, { type: "response.custom_tool_call_input.done", @@ -297,6 +304,7 @@ describe("constrained tool sampling", () => { }); expect(output.stopReason).toBe("toolUse"); + expect(starts).toEqual([{ payload: "a" }]); expect(output.content).toEqual([ { type: "toolCall", id: "call_1|ctc_1", name: "sample_tool", arguments: { payload: "abc" } }, ]); diff --git a/packages/ai/test/pre-generation-error.test.ts b/packages/ai/test/pre-generation-error.test.ts index d30fde904..991f37f85 100644 --- a/packages/ai/test/pre-generation-error.test.ts +++ b/packages/ai/test/pre-generation-error.test.ts @@ -23,22 +23,18 @@ function model(api: TApi): Model { }; } -async function expectPreGenerationError(create: () => AssistantMessageEventStream): Promise { - const stream = create(); - const events = []; - for await (const event of stream) events.push(event); - expect(events.map((event) => event.type)).toEqual(["error"]); - expect(await stream.result()).toMatchObject({ stopReason: "error", content: [] }); +function expectMissingAuthThrows(create: () => AssistantMessageEventStream): void { + expect(create).toThrow("No API key for provider: test-provider"); } -describe("direct API pre-generation errors", () => { - it("return an error stream instead of throwing synchronously when auth is missing", async () => { - await expectPreGenerationError(() => streamAnthropic(model("anthropic-messages"), { messages: [] }, {})); - await expectPreGenerationError(() => streamAzure(model("azure-openai-responses"), { messages: [] }, {})); - await expectPreGenerationError(() => streamGoogle(model("google-generative-ai"), { messages: [] }, {})); - await expectPreGenerationError(() => streamMistral(model("mistral-conversations"), { messages: [] }, {})); - await expectPreGenerationError(() => streamCodex(model("openai-codex-responses"), { messages: [] }, {})); - await expectPreGenerationError(() => streamOpenAICompletions(model("openai-completions"), { messages: [] }, {})); - await expectPreGenerationError(() => streamOpenAIResponses(model("openai-responses"), { messages: [] }, {})); +describe("direct API authentication", () => { + it("throws synchronously when auth is missing", () => { + expectMissingAuthThrows(() => streamAnthropic(model("anthropic-messages"), { messages: [] }, {})); + expectMissingAuthThrows(() => streamAzure(model("azure-openai-responses"), { messages: [] }, {})); + expectMissingAuthThrows(() => streamGoogle(model("google-generative-ai"), { messages: [] }, {})); + expectMissingAuthThrows(() => streamMistral(model("mistral-conversations"), { messages: [] }, {})); + expectMissingAuthThrows(() => streamCodex(model("openai-codex-responses"), { messages: [] }, {})); + expectMissingAuthThrows(() => streamOpenAICompletions(model("openai-completions"), { messages: [] }, {})); + expectMissingAuthThrows(() => streamOpenAIResponses(model("openai-responses"), { messages: [] }, {})); }); });