1use serde::{Deserialize, Serialize};
2use tokio::sync::broadcast;
3use tokio_util::sync::CancellationToken;
4
5use crate::error::RuntimeError;
6use crate::event::{NodeEvent, Observable};
7use crate::message::{Message, MessageOrigin, MessagePart, MessageRole};
8use crate::provider::{
9 AssistantMessage, CallTiming, DEFAULT_STREAM_BUFFER, LlmRequest, Provider, ReasoningEffort,
10 ReasoningSelection, ReasoningWireProfile, StopReason, TokenUsage, estimate_tokens,
11};
12use crate::providers::classify_attachment_error;
13use crate::tool::BoxFut;
14
15pub struct AnthropicProvider {
16 name: String,
17 api_key: String,
18 base_url: String,
19 client: reqwest::Client,
20 max_tokens: u32,
21 anthropic_version: String,
22}
23
24impl AnthropicProvider {
25 pub fn new(name: impl Into<String>, api_key: impl Into<String>) -> Self {
26 Self {
27 name: name.into(),
28 api_key: api_key.into(),
29 base_url: "https://api.anthropic.com".into(),
30 client: reqwest::Client::new(),
31 max_tokens: 16384,
32 anthropic_version: "2023-06-01".into(),
33 }
34 }
35
36 pub fn with_base_url(mut self, url: impl Into<String>) -> Self {
37 self.base_url = url.into();
38 self
39 }
40
41 pub fn with_max_tokens(mut self, n: u32) -> Self {
42 self.max_tokens = n;
43 self
44 }
45
46 pub fn with_anthropic_version(mut self, v: impl Into<String>) -> Self {
47 self.anthropic_version = v.into();
48 self
49 }
50
51 fn validate_reasoning(&self, selection: &ReasoningSelection) -> Result<(), RuntimeError> {
52 ReasoningWireProfile::AnthropicMessages
53 .validate(selection, Some(self.max_tokens))
54 .map_err(|error| RuntimeError::ToolFailed(format!("invalid request: {error}")))
55 }
56
57 fn build_body(&self, req: &LlmRequest, stream: bool) -> Result<MessagesRequest, RuntimeError> {
58 let raw_wire: Vec<WireMessage> = req
59 .messages
60 .iter()
61 .map(|m| build_wire_message(m, false, &req.tools))
62 .collect::<Result<Vec<_>, _>>()?
63 .into_iter()
64 .flatten()
65 .collect();
66 let wire_messages = merge_consecutive_same_role(raw_wire);
67 let tools: Vec<WireTool> = req
68 .tools
69 .iter()
70 .map(|t| WireTool {
71 name: crate::tool_naming::to_wire(&t.name),
72 description: t.description.clone(),
73 input_schema: t.input_schema.clone(),
74 })
75 .collect();
76 let (thinking, output_config) = anthropic_reasoning(&req.reasoning);
77 Ok(MessagesRequest {
78 model: req.model.clone(),
79 max_tokens: self.max_tokens,
80 stream,
81 system: req.system.clone().filter(|system| !system.is_empty()),
82 messages: wire_messages,
83 tools,
84 thinking,
85 output_config,
86 cache_control: if req.cache_prompt {
87 Some(CacheControl { kind: "ephemeral" })
88 } else {
89 None
90 },
91 })
92 }
93
94 fn build_request(
95 &self,
96 req: &LlmRequest,
97 stream: bool,
98 ) -> Result<reqwest::RequestBuilder, RuntimeError> {
99 let body = self.build_body(req, stream)?;
100 Ok(self
101 .client
102 .post(format!("{}/v1/messages", self.base_url))
103 .header("x-api-key", &self.api_key)
104 .header("anthropic-version", &self.anthropic_version)
105 .json(&body))
106 }
107
108 #[doc(hidden)]
109 pub fn wire_body_bytes(&self, req: &LlmRequest, stream: bool) -> Vec<u8> {
110 serde_json::to_vec(
111 &self
112 .build_body(req, stream)
113 .expect("build Anthropic wire body"),
114 )
115 .expect("serialize wire body")
116 }
117}
118
119fn anthropic_reasoning(
120 selection: &ReasoningSelection,
121) -> (Option<ThinkingConfig>, Option<OutputConfig>) {
122 match selection {
123 ReasoningSelection::ProviderDefault | ReasoningSelection::Disabled => (None, None),
124 ReasoningSelection::Auto { .. } => (
125 Some(ThinkingConfig {
126 kind: "adaptive",
127 budget_tokens: None,
128 }),
129 None,
130 ),
131 ReasoningSelection::Effort {
132 effort: ReasoningEffort::None,
133 ..
134 } => (None, None),
135 ReasoningSelection::Effort { effort, .. } => (
136 Some(ThinkingConfig {
137 kind: "adaptive",
138 budget_tokens: None,
139 }),
140 Some(OutputConfig {
141 effort: effort.to_string(),
142 }),
143 ),
144 ReasoningSelection::BudgetTokens { tokens } => (
145 Some(ThinkingConfig {
146 kind: "enabled",
147 budget_tokens: Some(*tokens),
148 }),
149 None,
150 ),
151 }
152}
153
154fn build_wire_message(
155 m: &Message,
156 apply_cache_control: bool,
157 tools: &[crate::tool::ToolSpec],
158) -> Result<Option<WireMessage>, RuntimeError> {
159 let role = match m.role {
160 MessageRole::User => "user",
161 MessageRole::Assistant => "assistant",
162 MessageRole::System => "user",
163 MessageRole::Tool => "user",
164 };
165 let mut blocks: Vec<ContentPart> = Vec::with_capacity(m.parts.len());
166 let last_idx = m.parts.len().saturating_sub(1);
167 for (i, part) in m.parts.iter().enumerate() {
168 let block = match part {
169 MessagePart::FinalAnswerSummary { .. } => continue,
170 MessagePart::ContextRecord(record) => ContentPart::Text {
171 text: record.render_for_model(),
172 cache_control: if apply_cache_control && i == last_idx {
173 Some(CacheControl { kind: "ephemeral" })
174 } else {
175 None
176 },
177 },
178 MessagePart::CompactSummary { summary, .. } => ContentPart::Text {
179 text: summary.clone(),
180 cache_control: if apply_cache_control && i == last_idx {
181 Some(CacheControl { kind: "ephemeral" })
182 } else {
183 None
184 },
185 },
186 MessagePart::Text { text } => ContentPart::Text {
187 text: text.clone(),
188 cache_control: if apply_cache_control && i == last_idx {
189 Some(CacheControl { kind: "ephemeral" })
190 } else {
191 None
192 },
193 },
194 MessagePart::Image { source } => {
195 let data = crate::attachment_store::image_base64(source)?;
196 ContentPart::Image {
197 source: ImageSourceWire {
198 kind: "base64",
199 media_type: source.media_type.clone(),
200 data,
201 },
202 }
203 }
204 MessagePart::ToolUse {
205 id,
206 name,
207 input,
208 intent,
209 } => ContentPart::ToolUse {
210 id: id.clone(),
211 name: crate::tool_naming::to_wire(name),
212 input: crate::message::encode_tool_call_input(input, intent.as_ref(), name, tools),
213 },
214 MessagePart::Thinking {
215 thinking,
216 signature,
217 } => {
218 if thinking.is_empty() || signature.is_none() {
219 continue;
220 }
221 ContentPart::Thinking {
222 thinking: thinking.clone(),
223 signature: signature.clone(),
224 }
225 }
226 MessagePart::ToolResult {
227 tool_use_id,
228 content,
229 is_error,
230 } => ContentPart::ToolResult {
231 tool_use_id: tool_use_id.clone(),
232 content: content.clone(),
233 is_error: *is_error,
234 },
235 };
236 if matches!(&block, ContentPart::Text { text, .. } if text.is_empty()) {
237 continue;
238 }
239 blocks.push(block);
240 }
241 if blocks.is_empty() {
242 return Ok(None);
243 }
244 Ok(Some(WireMessage {
245 role,
246 content: MessageContent::Blocks(blocks),
247 }))
248}
249
250fn merge_consecutive_same_role(wire: Vec<WireMessage>) -> Vec<WireMessage> {
251 let mut out: Vec<WireMessage> = Vec::with_capacity(wire.len());
252 for msg in wire {
253 let WireMessage { role, content } = msg;
254 let mut content = Some(content);
255 if let Some(last) = out.last_mut()
256 && last.role == role
257 && let Some(msg_content) = content.take()
258 {
259 let MessageContent::Blocks(last_blocks) = &mut last.content;
260 let MessageContent::Blocks(msg_blocks) = msg_content;
261 last_blocks.extend(msg_blocks);
262 }
263 if let Some(content) = content {
264 out.push(WireMessage { role, content });
265 }
266 }
267 out
268}
269
270impl Provider for AnthropicProvider {
271 fn name(&self) -> &str {
272 &self.name
273 }
274
275 fn capabilities(&self) -> crate::provider::ProviderCapabilities {
276 crate::provider::ProviderCapabilities {
277 context_prefix_profile: crate::context_plan::ContextPrefixProfile::AnthropicMessages,
278 ..Default::default()
279 }
280 }
281
282 fn context_prefix(
283 &self,
284 req: &LlmRequest,
285 ) -> Result<crate::context_plan::ContextPrefixSnapshot, RuntimeError> {
286 let body = self.build_body(req, true)?;
287 let record_texts: std::collections::HashSet<String> = req
288 .messages
289 .iter()
290 .flat_map(|message| &message.parts)
291 .filter_map(|part| match part {
292 MessagePart::ContextRecord(record) => Some(record.render_for_model()),
293 _ => None,
294 })
295 .collect();
296 let mut builder = crate::context_plan::ContextPrefixSnapshot::builder(
297 crate::context_plan::ContextPrefixProfile::AnthropicMessages,
298 req,
299 );
300 for tool in &body.tools {
301 builder.push(crate::context_plan::ContextPrefixLane::Tools, tool)?;
302 }
303 if let Some(system) = &body.system {
304 builder.push(crate::context_plan::ContextPrefixLane::Stable, system)?;
305 }
306 for message in &body.messages {
307 builder.push(
308 crate::context_plan::ContextPrefixLane::Messages,
309 &message.role,
310 )?;
311 let MessageContent::Blocks(parts) = &message.content;
312 for part in parts {
313 let lane = match part {
314 ContentPart::Text { text, .. } if record_texts.contains(text) => {
315 crate::context_plan::ContextPrefixLane::Records
316 }
317 _ => crate::context_plan::ContextPrefixLane::Messages,
318 };
319 builder.push(lane, part)?;
320 }
321 }
322 Ok(builder.finish())
323 }
324
325 fn call<'a>(&'a self, req: LlmRequest) -> BoxFut<'a, Result<AssistantMessage, RuntimeError>> {
326 if let Err(error) = self.validate_reasoning(&req.reasoning) {
327 return Box::pin(async move { Err(error) });
328 }
329 let request = match self.build_request(&req, false) {
330 Ok(request) => request,
331 Err(error) => return Box::pin(async move { Err(error) }),
332 };
333 Box::pin(async move {
334 let resp = request.send().await.map_err(net_err)?;
335 let status = resp.status();
336 let body: MessagesResponse = if status.is_success() {
337 resp.json().await.map_err(net_err)?
338 } else {
339 let body_text = resp.text().await.unwrap_or_default();
340 if let Some(reason) = classify_attachment_error(status.as_u16(), &body_text) {
341 return Err(RuntimeError::AttachmentError { reason });
342 }
343 return Err(RuntimeError::ToolFailed(format!(
344 "anthropic http {status}: {body_text}"
345 )));
346 };
347 Ok(response_to_assistant(
348 body,
349 next_turn_id_from_req(&req),
350 &req.tools,
351 ))
352 })
353 }
354
355 fn call_streaming(&self, req: LlmRequest) -> Observable<AssistantMessage> {
356 let preflight = self
357 .validate_reasoning(&req.reasoning)
358 .and_then(|()| self.build_request(&req, true));
359 let turn_id = next_turn_id_from_req(&req);
360 let tools: Vec<crate::tool::ToolSpec> = req.tools.clone();
361 let (tx, events) = broadcast::channel(DEFAULT_STREAM_BUFFER);
362 let cancel = CancellationToken::new();
363 let cancel_for_task = cancel.clone();
364 let output: BoxFut<'static, Result<AssistantMessage, RuntimeError>> = Box::pin(
365 async move {
366 let request = preflight?;
367 use eventsource_stream::Eventsource;
368 use futures::StreamExt;
369
370 let resp = tokio::select! {
371 biased;
372 _ = cancel_for_task.cancelled() => return Err(RuntimeError::Cancelled("anthropic cancelled before send".into())),
373 r = request.send() => r.map_err(net_err)?,
374 };
375 let status = resp.status();
376 if !status.is_success() {
377 let body = resp.text().await.unwrap_or_default();
378 if let Some(reason) = classify_attachment_error(status.as_u16(), &body) {
379 return Err(RuntimeError::AttachmentError { reason });
380 }
381 return Err(RuntimeError::ToolFailed(format!(
382 "anthropic http {status}: {body}"
383 )));
384 }
385
386 let mut stream = resp.bytes_stream().eventsource();
387 let mut acc_text = String::new();
388 let mut acc_thinking = String::new();
389 let mut acc_signature: Option<String> = None;
390 let mut cumulative = 0u64;
391 let mut input_tokens: u64 = 0;
392 let mut cache_read_tokens: u64 = 0;
393 let mut cache_write_tokens: u64 = 0;
394 let mut tool_use_partial: Vec<PartialToolUse> = Vec::new();
395 let mut stop_reason = StopReason::End;
396 while let Some(event) = tokio::select! {
397 biased;
398 _ = cancel_for_task.cancelled() => None,
399 next = stream.next() => next,
400 } {
401 let event = event.map_err(|e| RuntimeError::ToolFailed(format!("sse: {e}")))?;
402 if event.data.is_empty() {
403 continue;
404 }
405 let parsed: serde_json::Value = match serde_json::from_str(&event.data) {
406 Ok(v) => v,
407 Err(_) => continue,
408 };
409 let ty = parsed.get("type").and_then(|v| v.as_str()).unwrap_or("");
410 match ty {
411 "message_start" => {
412 if let Some(usage) = parsed.pointer("/message/usage") {
413 input_tokens = usage
414 .get("input_tokens")
415 .and_then(|v| v.as_u64())
416 .unwrap_or(0);
417 cache_read_tokens = usage
418 .get("cache_read_input_tokens")
419 .and_then(|v| v.as_u64())
420 .unwrap_or(0);
421 cache_write_tokens = usage
422 .get("cache_creation_input_tokens")
423 .and_then(|v| v.as_u64())
424 .unwrap_or(0);
425 }
426 }
427 "content_block_start" => {
428 if let Some(block) = parsed.get("content_block") {
429 match block.get("type").and_then(|v| v.as_str()) {
430 Some("tool_use") => {
431 if let (Some(id), Some(name)) = (
432 block.get("id").and_then(|v| v.as_str()),
433 block.get("name").and_then(|v| v.as_str()),
434 ) {
435 tool_use_partial.push(PartialToolUse {
436 id: id.to_string(),
437 name: name.to_string(),
438 input_json: String::new(),
439 });
440 }
441 }
442 Some("thinking") => {
443 if let Some(sig) =
444 block.get("signature").and_then(|v| v.as_str())
445 {
446 acc_signature = Some(sig.to_string());
447 }
448 }
449 _ => {}
450 }
451 }
452 }
453 "content_block_delta" => {
454 if let Some(delta) = parsed.get("delta") {
455 let delta_ty =
456 delta.get("type").and_then(|v| v.as_str()).unwrap_or("");
457 if delta_ty == "text_delta" {
458 if let Some(text) = delta.get("text").and_then(|v| v.as_str()) {
459 acc_text.push_str(text);
460 cumulative += estimate_tokens(text);
461 let _ = tx.send(NodeEvent::LlmChunk {
462 text: text.to_string(),
463 cumulative_tokens: cumulative,
464 });
465 }
466 } else if delta_ty == "thinking_delta" {
467 if let Some(text) =
468 delta.get("thinking").and_then(|v| v.as_str())
469 {
470 acc_thinking.push_str(text);
471 let _ = tx.send(NodeEvent::ThinkingChunk {
472 text: text.to_string(),
473 });
474 }
475 } else if let Some(text) =
476 delta.get("reasoning_content").and_then(|v| v.as_str())
477 {
478 acc_thinking.push_str(text);
479 let _ = tx.send(NodeEvent::ThinkingChunk {
480 text: text.to_string(),
481 });
482 } else if delta_ty == "signature_delta" {
483 if let Some(sig) =
484 delta.get("signature").and_then(|v| v.as_str())
485 {
486 acc_signature = Some(sig.to_string());
487 }
488 } else if delta_ty == "input_json_delta"
489 && let Some(partial) =
490 delta.get("partial_json").and_then(|v| v.as_str())
491 {
492 let index = tool_use_partial.len().saturating_sub(1);
493 if let Some(last) = tool_use_partial.last_mut() {
494 last.input_json.push_str(partial);
495 let _ = tx.send(NodeEvent::ToolCallDraft {
496 index,
497 call_id: last.id.clone(),
498 name: crate::tool_naming::from_wire(&last.name, &tools),
499 arguments_delta: partial.to_string(),
500 });
501 }
502 }
503 }
504 }
505 "message_delta" => {
506 if let Some(out) = parsed
507 .pointer("/usage/output_tokens")
508 .and_then(|v| v.as_u64())
509 {
510 cumulative = out;
511 }
512 if let Some(inp) = parsed
513 .pointer("/usage/input_tokens")
514 .and_then(|v| v.as_u64())
515 {
516 input_tokens = inp;
517 }
518 if let Some(cr) = parsed
519 .pointer("/usage/cache_read_input_tokens")
520 .and_then(|v| v.as_u64())
521 {
522 cache_read_tokens = cr;
523 }
524 if let Some(cw) = parsed
525 .pointer("/usage/cache_creation_input_tokens")
526 .and_then(|v| v.as_u64())
527 {
528 cache_write_tokens = cw;
529 }
530 if let Some(reason) = parsed
531 .pointer("/delta/stop_reason")
532 .and_then(|v| v.as_str())
533 {
534 stop_reason = parse_stop_reason(reason);
535 }
536 }
537 "message_stop" => break,
538 _ => {}
539 }
540 }
541 if cancel_for_task.is_cancelled() {
542 let _ = tx.send(NodeEvent::LlmDone {
543 total_tokens: cumulative,
544 });
545 return Err(RuntimeError::Cancelled(
546 "anthropic cancelled mid-stream".into(),
547 ));
548 }
549 let _ = tx.send(NodeEvent::LlmDone {
550 total_tokens: cumulative,
551 });
552
553 let mut parts: Vec<MessagePart> = Vec::new();
554 if !acc_thinking.is_empty() {
555 if req.reasoning.enabled() && acc_signature.is_none() {
556 return Err(RuntimeError::ThinkingSignatureMissing);
557 }
558 parts.push(MessagePart::Thinking {
559 thinking: acc_thinking,
560 signature: acc_signature,
561 });
562 }
563 if !acc_text.is_empty() {
564 parts.push(MessagePart::Text { text: acc_text });
565 }
566 for pu in tool_use_partial {
567 let input: serde_json::Value = if pu.input_json.is_empty() {
568 serde_json::Value::Object(Default::default())
569 } else {
570 serde_json::from_str(&pu.input_json).unwrap_or(serde_json::Value::Null)
571 };
572 let name = crate::tool_naming::from_wire(&pu.name, &tools);
573 let (input, intent) =
574 crate::message::decode_tool_call_input(input, &name, &tools);
575 parts.push(MessagePart::ToolUse {
576 id: pu.id,
577 name,
578 input,
579 intent,
580 });
581 }
582 Ok(AssistantMessage {
583 message: Message {
584 role: MessageRole::Assistant,
585 parts,
586 turn_id,
587 origin: MessageOrigin::User,
588 },
589 stop_reason,
590 token_usage: TokenUsage {
591 input: input_tokens,
592 cached_input: cache_read_tokens,
593 output: cumulative,
594 cache_write: cache_write_tokens,
595 ..Default::default()
596 },
597 timing: CallTiming::default(),
598 model: String::new(),
599 response_id: None,
600 })
601 },
602 );
603 Observable {
604 output,
605 events,
606 cancel,
607 }
608 }
609
610 fn test_connection(&self) -> BoxFut<'_, Result<String, String>> {
611 let base_url = self.base_url.clone();
612 let api_key = self.api_key.clone();
613 let name = self.name.clone();
614 Box::pin(async move {
615 let client = reqwest::Client::builder()
616 .timeout(std::time::Duration::from_secs(15))
617 .build()
618 .map_err(|e| e.to_string())?;
619 let resp = client
620 .get(format!("{}/v1/models", base_url.trim_end_matches('/')))
621 .header("x-api-key", &api_key)
622 .header("anthropic-version", "2023-06-01")
623 .send()
624 .await
625 .map_err(|e| format!("connection failed — {e}"))?;
626 let status = resp.status();
627 if status.is_success() {
628 Ok(format!("\"{name}\" responded OK"))
629 } else {
630 let body = resp.text().await.unwrap_or_default();
631 Err(format!(
632 "returned {status} — {}",
633 crate::provider::bounded_utf8_prefix(&body, 200)
634 ))
635 }
636 })
637 }
638}
639
640struct PartialToolUse {
641 id: String,
642 name: String,
643 input_json: String,
644}
645
646fn response_to_assistant(
647 body: MessagesResponse,
648 turn_id: crate::event::TurnId,
649 tools: &[crate::tool::ToolSpec],
650) -> AssistantMessage {
651 let mut parts: Vec<MessagePart> = Vec::new();
652 for block in body.content {
653 match block {
654 ContentBlock::Text { text } => parts.push(MessagePart::Text { text }),
655 ContentBlock::Thinking {
656 thinking,
657 signature,
658 } => parts.push(MessagePart::Thinking {
659 thinking,
660 signature,
661 }),
662 ContentBlock::ToolUse { id, name, input } => {
663 let name = crate::tool_naming::from_wire(&name, tools);
664 let (input, intent) = crate::message::decode_tool_call_input(input, &name, tools);
665 parts.push(MessagePart::ToolUse {
666 id,
667 name,
668 input,
669 intent,
670 });
671 }
672 ContentBlock::Other => {}
673 }
674 }
675 let stop_reason = body
676 .stop_reason
677 .as_deref()
678 .map(parse_stop_reason)
679 .unwrap_or(StopReason::End);
680 let usage = body
681 .usage
682 .map(|u| TokenUsage {
683 input: u.input_tokens.unwrap_or(0),
684 cached_input: u.cache_read_input_tokens.unwrap_or(0),
685 output: u.output_tokens.unwrap_or(0),
686 cache_write: u.cache_creation_input_tokens.unwrap_or(0),
687 ..Default::default()
688 })
689 .unwrap_or_default();
690 AssistantMessage {
691 message: Message {
692 role: MessageRole::Assistant,
693 parts,
694 turn_id,
695 origin: MessageOrigin::User,
696 },
697 stop_reason,
698 token_usage: usage,
699 timing: CallTiming::default(),
700 model: body.model.unwrap_or_default(),
701 response_id: body.id,
702 }
703}
704
705fn parse_stop_reason(s: &str) -> StopReason {
706 match s {
707 "tool_use" => StopReason::ToolUse,
708 "max_tokens" => StopReason::Length,
709 _ => StopReason::End,
710 }
711}
712
713fn next_turn_id_from_req(req: &LlmRequest) -> crate::event::TurnId {
714 req.messages
715 .first()
716 .map(|m| m.turn_id.clone())
717 .unwrap_or_else(crate::event::TurnId::now)
718}
719
720fn net_err(e: reqwest::Error) -> RuntimeError {
721 RuntimeError::ToolFailed(format!("anthropic net: {e}"))
722}
723
724#[derive(Serialize, Clone)]
725struct MessagesRequest {
726 model: String,
727 max_tokens: u32,
728 stream: bool,
729 #[serde(skip_serializing_if = "Option::is_none")]
730 system: Option<String>,
731 messages: Vec<WireMessage>,
732 #[serde(skip_serializing_if = "Vec::is_empty")]
733 tools: Vec<WireTool>,
734 #[serde(skip_serializing_if = "Option::is_none")]
735 thinking: Option<ThinkingConfig>,
736 #[serde(skip_serializing_if = "Option::is_none")]
737 output_config: Option<OutputConfig>,
738 #[serde(skip_serializing_if = "Option::is_none")]
739 cache_control: Option<CacheControl>,
740}
741
742#[derive(Serialize, Clone)]
743struct ThinkingConfig {
744 #[serde(rename = "type")]
745 kind: &'static str,
746 #[serde(skip_serializing_if = "Option::is_none")]
747 budget_tokens: Option<u32>,
748}
749
750#[derive(Serialize, Clone)]
751struct OutputConfig {
752 effort: String,
753}
754
755#[derive(Serialize, Clone)]
756struct WireTool {
757 name: String,
758 #[serde(skip_serializing_if = "Option::is_none")]
759 description: Option<String>,
760 input_schema: serde_json::Value,
761}
762
763#[derive(Serialize, Clone)]
764struct WireMessage {
765 role: &'static str,
766 content: MessageContent,
767}
768
769#[derive(Serialize, Clone)]
770#[serde(untagged)]
771enum MessageContent {
772 Blocks(Vec<ContentPart>),
773}
774
775#[derive(Serialize, Clone)]
776#[serde(tag = "type", rename_all = "snake_case")]
777enum ContentPart {
778 Text {
779 text: String,
780 #[serde(skip_serializing_if = "Option::is_none")]
781 cache_control: Option<CacheControl>,
782 },
783 Thinking {
784 thinking: String,
785 #[serde(skip_serializing_if = "Option::is_none")]
786 signature: Option<String>,
787 },
788 Image {
789 source: ImageSourceWire,
790 },
791 ToolUse {
792 id: String,
793 name: String,
794 input: serde_json::Value,
795 },
796 ToolResult {
797 tool_use_id: String,
798 content: String,
799 #[serde(skip_serializing_if = "core::ops::Not::not")]
800 is_error: bool,
801 },
802}
803
804#[derive(Serialize, Clone)]
805struct ImageSourceWire {
806 #[serde(rename = "type")]
807 kind: &'static str,
808 media_type: String,
809 data: String,
810}
811
812#[derive(Serialize, Clone)]
813struct CacheControl {
814 #[serde(rename = "type")]
815 kind: &'static str,
816}
817
818#[derive(Deserialize)]
819struct MessagesResponse {
820 content: Vec<ContentBlock>,
821 #[serde(default)]
822 stop_reason: Option<String>,
823 #[serde(default)]
824 usage: Option<AnthropicUsage>,
825 #[serde(default)]
826 model: Option<String>,
827 #[serde(default)]
828 id: Option<String>,
829}
830
831#[derive(Deserialize, Default)]
832struct AnthropicUsage {
833 #[serde(default)]
834 input_tokens: Option<u64>,
835 #[serde(default)]
836 output_tokens: Option<u64>,
837 #[serde(default)]
838 cache_read_input_tokens: Option<u64>,
839 #[serde(default)]
840 cache_creation_input_tokens: Option<u64>,
841}
842
843#[derive(Deserialize)]
844#[serde(tag = "type", rename_all = "snake_case")]
845enum ContentBlock {
846 Text {
847 text: String,
848 },
849 Thinking {
850 thinking: String,
851 #[serde(default)]
852 signature: Option<String>,
853 },
854 ToolUse {
855 id: String,
856 name: String,
857 input: serde_json::Value,
858 },
859 #[serde(other)]
860 Other,
861}
862
863#[cfg(test)]
864mod tests {
865 use super::*;
866
867 struct IntentTool;
868
869 impl crate::tool::Tool for IntentTool {
870 fn name(&self) -> &str {
871 "probe"
872 }
873
874 fn tier(&self) -> crate::tool::Tier {
875 crate::tool::Tier::Zero
876 }
877
878 fn call<'a>(
879 &'a self,
880 _args: crate::tool::ToolArgs,
881 _ctx: &'a crate::tool::ToolCtx,
882 ) -> crate::tool::BoxFut<'a, crate::tool::ToolResult> {
883 Box::pin(async { Ok(crate::Value::Unit) })
884 }
885 }
886
887 #[test]
888 fn tool_call_intent_round_trips_through_tool_use_input() {
889 let tools = vec![crate::tool::tool_spec(&IntentTool)];
890 let message = Message {
891 role: MessageRole::Assistant,
892 parts: vec![MessagePart::ToolUse {
893 id: "call-1".into(),
894 name: "probe".into(),
895 input: serde_json::json!({"value": 1}),
896 intent: crate::message::ToolCallIntent::new("Inspect provider state"),
897 }],
898 turn_id: crate::event::TurnId::now(),
899 origin: crate::message::MessageOrigin::User,
900 };
901 let wire = serde_json::to_value(
902 build_wire_message(&message, false, &tools)
903 .unwrap()
904 .unwrap(),
905 )
906 .unwrap();
907 assert_eq!(
908 wire["content"][0]["input"]["_atman_intent"],
909 "Inspect provider state"
910 );
911
912 let assistant = response_to_assistant(
913 MessagesResponse {
914 content: vec![ContentBlock::ToolUse {
915 id: "call-1".into(),
916 name: "probe".into(),
917 input: wire["content"][0]["input"].clone(),
918 }],
919 stop_reason: Some("tool_use".into()),
920 usage: None,
921 model: None,
922 id: None,
923 },
924 crate::event::TurnId::now(),
925 &tools,
926 );
927 assert!(matches!(
928 assistant.message.parts.as_slice(),
929 [MessagePart::ToolUse { input, intent: Some(intent), .. }]
930 if input == &serde_json::json!({"value": 1})
931 && intent.as_str() == "Inspect provider state"
932 ));
933 }
934
935 #[test]
936 fn unsigned_thinking_only_assistant_is_not_serialized_as_empty_content() {
937 let message = Message {
938 role: MessageRole::Assistant,
939 parts: vec![MessagePart::Thinking {
940 thinking: "provider-specific reasoning".into(),
941 signature: None,
942 }],
943 turn_id: crate::event::TurnId::now(),
944 origin: MessageOrigin::User,
945 };
946
947 assert!(build_wire_message(&message, false, &[]).unwrap().is_none());
948 }
949
950 #[test]
951 fn response_usage_keeps_cache_creation_in_its_own_lane() {
952 let assistant = response_to_assistant(
953 MessagesResponse {
954 content: Vec::new(),
955 stop_reason: Some("end_turn".into()),
956 usage: Some(AnthropicUsage {
957 input_tokens: Some(20),
958 output_tokens: Some(10),
959 cache_read_input_tokens: Some(80),
960 cache_creation_input_tokens: Some(50),
961 }),
962 model: Some("model".into()),
963 id: Some("message".into()),
964 },
965 crate::event::TurnId::now(),
966 &[],
967 );
968
969 assert_eq!(assistant.token_usage.input, 20);
970 assert_eq!(assistant.token_usage.cached_input, 80);
971 assert_eq!(assistant.token_usage.cache_write, 50);
972 assert_eq!(assistant.token_usage.prompt_input(), 150);
973 }
974
975 #[test]
976 fn context_prefix_uses_messages_projection_and_preserves_appended_messages() {
977 let provider = AnthropicProvider::new("anthropic", "test-key");
978 let mut request = LlmRequest {
979 model: "claude-test".into(),
980 messages: vec![Message::user_text(crate::event::TurnId::now(), "first")],
981 system: Some("stable".into()),
982 input: crate::Value::Unit,
983 schema: None,
984 cache_prompt: true,
985 prompt_cache_key: None,
986 tools: Vec::new(),
987 reasoning: ReasoningSelection::ProviderDefault,
988 stall_timeout_secs: 0,
989 };
990 let first = provider.context_prefix(&request).unwrap();
991 let first_bytes = first.initial_observation().wire_prefix_bytes;
992 request
993 .messages
994 .push(Message::user_text(crate::event::TurnId::now(), "second"));
995 let second = provider.context_prefix(&request).unwrap();
996 let observation = second.compare("anthropic", "anthropic", "model", "model", &first);
997
998 assert_eq!(
999 observation.profile,
1000 crate::context_plan::ContextPrefixProfile::AnthropicMessages
1001 );
1002 assert_eq!(observation.reset_reason, None);
1003 assert_eq!(observation.common_prefix_bytes, first_bytes);
1004 }
1005
1006 #[test]
1007 fn internal_context_record_projects_as_framed_user_context() {
1008 let provider = AnthropicProvider::new("anthropic", "test-key");
1009 let mut request = LlmRequest {
1010 model: "claude-test".into(),
1011 messages: vec![Message::user_text(crate::event::TurnId::now(), "before")],
1012 system: Some("stable".into()),
1013 input: crate::Value::Unit,
1014 schema: None,
1015 cache_prompt: true,
1016 prompt_cache_key: None,
1017 tools: Vec::new(),
1018 reasoning: ReasoningSelection::ProviderDefault,
1019 stall_timeout_secs: 0,
1020 };
1021 let before = provider.context_prefix(&request).unwrap();
1022 let before_bytes = before.initial_observation().wire_prefix_bytes;
1023 request.messages.push(Message::context_record(
1024 crate::event::TurnId::now(),
1025 crate::context_plan::ContextRecord::new(
1026 "session.goal",
1027 1,
1028 crate::context_plan::ContextRecordAuthority::User,
1029 crate::context_plan::ContextRecordRetention::Latest,
1030 crate::context_plan::ContextRecordBody::text("finish the task"),
1031 ),
1032 ));
1033
1034 let body = serde_json::to_value(provider.build_body(&request, true).unwrap()).unwrap();
1035 assert_eq!(body["messages"][0]["role"], "user");
1036 assert!(
1037 body["messages"][0]["content"][1]["text"]
1038 .as_str()
1039 .is_some_and(|content| content.contains("finish the task"))
1040 );
1041 let after = provider.context_prefix(&request).unwrap();
1042 let observation = after.compare("anthropic", "anthropic", "model", "model", &before);
1043 assert_eq!(observation.reset_reason, None);
1044 assert_eq!(observation.common_prefix_bytes, before_bytes);
1045 }
1046}