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"); +});