import * as fs from "node:fs";
import { scheduler } from "node:timers/promises";
import * as tls from "node:tls";
import { isAnthropicSigningProxyUrl, isOfficialAnthropicApiUrl } from "@oh-my-pi/pi-catalog/compat/anthropic";
import { hostMatchesUrl, isVertexRawPredictUrl } from "@oh-my-pi/pi-catalog/hosts";
import { mapEffortToAnthropicAdaptiveEffort } from "@oh-my-pi/pi-catalog/model-thinking";
import { calculateCost, getBundledModel } from "@oh-my-pi/pi-catalog/models";
import { isAnthropicOAuthToken } from "@oh-my-pi/pi-catalog/utils";
import { parseGitHubCopilotApiKey } from "@oh-my-pi/pi-catalog/wire/github-copilot";
import {
	$env,
	isEnoent,
	logger,
	parseJsonWithRepair,
	parseStreamingJsonThrottled,
	readSseEvents,
} from "@oh-my-pi/pi-utils";
import { NO_AUTH_SENTINEL } from "../auth-retry";
import { renderDemotedThinking } from "../dialect/demotion";
import * as AIError from "../error";
import { parseToolCallArguments } from "../utils/tool-call-arguments";
import { getEnvApiKey, OUTPUT_FALLBACK_BUFFER } from "../stream";
import type {
	AnthropicCompactionPayload,
	AnthropicFallbackContent,
	AnthropicMessagePayload,
	AnthropicOutputEffort,
	AnthropicRequestControls,
	AnthropicServerToolContent,
	AnthropicToolChange,
	Api,
	AssistantMessage,
	CacheRetention,
	Context,
	DeveloperMessage,
	FetchImpl,
	ImageContent,
	Message,
	Model,
	ProviderInputTransformation,
	ProviderPayload,
	ProviderSessionState,
	RawSseEvent,
	RedactedThinkingContent,
	ServiceTier,
	SimpleStreamOptions,
	StopReason,
	StreamFunction,
	StreamOptions,
	TextContent,
	ThinkingContent,
	Tool,
	ToolCall,
	ToolResultMessage,
	Usage,
} from "../types";
import {
	getHeaderCaseInsensitive,
	isRecord,
	normalizeSystemPrompts,
	normalizeToolCallId,
	resolveCacheRetention,
} from "../utils";
import { createAbortSourceTracker } from "../utils/abort";
import {
	clearStreamingPartialJson,
	copyPerCallContextMessage,
	type ConversationalUserCarrier,
	isConversationalUser,
	isPerCallContextMessage,
	isSyntheticUser,
	kConversationalUser,
	kStreamingBlockIndex,
	kStreamingLastParseLen,
	kStreamingPartialJson,
} from "../utils/block-symbols";
import { withReplaySafeStreamRetry } from "../utils/empty-completion-retry";
import { AssistantMessageEventStream } from "../utils/event-stream";
import { isFoundryEnabled } from "../utils/foundry";
import { finalizeErrorMessage, type RawHttpRequestDump } from "../utils/http-inspector";
import { getStreamFirstEventTimeoutMs, getStreamIdleTimeoutMs, iterateWithIdleTimeout } from "../utils/idle-iterator";
import {
	ANTHROPIC_SLOW_USAGE_LIMIT,
	ANTHROPIC_USAGE_LIMIT_HEADER,
	parseAnthropicSlowModeHeaders,
} from "./anthropic-slow-mode";
import { notifyProviderResponse } from "../utils/provider-response";
import { getHeadersFromError, getRetryAfterMsFromHeaders } from "../utils/retry-after";
import { COMBINATOR_KEYS, NO_STRICT, toolWireSchema } from "../utils/schema";
import { spillToDescription } from "../utils/schema/spill";
import { createSdkStreamRequestOptions } from "../utils/sdk-stream-timeout";
import { notifyRawSseEvent } from "../utils/sse-debug";
import { isForcedToolChoice } from "../utils/tool-choice";
import {
	AnthropicConnectionTimeoutError,
	type AnthropicFetchOptions,
	AnthropicMessagesClient,
	type AnthropicMessagesClientLike,
	calculateAnthropicRetryDelayMs,
} from "./anthropic-client";
import {
	type ToolInputSchema as AnthropicToolInputSchema,
	type Tool as AnthropicWireTool,
	type Usage as AnthropicWireUsage,
	COMPACTION_BETA,
	LEGACY_COMPACTION_BETA,
	type CompactionBlockParam,
	type CompactionEdit,
	type ContentBlockParam,
	type FallbackParam,
	isAnthropicServerToolHistoryBlock,
	type MessageCreateParams,
	type MessageCreateParamsStreaming,
	type MessageParam,
	parseAnthropicInputTransformations,
	type RawMessageStreamEvent,
	THINKING_BINDING_CONTROLS_BETA,
	type TextBlockParam,
} from "./anthropic-wire";
import {
	claudeCodeSdkVersion,
	claudeCodeSystemInstruction,
	adoptRequiredClaudeCodeVersion,
	claudeToolPrefix,
	getClaudeCodeUserAgent,
	getClaudeCodeVersion,
} from "./claude-code-fingerprint";
import {
	buildCopilotDynamicHeaders,
	getCachedCopilotIntegrationId,
	getCopilotIntegrationCacheKey,
	hasCopilotVisionInput,
	resolveCopilotRequestIdentity,
	resolveGitHubCopilotBaseUrl,
	wrapFetchForCopilotFallback,
} from "./github-copilot-headers";
import { servedModelFromAnthropicSignature } from "./anthropic-signature";
import { getOpenAIPromptCacheKey } from "./openai-shared";
import { applyInferenceHeaders } from "./inference-headers";
import { redactSensitiveCredentials, transformMessages } from "./transform-messages";
import { NON_VISION_IMAGE_PLACEHOLDER } from "./vision-guard";
import {
	injectedClientBaseUrl,
	resolvesToOfficialAnthropicEndpoint,
	supportsAnthropicCompaction,
	supportsAnthropicCompactionOnClient,
} from "./anthropic-compaction";
import {
	applyClaudeToolPrefix,
	deriveClaudeDeviceId,
	extractClaudeMetadataSessionId,
	generateClaudeCloakingUserId,
	isClaudeCloakingUserId,
	readAnthropicMetadataAccountId,
	readAnthropicMetadataString,
	resolveAnthropicMetadataUserId,
	stripClaudeToolPrefix,
} from "./anthropic-identity";
import { fitBedrockAnthropicPayload } from "./bedrock-anthropic";
import {
	anthropicProviderSessionStateKey,
	clearAnthropicFastModeFallback,
	isAnthropicFastModeFallbackDisabled,
	normalizeAnthropicBaseUrl,
	resolveDirectAnthropicBaseUrl,
} from "./anthropic-state";

export {
	applyClaudeToolPrefix,
	clearAnthropicFastModeFallback,
	deriveClaudeDeviceId,
	generateClaudeCloakingUserId,
	isAnthropicFastModeFallbackDisabled,
	isClaudeCloakingUserId,
	normalizeAnthropicBaseUrl,
	resolveAnthropicMetadataUserId,
	stripClaudeToolPrefix,
};
export { resolvesToOfficialAnthropicEndpoint, supportsAnthropicCompaction, supportsAnthropicCompactionOnClient };

export type AnthropicHeaderOptions = {
	apiKey: string;
	baseUrl?: string;
	isOAuth?: boolean;
	extraBetas?: string[];
	stream?: boolean;
	modelHeaders?: Record<string, string>;
	isCloudflareAiGateway?: boolean;
	claudeCodeSessionId?: string;
	claudeCodeBetas?: readonly string[];
	/** Allow explicit fingerprint headers to replace OAuth defaults on non-official endpoints. */
	allowAnthropicHeaderOverrides?: boolean;
};

// Build deduplicated beta header string
export function buildBetaHeader(baseBetas: readonly string[], extraBetas: readonly string[]): string {
	const seen = new Set<string>();
	const result: string[] = [];
	for (const beta of [...baseBetas, ...extraBetas]) {
		const trimmed = beta.trim();
		if (trimmed && !seen.has(trimmed)) {
			seen.add(trimmed);
			result.push(trimmed);
		}
	}
	return result.join(",");
}

/**
 * Merge an extra Anthropic beta into a caller-provided `anthropic-beta` header,
 * preserving the caller's key casing and deduping the tokens. Returns a
 * single-entry header record for a per-request `headers` override — used to
 * attach a required beta to injected SDK clients that bypass the client-level
 * beta construction.
 */
function mergeAnthropicBetaHeader(callerHeaders: Record<string, string>, beta: string): Record<string, string> {
	for (const key in callerHeaders) {
		if (key.toLowerCase() === "anthropic-beta") {
			return { [key]: buildBetaHeader(normalizeExtraBetas(callerHeaders[key]), [beta]) };
		}
	}
	return { "anthropic-beta": beta };
}
const oauthAuthBeta = "oauth-2025-04-20";
const midConversationSystemBeta = "mid-conversation-system-2026-04-07";
const midConversationSystemClearAtBeta = "mid-conversation-system-clear-at-2026-08-21";
const midConversationToolChangesBeta = "mid-conversation-tool-changes-2026-07-01";
const midConversationOutputConfigBeta = "mid-conversation-output-config-2026-07-01";
const contextManagementBeta = "context-management-2025-06-27";
const structuredOutputsBeta = "structured-outputs-2025-12-15";
const thinkingTokenCountBeta = "thinking-token-count-2026-05-13";
const fallbackCreditBeta = "fallback-credit-2026-06-01";
const claudeCodeUtilityBetaDefaults = [
	oauthAuthBeta,
	"interleaved-thinking-2025-05-14",
	thinkingTokenCountBeta,
	contextManagementBeta,
	"prompt-caching-scope-2026-01-05",
	structuredOutputsBeta,
] as const;
const claudeCodeAgentBetaDefaults = [
	"claude-code-20250219",
	oauthAuthBeta,
	"interleaved-thinking-2025-05-14",
	thinkingTokenCountBeta,
	contextManagementBeta,
	"prompt-caching-scope-2026-01-05",
	midConversationSystemBeta,
] as const;
const extendedCacheTtlBeta = "extended-cache-ttl-2025-04-11";
const fineGrainedToolStreamingBeta = "fine-grained-tool-streaming-2025-05-14";
const interleavedThinkingBeta = "interleaved-thinking-2025-05-14";
const fastModeBeta = "fast-mode-2026-02-01";
const taskBudgetBeta = "task-budgets-2026-03-13";
const effortBeta = "effort-2025-11-24";
const serverSideFallbackBeta = "server-side-fallback-2026-06-01";

function resolveAnthropicControlBetas(
	model: Model<"anthropic-messages">,
	prefixMismatchBehavior: "drop_block" | "error" | undefined,
): string[] {
	const betas: string[] = [];
	if (prefixMismatchBehavior) betas.push(THINKING_BINDING_CONTROLS_BETA);
	if (model.compat.supportsTurnScopedSystem) betas.push(midConversationSystemClearAtBeta);
	if (model.compat.supportsMidConversationToolChanges) betas.push(midConversationToolChangesBeta);
	if (model.compat.supportsPerMessageEffort) betas.push(midConversationOutputConfigBeta);
	return betas;
}

function buildClaudeCodeBetas({
	agentRequest,
	thinkingRequest,
	disableStrictTools = false,
	supportsContextManagement = true,
}: {
	agentRequest: boolean;
	thinkingRequest: boolean;
	disableStrictTools?: boolean;
	supportsContextManagement?: boolean;
}): readonly string[] {
	// `context-1m-2025-08-07` is intentionally never advertised. OAuth
	// subscription credentials have no long-context credit balance, so Anthropic
	// hard-429s ("Usage credits are required for long context requests") on any
	// beta-gated 1M model regardless of prompt size (#7238). Natively-1M models
	// (e.g. claude-sonnet-5) serve their full window without the beta anyway.
	if (!agentRequest && !disableStrictTools && supportsContextManagement) return claudeCodeUtilityBetaDefaults;
	const betas: string[] = [];
	for (const beta of agentRequest ? claudeCodeAgentBetaDefaults : claudeCodeUtilityBetaDefaults) {
		if (disableStrictTools && beta === structuredOutputsBeta) continue;
		if (!supportsContextManagement && beta === contextManagementBeta) continue;
		betas.push(beta);
	}
	if (!agentRequest) return betas;
	if (thinkingRequest) betas.push(effortBeta);
	betas.push(fallbackCreditBeta);
	return betas;
}

function isClaudeCodeClientUserAgent(userAgent: string | undefined): userAgent is string {
	if (!userAgent) return false;
	return userAgent.toLowerCase().startsWith("claude-cli");
}

const sharedHeaders = {
	"Accept-Encoding": "gzip, deflate, br, zstd",
	Connection: "keep-alive",
	"Content-Type": "application/json",
	"anthropic-version": "2023-06-01",
	"anthropic-dangerous-direct-browser-access": "true",
	"x-app": "cli",
};

export function buildAnthropicHeaders(options: AnthropicHeaderOptions): Record<string, string> {
	const oauthToken = options.isOAuth ?? isAnthropicOAuthToken(options.apiKey);
	const extraBetas = options.extraBetas ?? [];
	const stream = options.stream ?? false;
	// `enforcedHeaderKeys` strips User-Agent / X-Api-Key / Authorization out of
	// modelHeaders so a case-insensitive spread can't produce duplicate keys; each
	// branch re-adds the caller's value explicitly. User-Agent and X-Api-Key are
	// always honored (with branch-specific defaults filling in when absent), while
	// Authorization is honored for every non-OAuth, non-Cloudflare-gateway branch —
	// OAuth requests MUST carry `Authorization: Bearer <oauth-token>` (the OAuth
	// credential itself) and Cloudflare AI Gateway authenticates via
	// `cf-aig-authorization`, so user-supplied auth there would just leak. Both of
	// those cases drop + log the caller value (#3391).
	const incomingUserAgent = getHeaderCaseInsensitive(options.modelHeaders, "User-Agent");
	const incomingAuthorization = getHeaderCaseInsensitive(options.modelHeaders, "Authorization");
	const incomingApiKey = getHeaderCaseInsensitive(options.modelHeaders, "X-Api-Key");
	// Claude Code's beta profile is part of the OAuth fingerprint; API-key
	// requests default to extras only, matching the streaming path.
	const betaHeader = buildBetaHeader(
		options.claudeCodeBetas ??
			(oauthToken ? buildClaudeCodeBetas({ agentRequest: true, thinkingRequest: true }) : []),
		extraBetas,
	);
	const acceptHeader = oauthToken ? "application/json" : stream ? "text/event-stream" : "application/json";
	const isCloudflare = options.isCloudflareAiGateway ?? false;
	const honorAuthorization = !oauthToken && !isCloudflare;
	const allowAnthropicHeaderOverrides =
		oauthToken &&
		options.allowAnthropicHeaderOverrides === true &&
		!isCloudflare &&
		!isOfficialAnthropicApiUrl(options.baseUrl);
	const honorApiKey = !isCloudflare;
	const modelHeaders: Record<string, string> = {};
	const anthropicHeaderOverrides: Record<string, string> = {};
	const filteredEnforcedKeys: string[] = [];
	const headerSource = options.modelHeaders;
	if (headerSource) {
		for (const key in headerSource) {
			const value = headerSource[key];
			const lowerKey = key.toLowerCase();
			if (enforcedHeaderKeys.has(lowerKey) && (oauthToken || !coworkHeaderKeys.has(lowerKey))) {
				if (allowAnthropicHeaderOverrides && overridableAnthropicHeaderKeys.has(lowerKey)) {
					anthropicHeaderOverrides[key] = value;
					continue;
				}
				// user-agent is always re-applied explicitly. authorization / x-api-key
				// are silently re-applied in honoring branches and dropped + logged
				// where the branch enforces its own credential.
				if (lowerKey === "user-agent") continue;
				if (lowerKey === "authorization" && honorAuthorization) continue;
				if (lowerKey === "x-api-key" && honorApiKey) continue;
				filteredEnforcedKeys.push(key);
				continue;
			}
			modelHeaders[key] = value;
		}
	}
	if (filteredEnforcedKeys.length > 0) {
		// Caller/env-supplied values (options.headers, ANTHROPIC_CUSTOM_HEADERS)
		// for enforced headers are replaced by our own values; say so instead of
		// dropping them silently. Keys only — values may carry credentials.
		logger.debug("anthropic: ignoring caller-supplied enforced headers", {
			headers: filteredEnforcedKeys,
		});
	}

	if (isCloudflare) {
		return {
			...modelHeaders,
			Accept: acceptHeader,
			...sharedHeaders,
			...(incomingUserAgent ? { "User-Agent": incomingUserAgent } : {}),
			...(betaHeader ? { "anthropic-beta": betaHeader } : {}),
			"cf-aig-authorization": `Bearer ${options.apiKey}`,
		};
	}

	if (oauthToken) {
		const userAgent = isClaudeCodeClientUserAgent(incomingUserAgent) ? incomingUserAgent : getClaudeCodeUserAgent();
		const headers = {
			...modelHeaders,
			Accept: acceptHeader,
			"Content-Type": "application/json",
			"User-Agent": userAgent,
			...(options.claudeCodeSessionId ? { "X-Claude-Code-Session-Id": options.claudeCodeSessionId } : {}),
			...claudeCodeHeaders,
			...(betaHeader ? { "anthropic-beta": betaHeader } : {}),
			"anthropic-dangerous-direct-browser-access": "true",
			"anthropic-version": "2023-06-01",
			Authorization: `Bearer ${options.apiKey}`,
			"x-app": "cli",
			Connection: "keep-alive",
			"Accept-Encoding": "gzip, deflate, br, zstd",
			...(incomingApiKey ? { "X-Api-Key": incomingApiKey } : {}),
		};
		return allowAnthropicHeaderOverrides ? mergeHeaders(headers, anthropicHeaderOverrides) : headers;
	} else if (!isOfficialAnthropicApiUrl(options.baseUrl)) {
		// A keyless provider (`auth: none`) resolves to the `N/A` sentinel
		// rather than a real key; custom endpoints that authenticate via their
		// own headers may reject a bogus bearer, so send no Authorization —
		// same sentinel guard as the openai transports. A caller-supplied
		// Authorization in `model.headers` still wins.
		const bearer =
			incomingAuthorization ?? (options.apiKey !== NO_AUTH_SENTINEL ? `Bearer ${options.apiKey}` : undefined);
		return {
			...modelHeaders,
			Accept: acceptHeader,
			...(bearer ? { Authorization: bearer } : {}),
			...sharedHeaders,
			...(incomingUserAgent ? { "User-Agent": incomingUserAgent } : {}),
			...(betaHeader ? { "anthropic-beta": betaHeader } : {}),
			...(incomingApiKey ? { "X-Api-Key": incomingApiKey } : {}),
		};
	} else {
		return {
			...modelHeaders,
			Accept: acceptHeader,
			...sharedHeaders,
			...(incomingUserAgent ? { "User-Agent": incomingUserAgent } : {}),
			...(betaHeader ? { "anthropic-beta": betaHeader } : {}),
			...(incomingAuthorization ? { Authorization: incomingAuthorization } : {}),
			"X-Api-Key": incomingApiKey ?? options.apiKey,
		};
	}
}

type AnthropicCacheControl = NonNullable<TextBlockParam["cache_control"]>;
type AnthropicImageMediaType = "image/jpeg" | "image/png" | "image/gif" | "image/webp";

function normalizeAnthropicImageMediaType(mimeType: string): AnthropicImageMediaType | undefined {
	const normalized = mimeType.trim().toLowerCase();
	if (normalized === "image/jpg") return "image/jpeg";
	if (
		normalized === "image/jpeg" ||
		normalized === "image/png" ||
		normalized === "image/gif" ||
		normalized === "image/webp"
	) {
		return normalized;
	}
	return undefined;
}

function cloneAnthropicCacheControl(cacheControl: AnthropicCacheControl): AnthropicCacheControl {
	return { ...cacheControl };
}

type AnthropicOutputConfig = NonNullable<MessageCreateParamsStreaming["output_config"]>;

const ANTHROPIC_STOP_SEQUENCES_MAX = 4;
let warnedStopSequencesTrim = false;

type AnthropicProviderSessionState = ProviderSessionState & {
	strictToolsDisabled: boolean;
	fastModeDisabled: boolean;
	/**
	 * Runtime-learned: this endpoint rejected a replayed unsigned thinking
	 * block, so it must be treated as a signing proxy from now on. All
	 * subsequent requests demote unsigned thinking to text for this (baseUrl,
	 * modelId), same behavior as an explicit
	 * `compat.replayUnsignedThinking: false`. Cleared on session close.
	 */
	replayUnsignedThinkingDisabled: boolean;
	/**
	 * Runtime-learned: this endpoint kept rejecting replayed thinking
	 * signatures even after unsigned demotion — every surviving block is
	 * signed by a foreign signer (e.g. a failover proxy swapped upstreams
	 * mid-conversation and minted signatures the restored upstream cannot
	 * verify). All subsequent requests drop replayed thinking entirely for
	 * this (baseUrl, modelId). Cleared on session close.
	 */
	thinkingReplayDisabled: boolean;
	/** Thinking blocks the API permanently dropped after a prefix mismatch. */
	prefixDroppedThinkingBlocks: Set<string>;
};

function createAnthropicProviderSessionState(): AnthropicProviderSessionState {
	const state: AnthropicProviderSessionState = {
		strictToolsDisabled: false,
		fastModeDisabled: false,
		replayUnsignedThinkingDisabled: false,
		thinkingReplayDisabled: false,
		prefixDroppedThinkingBlocks: new Set(),
		close: () => {
			state.strictToolsDisabled = false;
			state.fastModeDisabled = false;
			state.replayUnsignedThinkingDisabled = false;
			state.thinkingReplayDisabled = false;
			state.prefixDroppedThinkingBlocks.clear();
		},
	};
	return state;
}

function getAnthropicProviderSessionState(
	providerSessionState: Map<string, ProviderSessionState> | undefined,
	baseUrl: string,
	modelId: string,
): AnthropicProviderSessionState | undefined {
	if (!providerSessionState) return undefined;
	const key = anthropicProviderSessionStateKey(baseUrl, modelId);
	const existing = providerSessionState.get(key) as AnthropicProviderSessionState | undefined;
	if (existing) {
		existing.prefixDroppedThinkingBlocks ??= new Set();
		return existing;
	}
	const created = createAnthropicProviderSessionState();
	providerSessionState.set(key, created);
	return created;
}

function hasStrictAnthropicTools(params: MessageCreateParamsStreaming): boolean {
	return params.tools?.some(tool => tool.strict === true) ?? false;
}

function dropAnthropicFastMode(params: MessageCreateParamsStreaming): void {
	delete params.speed;
}

function dropAnthropicStrictTools(params: MessageCreateParamsStreaming): void {
	if (!params.tools) return;
	for (const tool of params.tools) {
		delete tool.strict;
	}
}

function getCacheControl(
	model: Model<"anthropic-messages">,
	cacheRetention: CacheRetention | undefined,
	isOAuthToken = false,
): { retention: CacheRetention; cacheControl?: AnthropicCacheControl } {
	// Five-minute writes are the cheapest cache population strategy for pay-per-token API keys.
	// For OAuth (Claude Code subscriber seats), match Claude Code's native policy by defaulting
	// to 1h retention where supported, avoiding cold cache re-writes after 15m idle intervals.
	// An explicit cacheRetention ('short', 'long', 'none') or PI_CACHE_RETENTION always takes precedence.
	const defaultRetention = isOAuthToken && model.compat.supportsLongCacheRetention ? "long" : "short";
	const retention = resolveCacheRetention(cacheRetention, defaultRetention);
	if (retention === "none") {
		return { retention };
	}
	const ttl = retention === "long" && model.compat.supportsLongCacheRetention ? "1h" : undefined;
	return {
		retention,
		cacheControl: { type: "ephemeral", ...(ttl && { ttl }) },
	};
}

// Claude Code mode: mimic the CLI's direct inference transport. Constants live
// in the leaf module so registry/usage consumers avoid an init cycle.
export * from "./claude-code-fingerprint";

/** Maps Node's platform identifier to the Stainless wire value. */
export function mapStainlessOs(platform: string): "MacOS" | "Windows" | "Linux" | "FreeBSD" | `Other::${string}` {
	switch (platform.toLowerCase()) {
		case "darwin":
			return "MacOS";
		case "windows":
		case "win32":
			return "Windows";
		case "linux":
			return "Linux";
		case "freebsd":
			return "FreeBSD";
		default:
			return `Other::${platform.toLowerCase()}`;
	}
}

/** Maps Node's architecture identifier to the Stainless wire value. */
export function mapStainlessArch(arch: string): "x64" | "arm64" | "x86" | `other::${string}` {
	switch (arch.toLowerCase()) {
		case "amd64":
		case "x64":
			return "x64";
		case "arm64":
		case "aarch64":
			return "arm64";
		case "386":
		case "x86":
		case "ia32":
			return "x86";
		default:
			return `other::${arch.toLowerCase()}`;
	}
}

/** Static headers emitted by Claude Code's CLI runtime. */
export const claudeCodeHeaders = {
	"X-Stainless-Arch": mapStainlessArch(process.arch),
	"X-Stainless-Lang": "js",
	"X-Stainless-OS": mapStainlessOs(process.platform),
	"X-Stainless-Package-Version": claudeCodeSdkVersion,
	"X-Stainless-Retry-Count": "0",
	"X-Stainless-Runtime": "node",
	"X-Stainless-Runtime-Version": "v26.3.0",
	"X-Stainless-Timeout": "600",
};

/**
 * Cowork's fingerprint keys. Only the OAuth branch re-emits them, so only that
 * branch may enforce them: everywhere else nothing would fill the gap and
 * filtering would silently delete a client identity the caller built on
 * purpose (Factory Droid mirrors droid's Anthropic SDK fingerprint on this
 * wire, right down to the 600s client timeout).
 */
const coworkHeaderKeys = new Set(Object.keys(claudeCodeHeaders).map(key => key.toLowerCase()));

const enforcedHeaderKeys = new Set(
	[
		...Object.keys(claudeCodeHeaders),
		"Accept",
		"Accept-Encoding",
		"Connection",
		"Content-Type",
		"anthropic-version",
		"anthropic-dangerous-direct-browser-access",
		"anthropic-beta",
		"User-Agent",
		"x-app",
		"Authorization",
		"X-Api-Key",
		"X-Claude-Code-Session-Id",
		"x-client-request-id",
		"cf-aig-authorization",
	].map(key => key.toLowerCase()),
);

const overridableAnthropicHeaderKeys = new Set(
	[...Object.keys(claudeCodeHeaders), "anthropic-beta", "User-Agent", "x-app"].map(key => key.toLowerCase()),
);

const CLAUDE_BILLING_HEADER_PREFIX = "x-anthropic-billing-header:";

function createClaudeBillingHeader(firstUserMessageText: string): string {
	// Fingerprint: SHA256(salt + msg[4] + msg[7] + msg[20] + version)[:3]
	// Matches CC's computeFingerprint in utils/fingerprint.ts.
	// Uses chars from the first user message (not the system prompt).
	const k = [4, 7, 20].map(i => firstUserMessageText[i] ?? "0").join("");
	const version = getClaudeCodeVersion();
	const versionSuffix = Bun.SHA256.hash(`59cf53e54c78${k}${version}`, "hex").slice(0, 3);
	// cch=00000: placeholder replaced with the real attestation hash by wrapFetchForCch
	// before the request hits the wire (see below).
	return `${CLAUDE_BILLING_HEADER_PREFIX} cc_version=${version}.${versionSuffix}; cc_entrypoint=cli; ${CCH_PLACEHOLDER_STR};`;
}

// cch attestation: XXHash64(body_with_placeholder, seed) low-20-bits, 5 hex chars.
const CCH_SEED = 0x4d659218e32a3268n;
const CCH_PLACEHOLDER_STR = "cch=00000";
const cchEncoder = new TextEncoder();
const CCH_PLACEHOLDER = cchEncoder.encode(CCH_PLACEHOLDER_STR);
// Combined anchor for the billing-header placeholder inside system[0].
// "system":[{"type":"text","text":"x-anthropic-billing-header:
// Matches the exact JSON prefix of the first system block when
// createClaudeBillingHeader injects system[0].  "messages" serializes before
// "system" in Anthropic SDK payloads (~byte 29 vs ~byte 4705), so user content
// in the messages array can never match this sequence.  User system prompt text
// lives in system[2] and therefore also cannot match.
const BILLING_SYSTEM_MARKER = cchEncoder.encode(`"system":[{"type":"text","text":"${CLAUDE_BILLING_HEADER_PREFIX}`);
const CCH_BILLING_SEARCH_WINDOW = 150;

function patchCch(body: Uint8Array): "patched" | "no-billing-header" | "unanchored" {
	// Zero-copy Buffer view over the same memory; its `indexOf` is a native memmem,
	// ~7.5x faster than a hand-rolled byte loop here — the marker sits ~99% through
	// the body because `messages` serializes before `system`, so a JS scan would
	// walk almost the entire payload (benchmarked: 563µs -> 75µs on a 1MB body).
	const view = Buffer.from(body.buffer, body.byteOffset, body.byteLength);

	// Find the combined system[0] + billing-header prefix marker.
	const markerIdx = view.indexOf(BILLING_SYSTEM_MARKER);
	if (markerIdx === -1) return "no-billing-header"; // no CC billing header injected

	// Placeholder must sit within CCH_BILLING_SEARCH_WINDOW bytes after the marker.
	const searchFrom = markerIdx + BILLING_SYSTEM_MARKER.length;
	const idx = view.indexOf(CCH_PLACEHOLDER, searchFrom);
	if (idx === -1 || idx - searchFrom > CCH_BILLING_SEARCH_WINDOW) return "unanchored";

	// Hash the body with the placeholder in place (matches CC's in-place behaviour).
	const h = Bun.hash.xxHash64(body, CCH_SEED);
	const cch = (h & 0xfffffn).toString(16).padStart(5, "0");

	for (let i = 0; i < 5; i++) body[idx + 4 + i] = cch.charCodeAt(i);
	return "patched";
}

/**
 * Wraps a fetch implementation to patch the Claude Code billing-header `cch`
 * attestation into outgoing request bodies. Bodies without the placeholder
 * pass through untouched, so installing it on every OAuth flow is safe.
 */
export function wrapFetchForCch(base: FetchImpl): FetchImpl {
	return (input, init) => {
		if (init?.body && typeof init.body === "string" && init.body.includes(CCH_PLACEHOLDER_STR)) {
			const encoded = cchEncoder.encode(init.body);
			if (patchCch(encoded) === "unanchored") {
				// The OAuth billing placeholder is anchored to system[0] but we couldn't
				// patch it — e.g. an `onPayload` hook reordered the first system block's keys
				// so BILLING_SYSTEM_MARKER no longer matches. Send the body as-is (cch stays
				// `00000`, the prior behaviour) rather than failing the request, but surface the
				// fingerprint regression instead of letting it ship silently. A `cch=00000`
				// literal in user content alone ("no-billing-header") is not a regression.
				logger.warn("anthropic: cch billing placeholder present but not patched; sending unattested request");
			}
			return base(input, { ...init, body: encoded });
		}
		return base(input, init);
	};
}

const UMANS_WEBSEARCH_PROVIDER_HEADER = "X-Umans-Websearch-Provider";
const UMANS_WEBSEARCH_TOOL_NAME = "web_search";

function normalizeUmansWebSearchProvider(value: string | undefined): "native" | "exa" | undefined {
	const normalized = value?.trim().toLowerCase();
	return normalized === "native" || normalized === "exa" ? normalized : undefined;
}

function getUmansWebSearchProvider(headers: Record<string, string> | undefined): "native" | "exa" | undefined {
	const explicit = getHeaderCaseInsensitive(headers, UMANS_WEBSEARCH_PROVIDER_HEADER);
	if (explicit !== undefined) return normalizeUmansWebSearchProvider(explicit);
	return normalizeUmansWebSearchProvider($env.UMANS_WEBSEARCH_PROVIDER);
}

function isUmansAnthropicModel(model: Model<"anthropic-messages">): boolean {
	return model.provider === "umans" || model.baseUrl.toLowerCase().includes("api.code.umans.ai");
}

function getUmansWebSearchHeader(
	model: Model<"anthropic-messages">,
	headers: Record<string, string> | undefined,
): Record<string, string> | undefined {
	if (!isUmansAnthropicModel(model)) return undefined;
	const provider = getUmansWebSearchProvider(headers);
	return provider ? { [UMANS_WEBSEARCH_PROVIDER_HEADER]: provider } : undefined;
}

function shouldUseUmansGatewayWebSearch(name: string, enabled: boolean): boolean {
	return enabled && name.toLowerCase() === UMANS_WEBSEARCH_TOOL_NAME;
}

function encodeAnthropicToolName(
	name: string,
	isOAuthToken: boolean,
	escapeBuiltinToolNames: boolean,
	useUmansGatewayWebSearch = false,
): string {
	if (shouldUseUmansGatewayWebSearch(name, useUmansGatewayWebSearch)) return name;
	if (escapeBuiltinToolNames) return `${claudeToolPrefix}${name}`;
	return isOAuthToken ? applyClaudeToolPrefix(name) : name;
}

function decodeAnthropicToolName(name: string, isOAuthToken: boolean, escapeBuiltinToolNames: boolean): string {
	if (isOAuthToken || escapeBuiltinToolNames) return stripClaudeToolPrefix(name);
	return name;
}

const ANTHROPIC_MANY_IMAGE_THRESHOLD = 20;
const ANTHROPIC_MANY_IMAGE_MAX_DIMENSION = 2000;

function countAnthropicImageBlocks(messages: Message[]): number {
	let count = 0;
	for (const message of messages) {
		if (message.role !== "user" && message.role !== "developer" && message.role !== "toolResult") continue;
		if (!Array.isArray(message.content)) continue;
		for (const block of message.content) {
			if (block.type === "image") count++;
		}
	}
	return count;
}

const ANTHROPIC_IMAGE_RESIZE_CONCURRENCY = 4;

/**
 * Memoized resize results keyed on ImageContent identity. Callers keep message
 * objects stable across turns, so without this every request (and every
 * in-provider retry of a fresh turn) re-decodes and re-encodes the same
 * oversized screenshots. A cached value identical to the key means "already
 * within bounds / unresizable — skip the decode".
 */
const anthropicManyImageResizeCache = new WeakMap<ImageContent, ImageContent>();

type ResizeLimiter = <R>(fn: () => Promise<R>) => Promise<R>;

/**
 * Bounded-concurrency gate for image decode/encode work. The many-image path
 * fans out over every block of every message; unbounded, 100+ oversized images
 * would decode concurrently (two encode pipelines each) and spike memory by
 * gigabytes. Slots are handed off directly to the next waiter on release.
 */
function createResizeLimiter(limit: number): ResizeLimiter {
	let active = 0;
	const queue: (() => void)[] = [];
	return async fn => {
		if (active >= limit) {
			const { promise, resolve } = Promise.withResolvers<void>();
			queue.push(resolve);
			await promise;
		} else {
			active++;
		}
		try {
			return await fn();
		} finally {
			const next = queue.shift();
			if (next) next();
			else active--;
		}
	};
}

async function resizeAnthropicManyImageBlock(block: ImageContent): Promise<ImageContent> {
	try {
		const inputBuffer = Buffer.from(block.data, "base64");
		const { width, height } = await new Bun.Image(inputBuffer).metadata();
		if (!width || !height) return block;
		if (width <= ANTHROPIC_MANY_IMAGE_MAX_DIMENSION && height <= ANTHROPIC_MANY_IMAGE_MAX_DIMENSION) return block;

		const scale = Math.min(ANTHROPIC_MANY_IMAGE_MAX_DIMENSION / width, ANTHROPIC_MANY_IMAGE_MAX_DIMENSION / height);
		const targetWidth = Math.max(1, Math.min(ANTHROPIC_MANY_IMAGE_MAX_DIMENSION, Math.round(width * scale)));
		const targetHeight = Math.max(1, Math.min(ANTHROPIC_MANY_IMAGE_MAX_DIMENSION, Math.round(height * scale)));

		const [png, jpeg] = await Promise.all([
			new Bun.Image(inputBuffer).resize(targetWidth, targetHeight).png().bytes(),
			new Bun.Image(inputBuffer).resize(targetWidth, targetHeight).jpeg({ quality: 85 }).bytes(),
		]);
		const best =
			png.length <= jpeg.length ? { buffer: png, mimeType: "image/png" } : { buffer: jpeg, mimeType: "image/jpeg" };

		return {
			type: "image",
			data: Buffer.from(best.buffer).toString("base64"),
			mimeType: best.mimeType,
		};
	} catch (error) {
		logger.warn("anthropic: failed to resize oversized image for many-image request", {
			mimeType: block.mimeType,
			error: error instanceof Error ? error.message : String(error),
		});
		return block;
	}
}

async function resizeAnthropicManyImageContent(
	content: (TextContent | ImageContent)[],
	state: { resized: number },
	limit: ResizeLimiter,
): Promise<(TextContent | ImageContent)[]> {
	let changed = false;
	const next = await Promise.all(
		content.map(async block => {
			// Remotely referenced blocks never put base64 on the wire, so their size
			// cannot violate the many-image request budget — and resizing would
			// desync fallback bytes from the advertised remote image.
			if (
				block.type !== "image" ||
				block.url ||
				(block.providerFile?.provider === "anthropic" && block.providerFile.id)
			)
				return block;
			let resized = anthropicManyImageResizeCache.get(block);
			if (resized === undefined) {
				resized = await limit(() => resizeAnthropicManyImageBlock(block));
				anthropicManyImageResizeCache.set(block, resized);
			}
			if (resized !== block) {
				changed = true;
				state.resized++;
			}
			return resized;
		}),
	);
	return changed ? next : content;
}

async function resizeAnthropicManyImageMessage(
	message: Message,
	state: { resized: number },
	limit: ResizeLimiter,
): Promise<Message> {
	if (message.role === "user" || message.role === "developer") {
		if (!Array.isArray(message.content)) return message;
		const content = await resizeAnthropicManyImageContent(message.content, state, limit);
		return content === message.content ? message : { ...message, content };
	}
	if (message.role === "toolResult") {
		const content = await resizeAnthropicManyImageContent(message.content, state, limit);
		return content === message.content ? message : { ...message, content };
	}
	return message;
}

async function prepareAnthropicManyImageContext(context: Context, supportsImages: boolean): Promise<Context> {
	if (!supportsImages) return context;
	const imageCount = countAnthropicImageBlocks(context.messages);
	if (imageCount <= ANTHROPIC_MANY_IMAGE_THRESHOLD) return context;

	let changed = false;
	const state = { resized: 0 };
	const limit = createResizeLimiter(ANTHROPIC_IMAGE_RESIZE_CONCURRENCY);
	const messages = await Promise.all(
		context.messages.map(async message => {
			const next = await resizeAnthropicManyImageMessage(message, state, limit);
			if (next !== message) changed = true;
			return next;
		}),
	);
	if (!changed) return context;
	logger.debug("anthropic: resized oversized images for many-image request", {
		imageCount,
		resized: state.resized,
		maxDimension: ANTHROPIC_MANY_IMAGE_MAX_DIMENSION,
	});
	return { ...context, messages };
}

type AnthropicImageSource =
	| { type: "base64"; media_type: AnthropicImageMediaType; data: string }
	| { type: "url"; url: string }
	| { type: "file"; file_id: string };

type AnthropicToolResultContent =
	| string
	| Array<{ type: "text"; text: string } | { type: "image"; source: AnthropicImageSource }>;

/**
 * Convert content blocks to Anthropic API format
 */
function convertContentBlocks(
	content: (TextContent | ImageContent)[],
	supportsImages = true,
): AnthropicToolResultContent {
	const blocks: Array<{ type: "text"; text: string } | { type: "image"; source: AnthropicImageSource }> = [];
	let sawText = false;
	let sawImage = false;

	for (const block of content) {
		if (block.type === "text") {
			const text = block.text.toWellFormed();
			if (text.trim().length === 0) continue;
			sawText = true;
			blocks.push({ type: "text", text });
			continue;
		}

		if (!supportsImages) {
			blocks.push({ type: "text", text: NON_VISION_IMAGE_PLACEHOLDER });
			continue;
		}

		let source: AnthropicImageSource;
		if (block.providerFile?.provider === "anthropic" && block.providerFile.id) {
			source = { type: "file", file_id: block.providerFile.id };
		} else if (block.url) {
			source = { type: "url", url: block.url };
		} else {
			const mediaType = normalizeAnthropicImageMediaType(block.mimeType);
			if (!mediaType) {
				blocks.push({ type: "text", text: `[unsupported image: ${block.mimeType}]` });
				continue;
			}
			source = { type: "base64", media_type: mediaType, data: block.data };
		}

		sawImage = true;
		blocks.push({ type: "image", source });
	}

	if (!supportsImages) {
		return blocks
			.filter((block): block is { type: "text"; text: string } => block.type === "text")
			.map(block => block.text)
			.join("\n")
			.toWellFormed();
	}

	if (sawImage && !sawText) {
		blocks.unshift({
			type: "text",
			text: "(see attached image)",
		});
	}

	return blocks;
}

export type AnthropicEffort = AnthropicOutputEffort | "adaptive";
export type AnthropicThinkingDisplay = "summarized" | "omitted";

export interface AnthropicOptions extends StreamOptions {
	/**
	 * Enable extended thinking.
	 * For adaptive-capable models (Opus 4.6+, Sonnet 4.6+, Fable/Mythos 5):
	 * uses adaptive thinking (Claude decides when/how much to think). For older
	 * models: uses budget-based thinking with thinkingBudgetTokens.
	 */
	thinkingEnabled?: boolean;
	/**
	 * Token budget for extended thinking (older models only).
	 * Ignored for adaptive-capable models.
	 */
	thinkingBudgetTokens?: number;
	/**
	 * Upstream wire model id override for collapsed effort-tier variants.
	 * Serialized as `requestModelId ?? model.requestModelId ?? model.id`.
	 */
	requestModelId?: string;
	/**
	 * Effort level for adaptive thinking.
	 * Controls how much Claude allocates, or uses "adaptive" for MiniMax's
	 * binary adaptive-thinking tag:
	 * - "max": Always thinks with no constraints
	 * - "high": Always thinks, deep reasoning (default)
	 * - "medium": Moderate thinking, may skip for simple queries
	 * - "low": Minimal thinking, skips for simple tasks
	 * - "adaptive": Sends `thinking.type: "adaptive"` without `output_config.effort`
	 * Ignored for older models.
	 */
	effort?: AnthropicEffort;
	/**
	 * Optional reasoning level fallback for direct Anthropic provider usage.
	 * Converted to adaptive effort when effort is not explicitly provided.
	 */
	reasoning?: SimpleStreamOptions["reasoning"];
	/**
	 * Controls how Anthropic returns thinking content when the selected thinking
	 * transport supports a display option. Defaults to "summarized" where the
	 * API accepts it.
	 */
	thinkingDisplay?: AnthropicThinkingDisplay;
	interleavedThinking?: boolean;
	toolChoice?: "auto" | "any" | "none" | { type: "tool"; name: string };
	betas?: string[] | string;
	/**
	 * Realization of `serviceTier: "priority"` on Anthropic models. When
	 * `"priority"`, sets `speed: "fast"` on the request and appends the
	 * `fast-mode-2026-02-01` beta header. Anthropic rejects unsupported models
	 * with `invalid_request_error`, which triggers an in-provider one-shot
	 * fallback (see `fastModeDisabled` provider state).
	 *
	 * Other `ServiceTier` values are currently ignored on this provider.
	 */
	serviceTier?: ServiceTier;
	/** Force OAuth bearer auth mode for proxy tokens that don't match Anthropic token prefixes. */
	isOAuth?: boolean;
	/**
	 * Pre-built Anthropic Messages client. When provided, skips internal client
	 * construction entirely. Accepts any structurally compatible client,
	 * including SDK clients such as `AnthropicVertex`.
	 */
	client?: AnthropicMessagesClientLike;
	/**
	 * Server-side fallback beta chain (`server-side-fallback-2026-06-01`).
	 * When set, `fallbacks` is forwarded on the request body and the beta
	 * header is auto-attached; the response parser then honors mid-stream
	 * `fallback` content blocks and `usage.iterations` for served-model
	 * promotion and per-attempt pricing. Opt-in ONLY — leaving this
	 * undefined preserves the pre-fallback behavior on every code path.
	 */
	fallbacks?: FallbackParam[];
}

export type AnthropicClientOptionsArgs = {
	model: Model<"anthropic-messages">;
	apiKey: string;
	extraBetas?: string[];
	stream?: boolean;
	interleavedThinking?: boolean;
	headers?: Record<string, string>;
	dynamicHeaders?: Record<string, string>;
	isOAuth?: boolean;
	hasTools?: boolean;
	thinkingEnabled?: boolean;
	thinkingDisplay?: AnthropicThinkingDisplay;
	disableStrictTools?: boolean;
	fetch?: FetchImpl;
	maxRetryDelayMs?: number;
	sessionId?: string;
	/** Working-identity cache key for this credential+host; undefined off the Copilot path. */
	copilotCacheKey?: string;
	/**
	 * Build-time cache provenance for the wrapper: the cached value the
	 * outgoing headers were built from, or `null` when the cache was empty at
	 * build. `undefined` rereads the cache at dispatch.
	 */
	copilotCacheSnapshot?: string | null;
};

export type AnthropicClientOptionsResult = {
	isOAuthToken: boolean;
	apiKey: string | null;
	authToken?: string | null;
	baseURL?: string;
	maxRetries: number;
	maxRetryDelayMs?: number;
	defaultHeaders: Record<string, string>;
	fetch?: FetchImpl;
	fetchOptions?: AnthropicFetchOptions;
};

const CLAUDE_CODE_TLS_CIPHERS = tls.DEFAULT_CIPHERS;

type FoundryTlsOptions = {
	ca?: string | string[];
	cert?: string;
	key?: string;
};

const foundryTlsOptionsCache = new Map<string, FoundryTlsOptions | undefined>();

function foundryTlsCacheKeyComponent(value: string | undefined): string | null {
	if (!value) return null;
	const trimmed = value.trim();
	// For path-valued vars, fold the file mtime into the key so on-disk cert
	// rotation (common for short-lived corporate mTLS certs) invalidates the
	// cached TLS options instead of pinning the first read forever.
	if (trimmed && !trimmed.includes("-----BEGIN") && looksLikeFilePath(trimmed)) {
		try {
			return `${trimmed}@${fs.statSync(trimmed).mtimeMs}`;
		} catch {
			return trimmed;
		}
	}
	return value;
}

function foundryTlsOptionsCacheKey(): string {
	return JSON.stringify([
		foundryTlsCacheKeyComponent($env.NODE_EXTRA_CA_CERTS),
		foundryTlsCacheKeyComponent($env.CLAUDE_CODE_CLIENT_CERT),
		foundryTlsCacheKeyComponent($env.CLAUDE_CODE_CLIENT_KEY),
	]);
}

function resolveAnthropicBaseUrl(model: Model<"anthropic-messages">, apiKey?: string): string | undefined {
	if (model.provider === "github-copilot") {
		return normalizeAnthropicBaseUrl(resolveGitHubCopilotBaseUrl(model.baseUrl, apiKey) ?? model.baseUrl);
	}
	if (model.provider === "anthropic") return resolveDirectAnthropicBaseUrl(model);
	return normalizeAnthropicBaseUrl(model.baseUrl);
}

function resolveEagerToolInputStreamingSupport(
	model: Model<"anthropic-messages">,
	effectiveBaseUrl: string | undefined,
): boolean {
	if (!model.compat.supportsEagerToolInputStreaming) return false;
	// First-party Anthropic endpoints accept the per-tool flag.
	if (isOfficialAnthropicApiUrl(effectiveBaseUrl)) return true;
	// Non-official effective endpoint. `supportsEagerToolInputStreaming` may be
	// stale-true here because compat is materialized once at build time and is
	// never rebuilt for a baseUrl-only reroute — either a runtime provider
	// override (`pi.registerProvider("anthropic", { baseUrl })`) or Foundry
	// (`CLAUDE_CODE_USE_FOUNDRY`). Both leave the canonical model's resolved
	// compat in place. `officialEndpoint` records whether compat was built for
	// the canonical Anthropic URL, so only endpoints whose compat was authored
	// for a non-official host (an explicit `compat.supportsEagerToolInputStreaming`
	// opt-in on a custom `baseUrl`) still send the field.
	return !model.compat.officialEndpoint;
}

function parseAnthropicCustomHeaders(rawHeaders: string | undefined): Record<string, string> | undefined {
	const source = rawHeaders?.trim();
	if (!source) return undefined;

	const parsed: Record<string, string> = {};
	for (const token of source.split(/\r?\n|,/)) {
		const entry = token.trim();
		if (!entry) continue;
		const separatorIndex = entry.indexOf(":");
		if (separatorIndex <= 0) continue;
		const key = entry.slice(0, separatorIndex).trim();
		const value = entry.slice(separatorIndex + 1).trim();
		if (!key || !value) continue;
		parsed[key] = value;
	}

	return Object.keys(parsed).length > 0 ? parsed : undefined;
}

/**
 * Returns env-supplied custom headers (`ANTHROPIC_CUSTOM_HEADERS`) when they
 * should be forwarded to the upstream endpoint.
 *
 * Foundry mode forwards them unconditionally. Outside Foundry, they're applied
 * only when the configured base URL is a non-Anthropic host — i.e. an
 * enterprise/corporate gateway that may require its own proprietary auth
 * header. Stock `api.anthropic.com` would reject unknown headers, so they're
 * omitted there.
 */
export function resolveAnthropicCustomHeadersForBaseUrl(
	baseUrl: string | undefined,
): Record<string, string> | undefined {
	if (!isFoundryEnabled() && isOfficialAnthropicApiUrl(baseUrl)) return undefined;
	return parseAnthropicCustomHeaders($env.ANTHROPIC_CUSTOM_HEADERS);
}

function resolveAnthropicCustomHeaders(
	model: Model<"anthropic-messages">,
	baseUrl: string | undefined,
): Record<string, string> | undefined {
	if (model.provider !== "anthropic") return undefined;
	return resolveAnthropicCustomHeadersForBaseUrl(baseUrl);
}

function looksLikeFilePath(value: string): boolean {
	return value.includes("/") || value.includes("\\") || /\.(pem|crt|cer|key)$/i.test(value);
}

function resolvePemValue(value: string | undefined, name: string): string | undefined {
	const trimmed = value?.trim();
	if (!trimmed) return undefined;

	const inline = trimmed.replace(/\\n/g, "\n");
	if (inline.includes("-----BEGIN")) {
		return inline;
	}

	if (looksLikeFilePath(trimmed)) {
		try {
			return fs.readFileSync(trimmed, "utf8");
		} catch (error) {
			if (isEnoent(error)) {
				throw new AIError.ValidationError(`${name} path does not exist: ${trimmed}`);
			}
			throw error;
		}
	}

	return inline;
}

function resolveFoundryTlsOptions(model: Model<"anthropic-messages">): FoundryTlsOptions | undefined {
	if (model.provider !== "anthropic") return undefined;
	if (!isFoundryEnabled()) return undefined;

	const cacheKey = foundryTlsOptionsCacheKey();
	if (foundryTlsOptionsCache.has(cacheKey)) return foundryTlsOptionsCache.get(cacheKey);

	const ca = resolvePemValue($env.NODE_EXTRA_CA_CERTS, "NODE_EXTRA_CA_CERTS");
	const cert = resolvePemValue($env.CLAUDE_CODE_CLIENT_CERT, "CLAUDE_CODE_CLIENT_CERT");
	const key = resolvePemValue($env.CLAUDE_CODE_CLIENT_KEY, "CLAUDE_CODE_CLIENT_KEY");

	if ((cert && !key) || (!cert && key)) {
		throw new AIError.ConfigurationError(
			"Both CLAUDE_CODE_CLIENT_CERT and CLAUDE_CODE_CLIENT_KEY must be set for mTLS.",
		);
	}

	const options: FoundryTlsOptions = {};
	if (ca) options.ca = [...tls.rootCertificates, ca];
	if (cert) options.cert = cert;
	if (key) options.key = key;
	const resolved = Object.keys(options).length > 0 ? options : undefined;
	foundryTlsOptionsCache.set(cacheKey, resolved);
	return resolved;
}

function buildClaudeCodeTlsFetchOptions(
	model: Model<"anthropic-messages">,
	baseUrl: string | undefined,
): AnthropicFetchOptions | undefined {
	if (model.provider !== "anthropic") return undefined;
	if (!baseUrl) return undefined;

	let serverName: string;
	try {
		serverName = new URL(baseUrl).hostname;
	} catch {
		return undefined;
	}

	if (!serverName) return undefined;

	const foundryTlsOptions = resolveFoundryTlsOptions(model);

	return {
		tls: {
			rejectUnauthorized: true,
			serverName,
			...(CLAUDE_CODE_TLS_CIPHERS ? { ciphers: CLAUDE_CODE_TLS_CIPHERS } : {}),
			...foundryTlsOptions,
		},
	};
}
function mergeHeaders(...headerSources: (Record<string, string> | undefined)[]): Record<string, string> {
	// Case-insensitive merge: later sources win and keep their casing. A plain
	// Object.assign would let `authorization` and `Authorization` coexist, and
	// the Headers constructor then joins both values comma-separated on the wire.
	const merged: Record<string, string> = {};
	const keyByLower = new Map<string, string>();
	for (const headers of headerSources) {
		if (!headers) continue;
		for (const [key, value] of Object.entries(headers)) {
			const lower = key.toLowerCase();
			const existing = keyByLower.get(lower);
			if (existing !== undefined && existing !== key) delete merged[existing];
			keyByLower.set(lower, key);
			merged[key] = value;
		}
	}
	return merged;
}

const ANTHROPIC_MESSAGE_EVENTS: ReadonlySet<string> = new Set([
	"message_start",
	"message_delta",
	"message_stop",
	"content_block_start",
	"content_block_delta",
	"content_block_stop",
]);

/**
 * Iterate over Anthropic SSE events from a raw Response, preserving ping events
 * for liveness. Malformed event envelopes are logged and skipped (non-fatal)
 * rather than aborting the stream.
 */
type RawMessagePingEvent = { type: "ping" };
type AnthropicStreamEvent = RawMessageStreamEvent | RawMessagePingEvent;
const ANTHROPIC_PING_EVENT: RawMessagePingEvent = { type: "ping" };

/**
 * In-stream `error` SSE frames carry an Anthropic error envelope:
 * `{"type":"error","error":{"type":"overloaded_error","message":"Overloaded"}}`.
 * Surface the structured type + message instead of the raw JSON blob; the
 * error type token (e.g. `overloaded_error`, `rate_limit_error`) is kept in
 * the message so `isProviderRetryableError`'s classification keys off the
 * structured type rather than incidental JSON substrings.
 */
function createAnthropicSseStreamError(data: string): Error {
	try {
		const parsed = JSON.parse(data) as { error?: { type?: unknown; message?: unknown } };
		const errorType = typeof parsed?.error?.type === "string" ? parsed.error.type : undefined;
		const message = typeof parsed?.error?.message === "string" ? parsed.error.message : undefined;
		if (message) {
			return new AIError.ProviderResponseError(
				errorType ? `Anthropic stream error (${errorType}): ${message}` : `Anthropic stream error: ${message}`,
				{ provider: "anthropic", kind: "output" },
			);
		}
	} catch {
		// Not a JSON envelope; fall through to the raw payload.
	}
	return new AIError.ProviderResponseError(data, { provider: "anthropic", kind: "output" });
}

async function* iterateAnthropicEvents(
	response: Response,
	signal?: AbortSignal,
	onSseEvent?: AnthropicOptions["onSseEvent"],
): AsyncGenerator<AnthropicStreamEvent> {
	if (!response.body) {
		throw new AIError.AnthropicStreamEnvelopeError("Attempted to iterate over an Anthropic response with no body");
	}

	let sawMessageStart = false;
	let sawMessageEnd = false;

	// Capture `raw` only when the diagnostic observer exists; otherwise the
	// per-frame wire-line array is pure token-path garbage.
	for await (const sse of readSseEvents(response.body, signal, onSseEvent ? { captureRaw: true } : undefined)) {
		notifyRawSseEvent(onSseEvent, sse);
		if (sse.event === "error") {
			throw createAnthropicSseStreamError(sse.data);
		}

		if (sse.event === "ping") {
			// Surface keepalives so the idle watchdog treats them as liveness.
			yield ANTHROPIC_PING_EVENT;
			continue;
		}

		if (!ANTHROPIC_MESSAGE_EVENTS.has(sse.event ?? "")) {
			continue;
		}

		try {
			const event = JSON.parse(sse.data) as RawMessageStreamEvent;
			if (event.type !== sse.event) {
				reportAnthropicEnvelopeAnomaly(`event type ${event.type} does not match SSE event ${sse.event}`);
			}
			if (event.type === "message_start") {
				sawMessageStart = true;
			} else if (event.type === "message_stop") {
				sawMessageEnd = true;
			}
			yield event;
		} catch (error) {
			const message = error instanceof Error ? error.message : String(error);
			reportAnthropicEnvelopeAnomaly(
				`could not parse SSE event ${sse.event}: ${message}; skipping frame; data=${sse.data}`,
			);
		}
	}

	if (sawMessageStart && !sawMessageEnd && !signal?.aborted) {
		reportAnthropicEnvelopeAnomaly("stream ended before message_stop");
	}
}

type AnthropicRawResponseRequest = {
	asResponse(): Promise<Response>;
};

function hasAnthropicRawResponseRequest(request: unknown): request is AnthropicRawResponseRequest {
	return isRecord(request) && typeof request.asResponse === "function";
}

type AnthropicStreamWithResponseRequest = {
	withResponse(): Promise<{
		data: AsyncIterable<RawMessageStreamEvent>;
		response: Response;
		request_id: string | null;
	}>;
};

function hasAnthropicStreamWithResponseRequest(request: unknown): request is AnthropicStreamWithResponseRequest {
	return isRecord(request) && typeof request.withResponse === "function";
}

async function getAnthropicStreamResponse(
	request: unknown,
	signal?: AbortSignal,
	onSseEvent?: AnthropicOptions["onSseEvent"],
): Promise<{
	events: AsyncIterable<AnthropicStreamEvent>;
	response: Response;
	requestId: string | null;
	recordsRawSseEvents: boolean;
}> {
	if (hasAnthropicRawResponseRequest(request)) {
		const response = await request.asResponse();
		return {
			events: iterateAnthropicEvents(response, signal, onSseEvent),
			response,
			requestId: response.headers.get("request-id"),
			recordsRawSseEvents: true,
		};
	}
	if (hasAnthropicStreamWithResponseRequest(request)) {
		const { data, response, request_id } = await request.withResponse();
		return { events: data, response, requestId: request_id, recordsRawSseEvents: false };
	}
	throw new AIError.AnthropicStreamEnvelopeError("Anthropic SDK request did not expose a stream response");
}

async function* observeDecodedAnthropicSdkEvents(
	events: AsyncIterable<AnthropicStreamEvent>,
	observer: (event: RawSseEvent) => void,
): AsyncGenerator<AnthropicStreamEvent> {
	for await (const event of events) {
		const data = JSON.stringify(event);
		// Reconstructed from decoded SDK event; not literal wire bytes.
		notifyRawSseEvent(observer, { event: event.type, data, raw: [`event: ${event.type}`, `data: ${data}`] });
		yield event;
	}
}

const PROVIDER_MAX_RETRIES = 10;

/**
 * How long `ping` keepalives may keep extending the idle deadline without any
 * semantic stream progress, as a multiple of the idle timeout. Anthropic pings
 * across legitimate generation gaps, so pings count as liveness — but a wedged
 * upstream that pings forever while producing no events must eventually trip
 * the idle watchdog instead of hanging an active tool-call stream without a
 * recovery path (#4900).
 */
const PING_PROGRESS_MAX_IDLE_MULTIPLIER = 3;

/**
 * Log a malformed-stream-envelope anomaly without aborting the turn. The strict
 * parser would `throw new AnthropicStreamEnvelopeError(...)` here; we instead
 * surface a warning and let the caller skip the offending event (or finalize what
 * already streamed) so a non-conforming endpoint degrades to best-effort content
 * rather than failing the request.
 */
function reportAnthropicEnvelopeAnomaly(detail: string): void {
	logger.warn(`anthropic: ignoring malformed stream envelope: ${detail}`);
}

function shouldIgnoreAnthropicPreambleEvent(eventType: unknown): boolean {
	if (typeof eventType !== "string") return false;
	if (eventType === "ping") return true;
	return !ANTHROPIC_MESSAGE_EVENTS.has(eventType);
}

const THINKING_ENVELOPE_OPEN = "<thinking>";
const THINKING_ENVELOPE_CLOSE = "</thinking>";

function unwrapAnthropicThinkingEnvelope(text: string): string | undefined {
	let current = text.trim();
	let stripped = false;
	while (current.startsWith(THINKING_ENVELOPE_OPEN) && current.endsWith(THINKING_ENVELOPE_CLOSE)) {
		current = current.slice(THINKING_ENVELOPE_OPEN.length, current.length - THINKING_ENVELOPE_CLOSE.length).trim();
		stripped = true;
	}
	return stripped ? current : undefined;
}

/**
 * The refused response's content as the continuation prefix: client tool
 * calls (no matching tool_result) are omitted and a trailing text block is
 * right-trimmed, per the fallback-credit continuation contract.
 */
function refusalContinuationPrefix(content: AssistantMessage["content"]): AssistantMessage["content"] {
	const prefix = structuredClone(content.filter(block => block.type !== "toolCall"));
	const last = prefix.at(-1);
	if (last?.type === "text") last.text = last.text.trimEnd();
	return prefix;
}

/** Wire form of {@link refusalContinuationPrefix} for the appended assistant message. */
function formatEchoedRefusalContent(content: AssistantMessage["content"]): unknown[] {
	return refusalContinuationPrefix(content).map(block => {
		switch (block.type) {
			case "text":
				return { type: "text", text: block.text };
			case "thinking":
				return {
					type: "thinking",
					thinking: block.thinking,
					...(block.thinkingSignature ? { signature: block.thinkingSignature } : {}),
				};
			case "redactedThinking":
				return { type: "redacted_thinking", data: block.data };
			case "fallback":
				return { type: "fallback", from: block.from, to: block.to };
			case "anthropicServerTool":
				return block.block;
			default:
				return block;
		}
	});
}

function isAnthropicBadRequest(error: unknown): boolean {
	if (!error) return false;
	if (typeof error === "object") {
		const rec = error as Record<string, unknown>;
		if (rec.status === 400 || rec.statusCode === 400) return true;
		if (typeof rec.message === "string" && (/\b400\b/.test(rec.message) || rec.message.includes("BadRequestError"))) {
			return true;
		}
	}
	return false;
}

function createEmptyUsage(premiumRequests?: number): Usage {
	return {
		input: 0,
		output: 0,
		cacheRead: 0,
		cacheWrite: 0,
		totalTokens: 0,
		...(premiumRequests === undefined ? {} : { premiumRequests }),
		cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
	};
}

export type AnthropicUsageLike = {
	cache_creation?: { ephemeral_5m_input_tokens?: number | null; ephemeral_1h_input_tokens?: number | null } | null;
	server_tool_use?: { web_search_requests?: number | null; web_fetch_requests?: number | null } | null;
};

/**
 * Capture Anthropic's optional cache-creation TTL breakdown and server-tool-use
 * counters into the harness Usage shape. Omitted/null fields are no-ops; explicit
 * zero-valued objects clear prior extras from earlier stream usage snapshots.
 */
export function applyAnthropicUsageExtras(usage: Usage, source: AnthropicUsageLike): void {
	const cacheCreation = source.cache_creation;
	if (cacheCreation != null) {
		const fiveMinute = cacheCreation.ephemeral_5m_input_tokens ?? 0;
		const oneHour = cacheCreation.ephemeral_1h_input_tokens ?? 0;
		if (fiveMinute > 0 || oneHour > 0) {
			usage.cttl = {
				...(fiveMinute > 0 ? { ephemeral5m: fiveMinute } : {}),
				...(oneHour > 0 ? { ephemeral1h: oneHour } : {}),
			};
		} else {
			delete usage.cttl;
		}
	}
	const serverToolUse = source.server_tool_use;
	if (serverToolUse != null) {
		const webSearch = serverToolUse.web_search_requests ?? 0;
		const webFetch = serverToolUse.web_fetch_requests ?? 0;
		if (webSearch > 0 || webFetch > 0) {
			usage.server = {
				...(webSearch > 0 ? { webSearch } : {}),
				...(webFetch > 0 ? { webFetch } : {}),
			};
		} else {
			delete usage.server;
		}
	}
}

function parseAnthropicFallbackWireBlock(value: unknown): AnthropicFallbackContent | undefined {
	if (!isRecord(value) || value.type !== "fallback") return undefined;
	const from = isRecord(value.from) && typeof value.from.model === "string" ? value.from.model : undefined;
	const to = isRecord(value.to) && typeof value.to.model === "string" ? value.to.model : undefined;
	if (!from?.trim() || !to?.trim()) return undefined;
	return { type: "fallback", from: { model: from }, to: { model: to } };
}

/**
 * Whether a persisted compaction summary replays as a native `compaction`
 * block: only the provider that produced it may replay it, and only on a
 * request whose endpoint supports compaction (the caller decides that with
 * {@link supportsAnthropicCompaction}); every other model reads the text.
 */
function isReplayableAnthropicCompaction(
	payload: ProviderPayload | undefined,
	model: Model<"anthropic-messages">,
): payload is AnthropicCompactionPayload {
	return payload?.type === "anthropicCompaction" && payload.provider === model.provider && payload.content.length > 0;
}

/** Which persisted compaction summaries a request replays as native blocks. */
interface AnthropicCompactionReplay {
	model: Model<"anthropic-messages">;
	/** Replay persisted legacy threshold blocks (encrypted content) too. */
	legacy: boolean;
}

/**
 * Whether `payload` goes on the wire as a `compaction` block under `replay`:
 * replayable for the model, and carrying a signature or (when legacy replay is
 * on) the legacy ciphertext. Anything else is sent as the summary's text.
 */
function replaysAnthropicCompactionBlock(
	payload: ProviderPayload | undefined,
	replay: AnthropicCompactionReplay,
): payload is AnthropicCompactionPayload {
	return (
		isReplayableAnthropicCompaction(payload, replay.model) &&
		(payload.signature !== undefined || (replay.legacy && payload.encryptedContent !== undefined))
	);
}

/** The wire block for a replayed compaction payload, opaque state included. */
function compactionBlockParam(payload: AnthropicCompactionPayload): CompactionBlockParam {
	const { content, signature, encryptedContent } = payload;
	return {
		type: "compaction",
		content,
		...(signature !== undefined ? { signature } : {}),
		...(encryptedContent !== undefined ? { encrypted_content: encryptedContent } : {}),
	};
}

/**
 * Whether any message in `messages` replays a native compaction block: the
 * harness's user-role summary message, or the assistant message that
 * produced the block when a caller appends the response itself.
 */
function contextReplaysAnthropicCompaction(
	messages: readonly Message[],
	model: Model<"anthropic-messages">,
	format: "signed" | "legacy",
): boolean {
	return messages.some(message => {
		if (message.role !== "user" && message.role !== "developer" && message.role !== "assistant") return false;
		const payload = message.providerPayload;
		return (
			isReplayableAnthropicCompaction(payload, model) &&
			(format === "signed" ? payload.signature !== undefined : payload.encryptedContent !== undefined)
		);
	});
}

/**
 * Persisted encrypted threshold blocks still require a legacy strategy.
 * This edit is replay-only and cannot trigger a new threshold compaction.
 */
function buildAnthropicCompactionReplayEdit(model: Model<"anthropic-messages">): CompactionEdit {
	return {
		type: "compact_20260112",
		trigger: { type: "input_tokens", value: Math.max(50_000, model.contextWindow ?? 0) },
	};
}

/** Legacy threshold block replay requires its original beta and replay edit. */
function carriesLegacyCompactionEdit(params: MessageCreateParams): boolean {
	return params.context_management?.edits.some(edit => edit.type === "compact_20260112") ?? false;
}

/** Signed replay blocks and on-demand requests require the current compaction beta. */
function carriesSignedCompaction(params: MessageCreateParams): boolean {
	return (
		params.compaction !== undefined ||
		params.messages.some(
			message =>
				Array.isArray(message.content) &&
				message.content.some(block => block.type === "compaction" && block.signature !== undefined),
		)
	);
}

/**
 * Replace the top-level zero token counts with the sum over every sampling
 * iteration on an on-demand compaction response. The iteration list is the
 * documented billing total. Requests without a compaction iteration retain
 * their top-level usage.
 */
function applyCompactionIterationUsage(usage: Usage, source: AnthropicWireUsage): boolean {
	const iterations = source.iterations;
	if (!iterations?.some(iteration => iteration?.type === "compaction")) return false;
	let input = 0;
	let output = 0;
	let cacheRead = 0;
	let cacheWrite = 0;
	for (const iteration of iterations) {
		if (!iteration) continue;
		input += iteration.input_tokens ?? 0;
		output += iteration.output_tokens ?? 0;
		cacheRead += iteration.cache_read_input_tokens ?? 0;
		cacheWrite += iteration.cache_creation_input_tokens ?? 0;
	}
	usage.input = input;
	usage.output = output;
	usage.cacheRead = cacheRead;
	usage.cacheWrite = cacheWrite;
	return true;
}

/**
 * The definitive "served by fallback" signal per Anthropic's fallback
 * billing cookbook (§4): a `fallback_message` iteration in `usage.iterations`.
 * Any other iteration type is per-attempt bookkeeping for the requested model
 * (including its dated snapshot alias) and MUST NOT retag the assistant turn.
 */
function fallbackServedModelFromUsage(source: AnthropicWireUsage): string | undefined {
	const iterations = source.iterations ?? [];
	for (let index = iterations.length - 1; index >= 0; index -= 1) {
		const iteration = iterations[index];
		if (iteration?.type === "fallback_message" && iteration.model?.trim()) return iteration.model;
	}
	return undefined;
}

/**
 * Resolve a served/iteration model id to its bundled catalog entry when
 * possible so the per-iteration cost uses the served model's pricing
 * (e.g. Opus 4.8 rates for a Fable→Opus fallback). Falls back to
 * `requestModel` when the id is empty, matches the request, or the
 * catalog has no entry under it — the caller keeps the requested-model
 * pricing as the safe default and logs at the source.
 */
function resolveIterationModel(
	requestModel: Model<"anthropic-messages">,
	iterationModelId: string | null | undefined,
): Model<Api> {
	const id = iterationModelId?.trim();
	if (!id || id === requestModel.id) return requestModel;
	// Bundled catalog lookup: only Anthropic provider entries are safe to
	// reference (dated snapshots resolve to their alias entry when present).
	if (requestModel.provider === "anthropic") {
		const bundled = getBundledModel("anthropic", id);
		if (bundled?.api === "anthropic-messages") return bundled;
	}
	return requestModel;
}

/**
 * Price a turn per sampling iteration whenever the API reports one per
 * iteration — a server-side fallback or a server-side compaction. Each
 * iteration is priced on its own prompt size, so a long-context tier applies
 * only to an iteration that itself crosses the threshold, never to the
 * summed totals of two sub-threshold samplings. Fallback turns follow the
 * fallback billing cookbook §4 on top:
 *   • A pre-served attempt with zero output/cache-creation is not billed
 *     (waived classifier block); its iteration is skipped.
 *   • Mid-stream refusals bill their attempting model's input+output at
 *     that model's normal rates.
 *   • The `fallback_message` attempt's input tokens are rebilled at the
 *     served model's cache-read rate (fallback credit — 10% of base input).
 *
 * Top-level `usage.input/output/cacheRead/cacheWrite` keep their summed
 * counts; `usage.cost` reflects the per-iteration attributed total. Turns
 * without iterations use the requested model at the normal `calculateCost`
 * call.
 */
function calculateIterationTurnCost(
	requestModel: Model<"anthropic-messages">,
	usage: Usage,
	source: AnthropicWireUsage,
	timestamp: number,
): boolean {
	const iterations = source.iterations ?? [];
	if (iterations.length === 0) return false;
	const cost = { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 };
	const hasFallbackMessage = iterations.some(iter => iter.type === "fallback_message");
	let applied = false;
	for (const iteration of iterations) {
		const inputTokens = iteration.input_tokens ?? 0;
		const outputTokens = iteration.output_tokens ?? 0;
		const cacheReadTokens = iteration.cache_read_input_tokens ?? 0;
		const cacheWriteTokens = iteration.cache_creation_input_tokens ?? 0;
		const isFallback = iteration.type === "fallback_message";
		if (hasFallbackMessage && !isFallback && outputTokens === 0 && cacheWriteTokens === 0) continue;
		const iterationUsage = createEmptyUsage();
		if (isFallback) {
			iterationUsage.input = 0;
			iterationUsage.cacheRead = cacheReadTokens + inputTokens;
		} else {
			iterationUsage.input = inputTokens;
			iterationUsage.cacheRead = cacheReadTokens;
		}
		iterationUsage.output = outputTokens;
		iterationUsage.cacheWrite = cacheWriteTokens;
		iterationUsage.totalTokens =
			iterationUsage.input + iterationUsage.output + iterationUsage.cacheRead + iterationUsage.cacheWrite;
		calculateCost(resolveIterationModel(requestModel, iteration.model), iterationUsage, timestamp);
		cost.input += iterationUsage.cost.input;
		cost.output += iterationUsage.cost.output;
		cost.cacheRead += iterationUsage.cost.cacheRead;
		cost.cacheWrite += iterationUsage.cost.cacheWrite;
		cost.total += iterationUsage.cost.total;
		applied = true;
	}
	if (!applied) return false;
	usage.cost = cost;
	return true;
}

/**
 * Detects the two shapes a signature-enforcing endpoint uses to reject a
 * replayed unsigned thinking block (sent as `signature: ""`):
 *
 * - Anthropic and Anthropic-fronting proxies: `400 Invalid `signature` in
 *   `thinking` block`.
 * - Bedrock-backed proxies: the empty string fails schema validation before
 *   signature checking, so it comes back as `ValidationException: The model
 *   returned the following errors: messages.N.content.M.thinking.signature:
 *   Field required`.
 *
 * Exported for the compat tests.
 */
const INVALID_THINKING_SIGNATURE_PATTERN = /invalid\s+`?signature`?\s+in\s+`?thinking`?(?:\s+block)?/i;
const MISSING_THINKING_SIGNATURE_PATTERN = /thinking\.signature\b[^"\n]{0,32}\brequired\b/i;
const THINKING_PREFIX_BINDING_PATTERN =
	/(?:bound to a different conversation|block_binding\.prefix_mismatch_behavior|prefix_mismatch_behavior)/i;

/** Detects the preserved-thinking error caused by rewriting a signed block's conversation prefix. */
export function isThinkingPrefixBindingError(message: string): boolean {
	return (
		!/\bcompaction_[a-z_]+\b/i.test(message) &&
		INVALID_THINKING_SIGNATURE_PATTERN.test(message) &&
		THINKING_PREFIX_BINDING_PATTERN.test(message)
	);
}

export function isInvalidThinkingSignatureError(message: string): boolean {
	if (/\bcompaction_[a-z_]+\b/i.test(message)) return false;
	return INVALID_THINKING_SIGNATURE_PATTERN.test(message) || MISSING_THINKING_SIGNATURE_PATTERN.test(message);
}

const INPUT_TRANSFORMATION_PATH_PATTERN = /^messages\.(\d+)\.content\.(\d+)$/;
const PREFIX_BINDING_ERROR_PATH_PATTERN = /messages\.(\d+)\.content\.(\d+)/;

function thinkingReplayKey(block: ContentBlockParam): string | undefined {
	if (block.type === "thinking") return block.signature ? `thinking:${block.signature}` : undefined;
	if (block.type === "redacted_thinking") return block.data ? `redacted:${block.data}` : undefined;
	return undefined;
}

function rememberPrefixDroppedThinking(
	params: MessageCreateParamsStreaming,
	transformations: readonly ProviderInputTransformation[],
	state: AnthropicProviderSessionState | undefined,
): void {
	if (!state) return;
	let firstMessageIndex: number | undefined;
	let firstBlockIndex: number | undefined;
	for (const transformation of transformations) {
		if (transformation.reason !== "prefix_binding_mismatch" || typeof transformation.path !== "string") continue;
		const match = INPUT_TRANSFORMATION_PATH_PATTERN.exec(transformation.path);
		if (!match) continue;
		const messageIndex = Number(match[1]);
		const blockIndex = Number(match[2]);
		if (
			firstMessageIndex === undefined ||
			messageIndex < firstMessageIndex ||
			(messageIndex === firstMessageIndex && blockIndex < (firstBlockIndex ?? Number.POSITIVE_INFINITY))
		) {
			firstMessageIndex = messageIndex;
			firstBlockIndex = blockIndex;
		}
	}
	if (firstMessageIndex === undefined || firstBlockIndex === undefined) return;
	for (let messageIndex = firstMessageIndex; messageIndex < params.messages.length; messageIndex++) {
		const message = params.messages[messageIndex];
		if (!message || !Array.isArray(message.content)) continue;
		const blockStart = messageIndex === firstMessageIndex ? firstBlockIndex : 0;
		for (let blockIndex = blockStart; blockIndex < message.content.length; blockIndex++) {
			const key = thinkingReplayKey(message.content[blockIndex]!);
			if (key) state.prefixDroppedThinkingBlocks.add(key);
		}
	}
}

function rememberPrefixBindingFailure(
	params: MessageCreateParamsStreaming,
	message: string,
	state: AnthropicProviderSessionState | undefined,
): boolean {
	if (!state) return false;
	const match = PREFIX_BINDING_ERROR_PATH_PATTERN.exec(message);
	let path = match ? `messages.${match[1]}.content.${match[2]}` : undefined;
	if (!path) {
		for (let messageIndex = 0; messageIndex < params.messages.length && !path; messageIndex++) {
			const candidate = params.messages[messageIndex];
			if (!candidate || !Array.isArray(candidate.content)) continue;
			const blockIndex = candidate.content.findIndex(block => thinkingReplayKey(block) !== undefined);
			if (blockIndex >= 0) path = `messages.${messageIndex}.content.${blockIndex}`;
		}
	}
	if (!path) return false;
	rememberPrefixDroppedThinking(
		params,
		[{ type: "thinking_dropped", reason: "prefix_binding_mismatch", path }],
		state,
	);
	return true;
}

function applyReportedInputTransformations(
	output: AssistantMessage,
	params: MessageCreateParamsStreaming,
	state: AnthropicProviderSessionState | undefined,
	value: unknown,
	seen: Set<string>,
	replace = false,
): void {
	if (value === undefined || value === null) return;
	if (replace) {
		seen.clear();
		output.inputTransformations = [];
	}
	const fresh: ProviderInputTransformation[] = [];
	for (const transformation of parseAnthropicInputTransformations(value)) {
		const key = JSON.stringify(transformation);
		if (seen.has(key)) continue;
		seen.add(key);
		fresh.push(transformation);
	}
	if (fresh.length === 0) return;
	output.inputTransformations = [...(output.inputTransformations ?? []), ...fresh];
	rememberPrefixDroppedThinking(params, fresh, state);
	for (const transformation of fresh) {
		if (transformation.reason !== "prefix_binding_mismatch") continue;
		logger.warn("anthropic: dropped thinking block after conversation prefix changed", {
			model: output.model,
			path: transformation.path,
		});
	}
}

/**
 * Prepend a pointed remediation to a thinking-signature rejection 400 when the
 * model looks like an unmarked custom signing proxy
 * (opaque baseUrl, `spec.reasoning: true`, no explicit
 * `compat.replayUnsignedThinking` override). The default is native replay for
 * the 3p reasoning majority (#2005); this hint turns the misconfigured-proxy
 * case into a one-line fix instead of a silent retry loop (#4297).
 */
export function maybeAddReplayUnsignedThinkingHint(model: Model<"anthropic-messages">, message: string): string {
	if (!isInvalidThinkingSignatureError(message) || isThinkingPrefixBindingError(message)) return message;
	if (model.compat.officialEndpoint) return message;
	if (model.compatConfig?.replayUnsignedThinking !== undefined) return message;
	const hint = `Provider "${model.provider}" looks like an Anthropic-compatible signing proxy: it rejected a replayed unsigned thinking block. Set \`compat.replayUnsignedThinking: false\` under \`providers.${model.provider}\` in your models.yml and retry. See https://github.com/can1357/oh-my-pi/issues/4297.`;
	return `${hint}\n\n${message}`;
}

const streamAnthropicOnce = (
	model: Model<"anthropic-messages">,
	context: Context,
	options?: AnthropicOptions,
): AssistantMessageEventStream => {
	const stream = new AssistantMessageEventStream();

	(async () => {
		const startTime = performance.now();
		let firstTokenTime: number | undefined;

		const output: AssistantMessage = {
			role: "assistant",
			content: [],
			api: model.api as Api,
			provider: model.provider,
			model: model.id,
			usage: createEmptyUsage(),
			stopReason: "stop",
			timestamp: Date.now(),
		};
		let rawRequestDump: RawHttpRequestDump | undefined;
		let activeAbortTracker = createAbortSourceTracker(options?.signal);

		const onSseEvent = options?.onSseEvent;
		const rawSseObserver = onSseEvent ? (event: RawSseEvent) => onSseEvent(event, model) : undefined;

		try {
			// Built inside the try so a copilot credential/header failure surfaces as
			// an error event instead of an unhandled rejection that leaves the stream
			// (and any consumer awaiting `result()`) hanging forever.
			const apiKey = options?.apiKey ?? getEnvApiKey(model.provider) ?? "";
			const copilotApiKey = model.provider === "github-copilot" ? parseGitHubCopilotApiKey(apiKey) : undefined;
			const copilotBaseUrl =
				model.provider === "github-copilot"
					? (resolveAnthropicBaseUrl(model, apiKey) ?? "https://api.anthropic.com")
					: undefined;
			const copilotCacheKey =
				model.provider === "github-copilot" ? getCopilotIntegrationCacheKey(apiKey, copilotBaseUrl) : undefined;
			const copilotCached =
				model.provider === "github-copilot" ? getCachedCopilotIntegrationId(copilotCacheKey) : undefined;
			const copilotDynamicHeaders = copilotApiKey
				? buildCopilotDynamicHeaders({
						messages: context.messages,
						hasImages: hasCopilotVisionInput(context.messages),
						premiumMultiplier: model.premiumMultiplier,
						headers: { ...model.headers, ...options?.headers },
						integrationId: resolveCopilotRequestIdentity(options?.headers),
						initiatorOverride: options?.initiatorOverride,
						enterpriseUrl: copilotApiKey.enterpriseUrl,
						cachedIntegrationId: copilotCached,
					})
				: undefined;
			if (copilotDynamicHeaders?.premiumRequests !== undefined) {
				output.usage.premiumRequests = copilotDynamicHeaders.premiumRequests;
			}
			const baseUrl = copilotBaseUrl ?? resolveAnthropicBaseUrl(model, apiKey) ?? "https://api.anthropic.com";
			const supportsEagerToolInputStreaming = resolveEagerToolInputStreamingSupport(model, baseUrl);
			// A caller-owned client decides the endpoint itself (its `baseURL`, or an
			// explicit opt-in); it receives the compaction beta per request.
			const compactionSupported = options?.client
				? supportsAnthropicCompactionOnClient(model, options.client)
				: supportsAnthropicCompaction(model, baseUrl);
			const providerSessionState = getAnthropicProviderSessionState(
				options?.providerSessionState,
				baseUrl,
				model.id,
			);
			let disableStrictTools =
				(providerSessionState?.strictToolsDisabled ?? false) || (model.compat?.disableStrictTools ?? false);
			let dropFastMode = providerSessionState?.fastModeDisabled ?? false;
			let forceDemoteUnsignedThinking = providerSessionState?.replayUnsignedThinkingDisabled ?? false;
			let droppedAllThinkingForSignature = providerSessionState?.thinkingReplayDisabled ?? false;
			let dropAllThinking = droppedAllThinkingForSignature;
			let prefixBindingRetryAttempted = false;
			let prefixMismatchBehavior =
				model.thinking?.prefixBinding && model.compat.supportsThinkingBindingControls
					? (options?.anthropicPrefixMismatchBehavior ?? "drop_block")
					: undefined;
			const controlBetas = resolveAnthropicControlBetas(model, prefixMismatchBehavior);
			const mergedCallerHeaders = mergeHeaders(model.headers, options?.headers);
			const umansGatewayWebSearchHeader = getUmansWebSearchHeader(model, mergedCallerHeaders);
			// Keep fallback payloads aligned with the top-level Vertex effort gate:
			// no nested effort field means the fallback scan cannot re-add its beta.
			let fallbacks = options?.fallbacks;
			if (
				!model.compat.supportsOutputEffort &&
				fallbacks?.some(entry => entry.output_config?.effort !== undefined)
			) {
				fallbacks = fallbacks.map(entry => {
					const outputConfig = entry.output_config;
					if (outputConfig?.effort === undefined) return entry;
					return {
						...entry,
						output_config:
							outputConfig.task_budget === undefined ? undefined : { task_budget: outputConfig.task_budget },
					};
				});
			}

			let client: AnthropicMessagesClientLike;
			let isOAuthToken: boolean;
			// Retained so a Claude Code version bump can rebuild the client's fingerprint headers.
			let clientArgs: AnthropicClientOptionsArgs | undefined;
			let requestExtraBetas: readonly string[] = [];
			let clientDefaultHeaders: Record<string, string> | undefined;

			if (options?.client) {
				client = options.client;
				isOAuthToken = false;
			} else {
				const extraBetas = normalizeExtraBetas(options?.betas);
				// Skip the fast-mode beta when this session already learned the
				// endpoint+model rejects fast mode; `speed` is dropped from the params
				// too (dropFastMode), so the request stays a faithful non-fast request.
				if (wantsAnthropicFastMode(model, options) && !dropFastMode && !extraBetas.includes(fastModeBeta)) {
					extraBetas.push(fastModeBeta);
				}
				if (options?.taskBudget && !extraBetas.includes(taskBudgetBeta)) {
					extraBetas.push(taskBudgetBeta);
				}
				// `output_config.effort` ships on thinking-on requests, explicit
				// thinking-off adaptive pins, and forced-tool adaptive pins. The beta
				// must accompany the field even when direct streamAnthropic callers omit
				// thinkingEnabled (#6589). MiniMax uses `thinking.type:"adaptive"` itself
				// as the control surface, so the sentinel "adaptive" value intentionally
				// sends no output_config. Skip Vertex rawPredict: that adapter needs betas
				// in the body (`anthropic_beta`), not as an `anthropic-beta` HTTP header,
				// so the effort field is dropped from the body there too (see buildParams)
				// and advertising the beta would only earn a 400 (#5614).
				const sendsAdaptiveEffortPin =
					isAdaptiveOnlyThinking(model) &&
					(options?.thinkingEnabled === false ||
						(model.compat.supportsForcedToolChoice && isForcedToolChoice(options?.toolChoice)));
				// A resolved `effortBeta` (Factory Droid routes, whose Bedrock/Vertex
				// upstreams gate the beta themselves) replaces the heuristic: the beta
				// rides exactly when an output effort does.
				const emitsEffortBeta =
					model.compat.effortBeta !== undefined
						? model.compat.effortBeta && options?.effort !== undefined
						: model.reasoning &&
							model.compat.supportsOutputEffort &&
							((options?.thinkingEnabled && options.effort !== "adaptive") || sendsAdaptiveEffortPin);
				if (emitsEffortBeta && !extraBetas.includes(effortBeta)) {
					extraBetas.push(effortBeta);
				}
				if (!isVertexRawPredictUrl(baseUrl)) {
					for (const beta of controlBetas) {
						if (!extraBetas.includes(beta)) extraBetas.push(beta);
					}
				}
				// `context_management.clear_thinking_20251015` requires this beta. OAuth
				// requests carry it in `claudeCodeAgentBetaDefaults`; API-key requests
				// need it added explicitly so the field is honored instead of rejected
				// (#3288). Provider deployment contracts that cannot deliver or accept
				// context management disable it through model compatibility policy.
				if (
					model.reasoning &&
					options?.thinkingEnabled &&
					model.compat.supportsContextManagement !== false &&
					!(compactionSupported && options?.anthropicCompaction) &&
					!isVertexRawPredictUrl(baseUrl) &&
					!extraBetas.includes(contextManagementBeta)
				) {
					extraBetas.push(contextManagementBeta);
				}
				// Vertex rawPredict takes betas in the body; other routes use the
				// header. Legacy encrypted replay keeps its original beta.
				if (compactionSupported && !isVertexRawPredictUrl(baseUrl)) {
					if (
						(options?.anthropicCompaction !== undefined ||
							contextReplaysAnthropicCompaction(context.messages, model, "signed")) &&
						!extraBetas.includes(COMPACTION_BETA)
					) {
						extraBetas.push(COMPACTION_BETA);
					}
					if (
						options?.anthropicCompaction === undefined &&
						contextReplaysAnthropicCompaction(context.messages, model, "legacy") &&
						!extraBetas.includes(LEGACY_COMPACTION_BETA)
					) {
						extraBetas.push(LEGACY_COMPACTION_BETA);
					}
				}
				// `ttl: "1h"` requires the extended-cache-ttl beta on API-key
				// requests. OAuth requests never add it here: Anthropic honors
				// `ttl: "1h"` on the OAuth path without it (verified against live
				// traffic: writes land in the `ephemeral_1h` bucket), and utility
				// requests must not deviate from CC's header fingerprint.
				const isOAuth = options?.isOAuth ?? isAnthropicOAuthToken(apiKey);
				if (
					!isOAuth &&
					getCacheControl(model, options?.cacheRetention, isOAuth).cacheControl?.ttl === "1h" &&
					!extraBetas.includes(extendedCacheTtlBeta)
				) {
					extraBetas.push(extendedCacheTtlBeta);
				}
				if (!isOAuth && isOfficialAnthropicApiUrl(baseUrl) && !extraBetas.includes(fallbackCreditBeta)) {
					extraBetas.push(fallbackCreditBeta);
				}
				if (options?.fallbackCreditRedemption) {
					const frozenBetas = (options.fallbackCreditRedemption.betas ?? []).filter(
						b => !b.startsWith("server-side-fallback-"),
					);
					extraBetas.length = 0;
					for (const beta of frozenBetas) {
						extraBetas.push(beta);
					}
					if (!extraBetas.includes(fallbackCreditBeta) && !extraBetas.includes("fallback-credit-2026-06-01")) {
						extraBetas.push(fallbackCreditBeta);
					}
				}
				// Server-side fallback beta chain: opt-in via `options.fallbacks`.
				// Nested overrides (`speed`, `output_config.effort`,
				// `output_config.task_budget`) reuse the same top-level betas
				// Anthropic requires for the primary request, so scan the chain
				// and add every companion beta the fallback entries touch.
				if (fallbacks?.length) {
					if (!extraBetas.includes(serverSideFallbackBeta)) {
						extraBetas.push(serverSideFallbackBeta);
					}
					for (const entry of fallbacks) {
						if (entry.speed === "fast" && !extraBetas.includes(fastModeBeta)) {
							extraBetas.push(fastModeBeta);
						}
						if (entry.output_config?.effort && !extraBetas.includes(effortBeta)) {
							extraBetas.push(effortBeta);
						}
						if (entry.output_config?.task_budget && !extraBetas.includes(taskBudgetBeta)) {
							extraBetas.push(taskBudgetBeta);
						}
					}
				}

				clientArgs = {
					model,
					apiKey,
					extraBetas,
					stream: true,
					interleavedThinking: options?.interleavedThinking ?? true,
					headers: options?.headers,
					dynamicHeaders: copilotDynamicHeaders?.headers,
					isOAuth: options?.isOAuth,
					hasTools: !!context.tools?.length,
					thinkingEnabled: options?.thinkingEnabled,
					thinkingDisplay: options?.thinkingDisplay,
					fetch: options?.fetch,
					maxRetryDelayMs: options?.maxRetryDelayMs,
					copilotCacheKey,
					copilotCacheSnapshot: copilotCached ?? null,
					sessionId:
						options?.sessionId ??
						extractClaudeMetadataSessionId(options?.metadata?.user_id) ??
						options?.promptCacheKey,
					disableStrictTools,
				};
				const created = createClient(model, clientArgs);
				client = created.client;
				isOAuthToken = created.isOAuthToken;
				clientDefaultHeaders = created.defaultHeaders;
				requestExtraBetas = extraBetas;
			}
			const preparedContext = await prepareAnthropicManyImageContext(context, model.input.includes("image"));
			const prepareParams = async (): Promise<MessageCreateParamsStreaming> => {
				const built = buildParams(model, preparedContext, isOAuthToken, options, {
					compactionSupported,
					disableStrictTools,
					useUmansGatewayWebSearch: umansGatewayWebSearchHeader !== undefined,
					forceDemoteUnsignedThinking,
					supportsEagerToolInputStreaming,
					prefixMismatchBehavior,
					dropAllThinking,
					droppedThinkingBlocks: providerSessionState?.prefixDroppedThinkingBlocks,
					fallbacks,
					effectiveBaseUrl: baseUrl,
				});
				let nextParams = built.params;
				// The last build wins: a retry may rebuild with different controls.
				if (built.requestControls) output.requestControls = built.requestControls;
				else delete output.requestControls;
				if (disableStrictTools) {
					dropAnthropicStrictTools(nextParams);
				}
				if (dropFastMode) {
					dropAnthropicFastMode(nextParams);
				}
				const replacementPayload = await options?.onPayload?.(nextParams, model);
				if (replacementPayload !== undefined) {
					nextParams = replacementPayload as typeof nextParams;
				}
				if (nextParams.compaction) stripCompactionIncompatibleParams(nextParams);
				// After `onPayload`, so a hook cannot restore a field Bedrock rejects.
				if (model.compat.bedrockMessagesApi) fitBedrockAnthropicPayload(nextParams);
				nextParams = toWellFormedDeep(nextParams) as typeof nextParams;
				rawRequestDump = {
					provider: model.provider,
					api: output.api,
					model: model.id,
					method: "POST",
					url: `${baseUrl}/v1/messages${isOAuthToken ? "?beta=true" : ""}`,
					body: nextParams,
				};
				return nextParams;
			};
			let usingFallbackCredit = false;
			let fallbackCreditShape: "continuation" | "unchanged" | undefined = undefined;
			let fallbackCreditTransientRetries = 0;
			let params: MessageCreateParamsStreaming;
			if (
				options?.fallbackCreditRedemption &&
				Date.now() <= options.fallbackCreditRedemption.expiresAt &&
				options.fallbackCreditRedemption.params
			) {
				const redemption = options.fallbackCreditRedemption;
				usingFallbackCredit = true;
				const frozenParams = structuredClone(redemption.params as MessageCreateParamsStreaming);
				const targetModelId = options?.requestModelId ?? model.requestModelId ?? model.id;
				frozenParams.model = targetModelId;
				frozenParams.fallback_credit_token = redemption.token;
				delete frozenParams.fallbacks;
				if (
					redemption.prefillClaim !== false &&
					redemption.refusedContent &&
					redemption.refusedContent.length > 0
				) {
					fallbackCreditShape = "continuation";
					const echoed = formatEchoedRefusalContent(redemption.refusedContent);
					if (echoed.length > 0) {
						frozenParams.messages = [
							...frozenParams.messages,
							{ role: "assistant", content: echoed } as unknown as (typeof frozenParams.messages)[number],
						];
					}
				} else {
					fallbackCreditShape = "unchanged";
				}
				params = frozenParams;
				rawRequestDump = {
					provider: model.provider,
					api: output.api,
					model: model.id,
					method: "POST",
					url: `${baseUrl}/v1/messages${isOAuthToken ? "?beta=true" : ""}`,
					body: params,
				};
			} else {
				params = await prepareParams();
			}
			const seenInputTransformations = new Set<string>();
			const idleTimeoutMs = options?.streamIdleTimeoutMs ?? getStreamIdleTimeoutMs(model.compat.streamIdleTimeoutMs);
			const firstEventTimeoutMs = options?.streamFirstEventTimeoutMs ?? getStreamFirstEventTimeoutMs(idleTimeoutMs);
			const requestTimeoutMs =
				firstEventTimeoutMs !== undefined && firstEventTimeoutMs > 0 ? firstEventTimeoutMs : undefined;

			// Opt-in flag: the response parser only honors `fallback` content
			// blocks and `usage.iterations` when the current request opted into
			// server-side-fallback beta chain. Leaving `fallbacks` unset preserves
			// the pre-fallback stream shape on every event.
			const serverSideFallback = !!fallbacks?.length;
			type Block = (
				| ThinkingContent
				| RedactedThinkingContent
				| TextContent
				| AnthropicFallbackContent
				| (AnthropicServerToolContent & { [kStreamingPartialJson]?: string })
				| (ToolCall & { [kStreamingPartialJson]: string; [kStreamingLastParseLen]?: number })
			) & { [kStreamingBlockIndex]: number };
			const blocks = output.content as Block[];
			const finalizeStreamBlock = (block: Block, contentIndex: number): void => {
				if (block.type === "text") {
					stream.push({ type: "text_end", contentIndex, content: block.text, partial: output });
				} else if (block.type === "thinking") {
					const unwrappedThinking = unwrapAnthropicThinkingEnvelope(block.thinking);
					if (unwrappedThinking !== undefined) {
						block.thinking = unwrappedThinking;
						block.thinkingSignature = undefined;
					} else if (!output.upstreamModel && block.thinkingSignature) {
						// The signature names the model that actually produced the
						// block; a gateway serving a different model than requested
						// cannot mint one that says otherwise.
						output.upstreamModel = servedModelFromAnthropicSignature(block.thinkingSignature);
					}
					stream.push({ type: "thinking_end", contentIndex, content: block.thinking, partial: output });
				} else if (block.type === "anthropicServerTool" && block.block.type === "server_tool_use") {
					const partialJson = block[kStreamingPartialJson];
					if (partialJson) {
						try {
							const input = parseJsonWithRepair(partialJson);
							if (isRecord(input)) {
								block.block.input = input;
							} else {
								reportAnthropicEnvelopeAnomaly("server_tool_use input is not a JSON object");
							}
						} catch (parseError) {
							reportAnthropicEnvelopeAnomaly(
								`server_tool_use ${block.block.id} input is not valid JSON: ${parseError instanceof Error ? parseError.message : String(parseError)}`,
							);
						}
					}
					clearStreamingPartialJson(block);
				} else if (block.type === "toolCall") {
					const finalJson =
						block[kStreamingPartialJson].length > 0
							? block[kStreamingPartialJson]
							: JSON.stringify(block.arguments ?? {});
					// Keep malformed calls non-fatal, but never execute their auto-closed previews.
					block.arguments = parseToolCallArguments(finalJson);
					if (isRecord(block.arguments) && "__parseError" in block.arguments) {
						reportAnthropicEnvelopeAnomaly(
							`tool_use ${block.id} arguments are not valid JSON: ${block.arguments.__parseError}`,
						);
					}
					clearStreamingPartialJson(block);
					stream.push({ type: "toolcall_end", contentIndex, toolCall: block, partial: output });
				}
			};
			stream.push({ type: "start", partial: output });
			// Retry loop for transient errors from the stream.
			// Provider-level transport/rate-limit failures: only before any streamed content starts.
			// Malformed envelopes/JSON: only before replay-unsafe text/tool events are visible on this stream.
			let providerRetryAttempt = 0;
			const firstEventTimeoutAbortError = new AIError.StreamTimeoutError(
				"Anthropic stream timed out while waiting for the first event",
			);
			const idleTimeoutAbortError = new AIError.StreamTimeoutError(
				"Anthropic stream stalled while waiting for the next event",
			);
			const resetStreamOutputState = (): void => {
				providerRetryAttempt = 0;
				output.content.length = 0;
				output.model = model.id;
				output.responseId = undefined;
				output.upstreamModel = undefined;
				output.errorMessage = undefined;
				output.stopDetails = undefined;
				output.inputTransformations = undefined;
				output.providerPayload = undefined;
				output.usage = createEmptyUsage(copilotDynamicHeaders?.premiumRequests);
				output.stopReason = "stop";
				firstTokenTime = undefined;
			};
			// A rebuilt body no longer matches the refused request, so it cannot carry
			// the credit token. When the refused turn already ran server tools, a
			// tokenless retry would re-run and re-bill them: surface the failure instead.
			const forfeitFallbackCredit = (streamFailure: unknown): void => {
				if (
					options?.fallbackCreditRedemption?.refusedContent?.some(block => block.type === "anthropicServerTool")
				) {
					logger.warn("anthropic: fallback credit cannot be forfeited after server tools ran; surfacing error", {
						model: model.id,
					});
					throw streamFailure;
				}
				usingFallbackCredit = false;
				fallbackCreditShape = undefined;
			};
			const rebuildParams = async (streamFailure: unknown): Promise<MessageCreateParamsStreaming> => {
				if (usingFallbackCredit) forfeitFallbackCredit(streamFailure);
				return prepareParams();
			};
			// Subscription slow mode (Claude Code `/low-priority`): first-party OAuth
			// requests only. Capacity waits are tracked per request so the caller's
			// max-wait budget covers the whole wait, not one attempt.
			const slowMode =
				options?.anthropicSlowMode !== undefined &&
				model.provider === "anthropic" &&
				isOAuthToken &&
				!options.client &&
				isOfficialAnthropicApiUrl(baseUrl)
					? options.anthropicSlowMode
					: undefined;
			// Slow-lane state is per Claude account: key it by the stored credential
			// that served this request, else by a digest of the bearer.
			const slowLane =
				options?.credentialId !== undefined
					? `cred:${options.credentialId}`
					: `key:${Bun.hash(apiKey).toString(36)}`;
			let slowWaitSinceMs: number | undefined;
			let slowWaitAttempts = 0;
			let sentSlow = false;
			while (true) {
				activeAbortTracker = createAbortSourceTracker(options?.signal);
				const { requestSignal } = activeAbortTracker;
				// The provider loop owns retries: pin the client's internal retry loop
				// to zero even when no watchdog timeout is configured (the helper only
				// pins it alongside a timeout; a client retry budget of 5 would otherwise
				// multiply with PROVIDER_MAX_RETRIES into up to 66 wire attempts).
				// Injected SDK clients bypass client-level beta construction. Attach
				// every beta required by fields this request actually carries. Vertex
				// rawPredict is excluded because its betas live in `anthropic_beta`.
				let injectedClientBetaHeaders: Record<string, string> | undefined;
				// A caller-owned client targets its own endpoint: route betas by
				// the client's URL when it exposes one, not the model's routing.
				const injectedBetaRouteUrl =
					options?.client !== undefined ? (injectedClientBaseUrl(options.client) ?? baseUrl) : baseUrl;
				if (options?.client !== undefined && !isVertexRawPredictUrl(injectedBetaRouteUrl)) {
					for (const beta of controlBetas) {
						injectedClientBetaHeaders = mergeAnthropicBetaHeader(
							injectedClientBetaHeaders ?? mergedCallerHeaders,
							beta,
						);
					}
					if ((params.output_config as AnthropicOutputConfig | undefined)?.effort !== undefined) {
						injectedClientBetaHeaders = mergeAnthropicBetaHeader(
							injectedClientBetaHeaders ?? mergedCallerHeaders,
							effortBeta,
						);
					}
					if (carriesSignedCompaction(params)) {
						injectedClientBetaHeaders = mergeAnthropicBetaHeader(
							injectedClientBetaHeaders ?? mergedCallerHeaders,
							COMPACTION_BETA,
						);
					}
					if (carriesLegacyCompactionEdit(params)) {
						injectedClientBetaHeaders = mergeAnthropicBetaHeader(
							injectedClientBetaHeaders ?? mergedCallerHeaders,
							LEGACY_COMPACTION_BETA,
						);
					}
				}
				sentSlow = slowMode?.isActive(slowLane) === true;
				let perRequestHeaders: Record<string, string> | undefined =
					umansGatewayWebSearchHeader || injectedClientBetaHeaders || options?.userProfileId || sentSlow
						? {
								...umansGatewayWebSearchHeader,
								...injectedClientBetaHeaders,
								...(options?.userProfileId ? { "anthropic-user-profile-id": options.userProfileId } : {}),
								...(sentSlow ? { [ANTHROPIC_USAGE_LIMIT_HEADER]: ANTHROPIC_SLOW_USAGE_LIMIT } : {}),
							}
						: undefined;
				if (usingFallbackCredit && options?.fallbackCreditRedemption?.betaHeader) {
					const rawBetas = options.fallbackCreditRedemption.betaHeader
						.split(",")
						.map(b => b.trim())
						.filter(b => b && !b.startsWith("server-side-fallback-"));
					if (!rawBetas.includes(fallbackCreditBeta) && !rawBetas.includes("fallback-credit-2026-06-01")) {
						rawBetas.push(fallbackCreditBeta);
					}
					perRequestHeaders = {
						...perRequestHeaders,
						"anthropic-beta": rawBetas.join(","),
					};
				}
				const requestOptions = {
					...createSdkStreamRequestOptions(requestSignal, requestTimeoutMs),
					maxRetries: 0,
					...(perRequestHeaders ? { headers: perRequestHeaders } : {}),
				};
				const anthropicRequest: unknown =
					isOAuthToken && client.beta
						? client.beta.messages.create({ ...params, stream: true }, requestOptions)
						: client.messages.create({ ...params, stream: true }, requestOptions);
				let streamedReplayUnsafeContent = false;

				try {
					let requestTimeout: NodeJS.Timeout | undefined;
					if (requestTimeoutMs !== undefined) {
						requestTimeout = setTimeout(
							() => activeAbortTracker.abortLocally(firstEventTimeoutAbortError),
							requestTimeoutMs,
						);
					}
					let anthropicStream: AsyncIterable<AnthropicStreamEvent>;
					let response: Response;
					let requestId: string | null;
					let recordsRawSseEvents: boolean;
					try {
						({
							events: anthropicStream,
							response,
							requestId,
							recordsRawSseEvents,
						} = await getAnthropicStreamResponse(anthropicRequest, requestSignal, rawSseObserver));
					} catch (error) {
						if (error instanceof AnthropicConnectionTimeoutError && !activeAbortTracker.wasCallerAbort()) {
							throw firstEventTimeoutAbortError;
						}
						throw error;
					} finally {
						if (requestTimeout !== undefined) clearTimeout(requestTimeout);
					}
					await notifyProviderResponse(options, response, model, requestId);
					if (slowMode) {
						const slowSignal = parseAnthropicSlowModeHeaders(response.headers);
						if (slowSignal) slowMode.observe(slowSignal, slowLane);
					}
					let sawEvent = false;
					let sawMessageStart = false;
					let sawTerminalEnvelope = false;
					let sawMessageStop = false;
					// Set when a duplicate message_start splices a second envelope onto
					// the stream; closed indexes then refuse to reopen so replayed
					// content cannot duplicate (see content_block_start guard).
					let sawSplicedEnvelope = false;
					const closedBlockIndexes = new Set<number>();
					const openBlocks = new Map<
						number,
						{
							contentIndex: number;
							kind:
								| "text"
								| "thinking"
								| "redactedThinking"
								| "fallback"
								| "anthropicServerTool"
								| "toolCall"
								| "compaction"
								| "ignored";
						}
					>();
					// A compaction block is complete at content_block_start, not streamed
					// through deltas. Hold it until the stop reason confirms compaction.
					let compactionPayload: AnthropicCompactionPayload | undefined;

					// Pings keep the idle deadline alive once content is flowing (Anthropic
					// bridges legitimate generation gaps with keepalives), but only within a
					// bounded window: a wedged upstream that pings forever while the model
					// produces nothing must still trip the idle watchdog, otherwise an
					// active tool-call stream hangs unrecoverably with no retry (#4900).
					// A ping before message_start must not consume the first-event watchdog
					// either: it would flip the (retryable) pre-content stall classification
					// into a terminal mid-stream idle timeout.
					let sawNonPingEvent = false;
					let lastNonPingProgressAtMs = 0;
					const pingProgressCapMs =
						idleTimeoutMs !== undefined && idleTimeoutMs > 0
							? idleTimeoutMs * PING_PROGRESS_MAX_IDLE_MULTIPLIER
							: undefined;
					const timedAnthropicStream = iterateWithIdleTimeout(anthropicStream, {
						idleTimeoutMs,
						firstItemTimeoutMs: firstEventTimeoutMs,
						errorMessage: idleTimeoutAbortError.message,
						firstItemErrorMessage: firstEventTimeoutAbortError.message,
						onIdle: () => activeAbortTracker.abortLocally(idleTimeoutAbortError),
						onFirstItemTimeout: () => activeAbortTracker.abortLocally(firstEventTimeoutAbortError),
						abortSignal: options?.signal,
						isProgressItem: item => {
							if ((item as AnthropicStreamEvent).type === "ping") {
								if (!sawNonPingEvent) return false;
								if (pingProgressCapMs === undefined) return true;
								return Date.now() - lastNonPingProgressAtMs < pingProgressCapMs;
							}
							sawNonPingEvent = true;
							lastNonPingProgressAtMs = Date.now();
							return true;
						},
					});
					const observedAnthropicStream =
						rawSseObserver && !recordsRawSseEvents
							? observeDecodedAnthropicSdkEvents(timedAnthropicStream, rawSseObserver)
							: timedAnthropicStream;
					for await (const event of observedAnthropicStream) {
						sawEvent = true;

						if (event.type === "message_start") {
							if (sawMessageStart) {
								// Transparent reconnects can splice a fresh envelope onto the same
								// stream; keep the original message but surface the anomaly. Events
								// for blocks still open from the first envelope continue to apply,
								// but replayed blocks are dropped below (see closedBlockIndexes).
								reportAnthropicEnvelopeAnomaly("duplicate message_start event");
								sawSplicedEnvelope = true;
								continue;
							}
							sawMessageStart = true;
							const startMessage = event.message;
							if (startMessage?.id) output.responseId = startMessage.id;
							applyReportedInputTransformations(
								output,
								params,
								providerSessionState,
								startMessage?.input_transformations,
								seenInputTransformations,
							);
							const startUsage = startMessage?.usage;
							if (startUsage) {
								applyAnthropicUsageExtras(output.usage, startUsage);
								output.usage.input = startUsage.input_tokens || 0;
								output.usage.output = startUsage.output_tokens || 0;
								output.usage.cacheRead = startUsage.cache_read_input_tokens || 0;
								output.usage.cacheWrite = startUsage.cache_creation_input_tokens || 0;
								const compacted = applyCompactionIterationUsage(output.usage, startUsage);
								output.usage.totalTokens =
									output.usage.input + output.usage.output + output.usage.cacheRead + output.usage.cacheWrite;
								if (serverSideFallback) {
									const served = fallbackServedModelFromUsage(startUsage);
									if (served) output.model = served;
								}
								if (
									!(serverSideFallback || compacted) ||
									!calculateIterationTurnCost(model, output.usage, startUsage, output.timestamp)
								) {
									calculateCost(model, output.usage, output.timestamp);
								}
							} else {
								reportAnthropicEnvelopeAnomaly("message_start missing usage");
							}
							continue;
						}

						if (!sawMessageStart) {
							if (shouldIgnoreAnthropicPreambleEvent(event.type)) {
								continue;
							}
							throw new AIError.AnthropicStreamEnvelopeError(`received ${event.type} before message_start`);
						}

						if (event.type === "content_block_start") {
							if (sawTerminalEnvelope) {
								reportAnthropicEnvelopeAnomaly(`received ${event.type} after terminal stop signal`);
								continue;
							}
							if (openBlocks.has(event.index)) {
								reportAnthropicEnvelopeAnomaly(`duplicate content_block_start index ${event.index}`);
								continue;
							}
							if (sawSplicedEnvelope && closedBlockIndexes.has(event.index)) {
								// A spliced envelope replaying an index this stream already
								// completed would append duplicate text/tool calls; consume its
								// events silently instead.
								reportAnthropicEnvelopeAnomaly(
									`replayed content_block_start index ${event.index} after duplicate message_start`,
								);
								openBlocks.set(event.index, { contentIndex: -1, kind: "ignored" });
								continue;
							}
							if (!event.content_block?.type) {
								reportAnthropicEnvelopeAnomaly("content_block_start missing content_block payload");
								continue;
							}
							if (!firstTokenTime) firstTokenTime = performance.now();
							if (event.content_block.type === "fallback") {
								// Fallback boundary is only meaningful when the request
								// opted into the beta chain — silently drop otherwise so
								// unopted-in sessions never see the block persisted or
								// influence downstream converters.
								const fallback = parseAnthropicFallbackWireBlock(event.content_block);
								if (!serverSideFallback || !fallback) {
									if (!fallback) {
										reportAnthropicEnvelopeAnomaly("fallback content_block missing model refs");
									}
									openBlocks.set(event.index, { contentIndex: -1, kind: "ignored" });
									continue;
								}
								const block: Block = { ...fallback, [kStreamingBlockIndex]: event.index };
								output.content.push(block);
								openBlocks.set(event.index, {
									contentIndex: output.content.length - 1,
									kind: "fallback",
								});
								// A fallback content block is the mid-stream signal that a
								// classifier block on the primary was retried on the
								// fallback model. Adopt the served id immediately so
								// pricing decisions downstream (final usage.iterations may
								// arrive before/after) see the right model.
								output.model = fallback.to.model;
								continue;
							}
							if (event.content_block.type === "text") {
								streamedReplayUnsafeContent = true;
								const block: Block = {
									type: "text",
									text: "",
									[kStreamingBlockIndex]: event.index,
								};
								output.content.push(block);
								const contentIndex = output.content.length - 1;
								openBlocks.set(event.index, { contentIndex, kind: "text" });
								stream.push({
									type: "text_start",
									contentIndex,
									partial: output,
								});
							} else if (event.content_block.type === "thinking") {
								streamedReplayUnsafeContent = true;
								const block: Block = {
									type: "thinking",
									thinking: event.content_block.thinking ?? "",
									thinkingSignature: "",
									[kStreamingBlockIndex]: event.index,
								};
								output.content.push(block);
								const contentIndex = output.content.length - 1;
								openBlocks.set(event.index, { contentIndex, kind: "thinking" });
								stream.push({
									type: "thinking_start",
									contentIndex,
									partial: output,
								});
								if (block.thinking) {
									stream.push({
										type: "thinking_delta",
										contentIndex,
										delta: block.thinking,
										partial: output,
									});
								}
							} else if (event.content_block.type === "redacted_thinking") {
								streamedReplayUnsafeContent = true;
								const block: Block = {
									type: "redactedThinking",
									data: event.content_block.data,
									[kStreamingBlockIndex]: event.index,
								};
								output.content.push(block);
								openBlocks.set(event.index, {
									contentIndex: output.content.length - 1,
									kind: "redactedThinking",
								});
							} else if (
								isAnthropicServerToolHistoryBlock(event.content_block) &&
								(umansGatewayWebSearchHeader === undefined ||
									(event.content_block.type === "server_tool_use"
										? event.content_block.name !== "web_search"
										: event.content_block.type !== "web_search_tool_result"))
							) {
								streamedReplayUnsafeContent = true;
								const block: Block = {
									type: "anthropicServerTool",
									block: { ...event.content_block },
									[kStreamingPartialJson]: "",
									[kStreamingBlockIndex]: event.index,
								};
								output.content.push(block);
								openBlocks.set(event.index, {
									contentIndex: output.content.length - 1,
									kind: "anthropicServerTool",
								});
							} else if (event.content_block.type === "tool_use") {
								streamedReplayUnsafeContent = true;
								const block: Block = {
									type: "toolCall",
									id: event.content_block.id,
									name: decodeAnthropicToolName(
										event.content_block.name,
										isOAuthToken,
										model.compat.escapeBuiltinToolNames,
									),
									arguments: event.content_block.input ?? {},
									[kStreamingPartialJson]: "",
									[kStreamingBlockIndex]: event.index,
								};
								output.content.push(block);
								const contentIndex = output.content.length - 1;
								openBlocks.set(event.index, { contentIndex, kind: "toolCall" });
								stream.push({
									type: "toolcall_start",
									contentIndex,
									partial: output,
								});
							} else if (event.content_block.type === "compaction") {
								const { content, signature, encrypted_content } = event.content_block;
								if (content) {
									compactionPayload = {
										type: "anthropicCompaction",
										provider: model.provider,
										content,
										...(signature !== undefined ? { signature } : {}),
										...(encrypted_content ? { encryptedContent: encrypted_content } : {}),
									};
								}
								openBlocks.set(event.index, { contentIndex: -1, kind: "compaction" });
							} else {
								openBlocks.set(event.index, { contentIndex: -1, kind: "ignored" });
							}
						} else if (event.type === "content_block_delta") {
							if (sawTerminalEnvelope) {
								reportAnthropicEnvelopeAnomaly(`received ${event.type} after terminal stop signal`);
								continue;
							}
							const openBlock = openBlocks.get(event.index);
							if (!openBlock) {
								reportAnthropicEnvelopeAnomaly(
									`received content_block_delta for unopened index ${event.index}`,
								);
								continue;
							}
							if (openBlock.kind === "ignored") continue;
							if (!event.delta?.type) {
								reportAnthropicEnvelopeAnomaly("content_block_delta missing delta payload");
								continue;
							}
							const block = blocks[openBlock.contentIndex];
							if (event.delta.type === "text_delta") {
								if (openBlock.kind !== "text" || block?.type !== "text") {
									reportAnthropicEnvelopeAnomaly(`received text_delta for ${openBlock.kind} block`);
									continue;
								}
								streamedReplayUnsafeContent = true;
								block.text += event.delta.text;
								stream.push({
									type: "text_delta",
									contentIndex: openBlock.contentIndex,
									delta: event.delta.text,
									partial: output,
								});
							} else if (event.delta.type === "thinking_delta") {
								if (openBlock.kind !== "thinking" || block?.type !== "thinking") {
									reportAnthropicEnvelopeAnomaly(`received thinking_delta for ${openBlock.kind} block`);
									continue;
								}
								streamedReplayUnsafeContent = true;
								block.thinking += event.delta.thinking;
								stream.push({
									type: "thinking_delta",
									contentIndex: openBlock.contentIndex,
									delta: event.delta.thinking,
									partial: output,
								});
							} else if (event.delta.type === "input_json_delta") {
								if (
									openBlock.kind === "anthropicServerTool" &&
									block?.type === "anthropicServerTool" &&
									block.block.type === "server_tool_use"
								) {
									block[kStreamingPartialJson] =
										(block[kStreamingPartialJson] ?? "") + event.delta.partial_json;
									continue;
								}
								if (openBlock.kind !== "toolCall" || block?.type !== "toolCall") {
									reportAnthropicEnvelopeAnomaly(`received input_json_delta for ${openBlock.kind} block`);
									continue;
								}
								streamedReplayUnsafeContent = true;
								block[kStreamingPartialJson] += event.delta.partial_json;
								const throttled = parseStreamingJsonThrottled(
									block[kStreamingPartialJson],
									block[kStreamingLastParseLen] ?? 0,
								);
								if (throttled) {
									block.arguments = throttled.value;
									block[kStreamingLastParseLen] = throttled.parsedLen;
								}
								stream.push({
									type: "toolcall_delta",
									contentIndex: openBlock.contentIndex,
									delta: event.delta.partial_json,
									partial: output,
								});
							} else if (event.delta.type === "signature_delta") {
								if (openBlock.kind !== "thinking" || block?.type !== "thinking") {
									reportAnthropicEnvelopeAnomaly(`received signature_delta for ${openBlock.kind} block`);
									continue;
								}
								streamedReplayUnsafeContent = true;
								block.thinkingSignature = block.thinkingSignature || "";
								block.thinkingSignature += event.delta.signature;
							}
						} else if (event.type === "content_block_stop") {
							if (sawTerminalEnvelope) {
								reportAnthropicEnvelopeAnomaly(`received ${event.type} after terminal stop signal`);
								continue;
							}
							const openBlock = openBlocks.get(event.index);
							if (!openBlock) {
								reportAnthropicEnvelopeAnomaly(`received content_block_stop for unopened index ${event.index}`);
								continue;
							}
							if (openBlock.kind === "ignored") {
								openBlocks.delete(event.index);
								continue;
							}
							if (openBlock.kind === "compaction") {
								openBlocks.delete(event.index);
								closedBlockIndexes.add(event.index);
								continue;
							}
							const block = blocks[openBlock.contentIndex];
							if (!block || block.type !== openBlock.kind) {
								reportAnthropicEnvelopeAnomaly(`content_block_stop kind mismatch for index ${event.index}`);
								openBlocks.delete(event.index);
								continue;
							}
							openBlocks.delete(event.index);
							closedBlockIndexes.add(event.index);
							finalizeStreamBlock(block, openBlock.contentIndex);
						} else if (event.type === "message_delta") {
							if (sawTerminalEnvelope) {
								// A spliced reconnect's second envelope must not overwrite the
								// completed message's stop reason or usage.
								reportAnthropicEnvelopeAnomaly("received message_delta after terminal stop signal");
								continue;
							}
							const delta = event.delta;
							applyReportedInputTransformations(
								output,
								params,
								providerSessionState,
								event.input_transformations,
								seenInputTransformations,
								true,
							);
							const rawStopReason = delta?.stop_reason;
							if (rawStopReason) {
								output.stopReason = mapStopReason(rawStopReason);
								sawTerminalEnvelope = true;
								// On-demand compaction ends with no ordinary assistant content.
								// A non-compaction stop (e.g. max_tokens) is not a usable summary.
								if (rawStopReason === "compaction") {
									output.stopDetails = { type: "compaction" };
									output.providerPayload = compactionPayload;
								}
							}
							if (output.stopReason === "error") {
								const stopDetails = delta?.stop_details;
								output.stopDetails = stopDetails ?? (rawStopReason ? { type: rawStopReason } : null);
								if (stopDetails?.type === "refusal") {
									const explanation = stopDetails.explanation?.trim();
									const category = stopDetails.category;
									const label = category ? `Refusal (${category})` : "Refusal";
									output.errorMessage = explanation ? `${label}: ${explanation}` : label;
								}
								if (stopDetails?.fallback_credit_token) {
									const sentBetaHeader =
										getHeaderCaseInsensitive(perRequestHeaders ?? {}, "anthropic-beta") ??
										getHeaderCaseInsensitive(clientDefaultHeaders ?? {}, "anthropic-beta") ??
										requestExtraBetas.join(",");
									output.fallbackCreditHandle = {
										token: stopDetails.fallback_credit_token,
										prefillClaim: stopDetails.fallback_has_prefill_claim,
										params: structuredClone(params),
										betas: Array.from(requestExtraBetas),
										betaHeader: sentBetaHeader,
										expiresAt: Date.now() + 5 * 60 * 1000,
										refusedContent: output.content ? structuredClone(output.content) : undefined,
									};
								}
								if (!output.errorMessage) {
									// Anthropic flagged an error-class stop (refusal / sensitive) without
									// populating stop_details. Surface the raw reason instead of falling
									// through to the generic "unknown error" string when we throw below.
									output.errorMessage =
										rawStopReason === "refusal"
											? "Refusal (no details provided)"
											: rawStopReason === "sensitive"
												? "Content flagged by safety filters"
												: `Anthropic stream ended with stop_reason: ${rawStopReason ?? "unknown"}`;
								}
							}
							const deltaUsage = event.usage;
							if (deltaUsage) {
								if (deltaUsage.input_tokens != null) {
									output.usage.input = deltaUsage.input_tokens;
								}
								if (deltaUsage.output_tokens != null) {
									output.usage.output = deltaUsage.output_tokens;
								}
								if (deltaUsage.cache_read_input_tokens != null) {
									output.usage.cacheRead = deltaUsage.cache_read_input_tokens;
								}
								if (deltaUsage.cache_creation_input_tokens != null) {
									output.usage.cacheWrite = deltaUsage.cache_creation_input_tokens;
								}
								applyAnthropicUsageExtras(output.usage, deltaUsage);
								const compacted = applyCompactionIterationUsage(output.usage, deltaUsage);
								output.usage.totalTokens =
									output.usage.input + output.usage.output + output.usage.cacheRead + output.usage.cacheWrite;
								if (serverSideFallback) {
									const served = fallbackServedModelFromUsage(deltaUsage);
									if (served) output.model = served;
								}
								if (
									!(serverSideFallback || compacted) ||
									!calculateIterationTurnCost(model, output.usage, deltaUsage, output.timestamp)
								) {
									calculateCost(model, output.usage, output.timestamp);
								}
							}
						} else if (event.type === "message_stop") {
							sawTerminalEnvelope = true;
							sawMessageStop = true;
							// The protocol is complete even if a broken keep-alive leaves the HTTP body open.
							break;
						}
					}

					const firstEventTimeoutError = activeAbortTracker.getLocalAbortReason();
					if (firstEventTimeoutError) {
						throw firstEventTimeoutError;
					}
					if (activeAbortTracker.wasCallerAbort()) {
						throw new AIError.AbortError();
					}
					if (!sawEvent || !sawMessageStart) {
						throw new AIError.AnthropicStreamEnvelopeError("stream ended before message_start");
					}
					if (!sawTerminalEnvelope) {
						// Neither a message_delta stop_reason nor message_stop arrived: the
						// connection died mid-generation. Finalizing the partial message as
						// a clean "stop" would make the agent loop treat the truncated turn
						// as complete (silent mid-sentence halt), so fail the turn. The
						// envelope error is transparently retried before replay-unsafe
						// content streams; afterwards it surfaces as an error turn whose
						// complete tool calls the agent loop salvages
						// (`recoverTransientErrorToolTurn` recognizes the envelope-error
						// text and `retainCompletedToolCalls` drops half-streamed calls).
						throw new AIError.AnthropicStreamEnvelopeError("stream ended before message_stop");
					}
					if (!sawMessageStop) {
						// A stop_reason arrived via message_delta, so generation finished;
						// only the trailing message_stop frame is missing (non-conforming
						// gateway). Degrade to best-effort instead of discarding the turn.
						reportAnthropicEnvelopeAnomaly("stream ended before message_stop");
					}
					if (openBlocks.size > 0) {
						for (const [openIndex, openBlock] of openBlocks) {
							reportAnthropicEnvelopeAnomaly(
								`stream ended with an unterminated ${openBlock.kind} block at index ${openIndex}`,
							);
							if (openBlock.kind === "ignored" || openBlock.contentIndex < 0) continue;
							const danglingBlock = blocks[openBlock.contentIndex];
							if (danglingBlock) finalizeStreamBlock(danglingBlock, openBlock.contentIndex);
						}
						openBlocks.clear();
					}

					if (output.stopReason === "aborted" || output.stopReason === "error") {
						throw new AIError.ProviderResponseError(output.errorMessage ?? "An unknown error occurred", {
							provider: model.provider,
							kind: "output",
						});
					}
					break;
				} catch (streamError) {
					const streamFailure = activeAbortTracker.getLocalAbortReason() ?? streamError;
					if (
						!disableStrictTools &&
						firstTokenTime === undefined &&
						hasStrictAnthropicTools(params) &&
						AIError.isGrammarError(streamFailure)
					) {
						// Log-only: the retried turn must not carry an errorMessage on
						// success (consumers treat its presence as failure).
						logger.warn("anthropic: strict tools rejected, retrying without strict tools", {
							model: model.id,
							error: await finalizeErrorMessage(streamFailure, rawRequestDump),
						});
						if (providerSessionState) {
							providerSessionState.strictToolsDisabled = true;
						}
						disableStrictTools = true;
						params = await rebuildParams(streamFailure);
						resetStreamOutputState();
						continue;
					}
					if (usingFallbackCredit && firstTokenTime === undefined && isAnthropicBadRequest(streamFailure)) {
						const errMessage = streamFailure instanceof Error ? streamFailure.message : String(streamFailure);
						const redemption = options!.fallbackCreditRedemption!;
						if (errMessage.includes("redemption temporarily unavailable")) {
							if (Date.now() < redemption.expiresAt && fallbackCreditTransientRetries < 2) {
								fallbackCreditTransientRetries++;
								logger.warn(
									"anthropic: fallback credit redemption temporarily unavailable, retrying same shape",
									{
										model: model.id,
										attempt: fallbackCreditTransientRetries,
									},
								);
								if (options?.providerRetryWait) {
									await options.providerRetryWait(500, options.signal);
								} else {
									await scheduler.wait(500, { signal: options?.signal });
								}
								resetStreamOutputState();
								continue;
							}
							throw streamFailure;
						}
						if (fallbackCreditShape === "continuation") {
							logger.warn(
								"anthropic: fallback credit continuation shape rejected, retrying with unchanged body",
								{
									model: model.id,
									error: errMessage,
								},
							);
							fallbackCreditShape = "unchanged";
							const frozenParams = structuredClone(redemption.params as MessageCreateParamsStreaming);
							const targetModelId = options?.requestModelId ?? model.requestModelId ?? model.id;
							frozenParams.model = targetModelId;
							frozenParams.fallback_credit_token = redemption.token;
							delete frozenParams.fallbacks;
							params = frozenParams;
							resetStreamOutputState();
							continue;
						}
						if (errMessage.includes("fallback_credit_token")) {
							forfeitFallbackCredit(streamFailure);
							logger.warn(
								"anthropic: fallback credit token rejected, falling back to standard request without token",
								{
									model: model.id,
									error: errMessage,
								},
							);
							dropAllThinking = true;
							params = await prepareParams();
							resetStreamOutputState();
							continue;
						}
					}
					const streamFailureMessage =
						streamFailure instanceof Error ? streamFailure.message : String(streamFailure);
					if (
						isOAuthToken &&
						clientArgs &&
						firstTokenTime === undefined &&
						adoptRequiredClaudeCodeVersion(streamFailure)
					) {
						logger.warn("anthropic: Claude Code version rejected as too old, retrying with required version", {
							model: model.id,
							version: getClaudeCodeVersion(),
						});
						client = createClient(model, { ...clientArgs, disableStrictTools }).client;
						// The version only changes client headers; a redemption keeps its frozen body.
						if (!usingFallbackCredit) params = await prepareParams();
						providerRetryAttempt = 0;
						output.content.length = 0;
						output.model = model.id;
						output.responseId = undefined;
						output.upstreamModel = undefined;
						output.errorMessage = undefined;
						output.providerPayload = undefined;
						output.usage = createEmptyUsage(copilotDynamicHeaders?.premiumRequests);
						output.stopReason = "stop";
						firstTokenTime = undefined;
						continue;
					}
					if (
						!prefixBindingRetryAttempted &&
						options?.anthropicPrefixMismatchBehavior !== "error" &&
						firstTokenTime === undefined &&
						!streamedReplayUnsafeContent &&
						isThinkingPrefixBindingError(streamFailureMessage)
					) {
						logger.warn("anthropic: thinking prefix changed, stripping bound thinking and retrying", {
							provider: model.provider,
							model: model.id,
							baseUrl,
						});
						prefixBindingRetryAttempted = true;
						prefixMismatchBehavior = undefined;
						dropAllThinking = !rememberPrefixBindingFailure(params, streamFailureMessage, providerSessionState);
						params = await rebuildParams(streamFailure);
						providerRetryAttempt = 0;
						output.content.length = 0;
						output.model = model.id;
						output.responseId = undefined;
						output.upstreamModel = undefined;
						output.errorMessage = undefined;
						output.inputTransformations = undefined;
						output.providerPayload = undefined;
						output.usage = createEmptyUsage(copilotDynamicHeaders?.premiumRequests);
						output.stopReason = "stop";
						firstTokenTime = undefined;
						continue;
					}
					if (
						!forceDemoteUnsignedThinking &&
						firstTokenTime === undefined &&
						!streamedReplayUnsafeContent &&
						!isThinkingPrefixBindingError(streamFailureMessage) &&
						isInvalidThinkingSignatureError(streamFailureMessage)
					) {
						logger.warn(
							"anthropic: signing proxy detected (thinking signature rejected), demoting unsigned thinking and retrying",
							{
								provider: model.provider,
								model: model.id,
								baseUrl,
								error: streamFailureMessage,
							},
						);
						if (providerSessionState) {
							providerSessionState.replayUnsignedThinkingDisabled = true;
						}
						forceDemoteUnsignedThinking = true;
						params = await rebuildParams(streamFailure);
						providerRetryAttempt = 0;
						output.content.length = 0;
						output.model = model.id;
						output.responseId = undefined;
						output.upstreamModel = undefined;
						output.errorMessage = undefined;
						output.providerPayload = undefined;
						output.usage = createEmptyUsage(copilotDynamicHeaders?.premiumRequests);
						output.stopReason = "stop";
						firstTokenTime = undefined;
						continue;
					}
					if (
						!dropAllThinking &&
						firstTokenTime === undefined &&
						!streamedReplayUnsafeContent &&
						!isThinkingPrefixBindingError(streamFailureMessage) &&
						isInvalidThinkingSignatureError(streamFailureMessage)
					) {
						// The unsigned-demotion retry only rewrites UNSIGNED blocks;
						// when every replayed block carries a signature the signer no
						// longer accepts (e.g. a failover proxy swapped upstreams
						// mid-conversation and minted foreign signatures), the retry
						// resends a byte-identical body and the session 400s forever.
						// Escalate: drop all replayed thinking — prior-turn reasoning
						// is optional context — and retry once. Stored history keeps
						// its thinking blocks; only the wire payload changes.
						logger.warn(
							"anthropic: thinking signatures still rejected after unsigned demotion, dropping replayed thinking and retrying",
							{
								provider: model.provider,
								model: model.id,
								baseUrl,
								error: streamFailureMessage,
							},
						);
						if (providerSessionState) {
							providerSessionState.thinkingReplayDisabled = true;
						}
						droppedAllThinkingForSignature = true;
						dropAllThinking = true;
						params = await rebuildParams(streamFailure);
						providerRetryAttempt = 0;
						output.content.length = 0;
						output.model = model.id;
						output.responseId = undefined;
						output.errorMessage = undefined;
						output.inputTransformations = undefined;
						output.providerPayload = undefined;
						output.usage = createEmptyUsage(copilotDynamicHeaders?.premiumRequests);
						output.stopReason = "stop";
						firstTokenTime = undefined;
						continue;
					}
					if (
						!dropFastMode &&
						wantsAnthropicFastMode(model, options) &&
						firstTokenTime === undefined &&
						AIError.isFastModeUnsupported(streamFailure)
					) {
						logger.debug("anthropic: fast mode unsupported, retrying without speed", {
							model: model.id,
							error: streamFailure instanceof Error ? streamFailure.message : String(streamFailure),
						});
						if (providerSessionState) {
							providerSessionState.fastModeDisabled = true;
						}
						dropFastMode = true;
						params = await rebuildParams(streamFailure);
						providerRetryAttempt = 0;
						output.content.length = 0;
						output.model = model.id;
						output.responseId = undefined;
						output.upstreamModel = undefined;
						output.errorMessage = undefined;
						output.providerPayload = undefined;
						output.usage = createEmptyUsage(copilotDynamicHeaders?.premiumRequests);
						output.stopReason = "stop";
						firstTokenTime = undefined;
						continue;
					}
					if (
						slowMode &&
						firstTokenTime === undefined &&
						!streamedReplayUnsafeContent &&
						!activeAbortTracker.wasCallerAbort()
					) {
						const failureStatus = (streamFailure as { status?: unknown } | null)?.status;
						const httpStatus = typeof failureStatus === "number" ? failureStatus : undefined;
						const failureText = streamFailure instanceof Error ? streamFailure.message : String(streamFailure);
						const slowRetry = await slowMode.onFailure({
							lane: slowLane,
							httpStatus,
							overloaded: httpStatus === 529 || failureText.includes("overloaded_error"),
							signal: parseAnthropicSlowModeHeaders(getHeadersFromError(streamFailure)),
							sentSlow,
							waitedMs: slowWaitSinceMs === undefined ? 0 : Date.now() - slowWaitSinceMs,
							attempts: slowWaitAttempts,
						});
						if (slowRetry) {
							if (slowRetry.capacityWait) {
								slowWaitSinceMs ??= Date.now();
								slowWaitAttempts++;
							}
							logger.debug("anthropic: slow mode retry", {
								model: model.id,
								status: httpStatus,
								delayMs: slowRetry.delayMs,
								attempt: slowWaitAttempts,
							});
							if (slowRetry.delayMs > 0) {
								if (options?.providerRetryWait) {
									await options.providerRetryWait(slowRetry.delayMs, options.signal);
								} else {
									await scheduler.wait(slowRetry.delayMs, { signal: options?.signal });
								}
							}
							resetStreamOutputState();
							continue;
						}
					}
					const isTransientEnvelopeFailure =
						AIError.isTransientStreamParseError(streamFailure) || AIError.isStreamEnvelopeError(streamFailure);
					const isLocalIdleTimeout =
						streamFailure === idleTimeoutAbortError ||
						(streamFailure instanceof Error && streamFailure.message === idleTimeoutAbortError.message);
					const canRetryTransientEnvelopeFailure = isTransientEnvelopeFailure && !streamedReplayUnsafeContent;
					const canRetryProviderFailure =
						!isLocalIdleTimeout &&
						firstTokenTime === undefined &&
						!streamedReplayUnsafeContent &&
						AIError.isProviderRetryableError(streamFailure);
					if (
						activeAbortTracker.wasCallerAbort() ||
						providerRetryAttempt >= PROVIDER_MAX_RETRIES ||
						(!canRetryTransientEnvelopeFailure && !canRetryProviderFailure)
					) {
						throw streamFailure;
					}
					providerRetryAttempt++;
					const backoffDelayMs = calculateAnthropicRetryDelayMs(providerRetryAttempt - 1);
					// Honor the server's retry hint (`retry-after-ms`/`retry-after`) on
					// 429/529-style failures: retrying sooner than the server asked is a
					// guaranteed failure that just burns the retry budget.
					const headerDelayMs = getRetryAfterMsFromHeaders(getHeadersFromError(streamFailure));
					// Bound the server-directed wait so a multi-hour `retry-after` cannot
					// park the provider stream before higher-level recovery runs. A non-positive cap
					// disables the bound; an over-cap hint surfaces the original error immediately.
					const maxRetryDelayMs = options?.maxRetryDelayMs ?? 60_000;
					if (headerDelayMs !== undefined && maxRetryDelayMs > 0 && headerDelayMs > maxRetryDelayMs) {
						throw streamFailure;
					}
					const delayMs = headerDelayMs !== undefined ? Math.max(headerDelayMs, backoffDelayMs) : backoffDelayMs;
					if (options?.providerRetryWait) {
						await options.providerRetryWait(delayMs, options.signal);
					} else {
						await scheduler.wait(delayMs, { signal: options?.signal });
					}
					output.content.length = 0;
					output.model = model.id;
					output.responseId = undefined;
					output.errorMessage = undefined;
					output.stopDetails = undefined;
					output.providerPayload = undefined;
					output.usage = createEmptyUsage(copilotDynamicHeaders?.premiumRequests);
					output.stopReason = "stop";
					firstTokenTime = undefined;
				}
			}
			if (
				usingFallbackCredit &&
				fallbackCreditShape === "continuation" &&
				options?.fallbackCreditRedemption?.refusedContent
			) {
				// The response continues the echoed prefix; keep it in the stored turn in
				// AssistantMessage form so later replays keep signatures and server tools.
				output.content.unshift(...refusalContinuationPrefix(options.fallbackCreditRedemption.refusedContent));
			}
			output.duration = performance.now() - startTime;
			if (firstTokenTime) output.ttft = firstTokenTime - startTime;
			if (dropFastMode && model.provider === "anthropic" && options?.serviceTier === "priority") {
				output.disabledFeatures = [...(output.disabledFeatures ?? []), "priority"];
			}
			if (forceDemoteUnsignedThinking && model.compat.replayUnsignedThinking) {
				output.disabledFeatures = [...(output.disabledFeatures ?? []), "unsigned-thinking-replay"];
			}
			if (droppedAllThinkingForSignature) {
				output.disabledFeatures = [...(output.disabledFeatures ?? []), "thinking-replay"];
			}
			stream.push({ type: "done", reason: output.stopReason, message: output });
			stream.end();
		} catch (error) {
			for (const block of output.content) {
				if (block.type === "toolCall") clearStreamingPartialJson(block);
			}
			const result = await AIError.finalize(error, {
				api: model.api,
				provider: model.provider,
				abortTracker: activeAbortTracker,
				rawRequestDump,
			});
			output.stopReason = result.stopReason;
			output.errorStatus = result.status;
			output.errorId = result.id;
			output.errorMessage = maybeAddReplayUnsignedThinkingHint(model, result.message);
			output.duration = performance.now() - startTime;
			if (firstTokenTime) output.ttft = firstTokenTime - startTime;
			stream.push({ type: "error", reason: output.stopReason, error: output });
			stream.end();
		}
	})();

	return stream;
};

/**
 * Public entry: retry benign empty completions before they reach the agent
 * loop. The inner attempt owns Anthropic provider-failure retries.
 */
export const streamAnthropic: StreamFunction<"anthropic-messages"> = (model, context, options) =>
	withReplaySafeStreamRetry(model, context, options, streamAnthropicOnce, {
		retryEmptyCompletion: true,
	});

export type AnthropicSystemBlock = {
	type: "text";
	text: string;
	cache_control?: AnthropicCacheControl;
};
type SystemBlockOptions = {
	includeClaudeCodeInstruction?: boolean;
	extraInstructions?: string[];
	/** Text of the first user message — used as fingerprint seed for the billing header. */
	firstUserMessageText?: string;
	/** Cache lifetime shared by the OAuth system breakpoint and later message breakpoints. */
	cacheControl?: AnthropicCacheControl;
};

export function buildAnthropicSystemBlocks(
	systemPrompt: readonly string[] | undefined,
	options: SystemBlockOptions = {},
): AnthropicSystemBlock[] | undefined {
	const { includeClaudeCodeInstruction = false, extraInstructions = [], firstUserMessageText, cacheControl } = options;
	const sanitizedPrompts = normalizeSystemPrompts(systemPrompt);
	const trimmedInstructions = extraInstructions.map(instruction => instruction.trim()).filter(Boolean);
	const hasBillingHeader = sanitizedPrompts.some(prompt => prompt.startsWith(CLAUDE_BILLING_HEADER_PREFIX));

	if (includeClaudeCodeInstruction && !hasBillingHeader) {
		const blocks: AnthropicSystemBlock[] = [
			{ type: "text", text: createClaudeBillingHeader(firstUserMessageText ?? "") },
			{
				type: "text",
				text: claudeCodeSystemInstruction,
				cache_control: cacheControl ? cloneAnthropicCacheControl(cacheControl) : { type: "ephemeral" },
			},
		];

		for (const instruction of trimmedInstructions) {
			blocks.push({ type: "text", text: instruction });
		}
		for (const prompt of sanitizedPrompts) {
			blocks.push({ type: "text", text: prompt });
		}

		return blocks;
	}

	const blocks: AnthropicSystemBlock[] = [];
	for (const instruction of trimmedInstructions) {
		blocks.push({ type: "text", text: instruction });
	}
	for (const prompt of sanitizedPrompts) {
		blocks.push({ type: "text", text: prompt });
	}
	return blocks.length > 0 ? blocks : undefined;
}

export function normalizeExtraBetas(betas?: string[] | string): string[] {
	if (!betas) return [];
	const raw = Array.isArray(betas) ? betas : betas.split(",");
	return raw.map(beta => beta.trim()).filter(beta => beta.length > 0);
}

export function buildAnthropicClientOptions(args: AnthropicClientOptionsArgs): AnthropicClientOptionsResult {
	const {
		model,
		apiKey,
		extraBetas = [],
		stream = true,
		interleavedThinking = true,
		headers,
		dynamicHeaders,
		hasTools = false,
		thinkingEnabled = false,
		isOAuth,
		maxRetryDelayMs,
		sessionId,
		disableStrictTools: disableStrictToolsOverride,
		copilotCacheKey,
		copilotCacheSnapshot,
	} = args;
	const compat = model.compat;
	const disableStrictTools = disableStrictToolsOverride ?? compat.disableStrictTools;
	const baseUrl = resolveAnthropicBaseUrl(model, apiKey);
	// Adaptive models (`supportsDisplay`) get native interleaved thinking on the
	// official API, so only non-official signing routes need the beta (#6717).
	// Two classifications feed the predicate: the effective URL, because Foundry
	// and provider overrides can reroute a model without rebuilding its
	// materialized compat, and non-official `compat.signingEndpoint`, because
	// provider ids (e.g. ZenMux on a mirror URL) and explicit spec overrides on
	// opaque proxies are authoritative even when the URL isn't recognized.
	// Stale-official compat never qualifies: a canonical model rerouted to an
	// unrecognized proxy keeps `officialEndpoint: true` (see
	// resolveEagerToolInputStreamingSupport), and signing there is unknowable.
	// Two signing routes still can't take the beta as this `anthropic-beta` HTTP
	// header, so they're excluded: Vertex rawPredict accepts betas only in the
	// JSON body (`anthropic_beta`) and 400s on the header (#5614), and GitHub
	// Copilot rejects Anthropic betas outright — the `github-copilot` provider
	// branch below strips them, but a custom provider id or a canonical model
	// rerouted to `api.githubcopilot.com` / `copilot-api.*` reaches the generic
	// header builder instead, so exclude those effective URLs here too.
	const needsInterleavedBeta =
		interleavedThinking &&
		(!model.thinking?.supportsDisplay ||
			(!isOfficialAnthropicApiUrl(baseUrl) &&
				(isAnthropicSigningProxyUrl(baseUrl) || (compat.signingEndpoint && !compat.officialEndpoint)) &&
				!isVertexRawPredictUrl(baseUrl ?? "") &&
				!hostMatchesUrl(baseUrl, "githubCopilot")));
	const oauthToken = isOAuth ?? isAnthropicOAuthToken(apiKey);
	const supportsEagerToolInputStreaming = resolveEagerToolInputStreamingSupport(model, baseUrl);
	const needsFineGrainedToolStreamingBeta =
		hasTools && isOfficialAnthropicApiUrl(baseUrl) && !supportsEagerToolInputStreaming;
	const foundryCustomHeaders = resolveAnthropicCustomHeaders(model, baseUrl);
	const tlsFetchOptions = buildClaudeCodeTlsFetchOptions(model, baseUrl);
	// Disable Bun's native ~300s pre-response fetch timeout (issue #2422).
	// `AnthropicMessagesClient` already arms its own DEFAULT_TIMEOUT_MS timer
	// per request, so the native ceiling can only short-circuit slow-prefill
	// streams before the configured watchdog gets to govern them.
	const fetchOptions: AnthropicFetchOptions = { ...tlsFetchOptions, timeout: false };
	const baseFetch = args.fetch ?? fetch;
	// Only OAuth requests inject the CC billing header; no API-key request can ever
	// contain it, so there is no need to install the rewriter for those.
	const cchFetch = oauthToken ? wrapFetchForCch(baseFetch) : baseFetch;
	if (model.provider === "github-copilot") {
		const copilotApiKey = parseGitHubCopilotApiKey(apiKey).accessToken;
		// The GitHub Copilot Anthropic proxy doesn't accept Anthropic beta
		// features. Forward only caller-supplied betas.
		const betaFeatures = [...extraBetas];
		const defaultHeaders = mergeHeaders(
			{
				Accept: stream ? "text/event-stream" : "application/json",
				"Content-Type": "application/json",
				"anthropic-version": "2023-06-01",
				"Anthropic-Dangerous-Direct-Browser-Access": "true",
				Authorization: `Bearer ${copilotApiKey}`,
				...(betaFeatures.length > 0 ? { "anthropic-beta": buildBetaHeader([], betaFeatures) } : {}),
			},
			model.headers,
			dynamicHeaders,
			headers,
		);
		applyInferenceHeaders(defaultHeaders, {
			provider: model.provider,
			protocol: "anthropic",
			sessionId,
		});

		return {
			isOAuthToken: false,
			apiKey: null,
			authToken: copilotApiKey,
			baseURL: baseUrl,
			maxRetries: 5,
			maxRetryDelayMs,
			defaultHeaders,
			fetch: wrapFetchForCopilotFallback(
				cchFetch,
				true,
				resolveCopilotRequestIdentity(headers),
				copilotCacheKey ?? getCopilotIntegrationCacheKey(apiKey, baseUrl),
				copilotCacheSnapshot,
			),
			fetchOptions,
		};
	}

	const betaFeatures = [...extraBetas];
	if (needsFineGrainedToolStreamingBeta) {
		betaFeatures.push(fineGrainedToolStreamingBeta);
	}
	if (needsInterleavedBeta) {
		betaFeatures.push(interleavedThinkingBeta);
	}

	const requestModelHeaders = mergeHeaders(
		model.headers,
		foundryCustomHeaders,
		getUmansWebSearchHeader(model, mergeHeaders(model.headers, headers)),
		headers,
		dynamicHeaders,
	);
	const defaultHeaders = buildAnthropicHeaders({
		apiKey,
		baseUrl,
		isOAuth: oauthToken,
		extraBetas: betaFeatures,
		stream,
		modelHeaders: requestModelHeaders,
		isCloudflareAiGateway: model.provider === "cloudflare-ai-gateway",
		allowAnthropicHeaderOverrides: model.compat.allowAnthropicHeaderOverrides,
		claudeCodeSessionId: sessionId,
		claudeCodeBetas: oauthToken
			? buildClaudeCodeBetas({
					agentRequest: hasTools || thinkingEnabled,
					thinkingRequest: thinkingEnabled,
					disableStrictTools,
					supportsContextManagement: model.compat.supportsContextManagement,
				})
			: [],
	});
	applyInferenceHeaders(defaultHeaders, {
		provider: model.provider,
		protocol: "anthropic",
		sessionId,
	});

	if (model.provider === "cloudflare-ai-gateway") {
		return {
			isOAuthToken: false,
			apiKey: null,
			authToken: null,
			baseURL: baseUrl,
			maxRetries: 5,
			maxRetryDelayMs,
			defaultHeaders,
			fetch: cchFetch,
			fetchOptions,
		};
	}

	// OpenCode Go/Zen and Umans validate Anthropic-compatible API-key auth
	// through `X-Api-Key`; bearer-only requests reach the endpoint but fail auth
	// with `401 Missing API key` (#6510). Drop the auto-built `Authorization`
	// header and keep `apiKey` so the client emits `X-Api-Key`.
	if (model.provider === "opencode-go" || model.provider === "opencode-zen" || model.provider === "umans") {
		delete defaultHeaders.Authorization;
		return {
			isOAuthToken: false,
			apiKey,
			authToken: null,
			baseURL: baseUrl,
			maxRetries: 5,
			maxRetryDelayMs,
			defaultHeaders,
			fetch: cchFetch,
			fetchOptions,
		};
	}

	// Suppress the client-level `X-Api-Key` whenever an `Authorization` header
	// already sits in `defaultHeaders` for a non-official, non-OAuth endpoint —
	// either our auto-built `Bearer <apiKey>` or a caller-supplied custom auth
	// scheme via `model.headers` (#3391). Adding a bonus `X-Api-Key` would force
	// the proxy to deal with two competing credentials when the user explicitly
	// asked for one.
	const authorizationHeader = getHeaderCaseInsensitive(defaultHeaders, "Authorization");
	// A keyless provider resolves to the `N/A` sentinel, for which no
	// Authorization was built above; the client would otherwise inject a
	// bogus `X-Api-Key: N/A` of its own.
	const shouldSuppressClientApiKey =
		!oauthToken &&
		!model.compat.officialEndpoint &&
		(typeof authorizationHeader === "string" || apiKey === NO_AUTH_SENTINEL);

	return {
		isOAuthToken: oauthToken,
		apiKey: oauthToken || shouldSuppressClientApiKey ? null : apiKey,
		authToken: oauthToken ? apiKey : undefined,
		baseURL: baseUrl,
		maxRetries: 5,
		maxRetryDelayMs,
		defaultHeaders,
		fetch: cchFetch,
		fetchOptions,
	};
}

function createClient(
	model: Model<"anthropic-messages">,
	args: AnthropicClientOptionsArgs,
): { client: AnthropicMessagesClient; isOAuthToken: boolean; defaultHeaders?: Record<string, string> } {
	const { isOAuthToken: oauthToken, ...clientOptions } = buildAnthropicClientOptions({ ...args, model });
	const client = new AnthropicMessagesClient(clientOptions);
	return { client, isOAuthToken: oauthToken, defaultHeaders: clientOptions.defaultHeaders };
}

/** The compaction request is a standalone summary call, not a generation turn. */
function stripCompactionIncompatibleParams(params: MessageCreateParamsStreaming): void {
	delete params.context_management;
	delete params.stop_sequences;
	if (params.tool_choice?.type === "any" || params.tool_choice?.type === "tool") {
		params.tool_choice = { type: "auto" };
	}
	if (params.output_config) {
		delete params.output_config.format;
		if (params.output_config.task_budget?.remaining !== undefined) {
			params.output_config.task_budget = { ...params.output_config.task_budget };
			delete params.output_config.task_budget.remaining;
		}
		if (Object.keys(params.output_config).length === 0) delete params.output_config;
	}
}

function disableThinkingIfToolChoiceForced(
	params: MessageCreateParamsStreaming,
	model: Model<"anthropic-messages">,
): void {
	const toolChoice = params.tool_choice;
	if (!toolChoice) return;
	if (toolChoice.type !== "any" && toolChoice.type !== "tool") return;

	delete params.thinking;
	// Preserve the replay-only legacy edit when disabling thinking. Signed
	// compaction blocks need no edit and keep their normal context management.
	const compactionEdits = params.context_management?.edits.filter(edit => edit.type === "compact_20260112") ?? [];
	if (compactionEdits.length > 0) {
		params.context_management = { edits: compactionEdits };
	} else {
		delete params.context_management;
	}

	// Adaptive-only models can't be switched off by omitting `thinking` — a bare
	// omission defaults to adaptive thinking ON, so a forced-tool turn would still
	// reason instead of calling the tool (#6589). Pin the lowest adaptive effort
	// instead of dropping it, mirroring the disable branch in buildParams. Vertex
	// rawPredict is the sole exception: it can only carry the effort beta in the
	// body (dropped there too, see buildParams), so it keeps the delete behavior.
	// The effort beta itself is attached at the request site — including per-request
	// for injected SDK clients that bypass client-level beta construction.
	if (isAdaptiveOnlyThinking(model) && model.compat.supportsOutputEffort) {
		const outputConfig = (params.output_config as AnthropicOutputConfig | undefined) ?? {};
		outputConfig.effort = "low";
		params.output_config = outputConfig;
		return;
	}

	const outputConfig = params.output_config as AnthropicOutputConfig | undefined;
	if (!outputConfig) return;

	delete outputConfig.effort;
	if (Object.keys(outputConfig).length === 0) {
		delete params.output_config;
	}
}

function ensureMaxTokensForThinking(params: MessageCreateParamsStreaming, maxAllowedTokens: number): void {
	const thinking = params.thinking;
	if (thinking?.type !== "enabled") return;

	const budgetTokens = thinking.budget_tokens ?? 0;
	if (budgetTokens <= 0) return;

	const output = budgetThinkingOutput(params.max_tokens, budgetTokens, maxAllowedTokens);
	params.max_tokens = output.maxTokens;

	if (output.budgetTokens === budgetTokens) return;

	if (output.budgetTokens <= 0) {
		throw new AIError.ConfigurationError(
			`Anthropic thinking budget requires max_tokens greater than ${OUTPUT_FALLBACK_BUFFER}; got ${output.maxTokens}`,
		);
	}
	thinking.budget_tokens = output.budgetTokens;
}

function applyCacheControlToLastBlock(blocks: ContentBlockParam[], cacheControl: AnthropicCacheControl): boolean {
	for (let index = blocks.length - 1; index >= 0; index--) {
		const block = blocks[index];
		// Anthropic rejects cache_control on generated reasoning, fallback
		// boundaries, and mid-conversation tool-control blocks. Preserve the
		// requested trailing boundary on ordinary content, including tool use
		// and tool results.
		if (
			block.type === "thinking" ||
			block.type === "redacted_thinking" ||
			block.type === "fallback" ||
			block.type === "tool_addition" ||
			block.type === "tool_removal"
		) {
			continue;
		}
		if ("cache_control" in block && block.cache_control != null) return false;
		blocks[index] = { ...block, cache_control: cloneAnthropicCacheControl(cacheControl) };
		return true;
	}
	return false;
}

const ANTHROPIC_MAX_BREAKPOINTS = 4;
const ANTHROPIC_DECIMATION_INTERVAL = 15;

function countHeadBreakpoints(params: MessageCreateParamsStreaming): number {
	let count = 0;
	if (Array.isArray(params.system)) {
		for (const block of params.system) {
			if (typeof block !== "string" && block?.cache_control != null) count++;
		}
	}
	if (Array.isArray(params.tools)) {
		for (const tool of params.tools) {
			if (tool?.cache_control != null) count++;
		}
	}
	return count;
}

function applyCacheControlToMessage(message: MessageParam, cacheControl: AnthropicCacheControl): boolean {
	if (typeof message.content === "string") {
		message.content = [
			{ type: "text", text: message.content, cache_control: cloneAnthropicCacheControl(cacheControl) },
		];
		return true;
	} else if (Array.isArray(message.content)) {
		return applyCacheControlToLastBlock(message.content, cacheControl);
	}
	return false;
}

function applyPromptCaching(params: MessageCreateParamsStreaming, cacheControl?: AnthropicCacheControl): void {
	if (!cacheControl) return;

	const headBreakpoints = countHeadBreakpoints(params);
	const messageBudget = Math.max(0, ANTHROPIC_MAX_BREAKPOINTS - headBreakpoints);
	if (messageBudget <= 0 || params.messages.length === 0) return;
	// `convertAnthropicMessages` appends a neutral `Continue.` pad after a trailing
	// assistant because Anthropic rejects assistant-prefill endings. It is absent
	// from the next normal turn, so anchor the rolling window on the preceding
	// real assistant instead.
	const trailingIndex = params.messages.length - 1;
	const trailingMessage = params.messages[trailingIndex];
	const hasTrailingAssistantPad =
		trailingMessage?.role === "user" &&
		trailingMessage.content === "Continue." &&
		!isConversationalUser(trailingMessage) &&
		params.messages[trailingIndex - 1]?.role === "assistant";
	const messageEnd = hasTrailingAssistantPad ? trailingIndex - 1 : trailingIndex;

	// A breakpoint caches every preceding byte, not only the decorated message.
	// A per-call or turn-scoped message is rebuilt next request, so a prefix
	// spanning it cannot match — but only at its own position. Messages after
	// the mark are ordinary persisted history with stable bytes, so a later
	// breakpoint still matches everything after the mark. The cost is bounded
	// to re-billing the marked bytes themselves, not the growing tail.
	// Hence two anchors, not a truncation: the newest candidate at or before
	// the first per-call/turn-scoped message (when one exists) pins the
	// reusable prefix behind the mark, and the rolling tail candidates pin
	// the suffix after it. Turn-scoped `clear_at` messages are absent next
	// request, so they still truncate the decimation range (ordinals would
	// shift), but per-call marks no longer freeze the tail.
	let stableMessageEnd = messageEnd;
	for (let index = 0; index <= messageEnd; index++) {
		const message = params.messages[index];
		if (message && (message.clear_at === "next_user_message" || isPerCallContextMessage(message))) {
			stableMessageEnd = index - 1;
			break;
		}
	}

	// Decimation counts conversational turns, so it reads the provenance marker
	// `convertAnthropicMessages` records rather than the wire role. A wire `user`
	// can also be a serialized `developer` message, a tool_result run, or an
	// interior `Continue.` pad, none of which advance the user turn ordinal.
	const userIndices: number[] = [];
	for (let index = 0; index <= stableMessageEnd; index++) {
		const message = params.messages[index];
		if (message && isConversationalUser(message)) {
			userIndices.push(index);
		}
	}

	// Stable historical decimation checkpoint every 15 user turns (15th, 30th, 45th...)
	const decimationIndices = userIndices.filter((_, ordinal) => (ordinal + 1) % ANTHROPIC_DECIMATION_INTERVAL === 0);

	// Collect up to 2 trailing candidates from the message tail, skipping
	// per-call messages, turn-scoped messages, and mid-conversation
	// tool-control messages. A per-call tail candidate is rebuilt next request
	// (fresh timestamps on appended probes, fresh redaction bytes), so a
	// breakpoint on it cannot match — it would spend the tail anchor on bytes
	// that never repeat while the persisted history behind it goes uncached.
	// Turn-scoped messages are absent next request for the same reason, and
	// tool controls reject cache_control outright. The walk starts at the
	// message tail (not the truncated prefix end) so the anchor advances every
	// turn; the sub-prefix candidate below covers the reusable region behind
	// a mark.
	const trailingCandidates: number[] = [];
	for (let index = messageEnd; index >= 0 && trailingCandidates.length < 2; index--) {
		const message = params.messages[index];
		if (!message || message.clear_at === "next_user_message" || isPerCallContextMessage(message)) continue;
		if (
			message.role === "system" &&
			typeof message.content !== "string" &&
			Array.isArray(message.content) &&
			message.content.length > 0 &&
			message.content.every(block => block.type === "tool_addition" || block.type === "tool_removal")
		) {
			continue;
		}
		trailingCandidates.push(index);
	}
	// Prioritize:
	// 1. Most recent trailing message
	// 2. Latest decimation checkpoints (newest first) to maintain stable long-context anchors
	// 3. Newest message at or before the first per-call/turn-scoped mark, so a
	//    volatile interior message costs only its own re-billed bytes instead
	//    of invalidating the whole reusable prefix behind it
	// 4. Second trailing message
	const candidateIndices: number[] = [];
	if (trailingCandidates.length > 0) {
		candidateIndices.push(trailingCandidates[0]);
	}
	for (let i = decimationIndices.length - 1; i >= 0; i--) {
		if (!candidateIndices.includes(decimationIndices[i])) {
			candidateIndices.push(decimationIndices[i]);
		}
	}
	if (stableMessageEnd < messageEnd && stableMessageEnd >= 0 && !candidateIndices.includes(stableMessageEnd)) {
		candidateIndices.push(stableMessageEnd);
	}
	for (const index of trailingCandidates) {
		if (!candidateIndices.includes(index)) {
			candidateIndices.push(index);
		}
	}

	// Count only successful block decorations toward the message budget so an uncacheable
	// block (such as a thinking-only assistant) does not silently consume a breakpoint.
	let appliedCount = 0;
	for (const index of candidateIndices) {
		if (appliedCount >= messageBudget) break;
		const message = params.messages[index];
		if (message && applyCacheControlToMessage(message, cacheControl)) {
			appliedCount++;
		}
	}
}

/**
 * System-prompt segments whose bytes differ between sessions or turns of the
 * same agent: per-turn memory recall (`<memories>`) and the coding agent's
 * working-directory context (`<project-context>`: context files with their
 * paths, workspace tree, workspace roots, session append text; advisors use
 * the same tag for their context-file block). The coding agent renders them
 * as their own `systemPrompt` array elements after the large static prompt.
 * The system cache breakpoint anchors on the block right before the first
 * such segment, so the static head is shared byte-for-byte across sessions in
 * different directories (e.g. one git worktree per task) and a recall refresh
 * re-bills only the suffix and the message tail.
 *
 * Everything from the first volatile segment on sits after the head
 * breakpoint, including stable blocks appended behind it (per-spawn subagent
 * role text, `before_agent_start` extension policy): prefix caching is
 * positional, so bytes after a changing segment can never extend the cached
 * head anyway; the rolling message breakpoints still cover them.
 *
 * Detection is by our own markup, not model identity: only a leading tag
 * counts, so stable segments quoting these tags elsewhere are unaffected.
 */
const VOLATILE_SYSTEM_SEGMENT_MARKERS = ["<memories>", "<project-context>"];

function volatileSystemSuffixStart(systemBlocks: readonly AnthropicSystemBlock[]): number {
	const start = systemBlocks.findIndex(block =>
		VOLATILE_SYSTEM_SEGMENT_MARKERS.some(marker => block.text.startsWith(marker)),
	);
	return start === -1 ? systemBlocks.length : start;
}

/**
 * Anchor cache_control on the stable request head — the last (non-deferred)
 * tool definition and the last stable system block. The canonical cache order is
 * tools → system → messages, so a breakpoint on the final stable system block caches
 * the entire tools+stable-system prefix, and the extra tool breakpoint keeps the tool
 * definitions cached even when the system text changes. This guarantees the
 * large, unchanging head is a cache hit on every turn regardless of how the
 * message tail churns — the breakpoint placement first-party Anthropic clients
 * (Claude Code, Pi) use. Without it, the general API-key path anchors only the
 * moving message tail, so tail churn re-writes the whole head uncached.
 *
 * Volatile segments (memory recall, working-directory context) sit after the
 * breakpoint, so a recall refresh or a different cwd re-bills only the suffix
 * and the tail instead of the whole head. When every system block is volatile
 * there is no stable boundary and the breakpoint stays on the array tail.
 *
 * Anthropic allows at most 4 cache breakpoints per request. At most one is
 * spent on tools and one on system here, leaving the remaining budget for
 * the message tail and historical decimation checkpoints in `applyPromptCaching`.
 *
 * The tool array is anchored on both API-key and OAuth paths: the tool definitions
 * sit first in wire order and survive message rewrites, and sibling subagents of
 * the same definition share this prefix byte for byte.
 *
 * The OAuth Claude Code path pre-decorates its identity system block in
 * buildAnthropicSystemBlocks. That breakpoint moves to the anchor block instead
 * of a second one being added: the identity block is a prefix of the anchored
 * head, so the move keeps the budget at last tool + last stable system block +
 * 2 message breakpoints while the agent's static system prompt, not just the
 * identity line, becomes a cached prefix of its own.
 *
 * Runs on the fresh system blocks and wire tools built for this request, after
 * the declared tool list was derived from the transcript's request controls.
 */
function applyHeadCaching(
	systemBlocks: AnthropicSystemBlock[] | undefined,
	tools: AnthropicWireTool[] | undefined,
	cacheControl?: AnthropicCacheControl,
): void {
	if (!cacheControl) return;

	if (tools && tools.length > 0 && !tools.some(tool => tool.cache_control != null)) {
		// Deferred tools are not part of the checked prefix until referenced, so
		// anchor the last tool that actually sits in the stable prefix.
		for (let index = tools.length - 1; index >= 0; index--) {
			const tool = tools[index];
			if (!tool || tool.defer_loading) continue;
			tool.cache_control = cloneAnthropicCacheControl(cacheControl);
			break;
		}
	}

	if (systemBlocks && systemBlocks.length > 0) {
		// Anchor on the last block before the volatile suffix so a recall
		// refresh or a different working directory re-bills only the suffix,
		// not the whole head. An earlier system breakpoint (the OAuth identity
		// block) moves to the anchor rather than staying as a second one: a
		// breakpoint left on the identity block caches only tools + identity,
		// and keeping both would take a rolling message breakpoint from
		// `applyPromptCaching` (4 minus head breakpoints). All-volatile falls
		// back to tail anchoring.
		const suffixStart = volatileSystemSuffixStart(systemBlocks);
		const anchorIndex = suffixStart === 0 ? systemBlocks.length - 1 : suffixStart - 1;
		const anchor = systemBlocks[anchorIndex];
		if (anchor && anchor.cache_control == null) {
			for (const block of systemBlocks) delete block.cache_control;
			anchor.cache_control = cloneAnthropicCacheControl(cacheControl);
		}
	}
}

function usesAdaptiveThinkingTagOnly(model: Model<"anthropic-messages">): boolean {
	const thinking = model.thinking;
	if (thinking?.mode !== "anthropic-adaptive") return false;
	const effortMap = thinking.effortMap;
	if (!effortMap) return false;
	for (const effort of thinking.efforts) {
		if (effortMap[effort] !== "adaptive") return false;
	}
	return thinking.efforts.length > 0;
}

/**
 * True when enabled thinking on `model` is budget thinking
 * (`thinking.type: "enabled"` with `budget_tokens`) rather than adaptive.
 */
export function usesBudgetThinking(model: Model<"anthropic-messages">): boolean {
	return model.thinking?.mode !== "anthropic-adaptive" || model.compat.disableAdaptiveThinking === true;
}

/** The most output tokens a request to `model` may ask for (`max_tokens` ceiling). */
export function anthropicOutputLimit(model: Model<"anthropic-messages">): number {
	return model.maxTokens ?? UNKNOWN_MODEL_MAX_OUTPUT_TOKENS;
}

/**
 * The `max_tokens` and thinking budget of budget thinking: `max_tokens`
 * rises to leave {@link OUTPUT_FALLBACK_BUFFER} visible output tokens after
 * the budget, within `maxAllowedTokens`, and the budget shrinks when that
 * ceiling leaves less (a non-positive budget means the ceiling is too low).
 */
export function budgetThinkingOutput(
	maxTokens: number | undefined,
	budgetTokens: number,
	maxAllowedTokens: number,
): { maxTokens: number; budgetTokens: number } {
	const currentMaxTokens = Math.min(maxTokens ?? maxAllowedTokens, maxAllowedTokens);
	const raisedMaxTokens = Math.min(
		Math.max(currentMaxTokens, budgetTokens + OUTPUT_FALLBACK_BUFFER),
		maxAllowedTokens,
	);
	return {
		maxTokens: raisedMaxTokens,
		budgetTokens: Math.min(budgetTokens, raisedMaxTokens - OUTPUT_FALLBACK_BUFFER),
	};
}

/**
 * True for adaptive-only Claude models (Opus 4.6+, Sonnet 4.6+, Fable/Mythos 5)
 * that reject `thinking.type: "disabled"`. Turning thinking off on these models
 * means omitting the `thinking` field entirely and pinning the lowest adaptive
 * effort — a bare omission defaults to adaptive thinking ON. Excludes MiniMax,
 * which drives adaptive thinking through the `thinking.type: "adaptive"` tag
 * itself rather than `output_config.effort`.
 */
function isAdaptiveOnlyThinking(model: Model<"anthropic-messages">): boolean {
	return (
		model.thinking?.mode === "anthropic-adaptive" &&
		!model.compat.disableAdaptiveThinking &&
		!usesAdaptiveThinkingTagOnly(model)
	);
}

function resolveAnthropicAdaptiveEffort(
	model: Model<"anthropic-messages">,
	options: AnthropicOptions,
): AnthropicEffort | undefined {
	if (options.effort) return usesAdaptiveThinkingTagOnly(model) ? "adaptive" : options.effort;
	const requestedEffort = options.reasoning;
	if (!requestedEffort) return undefined;
	return mapEffortToAnthropicAdaptiveEffort(model, requestedEffort);
}

function extractClaudeCodeFirstUserMessageText(messages: readonly Message[]): string {
	for (const message of messages) {
		if (message.role !== "user") continue;
		const { content } = message;
		if (typeof content === "string") return content;
		if (!Array.isArray(content)) return "";
		for (const block of content) {
			if (block.type === "text") return block.text;
		}
		return "";
	}
	return "";
}

/**
 * Marks a developer message synthesized from {@link AnthropicRequestControls}
 * records. {@link convertAnthropicMessages} renders it as a bare
 * mid-conversation `role: "system"` control. Object spread copies symbol keys,
 * so the marker survives `transformMessages`.
 */
const ANTHROPIC_CONTROL = Symbol("anthropicControl");

/** `max_tokens` requested when the catalog has no output ceiling for the model. */
const UNKNOWN_MODEL_MAX_OUTPUT_TOKENS = 64_000;

/** One control message: tool changes (removals first) and an optional per-message effort. */
type AnthropicControlSpec = { toolChanges: AnthropicToolChange[]; effort?: AnthropicOutputEffort };

type AnthropicControlCarrier = object & { [ANTHROPIC_CONTROL]?: AnthropicControlSpec };

/**
 * An assistant message in `context.messages` whose request declared controls.
 * `live` when it still sits at the index it was written at: the history before
 * it is the one its request sent.
 */
type AnthropicControlRecord = { index: number; live: boolean; controls: AnthropicRequestControls };

/** A control to splice in before `context.messages[index]`. */
type AnthropicControlInsert = { index: number; spec: AnthropicControlSpec };

type AnthropicToolControls = NonNullable<AnthropicRequestControls["tools"]>;
type AnthropicEffortControls = NonNullable<AnthropicRequestControls["effort"]>;

function anthropicControlOf(message: Message): AnthropicControlSpec | undefined {
	const carrier: AnthropicControlCarrier = message;
	return message.role === "developer" ? carrier[ANTHROPIC_CONTROL] : undefined;
}

function collectAnthropicControlRecords(messages: readonly Message[]): AnthropicControlRecord[] {
	const records: AnthropicControlRecord[] = [];
	for (let index = 0; index < messages.length; index++) {
		const message = messages[index];
		if (message?.role === "assistant" && message.requestControls) {
			records.push({
				index,
				live: message.requestControls.messageIndex === index,
				controls: message.requestControls,
			});
		}
	}
	return records;
}

/**
 * Slot for an effort control of the request whose response lands at `end`:
 * before the turn's user message, where Anthropic applies it to that
 * response, or at `end` when the request continues a tool loop. It never
 * lands between a `tool_use` and its `tool_result`, where `transformMessages`
 * would flush synthetic aborted results; a mid-turn change therefore takes
 * effect from the next step.
 *
 * A summary replayed as a `compaction` block is never a slot: the block must
 * open the request, so nothing may precede it. A response opened by the block
 * (no real user turn between them) takes the change from its next step. A
 * summary sent as text is an ordinary user turn.
 */
function anthropicEffortInsertIndex(
	messages: readonly Message[],
	end: number,
	compactionReplay: AnthropicCompactionReplay | undefined,
): number {
	for (let i = end - 1; i >= 0; i--) {
		const message = messages[i];
		if (message?.role === "assistant") return end;
		if (message?.role !== "user") continue;
		if (!compactionReplay || !replaysAnthropicCompactionBlock(message.providerPayload, compactionReplay)) return i;
		let next = end;
		if (messages[next]?.role === "assistant") next++;
		while (messages[next]?.role === "toolResult") next++;
		return next;
	}
	return end;
}

/** `tool_removal`s (in `previous` order) then `tool_addition`s (in `next` order), limited to `declared`. */
function diffAnthropicActiveTools(
	previous: readonly string[],
	next: readonly string[],
	declared: ReadonlySet<string>,
): AnthropicToolChange[] {
	const previousSet = new Set(previous);
	const nextSet = new Set(next);
	const changes: AnthropicToolChange[] = [];
	for (const name of previous) {
		if (!nextSet.has(name) && declared.has(name)) changes.push({ type: "tool_removal", name });
	}
	for (const name of next) {
		if (!previousSet.has(name) && declared.has(name)) changes.push({ type: "tool_addition", name });
	}
	return changes;
}

/**
 * Keep top-level `tools` byte-stable across a conversation. The declared list
 * is the latest record's, minus names with no definition in `tools` or
 * `inactiveTools`, plus newly active tools appended with `defer_loading`.
 * Every recorded active-set change is replayed as a control before the
 * response of the request that made it; the change this request makes goes
 * at the tail. Records at a different index than they were written at are
 * rewritten history: they keep the declaration but emit no controls, so the
 * net change from the declared baseline lands at the first live position.
 */
function planAnthropicToolControls(
	context: Context,
	records: readonly AnthropicControlRecord[],
	enabled: boolean,
): { tools: Tool[] | undefined; inserts: AnthropicControlInsert[]; record: AnthropicToolControls | undefined } {
	if (!enabled || !context.tools) return { tools: context.tools, inserts: [], record: undefined };
	const definitions = new Map<string, Tool>();
	for (const tool of context.tools) definitions.set(tool.name, tool);
	for (const tool of context.inactiveTools ?? []) {
		if (!definitions.has(tool.name)) definitions.set(tool.name, tool);
	}
	const activeNames = context.tools.map(tool => tool.name);

	const toolRecords: { index: number; live: boolean; tools: AnthropicToolControls }[] = [];
	for (const record of records) {
		if (record.controls.tools) {
			toolRecords.push({ index: record.index, live: record.live, tools: record.controls.tools });
		}
	}
	const inserts: AnthropicControlInsert[] = [];
	let declared = activeNames;
	const deferred = new Set<string>();
	const latest = toolRecords.at(-1);
	if (latest) {
		// A declared name without a definition cannot be re-sent: accepted cache miss (e.g. after a resume).
		declared = latest.tools.declared.filter(name => definitions.has(name));
		const declaredSet = new Set(declared);
		for (const name of latest.tools.deferred) {
			if (declaredSet.has(name)) deferred.add(name);
		}
		for (const name of activeNames) {
			if (declaredSet.has(name)) continue;
			declared.push(name);
			declaredSet.add(name);
			deferred.add(name);
		}
		// The chain starts from what the top-level declaration makes active, so a
		// rewritten history converges on the current roster instead of trusting a
		// record whose earlier controls were summarized away.
		let previous = declared.filter(name => !deferred.has(name));
		for (const record of toolRecords) {
			if (!record.live) continue;
			const toolChanges = diffAnthropicActiveTools(previous, record.tools.active, declaredSet);
			if (toolChanges.length > 0) inserts.push({ index: record.index, spec: { toolChanges } });
			previous = record.tools.active;
		}
		const toolChanges = diffAnthropicActiveTools(previous, activeNames, declaredSet);
		if (toolChanges.length > 0) inserts.push({ index: context.messages.length, spec: { toolChanges } });
	}

	const tools: Tool[] = [];
	for (const name of declared) {
		const tool = definitions.get(name);
		if (tool) tools.push(deferred.has(name) ? { ...tool, deferLoading: true } : tool);
	}
	return {
		tools,
		inserts,
		record: {
			declared: tools.map(tool => tool.name),
			deferred: tools.filter(tool => tool.deferLoading === true).map(tool => tool.name),
			active: activeNames,
		},
	};
}

/**
 * Keep top-level `output_config.effort` byte-stable across a conversation and
 * replay every recorded change as a per-message effort control. An omitted
 * effort means the API's per-model default, which a per-message control cannot
 * restore, so a request without an explicit effort keeps the level in force.
 * Records at a different index than they were written at are rewritten history:
 * they keep the top-level effort but emit no controls, so the net change lands
 * at the first live position.
 */
function planAnthropicEffortControls(
	current: AnthropicOutputEffort | undefined,
	messages: readonly Message[],
	records: readonly AnthropicControlRecord[],
	enabled: boolean,
	compactionReplay: AnthropicCompactionReplay | undefined,
): {
	topLevel: AnthropicOutputEffort | undefined;
	inserts: AnthropicControlInsert[];
	record: AnthropicEffortControls | undefined;
} {
	if (!enabled) return { topLevel: current, inserts: [], record: undefined };
	const effortRecords: { index: number; live: boolean; effort: AnthropicEffortControls }[] = [];
	for (const record of records) {
		if (record.controls.effort) {
			effortRecords.push({ index: record.index, live: record.live, effort: record.controls.effort });
		}
	}
	const latest = effortRecords.at(-1);
	if (!latest) {
		return { topLevel: current, inserts: [], record: { topLevel: current ?? null, tail: current ?? null } };
	}
	const topLevel = latest.effort.topLevel ?? undefined;
	const inserts: AnthropicControlInsert[] = [];
	let tail = topLevel;
	for (const record of effortRecords) {
		if (!record.live) continue;
		const recorded = record.effort.tail;
		if (recorded !== null && recorded !== tail) {
			inserts.push({
				index: anthropicEffortInsertIndex(messages, record.index, compactionReplay),
				spec: { toolChanges: [], effort: recorded },
			});
		}
		tail = recorded ?? tail;
	}
	if (current !== undefined && current !== tail) {
		inserts.push({
			index: anthropicEffortInsertIndex(messages, messages.length, compactionReplay),
			spec: { toolChanges: [], effort: current },
		});
		tail = current;
	}
	return { topLevel, inserts, record: { topLevel: topLevel ?? null, tail: tail ?? null } };
}

/**
 * Splice control markers into `messages`. Inserts sharing an index come from
 * one request and merge into one control: tool changes first, then the effort.
 */
function insertAnthropicControlMarkers(messages: Message[], inserts: readonly AnthropicControlInsert[]): Message[] {
	if (inserts.length === 0) return messages;
	const merged = new Map<number, AnthropicControlSpec>();
	for (const { index, spec } of inserts) {
		const existing = merged.get(index);
		if (!existing) {
			merged.set(index, { ...spec, toolChanges: [...spec.toolChanges] });
			continue;
		}
		existing.toolChanges.push(...spec.toolChanges);
		if (spec.effort !== undefined) existing.effort = spec.effort;
	}
	const result = messages.slice();
	for (const [index, spec] of [...merged].sort((a, b) => b[0] - a[0])) {
		const marker: DeveloperMessage & AnthropicControlCarrier = {
			role: "developer",
			content: [],
			attribution: "agent",
			timestamp: messages[index - 1]?.timestamp ?? 0,
		};
		marker[ANTHROPIC_CONTROL] = spec;
		result.splice(index, 0, marker);
	}
	return result;
}

/** Wire `tool_addition`/`tool_removal` blocks for `changes`. */
function anthropicToolChangeBlocks(
	changes: readonly AnthropicToolChange[],
	isOAuthToken: boolean,
	model: Model<"anthropic-messages">,
): ContentBlockParam[] {
	return changes.map((change): ContentBlockParam => ({
		type: change.type,
		tool: {
			type: "tool_reference",
			name: encodeAnthropicToolName(change.name, isOAuthToken, model.compat.escapeBuiltinToolNames),
		},
	}));
}

type AnthropicParamBuildOptions = {
	disableStrictTools: boolean;
	useUmansGatewayWebSearch: boolean;
	forceDemoteUnsignedThinking: boolean;
	supportsEagerToolInputStreaming: boolean;
	prefixMismatchBehavior?: "drop_block" | "error";
	dropAllThinking: boolean;
	droppedThinkingBlocks?: ReadonlySet<string>;
	/** Sanitized server-side fallback entries; defaults to `options?.fallbacks` when omitted. */
	fallbacks?: AnthropicOptions["fallbacks"];
	/**
	 * Whether the endpoint this request reaches accepts server-side compaction
	 * (see {@link supportsAnthropicCompaction}); false keeps the edit, the beta,
	 * and block replay inert. Defaults to the model's static resolution.
	 */
	compactionSupported?: boolean;
	/**
	 * Already-resolved effective endpoint for this request (reroutes applied).
	 * Beta routing reads this instead of the spec URL so environment reroutes
	 * land on the right channel. Defaults to `model.baseUrl`.
	 */
	effectiveBaseUrl?: string;
};

/**
 * Fast mode (`speed: "fast"` plus its beta): a KDL fast-mode SKU, or a
 * priority-tier request to first-party Anthropic.
 */
function wantsAnthropicFastMode(model: Model<"anthropic-messages">, options: AnthropicOptions | undefined): boolean {
	return model.compat.fastMode === true || (model.provider === "anthropic" && options?.serviceTier === "priority");
}

/**
 * Native Factory Droid history predicate (ported from the CLI's request
 * builder): true when the conversation contains assistant turns but none of
 * them leads with a `thinking`/`redactedThinking` block.
 */
function hasThinkinglessAssistantHistory(messages: readonly Message[]): boolean {
	let hasAssistant = false;
	for (const message of messages) {
		if (message.role !== "assistant") continue;
		if (!Array.isArray(message.content) || message.content.length === 0) continue;
		hasAssistant = true;
		const first = message.content[0];
		if (
			first != null &&
			typeof first === "object" &&
			(first.type === "thinking" || first.type === "redactedThinking")
		) {
			return false;
		}
	}
	return hasAssistant;
}

/**
 * Native Factory Droid history predicate: true when the message after the last
 * user turn is an assistant turn whose first block is not
 * `thinking`/`redactedThinking` (the conversation is not resuming a
 * thinking-led chain).
 */
function hasNonThinkingTurnAfterLastUser(messages: readonly Message[]): boolean {
	let lastUser = -1;
	for (let i = messages.length - 1; i >= 0; i--) {
		if (messages[i].role === "user") {
			lastUser = i;
			break;
		}
	}
	if (lastUser === -1 || lastUser === messages.length - 1) return false;
	const next = messages[lastUser + 1];
	if (next.role !== "assistant") return false;
	if (!Array.isArray(next.content) || next.content.length === 0) return false;
	const first = next.content[0];
	if (first == null || typeof first !== "object") return false;
	return first.type !== "thinking" && first.type !== "redactedThinking";
}

/**
 * Whether the conversation has stopped being thinking-led, so routes with
 * `compat.stripThinkingHistory` drop the budget `thinking` config and replay
 * history without thinking blocks. Exported so the factory-droid provider can
 * gate its header-level interleaved beta on the same decision.
 */
export function shouldStripThinkingHistory(messages: readonly Message[]): boolean {
	return hasThinkinglessAssistantHistory(messages) || hasNonThinkingTurnAfterLastUser(messages);
}

function buildParams(
	model: Model<"anthropic-messages">,
	context: Context,
	isOAuthToken: boolean,
	options: AnthropicOptions | undefined,
	buildOptions: AnthropicParamBuildOptions,
): { params: MessageCreateParamsStreaming; requestControls: AnthropicRequestControls | undefined } {
	const {
		disableStrictTools,
		useUmansGatewayWebSearch,
		forceDemoteUnsignedThinking,
		supportsEagerToolInputStreaming,
		prefixMismatchBehavior,
		dropAllThinking,
		droppedThinkingBlocks,
		fallbacks = options?.fallbacks,
		compactionSupported = supportsAnthropicCompaction(model),
		effectiveBaseUrl,
	} = buildOptions;
	// A session-scoped auto-demote (learned from a live signing 400) clones the
	// resolved compat with `replayUnsignedThinking: false` so every subsequent
	// downstream read (convertAnthropicMessages, transformMessages) sees the
	// demoted default without mutating the shared `model` reference.
	const effectiveModel =
		forceDemoteUnsignedThinking && model.compat.replayUnsignedThinking
			? { ...model, compat: { ...model.compat, replayUnsignedThinking: false } }
			: model;
	const { cacheControl } = getCacheControl(model, options?.cacheRetention, isOAuthToken);

	// Pre-compute system blocks so they occupy the right slot in the serialized body.
	const shouldInjectClaudeCodeInstruction = isOAuthToken && model.compat.injectClaudeCodeInstruction !== false;
	const firstUserMessageText = shouldInjectClaudeCodeInstruction
		? extractClaudeCodeFirstUserMessageText(context.messages)
		: "";
	const systemBlocks = buildAnthropicSystemBlocks(context.systemPrompt, {
		includeClaudeCodeInstruction: shouldInjectClaudeCodeInstruction,
		firstUserMessageText,
		cacheControl,
	});

	// Controls earlier requests recorded on their responses fix the declared
	// tools, the top-level effort and every control message in between.
	const records = collectAnthropicControlRecords(context.messages);
	const toolPlan = planAnthropicToolControls(
		context,
		records,
		model.compat.supportsMidConversationToolChanges === true,
	);

	// Pre-compute tools.
	let tools: AnthropicWireTool[] | undefined;
	if (toolPlan.tools) {
		tools = convertTools(
			toolPlan.tools,
			isOAuthToken,
			disableStrictTools,
			supportsEagerToolInputStreaming,
			model.compat.escapeBuiltinToolNames,
			useUmansGatewayWebSearch,
		);
	} else if (isOAuthToken) {
		tools = [];
	}

	// Pre-compute metadata.
	const metadataAccountId = readAnthropicMetadataAccountId(options?.metadata);
	const metadataUserId = resolveAnthropicMetadataUserId(
		readAnthropicMetadataString(options?.metadata, "user_id") ??
			// Deliberately share the normalized affinity identity across Kimi's two transports.
			(model.provider === "kimi-code" ? getOpenAIPromptCacheKey(options) : undefined),
		isOAuthToken,
		options?.sessionId,
		metadataAccountId,
	);
	const metadata = metadataUserId ? { user_id: metadataUserId } : undefined;

	// Pre-compute thinking + output_config effort.
	let thinking: MessageCreateParamsStreaming["thinking"] | undefined;
	let outputConfigEffort: AnthropicOutputEffort | undefined;
	if (model.reasoning) {
		const disabledThinking = model.compat.disabledThinking;
		if (options?.thinkingEnabled === false && disabledThinking !== undefined) {
			if (disabledThinking !== "omit") {
				thinking =
					disabledThinking === "adaptive"
						? {
								type: "adaptive",
								...(model.thinking?.supportsDisplay
									? { display: options.thinkingDisplay ?? "summarized" }
									: {}),
							}
						: { type: "disabled" };
			}
		} else if (options?.thinkingEnabled || model.compat.requiresThinkingEnabled) {
			const thinkingOptions = options ?? {};
			const mode = model.thinking?.mode;
			const effort = resolveAnthropicAdaptiveEffort(model, thinkingOptions);
			if (!usesBudgetThinking(model)) {
				const adaptive: { type: "adaptive"; display?: AnthropicThinkingDisplay } = { type: "adaptive" };
				// Starting with Claude Opus 4.7 and Claude Fable/Mythos 5, adaptive thinking
				// content is omitted from the response by default. Opt into summarized
				// reasoning so thinking deltas keep streaming with human-readable content for
				// callers that rely on it. The `display` field is gated strictly on model
				// support: Opus 4.6 / Sonnet 4.6+ reject it with a 400, so an explicit
				// `thinkingDisplay` MUST NOT force it onto a model that can't accept it.
				if (model.thinking?.supportsDisplay) {
					adaptive.display = thinkingOptions.thinkingDisplay ?? "summarized";
				}
				thinking = adaptive;
				if (effort && effort !== "adaptive") outputConfigEffort = effort;
			} else {
				thinking = {
					type: "enabled",
					budget_tokens: thinkingOptions.thinkingBudgetTokens || 1024,
					// Budget thinking accepts `display` unless a host explicitly declares
					// it unsupported (proxies whose native clients never send it).
					...(model.thinking?.supportsDisplay === false
						? {}
						: { display: thinkingOptions.thinkingDisplay ?? "summarized" }),
				};
				if (mode === "anthropic-budget-effort" && effort && effort !== "adaptive") outputConfigEffort = effort;
			}
		} else if (options?.thinkingEnabled === false) {
			if (model.compat.supportsBetweenToolsThinking) {
				// Sonnet 5.5 rejects `disabled` with a 400; `between_tools` is its lowest
				// thinking setting. It takes no other field and leaves effort untouched:
				// pinning `low` here would cap the whole turn's quality, not only thinking.
				thinking = { type: "between_tools" };
			} else if (isAdaptiveOnlyThinking(model)) {
				// Adaptive-only Claude models (Opus 4.6+, Sonnet 4.6+, Fable/Mythos 5) reject
				// `thinking.type: "disabled"` — adaptive thinking cannot be switched off.
				// Omit the thinking field (the API defaults to adaptive) and pin the
				// lowest effort so "thinking off" calls stay cheap instead of failing
				// the request with a 400 (a hidden-thinking toggle must never break it).
				// The effort field requires the `effort-2025-11-24` beta; it is attached
				// at the request site, including per-request for injected SDK clients.
				outputConfigEffort = "low";
			} else {
				thinking = { type: "disabled" };
			}
		}
	}

	if (prefixMismatchBehavior) {
		if (
			!thinking &&
			model.thinking?.mode === "anthropic-adaptive" &&
			!(options?.thinkingEnabled === false && model.compat.disabledThinking === "omit")
		) {
			thinking = { type: "adaptive" };
		}
		if (thinking?.type === "adaptive" || thinking?.type === "enabled") {
			thinking.block_binding = { prefix_mismatch_behavior: prefixMismatchBehavior };
		}
	}

	// Factory Droid's native thinking-history rule for non-adaptive budget
	// models: when the conversation is not thinking-led (an assistant turn
	// exists but none opens with a thinking block, or the turn after the last
	// user does not), the CLI drops the `thinking` field and replays the
	// history without thinking blocks — budget-effort models keep
	// `output_config.effort`, interleaved budget models carry none. Off turns
	// never strip (native only applies this to an active non-adaptive
	// `thinking` config).
	const stripThinkingHistory =
		model.compat.stripThinkingHistory === true &&
		thinking?.type === "enabled" &&
		shouldStripThinkingHistory(context.messages);
	if (stripThinkingHistory) thinking = undefined;

	// Pre-compute context_management. Send keep: "all" for every enabled or
	// adaptive thinking request (OAuth + API-key) — not just OAuth. Without
	// this directive Anthropic-compatible backends (Z.AI, Kimi, DeepSeek, …)
	// strip the replayed thinking blocks `replayUnsignedThinking` puts back
	// on the wire, so the model loses the prior reasoning chain across turns
	// and the KV cache misses every turn (#3288). Narrowing this guard back
	// to `isOAuthToken` regresses every API-key thinking provider. Skip
	// injected clients because this code cannot add the required
	// `context-management-2025-06-27` beta to caller-owned SDK clients.
	// Providers that cannot deliver or accept context management disable it
	// through model compatibility policy.
	const shouldKeepThinkingContext =
		!options?.client &&
		model.compat.supportsContextManagement !== false &&
		(effectiveBaseUrl === undefined || !isVertexRawPredictUrl(effectiveBaseUrl)) &&
		(thinking?.type === "adaptive" || thinking?.type === "enabled");
	// A new on-demand compaction request cannot carry context_management.
	// Later turns carrying its signed block may keep clear_thinking as usual.
	// Persisted encrypted threshold blocks alone need a legacy replay edit.
	const compactionRequest = compactionSupported ? options?.anthropicCompaction : undefined;
	const signedReplay = compactionSupported && contextReplaysAnthropicCompaction(context.messages, model, "signed");
	const legacyReplay =
		compactionSupported && !signedReplay && contextReplaysAnthropicCompaction(context.messages, model, "legacy");
	const contextManagementEdits: NonNullable<MessageCreateParams["context_management"]>["edits"] = [];
	if (!compactionRequest && shouldKeepThinkingContext) {
		contextManagementEdits.push({ type: "clear_thinking_20251015", keep: "all" });
	}
	if (legacyReplay && !compactionRequest) {
		contextManagementEdits.push(buildAnthropicCompactionReplayEdit(model));
	}
	const contextManagement = contextManagementEdits.length > 0 ? { edits: contextManagementEdits } : undefined;

	// Pre-compute output_config. Skip `effort` on Vertex rawPredict: it requires
	// the `effort-2025-11-24` beta, which that adapter can only accept in the body
	// (`anthropic_beta`), never as the `anthropic-beta` HTTP header this path sets
	// — so the field is dropped alongside the beta to avoid a 400 (#5614).
	const compactionReplay: AnthropicCompactionReplay | undefined = compactionSupported
		? { model: effectiveModel, legacy: !compactionRequest && !signedReplay }
		: undefined;
	const effortPlan = planAnthropicEffortControls(
		outputConfigEffort,
		context.messages,
		records,
		model.compat.supportsPerMessageEffort === true,
		compactionReplay,
	);
	// `between_tools` returns a 400 at `xhigh`/`max` effort, and the effort in
	// force from earlier turns outlives a thinking toggle. Fall back to the
	// default adaptive request, which accepts every effort level.
	if (thinking?.type === "between_tools" && (effortPlan.topLevel === "xhigh" || effortPlan.topLevel === "max")) {
		thinking = undefined;
	}
	const wireMessages = convertAnthropicMessages(
		insertAnthropicControlMarkers(context.messages, [...toolPlan.inserts, ...effortPlan.inserts]),
		effectiveModel,
		isOAuthToken,
		{
			serverSideFallbackEnabled: !!fallbacks?.length,
			replayCompaction: compactionSupported,
			replayLegacyCompaction: compactionReplay?.legacy,
			dropAllThinking: dropAllThinking || stripThinkingHistory,
			droppedThinkingBlocks,
			credentialId: options?.credentialId,
		},
	);
	// Anchor the stable tools+system head so it stays cached across turns; the
	// moving message tail is anchored separately in applyPromptCaching below.
	applyHeadCaching(systemBlocks, tools, cacheControl);
	const requestControls: AnthropicRequestControls | undefined =
		toolPlan.record || effortPlan.record
			? {
					messageIndex: context.messages.length,
					...(toolPlan.record && { tools: toolPlan.record }),
					...(effortPlan.record && { effort: effortPlan.record }),
				}
			: undefined;

	const outputConfigEntries: AnthropicOutputConfig = {};
	if (effortPlan.topLevel && model.compat.supportsOutputEffort) outputConfigEntries.effort = effortPlan.topLevel;
	if (options?.taskBudget) {
		if (compactionRequest || signedReplay) {
			const taskBudget = { ...options.taskBudget };
			delete taskBudget.remaining;
			outputConfigEntries.task_budget = taskBudget;
		} else {
			outputConfigEntries.task_budget = options.taskBudget;
		}
	}
	const outputConfig = Object.keys(outputConfigEntries).length ? outputConfigEntries : undefined;

	// OAuth and API-key requests alike get the full model ceiling; Claude Code
	// itself requests 128k on Opus 5.5.
	const maxOutputTokens = anthropicOutputLimit(model);

	// A caller-owned client targets its own endpoint: route body betas by the
	// client's URL when it exposes one, not the model's routing. Otherwise the
	// already-resolved effective URL wins over the spec URL so environment
	// reroutes land on the right channel.
	const vertexRequestUrl =
		(options?.client !== undefined ? injectedClientBaseUrl(options.client) : undefined) ??
		effectiveBaseUrl ??
		model.baseUrl;
	const vertexControlBetas = isVertexRawPredictUrl(vertexRequestUrl)
		? resolveAnthropicControlBetas(model, prefixMismatchBehavior)
		: [];
	// Vertex rawPredict routes on-demand and legacy replay betas in the body.
	if (isVertexRawPredictUrl(vertexRequestUrl)) {
		if ((compactionRequest || signedReplay) && compactionSupported && !vertexControlBetas.includes(COMPACTION_BETA)) {
			vertexControlBetas.push(COMPACTION_BETA);
		}
		if (legacyReplay && !compactionRequest && !vertexControlBetas.includes(LEGACY_COMPACTION_BETA)) {
			vertexControlBetas.push(LEGACY_COMPACTION_BETA);
		}
	}

	// Build params in the canonical field order: model → messages → system → tools →
	// metadata → max_tokens → thinking → context_management/compaction → output_config → stream.
	const params: MessageCreateParamsStreaming = {
		model: options?.requestModelId ?? model.requestModelId ?? model.id,
		messages: wireMessages,
		...(systemBlocks && { system: systemBlocks }),
		...(tools !== undefined && { tools }),
		...(metadata && { metadata }),
		max_tokens: Math.min(maxOutputTokens, options?.maxTokens ?? maxOutputTokens),
		...(thinking && { thinking }),
		...(contextManagement && { context_management: contextManagement }),
		...(compactionRequest && {
			compaction: {
				type: "summarize",
				...(compactionRequest.instructions ? { instructions: compactionRequest.instructions } : {}),
			},
		}),
		...(outputConfig && { output_config: outputConfig }),
		...(fallbacks?.length ? { fallbacks } : {}),
		...(vertexControlBetas.length > 0 ? { anthropic_beta: vertexControlBetas } : {}),
		stream: true,
	};

	// Opus 4.7+ and Fable/Mythos 5 reject non-default sampling parameters with 400 error.
	const thinkingType = params.thinking?.type;
	const allowSamplingParams =
		model.compat.supportsSamplingParams && (thinkingType === undefined || thinkingType === "disabled");
	if (allowSamplingParams && options?.temperature !== undefined) {
		params.temperature = options.temperature;
	}
	if (allowSamplingParams && options?.topP !== undefined) {
		params.top_p = options.topP;
	}
	if (allowSamplingParams && options?.topK !== undefined) {
		params.top_k = options.topK;
	}
	if (!compactionRequest && options?.stopSequences?.length) {
		const seqs = options.stopSequences;
		if (seqs.length > ANTHROPIC_STOP_SEQUENCES_MAX && !warnedStopSequencesTrim) {
			warnedStopSequencesTrim = true;
			logger.warn("anthropic: stop_sequences exceeds 4; extra entries dropped", {
				received: seqs.length,
				kept: ANTHROPIC_STOP_SEQUENCES_MAX,
			});
		}
		params.stop_sequences =
			seqs.length > ANTHROPIC_STOP_SEQUENCES_MAX ? seqs.slice(0, ANTHROPIC_STOP_SEQUENCES_MAX) : seqs;
	}

	if (wantsAnthropicFastMode(model, options)) {
		params.speed = "fast";
	}

	if (options?.toolChoice) {
		if (typeof options.toolChoice === "string") {
			params.tool_choice = { type: options.toolChoice };
		} else if (options.toolChoice.name) {
			params.tool_choice = {
				...options.toolChoice,
				name: encodeAnthropicToolName(
					options.toolChoice.name,
					isOAuthToken,
					model.compat.escapeBuiltinToolNames,
					useUmansGatewayWebSearch,
				),
			};
		}
		// Claude Fable/Mythos 5 reject forced tool use outright ("tool_choice forces
		// tool use is not compatible with this model"). Downgrade any/tool → auto so the
		// request succeeds; the tool stays available and the caller's prompt steers
		// the model toward it.
		const choiceType = params.tool_choice?.type;
		if (
			(choiceType === "any" || choiceType === "tool") &&
			(compactionRequest || !model.compat.supportsForcedToolChoice)
		) {
			params.tool_choice = { type: "auto" };
		}
	}

	disableThinkingIfToolChoiceForced(params, model);
	ensureMaxTokensForThinking(params, maxOutputTokens);
	applyPromptCaching(params, cacheControl);

	return { params, requestControls };
}

const EMPTY_ERROR_TOOL_RESULT_TEXT = "Tool failed with no output.";

function isEmptyToolResultWireContent(content: AnthropicToolResultContent): boolean {
	if (typeof content === "string") {
		return content.trim().length === 0;
	}
	return content.length === 0;
}

function ensureErrorToolResultWireContent(
	content: AnthropicToolResultContent,
	isError: boolean | undefined,
): AnthropicToolResultContent {
	if (!isError || !isEmptyToolResultWireContent(content)) {
		return content;
	}
	return typeof content === "string"
		? EMPTY_ERROR_TOOL_RESULT_TEXT
		: [{ type: "text", text: EMPTY_ERROR_TOOL_RESULT_TEXT }];
}

function buildToolResultBlock(
	model: Model<"anthropic-messages">,
	msg: ToolResultMessage,
	hoistedImages: ContentBlockParam[],
): ContentBlockParam {
	let content = convertContentBlocks(msg.content, model.input.includes("image"));
	// Anthropic rejects images inside error tool results ("all content must be
	// type `text` if `is_error` is true") — keep the text in the block and
	// hoist the images after the message's tool_result run.
	if (msg.isError && typeof content !== "string" && content.some(block => block.type === "image")) {
		for (const block of content) {
			if (block.type === "image") hoistedImages.push(block);
		}
		content = content.filter(block => block.type === "text");
	}
	// An empty array is valid for the official API, but strict Anthropic-compatible
	// endpoints (Z.AI GLM: 400 code 1213 "The prompt parameter was not received
	// normally") reject it; the empty-string form is accepted by both.
	if (Array.isArray(content) && content.length === 0) {
		content = "";
	}
	content = ensureErrorToolResultWireContent(content, msg.isError);
	const block: ContentBlockParam = {
		type: "tool_result",
		tool_use_id: msg.toolCallId,
		content,
		is_error: msg.isError,
	};
	if (model.compat.requiresToolResultId) {
		// Z.AI workaround (issue #814): include `id` aliased to `tool_use_id`.
		(block as unknown as Record<string, unknown>).id = msg.toolCallId;
	}
	return block;
}

/**
 * A single Anthropic conversation turn, including the mid-conversation
 * `system` role (Opus 4.8+ and Fable/Mythos 5).
 */
export type AnthropicMessageParam = MessageParam;

/**
 * Recursively replace lone surrogates in string leaves. Identity-preserving:
 * returns the input object/array when nothing changed.
 */
function toWellFormedDeep(value: unknown): unknown {
	if (typeof value === "string") {
		const wellFormed = value.toWellFormed();
		return wellFormed === value ? value : wellFormed;
	}
	if (Array.isArray(value)) {
		let changed = false;
		const next = value.map(entry => {
			const sanitized = toWellFormedDeep(entry);
			if (sanitized !== entry) changed = true;
			return sanitized;
		});
		return changed ? next : value;
	}
	if (isRecord(value)) {
		let changed = false;
		const next: Record<string, unknown> = {};
		for (const [key, entry] of Object.entries(value)) {
			const sanitized = toWellFormedDeep(entry);
			if (sanitized !== entry) changed = true;
			next[key] = sanitized;
		}
		return changed ? next : value;
	}
	return value;
}

/**
 * Serialize omp {@link Message}s to Anthropic wire messages.
 *
 * `opts.serverSideFallbackEnabled` — when the CURRENT request itself
 * opts into the server-side-fallback beta chain. Only then may a persisted
 * `fallback` content block from a prior turn be replayed on the wire;
 * otherwise the block is dropped to avoid a 400 on non-fallback requests
 * that don't send the beta.
 *
 * `opts.replayCompaction` — replay a user-role compaction summary that
 * carries an {@link AnthropicCompactionPayload} from this provider as a
 * native `compaction` block instead of its text. The API drops every block
 * before the compaction block, so the assistant turn carrying it may open
 * the conversation; the request must send the compaction beta (the stream
 * entry point adds it whenever such a payload is present).
 */
export function convertAnthropicMessages(
	messages: Message[],
	model: Model<"anthropic-messages">,
	isOAuthToken: boolean,
	opts?: {
		serverSideFallbackEnabled?: boolean;
		replayCompaction?: boolean;
		replayLegacyCompaction?: boolean;
		dropAllThinking?: boolean;
		droppedThinkingBlocks?: ReadonlySet<string>;
		credentialId?: number;
	},
): AnthropicMessageParam[] {
	// Indices of params emitted from `developer` messages. After the main pass,
	// the ones whose placement satisfies Anthropic's mid-conversation rules are
	// upgraded from the `user` role to the authoritative `system` role.
	const developerParams: Array<{ index: number; payload?: AnthropicMessagePayload }> = [];
	const params: AnthropicMessageParam[] = [];
	// Controls rendered from request-control markers. The assistant repairs
	// and the developer upgrade below look through them: they were inserted
	// by this provider, not authored in the conversation.
	const controlParams = new Set<AnthropicMessageParam>();
	// Harness file metadata queued behind a replayed compaction block. Flushed
	// after the next param boundary that keeps it clear of both the block (the
	// fold below must still join the block with a following assistant turn, or
	// that turn's thinking prefix changes) and any open tool_use turn (its
	// results must follow it contiguously).
	const pendingCompactionFiles: string[] = [];
	const flushCompactionFiles = (): void => {
		while (pendingCompactionFiles.length > 0) {
			const filesText = pendingCompactionFiles.shift();
			if (filesText === undefined || filesText.trim().length === 0) continue;
			// The payload bypassed the `transformMessages` redaction pass, so
			// the metadata takes the same credential redaction here that the
			// dropped message text received there.
			params.push({ role: "user", content: redactSensitiveCredentials(filesText) });
		}
	};

	const transformedMessages = transformMessages(
		messages,
		model,
		normalizeToolCallId,
		undefined,
		undefined,
		undefined,
		opts?.credentialId,
	);

	for (let i = 0; i < transformedMessages.length; i++) {
		const msg = transformedMessages[i];

		const control = anthropicControlOf(msg);
		if (control) {
			const controlParam: AnthropicMessageParam = {
				role: "system",
				content: anthropicToolChangeBlocks(control.toolChanges, isOAuthToken, model),
				...(control.effort ? { output_config: { effort: control.effort } } : {}),
			};
			controlParams.add(controlParam);
			params.push(controlParam);
			continue;
		}
		if (
			opts?.replayCompaction &&
			(msg.role === "user" || msg.role === "developer") &&
			replaysAnthropicCompactionBlock(msg.providerPayload, { model, legacy: opts.replayLegacyCompaction !== false })
		) {
			const compactionParam: AnthropicMessageParam = {
				role: "assistant",
				content: [compactionBlockParam(msg.providerPayload)],
			};
			copyPerCallContextMessage(compactionParam, msg);
			params.push(compactionParam);
			// The block carries the verbatim API summary, so the message text
			// (which holds the harness file lists) would be dropped with it.
			// Queue the file metadata for after the block: it sits past the
			// compaction boundary the API enforces, unlike anything before it.
			// The flush waits past a following assistant turn (see above).
			if (msg.providerPayload.filesText !== undefined) {
				pendingCompactionFiles.push(msg.providerPayload.filesText);
			}
			continue;
		}
		if (msg.role === "user" || msg.role === "developer") {
			// Queued file metadata predates this message, so it emits first.
			flushCompactionFiles();
			const payload =
				msg.role === "developer" && msg.providerPayload?.type === "anthropicMessage"
					? msg.providerPayload
					: undefined;
			const hasProviderControls =
				payload?.clearAt !== undefined ||
				payload?.effort !== undefined ||
				(payload?.toolChanges !== undefined && payload.toolChanges.length > 0);

			let content: string | ContentBlockParam[];
			if (typeof msg.content === "string") {
				if (msg.content.trim().length === 0) {
					if (!hasProviderControls) continue;
					content = [];
				} else {
					content = msg.content.toWellFormed();
				}
			} else {
				const contentBlocks = convertContentBlocks(msg.content, model.input.includes("image"));
				if (typeof contentBlocks === "string") {
					if (contentBlocks.trim().length === 0) {
						if (!hasProviderControls) continue;
						content = [];
					} else {
						content = contentBlocks;
					}
				} else {
					if (contentBlocks.length === 0 && !hasProviderControls) continue;
					content = contentBlocks;
				}
			}
			if (payload?.toolChanges && model.compat.supportsMidConversationToolChanges) {
				const blocks: ContentBlockParam[] =
					typeof content === "string" ? [{ type: "text", text: content }] : content;
				blocks.push(...anthropicToolChangeBlocks(payload.toolChanges, isOAuthToken, model));
				content = blocks;
			}
			if (msg.role === "developer") developerParams.push({ index: params.length, payload });
			const param: AnthropicMessageParam & ConversationalUserCarrier = { role: "user", content };
			// Record that this wire `user` came from a real conversational turn, so
			// prompt-cache decimation can tell it apart from everything else that
			// serializes as `role: "user"`: a `developer` message, a tool_result run,
			// a synthetic `Continue.` pad, the stale-tool-result note, and any
			// agent-authored turn (`synthetic`, or `attribution: "agent"`, which is
			// what compaction and branch summaries carry).
			const agentAuthored = msg.synthetic === true || msg.attribution === "agent";
			if (msg.role === "user" && !agentAuthored && !isSyntheticUser(msg)) {
				param[kConversationalUser] = true;
			}
			copyPerCallContextMessage(param, msg);
			params.push(param);
		} else if (msg.role === "assistant") {
			const blocks: ContentBlockParam[] = [];
			const hasSignedThinking = msg.content.some(
				block =>
					block.type === "thinking" && !!block.thinkingSignature && block.thinkingSignature.trim().length > 0,
			);

			// A caller that appends the compacting response itself holds the block
			// on the assistant message; it opened that response, so it opens the
			// replayed turn.
			if (
				opts?.replayCompaction &&
				replaysAnthropicCompactionBlock(msg.providerPayload, {
					model,
					legacy: opts.replayLegacyCompaction !== false,
				})
			) {
				blocks.push(compactionBlockParam(msg.providerPayload));
			}

			for (const block of msg.content) {
				if (block.type === "text") {
					if (block.text.trim().length === 0) continue;
					blocks.push({
						type: "text",
						text: block.text.toWellFormed(),
					});
				} else if (block.type === "thinking") {
					if (
						opts?.dropAllThinking ||
						(block.thinkingSignature && opts?.droppedThinkingBlocks?.has(`thinking:${block.thinkingSignature}`))
					) {
						continue;
					}
					if (hasSignedThinking) {
						if (!block.thinkingSignature || block.thinkingSignature.trim().length === 0) {
							if (block.thinking.trim().length === 0) continue;
							blocks.push({
								type: "text",
								text: renderDemotedThinking(model.id, block.thinking),
							});
							continue;
						}
						blocks.push({
							type: "thinking",
							thinking: block.thinking,
							signature: block.thinkingSignature,
						});
						continue;
					}
					if (block.thinking.trim().length === 0) continue;
					if (!block.thinkingSignature || block.thinkingSignature.trim().length === 0) {
						if (model.compat.replayUnsignedThinking) {
							blocks.push({
								type: "thinking",
								thinking: block.thinking.toWellFormed(),
								signature: "",
							});
						} else {
							blocks.push({
								type: "text",
								text: renderDemotedThinking(model.id, block.thinking),
							});
						}
					} else {
						blocks.push({
							type: "thinking",
							thinking: block.thinking.toWellFormed(),
							signature: block.thinkingSignature,
						});
					}
				} else if (block.type === "redactedThinking") {
					if (opts?.dropAllThinking || opts?.droppedThinkingBlocks?.has(`redacted:${block.data}`)) continue;
					if (block.data.trim().length === 0) continue;
					blocks.push({
						type: "redacted_thinking",
						data: block.data,
					});
				} else if (block.type === "anthropicServerTool") {
					blocks.push(block.block);
				} else if (block.type === "fallback") {
					// Replay ONLY when both sides are aligned: the current
					// request opted into the beta chain, and the target is
					// official Anthropic (the only endpoint that accepts the
					// block on the wire). `transformMessages` already drops
					// the block for cross-provider / non-official replays, so
					// this is defense-in-depth for direct convert calls.
					if (!opts?.serverSideFallbackEnabled || !model.compat.officialEndpoint) continue;
					blocks.push({
						type: "fallback",
						from: block.from,
						to: block.to,
					});
				} else if (block.type === "toolCall") {
					blocks.push({
						type: "tool_use",
						id: block.id,
						name: encodeAnthropicToolName(block.name, isOAuthToken, model.compat.escapeBuiltinToolNames),
						// Always sanitize: the model itself can emit lone-surrogate escapes
						// in tool-argument JSON (streamed out fine, rejected with a 400 on
						// replay by Anthropic's strict UTF-8 validation). toWellFormedDeep
						// is identity-preserving, so well-formed arguments stay
						// byte-identical and prompt-cache prefixes are unaffected.
						input: toWellFormedDeep(block.arguments ?? {}),
					});
				}
			}
			// Anthropic's replay validator rejects any non-`tool_use` block that
			// appears after a `tool_use` inside an assistant turn (400:
			// "tool_use ids were found without tool_result blocks immediately
			// after: <id>"). A persisted turn can violate this when a mid-turn
			// server-side fallback handoff lands after the primary model already
			// emitted a tool_use — the replayed content is then e.g.
			// [thinking, text, tool_use, fallback, text, tool_use] — and also for
			// the older cross-provider [text, tool_use, text] shape (issue #544).
			// Stable-partition into [...non-tool_use, ...tool_use], preserving each
			// side's relative order: the non-tool_use chain (thinking → text →
			// fallback → text) carries thinking signatures and the fallback
			// boundary marker whose order Anthropic verifies, while tool_use blocks
			// are unsigned and safe to defer to the tail. Fast-path untouched when
			// already in order so prompt-cache prefixes stay byte-identical.
			let sawToolUse = false;
			let needsPartition = false;
			for (const block of blocks) {
				if (block.type === "tool_use") {
					sawToolUse = true;
				} else if (sawToolUse && block.type !== "thinking" && block.type !== "redacted_thinking") {
					needsPartition = true;
					break;
				}
			}
			if (needsPartition) {
				const nonToolUse: ContentBlockParam[] = [];
				const toolUse: ContentBlockParam[] = [];
				for (const block of blocks) {
					if (block.type === "tool_use") toolUse.push(block);
					else nonToolUse.push(block);
				}
				blocks.length = 0;
				blocks.push(...nonToolUse, ...toolUse);
			}
			if (blocks.length === 0) continue;
			const assistantParam: AnthropicMessageParam = {
				role: "assistant",
				content: blocks,
			};
			copyPerCallContextMessage(assistantParam, msg);
			params.push(assistantParam);
			// Flush queued file metadata unless this turn left tool calls open:
			// their results must follow the turn contiguously, so the metadata
			// waits for the merged result message (or the end of the list).
			if (!blocks.some(block => block.type === "tool_use")) {
				flushCompactionFiles();
			}
		} else if (msg.role === "toolResult") {
			// Collect all consecutive toolResult messages, needed for z.ai Anthropic endpoint
			const toolResults: ContentBlockParam[] = [];
			// Images stripped out of error tool results, re-attached after the run.
			const hoistedImages: ContentBlockParam[] = [];
			const toolResultParam: AnthropicMessageParam = {
				role: "user",
				content: toolResults,
			};

			// Add the current tool result
			toolResults.push(buildToolResultBlock(model, msg, hoistedImages));
			copyPerCallContextMessage(toolResultParam, msg);

			// Look ahead for consecutive toolResult messages
			let j = i + 1;
			while (j < transformedMessages.length && transformedMessages[j].role === "toolResult") {
				const nextMsg = transformedMessages[j] as ToolResultMessage; // We know it's a toolResult
				toolResults.push(buildToolResultBlock(model, nextMsg, hoistedImages));
				copyPerCallContextMessage(toolResultParam, nextMsg);
				j++;
			}

			// Skip the messages we've already processed
			i = j - 1;

			if (hoistedImages.length > 0) {
				toolResults.push(
					{ type: "text", text: "Attached image(s) from the tool result(s) above:" },
					...hoistedImages,
				);
			}

			// Add a single user message with all tool results
			params.push(toolResultParam);
			// An open tool_use turn's results are whole again; queued file
			// metadata can follow without splitting the pairing.
			flushCompactionFiles();
		}
	}

	// Upgrade developer-origin params to mid-conversation `system` messages where
	// Anthropic's placement rules allow it (Opus 4.8+ / Fable/Mythos 5 on first-party API).
	// Rules: a system message must immediately follow a `user` turn and must be
	// the last entry or be followed by an `assistant` turn — never first, and
	// never consecutive. Requiring the next param to be `assistant` (or absent)
	// covers both the "followed by assistant / last" and "no consecutive system"
	// constraints. Anything that does not qualify stays a `user` message.
	// Request controls are looked through, so recording one never demotes a
	// neighbouring developer message on a later request.
	const skipControls = (index: number, step: 1 | -1): number => {
		let next = index + step;
		while (next >= 0 && next < params.length && controlParams.has(params[next])) next += step;
		return next;
	};
	if (developerParams.length > 0 && model.compat.supportsMidConversationSystem) {
		for (const developer of developerParams.toReversed()) {
			const idx = developer.index;
			const followsUser = params[skipControls(idx, -1)]?.role === "user";
			const nextIndex = skipControls(idx, 1);
			const lastOrBeforeAssistant = nextIndex >= params.length || params[nextIndex]?.role === "assistant";
			const content = params[idx].content;
			const systemCompatible =
				typeof content === "string" ||
				content.every(
					block => block.type === "text" || block.type === "tool_addition" || block.type === "tool_removal",
				);
			const effortOnly = developer.payload?.effort !== undefined && Array.isArray(content) && content.length === 0;
			if (!((followsUser && lastOrBeforeAssistant && systemCompatible) || effortOnly)) continue;

			const turnScoped = developer.payload?.clearAt === "next_user_message" && model.compat.supportsTurnScopedSystem;
			const hasEffort = developer.payload?.effort !== undefined && model.compat.supportsPerMessageEffort;
			const hasToolChanges = (developer.payload?.toolChanges?.length ?? 0) > 0;
			if (turnScoped && (hasEffort || hasToolChanges) && Array.isArray(content)) {
				const scopedContent = content.filter(block => block.type === "text");
				const controlContent = content.filter(block => block.type !== "text");
				if (scopedContent.length > 0) {
					params[idx] = {
						...params[idx],
						role: "system",
						content: scopedContent,
						clear_at: "next_user_message",
					};
					const controlParam: AnthropicMessageParam = {
						role: "system",
						content: controlContent,
						...(hasEffort ? { output_config: { effort: developer.payload?.effort } } : {}),
					};
					copyPerCallContextMessage(controlParam, params[idx]);
					params.splice(idx + 1, 0, controlParam);
					continue;
				}
			}

			params[idx] = {
				...params[idx],
				role: "system",
				content,
				...(turnScoped && !hasEffort && !hasToolChanges ? { clear_at: "next_user_message" } : {}),
				...(hasEffort ? { output_config: { effort: developer.payload?.effort } } : {}),
			};
		}
	}
	// A replayed compaction block opens the assistant response it was produced
	// in, so when the retained tail begins with an assistant turn the block
	// belongs at the head of that turn — the API's own response shape — rather
	// than in a standalone assistant param that would force a synthetic
	// `Continue.` user turn between two assistants.
	for (let i = params.length - 2; i >= 0; i--) {
		const current = params[i];
		const next = params[i + 1];
		if (
			current.role !== "assistant" ||
			next?.role !== "assistant" ||
			typeof current.content === "string" ||
			current.content.length !== 1 ||
			current.content[0]?.type !== "compaction" ||
			typeof next.content === "string"
		) {
			continue;
		}
		params.splice(i, 2, { ...next, content: [current.content[0], ...next.content] });
	}
	// Dropped empty user/developer turns can leave two assistant params adjacent;
	// the API rejects consecutive assistant messages. Repair with the same neutral
	// nudge used for trailing-assistant prefill below, directly after the earlier
	// assistant so request controls between them keep their slot.
	for (let i = params.length - 1; i > 0; i--) {
		if (params[i].role !== "assistant") continue;
		const previous = skipControls(i, -1);
		if (params[previous]?.role !== "assistant") continue;
		params.splice(previous + 1, 0, { role: "user", content: "Continue." });
		i = previous + 1;
	}
	// A trailing compaction summary leaves its file metadata queued; emit it
	// before the prefill check so the list ends the request as a user turn.
	flushCompactionFiles();
	const last = skipControls(params.length, -1);
	if (params[last]?.role === "assistant") {
		params.splice(last + 1, 0, { role: "user", content: "Continue." });
	}

	return params;
}

/**
 * JSON Schema whitelist for Anthropic tool `input_schema` nodes.
 *
 * Tracks the Anthropic Python SDK's `lib/_parse/_transform.py::transform_schema`,
 * with live Messages API guardrails for keywords the SDK preserves but the API rejects.
 * We keep only structural/metadata keywords Anthropic's validator honors, and demote
 * anything else into the node's `description` as `\n\n{key: value, ...}` so the model
 * still sees the constraint as a natural-language hint.
 *
 * `Set` (not `Record<string, true>`) because membership is probed against arbitrary
 * user/Zod-derived schema keys: a literal Record would falsely match prototype names
 * like `"toString"` and silently strip valid properties.
 */
const ANTHROPIC_TOOL_SCHEMA_UNIVERSAL_KEEP = new Set([
	"$ref",
	"$defs",
	"$schema",
	"definitions",
	"type",
	"anyOf",
	"allOf",
	"enum",
	"const",
	"description",
	"title",
	"default",
	"nullable",
]);
/** Keys preserved on `type: "object"` nodes (in addition to the universal set). */
const ANTHROPIC_TOOL_SCHEMA_OBJECT_KEEP = new Set(["properties", "required", "additionalProperties"]);
/** Keys preserved on `type: "array"` nodes; `minItems` only when its value is 0 or 1. */
const ANTHROPIC_TOOL_SCHEMA_ARRAY_KEEP = new Set(["items", "prefixItems", "minItems"]);
/** Keys preserved on `type: "string"` nodes; `format` only when its value is in the supported list. */
const ANTHROPIC_TOOL_SCHEMA_STRING_KEEP = new Set(["format"]);
/**
 * String `format` values Anthropic accepts; everything else (including `pattern`-style
 * format hints) gets demoted into `description`. Matches `SupportedStringFormats` in the
 * Anthropic SDK's `_transform.py`.
 */
const ANTHROPIC_TOOL_SCHEMA_STRING_FORMATS = new Set([
	"date-time",
	"time",
	"date",
	"duration",
	"email",
	"hostname",
	"uri",
	"ipv4",
	"ipv6",
	"uuid",
]);
// Not `bash`: strict decoding fixes property order, so an optional key declared
// before one the model has already written can no longer be emitted, and bash's
// `timeout` vanished from every `async`-first call.
const ANTHROPIC_STRICT_TOOL_ALLOWLIST = new Set(["python", "edit", "find"]);
const MAX_ANTHROPIC_STRICT_TOOLS = 20;
const MAX_ANTHROPIC_STRICT_OPTIONAL_PARAMETERS = 24;
const MAX_ANTHROPIC_STRICT_UNION_PARAMETERS = 16;

/** `minItems` / `maxItems` apply to arrays; Anthropic rejects them on `type: "object"` (including `minItems: 0`/`1`). */
function isJsonSchemaArrayNode(schema: Record<string, unknown>): boolean {
	const t = schema.type;
	if (t === "array") return true;
	if (Array.isArray(t) && t.includes("array") && !t.includes("object")) return true;
	if (schema.items !== undefined || Array.isArray(schema.prefixItems)) return true;
	return false;
}

function isJsonSchemaObjectNode(schema: Record<string, unknown>): boolean {
	if (isJsonSchemaArrayNode(schema)) return false;
	if (schema.type === "object") return true;
	if (Array.isArray(schema.type) && schema.type.includes("object")) return true;
	if (isRecord(schema.properties)) return true;
	return false;
}

/**
 * Pick the principal non-null scalar type from a `type` keyword. Anthropic accepts
 * `type` as either a single string or an array (e.g. `["number", "null"]` for a
 * nullable value); the SDK whitelist is keyed off the scalar type, with `"null"`
 * ignored so nullable variants are normalized as their underlying type.
 */
function pickAnthropicScalarType(type: unknown): string | undefined {
	if (typeof type === "string") return type;
	if (Array.isArray(type)) {
		for (const entry of type) {
			if (typeof entry === "string" && entry !== "null") return entry;
		}
	}
	return undefined;
}
function pickAnthropicEffectiveScalarType(schema: Record<string, unknown>): string | undefined {
	const explicit = pickAnthropicScalarType(schema.type);
	if (explicit) return explicit;
	if (isRecord(schema.properties)) return "object";
	if (schema.items !== undefined || Array.isArray(schema.prefixItems)) return "array";
	return undefined;
}

function anthropicPerTypeKeep(scalarType: string | undefined): Set<string> | undefined {
	switch (scalarType) {
		case "object":
			return ANTHROPIC_TOOL_SCHEMA_OBJECT_KEEP;
		case "array":
			return ANTHROPIC_TOOL_SCHEMA_ARRAY_KEEP;
		case "string":
			return ANTHROPIC_TOOL_SCHEMA_STRING_KEEP;
		default:
			return undefined;
	}
}

/**
 * Normalize a JSON Schema node for Anthropic tool `input_schema`.
 *
 * Applies the full whitelist semantics from the Anthropic Python SDK's
 * `lib/_parse/_transform.py::transform_schema`:
 *
 * 1. Universal keys (`$ref`, `$defs`, `type`, `anyOf`, `allOf`, `enum`, `const`,
 *    `description`, `title`, `default`, `nullable`) are preserved on every node, with
 *    one position-dependent exception: the combinator keys. Root `anyOf`/`allOf` are
 *    spilled (recent Anthropic Messages validators reject combinators at the tool
 *    `input_schema` root) but kept when nested; `oneOf` is spilled at every position
 *    (it is not in the documented supported subset).
 * 2. Per-type keys are kept additively (object → `properties`/`required`/`additionalProperties`,
 *    array → `items`/`prefixItems` plus `minItems` only when 0 or 1, string → `format`
 *    only when in the supported value set).
 * 3. Everything else is demoted into the node's `description` as `\n\n{key: value, ...}`
 *
 * Object nodes default to `additionalProperties: false`, but explicit open-map
 * declarations (`additionalProperties: true` or a schema literal — Zod's
 * `z.record(z.string(), z.unknown())` produces `{}`) are preserved. The strict-mode
 * pass downstream demotes those shapes to non-strict instead of fabricating a closed
 * object, so callers like the resolve tool keep working open-map semantics.
 */
function normalizeAnthropicToolSchemaNode(
	schema: unknown,
	cache: WeakMap<Record<string, unknown>, Record<string, unknown>>,
	isRoot = false,
): unknown {
	if (Array.isArray(schema)) return schema.map(entry => normalizeAnthropicToolSchemaNode(entry, cache));
	if (!isRecord(schema)) return schema;

	const existing = cache.get(schema);
	if (existing !== undefined) return existing;

	const result: Record<string, unknown> = {};
	cache.set(schema, result);

	const scalarType = pickAnthropicEffectiveScalarType(schema);
	const perTypeKeep = anthropicPerTypeKeep(scalarType);
	const spill: Array<[string, unknown]> = [];

	for (const key in schema) {
		if (!Object.hasOwn(schema, key)) continue;
		const value = schema[key];
		const isRootCombinator = isRoot && COMBINATOR_KEYS.includes(key as (typeof COMBINATOR_KEYS)[number]);
		if (!isRootCombinator && (ANTHROPIC_TOOL_SCHEMA_UNIVERSAL_KEEP.has(key) || perTypeKeep?.has(key))) {
			result[key] = value;
		} else {
			spill.push([key, value]);
		}
	}

	// Per-type conditional keys: prune within the kept set.
	if (scalarType === "string") {
		const format = result.format;
		if (typeof format === "string" && !ANTHROPIC_TOOL_SCHEMA_STRING_FORMATS.has(format)) {
			spill.push(["format", format]);
			delete result.format;
		}
	}
	if (scalarType === "array" && result.minItems !== undefined) {
		const minItems = result.minItems;
		if (!(typeof minItems === "number" && (minItems === 0 || minItems === 1))) {
			spill.push(["minItems", minItems]);
			delete result.minItems;
		}
	}
	if (scalarType === "object" && result.additionalProperties === undefined) {
		result.additionalProperties = false;
	}

	// Recurse on structural keys.
	if (isRecord(result.properties)) {
		const normalizedProperties: Record<string, unknown> = {};
		const sourceProperties = result.properties as Record<string, unknown>;
		for (const propName in sourceProperties) {
			if (!Object.hasOwn(sourceProperties, propName)) continue;
			normalizedProperties[propName] = normalizeAnthropicToolSchemaNode(sourceProperties[propName], cache);
		}
		result.properties = normalizedProperties;
	}
	if (isRecord(result.additionalProperties)) {
		const normalized = normalizeAnthropicToolSchemaNode(result.additionalProperties, cache);
		if (isRecord(normalized) && Object.keys(normalized).length === 0) {
			result.additionalProperties = true;
		} else {
			result.additionalProperties = normalized;
		}
	}
	if (Array.isArray(result.items)) {
		result.items = result.items.map(item => normalizeAnthropicToolSchemaNode(item, cache));
	} else if (isRecord(result.items)) {
		result.items = normalizeAnthropicToolSchemaNode(result.items, cache);
	}
	if (Array.isArray(result.prefixItems)) {
		result.prefixItems = result.prefixItems.map(item => normalizeAnthropicToolSchemaNode(item, cache));
	}
	for (const key of COMBINATOR_KEYS) {
		const variants = result[key];
		if (Array.isArray(variants)) {
			result[key] = variants.map(variant => normalizeAnthropicToolSchemaNode(variant, cache));
		}
	}
	for (const defsKey of ["$defs", "definitions"] as const) {
		const definitions = result[defsKey];
		if (!isRecord(definitions)) continue;
		const normalizedDefs: Record<string, unknown> = {};
		const sourceDefs = definitions as Record<string, unknown>;
		for (const name in sourceDefs) {
			if (!Object.hasOwn(sourceDefs, name)) continue;
			normalizedDefs[name] = normalizeAnthropicToolSchemaNode(sourceDefs[name], cache);
		}
		result[defsKey] = normalizedDefs;
	}

	spillToDescription(result, spill);
	return result;
}

export function normalizeAnthropicToolSchema(schema: unknown): unknown {
	return normalizeAnthropicToolSchemaNode(schema, new WeakMap(), true);
}

type AnthropicToolSchemaPlan = {
	inputSchema: AnthropicToolInputSchema;
	strict: boolean;
};

type AnthropicStrictBudget = {
	optionalRemaining: number;
	unionRemaining: number;
	optionalCount: number;
	unionCount: number;
};

function hasAnthropicUnionType(schema: Record<string, unknown>): boolean {
	return Array.isArray(schema.type) || Array.isArray(schema.anyOf);
}

function hasNullVariant(schema: Record<string, unknown>): boolean {
	if (Array.isArray(schema.type) && schema.type.includes("null")) return true;
	return Array.isArray(schema.anyOf) && schema.anyOf.some(variant => isRecord(variant) && variant.type === "null");
}
function hasAnthropicSchemaDefiningKeyword(schema: Record<string, unknown>): boolean {
	if (
		schema.type !== undefined ||
		schema.properties !== undefined ||
		schema.additionalProperties !== undefined ||
		schema.items !== undefined ||
		schema.prefixItems !== undefined ||
		schema.enum !== undefined ||
		schema.const !== undefined ||
		schema.$ref !== undefined
	) {
		return true;
	}
	for (const key of COMBINATOR_KEYS) {
		if (schema[key] !== undefined) return true;
	}
	return schema.$defs !== undefined || schema.definitions !== undefined;
}

function makeAnthropicNullableSchema(schema: unknown, budget: AnthropicStrictBudget): unknown | undefined {
	if (isRecord(schema)) {
		if (hasNullVariant(schema)) return schema;
		if (Array.isArray(schema.anyOf)) {
			return { ...schema, anyOf: [...schema.anyOf, { type: "null" }] };
		}
		if (Array.isArray(schema.type)) {
			return { ...schema, type: [...schema.type, "null"] };
		}
	}

	if (budget.unionRemaining <= 0) return undefined;
	budget.unionRemaining--;
	budget.unionCount++;
	return { anyOf: [schema, { type: "null" }] };
}

function normalizeAnthropicStrictSchemaNode(
	schema: unknown,
	budget: AnthropicStrictBudget,
	cache: WeakMap<Record<string, unknown>, Record<string, unknown>>,
): unknown | undefined {
	if (Array.isArray(schema)) {
		const result: unknown[] = [];
		for (const entry of schema) {
			const normalized = normalizeAnthropicStrictSchemaNode(entry, budget, cache);
			if (normalized === undefined) return undefined;
			result.push(normalized);
		}
		return result;
	}

	if (!isRecord(schema)) return schema;

	const cached = cache.get(schema);
	if (cached) return cached;

	if (!hasAnthropicSchemaDefiningKeyword(schema)) return undefined;

	// Strict tool use only supports closed objects. Open maps stay available on
	// the non-strict schema plan instead of producing an Anthropic 400.
	if (isJsonSchemaObjectNode(schema) && schema.additionalProperties !== false) {
		return undefined;
	}

	const result: Record<string, unknown> = { ...schema };
	cache.set(schema, result);

	if (hasAnthropicUnionType(result)) {
		if (budget.unionRemaining <= 0) return undefined;
		budget.unionRemaining--;
		budget.unionCount++;
	}

	if (isRecord(result.properties)) {
		const originalRequired = new Set(
			Array.isArray(result.required)
				? result.required.filter((entry): entry is string => typeof entry === "string")
				: [],
		);
		const properties: Record<string, unknown> = {};
		const required: string[] = [];

		for (const [propertyName, propertySchema] of Object.entries(result.properties)) {
			const normalizedProperty = normalizeAnthropicStrictSchemaNode(propertySchema, budget, cache);
			if (normalizedProperty === undefined) return undefined;

			if (originalRequired.has(propertyName)) {
				properties[propertyName] = normalizedProperty;
				required.push(propertyName);
				continue;
			}

			if (budget.optionalRemaining > 0) {
				budget.optionalRemaining--;
				budget.optionalCount++;
				properties[propertyName] = normalizedProperty;
				continue;
			}

			const nullableProperty = makeAnthropicNullableSchema(normalizedProperty, budget);
			if (nullableProperty === undefined) return undefined;
			properties[propertyName] = nullableProperty;
			required.push(propertyName);
		}

		result.properties = properties;
		result.required = required;
	}

	if (Array.isArray(result.items)) {
		const items = normalizeAnthropicStrictSchemaNode(result.items, budget, cache);
		if (items === undefined) return undefined;
		result.items = items;
	} else if (isRecord(result.items)) {
		const items = normalizeAnthropicStrictSchemaNode(result.items, budget, cache);
		if (items === undefined) return undefined;
		result.items = items;
	}
	if (Array.isArray(result.prefixItems)) {
		const prefixItems = normalizeAnthropicStrictSchemaNode(result.prefixItems, budget, cache);
		if (prefixItems === undefined) return undefined;
		result.prefixItems = prefixItems;
	}

	for (const key of COMBINATOR_KEYS) {
		const variants = result[key];
		if (!Array.isArray(variants)) continue;
		const normalizedVariants = normalizeAnthropicStrictSchemaNode(variants, budget, cache);
		if (normalizedVariants === undefined) return undefined;
		result[key] = normalizedVariants;
	}

	for (const defsKey of ["$defs", "definitions"] as const) {
		const definitions = result[defsKey];
		if (!isRecord(definitions)) continue;
		const normalizedDefinitions: Record<string, unknown> = {};
		for (const [definitionName, definitionSchema] of Object.entries(definitions)) {
			const normalizedDefinition = normalizeAnthropicStrictSchemaNode(definitionSchema, budget, cache);
			if (normalizedDefinition === undefined) return undefined;
			normalizedDefinitions[definitionName] = normalizedDefinition;
		}
		result[defsKey] = normalizedDefinitions;
	}

	return result;
}

const ANTHROPIC_STRICT_INCOMPATIBLE_KEYWORDS = [
	"oneOf",
	"allOf",
	"$ref",
	"patternProperties",
	"propertyNames",
] as const;

/**
 * Anthropic's strict grammar subset supports anyOf/type-array unions only.
 * oneOf/allOf/$ref compile unpredictably (rejections arrive as 400s the
 * grammar-too-large fallback does not recognize, so they would hard-fail the
 * turn), and patternProperties/propertyNames describe open key sets that the
 * strict pipeline's injected `additionalProperties: false` would contradict.
 * Runs against the raw wire schema — the base normalizer spills several of
 * these keywords into the description, erasing the evidence.
 */
function hasAnthropicStrictIncompatibleKeyword(schema: unknown, seen = new Set<object>()): boolean {
	if (Array.isArray(schema)) {
		if (seen.has(schema)) return false;
		seen.add(schema);
		return schema.some(entry => hasAnthropicStrictIncompatibleKeyword(entry, seen));
	}
	if (!isRecord(schema)) return false;
	if (seen.has(schema)) return false;
	seen.add(schema);
	for (const keyword of ANTHROPIC_STRICT_INCOMPATIBLE_KEYWORDS) {
		if (schema[keyword] !== undefined) return true;
	}
	return Object.values(schema).some(value => hasAnthropicStrictIncompatibleKeyword(value, seen));
}

function normalizeAnthropicStrictSchema(
	schema: Record<string, unknown>,
	optionalRemaining: number,
	unionRemaining: number,
): { schema: Record<string, unknown>; optionalCount: number; unionCount: number } | undefined {
	const budget: AnthropicStrictBudget = {
		optionalRemaining,
		unionRemaining,
		optionalCount: 0,
		unionCount: 0,
	};
	const normalized = normalizeAnthropicStrictSchemaNode(schema, budget, new WeakMap());
	if (!isRecord(normalized)) return undefined;
	return { schema: normalized, optionalCount: budget.optionalCount, unionCount: budget.unionCount };
}

function buildAnthropicBaseToolInputSchema(tool: Tool): Record<string, unknown> {
	const jsonSchema = toolWireSchema(tool);
	return normalizeAnthropicToolSchema({
		...jsonSchema,
		type: "object",
		properties: isRecord(jsonSchema.properties) ? jsonSchema.properties : {},
		required: Array.isArray(jsonSchema.required)
			? jsonSchema.required.filter((entry): entry is string => typeof entry === "string")
			: [],
	}) as Record<string, unknown>;
}

function buildAnthropicToolSchemaPlans(tools: Tool[], disableStrictTools = false): AnthropicToolSchemaPlan[] {
	const plans = tools.map((tool): AnthropicToolSchemaPlan => ({
		inputSchema: buildAnthropicBaseToolInputSchema(tool) as AnthropicToolInputSchema,
		strict: false,
	}));
	if (NO_STRICT || disableStrictTools) return plans;

	const candidateIndexes = tools.flatMap((tool, index) => {
		if (!ANTHROPIC_STRICT_TOOL_ALLOWLIST.has(tool.name)) return [];
		if (tool.strict === false) return [];
		if (hasAnthropicStrictIncompatibleKeyword(toolWireSchema(tool))) return [];
		return [index];
	});

	let strictToolCount = 0;
	let strictOptionalParameterCount = 0;
	let strictUnionParameterCount = 0;
	for (const index of candidateIndexes) {
		if (strictToolCount >= MAX_ANTHROPIC_STRICT_TOOLS) break;

		const strictResult = normalizeAnthropicStrictSchema(
			plans[index].inputSchema as Record<string, unknown>,
			MAX_ANTHROPIC_STRICT_OPTIONAL_PARAMETERS - strictOptionalParameterCount,
			MAX_ANTHROPIC_STRICT_UNION_PARAMETERS - strictUnionParameterCount,
		);
		if (!strictResult) continue;

		plans[index] = {
			inputSchema: strictResult.schema as AnthropicToolInputSchema,
			strict: true,
		};
		strictToolCount++;
		strictOptionalParameterCount += strictResult.optionalCount;
		strictUnionParameterCount += strictResult.unionCount;
	}

	return plans;
}

function convertTools(
	tools: Tool[],
	isOAuthToken: boolean,
	disableStrictTools = false,
	supportsEagerToolInputStreaming = true,
	escapeBuiltinToolNames = false,
	useUmansGatewayWebSearch = false,
): AnthropicWireTool[] {
	if (!tools) return [];
	const schemaPlans = buildAnthropicToolSchemaPlans(tools, disableStrictTools);

	return tools.map((tool, index) => {
		const plan = schemaPlans[index];
		const baseTool = {
			name: encodeAnthropicToolName(tool.name, isOAuthToken, escapeBuiltinToolNames, useUmansGatewayWebSearch),
			description: tool.description || "",
			input_schema: plan.inputSchema,
		};
		return {
			...baseTool,
			...(supportsEagerToolInputStreaming ? { eager_input_streaming: true } : {}),
			...(plan.strict ? { strict: true } : {}),
			...(tool.deferLoading ? { defer_loading: true } : {}),
		};
	});
}

function mapStopReason(reason: string): StopReason {
	switch (reason) {
		case "end_turn":
			return "stop";
		case "max_tokens":
			return "length";
		// Generation ran into the model's context window (default behavior on
		// Sonnet 4.5+); the streamed content is valid, just truncated.
		case "model_context_window_exceeded":
			return "length";
		case "tool_use":
			return "toolUse";
		case "refusal":
			return "error";
		case "pause_turn": // Stop is good enough -> resubmit
			return "stop";
		case "compaction": // pause_after_compaction: the summary block is the whole response
			return "stop";
		case "stop_sequence":
			return "stop"; // A caller-supplied stop_sequences entry matched; the turn completed normally.
		case "sensitive": // Content flagged by safety filters (not yet in SDK types)
			return "error";
		default:
			// New stop reasons ship server-side first ("sensitive",
			// "model_context_window_exceeded") and arrive on the trailing
			// message_delta after all content has streamed. Degrade to a normal
			// stop instead of failing the fully streamed turn.
			reportAnthropicEnvelopeAnomaly(`unhandled stop reason: ${reason}`);
			return "stop";
	}
}