aither-claude 0.22.0

//! SSE response parsing for the Claude API.

use aither_core::llm::{Event as LLMEvent, ReasoningState, Usage as TokenUsage};

use crate::PROVIDER_NAME;
use serde::Deserialize;
use serde_json::Value;
use zenwave::sse::Event;

use crate::error::ClaudeError;

/// Initial `message_start` event data.
#[derive(Debug, Deserialize)]
#[allow(dead_code)]
pub struct MessageStartEvent {
    /// The message object.
    pub message: MessageObject,
}

/// Message object from `message_start`.
#[derive(Debug, Deserialize)]
#[allow(dead_code)]
pub struct MessageObject {
    /// Unique message ID.
    pub id: String,
    /// Model that generated the response.
    pub model: String,
    /// Reason the response stopped (null while streaming).
    #[serde(default)]
    pub stop_reason: Option<String>,
    /// Token usage information.
    #[serde(default)]
    pub usage: Option<Usage>,
}

/// Token usage information.
#[derive(Debug, Deserialize)]
#[allow(dead_code)]
pub struct Usage {
    /// Number of input tokens.
    pub input_tokens: u32,
    /// Number of output tokens.
    pub output_tokens: u32,
    /// Number of input tokens written into cache.
    #[serde(default)]
    pub cache_creation_input_tokens: Option<u32>,
    /// Number of input tokens read from cache.
    #[serde(default)]
    pub cache_read_input_tokens: Option<u32>,
    /// Detailed cache creation token counts by TTL.
    #[serde(default)]
    pub cache_creation: Option<CacheCreation>,
}

impl Usage {
    fn cache_write_tokens(&self) -> Option<u32> {
        self.cache_creation_input_tokens
            .or_else(|| self.cache_creation.as_ref().and_then(CacheCreation::total))
    }
}

/// Content block start event data.
#[derive(Debug, Deserialize)]
pub struct ContentBlockStartEvent {
    /// Index of this content block.
    pub index: usize,
    /// The content block being started.
    pub content_block: ContentBlockType,
}

/// Types of content blocks.
#[derive(Debug, Clone, Deserialize)]
#[serde(tag = "type")]
#[allow(dead_code)]
pub enum ContentBlockType {
    /// Text content block.
    #[serde(rename = "text")]
    Text {
        /// Initial text (usually empty).
        text: String,
    },
    /// Thinking/reasoning content block.
    #[serde(rename = "thinking")]
    Thinking {
        /// Initial thinking text. Empty when `display` is `omitted`.
        thinking: String,
        /// Encrypted copy of the full reasoning, replayed back unmodified.
        #[serde(default)]
        signature: String,
    },
    /// Safety-redacted thinking. Carries no readable text, only ciphertext.
    #[serde(rename = "redacted_thinking")]
    RedactedThinking {
        /// Opaque encrypted thinking, replayed back unmodified.
        data: String,
    },
    /// Tool use content block.
    #[serde(rename = "tool_use")]
    ToolUse {
        /// Unique ID for this tool use.
        id: String,
        /// Tool name.
        name: String,
        /// Tool input (builds up via deltas).
        #[serde(default)]
        input: Value,
    },
    /// Server-executed tool use — the API runs it, the caller only observes.
    #[serde(rename = "server_tool_use")]
    ServerToolUse {
        /// Unique ID for this tool use.
        id: String,
        /// Tool name (`web_search`, `web_fetch`, `code_execution`, ...).
        name: String,
        /// Tool input (builds up via `input_json_delta`).
        #[serde(default)]
        input: Value,
    },
    /// Result of a server-side web search.
    #[serde(rename = "web_search_tool_result")]
    WebSearchToolResult {
        /// ID of the `server_tool_use` block this answers.
        tool_use_id: String,
        /// Search hits or an error object.
        #[serde(default)]
        content: Value,
    },
    /// Result of a server-side web fetch.
    #[serde(rename = "web_fetch_tool_result")]
    WebFetchToolResult {
        /// ID of the `server_tool_use` block this answers.
        tool_use_id: String,
        /// Retrieved document or an error object.
        #[serde(default)]
        content: Value,
    },
    /// Result of a server-side code execution.
    #[serde(rename = "code_execution_tool_result")]
    CodeExecutionToolResult {
        /// ID of the `server_tool_use` block this answers.
        tool_use_id: String,
        /// Execution output or an error object.
        #[serde(default)]
        content: Value,
    },
    /// Result of a server-side bash code execution.
    #[serde(rename = "bash_code_execution_tool_result")]
    BashCodeExecutionToolResult {
        /// ID of the `server_tool_use` block this answers.
        tool_use_id: String,
        /// Execution output or an error object.
        #[serde(default)]
        content: Value,
    },
    /// Result of a server-side text-editor code execution.
    #[serde(rename = "text_editor_code_execution_tool_result")]
    TextEditorCodeExecutionToolResult {
        /// ID of the `server_tool_use` block this answers.
        tool_use_id: String,
        /// Execution output or an error object.
        #[serde(default)]
        content: Value,
    },
    /// A block type this crate does not model yet — ignored rather than
    /// fatal, matching the API's forwards-compatibility contract.
    #[serde(other)]
    Unknown,
}

/// Content block delta event data.
#[derive(Debug, Deserialize)]
pub struct ContentBlockDeltaEvent {
    /// Index of the content block being updated.
    pub index: usize,
    /// The delta update.
    pub delta: DeltaType,
}

/// Types of delta updates.
#[derive(Debug, Deserialize)]
#[serde(tag = "type")]
#[allow(clippy::enum_variant_names)]
pub enum DeltaType {
    /// Text delta.
    #[serde(rename = "text_delta")]
    TextDelta {
        /// Text fragment to append.
        text: String,
    },
    /// Thinking delta.
    #[serde(rename = "thinking_delta")]
    ThinkingDelta {
        /// Thinking fragment to append.
        thinking: String,
    },
    /// Signature delta, arriving once just before the thinking block closes.
    #[serde(rename = "signature_delta")]
    SignatureDelta {
        /// The block's signature.
        signature: String,
    },
    /// Input JSON delta for tool use.
    #[serde(rename = "input_json_delta")]
    InputJsonDelta {
        /// Partial JSON to append.
        partial_json: String,
    },
    /// A delta type this crate does not model yet — ignored rather than
    /// fatal, matching the API's forwards-compatibility contract.
    #[serde(other)]
    Unknown,
}

/// Message delta event data (final updates).
#[derive(Debug, Deserialize)]
#[allow(dead_code)]
pub struct MessageDeltaEvent {
    /// Delta updates to the message.
    pub delta: MessageDelta,
    /// Updated usage information.
    #[serde(default)]
    pub usage: Option<DeltaUsage>,
}

/// Message-level delta updates.
#[derive(Debug, Deserialize)]
pub struct MessageDelta {
    /// Reason the response stopped.
    pub stop_reason: Option<String>,
}

/// Usage information in message delta.
#[derive(Debug, Deserialize)]
#[allow(dead_code)]
pub struct DeltaUsage {
    /// Total output tokens.
    #[serde(default)]
    pub output_tokens: Option<u32>,
    /// Number of input tokens written into cache.
    #[serde(default)]
    pub cache_creation_input_tokens: Option<u32>,
    /// Number of input tokens read from cache.
    #[serde(default)]
    pub cache_read_input_tokens: Option<u32>,
    /// Detailed cache creation token counts by TTL.
    #[serde(default)]
    pub cache_creation: Option<CacheCreation>,
}

impl DeltaUsage {
    fn cache_write_tokens(&self) -> Option<u32> {
        self.cache_creation_input_tokens
            .or_else(|| self.cache_creation.as_ref().and_then(CacheCreation::total))
    }
}

/// Detailed cache creation token counts by TTL.
#[derive(Debug, Deserialize)]
pub struct CacheCreation {
    /// Tokens created with 5-minute TTL.
    #[serde(default)]
    pub ephemeral_5m_input_tokens: Option<u32>,
    /// Tokens created with 1-hour TTL.
    #[serde(default)]
    pub ephemeral_1h_input_tokens: Option<u32>,
}

impl CacheCreation {
    fn total(&self) -> Option<u32> {
        // `None` and `Some(0)` mean different things here: the former is
        // "the API said nothing about cache creation", the latter "nothing was
        // cached", so only sum when at least one field was reported.
        match (
            self.ephemeral_5m_input_tokens,
            self.ephemeral_1h_input_tokens,
        ) {
            (None, None) => None,
            (a, b) => Some(a.unwrap_or(0).saturating_add(b.unwrap_or(0))),
        }
    }
}

/// Parsed tool call from a completed `tool_use` block.
#[derive(Debug, Clone)]
pub struct ToolCall {
    /// Unique ID for this tool use.
    pub id: String,
    /// Tool name.
    pub name: String,
    /// Parsed tool input.
    pub input: Value,
}

/// Non-streaming `/v1/messages` response body.
///
/// The wire shape mirrors the terminal state an SSE stream ends in: the same
/// content blocks arrive complete rather than as start/delta/stop triples.
#[derive(Debug, Deserialize)]
pub struct MessagesResponse {
    /// Content blocks produced by the model.
    pub content: Vec<ContentBlockType>,
    /// Reason generation stopped.
    #[serde(default)]
    pub stop_reason: Option<String>,
    /// Token usage for the call.
    #[serde(default)]
    pub usage: Option<Usage>,
}

/// Projects a non-streaming [`MessagesResponse`] into the same event stream a
/// streaming response would have produced.
///
/// Each block is replayed through the same start/delta/stop handling the SSE
/// parser drives, so every block type the stream can carry is projected by
/// construction rather than re-implemented here.
#[must_use]
pub fn response_events(response: MessagesResponse) -> Vec<LLMEvent> {
    let mut state = StreamState::new();
    let mut events = Vec::new();
    for (index, block) in response.content.into_iter().enumerate() {
        events.extend(state.replay_block(index, block));
    }
    if let Some(usage) = response.usage {
        state.uncached_input_tokens = Some(usage.input_tokens);
        state.completion_tokens = Some(usage.output_tokens);
        state.cache_write_tokens = usage.cache_write_tokens();
        state.cache_read_tokens = usage.cache_read_input_tokens;
        state.refresh_prompt_tokens();
    }
    state.stop_reason = response.stop_reason;
    events.extend(state.maybe_usage_event());
    for call in state.tool_calls {
        events.push(LLMEvent::tool_call(call.id, call.name, call.input));
    }
    events
}

/// State for tracking content blocks during streaming.
#[derive(Debug, Default)]
pub struct StreamState {
    /// Current content blocks being built.
    pub blocks: Vec<BlockState>,
    /// Completed tool calls.
    pub tool_calls: Vec<ToolCall>,
    /// Final stop reason.
    pub stop_reason: Option<String>,
    /// Prompt/input token usage.
    pub prompt_tokens: Option<u32>,
    /// Uncached input tokens (`usage.input_tokens` from Claude).
    pub uncached_input_tokens: Option<u32>,
    /// Completion/output token usage.
    pub completion_tokens: Option<u32>,
    /// Prompt cache read token usage.
    pub cache_read_tokens: Option<u32>,
    /// Prompt cache write token usage.
    pub cache_write_tokens: Option<u32>,
    /// Whether usage has already been emitted.
    pub usage_emitted: bool,
}

/// State of an individual content block.
#[derive(Debug, Clone)]
pub enum BlockState {
    /// Text block with accumulated text.
    Text(String),
    /// Thinking block with accumulated reasoning and its signature.
    Thinking {
        /// Accumulated thinking text; empty when `display` is `omitted`.
        text: String,
        /// Signature, which arrives as a single delta before the block closes.
        signature: String,
    },
    /// Safety-redacted thinking block.
    RedactedThinking(String),
    /// Tool use block with accumulated JSON.
    ToolUse {
        /// Tool use ID.
        id: String,
        /// Tool name.
        name: String,
        /// Accumulated input JSON string.
        input_json: String,
    },
    /// Server-executed tool use with accumulated input JSON.
    ServerToolUse {
        /// Tool use ID.
        id: String,
        /// Tool name.
        name: String,
        /// Accumulated input JSON string.
        input_json: String,
    },
}

impl StreamState {
    /// Create a new empty stream state.
    pub fn new() -> Self {
        Self::default()
    }

    fn maybe_usage_event(&mut self) -> Option<LLMEvent> {
        if self.usage_emitted {
            return None;
        }
        if self.prompt_tokens.is_none()
            && self.completion_tokens.is_none()
            && self.cache_read_tokens.is_none()
            && self.cache_write_tokens.is_none()
            && self.stop_reason.is_none()
        {
            return None;
        }
        self.usage_emitted = true;
        let total_tokens = match (self.prompt_tokens, self.completion_tokens) {
            (Some(prompt), Some(completion)) => Some(prompt + completion),
            _ => None,
        };
        Some(LLMEvent::Usage(TokenUsage {
            prompt_tokens: self.prompt_tokens,
            completion_tokens: self.completion_tokens,
            total_tokens,
            reasoning_tokens: None,
            cache_read_tokens: self.cache_read_tokens,
            cache_write_tokens: self.cache_write_tokens,
            cost_usd: None,
            stop_reason: self.stop_reason.clone(),
        }))
    }

    fn refresh_prompt_tokens(&mut self) {
        // As above: report a total only when the API reported at least one of
        // the components, so "not stated" stays distinct from "zero".
        let parts = [
            self.uncached_input_tokens,
            self.cache_read_tokens,
            self.cache_write_tokens,
        ];
        self.prompt_tokens = parts.iter().any(Option::is_some).then(|| {
            parts
                .iter()
                .flatten()
                .fold(0u32, |total, part| total.saturating_add(*part))
        });
    }
}

/// Parse a single SSE event into LLM events.
///
/// Updates the stream state and returns events to emit.
///
/// # Errors
///
/// Returns [`ClaudeError`] if the event body does not parse, or if the API sent
/// an `error` event.
pub fn parse_event(event: &Event, state: &mut StreamState) -> Result<Vec<LLMEvent>, ClaudeError> {
    let data = event.text_data();

    match event.event().unwrap_or("") {
        "message_start" => {
            state.apply_message_start(serde_json::from_str(data)?);
            Ok(Vec::new())
        }
        "content_block_start" => Ok(state.begin_block(serde_json::from_str(data)?)),
        "content_block_delta" => Ok(state.apply_block_delta(serde_json::from_str(data)?)),
        "content_block_stop" => {
            #[derive(Deserialize)]
            struct StopEvent {
                index: usize,
            }
            let ev: StopEvent = serde_json::from_str(data)?;
            Ok(state.finish_block(ev.index))
        }
        "message_delta" => {
            state.apply_message_delta(serde_json::from_str(data)?);
            Ok(Vec::new())
        }
        "message_stop" => Ok(state.maybe_usage_event().into_iter().collect()),
        // Keepalives carry nothing to emit.
        "ping" | "" => Ok(Vec::new()),
        "error" => Err(parse_error_event(data)),
        unknown => {
            // Anthropic adds event types over time; an unfamiliar one is not a
            // reason to fail the stream.
            tracing::debug!("Unknown Claude SSE event type: {unknown}");
            Ok(Vec::new())
        }
    }
}

/// Reads the message out of an `error` event, falling back to its raw body.
fn parse_error_event(data: &str) -> ClaudeError {
    #[derive(Deserialize)]
    struct ErrorEvent {
        error: ErrorDetail,
    }
    #[derive(Deserialize)]
    struct ErrorDetail {
        #[serde(rename = "type")]
        error_type: String,
        message: String,
    }

    serde_json::from_str::<ErrorEvent>(data).map_or_else(
        |_| ClaudeError::Api(data.to_string()),
        |ev| ClaudeError::StreamEvent {
            error_type: ev.error.error_type,
            message: ev.error.message,
        },
    )
}

impl StreamState {
    /// Records the usage figures the API reports when a message opens.
    fn apply_message_start(&mut self, ev: MessageStartEvent) {
        let Some(usage) = ev.message.usage else {
            return;
        };
        self.uncached_input_tokens = Some(usage.input_tokens);
        self.completion_tokens = Some(usage.output_tokens);
        self.cache_write_tokens = usage.cache_write_tokens();
        self.cache_read_tokens = usage.cache_read_input_tokens;
        self.refresh_prompt_tokens();
    }

    /// Opens a content block, emitting whatever text arrived with it.
    fn begin_block(&mut self, ev: ContentBlockStartEvent) -> Vec<LLMEvent> {
        ensure_block_capacity(self, ev.index);

        // Server tool results carry their payload whole at block start.
        if let Some((tool, tool_use_id, content)) = ev.content_block.server_tool_result() {
            return vec![LLMEvent::builtin_result(
                tool_use_id,
                tool,
                server_tool_result_text(tool, content),
            )];
        }

        match ev.content_block {
            ContentBlockType::Text { text } => {
                self.blocks[ev.index] = BlockState::Text(text.clone());
                if text.is_empty() {
                    Vec::new()
                } else {
                    vec![LLMEvent::Text(text)]
                }
            }
            ContentBlockType::Thinking {
                thinking,
                signature,
            } => {
                self.blocks[ev.index] = BlockState::Thinking {
                    text: thinking.clone(),
                    signature,
                };
                if thinking.is_empty() {
                    Vec::new()
                } else {
                    vec![LLMEvent::Reasoning(thinking)]
                }
            }
            // Redacted thinking has no readable text, so it produces no
            // Reasoning event — only state, emitted when the block closes.
            ContentBlockType::RedactedThinking { data } => {
                self.blocks[ev.index] = BlockState::RedactedThinking(data);
                Vec::new()
            }
            ContentBlockType::ToolUse { id, name, .. } => {
                // Emit an initial delta so a UI can show the tool name before
                // any arguments have streamed in.
                let started = LLMEvent::ToolCallDelta {
                    id: id.clone(),
                    name: name.clone(),
                    arguments_fragment: String::new(),
                };
                self.blocks[ev.index] = BlockState::ToolUse {
                    id,
                    name,
                    input_json: String::new(),
                };
                vec![started]
            }
            ContentBlockType::ServerToolUse { id, name, .. } => {
                self.blocks[ev.index] = BlockState::ServerToolUse {
                    id,
                    name,
                    input_json: String::new(),
                };
                Vec::new()
            }
            // Server tool results were handled above; unknown blocks drop.
            _ => Vec::new(),
        }
    }

    /// Appends a delta to its block, emitting it if it is text the caller sees.
    ///
    /// A delta whose type does not match the block it names is dropped: the two
    /// disagreeing leaves nothing meaningful to append.
    fn apply_block_delta(&mut self, ev: ContentBlockDeltaEvent) -> Vec<LLMEvent> {
        let Some(block) = self.blocks.get_mut(ev.index) else {
            return Vec::new();
        };

        match (block, ev.delta) {
            (BlockState::Text(text), DeltaType::TextDelta { text: delta }) => {
                text.push_str(&delta);
                vec![LLMEvent::Text(delta)]
            }
            (BlockState::Thinking { text, .. }, DeltaType::ThinkingDelta { thinking: delta }) => {
                text.push_str(&delta);
                vec![LLMEvent::Reasoning(delta)]
            }
            // Arrives once, just before content_block_stop. With
            // `display: "omitted"` it is the only delta the block produces.
            (
                BlockState::Thinking { signature, .. },
                DeltaType::SignatureDelta {
                    signature: delta, ..
                },
            ) => {
                signature.push_str(&delta);
                Vec::new()
            }
            (
                BlockState::ToolUse {
                    id,
                    name,
                    input_json,
                },
                DeltaType::InputJsonDelta { partial_json },
            ) => {
                input_json.push_str(&partial_json);
                vec![LLMEvent::ToolCallDelta {
                    id: id.clone(),
                    name: name.clone(),
                    arguments_fragment: partial_json,
                }]
            }
            (
                BlockState::ServerToolUse { input_json, .. },
                DeltaType::InputJsonDelta { partial_json },
            ) => {
                input_json.push_str(&partial_json);
                Vec::new()
            }
            _ => Vec::new(),
        }
    }

    /// Closes a content block, turning a finished tool-use block into a call.
    ///
    /// Arguments that did not parse become an empty object rather than failing
    /// the stream: the tool reports the mismatch far more usefully than a
    /// truncated JSON error would.
    /// A closed thinking block emits its state here rather than at block start,
    /// because the signature arrives last: only now is the block complete
    /// enough to be replayed. The payload is the whole block re-serialized, so
    /// that `thinking` and `redacted_thinking` — which have different shapes —
    /// both round-trip without core knowing either.
    fn finish_block(&mut self, index: usize) -> Vec<LLMEvent> {
        match self.blocks.get(index) {
            Some(BlockState::ToolUse {
                id,
                name,
                input_json,
            }) => {
                let input = serde_json::from_str(input_json)
                    .unwrap_or_else(|_| Value::Object(serde_json::Map::new()));
                self.tool_calls.push(ToolCall {
                    id: id.clone(),
                    name: name.clone(),
                    input,
                });
                Vec::new()
            }
            Some(BlockState::Thinking { text, signature }) => {
                // An unsigned block cannot be verified on replay, so there is
                // nothing worth preserving.
                if signature.is_empty() {
                    return Vec::new();
                }
                let block = serde_json::json!({
                    "type": "thinking",
                    "thinking": text,
                    "signature": signature,
                });
                vec![LLMEvent::ReasoningState(ReasoningState::new(
                    PROVIDER_NAME,
                    block.to_string(),
                ))]
            }
            Some(BlockState::RedactedThinking(data)) => {
                let block = serde_json::json!({
                    "type": "redacted_thinking",
                    "data": data,
                });
                vec![LLMEvent::ReasoningState(ReasoningState::new(
                    PROVIDER_NAME,
                    block.to_string(),
                ))]
            }
            Some(BlockState::ServerToolUse {
                id,
                name,
                input_json,
            }) => {
                let input = serde_json::from_str::<Value>(input_json)
                    .unwrap_or_else(|_| Value::Object(serde_json::Map::new()));
                // The server runs the call between this block and the result
                // block, so the call event belongs at close — with the input
                // complete — not at open.
                vec![LLMEvent::builtin_call(
                    id,
                    name,
                    server_tool_call_detail(&input),
                )]
            }
            Some(BlockState::Text(_)) | None => Vec::new(),
        }
    }

    /// Folds in the running usage figures the API reports as it generates.
    fn apply_message_delta(&mut self, ev: MessageDeltaEvent) {
        self.stop_reason = ev.delta.stop_reason;
        let Some(usage) = ev.usage else {
            return;
        };
        if let Some(output_tokens) = usage.output_tokens {
            self.completion_tokens = Some(output_tokens);
        }
        if let Some(cache_write_tokens) = usage.cache_write_tokens() {
            self.cache_write_tokens = Some(cache_write_tokens);
        }
        if let Some(cache_read_tokens) = usage.cache_read_input_tokens {
            self.cache_read_tokens = Some(cache_read_tokens);
        }
        self.refresh_prompt_tokens();
    }

    /// Replays a complete content block through the block lifecycle —
    /// non-streaming responses deliver whole blocks, so this synthesizes the
    /// start/delta/stop sequence a stream would have carried for them.
    ///
    /// Tool input travels as `input_json_delta` events on a block that opens
    /// with an empty `input`, and a missing input means the stream sent none.
    fn replay_block(&mut self, index: usize, block: ContentBlockType) -> Vec<LLMEvent> {
        let input_json = match &block {
            ContentBlockType::ToolUse { input, .. }
            | ContentBlockType::ServerToolUse { input, .. }
                if !input.is_null() =>
            {
                Some(input.to_string())
            }
            _ => None,
        };
        let mut events = self.begin_block(ContentBlockStartEvent {
            index,
            content_block: block,
        });
        if let Some(partial_json) = input_json {
            events.extend(self.apply_block_delta(ContentBlockDeltaEvent {
                index,
                delta: DeltaType::InputJsonDelta { partial_json },
            }));
        }
        events.extend(self.finish_block(index));
        events
    }
}

/// Ensure the blocks vector has capacity for the given index.
fn ensure_block_capacity(state: &mut StreamState, index: usize) {
    while state.blocks.len() <= index {
        state.blocks.push(BlockState::Text(String::new()));
    }
}

/// Check if an SSE event should be skipped.
pub fn should_skip_event(event: &Event) -> bool {
    let data = event.text_data();
    data.is_empty() || event.event() == Some("ping")
}

impl ContentBlockType {
    /// `(tool_name, tool_use_id, content)` for server tool result blocks —
    /// blocks the API executes itself and delivers whole.
    fn server_tool_result(&self) -> Option<(&'static str, &str, &Value)> {
        Some(match self {
            Self::WebSearchToolResult {
                tool_use_id,
                content,
            } => ("web_search", tool_use_id, content),
            Self::WebFetchToolResult {
                tool_use_id,
                content,
            } => ("web_fetch", tool_use_id, content),
            Self::CodeExecutionToolResult {
                tool_use_id,
                content,
            } => ("code_execution", tool_use_id, content),
            Self::BashCodeExecutionToolResult {
                tool_use_id,
                content,
            } => ("bash_code_execution", tool_use_id, content),
            Self::TextEditorCodeExecutionToolResult {
                tool_use_id,
                content,
            } => ("text_editor_code_execution", tool_use_id, content),
            _ => return None,
        })
    }
}

/// Invocation detail for a server tool call: the query, URL or command the
/// block carried, compact JSON otherwise.
fn server_tool_call_detail(input: &Value) -> String {
    for key in ["query", "url", "command", "code"] {
        if let Some(value) = input.get(key).and_then(Value::as_str) {
            return value.to_string();
        }
    }
    match input {
        Value::Object(map) if map.is_empty() => String::new(),
        other => other.to_string(),
    }
}

/// One-line summary of a server tool result block: the error code on failure,
/// a result count or output excerpt on success.
fn server_tool_result_text(tool: &str, content: &Value) -> String {
    if let Some(code) = content.get("error_code").and_then(Value::as_str) {
        return format!("error: {code}");
    }
    if tool == "web_search" {
        // `web_search_result_block` wraps a list of hits.
        if let Some(hits) = content.get("content").and_then(Value::as_array) {
            return format!("{} results", hits.len());
        }
        if content.is_array() {
            return format!("{} results", content.as_array().map_or(0, Vec::len));
        }
    }
    if tool == "web_fetch"
        && let Some(url) = content.get("url").and_then(Value::as_str)
    {
        return format!("fetched {url}");
    }
    if let Some(text) = ["stdout", "stderr"]
        .iter()
        .filter_map(|key| content.get(*key).and_then(Value::as_str))
        .find(|text| !text.is_empty())
    {
        return text.chars().take(200).collect();
    }
    "completed".to_string()
}

#[cfg(test)]
mod tests {
    use super::*;

    fn block_start(data: &str) -> ContentBlockStartEvent {
        serde_json::from_str(data).expect("content_block_start parses")
    }

    #[test]
    fn server_tool_use_emits_builtin_call_at_block_close() {
        let mut state = StreamState::new();
        let events = state.begin_block(block_start(
            r#"{"index":0,"content_block":{"type":"server_tool_use","id":"srvtoolu_1","name":"web_search","input":{}}}"#,
        ));
        assert!(events.is_empty());

        let events = state.apply_block_delta(
            serde_json::from_str(
                r#"{"index":0,"delta":{"type":"input_json_delta","partial_json":"{\"query\":\"rust\"}"}}"#,
            )
            .expect("delta parses"),
        );
        assert!(events.is_empty());

        match state.finish_block(0).as_slice() {
            [LLMEvent::BuiltInToolCall { id, tool, detail }] => {
                assert_eq!(id, "srvtoolu_1");
                assert_eq!(tool, "web_search");
                assert_eq!(detail, "rust");
            }
            other => panic!("expected BuiltInToolCall, got {other:?}"),
        }
        // Server calls never join the client-side tool queue.
        assert!(state.tool_calls.is_empty());
    }

    #[test]
    fn web_search_tool_result_emits_builtin_result() {
        let mut state = StreamState::new();
        match state
            .begin_block(block_start(
                r#"{"index":1,"content_block":{"type":"web_search_tool_result","tool_use_id":"srvtoolu_1","content":{"type":"web_search_result_block","content":[{"type":"web_search_result","title":"t","url":"u"}]}}}"#,
            ))
            .as_slice()
        {
            [LLMEvent::BuiltInToolResult { id, tool, result }] => {
                assert_eq!(id, "srvtoolu_1");
                assert_eq!(tool, "web_search");
                assert_eq!(result, "1 results");
            }
            other => panic!("expected BuiltInToolResult, got {other:?}"),
        }
    }

    #[test]
    fn unknown_block_types_are_tolerated() {
        let mut state = StreamState::new();
        let events = state.begin_block(block_start(
            r#"{"index":0,"content_block":{"type":"some_future_block","id":"x"}}"#,
        ));
        assert!(events.is_empty());
    }

    #[test]
    fn usage_event_includes_claude_cache_token_stats() {
        let mut state = StreamState::new();
        state.prompt_tokens = Some(120);
        state.completion_tokens = Some(30);
        state.cache_read_tokens = Some(90);
        state.cache_write_tokens = Some(45);
        state.stop_reason = Some("end_turn".to_string());

        let event = state
            .maybe_usage_event()
            .expect("usage event should be emitted");
        match event {
            LLMEvent::Usage(usage) => {
                assert_eq!(usage.prompt_tokens, Some(120));
                assert_eq!(usage.completion_tokens, Some(30));
                assert_eq!(usage.total_tokens, Some(150));
                assert_eq!(usage.cache_read_tokens, Some(90));
                assert_eq!(usage.cache_write_tokens, Some(45));
                assert_eq!(usage.stop_reason.as_deref(), Some("end_turn"));
            }
            _ => panic!("expected usage event"),
        }
    }

    #[test]
    fn usage_event_is_not_reemitted() {
        let mut state = StreamState::new();
        state.prompt_tokens = Some(1);
        assert!(state.maybe_usage_event().is_some());
        assert!(state.maybe_usage_event().is_none());
    }

    #[test]
    fn cache_creation_total_sums_5m_and_1h_tokens() {
        let cache_creation = CacheCreation {
            ephemeral_5m_input_tokens: Some(20),
            ephemeral_1h_input_tokens: Some(30),
        };
        assert_eq!(cache_creation.total(), Some(50));
    }

    #[test]
    fn refresh_prompt_tokens_adds_uncached_and_cache_tokens() {
        let mut state = StreamState::new();
        state.uncached_input_tokens = Some(40);
        state.cache_read_tokens = Some(30);
        state.cache_write_tokens = Some(10);
        state.refresh_prompt_tokens();
        assert_eq!(state.prompt_tokens, Some(80));
    }

    #[test]
    fn response_events_matches_streaming_projection() {
        let response: MessagesResponse = serde_json::from_str(
            r#"{
                "content": [
                    {"type": "text", "text": "Checking the weather."},
                    {"type": "thinking", "thinking": "Need a forecast.", "signature": "sig_1"},
                    {"type": "tool_use", "id": "toolu_1", "name": "get_weather", "input": {"city": "Paris"}},
                    {"type": "server_tool_use", "id": "srvtoolu_1", "name": "web_search", "input": {"query": "Paris weather"}},
                    {"type": "web_search_tool_result", "tool_use_id": "srvtoolu_1", "content": [{"type": "web_search_result", "title": "t", "url": "u"}]}
                ],
                "stop_reason": "tool_use",
                "usage": {"input_tokens": 40, "output_tokens": 30, "cache_read_input_tokens": 10, "cache_creation_input_tokens": 5}
            }"#,
        )
        .expect("response parses");

        // The same response as the API would have streamed it.
        let streamed = [
            (
                "message_start",
                r#"{"message":{"id":"msg_1","model":"claude-test","usage":{"input_tokens":40,"output_tokens":1,"cache_read_input_tokens":10,"cache_creation_input_tokens":5}}}"#,
            ),
            (
                "content_block_start",
                r#"{"index":0,"content_block":{"type":"text","text":""}}"#,
            ),
            (
                "content_block_delta",
                r#"{"index":0,"delta":{"type":"text_delta","text":"Checking the weather."}}"#,
            ),
            ("content_block_stop", r#"{"index":0}"#),
            (
                "content_block_start",
                r#"{"index":1,"content_block":{"type":"thinking","thinking":""}}"#,
            ),
            (
                "content_block_delta",
                r#"{"index":1,"delta":{"type":"thinking_delta","thinking":"Need a forecast."}}"#,
            ),
            (
                "content_block_delta",
                r#"{"index":1,"delta":{"type":"signature_delta","signature":"sig_1"}}"#,
            ),
            ("content_block_stop", r#"{"index":1}"#),
            (
                "content_block_start",
                r#"{"index":2,"content_block":{"type":"tool_use","id":"toolu_1","name":"get_weather","input":{}}}"#,
            ),
            (
                "content_block_delta",
                r#"{"index":2,"delta":{"type":"input_json_delta","partial_json":"{\"city\":\"Paris\"}"}}"#,
            ),
            ("content_block_stop", r#"{"index":2}"#),
            (
                "content_block_start",
                r#"{"index":3,"content_block":{"type":"server_tool_use","id":"srvtoolu_1","name":"web_search","input":{}}}"#,
            ),
            (
                "content_block_delta",
                r#"{"index":3,"delta":{"type":"input_json_delta","partial_json":"{\"query\":\"Paris weather\"}"}}"#,
            ),
            ("content_block_stop", r#"{"index":3}"#),
            (
                "content_block_start",
                r#"{"index":4,"content_block":{"type":"web_search_tool_result","tool_use_id":"srvtoolu_1","content":[{"type":"web_search_result","title":"t","url":"u"}]}}"#,
            ),
            ("content_block_stop", r#"{"index":4}"#),
            (
                "message_delta",
                r#"{"delta":{"stop_reason":"tool_use"},"usage":{"output_tokens":30}}"#,
            ),
            ("message_stop", r#"{"type":"message_stop"}"#),
        ];

        let mut state = StreamState::new();
        let mut expected = Vec::new();
        for (kind, data) in streamed {
            expected.extend(
                parse_event(&Event::from_data(data).with_event(kind), &mut state)
                    .expect("SSE event parses"),
            );
        }
        // The client drains completed tool calls after the stream ends.
        expected.extend(
            std::mem::take(&mut state.tool_calls)
                .into_iter()
                .map(|call| LLMEvent::tool_call(call.id, call.name, call.input)),
        );

        assert!(
            expected
                .iter()
                .any(|event| matches!(event, LLMEvent::BuiltInToolCall { .. }))
        );
        assert!(
            expected
                .iter()
                .any(|event| matches!(event, LLMEvent::BuiltInToolResult { .. }))
        );
        assert_eq!(
            format!("{:?}", response_events(response)),
            format!("{expected:?}")
        );
    }
}