diff --git a/docs-site/src/content/docs/guides/providers.md b/docs-site/src/content/docs/guides/providers.md index 5fc5864481d..bbbf9a61325 100644 --- a/docs-site/src/content/docs/guides/providers.md +++ b/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 diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index ca3cea00b67..edab985cdac 100644 --- a/scripts/test-layout/layout.json +++ b/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", diff --git a/src/adapters/anthropic-rejection-memo.ts b/src/adapters/anthropic-rejection-memo.ts new file mode 100644 index 00000000000..1d24ec9d76f --- /dev/null +++ b/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; +type Policy = { mode: "blanket" } | { mode: "selective"; fingerprints: ReadonlySet }; +const requests = new WeakMap>(); +const responses = new WeakMap>(); +const policies = new WeakMap(); +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(); + 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 | 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; +} diff --git a/src/adapters/anthropic-replay.ts b/src/adapters/anthropic-replay.ts new file mode 100644 index 00000000000..e91f933d654 --- /dev/null +++ b/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; +export interface AnthropicReplaySnapshot { readonly content: readonly Block[]; release(): void } +const enabled = new WeakSet(); +const terminals = new WeakMap(); +const pending = new WeakMap(); +const owners = new WeakMap(); +const issued = new WeakSet(); +const messages = new WeakMap(); +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(); + 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): 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(); }, + }; +} diff --git a/src/adapters/anthropic-tool-catalog.ts b/src/adapters/anthropic-tool-catalog.ts new file mode 100644 index 00000000000..34bd84cd07d --- /dev/null +++ b/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; +} diff --git a/src/adapters/anthropic.ts b/src/adapters/anthropic.ts index b71e9a5b090..441f3e23e4c 100644 --- a/src/adapters/anthropic.ts +++ b/src/adapters/anthropic.ts @@ -1,3 +1,6 @@ +import { captureAnthropicRejectionPayloads, keepAnthropicReplayBlock } from "./anthropic-rejection-memo"; +import { anthropicDeclaredTools } from "./anthropic-tool-catalog"; +import { anthropicReplayContent, createAnthropicReplayCapture } from "./anthropic-replay"; import { withAnthropicProgressUpdates } from "./anthropic-progress"; import type { IncomingMeta, ProviderAdapter } from "./base"; import { createAdapterTierMetadata, type AdapterTierMetadata } from "../providers/fastwire"; @@ -17,7 +20,7 @@ import type { OcxToolResultMessage, OcxUsage, } from "../types"; -import { isAllowedToolChoice, namespacedToolName, resolveToolChoiceWireName, toolChoiceToolPredicate } from "../types"; +import { isAllowedToolChoice, namespacedToolName, resolveToolChoiceWireName } from "../types"; import { ANTHROPIC_OAUTH_BETA, CLAUDE_CODE_SYSTEM_INSTRUCTION, applyClaudeToolPrefix, stripClaudeToolPrefix } from "../oauth/anthropic"; import { parseDataUrl } from "./image"; import { enforceAnthropicImageLimits } from "./anthropic-image-guard"; @@ -283,7 +286,7 @@ function normalizeTtlOrdering(body: Record): void { } } -function isLikelyRealAnthropicThinkingSignature(signature: string | undefined): signature is string { +export function isLikelyRealAnthropicThinkingSignature(signature: string | undefined): signature is string { if (typeof signature !== "string" || signature.length < 16) return false; if (/^(fc|call|msg|rs|resp|reasoning|item|ws|tool|func|function)[-_]/i.test(signature)) return false; return /^[A-Za-z0-9+/_=-]+$/.test(signature); @@ -759,8 +762,8 @@ function messagesToAnthropicFormat( } } const toolCatalogNudge = buildNonOpenAIToolCatalogNudgeForTools( - parsed.context.tools, - parsed.options.toolChoice, + anthropicDeclaredTools(parsed), + undefined, tool => toolNames.toWire(namespacedToolName(tool.namespace, tool.name)), ); const systemParts = [...(parsed.context.systemPrompt ?? []), ...(toolCatalogNudge ? [toolCatalogNudge] : [])]; @@ -790,6 +793,7 @@ function messagesToAnthropicFormat( case "assistant": { const aMsg = msg as OcxAssistantMessage; const preface: unknown[] = []; + const toolUses: unknown[] = []; const originalOrder: unknown[] = []; const appendPreface = (block: unknown) => { preface.push(block); originalOrder.push(block); }; @@ -838,9 +842,15 @@ function messagesToAnthropicFormat( } // Anthropic treats text/thinking after tool_use as ending the tool turn, which makes // earlier tool_use ids look unpaired (#620 / common multi-step history shape). - // Signed native blocks bind the source ordering, including updates between tool calls. - // Keep the legacy repair only for unsigned/cross-provider history. - const content = signed ? originalOrder : [...preface, ...toolUses]; + const captured = anthropicReplayContent(aMsg); + if (captured) { + const expected = toolUses as Array<{ id: string; name: string }>; + const actual = captured.filter(block => block.type === "tool_use"); + if (actual.length !== expected.length || actual.some((block, index) => block.id !== expected[index]?.id || block.name !== expected[index]?.name)) { + throw new Error("Anthropic search replay tool identity changed"); + } + } + const content = (captured ?? (signed ? originalOrder : [...preface, ...toolUses])).filter(block => keepAnthropicReplayBlock(parsed, block as Record)); if (content.length === 0) break; messages.push({ role: "assistant", content }); if (toolUseIds.length > 0) { @@ -903,9 +913,7 @@ function messagesToAnthropicFormat( function toolsToAnthropicFormat(parsed: OcxParsedRequest, toolNames: { toWire: (name: string) => string }): unknown[] | undefined { if (!parsed.context.tools || parsed.context.tools.length === 0) return undefined; - const tools = isAllowedToolChoice(parsed.options.toolChoice) - ? parsed.context.tools.filter(toolChoiceToolPredicate(parsed.options.toolChoice, parsed.context.tools)) - : parsed.context.tools; + const tools = anthropicDeclaredTools(parsed); if (tools.length === 0) return undefined; const converted = tools.map(t => ({ name: toolNames.toWire(namespacedToolName(t.namespace, t.name)), @@ -1226,7 +1234,7 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti enforceCacheControlLimit(body, explicitLimit); normalizeTtlOrdering(body); - return { + const request = { url, method: "POST", headers, @@ -1240,6 +1248,8 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti fastSpeed?.value ?? null, ), }; + captureAnthropicRejectionPayloads(request, body, incoming?.translatorBudget); + return request; }, async *parseStream( @@ -1252,6 +1262,7 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti return; } + const replayCapture = createAnthropicReplayCapture(response, budget); const budgetEncoder = new TextEncoder(); let currentBlockType = ""; let currentToolCallId = ""; @@ -1286,11 +1297,13 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti yield anthropicContentFilterIncomplete(pendingStopReason, usageFromAnthropic(pendingUsage)); return; } - yield { + const done: AdapterEvent = { type: "done", usage: usageFromAnthropic(pendingUsage), ...(pendingStopReason ? { stopReason: pendingStopReason } : {}), }; + replayCapture?.finish(done); + yield done; }; try { @@ -1310,6 +1323,7 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti try { parsed = JSON.parse(payload); } catch { + replayCapture?.invalidate(); debugDroppedFrame("anthropic", payload); continue; } @@ -1317,10 +1331,12 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti // it and the `data.type` read below crashed the stream. Drop a non-record frame the same // way an unparseable one is dropped, so the message_stop check still governs the outcome. if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) { + replayCapture?.invalidate(); debugDroppedFrame("anthropic", payload); continue; } const data = parsed as Record; + replayCapture?.accept({ ...data, type: record.event || data.type }); switch (record.event || data.type) { case "message_start": { @@ -1333,6 +1349,9 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti case "content_block_start": { const block = data.content_block as { type: string; id?: string; name?: unknown; data?: string; thinking?: string; tool_use_id?: unknown; content?: unknown } | undefined; if (!block) break; + // Even an unmodeled server tool or empty block is semantic progress. + // Fence replays before a later refusal; unknown blocks must not look like pings. + yield { type: "heartbeat", replayUnsafe: true }; currentBlockType = block.type; if (block.type === "thinking") { // Preserve even a display:omitted block boundary. The bridge can then @@ -1469,8 +1488,13 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti break; } case "error": { - const err = data.error as { message?: string } | undefined; - yield { type: "error", message: err?.message ?? "Anthropic error" }; + const err = data.error as { message?: unknown; type?: unknown; code?: unknown; status?: unknown } | undefined; + yield { type: "error", message: typeof err?.message === "string" ? err.message : "Anthropic error", + ...(typeof err?.type === "string" ? { errorType: err.type } : {}), + ...(typeof err?.code === "string" ? { code: err.code } : {}), + ...(typeof err?.status === "number" && Number.isInteger(err.status) && err.status >= 400 && err.status <= 599 + ? { status: err.status } : {}), + }; return; } } @@ -1489,6 +1513,7 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti // one-terminal-event contract for consumers that keep draining the generator. return; } finally { + replayCapture?.discard(); if (currentToolCallId) budget.closeCall(currentToolCallId); } if (!emittedDone) { diff --git a/src/images/loop.ts b/src/images/loop.ts index 9d729678cd8..77ec229f71b 100644 --- a/src/images/loop.ts +++ b/src/images/loop.ts @@ -1,3 +1,5 @@ +import type { createAnthropicThinkingRecovery } from "../server/responses/anthropic-thinking-recovery"; +import { bindAnthropicRejectionResponse } from "../adapters/anthropic-rejection-memo"; /** * Media bridge agentic loop — supports both image and video generation sidecars. * @@ -23,7 +25,7 @@ import { bridgeToResponsesSSE } from "../bridge"; import { isTerminalRefusalIncomplete } from "../bridge/terminal-refusal"; import { clearableDeadline, idleDeadline } from "../lib/abort"; import { readBoundedResponseBody } from "../lib/bounded-body"; -import { applyUpstreamRecoveryInit, fetchWithResetRetry, prepareSameTarget429Wait } from "../lib/upstream-retry"; +import { applyUpstreamRecoveryInit, fetchWithResetRetry, prepareSameTarget429Wait, isNonReplayableResponse, SendBudgetExhaustedError } from "../lib/upstream-retry"; import { rateLimitRetryDelayMs } from "../providers/key-failover"; import { releaseProviderRequestSlot, sendTrackingRequestSlot, type ProviderRequestSlot } from "../providers/request-pacing"; import { @@ -33,7 +35,7 @@ import { TRANSLATOR_MAX_TURN_BYTES, TranslatorBudgetExceededError, } from "../lib/translator-budget"; -import { parseStreamWithProgress, RoutedModelInactivityError, WebSearchStreamProtocolError } from "../web-search/progress-stream"; +import { parseStreamWithProgress, RoutedModelInactivityError, WebSearchStreamProtocolError, WebSearchAdapterEventError } from "../web-search/progress-stream"; import { fulfillImageCall } from "./fulfill"; import { parseVideoCallArgs, pollVideoWithHeartbeats, buildVideoResult, createVideoBudget } from "./fulfill-video"; import { submitVideoJob } from "./xai-video-client"; @@ -270,17 +272,18 @@ function extractIterationThinking(events: AdapterEvent[]): OcxThinkingContent[] return parts; } -function jsonError(status: number, message: string): Response { - return new Response(JSON.stringify({ error: { message, type: "upstream_error", code: null } }), { +function jsonError(status: number, message: string, event?: Extract): Response { + return new Response(JSON.stringify({ error: { message, type: event?.errorType ?? "upstream_error", code: event?.code ?? null, + ...(event?.retryable === false ? { retryable: false } : {}) } }), { status, - headers: { "Content-Type": "application/json" }, + headers: { "Content-Type": "application/json", ...(event?.retryable === false ? { "x-should-retry": "false" } : {}) }, }); } /** Hard provider/parse failure inside an iteration. The eager first iteration converts it to a * non-2xx jsonError; later (already-streaming) iterations surface it as an in-stream error event. */ class LoopError extends Error { - constructor(readonly status: number, message: string) { + constructor(readonly status: number, message: string, readonly event?: Extract) { super(message); this.name = "LoopError"; } @@ -291,6 +294,7 @@ class LoopError extends Error { * metadata, and the optional image/video bridge plans. */ export interface ImageBridgeDeps { + thinkingRecovery?: ReturnType; parsed: OcxParsedRequest; clientToolCatalogRestriction?: ClientToolCatalogRestriction; adapter: ProviderAdapter; @@ -321,7 +325,7 @@ export interface ImageBridgeDeps { /** Provider-specific fetch (e.g. xAI transport wrapper). Falls back to global fetch. */ fetchImpl?: typeof globalThis.fetch; /** Bind physical dispatch to this iteration's built request; pacing remains owned by the loop. */ - fetchForRequest?: (request: AdapterRequest, parsed: OcxParsedRequest, adapter: ProviderAdapter) => typeof globalThis.fetch; + fetchForRequest?: (request: AdapterRequest, parsed: OcxParsedRequest, adapter: ProviderAdapter, recovery?: AttemptRecoveryKind) => typeof globalThis.fetch; /** Reserve the routed provider's next request-start slot before each adapter dispatch. */ waitForRequestSlot?: (signal?: AbortSignal) => Promise; /** Raw adapter usage at the terminal event, pre wire-normalization (see bridgeToResponsesSSE onUsage). */ @@ -464,6 +468,8 @@ export async function runWithImageBridge(deps: ImageBridgeDeps): Promise Promise; /** runTurn events are already bounded; consume them without replaying or charging twice. */ collectedEvents?: AdapterEvent[]; } @@ -634,7 +640,7 @@ export async function runWithImageBridge(deps: ImageBridgeDeps): Promise {}); } catch { /* closed */ } + cachedRequest = undefined; + prepared = await fetchOnce(adapter, "opaque-blob-rejection"); + } // Final headers have arrived. Clear only the deadline timer before ANY body read. headerDeadline.clear(); if (!prepared.response.ok) { + if (deps.thinkingRecovery && prepared.responseAdapter.name === "anthropic") { + const event = await deps.thinkingRecovery.projectHttpFailure(prepared.response, signal); + throw new LoopError(event.status ?? prepared.response.status, event.message, event); + } let body: Awaited>; try { body = await readBoundedResponseBody(prepared.response, { signal }); @@ -762,9 +778,21 @@ export async function runWithImageBridge(deps: ImageBridgeDeps): Promise { + cachedRequest = undefined; + headerDeadline = clearableDeadline(connectTimeoutMs, signal); + try { return await fetchOnce(adapter, "opaque-blob-rejection"); } + catch (error) { + if (signal.aborted) throw new LoopError(499, "client closed request during image-bridge"); + if (headerDeadline.didExpire()) throw new LoopError(504, + `Provider response-header timeout after ${connectTimeoutMs}ms during image-bridge`); + throw error; + } finally { headerDeadline.clear(); deps.onIterationEnd?.(); } + } }; } catch (error) { if (isTranslatorBudgetExceededError(error)) throw error; + if (error instanceof SendBudgetExhaustedError) throw new LoopError(429, error.message, + { type: "error", status: 429, errorType: "rate_limit_error", code: error.code, message: error.message, retryable: false }); if (headerDeadline.didExpire()) { throw new LoopError(504, `Provider response-header timeout after ${connectTimeoutMs}ms during image-bridge`); } @@ -791,14 +819,31 @@ export async function runWithImageBridge(deps: ImageBridgeDeps): Promise