Vakyartha documentation

1use futures::StreamExt;
2use serde_json::Value;
3use tokio_util::sync::CancellationToken;
4
5use crate::Provider;
6use crate::error::LlmError;
7use crate::gate::ProviderGate;
8use crate::sse::SseDecoder;
9use crate::stream::{EventSink, EventStream, StreamEvent, channel};
10use crate::turn::{current_turn_boundary, strip_thinking};
11use crate::types::{
12 AssistantMessage, ChatRequest, ContentBlock, Effort, Message, Role, StopReason, Usage,
13};
14
15/// Anthropic accepts at most 4 `cache_control` breakpoints per request. The
16/// stable system prompt always claims one when present, leaving the rest
17/// for message-level breakpoints named by `ChatRequest::cache`.
18const MAX_CACHE_BREAKPOINTS: usize = 4;
19
20/// Anthropic's server-side tool search tool (docs/design/68 §5/§11):
21/// prepended to `tools` whenever any tool in the request is deferred, so the
22/// model can discover a deferred schema without it ever entering the prefix.
23const TOOL_SEARCH_TOOL: &str = "tool_search_tool_regex_20251119";
24
25/// Beta flag for the opt-in fast-mode research preview (docs/design/68 §11
26/// "Anthropic" row / the Anthropic API "Fast Mode" quick reference): sent
27/// only alongside `"speed": "fast"`, and only for a model the capability
28/// cache has confirmed supports it.
29const FAST_MODE_BETA: &str = "fast-mode-2026-02-01";
30
31pub const ANTHROPIC_VERSION: &str = "2023-06-01";
32pub const DEFAULT_BASE_URL: &str = "https://api.anthropic.com";
33
34#[derive(Debug, Clone)]
35pub struct AnthropicConfig {
36 pub api_key: String,
37 pub base_url: String,
38 pub model: String,
39 /// Opt-in fast-mode research preview (config key under
40 /// `[providers.anthropic]`, default off): premium pricing, its own
41 /// rate-limit bucket, and restricted to specific models — the adapter
42 /// only ever sends `speed: "fast"` when this is true AND the model's
43 /// discovered capabilities confirm support
44 /// (`models::anthropic_fast_mode_allowed`).
45 pub fast_mode: bool,
46}
47
48#[derive(Clone)]
49pub struct AnthropicProvider {
50 http: reqwest::Client,
51 config: AnthropicConfig,
52 gate: ProviderGate,
53}
54
55impl AnthropicProvider {
56 pub fn new(config: AnthropicConfig) -> Result<Self, LlmError> {
57 let http = reqwest::Client::builder()
58 .connect_timeout(std::time::Duration::from_secs(30))
59 .build()
60 .map_err(|e| LlmError::Network(e.to_string()))?;
61 Ok(AnthropicProvider {
62 gate: ProviderGate::new(&config.base_url, &config.api_key),
63 http,
64 config,
65 })
66 }
67}
68
69/// `effort_allowed` and `fast_mode` are resolved by the caller (`stream`)
70/// from the in-process capability cache (`models.rs`) before this is
71/// called, so building the body stays a pure function of its inputs.
72pub fn build_body(
73 request: &ChatRequest,
74 effort_allowed: bool,
75 fast_mode: bool,
76) -> Result<Value, LlmError> {
77 let boundary = current_turn_boundary(&request.messages);
78 let system_takes_a_slot = request.system.is_some();
79 let message_breakpoints = select_message_breakpoints(
80 request.cache.as_ref(),
81 MAX_CACHE_BREAKPOINTS.saturating_sub(usize::from(system_takes_a_slot)),
82 );
83
84 let mut messages = Vec::with_capacity(request.messages.len());
85 for (i, m) in request.messages.iter().enumerate() {
86 validate_message(m)?;
87 let stripped;
88 let rendered = if i < boundary {
89 stripped = strip_thinking(m);
90 &stripped
91 } else {
92 m
93 };
94 let mut value =
95 serde_json::to_value(rendered).map_err(|e| LlmError::Parse(e.to_string()))?;
96 unwrap_provider_blocks(&mut value);
97 if message_breakpoints.contains(&i) {
98 mark_last_block_ephemeral(&mut value);
99 }
100 messages.push(value);
101 }
102
103 let mut body = serde_json::json!({
104 "model": request.model,
105 "max_tokens": request.max_tokens,
106 "messages": messages,
107 "stream": true,
108 });
109 if let Some(system) = &request.system {
110 // Prompt caching: the system prompt is stable across turns, so mark
111 // it ephemeral-cachable — long sessions stop re-paying full input
112 // cost every request.
113 body["system"] = serde_json::json!([{
114 "type": "text",
115 "text": system,
116 "cache_control": {"type": "ephemeral"}
117 }]);
118 }
119 if let Some(t) = request.temperature {
120 body["temperature"] = serde_json::json!(t);
121 }
122 if !request.tools.is_empty() {
123 let mut tools: Vec<Value> = Vec::with_capacity(request.tools.len() + 1);
124 if request.tools.iter().any(|t| t.defer) {
125 tools.push(serde_json::json!({
126 "type": TOOL_SEARCH_TOOL,
127 "name": "tool_search_tool_regex",
128 }));
129 }
130 for t in &request.tools {
131 let mut tool = serde_json::json!({
132 "name": t.name,
133 "description": t.description,
134 "input_schema": t.parameters,
135 });
136 // A deferred tool never carries `cache_control`: its schema is
137 // never in the stable prefix, so there is nothing to mark
138 // cacheable (docs/design/68 §5/§11).
139 if t.defer {
140 tool["defer_loading"] = serde_json::json!(true);
141 }
142 tools.push(tool);
143 }
144 body["tools"] = Value::Array(tools);
145 }
146 // Reasoning depth, never thinking on/off (docs/design/68 §11): current
147 // models either reject an explicit `{"type":"disabled"}`/`budget_tokens`
148 // outright or silently degrade tool-call reliability under it, so this
149 // adapter never sends either — `thinking` is simply omitted, which runs
150 // adaptive on every current model. `effort` is the one supported dial,
151 // and `think == Some(false)` (a side-dispatch that wants a fast, cheap
152 // answer) maps onto its lowest level only when the caller has not
153 // already asked for a specific one.
154 if fast_mode {
155 body["speed"] = serde_json::json!("fast");
156 }
157 if effort_allowed {
158 let effort = request
159 .effort
160 .or_else(|| (request.think == Some(false)).then_some(Effort::Low));
161 if let Some(effort) = effort {
162 body["output_config"] = serde_json::json!({ "effort": effort.as_str() });
163 }
164 }
165 Ok(body)
166}
167
168/// Replaces a serialized `ContentBlock::Provider` (`{"type":"provider",
169/// "kind":"server_tool_use", "raw": {...}}`) with its `raw` value, which
170/// already carries the provider's own `"type"` and every original field —
171/// so a `server_tool_use` / `tool_search_tool_result` block the API sent
172/// round-trips back to it byte-for-byte (docs/design/68 §5/§12). Other
173/// content blocks are left untouched.
174fn unwrap_provider_blocks(message: &mut Value) {
175 let Some(content) = message.get_mut("content").and_then(|c| c.as_array_mut()) else {
176 return;
177 };
178 for block in content.iter_mut() {
179 if block.get("type").and_then(|t| t.as_str()) == Some("provider")
180 && let Some(raw) = block.get("raw").cloned()
181 {
182 *block = raw;
183 }
184 }
185}
186
187/// Picks which message indices get a `cache_control` breakpoint, bounded by
188/// `budget`. When the hinted set is larger than the budget, the oldest
189/// (lowest-index) breakpoints are dropped first — the newest history is the
190/// one most likely to be replayed unchanged on the next turn.
191fn select_message_breakpoints(
192 cache: Option<&crate::types::CacheHints>,
193 budget: usize,
194) -> std::collections::BTreeSet<usize> {
195 let Some(cache) = cache else {
196 return std::collections::BTreeSet::new();
197 };
198 let mut indices: Vec<usize> = cache
199 .breakpoints
200 .iter()
201 .filter_map(|b| b.after_message)
202 .collect();
203 indices.sort_unstable();
204 indices.dedup();
205 if indices.len() > budget {
206 indices = indices.split_off(indices.len() - budget);
207 }
208 indices.into_iter().collect()
209}
210
211/// Attaches `cache_control: {"type": "ephemeral"}` to the last content
212/// block of an already-serialized message, matching Anthropic's rule that a
213/// breakpoint marks the end of the cached range.
214fn mark_last_block_ephemeral(message: &mut Value) {
215 if let Some(last) = message
216 .get_mut("content")
217 .and_then(|c| c.as_array_mut())
218 .and_then(|arr| arr.last_mut())
219 && let Some(obj) = last.as_object_mut()
220 {
221 obj.insert(
222 "cache_control".to_string(),
223 serde_json::json!({"type": "ephemeral"}),
224 );
225 }
226}
227
228fn validate_message(m: &Message) -> Result<(), LlmError> {
229 for block in &m.content {
230 match (m.role, block) {
231 (Role::Assistant, ContentBlock::ToolResult { .. }) => {
232 return Err(LlmError::InvalidRequest(
233 "tool_result blocks must appear in user messages".into(),
234 ));
235 }
236 (Role::User, ContentBlock::ToolUse { .. }) => {
237 return Err(LlmError::InvalidRequest(
238 "tool_use blocks must appear in assistant messages".into(),
239 ));
240 }
241 (Role::Assistant, ContentBlock::Image { .. }) => {
242 return Err(LlmError::InvalidRequest(
243 "image blocks must appear in user messages".into(),
244 ));
245 }
246 _ => {}
247 }
248 }
249 Ok(())
250}
251
252/// Provider phrasing that identifies a 400 as caused specifically by the
253/// `output_config.effort` parameter, distinct from an unrelated validation
254/// failure — same "key off the provider's own wording" approach as
255/// `LlmError::classify_400`'s over-length markers.
256fn rejects_effort(message: &str) -> bool {
257 let normalized = message.to_ascii_lowercase();
258 normalized.contains("effort") || normalized.contains("output_config")
259}
260
261pub fn map_status_error(status: u16, body: &str, retry_after: Option<u64>) -> LlmError {
262 let message = serde_json::from_str::<Value>(body)
263 .ok()
264 .and_then(|v| {
265 v.pointer("/error/message")
266 .and_then(|m| m.as_str().map(String::from))
267 })
268 .unwrap_or_else(|| body.chars().take(500).collect());
269
270 match status {
271 401 | 403 => LlmError::Auth(message),
272 400 => LlmError::classify_400(message),
273 404 | 413 | 422 => LlmError::InvalidRequest(message),
274 429 => LlmError::RateLimit {
275 message,
276 retry_after_secs: retry_after,
277 },
278 503 | 529 => LlmError::Overloaded(message),
279 _ => LlmError::Api { status, message },
280 }
281}
282
283struct Accumulator {
284 message: AssistantMessage,
285 index_map: std::collections::HashMap<u64, usize>,
286 raw_json: std::collections::HashMap<u64, String>,
287}
288
289impl Accumulator {
290 fn new(model: &str) -> Self {
291 Accumulator {
292 message: AssistantMessage::empty(model),
293 index_map: std::collections::HashMap::new(),
294 raw_json: std::collections::HashMap::new(),
295 }
296 }
297
298 fn convert(&mut self, data: &str) -> Result<Option<StreamEvent>, LlmError> {
299 let v: Value = serde_json::from_str(data)
300 .map_err(|e| LlmError::Parse(format!("bad event json: {e}")))?;
301 let kind = v
302 .get("type")
303 .and_then(|t| t.as_str())
304 .ok_or_else(|| LlmError::Parse("event missing type".into()))?
305 .to_string();
306
307 match kind.as_str() {
308 "ping" => Ok(None),
309 "message_start" => {
310 if let Some(model) = v.pointer("/message/model").and_then(|m| m.as_str()) {
311 self.message.model = model.to_string();
312 }
313 if let Some(usage) = parse_usage(&v) {
314 self.message.usage = usage;
315 }
316 Ok(Some(StreamEvent::Start {
317 partial: self.message.clone(),
318 }))
319 }
320 "content_block_start" => {
321 let idx = v.get("index").and_then(|i| i.as_u64()).unwrap_or(0);
322 let block = v.get("content_block").cloned().unwrap_or(Value::Null);
323 let block_type = block.get("type").and_then(|t| t.as_str()).unwrap_or("text");
324 let our_idx = self.message.content.len();
325 match block_type {
326 "tool_use" => {
327 let id = block
328 .get("id")
329 .and_then(|i| i.as_str())
330 .unwrap_or_default()
331 .to_string();
332 let name = block
333 .get("name")
334 .and_then(|n| n.as_str())
335 .unwrap_or_default()
336 .to_string();
337 self.message.content.push(ContentBlock::ToolUse {
338 id: id.clone(),
339 name: name.clone(),
340 input: Value::Object(Default::default()),
341 });
342 self.index_map.insert(idx, our_idx);
343 Ok(Some(StreamEvent::ToolUseStart {
344 index: our_idx,
345 id,
346 name,
347 partial: self.message.clone(),
348 }))
349 }
350 "thinking" => {
351 self.message.content.push(ContentBlock::Thinking {
352 text: String::new(),
353 signature: None,
354 });
355 self.index_map.insert(idx, our_idx);
356 Ok(None)
357 }
358 "text" => {
359 self.message.content.push(ContentBlock::text(String::new()));
360 self.index_map.insert(idx, our_idx);
361 Ok(None)
362 }
363 // A block type this build does not otherwise interpret
364 // (`server_tool_use`, `tool_search_tool_result`, and any
365 // future addition) is kept opaque rather than silently
366 // folded into an empty text block, so it round-trips
367 // unchanged (docs/design/68 §5/§12).
368 other => {
369 self.message.content.push(ContentBlock::Provider {
370 kind: other.to_string(),
371 raw: block.clone(),
372 });
373 self.index_map.insert(idx, our_idx);
374 Ok(None)
375 }
376 }
377 }
378 "content_block_delta" => {
379 let idx = v.get("index").and_then(|i| i.as_u64()).unwrap_or(0);
380 let delta = v.get("delta").cloned().unwrap_or(Value::Null);
381 let delta_type = delta.get("type").and_then(|t| t.as_str()).unwrap_or("");
382 let Some(&our_idx) = self.index_map.get(&idx) else {
383 return Ok(None);
384 };
385 match delta_type {
386 "text_delta" => {
387 let text = delta.get("text").and_then(|t| t.as_str()).unwrap_or("");
388 append_text(&mut self.message.content[our_idx], text);
389 Ok(Some(StreamEvent::TextDelta {
390 delta: text.to_string(),
391 partial: self.message.clone(),
392 }))
393 }
394 "thinking_delta" => {
395 let text = delta.get("thinking").and_then(|t| t.as_str()).unwrap_or("");
396 append_thinking(&mut self.message.content[our_idx], text);
397 Ok(Some(StreamEvent::ThinkingDelta {
398 delta: text.to_string(),
399 partial: self.message.clone(),
400 }))
401 }
402 "signature_delta" => {
403 let sig = delta
404 .get("signature")
405 .and_then(|t| t.as_str())
406 .unwrap_or("");
407 if let ContentBlock::Thinking { signature, .. } =
408 &mut self.message.content[our_idx]
409 {
410 *signature = Some(sig.to_string());
411 }
412 Ok(None)
413 }
414 "input_json_delta" => {
415 let json = delta
416 .get("partial_json")
417 .and_then(|t| t.as_str())
418 .unwrap_or("");
419 let raw = self.raw_json.entry(idx).or_default();
420 raw.push_str(json);
421 let parsed: Value =
422 serde_json::from_str(raw).unwrap_or(Value::Object(Default::default()));
423 match &mut self.message.content[our_idx] {
424 ContentBlock::ToolUse { input, .. } => {
425 *input = parsed;
426 }
427 // `server_tool_use` streams its input the same
428 // way `tool_use` does; keep the opaque block's
429 // raw JSON in sync so it round-trips complete.
430 ContentBlock::Provider { raw: block_raw, .. } => {
431 if let Some(obj) = block_raw.as_object_mut() {
432 obj.insert("input".to_string(), parsed);
433 }
434 return Ok(None);
435 }
436 _ => return Ok(None),
437 }
438 Ok(Some(StreamEvent::ToolInputDelta {
439 index: our_idx,
440 delta: json.to_string(),
441 partial: self.message.clone(),
442 }))
443 }
444 _ => Ok(None),
445 }
446 }
447 "content_block_stop" => {
448 let idx = v.get("index").and_then(|i| i.as_u64()).unwrap_or(0);
449 if let Some(&our_idx) = self.index_map.get(&idx) {
450 let raw = self.raw_json.get(&idx).cloned();
451 match &mut self.message.content[our_idx] {
452 ContentBlock::ToolUse { input, .. } => {
453 let raw = raw.unwrap_or_default();
454 *input = if raw.trim().is_empty() {
455 Value::Object(Default::default())
456 } else {
457 serde_json::from_str(&raw).map_err(|e| {
458 LlmError::Parse(format!(
459 "tool input json invalid at block stop: {e}"
460 ))
461 })?
462 };
463 }
464 ContentBlock::Provider { raw: block_raw, .. } => {
465 if let Some(raw) = raw
466 && !raw.trim().is_empty()
467 && let Ok(parsed) = serde_json::from_str::<Value>(&raw)
468 && let Some(obj) = block_raw.as_object_mut()
469 {
470 obj.insert("input".to_string(), parsed);
471 }
472 }
473 _ => {}
474 }
475 }
476 Ok(None)
477 }
478 "message_delta" => {
479 if let Some(reason) = v.pointer("/delta/stop_reason").and_then(|r| r.as_str()) {
480 self.message.stop_reason = match reason {
481 "tool_use" => StopReason::ToolUse,
482 "max_tokens" => StopReason::MaxTokens,
483 _ => StopReason::EndTurn,
484 };
485 }
486 if let Some(out) = v.pointer("/usage/output_tokens").and_then(|u| u.as_u64()) {
487 self.message.usage.output_tokens = out;
488 }
489 Ok(None)
490 }
491 "message_stop" => Ok(Some(StreamEvent::End {
492 message: self.message.clone(),
493 })),
494 "error" => {
495 let err_type = v
496 .pointer("/error/type")
497 .and_then(|t| t.as_str())
498 .unwrap_or("api_error");
499 let msg = v
500 .pointer("/error/message")
501 .and_then(|m| m.as_str())
502 .unwrap_or("unknown provider error");
503 Err(match err_type {
504 "overloaded_error" => LlmError::Overloaded(msg.to_string()),
505 "rate_limit_error" => LlmError::RateLimit {
506 message: msg.to_string(),
507 retry_after_secs: None,
508 },
509 "invalid_request_error" => LlmError::InvalidRequest(msg.to_string()),
510 "authentication_error" | "permission_error" => LlmError::Auth(msg.to_string()),
511 _ => LlmError::Api {
512 status: 0,
513 message: format!("{err_type}: {msg}"),
514 },
515 })
516 }
517 // Unknown event types are additive provider extensions; a new
518 // type from the server must never fail an in-flight request.
519 _other => Ok(None),
520 }
521 }
522}
523
524fn append_text(block: &mut ContentBlock, text: &str) {
525 if let ContentBlock::Text { text: t } = block {
526 t.push_str(text);
527 }
528}
529
530fn append_thinking(block: &mut ContentBlock, text: &str) {
531 if let ContentBlock::Thinking { text: t, .. } = block {
532 t.push_str(text);
533 }
534}
535
536fn parse_usage(v: &Value) -> Option<Usage> {
537 let u = v.pointer("/message/usage")?;
538 Some(Usage {
539 input_tokens: u.get("input_tokens").and_then(|x| x.as_u64()).unwrap_or(0),
540 output_tokens: u.get("output_tokens").and_then(|x| x.as_u64()).unwrap_or(0),
541 cache_read_input_tokens: u.get("cache_read_input_tokens").and_then(|x| x.as_u64()),
542 cache_creation_input_tokens: u
543 .get("cache_creation_input_tokens")
544 .and_then(|x| x.as_u64()),
545 ..Default::default()
546 })
547}
548
549#[async_trait::async_trait]
550impl Provider for AnthropicProvider {
551 fn name(&self) -> &str {
552 "anthropic"
553 }
554
555 fn circuit_key(&self) -> String {
556 crate::gate::route_identity(self.name(), &self.config.base_url, &self.config.api_key)
557 }
558
559 async fn stream(
560 &self,
561 request: ChatRequest,
562 cancel: CancellationToken,
563 ) -> Result<EventStream, LlmError> {
564 let provider_permit = self.gate.acquire(&cancel).await?;
565
566 // Fast mode is an opt-in, model-restricted research preview
567 // (docs/design/68 §11): discover support once per model id in the
568 // background — never blocking this request — so a later request
569 // for the same model has an answer cached. This request itself
570 // conservatively skips fast mode until that lookup lands.
571 if self.config.fast_mode && !crate::models::anthropic_fast_mode_known(&request.model) {
572 let auth = crate::registry::ProviderAuth {
573 api_key: self.config.api_key.clone(),
574 base_url: Some(self.config.base_url.clone()),
575 ..Default::default()
576 };
577 let model = request.model.clone();
578 tokio::spawn(async move {
579 if let Ok(caps) = crate::models::anthropic_model_capabilities(&auth, &model).await {
580 crate::models::record_anthropic_capabilities(&model, caps);
581 }
582 });
583 }
584
585 let mut effort_allowed = crate::models::anthropic_effort_allowed(&request.model);
586 let mut fast_mode =
587 self.config.fast_mode && crate::models::anthropic_fast_mode_allowed(&request.model);
588 let mut response = self
589 .send_once(&request, effort_allowed, fast_mode, &cancel)
590 .await?;
591
592 // One narrow, same-turn retry per failure class (docs/design/68
593 // §11): an unsupported `output_config.effort` 400s naming the
594 // parameter, and fast mode has its own rate-limit bucket that can
595 // 429 while standard speed would not. Each retry strips exactly the
596 // feature that failed and resends once; a second failure is final.
597 if !response.status().is_success() {
598 let status = response.status().as_u16();
599 if status == 400 && effort_allowed {
600 let retry_after = retry_after_header(&response);
601 let text = response.text().await.unwrap_or_default();
602 if rejects_effort(&text) {
603 crate::models::mark_anthropic_effort_unsupported(&request.model);
604 effort_allowed = false;
605 response = self
606 .send_once(&request, effort_allowed, fast_mode, &cancel)
607 .await?;
608 } else {
609 return Err(map_status_error(status, &text, retry_after));
610 }
611 } else if status == 429 && fast_mode {
612 fast_mode = false;
613 response = self
614 .send_once(&request, effort_allowed, fast_mode, &cancel)
615 .await?;
616 }
617 }
618
619 let status = response.status();
620 if !status.is_success() {
621 let retry_after = retry_after_header(&response);
622 let text = response.text().await.unwrap_or_default();
623 return Err(map_status_error(status.as_u16(), &text, retry_after));
624 }
625
626 let model = request.model.clone();
627 let (sink, stream_rx) = channel(256);
628 let mut byte_stream = response.bytes_stream();
629 let mut decoder = SseDecoder::new();
630 let mut acc = Accumulator::new(&model);
631
632 tokio::spawn(async move {
633 drive_stream(&mut byte_stream, &mut decoder, &mut acc, sink, cancel).await;
634 });
635
636 Ok(stream_rx.with_guard(provider_permit))
637 }
638}
639
640impl AnthropicProvider {
641 async fn send_once(
642 &self,
643 request: &ChatRequest,
644 effort_allowed: bool,
645 fast_mode: bool,
646 cancel: &CancellationToken,
647 ) -> Result<reqwest::Response, LlmError> {
648 let url = format!("{}/v1/messages", self.config.base_url.trim_end_matches('/'));
649 let body = build_body(request, effort_allowed, fast_mode)?;
650 let mut req = self
651 .http
652 .post(&url)
653 .header("x-api-key", &self.config.api_key)
654 .header("anthropic-version", ANTHROPIC_VERSION);
655 if fast_mode {
656 req = req.header("anthropic-beta", FAST_MODE_BETA);
657 }
658 let send_fut = req.json(&body).send();
659 tokio::select! {
660 _ = cancel.cancelled() => Err(LlmError::Aborted { partial: None }),
661 r = send_fut => r.map_err(|e| LlmError::Network(e.to_string())),
662 }
663 }
664}
665
666fn retry_after_header(response: &reqwest::Response) -> Option<u64> {
667 response
668 .headers()
669 .get("retry-after")
670 .and_then(|v| v.to_str().ok())
671 .and_then(|v| v.parse::<u64>().ok())
672}
673
674async fn drive_stream<S>(
675 byte_stream: &mut S,
676 decoder: &mut SseDecoder,
677 acc: &mut Accumulator,
678 mut sink: EventSink,
679 cancel: CancellationToken,
680) where
681 S: futures::Stream<Item = reqwest::Result<bytes::Bytes>> + Unpin,
682{
683 let mut saw_end = false;
684 loop {
685 tokio::select! {
686 _ = cancel.cancelled() => {
687 let partial = (!acc.message.content.is_empty()).then(|| Box::new(acc.message.clone()));
688 sink.close_error(LlmError::Aborted { partial }).await;
689 return;
690 }
691 chunk = byte_stream.next() => {
692 match chunk {
693 Some(Ok(bytes)) => {
694 decoder.push(&bytes);
695 while let Some(frame) = decoder.next_frame() {
696 match acc.convert(&frame.data) {
697 Ok(Some(event)) => {
698 if matches!(event, StreamEvent::End { .. }) {
699 saw_end = true;
700 }
701 sink.push(event);
702 }
703 Ok(None) => {}
704 Err(e) => {
705 sink.close_error(e).await;
706 return;
707 }
708 }
709 }
710 }
711 Some(Err(e)) => {
712 sink.close_error(LlmError::Network(e.to_string())).await;
713 return;
714 }
715 None => {
716 if saw_end {
717 sink.close_message(acc.message.clone()).await;
718 } else {
719 sink.close_error(LlmError::Parse(
720 "stream closed before message_stop".into(),
721 )).await;
722 }
723 return;
724 }
725 }
726 }
727 }
728 }
729}
730
731#[cfg(test)]
732mod build_body_tests {
733 #![allow(clippy::unwrap_used, clippy::expect_used)]
734 use super::*;
735 use crate::types::{CacheBreakpoint, CacheHints, ToolDefinition};
736
737 fn message_with_thinking(text: &str) -> Message {
738 Message::assistant(vec![
739 ContentBlock::Thinking {
740 text: "reasoning".into(),
741 signature: Some("sig".into()),
742 },
743 ContentBlock::text(text),
744 ])
745 }
746
747 #[test]
748 fn no_cache_hints_only_caches_the_system_prompt() {
749 let mut req = ChatRequest::new("claude-sonnet-4-5");
750 req.system = Some("be helpful".into());
751 req.messages = vec![Message::user_text("hi")];
752 let body = build_body(&req, true, false).unwrap();
753 assert_eq!(body["system"][0]["cache_control"]["type"], "ephemeral");
754 assert!(body["messages"][0].get("cache_control").is_none());
755 }
756
757 #[test]
758 fn breakpoints_mark_the_last_block_of_named_messages() {
759 let mut req = ChatRequest::new("claude-sonnet-4-5");
760 req.system = Some("be helpful".into());
761 req.messages = vec![
762 Message::user_text("first"),
763 Message::assistant(vec![ContentBlock::text("first answer")]),
764 Message::user_text("second"),
765 ];
766 req.cache = Some(CacheHints {
767 session_key: "s1".into(),
768 breakpoints: vec![CacheBreakpoint {
769 after_message: Some(1),
770 }],
771 });
772 let body = build_body(&req, true, false).unwrap();
773 let msgs = body["messages"].as_array().unwrap();
774 assert_eq!(msgs[1]["content"][0]["cache_control"]["type"], "ephemeral");
775 assert!(msgs[0].get("cache_control").is_none());
776 assert!(msgs[2]["content"][0].get("cache_control").is_none());
777 }
778
779 #[test]
780 fn breakpoint_budget_drops_the_oldest_first_when_over_four_total() {
781 let mut req = ChatRequest::new("claude-sonnet-4-5");
782 req.system = Some("be helpful".into()); // claims one of the 4 slots
783 req.messages = (0..6)
784 .map(|i| Message::user_text(format!("m{i}")))
785 .collect();
786 req.cache = Some(CacheHints {
787 session_key: "s1".into(),
788 breakpoints: (0..6)
789 .map(|i| CacheBreakpoint {
790 after_message: Some(i),
791 })
792 .collect(),
793 });
794 let body = build_body(&req, true, false).unwrap();
795 let msgs = body["messages"].as_array().unwrap();
796 let cached: Vec<usize> = (0..6)
797 .filter(|&i| msgs[i]["content"][0].get("cache_control").is_some())
798 .collect();
799 // 4 total breakpoints minus the system prompt's slot leaves 3
800 // message breakpoints, keeping the newest (highest-index) ones.
801 assert_eq!(cached, vec![3, 4, 5]);
802 }
803
804 #[test]
805 fn thinking_before_the_current_turn_is_stripped() {
806 let mut req = ChatRequest::new("claude-sonnet-4-5");
807 req.messages = vec![
808 Message::user_text("first"),
809 message_with_thinking("first answer"),
810 Message::user_text("second, a fresh directive"),
811 message_with_thinking("second answer"),
812 ];
813 let body = build_body(&req, true, false).unwrap();
814 let msgs = body["messages"].as_array().unwrap();
815 // Turn 1 (indices 0-1) is closed: thinking must not survive.
816 assert!(
817 !msgs[1]["content"]
818 .as_array()
819 .unwrap()
820 .iter()
821 .any(|b| b["type"] == "thinking")
822 );
823 // Turn 2 (indices 2-3) is the current turn: thinking is replayed.
824 assert!(
825 msgs[3]["content"]
826 .as_array()
827 .unwrap()
828 .iter()
829 .any(|b| b["type"] == "thinking")
830 );
831 }
832
833 #[test]
834 fn thinking_survives_mid_turn_tool_continuation() {
835 let mut req = ChatRequest::new("claude-sonnet-4-5");
836 req.messages = vec![
837 Message::user_text("do the thing"),
838 Message::assistant(vec![
839 ContentBlock::Thinking {
840 text: "reasoning".into(),
841 signature: Some("sig".into()),
842 },
843 ContentBlock::ToolUse {
844 id: "t1".into(),
845 name: "search".into(),
846 input: serde_json::json!({}),
847 },
848 ]),
849 Message {
850 role: Role::User,
851 content: vec![ContentBlock::tool_result("t1", "result")],
852 },
853 ];
854 let body = build_body(&req, true, false).unwrap();
855 let msgs = body["messages"].as_array().unwrap();
856 // The whole exchange is one turn (the tool-result message is not a
857 // fresh directive), so thinking in message 1 must still be present.
858 assert!(
859 msgs[1]["content"]
860 .as_array()
861 .unwrap()
862 .iter()
863 .any(|b| b["type"] == "thinking")
864 );
865 }
866
867 #[test]
868 fn deferred_tools_render_defer_loading_and_prepend_the_search_tool() {
869 let mut req = ChatRequest::new("claude-sonnet-4-5");
870 req.messages = vec![Message::user_text("hi")];
871 req.tools = vec![
872 ToolDefinition::new("core_tool", "always visible", serde_json::json!({})),
873 ToolDefinition::new("rare_tool", "rarely needed", serde_json::json!({})).deferred(),
874 ];
875 let body = build_body(&req, true, false).unwrap();
876 let tools = body["tools"].as_array().unwrap();
877 assert_eq!(tools[0]["type"], "tool_search_tool_regex_20251119");
878 let core = tools.iter().find(|t| t["name"] == "core_tool").unwrap();
879 assert!(core.get("defer_loading").is_none());
880 assert!(core.get("cache_control").is_none());
881 let rare = tools.iter().find(|t| t["name"] == "rare_tool").unwrap();
882 assert_eq!(rare["defer_loading"], true);
883 assert!(
884 rare.get("cache_control").is_none(),
885 "a deferred tool must never carry cache_control"
886 );
887 }
888
889 #[test]
890 fn no_deferred_tools_means_no_search_tool_is_sent() {
891 let mut req = ChatRequest::new("claude-sonnet-4-5");
892 req.messages = vec![Message::user_text("hi")];
893 req.tools = vec![ToolDefinition::new(
894 "core_tool",
895 "always visible",
896 serde_json::json!({}),
897 )];
898 let body = build_body(&req, true, false).unwrap();
899 let tools = body["tools"].as_array().unwrap();
900 assert_eq!(tools.len(), 1);
901 assert_eq!(tools[0]["name"], "core_tool");
902 }
903
904 #[test]
905 fn a_provider_block_replays_verbatim_on_the_wire() {
906 let raw = serde_json::json!({
907 "type": "server_tool_use",
908 "id": "srvtoolu_1",
909 "name": "tool_search_tool_regex",
910 "input": {"pattern": "weather"}
911 });
912 let mut req = ChatRequest::new("claude-sonnet-4-5");
913 req.messages = vec![Message::assistant(vec![ContentBlock::Provider {
914 kind: "server_tool_use".into(),
915 raw: raw.clone(),
916 }])];
917 let body = build_body(&req, true, false).unwrap();
918 let sent = &body["messages"][0]["content"][0];
919 assert_eq!(sent, &raw, "a provider block must round-trip unchanged");
920 assert_ne!(sent["type"], "provider");
921 }
922
923 #[test]
924 fn a_provider_block_streams_and_finalizes_its_input() {
925 let mut acc = Accumulator::new("claude-sonnet-4-5");
926 acc.convert(
927 &serde_json::json!({
928 "type": "content_block_start",
929 "index": 0,
930 "content_block": {
931 "type": "server_tool_use",
932 "id": "srvtoolu_1",
933 "name": "tool_search_tool_regex",
934 }
935 })
936 .to_string(),
937 )
938 .unwrap();
939 acc.convert(
940 &serde_json::json!({
941 "type": "content_block_delta",
942 "index": 0,
943 "delta": {"type": "input_json_delta", "partial_json": "{\"pattern\""}
944 })
945 .to_string(),
946 )
947 .unwrap();
948 acc.convert(
949 &serde_json::json!({
950 "type": "content_block_delta",
951 "index": 0,
952 "delta": {"type": "input_json_delta", "partial_json": ":\"weather\"}"}
953 })
954 .to_string(),
955 )
956 .unwrap();
957 acc.convert(&serde_json::json!({"type": "content_block_stop", "index": 0}).to_string())
958 .unwrap();
959 let ContentBlock::Provider { kind, raw } = &acc.message.content[0] else {
960 unreachable!("expected a Provider block");
961 };
962 assert_eq!(kind, "server_tool_use");
963 assert_eq!(raw["input"]["pattern"], "weather");
964 assert_eq!(raw["type"], "server_tool_use");
965 }
966
967 #[test]
968 fn explicit_effort_is_rendered_under_output_config() {
969 let mut req = ChatRequest::new("claude-opus-5");
970 req.messages = vec![Message::user_text("hi")];
971 req.effort = Some(Effort::XHigh);
972 let body = build_body(&req, true, false).unwrap();
973 assert_eq!(body["output_config"]["effort"], "xhigh");
974 }
975
976 #[test]
977 fn think_false_falls_back_to_low_effort_when_effort_is_unset() {
978 let mut req = ChatRequest::new("claude-haiku-4-5");
979 req.messages = vec![Message::user_text("classify this")];
980 req.think = Some(false);
981 let body = build_body(&req, true, false).unwrap();
982 assert_eq!(body["output_config"]["effort"], "low");
983 }
984
985 #[test]
986 fn explicit_effort_wins_over_the_think_false_fallback() {
987 let mut req = ChatRequest::new("claude-opus-5");
988 req.messages = vec![Message::user_text("hi")];
989 req.think = Some(false);
990 req.effort = Some(Effort::Max);
991 let body = build_body(&req, true, false).unwrap();
992 assert_eq!(body["output_config"]["effort"], "max");
993 }
994
995 #[test]
996 fn no_output_config_when_neither_effort_nor_think_false_is_set() {
997 let mut req = ChatRequest::new("claude-sonnet-5");
998 req.messages = vec![Message::user_text("hi")];
999 let body = build_body(&req, true, false).unwrap();
1000 assert!(body.get("output_config").is_none());

1use futures::StreamExt;
2use serde_json::Value;
3use tokio_util::sync::CancellationToken;
4
5use crate::Provider;
6use crate::error::LlmError;
7use crate::gate::ProviderGate;
8use crate::sse::SseDecoder;
9use crate::stream::{EventSink, EventStream, StreamEvent, channel};
10use crate::turn::{current_turn_boundary, strip_thinking};
11use crate::types::{
12 AssistantMessage, ChatRequest, ContentBlock, Effort, Message, Role, StopReason, Usage,
13};
14
15/// Anthropic accepts at most 4 `cache_control` breakpoints per request. The
16/// stable system prompt always claims one when present, leaving the rest
17/// for message-level breakpoints named by `ChatRequest::cache`.
18const MAX_CACHE_BREAKPOINTS: usize = 4;
19
20/// Anthropic's server-side tool search tool (docs/design/68 §5/§11):
21/// prepended to `tools` whenever any tool in the request is deferred, so the
22/// model can discover a deferred schema without it ever entering the prefix.
23const TOOL_SEARCH_TOOL: &str = "tool_search_tool_regex_20251119";
24
25/// Beta flag for the opt-in fast-mode research preview (docs/design/68 §11
26/// "Anthropic" row / the Anthropic API "Fast Mode" quick reference): sent
27/// only alongside `"speed": "fast"`, and only for a model the capability
28/// cache has confirmed supports it.
29const FAST_MODE_BETA: &str = "fast-mode-2026-02-01";
30
31pub const ANTHROPIC_VERSION: &str = "2023-06-01";
32pub const DEFAULT_BASE_URL: &str = "https://api.anthropic.com";
33
34#[derive(Debug, Clone)]
35pub struct AnthropicConfig {
36 pub api_key: String,
37 pub base_url: String,
38 pub model: String,
39 /// Opt-in fast-mode research preview (config key under
40 /// `[providers.anthropic]`, default off): premium pricing, its own
41 /// rate-limit bucket, and restricted to specific models — the adapter
42 /// only ever sends `speed: "fast"` when this is true AND the model's
43 /// discovered capabilities confirm support
44 /// (`models::anthropic_fast_mode_allowed`).
45 pub fast_mode: bool,
46}
47
48#[derive(Clone)]
49pub struct AnthropicProvider {
50 http: reqwest::Client,
51 config: AnthropicConfig,
52 gate: ProviderGate,
53}
54
55impl AnthropicProvider {
56 pub fn new(config: AnthropicConfig) -> Result<Self, LlmError> {
57 let http = reqwest::Client::builder()
58 .connect_timeout(std::time::Duration::from_secs(30))
59 .build()
60 .map_err(|e| LlmError::Network(e.to_string()))?;
61 Ok(AnthropicProvider {
62 gate: ProviderGate::new(&config.base_url, &config.api_key),
63 http,
64 config,
65 })
66 }
67}
68
69/// `effort_allowed` and `fast_mode` are resolved by the caller (`stream`)
70/// from the in-process capability cache (`models.rs`) before this is
71/// called, so building the body stays a pure function of its inputs.
72pub fn build_body(
73 request: &ChatRequest,
74 effort_allowed: bool,
75 fast_mode: bool,
76) -> Result<Value, LlmError> {
77 let boundary = current_turn_boundary(&request.messages);
78 let system_takes_a_slot = request.system.is_some();
79 let message_breakpoints = select_message_breakpoints(
80 request.cache.as_ref(),
81 MAX_CACHE_BREAKPOINTS.saturating_sub(usize::from(system_takes_a_slot)),
82 );
83
84 let mut messages = Vec::with_capacity(request.messages.len());
85 for (i, m) in request.messages.iter().enumerate() {
86 validate_message(m)?;
87 let stripped;
88 let rendered = if i < boundary {
89 stripped = strip_thinking(m);
90 &stripped
91 } else {
92 m
93 };
94 let mut value =
95 serde_json::to_value(rendered).map_err(|e| LlmError::Parse(e.to_string()))?;
96 unwrap_provider_blocks(&mut value);
97 if message_breakpoints.contains(&i) {
98 mark_last_block_ephemeral(&mut value);
99 }
100 messages.push(value);
101 }
102
103 let mut body = serde_json::json!({
104 "model": request.model,
105 "max_tokens": request.max_tokens,
106 "messages": messages,
107 "stream": true,
108 });
109 if let Some(system) = &request.system {
110 // Prompt caching: the system prompt is stable across turns, so mark
111 // it ephemeral-cachable — long sessions stop re-paying full input
112 // cost every request.
113 body["system"] = serde_json::json!([{
114 "type": "text",
115 "text": system,
116 "cache_control": {"type": "ephemeral"}
117 }]);
118 }
119 if let Some(t) = request.temperature {
120 body["temperature"] = serde_json::json!(t);
121 }
122 if !request.tools.is_empty() {
123 let mut tools: Vec<Value> = Vec::with_capacity(request.tools.len() + 1);
124 if request.tools.iter().any(|t| t.defer) {
125 tools.push(serde_json::json!({
126 "type": TOOL_SEARCH_TOOL,
127 "name": "tool_search_tool_regex",
128 }));
129 }
130 for t in &request.tools {
131 let mut tool = serde_json::json!({
132 "name": t.name,
133 "description": t.description,
134 "input_schema": t.parameters,
135 });
136 // A deferred tool never carries `cache_control`: its schema is
137 // never in the stable prefix, so there is nothing to mark
138 // cacheable (docs/design/68 §5/§11).
139 if t.defer {
140 tool["defer_loading"] = serde_json::json!(true);
141 }
142 tools.push(tool);
143 }
144 body["tools"] = Value::Array(tools);
145 }
146 // Reasoning depth, never thinking on/off (docs/design/68 §11): current
147 // models either reject an explicit `{"type":"disabled"}`/`budget_tokens`
148 // outright or silently degrade tool-call reliability under it, so this
149 // adapter never sends either — `thinking` is simply omitted, which runs
150 // adaptive on every current model. `effort` is the one supported dial,
151 // and `think == Some(false)` (a side-dispatch that wants a fast, cheap
152 // answer) maps onto its lowest level only when the caller has not
153 // already asked for a specific one.
154 if fast_mode {
155 body["speed"] = serde_json::json!("fast");
156 }
157 if effort_allowed {
158 let effort = request
159 .effort
160 .or_else(|| (request.think == Some(false)).then_some(Effort::Low));
161 if let Some(effort) = effort {
162 body["output_config"] = serde_json::json!({ "effort": effort.as_str() });
163 }
164 }
165 Ok(body)
166}
167
168/// Replaces a serialized `ContentBlock::Provider` (`{"type":"provider",
169/// "kind":"server_tool_use", "raw": {...}}`) with its `raw` value, which
170/// already carries the provider's own `"type"` and every original field —
171/// so a `server_tool_use` / `tool_search_tool_result` block the API sent
172/// round-trips back to it byte-for-byte (docs/design/68 §5/§12). Other
173/// content blocks are left untouched.
174fn unwrap_provider_blocks(message: &mut Value) {
175 let Some(content) = message.get_mut("content").and_then(|c| c.as_array_mut()) else {
176 return;
177 };
178 for block in content.iter_mut() {
179 if block.get("type").and_then(|t| t.as_str()) == Some("provider")
180 && let Some(raw) = block.get("raw").cloned()
181 {
182 *block = raw;
183 }
184 }
185}
186
187/// Picks which message indices get a `cache_control` breakpoint, bounded by
188/// `budget`. When the hinted set is larger than the budget, the oldest
189/// (lowest-index) breakpoints are dropped first — the newest history is the
190/// one most likely to be replayed unchanged on the next turn.
191fn select_message_breakpoints(
192 cache: Option<&crate::types::CacheHints>,
193 budget: usize,
194) -> std::collections::BTreeSet<usize> {
195 let Some(cache) = cache else {
196 return std::collections::BTreeSet::new();
197 };
198 let mut indices: Vec<usize> = cache
199 .breakpoints
200 .iter()
201 .filter_map(|b| b.after_message)
202 .collect();
203 indices.sort_unstable();
204 indices.dedup();
205 if indices.len() > budget {
206 indices = indices.split_off(indices.len() - budget);
207 }
208 indices.into_iter().collect()
209}
210
211/// Attaches `cache_control: {"type": "ephemeral"}` to the last content
212/// block of an already-serialized message, matching Anthropic's rule that a
213/// breakpoint marks the end of the cached range.
214fn mark_last_block_ephemeral(message: &mut Value) {
215 if let Some(last) = message
216 .get_mut("content")
217 .and_then(|c| c.as_array_mut())
218 .and_then(|arr| arr.last_mut())
219 && let Some(obj) = last.as_object_mut()
220 {
221 obj.insert(
222 "cache_control".to_string(),
223 serde_json::json!({"type": "ephemeral"}),
224 );
225 }
226}
227
228fn validate_message(m: &Message) -> Result<(), LlmError> {
229 for block in &m.content {
230 match (m.role, block) {
231 (Role::Assistant, ContentBlock::ToolResult { .. }) => {
232 return Err(LlmError::InvalidRequest(
233 "tool_result blocks must appear in user messages".into(),
234 ));
235 }
236 (Role::User, ContentBlock::ToolUse { .. }) => {
237 return Err(LlmError::InvalidRequest(
238 "tool_use blocks must appear in assistant messages".into(),
239 ));
240 }
241 (Role::Assistant, ContentBlock::Image { .. }) => {
242 return Err(LlmError::InvalidRequest(
243 "image blocks must appear in user messages".into(),
244 ));
245 }
246 _ => {}
247 }
248 }
249 Ok(())
250}
251
252/// Provider phrasing that identifies a 400 as caused specifically by the
253/// `output_config.effort` parameter, distinct from an unrelated validation
254/// failure — same "key off the provider's own wording" approach as
255/// `LlmError::classify_400`'s over-length markers.
256fn rejects_effort(message: &str) -> bool {
257 let normalized = message.to_ascii_lowercase();
258 normalized.contains("effort") || normalized.contains("output_config")
259}
260
261pub fn map_status_error(status: u16, body: &str, retry_after: Option<u64>) -> LlmError {
262 let message = serde_json::from_str::<Value>(body)
263 .ok()
264 .and_then(|v| {
265 v.pointer("/error/message")
266 .and_then(|m| m.as_str().map(String::from))
267 })
268 .unwrap_or_else(|| body.chars().take(500).collect());
269
270 match status {
271 401 | 403 => LlmError::Auth(message),
272 400 => LlmError::classify_400(message),
273 404 | 413 | 422 => LlmError::InvalidRequest(message),
274 429 => LlmError::RateLimit {
275 message,
276 retry_after_secs: retry_after,
277 },
278 503 | 529 => LlmError::Overloaded(message),
279 _ => LlmError::Api { status, message },
280 }
281}
282
283struct Accumulator {
284 message: AssistantMessage,
285 index_map: std::collections::HashMap<u64, usize>,
286 raw_json: std::collections::HashMap<u64, String>,
287}
288
289impl Accumulator {
290 fn new(model: &str) -> Self {
291 Accumulator {
292 message: AssistantMessage::empty(model),
293 index_map: std::collections::HashMap::new(),
294 raw_json: std::collections::HashMap::new(),
295 }
296 }
297
298 fn convert(&mut self, data: &str) -> Result<Option<StreamEvent>, LlmError> {
299 let v: Value = serde_json::from_str(data)
300 .map_err(|e| LlmError::Parse(format!("bad event json: {e}")))?;
301 let kind = v
302 .get("type")
303 .and_then(|t| t.as_str())
304 .ok_or_else(|| LlmError::Parse("event missing type".into()))?
305 .to_string();
306
307 match kind.as_str() {
308 "ping" => Ok(None),
309 "message_start" => {
310 if let Some(model) = v.pointer("/message/model").and_then(|m| m.as_str()) {
311 self.message.model = model.to_string();
312 }
313 if let Some(usage) = parse_usage(&v) {
314 self.message.usage = usage;
315 }
316 Ok(Some(StreamEvent::Start {
317 partial: self.message.clone(),
318 }))
319 }
320 "content_block_start" => {
321 let idx = v.get("index").and_then(|i| i.as_u64()).unwrap_or(0);
322 let block = v.get("content_block").cloned().unwrap_or(Value::Null);
323 let block_type = block.get("type").and_then(|t| t.as_str()).unwrap_or("text");
324 let our_idx = self.message.content.len();
325 match block_type {
326 "tool_use" => {
327 let id = block
328 .get("id")
329 .and_then(|i| i.as_str())
330 .unwrap_or_default()
331 .to_string();
332 let name = block
333 .get("name")
334 .and_then(|n| n.as_str())
335 .unwrap_or_default()
336 .to_string();
337 self.message.content.push(ContentBlock::ToolUse {
338 id: id.clone(),
339 name: name.clone(),
340 input: Value::Object(Default::default()),
341 });
342 self.index_map.insert(idx, our_idx);
343 Ok(Some(StreamEvent::ToolUseStart {
344 index: our_idx,
345 id,
346 name,
347 partial: self.message.clone(),
348 }))
349 }
350 "thinking" => {
351 self.message.content.push(ContentBlock::Thinking {
352 text: String::new(),
353 signature: None,
354 });
355 self.index_map.insert(idx, our_idx);
356 Ok(None)
357 }
358 "text" => {
359 self.message.content.push(ContentBlock::text(String::new()));
360 self.index_map.insert(idx, our_idx);
361 Ok(None)
362 }
363 // A block type this build does not otherwise interpret
364 // (`server_tool_use`, `tool_search_tool_result`, and any
365 // future addition) is kept opaque rather than silently
366 // folded into an empty text block, so it round-trips
367 // unchanged (docs/design/68 §5/§12).
368 other => {
369 self.message.content.push(ContentBlock::Provider {
370 kind: other.to_string(),
371 raw: block.clone(),
372 });
373 self.index_map.insert(idx, our_idx);
374 Ok(None)
375 }
376 }
377 }
378 "content_block_delta" => {
379 let idx = v.get("index").and_then(|i| i.as_u64()).unwrap_or(0);
380 let delta = v.get("delta").cloned().unwrap_or(Value::Null);
381 let delta_type = delta.get("type").and_then(|t| t.as_str()).unwrap_or("");
382 let Some(&our_idx) = self.index_map.get(&idx) else {
383 return Ok(None);
384 };
385 match delta_type {
386 "text_delta" => {
387 let text = delta.get("text").and_then(|t| t.as_str()).unwrap_or("");
388 append_text(&mut self.message.content[our_idx], text);
389 Ok(Some(StreamEvent::TextDelta {
390 delta: text.to_string(),
391 partial: self.message.clone(),
392 }))
393 }
394 "thinking_delta" => {
395 let text = delta.get("thinking").and_then(|t| t.as_str()).unwrap_or("");
396 append_thinking(&mut self.message.content[our_idx], text);
397 Ok(Some(StreamEvent::ThinkingDelta {
398 delta: text.to_string(),
399 partial: self.message.clone(),
400 }))
401 }
402 "signature_delta" => {
403 let sig = delta
404 .get("signature")
405 .and_then(|t| t.as_str())
406 .unwrap_or("");
407 if let ContentBlock::Thinking { signature, .. } =
408 &mut self.message.content[our_idx]
409 {
410 *signature = Some(sig.to_string());
411 }
412 Ok(None)
413 }
414 "input_json_delta" => {
415 let json = delta
416 .get("partial_json")
417 .and_then(|t| t.as_str())
418 .unwrap_or("");
419 let raw = self.raw_json.entry(idx).or_default();
420 raw.push_str(json);
421 let parsed: Value =
422 serde_json::from_str(raw).unwrap_or(Value::Object(Default::default()));
423 match &mut self.message.content[our_idx] {
424 ContentBlock::ToolUse { input, .. } => {
425 *input = parsed;
426 }
427 // `server_tool_use` streams its input the same
428 // way `tool_use` does; keep the opaque block's
429 // raw JSON in sync so it round-trips complete.
430 ContentBlock::Provider { raw: block_raw, .. } => {
431 if let Some(obj) = block_raw.as_object_mut() {
432 obj.insert("input".to_string(), parsed);
433 }
434 return Ok(None);
435 }
436 _ => return Ok(None),
437 }
438 Ok(Some(StreamEvent::ToolInputDelta {
439 index: our_idx,
440 delta: json.to_string(),
441 partial: self.message.clone(),
442 }))
443 }
444 _ => Ok(None),
445 }
446 }
447 "content_block_stop" => {
448 let idx = v.get("index").and_then(|i| i.as_u64()).unwrap_or(0);
449 if let Some(&our_idx) = self.index_map.get(&idx) {
450 let raw = self.raw_json.get(&idx).cloned();
451 match &mut self.message.content[our_idx] {
452 ContentBlock::ToolUse { input, .. } => {
453 let raw = raw.unwrap_or_default();
454 *input = if raw.trim().is_empty() {
455 Value::Object(Default::default())
456 } else {
457 serde_json::from_str(&raw).map_err(|e| {
458 LlmError::Parse(format!(
459 "tool input json invalid at block stop: {e}"
460 ))
461 })?
462 };
463 }
464 ContentBlock::Provider { raw: block_raw, .. } => {
465 if let Some(raw) = raw
466 && !raw.trim().is_empty()
467 && let Ok(parsed) = serde_json::from_str::<Value>(&raw)
468 && let Some(obj) = block_raw.as_object_mut()
469 {
470 obj.insert("input".to_string(), parsed);
471 }
472 }
473 _ => {}
474 }
475 }
476 Ok(None)
477 }
478 "message_delta" => {
479 if let Some(reason) = v.pointer("/delta/stop_reason").and_then(|r| r.as_str()) {
480 self.message.stop_reason = match reason {
481 "tool_use" => StopReason::ToolUse,
482 "max_tokens" => StopReason::MaxTokens,
483 _ => StopReason::EndTurn,
484 };
485 }
486 if let Some(out) = v.pointer("/usage/output_tokens").and_then(|u| u.as_u64()) {
487 self.message.usage.output_tokens = out;
488 }
489 Ok(None)
490 }
491 "message_stop" => Ok(Some(StreamEvent::End {
492 message: self.message.clone(),
493 })),
494 "error" => {
495 let err_type = v
496 .pointer("/error/type")
497 .and_then(|t| t.as_str())
498 .unwrap_or("api_error");
499 let msg = v
500 .pointer("/error/message")
501 .and_then(|m| m.as_str())
502 .unwrap_or("unknown provider error");
503 Err(match err_type {
504 "overloaded_error" => LlmError::Overloaded(msg.to_string()),
505 "rate_limit_error" => LlmError::RateLimit {
506 message: msg.to_string(),
507 retry_after_secs: None,
508 },
509 "invalid_request_error" => LlmError::InvalidRequest(msg.to_string()),
510 "authentication_error" | "permission_error" => LlmError::Auth(msg.to_string()),
511 _ => LlmError::Api {
512 status: 0,
513 message: format!("{err_type}: {msg}"),
514 },
515 })
516 }
517 // Unknown event types are additive provider extensions; a new
518 // type from the server must never fail an in-flight request.
519 _other => Ok(None),
520 }
521 }
522}
523
524fn append_text(block: &mut ContentBlock, text: &str) {
525 if let ContentBlock::Text { text: t } = block {
526 t.push_str(text);
527 }
528}
529
530fn append_thinking(block: &mut ContentBlock, text: &str) {
531 if let ContentBlock::Thinking { text: t, .. } = block {
532 t.push_str(text);
533 }
534}
535
536fn parse_usage(v: &Value) -> Option<Usage> {
537 let u = v.pointer("/message/usage")?;
538 Some(Usage {
539 input_tokens: u.get("input_tokens").and_then(|x| x.as_u64()).unwrap_or(0),
540 output_tokens: u.get("output_tokens").and_then(|x| x.as_u64()).unwrap_or(0),
541 cache_read_input_tokens: u.get("cache_read_input_tokens").and_then(|x| x.as_u64()),
542 cache_creation_input_tokens: u
543 .get("cache_creation_input_tokens")
544 .and_then(|x| x.as_u64()),
545 ..Default::default()
546 })
547}
548
549#[async_trait::async_trait]
550impl Provider for AnthropicProvider {
551 fn name(&self) -> &str {
552 "anthropic"
553 }
554
555 fn circuit_key(&self) -> String {
556 crate::gate::route_identity(self.name(), &self.config.base_url, &self.config.api_key)
557 }
558
559 async fn stream(
560 &self,
561 request: ChatRequest,
562 cancel: CancellationToken,
563 ) -> Result<EventStream, LlmError> {
564 let provider_permit = self.gate.acquire(&cancel).await?;
565
566 // Fast mode is an opt-in, model-restricted research preview
567 // (docs/design/68 §11): discover support once per model id in the
568 // background — never blocking this request — so a later request
569 // for the same model has an answer cached. This request itself
570 // conservatively skips fast mode until that lookup lands.
571 if self.config.fast_mode && !crate::models::anthropic_fast_mode_known(&request.model) {
572 let auth = crate::registry::ProviderAuth {
573 api_key: self.config.api_key.clone(),
574 base_url: Some(self.config.base_url.clone()),
575 ..Default::default()
576 };
577 let model = request.model.clone();
578 tokio::spawn(async move {
579 if let Ok(caps) = crate::models::anthropic_model_capabilities(&auth, &model).await {
580 crate::models::record_anthropic_capabilities(&model, caps);
581 }
582 });
583 }
584
585 let mut effort_allowed = crate::models::anthropic_effort_allowed(&request.model);
586 let mut fast_mode =
587 self.config.fast_mode && crate::models::anthropic_fast_mode_allowed(&request.model);
588 let mut response = self
589 .send_once(&request, effort_allowed, fast_mode, &cancel)
590 .await?;
591
592 // One narrow, same-turn retry per failure class (docs/design/68
593 // §11): an unsupported `output_config.effort` 400s naming the
594 // parameter, and fast mode has its own rate-limit bucket that can
595 // 429 while standard speed would not. Each retry strips exactly the
596 // feature that failed and resends once; a second failure is final.
597 if !response.status().is_success() {
598 let status = response.status().as_u16();
599 if status == 400 && effort_allowed {
600 let retry_after = retry_after_header(&response);
601 let text = response.text().await.unwrap_or_default();
602 if rejects_effort(&text) {
603 crate::models::mark_anthropic_effort_unsupported(&request.model);
604 effort_allowed = false;
605 response = self
606 .send_once(&request, effort_allowed, fast_mode, &cancel)
607 .await?;
608 } else {
609 return Err(map_status_error(status, &text, retry_after));
610 }
611 } else if status == 429 && fast_mode {
612 fast_mode = false;
613 response = self
614 .send_once(&request, effort_allowed, fast_mode, &cancel)
615 .await?;
616 }
617 }
618
619 let status = response.status();
620 if !status.is_success() {
621 let retry_after = retry_after_header(&response);
622 let text = response.text().await.unwrap_or_default();
623 return Err(map_status_error(status.as_u16(), &text, retry_after));
624 }
625
626 let model = request.model.clone();
627 let (sink, stream_rx) = channel(256);
628 let mut byte_stream = response.bytes_stream();
629 let mut decoder = SseDecoder::new();
630 let mut acc = Accumulator::new(&model);
631
632 tokio::spawn(async move {
633 drive_stream(&mut byte_stream, &mut decoder, &mut acc, sink, cancel).await;
634 });
635
636 Ok(stream_rx.with_guard(provider_permit))
637 }
638}
639
640impl AnthropicProvider {
641 async fn send_once(
642 &self,
643 request: &ChatRequest,
644 effort_allowed: bool,
645 fast_mode: bool,
646 cancel: &CancellationToken,
647 ) -> Result<reqwest::Response, LlmError> {
648 let url = format!("{}/v1/messages", self.config.base_url.trim_end_matches('/'));
649 let body = build_body(request, effort_allowed, fast_mode)?;
650 let mut req = self
651 .http
652 .post(&url)
653 .header("x-api-key", &self.config.api_key)
654 .header("anthropic-version", ANTHROPIC_VERSION);
655 if fast_mode {
656 req = req.header("anthropic-beta", FAST_MODE_BETA);
657 }
658 let send_fut = req.json(&body).send();
659 tokio::select! {
660 _ = cancel.cancelled() => Err(LlmError::Aborted { partial: None }),
661 r = send_fut => r.map_err(|e| LlmError::Network(e.to_string())),
662 }
663 }
664}
665
666fn retry_after_header(response: &reqwest::Response) -> Option<u64> {
667 response
668 .headers()
669 .get("retry-after")
670 .and_then(|v| v.to_str().ok())
671 .and_then(|v| v.parse::<u64>().ok())
672}
673
674async fn drive_stream<S>(
675 byte_stream: &mut S,
676 decoder: &mut SseDecoder,
677 acc: &mut Accumulator,
678 mut sink: EventSink,
679 cancel: CancellationToken,
680) where
681 S: futures::Stream<Item = reqwest::Result<bytes::Bytes>> + Unpin,
682{
683 let mut saw_end = false;
684 loop {
685 tokio::select! {
686 _ = cancel.cancelled() => {
687 let partial = (!acc.message.content.is_empty()).then(|| Box::new(acc.message.clone()));
688 sink.close_error(LlmError::Aborted { partial }).await;
689 return;
690 }
691 chunk = byte_stream.next() => {
692 match chunk {
693 Some(Ok(bytes)) => {
694 decoder.push(&bytes);
695 while let Some(frame) = decoder.next_frame() {
696 match acc.convert(&frame.data) {
697 Ok(Some(event)) => {
698 if matches!(event, StreamEvent::End { .. }) {
699 saw_end = true;
700 }
701 sink.push(event);
702 }
703 Ok(None) => {}
704 Err(e) => {
705 sink.close_error(e).await;
706 return;
707 }
708 }
709 }
710 }
711 Some(Err(e)) => {
712 sink.close_error(LlmError::Network(e.to_string())).await;
713 return;
714 }
715 None => {
716 if saw_end {
717 sink.close_message(acc.message.clone()).await;
718 } else {
719 sink.close_error(LlmError::Parse(
720 "stream closed before message_stop".into(),
721 )).await;
722 }
723 return;
724 }
725 }
726 }
727 }
728 }
729}
730
731#[cfg(test)]
732mod build_body_tests {
733 #![allow(clippy::unwrap_used, clippy::expect_used)]
734 use super::*;
735 use crate::types::{CacheBreakpoint, CacheHints, ToolDefinition};
736
737 fn message_with_thinking(text: &str) -> Message {
738 Message::assistant(vec![
739 ContentBlock::Thinking {
740 text: "reasoning".into(),
741 signature: Some("sig".into()),
742 },
743 ContentBlock::text(text),
744 ])
745 }
746
747 #[test]
748 fn no_cache_hints_only_caches_the_system_prompt() {
749 let mut req = ChatRequest::new("claude-sonnet-4-5");
750 req.system = Some("be helpful".into());
751 req.messages = vec![Message::user_text("hi")];
752 let body = build_body(&req, true, false).unwrap();
753 assert_eq!(body["system"][0]["cache_control"]["type"], "ephemeral");
754 assert!(body["messages"][0].get("cache_control").is_none());
755 }
756
757 #[test]
758 fn breakpoints_mark_the_last_block_of_named_messages() {
759 let mut req = ChatRequest::new("claude-sonnet-4-5");
760 req.system = Some("be helpful".into());
761 req.messages = vec![
762 Message::user_text("first"),
763 Message::assistant(vec![ContentBlock::text("first answer")]),
764 Message::user_text("second"),
765 ];
766 req.cache = Some(CacheHints {
767 session_key: "s1".into(),
768 breakpoints: vec![CacheBreakpoint {
769 after_message: Some(1),
770 }],
771 });
772 let body = build_body(&req, true, false).unwrap();
773 let msgs = body["messages"].as_array().unwrap();
774 assert_eq!(msgs[1]["content"][0]["cache_control"]["type"], "ephemeral");
775 assert!(msgs[0].get("cache_control").is_none());
776 assert!(msgs[2]["content"][0].get("cache_control").is_none());
777 }
778
779 #[test]
780 fn breakpoint_budget_drops_the_oldest_first_when_over_four_total() {
781 let mut req = ChatRequest::new("claude-sonnet-4-5");
782 req.system = Some("be helpful".into()); // claims one of the 4 slots
783 req.messages = (0..6)
784 .map(|i| Message::user_text(format!("m{i}")))
785 .collect();
786 req.cache = Some(CacheHints {
787 session_key: "s1".into(),
788 breakpoints: (0..6)
789 .map(|i| CacheBreakpoint {
790 after_message: Some(i),
791 })
792 .collect(),
793 });
794 let body = build_body(&req, true, false).unwrap();
795 let msgs = body["messages"].as_array().unwrap();
796 let cached: Vec<usize> = (0..6)
797 .filter(|&i| msgs[i]["content"][0].get("cache_control").is_some())
798 .collect();
799 // 4 total breakpoints minus the system prompt's slot leaves 3
800 // message breakpoints, keeping the newest (highest-index) ones.
801 assert_eq!(cached, vec![3, 4, 5]);
802 }
803
804 #[test]
805 fn thinking_before_the_current_turn_is_stripped() {
806 let mut req = ChatRequest::new("claude-sonnet-4-5");
807 req.messages = vec![
808 Message::user_text("first"),
809 message_with_thinking("first answer"),
810 Message::user_text("second, a fresh directive"),
811 message_with_thinking("second answer"),
812 ];
813 let body = build_body(&req, true, false).unwrap();
814 let msgs = body["messages"].as_array().unwrap();
815 // Turn 1 (indices 0-1) is closed: thinking must not survive.
816 assert!(
817 !msgs[1]["content"]
818 .as_array()
819 .unwrap()
820 .iter()
821 .any(|b| b["type"] == "thinking")
822 );
823 // Turn 2 (indices 2-3) is the current turn: thinking is replayed.
824 assert!(
825 msgs[3]["content"]
826 .as_array()
827 .unwrap()
828 .iter()
829 .any(|b| b["type"] == "thinking")
830 );
831 }
832
833 #[test]
834 fn thinking_survives_mid_turn_tool_continuation() {
835 let mut req = ChatRequest::new("claude-sonnet-4-5");
836 req.messages = vec![
837 Message::user_text("do the thing"),
838 Message::assistant(vec![
839 ContentBlock::Thinking {
840 text: "reasoning".into(),
841 signature: Some("sig".into()),
842 },
843 ContentBlock::ToolUse {
844 id: "t1".into(),
845 name: "search".into(),
846 input: serde_json::json!({}),
847 },
848 ]),
849 Message {
850 role: Role::User,
851 content: vec![ContentBlock::tool_result("t1", "result")],
852 },
853 ];
854 let body = build_body(&req, true, false).unwrap();
855 let msgs = body["messages"].as_array().unwrap();
856 // The whole exchange is one turn (the tool-result message is not a
857 // fresh directive), so thinking in message 1 must still be present.
858 assert!(
859 msgs[1]["content"]
860 .as_array()
861 .unwrap()
862 .iter()
863 .any(|b| b["type"] == "thinking")
864 );
865 }
866
867 #[test]
868 fn deferred_tools_render_defer_loading_and_prepend_the_search_tool() {
869 let mut req = ChatRequest::new("claude-sonnet-4-5");
870 req.messages = vec![Message::user_text("hi")];
871 req.tools = vec![
872 ToolDefinition::new("core_tool", "always visible", serde_json::json!({})),
873 ToolDefinition::new("rare_tool", "rarely needed", serde_json::json!({})).deferred(),
874 ];
875 let body = build_body(&req, true, false).unwrap();
876 let tools = body["tools"].as_array().unwrap();
877 assert_eq!(tools[0]["type"], "tool_search_tool_regex_20251119");
878 let core = tools.iter().find(|t| t["name"] == "core_tool").unwrap();
879 assert!(core.get("defer_loading").is_none());
880 assert!(core.get("cache_control").is_none());
881 let rare = tools.iter().find(|t| t["name"] == "rare_tool").unwrap();
882 assert_eq!(rare["defer_loading"], true);
883 assert!(
884 rare.get("cache_control").is_none(),
885 "a deferred tool must never carry cache_control"
886 );
887 }
888
889 #[test]
890 fn no_deferred_tools_means_no_search_tool_is_sent() {
891 let mut req = ChatRequest::new("claude-sonnet-4-5");
892 req.messages = vec![Message::user_text("hi")];
893 req.tools = vec![ToolDefinition::new(
894 "core_tool",
895 "always visible",
896 serde_json::json!({}),
897 )];
898 let body = build_body(&req, true, false).unwrap();
899 let tools = body["tools"].as_array().unwrap();
900 assert_eq!(tools.len(), 1);
901 assert_eq!(tools[0]["name"], "core_tool");
902 }
903
904 #[test]
905 fn a_provider_block_replays_verbatim_on_the_wire() {
906 let raw = serde_json::json!({
907 "type": "server_tool_use",
908 "id": "srvtoolu_1",
909 "name": "tool_search_tool_regex",
910 "input": {"pattern": "weather"}
911 });
912 let mut req = ChatRequest::new("claude-sonnet-4-5");
913 req.messages = vec![Message::assistant(vec![ContentBlock::Provider {
914 kind: "server_tool_use".into(),
915 raw: raw.clone(),
916 }])];
917 let body = build_body(&req, true, false).unwrap();
918 let sent = &body["messages"][0]["content"][0];
919 assert_eq!(sent, &raw, "a provider block must round-trip unchanged");
920 assert_ne!(sent["type"], "provider");
921 }
922
923 #[test]
924 fn a_provider_block_streams_and_finalizes_its_input() {
925 let mut acc = Accumulator::new("claude-sonnet-4-5");
926 acc.convert(
927 &serde_json::json!({
928 "type": "content_block_start",
929 "index": 0,
930 "content_block": {
931 "type": "server_tool_use",
932 "id": "srvtoolu_1",
933 "name": "tool_search_tool_regex",
934 }
935 })
936 .to_string(),
937 )
938 .unwrap();
939 acc.convert(
940 &serde_json::json!({
941 "type": "content_block_delta",
942 "index": 0,
943 "delta": {"type": "input_json_delta", "partial_json": "{\"pattern\""}
944 })
945 .to_string(),
946 )
947 .unwrap();
948 acc.convert(
949 &serde_json::json!({
950 "type": "content_block_delta",
951 "index": 0,
952 "delta": {"type": "input_json_delta", "partial_json": ":\"weather\"}"}
953 })
954 .to_string(),
955 )
956 .unwrap();
957 acc.convert(&serde_json::json!({"type": "content_block_stop", "index": 0}).to_string())
958 .unwrap();
959 let ContentBlock::Provider { kind, raw } = &acc.message.content[0] else {
960 unreachable!("expected a Provider block");
961 };
962 assert_eq!(kind, "server_tool_use");
963 assert_eq!(raw["input"]["pattern"], "weather");
964 assert_eq!(raw["type"], "server_tool_use");
965 }
966
967 #[test]
968 fn explicit_effort_is_rendered_under_output_config() {
969 let mut req = ChatRequest::new("claude-opus-5");
970 req.messages = vec![Message::user_text("hi")];
971 req.effort = Some(Effort::XHigh);
972 let body = build_body(&req, true, false).unwrap();
973 assert_eq!(body["output_config"]["effort"], "xhigh");
974 }
975
976 #[test]
977 fn think_false_falls_back_to_low_effort_when_effort_is_unset() {
978 let mut req = ChatRequest::new("claude-haiku-4-5");
979 req.messages = vec![Message::user_text("classify this")];
980 req.think = Some(false);
981 let body = build_body(&req, true, false).unwrap();
982 assert_eq!(body["output_config"]["effort"], "low");
983 }
984
985 #[test]
986 fn explicit_effort_wins_over_the_think_false_fallback() {
987 let mut req = ChatRequest::new("claude-opus-5");
988 req.messages = vec![Message::user_text("hi")];
989 req.think = Some(false);
990 req.effort = Some(Effort::Max);
991 let body = build_body(&req, true, false).unwrap();
992 assert_eq!(body["output_config"]["effort"], "max");
993 }
994
995 #[test]
996 fn no_output_config_when_neither_effort_nor_think_false_is_set() {
997 let mut req = ChatRequest::new("claude-sonnet-5");
998 req.messages = vec![Message::user_text("hi")];
999 let body = build_body(&req, true, false).unwrap();
1000 assert!(body.get("output_config").is_none());

Indexing the workspace…