feat(routing): fail over between policy candidates on retryable failures

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";

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<CoreHandler>[3];
+
+export interface PolicyFallbackDeps {
+  runCore?: CoreHandler;
+}
+
+function candidateKey(candidate: Pick<RouteCandidateTrace, "provider" | "model">): 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<string>,
+): 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<string, unknown>,
+  candidate: Pick<RouteCandidateTrace, "provider" | "model">,
+): 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<boolean> {
+  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<Response> {
+  const runCore = deps.runCore ?? handleResponsesCore;
+  let rawBody: Record<string, unknown> | null = null;
+  try {
+    const parsed = await readJsonRequestBody(req.clone());
+    if (parsed && typeof parsed === "object" && !Array.isArray(parsed)) rawBody = parsed as Record<string, unknown>;
+  } 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<string>([
+    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;

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