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 diff --git a/src/server/responses/policy-fallback.ts b/src/server/responses/policy-fallback.ts new file mode 100644 index 000000000..fd63b40c3 --- /dev/null +++ b/src/server/responses/policy-fallback.ts @@ -0,0 +1,152 @@ +import { comboFailureDecision } from "../../combos/failover"; +import { readBoundedResponseBody } from "../../lib/bounded-body"; +import { readJsonRequestBody } from "../request-decompress"; +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"; + +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); + 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 || 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"; + } catch { + return false; + } +} + +function isPolicyDecision(trace: RouteDecisionTraceV1 | undefined): trace is RouteDecisionTraceV1 { + 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 + * 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; + 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)); + + finishFailedPolicyAttempt(logCtx, response.status); + const retryRequest = requestWithCandidate(req, rawBody, next); + try { + response = await runCore(retryRequest, config, logCtx, options); + } finally { + logCtx.requestedModel = initialRequestedModel; + logCtx.routeDecision = initialTrace; + } + } + + return response; +} + +export const handleResponses = handleResponsesWithPolicyFallback; diff --git a/tests/routing-policy-fallback.test.ts b/tests/routing-policy-fallback.test.ts new file mode 100644 index 000000000..26503e5eb --- /dev/null +++ b/tests/routing-policy-fallback.test.ts @@ -0,0 +1,143 @@ +import { describe, expect, test } from "bun:test"; + +import type { OcxConfig } from "../src/types"; +import { beginRequestAttempt, 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(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 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 and keeps distinct physical attempts", 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); + 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; + 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); + }); + + 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); + }); + + 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); + }); +});