tenant.openagents/omega

1use anyhow::{Result, anyhow};
2use collections::HashMap;
3use futures::{Stream, StreamExt};
4use language_model_core::{
5    CompactedContext, CompactionUpdate, LanguageModelCompletionError, LanguageModelCompletionEvent,
6    LanguageModelProviderId, LanguageModelProviderName, LanguageModelRequest,
7    LanguageModelRequestToolInput, LanguageModelToolChoice, LanguageModelToolResultContent,
8    LanguageModelToolUse, LanguageModelToolUseInput, MessageContent, ProviderCompactionState, Role,
9    SharedString, StopReason, TokenUsage,
10    util::{fix_streamed_json, parse_tool_arguments},
11};
12use std::pin::Pin;
13use std::str::FromStr;
14use std::sync::Arc;
15
16use crate::{
17    AdaptiveThinkingDisplay, AnthropicError, AnthropicModelMode, CacheControl, CacheControlType,
18    CacheTtl, CompactionTrigger, ContentDelta, ContextManagement, ContextManagementEdit, Event,
19    ImageSource, Message, RequestContent, ResponseContent, StringOrContents, Thinking, Tool,
20    ToolChoice, ToolResultContent, ToolResultPart, Usage, completion_error_from_anthropic,
21    completion_error_from_anthropic_api,
22};
23
24pub const COMPACTION_STATE_FORMAT: &str = "anthropic.messages.encrypted-content.v1";
25
26/// Packages a compaction block's opaque `encrypted_content` into provider
27/// state owned by `owner`.
28///
29/// Anthropic requires the metadata to be round-tripped verbatim, and only the
30/// backend whose infrastructure produced it can make sense of it. The owner
31/// recorded here is what [`provider_compaction_encrypted_content`] later
32/// compares against, so it must identify that backend, not merely the wire
33/// protocol.
34pub fn provider_compaction_state_from_encrypted_content(
35    owner: LanguageModelProviderId,
36    encrypted_content: impl Into<Arc<str>>,
37) -> ProviderCompactionState {
38    ProviderCompactionState::new(
39        owner,
40        SharedString::new_static(COMPACTION_STATE_FORMAT),
41        encrypted_content,
42    )
43}
44
45/// Recovers the `encrypted_content` to round-trip from `state` if it is owned
46/// by `owner`, or `None` when the state belongs to a different backend and the
47/// summary should be replayed without it.
48pub fn provider_compaction_encrypted_content(
49    state: &ProviderCompactionState,
50    owner: &LanguageModelProviderId,
51) -> Result<Option<Arc<str>>> {
52    if state.provider_id() != owner {
53        return Ok(None);
54    }
55    if state.format() != COMPACTION_STATE_FORMAT {
56        return Err(anyhow!(
57            "unsupported Anthropic compaction state format: {}",
58            state.format()
59        ));
60    }
61    Ok(Some(state.payload().into()))
62}
63
64#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
65pub enum AnthropicPromptCacheMode {
66    Disabled,
67    Legacy,
68    #[default]
69    Automatic,
70}
71
72fn set_cache_control(content: &mut RequestContent, cache_control: Option<CacheControl>) -> bool {
73    match content {
74        RequestContent::RedactedThinking { .. } => false,
75        RequestContent::Text {
76            cache_control: target,
77            ..
78        }
79        | RequestContent::Thinking {
80            cache_control: target,
81            ..
82        }
83        | RequestContent::Image {
84            cache_control: target,
85            ..
86        }
87        | RequestContent::ToolUse {
88            cache_control: target,
89            ..
90        }
91        | RequestContent::ToolResult {
92            cache_control: target,
93            ..
94        }
95        | RequestContent::Compaction {
96            cache_control: target,
97            ..
98        } => {
99            *target = cache_control;
100            true
101        }
102    }
103}
104
105fn mark_last_cacheable_content(content: &mut [RequestContent], cache_control: CacheControl) {
106    for content in content.iter_mut().rev() {
107        if set_cache_control(content, Some(cache_control)) {
108            break;
109        }
110    }
111}
112
113fn to_anthropic_content(
114    content: MessageContent,
115    compaction_state_owner: &LanguageModelProviderId,
116) -> Result<Option<RequestContent>> {
117    match content {
118        MessageContent::Text(text) => {
119            let text = if text.chars().last().is_some_and(|c| c.is_whitespace()) {
120                text.trim_end().to_string()
121            } else {
122                text
123            };
124            if !text.is_empty() {
125                Ok(Some(RequestContent::Text {
126                    text,
127                    cache_control: None,
128                }))
129            } else {
130                Ok(None)
131            }
132        }
133        MessageContent::Thinking {
134            text: thinking,
135            signature,
136        } => {
137            if let Some(signature) = signature
138                && !thinking.is_empty()
139            {
140                Ok(Some(RequestContent::Thinking {
141                    thinking,
142                    signature,
143                    cache_control: None,
144                }))
145            } else {
146                Ok(None)
147            }
148        }
149        MessageContent::RedactedThinking(data) => {
150            if !data.is_empty() {
151                Ok(Some(RequestContent::RedactedThinking { data }))
152            } else {
153                Ok(None)
154            }
155        }
156        MessageContent::Image(image) => Ok(Some(RequestContent::Image {
157            source: ImageSource {
158                source_type: "base64".to_string(),
159                media_type: "image/png".to_string(),
160                data: image.source.to_string(),
161            },
162            cache_control: None,
163        })),
164        MessageContent::ToolUse(tool_use) => match tool_use.input {
165            LanguageModelToolUseInput::Json(input) => Ok(Some(RequestContent::ToolUse {
166                id: tool_use.id.to_string(),
167                name: tool_use.name.to_string(),
168                input,
169                cache_control: None,
170            })),
171            LanguageModelToolUseInput::Text(_) => Err(anyhow::anyhow!(
172                "Anthropic does not support custom tool calls"
173            )),
174        },
175        MessageContent::ToolResult(tool_result) => {
176            let content = match tool_result.content.as_slice() {
177                [LanguageModelToolResultContent::Text(text)] => {
178                    ToolResultContent::Plain(text.to_string())
179                }
180                _ => {
181                    let parts = tool_result
182                        .content
183                        .into_iter()
184                        .map(|part| match part {
185                            LanguageModelToolResultContent::Text(text) => ToolResultPart::Text {
186                                text: text.to_string(),
187                            },
188                            LanguageModelToolResultContent::Image(image) => ToolResultPart::Image {
189                                source: ImageSource {
190                                    source_type: "base64".to_string(),
191                                    media_type: "image/png".to_string(),
192                                    data: image.source.to_string(),
193                                },
194                            },
195                        })
196                        .collect();
197                    ToolResultContent::Multipart(parts)
198                }
199            };
200            Ok(Some(RequestContent::ToolResult {
201                tool_use_id: tool_result.tool_use_id.to_string(),
202                is_error: tool_result.is_error,
203                content,
204                cache_control: None,
205            }))
206        }
207        MessageContent::Compaction(CompactedContext::Summary {
208            content,
209            provider_state,
210        }) => {
211            let encrypted_content = match &provider_state {
212                Some(state) => {
213                    provider_compaction_encrypted_content(state, compaction_state_owner)?
214                }
215                None => None,
216            };
217            Ok(Some(RequestContent::Compaction {
218                content: Some(content),
219                encrypted_content,
220                cache_control: None,
221            }))
222        }
223        MessageContent::Compaction(CompactedContext::ProviderState(_)) => Ok(None),
224    }
225}
226
227pub fn into_anthropic(
228    request: LanguageModelRequest,
229    model: String,
230    default_temperature: f32,
231    max_output_tokens: u64,
232    mode: AnthropicModelMode,
233    cache_mode: AnthropicPromptCacheMode,
234    compaction_state_owner: &LanguageModelProviderId,
235) -> Result<crate::Request> {
236    let mut new_messages: Vec<Message> = Vec::new();
237    let mut system_message = String::new();
238    let mut any_message_wants_cache = false;
239
240    for message in request.messages {
241        if message.contents_empty() {
242            continue;
243        }
244
245        any_message_wants_cache |= message.cache;
246
247        match message.role {
248            Role::User | Role::Assistant => {
249                let mut anthropic_message_content = Vec::new();
250                for content in message.content {
251                    if let Some(content) = to_anthropic_content(content, compaction_state_owner)? {
252                        anthropic_message_content.push(content);
253                    }
254                }
255                let anthropic_role = match message.role {
256                    Role::User => crate::Role::User,
257                    Role::Assistant => crate::Role::Assistant,
258                    Role::System => unreachable!("System role should never occur here"),
259                };
260                if anthropic_message_content.is_empty() {
261                    continue;
262                }
263
264                if cache_mode == AnthropicPromptCacheMode::Legacy && message.cache {
265                    mark_last_cacheable_content(
266                        &mut anthropic_message_content,
267                        CacheControl {
268                            cache_type: CacheControlType::Ephemeral,
269                            ttl: None,
270                        },
271                    );
272                }
273
274                if let Some(last_message) = new_messages.last_mut()
275                    && last_message.role == anthropic_role
276                {
277                    last_message.content.extend(anthropic_message_content);
278                    continue;
279                }
280
281                new_messages.push(Message {
282                    role: anthropic_role,
283                    content: anthropic_message_content,
284                });
285            }
286            Role::System => {
287                if !system_message.is_empty() {
288                    system_message.push_str("\n\n");
289                }
290                system_message.push_str(&message.string_contents());
291            }
292        }
293    }
294
295    // When caching is enabled, mark the static prefix (tools + system) with an
296    // explicit long-TTL breakpoint, and let Anthropic's automatic top-level
297    // cache_control handle the short-TTL conversation breakpoint. Anthropic
298    // requires that longer TTLs appear earlier in the prefix, and the prefix
299    // order is tools → system → messages, so long-TTL tools/system before a
300    // short-TTL conversation breakpoint is a valid mix.
301    let long_lived_cache = (cache_mode == AnthropicPromptCacheMode::Automatic
302        && any_message_wants_cache)
303        .then_some(CacheControl {
304            cache_type: CacheControlType::Ephemeral,
305            ttl: Some(CacheTtl::OneHour),
306        });
307
308    let system = if system_message.is_empty() {
309        None
310    } else if let Some(cache_control) = long_lived_cache {
311        Some(StringOrContents::Content(vec![RequestContent::Text {
312            text: system_message,
313            cache_control: Some(cache_control),
314        }]))
315    } else {
316        Some(StringOrContents::String(system_message))
317    };
318
319    let mut tools: Vec<Tool> = request
320        .tools
321        .into_iter()
322        .map(|tool| match tool.input {
323            LanguageModelRequestToolInput::Function {
324                input_schema,
325                use_input_streaming,
326            } => Ok(Tool {
327                name: tool.name,
328                description: tool.description,
329                input_schema,
330                eager_input_streaming: use_input_streaming,
331                cache_control: None,
332            }),
333            LanguageModelRequestToolInput::Custom { .. } => {
334                Err(anyhow::anyhow!("Anthropic does not support custom tools"))
335            }
336        })
337        .collect::<Result<_>>()?;
338    if let Some(cache_control) = long_lived_cache
339        && let Some(last_tool) = tools.last_mut()
340    {
341        last_tool.cache_control = Some(cache_control);
342    }
343
344    let thinking = if request.thinking_allowed {
345        match mode {
346            AnthropicModelMode::Thinking { budget_tokens } => {
347                Some(Thinking::Enabled { budget_tokens })
348            }
349            AnthropicModelMode::AdaptiveThinking => Some(Thinking::Adaptive {
350                display: Some(AdaptiveThinkingDisplay::Summarized),
351            }),
352            AnthropicModelMode::Default => None,
353        }
354    } else if crate::requires_explicit_thinking_opt_out(&model) {
355        // On Claude Opus 5, omitting the `thinking` field no longer means
356        // "off": the model runs adaptive thinking by default, so features
357        // that suppress thinking (e.g. inline assist) must opt out
358        // explicitly. `disabled` is only accepted at effort `high` or below;
359        // that holds here because `output_config` is never sent when thinking
360        // is disallowed, and the server-side default effort is `high`.
361        // <https://platform.claude.com/docs/en/about-claude/models/migration-guide#migrating-to-claude-opus-5>
362        Some(Thinking::Disabled)
363    } else {
364        None
365    };
366
367    Ok(crate::Request {
368        model,
369        messages: new_messages,
370        max_tokens: max_output_tokens,
371        system,
372        // Opt into Anthropic's automatic prompt caching for the conversation
373        // tail. Omitting `ttl` uses the default (short) TTL, which refreshes
374        // for free on every cache hit — ideal for the rapidly-changing
375        // conversation suffix.
376        cache_control: (cache_mode == AnthropicPromptCacheMode::Automatic
377            && any_message_wants_cache)
378            .then_some(CacheControl {
379                cache_type: CacheControlType::Ephemeral,
380                ttl: None,
381            }),
382        thinking,
383        tools,
384        tool_choice: request.tool_choice.map(|choice| match choice {
385            LanguageModelToolChoice::Auto => ToolChoice::Auto,
386            LanguageModelToolChoice::Any => ToolChoice::Any,
387            LanguageModelToolChoice::None => ToolChoice::None,
388        }),
389        metadata: None,
390        output_config: if request.thinking_allowed
391            && matches!(mode, AnthropicModelMode::AdaptiveThinking)
392        {
393            request.thinking_effort.as_deref().and_then(|effort| {
394                let effort = match effort {
395                    "low" => Some(crate::Effort::Low),
396                    "medium" => Some(crate::Effort::Medium),
397                    "high" => Some(crate::Effort::High),
398                    "xhigh" => Some(crate::Effort::XHigh),
399                    "max" => Some(crate::Effort::Max),
400                    _ => None,
401                };
402                effort.map(|effort| crate::OutputConfig {
403                    effort: Some(effort),
404                })
405            })
406        } else {
407            None
408        },
409        stop_sequences: Vec::new(),
410        speed: request.speed.map(Into::into),
411        temperature: request.temperature.or(Some(default_temperature)),
412        top_k: None,
413        top_p: None,
414        context_management: request.compact_at_tokens.map(|value| ContextManagement {
415            edits: vec![ContextManagementEdit::Compact {
416                trigger: Some(CompactionTrigger::InputTokens { value }),
417            }],
418        }),
419    })
420}
421
422pub struct AnthropicEventMapper {
423    tool_uses_by_index: HashMap<usize, RawToolUse>,
424    compactions_by_index: HashMap<usize, RawCompaction>,
425    usage: Usage,
426    stop_reason: StopReason,
427    provider_name: LanguageModelProviderName,
428    compaction_state_owner: LanguageModelProviderId,
429}
430
431impl AnthropicEventMapper {
432    /// `compaction_state_owner` identifies the backend whose infrastructure
433    /// produced this stream, so that any `encrypted_content` it emits is only
434    /// ever round-tripped back to that same backend.
435    pub fn new(
436        provider_name: LanguageModelProviderName,
437        compaction_state_owner: LanguageModelProviderId,
438    ) -> Self {
439        Self {
440            tool_uses_by_index: HashMap::default(),
441            compactions_by_index: HashMap::default(),
442            usage: Usage::default(),
443            stop_reason: StopReason::EndTurn,
444            provider_name,
445            compaction_state_owner,
446        }
447    }
448
449    pub fn map_stream(
450        mut self,
451        events: Pin<Box<dyn Send + Stream<Item = Result<Event, AnthropicError>>>>,
452    ) -> impl Stream<Item = Result<LanguageModelCompletionEvent, LanguageModelCompletionError>>
453    {
454        events.flat_map(move |event| {
455            futures::stream::iter(match event {
456                Ok(event) => self.map_event(event),
457                Err(error) => vec![Err(completion_error_from_anthropic(
458                    error,
459                    self.provider_name.clone(),
460                ))],
461            })
462        })
463    }
464
465    pub fn map_event(
466        &mut self,
467        event: Event,
468    ) -> Vec<Result<LanguageModelCompletionEvent, LanguageModelCompletionError>> {
469        match event {
470            Event::ContentBlockStart {
471                index,
472                content_block,
473            } => match content_block {
474                ResponseContent::Text { text } => {
475                    vec![Ok(LanguageModelCompletionEvent::Text(text))]
476                }
477                ResponseContent::Thinking { thinking } => {
478                    vec![Ok(LanguageModelCompletionEvent::Thinking {
479                        text: thinking,
480                        signature: None,
481                    })]
482                }
483                ResponseContent::RedactedThinking { data } => {
484                    vec![Ok(LanguageModelCompletionEvent::RedactedThinking { data })]
485                }
486                ResponseContent::ToolUse { id, name, .. } => {
487                    self.tool_uses_by_index.insert(
488                        index,
489                        RawToolUse {
490                            id,
491                            name,
492                            input_json: String::new(),
493                        },
494                    );
495                    Vec::new()
496                }
497                ResponseContent::Compaction {
498                    content,
499                    encrypted_content,
500                } => {
501                    let mut events = vec![Ok(LanguageModelCompletionEvent::Compaction(
502                        CompactionUpdate::Started,
503                    ))];
504                    let compaction = self.compactions_by_index.entry(index).or_default();
505                    if let Some(encrypted_content) =
506                        encrypted_content.filter(|encrypted| !encrypted.is_empty())
507                    {
508                        compaction.encrypted_content = Some(encrypted_content);
509                    }
510                    if let Some(content) = content
511                        && !content.is_empty()
512                    {
513                        compaction.summary.push_str(&content);
514                        events.push(Ok(LanguageModelCompletionEvent::Compaction(
515                            CompactionUpdate::SummaryDelta(content),
516                        )));
517                    }
518                    events
519                }
520            },
521            Event::ContentBlockDelta { index, delta } => match delta {
522                ContentDelta::TextDelta { text } => {
523                    vec![Ok(LanguageModelCompletionEvent::Text(text))]
524                }
525                ContentDelta::ThinkingDelta { thinking } => {
526                    vec![Ok(LanguageModelCompletionEvent::Thinking {
527                        text: thinking,
528                        signature: None,
529                    })]
530                }
531                ContentDelta::SignatureDelta { signature } => {
532                    vec![Ok(LanguageModelCompletionEvent::Thinking {
533                        text: "".to_string(),
534                        signature: Some(signature),
535                    })]
536                }
537                ContentDelta::CompactionDelta {
538                    content,
539                    encrypted_content,
540                } => {
541                    let Some(compaction) = self.compactions_by_index.get_mut(&index) else {
542                        return vec![Err(LanguageModelCompletionError::Other(anyhow::anyhow!(
543                            "Anthropic streamed a compaction delta before starting its content block"
544                        )))];
545                    };
546                    // Unlike summary text, `encrypted_content` arrives whole:
547                    // a later delta carries a complete replacement value, not
548                    // a chunk to append (Anthropic's own SDKs assign it, the
549                    // way they do thinking signatures).
550                    if let Some(encrypted_content) =
551                        encrypted_content.filter(|encrypted| !encrypted.is_empty())
552                    {
553                        compaction.encrypted_content = Some(encrypted_content);
554                    }
555                    let Some(content) = content.filter(|content| !content.is_empty()) else {
556                        return Vec::new();
557                    };
558                    compaction.summary.push_str(&content);
559                    vec![Ok(LanguageModelCompletionEvent::Compaction(
560                        CompactionUpdate::SummaryDelta(content),
561                    ))]
562                }
563                ContentDelta::InputJsonDelta { partial_json } => {
564                    if let Some(tool_use) = self.tool_uses_by_index.get_mut(&index) {
565                        tool_use.input_json.push_str(&partial_json);
566
567                        // Try to convert invalid (incomplete) JSON into
568                        // valid JSON that serde can accept, e.g. by closing
569                        // unclosed delimiters. This way, we can update the
570                        // UI with whatever has been streamed back so far.
571                        if let Ok(input) =
572                            serde_json::Value::from_str(&fix_streamed_json(&tool_use.input_json))
573                        {
574                            return vec![Ok(LanguageModelCompletionEvent::ToolUse(
575                                LanguageModelToolUse {
576                                    id: tool_use.id.clone().into(),
577                                    name: tool_use.name.clone().into(),
578                                    is_input_complete: false,
579                                    raw_input: tool_use.input_json.clone(),
580                                    input: LanguageModelToolUseInput::Json(input),
581                                    thought_signature: None,
582                                },
583                            ))];
584                        }
585                    }
586                    vec![]
587                }
588            },
589            Event::ContentBlockStop { index } => {
590                if let Some(compaction) = self.compactions_by_index.remove(&index) {
591                    // A compaction block that closes without content is a
592                    // documented failed compaction, which the server treats
593                    // as a no-op: there is nothing to persist, and the
594                    // conversation continues on the uncompacted transcript.
595                    if compaction.summary.is_empty() {
596                        return vec![Ok(LanguageModelCompletionEvent::Compaction(
597                            CompactionUpdate::Failed,
598                        ))];
599                    }
600                    let provider_state = compaction.encrypted_content.map(|encrypted_content| {
601                        provider_compaction_state_from_encrypted_content(
602                            self.compaction_state_owner.clone(),
603                            encrypted_content,
604                        )
605                    });
606                    vec![Ok(LanguageModelCompletionEvent::Compaction(
607                        CompactionUpdate::Finished(CompactedContext::Summary {
608                            content: compaction.summary.into(),
609                            provider_state,
610                        }),
611                    ))]
612                } else if let Some(tool_use) = self.tool_uses_by_index.remove(&index) {
613                    let input_json = tool_use.input_json.trim();
614                    let event_result = match parse_tool_arguments(input_json) {
615                        Ok(input) => Ok(LanguageModelCompletionEvent::ToolUse(
616                            LanguageModelToolUse {
617                                id: tool_use.id.into(),
618                                name: tool_use.name.into(),
619                                is_input_complete: true,
620                                input: LanguageModelToolUseInput::Json(input),
621                                raw_input: tool_use.input_json.clone(),
622                                thought_signature: None,
623                            },
624                        )),
625                        Err(json_parse_err) => {
626                            Ok(LanguageModelCompletionEvent::ToolUseJsonParseError {
627                                id: tool_use.id.into(),
628                                tool_name: tool_use.name.into(),
629                                raw_input: input_json.into(),
630                                json_parse_error: json_parse_err.to_string(),
631                            })
632                        }
633                    };
634
635                    vec![event_result]
636                } else {
637                    Vec::new()
638                }
639            }
640            Event::MessageStart { message } => {
641                update_usage(&mut self.usage, &message.usage);
642                vec![
643                    Ok(LanguageModelCompletionEvent::UsageUpdate(convert_usage(
644                        &self.usage,
645                    ))),
646                    Ok(LanguageModelCompletionEvent::StartMessage {
647                        message_id: message.id,
648                    }),
649                ]
650            }
651            Event::MessageDelta { delta, usage } => {
652                update_usage(&mut self.usage, &usage);
653                if let Some(stop_reason) = delta.stop_reason.as_deref() {
654                    self.stop_reason = match stop_reason {
655                        "end_turn" => StopReason::EndTurn,
656                        "max_tokens" => StopReason::MaxTokens,
657                        "tool_use" => StopReason::ToolUse,
658                        "refusal" => StopReason::Refusal,
659                        _ => {
660                            log::error!("Unexpected anthropic stop_reason: {stop_reason}");
661                            StopReason::EndTurn
662                        }
663                    };
664                }
665                vec![Ok(LanguageModelCompletionEvent::UsageUpdate(
666                    convert_usage(&self.usage),
667                ))]
668            }
669            Event::MessageStop => {
670                // Anthropic closes every content block before ending the
671                // message, so an unclosed compaction block means the stream
672                // was malformed and its finalized summary never arrived.
673                // Consumers would otherwise see `Started` with no terminal
674                // event and treat the compaction as still in progress.
675                if !self.compactions_by_index.is_empty() {
676                    self.compactions_by_index.clear();
677                    return vec![Err(LanguageModelCompletionError::Other(anyhow::anyhow!(
678                        "Anthropic ended the stream without finishing its compaction summary"
679                    )))];
680                }
681                vec![Ok(LanguageModelCompletionEvent::Stop(self.stop_reason))]
682            }
683            Event::Error { error } => {
684                vec![Err(completion_error_from_anthropic_api(
685                    error,
686                    self.provider_name.clone(),
687                ))]
688            }
689            _ => Vec::new(),
690        }
691    }
692}
693
694struct RawToolUse {
695    id: String,
696    name: String,
697    input_json: String,
698}
699
700#[derive(Default)]
701struct RawCompaction {
702    summary: String,
703    encrypted_content: Option<Arc<str>>,
704}
705
706/// Updates usage data by preferring counts from `new`.
707fn update_usage(usage: &mut Usage, new: &Usage) {
708    if let Some(input_tokens) = new.input_tokens {
709        usage.input_tokens = Some(input_tokens);
710    }
711    if let Some(output_tokens) = new.output_tokens {
712        usage.output_tokens = Some(output_tokens);
713    }
714    if let Some(cache_creation_input_tokens) = new.cache_creation_input_tokens {
715        usage.cache_creation_input_tokens = Some(cache_creation_input_tokens);
716    }
717    if let Some(cache_read_input_tokens) = new.cache_read_input_tokens {
718        usage.cache_read_input_tokens = Some(cache_read_input_tokens);
719    }
720}
721
722fn convert_usage(usage: &Usage) -> TokenUsage {
723    TokenUsage {
724        input_tokens: usage.input_tokens.unwrap_or(0),
725        output_tokens: usage.output_tokens.unwrap_or(0),
726        cache_creation_input_tokens: usage.cache_creation_input_tokens.unwrap_or(0),
727        cache_read_input_tokens: usage.cache_read_input_tokens.unwrap_or(0),
728    }
729}
730
731#[cfg(test)]
732mod tests {
733    use super::*;
734    use crate::{AnthropicModelMode, UsageIteration, UsageIterationType};
735    use language_model_core::{
736        ANTHROPIC_PROVIDER_ID, ANTHROPIC_PROVIDER_NAME, LanguageModelImage,
737        LanguageModelRequestMessage, MessageContent,
738    };
739
740    #[test]
741    fn test_caching_uses_top_level_auto_and_long_lived_prefix() {
742        let request = LanguageModelRequest {
743            messages: vec![
744                LanguageModelRequestMessage {
745                    role: Role::System,
746                    content: vec![MessageContent::Text("You are helpful.".to_string())],
747                    cache: false,
748                    reasoning_details: None,
749                },
750                LanguageModelRequestMessage {
751                    role: Role::User,
752                    content: vec![
753                        MessageContent::Text("Some prompt".to_string()),
754                        MessageContent::Image(LanguageModelImage::empty()),
755                        MessageContent::Image(LanguageModelImage::empty()),
756                    ],
757                    cache: true,
758                    reasoning_details: None,
759                },
760            ],
761            thread_id: None,
762            prompt_id: None,
763            intent: None,
764            stop: vec![],
765            temperature: None,
766            tools: vec![language_model_core::LanguageModelRequestTool::function(
767                "do_thing".into(),
768                "Does a thing.".into(),
769                serde_json::json!({"type": "object"}),
770                false,
771            )],
772            tool_choice: None,
773            thinking_allowed: true,
774            thinking_effort: None,
775            speed: None,
776            compact_at_tokens: None,
777        };
778
779        let anthropic_request = into_anthropic(
780            request,
781            "claude-3-5-sonnet".to_string(),
782            0.7,
783            4096,
784            AnthropicModelMode::Default,
785            AnthropicPromptCacheMode::Automatic,
786            &ANTHROPIC_PROVIDER_ID,
787        )
788        .unwrap();
789
790        // No message content block should carry cache_control anymore; the
791        // conversation breakpoint is set via top-level automatic caching.
792        assert_eq!(anthropic_request.messages.len(), 1);
793        for block in &anthropic_request.messages[0].content {
794            let cache_control = match block {
795                RequestContent::Text { cache_control, .. }
796                | RequestContent::Thinking { cache_control, .. }
797                | RequestContent::Image { cache_control, .. }
798                | RequestContent::ToolUse { cache_control, .. }
799                | RequestContent::ToolResult { cache_control, .. }
800                | RequestContent::Compaction { cache_control, .. } => *cache_control,
801                RequestContent::RedactedThinking { .. } => None,
802            };
803            assert!(
804                cache_control.is_none(),
805                "message content blocks should no longer be individually marked",
806            );
807        }
808
809        // Top-level cache_control opts into automatic caching with the default
810        // 5-minute TTL for the conversation tail.
811        assert!(matches!(
812            anthropic_request.cache_control,
813            Some(CacheControl {
814                cache_type: CacheControlType::Ephemeral,
815                ttl: None,
816            })
817        ));
818
819        // System prompt is emitted in array form with a long-TTL breakpoint on
820        // the final text block.
821        match anthropic_request.system {
822            Some(StringOrContents::Content(ref blocks)) => {
823                assert_eq!(blocks.len(), 1);
824                assert!(matches!(
825                    blocks[0],
826                    RequestContent::Text {
827                        cache_control: Some(CacheControl {
828                            cache_type: CacheControlType::Ephemeral,
829                            ttl: Some(CacheTtl::OneHour),
830                        }),
831                        ..
832                    }
833                ));
834            }
835            other => panic!("expected system content array, got {other:?}"),
836        }
837
838        // The last (and only) tool carries a long-TTL breakpoint.
839        assert_eq!(anthropic_request.tools.len(), 1);
840        assert!(matches!(
841            anthropic_request.tools[0].cache_control,
842            Some(CacheControl {
843                cache_type: CacheControlType::Ephemeral,
844                ttl: Some(CacheTtl::OneHour),
845            })
846        ));
847    }
848
849    #[test]
850    fn test_legacy_caching_marks_last_message_content_block() {
851        let request = LanguageModelRequest {
852            messages: vec![
853                LanguageModelRequestMessage {
854                    role: Role::System,
855                    content: vec![MessageContent::Text("You are helpful.".to_string())],
856                    cache: false,
857                    reasoning_details: None,
858                },
859                LanguageModelRequestMessage {
860                    role: Role::User,
861                    content: vec![
862                        MessageContent::Text("Some prompt".to_string()),
863                        MessageContent::Image(LanguageModelImage::empty()),
864                    ],
865                    cache: true,
866                    reasoning_details: None,
867                },
868            ],
869            thread_id: None,
870            prompt_id: None,
871            intent: None,
872            stop: vec![],
873            temperature: None,
874            tools: vec![language_model_core::LanguageModelRequestTool::function(
875                "do_thing".into(),
876                "Does a thing.".into(),
877                serde_json::json!({"type": "object"}),
878                false,
879            )],
880            tool_choice: None,
881            thinking_allowed: true,
882            thinking_effort: None,
883            speed: None,
884            compact_at_tokens: None,
885        };
886
887        let anthropic_request = into_anthropic(
888            request,
889            "claude-3-5-sonnet".to_string(),
890            0.7,
891            4096,
892            AnthropicModelMode::Default,
893            AnthropicPromptCacheMode::Legacy,
894            &ANTHROPIC_PROVIDER_ID,
895        )
896        .unwrap();
897
898        assert!(anthropic_request.cache_control.is_none());
899        assert!(matches!(
900            anthropic_request.system,
901            Some(StringOrContents::String(_))
902        ));
903        assert_eq!(anthropic_request.tools.len(), 1);
904        assert!(anthropic_request.tools[0].cache_control.is_none());
905        assert_eq!(anthropic_request.messages.len(), 1);
906        assert!(matches!(
907            anthropic_request.messages[0].content[0],
908            RequestContent::Text {
909                cache_control: None,
910                ..
911            }
912        ));
913        assert!(matches!(
914            anthropic_request.messages[0].content[1],
915            RequestContent::Image {
916                cache_control: Some(CacheControl {
917                    cache_type: CacheControlType::Ephemeral,
918                    ttl: None,
919                }),
920                ..
921            }
922        ));
923    }
924
925    #[test]
926    fn test_xhigh_effort_is_serialized_for_adaptive_thinking() {
927        let request = LanguageModelRequest {
928            messages: vec![LanguageModelRequestMessage {
929                role: Role::User,
930                content: vec![MessageContent::Text("Hi".to_string())],
931                cache: false,
932                reasoning_details: None,
933            }],
934            thread_id: None,
935            prompt_id: None,
936            intent: None,
937            stop: vec![],
938            temperature: None,
939            tools: vec![],
940            tool_choice: None,
941            thinking_allowed: true,
942            thinking_effort: Some("xhigh".into()),
943            speed: None,
944            compact_at_tokens: None,
945        };
946
947        let anthropic_request = into_anthropic(
948            request,
949            "claude-opus-4-8".to_string(),
950            1.0,
951            128_000,
952            AnthropicModelMode::AdaptiveThinking,
953            AnthropicPromptCacheMode::Automatic,
954            &ANTHROPIC_PROVIDER_ID,
955        )
956        .unwrap();
957
958        assert_eq!(
959            anthropic_request
960                .output_config
961                .and_then(|config| config.effort),
962            Some(crate::Effort::XHigh)
963        );
964    }
965
966    #[test]
967    fn test_thinking_disallowed_sends_explicit_opt_out_only_where_required() {
968        // (model, expects_explicit_opt_out): Claude Opus 5 thinks by default
969        // when the `thinking` field is omitted, so suppressing thinking
970        // requires sending `{"type": "disabled"}`. Earlier Opus models treat
971        // omission as "off", and Fable rejects `disabled` outright, so both
972        // must keep omitting the field.
973        for (model, expects_explicit_opt_out) in [
974            ("claude-opus-5", true),
975            ("claude-opus-4-8", false),
976            ("claude-fable-5", false),
977        ] {
978            let request = LanguageModelRequest {
979                messages: vec![LanguageModelRequestMessage {
980                    role: Role::User,
981                    content: vec![MessageContent::Text("Hi".to_string())],
982                    cache: false,
983                    reasoning_details: None,
984                }],
985                thread_id: None,
986                prompt_id: None,
987                intent: None,
988                stop: vec![],
989                temperature: None,
990                tools: vec![],
991                tool_choice: None,
992                thinking_allowed: false,
993                thinking_effort: None,
994                speed: None,
995                compact_at_tokens: None,
996            };
997
998            let anthropic_request = into_anthropic(
999                request,
1000                model.to_string(),
1001                1.0,
1002                128_000,
1003                AnthropicModelMode::AdaptiveThinking,
1004                AnthropicPromptCacheMode::Automatic,
1005                &ANTHROPIC_PROVIDER_ID,
1006            )
1007            .unwrap();
1008
1009            if expects_explicit_opt_out {
1010                assert!(
1011                    matches!(anthropic_request.thinking, Some(Thinking::Disabled)),
1012                    "{model} should send an explicit thinking opt-out"
1013                );
1014                // `disabled` combined with effort `xhigh`/`max` is a 400, so
1015                // no effort may accompany the opt-out.
1016                assert!(
1017                    anthropic_request.output_config.is_none(),
1018                    "{model} must not send output_config with thinking disabled"
1019                );
1020            } else {
1021                assert!(
1022                    anthropic_request.thinking.is_none(),
1023                    "{model} should omit the thinking field entirely"
1024                );
1025            }
1026        }
1027    }
1028
1029    #[test]
1030    fn test_no_cache_control_when_caching_disabled() {
1031        let request = LanguageModelRequest {
1032            messages: vec![
1033                LanguageModelRequestMessage {
1034                    role: Role::System,
1035                    content: vec![MessageContent::Text("You are helpful.".to_string())],
1036                    cache: false,
1037                    reasoning_details: None,
1038                },
1039                LanguageModelRequestMessage {
1040                    role: Role::User,
1041                    content: vec![MessageContent::Text("Hi".to_string())],
1042                    cache: false,
1043                    reasoning_details: None,
1044                },
1045            ],
1046            thread_id: None,
1047            prompt_id: None,
1048            intent: None,
1049            stop: vec![],
1050            temperature: None,
1051            tools: vec![language_model_core::LanguageModelRequestTool::function(
1052                "do_thing".into(),
1053                "Does a thing.".into(),
1054                serde_json::json!({"type": "object"}),
1055                false,
1056            )],
1057            tool_choice: None,
1058            thinking_allowed: true,
1059            thinking_effort: None,
1060            speed: None,
1061            compact_at_tokens: None,
1062        };
1063
1064        let anthropic_request = into_anthropic(
1065            request,
1066            "claude-3-5-sonnet".to_string(),
1067            0.7,
1068            4096,
1069            AnthropicModelMode::Default,
1070            AnthropicPromptCacheMode::Automatic,
1071            &ANTHROPIC_PROVIDER_ID,
1072        )
1073        .unwrap();
1074
1075        assert!(anthropic_request.cache_control.is_none());
1076        assert!(matches!(
1077            anthropic_request.system,
1078            Some(StringOrContents::String(_))
1079        ));
1080        assert!(anthropic_request.tools[0].cache_control.is_none());
1081    }
1082
1083    fn request_with_assistant_content(assistant_content: Vec<MessageContent>) -> crate::Request {
1084        let mut request = LanguageModelRequest {
1085            messages: vec![LanguageModelRequestMessage {
1086                role: Role::User,
1087                content: vec![MessageContent::Text("Hello".to_string())],
1088                cache: false,
1089                reasoning_details: None,
1090            }],
1091            thinking_effort: None,
1092            thread_id: None,
1093            prompt_id: None,
1094            intent: None,
1095            stop: vec![],
1096            temperature: None,
1097            tools: vec![],
1098            tool_choice: None,
1099            thinking_allowed: true,
1100            speed: None,
1101            compact_at_tokens: None,
1102        };
1103        request.messages.push(LanguageModelRequestMessage {
1104            role: Role::Assistant,
1105            content: assistant_content,
1106            cache: false,
1107            reasoning_details: None,
1108        });
1109        into_anthropic(
1110            request,
1111            "claude-sonnet-4-5".to_string(),
1112            1.0,
1113            16000,
1114            AnthropicModelMode::Thinking {
1115                budget_tokens: Some(10000),
1116            },
1117            AnthropicPromptCacheMode::Automatic,
1118            &ANTHROPIC_PROVIDER_ID,
1119        )
1120        .unwrap()
1121    }
1122
1123    #[test]
1124    fn test_unsigned_thinking_blocks_stripped() {
1125        let result = request_with_assistant_content(vec![
1126            MessageContent::Thinking {
1127                text: "Cancelled mid-think, no signature".to_string(),
1128                signature: None,
1129            },
1130            MessageContent::Text("Some response text".to_string()),
1131        ]);
1132
1133        let assistant_message = result
1134            .messages
1135            .iter()
1136            .find(|m| m.role == crate::Role::Assistant)
1137            .expect("assistant message should still exist");
1138
1139        assert_eq!(
1140            assistant_message.content.len(),
1141            1,
1142            "Only the text content should remain; unsigned thinking block should be stripped"
1143        );
1144        assert!(matches!(
1145            &assistant_message.content[0],
1146            RequestContent::Text { text, .. } if text == "Some response text"
1147        ));
1148    }
1149
1150    #[test]
1151    fn test_signed_thinking_blocks_preserved() {
1152        let result = request_with_assistant_content(vec![
1153            MessageContent::Thinking {
1154                text: "Completed thinking".to_string(),
1155                signature: Some("valid-signature".to_string()),
1156            },
1157            MessageContent::Text("Response".to_string()),
1158        ]);
1159
1160        let assistant_message = result
1161            .messages
1162            .iter()
1163            .find(|m| m.role == crate::Role::Assistant)
1164            .expect("assistant message should exist");
1165
1166        assert_eq!(
1167            assistant_message.content.len(),
1168            2,
1169            "Both the signed thinking block and text should be preserved"
1170        );
1171        assert!(matches!(
1172            &assistant_message.content[0],
1173            RequestContent::Thinking { thinking, signature, .. }
1174                if thinking == "Completed thinking" && signature == "valid-signature"
1175        ));
1176    }
1177
1178    #[test]
1179    fn test_only_unsigned_thinking_block_omits_entire_message() {
1180        let result = request_with_assistant_content(vec![MessageContent::Thinking {
1181            text: "Cancelled before any text or signature".to_string(),
1182            signature: None,
1183        }]);
1184
1185        let assistant_messages: Vec<_> = result
1186            .messages
1187            .iter()
1188            .filter(|m| m.role == crate::Role::Assistant)
1189            .collect();
1190
1191        assert_eq!(
1192            assistant_messages.len(),
1193            0,
1194            "An assistant message whose only content was an unsigned thinking block \
1195             should be omitted entirely"
1196        );
1197    }
1198
1199    #[test]
1200    fn test_compact_at_tokens_maps_to_context_management() {
1201        let request = LanguageModelRequest {
1202            messages: vec![LanguageModelRequestMessage {
1203                role: Role::User,
1204                content: vec![MessageContent::Text("Hello".to_string())],
1205                cache: false,
1206                reasoning_details: None,
1207            }],
1208            compact_at_tokens: Some(100_000),
1209            ..Default::default()
1210        };
1211
1212        let anthropic_request = into_anthropic(
1213            request,
1214            "claude-sonnet-4-5".to_string(),
1215            1.0,
1216            4096,
1217            AnthropicModelMode::Default,
1218            AnthropicPromptCacheMode::Disabled,
1219            &ANTHROPIC_PROVIDER_ID,
1220        )
1221        .unwrap();
1222
1223        assert_eq!(
1224            serde_json::to_value(&anthropic_request.context_management).unwrap(),
1225            serde_json::json!({
1226                "edits": [{
1227                    "type": "compact_20260112",
1228                    "trigger": { "type": "input_tokens", "value": 100_000 }
1229                }]
1230            })
1231        );
1232    }
1233
1234    #[test]
1235    fn test_no_context_management_without_compact_at_tokens() {
1236        let result =
1237            request_with_assistant_content(vec![MessageContent::Text("Response".to_string())]);
1238
1239        assert!(result.context_management.is_none());
1240    }
1241
1242    #[test]
1243    fn test_compaction_content_replayed_as_compaction_block() {
1244        let result = request_with_assistant_content(vec![
1245            MessageContent::Compaction(CompactedContext::Summary {
1246                content: "Summary of the conversation so far.".into(),
1247                provider_state: None,
1248            }),
1249            MessageContent::Text("Response".to_string()),
1250        ]);
1251
1252        let assistant_message = result
1253            .messages
1254            .iter()
1255            .find(|m| m.role == crate::Role::Assistant)
1256            .expect("assistant message should exist");
1257
1258        assert_eq!(
1259            serde_json::to_value(&assistant_message.content[0]).unwrap(),
1260            serde_json::json!({
1261                "type": "compaction",
1262                "content": "Summary of the conversation so far."
1263            })
1264        );
1265    }
1266
1267    #[test]
1268    fn test_compaction_encrypted_content_replayed_only_for_owning_backend() {
1269        let summary_owned_by = |owner: LanguageModelProviderId| {
1270            MessageContent::Compaction(CompactedContext::Summary {
1271                content: "Summary of the conversation so far.".into(),
1272                provider_state: Some(provider_compaction_state_from_encrypted_content(
1273                    owner,
1274                    "opaque-compaction-payload",
1275                )),
1276            })
1277        };
1278
1279        let owned = to_anthropic_content(
1280            summary_owned_by(ANTHROPIC_PROVIDER_ID),
1281            &ANTHROPIC_PROVIDER_ID,
1282        )
1283        .unwrap()
1284        .expect("compaction block should be produced");
1285        assert_eq!(
1286            serde_json::to_value(&owned).unwrap(),
1287            serde_json::json!({
1288                "type": "compaction",
1289                "content": "Summary of the conversation so far.",
1290                "encrypted_content": "opaque-compaction-payload"
1291            })
1292        );
1293
1294        // State produced by a different Anthropic-protocol backend must not
1295        // be round-tripped: the summary is still replayed, but without the
1296        // foreign encrypted payload.
1297        let foreign = to_anthropic_content(
1298            summary_owned_by(LanguageModelProviderId::new("other-anthropic-backend")),
1299            &ANTHROPIC_PROVIDER_ID,
1300        )
1301        .unwrap()
1302        .expect("compaction block should be produced");
1303        assert_eq!(
1304            serde_json::to_value(&foreign).unwrap(),
1305            serde_json::json!({
1306                "type": "compaction",
1307                "content": "Summary of the conversation so far."
1308            })
1309        );
1310    }
1311
1312    #[test]
1313    fn test_event_mapper_maps_compaction_block_and_deltas() {
1314        let mut mapper = AnthropicEventMapper::new(ANTHROPIC_PROVIDER_NAME, ANTHROPIC_PROVIDER_ID);
1315
1316        let start_event: Event = serde_json::from_value(serde_json::json!({
1317            "type": "content_block_start",
1318            "index": 0,
1319            "content_block": { "type": "compaction", "content": "Summary " }
1320        }))
1321        .unwrap();
1322        let delta_event: Event = serde_json::from_value(serde_json::json!({
1323            "type": "content_block_delta",
1324            "index": 0,
1325            "delta": { "type": "compaction_delta", "content": "in " }
1326        }))
1327        .unwrap();
1328        let second_delta_event: Event = serde_json::from_value(serde_json::json!({
1329            "type": "content_block_delta",
1330            "index": 0,
1331            "delta": { "type": "compaction_delta", "content": "chunks" }
1332        }))
1333        .unwrap();
1334        let stop_event: Event = serde_json::from_value(serde_json::json!({
1335            "type": "content_block_stop",
1336            "index": 0
1337        }))
1338        .unwrap();
1339
1340        let mut events = Vec::new();
1341        events.extend(mapper.map_event(start_event));
1342        events.extend(mapper.map_event(delta_event));
1343        events.extend(mapper.map_event(second_delta_event));
1344        events.extend(mapper.map_event(stop_event));
1345        let events = events
1346            .into_iter()
1347            .collect::<Result<Vec<_>, _>>()
1348            .expect("all events should map successfully");
1349
1350        assert_eq!(
1351            events,
1352            vec![
1353                LanguageModelCompletionEvent::Compaction(CompactionUpdate::Started),
1354                LanguageModelCompletionEvent::Compaction(CompactionUpdate::SummaryDelta(
1355                    "Summary ".into()
1356                )),
1357                LanguageModelCompletionEvent::Compaction(CompactionUpdate::SummaryDelta(
1358                    "in ".into()
1359                )),
1360                LanguageModelCompletionEvent::Compaction(CompactionUpdate::SummaryDelta(
1361                    "chunks".into()
1362                )),
1363                LanguageModelCompletionEvent::Compaction(CompactionUpdate::Finished(
1364                    CompactedContext::Summary {
1365                        content: "Summary in chunks".into(),
1366                        provider_state: None,
1367                    }
1368                )),
1369            ]
1370        );
1371    }
1372
1373    /// Mirrors the stream shape in Anthropic's SDK fixtures: the block starts
1374    /// with both fields null, then a single delta carries the summary text
1375    /// alongside the opaque `encrypted_content` that must be round-tripped.
1376    #[test]
1377    fn test_event_mapper_captures_encrypted_content_as_provider_state() {
1378        let mut mapper = AnthropicEventMapper::new(ANTHROPIC_PROVIDER_NAME, ANTHROPIC_PROVIDER_ID);
1379
1380        let start_event: Event = serde_json::from_value(serde_json::json!({
1381            "type": "content_block_start",
1382            "index": 0,
1383            "content_block": { "type": "compaction", "content": null, "encrypted_content": null }
1384        }))
1385        .unwrap();
1386        let delta_event: Event = serde_json::from_value(serde_json::json!({
1387            "type": "content_block_delta",
1388            "index": 0,
1389            "delta": {
1390                "type": "compaction_delta",
1391                "content": "Earlier conversation summarized.",
1392                "encrypted_content": "opaque-compaction-payload"
1393            }
1394        }))
1395        .unwrap();
1396        let stop_event: Event = serde_json::from_value(serde_json::json!({
1397            "type": "content_block_stop",
1398            "index": 0
1399        }))
1400        .unwrap();
1401
1402        let mut events = Vec::new();
1403        events.extend(mapper.map_event(start_event));
1404        events.extend(mapper.map_event(delta_event));
1405        events.extend(mapper.map_event(stop_event));
1406        let mut events = events
1407            .into_iter()
1408            .collect::<Result<Vec<_>, _>>()
1409            .expect("all events should map successfully");
1410
1411        let Some(LanguageModelCompletionEvent::Compaction(CompactionUpdate::Finished(
1412            CompactedContext::Summary {
1413                content,
1414                provider_state: Some(state),
1415            },
1416        ))) = events.pop()
1417        else {
1418            panic!("expected a finished summary carrying provider state");
1419        };
1420        assert_eq!(content.as_ref(), "Earlier conversation summarized.");
1421        assert_eq!(
1422            provider_compaction_encrypted_content(&state, &ANTHROPIC_PROVIDER_ID)
1423                .unwrap()
1424                .as_deref(),
1425            Some("opaque-compaction-payload")
1426        );
1427        assert_eq!(
1428            provider_compaction_encrypted_content(
1429                &state,
1430                &LanguageModelProviderId::new("other-anthropic-backend")
1431            )
1432            .unwrap(),
1433            None
1434        );
1435    }
1436
1437    /// A compaction block that closes without any content is Anthropic's
1438    /// documented representation of a failed compaction, which the server
1439    /// treats as a no-op. It must surface as `Failed` -- not as an error that
1440    /// would kill the rest of the response.
1441    #[test]
1442    fn test_event_mapper_maps_null_content_compaction_to_failed() {
1443        let mut mapper = AnthropicEventMapper::new(ANTHROPIC_PROVIDER_NAME, ANTHROPIC_PROVIDER_ID);
1444        let start_event: Event = serde_json::from_value(serde_json::json!({
1445            "type": "content_block_start",
1446            "index": 0,
1447            "content_block": { "type": "compaction", "content": null, "encrypted_content": null }
1448        }))
1449        .unwrap();
1450        let stop_event: Event = serde_json::from_value(serde_json::json!({
1451            "type": "content_block_stop",
1452            "index": 0
1453        }))
1454        .unwrap();
1455
1456        assert_eq!(
1457            mapper
1458                .map_event(start_event)
1459                .into_iter()
1460                .collect::<Result<Vec<_>, _>>()
1461                .unwrap(),
1462            vec![LanguageModelCompletionEvent::Compaction(
1463                CompactionUpdate::Started
1464            )]
1465        );
1466        assert_eq!(
1467            mapper
1468                .map_event(stop_event)
1469                .into_iter()
1470                .collect::<Result<Vec<_>, _>>()
1471                .unwrap(),
1472            vec![LanguageModelCompletionEvent::Compaction(
1473                CompactionUpdate::Failed
1474            )]
1475        );
1476    }
1477
1478    #[test]
1479    fn test_event_mapper_rejects_compaction_delta_before_start() {
1480        let mut mapper = AnthropicEventMapper::new(ANTHROPIC_PROVIDER_NAME, ANTHROPIC_PROVIDER_ID);
1481        let delta_event: Event = serde_json::from_value(serde_json::json!({
1482            "type": "content_block_delta",
1483            "index": 0,
1484            "delta": { "type": "compaction_delta", "content": "Summary chunk" }
1485        }))
1486        .unwrap();
1487
1488        let error = mapper.map_event(delta_event).pop().unwrap().unwrap_err();
1489
1490        assert!(
1491            error
1492                .to_string()
1493                .contains("compaction delta before starting")
1494        );
1495    }
1496
1497    #[test]
1498    fn test_event_mapper_rejects_stream_end_with_unfinished_compaction() {
1499        let mut mapper = AnthropicEventMapper::new(ANTHROPIC_PROVIDER_NAME, ANTHROPIC_PROVIDER_ID);
1500        let start_event: Event = serde_json::from_value(serde_json::json!({
1501            "type": "content_block_start",
1502            "index": 0,
1503            "content_block": { "type": "compaction", "content": "Summary " }
1504        }))
1505        .unwrap();
1506        let stop_event: Event = serde_json::from_value(serde_json::json!({
1507            "type": "message_stop"
1508        }))
1509        .unwrap();
1510
1511        let started = mapper
1512            .map_event(start_event)
1513            .into_iter()
1514            .collect::<Result<Vec<_>, _>>()
1515            .unwrap();
1516        assert_eq!(
1517            started,
1518            vec![
1519                LanguageModelCompletionEvent::Compaction(CompactionUpdate::Started),
1520                LanguageModelCompletionEvent::Compaction(CompactionUpdate::SummaryDelta(
1521                    "Summary ".into()
1522                )),
1523            ]
1524        );
1525
1526        let error = mapper.map_event(stop_event).pop().unwrap().unwrap_err();
1527
1528        assert!(
1529            error
1530                .to_string()
1531                .contains("without finishing its compaction summary")
1532        );
1533    }
1534
1535    #[test]
1536    fn test_usage_iterations_parsed_from_message_delta() {
1537        let event: Event = serde_json::from_value(serde_json::json!({
1538            "type": "message_delta",
1539            "delta": { "stop_reason": "end_turn", "stop_sequence": null },
1540            "usage": {
1541                "input_tokens": 100,
1542                "output_tokens": 39,
1543                "iterations": [
1544                    { "type": "compaction", "input_tokens": 180000, "output_tokens": 1200 },
1545                    { "type": "message", "input_tokens": 100, "output_tokens": 39 }
1546                ]
1547            }
1548        }))
1549        .unwrap();
1550
1551        let Event::MessageDelta { usage, .. } = event else {
1552            panic!("expected message_delta event");
1553        };
1554        let iterations = usage.iterations.as_deref().expect("iterations expected");
1555        assert!(matches!(
1556            iterations[0],
1557            UsageIteration {
1558                iteration_type: UsageIterationType::Compaction,
1559                input_tokens: Some(180000),
1560                ..
1561            }
1562        ));
1563        assert!(matches!(
1564            iterations[1],
1565            UsageIteration {
1566                iteration_type: UsageIterationType::Message,
1567                input_tokens: Some(100),
1568                ..
1569            }
1570        ));
1571    }
1572}
1573