From 02bd2d1c628d0e01c71b7088443d3cb117495d2d Mon Sep 17 00:00:00 2001 From: Armin Ronacher Date: Fri, 7 Aug 2026 22:59:56 +0200 Subject: [PATCH] fix(ai): preserve Responses tool-call namespaces closes #7709 --- packages/agent/CHANGELOG.md | 4 + packages/agent/src/proxy.ts | 3 +- packages/agent/test/proxy.test.ts | 79 ++++++ packages/ai/CHANGELOG.md | 1 + .../ai/src/api/openai-responses-shared.ts | 19 +- packages/ai/src/types.ts | 2 + .../test/openai-responses-namespace.test.ts | 224 ++++++++++++++++++ packages/server/src/protocol.ts | 2 +- 8 files changed, 328 insertions(+), 6 deletions(-) create mode 100644 packages/agent/test/proxy.test.ts create mode 100644 packages/ai/test/openai-responses-namespace.test.ts diff --git a/packages/agent/CHANGELOG.md b/packages/agent/CHANGELOG.md index 7911fbc60..81e894d8c 100644 --- a/packages/agent/CHANGELOG.md +++ b/packages/agent/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Fixed + +- Fixed `streamProxy()` dropping finalized tool-call metadata such as OpenAI Responses namespaces ([#7709](https://github.com/earendil-works/pi/issues/7709)). + ## [0.84.1] - 2026-08-07 ### Added diff --git a/packages/agent/src/proxy.ts b/packages/agent/src/proxy.ts index 2e4cc8f48..678bc7abf 100644 --- a/packages/agent/src/proxy.ts +++ b/packages/agent/src/proxy.ts @@ -43,7 +43,7 @@ export type ProxyAssistantMessageEvent = | { type: "thinking_end"; contentIndex: number; contentSignature?: string } | { type: "toolcall_start"; contentIndex: number; id: string; toolName: string } | { type: "toolcall_delta"; contentIndex: number; delta: string } - | { type: "toolcall_end"; contentIndex: number } + | { type: "toolcall_end"; contentIndex: number; toolCall: ToolCall } | { type: "done"; reason: Extract; @@ -338,6 +338,7 @@ function processProxyEvent( case "toolcall_end": { const content = partial.content[proxyEvent.contentIndex]; if (content?.type === "toolCall") { + Object.assign(content, proxyEvent.toolCall); delete (content as any).partialJson; return { type: "toolcall_end", diff --git a/packages/agent/test/proxy.test.ts b/packages/agent/test/proxy.test.ts new file mode 100644 index 000000000..05f59b11f --- /dev/null +++ b/packages/agent/test/proxy.test.ts @@ -0,0 +1,79 @@ +import type { AssistantMessage, AssistantMessageEvent, Model } from "@earendil-works/pi-ai"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { type ProxyAssistantMessageEvent, streamProxy } from "../src/proxy.ts"; + +const model: Model<"openai-responses"> = { + id: "gpt-5.4", + name: "GPT-5.4", + 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: 400000, + maxTokens: 128000, +}; + +const usage: AssistantMessage["usage"] = { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, +}; + +afterEach(() => { + vi.unstubAllGlobals(); +}); + +describe("streamProxy", () => { + it("preserves tool-call metadata received only on toolcall_end", async () => { + const proxyEvents: ProxyAssistantMessageEvent[] = [ + { type: "start" }, + { type: "toolcall_start", contentIndex: 0, id: "call_test|fc_test", toolName: "lookup" }, + { type: "toolcall_delta", contentIndex: 0, delta: '{"value":"hello"}' }, + { + type: "toolcall_end", + contentIndex: 0, + toolCall: { + type: "toolCall", + id: "call_test|fc_test", + name: "lookup", + arguments: { value: "hello" }, + namespace: "dynamic_tools", + }, + }, + { type: "done", reason: "toolUse", usage }, + ]; + const body = proxyEvents.map((event) => `data: ${JSON.stringify(event)}\n\n`).join(""); + vi.stubGlobal( + "fetch", + vi.fn(async () => new Response(body, { status: 200 })), + ); + + const stream = streamProxy( + model, + { systemPrompt: "", messages: [] }, + { + authToken: "test-token", + proxyUrl: "https://proxy.example.com", + }, + ); + const events: AssistantMessageEvent[] = []; + for await (const event of stream) events.push(event); + const result = await stream.result(); + const endEvent = events.find((event) => event.type === "toolcall_end"); + + expect(endEvent).toMatchObject({ + type: "toolcall_end", + toolCall: { namespace: "dynamic_tools" }, + }); + expect(result.content[0]).toMatchObject({ + type: "toolCall", + arguments: { value: "hello" }, + namespace: "dynamic_tools", + }); + }); +}); diff --git a/packages/ai/CHANGELOG.md b/packages/ai/CHANGELOG.md index 86fc38efe..b18b16821 100644 --- a/packages/ai/CHANGELOG.md +++ b/packages/ai/CHANGELOG.md @@ -5,6 +5,7 @@ ### Fixed - Fixed upstream request buffer limit failures to trigger automatic assistant retries. +- Fixed OpenAI Responses function and custom tool calls to preserve namespaces during streaming, proxying, and replay ([#7709](https://github.com/earendil-works/pi/issues/7709)). ## [0.84.1] - 2026-08-07 diff --git a/packages/ai/src/api/openai-responses-shared.ts b/packages/ai/src/api/openai-responses-shared.ts index ad24c8d46..801c9cfcd 100644 --- a/packages/ai/src/api/openai-responses-shared.ts +++ b/packages/ai/src/api/openai-responses-shared.ts @@ -210,10 +210,9 @@ export function convertResponsesMessages( } else if (msg.role === "assistant") { const output: ResponseInput = []; const assistantMsg = msg as AssistantMessage; - const isDifferentModel = - assistantMsg.model !== model.id && - assistantMsg.provider === model.provider && - assistantMsg.api === model.api; + const isSameProviderAndApi = assistantMsg.provider === model.provider && assistantMsg.api === model.api; + const isSameModel = isSameProviderAndApi && assistantMsg.model === model.id; + const isDifferentModel = isSameProviderAndApi && assistantMsg.model !== model.id; let textBlockIndex = 0; for (const block of msg.content) { @@ -261,6 +260,8 @@ export function convertResponsesMessages( itemId = undefined; } + const canReplayNamespace = isSameModel || options?.deferredTools?.has(toolCall.name) === true; + if (customInputProperty !== undefined) { output.push({ type: "custom_tool_call", @@ -270,6 +271,9 @@ export function convertResponsesMessages( input: sanitizeSurrogates( getGrammarToolInput(toolCall.name, toolCall.arguments, customInputProperty), ), + ...(canReplayNamespace && toolCall.namespace !== undefined + ? { namespace: toolCall.namespace } + : {}), } satisfies ResponseOutputItem); } else { output.push({ @@ -278,6 +282,9 @@ export function convertResponsesMessages( call_id: callId, name: toolCall.name, arguments: JSON.stringify(toolCall.arguments), + ...(canReplayNamespace && toolCall.namespace !== undefined + ? { namespace: toolCall.namespace } + : {}), }); } } @@ -472,6 +479,7 @@ export async function processResponsesStream( id: `${item.call_id}|${item.id}`, name: item.name, arguments: {}, + ...(item.namespace !== undefined ? { namespace: item.namespace } : {}), partialJson: item.arguments || "", }; output.content.push(block); @@ -492,6 +500,7 @@ export async function processResponsesStream( id: `${item.call_id}|${item.id}`, name: item.name, arguments: { [inputProperty]: input }, + ...(item.namespace !== undefined ? { namespace: item.namespace } : {}), customInput: { property: inputProperty, jsonBuffer: { input: "", started: false, closed: false }, @@ -693,6 +702,7 @@ export async function processResponsesStream( slot.block.partialJson !== undefined ) { slot.block.arguments = parseStreamingJson(item.arguments || slot.block.partialJson || "{}"); + if (item.namespace !== undefined) slot.block.namespace = item.namespace; // Finalize in-place and strip the scratch buffer so replay only // carries parsed arguments. delete slot.block.partialJson; @@ -708,6 +718,7 @@ export async function processResponsesStream( slot, appendCustomToolCallInput(slot.block, item.input ?? getCustomToolCallInput(slot.block), true), ); + if (item.namespace !== undefined) slot.block.namespace = item.namespace; delete slot.block.customInput; stream.push({ type: "toolcall_end", diff --git a/packages/ai/src/types.ts b/packages/ai/src/types.ts index 5bf6c9fa6..9b11136b7 100644 --- a/packages/ai/src/types.ts +++ b/packages/ai/src/types.ts @@ -363,6 +363,8 @@ export interface ToolCall { name: string; arguments: Record; thoughtSignature?: string; // Google-specific: opaque signature for reusing thought context + /** OpenAI Responses namespace for calls to dynamically loaded or namespaced tools. */ + namespace?: string; } export interface Usage { diff --git a/packages/ai/test/openai-responses-namespace.test.ts b/packages/ai/test/openai-responses-namespace.test.ts new file mode 100644 index 000000000..75fc4eac1 --- /dev/null +++ b/packages/ai/test/openai-responses-namespace.test.ts @@ -0,0 +1,224 @@ +import type { ResponseStreamEvent } from "openai/resources/responses/responses.js"; +import { describe, expect, it } from "vitest"; +import { convertResponsesMessages, processResponsesStream } from "../src/api/openai-responses-shared.ts"; +import type { Api, AssistantMessage, Model, ToolCall } from "../src/types.ts"; +import { AssistantMessageEventStream } from "../src/utils/event-stream.ts"; + +const model: Model<"openai-responses"> = { + id: "gpt-5.4", + name: "GPT-5.4", + 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: 400000, + maxTokens: 128000, +}; + +function createOutput(): AssistantMessage { + return { + role: "assistant", + content: [], + 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: "pending", + timestamp: Date.now(), + }; +} + +async function* createFunctionCallEvents(): AsyncIterable { + yield { + type: "response.output_item.added", + sequence_number: 0, + output_index: 0, + item: { + type: "function_call", + id: "fc_test", + call_id: "call_test", + name: "lookup", + arguments: "", + }, + } as ResponseStreamEvent; + yield { + type: "response.output_item.done", + sequence_number: 1, + output_index: 0, + item: { + type: "function_call", + id: "fc_test", + call_id: "call_test", + name: "lookup", + arguments: '{"value":"hello"}', + namespace: "dynamic_tools", + }, + } as ResponseStreamEvent; + yield { + type: "response.completed", + sequence_number: 2, + response: { id: "resp_test", status: "completed" }, + } as ResponseStreamEvent; +} + +async function* createCustomToolCallEvents(): AsyncIterable { + yield { + type: "response.output_item.added", + sequence_number: 0, + output_index: 0, + item: { + type: "custom_tool_call", + id: "ctc_test", + call_id: "call_test", + name: "query", + input: "", + }, + } as ResponseStreamEvent; + yield { + type: "response.output_item.done", + sequence_number: 1, + output_index: 0, + item: { + type: "custom_tool_call", + id: "ctc_test", + call_id: "call_test", + name: "query", + input: "hello", + namespace: "dynamic_tools", + }, + } as ResponseStreamEvent; + yield { + type: "response.completed", + sequence_number: 2, + response: { id: "resp_test", status: "completed" }, + } as ResponseStreamEvent; +} + +function getToolCall(output: AssistantMessage): ToolCall { + const block = output.content[0]; + if (!block || block.type !== "toolCall") throw new Error("Expected toolCall block"); + return block; +} + +describe("OpenAI Responses tool-call namespaces", () => { + it("round-trips a function namespace received only on output_item.done", async () => { + const output = createOutput(); + await processResponsesStream(createFunctionCallEvents(), output, new AssistantMessageEventStream(), model); + + const toolCall = getToolCall(output); + expect(toolCall).toMatchObject({ + id: "call_test|fc_test", + name: "lookup", + arguments: { value: "hello" }, + namespace: "dynamic_tools", + }); + + const replayed = convertResponsesMessages(model, { messages: [output] }, new Set(["openai"])).find( + (item) => item.type === "function_call", + ); + expect(replayed).toMatchObject({ + type: "function_call", + id: "fc_test", + call_id: "call_test", + name: "lookup", + arguments: '{"value":"hello"}', + namespace: "dynamic_tools", + }); + }); + + it("round-trips a custom-tool namespace received only on output_item.done", async () => { + const output = createOutput(); + const grammarToolInputProperties = new Map([["query", "input"]]); + await processResponsesStream(createCustomToolCallEvents(), output, new AssistantMessageEventStream(), model, { + grammarToolInputProperties, + }); + + const toolCall = getToolCall(output); + expect(toolCall).toMatchObject({ + id: "call_test|ctc_test", + name: "query", + arguments: { input: "hello" }, + namespace: "dynamic_tools", + }); + + const replayed = convertResponsesMessages(model, { messages: [output] }, new Set(["openai"]), { + grammarToolInputProperties, + }).find((item) => item.type === "custom_tool_call"); + expect(replayed).toMatchObject({ + type: "custom_tool_call", + id: "ctc_test", + call_id: "call_test", + name: "query", + input: "hello", + namespace: "dynamic_tools", + }); + }); + + it("drops namespaces when the target cannot replay their load items", () => { + const output = createOutput(); + output.content.push( + { + type: "toolCall", + id: "call_function|fc_test", + name: "lookup", + arguments: { value: "hello" }, + namespace: "dynamic_tools", + }, + { + type: "toolCall", + id: "call_custom|ctc_test", + name: "query", + arguments: { input: "hello" }, + namespace: "dynamic_tools", + }, + ); + const targetModels: Model[] = [ + { ...model, id: "gpt-5.2", name: "GPT-5.2" }, + { ...model, provider: "azure-openai-responses" }, + { + ...model, + api: "openai-codex-responses", + provider: "openai-codex", + id: "gpt-5.3-codex-spark", + name: "GPT-5.3 Codex Spark", + }, + ]; + + for (const targetModel of targetModels) { + const replayed = convertResponsesMessages(targetModel, { messages: [output] }, new Set(["openai"]), { + grammarToolInputProperties: new Map([["query", "input"]]), + }); + const functionCall = replayed.find((item) => item.type === "function_call"); + const customToolCall = replayed.find((item) => item.type === "custom_tool_call"); + expect(functionCall).toBeDefined(); + expect(functionCall).not.toHaveProperty("namespace"); + expect(customToolCall).toBeDefined(); + expect(customToolCall).not.toHaveProperty("namespace"); + } + }); + + it("does not add a namespace to ordinary function calls", () => { + const output = createOutput(); + output.content.push({ + type: "toolCall", + id: "call_test|fc_test", + name: "lookup", + arguments: { value: "hello" }, + }); + + const replayed = convertResponsesMessages(model, { messages: [output] }, new Set(["openai"])).find( + (item) => item.type === "function_call", + ); + expect(replayed).toBeDefined(); + expect(replayed).not.toHaveProperty("namespace"); + }); +}); diff --git a/packages/server/src/protocol.ts b/packages/server/src/protocol.ts index 6180cf407..069828e59 100644 --- a/packages/server/src/protocol.ts +++ b/packages/server/src/protocol.ts @@ -44,7 +44,7 @@ type _AiThinkingContentFieldsAccountedFor = Assert< >; type _AiImageContentFieldsAccountedFor = Assert>; type _AiToolCallFieldsAccountedFor = Assert< - ExactKeys + ExactKeys >; type _AiUsageFieldsAccountedFor = Assert< ExactKeys<