Skip to content

fix(sse): normalize numeric provider ids to strings #3451

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
diegosouzapw merged 1 commit into from
Jun 9, 2026
Merged

fix(sse): normalize numeric provider ids to strings #3451

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 32 additions & 13 deletions open-sse/handlers/sseParser.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -130,7 +130,7 @@ export function parseSSEToOpenAIResponse(rawSSE, fallbackModel) {

const getToolCallKey = (toolCall: Record<string, unknown>) => {
if (Number.isInteger(toolCall?.index)) return `idx:${toolCall.index}`;
if (toolCall?.id) return `id:${toolCall.id}`;
if (toolCall?.id != null) return `id:${String(toolCall.id)}`;
unknownToolCallSeq += 1;
return `seq:${unknownToolCallSeq}`;
};
Expand Down Expand Up @@ -448,11 +448,16 @@ function toOutputIndex(value) {
return null;
}

function toIdString(value) {
return value === null || value === undefined ? "" : String(value);
}

function cloneResponseItem(item) {
const record = toRecord(item);
return {
...record,
id: record.id != null ? String(record.id) : record.id,
call_id: record.call_id != null ? String(record.call_id) : record.call_id,
...(Array.isArray(record.content)
? {
content: record.content.map((contentPart) => {
Expand Down Expand Up @@ -517,18 +522,28 @@ function ensureResponsesReasoningItem(outputItems, outputIndex, itemId) {

function ensureResponsesFunctionCallItem(outputItems, outputIndex, itemId, callId, name) {
const existing = outputItems.get(outputIndex);
const normalizedItemId = toIdString(itemId);
const normalizedCallId = toIdString(callId);
const existingId = existing?.id != null ? String(existing.id) : "";
const existingCallId = existing?.call_id != null ? String(existing.call_id) : "";

if (existing?.type === "function_call") {
if (callId && !existing.call_id) existing.call_id = callId;
if (existing.call_id != null) existing.call_id = String(existing.call_id);
if (existing.id != null) existing.id = String(existing.id);
if (normalizedCallId && !existing.call_id) existing.call_id = normalizedCallId;
if (name && !existing.name) existing.name = name;
if (itemId && !existing.id) existing.id = itemId;
if (normalizedItemId && !existing.id) existing.id = normalizedItemId;
return existing;
}

const next = {
...(existing && typeof existing === "object" ? existing : {}),
id: itemId || (existing?.id != null ? String(existing.id) : null) || `fc_${callId || `${Date.now()}_${outputIndex}`}`,
id:
normalizedItemId ||
existingId ||
`fc_${normalizedCallId || `${Date.now()}_${outputIndex}`}`,
type: "function_call",
call_id: callId || existing?.call_id || "",
call_id: normalizedCallId || existingCallId || "",
name: name || existing?.name || "",
arguments: typeof existing?.arguments === "string" ? existing.arguments : "",
};
Expand Down Expand Up @@ -626,7 +641,7 @@ export function parseSSEToResponsesOutput(rawSSE, fallbackModel) {
const reasoningItem = ensureResponsesReasoningItem(
outputItems,
outputIndex,
toString(evt.item_id)
toIdString(evt.item_id)
);
const summary = Array.isArray(reasoningItem.summary) ? reasoningItem.summary : [];
const firstPart =
Expand All @@ -641,7 +656,7 @@ export function parseSSEToResponsesOutput(rawSSE, fallbackModel) {
const reasoningItem = ensureResponsesReasoningItem(
outputItems,
outputIndex,
toString(evt.item_id)
toIdString(evt.item_id)
);
const summary = Array.isArray(reasoningItem.summary) ? reasoningItem.summary : [];
const firstPart =
Expand All @@ -656,7 +671,7 @@ export function parseSSEToResponsesOutput(rawSSE, fallbackModel) {
const functionCallItem = ensureResponsesFunctionCallItem(
outputItems,
outputIndex,
toString(evt.item_id),
toIdString(evt.item_id),
"",
""
);
Expand All @@ -667,7 +682,7 @@ export function parseSSEToResponsesOutput(rawSSE, fallbackModel) {
const functionCallItem = ensureResponsesFunctionCallItem(
outputItems,
outputIndex,
toString(evt.item_id),
toIdString(evt.item_id),
"",
""
);
Expand Down Expand Up @@ -709,10 +724,14 @@ export function parseSSEToResponsesOutput(rawSSE, fallbackModel) {
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).map((item) => ({
...item,
id: item.id != null ? String(item.id) : item.id,
})),
output: (pickedOutput.length > 0 ? pickedOutput : reconstructedOutput).map((item) => {
const record = toRecord(item);
return {
...record,
id: record.id != null ? String(record.id) : record.id,
call_id: record.call_id != null ? String(record.call_id) : record.call_id,
};
}),
usage: picked.usage || null,
status: picked.status || statusFallback,
created_at: picked.created_at || Math.floor(Date.now() / 1000),
Expand Down
9 changes: 7 additions & 2 deletions open-sse/utils/ollamaTransform.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,14 +44,19 @@ export function transformToOllama(response, model) {
for (const tc of toolCalls) {
const idx = tc.index;

const toolCallId = tc.id != null ? String(tc.id) : tc.id;

// T37: Prevent merging tool_calls on same index if ID changes
if (pendingToolCalls[idx] && tc.id && pendingToolCalls[idx].id !== tc.id) {
if (pendingToolCalls[idx] && toolCallId && pendingToolCalls[idx].id !== toolCallId) {
completedToolCalls.push(pendingToolCalls[idx]);
delete pendingToolCalls[idx];
}

if (!pendingToolCalls[idx]) {
pendingToolCalls[idx] = { id: tc.id != null ? String(tc.id) : tc.id, function: { name: "", arguments: "" } };
pendingToolCalls[idx] = {
id: toolCallId,
function: { name: "", arguments: "" },
};
}
if (tc.function?.name) pendingToolCalls[idx].function.name += tc.function.name;
if (tc.function?.arguments)
Expand Down
167 changes: 142 additions & 25 deletions open-sse/utils/stream.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,81 @@ export { COLORS, formatSSE };

type JsonRecord = Record<string, unknown>;

function stringifyIdValue(value: unknown): string | null {
return value === null || value === undefined ? null : String(value);
}

function normalizeResponsesOutputItemIds(item: unknown): unknown {
if (!item || typeof item !== "object" || Array.isArray(item)) {
return item;
}

const record = item as JsonRecord;
let changed = false;
const normalized = { ...record };

const id = stringifyIdValue(record.id);
if (id !== null && record.id !== id) {
normalized.id = id;
changed = true;
}

const callId = stringifyIdValue(record.call_id);
if (callId !== null && record.call_id !== callId) {
normalized.call_id = callId;
changed = true;
}

return changed ? normalized : item;
}

function normalizeResponsesSseIds(payload: JsonRecord): boolean {
let changed = false;

for (const key of ["response_id", "item_id", "call_id"] as const) {
const value = stringifyIdValue(payload[key]);
if (value !== null && payload[key] !== value) {
payload[key] = value;
changed = true;
}
}

if (payload.item && typeof payload.item === "object" && !Array.isArray(payload.item)) {
const normalizedItem = normalizeResponsesOutputItemIds(payload.item);
if (normalizedItem !== payload.item) {
payload.item = normalizedItem;
changed = true;
}
}

if (payload.response && typeof payload.response === "object" && !Array.isArray(payload.response)) {
const response = payload.response as JsonRecord;
let responseChanged = false;
const normalizedResponse = { ...response };

const responseId = stringifyIdValue(response.id);
if (responseId !== null && response.id !== responseId) {
normalizedResponse.id = responseId;
responseChanged = true;
}

if (Array.isArray(response.output)) {
const normalizedOutput = response.output.map(normalizeResponsesOutputItemIds);
if (normalizedOutput.some((item, index) => item !== response.output[index])) {
normalizedResponse.output = normalizedOutput;
responseChanged = true;
}
}

if (responseChanged) {
payload.response = normalizedResponse;
changed = true;
}
}

return changed;
}

export const PENDING_REQUEST_CLEARED_MARKER = "__omniroutePendingRequestCleared";

function markPendingRequestCleared(error: Error): Error {
Expand All @@ -76,8 +151,8 @@ function buildResponsesOutputItemKey(item: unknown): string | null {

const record = item as JsonRecord;
const type = typeof record.type === "string" ? record.type : "";
const id = typeof record.id === "string" ? record.id : "";
const callId = typeof record.call_id === "string" ? record.call_id : "";
const id = stringifyIdValue(record.id) ?? "";
const callId = stringifyIdValue(record.call_id) ?? "";
const outputIndex = typeof record.output_index === "number" ? record.output_index : "";
const name = typeof record.name === "string" ? record.name : "";

Expand Down Expand Up @@ -1025,22 +1100,21 @@ export function createSSEStream(options: StreamOptions = {}) {
};

const getResponsesReasoningKey = (payload: Record<string, unknown>): string | null => {
if (typeof payload.item_id === "string" && payload.item_id) {
return payload.item_id;
const itemId = stringifyIdValue(payload.item_id);
if (itemId) {
return itemId;
}

const item =
payload.item && typeof payload.item === "object" && !Array.isArray(payload.item)
? (payload.item as Record<string, unknown>)
: null;
if (item && typeof item.id === "string" && item.id) {
return item.id;
const outputItemId = item ? stringifyIdValue(item.id) : null;
if (outputItemId) {
return outputItemId;
}

const responseId =
typeof payload.response_id === "string" && payload.response_id
? payload.response_id
: passthroughResponsesId;
const responseId = stringifyIdValue(payload.response_id) || passthroughResponsesId;
const outputIndex =
typeof payload.output_index === "number" && Number.isInteger(payload.output_index)
? payload.output_index
Expand Down Expand Up @@ -1319,12 +1393,16 @@ export function createSSEStream(options: StreamOptions = {}) {
parsed.type === "error");

if (isResponsesSSE) {
const responsesIdsNormalized = normalizeResponsesSseIds(parsed as JsonRecord);
const parsedResponse =
parsed.response &&
typeof parsed.response === "object" &&
!Array.isArray(parsed.response)
? (parsed.response as JsonRecord)
: null;
const responseId =
typeof parsed.response?.id === "string"
? parsed.response.id
: typeof parsed.response_id === "string"
? parsed.response_id
: null;
(parsedResponse ? stringifyIdValue(parsedResponse.id) : null) ||
stringifyIdValue(parsed.response_id);
if (responseId) {
passthroughResponsesId = responseId;
}
Expand Down Expand Up @@ -1530,7 +1608,7 @@ export function createSSEStream(options: StreamOptions = {}) {
parsed,
passthroughResponsesOutputItems
);
if (stripped || backfilled || textualToolCallBackfilled) {
if (stripped || backfilled || textualToolCallBackfilled || responsesIdsNormalized) {
output = `data: ${JSON.stringify(parsed)}\n`;
injectedUsage = true;
}
Expand Down Expand Up @@ -1655,6 +1733,7 @@ export function createSSEStream(options: StreamOptions = {}) {
)
)
: false;
const hadNonStringTopLevelId = parsed?.id != null && typeof parsed.id !== "string";

parsed = sanitizeStreamingChunk(parsed);
if (
Expand All @@ -1666,7 +1745,7 @@ export function createSSEStream(options: StreamOptions = {}) {
continue;
}

const idFixed = fixInvalidId(parsed);
const idFixed = hadNonStringTopLevelId ? false : fixInvalidId(parsed);

if (!hasValuableContent(parsed, FORMATS.OPENAI)) {
continue;
Expand Down Expand Up @@ -1718,18 +1797,18 @@ 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;
}
// Note: sanitizeStreamingChunk above already coerces non-string
// tool_call IDs, but this defensive check catches edge cases
// where sanitize didn't run (e.g. flush path shortcuts).
if (tc?.id != null && typeof tc.id !== "string") {
tc.id = String(tc.id);
toolCallIdCoerced = true;
}
// Key by index first — id only appears on the first delta in OpenAI streaming
let key: string;
if (Number.isInteger(tc?.index)) {
key = `idx:${tc.index}`;
} else if (tc?.id) {
} else if (tc?.id != null) {
key = `id:${tc.id}`;
} else {
key = `seq:${++passthroughToolCallSeq}`;
Expand Down Expand Up @@ -1839,7 +1918,8 @@ export function createSSEStream(options: StreamOptions = {}) {
idFixed ||
needsReserialization ||
toolCallIdCoerced ||
hadNonStringToolCallId
hadNonStringToolCallId ||
hadNonStringTopLevelId
) {
output = `data: ${JSON.stringify(parsed)}\n`;
injectedUsage = true;
Expand Down Expand Up @@ -2098,6 +2178,43 @@ export function createSSEStream(options: StreamOptions = {}) {
updateClaudeEmptyResponseLifecycle(claudeEmptyResponseLifecycle, bufferedPayload);
}
clientPayloadCollector.push(bufferedPayload);

// Normalize numeric IDs for final buffered data: chunk (same as transform path)
if (typeof bufferedPayload === "object" && !Array.isArray(bufferedPayload)) {
const flushedParsed = bufferedPayload as JsonRecord;
const flushedType = typeof flushedParsed.type === "string" ? flushedParsed.type : "";
const isResponses = flushedType.startsWith("response.");
const isClaude = isClaudeEventPayload(flushedParsed);
if (isResponses) {
if (normalizeResponsesSseIds(flushedParsed)) {
output = `data: ${JSON.stringify(flushedParsed)}\n`;
}
} else if (!isClaude) {
let flushChanged = false;
const flushedHadNonStringTopLevelId =
flushedParsed?.id != null && typeof flushedParsed.id !== "string";
if (flushedHadNonStringTopLevelId) {
flushedParsed.id = String(flushedParsed.id);
flushChanged = true;
}
if (Array.isArray(flushedParsed.choices)) {
for (const choice of flushedParsed.choices as JsonRecord[]) {
const tcs = (choice as JsonRecord | undefined)?.delta as JsonRecord | undefined;
if (Array.isArray(tcs?.tool_calls)) {
for (const tc of tcs.tool_calls as JsonRecord[]) {
if (tc?.id != null && typeof tc.id !== "string") {
tc.id = String(tc.id);
flushChanged = true;
}
}
}
}
}
Comment on lines +2200 to +2212

@gemini-code-assist gemini-code-assist Bot Jun 8, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

If choice in flushedParsed.choices is null or undefined, accessing choice.delta directly will throw a TypeError. Using optional chaining (choice as JsonRecord | undefined)?.delta prevents potential crashes during the flush phase when handling malformed upstream choices.

Suggested change
if (Array.isArray(flushedParsed.choices)) {
for (const choice of flushedParsed.choices as JsonRecord[]) {
const tcs = (choice as JsonRecord).delta as JsonRecord | undefined;
if (Array.isArray(tcs?.tool_calls)) {
for (const tc of tcs.tool_calls as JsonRecord[]) {
if (tc?.id != null && typeof tc.id !== "string") {
tc.id = String(tc.id);
flushChanged = true;
}
}
}
}
}
if (Array.isArray(flushedParsed.choices)) {
for (const choice of flushedParsed.choices as JsonRecord[]) {
const tcs = (choice as JsonRecord | undefined)?.delta as JsonRecord | undefined;
if (Array.isArray(tcs?.tool_calls)) {
for (const tc of tcs.tool_calls as JsonRecord[]) {
if (tc?.id != null && typeof tc.id !== "string") {
tc.id = String(tc.id);
flushChanged = true;
}
}
}
}
}

if (flushChanged) {
output = `data: ${JSON.stringify(flushedParsed)}\n`;
}
}
}
}
if (!bufferedLine && pendingPassthroughEventLine && !pendingPassthroughEventEmitted) {
output = `${pendingPassthroughEventLine}\n${output}`;
Expand Down
Loading