adk-model 0.6.0

//! Anthropic client implementation.

use super::config::AnthropicConfig;
use super::convert;
use super::error::AnthropicApiError;
use super::rate_limit::RateLimitInfo;
use crate::retry::{RetryConfig, ServerRetryHint, execute_with_retry, is_retryable_model_error};
use adk_anthropic::{
    Anthropic, ContentBlock, ContentBlockDelta, ContentBlockDeltaEvent, MessageStreamEvent,
    StopReason, TextDelta,
};
use adk_core::{AdkError, ErrorCategory, ErrorComponent, FinishReason, Llm, LlmRequest, Part};
use async_stream::try_stream;
use async_trait::async_trait;
use futures::StreamExt;
use std::pin::pin;
use std::sync::Arc;
use tokio::sync::RwLock;
use tracing::field;
use tracing::{Span, debug};

/// Anthropic client for Claude models.
pub struct AnthropicClient {
    pub(super) client: Anthropic,
    pub(super) config: AnthropicConfig,
    pub(super) model: String,
    pub(super) max_tokens: u32,
    retry_config: RetryConfig,
    /// Latest rate-limit information from the most recent API response.
    latest_rate_limit: Arc<RwLock<RateLimitInfo>>,
}

impl AnthropicClient {
    /// Create a new Anthropic client.
    pub fn new(config: AnthropicConfig) -> Result<Self, AdkError> {
        let client = Anthropic::new(Some(config.api_key.clone()))
            .map_err(|e| AdkError::model(format!("Failed to create Anthropic client: {e}")))?;

        Ok(Self {
            client,
            model: config.model.clone(),
            max_tokens: config.max_tokens,
            config,
            retry_config: RetryConfig::default(),
            latest_rate_limit: Arc::new(RwLock::new(RateLimitInfo::default())),
        })
    }

    /// Create a client with just an API key (uses default model).
    pub fn from_api_key(api_key: impl Into<String>) -> Result<Self, AdkError> {
        Self::new(AnthropicConfig::new(api_key, "claude-sonnet-4-6"))
    }

    /// Access the underlying `adk_anthropic::Anthropic` HTTP client.
    ///
    /// Use this for direct API access to endpoints not covered by the `Llm` trait:
    /// batches, files, skills, models, token counting, and pricing.
    ///
    /// ```rust,ignore
    /// let inner = anthropic_client.inner();
    /// let models = inner.list_models(None).await?;
    /// let batch = inner.create_batch(requests).await?;
    /// ```
    pub fn inner(&self) -> &adk_anthropic::Anthropic {
        &self.client
    }

    /// Access the current Anthropic configuration.
    pub fn anthropic_config(&self) -> &AnthropicConfig {
        &self.config
    }

    #[must_use]
    pub fn with_retry_config(mut self, retry_config: RetryConfig) -> Self {
        self.retry_config = retry_config;
        self
    }

    pub fn set_retry_config(&mut self, retry_config: RetryConfig) {
        self.retry_config = retry_config;
    }

    pub fn retry_config(&self) -> &RetryConfig {
        &self.retry_config
    }

    /// Returns the latest rate-limit information from the most recent API response.
    ///
    /// Updated after each API call when the server provides rate-limit headers
    /// via `adk_anthropic::Error::RateLimit` or `adk_anthropic::Error::ServiceUnavailable`.
    /// Returns the default (all `None`) if no rate-limit info has been received.
    pub async fn latest_rate_limit_info(&self) -> RateLimitInfo {
        self.latest_rate_limit.read().await.clone()
    }

    pub(super) fn build_message_params(
        model: &str,
        max_tokens: u32,
        request: &LlmRequest,
        anthropic_config: &super::config::AnthropicConfig,
    ) -> Result<adk_anthropic::MessageCreateParams, AdkError> {
        let mut system_parts: Vec<String> = Vec::new();
        let mut messages = Vec::new();

        for content in &request.contents {
            if content.role == "system" {
                // Requirement 1.1: Extract system-role content text parts
                let text: String = content
                    .parts
                    .iter()
                    .filter_map(|p| match p {
                        Part::Text { text } => Some(text.clone()),
                        _ => None,
                    })
                    .collect::<Vec<_>>()
                    .join("\n");
                if !text.is_empty() {
                    system_parts.push(text);
                }
            } else {
                messages
                    .push(convert::content_to_message(content, anthropic_config.prompt_caching)?);
            }
        }

        // Requirement 1.2: Heuristic — re-route leading user-role text-only messages
        // to the system parameter when no explicit system-role content exists.
        // The agent layer injects instructions as role="user" before session history.
        // We detect consecutive user-only-text messages before the first assistant reply
        // and move them to the system parameter.
        if system_parts.is_empty() {
            let instruction_boundary = messages
                .iter()
                .position(|m| m.role == adk_anthropic::MessageRole::Assistant)
                .unwrap_or(0);

            if instruction_boundary > 0 {
                // Verify all leading messages are text-only user messages
                let all_text_only = messages[..instruction_boundary]
                    .iter()
                    .all(|m| m.role == adk_anthropic::MessageRole::User && is_text_only_message(m));

                if all_text_only {
                    let instruction_messages: Vec<_> =
                        messages.drain(..instruction_boundary).collect();
                    for msg in &instruction_messages {
                        if let Some(text) = extract_text_from_message(msg) {
                            if !text.is_empty() {
                                system_parts.push(text);
                            }
                        }
                    }
                }
            }
        }

        // Requirement 1.3: Concatenate multiple system entries with newline separators
        // Requirement 1.4: Omit system parameter when no system content found
        let system_prompt =
            if system_parts.is_empty() { None } else { Some(system_parts.join("\n")) };

        let mut tools = if request.tools.is_empty() {
            Vec::new()
        } else {
            // Requirement 19.3: When ToolSearchConfig is set, filter tools by regex pattern.
            // Requirement 19.4: When no ToolSearchConfig is set, load all tools.
            let filtered_tools = if let Some(ref tool_search) = anthropic_config.tool_search {
                request
                    .tools
                    .iter()
                    .filter(|(name, _)| tool_search.matches(name).unwrap_or(false))
                    .map(|(k, v)| (k.clone(), v.clone()))
                    .collect::<std::collections::HashMap<_, _>>()
            } else {
                request.tools.clone()
            };
            if filtered_tools.is_empty() {
                Vec::new()
            } else {
                convert::convert_tools(&filtered_tools)?
            }
        };

        // Read extensions["anthropic"]["built_in_tools"] → append to tools array
        let config = request.config.as_ref();
        let anthropic_ext =
            config.and_then(|c| c.extensions.get("anthropic")).and_then(|v| v.as_object());
        if let Some(built_in_tools) = anthropic_ext.and_then(|o| o.get("built_in_tools")) {
            if let Some(arr) = built_in_tools.as_array() {
                for (index, tool_value) in arr.iter().enumerate() {
                    let tool = serde_json::from_value::<adk_anthropic::ToolUnionParam>(
                        tool_value.clone(),
                    )
                    .map_err(|error| {
                        AdkError::new(
                            ErrorComponent::Model,
                            ErrorCategory::InvalidInput,
                            "model.anthropic.invalid_tool",
                            format!(
                                "failed to deserialize Anthropic built-in tool at index {index}: {error}"
                            ),
                        )
                        .with_provider("anthropic")
                    })?;
                    tools.push(tool);
                }
            }
        }

        // Requirement 7.3: Force temperature to 1.0 when thinking is enabled
        let temperature = if anthropic_config.thinking.is_some() {
            Some(1.0)
        } else {
            request.config.as_ref().and_then(|c| c.temperature)
        };
        let top_p = request.config.as_ref().and_then(|c| c.top_p);
        let top_k = request.config.as_ref().and_then(|c| c.top_k);
        let effective_max_tokens = request
            .config
            .as_ref()
            .and_then(|c| c.max_output_tokens)
            .map(|t| t as u32)
            .unwrap_or(max_tokens);

        // Merge consecutive messages with the same role.
        // This is critical for Anthropic parallel tool use — per the docs,
        // all tool results must be in a single user message. Without this,
        // Claude "learns to avoid parallel calls" from the conversation history.
        merge_consecutive_messages(&mut messages);

        Ok(convert::build_message_params(
            model,
            effective_max_tokens,
            messages,
            tools,
            system_prompt,
            temperature,
            top_p,
            top_k,
            anthropic_config.prompt_caching,
            anthropic_config.thinking.as_ref(),
            anthropic_config.effort,
            anthropic_config.fast_mode,
            anthropic_config.inference_geo.as_deref(),
            anthropic_config.service_tier.as_deref(),
            anthropic_config.context_management.as_ref(),
        ))
    }
}

/// Check if a `MessageParam` contains only text content (no tool use, tool results, images, etc.).
fn is_text_only_message(msg: &adk_anthropic::MessageParam) -> bool {
    match &msg.content {
        adk_anthropic::MessageParamContent::String(_) => true,
        adk_anthropic::MessageParamContent::Array(blocks) => {
            !blocks.is_empty() && blocks.iter().all(|block| matches!(block, ContentBlock::Text(_)))
        }
    }
}

/// Extract concatenated text from a `MessageParam`, returning `None` if empty.
fn extract_text_from_message(msg: &adk_anthropic::MessageParam) -> Option<String> {
    match &msg.content {
        adk_anthropic::MessageParamContent::String(s) => {
            if s.is_empty() {
                None
            } else {
                Some(s.clone())
            }
        }
        adk_anthropic::MessageParamContent::Array(blocks) => {
            let parts: Vec<&str> = blocks
                .iter()
                .filter_map(|block| match block {
                    ContentBlock::Text(tb) if !tb.text.is_empty() => Some(tb.text.as_str()),
                    _ => None,
                })
                .collect();
            if parts.is_empty() { None } else { Some(parts.join("\n")) }
        }
    }
}

/// Merge consecutive `MessageParam`s that share the same role into a single message.
///
/// This is required for Anthropic parallel tool use. Per the
/// [Anthropic docs](https://docs.anthropic.com/en/docs/build-with-claude/tool-use/parallel-tool-use),
/// all tool results must be in a single user message. Without merging, each tool
/// result becomes a separate user message, which "teaches Claude to avoid parallel calls."
///
/// Zero-cost when messages already alternate roles correctly.
fn merge_consecutive_messages(messages: &mut Vec<adk_anthropic::MessageParam>) {
    if messages.len() < 2 {
        return;
    }

    let mut merged = Vec::with_capacity(messages.len());
    let mut drain = messages.drain(..);

    if let Some(first) = drain.next() {
        merged.push(first);
    }

    for msg in drain {
        let last = merged.last_mut().unwrap();
        if last.role == msg.role {
            // Same role — merge content blocks into the existing message
            let blocks = match std::mem::replace(
                &mut last.content,
                adk_anthropic::MessageParamContent::Array(Vec::new()),
            ) {
                adk_anthropic::MessageParamContent::String(s) => {
                    if s.is_empty() {
                        Vec::new()
                    } else {
                        vec![ContentBlock::Text(adk_anthropic::TextBlock::new(s))]
                    }
                }
                adk_anthropic::MessageParamContent::Array(blocks) => blocks,
            };

            let new_blocks = match msg.content {
                adk_anthropic::MessageParamContent::String(s) => {
                    if s.is_empty() {
                        Vec::new()
                    } else {
                        vec![ContentBlock::Text(adk_anthropic::TextBlock::new(s))]
                    }
                }
                adk_anthropic::MessageParamContent::Array(blocks) => blocks,
            };

            let mut combined = blocks;
            combined.extend(new_blocks);
            last.content = adk_anthropic::MessageParamContent::Array(combined);
        } else {
            merged.push(msg);
        }
    }

    *messages = merged;
}

/// Convert an `adk_anthropic::Error` into an [`AnthropicApiError`], preserving
/// structured context (error type, message, status code, request ID).
///
/// The resulting `AnthropicApiError` is then converted to `AdkError` via its
/// `From` impl. The request ID, when present, is also recorded on the current
/// tracing span as `anthropic.request_id` (Requirement 4.2).
pub(super) fn convert_anthropic_error(e: adk_anthropic::Error) -> AdkError {
    let api_error = to_anthropic_api_error(&e);

    // Requirement 4.2: record request-id on the active tracing span when present
    if let Some(ref rid) = api_error.request_id {
        Span::current().record("anthropic.request_id", rid.as_str());
    }

    // Requirement 11.4: Record error type and message as a span event on failure
    tracing::error!(
        error.type_ = %api_error.error_type,
        error.message = %api_error.message,
        error.status_code = api_error.status_code,
        "anthropic api error"
    );

    api_error.into()
}

/// Build an [`AnthropicApiError`] from an `adk_anthropic::Error`, extracting the
/// error type, message, HTTP status code, and request ID from whichever
/// variant is present.
///
/// Requirements 4.1, 4.2, 4.4: Parse structured error body fields and
/// capture the request-id header value.
fn to_anthropic_api_error(e: &adk_anthropic::Error) -> AnthropicApiError {
    match e {
        adk_anthropic::Error::Api { status_code, error_type, message, request_id } => {
            AnthropicApiError {
                error_type: error_type.clone().unwrap_or_else(|| "api_error".to_string()),
                message: message.clone(),
                status_code: *status_code,
                request_id: request_id.clone(),
            }
        }
        adk_anthropic::Error::RateLimit { message, retry_after } => {
            let msg = match retry_after {
                Some(secs) => format!("{message} (retry-after: {secs}s)"),
                None => message.clone(),
            };
            AnthropicApiError {
                error_type: "rate_limit_error".to_string(),
                message: msg,
                status_code: 429,
                request_id: None,
            }
        }
        adk_anthropic::Error::ServiceUnavailable { message, retry_after } => {
            let msg = match retry_after {
                Some(secs) => format!("{message} (retry-after: {secs}s)"),
                None => message.clone(),
            };
            AnthropicApiError {
                error_type: "overloaded_error".to_string(),
                message: msg,
                status_code: 529,
                request_id: None,
            }
        }
        adk_anthropic::Error::Authentication { message } => AnthropicApiError {
            error_type: "authentication_error".to_string(),
            message: message.clone(),
            status_code: 401,
            request_id: None,
        },
        adk_anthropic::Error::Permission { message } => AnthropicApiError {
            error_type: "permission_error".to_string(),
            message: message.clone(),
            status_code: 403,
            request_id: None,
        },
        adk_anthropic::Error::NotFound { message, .. } => AnthropicApiError {
            error_type: "not_found_error".to_string(),
            message: message.clone(),
            status_code: 404,
            request_id: None,
        },
        adk_anthropic::Error::BadRequest { message, .. } => AnthropicApiError {
            error_type: "invalid_request_error".to_string(),
            message: message.clone(),
            status_code: 400,
            request_id: None,
        },
        adk_anthropic::Error::InternalServer { message, request_id } => AnthropicApiError {
            error_type: "api_error".to_string(),
            message: message.clone(),
            status_code: 500,
            request_id: request_id.clone(),
        },
        // All other adk_anthropic error variants (Connection, Timeout, Serialization, etc.)
        // are client-side errors without structured API error bodies.
        other => AnthropicApiError {
            error_type: "client_error".to_string(),
            message: format!("{other}"),
            status_code: 0,
            request_id: None,
        },
    }
}

/// Extract a [`ServerRetryHint`] from an `adk_anthropic::Error`, if the error
/// contains a server-provided `retry_after` value.
#[allow(dead_code)]
fn extract_retry_hint(e: &adk_anthropic::Error) -> Option<ServerRetryHint> {
    match e {
        adk_anthropic::Error::RateLimit { retry_after: Some(secs), .. }
        | adk_anthropic::Error::ServiceUnavailable { retry_after: Some(secs), .. } => {
            Some(ServerRetryHint { retry_after: Some(std::time::Duration::from_secs(*secs)) })
        }
        _ => None,
    }
}

#[async_trait]
impl Llm for AnthropicClient {
    fn name(&self) -> &str {
        &self.model
    }

    #[tracing::instrument(
        skip_all,
        fields(
            anthropic.model = %self.model,
            anthropic.request_type = if stream { "stream" } else { "unary" },
            anthropic.request_id = field::Empty,
        )
    )]
    async fn generate_content(
        &self,
        request: LlmRequest,
        stream: bool,
    ) -> Result<adk_core::LlmResponseStream, AdkError> {
        let usage_span = adk_telemetry::llm_generate_span("anthropic", &self.model, stream);
        let model = self.model.clone();
        let max_tokens = self.max_tokens;
        let client = self.client.clone();
        let retry_config = self.retry_config.clone();
        let request_for_retry = request.clone();
        let anthropic_config = self.config.clone();

        let response_stream = try_stream! {
            if stream {
                // Streaming mode
                let client_ref = &client;
                let model_ref = model.as_str();
                let event_stream = execute_with_retry(&retry_config, is_retryable_model_error, || {
                    let request = request_for_retry.clone();
                    let cfg = &anthropic_config;
                    async move {
                        let mut params = Self::build_message_params(model_ref, max_tokens, &request, cfg)?;
                        params.stream = true;
                        client_ref
                            .stream(&params)
                            .await
                            .map_err(convert_anthropic_error)
                    }
                })
                .await?;

                // Pin the stream for iteration
                let mut pinned_stream = pin!(event_stream);

                // Track tool calls being built
                let mut current_tool_calls: Vec<(String, String, String)> = Vec::new(); // (id, name, args_json)
                let mut current_tool_index: Option<usize> = None;
                let mut pending_server_parts: Vec<Part> = Vec::new();

                // Track usage from MessageStart for propagation to final MessageDelta
                let mut stream_input_tokens: i32 = 0;
                let mut stream_cache_read_tokens: Option<i32> = None;
                let mut stream_cache_creation_tokens: Option<i32> = None;

                while let Some(event_result) = pinned_stream.next().await {
                    // Requirement 3.4: Handle error events from the stream.
                    // The adk-anthropic SSE parser converts `event: error` into stream Err values
                    // with structured error info. We emit these as LlmResponse with error fields
                    // rather than propagating as AdkError.
                    let event = match event_result {
                        Ok(ev) => ev,
                        Err(ref e) => {
                            // Requirement 4.2: extract request-id from stream errors
                            let api_err = to_anthropic_api_error(e);
                            if let Some(ref rid) = api_err.request_id {
                                Span::current().record("anthropic.request_id", rid.as_str());
                            }
                            // Requirement 11.4: Record error details as a span event
                            tracing::error!(
                                error.type_ = %api_err.error_type,
                                error.message = %api_err.message,
                                error.status_code = api_err.status_code,
                                "anthropic stream error"
                            );
                            yield convert::from_stream_error(&api_err.error_type, &api_err.message);
                            continue;
                        }
                    };

                    match event {
                        MessageStreamEvent::ContentBlockStart(start_event) => {
                            // Check if this is a tool_use block
                            let index = start_event.index;
                            match start_event.content_block {
                                ContentBlock::ToolUse(tool_use) => {
                                    current_tool_index = Some(index);
                                    while current_tool_calls.len() <= index {
                                        current_tool_calls
                                            .push((String::new(), String::new(), String::new()));
                                    }
                                    current_tool_calls[index] = (
                                        tool_use.id.clone(),
                                        tool_use.name.clone(),
                                        String::new(),
                                    );
                                }
                                ContentBlock::ServerToolUse(server_tool_use) => {
                                    if let Ok(val) = serde_json::to_value(server_tool_use) {
                                        pending_server_parts
                                            .push(Part::ServerToolCall { server_tool_call: val });
                                    }
                                }
                                ContentBlock::WebSearchToolResult(web_search_result) => {
                                    if let Ok(val) = serde_json::to_value(web_search_result) {
                                        pending_server_parts.push(Part::ServerToolResponse {
                                            server_tool_response: val,
                                        });
                                    }
                                }
                                _ => {}
                            }
                        }
                        MessageStreamEvent::ContentBlockDelta(ContentBlockDeltaEvent { index, delta }) => {
                            match delta {
                                ContentBlockDelta::TextDelta(TextDelta { text }) => {
                                    if !text.is_empty() {
                                        yield convert::from_text_delta(&text);
                                    }
                                }
                                ContentBlockDelta::InputJsonDelta(json_delta) => {
                                    // Accumulate tool call arguments
                                    if let Some(idx) = current_tool_index {
                                        if idx < current_tool_calls.len() {
                                            current_tool_calls[idx].2.push_str(&json_delta.partial_json);
                                        }
                                    } else if index < current_tool_calls.len() {
                                        current_tool_calls[index].2.push_str(&json_delta.partial_json);
                                    }
                                }
                                // Requirement 3.1: Emit thinking deltas wrapped in <thinking> tags
                                ContentBlockDelta::ThinkingDelta(td) => {
                                    if !td.thinking.is_empty() {
                                        yield convert::from_thinking_delta(&td.thinking);
                                    }
                                }
                                // Requirement 3.2: Accumulate signature deltas silently
                                ContentBlockDelta::SignatureDelta(_) => {}
                                // Requirement 3.5: Log unrecognized deltas at debug level
                                ContentBlockDelta::CitationsDelta(cd) => {
                                    debug!(?cd, "citations delta received (not yet mapped)");
                                }
                            }
                        }
                        MessageStreamEvent::ContentBlockStop { .. } => {
                            current_tool_index = None;
                        }
                        MessageStreamEvent::MessageDelta(delta_event) => {
                            // Check for stop reason
                            if let Some(stop_reason) = &delta_event.delta.stop_reason {
                                let finish_reason = match stop_reason {
                                    StopReason::EndTurn => Some(FinishReason::Stop),
                                    StopReason::MaxTokens => Some(FinishReason::MaxTokens),
                                    StopReason::StopSequence => Some(FinishReason::Stop),
                                    StopReason::ToolUse => Some(FinishReason::Stop),
                                    StopReason::PauseTurn => Some(FinishReason::Stop),
                                    StopReason::Refusal => Some(FinishReason::Safety),
                                    StopReason::PauseRun => Some(FinishReason::Stop),
                                    StopReason::ModelContextWindowExceeded => Some(FinishReason::MaxTokens),
                                };

                                // If we have accumulated tool calls, emit them
                                let mut parts = std::mem::take(&mut pending_server_parts);
                                if !current_tool_calls.is_empty() {
                                    let tool_calls = current_tool_calls
                                        .drain(..)
                                        .filter(|(id, name, _)| !id.is_empty() && !name.is_empty())
                                        .map(|(id, name, args_str)| {
                                            let args: serde_json::Value = serde_json::from_str(&args_str)
                                                .unwrap_or(serde_json::json!({}));
                                            Part::FunctionCall {
                                                name,
                                                args,
                                                id: Some(id),
                                                thought_signature: None,
                                            }
                                        })
                                        .collect::<Vec<_>>();
                                    parts.extend(tool_calls);
                                }

                                if !parts.is_empty() {
                                    yield adk_core::LlmResponse {
                                        content: Some(adk_core::Content {
                                            role: "model".to_string(),
                                            parts,
                                        }),
                                        usage_metadata: Some(adk_core::UsageMetadata {
                                            prompt_token_count: stream_input_tokens,
                                            candidates_token_count: delta_event.usage.output_tokens,
                                            total_token_count: stream_input_tokens + delta_event.usage.output_tokens,
                                            cache_read_input_token_count: stream_cache_read_tokens,
                                            cache_creation_input_token_count: stream_cache_creation_tokens,
                                            ..Default::default()
                                        }),
                                        finish_reason,
                                        citation_metadata: None,
                                        partial: false,
                                        turn_complete: true,
                                        interrupted: false,
                                        error_code: None,
                                        error_message: None,
                                        provider_metadata: None,
                                    };
                                    continue;
                                }

                                // Emit final message
                                yield adk_core::LlmResponse {
                                    content: None,
                                    usage_metadata: Some(adk_core::UsageMetadata {
                                        prompt_token_count: stream_input_tokens,
                                        candidates_token_count: delta_event.usage.output_tokens,
                                        total_token_count: stream_input_tokens + delta_event.usage.output_tokens,
                                        cache_read_input_token_count: stream_cache_read_tokens,
                                        cache_creation_input_token_count: stream_cache_creation_tokens,
                                        ..Default::default()
                                    }),
                                    finish_reason,
                                    citation_metadata: None,
                                    partial: false,
                                    turn_complete: true,
                                    interrupted: false,
                                    error_code: None,
                                    error_message: None,
                                    provider_metadata: None,
                                };
                            }
                        }
                        MessageStreamEvent::MessageStop(_) => {
                            // Stream complete
                        }
                        // Requirement 3.3: Treat ping as keep-alive no-op
                        MessageStreamEvent::Ping => {}
                        // Requirement 3.5: Log unrecognized events at debug level
                        MessageStreamEvent::MessageStart(start_event) => {
                            debug!("message_start event received");
                            // Store input tokens for the final UsageMetadata
                            stream_input_tokens = start_event.message.usage.input_tokens;
                            // Store cache token counts for propagation to the final MessageDelta
                            stream_cache_read_tokens = start_event.message.usage.cache_read_input_tokens;
                            stream_cache_creation_tokens = start_event.message.usage.cache_creation_input_tokens;
                            // Requirement 6.3: Extract cache usage from the initial message usage
                            let cache_meta = convert::extract_cache_usage(&start_event.message.usage);
                            if !cache_meta.is_empty() {
                                debug!(
                                    cache_creation = ?start_event.message.usage.cache_creation_input_tokens,
                                    cache_read = ?start_event.message.usage.cache_read_input_tokens,
                                    "cache usage tokens received in stream"
                                );
                            }
                        }
                        // New adk-anthropic event variants — log at debug level for now
                        MessageStreamEvent::ToolInputStart { .. }
                        | MessageStreamEvent::ToolInputDelta { .. }
                        | MessageStreamEvent::CompactionEvent(_)
                        | MessageStreamEvent::StreamError { .. } => {
                            debug!("unhandled stream event variant received");
                        }
                    }
                }
            } else {
                // Non-streaming mode
                let client_ref = &client;
                let model_ref = model.as_str();
                let message = execute_with_retry(&retry_config, is_retryable_model_error, || {
                    let request = request_for_retry.clone();
                    let cfg = &anthropic_config;
                    async move {
                        let params = Self::build_message_params(model_ref, max_tokens, &request, cfg)?;
                        client_ref
                            .send(params)
                            .await
                            .map_err(convert_anthropic_error)
                    }
                })
                .await?;

                // Requirement 4.3: On success, propagate request-id to tracing span.
                // The adk-anthropic crate does not expose the raw `request-id` response
                // header on successful responses, but the message `id` field
                // (e.g. "msg_...") serves as the primary correlation identifier.
                // When adk-anthropic adds header access, this will be updated to use
                // the actual `request-id` header value.
                Span::current().record("anthropic.request_id", message.id.as_str());

                // Requirement 6.3: Extract cache usage tokens into provider metadata
                let (_response, _cache_metadata) = convert::from_anthropic_message(&message);

                yield convert::from_anthropic_message(&message).0;
            }
        };

        Ok(crate::usage_tracking::with_usage_tracking(Box::pin(response_stream), usage_span))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use adk_anthropic::SystemPrompt;
    use adk_core::{Content, GenerateContentConfig, LlmRequest, Part};

    fn make_request(contents: Vec<Content>) -> LlmRequest {
        LlmRequest {
            model: "claude-sonnet-4-5-20250929".to_string(),
            contents,
            tools: std::collections::HashMap::new(),
            config: None,
        }
    }

    /// Requirement 1.1: System-role content extracted to system parameter.
    #[test]
    fn test_system_role_extracted_to_system_param() {
        let request = make_request(vec![
            Content {
                role: "system".to_string(),
                parts: vec![Part::Text { text: "You are a helpful assistant.".to_string() }],
            },
            Content {
                role: "user".to_string(),
                parts: vec![Part::Text { text: "Hello".to_string() }],
            },
        ]);

        let params = AnthropicClient::build_message_params(
            "claude-sonnet-4-5-20250929",
            4096,
            &request,
            &AnthropicConfig::default(),
        )
        .unwrap();

        assert!(params.system.is_some());
        match &params.system.unwrap() {
            SystemPrompt::String(s) => assert_eq!(s, "You are a helpful assistant."),
            SystemPrompt::Blocks(blocks) => {
                let text: String =
                    blocks.iter().map(|b| b.block.text.as_str()).collect::<Vec<_>>().join("");
                assert_eq!(text, "You are a helpful assistant.");
            }
        }
        // The user message should remain in messages
        assert_eq!(params.messages.len(), 1);
    }

    /// Requirement 1.2: Leading user-role text-only messages re-routed to system
    /// when no explicit system content exists.
    #[test]
    fn test_instruction_rerouting_to_system() {
        let request = make_request(vec![
            Content {
                role: "user".to_string(),
                parts: vec![Part::Text { text: "You are a coding assistant.".to_string() }],
            },
            Content {
                role: "user".to_string(),
                parts: vec![Part::Text { text: "Always respond in Rust.".to_string() }],
            },
            Content {
                role: "model".to_string(),
                parts: vec![Part::Text { text: "Understood.".to_string() }],
            },
            Content {
                role: "user".to_string(),
                parts: vec![Part::Text { text: "Write a function.".to_string() }],
            },
        ]);

        let params = AnthropicClient::build_message_params(
            "claude-sonnet-4-5-20250929",
            4096,
            &request,
            &AnthropicConfig::default(),
        )
        .unwrap();

        // The two leading user messages should be in system
        assert!(params.system.is_some());
        match &params.system.unwrap() {
            SystemPrompt::String(s) => {
                assert!(s.contains("You are a coding assistant."));
                assert!(s.contains("Always respond in Rust."));
            }
            SystemPrompt::Blocks(blocks) => {
                let text: String =
                    blocks.iter().map(|b| b.block.text.as_str()).collect::<Vec<_>>().join("\n");
                assert!(text.contains("You are a coding assistant."));
                assert!(text.contains("Always respond in Rust."));
            }
        }
        // Messages should start with the assistant message
        assert_eq!(params.messages.len(), 2);
        assert_eq!(params.messages[0].role, adk_anthropic::MessageRole::Assistant);
    }

    /// Requirement 1.3: Multiple system entries concatenated with newline.
    #[test]
    fn test_multiple_system_entries_concatenated() {
        let request = make_request(vec![
            Content {
                role: "system".to_string(),
                parts: vec![Part::Text { text: "First system instruction.".to_string() }],
            },
            Content {
                role: "system".to_string(),
                parts: vec![Part::Text { text: "Second system instruction.".to_string() }],
            },
            Content {
                role: "user".to_string(),
                parts: vec![Part::Text { text: "Hello".to_string() }],
            },
        ]);

        let params = AnthropicClient::build_message_params(
            "claude-sonnet-4-5-20250929",
            4096,
            &request,
            &AnthropicConfig::default(),
        )
        .unwrap();

        assert!(params.system.is_some());
        match &params.system.unwrap() {
            SystemPrompt::String(s) => {
                assert_eq!(s, "First system instruction.\nSecond system instruction.");
            }
            SystemPrompt::Blocks(blocks) => {
                let text: String =
                    blocks.iter().map(|b| b.block.text.as_str()).collect::<Vec<_>>().join("");
                assert_eq!(text, "First system instruction.\nSecond system instruction.");
            }
        }
    }

    /// Requirement 1.4: No system content → system parameter omitted.
    #[test]
    fn test_no_system_content_omits_system_param() {
        // No system role, no assistant message → no instruction boundary → no system
        let request = make_request(vec![Content {
            role: "user".to_string(),
            parts: vec![Part::Text { text: "Hello".to_string() }],
        }]);

        let params = AnthropicClient::build_message_params(
            "claude-sonnet-4-5-20250929",
            4096,
            &request,
            &AnthropicConfig::default(),
        )
        .unwrap();

        assert!(params.system.is_none());
        assert_eq!(params.messages.len(), 1);
    }

    /// Heuristic should NOT re-route when explicit system content exists.
    #[test]
    fn test_heuristic_skipped_when_explicit_system_exists() {
        let request = make_request(vec![
            Content {
                role: "system".to_string(),
                parts: vec![Part::Text { text: "Explicit system.".to_string() }],
            },
            Content {
                role: "user".to_string(),
                parts: vec![Part::Text { text: "Instruction-like text.".to_string() }],
            },
            Content {
                role: "model".to_string(),
                parts: vec![Part::Text { text: "OK.".to_string() }],
            },
        ]);

        let params = AnthropicClient::build_message_params(
            "claude-sonnet-4-5-20250929",
            4096,
            &request,
            &AnthropicConfig::default(),
        )
        .unwrap();

        // System should only contain the explicit system content
        match &params.system.unwrap() {
            SystemPrompt::String(s) => assert_eq!(s, "Explicit system."),
            SystemPrompt::Blocks(blocks) => {
                let text: String =
                    blocks.iter().map(|b| b.block.text.as_str()).collect::<Vec<_>>().join("");
                assert_eq!(text, "Explicit system.");
            }
        }
        // The user message should remain in messages (not re-routed)
        assert_eq!(params.messages.len(), 2);
        assert_eq!(params.messages[0].role, adk_anthropic::MessageRole::User);
    }

    /// Heuristic should NOT re-route user messages containing non-text parts.
    #[test]
    fn test_heuristic_skips_non_text_user_messages() {
        let request = make_request(vec![
            Content {
                role: "user".to_string(),
                parts: vec![Part::FunctionResponse {
                    function_response: adk_core::FunctionResponseData::new(
                        "tool",
                        serde_json::json!({"result": "ok"}),
                    ),
                    id: Some("call_1".to_string()),
                }],
            },
            Content {
                role: "model".to_string(),
                parts: vec![Part::Text { text: "Got it.".to_string() }],
            },
        ]);

        let params = AnthropicClient::build_message_params(
            "claude-sonnet-4-5-20250929",
            4096,
            &request,
            &AnthropicConfig::default(),
        )
        .unwrap();

        // Should NOT re-route because the leading user message has non-text content
        assert!(params.system.is_none());
        assert_eq!(params.messages.len(), 2);
    }

    /// Empty contents produces no system and no messages.
    #[test]
    fn test_empty_contents() {
        let request = make_request(vec![]);
        let params = AnthropicClient::build_message_params(
            "claude-sonnet-4-5-20250929",
            4096,
            &request,
            &AnthropicConfig::default(),
        )
        .unwrap();

        assert!(params.system.is_none());
        assert!(params.messages.is_empty());
    }

    #[test]
    fn test_invalid_extension_builtin_tool_returns_error() {
        let mut request = make_request(vec![]);
        let mut extensions = serde_json::Map::new();
        extensions.insert(
            "anthropic".to_string(),
            serde_json::json!({
                "built_in_tools": [
                    {
                        "type": "web_fetch_20250910"
                    }
                ]
            }),
        );
        request.config = Some(GenerateContentConfig { extensions, ..Default::default() });

        let error = AnthropicClient::build_message_params(
            "claude-sonnet-4-5-20250929",
            4096,
            &request,
            &AnthropicConfig::default(),
        )
        .expect_err("invalid built-in tool should fail");

        assert_eq!(error.code, "model.anthropic.invalid_tool");
    }

    /// Requirement 3.3: Ping events are treated as keep-alive no-ops and don't produce
    /// LlmResponse emissions. Verifies that the Ping variant of MessageStreamEvent
    /// is handled silently without yielding any content.
    #[test]
    fn test_ping_event_produces_no_response() {
        // Construct a Ping event — the same type the streaming match arm receives
        let event = MessageStreamEvent::Ping;

        // Verify it matches the Ping variant (the match arm is `MessageStreamEvent::Ping => {}`)
        // This test documents and enforces that Ping is a recognized no-op event.
        let produces_response = !matches!(event, MessageStreamEvent::Ping);

        assert!(!produces_response, "Ping events must not produce any LlmResponse");
    }

    /// Requirement 3.2: Signature delta events are accumulated silently without
    /// emitting user-visible content. Verifies that SignatureDelta within a
    /// ContentBlockDelta event doesn't produce any LlmResponse.
    #[test]
    fn test_signature_delta_produces_no_visible_content() {
        use adk_anthropic::SignatureDelta;

        // Construct a signature delta event — the same type the streaming match arm receives
        let delta = ContentBlockDelta::SignatureDelta(SignatureDelta::new(
            "EqQBCgIYAhIM1gasdXOgGf4Bh".to_string(),
        ));

        // Verify the match arm behavior: SignatureDelta should not produce output
        let produces_response = match delta {
            ContentBlockDelta::SignatureDelta(_) => false,
            ContentBlockDelta::TextDelta(_) => true,
            ContentBlockDelta::ThinkingDelta(_) => true,
            ContentBlockDelta::InputJsonDelta(_) => true,
            ContentBlockDelta::CitationsDelta(_) => false, // also a no-op
        };

        assert!(!produces_response, "SignatureDelta events must not produce user-visible content");
    }

    /// Requirement 3.2: Multiple signature deltas should all be silent.
    /// Verifies that signature data of varying lengths is handled without output.
    #[test]
    fn test_multiple_signature_deltas_all_silent() {
        use adk_anthropic::SignatureDelta;

        let signatures = vec![
            "".to_string(),
            "abc".to_string(),
            "EqQBCgIYAhIM1gasdXOgGf4BhLqPEhIwxSigma".to_string(),
            "a".repeat(1000),
        ];

        for sig in signatures {
            let delta = ContentBlockDelta::SignatureDelta(SignatureDelta::new(sig.clone()));

            let produces_response = !matches!(delta, ContentBlockDelta::SignatureDelta(_));

            assert!(
                !produces_response,
                "SignatureDelta with signature '{sig}' must not produce output"
            );
        }
    }

    /// Requirement 3.3: Ping events interspersed with text deltas should not
    /// affect the text delta output. Only text deltas produce LlmResponse.
    #[test]
    fn test_ping_among_text_deltas_only_text_produces_output() {
        let events: Vec<MessageStreamEvent> = vec![
            MessageStreamEvent::Ping,
            MessageStreamEvent::ContentBlockDelta(ContentBlockDeltaEvent {
                index: 0,
                delta: ContentBlockDelta::TextDelta(TextDelta { text: "Hello".to_string() }),
            }),
            MessageStreamEvent::Ping,
            MessageStreamEvent::Ping,
            MessageStreamEvent::ContentBlockDelta(ContentBlockDeltaEvent {
                index: 0,
                delta: ContentBlockDelta::TextDelta(TextDelta { text: " world".to_string() }),
            }),
            MessageStreamEvent::Ping,
        ];

        let mut responses = Vec::new();
        for event in events {
            match event {
                MessageStreamEvent::Ping => {
                    // No-op, same as production code
                }
                MessageStreamEvent::ContentBlockDelta(ContentBlockDeltaEvent {
                    delta: ContentBlockDelta::TextDelta(TextDelta { ref text }),
                    ..
                }) => {
                    if !text.is_empty() {
                        responses.push(convert::from_text_delta(text));
                    }
                }
                _ => {}
            }
        }

        // Only the two text deltas should produce responses
        assert_eq!(responses.len(), 2);
        assert!(responses[0].partial);
        assert!(!responses[0].turn_complete);
        assert!(responses[1].partial);
        assert!(!responses[1].turn_complete);
    }

    /// Requirements 3.2, 3.3: Signature deltas and pings interspersed with text
    /// deltas should not affect output — only text deltas produce LlmResponse.
    #[test]
    fn test_signature_and_ping_among_text_deltas() {
        use adk_anthropic::SignatureDelta;

        let events: Vec<MessageStreamEvent> = vec![
            MessageStreamEvent::Ping,
            MessageStreamEvent::ContentBlockDelta(ContentBlockDeltaEvent {
                index: 0,
                delta: ContentBlockDelta::TextDelta(TextDelta {
                    text: "Response text".to_string(),
                }),
            }),
            MessageStreamEvent::ContentBlockDelta(ContentBlockDeltaEvent {
                index: 0,
                delta: ContentBlockDelta::SignatureDelta(SignatureDelta::new(
                    "sig_part_1".to_string(),
                )),
            }),
            MessageStreamEvent::Ping,
            MessageStreamEvent::ContentBlockDelta(ContentBlockDeltaEvent {
                index: 0,
                delta: ContentBlockDelta::SignatureDelta(SignatureDelta::new(
                    "sig_part_2".to_string(),
                )),
            }),
        ];

        let mut response_count = 0;
        for event in events {
            match event {
                MessageStreamEvent::Ping => {}
                MessageStreamEvent::ContentBlockDelta(ContentBlockDeltaEvent { delta, .. }) => {
                    match delta {
                        ContentBlockDelta::TextDelta(TextDelta { ref text })
                            if !text.is_empty() =>
                        {
                            response_count += 1;
                        }
                        ContentBlockDelta::SignatureDelta(_) => {}
                        _ => {}
                    }
                }
                _ => {}
            }
        }

        assert_eq!(
            response_count, 1,
            "Only the text delta should produce a response; ping and signature should be silent"
        );
    }

    // ---- Tests for structured error parsing (Requirement 4.1, 4.2, 4.4) ----

    /// Requirement 4.1, 4.4: Api variant preserves error_type, message, status_code, and request_id.
    #[test]
    fn test_to_anthropic_api_error_api_variant() {
        let err = adk_anthropic::Error::Api {
            status_code: 400,
            error_type: Some("invalid_request_error".to_string()),
            message: "Invalid model specified".to_string(),
            request_id: Some("req_abc123".to_string()),
        };
        let api_err = to_anthropic_api_error(&err);
        assert_eq!(api_err.error_type, "invalid_request_error");
        assert_eq!(api_err.message, "Invalid model specified");
        assert_eq!(api_err.status_code, 400);
        assert_eq!(api_err.request_id.as_deref(), Some("req_abc123"));
    }

    /// Requirement 4.1: Api variant with missing error_type defaults to "api_error".
    #[test]
    fn test_to_anthropic_api_error_api_variant_no_error_type() {
        let err = adk_anthropic::Error::Api {
            status_code: 500,
            error_type: None,
            message: "Internal error".to_string(),
            request_id: None,
        };
        let api_err = to_anthropic_api_error(&err);
        assert_eq!(api_err.error_type, "api_error");
        assert_eq!(api_err.status_code, 500);
        assert!(api_err.request_id.is_none());
    }

    /// Requirement 4.4: RateLimit variant maps to status 429 with retry-after in message.
    #[test]
    fn test_to_anthropic_api_error_rate_limit() {
        let err = adk_anthropic::Error::RateLimit {
            message: "Too many requests".to_string(),
            retry_after: Some(30),
        };
        let api_err = to_anthropic_api_error(&err);
        assert_eq!(api_err.error_type, "rate_limit_error");
        assert_eq!(api_err.status_code, 429);
        assert!(api_err.message.contains("retry-after: 30s"));
    }

    /// Requirement 4.4: ServiceUnavailable variant maps to status 529.
    #[test]
    fn test_to_anthropic_api_error_service_unavailable() {
        let err = adk_anthropic::Error::ServiceUnavailable {
            message: "Overloaded".to_string(),
            retry_after: None,
        };
        let api_err = to_anthropic_api_error(&err);
        assert_eq!(api_err.error_type, "overloaded_error");
        assert_eq!(api_err.status_code, 529);
        assert_eq!(api_err.message, "Overloaded");
    }

    /// Requirement 4.4: InternalServer variant preserves request_id.
    #[test]
    fn test_to_anthropic_api_error_internal_server() {
        let err = adk_anthropic::Error::InternalServer {
            message: "Server error".to_string(),
            request_id: Some("req_xyz789".to_string()),
        };
        let api_err = to_anthropic_api_error(&err);
        assert_eq!(api_err.error_type, "api_error");
        assert_eq!(api_err.status_code, 500);
        assert_eq!(api_err.request_id.as_deref(), Some("req_xyz789"));
    }

    /// Requirement 4.4: Authentication variant maps to status 401.
    #[test]
    fn test_to_anthropic_api_error_authentication() {
        let err = adk_anthropic::Error::Authentication { message: "Invalid API key".to_string() };
        let api_err = to_anthropic_api_error(&err);
        assert_eq!(api_err.error_type, "authentication_error");
        assert_eq!(api_err.status_code, 401);
    }

    /// Requirement 4.1: convert_anthropic_error produces AdkError::Model with structured info.
    #[test]
    fn test_convert_anthropic_error_preserves_structure() {
        let err = adk_anthropic::Error::Api {
            status_code: 429,
            error_type: Some("rate_limit_error".to_string()),
            message: "Rate limited".to_string(),
            request_id: Some("req_test".to_string()),
        };
        let adk_err = convert_anthropic_error(err);
        let msg = adk_err.to_string();
        assert!(msg.contains("429"), "Should contain status code");
        assert!(msg.contains("rate_limit_error"), "Should contain error type");
        assert!(msg.contains("Rate limited"), "Should contain message");
        assert!(msg.contains("req_test"), "Should contain request_id");
    }

    // ---- Property-based tests for system prompt routing ----

    use proptest::prelude::*;

    /// Extract the system prompt string from a SystemPrompt, regardless of variant.
    fn extract_system_text(sp: &SystemPrompt) -> String {
        match sp {
            SystemPrompt::String(s) => s.clone(),
            SystemPrompt::Blocks(blocks) => {
                blocks.iter().map(|b| b.block.text.as_str()).collect::<Vec<_>>().join("\n")
            }
        }
    }

    /// Generator for non-empty text strings suitable for system prompt content.
    fn arb_system_text() -> impl Strategy<Value = String> {
        "[A-Za-z0-9 .,!?]{1,80}".prop_map(String::from)
    }

    /// Generator for a Content with role "system" containing 1..3 text parts.
    fn arb_system_content() -> impl Strategy<Value = Content> {
        prop::collection::vec(arb_system_text(), 1..=3).prop_map(|texts| Content {
            role: "system".to_string(),
            parts: texts.into_iter().map(|t| Part::Text { text: t }).collect(),
        })
    }

    /// Generator for a Content with role "user" containing a single text part.
    fn arb_user_text_content() -> impl Strategy<Value = Content> {
        arb_system_text()
            .prop_map(|text| Content { role: "user".to_string(), parts: vec![Part::Text { text }] })
    }

    /// Generator for a Content with role "model" (assistant) containing a single text part.
    fn arb_assistant_content() -> impl Strategy<Value = Content> {
        arb_system_text().prop_map(|text| Content {
            role: "model".to_string(),
            parts: vec![Part::Text { text }],
        })
    }

    proptest! {
        #![proptest_config(ProptestConfig::with_cases(100))]

        /// **Feature: anthropic-deep-integration, Property 1: System prompt extraction preserves all system text**
        /// *For any* LlmRequest containing N >= 0 Content entries with role "system",
        /// each containing arbitrary text parts, the resulting Anthropic `system`
        /// parameter SHALL equal the newline-joined concatenation of all text parts
        /// from all system-role entries (or be None when N = 0).
        /// **Validates: Requirements 1.1, 1.3, 1.4**
        #[test]
        fn prop_system_prompt_extraction_preserves_all_text(
            system_contents in prop::collection::vec(arb_system_content(), 0..=4),
            trailing_user in arb_user_text_content(),
        ) {
            // Build the expected system text: join each Content's text parts with
            // newline, then join all Contents' texts with newline.
            let expected_parts: Vec<String> = system_contents
                .iter()
                .map(|c| {
                    c.parts
                        .iter()
                        .filter_map(|p| match p {
                            Part::Text { text } => Some(text.clone()),
                            _ => None,
                        })
                        .collect::<Vec<_>>()
                        .join("\n")
                })
                .filter(|s| !s.is_empty())
                .collect();

            let mut contents: Vec<Content> = system_contents;
            // Always add a trailing user message so the request is valid
            contents.push(trailing_user);

            let request = make_request(contents);
            let params = AnthropicClient::build_message_params(
                "claude-sonnet-4-5-20250929",
                4096,
                &request,
            &AnthropicConfig::default(),
            ).unwrap();

            if expected_parts.is_empty() {
                // Requirement 1.4: no system content → system parameter omitted
                // (heuristic may or may not fire depending on message structure,
                // but with no assistant message the heuristic boundary is 0)
                // We only assert None when there's truly no system text AND
                // the heuristic doesn't apply (no assistant message to form boundary).
                // Since we only have user messages and no assistant, boundary = 0,
                // so heuristic won't fire.
                prop_assert!(
                    params.system.is_none(),
                    "Expected no system param when no system-role content and no assistant boundary"
                );
            } else {
                // Requirements 1.1, 1.3: system text preserved and concatenated
                let expected = expected_parts.join("\n");
                prop_assert!(params.system.is_some(), "Expected system param to be present");
                let actual = extract_system_text(params.system.as_ref().unwrap());
                prop_assert_eq!(
                    actual,
                    expected,
                    "System prompt text mismatch"
                );
            }
        }

        /// **Feature: anthropic-deep-integration, Property 2: Instruction re-routing to system parameter**
        /// *For any* LlmRequest where the conversation starts with K >= 1 consecutive
        /// user-role text-only Content entries followed by an assistant-role entry,
        /// and no explicit system-role entries exist, the resulting Anthropic `system`
        /// parameter SHALL contain the text from those K leading user entries, and
        /// the `messages` array SHALL start with the assistant-role entry.
        /// **Validates: Requirements 1.2**
        #[test]
        fn prop_instruction_rerouting_to_system(
            leading_user in prop::collection::vec(arb_user_text_content(), 1..=4),
            assistant in arb_assistant_content(),
            trailing_user in arb_user_text_content(),
        ) {
            let k = leading_user.len();

            // Collect expected system text from the leading user messages
            let expected_system_parts: Vec<String> = leading_user
                .iter()
                .filter_map(|c| {
                    let text: String = c
                        .parts
                        .iter()
                        .filter_map(|p| match p {
                            Part::Text { text } => Some(text.clone()),
                            _ => None,
                        })
                        .collect::<Vec<_>>()
                        .join("\n");
                    if text.is_empty() { None } else { Some(text) }
                })
                .collect();

            // Build contents: leading user messages, then assistant, then trailing user
            let mut contents: Vec<Content> = leading_user;
            contents.push(assistant);
            contents.push(trailing_user);

            let request = make_request(contents);
            let params = AnthropicClient::build_message_params(
                "claude-sonnet-4-5-20250929",
                4096,
                &request,
            &AnthropicConfig::default(),
            ).unwrap();

            // The leading user messages should be re-routed to system
            prop_assert!(
                params.system.is_some(),
                "Expected system param from re-routed instructions"
            );
            let actual_system = extract_system_text(params.system.as_ref().unwrap());
            let expected_system = expected_system_parts.join("\n");
            prop_assert_eq!(
                actual_system,
                expected_system,
                "Re-routed system text mismatch"
            );

            // Messages should start with the assistant message (K user messages removed)
            // Total messages = original (K user + 1 assistant + 1 trailing user) - K re-routed = 2
            let _ = k; // used for documentation clarity
            prop_assert_eq!(
                params.messages.len(),
                2,
                "Expected 2 messages after re-routing leading user messages"
            );
            prop_assert_eq!(
                params.messages[0].role,
                adk_anthropic::MessageRole::Assistant,
                "First message should be assistant after re-routing"
            );
        }
    }
}