use std::{env, mem, time::Duration};

use async_anthropic::{
    Client,
    errors::AnthropicError,
    messages::DEFAULT_MAX_TOKENS,
    types::{
        self, Effort, JsonOutputFormat, ListModelsResponse, OutputConfig, System, Thinking,
        ToolBash, ToolCodeExecution, ToolComputerUse, ToolTextEditor, ToolWebSearch,
    },
};
use async_stream::try_stream;
use async_trait::async_trait;
use base64::Engine as _;
use chrono::NaiveDate;
use futures::{StreamExt as _, TryStreamExt as _, pin_mut, stream};
use jp_attachment::AttachmentContent;
use jp_config::{
    assistant::tool_choice::ToolChoice,
    model::{
        id::{Name, ProviderId},
        parameters::ReasoningEffort,
    },
    providers::llm::anthropic::AnthropicConfig,
};
use jp_conversation::{
    ConversationStream,
    event::{ChatResponse, ConversationEvent, EventKind},
};
use serde_json::{Map, Value, json};
use tracing::{debug, info, trace, warn};

use super::{Provider, trace_to_tmpfile};
use crate::{
    error::{
        Error, Result, StreamError, StreamErrorKind, extract_retry_from_text,
        looks_like_quota_error,
    },
    event::{Event, EventMatcher, EventPart, EventPatch, FinishReason, PatchAction, ToolCallPart},
    event_builder::EventBuilder,
    model::{ModelDeprecation, ModelDetails, ReasoningDetails},
    query::ChatQuery,
    stream::{EventStream, chain::find_merge_point, with_tool_call_keepalive},
    tool::ToolDefinition,
};

static PROVIDER: ProviderId = ProviderId::Anthropic;

/// Anthropic limits the number of explicit cache breakpoints to 4 per request,
/// returning an API error if the request exceeds this limit.
///
/// We use automatic caching (top-level `cache_control` on the request) to
/// handle the growing message history, which consumes 1 of the 4 slots.
/// The remaining 3 are allocated to stable, explicitly-cached content: system
/// prompt, attachments, and tools.
///
/// See:
/// <https://platform.claude.com/docs/en/build-with-claude/prompt-caching#combining-with-block-level-caching>
const MAX_EXPLICIT_CACHE_CONTROL_COUNT: usize = 3;

const THINKING_SIGNATURE_KEY: &str = "anthropic_thinking_signature";
const REDACTED_THINKING_KEY: &str = "anthropic_redacted_thinking";

/// Known Anthropic error types that are safe to retry.
///
/// See:
/// <https://docs.claude.com/en/docs/build-with-claude/streaming#error-events>
/// See: <https://docs.claude.com/en/api/errors#http-errors>
const RETRYABLE_ANTHROPIC_ERROR_TYPES: &[&str] =
    &["rate_limit_error", "overloaded_error", "api_error"];

/// Supported string `format` values for Anthropic structured output.
const SUPPORTED_STRING_FORMATS: &[&str] = &[
    "date-time",
    "time",
    "date",
    "duration",
    "email",
    "hostname",
    "uri",
    "ipv4",
    "ipv6",
    "uuid",
];

/// Synthetic user message injected when the conversation ends with an assistant
/// message and the model does not support assistant prefill.
///
/// This replaces the prefill technique (where an assistant message as the last
/// event causes the model to continue from that point) for models that require
/// conversations to end with a user message.
const CONTINUE_MESSAGE: &str = "Continue your response exactly from where you left off. Do not \
                                repeat content you already produced.";

/// Default minimum overlap for merge point detection during chaining.
const CHAIN_MIN_OVERLAP: usize = 5;

/// How often to inject a synthetic keep-alive while a tool call is streaming.
///
/// Anthropic emits a large tool-call argument as a single key/value burst after
/// a potentially long silent gap, which the idle timeout would otherwise treat
/// as a dead connection.
/// This interval stays comfortably below the enforced minimum
/// `stream_idle_timeout_secs` (10s), so the heartbeat always lands before the
/// idle window elapses.
///
/// See:
/// <https://platform.claude.com/docs/en/build-with-claude/streaming#input-json-delta>
const TOOL_CALL_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(5);

#[derive(Debug, Clone)]
pub struct Anthropic {
    client: Client,

    /// See [`AnthropicConfig::chain_on_max_tokens`].
    chain_on_max_tokens: bool,

    /// Which beta features are enabled.
    beta: BetaFeatures,
}

#[async_trait]
impl Provider for Anthropic {
    async fn model_details(&self, name: &Name) -> Result<ModelDetails> {
        let model = self.client.models().get(name).await?;
        map_model(model)
    }

    async fn models(&self) -> Result<Vec<ModelDetails>> {
        let mut all_models = vec![];
        let mut after_id = None;

        loop {
            let path = match after_id {
                Some(id) => format!("/v1/models?after_id={id}"),
                None => "/v1/models".to_string(),
            };

            let models;
            (after_id, models) = self
                .client
                .get::<ListModelsResponse>(&path)
                .await
                .map(|list| (list.has_more.then_some(list.last_id).flatten(), list.data))?;

            all_models.extend(models);

            if after_id.is_none() {
                break;
            }
        }

        all_models.into_iter().map(map_model).collect::<Result<_>>()
    }

    async fn chat_completion_stream(
        &self,
        model: &ModelDetails,
        query: ChatQuery,
    ) -> Result<EventStream> {
        let client = self.client.clone();
        let max_tokens_config = query
            .thread
            .events
            .config()?
            .assistant
            .model
            .parameters
            .max_tokens;

        let (request, is_structured, forced_tool) = create_request(model, query, true, &self.beta)?;

        // Chaining is disabled for structured output — the provider guarantees
        // schema compliance so the response won't hit max_tokens for a
        // well-constrained schema.
        //
        // It is also disabled when the user has explicitly configured a max
        // tokens value, or when chaining is disabled in the provider config.
        let chain_on_max_tokens =
            !is_structured && max_tokens_config.is_none() && self.chain_on_max_tokens;

        debug!(stream = true, "Anthropic chat completion stream request.");
        trace!(
            request = %trace_to_tmpfile("jp-anthropic-request", &request),
            "Request payload."
        );

        Ok(with_tool_call_keepalive(
            call(
                client,
                request,
                chain_on_max_tokens,
                is_structured,
                forced_tool,
            ),
            TOOL_CALL_KEEPALIVE_INTERVAL,
        ))
    }
}

/// Context for retrying with forced tool use when an initial soft-forced
/// request (reasoning enabled + `tool_choice: auto` with a system prompt nudge)
/// completes without the model calling any tool.
///
/// See:
/// <https://docs.claude.com/en/docs/build-with-claude/extended-thinking#extended-thinking-with-tool-use>
#[derive(Debug, Clone)]
struct ForcedToolFallback {
    /// The Anthropic API `tool_choice` to use on the retry request (the
    /// original forced value before we downgraded it to `auto`).
    tool_choice: types::ToolChoice,
}

impl ForcedToolFallback {
    /// Check whether the forced tool constraint is satisfied by the tool calls
    /// observed so far.
    ///
    /// - `Any` — any tool call satisfies the constraint.
    /// - `Tool { name }` — only a call to that specific tool counts.
    fn is_satisfied_by(&self, tool_names_called: &[String]) -> bool {
        match &self.tool_choice {
            types::ToolChoice::Tool { name, .. } => tool_names_called.iter().any(|n| n == name),
            types::ToolChoice::Any { .. } => !tool_names_called.is_empty(),
            types::ToolChoice::None | types::ToolChoice::Auto { .. } => true,
        }
    }
}

/// Create a streaming request and return a stream of [`Event`]s.
///
/// If `chain_on_max_tokens` is `true`, a new request is created when the last
/// one ends with a [`FinishReason::MaxTokens`] event, allowing the assistant to
/// continue from where it left off and the caller to receive the full response
/// as a single stream of events.
///
/// If `forced_tool_fallback` is `Some`, the model was asked to call a tool but
/// the API restriction on reasoning + forced tool use required us to downgrade
/// to `auto`.
/// If the response completes without any tool call, a follow-up request is
/// issued with thinking disabled and the original forced `tool_choice`
/// restored.
///
/// If the API rejects the request due to an invalid thinking-block signature,
/// emits [`Event::Patch`] instructions to fix the conversation stream and
/// finishes with [`FinishReason::Retry`] so the caller can rebuild and retry.
fn call(
    client: Client,
    request: types::CreateMessagesRequest,
    chain_on_max_tokens: bool,
    is_structured: bool,
    forced_tool_fallback: Option<ForcedToolFallback>,
) -> EventStream {
    Box::pin(try_stream!({
        // Buffer flushed ConversationEvents for chaining. When MaxTokens hits,
        // these are passed to chain() to build the continuation request. We use
        // EventBuilder to accumulate streaming parts into complete events with
        // merged metadata (needed for thinking signatures).
        //
        // The same buffer is used for the forced-tool fallback: if the model
        // finishes without calling a tool, we need the accumulated events to
        // build the assistant message for the retry request.
        let needs_event_buffer = chain_on_max_tokens || forced_tool_fallback.is_some();
        let mut chain_builder = EventBuilder::new();
        let mut chain_events = vec![];

        // If a tool call is requested, we cannot chain on max tokens, as the
        // Anthropic API expects the response to a tool call to contain the tool
        // result.
        let mut tool_calls_requested = false;

        // Track which tool names were called, so the forced-tool fallback can
        // check whether the *specific* required tool was invoked (not just any
        // tool).
        let mut tool_names_called: Vec<String> = vec![];

        let stream = client
            .messages()
            .create_stream(request.clone())
            .await
            .map_err(StreamError::from)
            .map_ok(|v| stream::iter(map_event(v, is_structured)))
            .try_flatten()
            .peekable();

        pin_mut!(stream);
        while let Some(result) = stream.next().await {
            // Anthropic rejects requests with stale thinking signatures as a
            // 400 `invalid_request_error`. Emit a patch to strip the offending
            // metadata and ask the caller to retry.
            if let Err(ref err) = result
                && is_invalid_thinking_signature(err)
                && let Some(patches) = build_thinking_patches(&request, err)
            {
                warn!("Invalid thinking signature, requesting retry.");
                yield Event::Patch(patches);
                yield Event::Finished(FinishReason::Retry);
                return;
            }

            match result? {
                // If the assistant has reached the maximum number of
                // tokens, and we are in a state in which we can request
                // more tokens, we do so by sending a new request and
                // chaining those events onto the previous ones, keeping the
                // existing stream of events alive.
                //
                // TODO: generalize this for any provider.
                event if should_chain(&event, tool_calls_requested, chain_on_max_tokens) => {
                    debug!("Max tokens reached, auto-requesting more tokens.");

                    chain_events.extend(chain_builder.drain());
                    for await event in
                        chain(client.clone(), request.clone(), chain_events, is_structured)
                    {
                        yield event?;
                    }
                    return;
                }

                // The model finished without calling the required tool, but the
                // caller originally requested a forced tool call that was
                // downgraded due to the reasoning + tool_choice API
                // restriction. Retry with thinking disabled and the real forced
                // tool_choice.
                Event::Finished(FinishReason::Completed)
                    if let Some(fallback) = forced_tool_fallback
                        .as_ref()
                        .filter(|fb| !fb.is_satisfied_by(&tool_names_called)) =>
                {
                    info!(
                        "Model did not call the required tool during soft-forced request. \
                         Retrying with thinking disabled and forced tool_choice."
                    );

                    chain_events.extend(chain_builder.drain());
                    for await event in force_tool_retry(
                        client.clone(),
                        request.clone(),
                        chain_events,
                        fallback,
                        is_structured,
                    ) {
                        yield event?;
                    }

                    return;
                }

                done @ Event::Finished(_) => {
                    yield done;
                    return;
                }
                Event::Part {
                    index,
                    part,
                    metadata,
                } => {
                    if let EventPart::ToolCall(ToolCallPart::Start { name, .. }) = &part {
                        tool_calls_requested = true;
                        tool_names_called.push(name.clone());
                    } else if part.is_tool_call() {
                        tool_calls_requested = true;
                    } else if needs_event_buffer {
                        chain_builder.handle_part(index, part.clone(), metadata.clone());
                    }

                    yield Event::Part {
                        index,
                        part,
                        metadata,
                    };
                }
                flush @ Event::Flush { .. } => {
                    let next_event = stream.as_mut().peek().await.and_then(|e| e.as_ref().ok());

                    // If we try to flush, but we're about to continue with a
                    // chained request, we ignore the flush event, to allow more
                    // event parts to be generated by the next response.
                    if let Some(event) = next_event
                        && should_chain(event, tool_calls_requested, chain_on_max_tokens)
                    {
                        continue;
                    }

                    // Flush the chain builder so accumulated parts become
                    // complete ConversationEvents with merged metadata.
                    if needs_event_buffer && let Event::Flush { index, metadata } = &flush {
                        chain_events.extend(chain_builder.handle_flush(*index, metadata.clone()));
                    }

                    yield flush;
                }
                patch @ Event::Patch(_) => yield patch,
                keep_alive @ Event::KeepAlive => yield keep_alive,
            }
        }
    }))
}

/// Check if we should chain more events from a new request.
fn should_chain(event: &Event, tool_calls_requested: bool, chain_on_max_tokens: bool) -> bool {
    !tool_calls_requested
        && chain_on_max_tokens
        && matches!(event, Event::Finished(FinishReason::MaxTokens))
}

/// Create a new `EventStream` by asking the assistant to continue from where it
/// left off.
fn chain(
    client: Client,
    mut request: types::CreateMessagesRequest,
    events: Vec<ConversationEvent>,
    is_structured: bool,
) -> EventStream {
    debug_assert!(!events.iter().any(ConversationEvent::is_tool_call_request));

    let mut should_merge = true;
    let previous_content = events
        .last()
        .and_then(|e| match e.as_chat_response() {
            Some(ChatResponse::Message { message }) => Some(message.as_str()),
            Some(ChatResponse::Reasoning { reasoning }) => Some(reasoning.as_str()),
            _ => None,
        })
        .unwrap_or_default()
        .to_owned();

    let message = events
        .into_iter()
        .filter_map(|event| convert_event(event, true).map(|v| v.1))
        .fold(
            types::Message {
                role: types::MessageRole::Assistant,
                ..Default::default()
            },
            |mut message, content| {
                message.content.push(content);
                message
            },
        );

    request.messages.push(message);
    request.messages.push(types::Message {
        role: types::MessageRole::User,
        content: types::MessageContentList(vec![types::MessageContent::Text(
            "Please continue from where you left off. DO NOT reason or mention being asked to \
             continue the conversation, just do it. YOU MUST repeat between 10 and 100 characters \
             from the end of the previous message, so that our system can find the correct merge \
             point between the two messages."
                .into(),
        )]),
    });

    Box::pin(try_stream!({
        for await event in call(client, request, true, is_structured, None) {
            let mut event = event?;

            // When chaining new events, the reasoning content is irrelevant, as
            // it will contain text such as "the user asked me to continue
            // [...]".
            //
            // NOTE: we should never end up in a situation in which we want to
            // continue an existing reasoning stint from the previous request,
            // as the model is configured to reason with less tokens than the
            // total maximum number of tokens allowed, so we do not need to
            // worry about omitting valid reasoning content here.
            if event.as_part().is_some_and(EventPart::is_reasoning) {
                continue;
            }

            // Merge the new content with the previous content, if there is any
            // overlap. Sometimes the assistant will start a chaining response
            // with a small amount of content that was already seen in the
            // previous response, and we want to avoid duplicating that.
            if let Some(content) = event.as_part_mut().and_then(EventPart::as_text_mut) {
                if should_merge {
                    let merge_point =
                        find_merge_point(&previous_content, content, 500, CHAIN_MIN_OVERLAP);
                    content.replace_range(..merge_point, "");
                }

                // After receiving the first content event, we can
                // stop merging.
                should_merge = false;
            }

            yield event;
        }

        return;
    }))
}

/// Retry with thinking disabled and forced `tool_choice` after the initial
/// soft-forced request (reasoning + `tool_choice: auto`) completed without
/// producing a tool call.
///
/// Appends the first response's content as an assistant message, adds a user
/// message requesting the tool call, then issues a new request with thinking
/// disabled and the original forced `tool_choice`.
fn force_tool_retry(
    client: Client,
    mut request: types::CreateMessagesRequest,
    events: Vec<ConversationEvent>,
    fallback: &ForcedToolFallback,
    is_structured: bool,
) -> EventStream {
    // Append the assistant's response from the first request.
    let message = events
        .into_iter()
        .filter_map(|event| convert_event(event, true).map(|v| v.1))
        .fold(
            types::Message {
                role: types::MessageRole::Assistant,
                ..Default::default()
            },
            |mut message, content| {
                message.content.push(content);
                message
            },
        );

    request.messages.push(message);

    // Ask the model to call the required tool.
    let prompt = match &fallback.tool_choice {
        types::ToolChoice::Tool { name, .. } => {
            format!(
                "You did not call the required tool. You MUST call the tool named '{name}' now. \
                 Do not respond with text."
            )
        }
        _ => "You did not call any tool. You MUST call at least one tool now. Do not respond with \
              text."
            .to_string(),
    };

    request.messages.push(types::Message {
        role: types::MessageRole::User,
        content: types::MessageContentList(vec![types::MessageContent::Text(prompt.into())]),
    });

    // Disable thinking and restore the original forced tool_choice.
    request.thinking = Some(types::ExtendedThinking::Disabled);
    request.tool_choice = Some(fallback.tool_choice.clone());

    // Remove effort from output config since thinking is disabled.
    if let Some(ref mut config) = request.output_config {
        config.effort = None;
        if config.format.is_none() {
            request.output_config = None;
        }
    }

    Box::pin(try_stream!({
        // Pass `None` for forced_tool_fallback to prevent infinite retries.
        for await event in call(client, request, false, is_structured, None) {
            let event = event?;

            // Skip reasoning (shouldn't appear with thinking disabled, but
            // be defensive).
            if event.as_part().is_some_and(EventPart::is_reasoning) {
                continue;
            }

            yield event;
        }

        return;
    }))
}

/// Returns `true` if the error is an Anthropic `invalid_request_error`
/// complaining about an invalid `signature` in a `thinking` block.
fn is_invalid_thinking_signature(error: &StreamError) -> bool {
    error.kind == StreamErrorKind::Other
        && error.message().contains("invalid_request_error")
        && error.message().contains("signature")
        && error.message().contains("thinking")
}

/// Extract the `(turn_index, flat_content_index)` from an Anthropic error
/// message like `"messages.1.content.0: Invalid 'signature' …"`.
///
/// These are **logical turn** indices — not our `request.messages[]` array
/// positions.
/// Use [`resolve_turn_position`] to map them to actual message and
/// content-block indices in the request.
fn parse_signature_error_position(msg: &str) -> Option<(usize, usize)> {
    let rest = msg.split("messages.").nth(1)?;
    let dot = rest.find('.')?;

    let turn_idx = rest[..dot].parse().ok()?;
    let rest = rest.get(dot + 1..)?.strip_prefix("content.")?;
    let end = rest
        .find(|c: char| !c.is_ascii_digit())
        .unwrap_or(rest.len());
    let flat_content_idx = rest[..end].parse().ok()?;

    Some((turn_idx, flat_content_idx))
}

/// Map an Anthropic logical-turn position to a `(msg_idx, content_idx)` pair in
/// our `request.messages[]` array.
///
/// The Anthropic API groups messages into logical turns: a user turn is a
/// single non-tool-result user message, and an assistant turn spans from the
/// first assistant message through all subsequent tool-result / assistant pairs
/// until the next real user message.
/// Content blocks are flat-indexed across all messages within a turn.
fn resolve_turn_position(
    messages: &[types::Message],
    turn_idx: usize,
    flat_content_idx: usize,
) -> Option<(usize, usize)> {
    let mut current_turn = 0;
    let mut i = 0;

    while i < messages.len() {
        let turn_start = i;

        // Advance past all messages belonging to this turn.
        i += 1;
        if messages[turn_start].role == types::MessageRole::Assistant {
            // Assistant turns include subsequent tool-result user messages and
            // assistant continuations until a non-tool-result user message
            // appears.
            while i < messages.len() {
                if messages[i].role == types::MessageRole::User
                    && !is_tool_result_message(&messages[i])
                {
                    break;
                }
                i += 1;
            }
        }

        if current_turn == turn_idx {
            let mut flat = 0;
            for (mi, msg) in messages.iter().enumerate().take(i).skip(turn_start) {
                for (ci, _) in msg.content.0.iter().enumerate() {
                    if flat == flat_content_idx {
                        return Some((mi, ci));
                    }
                    flat += 1;
                }
            }
            return None;
        }

        current_turn += 1;
    }

    None
}

/// Returns `true` if every content block in the message is a `ToolResult`.
fn is_tool_result_message(msg: &types::Message) -> bool {
    !msg.content.0.is_empty()
        && msg
            .content
            .0
            .iter()
            .all(|c| matches!(c, types::MessageContent::ToolResult(_)))
}

/// Build [`EventPatch`] instructions for the thinking block identified by the
/// API error.
/// The caller yields these as [`Event::Patch`] so the CLI can remove the stale
/// signature metadata from the persisted conversation stream.
/// On the next request rebuild, [`convert_event`] will fall through to the
/// `<think>` tag path for events that no longer carry a signature.
///
/// Returns `None` if the offending block cannot be located.
fn build_thinking_patches(
    request: &types::CreateMessagesRequest,
    error: &StreamError,
) -> Option<Vec<EventPatch>> {
    let api_position = parse_signature_error_position(error.message());
    let (msg_idx, content_idx) = api_position
        .and_then(|(turn, flat)| resolve_turn_position(&request.messages, turn, flat))
        .or_else(|| find_oldest_thinking_block(request))?;

    let (key, value) = identify_thinking_block(request, msg_idx, content_idx).or_else(|| {
        // Resolved position wasn't a thinking block. Fall back to oldest.
        let (msg, idx) = find_oldest_thinking_block(request)?;
        identify_thinking_block(request, msg, idx)
    })?;

    Some(vec![EventPatch {
        matcher: EventMatcher::MetadataValue {
            key: key.to_owned(),
            value,
        },
        action: PatchAction::RemoveMetadata(key.to_owned()),
    }])
}

/// Find the first (oldest) `Thinking` or `RedactedThinking` content block in
/// the request.
/// Used as a fallback when [`parse_signature_error_position`] cannot parse the
/// API error.
fn find_oldest_thinking_block(request: &types::CreateMessagesRequest) -> Option<(usize, usize)> {
    for (i, msg) in request.messages.iter().enumerate() {
        for (j, content) in msg.content.0.iter().enumerate() {
            if matches!(
                content,
                types::MessageContent::Thinking(_) | types::MessageContent::RedactedThinking { .. }
            ) {
                return Some((i, j));
            }
        }
    }

    None
}

/// Identify a thinking block at the given position and return its metadata key
/// and value, without mutating the request.
///
/// Returns `(metadata_key, metadata_value)` for the block, or `None` if the
/// target isn't a thinking block.
fn identify_thinking_block(
    request: &types::CreateMessagesRequest,
    msg_idx: usize,
    content_idx: usize,
) -> Option<(&'static str, String)> {
    let content = request.messages.get(msg_idx)?.content.0.get(content_idx)?;

    match content {
        types::MessageContent::Thinking(thinking) => Some((
            THINKING_SIGNATURE_KEY,
            thinking.signature.clone().unwrap_or_default(),
        )),
        types::MessageContent::RedactedThinking { data } => {
            Some((REDACTED_THINKING_KEY, data.clone()))
        }
        _ => None,
    }
}

#[derive(Debug, Clone, Default)]
struct BetaFeatures(Vec<String>);

impl BetaFeatures {
    /// See:
    /// <https://docs.claude.com/en/docs/build-with-claude/extended-thinking#interleaved-thinking>
    fn interleaved_thinking(&self) -> bool {
        self.0
            .iter()
            .any(|h| h == "interleaved-thinking-2025-05-14")
    }

    /// See: <https://docs.claude.com/en/docs/build-with-claude/context-editing>
    fn context_editing(&self) -> bool {
        self.0.iter().any(|h| h == "context-editing-2025-06-27")
    }

    /// See:
    /// <https://platform.claude.com/docs/en/build-with-claude/structured-outputs>
    fn structured_outputs(&self) -> bool {
        self.0.iter().any(|h| h == "structured-outputs-2025-10-27")
    }
}

#[expect(clippy::too_many_lines)]
fn create_request(
    model: &ModelDetails,
    query: ChatQuery,
    stream: bool,
    beta: &BetaFeatures,
) -> Result<(
    types::CreateMessagesRequest,
    bool,
    Option<ForcedToolFallback>,
)> {
    let ChatQuery {
        thread,
        tools,
        mut tool_choice,
    } = query;

    let mut builder = types::CreateMessagesRequestBuilder::default();

    builder.stream(stream);

    // Extract schema and config before into_parts() consumes the thread.
    let format = thread
        .events
        .schema()
        .map(|schema| JsonOutputFormat::JsonSchema {
            schema: transform_schema(schema),
        });

    let mut cache_budget = MAX_EXPLICIT_CACHE_CONTROL_COUNT;
    let config = thread.events.config()?;

    let thread_parts = thread.into_parts();

    // Build messages from conversation events, then prepend any attachments as
    // native content blocks in the first user message.
    let mut messages = convert_events(thread_parts.events);

    if !thread_parts.attachments.is_empty() {
        let mut doc_blocks: Vec<_> = thread_parts
            .attachments
            .into_iter()
            .filter_map(|attachment| match attachment.content {
                AttachmentContent::Text(text) => {
                    Some(types::MessageContent::Document(types::Document {
                        source: types::DocumentSource::Text {
                            media_type: types::PlainTextMediaType::default(),
                            data: text,
                        },
                        title: Some(attachment.source),
                        context: attachment.description,
                        citations: None,
                        cache_control: None,
                    }))
                }
                AttachmentContent::Binary { data, media_type } => {
                    let b64 = base64::engine::general_purpose::STANDARD.encode(&data);
                    let source = if media_type == "application/pdf" {
                        types::DocumentSource::Base64 {
                            data: b64,
                            media_type: types::PdfMediaType::default(),
                        }
                    } else if let Some(image_mt) = to_anthropic_image_media_type(&media_type) {
                        types::DocumentSource::Content {
                            content: types::DocumentSourceContent::Blocks(vec![
                                types::ContentBlockSourceContent::Image {
                                    source: types::ImageSource::Base64 {
                                        data: b64,
                                        media_type: image_mt,
                                    },
                                    cache_control: None,
                                },
                            ]),
                        }
                    } else {
                        warn!(
                            source = %attachment.source,
                            media_type,
                            "Unsupported binary attachment media type for Anthropic, skipping."
                        );

                        return None;
                    };

                    Some(types::MessageContent::Document(types::Document {
                        source,
                        title: Some(attachment.source),
                        context: attachment.description,
                        citations: None,
                        cache_control: None,
                    }))
                }
            })
            .collect();

        // Place a cache breakpoint on the last document block so the
        // entire prefix (system prompt + documents) is cached.
        if let Some(types::MessageContent::Document(doc)) = doc_blocks.last_mut()
            && cache_budget > 0
        {
            doc.cache_control = Some(types::CacheControl::default());
            cache_budget -= 1;
        }

        // Prepend document blocks to the first user message.
        if let Some(first_user) = messages
            .iter_mut()
            .find(|m| m.role == types::MessageRole::User)
        {
            let existing = mem::take(&mut first_user.content.0);
            first_user.content.0 = doc_blocks;
            first_user.content.0.extend(existing);
        }
    }

    // Check if we used an explicit breakpoint in the messages (via
    // CACHE_BREAKPOINT_KEY metadata).
    let has_message_breakpoint = messages.iter().any(|m| {
        m.content.0.iter().any(|c| match c {
            types::MessageContent::Text(t) => t.cache_control.is_some(),
            types::MessageContent::ToolUse(t) => t.cache_control.is_some(),
            types::MessageContent::ToolResult(t) => t.cache_control.is_some(),
            _ => false,
        })
    });

    if has_message_breakpoint {
        // If we explicitly placed a breakpoint in the messages, we consume 1
        // budget
        cache_budget = cache_budget.saturating_sub(1);
    } else {
        // Automatic caching handles the growing message history. The API places
        // a cache breakpoint on the last cacheable block (the latest message),
        // consuming 1 of the 4 available slots.
        builder.cache_control(types::CacheControl::default());
    }

    // The trailing assistant message, if any, is the continuation target.
    // Anthropic rejects native thinking blocks there once the turn has been
    // resumed, so rewrite them as `<think>` text before deciding how to
    // continue.
    downgrade_trailing_thinking(&mut messages);

    // Models that don't support assistant prefill require the conversation to
    // end with a user message. When the last message is from the assistant
    // (e.g. after an interrupt or retry that flushed partial content), inject a
    // synthetic "continue" message so the model picks up where it left off.
    if messages
        .last()
        .is_some_and(|m| m.role == types::MessageRole::Assistant)
        && !model.supports_prefill()
    {
        debug!("Injecting synthetic continue message (model does not support prefill).");
        messages.push(types::Message {
            role: types::MessageRole::User,
            content: types::MessageContentList(vec![types::MessageContent::Text(
                CONTINUE_MESSAGE.into(),
            )]),
        });
    }

    builder.model(model.id.name.clone()).messages(messages);

    // Explicit breakpoints are allocated to stable content, in request
    // structure order (system > documents > tools). System content gets
    // priority because it changes least frequently.
    //
    // Cache breakpoints (budget permitting):
    // - Last system Prompt block (captures system prompt + all sections)
    // - Last document block in first user message (attachments)
    // - Last tool definition (extends through tools)
    let last_prompt_idx = thread_parts.system_parts.len().checked_sub(1);

    let mut system_content: Vec<types::SystemContent> = thread_parts
        .system_parts
        .into_iter()
        .enumerate()
        .map(|(i, text)| {
            let cache_control = if Some(i) == last_prompt_idx && cache_budget > 0 {
                cache_budget -= 1;
                Some(types::CacheControl::default())
            } else {
                None
            };

            types::SystemContent::Text(types::Text {
                text,
                cache_control,
            })
        })
        .collect();

    let strict_tools = model.supports_structured_output() && beta.structured_outputs();
    let tools = convert_tools(tools, strict_tools, &mut cache_budget);

    // From testing, it seems that sending a single tool with the
    // "function" tool choice can result in incorrect API responses from
    // Anthropic. I (Jean) have an open support case with Anthropic to dig into
    // this finding more. UPDATE: This has been confirmed as a "quirk" of their
    // API, and the fix here is what they recommend as well.
    if tools.len() == 1 && matches!(tool_choice, ToolChoice::Function(_)) {
        tool_choice = ToolChoice::Required;
    }

    let parameters = &config.assistant.model.parameters;
    let max_tokens = parameters
        .max_tokens
        .or(model.max_output_tokens)
        .unwrap_or_else(|| {
            warn!(
                %model.id,
                %DEFAULT_MAX_TOKENS,
                "Model `max_tokens` parameter not found, using default value."
            );

            DEFAULT_MAX_TOKENS as u32
        });

    let reasoning_config = model.custom_reasoning_config(parameters.reasoning);

    // See: <https://docs.claude.com/en/docs/build-with-claude/extended-thinking#extended-thinking-with-tool-use>
    //
    // Anthropic does not support forced tool_choice with extended thinking.
    // We downgrade to auto + a system prompt nudge, and if the model still
    // doesn't call a tool, `call()` retries with thinking disabled and the
    // real forced tool_choice.
    let forced_tool_fallback = if reasoning_config.is_some() && tool_choice.is_forced_call() {
        info!(
            ?tool_choice,
            "Anthropic API does not support reasoning when tool_choice forces tool use. Switching \
             to soft-force mode with forced-tool fallback."
        );

        // Capture the original forced choice for the fallback before
        // downgrading to auto.
        let fallback = ForcedToolFallback {
            tool_choice: convert_tool_choice(tool_choice.clone()),
        };

        // Build the system prompt nudge.
        let text = if let Some(tool) = tool_choice.function_name() {
            format!(
                "IMPORTANT: You MUST call the tool named '{tool}'. DO NOT QUESTION THIS \
                 DIRECTIVE. DO NOT PROMPT FOR MORE CONTEXT OR DETAILS. JUST RUN IT."
            )
        } else {
            "IMPORTANT: You MUST use AT LEAST ONE tool available to you. DO NOT QUESTION THIS \
             DIRECTIVE. DO NOT PROMPT FOR MORE CONTEXT OR DETAILS. JUST RUN IT."
                .to_string()
        };

        tool_choice = ToolChoice::Auto;
        system_content.push(types::SystemContent::Text(types::Text {
            text,
            cache_control: None,
        }));

        Some(fallback)
    } else {
        None
    };

    let tool_choice = convert_tool_choice(tool_choice);

    if !tools.is_empty() {
        builder.tools(tools).tool_choice(tool_choice);
    }

    if !system_content.is_empty() {
        builder.system(System::Content(system_content));
    }

    // Track the effort from reasoning config so we can merge it with the
    // structured output format into a single OutputConfig at the end.
    let mut effort = None;
    let supports_thinking = model.reasoning.is_some_and(|r| !r.is_unsupported());

    if let Some(config) = reasoning_config {
        match model.reasoning {
            // Adaptive thinking for Opus 4.6+
            Some(ReasoningDetails::Adaptive {
                xhigh: supports_xhigh,
                max: supports_max,
            }) => {
                // Always request summarized thinking so reasoning is
                // captured in the conversation. The display layer handles
                // visibility.
                builder.thinking(types::ExtendedThinking::Adaptive {
                    display: Some(types::ThinkingDisplay::Summarized),
                });

                effort = match config
                    .effort
                    .abs_to_rel(model.max_output_tokens)
                    .unwrap_or(ReasoningEffort::Auto)
                {
                    ReasoningEffort::Max if supports_max => Some(Effort::Max),
                    ReasoningEffort::Max | ReasoningEffort::XHigh if supports_xhigh => {
                        Some(Effort::XHigh)
                    }
                    ReasoningEffort::Max
                    | ReasoningEffort::XHigh
                    | ReasoningEffort::High
                    | ReasoningEffort::Absolute(_) => Some(Effort::High),
                    ReasoningEffort::Medium => Some(Effort::Medium),
                    ReasoningEffort::Low | ReasoningEffort::Xlow | ReasoningEffort::None => {
                        Some(Effort::Low)
                    }
                    ReasoningEffort::Auto => None,
                };
            }

            // Budget-based thinking for older models
            Some(ReasoningDetails::Budgetted {
                min_tokens,
                max_tokens: reasoning_max_tokens,
            }) => {
                let mut max_budget = reasoning_max_tokens.unwrap_or(u32::MAX);

                // With interleaved thinking, the `budget_tokens` can exceed the
                // `max_tokens` parameter, as it represents the total budget across all
                // thinking blocks within one assistant turn.
                //
                // See: <https://docs.claude.com/en/docs/build-with-claude/extended-thinking#interleaved-thinking>
                //
                // This is only enabled if the model supports it, otherwise an error is
                // returned if the `max_tokens` parameter is larger than the model's
                // supported range.
                if beta.interleaved_thinking() && model.features.contains(&"interleaved-thinking") {
                    max_budget = model.context_window.unwrap_or(max_budget);
                }

                builder.thinking(types::ExtendedThinking::Enabled {
                    budget_tokens: config
                        .effort
                        .to_tokens(max_tokens)
                        .max(min_tokens)
                        .min(max_budget),
                    display: Some(types::ThinkingDisplay::Summarized),
                });
            }

            // Other reasoning details (Leveled, Unsupported) - no thinking config
            _ => {}
        }
    } else if supports_thinking {
        // Reasoning is off but the model supports it — explicitly disable
        // to prevent the model from thinking by default.
        builder.thinking(types::ExtendedThinking::Disabled);
    }

    let is_structured = format.is_some();
    if effort.is_some() || is_structured {
        builder.output_config(OutputConfig { effort, format });
    }

    if let Some(temperature) = parameters.temperature {
        builder.temperature(temperature);
    }

    #[expect(clippy::cast_possible_wrap)]
    builder.max_tokens(max_tokens as i32);

    if let Some(top_p) = parameters.top_p {
        builder.top_p(top_p);
    }

    if let Some(top_k) = parameters.top_k {
        builder.top_k(top_k);
    }

    // See: <https://docs.claude.com/en/docs/build-with-claude/context-editing>
    if beta.context_editing() {
        let strategy = match parameters.other.get("context_management").map(|v| &v.0) {
            Some(Value::Object(strategy)) => strategy.clone(),
            // If no strategy is provided, but the `context_editing` feature is
            // enabled, use the default strategy.
            _ => Map::from_iter([(
                "edits".into(),
                json!([
                  {
                    "type": "clear_tool_uses_20250919"
                  }
                ]),
            )]),
        };

        builder.context_management(strategy);
    }

    builder
        .build()
        .map(|req| (req, is_structured, forced_tool_fallback))
        .map_err(Into::into)
}

#[expect(clippy::too_many_lines)]
fn map_model(model: types::Model) -> Result<ModelDetails> {
    #[expect(clippy::match_same_arms)]
    let details = match model.id.as_str() {
        "claude-opus-4-8" => ModelDetails {
            id: (PROVIDER, model.id).try_into()?,
            display_name: Some(model.display_name),
            context_window: Some(1_000_000),
            max_output_tokens: Some(128_000),
            reasoning: Some(ReasoningDetails::adaptive(true, true)),
            knowledge_cutoff: Some(NaiveDate::from_ymd_opt(2026, 1, 1).unwrap()),
            deprecated: Some(ModelDeprecation::Active),
            structured_output: Some(true),
            features: vec![
                "interleaved-thinking",
                "context-editing",
                "adaptive-thinking",
            ],
        },
        "claude-opus-4-7" | "claude-opus-4-7-20260416" => ModelDetails {
            id: (PROVIDER, model.id).try_into()?,
            display_name: Some(model.display_name),
            context_window: Some(1_000_000),
            max_output_tokens: Some(128_000),
            reasoning: Some(ReasoningDetails::adaptive(true, true)),
            knowledge_cutoff: Some(NaiveDate::from_ymd_opt(2026, 1, 1).unwrap()),
            deprecated: Some(ModelDeprecation::Active),
            structured_output: Some(true),
            features: vec![
                "interleaved-thinking",
                "context-editing",
                "adaptive-thinking",
            ],
        },
        "claude-sonnet-4-6" => ModelDetails {
            id: (PROVIDER, model.id).try_into()?,
            display_name: Some(model.display_name),
            context_window: Some(1_000_000),
            max_output_tokens: Some(64_000),
            reasoning: Some(ReasoningDetails::adaptive(false, true)),
            knowledge_cutoff: Some(NaiveDate::from_ymd_opt(2025, 8, 1).unwrap()),
            deprecated: Some(ModelDeprecation::Active),
            structured_output: Some(true),
            features: vec![
                "interleaved-thinking",
                "context-editing",
                "adaptive-thinking",
            ],
        },
        "claude-opus-4-6" | "claude-opus-4-6-20260205" => ModelDetails {
            id: (PROVIDER, model.id).try_into()?,
            display_name: Some(model.display_name),
            context_window: Some(1_000_000),
            max_output_tokens: Some(128_000),
            reasoning: Some(ReasoningDetails::adaptive(false, true)),
            knowledge_cutoff: Some(NaiveDate::from_ymd_opt(2025, 5, 1).unwrap()),
            deprecated: Some(ModelDeprecation::Active),
            structured_output: Some(true),
            features: vec![
                "interleaved-thinking",
                "context-editing",
                "adaptive-thinking",
            ],
        },
        "claude-opus-4-5" | "claude-opus-4-5-20251101" => ModelDetails {
            id: (PROVIDER, model.id).try_into()?,
            display_name: Some(model.display_name),
            context_window: Some(200_000),
            max_output_tokens: Some(64_000),
            reasoning: Some(ReasoningDetails::budgetted(1024, None)),
            knowledge_cutoff: Some(NaiveDate::from_ymd_opt(2025, 8, 1).unwrap()),
            deprecated: Some(ModelDeprecation::Active),
            structured_output: Some(true),
            features: vec!["interleaved-thinking", "context-editing", "prefill"],
        },
        "claude-haiku-4-5" | "claude-haiku-4-5-20251001" => ModelDetails {
            id: (PROVIDER, model.id).try_into()?,
            display_name: Some(model.display_name),
            context_window: Some(200_000),
            max_output_tokens: Some(64_000),
            reasoning: Some(ReasoningDetails::budgetted(1024, None)),
            knowledge_cutoff: Some(NaiveDate::from_ymd_opt(2025, 7, 1).unwrap()),
            deprecated: Some(ModelDeprecation::Active),
            structured_output: Some(true),
            features: vec!["interleaved-thinking", "context-editing", "prefill"],
        },
        "claude-sonnet-4-5" | "claude-sonnet-4-5-20250929" => ModelDetails {
            id: (PROVIDER, model.id).try_into()?,
            display_name: Some(model.display_name),
            context_window: Some(1_000_000),
            max_output_tokens: Some(64_000),
            reasoning: Some(ReasoningDetails::budgetted(1024, None)),
            knowledge_cutoff: Some(NaiveDate::from_ymd_opt(2025, 7, 1).unwrap()),
            deprecated: Some(ModelDeprecation::Active),
            structured_output: Some(true),
            features: vec!["interleaved-thinking", "context-editing", "prefill"],
        },
        "claude-opus-4-1" | "claude-opus-4-1-20250805" => ModelDetails {
            id: (PROVIDER, model.id).try_into()?,
            display_name: Some(model.display_name),
            context_window: Some(200_000),
            max_output_tokens: Some(32_000),
            reasoning: Some(ReasoningDetails::budgetted(1024, None)),
            knowledge_cutoff: Some(NaiveDate::from_ymd_opt(2025, 3, 1).unwrap()),
            deprecated: Some(ModelDeprecation::Active),
            structured_output: Some(true),
            features: vec!["interleaved-thinking", "context-editing", "prefill"],
        },
        "claude-opus-4-0" | "claude-opus-4-20250514" => ModelDetails {
            id: (PROVIDER, model.id).try_into()?,
            display_name: Some(model.display_name),
            context_window: Some(200_000),
            max_output_tokens: Some(32_000),
            reasoning: Some(ReasoningDetails::budgetted(1024, None)),
            knowledge_cutoff: Some(NaiveDate::from_ymd_opt(2025, 3, 1).unwrap()),
            deprecated: Some(ModelDeprecation::Active),
            structured_output: Some(false),
            features: vec!["interleaved-thinking", "context-editing", "prefill"],
        },
        "claude-sonnet-4-0" | "claude-sonnet-4-20250514" => ModelDetails {
            id: (PROVIDER, model.id).try_into()?,
            display_name: Some(model.display_name),
            context_window: Some(1_000_000),
            max_output_tokens: Some(64_000),
            reasoning: Some(ReasoningDetails::budgetted(1024, None)),
            knowledge_cutoff: Some(NaiveDate::from_ymd_opt(2025, 3, 1).unwrap()),
            deprecated: Some(ModelDeprecation::Active),
            structured_output: Some(false),
            features: vec!["interleaved-thinking", "context-editing", "prefill"],
        },
        "claude-3-haiku-20240307" => ModelDetails {
            id: (PROVIDER, model.id).try_into()?,
            display_name: Some(model.display_name),
            context_window: Some(200_000),
            max_output_tokens: Some(4_096),
            reasoning: Some(ReasoningDetails::unsupported()),
            knowledge_cutoff: Some(NaiveDate::from_ymd_opt(2024, 8, 1).unwrap()),
            deprecated: Some(ModelDeprecation::deprecated(
                &"recommended replacement: claude-haiku-4-5-20251001",
                Some(NaiveDate::from_ymd_opt(2026, 4, 20).unwrap()),
            )),
            structured_output: Some(false),
            features: vec!["prefill"],
        },
        id => {
            debug!(model = id, ?model, "Missing model details.");
            let mut model = ModelDetails::empty((PROVIDER, id).try_into()?);
            model.display_name = Some(id.to_string());
            model
        }
    };

    Ok(details)
}

fn map_event(
    event: types::MessagesStreamEvent,
    is_structured: bool,
) -> Vec<std::result::Result<Event, StreamError>> {
    use types::MessagesStreamEvent::*;

    trace!(
        event = serde_json::to_string(&event).unwrap_or_default(),
        "Received event from Anthropic API."
    );

    match event {
        ContentBlockStart {
            content_block,
            index,
        } => map_content_start(content_block, index, is_structured)
            .into_iter()
            .map(Ok)
            .collect(),
        ContentBlockDelta { delta, index } => {
            vec![Ok(map_content_delta(delta, index, is_structured))]
        }
        ContentBlockStop { index } => vec![Ok(Event::flush(index))],
        MessageDelta { delta, .. } => map_message_delta(&delta).into_iter().map(Ok).collect(),
        // `message_stop` is Anthropic's terminal event. Emit `Finished` here so
        // the stream carries its own completion signal: a stream that ends
        // without it (a dropped connection mid-response) is treated as
        // incomplete by the consumer and retried, rather than being mistaken
        // for a clean completion.
        MessageStop => vec![Ok(Event::Finished(FinishReason::Completed))],
        MessageStart { .. } => vec![],
        // Keep-alive heartbeat: surface it so the idle-timeout layer treats the
        // connection as live, but it carries no content.
        Ping => vec![Ok(Event::KeepAlive)],
    }
}

impl TryFrom<&AnthropicConfig> for Anthropic {
    type Error = Error;

    fn try_from(config: &AnthropicConfig) -> Result<Self> {
        let api_key = env::var(&config.api_key_env)
            .map_err(|_| Error::MissingEnv(config.api_key_env.clone()))?;

        let mut builder = Client::builder();
        builder
            .api_key(api_key)
            .base_url(config.base_url.clone())
            .version("2023-06-01");

        if !config.beta_headers.is_empty() {
            builder.beta(config.beta_headers.join(","));
        }

        Ok(Anthropic {
            beta: BetaFeatures(config.beta_headers.clone()),
            chain_on_max_tokens: config.chain_on_max_tokens,
            client: builder
                .build()
                .map_err(|e| Error::Anthropic(AnthropicError::Unknown(e.to_string())))?,
        })
    }
}

/// Transform a JSON schema to conform to Anthropic's structured output
/// constraints.
///
/// Anthropic's structured output supports a subset of JSON Schema.
/// Unsupported properties are stripped and appended to the `description` field
/// so the model can still see them as soft hints.
///
/// Mirrors the logic from Anthropic's Python SDK `transform_schema`.
///
/// See:
/// <https://docs.claude.com/en/docs/build-with-claude/structured-outputs#json-schema-limitations>
fn transform_schema(mut src: Map<String, Value>) -> Map<String, Value> {
    if let Some(r) = src.remove("$ref") {
        return Map::from_iter([("$ref".into(), r)]);
    }

    let mut out = Map::new();

    // Helper macro to move a field from src to out.
    macro_rules! move_field {
        ($key:literal) => {
            if let Some(v) = src.remove($key) {
                out.insert($key.into(), v);
            }
        };
    }

    // Extract common fields
    move_field!("title");
    move_field!("description");

    // Recursive Transformation Helpers
    let transform_val = |v: Value| match v {
        Value::Object(o) => Value::Object(transform_schema(o)),
        other => other,
    };

    let transform_map = |m: Map<String, Value>| -> Map<String, Value> {
        m.into_iter().map(|(k, v)| (k, transform_val(v))).collect()
    };

    let transform_vec =
        |v: Vec<Value>| -> Vec<Value> { v.into_iter().map(transform_val).collect() };

    // Handle Recursive Dictionaries
    for key in ["$defs", "definitions"] {
        if let Some(Value::Object(defs)) = src.remove(key) {
            out.insert(key.into(), Value::Object(transform_map(defs)));
        }
    }

    // Handle Combinators
    if let Some(Value::Array(variants)) = src.remove("anyOf") {
        out.insert("anyOf".into(), Value::Array(transform_vec(variants)));
    } else if let Some(Value::Array(variants)) = src.remove("oneOf") {
        // Remap oneOf -> anyOf
        out.insert("anyOf".into(), Value::Array(transform_vec(variants)));
    } else if let Some(Value::Array(variants)) = src.remove("allOf") {
        out.insert("allOf".into(), Value::Array(transform_vec(variants)));
    }

    // Handle Type-Specific Logic
    //
    // We remove "type" now so it doesn't get caught in the "leftovers" logic
    // later.
    let type_val = src.remove("type");
    match type_val.as_ref().and_then(Value::as_str) {
        Some("object") => {
            if let Some(Value::Object(props)) = src.remove("properties") {
                out.insert("properties".into(), Value::Object(transform_map(props)));
            }

            move_field!("required");

            // Force strictness
            src.remove("additionalProperties");
            out.insert("additionalProperties".into(), Value::Bool(false));
        }
        Some("array") => {
            if let Some(items) = src.remove("items") {
                out.insert("items".into(), transform_val(items));
            }

            // Enforce minItems logic
            if let Some(min) = src.remove("minItems") {
                if min.as_u64().is_some_and(|n| n <= 1) {
                    out.insert("minItems".into(), min);
                } else {
                    // Put it back into src so it falls into description
                    src.insert("minItems".into(), min);
                }
            }
        }
        Some("string") => {
            if let Some(format) = src.remove("format") {
                let is_supported = format
                    .as_str()
                    .is_some_and(|f| SUPPORTED_STRING_FORMATS.contains(&f));

                if is_supported {
                    out.insert("format".into(), format);
                } else {
                    src.insert("format".into(), format);
                }
            }
        }
        _ => {}
    }

    // Re-insert type.
    if let Some(t) = type_val {
        out.insert("type".into(), t);
    }

    // 7. Handle "Leftovers" (Unsupported fields -> Description)
    if !src.is_empty() {
        let extra_info = src
            .iter()
            .map(|(k, v)| format!("{k}: {v}"))
            .collect::<Vec<_>>()
            .join(", ");

        out.entry("description")
            .and_modify(|v| {
                if let Some(s) = v.as_str() {
                    *v = Value::from(format!("{s}\n\n{{{extra_info}}}"));
                }
            })
            .or_insert_with(|| Value::from(format!("{{{extra_info}}}")));
    }

    out
}

/// Map a MIME type string to Anthropic's `ImageMediaType` enum.
fn to_anthropic_image_media_type(media_type: &str) -> Option<types::ImageMediaType> {
    match media_type {
        "image/jpeg" => Some(types::ImageMediaType::Jpeg),
        "image/png" => Some(types::ImageMediaType::Png),
        "image/gif" => Some(types::ImageMediaType::Gif),
        "image/webp" => Some(types::ImageMediaType::Webp),
        _ => None,
    }
}

fn convert_tool_choice(choice: ToolChoice) -> types::ToolChoice {
    match choice {
        ToolChoice::None => types::ToolChoice::none(),
        ToolChoice::Auto => types::ToolChoice::auto(),
        ToolChoice::Required => types::ToolChoice::any(),
        ToolChoice::Function(name) => types::ToolChoice::tool(name),
    }
}

fn convert_tools(
    tools: Vec<ToolDefinition>,
    strict: bool,
    cache_budget: &mut usize,
) -> Vec<types::Tool> {
    let mut tools: Vec<_> = tools
        .into_iter()
        .map(|tool| {
            types::Tool::Custom(types::CustomTool {
                name: tool.name,
                description: tool.docs.schema_description().map(str::to_owned),
                strict: strict.then_some(true),
                input_schema: {
                    let required = tool
                        .parameters
                        .iter()
                        .filter(|(_, cfg)| cfg.required)
                        .map(|(key, _)| key.clone())
                        .collect();

                    let properties = tool
                        .parameters
                        .into_iter()
                        .map(|(key, cfg)| (key, cfg.to_json_schema()))
                        .collect();

                    types::ToolInputSchema {
                        kind: types::ToolInputSchemaKind::Object,
                        properties,
                        required,
                        additional_properties: strict.then_some(false),
                    }
                },
                cache_control: None,
            })
        })
        .collect();

    // Cache tool definitions, as they are unlikely to change.
    if *cache_budget > 0
        && let Some(tool) = tools.last_mut()
    {
        let cache_control = match tool {
            types::Tool::Custom(tool) => &mut tool.cache_control,
            types::Tool::Bash(ToolBash::Bash20241022(tool)) => &mut tool.cache_control,
            types::Tool::Bash(ToolBash::Bash20250124(tool)) => &mut tool.cache_control,
            types::Tool::CodeExecution(ToolCodeExecution::CodeExecution20250522(tool)) => {
                &mut tool.cache_control
            }
            types::Tool::ComputerUse(ToolComputerUse::ComputerUse20241022(tool)) => {
                &mut tool.cache_control
            }
            types::Tool::ComputerUse(ToolComputerUse::ComputerUse20250124(tool)) => {
                &mut tool.cache_control
            }
            types::Tool::TextEditor(ToolTextEditor::TextEditor20241022(tool)) => {
                &mut tool.cache_control
            }
            types::Tool::TextEditor(ToolTextEditor::TextEditor20250124(tool)) => {
                &mut tool.cache_control
            }
            types::Tool::TextEditor(ToolTextEditor::TextEditor20250429(tool)) => {
                &mut tool.cache_control
            }
            types::Tool::WebSearch(ToolWebSearch::WebSearch20250305(tool)) => {
                &mut tool.cache_control
            }
        };

        *cache_control = Some(types::CacheControl::default());
        *cache_budget = cache_budget.saturating_sub(1);
    }

    tools
}

/// Groups consecutive events into messages by role.
///
/// Events from the same role are combined into a single message with multiple
/// content blocks.
fn convert_events(events: ConversationStream) -> Vec<types::Message> {
    events
        .into_iter()
        .filter_map(|event| {
            let aliases = &event.config.providers.llm.aliases;

            let is_anthropic = event
                .config
                .assistant
                .model
                .id
                .finalize(aliases)
                .is_ok_and(|id| id.provider == Some(PROVIDER));

            let has_cache_breakpoint = event
                .event
                .metadata
                .get(jp_conversation::event::CACHE_BREAKPOINT_KEY)
                .and_then(serde_json::Value::as_bool)
                .unwrap_or(false);

            convert_event(event.event, is_anthropic).map(|(role, mut content)| {
                if has_cache_breakpoint {
                    apply_cache_control(&mut content);
                }

                (role, content)
            })
        })
        .fold(vec![], |mut messages, (role, content)| {
            match messages.last_mut() {
                // If the last message has the same role, append content to it.
                Some(last) if last.role == role => last.content.0.push(content),
                // Different role or no messages yet, start a new message.
                _ => messages.push(types::Message {
                    role,
                    content: types::MessageContentList(vec![content]),
                }),
            }

            messages
        })
}

fn apply_cache_control(content: &mut types::MessageContent) {
    match content {
        types::MessageContent::Text(text) => {
            text.cache_control = Some(types::CacheControl::default());
        }
        types::MessageContent::ToolUse(tu) => {
            tu.cache_control = Some(types::CacheControl::default());
        }
        types::MessageContent::ToolResult(tr) => {
            tr.cache_control = Some(types::CacheControl::default());
        }
        _ => {}
    }
}

/// Rewrite thinking blocks in the trailing assistant message as `<think>` text.
///
/// When the conversation ends with an assistant message, that turn is being
/// continued rather than answered with a tool result.
/// Anthropic rejects native `thinking`/`redacted_thinking` blocks there: once a
/// turn has been resumed or replayed, the block sequence no longer matches a
/// single original generation and the API fails with `thinking blocks ...
/// cannot be modified`.
/// Rewriting readable thinking as ordinary `<think>` text preserves the
/// reasoning as context while sidestepping that validation; redacted
/// (encrypted) thinking has no readable form, so it is dropped.
///
/// Only the final assistant message is rewritten.
/// Thinking that precedes a tool result in an earlier turn is left native, as
/// the API requires for tool-use reasoning continuity.
fn downgrade_trailing_thinking(messages: &mut [types::Message]) {
    let Some(last) = messages.last_mut() else {
        return;
    };

    if last.role != types::MessageRole::Assistant {
        return;
    }

    let blocks = mem::take(&mut last.content.0);
    last.content.0 = blocks
        .into_iter()
        .filter_map(|content| match content {
            types::MessageContent::Thinking(types::Thinking { thinking, .. }) => {
                Some(think_tag_text(&thinking))
            }
            types::MessageContent::RedactedThinking { .. } => None,
            other => Some(other),
        })
        .collect();
}

/// Wrap reasoning text in `<think>` tags as a plain text content block.
///
/// Used wherever reasoning must be sent to Anthropic as ordinary text rather
/// than a native thinking block: reasoning from other providers replayed
/// through this provider, and signed thinking in a continuation target that
/// would otherwise be rejected.
fn think_tag_text(reasoning: &str) -> types::MessageContent {
    types::MessageContent::Text(format!("<think>\n{reasoning}\n</think>\n\n").into())
}

fn convert_event(
    event: ConversationEvent,
    is_anthropic: bool,
) -> Option<(types::MessageRole, types::MessageContent)> {
    let ConversationEvent { kind, metadata, .. } = event;

    match kind {
        EventKind::ChatRequest(request) if !request.content.is_empty() => Some((
            types::MessageRole::User,
            types::MessageContent::Text(request.content.into()),
        )),
        EventKind::ChatResponse(response) => {
            let content = if is_anthropic
                && response.is_reasoning()
                && let signature = metadata
                    .get(THINKING_SIGNATURE_KEY)
                    .and_then(Value::as_str)
                    .map(str::to_owned)
                && signature.is_some()
            {
                types::MessageContent::Thinking(Thinking {
                    thinking: match response {
                        ChatResponse::Reasoning { reasoning } => reasoning,
                        ChatResponse::Message { message } => message,
                        ChatResponse::Structured { data } => data.to_string(),
                    },
                    signature,
                })
            } else if is_anthropic
                && response.is_reasoning()
                && let Some(data) = metadata
                    .get(REDACTED_THINKING_KEY)
                    .and_then(Value::as_str)
                    .map(str::to_owned)
            {
                types::MessageContent::RedactedThinking { data }
            } else {
                match response {
                    // Reasoning from other providers - wrap in <think> tags.
                    ChatResponse::Reasoning { reasoning } => think_tag_text(&reasoning),
                    ChatResponse::Message { message } => {
                        types::MessageContent::Text(message.into())
                    }
                    ChatResponse::Structured { data } => {
                        types::MessageContent::Text(data.to_string().into())
                    }
                }
            };

            Some((types::MessageRole::Assistant, content))
        }
        EventKind::ToolCallRequest(request) => Some((
            types::MessageRole::Assistant,
            types::MessageContent::ToolUse(types::ToolUse {
                id: request.id,
                name: request.name,
                input: Value::Object(request.arguments),
                cache_control: None,
            }),
        )),
        EventKind::ToolCallResponse(response) => {
            let (content, is_error) = match response.result {
                Ok(c) => (Some(c), false),
                Err(e) => (Some(e), true),
            };

            Some((
                types::MessageRole::User,
                types::MessageContent::ToolResult(types::ToolResult {
                    tool_use_id: response.id,
                    content,
                    is_error,
                    cache_control: None,
                }),
            ))
        }
        // ChatRequest is handled as the role boundary (not a message).
        // Internal events are filtered by into_parts().
        _ => None,
    }
}

fn map_content_start(
    item: types::MessageContent,
    index: usize,
    is_structured: bool,
) -> Option<Event> {
    use types::MessageContent::*;

    let mut metadata = Map::new();
    match item {
        ToolUse(types::ToolUse { id, name, .. }) => Some(Event::tool_call_start(index, id, name)),
        Text(text) if is_structured => Some(Event::structured(index, text.text)),
        Text(text) if !text.text.is_empty() => Some(Event::message(index, text.text)),
        Document(_) | Text(_) => None,
        Thinking(types::Thinking {
            thinking,
            signature,
        }) => {
            if let Some(signature) = signature
                && !signature.is_empty()
            {
                metadata.insert(THINKING_SIGNATURE_KEY.to_owned(), signature.into());
            }

            Some(Event::Part {
                index,
                part: EventPart::Reasoning(thinking),
                metadata,
            })
        }
        RedactedThinking { data } => {
            metadata.insert(REDACTED_THINKING_KEY.to_owned(), data.into());
            Some(Event::Part {
                index,
                part: EventPart::Reasoning(String::new()),
                metadata,
            })
        }
        ToolResult(_) => unreachable!("never triggered by the API"),
    }
}

fn map_content_delta(delta: types::ContentBlockDelta, index: usize, is_structured: bool) -> Event {
    match delta {
        types::ContentBlockDelta::TextDelta { text } if is_structured => {
            Event::structured(index, text)
        }
        types::ContentBlockDelta::TextDelta { text } => Event::message(index, text),
        types::ContentBlockDelta::ThinkingDelta { thinking } => Event::reasoning(index, thinking),
        // This is only used for thinking blocks, and we need to store this
        // signature to pass it back to the assistant in the message
        // history.
        //
        // See: <https://docs.anthropic.com/en/docs/build-with-claude/streaming#thinking-delta>
        types::ContentBlockDelta::SignatureDelta { signature } => Event::Part {
            index,
            part: EventPart::Reasoning(String::new()),
            metadata: Map::from_iter([(THINKING_SIGNATURE_KEY.to_owned(), signature.into())]),
        },
        types::ContentBlockDelta::InputJsonDelta { partial_json } => {
            Event::tool_call_args(index, partial_json)
        }
    }
}

fn map_message_delta(delta: &types::MessageDelta) -> Option<Event> {
    match delta.stop_reason.as_deref()? {
        "max_tokens" => Some(Event::Finished(FinishReason::MaxTokens)),
        _ => None,
    }
}

impl From<AnthropicError> for StreamError {
    fn from(error: AnthropicError) -> Self {
        use AnthropicError as E;

        match error {
            E::Network(error) => Self::from(error),
            E::StreamTransport(_) => StreamError::transient(error.to_string()).with_source(error),
            E::RateLimit { retry_after } => {
                Self::rate_limit(retry_after.map(Duration::from_secs)).with_source(error)
            }

            // Anthropic's API is notoriously unreliable, so we special-case a
            // few common errors that most of the times resolve themselves when
            // retried.
            //
            // See: <https://docs.claude.com/en/docs/build-with-claude/streaming#error-events>
            // See: <https://docs.claude.com/en/api/errors#http-errors>
            E::Api(ref api_error)
                if RETRYABLE_ANTHROPIC_ERROR_TYPES.contains(&api_error.error_type.as_str()) =>
            {
                let retry_after = api_error
                    .message
                    .as_deref()
                    .and_then(extract_retry_from_text)
                    .unwrap_or(Duration::from_secs(3));

                StreamError::transient(error.to_string())
                    .with_retry_after(retry_after)
                    .with_source(error)
            }

            // Detect billing/quota errors before falling through to generic.
            E::Api(ref api_error)
                if looks_like_quota_error(&api_error.error_type)
                    || api_error
                        .message
                        .as_deref()
                        .is_some_and(looks_like_quota_error) =>
            {
                StreamError::new(
                    StreamErrorKind::InsufficientQuota,
                    format!(
                        "Insufficient API quota. Check your plan and billing details \
                         at https://console.anthropic.com/settings/billing. ({error})"
                    ),
                )
                .with_source(error)
            }

            error => StreamError::other(error.to_string()).with_source(error),
        }
    }
}

#[cfg(test)]
#[path = "anthropic_tests.rs"]
mod tests;
