smashimo/oh-my-pi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(ai): fixed OpenAI provider stream assembly and gateway round-trips

Browse Source 
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.
This commit is contained in:
can1357
2026-06-10 01:26:31 +02:00
parent 2e9746ad61
commit 694a45040f
10 changed files with 258 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/ai/src/providers/azure-openai-responses.ts
+2 -2
View File
@@ -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<string, string> {
export function parseAzureDeploymentNameMap(value: string | undefined): Map<string, string> {
const map = new Map<string, string>();
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;
}
packages/ai/src/providers/openai-chat-server-schema.ts
+5
View File
@@ -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({
packages/ai/src/providers/openai-chat-server.ts
+46 -2
View File
@@ -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<number, number>();
// 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<number, { id: string; name: string }>();
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;
}
packages/ai/src/providers/openai-codex-responses.ts
+29 -3
View File
@@ -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);
packages/ai/src/providers/openai-completions.ts
+39 -9
View File
@@ -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
packages/ai/src/providers/openai-responses-server-schema.ts
+11 -6
View File
@@ -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"),
packages/ai/src/providers/openai-responses-server.ts
+14 -3
View File
@@ -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
? {
packages/ai/src/providers/openai-responses-shared.ts
+30 -20
View File
@@ -664,32 +664,42 @@ export async function processResponsesStream<TApi extends Api>(
} 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);
packages/ai/src/providers/openai-responses.ts
+4 -2
View File
@@ -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<unknown>(item) as unknown as Record<string, unknown>);
// `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<string, unknown>);
},
});
packages/ai/test/openai-completions-compat.test.ts
+78
View File
@@ -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"),