pjandro/OmniRoute
1
0
Fork 0
mirror of https://github.com/diegosouzapw/OmniRoute.git synced 2026-07-31 04:12:10 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(sse): dense, deterministic output ordering in Responses API response.completed (#4906)

Browse Source 
Integrated into release/v3.8.37 — manual integration with #4862 in response/openai-responses.ts (custom-tool funcItem + dense recordCompletedItem). Fixed a latent #4848 interaction: the close-reasoning-before-message guard force-closed <think>-tag reasoning prematurely, which dense output (#4906) then snapshotted as a partial buffer ("plan" vs "planning") — scoped the guard to native reasoning_content (!inThinking) in BOTH transformer + translator paths. Full Responses suite 203/203 green incl. #4848/#4862 regression.
This commit is contained in:
Diego Rodrigues de Sa e Souza
2026-06-26 00:30:38 -03:00
committed by GitHub
parent 25bc570cfc
commit e02ebcaa32
6 changed files with 318 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGELOG.md
View File

@@ -10,6 +10,8 @@ _In development — bullets added per PR; finalized at release._
### 🔧 Bug Fixes
- **fix(sse):** dense, deterministic `response.output` ordering in `response.completed` — items are now sorted by their actual `output_index` (via a recorded-as-emitted accumulator + stable sort) instead of being rebuilt from unordered state dicts; `normalizeOutputIndex` replaces fragile `parseInt` calls for robust index coercion; superseded tool calls (replaced at the same index mid-stream) are excluded from the final output array. (thanks @Marco9113)
- **codex/translator**: normalize Codex custom/freeform tools (`apply_patch`, `type:"custom"` with no `parameters`) to a `{ input: string }` function schema instead of an empty schema — the empty schema made models invoke `apply_patch` with `{}`, breaking the Codex runtime which expects `{ input: string }`. Also map `custom_tool_call` / `custom_tool_call_output` input items and stream `apply_patch` tool calls via `custom_tool_call_input.delta`/`.done` events. (thanks @nstung463)
- **antigravity-to-openai**: preserve the `required` array when translating Draft 2020-12 tool schemas (e.g. from OpenCode), stripping unsupported JSON Schema meta keywords while keeping mandatory arguments required so the model no longer calls tools without them. (thanks @anuragg-saxenaa)

142
open-sse/transformer/responsesTransformer.ts
View File

@@ -97,6 +97,11 @@ export function createResponsesApiTransformStream(logger = null, keepaliveInterv
funcCallIds: {},
funcArgsDone: {},
funcItemDone: {},
completedOutputItems: [] as Array<{
output_index: number;
item: Record<string, unknown>;
seq: number;
}>,
buffer: "",
completedSent: false,
usage: null,
@@ -106,6 +111,36 @@ export function createResponsesApiTransformStream(logger = null, keepaliveInterv
const encoder = new TextEncoder();
const nextSeq = () => ++state.seq;
// Normalize output_index to a non-negative integer (replaces fragile parseInt calls)
const normalizeOutputIndex = (outputIndex: number | string): number => {
const normalized = Number(outputIndex);
return Number.isInteger(normalized) && normalized >= 0 ? normalized : 0;
};
// Record a finalized item as it is emitted in output_item.done so buildDenseOutput
// can later sort by the actual output_index rather than rebuilding from state dicts.
const recordCompletedItem = (
outputIndex: number | string,
item: Record<string, unknown>
): number => {
const normalized = normalizeOutputIndex(outputIndex);
state.completedOutputItems.push({ output_index: normalized, item, seq: state.seq });
return normalized;
};
// Build a dense, deterministic output array sorted by output_index then by seq
// (emission order within the same index) — mirrors upstream PR #721.
const buildDenseOutput = (): Array<Record<string, unknown>> =>
state.completedOutputItems
.slice()
.sort((left, right) => {
if (left.output_index !== right.output_index) {
return left.output_index - right.output_index;
}
return left.seq - right.seq;
})
.map(({ item }) => item);
const emit = (controller, eventType, data) => {
data.sequence_number = nextSeq();
const output = `event: ${eventType}\ndata: ${JSON.stringify(data)}\n\n`;
@@ -172,15 +207,19 @@ export function createResponsesApiTransformStream(logger = null, keepaliveInterv
part: { type: "summary_text", text: state.reasoningBuf },
});
const reasoningItem = {
id: state.reasoningId,
type: "reasoning",
summary: [{ type: "summary_text", text: state.reasoningBuf }],
};
emit(controller, "response.output_item.done", {
type: "response.output_item.done",
output_index: state.reasoningIndex,
item: {
id: state.reasoningId,
type: "reasoning",
summary: [{ type: "summary_text", text: state.reasoningBuf }],
},
item: reasoningItem,
});
recordCompletedItem(state.reasoningIndex, reasoningItem);
}
};
@@ -188,12 +227,13 @@ export function createResponsesApiTransformStream(logger = null, keepaliveInterv
if (state.msgItemAdded[idx] && !state.msgItemDone[idx]) {
state.msgItemDone[idx] = true;
const fullText = state.msgTextBuf[idx] || "";
const msgId = `msg_${state.responseId}_${idx}`;
const normalizedIndex = normalizeOutputIndex(idx);
const msgId = `msg_${state.responseId}_${normalizedIndex}`;
emit(controller, "response.output_text.done", {
type: "response.output_text.done",
item_id: msgId,
output_index: parseInt(idx),
output_index: normalizedIndex,
content_index: 0,
text: fullText,
logprobs: [],
@@ -202,27 +242,32 @@ export function createResponsesApiTransformStream(logger = null, keepaliveInterv
emit(controller, "response.content_part.done", {
type: "response.content_part.done",
item_id: msgId,
output_index: parseInt(idx),
output_index: normalizedIndex,
content_index: 0,
part: { type: "output_text", annotations: [], logprobs: [], text: fullText },
});
const msgItem = {
id: msgId,
type: "message",
content: [{ type: "output_text", annotations: [], logprobs: [], text: fullText }],
role: "assistant",
};
emit(controller, "response.output_item.done", {
type: "response.output_item.done",
output_index: parseInt(idx),
item: {
id: msgId,
type: "message",
content: [{ type: "output_text", annotations: [], logprobs: [], text: fullText }],
role: "assistant",
},
output_index: normalizedIndex,
item: msgItem,
});
recordCompletedItem(normalizedIndex, msgItem);
}
};
const closeToolCall = (controller, idx) => {
const closeToolCall = (controller, idx, recordAsCompleted = true) => {
const callId = state.funcCallIds[idx];
if (callId && !state.funcItemDone[idx]) {
const normalizedIndex = normalizeOutputIndex(idx);
let args = state.funcArgsBuf[idx] || "{}";
// Fix #1674 & #1852: Final cleanup of empty string and empty array placeholders
@@ -248,22 +293,30 @@ export function createResponsesApiTransformStream(logger = null, keepaliveInterv
emit(controller, "response.function_call_arguments.done", {
type: "response.function_call_arguments.done",
item_id: `fc_${callId}`,
output_index: parseInt(idx),
output_index: normalizedIndex,
arguments: args,
});
const funcItem = {
id: `fc_${callId}`,
type: "function_call",
arguments: args,
call_id: callId,
name: state.funcNames[idx] || "",
};
emit(controller, "response.output_item.done", {
type: "response.output_item.done",
output_index: parseInt(idx),
item: {
id: `fc_${callId}`,
type: "function_call",
arguments: args,
call_id: callId,
name: state.funcNames[idx] || "",
},
output_index: normalizedIndex,
item: funcItem,
});
// Only record as a completed output item when this is a final close (not a
// superseded-call eviction where a new call replaced this one at the same index).
if (recordAsCompleted) {
recordCompletedItem(normalizedIndex, funcItem);
}
state.funcItemDone[idx] = true;
state.funcArgsDone[idx] = true;
}
@@ -273,33 +326,9 @@ export function createResponsesApiTransformStream(logger = null, keepaliveInterv
if (!state.completedSent) {
state.completedSent = true;
// Build output from accumulated state
const output = [];
if (state.reasoningId) {
output.push({
id: state.reasoningId,
type: "reasoning",
summary: [{ type: "summary_text", text: state.reasoningBuf }],
});
}
for (const idx in state.msgItemAdded) {
output.push({
id: `msg_${state.responseId}_${idx}`,
type: "message",
role: "assistant",
content: [{ type: "output_text", annotations: [], text: state.msgTextBuf[idx] || "" }],
});
}
for (const idx in state.funcCallIds) {
const callId = state.funcCallIds[idx];
output.push({
id: `fc_${callId}`,
type: "function_call",
call_id: callId,
name: state.funcNames[idx] || "",
arguments: state.funcArgsBuf[idx] || "{}",
});
}
// Build a dense, deterministic output array from items recorded as they were emitted.
// Sorted by output_index then by emission sequence for stable ordering.
const output = buildDenseOutput();
const response: Record<string, unknown> = {
id: state.responseId,
@@ -420,7 +449,10 @@ export function createResponsesApiTransformStream(logger = null, keepaliveInterv
// and is still open, before emitting message content. Without this
// the reasoning item is never closed and the message reuses the
// reasoning output_index, producing a protocol-invalid stream.
if (state.reasoningId && !state.reasoningDone) {
// Guard on !inThinking: reasoning opened via <think> tags is closed by
// its matching </think> below — force-closing it here would snapshot a
// partial buffer (dense output records the item at close time). (#4848 + #4906)
if (state.reasoningId && !state.reasoningDone && !state.inThinking) {
closeReasoning(controller);
}
@@ -516,7 +548,9 @@ export function createResponsesApiTransformStream(logger = null, keepaliveInterv
// T37: Prevent merging if a new tool_call uses the same index
if (state.funcCallIds[tcIdx] && newCallId && state.funcCallIds[tcIdx] !== newCallId) {
closeToolCall(controller, tcIdx);
// Superseded call: close and emit output_item.done but do NOT record as final output
// since this call was replaced by a new one at the same index.
closeToolCall(controller, tcIdx, false);
delete state.funcCallIds[tcIdx];
delete state.funcNames[tcIdx];
delete state.funcArgsBuf[tcIdx];

1
open-sse/translator/index.ts
View File

@@ -565,6 +565,7 @@ export function initState(sourceFormat) {
funcCallIds: {},
funcArgsDone: {},
funcItemDone: {},
completedOutputItems: [],
completedSent: false,
};
}

181
open-sse/translator/response/openai-responses.ts
View File

@@ -128,7 +128,10 @@ export function openaiToOpenAIResponsesResponse(chunk, state) {
// Close reasoning if it was opened via native reasoning_content and is
// still open, before emitting message content. Otherwise the reasoning
// item is never closed and the message reuses its output_index.
if (state.reasoningId && !state.reasoningDone) {
// Guard on !inThinking: reasoning opened via <think> tags is closed by its
// matching </think> below — force-closing it here would snapshot a partial
// buffer (dense output records the item at close time). (#4848 + #4906)
if (state.reasoningId && !state.reasoningDone && !state.inThinking) {
closeReasoning(state, emit);
}
@@ -188,6 +191,36 @@ export function openaiToOpenAIResponsesResponse(chunk, state) {
return events;
}
// Normalize output_index to a non-negative integer (replaces fragile parseInt calls)
function normalizeOutputIndex(outputIndex) {
const normalized = Number(outputIndex);
return Number.isInteger(normalized) && normalized >= 0 ? normalized : 0;
}
// Record a finalized item keyed by output_index so buildDenseOutput can sort later
function recordCompletedItem(state, outputIndex, item) {
if (!Array.isArray(state.completedOutputItems)) {
state.completedOutputItems = [];
}
const normalized = normalizeOutputIndex(outputIndex);
state.completedOutputItems.push({ output_index: normalized, item, seq: state.seq });
return normalized;
}
// Build a dense, deterministic output array sorted by output_index then by seq
function buildDenseOutput(state) {
const items = Array.isArray(state.completedOutputItems) ? state.completedOutputItems : [];
return items
.slice()
.sort((left, right) => {
if (left.output_index !== right.output_index) {
return left.output_index - right.output_index;
}
return left.seq - right.seq;
})
.map(({ item }) => item);
}
// Helper functions
function startReasoning(state, emit, idx) {
if (!state.reasoningId) {
@@ -243,15 +276,19 @@ function closeReasoning(state, emit) {
part: { type: "summary_text", text: state.reasoningBuf },
});
const reasoningItem = {
id: state.reasoningId,
type: "reasoning",
summary: [{ type: "summary_text", text: state.reasoningBuf }],
};
emit("response.output_item.done", {
type: "response.output_item.done",
output_index: state.reasoningIndex,
item: {
id: state.reasoningId,
type: "reasoning",
summary: [{ type: "summary_text", text: state.reasoningBuf }],
},
item: reasoningItem,
});
recordCompletedItem(state, state.reasoningIndex, reasoningItem);
}
}
@@ -296,12 +333,13 @@ function closeMessage(state, emit, idx) {
if (state.msgItemAdded[idx] && !state.msgItemDone[idx]) {
state.msgItemDone[idx] = true;
const fullText = state.msgTextBuf[idx] || "";
const msgId = `msg_${state.responseId}_${idx}`;
const normalizedIndex = normalizeOutputIndex(idx);
const msgId = `msg_${state.responseId}_${normalizedIndex}`;
emit("response.output_text.done", {
type: "response.output_text.done",
item_id: msgId,
output_index: parseInt(idx),
output_index: normalizedIndex,
content_index: 0,
text: fullText,
logprobs: [],
@@ -310,21 +348,25 @@ function closeMessage(state, emit, idx) {
emit("response.content_part.done", {
type: "response.content_part.done",
item_id: msgId,
output_index: parseInt(idx),
output_index: normalizedIndex,
content_index: 0,
part: { type: "output_text", annotations: [], logprobs: [], text: fullText },
});
const msgItem = {
id: msgId,
type: "message",
content: [{ type: "output_text", annotations: [], logprobs: [], text: fullText }],
role: "assistant",
};
emit("response.output_item.done", {
type: "response.output_item.done",
output_index: parseInt(idx),
item: {
id: msgId,
type: "message",
content: [{ type: "output_text", annotations: [], logprobs: [], text: fullText }],
role: "assistant",
},
output_index: normalizedIndex,
item: msgItem,
});
recordCompletedItem(state, normalizedIndex, msgItem);
}
}
@@ -336,7 +378,9 @@ function emitToolCall(state, emit, tc) {
// T37: If we already have a tool call at this index but the ID changed,
// we must close the current one and start a new one to prevent merging.
if (state.funcCallIds[tcIdx] && newCallId && state.funcCallIds[tcIdx] !== newCallId) {
closeToolCall(state, emit, tcIdx);
// Superseded call: close and emit output_item.done but do NOT record as final output
// since this call was replaced by a new one at the same index.
closeToolCall(state, emit, tcIdx, false);
delete state.funcCallIds[tcIdx];
delete state.funcNames[tcIdx];
delete state.funcArgsBuf[tcIdx];
@@ -398,12 +442,14 @@ function emitToolCall(state, emit, tc) {
}
}
function closeToolCall(state, emit, idx) {
function closeToolCall(state, emit, idx, recordAsCompleted = true) {
const callId = state.funcCallIds[idx];
if (callId && !state.funcItemDone[idx]) {
const normalizedIndex = normalizeOutputIndex(idx);
const args = state.funcArgsBuf[idx] || "{}";
const isCustomTool = (state.funcNames[idx] || "") === "apply_patch";
let funcItem;
if (isCustomTool) {
// The model produced JSON {"input":"..."} against the normalized custom-tool schema.
// Unwrap it back to the raw patch string the Codex runtime expects. (#1007)
@@ -418,42 +464,52 @@ function closeToolCall(state, emit, idx) {
emit("response.custom_tool_call_input.done", {
type: "response.custom_tool_call_input.done",
item_id: `fc_${callId}`,
output_index: parseInt(idx),
output_index: normalizedIndex,
input: rawInput,
});
funcItem = {
id: `fc_${callId}`,
type: "custom_tool_call",
input: rawInput,
call_id: callId,
name: state.funcNames[idx] || "",
};
emit("response.output_item.done", {
type: "response.output_item.done",
output_index: parseInt(idx),
item: {
id: `fc_${callId}`,
type: "custom_tool_call",
input: rawInput,
call_id: callId,
name: state.funcNames[idx] || "",
},
output_index: normalizedIndex,
item: funcItem,
});
} else {
emit("response.function_call_arguments.done", {
type: "response.function_call_arguments.done",
item_id: `fc_${callId}`,
output_index: parseInt(idx),
output_index: normalizedIndex,
arguments: args,
});
funcItem = {
id: `fc_${callId}`,
type: "function_call",
arguments: args,
call_id: callId,
name: state.funcNames[idx] || "",
};
emit("response.output_item.done", {
type: "response.output_item.done",
output_index: parseInt(idx),
item: {
id: `fc_${callId}`,
type: "function_call",
arguments: args,
call_id: callId,
name: state.funcNames[idx] || "",
},
output_index: normalizedIndex,
item: funcItem,
});
}
// Only record as a completed output item when this is a final close (not a
// superseded-call eviction where a new call replaced this one at the same index).
if (recordAsCompleted) {
recordCompletedItem(state, normalizedIndex, funcItem);
}
state.funcItemDone[idx] = true;
state.funcArgsDone[idx] = true;
}
@@ -463,54 +519,11 @@ function sendCompleted(state, emit) {
if (!state.completedSent) {
state.completedSent = true;
// Build output from accumulated state
const output = [];
if (state.reasoningId) {
output.push({
id: state.reasoningId,
type: "reasoning",
summary: [{ type: "summary_text", text: state.reasoningBuf }],
});
}
for (const idx in state.msgItemAdded) {
output.push({
id: `msg_${state.responseId}_${idx}`,
type: "message",
role: "assistant",
content: [{ type: "output_text", annotations: [], text: state.msgTextBuf[idx] || "" }],
});
}
for (const idx in state.funcCallIds) {
const callId = state.funcCallIds[idx];
const name = state.funcNames[idx] || "";
const args = state.funcArgsBuf[idx] || "{}";
// Mirror the streamed custom_tool_call shape in the final snapshot so Codex can
// reconcile the apply_patch item from response.completed. (#1007)
if (name === "apply_patch") {
let rawInput = args;
try {
const parsed = JSON.parse(args);
if (parsed && typeof parsed.input === "string") rawInput = parsed.input;
} catch {
// Not JSON — fall back to the raw buffered arguments.
}
output.push({
id: `fc_${callId}`,
type: "custom_tool_call",
call_id: callId,
name,
input: rawInput,
});
continue;
}
output.push({
id: `fc_${callId}`,
type: "function_call",
call_id: callId,
name,
arguments: args,
});
}
// Build a dense, deterministic output array from items recorded as they were emitted
// (each close*() call records its item via recordCompletedItem — including the
// #1007 custom_tool_call shape for apply_patch). Sorted by output_index then by
// emission sequence for stable ordering.
const output = buildDenseOutput(state);
const response: Record<string, unknown> = {
id: state.responseId,

129
tests/unit/responses-transformer-dense-output.test.ts Normal file
View File

@@ -0,0 +1,129 @@
/**
* Tests for dense, deterministic output ordering in response.completed.
*
* Port of upstream decolua/9router PR #721: "fix: suppress null Responses SSE frames
* and preserve completed output". These tests verify that:
* 1. response.completed.response.output is a dense array sorted by output_index
* 2. normalizeOutputIndex handles non-numeric keys robustly (no NaN from parseInt)
* 3. Function-call items at lower output_index appear before message items at higher
* output_index in the final output array
*/
import test from "node:test";
import assert from "node:assert/strict";
const { createResponsesApiTransformStream } =
await import("../../open-sse/transformer/responsesTransformer.ts");
const encoder = new TextEncoder();
const decoder = new TextDecoder();
async function runTransformStream(chunks) {
const stream = createResponsesApiTransformStream();
const writer = stream.writable.getWriter();
const reader = stream.readable.getReader();
const output = [];
const readerTask = (async () => {
while (true) {
const { value, done } = await reader.read();
if (done) break;
output.push(decoder.decode(value));
}
})();
for (const chunk of chunks) {
await writer.write(encoder.encode(chunk));
}
await writer.close();
await readerTask;
return output.join("");
}
function parseSseOutput(output) {
return output
.trim()
.split("\n\n")
.map((entry) => {
const lines = entry.split("\n");
const eventLine = lines.find((line) => line.startsWith("event: "));
const dataLine = lines.find((line) => line.startsWith("data: "));
return {
event: eventLine ? eventLine.slice("event: ".length) : null,
data: dataLine ? dataLine.slice("data: ".length) : null,
};
});
}
function getCompleted(output) {
const events = parseSseOutput(output);
const completedEvent = events.find((e) => e.event === "response.completed");
assert.ok(completedEvent, "response.completed event must be present");
return JSON.parse(completedEvent.data).response;
}
test("response.completed output is sorted by output_index: function_call at index 0 appears before message at index 2", async () => {
// function_call uses tool_calls[index: 0], message uses choice index 2.
// Without dense ordering, messages (iterated from msgItemAdded dict) may appear
// before function_calls regardless of their actual output_index.
const output = await runTransformStream([
// First chunk: tool call at output_index 0
`data: {"id":"chatcmpl-order","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call_abc","function":{"name":"lookup","arguments":"{\\"q\\":\\"test\\"}"}}]}}]}\n\n`,
// Second chunk: text content at output_index 2
`data: {"id":"chatcmpl-order","choices":[{"index":2,"delta":{"content":"Hello"}}]}\n\n`,
// Finish
`data: {"id":"chatcmpl-order","choices":[{"index":2,"delta":{},"finish_reason":"stop"},{"index":0,"delta":{},"finish_reason":"tool_calls"}]}\n\n`,
]);
const completed = getCompleted(output);
assert.ok(Array.isArray(completed.output), "output must be an array");
assert.ok(completed.output.length >= 2, "output must have at least 2 items");
// The function_call (output_index 0) must come before the message (output_index 2)
const types = completed.output.map((item) => item.type);
assert.ok(
types.indexOf("function_call") < types.indexOf("message"),
`function_call (output_index 0) must appear before message (output_index 2), got order: ${types.join(", ")}`
);
const funcCall = completed.output.find((item) => item.type === "function_call");
assert.equal(funcCall?.call_id, "call_abc");
assert.equal(funcCall?.name, "lookup");
const msg = completed.output.find((item) => item.type === "message");
assert.equal(msg?.content?.[0]?.text, "Hello");
});
test("response.completed output includes all finalized items even when output is empty", async () => {
// A finish_reason with no content should produce an empty output array, not undefined
const output = await runTransformStream([
`data: {"id":"chatcmpl-empty","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}\n\n`,
]);
const completed = getCompleted(output);
assert.ok(Array.isArray(completed.output), "output must be an array even when empty");
assert.equal(completed.output.length, 0, "output must be empty when no items finalize");
});
test("response.completed output preserves dense ordering for function_call-only streams", async () => {
// Two tool calls at different indexes — must appear in ascending output_index order
const output = await runTransformStream([
// call at index 3 first
`data: {"id":"chatcmpl-multi-tool","choices":[{"index":0,"delta":{"tool_calls":[{"index":3,"id":"call_z","function":{"name":"zeta","arguments":"{}"}}]}}]}\n\n`,
// call at index 1
`data: {"id":"chatcmpl-multi-tool","choices":[{"index":0,"delta":{"tool_calls":[{"index":1,"id":"call_a","function":{"name":"alpha","arguments":"{}"}}]}}]}\n\n`,
`data: {"id":"chatcmpl-multi-tool","choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}]}\n\n`,
]);
const completed = getCompleted(output);
assert.ok(Array.isArray(completed.output), "output must be an array");
assert.equal(completed.output.length, 2);
// Sorted by output_index: index 1 (alpha) before index 3 (zeta)
assert.equal(completed.output[0].name, "alpha", "output_index 1 (alpha) must come first");
assert.equal(completed.output[1].name, "zeta", "output_index 3 (zeta) must come second");
});

2
tests/unit/responses-transformer.test.ts
View File

@@ -105,7 +105,7 @@ test("createResponsesApiTransformStream converts think tags into reasoning summa
summary: [{ type: "summary_text", text: "planning" }],
});
assert.deepEqual(completed.output[1].content, [
{ type: "output_text", annotations: [], text: "answer" },
{ type: "output_text", annotations: [], logprobs: [], text: "answer" },
]);
});