fix(responses): carry WS dispatch override, pooled key rotation, usage accounting and Kiro refusal hop (B3)

scripts/test-layout/layout.json

@@ -744,7 +744,7 @@
     "combo-stream-preflight.test.ts": "routing",
     "combo-workspace-data.test.ts": "gui",
     "combo-workspace-jev-decision.test.ts": "gui",
-    "combos.test.ts": "codex-integration",
+    "combos.test.ts": "codex-integration", "combos-kiro-profile-hop.test.ts": "codex-integration",
     "command-code-error-finish.test.ts": "providers",
     "command-code-efforts.test.ts": "providers",
     "command-code-project-context.test.ts": "providers",

src/adapters/kiro-refusal.ts

@@ -1,8 +1,13 @@
 /** Account-scoped Kiro refusals. Unknown data never convicts an account. */
 import { BOUNDED_BODY_MAX_BYTES } from "../lib/bounded-body";
-export type KiroRefusalKind = "rate" | "monthly_quota" | "suspended" | "other";
+export type KiroRefusalKind = "rate" | "monthly_quota" | "suspended" | "profile" | "other";
 export interface KiroRefusal { kind: KiroRefusalKind; resetAt?: number }
 
+/** The vendor's own name for the body validator that rejects a request for this identity. */
+const VALIDATION_EXCEPTION = "ValidationException";
+/** Kiro's wording for that rejection; the code alone is also what a genuinely bad payload gets. */
+const IMPROPER_REQUEST_WORD = /improperly formed request/i;
+
 const SUSPENSION_WORDS = [
   "temporarily suspended", "temporarily is suspended",
   "locked your account", "locked it as a",
@@ -22,9 +27,44 @@ export function classifyKiroRefusal(status: number, bodyText: string): KiroRefus
   } catch { /* Unrecognised bodies cannot establish an account verdict. */ }
   if ((status === 400 || status === 429) && reason === "MONTHLY_REQUEST_COUNT")
     return { kind: "monthly_quota" };
+  // A 400 whose own validator calls the body improperly formed is not proof that the request is
+  // impossible: Kiro answers this shape when the identity a request was routed under cannot serve
+  // it. Treating it as `other` leaves the account selected and the same 400 reaches the client.
+  // Both vendor tokens are required, so a bad payload or a caller echoing one token stays terminal.
+  if (status === 400 && typeof message === "string" && IMPROPER_REQUEST_WORD.test(message)
+    && (reason === VALIDATION_EXCEPTION || message.includes(VALIDATION_EXCEPTION)))
+    return { kind: "profile" };
   if (status === 403 && (reason === "TEMPORARILY_SUSPENDED"
     || (typeof message === "string" && SUSPENSION_WORDS.some(word => message.toLowerCase().includes(word)))))
     return { kind: "suspended" };
   if (status === 429) return { kind: "rate" };
   return { kind: "other" };
 }
+
+/** The adapter's own prefix for a normalized 400 (`normalizedKiroErrorMessage` in kiro-errors.ts). */
+const KIRO_INVALID_REQUEST_PREFIX = "Kiro invalid request: ";
+
+/**
+ * The same verdict at the combo layer. There the text is either the adapter's normalized message
+ * ("Kiro invalid request: ValidationException: Improperly formed request", what a combo child's
+ * failure carries) or a raw JSON body, each possibly behind the "Provider error 400: " frame.
+ * JSON goes through the parsed `reason`/`message` rules of `classifyKiroRefusal` (tokens in
+ * unrelated fields never count); prose counts only behind the adapter's own prefix. Callers must
+ * also know the failed target is a Kiro adapter.
+ */
+export function isKiroAccountProfileRefusal(status: number, bodyText: string): boolean {
+  if (status !== 400 || Buffer.byteLength(bodyText, "utf8") > BOUNDED_BODY_MAX_BYTES) return false;
+  const text = bodyText.replace(/^Provider error \d{3}: /, "");
+  if (text.startsWith(KIRO_INVALID_REQUEST_PREFIX)) {
+    const detail = text.slice(KIRO_INVALID_REQUEST_PREFIX.length);
+    return detail.includes(VALIDATION_EXCEPTION) && IMPROPER_REQUEST_WORD.test(detail);
+  }
+  try {
+    const row: unknown = JSON.parse(text);
+    if (!row || typeof row !== "object" || Array.isArray(row)) return false;
+    const envelope = (row as Record<string, unknown>).error;
+    const body = envelope && typeof envelope === "object" && !Array.isArray(envelope) ? envelope : row;
+    const { reason, message } = body as Record<string, unknown>;
+    return classifyKiroRefusal(400, JSON.stringify({ reason, message })).kind === "profile";
+  } catch { return false; }
+}

src/combos/failover.ts

@@ -2,6 +2,7 @@ import { parseResetCooldownMs } from "../codex/routing";
 import { classifyError, isCyberPolicyCode } from "../lib/errors";
 import { isNonReplayableUpstreamCode } from "../lib/upstream-retry";
 import { isAnthropicBillingFailure } from "../providers/anthropic-billing-refusal";
+import { isKiroAccountProfileRefusal } from "../adapters/kiro-refusal";
 import type { OcxComboTarget } from "../types";
 import { targetKey } from "./types";
 import {
@@ -772,7 +773,7 @@ export function codexAccountModelRefusal(
 export function comboFailureDecision(
   status: number,
   message: string,
-  options?: { code?: string | null; codexModelRefusal?: CodexAccountModelRefusal },
+  options?: { code?: string | null; codexModelRefusal?: CodexAccountModelRefusal; failedAdapter?: string },
 ): ComboFailureDecision {
   if (status === 499) return "stop";
   if (message.toLowerCase().includes("origin_rejected")) return "stop";
@@ -856,6 +857,12 @@ export function comboFailureDecision(
   // could have served the request (#4903).
   if (isResponseFormatCapabilityRefusal(status, message, options?.code)) return "hop";
   if (isAnthropicBillingFailure(status, message, options?.code)) return "hop";
+  // Kiro's body validator rejects the request for the identity it was routed under. The account
+  // layer rotates logins first; this is the no-alternate case, so the next combo target is the
+  // only remaining chance to serve the turn. Only a caller that knows the failed target is a Kiro
+  // adapter grants it (the words alone prove nothing for another provider); it must precede the
+  // generic invalid_request_error stop.
+  if (options?.failedAdapter === "kiro" && isKiroAccountProfileRefusal(status, message)) return "hop";
   if (["origin_rejected", "context_length_exceeded", "invalid_request_error"].includes(error.code ?? "")) {
     return "stop";
   }

src/server/messages-native.ts

@@ -849,6 +849,17 @@ export async function handleNativeMessages(options: HandleNativeMessagesOptions)
   }
   releaseRetainedRequest();
 
+  // The last pool account refused on spent quota and rotation found no replacement: answer the
+  // usage limit with its reset (#6991), as the bridge lane does, not the upstream "try again later".
+  const poolUsageLimit = response.status === 429 && oauthBinding
+    ? anthropicRoutingFor(nativeInstance!).anthropicPoolUsageLimitAfterRefusal(Date.now(), oauthBinding.routeDecision, route.modelId)
+    : null;
+  if (poolUsageLimit) {
+    discard(response);
+    cleanupAbort();
+    logCtx.errorCode = "usage_limit_reached";
+    return retainUpstreamMessagesRequestId(localOAuthFailure(poolUsageLimit)!, response.headers);
+  }
   if (!response.ok) {
     let bodyText = "";
     try {

src/server/request-log.ts

@@ -1720,8 +1720,10 @@ export function noteProviderAttemptSend(
   if (attempt && attempt.sendCount > 0 && attemptAccountChanged(previous, next, attempt.provider, logCtx.provider)) {
     // An input estimate is not evidence that a failed send used that many tokens.
     delete attempt.inputTokenEstimate;
+    // An account-scoped recovery kind names the status the sealed account answered.
     finishRequestAttempt(attempt, attempt.status >= 100 ? attempt.status
-      : recovery === "key-401" ? 401 : recovery?.includes("429") ? 429 : 502,
+      : recovery === "key-401" || recovery === "oauth-401" ? 401 : recovery === "oauth-account-403" ? 403
+        : recovery === "oauth-account-402" ? 402 : recovery?.includes("429") ? 429 : 502,
     Date.now() - (logCtx.activeAttemptStartedAt ?? Date.now()), attempt.usage);
     // This attempt is being sealed because a NAMED recovery rejected it, so the recovery kind
     // is direct evidence here rather than an inference from history. Without this the sealed
@@ -1850,16 +1852,20 @@ export function finishRequestAttempt(
 export function aggregateAttemptUsage(
   attempts: readonly PersistedUsageAttempt[],
 ): FinalizedUsageResult {
-  const status: UsageStatus = attempts.length > 0
-    && attempts.every(attempt => attempt.usageStatus === "unsupported")
+  // A usage-less attempt that never reached upstream spent nothing (summary.ts `spentNothing`),
+  // so it cannot leave the row's usage unknown.
+  const measured = attempts.filter(attempt => attempt.usage
+    || (attempt.sendCount !== 0 && attempt.locallyAnswered !== true));
+  const status: UsageStatus = measured.length > 0
+    && measured.every(attempt => attempt.usageStatus === "unsupported")
     ? "unsupported"
-    : attempts.some(attempt => (
+    : measured.some(attempt => (
         attempt.usageStatus === "unreported" || attempt.usageStatus === "unsupported"
       ))
       ? "unreported"
-      : attempts.some(attempt => attempt.usageStatus === "estimated")
+      : measured.some(attempt => attempt.usageStatus === "estimated")
         ? "estimated"
-        : attempts.length > 0
+        : measured.length > 0
           ? "reported"
           : "unreported";
 
@@ -1876,7 +1882,10 @@ export function aggregateAttemptUsage(
     return present.length > 0 ? present.reduce((sum, value) => sum + value, 0) : undefined;
   };
   const cachedInputTokens = sumOptional("cachedInputTokens");
-  const cacheReadInputTokens = sumOptional("cacheReadInputTokens");
+  // Readers take `cacheReadInputTokens ?? cachedInputTokens` (usage/cost.ts), so once any attempt
+  // reports the read field, an attempt that only reports `cachedInputTokens` must count in it too.
+  const cacheReadInputTokens = sumOptional("cacheReadInputTokens") === undefined ? undefined
+    : usages.reduce((sum, usage) => sum + (usage.cacheReadInputTokens ?? usage.cachedInputTokens ?? 0), 0);
   const cacheCreationInputTokens = sumOptional("cacheCreationInputTokens");
   const cacheCreation1hInputTokens = sumOptional("cacheCreation1hInputTokens");
   const reasoningOutputTokens = sumOptional("reasoningOutputTokens");

src/server/responses/core-combo.ts

@@ -1165,6 +1165,9 @@ export async function executeComboResponses(
       : comboFailureDecision(failure.response.status, failure.classificationText, {
         code: failure.upstreamCode,
         codexModelRefusal: failure.codexModelRefusal,
+        // The adapter the failed leg actually resolved: a configured `kiro` provider can route a
+        // model through another wire via `modelAdapters`, and an unresolved leg grants no hop.
+        failedAdapter: childLog.providerAdapter,
       });
     const wantsStream = (rawBody as { stream?: unknown } | null)?.stream === true;
     // Local byte admission has its own diagnostic; do not relabel it as an upstream refusal.

src/server/responses/fetch-helpers.ts

@@ -248,6 +248,20 @@ export function sendWithConnectionPolicy(
   });
 }
 
+interface CodexWsExecutor { provider: OcxProviderConfig; http: typeof globalThis.fetch }
+const codexWsExecutors = new WeakMap<object, CodexWsExecutor>();
+const codexWsExecutor = (send: (input: Parameters<typeof globalThis.fetch>[0], init?: RequestInit) => Promise<Response>, binding: CodexWsExecutor) => {
+  codexWsExecutors.set(send, binding);
+  return send as typeof globalThis.fetch;
+};
+/**
+ * Set for the executor a `dispatchOverride` receives on the WS branch: it must not get HTTP-only
+ * send policy. `provider` is the one WS eligibility was checked against; a dispatch-time
+ * replacement was not, so it sends through `http` (the same fallback) instead.
+ */
+export const codexWsExecutorBinding = (execute: unknown): CodexWsExecutor | undefined =>
+  typeof execute === "function" ? codexWsExecutors.get(execute) : undefined;
+
 export interface ProviderFetchOptions {
   accountAutoPause?: (name: string, provider: OcxProviderConfig, headers: Headers) => AccountAutoPauseObserver | undefined;
   /**
@@ -330,7 +344,9 @@ export function providerFetch(
   ) as typeof globalThis.fetch);
   const httpFetch = Object.assign(
     async (input: Parameters<typeof globalThis.fetch>[0], init?: RequestInit,
-      reacquirePacing?: (provider: OcxProviderConfig, modelId?: string) => Promise<void>) => {
+      reacquirePacing?: (provider: OcxProviderConfig, modelId?: string) => Promise<void>,
+      // False only for the HTTP fallback of a WS send the override already wrapped: one logical send.
+      override: ProviderFetchOptions["dispatchOverride"] | null = options.dispatchOverride) => {
       // Refuse before any dispatch side effect where that is sound. `beforeDispatch` commits
       // attempt accounting and consumes admission state, so a refusal firing after it would
       // charge an attempt for a send that never happens, and a throwing hook would mask the
@@ -340,7 +356,7 @@ export function providerFetch(
       // full decision can be made now. With an override, only the configured value is checked:
       // the override may rebuild against a different host and select a different transport, and
       // refusing on this destination would reject a request whose real route is fine.
-      if (options.dispatchOverride) egressFor(input);
+      if (override) egressFor(input);
       else providerEgressSendInit(egressBinding, transport, input);
       // The hook inspects the outgoing headers and refuses the send by throwing; it is not a
       // mutator, and the copy it receives is deliberately not threaded onward. `Connection`
@@ -357,8 +373,8 @@ export function providerFetch(
           else if (!notified) { options.onPhysicalDispatch!(); notified = true; }
         } } : {}),
       };
-      return options.dispatchOverride
-        ? options.dispatchOverride(input, dispatchInit, dispatch, reacquirePacing)
+      return override
+        ? override(input, dispatchInit, dispatch, reacquirePacing)
         : dispatch(input, dispatchInit);
     },
     { preconnect },
@@ -397,9 +413,15 @@ export function providerFetch(
         options.onPhysicalDispatch?.();
       };
       const fallback = ((url: Parameters<typeof globalThis.fetch>[0], fallbackInit?: RequestInit) =>
-        httpFetch(url, { ...fallbackInit, [WS_FALLBACK_RECEIPT]: notifyOnce } as RequestInit, reacquirePacing)) as typeof globalThis.fetch;
-      return codexWsUpstreamFetch(input, init, fallback, runtime, options.onCodexWsQuota, options.beforeDispatch, options.nativeControl,
-        async () => { (await waitForPacing(init.signal ?? undefined))?.release(); }, notifyOnce, options.preOutputRefusalHold);
+        httpFetch(url, { ...fallbackInit, [WS_FALLBACK_RECEIPT]: notifyOnce } as RequestInit, reacquirePacing, null)) as typeof globalThis.fetch;
+      const wsSend = codexWsExecutor((url: Parameters<typeof globalThis.fetch>[0], wsInit?: RequestInit) => codexWsUpstreamFetch(String(url), wsInit ?? init,
+        fallback, runtime, options.onCodexWsQuota, options.beforeDispatch, options.nativeControl,
+        async () => { (await waitForPacing(init.signal ?? undefined))?.release(); }, notifyOnce, options.preOutputRefusalHold),
+        { provider, http: markEgressTransparentExecutor(fallback) });
+      // A WS frame is a physical send like any other, so it runs the same override: that is where a
+      // key-auth send is committed to its attempt row and the send budget is rebound to the
+      // credential that left. Without it a key-auth WS turn logged `sendCount: 0`.
+      return options.dispatchOverride ? options.dispatchOverride(input, init, wsSend, reacquirePacing) : wsSend(input, init);
     }
     return httpFetch(input, init, reacquirePacing);
   };

src/server/responses/passthrough-dispatch.ts

@@ -1,4 +1,4 @@
-import { recoverAutoPausedAccount } from "./account-auto-pause-recovery";
+import { recoverAutoPausedAccount, rotatePassthroughKeyOn429 } from "./passthrough-key-rotation";
 import { adoptCodexWsStage, capturePassthroughAffinity } from "./passthrough-observations";
 import { spendLedgerStorageErrorResponse } from "./spend-storage-error";
 import { requestPacingOverloadResponse } from "./pacing-overload";
@@ -1606,6 +1606,7 @@ export async function preparePassthroughExchange(
         return transportFailureResponse(err);
       }
     }
+    const keyRotation = await rotatePassthroughKeyOn429({ config, route, parsed, response: upstreamResponse, adapterName: transportState.adapter.name, budgetSpent: () => sendBudgetExhausted(transientSendAttempts()), refetch: () => rebuildAndRefetch("key-429") }); if (keyRotation) { if ("failed" in keyRotation) return keyRotation.failed; upstreamResponse = keyRotation; continue passthroughRecovery; }
     const captureAffinityResponse = (
       response: Response,
       captureAuthCtx: CodexAuthContext = admissionState.authCtx,

src/server/responses/passthrough-key-rotation.ts

@@ -0,0 +1,40 @@
+import { isNonReplayableResponse } from "../../lib/upstream-retry";
+import { hasKeyPoolFailover, rotateProviderTransportOn429 } from "../../providers/key-failover";
+import type { RouteResult } from "../../router";
+import type { OcxConfig, OcxParsedRequest } from "../../types";
+import { bindRouteReasoningReplayScope } from "./core-replay";
+
+// Passthrough's credential moves share one import site so passthrough-dispatch.ts stays under its line cap.
+export { recoverAutoPausedAccount } from "./account-auto-pause-recovery";
+
+/**
+ * Key-pool failover for a native passthrough 429 before any output. This wire returns before the
+ * adapter loop's key-429 arm, so without it a pooled key's 429 reached the client while the same
+ * refusal rotates on the adapter and chat-native wires. Distributed (round-robin) selection already
+ * cools the key on the send and deliberately answers without a replay. A synthesized replay refusal
+ * is not a provider rate limit and must not cool a key. A spent send budget is checked before any
+ * key is cooled, so it keeps the real 429 body. Returns undefined when nothing rotated.
+ */
+export async function rotatePassthroughKeyOn429<T>(options: {
+  config: OcxConfig;
+  route: Pick<RouteResult, "provider" | "providerName">;
+  parsed: OcxParsedRequest;
+  response: Response;
+  adapterName: string;
+  budgetSpent: () => boolean;
+  refetch: () => Promise<T>;
+}): Promise<T | undefined> {
+  const { config, route, parsed, response, adapterName, refetch } = options;
+  if (response.status !== 429 || isNonReplayableResponse(response) || !hasKeyPoolFailover(route.provider)
+    || route.provider._apiKeyAttempt?.distributed || options.budgetSpent()) return;
+  const rotated = rotateProviderTransportOn429(config, route.providerName, route.provider, {
+    retryAfter: response.headers.get("retry-after"),
+    attemptedKey: route.provider.apiKey,
+    promptCacheKey: parsed.options.promptCacheKey,
+  });
+  if (!rotated) return;
+  try { void response.body?.cancel().catch(() => {}); } catch { /* already closed */ }
+  route.provider = rotated;
+  bindRouteReasoningReplayScope({ parsed, providerName: route.providerName, provider: rotated, adapterName });
+  return refetch();
+}