From 6bbff5eac9cc208490af45e10fd6d73599f0570a Mon Sep 17 00:00:00 2001 From: Erick Kinnee Date: Fri, 17 Jul 2026 21:37:07 +0000 Subject: [PATCH] fix(translator): guard against double-emission for incrementally-streamed tool calls Add a guard that skips response.completed synthesis for call_ids already tracked via incremental output_item.added/.done events. Without this, providers that stream incrementally AND echo function_call items in the response.completed output[] snapshot get duplicate tool call chunks. Also adds a regression test combining both incremental events and a response.completed snapshot in the same turn. Refs: diegosouzapw/OmniRoute#7613 --- ...responses-completed-guard-seen-call-ids.md | 1 + docker-compose.yml.bak | 59 +++++++++ .../translator/response/openai-responses.ts | 16 ++- .../translator-resp-openai-responses.test.ts | 120 ++++++++++++++++++ 4 files changed, 195 insertions(+), 1 deletion(-) create mode 100644 changelog.d/fixes/7613-responses-completed-guard-seen-call-ids.md create mode 100644 docker-compose.yml.bak diff --git a/changelog.d/fixes/7613-responses-completed-guard-seen-call-ids.md b/changelog.d/fixes/7613-responses-completed-guard-seen-call-ids.md new file mode 100644 index 0000000000..b2e32f26be --- /dev/null +++ b/changelog.d/fixes/7613-responses-completed-guard-seen-call-ids.md @@ -0,0 +1 @@ +- **fix(translator):** guard against double-emission when `response.completed` echoes `function_call` items already streamed via incremental `output_item.added`/`done` events — skip synthesis for `call_id`s already tracked, preventing duplicate tool call chunks for incrementally-streaming providers diff --git a/docker-compose.yml.bak b/docker-compose.yml.bak new file mode 100644 index 0000000000..f44158773f --- /dev/null +++ b/docker-compose.yml.bak @@ -0,0 +1,59 @@ +services: + redis: + image: docker.io/library/redis:7-alpine + container_name: omniroute-redis + restart: unless-stopped + networks: + - dockernet + volumes: + - redis-data:/data + command: redis-server --save 60 1 --loglevel warning + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 10s + timeout: 5s + retries: 3 + + omniroute: + image: diegosouzapw/omniroute:latest + container_name: omniroute + restart: unless-stopped + stop_grace_period: 40s + networks: + - dockernet + depends_on: + redis: + condition: service_healthy + environment: + - JWT_SECRET=6E1dyjMlUIZzIMvN5sH+3/dEzgm99HpxLKtjcAzFg1m3x17wZyF31U2i6rlYJZaF + - API_KEY_SECRET=3b6888cbf398a151972f193402cd193caba6ce926f20040537b82c5efcbcf964 + - INITIAL_PASSWORD=CHANGEME + - DATA_DIR=/app/data + - REDIS_URL=redis://redis:6379 + - PORT=20128 + - NODE_ENV=production + - CONTAINER_HOST=docker + - AUTH_COOKIE_SECURE=false + - ALLOW_API_KEY_REVEAL=true + - OMNIROUTE_ALLOW_PRIVATE_PROVIDER_URLS=true + - OMNIROUTE_ALLOW_LOCAL_PROVIDER_URLS=true + - BASE_URL=http://omniroute:20128 + - NEXT_PUBLIC_BASE_URL=http://omniroute:20128 + volumes: + - omniroute-data:/app/data + healthcheck: + test: ["CMD", "node", "healthcheck.mjs"] + interval: 30s + timeout: 5s + retries: 3 + start_period: 15s + +networks: + dockernet: + external: true + +volumes: + redis-data: + name: omniroute-redis-data + omniroute-data: + name: omniroute-data diff --git a/open-sse/translator/response/openai-responses.ts b/open-sse/translator/response/openai-responses.ts index 4b2b8c6a7d..4aae615e90 100644 --- a/open-sse/translator/response/openai-responses.ts +++ b/open-sse/translator/response/openai-responses.ts @@ -784,6 +784,10 @@ function openaiResponsesToOpenAIResponseStream(chunk, state) { state.currentToolCallArgsBuffer = ""; // reset per-call arg buffer state.currentToolCallDeferred = false; + // Track this call_id so response.completed doesn't synthesize a duplicate + if (!state.toolCallIdsSeen) state.toolCallIdsSeen = new Set(); + if (state.currentToolCallId) state.toolCallIdsSeen.add(state.currentToolCallId); + const toolName = normalizeToolName(item.name); if (!toolName) { // Some Responses providers briefly emit placeholder/empty tool names. @@ -863,6 +867,10 @@ function openaiResponsesToOpenAIResponseStream(chunk, state) { const toolName = normalizeToolName(item.name); const toolSchema = state.toolSchemas?.get(toolName); + // Track this call_id so response.completed doesn't synthesize a duplicate + if (!state.toolCallIdsSeen) state.toolCallIdsSeen = new Set(); + if (callId) state.toolCallIdsSeen.add(callId); + if (state.currentToolCallDeferred) { state.currentToolCallDeferred = false; state.currentToolCallArgsBuffer = ""; @@ -1002,7 +1010,13 @@ function openaiResponsesToOpenAIResponseStream(chunk, state) { // agent loop for downstream Chat Completions clients. // --------------------------------------------------------------------------- const outputItems = Array.isArray(data.response?.output) ? data.response.output : []; - const functionCallItems = outputItems.filter((item) => item?.type === "function_call"); + // Filter out function_call items whose call_ids were already tracked via + // incremental events (output_item.added → .done). This prevents double-emission + // when the provider streams incrementally AND response.completed echoes the same + // function_call items in its output[] snapshot. + const functionCallItems = outputItems.filter( + (item) => item?.type === "function_call" && !state.toolCallIdsSeen?.has(item.call_id) + ); if (functionCallItems.length > 0 && !state.finishReasonSent) { const synthesizedChunks: Record[] = []; diff --git a/tests/unit/translator-resp-openai-responses.test.ts b/tests/unit/translator-resp-openai-responses.test.ts index 7fb20ee60c..e9ffb6f31f 100644 --- a/tests/unit/translator-resp-openai-responses.test.ts +++ b/tests/unit/translator-resp-openai-responses.test.ts @@ -801,3 +801,123 @@ test("Responses -> OpenAI: response.completed with function_call in output[] set assert.equal(result[0].choices[0].delta.role, "assistant"); assert.equal(state.roleEmitted, true); }); + +test("Responses -> OpenAI: incremental tool call events + response.completed snapshot does NOT double-emit", () => { + // Regression test: when a provider streams incrementally (output_item.added -> + // function_call_arguments.delta -> output_item.done) and then response.completed + // echoes the same function_call items in its output[] snapshot, we must NOT + // synthesize a second set of tool call chunks for already-seen call_ids. + const state = {}; + + // Phase 1: incremental events for call_a (identical to real streaming provider) + const added = openaiResponsesToOpenAIResponse( + { + type: "response.output_item.added", + item: { type: "function_call", call_id: "call_a", name: "read_file" }, + }, + state + ); + assert.ok(added, "should emit header chunk for incremental tool call"); + assert.equal(added.choices[0].delta.tool_calls[0].id, "call_a"); + + const args = openaiResponsesToOpenAIResponse( + { + type: "response.function_call_arguments.delta", + delta: '{"path":"/tmp/a"}', + }, + state + ); + assert.ok(args, "should emit args delta chunk"); + + openaiResponsesToOpenAIResponse( + { + type: "response.output_item.done", + item: { + type: "function_call", + call_id: "call_a", + name: "read_file", + arguments: { path: "/tmp/a" }, + }, + }, + state + ); + + // Phase 2: incremental events for call_b + const addedB = openaiResponsesToOpenAIResponse( + { + type: "response.output_item.added", + item: { type: "function_call", call_id: "call_b", name: "write_file" }, + }, + state + ); + assert.ok(addedB); + + openaiResponsesToOpenAIResponse( + { + type: "response.function_call_arguments.delta", + delta: '{"path":"/tmp/b","content":"hello"}', + }, + state + ); + + openaiResponsesToOpenAIResponse( + { + type: "response.output_item.done", + item: { + type: "function_call", + call_id: "call_b", + name: "write_file", + arguments: { path: "/tmp/b", content: "hello" }, + }, + }, + state + ); + + // Phase 3: response.completed echoes BOTH call_a and call_b in output[] + // This is the test: the guard should skip synthesis since both call_ids were + // already tracked via the incremental events above. + const completed = openaiResponsesToOpenAIResponse( + { + type: "response.completed", + response: { + id: "resp_combined", + status: "completed", + model: "deepseek-v4", + output: [ + { + type: "function_call", + call_id: "call_a", + name: "read_file", + arguments: { path: "/tmp/a" }, + }, + { + type: "function_call", + call_id: "call_b", + name: "write_file", + arguments: { path: "/tmp/b", content: "hello" }, + }, + ], + usage: { + input_tokens: 10, + output_tokens: 5, + total_tokens: 15, + }, + }, + }, + state + ); + + // response.completed should return a single chunk (not array of chunks) + // because both call_ids were already seen — no synthesis needed. + assert.ok(!Array.isArray(completed), "should NOT return array — no synthesis for seen call_ids"); + assert.equal( + completed.choices[0].finish_reason, + "tool_calls", + "finish_reason should still be tool_calls" + ); + assert.equal(completed.usage.prompt_tokens, 10); + assert.equal(completed.usage.completion_tokens, 5); + + // toolCallIndex should be 2 (two tool calls were processed via incremental events) + assert.equal(state.toolCallIndex, 2, "toolCallIndex should reflect both incremental tool calls"); +});