refactor(provider): share error-body reading, adapter mechanics, protocol sets

src/auth/codex/usage-limit-error.ts

@@ -13,6 +13,12 @@
  * other providers as Codex.
  */
 
+import {
+  coerceBody,
+  readErrorNode,
+  resetSecondsFromNode,
+} from "../../provider/error-body.js";
+
 export interface CodexUsageLimitError {
   readonly code: string;
   readonly message: string;
@@ -26,48 +32,6 @@ const USAGE_LIMIT_CODES = new Set([
   "usage_limit_exceeded",
 ]);
 
-function asRecord(value: unknown): Record<string, unknown> | undefined {
-  if (typeof value !== "object" || value === null || Array.isArray(value))
-    return undefined;
-  return value as Record<string, unknown>;
-}
-
-function tryParseJSON(text: string): unknown {
-  try {
-    return JSON.parse(text) as unknown;
-  } catch {
-    return undefined;
-  }
-}
-
-function coerceBody(raw: unknown): unknown {
-  if (typeof raw === "string") {
-    const trimmed = raw.trim();
-    if (trimmed.length === 0) return undefined;
-    return tryParseJSON(trimmed) ?? raw;
-  }
-  return raw;
-}
-
-function readErrorNode(body: unknown): Record<string, unknown> | undefined {
-  const root = asRecord(body);
-  if (root === undefined) return undefined;
-
-  const detail = asRecord(root["detail"]);
-  if (detail !== undefined) {
-    const nested = asRecord(detail["error"]);
-    if (nested !== undefined) return nested;
-    // Some gateways put the fields directly under detail.
-    if (typeof detail["code"] === "string") return detail;
-  }
-
-  const top = asRecord(root["error"]);
-  if (top !== undefined) return top;
-
-  if (typeof root["code"] === "string") return root;
-  return undefined;
-}
-
 /**
  * Nested Codex `detail.error.message` when the harness left it on `raw` while
  * classifying the HTTP statusText. Unlike `parseCodexUsageLimitError`, this
@@ -106,16 +70,7 @@ export function parseCodexUsageLimitError(
         ? node["planType"]
         : undefined;
 
-  const resetsRaw =
-    node["resets_in_seconds"] ??
-    node["resetsInSeconds"] ??
-    node["reset_after_seconds"];
-  const resetsInSeconds =
-    typeof resetsRaw === "number" &&
-    Number.isFinite(resetsRaw) &&
-    resetsRaw >= 0
-      ? Math.floor(resetsRaw)
-      : undefined;
+  const resetsInSeconds = resetSecondsFromNode(node);
 
   return {
     code,

src/inference-gateway-error.test.ts

@@ -308,6 +308,39 @@ describe("normalizeInferenceErrorForRetry", () => {
     );
   });
 
+  test("Codex bare 429 with a body reset window retries on the body, not the header (CL-10664)", () => {
+    const normalized = normalizeInferenceErrorForRetry({
+      ...BARE_429,
+      providerId: "codex/acme-labs",
+      raw: {
+        error: {
+          code: "rate_limit_exceeded",
+          message: "Too Many Requests",
+          resets_in_seconds: 7,
+        },
+      },
+    });
+    expect(normalized.category).toBe("retryable");
+    expect(normalized.message).toBe("Rate limited");
+    expect(normalized.retryAfterMs).toBe(7_000);
+  });
+
+  test("non-Codex 429s keep the header retry-after; a body reset is not read (CL-10664)", () => {
+    const normalized = normalizeInferenceErrorForRetry({
+      ...BARE_429,
+      providerId: "xai/alice",
+      raw: {
+        error: {
+          code: "rate_limit_exceeded",
+          message: "Too Many Requests",
+          resets_in_seconds: 7,
+        },
+      },
+    });
+    expect(normalized.category).toBe("retryable");
+    expect(normalized.retryAfterMs).toBe(45_000);
+  });
+
   test.each([{}, { providerId: "openai" }])(
     "bare 429 without a known provider keeps intx's quota_exhausted (%j)",
     (context) => {

src/inference-gateway-error.ts

@@ -19,6 +19,12 @@ import {
   xaiProfileFromProviderName,
 } from "./config/xai-providers.js";
 import { isXaiGrokLeafProvider } from "./subagent/provider-family.js";
+import {
+  coerceBody,
+  readErrorNode,
+  resetSecondsFromNode,
+  tryParseJSON,
+} from "./provider/error-body.js";
 
 export interface InferenceErrorLike {
   category: string;
@@ -155,14 +161,6 @@ export function isGatewayOverloadInferenceError(
   return false;
 }
 
-function tryParseJSON(text: string): unknown {
-  try {
-    return JSON.parse(text) as unknown;
-  } catch {
-    return undefined;
-  }
-}
-
 /**
  * Known OpenCode Go errors: provider context (id/flag/request URL) or Go
  * body markers; bare `provider_rate_limit_exceeded` alone is not enough —
@@ -352,25 +350,28 @@ export function isXaiShortRateLimitInferenceError(
 /**
  * Shared skeleton behind the twin rate-limit normalizers: known-provider
  * quota 429s without usage-limit bodies become retryable with scrubbed copy.
+ * An optional body reader lifts a carrier's reset window onto `retryAfterMs`
+ * when the wire omits a Retry-After header (body wins, matching the
+ * usage-limit path).
  */
 function normalizeProviderRateLimitError(
   error: InferenceErrorWithGoContext,
   isKnownProvider: (providerId: string | undefined) => boolean,
   isUsageLimit: (error: InferenceErrorLike) => boolean,
+  retryFromBody?: (error: InferenceErrorLike) => number | undefined,
 ): InferenceError {
   if (error.statusCode !== 429) return error;
   if (error.category !== "quota_exhausted") return error;
   if (!isKnownProvider(error.providerId)) return error;
   if (isUsageLimit(error)) return error;
 
+  const retryAfterMs = retryFromBody?.(error) ?? error.retryAfterMs;
   return {
     category: "retryable",
     message: RATE_LIMIT_USER_MESSAGE,
     statusCode: 429,
     ...(error.raw !== undefined ? { raw: error.raw } : {}),
-    ...(error.retryAfterMs !== undefined
-      ? { retryAfterMs: error.retryAfterMs }
-      : {}),
+    ...(retryAfterMs !== undefined ? { retryAfterMs } : {}),
   };
 }
 
@@ -430,6 +431,20 @@ export function isCodexShortRateLimitInferenceError(
   );
 }
 
+/**
+ * Codex sends its reset window in the 429 body, not a header; when a bare
+ * rate-limit body carries one, retry on that window instead of the policy
+ * default. Exact usage-limit codes never reach here (the skeleton vetoes).
+ */
+function codexRetryFromBody(error: InferenceErrorLike): number | undefined {
+  const node = readErrorNode(coerceBody(error.raw));
+  if (node === undefined) return undefined;
+  const seconds = resetSecondsFromNode(node);
+  if (seconds === undefined) return undefined;
+  if (seconds <= 0) return 0;
+  return seconds * 1000;
+}
+
 /**
  * Bare 429 in a known-Codex context reclassifies as retryable; intx
  * defaults bare 429 to quota_exhausted; nested `usage_limit_reached` keeps
@@ -442,6 +457,7 @@ export function normalizeCodexRateLimitError(
     error,
     isKnownCodexProviderId,
     hasCodexUsageLimit,
+    codexRetryFromBody,
   );
 }
 

src/provider/adapter-decorators.ts

@@ -0,0 +1,90 @@
+/**
+ * Shared adapter decorator mechanics. Only the JSON-body / extra-header /
+ * registry plumbing is shared; the per-adapter semantics stay in each
+ * adapter's rewrite or header factory (codex parallel_tool_calls, DeepSeek
+ * reasoning strip and V4 quirks, Bifrost virtual-key sentinel, anthropic
+ * session header). Every decorator is a null overhead: when its hook bails,
+ * the base request's body and headers pass through byte-identical.
+ */
+
+import type {
+  AdapterRegistry,
+  BuiltRequest,
+  ProviderAdapter,
+} from "@intx/inference";
+import type {
+  ConversationTurn,
+  InferenceOptions,
+  LastCycleSource,
+} from "@intx/types/runtime";
+
+export type DecoratorContext = {
+  readonly messages: readonly ConversationTurn[];
+  readonly model: string;
+  readonly options: InferenceOptions;
+};
+
+/** Parse the built body, apply the rewrite, and re-stringify only when the
+ * rewrite reports a change. An optional gate keeps adapters that only
+ * sometimes rewrite on their zero-parse fast path. */
+export function withJsonBodyRewrite(
+  adapter: ProviderAdapter,
+  rewrite: (
+    body: Record<string, unknown>,
+    context: DecoratorContext,
+  ) => boolean,
+  gate?: (request: BuiltRequest, context: DecoratorContext) => boolean,
+): ProviderAdapter {
+  return {
+    ...adapter,
+    buildRequest: (messages, model, options) => {
+      const built = adapter.buildRequest(messages, model, options);
+      const context: DecoratorContext = { messages, model, options };
+      if (gate !== undefined && !gate(built, context)) return built;
+      let body: Record<string, unknown>;
+      try {
+        body = JSON.parse(built.body) as Record<string, unknown>;
+      } catch {
+        return built;
+      }
+      if (!rewrite(body, context)) return built;
+      return { ...built, body: JSON.stringify(body) };
+    },
+  };
+}
+
+/** Merge the factory's headers over the built request's own; the spread
+ * never removes or rewrites a header the base adapter emitted. */
+export function withExtraHeaders(
+  adapter: ProviderAdapter,
+  headers: (context: DecoratorContext) => Record<string, string>,
+): ProviderAdapter {
+  return {
+    ...adapter,
+    buildRequest: (messages, model, options) => {
+      const built = adapter.buildRequest(messages, model, options);
+      const context: DecoratorContext = { messages, model, options };
+      return {
+        ...built,
+        headers: { ...built.headers, ...headers(context) },
+      };
+    },
+  };
+}
+
+/** Wrap an adapter registry so every resolved adapter passes through the
+ * decorator; membership is delegated untouched. */
+export function withDecoratedAdapters(
+  adapters: AdapterRegistry,
+  decorate: (
+    adapter: ProviderAdapter,
+    source: LastCycleSource,
+  ) => ProviderAdapter,
+): AdapterRegistry {
+  return {
+    has: (provider) => adapters.has(provider),
+    resolve(source, quirks) {
+      return decorate(adapters.resolve(source, quirks), source);
+    },
+  };
+}

src/provider/anthropic-cache-breakpoint.ts

@@ -1,19 +1,13 @@
 import type {
   AdapterRegistry,
-  BuiltRequest,
   ExtendedInferenceOptions,
   ProviderAdapter,
 } from "@intx/inference";
 import {
-  OPENCODE_GO_MESSAGES_PROVIDER,
-  ZEN_MESSAGES_PROVIDER,
-} from "./anthropic-session-adapter.js";
-
-const ANTHROPIC_MESSAGES_PROVIDERS: ReadonlySet<string> = new Set([
-  "anthropic",
-  ZEN_MESSAGES_PROVIDER,
-  OPENCODE_GO_MESSAGES_PROVIDER,
-]);
+  withDecoratedAdapters,
+  withJsonBodyRewrite,
+} from "./adapter-decorators.js";
+import { ANTHROPIC_MESSAGES_PROTOCOLS } from "./provider-protocols.js";
 
 type WireBlock = Record<string, unknown>;
 
@@ -57,46 +51,35 @@ function placeBreakpoint(messages: WireMessage[], before: number): boolean {
 }
 
 function moveBreakpointToPersistedTail(
-  built: BuiltRequest,
+  body: Record<string, unknown>,
   options: ExtendedInferenceOptions,
-): BuiltRequest {
+): boolean {
   const suffix = ephemeralSuffixLength(options);
-  if (suffix === 0) return built;
-  let body: Record<string, unknown>;
-  try {
-    body = JSON.parse(built.body) as Record<string, unknown>;
-  } catch {
-    return built;
-  }
-  if (!Array.isArray(body.messages)) return built;
+  if (suffix === 0) return false;
+  if (!Array.isArray(body.messages)) return false;
   const messages = body.messages as WireMessage[];
   const split = messages.length - suffix;
-  if (split <= 0) return built;
+  if (split <= 0) return false;
   stripBreakpoints(messages, split);
-  if (!placeBreakpoint(messages, split)) return built;
-  return { ...built, body: JSON.stringify(body) };
+  return placeBreakpoint(messages, split);
 }
 
+// Only Anthropic-protocol sources are decorated, so non-Anthropic requests
+// skip both the gate and the body parse (byte-identical to the base adapter).
 function withMovedBreakpoint(adapter: ProviderAdapter): ProviderAdapter {
-  return {
-    ...adapter,
-    buildRequest: (turns, model, options) =>
-      moveBreakpointToPersistedTail(
-        adapter.buildRequest(turns, model, options),
-        options,
-      ),
-  };
+  return withJsonBodyRewrite(
+    adapter,
+    (body, { options }) => moveBreakpointToPersistedTail(body, options),
+    (_request, { options }) => ephemeralSuffixLength(options) > 0,
+  );
 }
 
 export function withAnthropicCacheBreakpoint(
   adapters: AdapterRegistry,
 ): AdapterRegistry {
-  return {
-    has: (provider) => adapters.has(provider),
-    resolve(source, quirks) {
-      const adapter = adapters.resolve(source, quirks);
-      if (!ANTHROPIC_MESSAGES_PROVIDERS.has(source.provider)) return adapter;
-      return withMovedBreakpoint(adapter);
-    },
-  };
+  return withDecoratedAdapters(adapters, (adapter, source) =>
+    ANTHROPIC_MESSAGES_PROTOCOLS.has(source.provider)
+      ? withMovedBreakpoint(adapter)
+      : adapter,
+  );
 }