diff --git a/packages/ai/src/providers/azure-openai-responses.test.ts b/packages/ai/src/providers/azure-openai-responses.test.ts index e11469c4b5cb..7fc13ae8ccae 100644 --- a/packages/ai/src/providers/azure-openai-responses.test.ts +++ b/packages/ai/src/providers/azure-openai-responses.test.ts @@ -104,4 +104,24 @@ describe("azure-openai-responses", () => { } } }); + + it("disables response storage and clamps small output limits", async () => { + let sentParams: { max_output_tokens?: unknown; store?: unknown } | undefined; + const hostFetch: typeof fetch = async (input, init) => { + sentParams = (await new Request(input, init).json()) as typeof sentParams; + return Response.json({ error: { message: "captured" } }, { status: 400 }); + }; + + configureAiTransportHost({ buildModelFetch: () => hostFetch }); + try { + await streamSimpleAzureOpenAIResponses(azureResponsesModel, context, { + apiKey: "test-api-key", + maxTokens: 1, + }).result(); + + expect(sentParams).toMatchObject({ max_output_tokens: 16, store: false }); + } finally { + configureAiTransportHost({}); + } + }); }); diff --git a/packages/ai/src/providers/azure-openai-responses.ts b/packages/ai/src/providers/azure-openai-responses.ts index 9e6bfc0db105..41a77cac04a7 100644 --- a/packages/ai/src/providers/azure-openai-responses.ts +++ b/packages/ai/src/providers/azure-openai-responses.ts @@ -239,6 +239,7 @@ function buildParams( options?.cacheRetention === "none" ? undefined : clampOpenAIPromptCacheKey(options?.promptCacheKey ?? options?.sessionId), + store: false, }; applyCommonResponsesParams(params, model, context, options); diff --git a/packages/ai/src/providers/openai-chatgpt-responses.test.ts b/packages/ai/src/providers/openai-chatgpt-responses.test.ts index 0e359d6d106d..c80bfb36a463 100644 --- a/packages/ai/src/providers/openai-chatgpt-responses.test.ts +++ b/packages/ai/src/providers/openai-chatgpt-responses.test.ts @@ -160,6 +160,28 @@ describe("streamOpenAICodexResponses transport", () => { expect(providerToken).toBe(`Bearer ${realToken}`); }); + it("clamps session headers to the backend limit", async () => { + const jwt = createJwt({ "https://api.openai.com/auth": { chatgpt_account_id: "acct" } }); + let headers: Headers | undefined; + vi.stubGlobal( + "fetch", + vi.fn(async (_input, init) => { + headers = new Headers(init?.headers); + return completedSseResponse(); + }), + ); + + const result = await streamOpenAICodexResponses(model, context, { + apiKey: jwt, + sessionId: "s".repeat(80), + transport: "sse", + }).result(); + + expect(result.stopReason).toBe("stop"); + expect(headers?.get("session_id")).toBe("s".repeat(64)); + expect(headers?.get("x-client-request-id")).toBe("s".repeat(64)); + }); + it("builds the first Node request with an OS-specific user agent", async () => { vi.resetModules(); const freshProvider = await import("./openai-chatgpt-responses.js"); @@ -822,6 +844,28 @@ describe("streamOpenAICodexResponses transport", () => { expect(payload).toMatchObject({ prompt_cache_key: "stable-cache-key" }); }); + it("does not retry the ChatGPT transport when maxRetries is zero", async () => { + const jwt = createJwt({ "https://api.openai.com/auth": { chatgpt_account_id: "acct" } }); + const fetchMock = vi.fn().mockResolvedValue( + new Response("rate limited", { + status: 429, + headers: { "retry-after": "1" }, + }), + ); + vi.stubGlobal("fetch", fetchMock); + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + + const result = await streamOpenAICodexResponses(model, context, { + apiKey: jwt, + maxRetries: 0, + transport: "sse", + }).result(); + + expect(result.stopReason).toBe("error"); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(setTimeoutSpy).not.toHaveBeenCalled(); + }); + it.each([ "1.5", "0x10", diff --git a/packages/ai/src/providers/openai-chatgpt-responses.ts b/packages/ai/src/providers/openai-chatgpt-responses.ts index e118866fbe58..2a3b486dd4c7 100644 --- a/packages/ai/src/providers/openai-chatgpt-responses.ts +++ b/packages/ai/src/providers/openai-chatgpt-responses.ts @@ -75,7 +75,7 @@ import { buildBaseOptions } from "./simple-options.js"; // ============================================================================ const DEFAULT_CODEX_BASE_URL = "https://chatgpt.com/backend-api"; -const MAX_RETRIES = 3; +const DEFAULT_MAX_RETRIES = 3; const BASE_DELAY_MS = 1000; const REQUEST_COMPRESSION_ZSTD_LEVEL = 3; const CODEX_TOOL_CALL_PROVIDERS = new Set(["openai", "opencode"]); @@ -271,14 +271,9 @@ export const streamOpenAICodexResponses: StreamFunction< // per request, which forfeits session-affinity routing on the WS transport (the // backend routes by session_id/x-client-request-id). Left as-is for this fix; // see the SSE-path session_id addition in buildOpenAIClientHeaders (agents/openai-transport-stream.ts). - const websocketRequestId = options?.sessionId || createCodexRequestId(); - const sseHeaders = buildSSEHeaders( - modelHeaders, - optionHeaders, - accountId, - apiKey, - options?.sessionId, - ); + const sessionId = clampOpenAIPromptCacheKey(options?.sessionId); + const websocketRequestId = sessionId || createCodexRequestId(); + const sseHeaders = buildSSEHeaders(modelHeaders, optionHeaders, accountId, apiKey, sessionId); const websocketHeaders = buildWebSocketHeaders( modelHeaders, optionHeaders, @@ -371,8 +366,9 @@ export const streamOpenAICodexResponses: StreamFunction< // Fetch with retry logic for rate limits and transient errors let response: Response | undefined; let lastError: Error | undefined; + const maxRetries = options?.maxRetries ?? DEFAULT_MAX_RETRIES; - for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) { + for (let attempt = 0; attempt <= maxRetries; attempt++) { if (activeSignal?.aborted) { throw new Error("Request was aborted"); } @@ -394,7 +390,7 @@ export const streamOpenAICodexResponses: StreamFunction< } const errorText = await readChatGptResponsesErrorTextLimited(response); - if (attempt < MAX_RETRIES && isRetryableError(response.status, errorText)) { + if (attempt < maxRetries && isRetryableError(response.status, errorText)) { let delayMs = BASE_DELAY_MS * 2 ** attempt; const retryAfterMs = response.headers.get("retry-after-ms"); @@ -453,7 +449,7 @@ export const streamOpenAICodexResponses: StreamFunction< } lastError = error instanceof Error ? error : new Error(String(error)); // Network errors are retryable - if (attempt < MAX_RETRIES && !lastError.message.includes("usage limit")) { + if (attempt < maxRetries && !lastError.message.includes("usage limit")) { const delayMs = BASE_DELAY_MS * 2 ** attempt; await sleepWithAbort(delayMs, activeSignal); continue; diff --git a/packages/ai/src/providers/openai-responses-shared.test.ts b/packages/ai/src/providers/openai-responses-shared.test.ts index 5c11028272c5..4e69b0524090 100644 --- a/packages/ai/src/providers/openai-responses-shared.test.ts +++ b/packages/ai/src/providers/openai-responses-shared.test.ts @@ -1,5 +1,6 @@ // OpenAI Responses shared tests cover tool conversion and response item mapping. import type { + ResponseCreateParamsStreaming, ResponseStreamEvent, Tool as OpenAIResponsesTool, } from "openai/resources/responses/responses.js"; @@ -22,9 +23,9 @@ type ResponsesFunctionTool = Extract; type OpenAIResponsesStreamEvent = Parameters[0] extends AsyncIterable ? Event : never; -async function* streamResponsesEvents( - events: readonly OpenAIResponsesStreamEvent[], -): AsyncGenerator { +async function* streamResponsesEvents( + events: readonly T[], +): AsyncGenerator { for (const event of events) { yield event; } @@ -303,6 +304,20 @@ describe("Responses temperature support", () => { }); }); +describe("Responses output token limits", () => { + it.each([ + [1, 16], + [15, 16], + [16, 16], + [32, 32], + ])("normalizes maxTokens %i to %i", (maxTokens, expected) => { + const params = {} as ResponseCreateParamsStreaming; + applyCommonResponsesParams(params, nativeOpenAIModel, { messages: [] }, { maxTokens }); + + expect(params.max_output_tokens).toBe(expected); + }); +}); + describe("Responses reasoning effort", () => { it("omits unsupported default-off reasoning for GPT-5.6 Sol", () => { const params = {} as never; @@ -365,6 +380,21 @@ describe("convertResponsesMessages", () => { }); }); + it("uses the system role when a Responses-compatible provider opts out of developer", () => { + const compatModel = { + ...nativeOpenAIModel, + compat: { supportsDeveloperRole: false }, + } satisfies Model<"openai-responses">; + + const input = convertResponsesMessages( + compatModel, + { systemPrompt: "system", messages: [] }, + allowedToolCallProviders, + ); + + expect(input[0]).toMatchObject({ type: "message", role: "system" }); + }); + it("strips the internal cache boundary marker from the system prompt message", () => { const input = convertResponsesMessages( nativeOpenAIModel, @@ -899,6 +929,484 @@ describe("processResponsesStream", () => { } }); + it.each([ + [undefined, 0], + [2, 2], + ])("passes SDK maxRetries %s as %i", async (maxRetries, expected) => { + let requestMaxRetries: number | undefined; + const output = createAssistantOutput(); + const stream = new AssistantMessageEventStream(); + + await runResponsesStreamLifecycle({ + stream, + model: nativeOpenAIModel, + output, + options: maxRetries === undefined ? undefined : { maxRetries }, + createClient: () => ({ + responses: { + create: (_params, requestOptions) => { + requestMaxRetries = requestOptions.maxRetries; + return { + withResponse: async () => ({ + data: streamResponsesEvents([ + { + type: "response.completed", + sequence_number: 1, + response: { id: "resp_retry", status: "completed" }, + } as ResponseStreamEvent, + ]), + response: new Response(null, { status: 200 }), + }), + }; + }, + }, + }), + buildParams: () => ({ model: nativeOpenAIModel.id, input: [], stream: true }), + formatError: (error) => (error instanceof Error ? error.message : String(error)), + }); + + expect(requestMaxRetries).toBe(expected); + expect(output.stopReason).toBe("stop"); + }); + + it("keeps interleaved reasoning items bound to their output indices", async () => { + const output = createAssistantOutput(); + const { stream, events } = createCapturedAssistantMessageEventStream(); + + await processResponsesStream( + responseEvents([ + { + type: "response.output_item.added", + output_index: 0, + item: { type: "reasoning", id: "rs_first", summary: [] }, + }, + { + type: "response.output_item.added", + output_index: 1, + item: { type: "reasoning", id: "rs_second", summary: [] }, + }, + { type: "response.reasoning_text.delta", output_index: 0, delta: "first" }, + { type: "response.reasoning_text.delta", output_index: 1, delta: "second" }, + { + type: "response.output_item.done", + output_index: 1, + item: { + type: "reasoning", + id: "rs_second", + summary: [], + content: [{ type: "reasoning_text", text: "second" }], + encrypted_content: "cipher-second", + }, + }, + { + type: "response.output_item.done", + output_index: 0, + item: { + type: "reasoning", + id: "rs_first", + summary: [], + content: [{ type: "reasoning_text", text: "first" }], + encrypted_content: "cipher-first", + }, + }, + { type: "response.completed", response: { id: "resp_reasoning", status: "completed" } }, + ]), + output, + stream, + nativeOpenAIModel, + ); + + expect(output.content).toMatchObject([ + { type: "thinking", thinking: "first" }, + { type: "thinking", thinking: "second" }, + ]); + const signatures = output.content.map((block) => + block.type === "thinking" && block.thinkingSignature + ? JSON.parse(block.thinkingSignature) + : undefined, + ); + expect(signatures).toMatchObject([ + { id: "rs_first", encrypted_content: "cipher-first" }, + { id: "rs_second", encrypted_content: "cipher-second" }, + ]); + expect( + events.map((event) => [event.type, "contentIndex" in event ? event.contentIndex : undefined]), + ).toEqual([ + ["thinking_start", 0], + ["thinking_start", 1], + ["thinking_delta", 0], + ["thinking_delta", 1], + ["thinking_end", 1], + ["thinking_end", 0], + ]); + }); + + it("preserves reasoning while a tool call streams on another output index", async () => { + const output = createAssistantOutput(); + const { stream, events } = createCapturedAssistantMessageEventStream(); + + await processResponsesStream( + responseEvents([ + { + type: "response.output_item.added", + output_index: 0, + item: { type: "reasoning", id: "rs_tool", summary: [] }, + }, + { type: "response.reasoning_text.delta", output_index: 0, delta: "before " }, + { + type: "response.output_item.added", + output_index: 1, + item: { + type: "function_call", + id: "fc_read", + call_id: "call_read", + name: "read", + arguments: "", + }, + }, + { + type: "response.function_call_arguments.delta", + output_index: 1, + item_id: "fc_read", + delta: '{"path":"README.md"}', + }, + { type: "response.reasoning_text.delta", output_index: 0, delta: "after" }, + { + type: "response.output_item.done", + output_index: 1, + item: { + type: "function_call", + id: "fc_read", + call_id: "call_read", + name: "read", + arguments: '{"path":"README.md"}', + }, + }, + { + type: "response.output_item.done", + output_index: 0, + item: { + type: "reasoning", + id: "rs_tool", + summary: [], + encrypted_content: "cipher-tool", + }, + }, + { type: "response.completed", response: { id: "resp_tool", status: "completed" } }, + ]), + output, + stream, + nativeOpenAIModel, + ); + + expect(output.content).toMatchObject([ + { type: "thinking", thinking: "before after" }, + { type: "toolCall", id: "call_read|fc_read", arguments: { path: "README.md" } }, + ]); + expect( + events + .filter((event) => event.type.endsWith("_delta")) + .map((event) => [event.type, "contentIndex" in event ? event.contentIndex : undefined]), + ).toEqual([ + ["thinking_delta", 0], + ["toolcall_delta", 1], + ["thinking_delta", 0], + ]); + }); + + it("backfills terminal encrypted reasoning for stateless replay", async () => { + const output = createAssistantOutput(); + + await processResponsesStream( + responseEvents([ + { + type: "response.output_item.added", + output_index: 0, + item: { type: "reasoning", id: "rs_backfill", summary: [] }, + }, + { + type: "response.output_item.done", + output_index: 0, + item: { type: "reasoning", id: "rs_backfill", summary: [] }, + }, + { + type: "response.completed", + response: { + id: "resp_backfill", + status: "completed", + output: [ + { + type: "reasoning", + id: "rs_backfill", + summary: [], + encrypted_content: "cipher-from-terminal", + }, + ], + }, + }, + ]), + output, + new AssistantMessageEventStream(), + nativeOpenAIModel, + ); + + const replay = convertResponsesMessages( + nativeOpenAIModel, + { + messages: [ + { role: "user", content: "first", timestamp: 1 }, + output, + { role: "user", content: "again", timestamp: 2 }, + ], + }, + testAllowedToolCallProviders, + ); + expect(replay.find((item) => item.type === "reasoning")).toMatchObject({ + id: "rs_backfill", + encrypted_content: "cipher-from-terminal", + }); + }); + + it("tolerates a completed message item with null content", async () => { + const output = createAssistantOutput(); + + await processResponsesStream( + responseEvents([ + { + type: "response.output_item.added", + output_index: 0, + item: { + type: "message", + id: "msg_null", + content: [{ type: "output_text", text: "" }], + }, + }, + { type: "response.output_text.delta", output_index: 0, delta: "streamed" }, + { + type: "response.output_item.done", + output_index: 0, + item: { type: "message", id: "msg_null", content: null }, + }, + { type: "response.completed", response: { id: "resp_null", status: "completed" } }, + ]), + output, + new AssistantMessageEventStream(), + nativeOpenAIModel, + ); + + expect(output.content).toMatchObject([{ type: "text", text: "streamed" }]); + }); + + it("keeps deferred text before an interleaved output item", async () => { + const output = createAssistantOutput(); + const { stream, events } = createCapturedAssistantMessageEventStream(); + + await processResponsesStream( + responseEvents([ + { + type: "response.output_item.added", + output_index: 0, + item: { + type: "message", + id: "msg_first", + content: [{ type: "output_text", text: "" }], + }, + }, + { + type: "response.output_item.done", + output_index: 0, + item: { + type: "message", + id: "msg_first", + content: [{ type: "output_text", text: "first" }], + }, + }, + { + type: "response.output_item.added", + output_index: 1, + item: { + type: "message", + id: "msg_second", + content: [{ type: "output_text", text: "" }], + }, + }, + { type: "response.output_text.delta", output_index: 1, delta: "first extended" }, + { + type: "response.output_item.added", + output_index: 2, + item: { type: "reasoning", id: "rs_after", summary: [] }, + }, + { + type: "response.output_item.done", + output_index: 2, + item: { + type: "reasoning", + id: "rs_after", + summary: [], + content: [{ type: "reasoning_text", text: "thought" }], + }, + }, + { + type: "response.output_item.done", + output_index: 1, + item: { + type: "message", + id: "msg_second", + content: [{ type: "output_text", text: "first extended" }], + }, + }, + { type: "response.completed", response: { id: "resp_order", status: "completed" } }, + ]), + output, + stream, + nativeOpenAIModel, + ); + + expect(output.content).toMatchObject([ + { type: "text", text: "first" }, + { type: "text", text: "first extended" }, + { type: "thinking", thinking: "thought" }, + ]); + expect( + events + .filter((event) => event.type.endsWith("_start")) + .map((event) => [event.type, "contentIndex" in event ? event.contentIndex : undefined]), + ).toEqual([ + ["text_start", 0], + ["text_start", 1], + ["thinking_start", 2], + ]); + }); + + it("finalizes incomplete terminal events with usage", async () => { + const output = createAssistantOutput(); + + await processResponsesStream( + responseEvents([ + { + type: "response.incomplete", + response: { + id: "resp_incomplete", + status: "incomplete", + usage: { + input_tokens: 30, + output_tokens: 12, + total_tokens: 42, + input_tokens_details: { cached_tokens: 5, cache_write_tokens: 3 }, + }, + }, + }, + ]), + output, + new AssistantMessageEventStream(), + nativeOpenAIModel, + ); + + expect(output.stopReason).toBe("length"); + expect(output.usage).toMatchObject({ + input: 22, + output: 12, + cacheRead: 5, + cacheWrite: 3, + totalTokens: 42, + }); + }); + + it("reports content-filtered incomplete responses as errors", async () => { + const output = createAssistantOutput(); + + await processResponsesStream( + responseEvents([ + { + type: "response.incomplete", + response: { + id: "resp_filtered", + status: "incomplete", + incomplete_details: { reason: "content_filter" }, + }, + }, + ]), + output, + new AssistantMessageEventStream(), + nativeOpenAIModel, + ); + + expect(output.stopReason).toBe("error"); + expect(output.errorMessage).toBe("Provider incomplete_reason: content_filter"); + + const lifecycleOutput = createAssistantOutput(); + await runResponsesStreamLifecycle({ + stream: new AssistantMessageEventStream(), + model: nativeOpenAIModel, + output: lifecycleOutput, + createClient: () => ({ + responses: { + create: () => ({ + withResponse: async () => ({ + data: streamResponsesEvents([ + { + type: "response.incomplete", + response: { + id: "resp_filtered_lifecycle", + status: "incomplete", + incomplete_details: { reason: "content_filter" }, + }, + } as ResponseStreamEvent, + ]), + response: new Response(null, { status: 200 }), + }), + }), + }, + }), + buildParams: () => ({ model: nativeOpenAIModel.id, input: [], stream: true }), + formatError: (error) => (error instanceof Error ? error.message : String(error)), + }); + + expect(lifecycleOutput.stopReason).toBe("error"); + expect(lifecycleOutput.errorMessage).toBe("Provider incomplete_reason: content_filter"); + }); + + it("preserves failed terminal response details", async () => { + const output = createAssistantOutput(); + + await processResponsesStream( + responseEvents([ + { + type: "response.failed", + response: { + id: "resp_failed", + status: "failed", + error: { code: "server_error", message: "provider failed" }, + }, + }, + ]), + output, + new AssistantMessageEventStream(), + nativeOpenAIModel, + ); + + expect(output).toMatchObject({ + responseId: "resp_failed", + stopReason: "error", + errorMessage: "server_error: provider failed", + }); + }); + + it("rejects streams that end without a terminal response event", async () => { + const output = createAssistantOutput(); + output.usage.input = 7; + + await expect( + processResponsesStream( + responseEvents([{ type: "response.created", response: { id: "resp_truncated" } }]), + output, + new AssistantMessageEventStream(), + nativeOpenAIModel, + ), + ).rejects.toThrow("OpenAI Responses stream ended before a terminal response event"); + expect(output.usage.input).toBe(7); + }); + it.each([ ["omits arguments", undefined], ["sends empty arguments", ""], @@ -990,6 +1498,7 @@ describe("processResponsesStream", () => { type: "response.output_item.done", item: { type: "function_call", name: "computer", arguments: "{}" }, }, + { type: "response.completed", response: { id: "resp_idless", status: "completed" } }, ]), output, stream, @@ -1037,6 +1546,11 @@ describe("processResponsesStream", () => { status: "completed", }, }, + { + type: "response.completed", + sequence_number: 3, + response: { id: "resp_without_item_id", status: "completed" }, + } as ResponseStreamEvent, ]; const output = createAssistantOutput(); @@ -1084,6 +1598,11 @@ describe("processResponsesStream", () => { status: "completed", }, }, + { + type: "response.completed", + sequence_number: 3, + response: { id: "resp_weather", status: "completed" }, + } as ResponseStreamEvent, ]; const output = createAssistantOutput(); const { stream, events } = createCapturedAssistantMessageEventStream(); @@ -1221,6 +1740,11 @@ describe("processResponsesStream", () => { status: "completed", }, }, + { + type: "response.completed", + sequence_number: 7, + response: { id: "resp_interleaved", status: "completed" }, + } as ResponseStreamEvent, ]; const output = createAssistantOutput(); const { stream, events } = createCapturedAssistantMessageEventStream(); @@ -1740,6 +2264,7 @@ describe("processResponsesStream", () => { arguments: '{"slot":1}', }, }, + { type: "response.completed", response: { id: "resp_suffix", status: "completed" } }, ]), output, stream, @@ -1811,6 +2336,10 @@ describe("processResponsesStream", () => { arguments: '{"slot":0}', }, }, + { + type: "response.completed", + response: { id: "resp_omitted_completions", status: "completed" }, + }, ]), output, stream, @@ -1867,6 +2396,10 @@ describe("processResponsesStream", () => { arguments: '{"slot":0}', }, }, + { + type: "response.completed", + response: { id: "resp_identity_mismatch", status: "completed" }, + }, ]), output, stream, @@ -1931,6 +2464,7 @@ describe("processResponsesStream", () => { arguments: '{"slot":1}', }, }, + { type: "response.completed", response: { id: "resp_sequential", status: "completed" } }, ]), output, stream, @@ -2013,6 +2547,7 @@ describe("processResponsesStream", () => { status: "completed", }, }, + { type: "response.completed", response: { id: "resp_item_only", status: "completed" } }, ]), output, stream, diff --git a/packages/ai/src/providers/openai-responses-shared.ts b/packages/ai/src/providers/openai-responses-shared.ts index be1a60ed413b..10cdc2ff164f 100644 --- a/packages/ai/src/providers/openai-responses-shared.ts +++ b/packages/ai/src/providers/openai-responses-shared.ts @@ -4,12 +4,12 @@ import type OpenAI from "openai"; import type { ResponseCreateParamsStreaming, ResponseFunctionCallOutputItemList, - ResponseFunctionToolCall, ResponseInput, ResponseInputItem, ResponseInputContent, ResponseInputImage, ResponseInputText, + ResponseOutputItem, ResponseOutputMessage, ResponseReasoningItem, ResponseStreamEvent, @@ -326,7 +326,9 @@ export function convertResponsesMessages( const includeSystemPrompt = options?.includeSystemPrompt ?? true; if (includeSystemPrompt && context.systemPrompt) { - const role = model.reasoning ? "developer" : "system"; + const compat = model.compat as { supportsDeveloperRole?: boolean } | undefined; + const role = + model.reasoning && compat?.supportsDeveloperRole !== false ? "developer" : "system"; messages.push({ type: "message", role, @@ -554,7 +556,7 @@ export function applyCommonResponsesParams( config?: { setDefaultReasoningOff?: boolean }, ): void { if (options?.maxTokens) { - params.max_output_tokens = options.maxTokens; + params.max_output_tokens = Math.max(options.maxTokens, 16); } if (options?.temperature !== undefined && supportsOpenAITemperature(model)) { @@ -596,7 +598,7 @@ function buildResponsesRequestOptions( return { ...(options?.signal ? { signal: options.signal } : {}), ...(options?.timeoutMs !== undefined ? { timeout: options.timeoutMs } : {}), - ...(options?.maxRetries !== undefined ? { maxRetries: options.maxRetries } : {}), + maxRetries: options?.maxRetries ?? 0, }; } @@ -665,7 +667,7 @@ export async function runResponsesStreamLifecycle(params: { } if (output.stopReason === "aborted" || output.stopReason === "error") { - throw new Error("An unknown error occurred"); + throw new Error(output.errorMessage ?? "An unknown error occurred"); } stream.push({ type: "done", reason: output.stopReason, message: output }); @@ -692,30 +694,101 @@ export async function processResponsesStream( model: Model, options?: OpenAIResponsesProcessStreamOptions, ): Promise { - let currentItem: - | ResponseReasoningItem - | ResponsesStreamOutputMessage - | ResponseFunctionToolCall - | null = null; - let currentBlock: ThinkingContent | TextContent | (ToolCall & { partialJson: string }) | null = - null; type StreamingToolCallBlock = ToolCall & { partialJson: string }; type StreamingToolCallState = ResponsesToolCallState & { block: StreamingToolCallBlock; contentIndex: number; }; - const streamingToolCalls = createResponsesToolCallTracker(); - let lastTextBlock: { + type TextBlockReference = { block: TextContent; index: number; phase: TextSignatureV1["phase"] | undefined; - } | null = null; - // While a message item may still be a cumulative snapshot of lastTextBlock, - // its public block is deferred so a collapsed item never leaves an - // unbalanced text_start behind (#91959). null = no deferral in progress. - let pendingMessageText: string | null = null; + }; + type ResponsesOutputSlot = + | { + type: "thinking"; + item: ResponseReasoningItem; + block: ThinkingContent; + contentIndex: number; + } + | { + type: "text"; + item: ResponsesStreamOutputMessage; + block: TextContent | null; + contentIndex: number | undefined; + pendingText: string | null; + collapseCandidate: TextBlockReference | null; + } + | { type: "toolCall"; toolCall: StreamingToolCallState }; + const streamingToolCalls = createResponsesToolCallTracker(); + const outputSlots = new Map(); + const reasoningBlocksById = new Map(); + let unindexedOutputSlot: ResponsesOutputSlot | undefined; + let terminalResponseEvent: "finalized" | "failed" | undefined; + let lastTextBlock: TextBlockReference | null = null; const blocks = output.content; const blockIndex = () => blocks.length - 1; + const readOutputIndex = (event: object): number | undefined => { + const outputIndex = (event as { output_index?: unknown }).output_index; + return typeof outputIndex === "number" && Number.isInteger(outputIndex) && outputIndex >= 0 + ? outputIndex + : undefined; + }; + const registerOutputSlot = (event: object, slot: ResponsesOutputSlot): void => { + const outputIndex = readOutputIndex(event); + if (outputIndex === undefined) { + if (unindexedOutputSlot) { + throw new Error("Responses stream added overlapping unindexed output items"); + } + unindexedOutputSlot = slot; + return; + } + if (outputSlots.has(outputIndex)) { + throw new Error(`Responses stream reused active output index ${outputIndex}`); + } + outputSlots.set(outputIndex, slot); + }; + const resolveOutputSlot = ( + event: object, + type: TType, + ): Extract | undefined => { + const outputIndex = readOutputIndex(event); + let slot = outputIndex === undefined ? unindexedOutputSlot : outputSlots.get(outputIndex); + if (outputIndex === undefined && !slot) { + const matchingSlots = [...outputSlots.values()].filter( + (candidate) => candidate.type === type, + ); + slot = matchingSlots.length === 1 ? matchingSlots[0] : undefined; + } + return slot?.type === type + ? (slot as Extract) + : undefined; + }; + const forgetOutputSlot = (event: object, slot: ResponsesOutputSlot): void => { + const outputIndex = readOutputIndex(event); + if (outputIndex === undefined) { + if (unindexedOutputSlot === slot) { + unindexedOutputSlot = undefined; + } else { + for (const [indexedOutput, indexedSlot] of outputSlots) { + if (indexedSlot === slot) { + outputSlots.delete(indexedOutput); + } + } + } + return; + } + if (outputSlots.get(outputIndex) === slot) { + outputSlots.delete(outputIndex); + } + }; + const forgetToolCallOutputSlot = (toolCall: StreamingToolCallState): void => { + for (const [outputIndex, slot] of outputSlots) { + if (slot.type === "toolCall" && slot.toolCall === toolCall) { + outputSlots.delete(outputIndex); + } + } + }; const readIdentityValue = (value: unknown): string | undefined => { const identity = typeof value === "string" ? value.trim() : ""; return identity || undefined; @@ -739,30 +812,194 @@ export async function processResponsesStream( } return name; }; - const appendPendingMessageDelta = (delta: string) => { - pendingMessageText = `${pendingMessageText ?? ""}${delta}`; - const priorText = lastTextBlock?.block.text ?? ""; - if (priorText.startsWith(pendingMessageText) || pendingMessageText.startsWith(priorText)) { + const createOutputSlot = ( + event: object, + item: ResponseOutputItem | ResponsesStreamOutputMessage, + ): ResponsesOutputSlot | undefined => { + if (item.type === "reasoning") { + const block: ThinkingContent = { type: "thinking", thinking: "" }; + const slot = { + type: "thinking", + item, + block, + contentIndex: blocks.length, + } satisfies ResponsesOutputSlot; + blocks.push(block); + registerOutputSlot(event, slot); + stream.push({ type: "thinking_start", contentIndex: slot.contentIndex, partial: output }); + return slot; + } + if (item.type === "message") { + const messageItem = item as ResponsesStreamOutputMessage; + const collapseCandidate = lastTextBlock; + const block: TextContent | null = collapseCandidate + ? null + : { + type: "text", + text: "", + ...(messageItem.phase + ? { textSignature: encodeTextSignatureV1(messageItem.id, messageItem.phase) } + : {}), + }; + const slot = { + type: "text", + item: messageItem, + block, + contentIndex: block ? blocks.length : undefined, + pendingText: collapseCandidate ? "" : null, + collapseCandidate, + } satisfies ResponsesOutputSlot; + if (block) { + blocks.push(block); + } + registerOutputSlot(event, slot); + if (slot.contentIndex !== undefined) { + stream.push({ type: "text_start", contentIndex: slot.contentIndex, partial: output }); + } + return slot; + } + return undefined; + }; + const resolveOutputItemSlot = ( + event: object, + item: ResponseOutputItem | ResponsesStreamOutputMessage, + ): ResponsesOutputSlot | undefined => { + if (item.type === "reasoning") { + return resolveOutputSlot(event, "thinking"); + } + if (item.type === "message") { + return resolveOutputSlot(event, "text"); + } + const outputIndex = readOutputIndex(event); + return outputIndex === undefined ? undefined : outputSlots.get(outputIndex); + }; + const getOrCreateOutputSlot = ( + event: object, + item: ResponseOutputItem | ResponsesStreamOutputMessage, + ): ResponsesOutputSlot | undefined => { + return resolveOutputItemSlot(event, item) ?? createOutputSlot(event, item); + }; + const materializeDeferredTextSlot = ( + slot: Extract, + ): void => { + if (slot.block || slot.pendingText === null) { + return; + } + const text = slot.pendingText; + slot.block = { + type: "text", + text, + ...(slot.item.phase + ? { textSignature: encodeTextSignatureV1(slot.item.id, slot.item.phase) } + : {}), + }; + blocks.push(slot.block); + slot.contentIndex = blockIndex(); + stream.push({ type: "text_start", contentIndex: slot.contentIndex, partial: output }); + if (text) { + stream.push({ + type: "text_delta", + contentIndex: slot.contentIndex, + delta: text, + partial: output, + }); + } + if (lastTextBlock === slot.collapseCandidate) { + lastTextBlock = null; + } + slot.pendingText = null; + slot.collapseCandidate = null; + }; + const materializeDeferredTextSlots = (except?: ResponsesOutputSlot): void => { + for (const slot of outputSlots.values()) { + if (slot !== except && slot.type === "text") { + materializeDeferredTextSlot(slot); + } + } + if (unindexedOutputSlot !== except && unindexedOutputSlot?.type === "text") { + materializeDeferredTextSlot(unindexedOutputSlot); + } + }; + const appendPendingMessageDelta = ( + slot: Extract, + delta: string, + ) => { + slot.pendingText = `${slot.pendingText ?? ""}${delta}`; + const priorText = slot.collapseCandidate?.block.text ?? ""; + if (priorText.startsWith(slot.pendingText) || slot.pendingText.startsWith(priorText)) { return; } // Diverged from the prior text: this is a distinct message, so open its // block now and replay the withheld text as one delta. - currentBlock = { - type: "text", - text: pendingMessageText, - ...(currentItem?.type === "message" && currentItem.phase - ? { textSignature: encodeTextSignatureV1(currentItem.id, currentItem.phase ?? undefined) } - : {}), - }; - blocks.push(currentBlock); - stream.push({ type: "text_start", contentIndex: blockIndex(), partial: output }); - stream.push({ - type: "text_delta", - contentIndex: blockIndex(), - delta: pendingMessageText, - partial: output, - }); - pendingMessageText = null; + materializeDeferredTextSlot(slot); + }; + const backfillReasoningSignatures = (responseOutput: ResponseOutputItem[]): void => { + for (const item of responseOutput) { + if (item.type !== "reasoning" || !item.encrypted_content) { + continue; + } + const block = reasoningBlocksById.get(item.id); + if (!block?.thinkingSignature) { + continue; + } + + const storedItem = JSON.parse(block.thinkingSignature) as ResponseReasoningItem; + if (storedItem.encrypted_content) { + continue; + } + block.thinkingSignature = JSON.stringify({ + ...storedItem, + encrypted_content: item.encrypted_content, + }); + } + }; + const finalizeResponse = ( + response: Extract< + ResponseStreamEvent, + { type: "response.completed" | "response.incomplete" } + >["response"], + ): void => { + terminalResponseEvent = "finalized"; + backfillReasoningSignatures(response.output ?? []); + if (response.id) { + output.responseId = response.id; + } + if (response.usage) { + const inputTokenDetails = response.usage.input_tokens_details as + | ResponsesInputTokensDetails + | null + | undefined; + const cachedTokens = inputTokenDetails?.cached_tokens || 0; + const cacheWriteTokens = inputTokenDetails?.cache_write_tokens || 0; + output.usage = { + // OpenAI includes cache reads and writes in input_tokens, so split both priced buckets. + input: Math.max(0, (response.usage.input_tokens || 0) - cachedTokens - cacheWriteTokens), + output: response.usage.output_tokens || 0, + cacheRead: cachedTokens, + cacheWrite: cacheWriteTokens, + totalTokens: response.usage.total_tokens || 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }; + } + calculateCost(model, output.usage); + if (options?.applyServiceTierPricing) { + const serviceTier = options.resolveServiceTier + ? options.resolveServiceTier(response.service_tier, options.serviceTier) + : (response.service_tier ?? options.serviceTier); + options.applyServiceTierPricing(output.usage, serviceTier); + } + if ( + response.status === "incomplete" && + response.incomplete_details?.reason === "content_filter" + ) { + output.stopReason = "error"; + output.errorMessage = "Provider incomplete_reason: content_filter"; + } else { + output.stopReason = mapStopReason(response.status); + } + if (output.content.some((block) => block.type === "toolCall") && output.stopReason === "stop") { + output.stopReason = "toolUse"; + } }; const guardedStream = withFirstStreamEventTimeout(openaiStream, { @@ -779,32 +1016,15 @@ export async function processResponsesStream( if (event.type === "response.created") { output.responseId = event.response.id; } else if (event.type === "response.output_item.added") { + materializeDeferredTextSlots(); const item = event.item; if (item.type !== "message") { // Snapshot collapse only applies to back-to-back message items; any // other item is a real boundary (see resolveResponsesMessageSnapshotCollapse). lastTextBlock = null; - pendingMessageText = null; } - if (item.type === "reasoning") { - currentItem = item; - currentBlock = { type: "thinking", thinking: "" }; - output.content.push(currentBlock); - stream.push({ type: "thinking_start", contentIndex: blockIndex(), partial: output }); - } else if (item.type === "message") { - currentItem = item; - if (lastTextBlock) { - currentBlock = null; - pendingMessageText = ""; - } else { - currentBlock = { - type: "text", - text: "", - ...(item.phase ? { textSignature: encodeTextSignatureV1(item.id, item.phase) } : {}), - }; - output.content.push(currentBlock); - stream.push({ type: "text_start", contentIndex: blockIndex(), partial: output }); - } + if (item.type === "reasoning" || item.type === "message") { + createOutputSlot(event, item); } else if (item.type === "function_call") { const toolCallBlock: StreamingToolCallBlock = { type: "toolCall", @@ -821,129 +1041,144 @@ export async function processResponsesStream( ...readResponsesToolCallItemIdentity(item), }; streamingToolCalls.register(event, toolCallState); - currentItem = item; - currentBlock = toolCallBlock; + if (readOutputIndex(event) !== undefined) { + registerOutputSlot(event, { type: "toolCall", toolCall: toolCallState }); + } output.content.push(toolCallBlock); stream.push({ type: "toolcall_start", contentIndex, partial: output }); } } else if (event.type === "response.reasoning_summary_part.added") { - if (currentItem && currentItem.type === "reasoning") { - currentItem.summary = currentItem.summary || []; - currentItem.summary.push(event.part); + const slot = resolveOutputSlot(event, "thinking"); + if (!slot) { + continue; } + slot.item.summary = slot.item.summary || []; + slot.item.summary.push(event.part); } else if (event.type === "response.reasoning_summary_text.delta") { - if (currentItem?.type === "reasoning" && currentBlock?.type === "thinking") { - currentItem.summary = currentItem.summary || []; - const lastPart = currentItem.summary[currentItem.summary.length - 1]; - if (lastPart) { - currentBlock.thinking += event.delta; - lastPart.text += event.delta; - stream.push({ - type: "thinking_delta", - contentIndex: blockIndex(), - delta: event.delta, - partial: output, - }); - } + const slot = resolveOutputSlot(event, "thinking"); + if (!slot) { + continue; } + slot.item.summary = slot.item.summary || []; + const lastPart = slot.item.summary[slot.item.summary.length - 1]; + if (!lastPart) { + continue; + } + slot.block.thinking += event.delta; + lastPart.text += event.delta; + stream.push({ + type: "thinking_delta", + contentIndex: slot.contentIndex, + delta: event.delta, + partial: output, + }); } else if (event.type === "response.reasoning_summary_part.done") { - if (currentItem?.type === "reasoning" && currentBlock?.type === "thinking") { - currentItem.summary = currentItem.summary || []; - const lastPart = currentItem.summary[currentItem.summary.length - 1]; - if (lastPart) { - currentBlock.thinking += "\n\n"; - lastPart.text += "\n\n"; - stream.push({ - type: "thinking_delta", - contentIndex: blockIndex(), - delta: "\n\n", - partial: output, - }); - } + const slot = resolveOutputSlot(event, "thinking"); + if (!slot) { + continue; } + slot.item.summary = slot.item.summary || []; + const lastPart = slot.item.summary[slot.item.summary.length - 1]; + if (!lastPart) { + continue; + } + slot.block.thinking += "\n\n"; + lastPart.text += "\n\n"; + stream.push({ + type: "thinking_delta", + contentIndex: slot.contentIndex, + delta: "\n\n", + partial: output, + }); } else if (event.type === "response.reasoning_text.delta") { - if (currentItem?.type === "reasoning" && currentBlock?.type === "thinking") { - currentBlock.thinking += event.delta; + const slot = resolveOutputSlot(event, "thinking"); + if (!slot) { + continue; + } + slot.block.thinking += event.delta; + stream.push({ + type: "thinking_delta", + contentIndex: slot.contentIndex, + delta: event.delta, + partial: output, + }); + } else if (event.type === "response.content_part.added") { + const slot = resolveOutputSlot(event, "text"); + if (!slot) { + continue; + } + slot.item.content = slot.item.content || []; + if ( + event.part.type === OPENAI_RESPONSES_OUTPUT_TEXT_CONTENT_PART_TYPE || + event.part.type === AZURE_RESPONSES_TEXT_CONTENT_PART_TYPE || + event.part.type === "refusal" + ) { + slot.item.content.push(event.part); + } + } else if (event.type === "response.output_text.delta") { + const slot = resolveOutputSlot(event, "text"); + if (!slot?.item.content || slot.item.content.length === 0) { + continue; + } + const lastPart = slot.item.content[slot.item.content.length - 1]; + if (!isResponsesTextContentPartType(lastPart?.type)) { + continue; + } + lastPart.text += event.delta; + if (slot.pendingText !== null) { + appendPendingMessageDelta(slot, event.delta); + } else if (slot.block && slot.contentIndex !== undefined) { + slot.block.text += event.delta; stream.push({ - type: "thinking_delta", - contentIndex: blockIndex(), + type: "text_delta", + contentIndex: slot.contentIndex, delta: event.delta, partial: output, }); } - } else if (event.type === "response.content_part.added") { - if (currentItem?.type === "message") { - currentItem.content = currentItem.content || []; - if ( - event.part.type === OPENAI_RESPONSES_OUTPUT_TEXT_CONTENT_PART_TYPE || - event.part.type === AZURE_RESPONSES_TEXT_CONTENT_PART_TYPE || - event.part.type === "refusal" - ) { - currentItem.content.push(event.part); - } - } - } else if (event.type === "response.output_text.delta") { - if (currentItem?.type === "message") { - if (!currentItem.content || currentItem.content.length === 0) { - continue; - } - const lastPart = currentItem.content[currentItem.content.length - 1]; - if (isResponsesTextContentPartType(lastPart?.type)) { - lastPart.text += event.delta; - if (pendingMessageText !== null) { - appendPendingMessageDelta(event.delta); - } else if (currentBlock?.type === "text") { - currentBlock.text += event.delta; - stream.push({ - type: "text_delta", - contentIndex: blockIndex(), - delta: event.delta, - partial: output, - }); - } - } - } } else if (isAzureResponsesTextDeltaEvent(event)) { - if (currentItem?.type === "message") { - currentItem.content = currentItem.content || []; - let lastPart = currentItem.content[currentItem.content.length - 1]; - if (lastPart?.type !== "text") { - lastPart = { type: "text", text: "" }; - currentItem.content.push(lastPart); - } - lastPart.text += event.delta; - if (pendingMessageText !== null) { - appendPendingMessageDelta(event.delta); - } else if (currentBlock?.type === "text") { - currentBlock.text += event.delta; - stream.push({ - type: "text_delta", - contentIndex: blockIndex(), - delta: event.delta, - partial: output, - }); - } + const slot = resolveOutputSlot(event, "text"); + if (!slot) { + continue; + } + slot.item.content = slot.item.content || []; + let lastPart = slot.item.content[slot.item.content.length - 1]; + if (lastPart?.type !== "text") { + lastPart = { type: "text", text: "" }; + slot.item.content.push(lastPart); + } + lastPart.text += event.delta; + if (slot.pendingText !== null) { + appendPendingMessageDelta(slot, event.delta); + } else if (slot.block && slot.contentIndex !== undefined) { + slot.block.text += event.delta; + stream.push({ + type: "text_delta", + contentIndex: slot.contentIndex, + delta: event.delta, + partial: output, + }); } } else if (event.type === "response.refusal.delta") { - if (currentItem?.type === "message") { - if (!currentItem.content || currentItem.content.length === 0) { - continue; - } - const lastPart = currentItem.content[currentItem.content.length - 1]; - if (lastPart?.type === "refusal") { - lastPart.refusal += event.delta; - if (pendingMessageText !== null) { - appendPendingMessageDelta(event.delta); - } else if (currentBlock?.type === "text") { - currentBlock.text += event.delta; - stream.push({ - type: "text_delta", - contentIndex: blockIndex(), - delta: event.delta, - partial: output, - }); - } - } + const slot = resolveOutputSlot(event, "text"); + if (!slot?.item.content || slot.item.content.length === 0) { + continue; + } + const lastPart = slot.item.content[slot.item.content.length - 1]; + if (lastPart?.type !== "refusal") { + continue; + } + lastPart.refusal += event.delta; + if (slot.pendingText !== null) { + appendPendingMessageDelta(slot, event.delta); + } else if (slot.block && slot.contentIndex !== undefined) { + slot.block.text += event.delta; + stream.push({ + type: "text_delta", + contentIndex: slot.contentIndex, + delta: event.delta, + partial: output, + }); } } else if (event.type === "response.function_call_arguments.delta") { const toolCall = streamingToolCalls.resolve(event); @@ -992,78 +1227,98 @@ export async function processResponsesStream( const item = event.item; if (item.type !== "message") { lastTextBlock = null; - pendingMessageText = null; } - if (item.type === "reasoning" && currentBlock?.type === "thinking") { + const existingOutputSlot = resolveOutputItemSlot(event, item); + materializeDeferredTextSlots(existingOutputSlot); + const outputSlot = existingOutputSlot ?? getOrCreateOutputSlot(event, item); + if (item.type === "reasoning" && outputSlot?.type === "thinking") { const summaryText = item.summary?.map((s) => s.text).join("\n\n") || ""; const contentText = item.content?.map((c) => c.text).join("\n\n") || ""; - currentBlock.thinking = summaryText || contentText || currentBlock.thinking; - currentBlock.thinkingSignature = JSON.stringify(item); + outputSlot.block.thinking = summaryText || contentText || outputSlot.block.thinking; + outputSlot.block.thinkingSignature = JSON.stringify(item); + if (typeof item.id === "string") { + reasoningBlocksById.set(item.id, outputSlot.block); + } stream.push({ type: "thinking_end", - contentIndex: blockIndex(), - content: currentBlock.thinking, + contentIndex: outputSlot.contentIndex, + content: outputSlot.block.thinking, partial: output, }); - currentBlock = null; + forgetOutputSlot(event, outputSlot); } else if ( item.type === "message" && - (currentBlock?.type === "text" || pendingMessageText !== null) + outputSlot?.type === "text" && + (outputSlot.block || outputSlot.pendingText !== null) ) { // Support both OpenAI "output_text" and Azure "text" content types - const finalText = item.content - .map((c) => (c.type === "output_text" || c.type === "text" ? c.text : c.refusal)) - .join(""); + const streamedText = outputSlot.pendingText ?? outputSlot.block?.text ?? ""; + const finalText = + item.content == null + ? streamedText + : item.content + .map((c) => (c.type === "output_text" || c.type === "text" ? c.text : c.refusal)) + .join(""); const phase = item.phase ?? undefined; const collapse = - pendingMessageText !== null + outputSlot.pendingText !== null ? resolveResponsesMessageSnapshotCollapse({ - prior: lastTextBlock && { - text: lastTextBlock.block.text, - phase: lastTextBlock.phase, + prior: outputSlot.collapseCandidate && { + text: outputSlot.collapseCandidate.block.text, + phase: outputSlot.collapseCandidate.phase, }, nextText: finalText, nextPhase: phase, }) : ({ kind: "keep" } as const); - pendingMessageText = null; - if (collapse.kind === "extend" && lastTextBlock) { + outputSlot.pendingText = null; + if (collapse.kind === "extend" && outputSlot.collapseCandidate) { // Cumulative snapshot of the prior message item: replace its text // instead of appending another copy. The deferred block was never // started publicly, and the newest item's signature is kept so // replay carries the item that produced this content (#91959). - lastTextBlock.block.text = collapse.text; - lastTextBlock.block.textSignature = encodeTextSignatureV1(item.id, phase); + outputSlot.collapseCandidate.block.text = collapse.text; + outputSlot.collapseCandidate.block.textSignature = encodeTextSignatureV1(item.id, phase); stream.push({ type: "text_end", - contentIndex: lastTextBlock.index, + contentIndex: outputSlot.collapseCandidate.index, content: collapse.text, partial: output, }); + lastTextBlock = outputSlot.collapseCandidate; } else { - if (currentBlock?.type !== "text") { + if (!outputSlot.block) { // Deferred distinct message: open its block now, balanced with the // text_end below. - currentBlock = { + outputSlot.block = { type: "text", text: "", ...(phase ? { textSignature: encodeTextSignatureV1(item.id, phase) } : {}), }; - blocks.push(currentBlock); - stream.push({ type: "text_start", contentIndex: blockIndex(), partial: output }); + blocks.push(outputSlot.block); + outputSlot.contentIndex = blockIndex(); + stream.push({ + type: "text_start", + contentIndex: outputSlot.contentIndex, + partial: output, + }); } - currentBlock.text = finalText; - currentBlock.textSignature = encodeTextSignatureV1(item.id, phase); - lastTextBlock = { block: currentBlock, index: blockIndex(), phase }; + outputSlot.block.text = finalText; + outputSlot.block.textSignature = encodeTextSignatureV1(item.id, phase); + const contentIndex = outputSlot.contentIndex; + if (contentIndex === undefined) { + throw new Error("Responses stream finalized text without a content index"); + } + lastTextBlock = { block: outputSlot.block, index: contentIndex, phase }; stream.push({ type: "text_end", - contentIndex: blockIndex(), - content: currentBlock.text, + contentIndex, + content: outputSlot.block.text, partial: output, }); } - currentBlock = null; + forgetOutputSlot(event, outputSlot); } else if (item.type === "function_call") { const streamingToolCall = streamingToolCalls.resolve( event, @@ -1117,10 +1372,7 @@ export async function processResponsesStream( if (streamingToolCall) { streamingToolCalls.forget(streamingToolCall); - } - if (currentBlock === toolCall) { - currentBlock = null; - currentItem = null; + forgetToolCallOutputSlot(streamingToolCall); } stream.push({ type: "toolcall_end", @@ -1129,43 +1381,11 @@ export async function processResponsesStream( partial: output, }); } - } else if (event.type === "response.completed") { + } else if (event.type === "response.completed" || event.type === "response.incomplete") { if (streamingToolCalls.hasActive()) { throw new Error("Responses stream completed with unresolved tool calls"); } - const response = event.response; - if (response?.id) { - output.responseId = response.id; - } - if (response?.usage) { - const inputTokenDetails = response.usage.input_tokens_details as - | ResponsesInputTokensDetails - | null - | undefined; - const cachedTokens = inputTokenDetails?.cached_tokens || 0; - const cacheWriteTokens = inputTokenDetails?.cache_write_tokens || 0; - output.usage = { - // OpenAI includes cache reads and writes in input_tokens, so split both priced buckets. - input: Math.max(0, (response.usage.input_tokens || 0) - cachedTokens - cacheWriteTokens), - output: response.usage.output_tokens || 0, - cacheRead: cachedTokens, - cacheWrite: cacheWriteTokens, - totalTokens: response.usage.total_tokens || 0, - cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, - }; - } - calculateCost(model, output.usage); - if (options?.applyServiceTierPricing) { - const serviceTier = options.resolveServiceTier - ? options.resolveServiceTier(response?.service_tier, options.serviceTier) - : (response?.service_tier ?? options.serviceTier); - options.applyServiceTierPricing(output.usage, serviceTier); - } - // Map status to stop reason - output.stopReason = mapStopReason(response?.status); - if (output.content.some((b) => b.type === "toolCall") && output.stopReason === "stop") { - output.stopReason = "toolUse"; - } + finalizeResponse(event.response); } else if (event.type === "error") { throw new Error( event.message ? `Error Code ${event.code}: ${event.message}` : "Unknown error", @@ -1173,17 +1393,26 @@ export async function processResponsesStream( } else if (event.type === "response.failed") { const error = event.response?.error; const details = event.response?.incomplete_details; - const msg = error + output.responseId = event.response.id; + output.stopReason = "error"; + output.errorMessage = error ? `${error.code || "unknown"}: ${error.message || "no message"}` : details?.reason ? `incomplete: ${details.reason}` : "Unknown error (no error details in response)"; - throw new Error(msg); + terminalResponseEvent = "failed"; + break; } } + if (terminalResponseEvent === "failed") { + return; + } if (streamingToolCalls.hasActive()) { throw new Error("Responses stream ended with unresolved tool calls"); } + if (!terminalResponseEvent) { + throw new Error("OpenAI Responses stream ended before a terminal response event"); + } } function mapStopReason(status: OpenAI.Responses.ResponseStatus | undefined): StopReason { diff --git a/packages/ai/src/providers/openai-responses.live.test.ts b/packages/ai/src/providers/openai-responses.live.test.ts new file mode 100644 index 000000000000..0a8f67e45885 --- /dev/null +++ b/packages/ai/src/providers/openai-responses.live.test.ts @@ -0,0 +1,116 @@ +import { Type } from "typebox"; +import { describe, expect, it } from "vitest"; +import type { Context, Model } from "../types.js"; +import { streamOpenAIResponses } from "./openai-responses.js"; + +// Live coverage for the Responses stream state machine: real streams interleave +// reasoning/message/tool items, so unit fakes alone cannot prove slot tracking, +// terminal-event handling, or the max_output_tokens floor against the real API. +const LIVE = process.env.OPENCLAW_LIVE_TEST === "1"; +const OPENAI_KEY = process.env.OPENAI_API_KEY ?? ""; +const describeLive = LIVE && OPENAI_KEY ? describe : describe.skip; + +const LIVE_MODEL_ID = process.env.OPENCLAW_LIVE_RESPONSES_MODEL || "gpt-5.6-luna"; +const LIVE_TIMEOUT_MS = 120_000; + +function liveModel(overrides: Partial> = {}) { + return { + id: LIVE_MODEL_ID, + name: LIVE_MODEL_ID, + api: "openai-responses", + provider: "openai", + baseUrl: "https://api.openai.com/v1", + reasoning: true, + input: ["text"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 200_000, + maxTokens: 8192, + ...overrides, + } satisfies Model<"openai-responses">; +} + +describeLive("OpenAI Responses live", () => { + it( + "streams a reply through a terminal event with usage accounting", + async () => { + const context: Context = { + messages: [{ role: "user", content: "Reply with exactly: OK", timestamp: 0 }], + }; + const result = await streamOpenAIResponses(liveModel(), context, { + apiKey: OPENAI_KEY, + maxTokens: 256, + }).result(); + + expect(result.errorMessage).toBeUndefined(); + expect(result.stopReason).toBe("stop"); + const text = result.content + .filter((block) => block.type === "text") + .map((block) => block.text) + .join(""); + expect(text).toContain("OK"); + expect(result.usage.totalTokens).toBeGreaterThan(0); + expect(result.usage.output).toBeGreaterThan(0); + }, + LIVE_TIMEOUT_MS, + ); + + it( + "clamps a sub-floor max token budget instead of failing the request", + async () => { + const context: Context = { + messages: [{ role: "user", content: "Reply with exactly: OK", timestamp: 0 }], + }; + const result = await streamOpenAIResponses(liveModel(), context, { + apiKey: OPENAI_KEY, + maxTokens: 1, + }).result(); + + // Pre-floor behavior was a hard 400 from the API; truncation is expected. + expect(result.errorMessage).toBeUndefined(); + expect(["stop", "length"]).toContain(result.stopReason); + }, + LIVE_TIMEOUT_MS, + ); + + it( + "keeps reasoning and tool-call items separable on a real interleaved stream", + async () => { + const context: Context = { + messages: [ + { + role: "user", + content: "Call the live_probe tool with value set to exactly LIVE_OK.", + timestamp: 0, + }, + ], + tools: [ + { + name: "live_probe", + description: "Records a probe value.", + parameters: Type.Object({ value: Type.String() }), + }, + ], + }; + const result = await streamOpenAIResponses(liveModel(), context, { + apiKey: OPENAI_KEY, + maxTokens: 1024, + reasoningEffort: "low", + }).result(); + + expect(result.errorMessage).toBeUndefined(); + expect(result.stopReason).toBe("toolUse"); + const toolCalls = result.content.filter((block) => block.type === "toolCall"); + expect(toolCalls).toHaveLength(1); + const probeCall = toolCalls[0]; + expect(probeCall?.name).toBe("live_probe"); + const probeArguments = (probeCall?.arguments ?? {}) as { value?: string }; + expect(probeArguments.value).toBe("LIVE_OK"); + for (const block of result.content) { + if (block.type === "thinking") { + expect(block.thinkingSignature).toBeTruthy(); + } + } + }, + LIVE_TIMEOUT_MS, + ); +}); diff --git a/packages/ai/src/providers/openai-responses.test.ts b/packages/ai/src/providers/openai-responses.test.ts index 620565adff52..50f0cec4f0c1 100644 --- a/packages/ai/src/providers/openai-responses.test.ts +++ b/packages/ai/src/providers/openai-responses.test.ts @@ -2,12 +2,18 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { configureAiTransportHost } from "../host.js"; import type { Context, Model } from "../types.js"; -const openAiMockState = vi.hoisted(() => ({ configs: [] as unknown[] })); +const openAiMockState = vi.hoisted(() => ({ + configs: [] as unknown[], + params: [] as unknown[], + requestOptions: [] as unknown[], +})); vi.mock("openai", () => ({ default: class MockOpenAI { responses = { - create: vi.fn(() => { + create: vi.fn((params: unknown, requestOptions: unknown) => { + openAiMockState.params.push(params); + openAiMockState.requestOptions.push(requestOptions); throw new Error("stop after constructor"); }), }; @@ -43,6 +49,8 @@ function model(overrides: Partial> = {}) { describe("OpenAI Responses provider", () => { afterEach(() => { openAiMockState.configs = []; + openAiMockState.params = []; + openAiMockState.requestOptions = []; configureAiTransportHost({}); }); @@ -83,4 +91,15 @@ describe("OpenAI Responses provider", () => { ); expect(config.fetch).toBe(hostFetch); }); + + it("clamps small output limits and disables implicit SDK retries", async () => { + const result = await streamOpenAIResponses(model(), context, { + apiKey: String(1), + maxTokens: 1, + }).result(); + + expect(result.stopReason).toBe("error"); + expect(openAiMockState.params[0]).toMatchObject({ max_output_tokens: 16, store: false }); + expect(openAiMockState.requestOptions[0]).toMatchObject({ maxRetries: 0 }); + }); }); diff --git a/packages/llm-core/src/types.ts b/packages/llm-core/src/types.ts index 0d421770add8..e11741beecd9 100644 --- a/packages/llm-core/src/types.ts +++ b/packages/llm-core/src/types.ts @@ -475,6 +475,8 @@ export interface OpenAICompletionsCompat { /** Compatibility settings for OpenAI Responses APIs. */ export interface OpenAIResponsesCompat { + /** Whether the provider supports the `developer` role (vs `system`). Default: true. */ + supportsDeveloperRole?: boolean; /** Whether the model accepts the `temperature` parameter. Default: true. */ supportsTemperature?: boolean; /** Whether to send the OpenAI `session_id` cache-affinity header from `options.sessionId` when caching is enabled. Default: true. */ diff --git a/src/infra/vitest-live-config.test.ts b/src/infra/vitest-live-config.test.ts index 3d44883bd8a7..0215a5efab3d 100644 --- a/src/infra/vitest-live-config.test.ts +++ b/src/infra/vitest-live-config.test.ts @@ -22,6 +22,7 @@ describe("live vitest config", () => { expect(liveConfig.test?.include).toEqual([ "src/**/*.live.test.ts", "test/**/*.live.test.ts", + "packages/*/src/**/*.live.test.ts", BUNDLED_PLUGIN_LIVE_TEST_GLOB, ]); expect(normalizeConfigPaths(liveConfig.test?.setupFiles)).toEqual([ diff --git a/test/vitest/vitest.live.config.ts b/test/vitest/vitest.live.config.ts index d8f515fbcb4b..2db9cd328e0f 100644 --- a/test/vitest/vitest.live.config.ts +++ b/test/vitest/vitest.live.config.ts @@ -27,7 +27,12 @@ export default defineConfig({ [...(baseTest.setupFiles ?? []), "test/setup-openclaw-runtime.ts"].map(resolveRepoRootPath), ), ], - include: ["src/**/*.live.test.ts", "test/**/*.live.test.ts", BUNDLED_PLUGIN_LIVE_TEST_GLOB], + include: [ + "src/**/*.live.test.ts", + "test/**/*.live.test.ts", + "packages/*/src/**/*.live.test.ts", + BUNDLED_PLUGIN_LIVE_TEST_GLOB, + ], exclude, }, });