anthropic.rs - CodeWhale

anthropic.rs

1	//! Native Anthropic Messages API adapter (#3014).
2	//!
3	//! CodeWhale's internal wire types are already Anthropic-shaped (the harness
4	//! speaks Messages internally and translates out to OpenAI dialects), so
5	//! this adapter is mostly native serialization plus an SSE pass-through:
6	//! `StreamEvent` deserializes Anthropic's `message_start` /
7	//! `content_block_*` / `message_delta` / `message_stop` / `ping` events
8	//! directly. What the adapter adds on top:
9	//!
10	//! - request shaping: adaptive thinking + `output_config.effort` from
11	//! CodeWhale's `reasoning_effort` tiers, sampling-parameter rules for
12	//! models that reject them, and `cache_control` breakpoint placement
13	//! aligned with the prefix-zone model in `prefix_cache.rs`;
14	//! - usage normalization (#2961 / #4318): `prompt_cache_hit_tokens` comes from
15	//! `cache_read_input_tokens`, `prompt_cache_write_tokens` from
16	//! `cache_creation_input_tokens`, `prompt_cache_miss_tokens` is the raw
17	//! non-cached `input_tokens`, and the normalized `input_tokens` is the sum
18	//! of all three (total prompt, the DeepSeek convention);
19	//! - signed-thinking handling: `signature_delta` is captured into
20	//! [`codewhale_models::Delta::SignatureDelta`] and assistant thinking blocks
21	//! replay verbatim (signature included); unsigned thinking blocks are
22	//! dropped from replay because the API rejects them.
23	//!
24	//! Modeled on `client/responses.rs` (separate file per dialect, no protocol
25	//! hacks in the shared paths).
26
27	use anyhow::{Context, Result};
28	use serde::Deserialize;
29	use serde_json::{Value, json};
30
31	use crate::config::{ApiProvider, wire_model_for_provider_route};
32	use crate::llm_client::StreamEventBox;
33	use crate::logging;
34	use crate::tools::schema_sanitize;
35	use codewhale_models::{ContentBlock, MessageRequest, MessageResponse, StreamEvent, Usage};
36
37	use super::prepared::WireDialect;
38	use super::role_placement::{RolePlacement, role_placement};
39	use super::wire::{extract_sse_data_value, next_sse_line};
40	use super::{CodewhaleClient, ERROR_BODY_MAX_BYTES, bounded_error_text};
41
42	/// Maximum `cache_control` breakpoints Anthropic accepts per request.
43	const MAX_CACHE_BREAKPOINTS: usize = 4;
44
45	impl CodewhaleClient {
46	/// Build the native Messages API request body from a [`MessageRequest`].
47	pub(super) fn build_anthropic_body(&self, request: &MessageRequest, stream: bool) -> Value {
48	let model =
49	wire_model_for_provider_route(self.api_provider, &self.base_url, &request.model);
50	let mut body = json!({
51	"model": model,
52	"max_tokens": request.max_tokens,
53	"stream": stream,
54	});
55
56	if let Some(system) = request.system.as_ref() {
57	body["system"] = match system {
58	codewhale_models::SystemPrompt::Text(text) => json!(text),
59	codewhale_models::SystemPrompt::Blocks(blocks) => json!(
60	blocks
61	.iter()
62	.map(\|block\| {
63	let mut value = json!({
64	"type": "text",
65	"text": block.text,
66	});
67	if let Some(cache) = block.cache_control.as_ref() {
68	value["cache_control"] = json!({ "type": cache.cache_type });
69	}
70	value
71	})
72	.collect::<Vec<_>>()
73	),
74	};
75	}
76
77	let mut messages: Vec<Value> = request
78	.messages
79	.iter()
80	.filter_map(message_to_anthropic)
81	.collect();
82	repair_dangling_tool_uses(&mut messages);
83	body["messages"] = Value::Array(messages);
84
85	if let Some(tools) = request.tools.as_ref()
86	&& !tools.is_empty()
87	{
88	body["tools"] = json!(
89	tools
90	.iter()
91	.map(\|tool\| {
92	// Sanitize the tool's input_schema the same way the
93	// OpenAI Responses adapter does: strip top-level
94	// oneOf/anyOf/allOf (which Anthropic rejects), merge
95	// alternative properties into the root, and surface
96	// the dropped constraint as a description note so the
97	// model still knows which parameters are expected.
98	let mut schema = tool.input_schema.clone();
99	let constraint_note = schema_sanitize::sanitize_for_responses(&mut schema);
100	let description = match constraint_note {
101	Some(note) if tool.description.trim().is_empty() => note,
102	Some(note) => format!("{}\n\n{}", tool.description.trim(), note),
103	None => tool.description.clone(),
104	};
105	let mut value = json!({
106	"name": tool.name,
107	"description": description,
108	"input_schema": schema,
109	});
110	if let Some(strict) = tool.strict {
111	value["strict"] = json!(strict);
112	}
113	if let Some(cache) = tool.cache_control.as_ref() {
114	value["cache_control"] = json!({ "type": cache.cache_type });
115	}
116	value
117	})
118	.collect::<Vec<_>>()
119	);
120	}
121
122	if let Some(tool_choice) = request.tool_choice.as_ref() {
123	body["tool_choice"] = anthropic_tool_choice(tool_choice);
124	}
125
126	// Thinking + effort shaping. MiniMax supports adaptive/disabled but
127	// not Anthropic's output_config effort field; native Anthropic routes
128	// keep the existing effort mapping. Other Messages-compatible
129	// gateways (#4978, e.g. Sensenova) only accept the documented
130	// enabled/disabled/auto thinking types, so non-native routes get the
131	// portable `{"type":"enabled","budget_tokens":N}` shape instead.
132	let thinking_capable = codewhale_models::model_supports_reasoning(&model);
133	let is_minimax_provider = self.api_provider == ApiProvider::MinimaxAnthropic;
134	let is_minimax = crate::config::is_exact_minimax_anthropic_m3_route(
135	self.api_provider,
136	&self.base_url,
137	&model,
138	);
139	let is_deepseek = self.api_provider == ApiProvider::DeepseekAnthropic;
140	// Model Studio's Anthropic-compatible endpoint documents the portable
141	// `{"type":"enabled","budget_tokens":N}` shape AND `{"type":"disabled"}`
142	// (alibabacloud.com/help/en/model-studio/anthropic-api-messages), so
143	// an explicit "off" can be honored on the wire instead of silently
144	// falling through to the server default (which is thinking-ON for the
145	// qwen3.x families).
146	let is_modelstudio = matches!(
147	self.api_provider,
148	ApiProvider::ModelstudioTokenPlan
149	\| ApiProvider::ModelstudioTokenPlanAnthropic
150	\| ApiProvider::ModelstudioCodingPlan
151	\| ApiProvider::ModelstudioCodingPlanAnthropic
152	);
153	// MiniMax's exact M3 route and DeepSeek's Messages dialect both
154	// document adaptive support; everything else needs the native host.
155	let supports_adaptive =
156	is_native_anthropic_base_url(&self.base_url) \|\| is_minimax \|\| is_deepseek;
157	let effort = request
158	.reasoning_effort
159	.as_deref()
160	.map(\|raw\| raw.trim().to_ascii_lowercase());
161	match effort.as_deref() {
162	_ if is_minimax_provider && !is_minimax => {}
163	Some("off" \| "disabled" \| "none" \| "false")
164	if (is_minimax \|\| is_deepseek \|\| is_modelstudio) && thinking_capable =>
165	{
166	// Deliberately includes thinking-only Model Studio models
167	// (qwen3.8-max family): unlike the chat dialect's
168	// enable_thinking switch, the Messages endpoint documents the
169	// portable {"type":"disabled"} shape for them
170	// (alibabacloud.com/help/en/model-studio/anthropic-api-messages)
171	// — pinned by modelstudio_messages_body_requests_thinking_
172	// with_budget. Re-checked 2026-08-04.
173	body["thinking"] = json!({ "type": "disabled" });
174	}
175	Some("off" \| "disabled" \| "none" \| "false") => {}
176	Some(level) if thinking_capable && supports_adaptive => {
177	body["thinking"] = json!({ "type": "adaptive" });
178	if !is_minimax {
179	let mapped = match level {
180	"low" \| "minimal" => "low",
181	"medium" \| "mid" => "medium",
182	"max" \| "xhigh" \| "highest" => "max",
183	_ => "high",
184	};
185	body["output_config"] = json!({ "effort": mapped });
186	}
187	}
188	None if thinking_capable && supports_adaptive => {
189	body["thinking"] = json!({ "type": "adaptive" });
190	}
191	_ if thinking_capable => {
192	if let Some(budget) = compat_thinking_budget(effort.as_deref(), request.max_tokens)
193	{
194	body["thinking"] = json!({ "type": "enabled", "budget_tokens": budget });
195	}
196	}
197	_ => {}
198	}
199
200	// Sampling parameters: Claude 4.7+ rejects temperature/top_p
201	// entirely; earlier models reject the two together. Send at most one
202	// (temperature wins), or neither for models that forbid them.
203	if !anthropic_model_rejects_sampling(&request.model) {
204	if let Some(temperature) = request.temperature {
205	body["temperature"] = json!(temperature);
206	} else if let Some(top_p) = request.top_p {
207	body["top_p"] = json!(top_p);
208	}
209	}
210
211	apply_anthropic_cache_breakpoints(&mut body);
212	body
213	}
214
215	async fn send_anthropic_request(&self, url: &str, body: &Value) -> Result<reqwest::Response> {
216	let url = self.messages_transport_url(url);
217	self.wait_for_rate_limit().await;
218	let response = self
219	.http_client
220	.post(&url)
221	.header("Accept", "text/event-stream")
222	.json(body)
223	.send()
224	.await
225	.context("Anthropic Messages API request failed")?;
226	self.check_anthropic_response(response).await
227	}
228
229	/// Shared status/error-envelope handling for streaming and
230	/// non-streaming Messages responses.
231	async fn check_anthropic_response(
232	&self,
233	response: reqwest::Response,
234	) -> Result<reqwest::Response> {
235	let status = response.status();
236	if !status.is_success() {
237	let raw = bounded_error_text(response, ERROR_BODY_MAX_BYTES).await;
238	let (error_type, message) = parse_anthropic_error_envelope(&raw);
239	self.mark_request_failure(&format!("anthropic status={status}"))
240	.await;
241	anyhow::bail!("Anthropic API error (HTTP {status} {error_type}): {message}");
242	}
243	self.mark_request_success().await;
244	Ok(response)
245	}
246
247	/// Open the streaming Messages request through the shared stream-entry
248	/// transport policy: bounded header wait, dual-client selection, and at
249	/// most one HTTP/1.1 fallback retry on a classified H2 header stall.
250	/// Wire-specific request construction (headers, endpoint, body) stays
251	/// here at the adapter edge.
252	async fn open_anthropic_stream_response(
253	&self,
254	url: &str,
255	body: &Value,
256	) -> Result<reqwest::Response> {
257	let url = self.messages_transport_url(url);
258	let open_req = super::stream_entry::StreamOpenRequest::new(
259	super::stream_entry::stream_open_timeout(),
260	self.stream_idle_timeout,
261	);
262	let opened = super::stream_entry::open_sse_response(&open_req, \|policy\| {
263	let url = url.clone();
264	async move {
265	self.wait_for_rate_limit().await;
266	let client = super::stream_entry::client_for_policy(
267	&self.http_client,
268	self.http1_fallback_client(),
269	policy,
270	);
271	client
272	.post(&url)
273	.header("Accept", "text/event-stream")
274	.json(body)
275	.send()
276	.await
277	.context("Anthropic Messages API request failed")
278	}
279	})
280	.await;
281	let response = match opened {
282	Ok(response) => response,
283	Err(err) => {
284	self.mark_request_failure(&format!("anthropic stream open: {err}"))
285	.await;
286	return Err(err);
287	}
288	};
289	self.check_anthropic_response(response).await
290	}
291
292	/// Handle a streaming Messages API request.
293	pub(super) async fn handle_anthropic_stream(
294	&self,
295	prepared: &super::PreparedOutboundRequest,
296	) -> Result<StreamEventBox> {
297	// Body and endpoint come from the shared prepared-request seam
298	// (`prepare_outbound_request`), never from a second builder.
299	let body = &prepared.body;
300	let response = self
301	.open_anthropic_stream_response(&prepared.endpoint.url, body)
302	.await?;
303
304	let stream_idle_timeout = self.stream_idle_timeout;
305	let byte_stream = response.bytes_stream();
306
307	let stream = async_stream::stream! {
308	use futures_util::StreamExt;
309
310	// Raw byte buffer: decode only COMPLETE lines (or the stream-end
311	// tail) via the shared take_sse_line / flush_sse_line helpers so a
312	// multi-byte UTF-8 char (CJK/emoji) split across HTTP/2 DATA is
313	// never corrupted to U+FFFD. Genuine invalid bytes fail closed.
314	let mut buffer: Vec<u8> = Vec::new();
315	let stream_start = std::time::Instant::now();
316	let mut last_chunk_at = std::time::Instant::now();
317	let mut bytes_received: usize = 0;
318	let mut ended = false;
319	tokio::pin!(byte_stream);
320
321	loop {
322	if !ended {
323	match tokio::time::timeout(stream_idle_timeout, byte_stream.next()).await {
324	Ok(Some(Ok(chunk))) => {
325	bytes_received += chunk.len();
326	last_chunk_at = std::time::Instant::now();
327	buffer.extend_from_slice(&chunk);
328	}
329	Ok(Some(Err(e))) => {
330	yield Err(anyhow::anyhow!("Stream read error: {e}"));
331	return;
332	}
333	Ok(None) => ended = true,
334	Err(_) => {
335	yield Err(anyhow::anyhow!(super::stream_entry::idle_timeout_message(
336	stream_idle_timeout,
337	bytes_received,
338	stream_start.elapsed(),
339	last_chunk_at.elapsed(),
340	)));
341	return;
342	}
343	}
344	}
345
346	loop {
347	let line = match next_sse_line(&mut buffer, ended) {
348	Ok(Some(line)) => line,
349	Ok(None) => break,
350	Err(err) => {
351	yield Err(anyhow::anyhow!("{err}"));
352	return;
353	}
354	};
355
356	// `event:` lines are redundant (the data payload carries
357	// `type`) and comment/heartbeat lines are ignorable.
358	let Some(data) = extract_sse_data_value(&line) else {
359	continue;
360	};
361
362	match convert_anthropic_sse_data(data) {
363	Some(Ok(StreamEvent::Error { error })) => {
364	let (error_type, message) = anthropic_error_fields(&error);
365	yield Err(anyhow::anyhow!(
366	"Anthropic stream error ({error_type}): {message}"
367	));
368	return;
369	}
370	Some(Ok(event)) => {
371	let is_stop = matches!(event, StreamEvent::MessageStop);
372	yield Ok(event);
373	if is_stop {
374	return;
375	}
376	}
377	Some(Err(e)) => {
378	logging::warn(format!("Failed to parse Anthropic SSE event: {e}"));
379	}
380	None => {}
381	}
382	}
383
384	if ended {
385	break;
386	}
387	}
388	};
389
390	Ok(Box::pin(stream))
391	}
392
393	/// Handle a non-streaming Messages API request.
394	pub(super) async fn handle_anthropic_message(
395	&self,
396	prepared: &super::PreparedOutboundRequest,
397	) -> Result<MessageResponse> {
398	let response = self
399	.send_anthropic_request(&prepared.endpoint.url, &prepared.body)
400	.await?;
401	let mut value: Value = response
402	.json()
403	.await
404	.context("Failed to parse Anthropic Messages response")?;
405	if let Some(usage) = value.get_mut("usage") {
406	*usage = json!(parse_anthropic_usage(usage));
407	}
408	serde_json::from_value(value).context("Failed to decode Anthropic Messages response")
409	}
410	}
411
412	/// Build the `/v1/messages` endpoint URL, tolerating base URLs that already
413	/// carry a `/v1` suffix.
414	pub(super) fn anthropic_messages_url(base_url: &str) -> String {
415	let trimmed = base_url.trim_end_matches('/');
416	if trimmed.ends_with("/v1") {
417	format!("{trimmed}/messages")
418	} else {
419	format!("{trimmed}/v1/messages")
420	}
421	}
422
423	/// Whether the route targets first-party Anthropic (`api.anthropic.com`),
424	/// where the `{"type":"adaptive"}` thinking control is valid. Strict
425	/// Anthropic-compatible gateways reject it (#4978).
426	fn is_native_anthropic_base_url(base_url: &str) -> bool {
427	let rest = base_url
428	.trim()
429	.trim_start_matches("https://")
430	.trim_start_matches("http://");
431	let host = rest
432	.split(['/', ':', '?', '#'])
433	.next()
434	.unwrap_or("")
435	.to_ascii_lowercase();
436	host == "api.anthropic.com" \|\| host.ends_with(".anthropic.com")
437	}
438
439	/// Minimum `budget_tokens` the Messages API accepts for extended thinking.
440	const MIN_THINKING_BUDGET_TOKENS: u32 = 1024;
441
442	/// Effort-tier `budget_tokens` for gateways that only accept the documented
443	/// `{"type":"enabled","budget_tokens":N}` thinking shape (#4978). The wire
444	/// contract requires `budget_tokens >= 1024` and `< max_tokens`, so requests
445	/// too small to fit the minimum budget send no thinking block at all.
446	fn compat_thinking_budget(effort: Option<&str>, max_tokens: u32) -> Option<u32> {
447	let tier: u32 = match effort {
448	Some("low" \| "minimal") => 4_096,
449	Some("medium" \| "mid") => 8_192,
450	Some("max" \| "xhigh" \| "highest") => 32_768,
451	// "high" and unspecified effort share the adaptive default tier.
452	_ => 16_384,
453	};
454	let budget = tier.min(max_tokens.checked_sub(1)?);
455	(budget >= MIN_THINKING_BUDGET_TOKENS).then_some(budget)
456	}
457
458	/// Placeholder body for a `tool_use` that never produced a `tool_result`.
459	const UNEXECUTED_TOOL_RESULT: &str = "tool call was not executed";
460
461	/// Defensive wire repair (#5002): every assistant `tool_use` must be answered
462	/// by a `tool_result` in the immediately following user message, or the API
463	/// rejects the whole conversation with a 400 on every retry. Pre-dispatch
464	/// failure paths (e.g. the model calling an unavailable tool) can strand an
465	/// orphaned `tool_use` in history, so missing results get an explicit
466	/// error placeholder instead of poisoning the session.
467	fn repair_dangling_tool_uses(messages: &mut Vec<Value>) {
468	let mut index = 0;
469	while index < messages.len() {
470	let ids = assistant_tool_use_ids(&messages[index]);
471	if ids.is_empty() {
472	index += 1;
473	continue;
474	}
475	let next_is_user = messages
476	.get(index + 1)
477	.and_then(\|message\| message.get("role"))
478	.and_then(Value::as_str)
479	== Some("user");
480	if !next_is_user {
481	messages.insert(index + 1, json!({ "role": "user", "content": [] }));
482	}
483	if let Some(blocks) = messages[index + 1]
484	.get_mut("content")
485	.and_then(Value::as_array_mut)
486	{
487	let answered: std::collections::HashSet<String> = blocks
488	.iter()
489	.filter(\|block\| block.get("type").and_then(Value::as_str) == Some("tool_result"))
490	.filter_map(\|block\| block.get("tool_use_id").and_then(Value::as_str))
491	.map(str::to_string)
492	.collect();
493	// tool_result blocks must lead the user turn, so placeholders are
494	// prepended in tool_use order.
495	for (offset, id) in ids
496	.iter()
497	.filter(\|id\| !answered.contains(id.as_str()))
498	.enumerate()
499	{
500	blocks.insert(
501	offset,
502	json!({
503	"type": "tool_result",
504	"tool_use_id": id,
505	"content": UNEXECUTED_TOOL_RESULT,
506	"is_error": true,
507	}),
508	);
509	}
510	}
511	index += 1;
512	}
513	}
514
515	fn assistant_tool_use_ids(message: &Value) -> Vec<String> {
516	if message.get("role").and_then(Value::as_str) != Some("assistant") {
517	return Vec::new();
518	}
519	message
520	.get("content")
521	.and_then(Value::as_array)
522	.map(\|blocks\| {
523	blocks
524	.iter()
525	.filter(\|block\| block.get("type").and_then(Value::as_str) == Some("tool_use"))
526	.filter_map(\|block\| block.get("id").and_then(Value::as_str))
527	.map(str::to_string)
528	.collect()
529	})
530	.unwrap_or_default()
531	}
532
533	/// Models that reject `temperature` / `top_p` outright (Claude 4.7+).
534	fn anthropic_model_rejects_sampling(model: &str) -> bool {
535	let lower = model.to_ascii_lowercase();
536	lower.contains("opus-4-7")
537	\|\| lower.contains("opus-4-8")
538	\|\| lower.contains("fable")
539	\|\| lower.contains("mythos")
540	}
541
542	/// Convert the engine's `tool_choice` value (OpenAI-style string or object)
543	/// to the Anthropic object form.
544	fn anthropic_tool_choice(tool_choice: &Value) -> Value {
545	match tool_choice.as_str() {
546	Some("auto") => json!({ "type": "auto" }),
547	Some("none") => json!({ "type": "none" }),
548	Some("any" \| "required") => json!({ "type": "any" }),
549	Some(name) => json!({ "type": "tool", "name": name }),
550	None => tool_choice.clone(),
551	}
552	}
553
554	/// Convert one internal message to the Anthropic wire shape. Returns `None`
555	/// when no blocks survive conversion (Anthropic rejects empty content) or
556	/// when the role has no Anthropic channel.
557	///
558	/// The wire role used to be `message.role` forwarded verbatim, which is how a
559	/// `system` message ended up on the wire for the provider to 400 on. It now
560	/// comes from the shared placement table, and pairs the table rejects are
561	/// refused at the outbound seam before this function ever runs.
562	pub(super) fn message_to_anthropic(message: &codewhale_models::Message) -> Option<Value> {
563	let placement = role_placement(&message.role, WireDialect::AnthropicMessages);
564	let wire_role = match placement {
565	RolePlacement::User \| RolePlacement::Developer => "user",
566	RolePlacement::Assistant \| RolePlacement::InterruptedAssistant => "assistant",
567	// Unreachable in production: `reject_unsupported_roles` refuses these
568	// pairs at the seam. Failing closed here keeps a future caller that
569	// skips the seam from putting an unrepresentable role on the wire.
570	RolePlacement::System \| RolePlacement::Omitted \| RolePlacement::Rejected => return None,
571	};
572	let mut blocks: Vec<Value> = message
573	.content
574	.iter()
575	.filter_map(content_block_to_anthropic)
576	.collect();
577	if blocks.is_empty() {
578	return None;
579	}
580	if placement == RolePlacement::InterruptedAssistant
581	&& let Some(text) = blocks
582	.iter_mut()
583	.find(\|block\| block.get("type").and_then(Value::as_str) == Some("text"))
584	{
585	let existing = text
586	.get("text")
587	.and_then(Value::as_str)
588	.unwrap_or_default()
589	.to_string();
590	text["text"] = json!(format!(
591	"{}{}",
592	codewhale_models::INTERRUPTED_ASSISTANT_CONTEXT_PREFIX,
593	existing
594	));
595	}
596	Some(json!({ "role": wire_role, "content": blocks }))
597	}
598
599	/// Project the shared `ImageUrl` block onto Anthropic's tagged image source.
600	///
601	/// The OpenAI dialects carry an image as a single URL string, so that is what
602	/// [`ContentBlock::ImageUrl`] stores. Anthropic instead models the source as a
603	/// tagged union, and — this is the part that used to be wrong here — it does
604	/// not accept a `data:` URL under `{"type":"url"}`. Sending a local
605	/// screenshot that way earns an opaque provider-side 400, which is exactly the
606	/// confusing failure this whole path exists to avoid, so the data URL is taken
607	/// back apart into `{"type":"base64", media_type, data}`.
608	fn anthropic_image_block(url: &str) -> Value {
609	if let Some((media_type, data)) = crate::image_attach::parse_data_url(url) {
610	return json!({
611	"type": "image",
612	"source": { "type": "base64", "media_type": media_type, "data": data },
613	});
614	}
615	if crate::image_attach::is_remote_image_url(url) {
616	return json!({
617	"type": "image",
618	"source": { "type": "url", "url": url },
619	});
620	}
621	// Anything else (a bare path, a `file://`, a truncated data URL) has no
622	// Anthropic representation. Degrade to visible text rather than emitting a
623	// source the API will reject: the turn survives and the model can see that
624	// something was meant to be here.
625	json!({
626	"type": "text",
627	"text": format!("[unsupported image reference: {url}]"),
628	})
629	}
630
631	pub(super) fn anthropic_tool_result_content(
632	content: &str,
633	content_blocks: Option<&[Value]>,
634	) -> Value {
635	let (image, omitted) = crate::image_attach::provider_tool_result_image_refs(content_blocks);
636	let content = crate::image_attach::tool_result_text_with_omission(content, omitted);
637	let Some((mime_type, data)) = image else {
638	return json!(content);
639	};
640	let mut blocks = Vec::with_capacity(2);
641	if !content.is_empty() {
642	blocks.push(json!({ "type": "text", "text": content }));
643	}
644	blocks.push(json!({
645	"type": "image",
646	"source": { "type": "base64", "media_type": mime_type, "data": data },
647	}));
648	json!(blocks)
649	}
650
651	fn content_block_to_anthropic(block: &ContentBlock) -> Option<Value> {
652	match block {
653	ContentBlock::Text {
654	text,
655	cache_control,
656	} => {
657	let mut value = json!({ "type": "text", "text": text });
658	if let Some(cache) = cache_control {
659	value["cache_control"] = json!({ "type": cache.cache_type });
660	}
661	Some(value)
662	}
663	ContentBlock::Thinking {
664	thinking,
665	signature,
666	..
667	} => {
668	// Anthropic rejects unsigned thinking blocks on replay (and the
669	// DeepSeek-era "(reasoning omitted)" placeholders mean nothing to
670	// it), so only signed blocks are replayed — verbatim, signature
671	// included.
672	signature.as_ref().map(\|signature\| {
673	json!({
674	"type": "thinking",
675	"thinking": thinking,
676	"signature": signature,
677	})
678	})
679	}
680	ContentBlock::ToolUse {
681	id, name, input, ..
682	} => Some(json!({
683	"type": "tool_use",
684	"id": id,
685	"name": name,
686	"input": input,
687	})),
688	ContentBlock::ToolResult {
689	tool_use_id,
690	content,
691	is_error,
692	content_blocks,
693	} => {
694	let mut value = json!({
695	"type": "tool_result",
696	"tool_use_id": tool_use_id,
697	"content": anthropic_tool_result_content(content, content_blocks.as_deref()),
698	});
699	if let Some(is_error) = is_error {
700	value["is_error"] = json!(is_error);
701	}
702	Some(value)
703	}
704	ContentBlock::ImageUrl { image_url } => Some(anthropic_image_block(&image_url.url)),
705	// Server-tool block types are DeepSeek/internal concepts with no
706	// Anthropic client-side wire equivalent.
707	ContentBlock::ServerToolUse { .. }
708	\| ContentBlock::ToolSearchToolResult { .. }
709	\| ContentBlock::CodeExecutionToolResult { .. } => None,
710	}
711	}
712
713	/// Enforce the prefix-zone breakpoint policy (#3014):
714	/// 1. the last tool in the catalog (or, with no tools, the last system
715	/// block) — caches the immutable prefix;
716	/// 2. the last content block of the most recent user turn — caches the
717	/// append-only history.
718	///
719	/// Caller-provided breakpoints are preserved, but the total is capped at
720	/// [`MAX_CACHE_BREAKPOINTS`] by dropping the earliest markers first (the
721	/// latest markers cover the longest prefixes).
722	fn apply_anthropic_cache_breakpoints(body: &mut Value) {
723	// Place breakpoint 1: prefer the last tool; otherwise last system block.
724	let mut placed_prefix = false;
725	if let Some(tools) = body.get_mut("tools").and_then(Value::as_array_mut)
726	&& let Some(last) = tools.last_mut()
727	{
728	last["cache_control"] = json!({ "type": "ephemeral" });
729	placed_prefix = true;
730	}
731	if !placed_prefix
732	&& let Some(system) = body.get_mut("system").and_then(Value::as_array_mut)
733	&& let Some(last) = system.last_mut()
734	{
735	last["cache_control"] = json!({ "type": "ephemeral" });
736	}
737
738	// Place breakpoint 2: last content block of the latest user message.
739	if let Some(messages) = body.get_mut("messages").and_then(Value::as_array_mut)
740	&& let Some(last_user) = messages
741	.iter_mut()
742	.rev()
743	.find(\|message\| message.get("role").and_then(Value::as_str) == Some("user"))
744	&& let Some(last_block) = last_user
745	.get_mut("content")
746	.and_then(Value::as_array_mut)
747	.and_then(\|blocks\| blocks.last_mut())
748	{
749	last_block["cache_control"] = json!({ "type": "ephemeral" });
750	}
751
752	// Cap at MAX_CACHE_BREAKPOINTS in render order (tools → system →
753	// messages), dropping the earliest extras.
754	let mut marked: Vec<*mut Value> = Vec::new();
755	let collect = \|value: Option<&mut Value>\| {
756	let Some(array) = value.and_then(Value::as_array_mut) else {
757	return Vec::new();
758	};
759	array
760	.iter_mut()
761	.filter(\|item\| item.get("cache_control").is_some())
762	.map(\|item\| item as *mut Value)
763	.collect::<Vec<_>>()
764	};
765	marked.extend(collect(body.get_mut("tools")));
766	marked.extend(collect(body.get_mut("system")));
767	if let Some(messages) = body.get_mut("messages").and_then(Value::as_array_mut) {
768	for message in messages.iter_mut() {
769	if let Some(blocks) = message.get_mut("content").and_then(Value::as_array_mut) {
770	marked.extend(
771	blocks
772	.iter_mut()
773	.filter(\|block\| block.get("cache_control").is_some())
774	.map(\|block\| block as *mut Value),
775	);
776	}
777	}
778	}
779	if marked.len() > MAX_CACHE_BREAKPOINTS {
780	let excess = marked.len() - MAX_CACHE_BREAKPOINTS;
781	for pointer in marked.into_iter().take(excess) {
782	// SAFETY: the pointers were collected from `body`, which is
783	// exclusively borrowed for the duration of this function, and
784	// each pointer targets a distinct JSON node.
785	unsafe {
786	if let Some(map) = (*pointer).as_object_mut() {
787	map.remove("cache_control");
788	}
789	}
790	}
791	}
792	}
793
794	/// Provider event types [`convert_anthropic_sse_data`] accepts. Anything else
795	/// with a string `type` is tolerated as `None` (future additions); note
796	/// `tool_projection_warning` is deliberately absent — it is local-only and
797	/// must never decode from provider SSE.
798	fn is_known_sse_type(event_type: &str) -> bool {
799	matches!(
800	event_type,
801	"message_start"
802	\| "content_block_start"
803	\| "content_block_delta"
804	\| "content_block_stop"
805	\| "message_delta"
806	\| "message_stop"
807	\| "ping"
808	\| "error"
809	)
810	}
811
812	/// Peek at an SSE payload's `type` without building a DOM.
813	#[derive(Deserialize)]
814	struct SseTagPeek<'a> {
815	#[serde(borrow)]
816	r#type: Option<&'a str>,
817	}
818
819	/// Convert one SSE `data:` payload into a [`StreamEvent`], normalizing usage
820	/// objects to the #2961 convention. Returns `None` for ignorable payloads.
821	///
822	/// #6213 T7: the per-token path deserializes directly into the tagged
823	/// [`StreamEvent`] instead of building a `Value` DOM and converting it.
824	/// Usage-bearing events (two per stream) keep the exact legacy path — the
825	/// usage rewrite reads wire fields the normalized [`Usage`] cannot
826	/// represent — and decode failures keep their exact legacy outcomes.
827	fn convert_anthropic_sse_data(data: &str) -> Option<Result<StreamEvent>> {
828	let trimmed = data.trim();
829	if trimmed.is_empty() {
830	return None;
831	}
832	let usage_event = matches!(
833	serde_json::from_str::<SseTagPeek>(trimmed).map(\|peek\| peek.r#type),
834	Ok(Some("message_start" \| "message_delta"))
835	);
836	if usage_event {
837	return convert_anthropic_sse_usage_event(trimmed);
838	}
839	match serde_json::from_str::<StreamEvent>(trimmed) {
840	// Local-only receipt: the legacy path ignored it (not a provider
841	// type), so it stays ignored rather than decoding.
842	Ok(StreamEvent::ToolProjectionWarning { .. }) => None,
843	Ok(event) => Some(Ok(event)),
844	Err(error) => {
845	// Cold path, reached only when direct decode fails: invalid JSON
846	// and unknown types keep their exact legacy outcomes.
847	let value: Value = match serde_json::from_str(trimmed) {
848	Ok(value) => value,
849	Err(e) => return Some(Err(anyhow::anyhow!("invalid SSE JSON: {e}"))),
850	};
851	match value.get("type").and_then(Value::as_str) {
852	// Tolerate unknown event types (e.g. future additions) silently.
853	Some(known) if !is_known_sse_type(known) => None,
854	_ => Some(Err(anyhow::anyhow!("unrecognized SSE event: {error}"))),
855	}
856	}
857	}
858	}
859
860	/// Legacy `Value` path for `message_start`/`message_delta`: the usage
861	/// rewrite reads wire fields the normalized [`Usage`] cannot represent, so
862	/// these two events normalize before decoding, exactly as before.
863	fn convert_anthropic_sse_usage_event(trimmed: &str) -> Option<Result<StreamEvent>> {
864	let mut value: Value = match serde_json::from_str(trimmed) {
865	Ok(value) => value,
866	Err(e) => return Some(Err(anyhow::anyhow!("invalid SSE JSON: {e}"))),
867	};
868
869	match value.get("type").and_then(Value::as_str) {
870	Some("message_start") => {
871	if let Some(usage) = value
872	.get_mut("message")
873	.and_then(\|message\| message.get_mut("usage"))
874	{
875	*usage = json!(parse_anthropic_usage(usage));
876	}
877	}
878	Some("message_delta") => {
879	if let Some(usage) = value.get_mut("usage") {
880	*usage = json!(parse_anthropic_usage(usage));
881	}
882	}
883	// Tolerate unknown event types (e.g. future additions) silently.
884	Some(known) if !is_known_sse_type(known) => {
885	return None;
886	}
887	_ => {}
888	}
889
890	Some(serde_json::from_value(value).map_err(\|e\| anyhow::anyhow!("unrecognized SSE event: {e}")))
891	}
892
893	/// Map Anthropic's usage payload onto the normalized [`Usage`] convention
894	/// (#2961 / #4318): hit = cache reads, write = cache creation, miss = raw
895	/// uncached input, `input_tokens` = the total prompt across all three.
896	fn parse_anthropic_usage(usage: &Value) -> Usage {
897	let field = \|name: &str\| {
898	usage
899	.get(name)
900	.and_then(Value::as_u64)
901	.and_then(\|value\| u32::try_from(value).ok())
902	.unwrap_or(0)
903	};
904	let input_raw = field("input_tokens");
905	let cache_creation = field("cache_creation_input_tokens");
906	let cache_read = field("cache_read_input_tokens");
907	let output = field("output_tokens");
908
909	Usage {
910	input_tokens: input_raw
911	.saturating_add(cache_creation)
912	.saturating_add(cache_read),
913	output_tokens: output,
914	prompt_cache_hit_tokens: Some(cache_read),
915	prompt_cache_miss_tokens: Some(input_raw),
916	prompt_cache_write_tokens: Some(cache_creation),
917	reasoning_tokens: None,
918	reasoning_replay_tokens: None,
919	server_tool_use: None,
920	}
921	}
922
923	/// Extract `error.type` / `error.message` from an Anthropic error envelope
924	/// (`{"type":"error","error":{"type":...,"message":...}}`), falling back to
925	/// the raw body so nothing is swallowed.
926	fn parse_anthropic_error_envelope(raw: &str) -> (String, String) {
927	let Ok(value) = serde_json::from_str::<Value>(raw) else {
928	return ("unknown".to_string(), raw.to_string());
929	};
930	let error = value.get("error").unwrap_or(&value);
931	anthropic_error_fields(error)
932	}
933
934	fn anthropic_error_fields(error: &Value) -> (String, String) {
935	let error_type = error
936	.get("type")
937	.and_then(Value::as_str)
938	.unwrap_or("unknown")
939	.to_string();
940	let message = error
941	.get("message")
942	.and_then(Value::as_str)
943	.map(str::to_string)
944	.unwrap_or_else(\|\| error.to_string());
945	(error_type, message)
946	}
947
948	#[cfg(test)]
949	mod tests {
950	use super::*;
951	use codewhale_models::Role;
952	use codewhale_models::{CacheControl, Message, SystemBlock, SystemPrompt, Tool};
953
954	fn request_with(
955	model: &str,
956	reasoning_effort: Option<&str>,
957	temperature: Option<f32>,
958	top_p: Option<f32>,
959	) -> MessageRequest {
960	MessageRequest {
961	model: model.to_string(),
962	messages: vec![Message {
963	role: Role::User,
964	content: vec![ContentBlock::Text {
965	text: "hello".to_string(),
966	cache_control: None,
967	}],
968	}],
969	max_tokens: 1024,
970	system: Some(SystemPrompt::Blocks(vec![SystemBlock {
971	block_type: "text".to_string(),
972	text: "be helpful".to_string(),
973	cache_control: Some(CacheControl {
974	cache_type: "ephemeral".to_string(),
975	}),
976	}])),
977	tools: None,
978	tool_choice: None,
979	metadata: None,
980	thinking: None,
981	reasoning_effort: reasoning_effort.map(str::to_string),
982	stream: Some(true),
983	temperature,
984	top_p,
985	}
986	}
987
988	fn test_client() -> CodewhaleClient {
989	anthropic_test_client(None)
990	}
991
992	fn anthropic_test_client(base_url: Option<&str>) -> CodewhaleClient {
993	let _ = rustls::crypto::ring::default_provider().install_default();
994	let config = crate::config::Config {
995	provider: Some("anthropic".to_string()),
996	providers: Some(crate::config::ProvidersConfig {
997	anthropic: crate::config::ProviderConfig {
998	api_key: Some("test-key".to_string()),
999	base_url: base_url.map(str::to_string),
1000	..Default::default()
1001	},
1002	..Default::default()
1003	}),
1004	..Default::default()
1005	};
1006	CodewhaleClient::new(&config).expect("anthropic client constructs")
1007	}
1008
1009	fn minimax_test_client() -> CodewhaleClient {
1010	minimax_test_client_for(crate::config::DEFAULT_MINIMAX_ANTHROPIC_BASE_URL)
1011	}
1012
1013	fn minimax_test_client_for(base_url: &str) -> CodewhaleClient {
1014	let _ = rustls::crypto::ring::default_provider().install_default();
1015	let config = crate::config::Config {
1016	provider: Some("minimax-anthropic".to_string()),
1017	providers: Some(crate::config::ProvidersConfig {
1018	minimax_anthropic: crate::config::ProviderConfig {
1019	api_key: Some("test-key".to_string()),
1020	base_url: Some(base_url.to_string()),
1021	..Default::default()
1022	},
1023	..Default::default()
1024	}),
1025	..Default::default()
1026	};
1027	CodewhaleClient::new(&config).expect("MiniMax Messages client constructs")
1028	}
1029
1030	fn deepseek_test_client(base_url: &str) -> CodewhaleClient {
1031	let _ = rustls::crypto::ring::default_provider().install_default();
1032	let config = crate::config::Config {
1033	provider: Some("deepseek-anthropic".to_string()),
1034	providers: Some(crate::config::ProvidersConfig {
1035	deepseek_anthropic: crate::config::ProviderConfig {
1036	api_key: Some("test-key".to_string()),
1037	base_url: Some(base_url.to_string()),
1038	..Default::default()
1039	},
1040	..Default::default()
1041	}),
1042	..Default::default()
1043	};
1044	CodewhaleClient::new(&config).expect("DeepSeek Messages client constructs")
1045	}
1046
1047	fn modelstudio_test_client(base_url: &str) -> CodewhaleClient {
1048	let _ = rustls::crypto::ring::default_provider().install_default();
1049	let config = crate::config::Config {
1050	provider: Some("modelstudio-token-plan-anthropic".to_string()),
1051	providers: Some(crate::config::ProvidersConfig {
1052	// Durable secret-store keys share a family slot, but a literal
1053	// config key belongs to the selected route's own table.
1054	modelstudio_token_plan_anthropic: crate::config::ProviderConfig {
1055	api_key: Some("test-key".to_string()),
1056	base_url: Some(base_url.to_string()),
1057	..Default::default()
1058	},
1059	..Default::default()
1060	}),
1061	..Default::default()
1062	};
1063	CodewhaleClient::new(&config).expect("Model Studio Messages client constructs")
1064	}
1065
1066	#[test]
1067	fn body_keeps_native_cache_control_on_system_and_tools() {
1068	let client = test_client();
1069	let mut request = request_with("claude-sonnet-4-6", Some("high"), None, None);
1070	request.tools = Some(vec![Tool {
1071	tool_type: None,
1072	name: "read_file".to_string(),
1073	description: "Read a file".to_string(),
1074	input_schema: json!({"type": "object", "additionalProperties": false}),
1075	allowed_callers: None,
1076	defer_loading: None,
1077	input_examples: None,
1078	strict: Some(true),
1079	cache_control: None,
1080	}]);
1081
1082	let body = client.build_anthropic_body(&request, true);
1083
1084	assert_eq!(
1085	body.pointer("/system/0/cache_control/type")
1086	.and_then(Value::as_str),
1087	Some("ephemeral"),
1088	"system cache_control must survive natively: {body}"
1089	);
1090	assert_eq!(
1091	body.pointer("/tools/0/strict").and_then(Value::as_bool),
1092	Some(true)
1093	);
1094	assert_eq!(
1095	body.pointer("/tools/0/cache_control/type")
1096	.and_then(Value::as_str),
1097	Some("ephemeral"),
1098	"breakpoint 1 lands on the last tool: {body}"
1099	);
1100	// Breakpoint 2 lands on the latest user turn's last block.
1101	assert_eq!(
1102	body.pointer("/messages/0/content/0/cache_control/type")
1103	.and_then(Value::as_str),
1104	Some("ephemeral")
1105	);
1106	}
1107
1108	#[test]
1109	fn body_maps_reasoning_effort_to_adaptive_thinking_and_effort() {
1110	let client = test_client();
1111
1112	let body = client.build_anthropic_body(
1113	&request_with("claude-sonnet-4-6", Some("high"), None, None),
1114	true,
1115	);
1116	assert_eq!(
1117	body.pointer("/thinking/type").and_then(Value::as_str),
1118	Some("adaptive")
1119	);
1120	assert_eq!(
1121	body.pointer("/output_config/effort")
1122	.and_then(Value::as_str),
1123	Some("high")
1124	);
1125
1126	let body = client.build_anthropic_body(
1127	&request_with("claude-opus-4-8", Some("xhigh"), None, None),
1128	true,
1129	);
1130	assert_eq!(
1131	body.pointer("/output_config/effort")
1132	.and_then(Value::as_str),
1133	Some("max")
1134	);
1135
1136	let body = client.build_anthropic_body(
1137	&request_with("claude-sonnet-4-6", Some("off"), None, None),
1138	true,
1139	);
1140	assert!(body.get("thinking").is_none(), "off omits thinking: {body}");
1141	assert!(body.get("output_config").is_none());
1142
1143	// Haiku is not thinking-capable: no thinking, no effort.
1144	let body = client.build_anthropic_body(
1145	&request_with("claude-haiku-4-5", Some("high"), None, None),
1146	true,
1147	);
1148	assert!(body.get("thinking").is_none(), "{body}");
1149	assert!(body.get("output_config").is_none(), "{body}");
1150	}
1151
1152	#[test]
1153	fn compat_gateway_sends_enabled_budget_thinking_instead_of_adaptive() {
1154	// #4978: strict Anthropic-compatible gateways (e.g. Sensenova) reject
1155	// {"type":"adaptive"} with a 400; non-native routes must send the
1156	// documented enabled+budget shape and no output_config.
1157	let client = anthropic_test_client(Some("https://api.sensenova.example/v1"));
1158
1159	let mut request = request_with("claude-sonnet-4-6", Some("high"), None, None);
1160	request.max_tokens = 64_000;
1161	let body = client.build_anthropic_body(&request, true);
1162	assert_eq!(
1163	body.pointer("/thinking/type").and_then(Value::as_str),
1164	Some("enabled"),
1165	"{body}"
1166	);
1167	assert_eq!(
1168	body.pointer("/thinking/budget_tokens")
1169	.and_then(Value::as_u64),
1170	Some(16_384)
1171	);
1172	assert!(body.get("output_config").is_none(), "{body}");
1173
1174	// Effort tiers map onto budgets, capped below max_tokens.
1175	let mut request = request_with("claude-sonnet-4-6", Some("max"), None, None);
1176	request.max_tokens = 64_000;
1177	let body = client.build_anthropic_body(&request, true);
1178	assert_eq!(
1179	body.pointer("/thinking/budget_tokens")
1180	.and_then(Value::as_u64),
1181	Some(32_768)
1182	);
1183	let mut request = request_with("claude-sonnet-4-6", Some("max"), None, None);
1184	request.max_tokens = 8_000;
1185	let body = client.build_anthropic_body(&request, true);
1186	assert_eq!(
1187	body.pointer("/thinking/budget_tokens")
1188	.and_then(Value::as_u64),
1189	Some(7_999),
1190	"budget stays below max_tokens: {body}"
1191	);
1192
1193	// Unspecified effort defaults to the "high" tier.
1194	let mut request = request_with("claude-sonnet-4-6", None, None, None);
1195	request.max_tokens = 64_000;
1196	let body = client.build_anthropic_body(&request, true);
1197	assert_eq!(
1198	body.pointer("/thinking/type").and_then(Value::as_str),
1199	Some("enabled")
1200	);
1201	assert_eq!(
1202	body.pointer("/thinking/budget_tokens")
1203	.and_then(Value::as_u64),
1204	Some(16_384)
1205	);
1206
1207	// "off" and requests too small for the 1024-token minimum budget
1208	// omit thinking entirely.
1209	let mut request = request_with("claude-sonnet-4-6", Some("off"), None, None);
1210	request.max_tokens = 64_000;
1211	let body = client.build_anthropic_body(&request, true);
1212	assert!(body.get("thinking").is_none(), "{body}");
1213	let body = client.build_anthropic_body(
1214	&request_with("claude-sonnet-4-6", Some("high"), None, None),
1215	true,
1216	);
1217	assert!(
1218	body.get("thinking").is_none(),
1219	"max_tokens=1024 cannot fit the minimum budget: {body}"
1220	);
1221
1222	// The native route keeps adaptive; the compat shape is only for
1223	// non-anthropic.com hosts.
1224	let native = test_client().build_anthropic_body(
1225	&request_with("claude-sonnet-4-6", Some("high"), None, None),
1226	true,
1227	);
1228	assert_eq!(
1229	native.pointer("/thinking/type").and_then(Value::as_str),
1230	Some("adaptive")
1231	);
1232	}
1233
1234	#[test]
1235	fn dangling_tool_use_gets_placeholder_tool_result() {
1236	// #5002: an orphaned tool_use with no matching tool_result poisons
1237	// the conversation with repeated 400s; request preparation must
1238	// repair it with an explicit placeholder result.
1239	let client = test_client();
1240	let mut request = request_with("claude-sonnet-4-6", None, None, None);
1241	request.messages = vec![
1242	Message {
1243	role: Role::User,
1244	content: vec![ContentBlock::Text {
1245	text: "run both tools".to_string(),
1246	cache_control: None,
1247	}],
1248	},
1249	Message {
1250	role: Role::Assistant,
1251	content: vec![
1252	ContentBlock::ToolUse {
1253	id: "toolu_ok".to_string(),
1254	name: "read_file".to_string(),
1255	input: json!({"path": "a.txt"}),
1256	caller: None,
1257	thought_signature: None,
1258	},
1259	ContentBlock::ToolUse {
1260	id: "toolu_orphan".to_string(),
1261	name: "task".to_string(),
1262	input: json!({}),
1263	caller: None,
1264	thought_signature: None,
1265	},
1266	],
1267	},
1268	// Pre-dispatch failure left only one tool_result behind.
1269	Message {
1270	role: Role::User,
1271	content: vec![ContentBlock::ToolResult {
1272	tool_use_id: "toolu_ok".to_string(),
1273	content: "contents".to_string(),
1274	is_error: None,
1275	content_blocks: None,
1276	}],
1277	},
1278	// Trailing assistant tool_use with no user turn at all.
1279	Message {
1280	role: Role::Assistant,
1281	content: vec![ContentBlock::ToolUse {
1282	id: "toolu_tail".to_string(),
1283	name: "task".to_string(),
1284	input: json!({}),
1285	caller: None,
1286	thought_signature: None,
1287	}],
1288	},
1289	];
1290
1291	let body = client.build_anthropic_body(&request, true);
1292	let messages = body["messages"].as_array().expect("messages array");
1293	assert_eq!(messages.len(), 5, "a repair turn is appended: {body}");
1294
1295	// The orphaned id gets a leading placeholder; the answered one is
1296	// untouched (no duplicate result).
1297	let repaired = messages[2]["content"].as_array().expect("user content");
1298	assert_eq!(repaired.len(), 2, "{body}");
1299	assert_eq!(repaired[0]["type"].as_str(), Some("tool_result"));
1300	assert_eq!(repaired[0]["tool_use_id"].as_str(), Some("toolu_orphan"));
1301	assert_eq!(
1302	repaired[0]["content"].as_str(),
1303	Some(UNEXECUTED_TOOL_RESULT)
1304	);
1305	assert_eq!(repaired[0]["is_error"].as_bool(), Some(true));
1306	assert_eq!(repaired[1]["tool_use_id"].as_str(), Some("toolu_ok"));
1307	assert_eq!(repaired[1]["content"].as_str(), Some("contents"));
1308
1309	// The trailing tool_use gains a synthesized user turn.
1310	assert_eq!(messages[4]["role"].as_str(), Some("user"));
1311	let tail = messages[4]["content"].as_array().expect("tail content");
1312	assert_eq!(tail.len(), 1, "{body}");
1313	assert_eq!(tail[0]["type"].as_str(), Some("tool_result"));
1314	assert_eq!(tail[0]["tool_use_id"].as_str(), Some("toolu_tail"));
1315	assert_eq!(tail[0]["content"].as_str(), Some(UNEXECUTED_TOOL_RESULT));
1316
1317	// A fully answered history is left alone.
1318	request.messages.truncate(3);
1319	request.messages[1].content.retain(
1320	\|block\| !matches!(block, ContentBlock::ToolUse { id, ..} if id == "toolu_orphan"),
1321	);
1322	let body = client.build_anthropic_body(&request, true);
1323	let messages = body["messages"].as_array().expect("messages array");
1324	assert_eq!(messages.len(), 3, "no repair turn appended: {body}");
1325	let untouched = messages[2]["content"].as_array().expect("user content");
1326	assert_eq!(untouched.len(), 1, "{body}");
1327	assert_eq!(untouched[0]["tool_use_id"].as_str(), Some("toolu_ok"));
1328	}
1329
1330	#[test]
1331	fn modelstudio_messages_body_requests_thinking_with_budget() {
1332	// Model Studio's Anthropic-compatible endpoint documents the portable
1333	// {"type":"enabled","budget_tokens":N} shape plus {"type":"disabled"}
1334	// (alibabacloud.com/help/en/model-studio/anthropic-api-messages).
1335	let client = modelstudio_test_client(
1336	"https://token-plan.ap-southeast-1.maas.aliyuncs.com/apps/anthropic",
1337	);
1338
1339	let mut request = request_with("qwen3.8-max", Some("high"), None, None);
1340	request.max_tokens = 64_000;
1341	let body = client.build_anthropic_body(&request, true);
1342	assert_eq!(
1343	body.pointer("/thinking/type").and_then(Value::as_str),
1344	Some("enabled"),
1345	"{body}"
1346	);
1347	assert!(
1348	body.pointer("/thinking/budget_tokens")
1349	.and_then(Value::as_u64)
1350	.is_some(),
1351	"{body}"
1352	);
1353	assert!(body.get("output_config").is_none(), "{body}");
1354	assert_eq!(
1355	body.get("model").and_then(Value::as_str),
1356	Some("qwen3.8-max"),
1357	"{body}"
1358	);
1359
1360	// An explicit "off" is honored on the wire instead of silently
1361	// falling through to the server default (thinking-ON for qwen3.x).
1362	let mut request = request_with("qwen3.8-max", Some("off"), None, None);
1363	request.max_tokens = 64_000;
1364	let body = client.build_anthropic_body(&request, true);
1365	assert_eq!(
1366	body.pointer("/thinking/type").and_then(Value::as_str),
1367	Some("disabled"),
1368	"{body}"
1369	);
1370	}
1371
1372	#[test]
1373	fn deepseek_messages_body_retires_aliases_and_keeps_thinking_control() {
1374	let client = deepseek_test_client(crate::config::DEFAULT_DEEPSEEK_ANTHROPIC_BASE_URL);
1375
1376	let chat = client.build_anthropic_body(
1377	&request_with("deepseek-chat", Some("off"), None, None),
1378	true,
1379	);
1380	assert_eq!(
1381	chat.get("model").and_then(Value::as_str),
1382	Some(crate::config::DEEPSEEK_ALIAS_REPLACEMENT)
1383	);
1384	assert_eq!(
1385	chat.pointer("/thinking/type").and_then(Value::as_str),
1386	Some("disabled")
1387	);
1388
1389	let reasoner = client.build_anthropic_body(
1390	&request_with("deepseek-reasoner", Some("high"), None, None),
1391	true,
1392	);
1393	assert_eq!(
1394	reasoner.get("model").and_then(Value::as_str),
1395	Some(crate::config::DEEPSEEK_ALIAS_REPLACEMENT)
1396	);
1397	assert_eq!(
1398	reasoner.pointer("/thinking/type").and_then(Value::as_str),
1399	Some("adaptive")
1400	);
1401	assert_eq!(
1402	reasoner
1403	.pointer("/output_config/effort")
1404	.and_then(Value::as_str),
1405	Some("high")
1406	);
1407
1408	let custom = deepseek_test_client("https://messages.example/v1");
1409	let custom_body = custom.build_anthropic_body(
1410	&request_with("deepseek-reasoner", Some("high"), None, None),
1411	true,
1412	);
1413	assert_eq!(
1414	custom_body.get("model").and_then(Value::as_str),
1415	Some("deepseek-reasoner")
1416	);
1417	}
1418
1419	#[test]
1420	fn omitted_alias_effort_is_migrated_into_deepseek_messages_body() {
1421	for (alias, expected_effort, expected_thinking) in [
1422	("deepseek-chat", "off", "disabled"),
1423	("deepseek-reasoner", "high", "adaptive"),
1424	] {
1425	let mut config = crate::config::Config {
1426	provider: Some("deepseek-anthropic".to_string()),
1427	providers: Some(crate::config::ProvidersConfig {
1428	deepseek_anthropic: crate::config::ProviderConfig {
1429	api_key: Some("test-key".to_string()),
1430	model: Some(alias.to_string()),
1431	..Default::default()
1432	},
1433	..Default::default()
1434	}),
1435	..Default::default()
1436	};
1437	assert!(
1438	config.reasoning_effort().is_none(),
1439	"fixture must omit effort"
1440	);
1441
1442	crate::config::normalize_model_config_for_test(&mut config);
1443	let client = CodewhaleClient::new(&config).expect("DeepSeek Messages client");
1444	let model = config.default_model();
1445	let body = client.build_anthropic_body(
1446	&request_with(&model, config.reasoning_effort(), None, None),
1447	true,
1448	);
1449
1450	assert_eq!(
1451	body.get("model").and_then(Value::as_str),
1452	Some(crate::config::DEEPSEEK_ALIAS_REPLACEMENT),
1453	"{alias}: {body}"
1454	);
1455	assert_eq!(config.reasoning_effort(), Some(expected_effort));
1456	assert_eq!(
1457	body.pointer("/thinking/type").and_then(Value::as_str),
1458	Some(expected_thinking),
1459	"{alias}: {body}"
1460	);
1461	if alias == "deepseek-reasoner" {
1462	assert_eq!(
1463	body.pointer("/output_config/effort")
1464	.and_then(Value::as_str),
1465	Some("high"),
1466	"{body}"
1467	);
1468	} else {
1469	assert!(body.get("output_config").is_none(), "{body}");
1470	}
1471	}
1472	}
1473
1474	#[test]
1475	fn minimax_body_uses_supported_thinking_controls() {
1476	let client = minimax_test_client();
1477	let body =
1478	client.build_anthropic_body(&request_with("MiniMax-M3", Some("off"), None, None), true);
1479	assert_eq!(
1480	body.pointer("/thinking/type").and_then(Value::as_str),
1481	Some("disabled")
1482	);
1483	assert!(body.get("output_config").is_none(), "{body}");
1484
1485	let mut enabled_bodies = Vec::new();
1486	for effort in ["high", "max"] {
1487	let body = client
1488	.build_anthropic_body(&request_with("MiniMax-M3", Some(effort), None, None), true);
1489	assert_eq!(
1490	body.pointer("/thinking/type").and_then(Value::as_str),
1491	Some("adaptive"),
1492	"{effort}: {body}"
1493	);
1494	assert!(body.get("output_config").is_none(), "{effort}: {body}");
1495	enabled_bodies.push(body);
1496	}
1497	assert_eq!(
1498	enabled_bodies[0].get("thinking"),
1499	enabled_bodies[1].get("thinking"),
1500	"MiniMax high/max select the same untiered adaptive wire control"
1501	);
1502	}
1503
1504	#[test]
1505	fn minimax_messages_reasoning_controls_require_exact_first_party_m3_route() {
1506	for (base_url, model) in [
1507	(
1508	"https://gateway.example/anthropic",
1509	crate::config::DEFAULT_MINIMAX_MODEL,
1510	),
1511	(
1512	crate::config::DEFAULT_MINIMAX_ANTHROPIC_BASE_URL,
1513	"MiniMax-M2",
1514	),
1515	] {
1516	let client = minimax_test_client_for(base_url);
1517	for effort in ["off", "high", "max"] {
1518	let body = client
1519	.build_anthropic_body(&request_with(model, Some(effort), None, None), true);
1520	assert!(
1521	body.get("thinking").is_none(),
1522	"{base_url} {model} {effort}: {body}"
1523	);
1524	assert!(
1525	body.get("output_config").is_none(),
1526	"{base_url} {model} {effort}: {body}"
1527	);
1528	}
1529	}
1530	}
1531
1532	#[test]
1533	fn body_drops_sampling_params_for_models_that_reject_them() {
1534	let client = test_client();
1535
1536	let body = client.build_anthropic_body(
1537	&request_with("claude-opus-4-8", None, Some(0.7), Some(0.9)),
1538	true,
1539	);
1540	assert!(body.get("temperature").is_none(), "{body}");
1541	assert!(body.get("top_p").is_none(), "{body}");
1542
1543	// Older models accept ONE of temperature / top_p (temperature wins).
1544	let body = client.build_anthropic_body(
1545	&request_with("claude-sonnet-4-6", None, Some(0.7), Some(0.9)),
1546	true,
1547	);
1548	assert_eq!(
1549	body.get("temperature").and_then(Value::as_f64),
1550	Some(f64::from(0.7f32))
1551	);
1552	assert!(body.get("top_p").is_none(), "never send both: {body}");
1553	}
1554
1555	#[test]
1556	fn body_replays_signed_thinking_and_drops_unsigned_placeholders() {
1557	let client = test_client();
1558	let mut request = request_with("claude-sonnet-4-6", None, None, None);
1559	request.messages = vec![
1560	Message {
1561	role: Role::User,
1562	content: vec![ContentBlock::Text {
1563	text: "do the thing".to_string(),
1564	cache_control: None,
1565	}],
1566	},
1567	Message {
1568	role: Role::Assistant,
1569	content: vec![
1570	ContentBlock::Thinking {
1571	thinking: "signed reasoning".to_string(),
1572	signature: Some("sig-abc".to_string()),
1573	state: None,
1574	},
1575	ContentBlock::Thinking {
1576	thinking: "(reasoning omitted)".to_string(),
1577	signature: None,
1578	state: None,
1579	},
1580	ContentBlock::ToolUse {
1581	id: "toolu_1".to_string(),
1582	name: "read_file".to_string(),
1583	input: json!({"path": "a.txt"}),
1584	caller: None,
1585	thought_signature: None,
1586	},
1587	],
1588	},
1589	Message {
1590	role: Role::User,
1591	content: vec![ContentBlock::ToolResult {
1592	tool_use_id: "toolu_1".to_string(),
1593	content: "contents".to_string(),
1594	is_error: None,
1595	content_blocks: None,
1596	}],
1597	},
1598	];
1599
1600	let body = client.build_anthropic_body(&request, true);
1601	let assistant = &body["messages"][1]["content"];
1602	assert_eq!(assistant.as_array().map(Vec::len), Some(2));
1603	assert_eq!(
1604	assistant[0]["signature"].as_str(),
1605	Some("sig-abc"),
1606	"signed thinking replays verbatim: {assistant}"
1607	);
1608	assert_eq!(assistant[1]["type"].as_str(), Some("tool_use"));
1609	assert!(
1610	assistant[1].get("caller").is_none(),
1611	"internal caller metadata must not reach the wire"
1612	);
1613	assert_eq!(
1614	body["messages"][2]["content"][0]["type"].as_str(),
1615	Some("tool_result")
1616	);
1617	}
1618
1619	#[test]
1620	fn breakpoints_are_capped_at_four_dropping_earliest() {
1621	let client = test_client();
1622	let mut request = request_with("claude-sonnet-4-6", None, None, None);
1623	// Five caller-marked user turns + the two placed breakpoints.
1624	request.messages = (0..5)
1625	.map(\|i\| Message {
1626	role: Role::User,
1627	content: vec![ContentBlock::Text {
1628	text: format!("turn {i}"),
1629	cache_control: Some(CacheControl {
1630	cache_type: "ephemeral".to_string(),
1631	}),
1632	}],
1633	})
1634	.collect();
1635
1636	let body = client.build_anthropic_body(&request, true);
1637	let mut count = 0;
1638	if body.pointer("/system/0/cache_control").is_some() {
1639	count += 1;
1640	}
1641	for message in body["messages"].as_array().unwrap() {
1642	for block in message["content"].as_array().unwrap() {
1643	if block.get("cache_control").is_some() {
1644	count += 1;
1645	}
1646	}
1647	}
1648	assert!(
1649	count <= MAX_CACHE_BREAKPOINTS,
1650	"breakpoints must be capped at {MAX_CACHE_BREAKPOINTS}, got {count}: {body}"
1651	);
1652	// The latest user turn keeps its marker (longest prefix coverage).
1653	assert!(
1654	body.pointer("/messages/4/content/0/cache_control")
1655	.is_some(),
1656	"{body}"
1657	);
1658	}
1659
1660	#[test]
1661	fn sse_fixture_decodes_text_thinking_signature_and_tool_use() {
1662	use codewhale_models::{ContentBlockStart, Delta};
1663
1664	let events = [
1665	r#"{"type":"message_start","message":{"id":"msg_01","type":"message","role":"assistant","content":[],"model":"claude-sonnet-4-6","stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":3,"cache_creation_input_tokens":2045,"cache_read_input_tokens":18000,"output_tokens":1}}}"#,
1666	r#"{"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":""}}"#,
1667	r#"{"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"Let me check"}}"#,
1668	r#"{"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"sig-xyz"}}"#,
1669	r#"{"type":"content_block_stop","index":0}"#,
1670	r#"{"type":"content_block_start","index":1,"content_block":{"type":"text","text":""}}"#,
1671	r#"{"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"Reading the file."}}"#,
1672	r#"{"type":"content_block_stop","index":1}"#,
1673	r#"{"type":"content_block_start","index":2,"content_block":{"type":"tool_use","id":"toolu_9","name":"read_file","input":{}}}"#,
1674	r#"{"type":"content_block_delta","index":2,"delta":{"type":"input_json_delta","partial_json":"{\"path\":"}}"#,
1675	r#"{"type":"content_block_delta","index":2,"delta":{"type":"input_json_delta","partial_json":"\"a.txt\"}"}}"#,
1676	r#"{"type":"content_block_stop","index":2}"#,
1677	r#"{"type":"ping"}"#,
1678	r#"{"type":"message_delta","delta":{"stop_reason":"tool_use","stop_sequence":null},"usage":{"output_tokens":42}}"#,
1679	r#"{"type":"message_stop"}"#,
1680	];
1681
1682	let decoded: Vec<StreamEvent> = events
1683	.iter()
1684	.map(\|data\| {
1685	convert_anthropic_sse_data(data)
1686	.expect("known event")
1687	.expect("decodes")
1688	})
1689	.collect();
1690
1691	// message_start usage normalized to the #2961 convention.
1692	let StreamEvent::MessageStart { message } = &decoded[0] else {
1693	panic!("expected MessageStart, got {:?}", decoded[0]);
1694	};
1695	assert_eq!(message.usage.input_tokens, 3 + 2045 + 18000);
1696	assert_eq!(message.usage.prompt_cache_hit_tokens, Some(18000));
1697	assert_eq!(message.usage.prompt_cache_miss_tokens, Some(3));
1698	assert_eq!(message.usage.prompt_cache_write_tokens, Some(2045));
1699
1700	assert!(matches!(
1701	&decoded[1],
1702	StreamEvent::ContentBlockStart {
1703	content_block: ContentBlockStart::Thinking { .. },
1704	..
1705	}
1706	));
1707	assert!(matches!(
1708	&decoded[3],
1709	StreamEvent::ContentBlockDelta {
1710	delta: Delta::SignatureDelta { signature },
1711	..
1712	} if signature == "sig-xyz"
1713	));
1714	assert!(matches!(
1715	&decoded[6],
1716	StreamEvent::ContentBlockDelta {
1717	delta: Delta::TextDelta { text },
1718	..
1719	} if text == "Reading the file."
1720	));
1721	let mut tool_json = String::new();
1722	for event in &decoded {
1723	if let StreamEvent::ContentBlockDelta {
1724	delta: Delta::InputJsonDelta { partial_json },
1725	..
1726	} = event
1727	{
1728	tool_json.push_str(partial_json);
1729	}
1730	}
1731	assert_eq!(
1732	serde_json::from_str::<Value>(&tool_json).expect("accumulated tool args parse"),
1733	json!({"path": "a.txt"})
1734	);
1735	assert!(matches!(&decoded[12], StreamEvent::Ping));
1736	let StreamEvent::MessageDelta { delta, usage } = &decoded[13] else {
1737	panic!("expected MessageDelta");
1738	};
1739	assert_eq!(delta.stop_reason.as_deref(), Some("tool_use"));
1740	assert_eq!(usage.as_ref().map(\|u\| u.output_tokens), Some(42));
1741	assert!(matches!(&decoded[14], StreamEvent::MessageStop));
1742	}
1743
1744	#[test]
1745	fn sse_error_event_and_unknown_events_are_handled() {
1746	let error = convert_anthropic_sse_data(
1747	r#"{"type":"error","error":{"type":"overloaded_error","message":"Overloaded"}}"#,
1748	)
1749	.expect("error event decodes")
1750	.expect("error event is a StreamEvent");
1751	let StreamEvent::Error { error } = error else {
1752	panic!("expected StreamEvent::Error");
1753	};
1754	let (error_type, message) = anthropic_error_fields(&error);
1755	assert_eq!(error_type, "overloaded_error");
1756	assert_eq!(message, "Overloaded");
1757
1758	assert!(
1759	convert_anthropic_sse_data(r#"{"type":"content_block_started_v2","index":0}"#)
1760	.is_none(),
1761	"unknown event types are tolerated"
1762	);
1763	assert!(convert_anthropic_sse_data(" ").is_none());
1764	}
1765
1766	#[test]
1767	fn sse_decode_failures_keep_legacy_outcomes_on_the_direct_path() {
1768	// Malformed JSON: the invalid-input error, not the unrecognized one.
1769	let error = convert_anthropic_sse_data("{oops")
1770	.expect("malformed is Some")
1771	.expect_err("malformed is Err");
1772	assert!(error.to_string().contains("invalid SSE JSON"), "{error:?}");
1773	// Structurally invalid known event: unrecognized, not tolerated.
1774	let error = convert_anthropic_sse_data(r#"{"type":"content_block_stop"}"#)
1775	.expect("known type is Some")
1776	.expect_err("missing index is Err");
1777	assert!(
1778	error.to_string().contains("unrecognized SSE event"),
1779	"{error:?}"
1780	);
1781	// Local-only receipt: never provider SSE, stays ignored.
1782	assert!(
1783	convert_anthropic_sse_data(
1784	r#"{"type":"tool_projection_warning","provider":"x","omitted_tool_names":[],"omitted_tool_count":0}"#
1785	)
1786	.is_none()
1787	);
1788	}
1789
1790	#[test]
1791	fn usage_mapping_handles_missing_cache_fields() {
1792	let usage = parse_anthropic_usage(&json!({"input_tokens": 10, "output_tokens": 5}));
1793	assert_eq!(usage.input_tokens, 10);
1794	assert_eq!(usage.output_tokens, 5);
1795	assert_eq!(usage.prompt_cache_hit_tokens, Some(0));
1796	assert_eq!(usage.prompt_cache_miss_tokens, Some(10));
1797	assert_eq!(usage.prompt_cache_write_tokens, Some(0));
1798	}
1799
1800	#[test]
1801	fn usage_mapping_keeps_cache_write_separate_from_miss() {
1802	let usage = parse_anthropic_usage(&json!({
1803	"input_tokens": 3,
1804	"cache_creation_input_tokens": 2045,
1805	"cache_read_input_tokens": 18000,
1806	"output_tokens": 1,
1807	}));
1808	assert_eq!(usage.input_tokens, 3 + 2045 + 18000);
1809	assert_eq!(usage.prompt_cache_hit_tokens, Some(18000));
1810	assert_eq!(usage.prompt_cache_miss_tokens, Some(3));
1811	assert_eq!(usage.prompt_cache_write_tokens, Some(2045));
1812	}
1813
1814	#[test]
1815	fn error_envelope_parses_type_and_message() {
1816	let (error_type, message) = parse_anthropic_error_envelope(
1817	r#"{"type":"error","error":{"type":"rate_limit_error","message":"Too many requests"},"request_id":"req_1"}"#,
1818	);
1819	assert_eq!(error_type, "rate_limit_error");
1820	assert_eq!(message, "Too many requests");
1821
1822	let (error_type, message) = parse_anthropic_error_envelope("upstream blew up");
1823	assert_eq!(error_type, "unknown");
1824	assert_eq!(message, "upstream blew up");
1825	}
1826
1827	#[test]
1828	fn data_url_image_becomes_a_base64_source_not_a_url_source() {
1829	// Anthropic rejects a `data:` URL under `{"type":"url"}`. This is the
1830	// whole reason the projection exists; if it regresses, every locally
1831	// attached screenshot 400s on the native route.
1832	let block = content_block_to_anthropic(&ContentBlock::ImageUrl {
1833	image_url: codewhale_models::ImageUrlContent {
1834	url: "data:image/png;base64,QUJD".to_string(),
1835	},
1836	})
1837	.expect("image block");
1838
1839	assert_eq!(block["type"], "image");
1840	assert_eq!(block["source"]["type"], "base64");
1841	assert_eq!(block["source"]["media_type"], "image/png");
1842	assert_eq!(block["source"]["data"], "QUJD");
1843	assert!(
1844	block["source"].get("url").is_none(),
1845	"base64 sources must not carry a url field: {block}"
1846	);
1847	}
1848
1849	#[test]
1850	fn tool_result_image_stays_inside_the_native_tool_result_block() {
1851	let content = anthropic_tool_result_content(
1852	"screenshot captured",
1853	Some(&[json!({
1854	"type": "image",
1855	"mime_type": "image/png",
1856	"data": "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR4nGP4z8DwHwAFAAH/iZk9HQAAAABJRU5ErkJggg==",
1857	})]),
1858	);
1859	let blocks = content.as_array().expect("rich tool_result content");
1860
1861	assert_eq!(
1862	blocks[0],
1863	json!({"type": "text", "text": "screenshot captured"})
1864	);
1865	assert_eq!(blocks[1]["type"], "image");
1866	assert_eq!(blocks[1]["source"]["type"], "base64");
1867	assert_eq!(blocks[1]["source"]["media_type"], "image/png");
1868	assert_eq!(
1869	blocks[1]["source"]["data"],
1870	"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR4nGP4z8DwHwAFAAH/iZk9HQAAAABJRU5ErkJggg=="
1871	);
1872	}
1873
1874	#[test]
1875	fn remote_image_url_stays_a_url_source() {
1876	let block = content_block_to_anthropic(&ContentBlock::ImageUrl {
1877	image_url: codewhale_models::ImageUrlContent {
1878	url: "https://example.com/shot.png".to_string(),
1879	},
1880	})
1881	.expect("image block");
1882
1883	assert_eq!(block["type"], "image");
1884	assert_eq!(block["source"]["type"], "url");
1885	assert_eq!(block["source"]["url"], "https://example.com/shot.png");
1886	}
1887
1888	#[test]
1889	fn unrepresentable_image_reference_degrades_to_visible_text() {
1890	for url in [
1891	"file:///tmp/shot.png",
1892	"/tmp/shot.png",
1893	"data:image/png,QUJD",
1894	] {
1895	let block = content_block_to_anthropic(&ContentBlock::ImageUrl {
1896	image_url: codewhale_models::ImageUrlContent {
1897	url: url.to_string(),
1898	},
1899	})
1900	.expect("block");
1901
1902	assert_eq!(block["type"], "text", "{url} should degrade: {block}");
1903	assert!(
1904	block["text"].as_str().expect("text").contains(url),
1905	"the degraded text should name the reference: {block}"
1906	);
1907	}
1908	}
1909
1910	#[test]
1911	fn messages_url_tolerates_v1_suffix() {
1912	assert_eq!(
1913	anthropic_messages_url("https://api.anthropic.com"),
1914	"https://api.anthropic.com/v1/messages"
1915	);
1916	assert_eq!(
1917	anthropic_messages_url("https://api.anthropic.com/"),
1918	"https://api.anthropic.com/v1/messages"
1919	);
1920	assert_eq!(
1921	anthropic_messages_url("https://gateway.example/v1"),
1922	"https://gateway.example/v1/messages"
1923	);
1924	assert_eq!(
1925	anthropic_messages_url("https://api.deepseek.com/anthropic"),
1926	"https://api.deepseek.com/anthropic/v1/messages"
1927	);
1928	assert_eq!(
1929	anthropic_messages_url("https://api.minimax.io/anthropic"),
1930	"https://api.minimax.io/anthropic/v1/messages"
1931	);
1932	assert_eq!(
1933	anthropic_messages_url("https://api.minimaxi.com/anthropic"),
1934	"https://api.minimaxi.com/anthropic/v1/messages"
1935	);
1936	}
1937
1938	#[test]
1939	fn anthropic_body_serializes_the_child_catalog_without_duplication() {
1940	// The real child catalog fixture (not a hand-built tool list) must
1941	// survive Messages serialization with exactly one canonical `read`
1942	// entry — no dedup, filter, or sanitizer may drop or duplicate it.
1943	// Skills are discoverable through tool_search, so the child wire
1944	// catalog carries no load_skill at all.
1945	let tools = crate::tools::subagent::kimi_general_child_request_tools_fixture();
1946	assert_eq!(
1947	tools.iter().filter(\|tool\| tool.name == "read").count(),
1948	1,
1949	"catalog fixture carries one canonical read"
1950	);
1951	assert_eq!(
1952	tools
1953	.iter()
1954	.filter(\|tool\| tool.name == "load_skill")
1955	.count(),
1956	0,
1957	"load_skill is not part of the child wire catalog"
1958	);
1959	let client = test_client();
1960	let mut request = request_with("claude-sonnet-4-6", None, None, None);
1961	request.tools = Some(tools);
1962	let body = client.build_anthropic_body(&request, true);
1963	let serialized = body["tools"]
1964	.as_array()
1965	.expect("tools serialize as an array");
1966	let reads: Vec<_> = serialized
1967	.iter()
1968	.filter(\|tool\| tool["name"] == "read")
1969	.collect();
1970	assert_eq!(
1971	reads.len(),
1972	1,
1973	"exactly one canonical read definition reaches the Messages wire"
1974	);
1975	assert!(
1976	reads[0]["input_schema"]["properties"].is_object(),
1977	"read keeps a valid object schema: {}",
1978	reads[0]
1979	);
1980	assert!(
1981	serialized.iter().all(\|tool\| tool["name"] != "load_skill"),
1982	"load_skill must not appear on the child Messages wire"
1983	);
1984	}
1985
1986	#[tokio::test]
1987	async fn anthropic_stream_opens_through_shared_seam_preserving_headers() {
1988	use futures_util::StreamExt;
1989	use wiremock::matchers::{header, method, path};
1990	use wiremock::{Mock, MockServer, ResponseTemplate};
1991
1992	let server = MockServer::start().await;
1993	// The wire-specific Accept header must survive the shared stream-entry
1994	// open path; the mock only answers when it is present.
1995	Mock::given(method("POST"))
1996	.and(path("/v1/messages"))
1997	.and(header("Accept", "text/event-stream"))
1998	.respond_with(
1999	ResponseTemplate::new(200)
2000	.insert_header("Content-Type", "text/event-stream")
2001	.set_body_string("data: {\"type\":\"message_stop\"}\n\n"),
2002	)
2003	.expect(1)
2004	.mount(&server)
2005	.await;
2006
2007	let client = deepseek_test_client(&server.uri());
2008	let mut stream = client
2009	.handle_anthropic_stream(
2010	&client
2011	.prepare_outbound_request(request_with("deepseek-v4", None, None, None), true)
2012	.expect("anthropic request prepares"),
2013	)
2014	.await
2015	.expect("stream opens through the shared seam");
2016
2017	let mut saw_stop = false;
2018	tokio::time::timeout(std::time::Duration::from_secs(5), async {
2019	while let Some(event) = stream.next().await {
2020	if matches!(event.expect("stream event"), StreamEvent::MessageStop) {
2021	saw_stop = true;
2022	}
2023	}
2024	})
2025	.await
2026	.expect("stream finishes after message_stop");
2027	assert!(saw_stop, "message_stop should arrive through the seam");
2028	}
2029
2030	#[tokio::test]
2031	async fn anthropic_stream_open_error_is_not_retried() {
2032	use wiremock::matchers::{method, path};
2033	use wiremock::{Mock, MockServer, ResponseTemplate};
2034
2035	let server = MockServer::start().await;
2036	// A definitive provider error before any stream body must fail fast:
2037	// exactly one request, no H1 fallback, envelope preserved.
2038	Mock::given(method("POST"))
2039	.and(path("/v1/messages"))
2040	.respond_with(ResponseTemplate::new(401).set_body_string(
2041	"{\"error\":{\"type\":\"authentication_error\",\"message\":\"bad key\"}}",
2042	))
2043	.expect(1)
2044	.mount(&server)
2045	.await;
2046
2047	let client = deepseek_test_client(&server.uri());
2048	let err = match client
2049	.handle_anthropic_stream(
2050	&client
2051	.prepare_outbound_request(request_with("deepseek-v4", None, None, None), true)
2052	.expect("anthropic request prepares"),
2053	)
2054	.await
2055	{
2056	Ok(_) => panic!("auth errors must fail fast"),
2057	Err(err) => err,
2058	};
2059	let text = err.to_string();
2060	assert!(
2061	text.contains("HTTP 401") && text.contains("authentication_error"),
2062	"error envelope should be preserved: {text}"
2063	);
2064	}
2065
2066	/// A `system`-role history message — what a compaction summary, a branch
2067	/// summary, or an imported journal `system` entry becomes once it reaches
2068	/// `MessageRequest::messages` — must not be emitted verbatim: the Messages
2069	/// API accepts only `user` and `assistant` in `messages[].role` and 400s
2070	/// the whole conversation otherwise, on every retry.
2071	#[test]
2072	fn system_role_history_message_is_not_emitted_verbatim_on_the_messages_wire() {
2073	let mut request = request_with("claude-opus-4-6", None, None, None);
2074	request.messages.insert(
2075	0,
2076	Message {
2077	role: Role::System,
2078	content: vec![ContentBlock::Text {
2079	text: "[compaction summary] the user is porting the parser".to_string(),
2080	cache_control: None,
2081	}],
2082	},
2083	);
2084
2085	let body = test_client().build_anthropic_body(&request, false);
2086	let messages = body["messages"].as_array().expect("messages array");
2087
2088	for message in messages {
2089	let role = message["role"].as_str().expect("role is a string");
2090	assert!(
2091	role == "user" \|\| role == "assistant",
2092	"Messages API rejects role {role:?}"
2093	);
2094	}
2095	let carried = messages.iter().any(\|message\| {
2096	message["content"].as_array().is_some_and(\|blocks\| {
2097	blocks.iter().any(\|block\| {
2098	block["text"].as_str()
2099	== Some("[compaction summary] the user is porting the parser")
2100	})
2101	})
2102	});
2103	assert!(carried, "the summary text must survive: {messages:?}");
2104	}
2105	}
2106

2106 lines RUST