From 3e5a4dd9bb6d7a1bc274e63dd569169ea0d972f1 Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Mon, 21 Sep 2026 03:57:44 +0900 Subject: [PATCH 1/4] fix(responses): isolate Cursor combo shadow calls --- src/server/responses/core-options.ts | 2 ++ src/server/responses/request-prepare.ts | 6 ++++++ .../responses-shadow-intercept.test.ts | 20 ++++++++++++++++++- 3 files changed, 27 insertions(+), 1 deletion(-) diff --git a/src/server/responses/core-options.ts b/src/server/responses/core-options.ts index 5334cf4e02..110f56f724 100644 --- a/src/server/responses/core-options.ts +++ b/src/server/responses/core-options.ts @@ -108,6 +108,8 @@ export interface HandleResponsesOptions { callerDirectAuth?: CallerDirectAuth | null; /** Internal recursion guard; callers outside this module must not set it. */ comboAttempt?: boolean; + /** Internal handoff: this combo was selected by shadow-call interception. */ + shadowCallIntercepted?: boolean; compactionRoutingOverride?: CompactionRoutingOverride | null; /** Internal combo handoff for one parent-validated continuation snapshot. */ comboReplaySnapshot?: { diff --git a/src/server/responses/request-prepare.ts b/src/server/responses/request-prepare.ts index 867f816414..97eaf3985f 100644 --- a/src/server/responses/request-prepare.ts +++ b/src/server/responses/request-prepare.ts @@ -225,6 +225,7 @@ export async function prepareResponsesRequest( // hops — which only exist inside that loop — are unreachable (#4129). Rewrite the selector // here instead, before comboIdFromRawBody reads `model`, and identify the combo by CONFIG // LOOKUP so the check can never observe a one-candidate collapse. + let shadowCallIntercepted = false; if (!options.comboAttempt && !options.compactionRoutingOverride && body && typeof body === "object" && !Array.isArray(body)) { const shadowIntercept = config.shadowCallIntercept; const rawShadowModel = (body as { model?: unknown }).model; @@ -232,6 +233,7 @@ export async function prepareResponsesRequest( && isShadowSourceModel(rawShadowModel, shadowIntercept.sourceModels)) { const shadowComboId = resolveComboId(config, shadowIntercept.model); if (shadowComboId && Object.hasOwn(config.combos ?? {}, shadowComboId)) { + shadowCallIntercepted = true; (body as Record).model = shadowIntercept.model; // Same rule as the late intercept site: record the operator-configured prefix that // matched, never the caller's raw model string. Matching is by prefix, so the raw @@ -247,6 +249,9 @@ export async function prepareResponsesRequest( options.onRequestBodyRead?.(); return requestDispatchers.handleComboResponses(req, body, comboId, config, logCtx, { ...options, + // Concrete combo child selectors no longer match the shadow source model. Carry the + // interception decision explicitly so provider-specific helper isolation still applies. + shadowCallIntercepted, // The original request body was accepted above. Combo children are synthetic // replays and must not repeat the caller-owned timeout transition. onRequestBodyRead: undefined, @@ -369,6 +374,7 @@ export async function prepareResponsesRequest( } } if (cursorClientThreadId) parsed._cursorClientThreadId = cursorClientThreadId; + if (options.shadowCallIntercepted === true) parsed._cursorIsolateConversation = true; } catch (err) { if (isTranslatorBudgetExceededError(err)) { return formatErrorResponse(413, "request_too_large", "request translation buffer exceeded the safe limit", { diff --git a/tests/responses/responses-shadow-intercept.test.ts b/tests/responses/responses-shadow-intercept.test.ts index d5cbdd680c..ab3800136c 100644 --- a/tests/responses/responses-shadow-intercept.test.ts +++ b/tests/responses/responses-shadow-intercept.test.ts @@ -4,7 +4,7 @@ * default follows modern clients, while sourceModels keeps an escape hatch. */ import { afterEach, describe, expect, test } from "bun:test"; -import { mkdtempSync} from "node:fs"; +import { mkdtempSync, readFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { handleResponses, isShadowSourceModel } from "../../src/server/responses"; @@ -15,6 +15,7 @@ import type { OcxConfig } from "../../src/types"; import { catalogConvergenceFactory } from "../helpers/catalog-convergence"; import { removeTreeWithRetry } from "../helpers/remove-tree"; import { acquireOwnedSpendHome } from "../helpers/owned-spend-home"; +import { repoPath } from "../helpers/repo-root"; const originalFetch = globalThis.fetch; let releaseSpendHome: (() => void) | undefined; @@ -307,6 +308,23 @@ function chatOk(text: string): Response { } describe("a combo shadow-call target enters the failover loop (#4129)", () => { + test("carries helper conversation isolation into concrete combo children", () => { + const prepare = readFileSync(repoPath("src/server/responses/request-prepare.ts"), "utf8"); + const comboDispatch = prepare.slice( + prepare.indexOf("const comboId = !options.comboAttempt"), + prepare.indexOf("let unreadableEncryptedAgentTask"), + ); + const parsedHandoff = prepare.slice( + prepare.indexOf("if (cursorClientThreadId) parsed._cursorClientThreadId"), + prepare.indexOf("} catch (err)", prepare.indexOf("if (cursorClientThreadId) parsed._cursorClientThreadId")), + ); + + expect(comboDispatch).toContain("shadowCallIntercepted,"); + expect(parsedHandoff).toContain( + "if (options.shadowCallIntercepted === true) parsed._cursorIsolateConversation = true;", + ); + }); + test("a helper call rewritten to a combo hops past a 429 to the second target", async () => { takeSpendHome(); const urls: string[] = []; From 1fb6005335a9aee7f61b9dda1171ccb6b7e8272e Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Mon, 21 Sep 2026 03:58:15 +0900 Subject: [PATCH 2/4] fix(responses): repair terminal-less bridged search legs --- src/server/responses/passthrough-delivery.ts | 29 ++++++++++---------- tests/responses/passthrough-abort.test.ts | 9 +++++- 2 files changed, 23 insertions(+), 15 deletions(-) diff --git a/src/server/responses/passthrough-delivery.ts b/src/server/responses/passthrough-delivery.ts index 969cd84e58..4757802f7f 100644 --- a/src/server/responses/passthrough-delivery.ts +++ b/src/server/responses/passthrough-delivery.ts @@ -380,12 +380,22 @@ export async function deliverPassthroughResponse( }); // Capture the binding that actually served the first leg, after its permitted reselection. const webSearchBridgeBinding = requestBindings.get(nativeExchange.request); - // The bridge wraps the RAW upstream body, so terminal repair below still owns the single - // client-facing terminal — the bridge drops the terminal of every intercepted leg. - const upstreamSseBody = webSearchBridgePlan + // Repair must observe the raw first leg before the bridge suppresses an intercepted search + // lifecycle. Otherwise a provider that leaves that complete call open never arms repair's + // grace timer, so the bridge cannot execute the search or begin its continuation. + let passthroughSseBody = terminalRepairPolicy + ? relayResponsesSseWithTerminalRepair( + upstreamResponse.body, + upstream, + terminalRepairPolicy, + translatorBudget, + options.responsesTerminalRepairScheduler, + ) + : upstreamResponse.body; + passthroughSseBody = webSearchBridgePlan ? createPassthroughWebSearchBridgeStream({ plan: webSearchBridgePlan, - firstLeg: upstreamResponse.body, + firstLeg: passthroughSseBody, requestBody: nativeExchange.request.body, // Continuation legs replay the same built request with the executed search appended. // The first leg already passed the recovery ladder, the outbound size ceiling, and the @@ -433,16 +443,7 @@ export async function deliverPassthroughResponse( onFinalize: () => releaseCodexAuthContextProbeLease(openAiSidecar?.authContext), signal: upstream.signal, }) - : upstreamResponse.body; - const passthroughSseBody = terminalRepairPolicy - ? relayResponsesSseWithTerminalRepair( - upstreamSseBody, - upstream, - terminalRepairPolicy, - translatorBudget, - options.responsesTerminalRepairScheduler, - ) - : upstreamSseBody; + : passthroughSseBody; const repairConfig = route.provider.responsesItemIdRepair; // Grok Build renders deltas live but reconstructs its durable assistant // turn from the completed response snapshot. Native Responses streams diff --git a/tests/responses/passthrough-abort.test.ts b/tests/responses/passthrough-abort.test.ts index bbdd622d3d..cc50132307 100644 --- a/tests/responses/passthrough-abort.test.ts +++ b/tests/responses/passthrough-abort.test.ts @@ -62,8 +62,15 @@ describe("passthrough relayWithAbort (RC2, passthrough path)", () => { // The captured static policy now supplies the repair decision; the real platform gate and // pure native relay invariants below are unchanged. expect(sseBranch).toContain("const terminalRepairPolicy = route.staticPolicy.model.responsesTerminalRepair;"); - expect(sseBranch).toContain("const passthroughSseBody = terminalRepairPolicy"); + expect(sseBranch).toContain("let passthroughSseBody = terminalRepairPolicy"); expect(sseBranch).toContain(": upstreamResponse.body;"); + // Repair has to wrap the raw first leg before the bridge hides its completed web-search call; + // otherwise a terminal-less open leg cannot trigger the repair timer and continuation stalls. + const terminalRepair = sseBranch.indexOf("relayResponsesSseWithTerminalRepair("); + const webSearchBridge = sseBranch.indexOf("createPassthroughWebSearchBridgeStream({"); + expect(terminalRepair).toBeGreaterThanOrEqual(0); + expect(webSearchBridge).toBeGreaterThan(terminalRepair); + expect(sseBranch.slice(webSearchBridge)).toContain("firstLeg: passthroughSseBody,"); // Native tee stays inside the bounded observer. The production owner passes // the raw stream and disconnect signal before any client-side rewrite. expect(sseBranch).toMatch(/const \[nativeBody, inspectBody\] = teeWithBoundedInspection\(passthroughSseBody, \{ clientGoneSignal \}\)/); From 3948970a6459b803e5b1be4604fdeed78566f74f Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Mon, 21 Sep 2026 05:43:15 +0900 Subject: [PATCH 3/4] fix(responses): repair continuation legs and prove behavior in tests --- src/server/responses/passthrough-delivery.ts | 56 +++--- tests/responses/passthrough-abort.test.ts | 166 ++++++++++++++++++ .../responses-shadow-intercept.test.ts | 40 +++++ 3 files changed, 242 insertions(+), 20 deletions(-) diff --git a/src/server/responses/passthrough-delivery.ts b/src/server/responses/passthrough-delivery.ts index 4757802f7f..63e9438a20 100644 --- a/src/server/responses/passthrough-delivery.ts +++ b/src/server/responses/passthrough-delivery.ts @@ -400,26 +400,42 @@ export async function deliverPassthroughResponse( // Continuation legs replay the same built request with the executed search appended. // The first leg already passed the recovery ladder, the outbound size ceiling, and the // host circuit; a KEY-auth destination has no OAuth refresh to replay on a later leg. - send: (continuationBody: string) => fetchWithHeaderTimeout( - nativeExchange.request.url, - { method: nativeExchange.request.method, headers: nativeExchange.request.headers, body: continuationBody }, - upstream.signal, - connectMs, - true, - providerFetch(route.provider, options.codexWsRuntimeIdentity, { - // Pacing can outlive a manual selection change. A continuation must retain the - // first leg's key and appended search result, never rebuild from the original turn. - beforeDispatch: () => { - if (webSearchBridgeBinding?.kind !== "api-key" - || !providerApiKeySelectionIsCurrent(config, route.providerName, webSearchBridgeBinding.provider)) { - throw new Error("API key selection changed during a web-search continuation"); - } - }, - providerName: route.providerName, - modelId: route.modelId, - }), - false, - ), + send: async (continuationBody: string) => { + const continuation = await fetchWithHeaderTimeout( + nativeExchange.request.url, + { method: nativeExchange.request.method, headers: nativeExchange.request.headers, body: continuationBody }, + upstream.signal, + connectMs, + true, + providerFetch(route.provider, options.codexWsRuntimeIdentity, { + // Pacing can outlive a manual selection change. A continuation must retain the + // first leg's key and appended search result, never rebuild from the original turn. + beforeDispatch: () => { + if (webSearchBridgeBinding?.kind !== "api-key" + || !providerApiKeySelectionIsCurrent(config, route.providerName, webSearchBridgeBinding.provider)) { + throw new Error("API key selection changed during a web-search continuation"); + } + }, + providerName: route.providerName, + modelId: route.modelId, + }), + false, + ); + // The same provider can leave a complete continuation open without a terminal, which + // stalls the bridge's decide loop exactly like the first leg — so every leg gets the + // same repair, not only the intercepted first one. + if (!terminalRepairPolicy || !continuation.ok || !continuation.body) return continuation; + return new Response( + relayResponsesSseWithTerminalRepair( + continuation.body, + upstream, + terminalRepairPolicy, + translatorBudget, + options.responsesTerminalRepairScheduler, + ), + continuation, + ); + }, execute: createPassthroughWebSearchBridgeExecutor(webSearchBridgePlan, { providerApiKey: route.provider.apiKey ?? "", auth: webSearchBridgeAuth, diff --git a/tests/responses/passthrough-abort.test.ts b/tests/responses/passthrough-abort.test.ts index cc50132307..74cfcddca7 100644 --- a/tests/responses/passthrough-abort.test.ts +++ b/tests/responses/passthrough-abort.test.ts @@ -2,6 +2,9 @@ import { describe, expect, test } from "bun:test"; import { consumeForInspection, linkAbortSignal, relaySseWithFailedTail, relaySseWithHeartbeat, relayWithAbort } from "../../src/server"; import { pathToFileURL } from "node:url"; import { repoRoot } from "../helpers/repo-root"; +import { relayResponsesSseWithTerminalRepair, type ResponsesTerminalRepairScheduler } from "../../src/server/responses-terminal-repair"; +import { createPassthroughWebSearchBridgeStream, type PassthroughWebSearchBridgePlan } from "../../src/web-search/passthrough-bridge"; +import { createTestTranslatorBudget } from "../helpers/translator-budget"; const root = pathToFileURL(repoRoot() + "/"); @@ -627,3 +630,166 @@ describe("passthrough relayWithAbort (RC2, passthrough path)", () => { expect(upstream.signal.reason).toBe("replacement turn"); }); }); + +/** + * The reported stall: a provider emits a complete intercepted `web_search` call but never sends + * a terminal and holds the leg open. With repair wrapped around the raw first leg, the grace + * timer still arms and the bridge can execute the search and continue upstream. + */ +describe("terminal repair ahead of the passthrough web-search bridge", () => { + class ManualScheduler implements ResponsesTerminalRepairScheduler { + private current = 0; + private nextId = 1; + private readonly jobs = new Map void }>(); + + nowMs(): number { return this.current; } + + schedule(callback: () => void, delayMs: number): unknown { + const id = this.nextId++; + this.jobs.set(id, { at: this.current + delayMs, callback }); + return id; + } + + cancel(handle: unknown): void { + this.jobs.delete(handle as number); + } + + advance(ms: number): void { + this.current += ms; + for (;;) { + const due = [...this.jobs.entries()] + .filter(([, job]) => job.at <= this.current) + .sort((left, right) => left[1].at - right[1].at); + if (due.length === 0) return; + for (const [id, job] of due) { + if (!this.jobs.delete(id)) continue; + job.callback(); + } + } + } + + pending(): number { return this.jobs.size; } + } + + const searchCall = { + type: "function_call", + id: "fc_1", + status: "completed", + call_id: "call_1", + name: "web_search", + arguments: "{\"query\":\"opencodex release\"}", + }; + + const preamble = { + type: "message", + id: "msg_1", + status: "completed", + role: "assistant", + content: [{ type: "output_text", text: "Let me look that up." }], + }; + + const answer = { + type: "message", + id: "msg_2", + role: "assistant", + content: [{ type: "output_text", text: "The current release is 2.50.0." }], + }; + + function frame(type: string, payload: Record): string { + return "event: " + type + "\ndata: " + JSON.stringify({ type, ...payload }); + } + + function sseBody(...blocks: string[]): string { + return blocks.concat("data: [DONE]").join("\n\n") + "\n\n"; + } + + /** Every output item complete, no terminal event, and the leg is never closed. */ + function terminallessSearchLeg(): ReadableStream { + const text = [ + frame("response.created", { response: { id: "resp_1", status: "in_progress" } }), + frame("response.output_item.added", { output_index: 0, item: { ...preamble, content: [] } }), + frame("response.output_item.done", { output_index: 0, item: preamble }), + frame("response.output_item.added", { output_index: 1, item: { ...searchCall, arguments: "" } }), + frame("response.function_call_arguments.done", { + output_index: 1, + item_id: "fc_1", + arguments: searchCall.arguments, + }), + frame("response.output_item.done", { output_index: 1, item: searchCall }), + ].join("\n\n") + "\n\n"; + return new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(text)); + }, + }); + } + + function answerLeg(): ReadableStream { + return streamFromChunks([new TextEncoder().encode(sseBody( + frame("response.created", { response: { id: "resp_2", status: "in_progress" } }), + frame("response.output_item.added", { output_index: 0, item: { ...answer, content: [] } }), + frame("response.output_item.done", { output_index: 0, item: answer }), + frame("response.completed", { + response: { id: "resp_2", status: "completed", output: [answer] }, + }), + ))]); + } + + test("a terminal-less first search leg reaches the bridge once repaired", async () => { + const scheduler = new ManualScheduler(); + const upstream = new AbortController(); + const plan: PassthroughWebSearchBridgePlan = { + backend: "ollama", + endpoint: "https://ollama.com/api/web_search", + maxSearches: 3, + timeoutMs: 60_000, + }; + const sent: string[] = []; + const executed: string[][] = []; + const stream = createPassthroughWebSearchBridgeStream({ + plan, + firstLeg: relayResponsesSseWithTerminalRepair( + terminallessSearchLeg(), + upstream, + { graceMs: 5_000 }, + createTestTranslatorBudget(), + scheduler, + ), + requestBody: JSON.stringify({ + model: "glm-4.7", + stream: true, + input: [{ role: "user", content: [{ type: "input_text", text: "what is the latest release?" }] }], + tools: [{ type: "web_search" }], + }), + send: async (body) => { + sent.push(body); + return new Response(answerLeg(), { + headers: { "content-type": "text/event-stream" }, + }); + }, + execute: async (queries) => { + executed.push(queries); + return { text: "opencodex 2.50.0 shipped", sources: [] }; + }, + }); + + const bodyPromise = new Response(stream).text(); + // The bridge is pull-driven: let it drain the pushed leg frames so repair arms the timer. + for (let i = 0; i < 1_000 && scheduler.pending() === 0; i += 1) { + await new Promise(resolve => setTimeout(resolve, 0)); + } + expect(scheduler.pending()).toBe(1); + scheduler.advance(5_000); + const body = await bodyPromise; + + expect(executed).toEqual([["opencodex release"]]); + expect(sent).toHaveLength(1); + const events = body + .split(/\r?\n/) + .filter(line => line.startsWith("data:")) + .map(line => line.slice(5).trim()) + .filter(payload => payload.length > 0 && payload !== "[DONE]") + .map(payload => JSON.parse(payload) as Record); + expect(events.some(event => event.type === "response.completed")).toBe(true); + }); +}); diff --git a/tests/responses/responses-shadow-intercept.test.ts b/tests/responses/responses-shadow-intercept.test.ts index ab3800136c..fd5befeb7d 100644 --- a/tests/responses/responses-shadow-intercept.test.ts +++ b/tests/responses/responses-shadow-intercept.test.ts @@ -16,6 +16,8 @@ import { catalogConvergenceFactory } from "../helpers/catalog-convergence"; import { removeTreeWithRetry } from "../helpers/remove-tree"; import { acquireOwnedSpendHome } from "../helpers/owned-spend-home"; import { repoPath } from "../helpers/repo-root"; +import { createTestTranslatorBudget } from "../helpers/translator-budget"; +import { prepareResponsesRequest } from "../../src/server/responses/request-prepare"; const originalFetch = globalThis.fetch; let releaseSpendHome: (() => void) | undefined; @@ -308,6 +310,44 @@ function chatOk(text: string): Response { } describe("a combo shadow-call target enters the failover loop (#4129)", () => { + test("a combo child of a shadow-intercepted call gets Cursor conversation isolation", async () => { + const config = comboInterceptConfig([{ provider: "xai", model: "grok-4.5" }]); + const logCtx: RequestLogContext = { model: "", provider: "" }; + const mkreq = () => new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: "grok-4.5", + input: [{ type: "message", role: "user", content: [{ type: "input_text", text: "hi" }] }], + stream: false, + }), + }); + const dispatchers = { + handleResponses: () => Promise.reject(new Error("unexpected recursion")), + handleComboResponses: () => Promise.reject(new Error("unexpected combo dispatch")), + }; + const admission = () => ({ pendingHostAdmissionLease: null, authCtx: { kind: "main", accountId: null } }) as never; + + const intercepted = await prepareResponsesRequest( + { req: mkreq(), config, logCtx, options: { comboAttempt: true, shadowCallIntercepted: true, translatorBudget: createTestTranslatorBudget() } }, + admission(), + dispatchers, + ); + expect(intercepted).not.toBeInstanceOf(Response); + if (intercepted instanceof Response) throw new Error("expected a prepared request, got HTTP " + intercepted.status); + expect(intercepted.parsed._cursorIsolateConversation).toBe(true); + + // A plain combo child (no interception marker) must not be isolated. + const plain = await prepareResponsesRequest( + { req: mkreq(), config, logCtx, options: { comboAttempt: true, translatorBudget: createTestTranslatorBudget() } }, + admission(), + dispatchers, + ); + expect(plain).not.toBeInstanceOf(Response); + if (plain instanceof Response) throw new Error("expected a prepared request, got HTTP " + plain.status); + expect(plain.parsed._cursorIsolateConversation).not.toBe(true); + }); + test("carries helper conversation isolation into concrete combo children", () => { const prepare = readFileSync(repoPath("src/server/responses/request-prepare.ts"), "utf8"); const comboDispatch = prepare.slice( From 4bfcc0a8ea657b4f659355385a87c3b11a542365 Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Mon, 21 Sep 2026 11:11:56 +0900 Subject: [PATCH 4/4] fix(responses): close shadow combo intersection and continuation coverage Apply the existing source-target non-intersection rule before early combo interception, and add production-path behavioral coverage for terminal-less continuation repair. --- src/server/responses/request-prepare.ts | 23 ++- tests/responses/passthrough-abort.test.ts | 161 ++++++++++++++++++ .../responses-shadow-intercept.test.ts | 12 +- 3 files changed, 181 insertions(+), 15 deletions(-) diff --git a/src/server/responses/request-prepare.ts b/src/server/responses/request-prepare.ts index 97eaf3985f..5eea479b7e 100644 --- a/src/server/responses/request-prepare.ts +++ b/src/server/responses/request-prepare.ts @@ -233,14 +233,21 @@ export async function prepareResponsesRequest( && isShadowSourceModel(rawShadowModel, shadowIntercept.sourceModels)) { const shadowComboId = resolveComboId(config, shadowIntercept.model); if (shadowComboId && Object.hasOwn(config.combos ?? {}, shadowComboId)) { - shadowCallIntercepted = true; - (body as Record).model = shadowIntercept.model; - // Same rule as the late intercept site: record the operator-configured prefix that - // matched, never the caller's raw model string. Matching is by prefix, so the raw - // value is caller-controlled and reaches usage.jsonl and /api/logs. - logCtx.shadowCallRewrittenFrom = sanitizeLogMetadataString( - shadowSourceModelPrefix(rawShadowModel, shadowIntercept.sourceModels), - ); + const sourcePrefix = shadowSourceModelPrefix(rawShadowModel, shadowIntercept.sourceModels)!; + let sourceIdentity = { providerName: OPENAI_CODEX_PROVIDER_ID, modelId: sourcePrefix }; + try { + const resolvedSource = routeConcreteModel(config, rawShadowModel); + sourceIdentity = { providerName: resolvedSource.providerName, modelId: sourcePrefix }; + } catch { /* Native Codex helper calls remain OpenAI-owned without an enabled OpenAI route. */ } + const targetRoute = routeModel(config, shadowIntercept.model, evidenceFromBody(body)); + if (shouldInterceptShadowCall(rawShadowModel, shadowIntercept.sourceModels, sourceIdentity, targetRoute)) { + shadowCallIntercepted = true; + (body as Record).model = shadowIntercept.model; + // Same rule as the late intercept site: record the operator-configured prefix that + // matched, never the caller's raw model string. Matching is by prefix, so the raw + // value is caller-controlled and reaches usage.jsonl and /api/logs. + logCtx.shadowCallRewrittenFrom = sanitizeLogMetadataString(sourcePrefix); + } } } } diff --git a/tests/responses/passthrough-abort.test.ts b/tests/responses/passthrough-abort.test.ts index 74cfcddca7..5d5818bd6c 100644 --- a/tests/responses/passthrough-abort.test.ts +++ b/tests/responses/passthrough-abort.test.ts @@ -4,6 +4,8 @@ import { pathToFileURL } from "node:url"; import { repoRoot } from "../helpers/repo-root"; import { relayResponsesSseWithTerminalRepair, type ResponsesTerminalRepairScheduler } from "../../src/server/responses-terminal-repair"; import { createPassthroughWebSearchBridgeStream, type PassthroughWebSearchBridgePlan } from "../../src/web-search/passthrough-bridge"; +import { deliverPassthroughResponse } from "../../src/server/responses/passthrough-delivery"; +import { routedProviderConfig } from "../../src/router"; import { createTestTranslatorBudget } from "../helpers/translator-budget"; const root = pathToFileURL(repoRoot() + "/"); @@ -735,6 +737,20 @@ describe("terminal repair ahead of the passthrough web-search bridge", () => { ))]); } + /** Complete answer output, no terminal event, and the continuation remains open. */ + function terminallessAnswerLeg(): ReadableStream { + const text = [ + frame("response.created", { response: { id: "resp_2", status: "in_progress" } }), + frame("response.output_item.added", { output_index: 0, item: { ...answer, content: [] } }), + frame("response.output_item.done", { output_index: 0, item: { ...answer, status: "completed" } }), + ].join("\n\n") + "\n\n"; + return new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(text)); + }, + }); + } + test("a terminal-less first search leg reaches the bridge once repaired", async () => { const scheduler = new ManualScheduler(); const upstream = new AbortController(); @@ -792,4 +808,149 @@ describe("terminal repair ahead of the passthrough web-search bridge", () => { .map(payload => JSON.parse(payload) as Record); expect(events.some(event => event.type === "response.completed")).toBe(true); }); + + /** + * The production continuation sender in deliverPassthroughResponse must apply the same + * terminal repair to every leg, not only the first one. This drives the real function: + * the first leg is a terminal-less intercepted web_search call, the ollama search fetch is + * stubbed, and the provider's own fetch returns a terminal-less continuation — which only + * reaches the client when the sender's repair wrap synthesizes response.completed. + */ + test("deliverPassthroughResponse repairs a terminal-less continuation leg", async () => { + const scheduler = new ManualScheduler(); + const upstream = new AbortController(); + const originalFetch = globalThis.fetch; + + const provider = routedProviderConfig("bridge-test", { + adapter: "openai-responses", + baseUrl: "https://bridge-test.example/v1", + authMode: "key", + apiKey: "test-bridge-key", + webSearchBridge: { + enabled: true, + backend: "ollama", + endpoint: "https://bridge-test.example/api/web_search", + maxSearches: 3, + timeoutMs: 60_000, + }, + fetch: (async () => new Response(terminallessAnswerLeg(), { + headers: { "content-type": "text/event-stream" }, + })) as unknown as typeof globalThis.fetch, + } as never); + const config = { + providers: { "bridge-test": provider }, + maxUpstreamBodyBytes: 8 * 1024 * 1024, + }; + const upstreamRequest = { + url: "https://bridge-test.example/v1/responses", + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: "glm-4.7", + stream: true, + input: [{ role: "user", content: [{ type: "input_text", text: "what is the latest release?" }] }], + tools: [{ type: "web_search" }], + }), + }; + const requestBindings = new WeakMap(); + requestBindings.set(upstreamRequest, { kind: "api-key", provider }); + + globalThis.fetch = (async () => Response.json({ + results: [{ url: "https://example.com/release", title: "Release notes", content: "2.50.0 shipped" }], + })) as typeof globalThis.fetch; + try { + const response = await deliverPassthroughResponse( + { + logCtx: { model: "", provider: "" }, + config, + options: { responsesTerminalRepairScheduler: scheduler }, + req: new Request("http://localhost/v1/responses", { method: "POST" }), + }, + { authCtx: { kind: "main", accountId: null } }, + { + parsed: { + modelId: "glm-4.7", + stream: true, + options: {}, + _webSearch: { type: "web_search" }, + }, + route: { + providerName: "bridge-test", + provider, + modelId: "glm-4.7", + staticPolicy: { model: { responsesTerminalRepair: { graceMs: 5_000 } } }, + }, + subagentQuotaFailureModel: undefined, + subagentFallbackAccountId: undefined, + clientRequestedStream: true, + translatorBudget: createTestTranslatorBudget(), + }, + { requestBindings }, + { openAiSidecar: undefined }, + { + plaintextV2AgentMessageToolNames: new Set(), + commitReasoningReplayServingRoute: () => {}, + routedMuseToolNameAliases: new Map(), + routedNamespaceToolAliases: new Map(), + plaintextV2AgentMessageAliasedToolNames: new Set(), + recordTerminalOutcomes: false, + responseCompletionCancelled: false, + }, + { + upstreamResponse: new Response(terminallessSearchLeg(), { + headers: { "content-type": "text/event-stream" }, + }), + codexSafetyBufferingOptions: undefined, + upstream, + request: upstreamRequest, + connectMs: 5_000, + imageGenCallAliases: new Map(), + selfNamedNamespaceScrubAuthorization: undefined, + authorizedBareNamespaceToolAliases: new Map(), + rememberPassthroughResponseChecked: () => {}, + routedCustomToolNames: new Set(), + routedCustomToolRepairNames: new Set(), + declaredWireToolNames: new Set(), + routedToolSearchNames: new Set(), + outboundRequestBody: undefined, + functionRepairSchemas: new Map(), + undeclaredToolGuardActive: false, + declaredNamelessClientCallTypes: new Set(), + providerExecutedCallTypes: new Set(), + declaredBareWireToolNames: new Set(), + rememberPassthroughResponse: false, + noteInspectedPayload: () => {}, + normalizeFunctionCompletionJson: (text: string) => text, + }, + ); + + expect(response.ok).toBe(true); + const bodyPromise = response.text(); + // First leg: repair arms once every output item is complete and the grace timer + // synthesizes the terminal that lets the bridge dispatch its continuation. + for (let i = 0; i < 1_000 && scheduler.pending() === 0; i += 1) { + await new Promise(resolve => setTimeout(resolve, 0)); + } + expect(scheduler.pending()).toBe(1); + scheduler.advance(5_000); + // Continuation leg: the production sender wraps the fetch result in the same repair, + // so its own terminal-less body re-arms the timer instead of stalling the stream. + for (let i = 0; i < 1_000 && scheduler.pending() === 0; i += 1) { + await new Promise(resolve => setTimeout(resolve, 0)); + } + expect(scheduler.pending()).toBe(1); + scheduler.advance(5_000); + const body = await bodyPromise; + + const events = body + .split(/\r?\n/) + .filter(line => line.startsWith("data:")) + .map(line => line.slice(5).trim()) + .filter(payload => payload.length > 0 && payload !== "[DONE]") + .map(payload => JSON.parse(payload) as Record); + expect(events.some(event => event.type === "response.completed")).toBe(true); + } finally { + globalThis.fetch = originalFetch; + } + }); }); diff --git a/tests/responses/responses-shadow-intercept.test.ts b/tests/responses/responses-shadow-intercept.test.ts index fd5befeb7d..fc07e8bb8a 100644 --- a/tests/responses/responses-shadow-intercept.test.ts +++ b/tests/responses/responses-shadow-intercept.test.ts @@ -397,7 +397,7 @@ describe("a combo shadow-call target enters the failover loop (#4129)", () => { .toEqual(["xai/grok-4.5", "alt/grok-4.5"]); }); - test("a combo whose first target intersects the source still routes as a combo", async () => { + test("a combo whose selected target intersects the source is not intercepted", async () => { takeSpendHome(); const urls: string[] = []; const logCtx: RequestLogContext = { model: "", provider: "" }; @@ -423,12 +423,10 @@ describe("a combo shadow-call target enters the failover loop (#4129)", () => { // A healthy first target still costs exactly one upstream call. expect(urls).toHaveLength(1); expect(urls[0]).toContain("api.x.ai"); - expect(logCtx.provider).toBe("combo"); - expect(logCtx.comboId).toBe("shadow"); - expect(logCtx.routeDecision?.routeKind).toBe("combo"); - // Red before the fix: shouldInterceptShadowCall saw the collapsed pick as a self-target, - // skipped the rewrite, and the request left as a plain native route with no marker. - expect(logCtx.shadowCallRewrittenFrom).toBe("custom-helper"); + expect(logCtx.provider).toBe("xai"); + expect(logCtx.comboId).toBeUndefined(); + expect(logCtx.routeDecision?.routeKind).not.toBe("combo"); + expect(logCtx.shadowCallRewrittenFrom).toBeUndefined(); }); test("a non-combo replacement still takes the ordinary late intercept", async () => {