pjandro/OmniRoute
1
0
Fork 0
mirror of https://github.com/diegosouzapw/OmniRoute.git synced 2026-07-26 09:52:11 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(sse): normalize provider ids to strings (#3427)

Browse Source 
Integrated into release/v3.8.17
This commit is contained in:
Dmitrii Safronov
2026-06-09 01:41:33 +04:00
committed by GitHub
parent 70c6610fa8
commit fc437ddecd
10 changed files with 457 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
open-sse/handlers/rerank.ts
View File

@@ -40,10 +40,10 @@ function transformRequestForProvider(providerConfig, body) {
/**
* Transform response from provider-specific formats back to Cohere format
*/
function transformResponseFromProvider(providerConfig, data) {
/* @testonly */ export function transformResponseFromProvider(providerConfig, data) {
if (providerConfig.format === "nvidia") {
return {
id: data.id || `rerank-${Date.now()}`,
id: data.id != null ? String(data.id) : `rerank-${Date.now()}`,
results: (data.rankings || []).map((r) => ({
index: r.index,
relevance_score: r.logit || r.score || 0,

19
open-sse/handlers/responseSanitizer.ts
View File

@@ -985,8 +985,10 @@ export function sanitizeStreamingChunk(parsed: unknown): unknown {
// Build sanitized chunk
const sanitized: JsonRecord = {};
// Keep only standard fields
if (parsedRecord.id !== undefined) sanitized.id = parsedRecord.id;
// Keep only standard fields — normalize id to string to avoid AI_InvalidResponseDataError
if (parsedRecord.id !== undefined && parsedRecord.id !== null) {
sanitized.id = normalizeResponseId(typeof parsedRecord.id === "string" ? parsedRecord.id : String(parsedRecord.id));
}
sanitized.object = toString(parsedRecord.object) || "chat.completion.chunk";
if (parsedRecord.created !== undefined) sanitized.created = parsedRecord.created;
if (parsedRecord.model !== undefined) sanitized.model = parsedRecord.model;
@@ -1037,7 +1039,18 @@ export function sanitizeStreamingChunk(parsed: unknown): unknown {
delta.reasoning_content = parts.join("");
}
}
if (deltaRecord.tool_calls !== undefined) delta.tool_calls = deltaRecord.tool_calls;
if (deltaRecord.tool_calls !== undefined) {
delta.tool_calls = Array.isArray(deltaRecord.tool_calls)
? deltaRecord.tool_calls.map((tc) => {
const t = toRecord(tc);
if (!t) return tc;
if (t.id !== undefined && t.id !== null && typeof t.id !== "string") {
return { ...t, id: String(t.id) };
}
return t;
})
: deltaRecord.tool_calls;
}
if (deltaRecord.function_call !== undefined)
delta.function_call = deltaRecord.function_call;
c.delta = delta;

20
open-sse/handlers/sseParser.ts
View File

@@ -163,7 +163,7 @@ export function parseSSEToOpenAIResponse(rawSSE, fallbackModel) {
if (!existing) {
accumulatedToolCalls.set(key, {
id: tc?.id ?? null,
id: tc?.id != null ? String(tc.id) : null,
index: Number.isInteger(tc?.index) ? tc.index : accumulatedToolCalls.size,
type: tc?.type || "function",
function: {
@@ -172,7 +172,7 @@ export function parseSSEToOpenAIResponse(rawSSE, fallbackModel) {
},
});
} else {
existing.id = existing.id || tc?.id || null;
existing.id = existing.id || (tc?.id != null ? String(tc.id) : null);
if (!Number.isInteger(existing.index) && Number.isInteger(tc?.index)) {
existing.index = tc.index;
}
@@ -216,7 +216,7 @@ export function parseSSEToOpenAIResponse(rawSSE, fallbackModel) {
}
const result: Record<string, unknown> = {
id: first.id || `chatcmpl-${Date.now()}`,
id: first.id != null ? String(first.id) : `chatcmpl-${Date.now()}`,
object: "chat.completion",
created: first.created || Math.floor(Date.now() / 1000),
model: first.model || fallbackModel || "unknown",
@@ -452,6 +452,7 @@ function cloneResponseItem(item) {
const record = toRecord(item);
return {
...record,
id: record.id != null ? String(record.id) : record.id,
...(Array.isArray(record.content)
? {
content: record.content.map((contentPart) => {
@@ -477,7 +478,7 @@ function ensureResponsesMessageItem(outputItems, outputIndex) {
const next = {
...(existing && typeof existing === "object" ? existing : {}),
id: existing?.id || `msg_${Date.now()}_${outputIndex}`,
id: existing?.id != null ? String(existing.id) : `msg_${Date.now()}_${outputIndex}`,
type: "message",
role: "assistant",
content: Array.isArray(existing?.content)
@@ -499,7 +500,7 @@ function ensureResponsesReasoningItem(outputItems, outputIndex, itemId) {
const next = {
...(existing && typeof existing === "object" ? existing : {}),
id: itemId || existing?.id || `rs_${Date.now()}_${outputIndex}`,
id: itemId || (existing?.id != null ? String(existing.id) : null) || `rs_${Date.now()}_${outputIndex}`,
type: "reasoning",
summary: Array.isArray(existing?.summary)
? existing.summary.map((summaryPart) => ({ ...toRecord(summaryPart) }))
@@ -525,7 +526,7 @@ function ensureResponsesFunctionCallItem(outputItems, outputIndex, itemId, callI
const next = {
...(existing && typeof existing === "object" ? existing : {}),
id: itemId || existing?.id || `fc_${callId || `${Date.now()}_${outputIndex}`}`,
id: itemId || (existing?.id != null ? String(existing.id) : null) || `fc_${callId || `${Date.now()}_${outputIndex}`}`,
type: "function_call",
call_id: callId || existing?.call_id || "",
name: name || existing?.name || "",
@@ -705,10 +706,13 @@ export function parseSSEToResponsesOutput(rawSSE, fallbackModel) {
: "in_progress";
return {
id: picked.id || `resp_${Date.now()}`,
id: picked.id != null ? String(picked.id) : `resp_${Date.now()}`,
object: picked.object || "response",
model: picked.model || fallbackModel || "unknown",
output: pickedOutput.length > 0 ? pickedOutput : reconstructedOutput,
output: (pickedOutput.length > 0 ? pickedOutput : reconstructedOutput).map((item) => ({
...item,
id: item.id != null ? String(item.id) : item.id,
})),
usage: picked.usage || null,
status: picked.status || statusFallback,
created_at: picked.created_at || Math.floor(Date.now() / 1000),

2
open-sse/utils/ollamaTransform.ts
View File

@@ -51,7 +51,7 @@ export function transformToOllama(response, model) {
}
if (!pendingToolCalls[idx]) {
pendingToolCalls[idx] = { id: tc.id, function: { name: "", arguments: "" } };
pendingToolCalls[idx] = { id: tc.id != null ? String(tc.id) : tc.id, function: { name: "", arguments: "" } };
}
if (tc.function?.name) pendingToolCalls[idx].function.name += tc.function.name;
if (tc.function?.arguments)

39
open-sse/utils/stream.ts
View File

@@ -269,10 +269,10 @@ function collectPassthroughTextualToolCall(
return toolCall;
}
function toStreamingToolCallDelta(toolCall: ToolCall) {
/* @testonly */ export function toStreamingToolCallDelta(toolCall: ToolCall) {
return {
index: toolCall.index,
id: toolCall.id,
id: toolCall.id != null ? String(toolCall.id) : null,
type: toolCall.type,
function: {
name: toolCall.function.name,
@@ -281,11 +281,11 @@ function toStreamingToolCallDelta(toolCall: ToolCall) {
};
}
function toResponsesFunctionCallItem(toolCall: ToolCall) {
/* @testonly */ export function toResponsesFunctionCallItem(toolCall: ToolCall) {
return {
type: "function_call",
id: toolCall.id || `fc_${toolCall.index}`,
call_id: toolCall.id || `call_${toolCall.index}`,
id: (toolCall.id != null ? String(toolCall.id) : null) || `fc_${toolCall.index}`,
call_id: (toolCall.id != null ? String(toolCall.id) : null) || `call_${toolCall.index}`,
name: toolCall.function.name,
arguments: toolCall.function.arguments,
status: "completed",
@@ -1604,6 +1604,14 @@ export function createSSEStream(options: StreamOptions = {}) {
typeof parsed.choices[0].delta.reasoning === "string" &&
!parsed.choices[0].delta.reasoning_content
);
const hadNonStringToolCallId = Array.isArray(parsed.choices)
? parsed.choices.some((choice) =>
Array.isArray(choice?.delta?.tool_calls) &&
choice.delta.tool_calls.some(
(tc) => tc?.id != null && typeof tc.id !== "string"
)
)
: false;
parsed = sanitizeStreamingChunk(parsed);
if (
@@ -1623,6 +1631,7 @@ export function createSSEStream(options: StreamOptions = {}) {
const delta = parsed.choices?.[0]?.delta;
let textualToolCallConverted = false;
let toolCallIdCoerced = false;
// Extract <think> tags from streaming content
if (delta?.content && typeof delta.content === "string") {
@@ -1666,6 +1675,13 @@ export function createSSEStream(options: StreamOptions = {}) {
passthroughHasToolCalls = true;
lastToolCallChunkTime = Date.now();
for (const tc of delta.tool_calls) {
if (tc?.id != null) {
const stringId = String(tc.id);
if (tc.id !== stringId) {
tc.id = stringId;
toolCallIdCoerced = true;
}
}
// Key by index first — id only appears on the first delta in OpenAI streaming
let key: string;
if (Number.isInteger(tc?.index)) {
@@ -1680,7 +1696,7 @@ export function createSSEStream(options: StreamOptions = {}) {
typeof tc?.function?.arguments === "string" ? tc.function.arguments : "";
if (!existing) {
passthroughToolCalls.set(key, {
id: tc?.id ?? null,
id: tc?.id != null ? String(tc.id) : null,
index: Number.isInteger(tc?.index) ? tc.index : passthroughToolCalls.size,
type: tc?.type || "function",
function: {
@@ -1689,7 +1705,7 @@ export function createSSEStream(options: StreamOptions = {}) {
},
});
} else {
if (tc?.id) existing.id = existing.id || tc.id;
if (tc?.id) existing.id = existing.id || String(tc.id);
if (tc?.function?.name && !existing.function.name)
existing.function.name = tc.function.name;
existing.function.arguments += deltaArgs;
@@ -1772,7 +1788,12 @@ export function createSSEStream(options: StreamOptions = {}) {
} else if (textualToolCallConverted) {
output = `data: ${JSON.stringify(parsed)}\n`;
injectedUsage = true;
} else if (idFixed || needsReserialization) {
} else if (
idFixed ||
needsReserialization ||
toolCallIdCoerced ||
hadNonStringToolCallId
) {
output = `data: ${JSON.stringify(parsed)}\n`;
injectedUsage = true;
}
@@ -2385,7 +2406,7 @@ export function createSSEStream(options: StreamOptions = {}) {
? [...state.toolCalls.values()]
.map(
(tc: Record<string, unknown>): ToolCall => ({
id: (tc.id as string) ?? null,
id: tc.id != null ? String(tc.id) : null,
index: (tc.index as number) ?? (tc.blockIndex as number) ?? 0,
type: (tc.type as string) ?? "function",
function: (tc.function as ToolCall["function"]) ?? {

90
tests/unit/ollama-transform.test.ts Normal file
View File

@@ -0,0 +1,90 @@
import test from "node:test";
import assert from "node:assert/strict";
const { transformToOllama } = await import("../../open-sse/utils/ollamaTransform.ts");
test("transformToOllama coerces numeric tool_call id to string without crashing", async () => {
const inputSSE = [
`data: ${JSON.stringify({
id: "chatcmpl_1",
object: "chat.completion.chunk",
created: 1,
model: "gpt-4",
choices: [{
index: 0,
delta: {
tool_calls: [{
index: 0,
id: 12345,
type: "function",
function: { name: "test", arguments: "{}" }
}]
},
finish_reason: "tool_calls"
}]
})}\n`,
].join("");
const inputStream = new ReadableStream({
start(controller) {
controller.enqueue(new TextEncoder().encode(inputSSE));
controller.close();
},
});
const mockResponse = new Response(inputStream, {
headers: { "Content-Type": "text/event-stream" },
});
const result = transformToOllama(mockResponse, "test-model");
const text = await result.text();
// Should produce valid JSON lines without crashing
const lines = text.trim().split("\n");
assert.ok(lines.length > 0, "Should produce at least one line of output");
for (const line of lines) {
const parsed = JSON.parse(line);
assert.ok(parsed, "Each line should be valid JSON");
}
});
test("transformToOllama handles string tool_call id normally", async () => {
const inputSSE = [
`data: ${JSON.stringify({
id: "chatcmpl_1",
object: "chat.completion.chunk",
created: 1,
model: "gpt-4",
choices: [{
index: 0,
delta: {
tool_calls: [{
index: 0,
id: "call_abc",
type: "function",
function: { name: "test", arguments: "{}" }
}]
},
finish_reason: "tool_calls"
}]
})}\n`,
].join("");
const inputStream = new ReadableStream({
start(controller) {
controller.enqueue(new TextEncoder().encode(inputSSE));
controller.close();
},
});
const mockResponse = new Response(inputStream, {
headers: { "Content-Type": "text/event-stream" },
});
const result = transformToOllama(mockResponse, "test-model");
const text = await result.text();
const lines = text.trim().split("\n").map(l => JSON.parse(l));
const toolCallLine = lines.find(l => l.message?.tool_calls);
assert.ok(toolCallLine, "Should produce a tool call line");
});

52
tests/unit/rerank.test.ts Normal file
View File

@@ -0,0 +1,52 @@
import test from "node:test";
import assert from "node:assert/strict";
const { transformResponseFromProvider } = await import("../../open-sse/handlers/rerank.ts");
test("transformResponseFromProvider coerces numeric data.id to string for nvidia format", () => {
const result = transformResponseFromProvider(
{ format: "nvidia" },
{ id: 12345, rankings: [] }
);
assert.equal(typeof result.id, "string");
assert.equal(result.id, "12345");
});
test("transformResponseFromProvider coerces numeric zero data.id to string", () => {
const result = transformResponseFromProvider(
{ format: "nvidia" },
{ id: 0, rankings: [] }
);
assert.equal(typeof result.id, "string");
assert.equal(result.id, "0");
});
test("transformResponseFromProvider uses fallback when data.id is null", () => {
const result = transformResponseFromProvider(
{ format: "nvidia" },
{ id: null, rankings: [] }
);
assert.ok(result.id.startsWith("rerank-"));
});
test("transformResponseFromProvider uses fallback when data.id is undefined", () => {
const result = transformResponseFromProvider(
{ format: "nvidia" },
{ rankings: [] }
);
assert.ok(result.id.startsWith("rerank-"));
});
test("transformResponseFromProvider passes through string id unchanged", () => {
const result = transformResponseFromProvider(
{ format: "nvidia" },
{ id: "nvidia-abc", rankings: [] }
);
assert.equal(typeof result.id, "string");
assert.equal(result.id, "nvidia-abc");
});

79
tests/unit/sse-parser-numeric-ids.test.ts Normal file
View File

@@ -0,0 +1,79 @@
import test from "node:test";
import assert from "node:assert/strict";
const { parseSSEToOpenAIResponse, parseSSEToResponsesOutput } =
await import("../../open-sse/handlers/sseParser.ts");
test("parseSSEToOpenAIResponse coerces numeric top-level id to string", () => {
const rawSSE = [
'data: {"id":123,"model":"gpt-4o-mini","choices":[{"index":0,"delta":{"content":"hello"},"finish_reason":"stop"}]}',
"",
"data: [DONE]",
"",
].join("\n");
const parsed = parseSSEToOpenAIResponse(rawSSE, "fallback-model");
assert.equal(typeof parsed.id, "string");
assert.equal(parsed.id, "123");
});
test("parseSSEToOpenAIResponse coerces numeric tool_call id to string", () => {
const rawSSE = [
'data: {"id":"chatcmpl_tc1","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":123,"function":{"name":"foo","arguments":"{}"}}]},"finish_reason":"tool_calls"}]}',
"",
"data: [DONE]",
"",
].join("\n");
const parsed = parseSSEToOpenAIResponse(rawSSE, "fallback-model");
assert.equal(typeof parsed.choices[0].message.tool_calls[0].id, "string");
assert.equal(parsed.choices[0].message.tool_calls[0].id, "123");
});
test("parseSSEToResponsesOutput coerces numeric output item id to string", () => {
const rawSSE = [
'data: {"type":"response.completed","response":{"id":"resp_num","model":"gpt-4.1","status":"completed","output":[{"id":456,"type":"message","role":"assistant","content":[{"type":"output_text","text":"hi"}]}]}}',
"data: [DONE]",
].join("\n");
const parsed = parseSSEToResponsesOutput(rawSSE, "fallback-model");
assert.equal(typeof parsed.id, "string");
assert.equal(parsed.id, "resp_num");
assert.equal(typeof parsed.output[0].id, "string");
assert.equal(parsed.output[0].id, "456");
});
test("parseSSEToResponsesOutput coerces numeric function call item id from incremental events", () => {
const rawSSE = [
'data: {"type":"response.created","response":{"id":"resp_fc","model":"gpt-4.1","status":"in_progress","output":[]}}',
'data: {"type":"response.output_item.added","output_index":0,"item":{"id":789,"type":"function_call","status":"in_progress","name":"foo","arguments":""}}',
'data: {"type":"response.function_call_arguments.delta","output_index":0,"delta":"{\\"key\\":\\"val\\"}"}',
'data: {"type":"response.function_call_arguments.done","output_index":0,"arguments":"{\\"key\\":\\"val\\"}","status":"completed"}',
'data: {"type":"response.completed","response":{"id":"resp_fc","model":"gpt-4.1","status":"completed","output":null}}',
"data: [DONE]",
].join("\n");
const parsed = parseSSEToResponsesOutput(rawSSE, "fallback-model");
assert.equal(typeof parsed.output[0].id, "string");
assert.equal(parsed.output[0].id, "789");
});
test("parseSSEToResponsesOutput coerces numeric reasoning item id from incremental events", () => {
const rawSSE = [
'data: {"type":"response.created","response":{"id":"resp_rs","model":"gpt-4.1","status":"in_progress","output":[]}}',
'data: {"type":"response.output_item.added","output_index":0,"item":{"id":101,"type":"reasoning","status":"in_progress","summary":[]}}',
'data: {"type":"response.reasoning_summary_text.delta","output_index":0,"delta":"thinking step 1"}',
'data: {"type":"response.reasoning_summary_text.done","output_index":0,"text":"thinking step 1"}',
'data: {"type":"response.completed","response":{"id":"resp_rs","model":"gpt-4.1","status":"completed","output":null}}',
"data: [DONE]",
].join("\n");
const parsed = parseSSEToResponsesOutput(rawSSE, "fallback-model");
assert.equal(typeof parsed.output[0].id, "string");
assert.equal(parsed.output[0].id, "101");
});

68
tests/unit/stream-helpers.test.ts Normal file
View File

@@ -0,0 +1,68 @@
import test from "node:test";
import assert from "node:assert/strict";
const { toStreamingToolCallDelta, toResponsesFunctionCallItem } =
await import("../../open-sse/utils/stream.ts");
type StreamingToolCallInput = Parameters<typeof toStreamingToolCallDelta>[0];
type ResponsesToolCallInput = Parameters<typeof toResponsesFunctionCallItem>[0];
function streamingToolCall(id: unknown): StreamingToolCallInput {
return {
id,
index: 0,
type: "function",
function: { name: "foo", arguments: "{}" },
} as unknown as StreamingToolCallInput;
}
function responsesToolCall(id: unknown, index = 0): ResponsesToolCallInput {
return {
id,
index,
type: "function",
function: { name: "bar", arguments: "{}" },
} as unknown as ResponsesToolCallInput;
}
test("toStreamingToolCallDelta coerces numeric id to string", () => {
const result = toStreamingToolCallDelta(streamingToolCall(42));
assert.equal(typeof result.id, "string");
assert.equal(result.id, "42");
});
test("toStreamingToolCallDelta passes null id as null", () => {
const result = toStreamingToolCallDelta(streamingToolCall(null));
assert.equal(result.id, null);
});
test("toStreamingToolCallDelta passes string id unchanged", () => {
const result = toStreamingToolCallDelta(streamingToolCall("call_abc"));
assert.equal(result.id, "call_abc");
});
test("toResponsesFunctionCallItem coerces numeric id to string", () => {
const result = toResponsesFunctionCallItem(responsesToolCall(99));
assert.equal(typeof result.id, "string");
assert.equal(result.id, "99");
assert.equal(typeof result.call_id, "string");
assert.equal(result.call_id, "99");
});
test("toResponsesFunctionCallItem falls back to fc_ pattern for null id", () => {
const result = toResponsesFunctionCallItem(responsesToolCall(null, 5));
assert.equal(result.id, "fc_5");
assert.equal(result.call_id, "call_5");
});
test("toResponsesFunctionCallItem passes string id unchanged", () => {
const result = toResponsesFunctionCallItem(responsesToolCall("call_xyz"));
assert.equal(result.id, "call_xyz");
assert.equal(result.call_id, "call_xyz");
});

107
tests/unit/stream-numeric-ids.test.ts Normal file
View File

@@ -0,0 +1,107 @@
import test from "node:test";
import assert from "node:assert/strict";
import fs from "node:fs";
import os from "node:os";
import path from "node:path";
const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-stream-numeric-ids-"));
process.env.DATA_DIR = TEST_DATA_DIR;
const core = await import("../../src/lib/db/core.ts");
const { createSSEStream } = await import("../../open-sse/utils/stream.ts");
const { FORMATS } = await import("../../open-sse/translator/formats.ts");
const textEncoder = new TextEncoder();
async function readTransformed(chunks, options) {
const source = new ReadableStream({
start(controller) {
for (const chunk of chunks) {
controller.enqueue(textEncoder.encode(chunk));
}
controller.close();
},
});
return new Response(source.pipeThrough(createSSEStream(options))).text();
}
test.after(() => {
core.resetDbInstance();
if (fs.existsSync(TEST_DATA_DIR)) {
fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true });
}
});
test("createSSEStream passthrough coerces numeric tool_call id to string", async () => {
let onCompletePayload = null;
const text = await readTransformed(
[
`data: ${JSON.stringify({
id: "chatcmpl_num",
object: "chat.completion.chunk",
created: 1,
model: "gpt-4.1-mini",
choices: [{ index: 0, delta: { role: "assistant", content: "Hello " } }],
})}\n\n`,
`data: ${JSON.stringify({
id: "chatcmpl_num",
object: "chat.completion.chunk",
created: 1,
model: "gpt-4.1-mini",
choices: [
{
index: 0,
delta: {
tool_calls: [
{
index: 0,
id: 12345,
type: "function",
function: { name: "read_file", arguments: '{"path":"/tmp/a"}' },
},
],
},
},
],
})}\n\n`,
`data: ${JSON.stringify({
id: "chatcmpl_num",
object: "chat.completion.chunk",
created: 1,
model: "gpt-4.1-mini",
choices: [{ index: 0, delta: {}, finish_reason: "stop" }],
})}\n\n`,
],
{
mode: "passthrough",
sourceFormat: FORMATS.OPENAI,
provider: "openai",
model: "gpt-4.1-mini",
body: { messages: [{ role: "user", content: "hello" }] },
onComplete(payload) {
onCompletePayload = payload;
},
}
);
const lines = text
.trim()
.split("\n")
.filter((line) => line.startsWith("data: ") && !line.includes("[DONE]"));
let sawStreamedToolCall = false;
for (const line of lines) {
const payload = JSON.parse(line.slice(6));
const tc = payload?.choices?.[0]?.delta?.tool_calls?.[0];
if (tc?.id) {
assert.equal(typeof tc.id, "string", "tool_call.id should be a string");
assert.equal(tc.id, "12345");
sawStreamedToolCall = true;
}
}
assert.equal(sawStreamedToolCall, true, "expected streamed tool_call.id to be present");
const finalId = onCompletePayload?.responseBody?.choices?.[0]?.message?.tool_calls?.[0]?.id;
assert.equal(typeof finalId, "string", "tool_call.id in final message should be a string");
assert.equal(finalId, "12345");
});