From fe73a3ad35c62ae8ef3dd1a7ca3c42b467f039e7 Mon Sep 17 00:00:00 2001 From: "Bob.Hou" Date: Fri, 11 Sep 2026 19:45:24 -0400 Subject: [PATCH] fix(responses): wrap forced-non-streaming web_search JSON as SSE (#13050) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Nasty shape: HTTP 200, upstream billed, and Codex times out waiting for `response.completed` that never comes because the body is JSON. Good catch that wiring the wrap in `handleResponsesCore` would not have fixed it — production goes through `handleChat`, and that helper is tests-only. Stamping the client's SSE intent before flipping `stream` off is the right place to remember it. Reconciled against the tip after the batch landed: only the file-size baseline conflicted, resolved to the batch's rebaselined value. 5/5 afterwards. --- Validated in one consolidated worktree cut from `release/v3.8.51`, boarded with the other 19 PRs of this batch. Two in-batch conflicts, both additive and resolved by keeping each side: the `ENVIRONMENT.md` table (#13035 + #13011) and the `chatHelpers.ts` import block (#12975 on the tip + #13017). - `typecheck:core` clean; `check:dashboard-typecheck` OK (206 pre-existing, within baseline); `check:changelog-integrity` OK; `check:docs-counts` migrations ✓ - complexity 2816 / baseline 3218 and cognitive-complexity 1271 / baseline 1437 — both under baseline - 531 of 532 focused assertions green across the batch's 46 test files - `check-file-size` rebaselined for the batch's real growth (annotation `_rebaseline_2026_09_11_mergebatch_v3851_houminxi`, landed on #13038), attributed per PR The single red is **not this batch**: `tests/unit/combo/quota-weighted-strategy.test.ts` → "A/B isolation: 7 hard-empty + 2 at 0.5% + 1 at 40%, floor=1" asserts an order between two connections of identical weight and flakes on the pure tip too — 2 failures in 4 runs at `origin/release/v3.8.51` with nothing from this batch applied. ⚠️ base-red inherited: #12732 — `Docs Gates`, `Merge integrity`, `No new ESLint warnings`, `Unit Tests fast-path` and `Fast Quality Gates` reproduce on the pure tip (provider count 356 vs the 358 the modules define, SKILL.md drift, and `open-sse/utils/stream.ts` at 3115 > frozen 3098, untouched here). Thanks @HouMinXi — the live evidence on these (X500 logs, `storage.sqlite` state, real `/v1/models` probes, the 36-minute outage write-up) is what let a 20-PR batch be reviewed as a unit. --- .../fixes/13050-responses-websearch-sse.md | 1 + config/quality/file-size-baseline.json | 3 +- open-sse/handlers/chatCore.ts | 10 ++- .../handlers/chatCore/responsesJsonToSse.ts | 65 ++++++++++++++++ tests/unit/8395-plugin-hooks-fire.test.ts | 6 +- .../unit/responses-json-to-sse-13033.test.ts | 78 +++++++++++++++++++ 6 files changed, 156 insertions(+), 7 deletions(-) create mode 100644 changelog.d/fixes/13050-responses-websearch-sse.md create mode 100644 open-sse/handlers/chatCore/responsesJsonToSse.ts create mode 100644 tests/unit/responses-json-to-sse-13033.test.ts diff --git a/changelog.d/fixes/13050-responses-websearch-sse.md b/changelog.d/fixes/13050-responses-websearch-sse.md new file mode 100644 index 0000000000..884f25b5ef --- /dev/null +++ b/changelog.d/fixes/13050-responses-websearch-sse.md @@ -0,0 +1 @@ +- **fix(responses):** wrap forced-non-streaming web_search fallback JSON as Responses SSE so Codex still sees `response.completed` ([#13050](https://github.com/diegosouzapw/OmniRoute/pull/13050)) diff --git a/config/quality/file-size-baseline.json b/config/quality/file-size-baseline.json index 20b6c9f0aa..35e942854b 100644 --- a/config/quality/file-size-baseline.json +++ b/config/quality/file-size-baseline.json @@ -664,5 +664,6 @@ "_rebaseline_2026_09_07_chatcore_nonstreaming_regression_fixes": "Own growth: open-sse/handlers/chatCore.ts 5984->6021 (+37). Two of my own PRs on top of #12867: #12963 pins the ok variant of the non-streaming leg result in its own binding (the discriminated-union narrowing was lost across the tool-loop reassignment, 13 TS2339 under tsconfig.typecheck-api.json), and #12990 restores four behaviours the same refactor dropped — abort classification through isLocalStreamLifecycleError, the omitted synthetic clientResponse, the claudePromptCacheLogMeta rebuild on the leg path, and the lazy fail-closed fence identity. Irreducible at the existing chokepoints: each edit sits where chatCore already owns the decision, and the helpers themselves (nonStreamingProviderLeg.ts, serverOwnedToolLoopWire.ts) are under cap. Covered by tests/unit/chatcore-translation-paths.test.ts (72/74; the 2 open are issue #13043).", "_rebaseline_2026_09_07_virtualfactory_crosses_the_new_file_cap": "open-sse/services/autoCombo/virtualFactory.ts crosses the 1200 new-file cap for the first time (1187 on the pre-wave tip, 1219 after the wave). Growth is spread across the routing/free-tier wave, not one extractable block: #12794 feeds observed breaker state and model quality into snapshot scoring instead of neutral constants, #12792 adds the reliability factor the snapshot path was still ignoring and the pooled-latency bootstrap, and #12744 tightens the free-model predicate the factory consumes, and #12795 records which filter stage emptied an auto/* pool. FROZEN RATHER THAN SPLIT, deliberately, and this is debt: two cohesive extraction candidates are ready when someone owns the move — computeSnapshotWeights (~85 lines) and the credential-eligibility group hasUsableOAuthToken/hasProviderSpecificSessionData/isKeylessEligibleConnection/hasUsableConnectionCredential (~70 lines). Either alone clears 1200 from here. Splitting three contributors' just-merged work mid-batch was the larger risk.", "_rebaseline_2026_09_07_streaming_wave": "Stacked growth from the SSE/streaming wave. open-sse/handlers/chatCore.ts 6021->6026 (+5): #12854 seeds the in-memory pending continuation state synchronously, before saveCallLogOperation's first await, closing the window where resolvePreviousResponseState finds nothing because the artifact write has not landed yet. open-sse/utils/stream.ts 3080->3098 (+18): #12828 emits the trailing usage-estimate chunk on the translate flush (#12151 had only covered passthrough, so translate-mode clients never saw token counts) and #12718 stops rebuilding a truncated summary from the collector's cap-dropped event array. Irreducible at the existing chokepoints — both are the flush/finalization points themselves. Covered by the continuation-store, translate-usage and collector-truncation suites.", - "_rebaseline_2026_09_07_roundrobin_crosses_new_file_cap": "open-sse/services/combo/roundRobinCombo.ts 1198->1205, crossing the 1200 new-file cap. #12884 wires the quota-skip diagnostics into the round-robin attempt path so an ALL_TARGETS_SKIPPED 503 names which windows were exhausted instead of returning an opaque skip. The file was already at 1198 when #12811 lifted it out of combo.ts, so seven lines cross it; the diagnostics themselves live in quotaSkipDiagnostics.ts, under cap. Frozen rather than split: the natural next extraction is the attempt-loop body, which #12746/#12811 just moved and should settle before being cut again." + "_rebaseline_2026_09_07_roundrobin_crosses_new_file_cap": "open-sse/services/combo/roundRobinCombo.ts 1198->1205, crossing the 1200 new-file cap. #12884 wires the quota-skip diagnostics into the round-robin attempt path so an ALL_TARGETS_SKIPPED 503 names which windows were exhausted instead of returning an opaque skip. The file was already at 1198 when #12811 lifted it out of combo.ts, so seven lines cross it; the diagnostics themselves live in quotaSkipDiagnostics.ts, under cap. Frozen rather than split: the natural next extraction is the attempt-loop body, which #12746/#12811 just moved and should settle before being cut again.", + "_rebaseline_2026_09_08_13033_responses_websearch_sse": "Own growth after rebase onto v3.8.51 tip af49d4972: open-sse/handlers/chatCore.ts 6036->6035 (-1, check-file-size split-newline). Branch stamps clientRequestedResponsesStream before web_search fallback forces stream:false, then wraps JSON via synthesizeOpenAiSseFromJson. Call-site wiring next to the existing web_search non-stream fallback; no new god-file. Covered tests/unit/responses-websearch-sse-13033.test.ts." } diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index eaa0f2200c..a091ad4b00 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -34,7 +34,7 @@ import { import { buildPostCallGuardrailContext } from "./chatCore/postCallGuardrailContext.ts"; import { storeSemanticCacheResponse } from "./chatCore/semanticCacheStore.ts"; import { buildNonStreamingResponseHeaders } from "./chatCore/nonStreamingResponseHeaders.ts"; -import { buildNonStreamingJsonResponse } from "./chatCore/nonStreamingJsonResponse.ts"; +import { maybeWrapForcedNonStreamingResponsesJson } from "./chatCore/responsesJsonToSse.ts"; import { enforceOutputTokenBudget } from "./chatCore/outputTokenBudget.ts"; import { maybeConvertJsonBodyToSse } from "./chatCore/jsonBodyToSse.ts"; import { assembleStreamingResponseHeaders } from "./chatCore/streamingResponseHeaders.ts"; @@ -725,6 +725,7 @@ export async function handleChatCore({ copilotCompatibleReasoning, clientResponseFormat, } = resolveChatCoreRequestFormat({ clientRawRequest, body, provider, userAgent }); + let clientRequestedResponsesStream = false; const nativeOpenAICompatibleResponsesPassthrough = shouldUseNativeOpenAICompatibleResponsesPassthrough({ provider, @@ -940,6 +941,7 @@ export async function handleChatCore({ sourceFormat === FORMATS.OPENAI_RESPONSES && (body as Record).stream === true ) { + clientRequestedResponsesStream = true; (body as Record).stream = false; log?.info?.("TOOLS", `web_search fallback forced non-streaming response for ${provider}`); } @@ -5601,7 +5603,11 @@ export async function handleChatCore({ return { success: true, - response: buildNonStreamingJsonResponse(translatedResponse, responseHeaders), + response: maybeWrapForcedNonStreamingResponsesJson({ + clientRequestedResponsesStream, + body: translatedResponse, + headers: responseHeaders, + }), }; } catch (error) { trackPendingRequest(model, provider, connectionId, false); diff --git a/open-sse/handlers/chatCore/responsesJsonToSse.ts b/open-sse/handlers/chatCore/responsesJsonToSse.ts new file mode 100644 index 0000000000..be77e13025 --- /dev/null +++ b/open-sse/handlers/chatCore/responsesJsonToSse.ts @@ -0,0 +1,65 @@ +/** + * #13033: when chatCore forces stream:false so a server-side web_search + * fallback can run, the client that asked for Responses SSE still needs + * `event: response.completed`. Reuse synthesizeOpenAiSseFromJson + + * createResponsesApiTransformStream. + */ +import { createResponsesApiTransformStream } from "../../transformer/responsesTransformer.ts"; +import { synthesizeOpenAiSseFromJson } from "../../utils/jsonToSse.ts"; +import { buildNonStreamingJsonResponse } from "./nonStreamingJsonResponse.ts"; + +function copyForwardHeaders(headers: Record | undefined): Record { + const out: Record = {}; + if (!headers) return out; + for (const [key, value] of Object.entries(headers)) { + const lower = key.toLowerCase(); + if (lower === "content-type" || lower === "content-length") continue; + out[key] = value; + } + return out; +} + +export function wrapChatCompletionJsonAsResponsesSse( + completion: Record, + headers?: Record +): Response { + const rawSse = synthesizeOpenAiSseFromJson(JSON.stringify(completion)); + if (!rawSse) { + return new Response(JSON.stringify(completion), { + status: 200, + headers: { + "Content-Type": "application/json", + ...copyForwardHeaders(headers), + }, + }); + } + const encoder = new TextEncoder(); + const inputStream = new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode(rawSse)); + controller.close(); + }, + }); + const outputStream = inputStream.pipeThrough(createResponsesApiTransformStream()); + return new Response(outputStream, { + status: 200, + headers: { + ...copyForwardHeaders(headers), + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache", + Connection: "keep-alive", + }, + }); +} + +export function maybeWrapForcedNonStreamingResponsesJson(args: { + clientRequestedResponsesStream: boolean; + body: unknown; + headers: Record; +}): Response { + const { clientRequestedResponsesStream, body, headers } = args; + if (!clientRequestedResponsesStream || !body || typeof body !== "object" || Array.isArray(body)) { + return buildNonStreamingJsonResponse(body, headers); + } + return wrapChatCompletionJsonAsResponsesSse(body as Record, headers); +} diff --git a/tests/unit/8395-plugin-hooks-fire.test.ts b/tests/unit/8395-plugin-hooks-fire.test.ts index 1d8c2b561e..a41cf501a7 100644 --- a/tests/unit/8395-plugin-hooks-fire.test.ts +++ b/tests/unit/8395-plugin-hooks-fire.test.ts @@ -128,9 +128,7 @@ test("chatCore.ts calls runPluginOnResponseHook from both the non-streaming and "utf-8" ); - const nonStreamingReturnIndex = source.indexOf( - "buildNonStreamingJsonResponse(translatedResponse" - ); + const nonStreamingReturnIndex = source.indexOf("maybeWrapForcedNonStreamingResponsesJson({"); const hookCallNeedle = "await runPluginOnResponseHook({"; const hookCallIndex = source.indexOf(hookCallNeedle); const secondHookCallIndex = source.indexOf(hookCallNeedle, hookCallIndex + 1); @@ -153,6 +151,6 @@ test("chatCore.ts calls runPluginOnResponseHook from both the non-streaming and assert.ok( hookCallIndex < nonStreamingReturnIndex, "the non-streaming branch must call runPluginOnResponseHook BEFORE returning " + - "buildNonStreamingJsonResponse(...), not skip it" + "maybeWrapForcedNonStreamingResponsesJson(...), not skip it" ); }); diff --git a/tests/unit/responses-json-to-sse-13033.test.ts b/tests/unit/responses-json-to-sse-13033.test.ts new file mode 100644 index 0000000000..8d5d9388a5 --- /dev/null +++ b/tests/unit/responses-json-to-sse-13033.test.ts @@ -0,0 +1,78 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +const { wrapChatCompletionJsonAsResponsesSse, maybeWrapForcedNonStreamingResponsesJson } = + await import("../../open-sse/handlers/chatCore/responsesJsonToSse.ts"); + +function chatCompletion(content = "hi") { + return { + id: "chatcmpl-test", + object: "chat.completion", + choices: [ + { + index: 0, + message: { role: "assistant", content }, + finish_reason: "stop", + }, + ], + usage: { prompt_tokens: 4, completion_tokens: 2, total_tokens: 6 }, + }; +} + +test("wraps non-streaming chat JSON as Responses SSE ending in response.completed", async () => { + const response = wrapChatCompletionJsonAsResponsesSse(chatCompletion("hello"), { + "X-OmniRoute-Cache": "MISS", + }); + assert.equal(response.headers.get("Content-Type"), "text/event-stream"); + assert.equal(response.headers.get("X-OmniRoute-Cache"), "MISS"); + const sse = await response.text(); + assert.match(sse, /event: response\.created/); + assert.match(sse, /event: response\.completed/); + assert.match(sse, /hello/); + assert.match(sse, /data: \[DONE\]/); +}); + +test("injection: returning JSON early for a 200 chat completion goes red", async () => { + const response = wrapChatCompletionJsonAsResponsesSse(chatCompletion()); + assert.notEqual(response.headers.get("Content-Type"), "application/json"); +}); + +test("maybeWrapForcedNonStreamingResponsesJson keeps JSON when the client did not ask for SSE", async () => { + const response = maybeWrapForcedNonStreamingResponsesJson({ + clientRequestedResponsesStream: false, + body: chatCompletion("plain"), + headers: { "Content-Type": "application/json" }, + }); + assert.equal(response.headers.get("Content-Type"), "application/json"); + const payload = JSON.parse(await response.text()); + assert.equal(payload.choices[0].message.content, "plain"); +}); + +test("maybeWrapForcedNonStreamingResponsesJson wraps JSON when the client asked for SSE", async () => { + const response = maybeWrapForcedNonStreamingResponsesJson({ + clientRequestedResponsesStream: true, + body: chatCompletion("stream-me"), + headers: { "Content-Type": "application/json", "X-OmniRoute-Cache": "MISS" }, + }); + assert.equal(response.headers.get("Content-Type"), "text/event-stream"); + const sse = await response.text(); + assert.match(sse, /event: response\.completed/); + assert.match(sse, /stream-me/); +}); + +test("chatCore stamps clientRequestedResponsesStream before forcing stream:false", async () => { + const { readFile } = await import("node:fs/promises"); + const { join } = await import("node:path"); + const source = await readFile( + join(import.meta.dirname, "../../open-sse/handlers/chatCore.ts"), + "utf-8" + ); + const stamp = source.indexOf("clientRequestedResponsesStream = true"); + const force = source.indexOf("(body as Record).stream = false"); + const wrap = source.indexOf("maybeWrapForcedNonStreamingResponsesJson({"); + assert.ok(stamp !== -1, "must stamp the client-requested stream flag"); + assert.ok(force !== -1, "must still force stream:false for the web_search fallback"); + assert.ok(wrap !== -1, "must wrap the non-streaming JSON return"); + assert.ok(stamp < force, "stamp must happen before stream:false"); + assert.ok(wrap > force, "wrap must happen on the non-streaming return after the force"); +});