From 694a45040f1cc339d326f2f08cf2e5d4d6fa83ad Mon Sep 17 00:00:00 2001 From: can1357 Date: Wed, 10 Jun 2026 01:26:30 +0200 Subject: [PATCH] fix(ai): fixed OpenAI provider stream assembly and gateway round-trips Mistral thinking-as-text replay no longer crashes on string content; encrypted_content survives inbound reasoning parse; codex completed-handler sweeps unfinished tool calls; missing content_part synthesized on first delta; interleaved content/tool_calls no longer fragment calls; Azure completions honor AZURE_OPENAI_DEPLOYMENT_NAME_MAP; chat gateway round-trips reasoning_content; wire call_id loses internal item-id suffix. --- .../src/providers/azure-openai-responses.ts | 4 +- .../providers/openai-chat-server-schema.ts | 5 ++ .../ai/src/providers/openai-chat-server.ts | 48 +++++++++++- .../src/providers/openai-codex-responses.ts | 32 +++++++- .../ai/src/providers/openai-completions.ts | 48 +++++++++--- .../openai-responses-server-schema.ts | 17 ++-- .../src/providers/openai-responses-server.ts | 17 +++- .../src/providers/openai-responses-shared.ts | 50 +++++++----- packages/ai/src/providers/openai-responses.ts | 6 +- .../ai/test/openai-completions-compat.test.ts | 78 +++++++++++++++++++ 10 files changed, 258 insertions(+), 47 deletions(-) diff --git a/packages/ai/src/providers/azure-openai-responses.ts b/packages/ai/src/providers/azure-openai-responses.ts index 1ce6d0102..36bf5c58e 100644 --- a/packages/ai/src/providers/azure-openai-responses.ts +++ b/packages/ai/src/providers/azure-openai-responses.ts @@ -50,7 +50,7 @@ const DEFAULT_AZURE_API_VERSION = "v1"; const AZURE_OPENAI_RESPONSES_FIRST_EVENT_TIMEOUT_MESSAGE = "Azure OpenAI responses stream timed out while waiting for the first event"; -function parseDeploymentNameMap(value: string | undefined): Map { +export function parseAzureDeploymentNameMap(value: string | undefined): Map { const map = new Map(); if (!value) return map; for (const entry of value.split(",")) { @@ -67,7 +67,7 @@ function resolveDeploymentName(model: Model<"azure-openai-responses">, options?: if (options?.azureDeploymentName) { return options.azureDeploymentName; } - const mappedDeployment = parseDeploymentNameMap($env.AZURE_OPENAI_DEPLOYMENT_NAME_MAP).get(model.id); + const mappedDeployment = parseAzureDeploymentNameMap($env.AZURE_OPENAI_DEPLOYMENT_NAME_MAP).get(model.id); return mappedDeployment ?? model.id; } diff --git a/packages/ai/src/providers/openai-chat-server-schema.ts b/packages/ai/src/providers/openai-chat-server-schema.ts index 4a2cef612..854020cad 100644 --- a/packages/ai/src/providers/openai-chat-server-schema.ts +++ b/packages/ai/src/providers/openai-chat-server-schema.ts @@ -145,6 +145,11 @@ export const assistantMessageSchema = z.object({ role: z.literal("assistant"), content: baseContent.optional(), tool_calls: z.array(toolCallSchema).optional(), + // DeepSeek-style reasoning channel. The gateway emits it on the way out + // (encodeResponse/encodeStream); accept it back so thinking-mode + // continuations replay the model's actual reasoning instead of a + // synthesized placeholder. + reasoning_content: z.string().nullish(), }); export const toolMessageSchema = z.object({ diff --git a/packages/ai/src/providers/openai-chat-server.ts b/packages/ai/src/providers/openai-chat-server.ts index 053789f46..49a9d3217 100644 --- a/packages/ai/src/providers/openai-chat-server.ts +++ b/packages/ai/src/providers/openai-chat-server.ts @@ -91,6 +91,7 @@ export function parseRequest(body: unknown, headers?: Headers): ParsedRequest { buildAssistantMessage( (m.content ?? undefined) as string | OpenAIChatContentPart[] | undefined, m.tool_calls, + (m as { reasoning_content?: string | null }).reasoning_content ?? undefined, data.model, now, ), @@ -227,10 +228,17 @@ function decodeDataUri(url: string): { data: string; mimeType: string } | undefi function buildAssistantMessage( content: string | OpenAIChatContentPart[] | undefined, toolCalls: OpenAIChatToolCall[] | undefined, + reasoningContent: string | undefined, modelId: string, now: number, ): AssistantMessage { const parts: AssistantMessage["content"] = []; + if (reasoningContent !== undefined && reasoningContent.length > 0) { + // Replayed reasoning channel. The signature names the wire field so + // completions providers that demand exact `reasoning_content` replay + // (DeepSeek/Kimi) echo the model's actual reasoning back verbatim. + parts.push({ type: "thinking", thinking: reasoningContent, thinkingSignature: "reasoning_content" }); + } const text = stringifyContent(content); if (text.length > 0) parts.push({ type: "text", text }); if (toolCalls) { @@ -529,6 +537,9 @@ export function encodeStream( async start(controller) { // contentIndex (from pi-ai events) -> tool_calls index on the wire. const toolIndexByContentIndex = new Map(); + // wire index -> id/name emitted on the start chunk, to detect late-arriving + // upstream id/name that needs a corrective chunk before the finish. + const sentToolMeta = new Map(); let nextToolIndex = 0; let hasToolCalls = false; let finishReason: string = "stop"; @@ -559,6 +570,7 @@ export function encodeStream( toolIndexByContentIndex.set(event.contentIndex, idx); const partial = event.partial.content[event.contentIndex]; const call = partial && partial.type === "toolCall" ? partial : undefined; + sentToolMeta.set(idx, { id: call?.id ?? "", name: call?.name ?? "" }); writeSse( controller, baseChunk( @@ -588,6 +600,38 @@ export function encodeStream( break; } + case "toolcall_end": { + const idx = toolIndexByContentIndex.get(event.contentIndex); + if (idx === undefined) break; + const sent = sentToolMeta.get(idx); + if (sent === undefined) break; + // Upstream completions providers can receive the real id/name in a + // later chunk than toolcall_start. Emit a corrective chunk only when + // the streamed value was empty: accumulating clients concatenate + // string fields, so "" + value is the only safe correction. + const correctId = sent.id === "" && event.toolCall.id !== "" ? event.toolCall.id : undefined; + const correctName = + sent.name === "" && event.toolCall.name !== "" ? event.toolCall.name : undefined; + if (correctId !== undefined || correctName !== undefined) { + writeSse( + controller, + baseChunk( + { + tool_calls: [ + { + index: idx, + ...(correctId !== undefined ? { id: correctId } : {}), + ...(correctName !== undefined ? { function: { name: correctName } } : {}), + }, + ], + }, + null, + ), + ); + } + break; + } + case "done": finishReason = event.reason === "toolUse" @@ -610,8 +654,8 @@ export function encodeStream( return; } - // Drop start / *_start / *_end — chat-completions wire only - // surfaces deltas and the terminal finish_reason. + // Drop start / *_start and text/thinking *_end — chat-completions + // wire only surfaces deltas and the terminal finish_reason. default: break; } diff --git a/packages/ai/src/providers/openai-codex-responses.ts b/packages/ai/src/providers/openai-codex-responses.ts index 021044bdc..f7a7d283a 100644 --- a/packages/ai/src/providers/openai-codex-responses.ts +++ b/packages/ai/src/providers/openai-codex-responses.ts @@ -1269,9 +1269,17 @@ function handleMessageTextDelta( partType: "output_text" | "refusal", ): void { if (currentItem?.type !== "message" || currentBlock?.type !== "text") return; - if (!currentItem.content || currentItem.content.length === 0) return; - const lastPart = currentItem.content[currentItem.content.length - 1]; - if (!lastPart || lastPart.type !== partType) return; + currentItem.content = currentItem.content || []; + let lastPart = currentItem.content[currentItem.content.length - 1]; + if (lastPart?.type !== partType) { + // `content_part.added` never arrived (lossy proxy) — synthesize the part + // so live text still streams instead of freezing until output_item.done. + lastPart = + partType === "output_text" + ? { type: "output_text", text: "", annotations: [] } + : { type: "refusal", refusal: "" }; + currentItem.content.push(lastPart); + } const delta = (rawEvent as { delta?: string }).delta || ""; currentBlock.text += delta; if (lastPart.type === "output_text") { @@ -1493,6 +1501,24 @@ function handleResponseCompleted( } } + // Finalize any toolCall block whose output_item.done never arrived: the + // throttled delta parser may have left block.arguments stale, and the + // toolUse promotion below would hand the agent incomplete arguments. + // Mirrors the shared decoder's response.completed sweep; also strips the + // transient partialJson/lastParseLen fields so they never persist. + for (const block of output.content) { + if (block.type !== "toolCall") continue; + const pending = block as ToolCall & { partialJson?: string; lastParseLen?: number }; + if (pending.partialJson) { + pending.arguments = + pending.customWireName !== undefined + ? { input: pending.partialJson } + : parseStreamingJson(pending.partialJson); + } + delete pending.partialJson; + delete pending.lastParseLen; + } + calculateCost(model, output.usage); applyCodexServiceTierPricing(model, output.usage, response?.service_tier, runtime.requestBodyForState.service_tier); output.stopReason = mapOpenAIResponsesStopReason(response?.status as OpenAI.Responses.ResponseStatus | undefined); diff --git a/packages/ai/src/providers/openai-completions.ts b/packages/ai/src/providers/openai-completions.ts index 661c12704..250f99ff2 100644 --- a/packages/ai/src/providers/openai-completions.ts +++ b/packages/ai/src/providers/openai-completions.ts @@ -67,6 +67,7 @@ import { type StreamMarkupHealingEvent, } from "../utils/stream-markup-healing"; import { isForcedToolChoice, mapToOpenAICompletionsToolChoice } from "../utils/tool-choice"; +import { parseAzureDeploymentNameMap } from "./azure-openai-responses"; import { buildCopilotDynamicHeaders, hasCopilotVisionInput, @@ -460,6 +461,10 @@ export const streamOpenAICompletions: StreamFunction<"openai-completions"> = ( const { requestAbortController, requestSignal } = abortTracker; const onSseEvent = options?.onSseEvent; const rawSseObserver = onSseEvent ? (event: RawSseEvent) => onSseEvent(event, model) : undefined; + // Assigned once the block helpers exist (they are scoped to the `try`); + // the catch handler uses it to close any open blocks before emitting the + // terminal error so both exit paths obey the same block lifecycle. + let finishOpenBlocksOnError: () => void = () => {}; try { const apiKey = options?.apiKey || getEnvApiKey(model.provider) || ""; @@ -634,13 +639,21 @@ export const streamOpenAICompletions: StreamFunction<"openai-completions"> = ( } finishToolCallBlock(block); }; + finishOpenBlocksOnError = () => { + if (currentBlock?.type !== "toolCall") finishCurrentBlock(currentBlock); + finishPendingToolCallBlocks(); + }; const appendText = ( message: AssistantMessage, eventStream: AssistantMessageEventStream, text: string, ): void => { if (currentBlock?.type !== "text") { - finishCurrentBlock(currentBlock); + // Leave toolCall blocks pending across text transitions: chunks after + // the first typically carry only `index`, so a finished (de-registered) + // call would be reborn as a nameless phantom block when its arguments + // resume. The stream-end sweep finalizes pending calls. + if (currentBlock?.type !== "toolCall") finishCurrentBlock(currentBlock); currentBlock = { type: "text", text: "" }; message.content.push(currentBlock); eventStream.push({ type: "text_start", contentIndex: blockIndex(currentBlock), partial: message }); @@ -663,7 +676,9 @@ export const streamOpenAICompletions: StreamFunction<"openai-completions"> = ( currentBlock?.type !== "thinking" || (signature !== undefined && currentBlock.thinkingSignature !== signature) ) { - finishCurrentBlock(currentBlock); + // Same as appendText: leave toolCall blocks pending so index-only + // continuation deltas can still find them. + if (currentBlock?.type !== "toolCall") finishCurrentBlock(currentBlock); currentBlock = { type: "thinking", thinking: "", thinkingSignature: signature }; message.content.push(currentBlock); eventStream.push({ @@ -896,6 +911,11 @@ export const streamOpenAICompletions: StreamFunction<"openai-completions"> = ( partial: output, }); } else { + // Resuming a pending call after interleaved text/thinking: + // close the text/thinking block we drifted into. + if (currentBlock !== block && currentBlock && currentBlock.type !== "toolCall") { + finishCurrentBlock(currentBlock); + } currentBlock = block; if (streamIndex !== undefined && block.streamIndex === undefined) { block.streamIndex = streamIndex; @@ -1037,6 +1057,12 @@ export const streamOpenAICompletions: StreamFunction<"openai-completions"> = ( stream.push({ type: "done", reason: output.stopReason, message: output }); stream.end(); } catch (error) { + // Close open blocks first so consumers tracking text_/thinking_/toolcall_ + // lifecycles never see orphaned starts on the error path. Best-effort: a + // throw here must not prevent the terminal error event below. + try { + finishOpenBlocksOnError(); + } catch {} for (const block of output.content) delete (block as any).index; const firstEventTimeoutError = abortTracker.getLocalAbortReason(); output.stopReason = abortTracker.wasCallerAbort() ? "aborted" : "error"; @@ -1129,7 +1155,11 @@ async function createClient( if (baseUrl?.includes(".openai.azure.com")) { const apiVersion = $env.AZURE_OPENAI_API_VERSION || "2024-10-21"; if (!baseUrl.includes("/deployments/")) { - baseUrl = `${baseUrl}/deployments/${model.id}`; + // Honor AZURE_OPENAI_DEPLOYMENT_NAME_MAP like the responses provider: + // deployment names routinely differ from catalog model ids. + const deploymentName = + parseAzureDeploymentNameMap($env.AZURE_OPENAI_DEPLOYMENT_NAME_MAP).get(model.id) ?? model.id; + baseUrl = `${baseUrl}/deployments/${deploymentName}`; } azureDefaultQuery = { "api-version": apiVersion }; } @@ -1738,12 +1768,12 @@ export function convertMessages( if (compat.requiresThinkingAsText) { // Convert thinking blocks to plain text (no tags to avoid model mimicking them) const thinkingText = nonEmptyThinkingBlocks.map(b => b.thinking).join("\n\n"); - const textContent = assistantMsg.content as Array<{ type: "text"; text: string }> | null; - if (textContent) { - textContent.unshift({ type: "text", text: thinkingText }); - } else { - assistantMsg.content = [{ type: "text", text: thinkingText }]; - } + // `content` is a plain string at this point (set above) or null — + // never an array. Prepend the thinking text to the string form. + assistantMsg.content = + typeof assistantMsg.content === "string" && assistantMsg.content.length > 0 + ? `${thinkingText}\n\n${assistantMsg.content}` + : thinkingText; } else if (compat.requiresReasoningContentForToolCalls) { // Use the streamed signature when the backend accepts whichever // recognized field name was emitted (allowsSynthetic=true). Backends diff --git a/packages/ai/src/providers/openai-responses-server-schema.ts b/packages/ai/src/providers/openai-responses-server-schema.ts index 144853b6b..ea1be4bff 100644 --- a/packages/ai/src/providers/openai-responses-server-schema.ts +++ b/packages/ai/src/providers/openai-responses-server-schema.ts @@ -97,12 +97,17 @@ const assistantMessageItemSchema = z.object({ content: z.union([z.string(), z.array(outputContentBlockSchema)]).optional(), }); -const reasoningItemSchema = z.object({ - type: z.literal("reasoning"), - id: z.string().optional(), - summary: z.array(summaryTextSchema).optional(), - content: z.array(reasoningTextSchema).optional(), -}); +const reasoningItemSchema = z + .object({ + type: z.literal("reasoning"), + id: z.string().optional(), + summary: z.array(summaryTextSchema).optional(), + content: z.array(reasoningTextSchema).optional(), + }) + // Loose: unknown keys like `encrypted_content` must survive the parse — + // the outbound encoder replays them verbatim (buildReasoningItem spreads + // the persisted item to preserve encrypted reasoning round-trips). + .loose(); const functionCallItemSchema = z.object({ type: z.literal("function_call"), diff --git a/packages/ai/src/providers/openai-responses-server.ts b/packages/ai/src/providers/openai-responses-server.ts index 5c507cd67..457713c79 100644 --- a/packages/ai/src/providers/openai-responses-server.ts +++ b/packages/ai/src/providers/openai-responses-server.ts @@ -573,6 +573,17 @@ function reasoningItemId(part: ThinkingContent): string { return makeReasoningId(); } +/** + * pi-ai responses providers mint composite `"{call_id}|{item_id}"` tool-call + * ids ({@link encodeResponsesToolCallId}). Only the call_id half belongs on + * the wire: third-party clients validate the `call_id` charset + * (`^[a-zA-Z0-9_-]+$`) or echo it to other backends, and `|` fails both. + */ +function wireCallId(id: string): string { + const sep = id.indexOf("|"); + return sep >= 0 ? id.slice(0, sep) : id; +} + /** * Walk the assistant content array and group consecutive TextContent into a * single message item; each ThinkingContent / ToolCall is its own item. @@ -609,7 +620,7 @@ function buildOutputItems(message: AssistantMessage): OutputItem[] { out.push({ type: "custom_tool_call", id: part.thoughtSignature ?? makeCustomCallId(), - call_id: part.id, + call_id: wireCallId(part.id), name: part.customWireName, input: rawInput, status: "completed", @@ -618,7 +629,7 @@ function buildOutputItems(message: AssistantMessage): OutputItem[] { out.push({ type: "function_call", id: part.thoughtSignature ?? makeFuncCallId(), - call_id: part.id, + call_id: wireCallId(part.id), name: part.name, arguments: JSON.stringify(part.arguments ?? {}), status: "completed", @@ -801,7 +812,7 @@ export function encodeStream( : undefined; const isCustom = customWireName !== undefined; const itemId = tc?.thoughtSignature ?? (isCustom ? makeCustomCallId() : makeFuncCallId()); - const callId = tc?.id ?? ""; + const callId = wireCallId(tc?.id ?? ""); const name = customWireName ?? tc?.name ?? ""; const item = isCustom ? { diff --git a/packages/ai/src/providers/openai-responses-shared.ts b/packages/ai/src/providers/openai-responses-shared.ts index 2652d1821..6c399b9e5 100644 --- a/packages/ai/src/providers/openai-responses-shared.ts +++ b/packages/ai/src/providers/openai-responses-shared.ts @@ -664,32 +664,42 @@ export async function processResponsesStream( } else if (event.type === "response.output_text.delta") { const entry = lookupOpenItem(event); if (entry?.item.type === "message" && entry.block.type === "text") { - const lastPart = entry.item.content?.[entry.item.content.length - 1]; - if (lastPart?.type === "output_text") { - entry.block.text += event.delta; - lastPart.text += event.delta; - stream.push({ - type: "text_delta", - contentIndex: contentIndexOf(entry.block), - delta: event.delta, - partial: output, - }); + entry.item.content = entry.item.content || []; + let lastPart = entry.item.content[entry.item.content.length - 1]; + if (lastPart?.type !== "output_text") { + // `content_part.added` never arrived (lossy proxy) — synthesize the + // part so live text still streams instead of freezing until the + // item's output_item.done recovers the final text. + lastPart = { type: "output_text", text: "", annotations: [] }; + entry.item.content.push(lastPart); } + entry.block.text += event.delta; + lastPart.text += event.delta; + stream.push({ + type: "text_delta", + contentIndex: contentIndexOf(entry.block), + delta: event.delta, + partial: output, + }); } } else if (event.type === "response.refusal.delta") { const entry = lookupOpenItem(event); if (entry?.item.type === "message" && entry.block.type === "text") { - const lastPart = entry.item.content?.[entry.item.content.length - 1]; - if (lastPart?.type === "refusal") { - entry.block.text += event.delta; - lastPart.refusal += event.delta; - stream.push({ - type: "text_delta", - contentIndex: contentIndexOf(entry.block), - delta: event.delta, - partial: output, - }); + entry.item.content = entry.item.content || []; + let lastPart = entry.item.content[entry.item.content.length - 1]; + if (lastPart?.type !== "refusal") { + // Same lossy-proxy hardening as the output_text branch above. + lastPart = { type: "refusal", refusal: "" }; + entry.item.content.push(lastPart); } + entry.block.text += event.delta; + lastPart.refusal += event.delta; + stream.push({ + type: "text_delta", + contentIndex: contentIndexOf(entry.block), + delta: event.delta, + partial: output, + }); } } else if (event.type === "response.function_call_arguments.delta") { const entry = lookupOpenFunctionCallItem(event); diff --git a/packages/ai/src/providers/openai-responses.ts b/packages/ai/src/providers/openai-responses.ts index d385d0e06..05591b3a1 100644 --- a/packages/ai/src/providers/openai-responses.ts +++ b/packages/ai/src/providers/openai-responses.ts @@ -1,4 +1,4 @@ -import { $env, extractHttpStatusFromError, structuredCloneJSON } from "@oh-my-pi/pi-utils"; +import { $env, extractHttpStatusFromError } from "@oh-my-pi/pi-utils"; import OpenAI, { APIConnectionTimeoutError as OpenAIConnectionTimeoutError } from "openai"; import type { Tool as OpenAITool, @@ -312,7 +312,9 @@ export const streamOpenAIResponses: StreamFunction<"openai-responses"> = ( if (!firstTokenTime) firstTokenTime = Date.now(); }, onOutputItemDone: item => { - nativeOutputItems.push(structuredCloneJSON(item) as unknown as Record); + // `processResponsesStream` hands over a private clone already; no + // second deep copy needed (reasoning items carry multi-KB blobs). + nativeOutputItems.push(item as unknown as Record); }, }); diff --git a/packages/ai/test/openai-completions-compat.test.ts b/packages/ai/test/openai-completions-compat.test.ts index 3e9118638..7ca6085a7 100644 --- a/packages/ai/test/openai-completions-compat.test.ts +++ b/packages/ai/test/openai-completions-compat.test.ts @@ -172,6 +172,84 @@ describe("openai-completions compatibility", () => { expect(assistant.content).toBe("hello world"); }); + it("prepends thinking text to string assistant content when requiresThinkingAsText is set", () => { + const model: Model<"openai-completions"> = { + ...getBundledModel("openai", "gpt-4o-mini"), + api: "openai-completions", + }; + const assistantMessage: AssistantMessage = { + role: "assistant", + content: [ + { type: "thinking", thinking: "chain of thought" }, + { type: "text", text: "final answer" }, + ], + api: model.api, + provider: model.provider, + model: model.id, + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "stop", + timestamp: Date.now(), + }; + const messages = convertMessages( + model, + { messages: [assistantMessage] }, + { + ...detectCompat(model), + requiresThinkingAsText: true, + }, + ); + const assistant = messages.find(message => message.role === "assistant"); + expect(assistant).toBeDefined(); + if (assistant?.role !== "assistant") throw new Error("assistant message missing"); + // Regression: thinking+text replay used to call `.unshift` on the string + // content set above (TypeError). Both blocks must survive as one string. + expect(typeof assistant.content).toBe("string"); + expect(assistant.content).toBe("chain of thought\n\nfinal answer"); + }); + + it("emits thinking-only assistant content as a plain string when requiresThinkingAsText is set", () => { + const model: Model<"openai-completions"> = { + ...getBundledModel("openai", "gpt-4o-mini"), + api: "openai-completions", + }; + const assistantMessage: AssistantMessage = { + role: "assistant", + content: [{ type: "thinking", thinking: "only thoughts" }], + api: model.api, + provider: model.provider, + model: model.id, + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "stop", + timestamp: Date.now(), + }; + const messages = convertMessages( + model, + { messages: [assistantMessage] }, + { + ...detectCompat(model), + requiresThinkingAsText: true, + }, + ); + const assistant = messages.find(message => message.role === "assistant"); + expect(assistant).toBeDefined(); + if (assistant?.role !== "assistant") throw new Error("assistant message missing"); + expect(assistant.content).toBe("only thoughts"); + }); + it("preserves multiple system prompts as leading system messages for chat completions", () => { const model: Model<"openai-completions"> = { ...getBundledModel("openai", "gpt-4o-mini"),