From 07e1011d3b8be31edbb7f7faf2620448dd89f05b Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza <8016841+diegosouzapw@users.noreply.github.com> Date: Wed, 15 Jul 2026 23:49:22 -0300 Subject: [PATCH] fix(stream): reconcile encrypted Codex reasoning visibility without mutating upstream item (#7304) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(stream): reconcile encrypted Codex reasoning visibility without mutating upstream item Resolves the collision between two open PRs on ensureVisibleResponsesReasoningSummary: #7095 (xz-dev) found that chat clients see nothing when Codex exposes reasoning only as encrypted_content, and added a visible placeholder — but did so by mutating item.summary in place. #7176 (JxnLexn) found that same mutation corrupts the forwarded response item, discarding the encrypted_content shape Codex needs for follow-up requests, and removed the mutation — but that also silently dropped the placeholder, so chat clients went back to seeing nothing. The mutation existed only so a later line could read the summary text back off the same item. getVisibleResponsesReasoningSummaryText() computes that text without touching the item, so: - synthetic response.reasoning_summary_text.delta / .part.done events still carry the placeholder for chat clients (#7095's goal), and - the forwarded response.output_item.done payload keeps its original encrypted_content intact with no fabricated summary field (#7176's goal). Applied at both call sites #7095 identified: the native Responses passthrough in stream.ts/passthroughTailProcessor.ts, and the Responses-to-Chat-Completions translator in openai-responses.ts. Closes #7095, closes #7176. Co-authored-by: Xiangzhe Co-authored-by: Jan Leon * test(stream): guard the encrypted-reasoning mutation via the completed backfill path The output_item.done line is echoed verbatim on the wire, so a re-introduced item.summary mutation does NOT surface in that event — verified by re-injecting the mutation, which left the existing assertion green. The mutation does surface in the response.completed snapshot, where the captured reasoning item is re-serialized when upstream sends an empty output (store: false). Adds that case, which fails as expected when the mutation is re-introduced, making the #7176 half of the reconciliation an enforced regression guard rather than an incidental property of the current code path. Co-authored-by: Xiangzhe Co-authored-by: Jan Leon --------- Co-authored-by: Xiangzhe Co-authored-by: Jan Leon --- .../translator/response/openai-responses.ts | 8 +- .../response/openai-responses/pureHelpers.ts | 25 ++ open-sse/utils/passthroughTailProcessor.ts | 5 - open-sse/utils/stream.ts | 56 +--- .../codex-chat-reasoning-http-e2e.test.ts | 302 ++++++++++++++++++ tests/unit/stream-utilities.test.ts | 78 ++++- 6 files changed, 414 insertions(+), 60 deletions(-) create mode 100644 tests/integration/codex-chat-reasoning-http-e2e.test.ts diff --git a/open-sse/translator/response/openai-responses.ts b/open-sse/translator/response/openai-responses.ts index 42fca56462..fdd214d06d 100644 --- a/open-sse/translator/response/openai-responses.ts +++ b/open-sse/translator/response/openai-responses.ts @@ -12,7 +12,7 @@ import { stripEmptyOptionalToolArgs, normalizeOutputIndex, normalizeUpstreamFailure, - extractResponsesReasoningSummaryText, + getVisibleResponsesReasoningSummaryText, } from "./openai-responses/pureHelpers.ts"; import { createEventEmitter } from "./openai-responses/eventEmitter.ts"; @@ -1070,7 +1070,11 @@ function openaiResponsesToOpenAIResponseStream(chunk, state) { !(state.reasoningItemsWithDelta instanceof Set && state.reasoningItemsWithDelta.size > 0); if (emittedForItem || emittedWithoutItemId) return null; - const summaryText = extractResponsesReasoningSummaryText(item); + // #7095/#7176 reconciliation: computed WITHOUT mutating `item`, so an + // encrypted-only reasoning item (and its `encrypted_content`) is never + // rewritten with a fabricated `summary` — the placeholder only feeds this + // synthetic client-facing delta chunk. + const summaryText = getVisibleResponsesReasoningSummaryText(item); if (!summaryText) return null; return buildResponsesReasoningDeltaChunk(state, summaryText); } diff --git a/open-sse/translator/response/openai-responses/pureHelpers.ts b/open-sse/translator/response/openai-responses/pureHelpers.ts index 094d7dc58d..50e9cfe6c6 100644 --- a/open-sse/translator/response/openai-responses/pureHelpers.ts +++ b/open-sse/translator/response/openai-responses/pureHelpers.ts @@ -163,3 +163,28 @@ export function extractResponsesReasoningSummaryText(item) { ) .join(""); } + +// #7095/#7176 — when Codex exposes a reasoning item only as encrypted private +// reasoning (no plaintext summary), chat clients would otherwise see nothing in +// their thinking panel. Reconciles two goals that used to be in tension: +// - #7095 wants a visible placeholder in the chat client. +// - #7176 wants the upstream response item left untouched, so `encrypted_content` +// (needed by Codex for subsequent requests) is never overwritten by a +// fabricated `summary`. +// This function computes the placeholder text WITHOUT mutating `item` — callers +// use the returned text for synthetic client-facing events only. +const ENCRYPTED_REASONING_PLACEHOLDER = + "Codex is reasoning, but the upstream Responses API exposed this reasoning block only as encrypted private reasoning. OmniRoute cannot recover the plaintext."; + +export function getVisibleResponsesReasoningSummaryText(item) { + const existingSummary = extractResponsesReasoningSummaryText(item); + if (existingSummary) return existingSummary; + + const hasEncryptedReasoning = + item && + item.type === "reasoning" && + typeof item.encrypted_content === "string" && + item.encrypted_content.length > 0; + + return hasEncryptedReasoning ? ENCRYPTED_REASONING_PLACEHOLDER : ""; +} diff --git a/open-sse/utils/passthroughTailProcessor.ts b/open-sse/utils/passthroughTailProcessor.ts index 82b57da8aa..845bc6e352 100644 --- a/open-sse/utils/passthroughTailProcessor.ts +++ b/open-sse/utils/passthroughTailProcessor.ts @@ -37,7 +37,6 @@ export type PassthroughTailProcessorContext = { appendPassthroughReasoning: (value: string) => void; getResponsesReasoningKey: (payload: Record) => string | null; markResponsesReasoningSummarySeen: (key: string) => void; - ensureVisibleResponsesReasoningSummary: (payload: Record) => boolean; emitSyntheticResponsesReasoningSummary: (payload: Record) => void; passthroughResponsesOutputItems: unknown[]; passthroughResponsesPendingFunctionCalls: Map; @@ -136,12 +135,8 @@ function handleResponsesTailPayload( } } if (parsed.type === "response.output_item.done" && parsed.item) { - const reasoningSummaryInjected = context.ensureVisibleResponsesReasoningSummary(parsed); context.emitSyntheticResponsesReasoningSummary(parsed); pushUniqueResponsesOutputItems(context.passthroughResponsesOutputItems, [parsed.item]); - if (reasoningSummaryInjected) { - output = `data: ${JSON.stringify(parsed)}\n\n`; - } const item = asRecord(parsed.item); if (item.type === "function_call") { const pendingKey = getFunctionCallPendingKey(item); diff --git a/open-sse/utils/stream.ts b/open-sse/utils/stream.ts index 581f1f6342..974d532e9e 100644 --- a/open-sse/utils/stream.ts +++ b/open-sse/utils/stream.ts @@ -52,6 +52,7 @@ import { stripResponsesLifecycleEcho, } from "./responsesStreamHelpers.ts"; import { processBufferedPassthroughLine } from "./passthroughTailProcessor.ts"; +import { getVisibleResponsesReasoningSummaryText } from "../translator/response/openai-responses/pureHelpers.ts"; import { getAnyReasoningValue, getReadableReasoningValue, @@ -1006,49 +1007,6 @@ export function createSSEStream(options: StreamOptions = {}) { return responseId !== null && outputIndex !== null ? `${responseId}:${outputIndex}` : null; }; - const getResponsesReasoningSummaryText = (item: Record): string => { - return Array.isArray(item.summary) - ? item.summary - .map((part) => { - if (!part || typeof part !== "object" || Array.isArray(part)) { - return ""; - } - return typeof (part as Record).text === "string" - ? ((part as Record).text as string) - : ""; - }) - .join("") - : ""; - }; - - const ensureVisibleResponsesReasoningSummary = (payload: Record): boolean => { - const item = - payload.item && typeof payload.item === "object" && !Array.isArray(payload.item) - ? (payload.item as Record) - : null; - if (!item || item.type !== "reasoning") { - return false; - } - - if (getResponsesReasoningSummaryText(item)) { - return false; - } - - const hasEncryptedReasoning = - typeof item.encrypted_content === "string" && item.encrypted_content.length > 0; - if (!hasEncryptedReasoning) { - return false; - } - - item.summary = [ - { - type: "summary_text", - text: "Codex is reasoning, but the upstream Responses API exposed this reasoning block only as encrypted state. OmniRoute cannot recover the private reasoning text.", - }, - ]; - return true; - }; - const emitSyntheticResponsesReasoningSummary = ( controller: TransformStreamDefaultController, payload: Record @@ -1061,8 +1019,10 @@ export function createSSEStream(options: StreamOptions = {}) { return; } - ensureVisibleResponsesReasoningSummary(payload); - const visibleSummary = getResponsesReasoningSummaryText(item); + // #7095/#7176 reconciliation: compute the visible placeholder WITHOUT + // mutating `item` — the encrypted reasoning item (and its `encrypted_content`, + // required by Codex for subsequent requests) is forwarded to the client intact. + const visibleSummary = getVisibleResponsesReasoningSummaryText(item); if (!visibleSummary) { return; @@ -1485,13 +1445,8 @@ export function createSSEStream(options: StreamOptions = {}) { // response.completed snapshot can be backfilled when upstream // returns an empty `output` (happens with store: false). if (parsed.type === "response.output_item.done" && parsed.item) { - const reasoningSummaryInjected = ensureVisibleResponsesReasoningSummary(parsed); emitSyntheticResponsesReasoningSummary(controller, parsed); pushUniqueResponsesOutputItems(passthroughResponsesOutputItems, [parsed.item]); - if (reasoningSummaryInjected) { - output = `data: ${JSON.stringify(parsed)}\n\n`; - injectedUsage = true; - } if (parsed.item?.type === "function_call") { const pendingKey = typeof parsed.item.id === "string" @@ -2181,7 +2136,6 @@ export function createSSEStream(options: StreamOptions = {}) { markResponsesReasoningSummarySeen: (key: string) => { passthroughResponsesReasoningSummarySeen.add(key); }, - ensureVisibleResponsesReasoningSummary, emitSyntheticResponsesReasoningSummary: (payload: Record) => emitSyntheticResponsesReasoningSummary(controller, payload), passthroughResponsesOutputItems, diff --git a/tests/integration/codex-chat-reasoning-http-e2e.test.ts b/tests/integration/codex-chat-reasoning-http-e2e.test.ts new file mode 100644 index 0000000000..02b9fc9aae --- /dev/null +++ b/tests/integration/codex-chat-reasoning-http-e2e.test.ts @@ -0,0 +1,302 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import fsp from "node:fs/promises"; +import http from "node:http"; +import os from "node:os"; +import path from "node:path"; +import { once } from "node:events"; + +const CODEX_RESPONSES_URL = "https://chatgpt.com/backend-api/codex/responses"; +const ENCRYPTED_CONTENT_SENTINEL = "encrypted-codex-state:" + "A".repeat(910); +const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-codex-chat-http-")); + +process.env.DATA_DIR = TEST_DATA_DIR; +process.env.DISABLE_SQLITE_AUTO_BACKUP = "true"; +process.env.API_KEY_SECRET = "codex-chat-http-e2e-secret-123456"; +process.env.REQUIRE_API_KEY = "false"; +process.env.OMNIROUTE_LOG_REQUEST_SHAPE = "0"; + +const core = await import("../../src/lib/db/core.ts"); +const providersDb = await import("../../src/lib/db/providers.ts"); +const chatRoute = await import("../../src/app/api/v1/chat/completions/route.ts"); + +const originalFetch = globalThis.fetch; + +type RecordedRequest = { + url: string; + method: string; + body: Record; +}; + +function responsesEvents() { + const response = { + id: "resp_reasoning_http", + object: "response", + status: "in_progress", + model: "gpt-5.6-sol", + output: [], + }; + return [ + { type: "response.created", response }, + { + type: "response.output_item.added", + output_index: 0, + item: { + id: "rs_reasoning_http", + type: "reasoning", + encrypted_content: ENCRYPTED_CONTENT_SENTINEL, + summary: [], + }, + }, + { + type: "response.output_item.done", + output_index: 0, + item: { + id: "rs_reasoning_http", + type: "reasoning", + encrypted_content: ENCRYPTED_CONTENT_SENTINEL, + summary: [], + }, + }, + { + type: "response.output_item.added", + output_index: 1, + item: { id: "msg_reasoning_http", type: "message", role: "assistant", content: [] }, + }, + { + type: "response.content_part.added", + item_id: "msg_reasoning_http", + output_index: 1, + content_index: 0, + part: { type: "output_text", text: "", annotations: [] }, + }, + { + type: "response.output_text.delta", + item_id: "msg_reasoning_http", + output_index: 1, + content_index: 0, + delta: "The answer is 42.", + }, + { + type: "response.output_text.done", + item_id: "msg_reasoning_http", + output_index: 1, + content_index: 0, + text: "The answer is 42.", + }, + { + type: "response.output_item.done", + output_index: 1, + item: { + id: "msg_reasoning_http", + type: "message", + role: "assistant", + content: [{ type: "output_text", text: "The answer is 42.", annotations: [] }], + }, + }, + { + type: "response.completed", + response: { + ...response, + status: "completed", + output: [ + { + id: "rs_reasoning_http", + type: "reasoning", + summary: [{ type: "summary_text", text: "I checked the contract. " }], + }, + { + id: "msg_reasoning_http", + type: "message", + role: "assistant", + content: [{ type: "output_text", text: "The answer is 42.", annotations: [] }], + }, + ], + usage: { input_tokens: 8, output_tokens: 9, total_tokens: 17 }, + }, + }, + ]; +} + +function mockResponsesSse() { + const nativeFraming = process.env.CODEX_NATIVE_EVENT_FRAMING === "1"; + return responsesEvents() + .map((event) => { + const eventLine = nativeFraming ? `event: ${event.type}\n` : ""; + return `${eventLine}data: ${JSON.stringify(event)}\n\n`; + }) + .join(""); +} + +async function readIncomingBody(request: http.IncomingMessage) { + const chunks: Buffer[] = []; + for await (const chunk of request) + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); + return Buffer.concat(chunks); +} + +async function bridgeRouteResponse(response: Response, outgoing: http.ServerResponse) { + outgoing.writeHead(response.status, Object.fromEntries(response.headers.entries())); + if (!response.body) { + outgoing.end(); + return; + } + + const reader = response.body.getReader(); + try { + while (true) { + const { done, value } = await reader.read(); + if (done) break; + if (!outgoing.write(value)) await once(outgoing, "drain"); + } + outgoing.end(); + } finally { + reader.releaseLock(); + } +} + +async function startRouteServer() { + const server = http.createServer(async (incoming, outgoing) => { + try { + if (incoming.method !== "POST" || incoming.url !== "/v1/chat/completions") { + outgoing.writeHead(404).end(); + return; + } + + const body = await readIncomingBody(incoming); + const address = server.address(); + assert(address && typeof address !== "string"); + const headers = new Headers(); + for (const [name, value] of Object.entries(incoming.headers)) { + if (Array.isArray(value)) value.forEach((item) => headers.append(name, item)); + else if (value !== undefined) headers.set(name, value); + } + const request = new Request(`http://127.0.0.1:${address.port}${incoming.url}`, { + method: incoming.method, + headers, + body, + }); + await bridgeRouteResponse(await chatRoute.POST(request), outgoing); + } catch (error) { + outgoing.writeHead(500, { "content-type": "text/plain" }); + outgoing.end(error instanceof Error ? error.stack : String(error)); + } + }); + + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", resolve); + }); + const address = server.address(); + assert(address && typeof address !== "string"); + return { server, url: `http://127.0.0.1:${address.port}/v1/chat/completions` }; +} + +function parseSse(raw: string) { + return raw + .split(/\n\n+/) + .map((block) => + block + .split("\n") + .find((line) => line.startsWith("data: ")) + ?.slice(6) + ) + .filter((data): data is string => Boolean(data)); +} + +async function closeServer(server: http.Server) { + if (!server.listening) return; + await new Promise((resolve, reject) => + server.close((error) => (error ? reject(error) : resolve())) + ); +} + +test("chat completions streams Codex Responses reasoning through real route HTTP", async () => { + const recorded: RecordedRequest[] = []; + let routeServer: http.Server | undefined; + + try { + await providersDb.createProviderConnection({ + provider: "codex", + authType: "oauth", + name: "codex-http-reasoning", + email: "codex-http@example.test", + accessToken: "mock-codex-access-token", + refreshToken: "mock-codex-refresh-token", + tokenType: "Bearer", + expiresAt: new Date(Date.now() + 60 * 60 * 1000).toISOString(), + isActive: true, + testStatus: "active", + providerSpecificData: {}, + }); + + const routeHarness = await startRouteServer(); + routeServer = routeHarness.server; + + globalThis.fetch = async (input, init) => { + const request = input instanceof Request ? input : new Request(input, init); + if (request.url !== CODEX_RESPONSES_URL) { + throw new Error(`Unexpected external fetch in Codex HTTP test: ${request.url}`); + } + recorded.push({ + url: request.url, + method: request.method, + body: JSON.parse(await request.text()) as Record, + }); + return new Response(mockResponsesSse(), { + status: 200, + headers: { "content-type": "text/event-stream; charset=utf-8" }, + }); + }; + + const response = await originalFetch(routeHarness.url, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: "codex/gpt-5.6-sol", + stream: true, + reasoning_effort: "high", + messages: [{ role: "user", content: "What is the answer?" }], + }), + }); + const raw = await response.text(); + + assert.equal(response.status, 200, raw); + assert.match(response.headers.get("content-type") ?? "", /^text\/event-stream/); + assert.equal(recorded.length, 1); + assert.equal(recorded[0].url, CODEX_RESPONSES_URL); + assert.equal(recorded[0].method, "POST"); + assert.deepEqual(recorded[0].body.reasoning, { effort: "high", summary: "auto" }); + assert.deepEqual(recorded[0].body.input, [ + { + type: "message", + role: "user", + content: [{ type: "input_text", text: "What is the answer?" }], + }, + ]); + + const chunks = parseSse(raw); + assert.equal(chunks.at(-1), "[DONE]"); + const payloads = chunks.slice(0, -1).map((chunk) => JSON.parse(chunk)); + const reasoningContentDeltas = payloads + .map((payload) => payload.choices?.[0]?.delta?.reasoning_content) + .filter((content): content is string => Boolean(content)); + assert.equal(reasoningContentDeltas.length, 1); + const reasoningContent = reasoningContentDeltas.join(""); + assert.match(reasoningContent, /encrypted (?:state|private reasoning)/i); + assert(!raw.includes(ENCRYPTED_CONTENT_SENTINEL), raw); + assert(!reasoningContent.includes(ENCRYPTED_CONTENT_SENTINEL), reasoningContent); + assert( + payloads.some((payload) => payload.choices?.[0]?.delta?.content === "The answer is 42.") + ); + assert(!raw.includes("response.reasoning_summary_text.delta"), raw); + assert(!raw.includes('"type":"error"'), raw); + assert(!raw.includes('"error"'), raw); + } finally { + globalThis.fetch = originalFetch; + if (routeServer) await closeServer(routeServer); + core.closeDbInstance({ checkpointMode: null }); + await fsp.rm(TEST_DATA_DIR, { recursive: true, force: true }); + } +}); diff --git a/tests/unit/stream-utilities.test.ts b/tests/unit/stream-utilities.test.ts index 0f3404869c..33a1b2bdd7 100644 --- a/tests/unit/stream-utilities.test.ts +++ b/tests/unit/stream-utilities.test.ts @@ -133,7 +133,16 @@ test("createPassthroughStreamWithLogger synthesizes reasoning summary events fro assert.match(result, /event: response\.output_item\.done/); }); -test("createPassthroughStreamWithLogger shows a placeholder for encrypted reasoning items", async () => { +// Reconciles #7095 (xz-dev — chat clients never saw ANY signal that Codex was +// reasoning when the upstream Responses API exposed only encrypted private +// reasoning) with #7176 (JxnLexn — mutating `item.summary` with a fabricated +// placeholder corrupted the response item forwarded downstream, discarding the +// `encrypted_content` shape Codex needs for follow-up requests). Both goals are +// satisfied simultaneously: the client-facing synthetic delta/part events still +// carry the placeholder text, but the forwarded `response.output_item.done` +// payload is untouched — `encrypted_content` survives and no `summary` is +// fabricated onto the wire item. +test("createPassthroughStreamWithLogger shows a placeholder for encrypted reasoning items without mutating the forwarded item", async () => { const transform = createPassthroughStreamWithLogger( "codex", null, @@ -169,12 +178,77 @@ test("createPassthroughStreamWithLogger shows a placeholder for encrypted reason result += decoder.decode(value); } + // #7095: chat clients still see the placeholder via the synthetic events. assert.match(result, /event: response\.reasoning_summary_text\.delta/); assert.match(result, /Codex is reasoning/); - assert.match(result, /"summary":\[\{"type":"summary_text","text":"Codex is reasoning/); + + // #7176: the forwarded response.output_item.done payload is untouched — + // encrypted_content survives intact and no fabricated `summary` is present. + assert.match(result, /"encrypted_content":"enc_opaque_state"/); + assert.doesNotMatch(result, /"summary":/); assert.match(result, /event: response\.output_item\.done/); }); +// Companion guard for the test above. The output_item.done line is echoed +// verbatim, so a re-introduced `item.summary` mutation would NOT surface there. +// It does surface here: the reasoning item is captured into +// passthroughResponsesOutputItems and re-serialized into the response.completed +// snapshot when upstream sends an empty `output` (store: false). This is the +// path that actually fails if the mutation comes back — it is what makes the +// #7176 half of the reconciliation enforceable rather than incidental. +test("createPassthroughStreamWithLogger backfills completed output with encrypted reasoning unmutated", async () => { + const transform = createPassthroughStreamWithLogger( + "codex", + null, + null, + "gpt-5.5-low", + null, + null, + null, + null, + null, + "openai-responses" + ); + + const writer = transform.writable.getWriter(); + await writer.write( + new TextEncoder().encode( + [ + "event: response.output_item.done", + 'data: {"type":"response.output_item.done","response_id":"resp_reasoning_3","output_index":0,"item":{"id":"rs_resp_reasoning_3_0","type":"reasoning","encrypted_content":"enc_opaque_state"}}', + "", + "event: response.completed", + 'data: {"type":"response.completed","response":{"id":"resp_reasoning_3","model":"gpt-5.5-low","status":"completed","output":[],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}', + "", + ].join("\n") + ) + ); + await writer.close(); + + const reader = transform.readable.getReader(); + const decoder = new TextDecoder(); + let result = ""; + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + result += decoder.decode(value); + } + + // The placeholder still reached the client (#7095). + assert.match(result, /event: response\.reasoning_summary_text\.delta/); + assert.match(result, /Codex is reasoning/); + + // The item re-serialized into the completed snapshot carries the original + // encrypted_content and never a fabricated summary (#7176). + const completed = result + .split(/\n\n+/) + .find((block) => block.includes('"type":"response.completed"')); + assert.ok(completed, "expected a response.completed event on the wire"); + assert.match(completed, /"encrypted_content":"enc_opaque_state"/); + assert.doesNotMatch(completed, /"summary":/); +}); + test("createPassthroughStreamWithLogger backfills completed output from function_call arguments events", async () => { const transform = createPassthroughStreamWithLogger( "codex",