anthropic-messages.ts

1import Anthropic from "@anthropic-ai/sdk";
2import type {
3	CacheControlEphemeral,
4	ContentBlockParam,
5	MessageCreateParamsStreaming,
6	MessageParam,
7	RawMessageStreamEvent,
8	RefusalStopDetails,
9} from "@anthropic-ai/sdk/resources/messages.js";
10import { calculateCost } from "../models.ts";
11import type {
12	AnthropicMessagesCompat,
13	Api,
14	AssistantMessage,
15	CacheRetention,
16	Context,
17	ImageContent,
18	Message,
19	Model,
20	ProviderEnv,
21	ProviderHeaders,
22	SimpleStreamOptions,
23	StopReason,
24	StreamFunction,
25	StreamOptions,
26	TextContent,
27	ThinkingContent,
28	Tool,
29	ToolCall,
30	ToolResultMessage,
31} from "../types.ts";
32import { splitDeferredTools } from "../utils/deferred-tools.ts";
33import { AssistantMessageEventStream } from "../utils/event-stream.ts";
34import { headersToRecord } from "../utils/headers.ts";
35import { parseJsonWithRepair, parseStreamingJson } from "../utils/json-parse.ts";
36import { getProviderEnvValue } from "../utils/provider-env.ts";
37import { retryProviderRequest } from "../utils/provider-retry.ts";
38import { sanitizeSurrogates } from "../utils/sanitize-unicode.ts";
39
40import { resolveJsonSchemaStrictSampling } from "./constrained-sampling.ts";
41import { buildCopilotDynamicHeaders, hasCopilotVisionInput } from "./github-copilot-headers.ts";
42import { adjustMaxTokensForThinking, buildBaseOptions, clampMaxTokensToContext } from "./simple-options.ts";
43import { transformMessages } from "./transform-messages.ts";
44
45/**
46 * Resolve cache retention preference.
47 * Defaults to "short" and uses PI_CACHE_RETENTION for backward compatibility.
48 */
49function resolveCacheRetention(cacheRetention?: CacheRetention, env?: ProviderEnv): CacheRetention {
50	if (cacheRetention) {
51		return cacheRetention;
52	}
53	if (getProviderEnvValue("PI_CACHE_RETENTION", env) === "long") {
54		return "long";
55	}
56	return "short";
57}
58
59function getCacheControl(
60	model: Model<"anthropic-messages">,
61	cacheRetention?: CacheRetention,
62	env?: ProviderEnv,
63): { retention: CacheRetention; cacheControl?: CacheControlEphemeral } {
64	const retention = resolveCacheRetention(cacheRetention, env);
65	if (retention === "none") {
66		return { retention };
67	}
68	const ttl = retention === "long" && getAnthropicCompat(model).supportsLongCacheRetention ? "1h" : undefined;
69	return {
70		retention,
71		cacheControl: { type: "ephemeral", ...(ttl && { ttl }) },
72	};
73}
74
75// Stealth mode: Mimic Claude Code's tool naming exactly
76const claudeCodeVersion = "2.1.75";
77
78// Claude Code 2.x tool names (canonical casing)
79// Source: https://cchistory.mariozechner.at/data/prompts-2.1.11.md
80// To update: https://github.com/badlogic/cchistory
81const claudeCodeTools = [
82	"Read",
83	"Write",
84	"Edit",
85	"Bash",
86	"Grep",
87	"Glob",
88	"AskUserQuestion",
89	"EnterPlanMode",
90	"ExitPlanMode",
91	"KillShell",
92	"NotebookEdit",
93	"Skill",
94	"Task",
95	"TaskOutput",
96	"TodoWrite",
97	"WebFetch",
98	"WebSearch",
99];
100
101const ccToolLookup = new Map(claudeCodeTools.map((t) => [t.toLowerCase(), t]));
102
103// Convert tool name to CC canonical casing if it matches (case-insensitive)
104const toClaudeCodeName = (name: string) => ccToolLookup.get(name.toLowerCase()) ?? name;
105const fromClaudeCodeName = (name: string, tools?: Tool[]) => {
106	if (tools && tools.length > 0) {
107		const lowerName = name.toLowerCase();
108		const matchedTool = tools.find((tool) => tool.name.toLowerCase() === lowerName);
109		if (matchedTool) return matchedTool.name;
110	}
111	return name;
112};
113
114/**
115 * Convert content blocks to Anthropic API format
116 */
117function convertContentBlocks(content: (TextContent | ImageContent)[]):
118	| string
119	| Array<
120			| { type: "text"; text: string }
121			| {
122					type: "image";
123					source: {
124						type: "base64";
125						media_type: "image/jpeg" | "image/png" | "image/gif" | "image/webp";
126						data: string;
127					};
128			  }
129	  > {
130	// If only text blocks, return as concatenated string for simplicity
131	const hasImages = content.some((c) => c.type === "image");
132	if (!hasImages) {
133		return sanitizeSurrogates(content.map((c) => (c as TextContent).text).join("\n"));
134	}
135
136	// If we have images, convert to content block array
137	const blocks = content.map((block) => {
138		if (block.type === "text") {
139			return {
140				type: "text" as const,
141				text: sanitizeSurrogates(block.text),
142			};
143		}
144		return {
145			type: "image" as const,
146			source: {
147				type: "base64" as const,
148				media_type: block.mimeType as "image/jpeg" | "image/png" | "image/gif" | "image/webp",
149				data: block.data,
150			},
151		};
152	});
153
154	// If only images (no text), add placeholder text block
155	const hasText = blocks.some((b) => b.type === "text");
156	if (!hasText) {
157		blocks.unshift({
158			type: "text" as const,
159			text: "(see attached image)",
160		});
161	}
162
163	return blocks;
164}
165
166export type AnthropicEffort = "low" | "medium" | "high" | "xhigh" | "max";
167
168export type AnthropicThinkingDisplay = "summarized" | "omitted";
169
170const FINE_GRAINED_TOOL_STREAMING_BETA = "fine-grained-tool-streaming-2025-05-14";
171const INTERLEAVED_THINKING_BETA = "interleaved-thinking-2025-05-14";
172
173function getAnthropicCompat(
174	model: Model<"anthropic-messages">,
175): Required<Omit<AnthropicMessagesCompat, "forceAdaptiveThinking">> {
176	return {
177		supportsEagerToolInputStreaming: model.compat?.supportsEagerToolInputStreaming ?? true,
178		supportsLongCacheRetention: model.compat?.supportsLongCacheRetention ?? true,
179		sendSessionAffinityHeaders: model.compat?.sendSessionAffinityHeaders ?? false,
180		supportsCacheControlOnTools: model.compat?.supportsCacheControlOnTools ?? true,
181		supportsTemperature: model.compat?.supportsTemperature ?? true,
182		allowEmptySignature: model.compat?.allowEmptySignature ?? false,
183		supportsStrictTools: model.compat?.supportsStrictTools ?? false,
184		supportsToolReferences: model.compat?.supportsToolReferences ?? defaultSupportsToolReferences(model),
185	};
186}
187
188/**
189 * Default for `supportsToolReferences`: first-party Anthropic models except
190 * Haiku (rejects client-side tool_reference blocks) and models that predate
191 * tool search (Claude 3.x, Opus/Sonnet 4.0, Opus 4.1).
192 */
193function defaultSupportsToolReferences(model: Model<"anthropic-messages">): boolean {
194	if (model.provider !== "anthropic" || model.id.includes("haiku")) return false;
195	const version = model.id.match(/^claude-(?:opus|sonnet|fable)-(\d+)(?:-(\d+))?(?:-|$)/);
196	if (!version) return false;
197	const major = Number(version[1]);
198	const minor = version[2] && version[2].length < 8 ? Number(version[2]) : 0;
199	return major > 4 || (major === 4 && minor >= 5);
200}
201
202export interface AnthropicOptions extends StreamOptions {
203	/**
204	 * Enable extended thinking.
205	 * For adaptive thinking models: the model decides when/how much to think.
206	 * For older models: uses budget-based thinking with thinkingBudgetTokens.
207	 * Default: undefined (thinking is omitted unless `streamSimple()` maps
208	 * a simple reasoning level to this option, or callers set it explicitly).
209	 */
210	thinkingEnabled?: boolean;
211	/**
212	 * Token budget for extended thinking (older models only).
213	 * Ignored for adaptive thinking models.
214	 * Default: 1024 when `thinkingEnabled` is true and no budget is provided.
215	 */
216	thinkingBudgetTokens?: number;
217	/**
218	 * Effort level for adaptive thinking models.
219	 * Controls how much thinking Claude allocates:
220	 * - "max": Always thinks with no constraints (Opus 4.6 only)
221	 * - "xhigh": Highest reasoning level (Opus 4.7+, Fable 5)
222	 * - "high": Always thinks, deep reasoning
223	 * - "medium": Moderate thinking, may skip for simple queries
224	 * - "low": Minimal thinking, skips for simple tasks
225	 * Ignored for older models.
226	 * Default: omitted unless `streamSimple()` maps a simple reasoning
227	 * level to this option.
228	 */
229	effort?: AnthropicEffort;
230	/**
231	 * Controls how thinking content is returned in API responses.
232	 * - "summarized": Thinking blocks contain summarized thinking text.
233	 * - "omitted": Thinking blocks return an empty thinking field; the encrypted
234	 *   signature still travels back for multi-turn continuity. Use for faster
235	 *   time-to-first-text-token when your UI does not surface thinking.
236	 *
237	 * Note: Anthropic's API default for Claude Opus 4.7 and Claude Mythos Preview
238	 * is "omitted". We default to "summarized" here to keep behavior consistent
239	 * with older Claude 4 models. Set this explicitly to "omitted" to opt in.
240	 * Default: "summarized" when thinking is enabled.
241	 */
242	thinkingDisplay?: AnthropicThinkingDisplay;
243	/**
244	 * Whether to request the interleaved thinking beta header for non-adaptive
245	 * thinking models. Adaptive thinking models have interleaved thinking built in,
246	 * so the header is skipped for them regardless of this setting.
247	 * Default: true.
248	 */
249	interleavedThinking?: boolean;
250	/**
251	 * Anthropic tool choice behavior. String values map to Anthropic's built-in
252	 * choices; `{ type: "tool", name }` forces a specific tool.
253	 * Default: omitted (Anthropic default behavior, currently equivalent to auto).
254	 */
255	toolChoice?: "auto" | "any" | "none" | { type: "tool"; name: string };
256	/**
257	 * Pre-built Anthropic client instance. When provided, skips internal client
258	 * construction entirely. Use this to inject alternative SDK clients such as
259	 * `AnthropicVertex` that shares the same messaging API.
260	 */
261	client?: Anthropic;
262}
263
264function mergeHeaders(...headerSources: (ProviderHeaders | undefined)[]): ProviderHeaders {
265	const merged: ProviderHeaders = {};
266	for (const headers of headerSources) {
267		if (headers) {
268			Object.assign(merged, headers);
269		}
270	}
271	return merged;
272}
273
274function hasHeader(headers: ProviderHeaders | undefined, name: string): boolean {
275	if (!headers) return false;
276	const expected = name.toLowerCase();
277	for (const [key, value] of Object.entries(headers)) {
278		if (key.toLowerCase() === expected && value !== null && value.trim().length > 0) return true;
279	}
280	return false;
281}
282
283function assertRequestAuth(provider: string, apiKey: string | undefined, headers: ProviderHeaders | undefined): void {
284	if (apiKey) return;
285	if (
286		hasHeader(headers, "authorization") ||
287		hasHeader(headers, "x-api-key") ||
288		hasHeader(headers, "cf-aig-authorization")
289	) {
290		return;
291	}
292	throw new Error(`No API key for provider: ${provider}`);
293}
294
295interface ServerSentEvent {
296	event: string | null;
297	data: string;
298	raw: string[];
299}
300
301interface SseDecoderState {
302	event: string | null;
303	data: string[];
304	raw: string[];
305}
306
307const ANTHROPIC_MESSAGE_EVENTS: ReadonlySet<string> = new Set([
308	"message_start",
309	"message_delta",
310	"message_stop",
311	"content_block_start",
312	"content_block_delta",
313	"content_block_stop",
314]);
315
316function flushSseEvent(state: SseDecoderState): ServerSentEvent | null {
317	if (!state.event && state.data.length === 0) {
318		return null;
319	}
320
321	const event: ServerSentEvent = {
322		event: state.event,
323		data: state.data.join("\n"),
324		raw: [...state.raw],
325	};
326	state.event = null;
327	state.data = [];
328	state.raw = [];
329	return event;
330}
331
332function decodeSseLine(line: string, state: SseDecoderState): ServerSentEvent | null {
333	if (line === "") {
334		return flushSseEvent(state);
335	}
336
337	state.raw.push(line);
338	if (line.startsWith(":")) {
339		return null;
340	}
341
342	const delimiterIndex = line.indexOf(":");
343	const fieldName = delimiterIndex === -1 ? line : line.slice(0, delimiterIndex);
344	let value = delimiterIndex === -1 ? "" : line.slice(delimiterIndex + 1);
345	if (value.startsWith(" ")) {
346		value = value.slice(1);
347	}
348
349	if (fieldName === "event") {
350		state.event = value;
351	} else if (fieldName === "data") {
352		state.data.push(value);
353	}
354
355	return null;
356}
357
358function nextLineBreakIndex(text: string): number {
359	const carriageReturnIndex = text.indexOf("\r");
360	const newlineIndex = text.indexOf("\n");
361	if (carriageReturnIndex === -1) {
362		return newlineIndex;
363	}
364	if (newlineIndex === -1) {
365		return carriageReturnIndex;
366	}
367	return Math.min(carriageReturnIndex, newlineIndex);
368}
369
370function consumeLine(text: string): { line: string; rest: string } | null {
371	const lineBreakIndex = nextLineBreakIndex(text);
372	if (lineBreakIndex === -1) {
373		return null;
374	}
375
376	let nextIndex = lineBreakIndex + 1;
377	if (text[lineBreakIndex] === "\r" && text[nextIndex] === "\n") {
378		nextIndex += 1;
379	}
380
381	return {
382		line: text.slice(0, lineBreakIndex),
383		rest: text.slice(nextIndex),
384	};
385}
386
387async function* iterateSseMessages(
388	body: ReadableStream<Uint8Array>,
389	signal?: AbortSignal,
390): AsyncGenerator<ServerSentEvent> {
391	const reader = body.getReader();
392	const decoder = new TextDecoder();
393	const state: SseDecoderState = { event: null, data: [], raw: [] };
394	let buffer = "";
395
396	try {
397		while (true) {
398			if (signal?.aborted) {
399				throw new Error("Request was aborted");
400			}
401
402			const { value, done } = await reader.read();
403			if (done) {
404				break;
405			}
406
407			buffer += decoder.decode(value, { stream: true });
408			let consumed = consumeLine(buffer);
409			while (consumed) {
410				buffer = consumed.rest;
411				const event = decodeSseLine(consumed.line, state);
412				if (event) {
413					yield event;
414				}
415				consumed = consumeLine(buffer);
416			}
417		}
418
419		buffer += decoder.decode();
420		let consumed = consumeLine(buffer);
421		while (consumed) {
422			buffer = consumed.rest;
423			const event = decodeSseLine(consumed.line, state);
424			if (event) {
425				yield event;
426			}
427			consumed = consumeLine(buffer);
428		}
429
430		if (buffer.length > 0) {
431			const event = decodeSseLine(buffer, state);
432			if (event) {
433				yield event;
434			}
435		}
436
437		const trailingEvent = flushSseEvent(state);
438		if (trailingEvent) {
439			yield trailingEvent;
440		}
441	} finally {
442		reader.releaseLock();
443	}
444}
445
446async function* iterateAnthropicEvents(
447	response: Response,
448	signal?: AbortSignal,
449): AsyncGenerator<RawMessageStreamEvent> {
450	if (!response.body) {
451		throw new Error("Attempted to iterate over an Anthropic response with no body");
452	}
453
454	let sawMessageStart = false;
455	let sawMessageEnd = false;
456
457	for await (const sse of iterateSseMessages(response.body, signal)) {
458		if (sse.event === "error") {
459			throw new Error(sse.data);
460		}
461
462		if (!ANTHROPIC_MESSAGE_EVENTS.has(sse.event ?? "")) {
463			continue;
464		}
465
466		try {
467			const event = parseJsonWithRepair<RawMessageStreamEvent>(sse.data);
468			if (event.type === "message_start") {
469				sawMessageStart = true;
470			} else if (event.type === "message_stop") {
471				sawMessageEnd = true;
472			}
473			yield event;
474		} catch (error) {
475			const message = error instanceof Error ? error.message : String(error);
476			throw new Error(
477				`Could not parse Anthropic SSE event ${sse.event}: ${message}; data=${sse.data}; raw=${sse.raw.join("\\n")}`,
478			);
479		}
480	}
481
482	if (sawMessageStart && !sawMessageEnd) {
483		throw new Error("Anthropic stream ended before message_stop");
484	}
485}
486
487export const stream: StreamFunction<"anthropic-messages", AnthropicOptions> = (
488	model: Model<"anthropic-messages">,
489	context: Context,
490	options?: AnthropicOptions,
491): AssistantMessageEventStream => {
492	const stream = new AssistantMessageEventStream();
493
494	(async () => {
495		const output: AssistantMessage = {
496			role: "assistant",
497			content: [],
498			api: model.api as Api,
499			provider: model.provider,
500			model: model.id,
501			usage: {
502				input: 0,
503				output: 0,
504				cacheRead: 0,
505				cacheWrite: 0,
506				totalTokens: 0,
507				cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
508			},
509			stopReason: "pending",
510			timestamp: Date.now(),
511		};
512
513		try {
514			let client: Anthropic;
515			let isOAuth: boolean;
516
517			if (options?.client) {
518				client = options.client;
519				isOAuth = false;
520			} else {
521				const apiKey = options?.apiKey;
522				assertRequestAuth(model.provider, apiKey, options?.headers);
523
524				let copilotDynamicHeaders: Record<string, string> | undefined;
525				if (model.provider === "github-copilot") {
526					const hasImages = hasCopilotVisionInput(context.messages);
527					copilotDynamicHeaders = buildCopilotDynamicHeaders({
528						messages: context.messages,
529						hasImages,
530					});
531				}
532
533				const cacheRetention = resolveCacheRetention(options?.cacheRetention, options?.env);
534				const cacheSessionId = cacheRetention === "none" ? undefined : options?.sessionId;
535
536				const created = createClient(
537					model,
538					apiKey,
539					options?.interleavedThinking ?? true,
540					shouldUseFineGrainedToolStreamingBeta(model, context),
541					options?.headers,
542					copilotDynamicHeaders,
543					cacheSessionId,
544				);
545				client = created.client;
546				isOAuth = created.isOAuthToken;
547			}
548			let params = buildParams(model, context, isOAuth, options);
549			const nextParams = await options?.onPayload?.(params, model);
550			if (nextParams !== undefined) {
551				params = nextParams as MessageCreateParamsStreaming;
552			}
553			const requestOptions = {
554				...(options?.signal ? { signal: options.signal } : {}),
555				...(options?.timeoutMs !== undefined ? { timeout: options.timeoutMs } : {}),
556				maxRetries: 0,
557			};
558			const response = await retryProviderRequest(
559				() => client.messages.create({ ...params, stream: true }, requestOptions).asResponse(),
560				{
561					maxRetries: options?.maxRetries,
562					maxRetryDelayMs: options?.maxRetryDelayMs,
563					signal: options?.signal,
564				},
565			);
566			await options?.onResponse?.({ status: response.status, headers: headersToRecord(response.headers) }, model);
567			stream.push({ type: "start", partial: output });
568
569			type Block = (ThinkingContent | TextContent | (ToolCall & { partialJson: string })) & { index: number };
570			const blocks = output.content as Block[];
571
572			for await (const event of iterateAnthropicEvents(response, options?.signal)) {
573				if (event.type === "message_start") {
574					output.responseId = event.message.id;
575					// Capture initial token usage from message_start event
576					// This ensures we have input token counts even if the stream is aborted early
577					output.usage.input = event.message.usage.input_tokens || 0;
578					output.usage.output = event.message.usage.output_tokens || 0;
579					output.usage.cacheRead = event.message.usage.cache_read_input_tokens || 0;
580					output.usage.cacheWrite = event.message.usage.cache_creation_input_tokens || 0;
581					output.usage.cacheWrite1h = event.message.usage.cache_creation?.ephemeral_1h_input_tokens || 0;
582					// Anthropic doesn't provide total_tokens, compute from components
583					output.usage.totalTokens =
584						output.usage.input + output.usage.output + output.usage.cacheRead + output.usage.cacheWrite;
585					calculateCost(model, output.usage);
586				} else if (event.type === "content_block_start") {
587					if (event.content_block.type === "text") {
588						const block: Block = {
589							type: "text",
590							text: "",
591							index: event.index,
592						};
593						output.content.push(block);
594						stream.push({ type: "text_start", contentIndex: output.content.length - 1, partial: output });
595					} else if (event.content_block.type === "thinking") {
596						const block: Block = {
597							type: "thinking",
598							thinking: "",
599							thinkingSignature: "",
600							index: event.index,
601						};
602						output.content.push(block);
603						stream.push({ type: "thinking_start", contentIndex: output.content.length - 1, partial: output });
604					} else if (event.content_block.type === "redacted_thinking") {
605						const block: Block = {
606							type: "thinking",
607							thinking: "[Reasoning redacted]",
608							thinkingSignature: event.content_block.data,
609							redacted: true,
610							index: event.index,
611						};
612						output.content.push(block);
613						stream.push({ type: "thinking_start", contentIndex: output.content.length - 1, partial: output });
614					} else if (event.content_block.type === "tool_use") {
615						const block: Block = {
616							type: "toolCall",
617							id: event.content_block.id,
618							name: isOAuth
619								? fromClaudeCodeName(event.content_block.name, context.tools)
620								: event.content_block.name,
621							arguments: (event.content_block.input as Record<string, any>) ?? {},
622							partialJson: "",
623							index: event.index,
624						};
625						output.content.push(block);
626						stream.push({ type: "toolcall_start", contentIndex: output.content.length - 1, partial: output });
627					}
628				} else if (event.type === "content_block_delta") {
629					if (event.delta.type === "text_delta") {
630						const index = blocks.findIndex((b) => b.index === event.index);
631						const block = blocks[index];
632						if (block && block.type === "text") {
633							block.text += event.delta.text;
634							stream.push({
635								type: "text_delta",
636								contentIndex: index,
637								delta: event.delta.text,
638								partial: output,
639							});
640						}
641					} else if (event.delta.type === "thinking_delta") {
642						const index = blocks.findIndex((b) => b.index === event.index);
643						const block = blocks[index];
644						if (block && block.type === "thinking") {
645							block.thinking += event.delta.thinking;
646							stream.push({
647								type: "thinking_delta",
648								contentIndex: index,
649								delta: event.delta.thinking,
650								partial: output,
651							});
652						}
653					} else if (event.delta.type === "input_json_delta") {
654						const index = blocks.findIndex((b) => b.index === event.index);
655						const block = blocks[index];
656						if (block && block.type === "toolCall") {
657							block.partialJson += event.delta.partial_json;
658							block.arguments = parseStreamingJson(block.partialJson);
659							stream.push({
660								type: "toolcall_delta",
661								contentIndex: index,
662								delta: event.delta.partial_json,
663								partial: output,
664							});
665						}
666					} else if (event.delta.type === "signature_delta") {
667						const index = blocks.findIndex((b) => b.index === event.index);
668						const block = blocks[index];
669						if (block && block.type === "thinking") {
670							block.thinkingSignature = block.thinkingSignature || "";
671							block.thinkingSignature += event.delta.signature;
672						}
673					}
674				} else if (event.type === "content_block_stop") {
675					const index = blocks.findIndex((b) => b.index === event.index);
676					const block = blocks[index];
677					if (block) {
678						delete (block as any).index;
679						if (block.type === "text") {
680							stream.push({
681								type: "text_end",
682								contentIndex: index,
683								content: block.text,
684								partial: output,
685							});
686						} else if (block.type === "thinking") {
687							stream.push({
688								type: "thinking_end",
689								contentIndex: index,
690								content: block.thinking,
691								partial: output,
692							});
693						} else if (block.type === "toolCall") {
694							block.arguments = parseStreamingJson(block.partialJson);
695							// Finalize in-place and strip the scratch buffer so replay only
696							// carries parsed arguments.
697							delete (block as { partialJson?: string }).partialJson;
698							stream.push({
699								type: "toolcall_end",
700								contentIndex: index,
701								toolCall: block,
702								partial: output,
703							});
704						}
705					}
706				} else if (event.type === "message_delta") {
707					if (event.delta.stop_reason) {
708						const stopReasonResult = mapStopReason(event.delta.stop_reason, event.delta.stop_details);
709						output.stopReason = stopReasonResult.stopReason;
710						if (stopReasonResult.errorMessage) {
711							output.errorMessage = stopReasonResult.errorMessage;
712						}
713					}
714					// Only update usage fields if present (not null).
715					// Preserves input_tokens from message_start when proxies omit it in message_delta.
716					if (event.usage) {
717						if (event.usage.input_tokens != null) {
718							output.usage.input = event.usage.input_tokens;
719						}
720						if (event.usage.output_tokens != null) {
721							output.usage.output = event.usage.output_tokens;
722						}
723						if (event.usage.cache_read_input_tokens != null) {
724							output.usage.cacheRead = event.usage.cache_read_input_tokens;
725						}
726						if (event.usage.cache_creation_input_tokens != null) {
727							output.usage.cacheWrite = event.usage.cache_creation_input_tokens;
728						}
729						// Anthropic reports reasoning tokens in `output_tokens_details.thinking_tokens` on the
730						// final message_delta usage (a subset of output_tokens). SDK 0.91.1 omits the field from
731						// its Usage type, so read it through a narrow cast. Verified against the live API.
732						const thinkingTokens = (event.usage as { output_tokens_details?: { thinking_tokens?: number } })
733							.output_tokens_details?.thinking_tokens;
734						if (thinkingTokens != null) {
735							output.usage.reasoning = thinkingTokens;
736						}
737					}
738					// Anthropic doesn't provide total_tokens, compute from components
739					output.usage.totalTokens =
740						output.usage.input + output.usage.output + output.usage.cacheRead + output.usage.cacheWrite;
741					calculateCost(model, output.usage);
742				}
743			}
744
745			if (options?.signal?.aborted) {
746				throw new Error("Request was aborted");
747			}
748
749			if (output.stopReason === "pending") {
750				throw new Error("Anthropic stream ended without a stop reason");
751			}
752			if (output.stopReason === "aborted" || output.stopReason === "error") {
753				throw new Error(output.errorMessage || "An unknown error occurred");
754			}
755
756			stream.push({ type: "done", reason: output.stopReason, message: output });
757			stream.end();
758		} catch (error) {
759			for (const block of output.content) {
760				delete (block as { index?: number }).index;
761				// partialJson is only a streaming scratch buffer; never persist it.
762				delete (block as { partialJson?: string }).partialJson;
763			}
764			output.stopReason = options?.signal?.aborted ? "aborted" : "error";
765			output.errorMessage = error instanceof Error ? error.message : JSON.stringify(error);
766			stream.push({ type: "error", reason: output.stopReason, error: output });
767			stream.end();
768		}
769	})();
770
771	return stream;
772};
773
774/**
775 * Map ThinkingLevel to Anthropic effort levels for adaptive thinking.
776 * Note: effort "max" is available on all adaptive-thinking Claude models, while native
777 * "xhigh" is only available on Opus 4.7/4.8, Sonnet 5, and Fable 5.
778 */
779function mapThinkingLevelToEffort(
780	model: Model<"anthropic-messages">,
781	level: SimpleStreamOptions["reasoning"],
782): AnthropicEffort {
783	const mapped = level ? model.thinkingLevelMap?.[level] : undefined;
784	if (typeof mapped === "string") return mapped as AnthropicEffort;
785
786	switch (level) {
787		case "minimal":
788		case "low":
789			return "low";
790		case "medium":
791			return "medium";
792		case "high":
793			return "high";
794		default:
795			return "high";
796	}
797}
798
799export const streamSimple: StreamFunction<"anthropic-messages", SimpleStreamOptions> = (
800	model: Model<"anthropic-messages">,
801	context: Context,
802	options?: SimpleStreamOptions,
803): AssistantMessageEventStream => {
804	assertRequestAuth(model.provider, options?.apiKey, options?.headers);
805
806	const base = buildBaseOptions(model, context, options, options?.apiKey);
807	if (!options?.reasoning) {
808		return stream(model, context, { ...base, thinkingEnabled: false } satisfies AnthropicOptions);
809	}
810
811	// For models with adaptive thinking: use an effort level.
812	// For older models: use budget-based thinking.
813	if (model.compat?.forceAdaptiveThinking === true) {
814		const effort = mapThinkingLevelToEffort(model, options.reasoning);
815		return stream(model, context, {
816			...base,
817			thinkingEnabled: true,
818			effort,
819		} satisfies AnthropicOptions);
820	}
821
822	// Undefined means the caller did not request an output cap; let the helper use the model cap.
823	// Do not coerce to 0 here, or the thinking budget would become the entire max_tokens value.
824	const adjusted = adjustMaxTokensForThinking(
825		base.maxTokens,
826		model.maxTokens,
827		options.reasoning,
828		options.thinkingBudgets,
829	);
830
831	const maxTokens = clampMaxTokensToContext(model, context, adjusted.maxTokens);
832
833	return stream(model, context, {
834		...base,
835		maxTokens,
836		thinkingEnabled: true,
837		thinkingBudgetTokens: Math.min(adjusted.thinkingBudget, Math.max(0, maxTokens - 1024)),
838	} satisfies AnthropicOptions);
839};
840
841function isOAuthToken(apiKey: string): boolean {
842	return apiKey.includes("sk-ant-oat");
843}
844
845function createClient(
846	model: Model<"anthropic-messages">,
847	apiKey: string | undefined,
848	interleavedThinking: boolean,
849	useFineGrainedToolStreamingBeta: boolean,
850	optionsHeaders?: ProviderHeaders,
851	dynamicHeaders?: Record<string, string>,
852	sessionId?: string,
853): { client: Anthropic; isOAuthToken: boolean } {
854	// Adaptive thinking models have interleaved thinking built in, so skip the beta header.
855	const needsInterleavedBeta = interleavedThinking && model.compat?.forceAdaptiveThinking !== true;
856	const betaFeatures: string[] = [];
857	if (useFineGrainedToolStreamingBeta) {
858		betaFeatures.push(FINE_GRAINED_TOOL_STREAMING_BETA);
859	}
860	if (needsInterleavedBeta) {
861		betaFeatures.push(INTERLEAVED_THINKING_BETA);
862	}
863
864	// Copilot: Bearer auth, selective betas.
865	if (model.provider === "github-copilot") {
866		const client = new Anthropic({
867			apiKey: null,
868			authToken: apiKey ?? null,
869			baseURL: model.baseUrl,
870			dangerouslyAllowBrowser: true,
871			defaultHeaders: mergeHeaders(
872				{
873					accept: "application/json",
874					"anthropic-dangerous-direct-browser-access": "true",
875					...(betaFeatures.length > 0 ? { "anthropic-beta": betaFeatures.join(",") } : {}),
876				},
877				model.headers,
878				dynamicHeaders,
879				optionsHeaders,
880			),
881		});
882
883		return { client, isOAuthToken: false };
884	}
885
886	// OAuth: Bearer auth, Claude Code identity headers
887	if (apiKey && isOAuthToken(apiKey)) {
888		const client = new Anthropic({
889			apiKey: null,
890			authToken: apiKey,
891			baseURL: model.baseUrl,
892			dangerouslyAllowBrowser: true,
893			defaultHeaders: mergeHeaders(
894				{
895					accept: "application/json",
896					"anthropic-dangerous-direct-browser-access": "true",
897					"anthropic-beta": ["claude-code-20250219", "oauth-2025-04-20", ...betaFeatures].join(","),
898					"user-agent": `claude-cli/${claudeCodeVersion}`,
899					"x-app": "cli",
900				},
901				model.headers,
902				optionsHeaders,
903			),
904		});
905
906		return { client, isOAuthToken: true };
907	}
908
909	// API key or header-owned auth.
910	const sessionAffinityHeaders: ProviderHeaders =
911		sessionId && getAnthropicCompat(model).sendSessionAffinityHeaders ? { "x-session-affinity": sessionId } : {};
912	const defaultHeaders = mergeHeaders(
913		{
914			accept: "application/json",
915			"anthropic-dangerous-direct-browser-access": "true",
916			...(betaFeatures.length > 0 ? { "anthropic-beta": betaFeatures.join(",") } : {}),
917		},
918		sessionAffinityHeaders,
919		model.headers,
920		optionsHeaders,
921	);
922	const client = new Anthropic({
923		apiKey: apiKey ?? null,
924		authToken: null,
925		baseURL: model.baseUrl,
926		dangerouslyAllowBrowser: true,
927		defaultHeaders,
928	});
929
930	return { client, isOAuthToken: false };
931}
932
933function buildParams(
934	model: Model<"anthropic-messages">,
935	context: Context,
936	isOAuthToken: boolean,
937	options?: AnthropicOptions,
938): MessageCreateParamsStreaming {
939	const { cacheControl } = getCacheControl(model, options?.cacheRetention, options?.env);
940	const compat = getAnthropicCompat(model);
941	const transformedMessages = transformMessages(context.messages, model, normalizeToolCallId);
942	const normalizeToolName = isOAuthToken ? toClaudeCodeName : (name: string) => name;
943	const toolPlacement = splitDeferredTools(
944		{ ...context, messages: transformedMessages },
945		compat.supportsToolReferences,
946		normalizeToolName,
947	);
948	let immediateTools = toolPlacement.immediate;
949	let deferredTools = [...toolPlacement.deferred.values()];
950	if (immediateTools.length === 0 && deferredTools.length > 0) {
951		immediateTools = deferredTools;
952		deferredTools = [];
953	}
954	const deferredToolNames = new Set(deferredTools.map((tool) => normalizeToolName(tool.name)));
955	const params: MessageCreateParamsStreaming = {
956		model: model.id,
957		messages: convertMessages(
958			transformedMessages,
959			isOAuthToken,
960			cacheControl,
961			compat.allowEmptySignature,
962			deferredToolNames,
963			normalizeToolName,
964		),
965		max_tokens: options?.maxTokens ?? model.maxTokens,
966		stream: true,
967	};
968
969	// For OAuth tokens, we MUST include Claude Code identity
970	if (isOAuthToken) {
971		params.system = [
972			{
973				type: "text",
974				text: "You are Claude Code, Anthropic's official CLI for Claude.",
975				...(cacheControl ? { cache_control: cacheControl } : {}),
976			},
977		];
978		if (context.systemPrompt) {
979			params.system.push({
980				type: "text",
981				text: sanitizeSurrogates(context.systemPrompt),
982				...(cacheControl ? { cache_control: cacheControl } : {}),
983			});
984		}
985	} else if (context.systemPrompt) {
986		// Add cache control to system prompt for non-OAuth tokens
987		params.system = [
988			{
989				type: "text",
990				text: sanitizeSurrogates(context.systemPrompt),
991				...(cacheControl ? { cache_control: cacheControl } : {}),
992			},
993		];
994	}
995
996	// Temperature is incompatible with extended thinking and unsupported on Claude Opus 4.7+.
997	if (options?.temperature !== undefined && !options?.thinkingEnabled && compat.supportsTemperature) {
998		params.temperature = options.temperature;
999	}
1000
1001	if (immediateTools.length > 0 || deferredTools.length > 0) {
1002		params.tools = [
1003			...convertTools(
1004				immediateTools,
1005				isOAuthToken,
1006				compat.supportsEagerToolInputStreaming,
1007				compat.supportsStrictTools,
1008				compat.supportsCacheControlOnTools ? cacheControl : undefined,
1009			),
1010			...convertTools(
1011				deferredTools,
1012				isOAuthToken,
1013				compat.supportsEagerToolInputStreaming,
1014				compat.supportsStrictTools,
1015				undefined,
1016				true,
1017			),
1018		];
1019	}
1020
1021	// Configure thinking mode: adaptive, budget-based, or explicitly disabled.
1022	if (model.reasoning) {
1023		if (options?.thinkingEnabled) {
1024			// Default to "summarized" so Opus 4.7 and Mythos Preview behave like
1025			// older Claude 4 models (whose API default is also "summarized").
1026			const display: AnthropicThinkingDisplay = options.thinkingDisplay ?? "summarized";
1027			if (model.compat?.forceAdaptiveThinking === true) {
1028				// Adaptive thinking: Claude decides when and how much to think.
1029				params.thinking = { type: "adaptive", display };
1030				if (options.effort) {
1031					// The Anthropic SDK types can lag newly supported effort values such as "xhigh".
1032					params.output_config =
1033						options.effort === "xhigh"
1034							? ({ effort: options.effort } as unknown as NonNullable<
1035									MessageCreateParamsStreaming["output_config"]
1036								>)
1037							: { effort: options.effort };
1038				}
1039			} else {
1040				// Budget-based thinking for older models
1041				params.thinking = {
1042					type: "enabled",
1043					budget_tokens: options.thinkingBudgetTokens || 1024,
1044					display,
1045				};
1046			}
1047		} else if (options?.thinkingEnabled === false && model.thinkingLevelMap?.off !== null) {
1048			params.thinking = { type: "disabled" };
1049		}
1050	}
1051
1052	if (options?.metadata) {
1053		const userId = options.metadata.user_id;
1054		if (typeof userId === "string") {
1055			params.metadata = { user_id: userId };
1056		}
1057	}
1058
1059	if (options?.toolChoice) {
1060		if (typeof options.toolChoice === "string") {
1061			params.tool_choice = { type: options.toolChoice };
1062		} else {
1063			params.tool_choice = options.toolChoice;
1064		}
1065	}
1066
1067	return params;
1068}
1069
1070// Normalize tool call IDs to match Anthropic's required pattern and length
1071function normalizeToolCallId(id: string): string {
1072	return id.replace(/[^a-zA-Z0-9_-]/g, "_").slice(0, 64);
1073}
1074
1075function convertToolResult(
1076	msg: ToolResultMessage,
1077	isOAuthToken: boolean,
1078	deferredToolNames: ReadonlySet<string>,
1079	loadedToolNames: Set<string>,
1080	normalizeToolName: (name: string) => string,
1081): { toolResult: ContentBlockParam; siblingContent: ContentBlockParam[] } {
1082	const references: Array<{ type: "tool_reference"; tool_name: string }> = [];
1083	for (const name of msg.addedToolNames ?? []) {
1084		const normalizedName = normalizeToolName(name);
1085		if (!deferredToolNames.has(normalizedName) || loadedToolNames.has(normalizedName)) continue;
1086		loadedToolNames.add(normalizedName);
1087		references.push({
1088			type: "tool_reference",
1089			tool_name: isOAuthToken ? toClaudeCodeName(name) : name,
1090		});
1091	}
1092	const convertedContent = convertContentBlocks(msg.content);
1093	// Anthropic rejects tool references mixed with ordinary tool-result content.
1094	return {
1095		toolResult: {
1096			type: "tool_result",
1097			tool_use_id: msg.toolCallId,
1098			content: references.length > 0 ? references : convertedContent,
1099			is_error: msg.isError,
1100		},
1101		siblingContent:
1102			references.length === 0
1103				? []
1104				: typeof convertedContent === "string"
1105					? [{ type: "text", text: convertedContent }]
1106					: convertedContent,
1107	};
1108}
1109
1110function convertMessages(
1111	transformedMessages: Message[],
1112	isOAuthToken: boolean,
1113	cacheControl?: CacheControlEphemeral,
1114	allowEmptySignature = false,
1115	deferredToolNames: ReadonlySet<string> = new Set(),
1116	normalizeToolName: (name: string) => string = (name) => name,
1117): MessageParam[] {
1118	const params: MessageParam[] = [];
1119	const loadedToolNames = new Set<string>();
1120
1121	for (let i = 0; i < transformedMessages.length; i++) {
1122		const msg = transformedMessages[i];
1123
1124		if (msg.role === "user") {
1125			if (typeof msg.content === "string") {
1126				if (msg.content.trim().length > 0) {
1127					params.push({
1128						role: "user",
1129						content: sanitizeSurrogates(msg.content),
1130					});
1131				}
1132			} else {
1133				const blocks: ContentBlockParam[] = msg.content.map((item) => {
1134					if (item.type === "text") {
1135						return {
1136							type: "text",
1137							text: sanitizeSurrogates(item.text),
1138						};
1139					} else {
1140						return {
1141							type: "image",
1142							source: {
1143								type: "base64",
1144								media_type: item.mimeType as "image/jpeg" | "image/png" | "image/gif" | "image/webp",
1145								data: item.data,
1146							},
1147						};
1148					}
1149				});
1150				const filteredBlocks = blocks.filter((b) => {
1151					if (b.type === "text") {
1152						return b.text.trim().length > 0;
1153					}
1154					return true;
1155				});
1156				if (filteredBlocks.length === 0) continue;
1157				params.push({
1158					role: "user",
1159					content: filteredBlocks,
1160				});
1161			}
1162		} else if (msg.role === "assistant") {
1163			const blocks: ContentBlockParam[] = [];
1164
1165			for (const block of msg.content) {
1166				if (block.type === "text") {
1167					if (block.text.trim().length === 0) continue;
1168					blocks.push({
1169						type: "text",
1170						text: sanitizeSurrogates(block.text),
1171					});
1172				} else if (block.type === "thinking") {
1173					// Redacted thinking: pass the opaque payload back as redacted_thinking
1174					if (block.redacted) {
1175						blocks.push({
1176							type: "redacted_thinking",
1177							data: block.thinkingSignature!,
1178						});
1179						continue;
1180					}
1181					const thinkingSignature = block.thinkingSignature;
1182					const hasThinkingSignature = !!thinkingSignature && thinkingSignature.trim().length > 0;
1183					if (block.thinking.trim().length === 0 && !hasThinkingSignature) continue;
1184					// If thinking signature is missing/empty (e.g., from aborted stream),
1185					// convert to plain text for Anthropic. Some compatible providers emit
1186					// and accept empty signatures, so let marked models preserve the block.
1187					if (!hasThinkingSignature) {
1188						blocks.push(
1189							allowEmptySignature
1190								? {
1191										type: "thinking",
1192										thinking: sanitizeSurrogates(block.thinking),
1193										signature: "",
1194									}
1195								: {
1196										type: "text",
1197										text: sanitizeSurrogates(block.thinking),
1198									},
1199						);
1200					} else {
1201						blocks.push({
1202							type: "thinking",
1203							thinking: sanitizeSurrogates(block.thinking),
1204							signature: thinkingSignature,
1205						});
1206					}
1207				} else if (block.type === "toolCall") {
1208					blocks.push({
1209						type: "tool_use",
1210						id: block.id,
1211						name: isOAuthToken ? toClaudeCodeName(block.name) : block.name,
1212						input: block.arguments ?? {},
1213					});
1214				}
1215			}
1216			if (blocks.length === 0) continue;
1217			params.push({
1218				role: "assistant",
1219				content: blocks,
1220			});
1221		} else if (msg.role === "toolResult") {
1222			// Collect all consecutive toolResult messages, needed for z.ai Anthropic endpoint.
1223			const toolResults: ContentBlockParam[] = [];
1224			const siblingContent: ContentBlockParam[] = [];
1225			let j = i;
1226			while (j < transformedMessages.length && transformedMessages[j].role === "toolResult") {
1227				const converted = convertToolResult(
1228					transformedMessages[j] as ToolResultMessage,
1229					isOAuthToken,
1230					deferredToolNames,
1231					loadedToolNames,
1232					normalizeToolName,
1233				);
1234				toolResults.push(converted.toolResult);
1235				siblingContent.push(...converted.siblingContent);
1236				j++;
1237			}
1238
1239			// Skip the messages we've already processed.
1240			i = j - 1;
1241
1242			// Displaced reference-bearing results must follow every tool_result block.
1243			params.push({
1244				role: "user",
1245				content: [...toolResults, ...siblingContent],
1246			});
1247		}
1248	}
1249
1250	// Add cache_control to the last user message to cache conversation history
1251	if (cacheControl && params.length > 0) {
1252		const lastMessage = params[params.length - 1];
1253		if (lastMessage.role === "user") {
1254			if (Array.isArray(lastMessage.content)) {
1255				const lastBlock = lastMessage.content[lastMessage.content.length - 1];
1256				if (
1257					lastBlock &&
1258					(lastBlock.type === "text" || lastBlock.type === "image" || lastBlock.type === "tool_result")
1259				) {
1260					(lastBlock as any).cache_control = cacheControl;
1261				}
1262			} else if (typeof lastMessage.content === "string") {
1263				lastMessage.content = [
1264					{
1265						type: "text",
1266						text: lastMessage.content,
1267						cache_control: cacheControl,
1268					},
1269				] as any;
1270			}
1271		}
1272	}
1273
1274	return params;
1275}
1276
1277function shouldUseFineGrainedToolStreamingBeta(model: Model<"anthropic-messages">, context: Context): boolean {
1278	return !!context.tools?.length && !getAnthropicCompat(model).supportsEagerToolInputStreaming;
1279}
1280
1281function convertTools(
1282	tools: Tool[],
1283	isOAuthToken: boolean,
1284	supportsEagerToolInputStreaming: boolean,
1285	supportsStrictTools: boolean,
1286	cacheControl?: CacheControlEphemeral,
1287	deferLoading = false,
1288): Anthropic.Messages.Tool[] {
1289	if (!tools) return [];
1290
1291	return tools.map((tool, index) => {
1292		const strict = resolveJsonSchemaStrictSampling(tool, supportsStrictTools);
1293		const schema = tool.parameters as { properties?: unknown; required?: string[] };
1294		const legacyInputSchema = {
1295			type: "object" as const,
1296			properties: schema.properties ?? {},
1297			required: schema.required ?? [],
1298		};
1299		const inputSchema =
1300			strict === true
1301				? {
1302						...(tool.parameters as Record<string, unknown>),
1303						...legacyInputSchema,
1304					}
1305				: legacyInputSchema;
1306
1307		return {
1308			name: isOAuthToken ? toClaudeCodeName(tool.name) : tool.name,
1309			description: tool.description,
1310			...(supportsEagerToolInputStreaming ? { eager_input_streaming: true } : {}),
1311			...(strict === true ? { strict: true } : {}),
1312			input_schema: inputSchema,
1313			...(deferLoading ? { defer_loading: true } : {}),
1314			...(cacheControl && index === tools.length - 1 ? { cache_control: cacheControl } : {}),
1315		};
1316	});
1317}
1318
1319function mapStopReason(
1320	reason: Anthropic.Messages.StopReason | string,
1321	stopDetails?: RefusalStopDetails | null,
1322): { stopReason: StopReason; errorMessage?: string } {
1323	switch (reason) {
1324		case "end_turn":
1325			return { stopReason: "stop" };
1326		case "max_tokens":
1327			return { stopReason: "length" };
1328		case "tool_use":
1329			return { stopReason: "toolUse" };
1330		case "refusal":
1331			return {
1332				stopReason: "error",
1333				errorMessage: stopDetails?.explanation || `The model refused to complete the request`,
1334			};
1335		case "pause_turn": // Stop is good enough -> resubmit
1336			return { stopReason: "stop" };
1337		case "stop_sequence":
1338			return { stopReason: "stop" }; // We don't supply stop sequences, so this should never happen
1339		case "sensitive": // Content flagged by safety filters (not yet in SDK types)
1340			return { stopReason: "error" };
1341		default:
1342			// Handle unknown stop reasons gracefully (API may add new values)
1343			throw new Error(`Unhandled stop reason: ${reason}`);
1344	}
1345}
1346