From 82afb71db7064a450f2a1693a73418a3f55deded Mon Sep 17 00:00:00 2001 From: Wibias <37517432+Wibias@users.noreply.github.com> Date: Sat, 8 Aug 2026 13:30:08 +0200 Subject: [PATCH 1/5] test(routing): define policy candidate fallback behavior --- tests/routing-policy-fallback.test.ts | 155 ++++++++++++++++++++++++++ 1 file changed, 155 insertions(+) create mode 100644 tests/routing-policy-fallback.test.ts diff --git a/tests/routing-policy-fallback.test.ts b/tests/routing-policy-fallback.test.ts new file mode 100644 index 000000000..11e9360d9 --- /dev/null +++ b/tests/routing-policy-fallback.test.ts @@ -0,0 +1,155 @@ +import { describe, expect, test } from "bun:test"; + +import type { OcxConfig } from "../src/types"; +import type { RequestLogContext } from "../src/server/request-log"; +import type { RouteDecisionTraceV1 } from "../src/routing/trace"; +import { + handleResponsesWithPolicyFallback, + rankPolicyFallbackCandidates, +} from "../src/server/responses/policy-fallback"; + +function policyTrace(): RouteDecisionTraceV1 { + return { + version: 1, + decisionId: "decision-1", + createdAt: 1, + requestedModel: "policy/daily", + routeKind: "policy", + profile: { id: "daily", revision: "rev-1" }, + requirements: [], + candidates: [ + { + provider: "provider-a", + model: "model-a", + eligible: true, + exclusions: [], + score: { total: 0.90, components: {} }, + }, + { + provider: "provider-b", + model: "model-b", + eligible: true, + exclusions: [], + score: { total: 0.80, components: {} }, + }, + { + provider: "provider-c", + model: "model-c", + eligible: true, + exclusions: [], + score: { total: 0.80, components: {} }, + }, + { + provider: "provider-d", + model: "model-d", + eligible: false, + exclusions: [{ code: "tools" }], + score: { total: 1, components: {} }, + }, + ], + selected: { + candidateIndex: 0, + provider: "provider-a", + model: "model-a", + reason: "highest-score", + }, + }; +} + +function request(): Request { + return new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "policy/daily", input: "hello", stream: false }), + }); +} + +describe("policy candidate fallback", () => { + test("ranks only eligible untried candidates by score and stable original order", () => { + const trace = policyTrace(); + const ranked = rankPolicyFallbackCandidates( + trace, + new Set(["provider-a\u0000model-a"]), + ); + + expect(ranked.map(candidate => `${candidate.provider}/${candidate.model}`)).toEqual([ + "provider-b/model-b", + "provider-c/model-c", + ]); + }); + + test("retries the next policy candidate after a retryable pre-stream failure", async () => { + const trace = policyTrace(); + const logCtx = { + requestedModel: "policy/daily", + routeDecision: trace, + attempts: [], + } as unknown as RequestLogContext; + const seenModels: string[] = []; + + const response = await handleResponsesWithPolicyFallback( + request(), + {} as OcxConfig, + logCtx, + {}, + { + runCore: async (req, _config, childLog) => { + const body = await req.json() as { model: string }; + seenModels.push(body.model); + if (seenModels.length === 1) { + childLog.requestedModel = "policy/daily"; + childLog.routeDecision = trace; + return new Response( + JSON.stringify({ error: { message: "rate limited", type: "rate_limit_error" } }), + { status: 429, headers: { "content-type": "application/json" } }, + ); + } + childLog.requestedModel = body.model; + childLog.routeDecision = { + ...trace, + requestedModel: body.model, + routeKind: "explicit-provider", + profile: undefined, + }; + return new Response(JSON.stringify({ status: "completed" }), { status: 200 }); + }, + }, + ); + + expect(response.status).toBe(200); + expect(seenModels).toEqual(["policy/daily", "provider-b/model-b"]); + expect(logCtx.requestedModel).toBe("policy/daily"); + expect(logCtx.routeDecision).toBe(trace); + }); + + test("does not switch candidates for terminal client/input failures", async () => { + const trace = policyTrace(); + const logCtx = { + requestedModel: "policy/daily", + routeDecision: trace, + attempts: [], + } as unknown as RequestLogContext; + let calls = 0; + + const response = await handleResponsesWithPolicyFallback( + request(), + {} as OcxConfig, + logCtx, + {}, + { + runCore: async (_req, _config, childLog) => { + calls += 1; + childLog.requestedModel = "policy/daily"; + childLog.routeDecision = trace; + return new Response( + JSON.stringify({ error: { message: "invalid request", type: "invalid_request_error" } }), + { status: 400, headers: { "content-type": "application/json" } }, + ); + }, + }, + ); + + expect(response.status).toBe(400); + expect(calls).toBe(1); + }); +}); From cd7ea8a8822f81d352e1e2dfdb726ba1d7afb0c3 Mon Sep 17 00:00:00 2001 From: Wibias <37517432+Wibias@users.noreply.github.com> Date: Sat, 8 Aug 2026 13:41:17 +0200 Subject: [PATCH 2/5] feat(routing): add policy candidate fallback wrapper --- src/server/responses/policy-fallback.ts | 156 ++++++++++++++++++++++++ 1 file changed, 156 insertions(+) create mode 100644 src/server/responses/policy-fallback.ts diff --git a/src/server/responses/policy-fallback.ts b/src/server/responses/policy-fallback.ts new file mode 100644 index 000000000..5a701c6f4 --- /dev/null +++ b/src/server/responses/policy-fallback.ts @@ -0,0 +1,156 @@ +import { comboFailureDecision } from "../../combos/failover"; +import { readBoundedResponseBody } from "../../lib/bounded-body"; +import { readJsonRequestBody } from "../request-decompress"; +import type { RequestLogContext } from "../request-log"; +import type { OcxConfig } from "../../types"; +import type { RouteCandidateTrace, RouteDecisionTraceV1 } from "../../routing/trace"; +import { handleResponses as handleResponsesCore } from "./core"; + +type CoreHandler = typeof handleResponsesCore; +type CoreOptions = Parameters[3]; + +export interface PolicyFallbackDeps { + runCore?: CoreHandler; +} + +function candidateKey(candidate: Pick): string { + return `${candidate.provider}\u0000${candidate.model}`; +} + +/** + * Rank the remaining candidates from the ORIGINAL policy trace. The initial + * decision stays immutable; fallback execution belongs in attempts[], not in a + * rewritten decision trace. + */ +export function rankPolicyFallbackCandidates( + trace: RouteDecisionTraceV1, + tried: ReadonlySet, +): RouteCandidateTrace[] { + return trace.candidates + .map((candidate, index) => ({ candidate, index })) + .filter(({ candidate }) => + candidate.eligible + && candidate.exclusions.length === 0 + && !tried.has(candidateKey(candidate))) + .sort((left, right) => { + const scoreDelta = (right.candidate.score?.total ?? Number.NEGATIVE_INFINITY) + - (left.candidate.score?.total ?? Number.NEGATIVE_INFINITY); + return scoreDelta || left.index - right.index; + }) + .map(({ candidate }) => candidate); +} + +function requestWithCandidate( + req: Request, + rawBody: Record, + candidate: Pick, +): Request { + const headers = new Headers(req.headers); + // The retry body is re-serialized JSON. Carrying the original compression or + // byte length would make the child request malformed. + headers.delete("content-encoding"); + headers.delete("content-length"); + headers.set("content-type", "application/json"); + return new Request(req.url, { + method: req.method, + headers, + body: JSON.stringify({ + ...rawBody, + model: `${candidate.provider}/${candidate.model}`, + }), + signal: req.signal, + }); +} + +function errorCodeFromText(text: string): string | undefined { + if (!text) return undefined; + try { + const payload = JSON.parse(text) as { + error?: { code?: unknown; type?: unknown }; + code?: unknown; + }; + const candidate = payload.error?.code ?? payload.error?.type ?? payload.code; + return typeof candidate === "string" ? candidate : undefined; + } catch { + return undefined; + } +} + +async function shouldHopPolicyCandidate(response: Response, signal?: AbortSignal): Promise { + if (response.status < 400) return false; + try { + const inspected = await readBoundedResponseBody(response.clone(), { signal }); + const text = inspected.displaySafe ? inspected.text : ""; + return comboFailureDecision(response.status, text, { + code: errorCodeFromText(text), + }) === "hop"; + } catch { + // If the error body cannot be inspected safely, do not invent a retry. + return false; + } +} + +function isPolicyDecision(trace: RouteDecisionTraceV1 | undefined): trace is RouteDecisionTraceV1 { + return trace?.routeKind === "policy" && !!trace.profile; +} + +/** + * Run a Responses request and, only for an explicitly selected policy profile, + * hop to the next eligible policy candidate after a retryable pre-success + * failure. The initial policy trace remains the canonical selection evidence; + * physical retries continue to accumulate in the existing request attempts. + */ +export async function handleResponsesWithPolicyFallback( + req: Request, + config: OcxConfig, + logCtx: RequestLogContext, + options: CoreOptions = {}, + deps: PolicyFallbackDeps = {}, +): Promise { + const runCore = deps.runCore ?? handleResponsesCore; + + // Capture a replayable, decompressed body before the core consumes the + // request. If decoding fails, defer entirely to the canonical core path so + // its existing error semantics remain unchanged. + let rawBody: Record | null = null; + try { + const parsed = await readJsonRequestBody(req.clone()); + if (parsed && typeof parsed === "object" && !Array.isArray(parsed)) { + rawBody = parsed as Record; + } + } catch { + // Core owns the client-facing parse/decompression error. + } + + let response = await runCore(req, config, logCtx, options); + const initialTrace = logCtx.routeDecision; + const initialRequestedModel = logCtx.requestedModel; + if (!rawBody || !isPolicyDecision(initialTrace)) return response; + + const tried = new Set([ + candidateKey({ provider: initialTrace.selected.provider, model: initialTrace.selected.model }), + ]); + + while (await shouldHopPolicyCandidate(response, req.signal)) { + if (req.signal.aborted) return response; + const next = rankPolicyFallbackCandidates(initialTrace, tried)[0]; + if (!next) return response; + tried.add(candidateKey(next)); + + const retryRequest = requestWithCandidate(req, rawBody, next); + try { + response = await runCore(retryRequest, config, logCtx, options); + } finally { + // A fallback child routes explicitly and therefore produces its own + // explicit-provider trace. Keep the original policy decision as the + // request-level WHY while retaining the child's physical model/provider + // and attempts on the mutable log context. + logCtx.requestedModel = initialRequestedModel; + logCtx.routeDecision = initialTrace; + } + } + + return response; +} + +export const handleResponses = handleResponsesWithPolicyFallback; From 457c33675715b0a5f5ccee314a5ef94f42e4b62a Mon Sep 17 00:00:00 2001 From: Wibias <37517432+Wibias@users.noreply.github.com> Date: Sat, 8 Aug 2026 13:41:35 +0200 Subject: [PATCH 3/5] feat(routing): route Responses through policy fallback --- src/server/responses.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/server/responses.ts b/src/server/responses.ts index 8c33c5bca..172baccbc 100644 --- a/src/server/responses.ts +++ b/src/server/responses.ts @@ -5,5 +5,6 @@ export type { MultiAgentGuidanceOptions, MultiAgentGuidanceDeps } from "./respon export { hasUnreadableEncryptedAgentTask, sanitizeEncryptedContentInPlace } from "./responses/encrypted-payload"; export { COMPACT_RESPONSE_MAX_BYTES, bufferCompactResponse, handleResponsesCompact } from "./responses/compact"; export { disableResponsesRequestTimeout, safeHostLabel, fetchWithHeaderTimeout } from "./responses/fetch-helpers"; -export { sidecarOutcomeRecorder, isShadowSourceModel, codexLogAccountId, usesCodexForwardPoolAuth, codexForwardTerminalOutcomeRecorder, decodeRequestErrorResponse, buildComboChildHeaders, handleResponses, linkAbortSignal } from "./responses/core"; -export { adapterNeedsForcedContinuation } from "./responses/core"; +export { sidecarOutcomeRecorder, isShadowSourceModel, codexLogAccountId, usesCodexForwardPoolAuth, codexForwardTerminalOutcomeRecorder, decodeRequestErrorResponse, buildComboChildHeaders, linkAbortSignal } from "./responses/core"; +export { handleResponses, handleResponsesWithPolicyFallback, rankPolicyFallbackCandidates } from "./responses/policy-fallback"; +export { adapterNeedsForcedContinuation } from "./responses/core"; \ No newline at end of file From 73c2c14ede91211e3d7dae8e7d5e502e47996aae Mon Sep 17 00:00:00 2001 From: Wibias <37517432+Wibias@users.noreply.github.com> Date: Sat, 8 Aug 2026 13:54:55 +0200 Subject: [PATCH 4/5] fix(routing): preserve distinct policy fallback attempts --- src/server/responses/policy-fallback.ts | 50 ++++++++++++------------- 1 file changed, 23 insertions(+), 27 deletions(-) diff --git a/src/server/responses/policy-fallback.ts b/src/server/responses/policy-fallback.ts index 5a701c6f4..fd63b40c3 100644 --- a/src/server/responses/policy-fallback.ts +++ b/src/server/responses/policy-fallback.ts @@ -1,7 +1,7 @@ import { comboFailureDecision } from "../../combos/failover"; import { readBoundedResponseBody } from "../../lib/bounded-body"; import { readJsonRequestBody } from "../request-decompress"; -import type { RequestLogContext } from "../request-log"; +import { finishRequestAttempt, type RequestLogContext } from "../request-log"; import type { OcxConfig } from "../../types"; import type { RouteCandidateTrace, RouteDecisionTraceV1 } from "../../routing/trace"; import { handleResponses as handleResponsesCore } from "./core"; @@ -46,18 +46,13 @@ function requestWithCandidate( candidate: Pick, ): Request { const headers = new Headers(req.headers); - // The retry body is re-serialized JSON. Carrying the original compression or - // byte length would make the child request malformed. headers.delete("content-encoding"); headers.delete("content-length"); headers.set("content-type", "application/json"); return new Request(req.url, { method: req.method, headers, - body: JSON.stringify({ - ...rawBody, - model: `${candidate.provider}/${candidate.model}`, - }), + body: JSON.stringify({ ...rawBody, model: `${candidate.provider}/${candidate.model}` }), signal: req.signal, }); } @@ -65,10 +60,7 @@ function requestWithCandidate( function errorCodeFromText(text: string): string | undefined { if (!text) return undefined; try { - const payload = JSON.parse(text) as { - error?: { code?: unknown; type?: unknown }; - code?: unknown; - }; + const payload = JSON.parse(text) as { error?: { code?: unknown; type?: unknown }; code?: unknown }; const candidate = payload.error?.code ?? payload.error?.type ?? payload.code; return typeof candidate === "string" ? candidate : undefined; } catch { @@ -77,15 +69,12 @@ function errorCodeFromText(text: string): string | undefined { } async function shouldHopPolicyCandidate(response: Response, signal?: AbortSignal): Promise { - if (response.status < 400) return false; + if (response.status < 400 || signal?.aborted) return false; try { const inspected = await readBoundedResponseBody(response.clone(), { signal }); const text = inspected.displaySafe ? inspected.text : ""; - return comboFailureDecision(response.status, text, { - code: errorCodeFromText(text), - }) === "hop"; + return comboFailureDecision(response.status, text, { code: errorCodeFromText(text) }) === "hop"; } catch { - // If the error body cannot be inspected safely, do not invent a retry. return false; } } @@ -94,6 +83,22 @@ function isPolicyDecision(trace: RouteDecisionTraceV1 | undefined): trace is Rou return trace?.routeKind === "policy" && !!trace.profile; } +/** Finalize the failed physical attempt so the retry receives a fresh attempt row. */ +function finishFailedPolicyAttempt(logCtx: RequestLogContext, status: number): void { + const attempt = logCtx.activeAttempt; + if (attempt) { + const startedAt = logCtx.activeAttemptStartedAt ?? Date.now(); + finishRequestAttempt(attempt, status, Math.max(0, Date.now() - startedAt), attempt.usage ?? logCtx.usage); + } + delete logCtx.activeAttempt; + delete logCtx.activeAttemptStartedAt; + delete logCtx.usage; + delete logCtx.usageFromBridge; + delete logCtx.upstreamError; + delete logCtx.terminalHttpStatus; + delete logCtx.terminalIncompleteReason; +} + /** * Run a Responses request and, only for an explicitly selected policy profile, * hop to the next eligible policy candidate after a retryable pre-success @@ -108,16 +113,10 @@ export async function handleResponsesWithPolicyFallback( deps: PolicyFallbackDeps = {}, ): Promise { const runCore = deps.runCore ?? handleResponsesCore; - - // Capture a replayable, decompressed body before the core consumes the - // request. If decoding fails, defer entirely to the canonical core path so - // its existing error semantics remain unchanged. let rawBody: Record | null = null; try { const parsed = await readJsonRequestBody(req.clone()); - if (parsed && typeof parsed === "object" && !Array.isArray(parsed)) { - rawBody = parsed as Record; - } + if (parsed && typeof parsed === "object" && !Array.isArray(parsed)) rawBody = parsed as Record; } catch { // Core owns the client-facing parse/decompression error. } @@ -137,14 +136,11 @@ export async function handleResponsesWithPolicyFallback( if (!next) return response; tried.add(candidateKey(next)); + finishFailedPolicyAttempt(logCtx, response.status); const retryRequest = requestWithCandidate(req, rawBody, next); try { response = await runCore(retryRequest, config, logCtx, options); } finally { - // A fallback child routes explicitly and therefore produces its own - // explicit-provider trace. Keep the original policy decision as the - // request-level WHY while retaining the child's physical model/provider - // and attempts on the mutable log context. logCtx.requestedModel = initialRequestedModel; logCtx.routeDecision = initialTrace; } From f8a683b28490eece886c1db0e8bebdc0f5ac0c83 Mon Sep 17 00:00:00 2001 From: Wibias <37517432+Wibias@users.noreply.github.com> Date: Sat, 8 Aug 2026 13:55:31 +0200 Subject: [PATCH 5/5] test(routing): cover fallback attempt and no-hop guards --- tests/routing-policy-fallback.test.ts | 182 ++++++++++++-------------- 1 file changed, 85 insertions(+), 97 deletions(-) diff --git a/tests/routing-policy-fallback.test.ts b/tests/routing-policy-fallback.test.ts index 11e9360d9..26503e5eb 100644 --- a/tests/routing-policy-fallback.test.ts +++ b/tests/routing-policy-fallback.test.ts @@ -1,7 +1,7 @@ import { describe, expect, test } from "bun:test"; import type { OcxConfig } from "../src/types"; -import type { RequestLogContext } from "../src/server/request-log"; +import { beginRequestAttempt, type RequestLogContext } from "../src/server/request-log"; import type { RouteDecisionTraceV1 } from "../src/routing/trace"; import { handleResponsesWithPolicyFallback, @@ -18,138 +18,126 @@ function policyTrace(): RouteDecisionTraceV1 { profile: { id: "daily", revision: "rev-1" }, requirements: [], candidates: [ - { - provider: "provider-a", - model: "model-a", - eligible: true, - exclusions: [], - score: { total: 0.90, components: {} }, - }, - { - provider: "provider-b", - model: "model-b", - eligible: true, - exclusions: [], - score: { total: 0.80, components: {} }, - }, - { - provider: "provider-c", - model: "model-c", - eligible: true, - exclusions: [], - score: { total: 0.80, components: {} }, - }, - { - provider: "provider-d", - model: "model-d", - eligible: false, - exclusions: [{ code: "tools" }], - score: { total: 1, components: {} }, - }, + { provider: "provider-a", model: "model-a", eligible: true, exclusions: [], score: { total: 0.90, components: {} } }, + { provider: "provider-b", model: "model-b", eligible: true, exclusions: [], score: { total: 0.80, components: {} } }, + { provider: "provider-c", model: "model-c", eligible: true, exclusions: [], score: { total: 0.80, components: {} } }, + { provider: "provider-d", model: "model-d", eligible: false, exclusions: [{ code: "tools" }], score: { total: 1, components: {} } }, ], - selected: { - candidateIndex: 0, - provider: "provider-a", - model: "model-a", - reason: "highest-score", - }, + selected: { candidateIndex: 0, provider: "provider-a", model: "model-a", reason: "highest-score" }, }; } -function request(): Request { +function request(signal?: AbortSignal): Request { return new Request("http://localhost/v1/responses", { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ model: "policy/daily", input: "hello", stream: false }), + signal, }); } +function seedAttempt(logCtx: RequestLogContext, provider: string, model: string): void { + if (logCtx.activeAttempt) return; + const attempt = beginRequestAttempt((logCtx.attempts?.length ?? 0) + 1, provider, model, "test"); + (logCtx.attempts ??= []).push(attempt); + logCtx.activeAttempt = attempt; + logCtx.activeAttemptStartedAt = Date.now(); +} + describe("policy candidate fallback", () => { test("ranks only eligible untried candidates by score and stable original order", () => { - const trace = policyTrace(); - const ranked = rankPolicyFallbackCandidates( - trace, - new Set(["provider-a\u0000model-a"]), - ); - + const ranked = rankPolicyFallbackCandidates(policyTrace(), new Set(["provider-a\u0000model-a"])); expect(ranked.map(candidate => `${candidate.provider}/${candidate.model}`)).toEqual([ "provider-b/model-b", "provider-c/model-c", ]); }); - test("retries the next policy candidate after a retryable pre-stream failure", async () => { + test("retries the next policy candidate and keeps distinct physical attempts", async () => { const trace = policyTrace(); - const logCtx = { - requestedModel: "policy/daily", - routeDecision: trace, - attempts: [], - } as unknown as RequestLogContext; + const logCtx = { requestedModel: "policy/daily", routeDecision: trace, attempts: [] } as unknown as RequestLogContext; const seenModels: string[] = []; - const response = await handleResponsesWithPolicyFallback( - request(), - {} as OcxConfig, - logCtx, - {}, - { - runCore: async (req, _config, childLog) => { - const body = await req.json() as { model: string }; - seenModels.push(body.model); - if (seenModels.length === 1) { - childLog.requestedModel = "policy/daily"; - childLog.routeDecision = trace; - return new Response( - JSON.stringify({ error: { message: "rate limited", type: "rate_limit_error" } }), - { status: 429, headers: { "content-type": "application/json" } }, - ); - } - childLog.requestedModel = body.model; - childLog.routeDecision = { - ...trace, - requestedModel: body.model, - routeKind: "explicit-provider", - profile: undefined, - }; - return new Response(JSON.stringify({ status: "completed" }), { status: 200 }); - }, + const response = await handleResponsesWithPolicyFallback(request(), {} as OcxConfig, logCtx, {}, { + runCore: async (req, _config, childLog) => { + const body = await req.json() as { model: string }; + seenModels.push(body.model); + const first = seenModels.length === 1; + seedAttempt(childLog, first ? "provider-a" : "provider-b", first ? "model-a" : "model-b"); + if (first) { + childLog.requestedModel = "policy/daily"; + childLog.routeDecision = trace; + return new Response(JSON.stringify({ error: { message: "rate limited", type: "rate_limit_error" } }), { + status: 429, + headers: { "content-type": "application/json" }, + }); + } + childLog.requestedModel = body.model; + childLog.routeDecision = { ...trace, requestedModel: body.model, routeKind: "explicit-provider", profile: undefined }; + return new Response(JSON.stringify({ status: "completed" }), { status: 200 }); }, - ); + }); expect(response.status).toBe(200); expect(seenModels).toEqual(["policy/daily", "provider-b/model-b"]); expect(logCtx.requestedModel).toBe("policy/daily"); expect(logCtx.routeDecision).toBe(trace); + expect(logCtx.attempts).toHaveLength(2); + expect(logCtx.attempts?.[0]).toMatchObject({ provider: "provider-a", model: "model-a", status: 429 }); + expect(logCtx.attempts?.[1]).toMatchObject({ provider: "provider-b", model: "model-b" }); + expect(logCtx.activeAttempt).toBe(logCtx.attempts?.[1]); }); test("does not switch candidates for terminal client/input failures", async () => { const trace = policyTrace(); - const logCtx = { - requestedModel: "policy/daily", - routeDecision: trace, - attempts: [], - } as unknown as RequestLogContext; + const logCtx = { requestedModel: "policy/daily", routeDecision: trace, attempts: [] } as unknown as RequestLogContext; let calls = 0; + const response = await handleResponsesWithPolicyFallback(request(), {} as OcxConfig, logCtx, {}, { + runCore: async (_req, _config, childLog) => { + calls += 1; + childLog.requestedModel = "policy/daily"; + childLog.routeDecision = trace; + return new Response(JSON.stringify({ error: { message: "invalid request", type: "invalid_request_error" } }), { + status: 400, + headers: { "content-type": "application/json" }, + }); + }, + }); + expect(response.status).toBe(400); + expect(calls).toBe(1); + }); - const response = await handleResponsesWithPolicyFallback( - request(), - {} as OcxConfig, - logCtx, - {}, - { - runCore: async (_req, _config, childLog) => { - calls += 1; - childLog.requestedModel = "policy/daily"; - childLog.routeDecision = trace; - return new Response( - JSON.stringify({ error: { message: "invalid request", type: "invalid_request_error" } }), - { status: 400, headers: { "content-type": "application/json" } }, - ); - }, + test("does not switch candidates after client cancellation", async () => { + const trace = policyTrace(); + const controller = new AbortController(); + const logCtx = { requestedModel: "policy/daily", routeDecision: trace, attempts: [] } as unknown as RequestLogContext; + let calls = 0; + const response = await handleResponsesWithPolicyFallback(request(controller.signal), {} as OcxConfig, logCtx, {}, { + runCore: async (_req, _config, childLog) => { + calls += 1; + childLog.routeDecision = trace; + controller.abort(); + return new Response(JSON.stringify({ error: { type: "rate_limit_error" } }), { status: 429 }); }, - ); + }); + expect(response.status).toBe(429); + expect(calls).toBe(1); + }); - expect(response.status).toBe(400); + test("does not switch candidates after a streaming response has started", async () => { + const trace = policyTrace(); + const logCtx = { requestedModel: "policy/daily", routeDecision: trace, attempts: [] } as unknown as RequestLogContext; + let calls = 0; + const body = "data: {\"type\":\"response.output_text.delta\",\"delta\":\"hello\"}\n\ndata: {\"type\":\"response.failed\"}\n\n"; + const response = await handleResponsesWithPolicyFallback(request(), {} as OcxConfig, logCtx, {}, { + runCore: async (_req, _config, childLog) => { + calls += 1; + childLog.routeDecision = trace; + return new Response(body, { status: 200, headers: { "content-type": "text/event-stream" } }); + }, + }); + expect(response.status).toBe(200); + expect(await response.text()).toContain("hello"); expect(calls).toBe(1); }); });