mirror of
https://github.com/diegosouzapw/OmniRoute.git
synced 2026-07-26 09:52:11 +03:00
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
This commit is contained in:
@@ -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
|
||||
59
docker-compose.yml.bak
Normal file
59
docker-compose.yml.bak
Normal file
@@ -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
|
||||
@@ -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<string, unknown>[] = [];
|
||||
|
||||
@@ -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");
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user