fix(anthropic): recover rejected thinking without local prefix validation

docs-site/src/content/docs/guides/providers.md

@@ -87,6 +87,26 @@ modalities are declared; Kimi omits `capabilities` for unknown or text-only mode
 a supported capability field keep their existing configuration shape. After updating opencodex,
 regenerate or refresh the client configuration managed by opencodex to receive the updated metadata.
 
+## Replayed Anthropic thinking signatures
+
+On the Responses endpoint, an Anthropic-compatible upstream can reject a replayed thinking
+signature because it belongs to a different conversation prefix. When the error explicitly
+identifies this signature-binding problem, opencodex can remove replayed signed/redacted thinking
+and retry once within the request's send budget. This covers HTTP 400 refusals and initial stream
+errors before output or tool activity. Hosted web search and image/video bridges share one recovery
+limit across their model iterations: only the rejected inference is resent, while completed sidecar
+calls and tool results are retained. An empty-completion retry shares its request's allowance;
+ordinary continuations after output cannot use it. A failed replacement is
+returned without another signature retry. Other errors keep their existing handling.
+
+Sidecar recovery is disabled when enforced spend accounting cannot establish receipts for earlier
+ordinary sidecar sends. It never increases a caller or Combo send allowance.
+
+Only a successful, nonempty completion remembers the rejection for that conversation and serving
+identity for five minutes. Only the refused opaque payloads are suppressed when they can be
+identified; newly issued thinking remains replayable. Plaintext history remains available; a failed or cancelled retry does
+not disable thinking replay on later requests.
+
 ## Auth modes
 
 Provider configs accept three `authMode` values (`key` is the default). The built-in registry also

scripts/test-layout/layout.json

@@ -219,6 +219,7 @@
     "anthropic-tail-guard.test.ts": "adapters/anthropic",
     "anthropic-progress-updates.test.ts": "adapters/anthropic",
     "anthropic-thinking-signature.test.ts": "adapters/anthropic",
+    "anthropic-replay.test.ts": "adapters/anthropic",
     "anthropic-opaque-strip.test.ts": "adapters/anthropic",
     "anthropic-tool-call-id.test.ts": "adapters/anthropic",
     "anthropic-messages-passthrough.test.ts": "adapters/anthropic",
@@ -1508,6 +1509,10 @@
     "responses-account-change-scrub.test.ts": "responses",
     "responses-account-label.test.ts": "responses",
     "responses-anthropic-fast-downgrade.test.ts": "responses",
+    "responses-anthropic-thinking-signature-recovery.test.ts": "responses",
+    "responses-web-search-thinking-signature-recovery.test.ts": "responses",
+    "responses-anthropic-thinking-recovery.test.ts": "responses",
+    "responses-anthropic-legal-history.test.ts": "responses",
     "responses-azure-opaque-blob-recovery.test.ts": "responses",
     "responses-bare-echo-helper-fence.test.ts": "responses",
     "responses-canonical-only-top-level-fields.test.ts": "responses",

src/adapters/anthropic-rejection-memo.ts

@@ -0,0 +1,71 @@
+import { createHash } from "node:crypto";
+import type { TranslatorBudget } from "../lib/translator-budget";
+import type { AdapterRequest } from "./base";
+import type { OcxParsedRequest } from "../types";
+
+type Block = Record<string, unknown>;
+type Policy = { mode: "blanket" } | { mode: "selective"; fingerprints: ReadonlySet<string> };
+const requests = new WeakMap<AdapterRequest, ReadonlySet<string>>();
+const responses = new WeakMap<Response, ReadonlySet<string>>();
+const policies = new WeakMap<object, Policy>();
+const owner = (parsed: OcxParsedRequest) => parsed._reasoningReplayScope ?? parsed;
+/** Domain-separated opaque bytes, deliberately independent of plaintext and untrusted hints. */
+export function anthropicOpaqueFingerprint(block: Block): string | undefined {
+  const value = block.type === "thinking" ? block.signature : block.type === "redacted_thinking" ? block.data : undefined;
+  if (typeof value !== "string" || value.length === 0) return;
+  return createHash("sha256").update(String(block.type)).update("\0").update(value).digest("hex");
+}
+export function captureAnthropicRejectionPayloads(request: AdapterRequest, body: Block, budget?: TranslatorBudget): void {
+  requests.delete(request);
+  if (!budget || !Array.isArray(body.messages)) return;
+  let retained = 0;
+  let complete = false;
+  const fingerprints = new Set<string>();
+  try {
+  for (const message of body.messages) {
+    if (!message || typeof message !== "object") return;
+    if (typeof message.content === "string") continue;
+    if (!Array.isArray(message.content)) return;
+    for (const block of message.content) {
+      if (!block || typeof block !== "object") return;
+      if (block.type !== "thinking" && block.type !== "redacted_thinking") continue;
+      const value = block.type === "thinking" ? block.signature : block.data;
+      if (typeof value !== "string" || value.length === 0) return;
+      const temporary = budget.reserveTransient(Buffer.byteLength(value) * 2, { kind: "reasoning" });
+      let fingerprint: string | undefined;
+      try { fingerprint = anthropicOpaqueFingerprint(block); } finally { temporary.release(); }
+      if (!fingerprint) return;
+      if (!fingerprints.has(fingerprint)) {
+        if (fingerprints.size >= 256) return;
+        // Includes Set/hex storage and later defensive copies; owned until turn disposal.
+        budget.chargeRetained(256, { kind: "reasoning" }); retained += 256;
+        fingerprints.add(fingerprint);
+      }
+    }
+  }
+  if (fingerprints.size) { requests.set(request, fingerprints); complete = true; }
+  } catch { /* An uncollectable set keeps the legacy blanket policy. */ }
+  finally { if (!complete) budget.releaseRetained(retained, { kind: "reasoning" }); }
+}
+export function bindAnthropicRejectionResponse(request: AdapterRequest | undefined, response: Response): void {
+  const fingerprints = request && requests.get(request);
+  if (fingerprints) responses.set(response, fingerprints); else responses.delete(response);
+}
+export function rejectedAnthropicPayloads(response: Response | undefined): ReadonlySet<string> | undefined {
+  const value = response && responses.get(response);
+  return value && new Set(value);
+}
+export function bindAnthropicRejectionPolicy(parsed: OcxParsedRequest, policy: Policy | undefined): void {
+  if (policy) policies.set(owner(parsed), policy); else policies.delete(owner(parsed));
+}
+export function anthropicRejectionPolicy(parsed: OcxParsedRequest): Policy | undefined { return policies.get(owner(parsed)); }
+
+/** Filter only refused opaque bytes; newly issued thinking in this request remains replayable. */
+export function keepAnthropicReplayBlock(parsed: OcxParsedRequest, block: Block): boolean {
+  const fingerprint = anthropicOpaqueFingerprint(block);
+  if (!fingerprint) return true;
+  if (parsed._stripReasoningEncryptedContent === true) return false;
+  const policy = anthropicRejectionPolicy(parsed);
+  return policy ? policy.mode === "selective" && !policy.fingerprints.has(fingerprint)
+    : true;
+}

src/adapters/anthropic-replay.ts

@@ -0,0 +1,135 @@
+import { isConformingToolCallId } from "./tool-call-id";
+import type { AdapterEvent, OcxAssistantMessage } from "../types";
+import { isTranslatorBudgetExceededError, type TranslatorBudget, type TranslatorTransientReservation } from "../lib/translator-budget";
+import { jsonUtf8Bytes } from "../lib/json-byte-size";
+
+type Block = Record<string, unknown>;
+export interface AnthropicReplaySnapshot { readonly content: readonly Block[]; release(): void }
+const enabled = new WeakSet<Response>();
+const terminals = new WeakMap<object, AnthropicReplaySnapshot>();
+const pending = new WeakMap<Response, AnthropicReplaySnapshot>();
+const owners = new WeakMap<object, Response>();
+const issued = new WeakSet<object>();
+const messages = new WeakMap<object, AnthropicReplaySnapshot>();
+export function enableAnthropicReplayCapture(response: Response): () => void {
+  enabled.add(response);
+  return () => { enabled.delete(response); pending.get(response)?.release(); pending.delete(response); };
+}
+export function takeAnthropicReplay(event: object): AnthropicReplaySnapshot | undefined {
+  const snapshot = terminals.get(event); terminals.delete(event);
+  const owner = owners.get(event); if (owner) pending.delete(owner); owners.delete(event); return snapshot;
+}
+export function bindAnthropicReplay(message: OcxAssistantMessage, snapshot: AnthropicReplaySnapshot): void {
+  if (!issued.has(snapshot)) throw new Error("Untrusted Anthropic replay snapshot");
+  messages.set(message, snapshot);
+}
+/** Top-level copies let cache-control decoration leave the captured provider response untouched. */
+export function anthropicReplayContent(message: OcxAssistantMessage): Block[] | undefined {
+  const snapshot = messages.get(message);
+  if (snapshot && !issued.has(snapshot)) throw new Error("Expired Anthropic replay snapshot");
+  return snapshot?.content.map(block => ({ ...block }));
+}
+
+/** Capture only explicitly opted-in search streams; no client data can manufacture a snapshot. */
+export function createAnthropicReplayCapture(response: Response, budget: TranslatorBudget) {
+  if (!enabled.delete(response)) return undefined;
+  let bytes = 0, valid = true, transferred = false;
+  let active: number | undefined;
+  let started = false, stopped = false;
+  const toolIds = new Set<string>();
+  const blocks: Array<{ value: Block; text: string[]; thinking: string[]; args: string[] }> = [];
+  const release = () => { if (bytes) budget.releaseRetained(bytes, { kind: "retained_collectors" }); bytes = 0; };
+  const invalidate = () => { valid = false; release(); blocks.length = 0; toolIds.clear(); active = undefined; };
+  const retain = (value: unknown) => {
+    // Charge each frame once, before retaining its values. This conservative linear bound also
+    // covers array/object overhead and replacement signatures without serializing a growing block.
+    try {
+      const size = jsonUtf8Bytes(value) * 4 + 64;
+      budget.chargeRetained(size, { kind: "retained_collectors" }); bytes += size;
+      return true;
+    } catch (error) {
+      if (!isTranslatorBudgetExceededError(error)) throw error;
+      invalidate();
+      return false;
+    }
+  };
+  return {
+    invalidate,
+    accept(data: Record<string, unknown>): void {
+      if (!valid) return;
+      const type = data.type;
+      if (type === "ping") return;
+      if (type === "message_start") {
+        const message = data.message as Block | undefined;
+        if (started || (message?.content !== undefined && (!Array.isArray(message.content) || message.content.length > 0))) invalidate();
+        else started = true;
+        return;
+      }
+      if (!started || stopped) { invalidate(); return; }
+      if (type === "message_stop") { stopped = true; if (active !== undefined) invalidate(); return; }
+      if (type === "content_block_start") {
+        const block = data.content_block as Block | undefined;
+        if (active !== undefined || data.index !== blocks.length || !block
+          || !(block.type === "text" && typeof block.text === "string"
+            || block.type === "thinking" && typeof block.thinking === "string"
+            || block.type === "redacted_thinking" && typeof block.data === "string"
+            || block.type === "tool_use" && typeof block.id === "string" && typeof block.name === "string"
+              && !!block.input && typeof block.input === "object" && !Array.isArray(block.input))) { invalidate(); return; }
+        if (block.type === "tool_use") {
+          const id = block.id as string;
+          if (!isConformingToolCallId(id) || toolIds.has(id)) { invalidate(); return; }
+          toolIds.add(id);
+        }
+        if (!retain(block)) return;
+        active = blocks.length;
+        blocks.push({ value: { ...block, ...(Array.isArray(block.citations) ? { citations: [...block.citations] } : {}) }, text: typeof block.text === "string" ? [block.text] : [],
+          thinking: typeof block.thinking === "string" ? [block.thinking] : [], args: [] });
+      } else if (type === "content_block_delta") {
+        const delta = data.delta as Block | undefined;
+        if (active === undefined || data.index !== active || !delta) { invalidate(); return; }
+        const block = blocks[active]!;
+        if (!retain(delta)) return;
+        if (delta.type === "text_delta" && block.value.type === "text" && typeof delta.text === "string") block.text.push(delta.text);
+        else if (delta.type === "thinking_delta" && block.value.type === "thinking" && typeof delta.thinking === "string") block.thinking.push(delta.thinking);
+        else if (delta.type === "signature_delta" && block.value.type === "thinking" && typeof delta.signature === "string") block.value.signature = delta.signature;
+        else if (delta.type === "input_json_delta" && block.value.type === "tool_use" && typeof delta.partial_json === "string") block.args.push(delta.partial_json);
+        else if (delta.type === "citations_delta" && block.value.type === "text" && delta.citation && typeof delta.citation === "object") {
+          const citations = block.value.citations;
+          if (citations !== undefined && !Array.isArray(citations)) { invalidate(); return; }
+          if (citations === undefined) block.value.citations = [delta.citation];
+          else (citations as unknown[]).push(delta.citation);
+        } else invalidate();
+      } else if (type === "content_block_stop") {
+        if (active === undefined || data.index !== active) { invalidate(); return; }
+        active = undefined;
+      } else if (type !== "message_start" && type !== "message_delta" && type !== "message_stop" && type !== "ping") invalidate();
+    },
+    finish(event: AdapterEvent): void {
+      if (!valid || !stopped || active !== undefined || event.type !== "done") { invalidate(); return; }
+      let reservation: TranslatorTransientReservation | undefined;
+      try {
+        reservation = budget.reserveTransient(bytes * 2, { kind: "retained_collectors" });
+        const content = blocks.map(block => {
+          const value = block.value;
+          if (value.type === "text") value.text = block.text.join("");
+          if (value.type === "thinking") {
+            value.thinking = block.thinking.join("");
+            if (typeof value.signature !== "string" || value.signature.length === 0) throw new Error("incomplete thinking signature");
+          }
+          if (value.type === "tool_use" && block.args.length) {
+            value.input = JSON.parse(block.args.join(""));
+            if (!value.input || typeof value.input !== "object" || Array.isArray(value.input)) throw new Error("invalid tool input");
+          }
+          return Object.freeze(value);
+        });
+        const snapshot = { content: Object.freeze(content), release: () => { issued.delete(snapshot); release(); } };
+        issued.add(snapshot);
+        terminals.set(event, snapshot); owners.set(event, response); pending.set(response, snapshot);
+        blocks.length = 0; toolIds.clear();
+        transferred = true;
+      } catch { invalidate(); }
+      finally { reservation?.release(); }
+    },
+    discard(): void { if (!transferred) invalidate(); },
+  };
+}

src/adapters/anthropic-tool-catalog.ts

@@ -0,0 +1,11 @@
+import { isAllowedToolChoice, toolChoiceToolPredicate, type OcxParsedRequest } from "../types";
+import { rejectsForcedToolChoice } from "./anthropic-model-contract";
+
+/** The catalog actually advertised on the wire, independently of whether calling it is enabled. */
+export function anthropicDeclaredTools(parsed: OcxParsedRequest) {
+  const tools = parsed.context.tools ?? [];
+  const choice = parsed.options.toolChoice;
+  return isAllowedToolChoice(choice)
+    || (rejectsForcedToolChoice(parsed.modelId) && typeof choice === "object" && choice !== null && "name" in choice)
+    ? tools.filter(toolChoiceToolPredicate(choice, tools)) : tools;
+}