recursive/llm/

openai.rs

//! OpenAI-compatible chat-completions adapter.
//!
//! Targets the `/chat/completions` shape that OpenAI, Azure (via gateway),
//! GLM (Zhipu), DeepSeek, Moonshot, Together, Ollama and many others speak.
//! The only thing that varies is the base URL + model name + API key, which
//! is all driven by config.
//!
//! ## Deferred tool loading (software-layer ToolSearch)
//!
//! This provider implements the same deferred-tool pattern as `AnthropicProvider`
//! but in pure software — no API-level `defer_loading` or `tool_reference` needed.
//! On each `complete_with_search` / `stream_with_search` call:
//!
//! 1. Only eager tools + `ToolSearchTool` are sent in the initial request.
//! 2. If the model calls `ToolSearchTool`, the query is resolved against the
//!    deferred list, the matched schemas are returned as plain JSON in a
//!    `tool_result` message, and a new request is sent with the matched tools
//!    appended to the eager list.
//! 3. Capped at `max_search_rounds` to prevent infinite loops.

use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;

use async_trait::async_trait;
use futures_util::StreamExt;
use reqwest::Client;
use serde::{Deserialize, Serialize};
use serde_json::Value;

use super::search::{KeywordSearchEngine, SpecWithHint, ToolSearchEngine};
use super::StructuredRequest;
use super::{
    ChatProvider, Completion, RetryPolicy, StreamChunk, StreamSender, TokenUsage, ToolCall,
    ToolSpec,
};
use crate::error::{Error, Result};
use crate::message::{Message, Role};

const TOOL_SEARCH_TOOL_NAME: &str = "ToolSearchTool";

#[derive(Debug, Clone)]
pub struct OpenAiProvider {
    base_url: String,
    api_key: String,
    model: String,
    client: Client,
    temperature: f64,
    max_tokens: u32,
    retry: RetryPolicy,
    stream_tx: Option<StreamSender>,
    /// Algorithm used to resolve a `ToolSearchTool` query into a list of
    /// deferred tool names. Defaults to `KeywordSearchEngine`.
    search_engine: Arc<dyn ToolSearchEngine>,
    /// Maximum number of ToolSearchTool round-trips per
    /// `complete_with_search` / `stream_with_search` call.
    max_search_rounds: usize,
}

impl OpenAiProvider {
    pub fn new(
        base_url: impl Into<String>,
        api_key: impl Into<String>,
        model: impl Into<String>,
    ) -> crate::error::Result<Self> {
        let client = Client::builder()
            .timeout(Duration::from_secs(180))
            .build()
            .map_err(|e| crate::error::Error::Config {
                message: format!("failed to build HTTP client: {e}"),
            })?;
        Ok(Self {
            base_url: base_url.into().trim_end_matches('/').to_string(),
            api_key: api_key.into(),
            model: model.into(),
            client,
            temperature: 0.2,
            // Per-response output token cap. Seed from the crate default
            // (DEFAULT_MAX_TOKENS = 64K); callers override via
            // `with_max_tokens`, which `Config::from_env` drives from
            // `RECURSIVE_MAX_TOKENS` (env > provider ModelSpec.max_tokens
            // > agent.max_tokens > this default). The historical 16384
            // default + the comment below ("DeepSeek ceiling 8192 for v3")
            // were stale: DeepSeek V4 supports up to 384K output tokens
            // (flash == pro), and 16384 truncated reasoning-heavy turns
            // mid-generation (`provider_stop:length` with empty content).
            max_tokens: crate::llm::DEFAULT_MAX_TOKENS,
            retry: RetryPolicy::default(),
            stream_tx: None,
            search_engine: Arc::new(KeywordSearchEngine::new()),
            max_search_rounds: 3,
        })
    }

    /// Replace the search engine used to resolve `ToolSearchTool` queries.
    pub fn with_search_engine(mut self, engine: Arc<dyn ToolSearchEngine>) -> Self {
        self.search_engine = engine;
        self
    }

    /// Set the maximum number of ToolSearchTool round-trips per
    /// `complete_with_search` / `stream_with_search` call.
    pub fn with_max_search_rounds(mut self, n: usize) -> Self {
        self.max_search_rounds = n;
        self
    }

    /// Enable streaming by providing a channel sender for partial tokens.
    pub fn with_stream_tx(mut self, tx: StreamSender) -> Self {
        self.stream_tx = Some(tx);
        self
    }

    /// Build an `Error::Llm` with the model name prefixed.
    fn make_err(&self, ctx: impl Into<String>) -> Error {
        Error::Llm {
            provider: self.model.clone(),
            message: ctx.into(),
        }
    }

    pub fn with_temperature(mut self, t: f64) -> Self {
        self.temperature = t;
        self
    }

    pub fn with_max_tokens(mut self, n: u32) -> Self {
        self.max_tokens = n;
        self
    }

    pub fn with_retry_policy(mut self, policy: RetryPolicy) -> Self {
        self.retry = policy;
        self
    }

    /// POST `body` to `url` with retry, returning the raw response text on success.
    ///
    /// Handles: network errors, 5xx transient errors, and HTTP 200 with empty body
    /// (MiniMax transient failure). All three are retried with exponential back-off
    /// according to `self.retry`. Non-transient 4xx errors are returned immediately.
    async fn post_json_with_retry(&self, url: &str, body: &Value, label: &str) -> Result<String> {
        let mut attempt = 0;
        loop {
            tracing::debug!(target: "recursive::llm", request = %body, "POST {url} ({label})");
            let result = self
                .client
                .post(url)
                .bearer_auth(&self.api_key)
                .json(body)
                .send()
                .await;

            match result {
                Ok(resp) => {
                    let status = resp.status();
                    if status.is_success() {
                        let text = resp.text().await?;
                        if text.trim().is_empty() {
                            if let Some(backoff) = self.retry.backoff_for(attempt, None, true) {
                                tracing::warn!(
                                    target: "recursive::llm",
                                    attempt,
                                    backoff_ms = backoff.as_millis(),
                                    "HTTP 200 but empty body ({label}), retrying"
                                );
                                tokio::time::sleep(backoff).await;
                                attempt += 1;
                                continue;
                            }
                            return Err(self.make_err("HTTP 200 but response body is empty"));
                        }
                        return Ok(text);
                    }

                    let text = resp.text().await?;
                    tracing::debug!(target: "recursive::llm", body = %text, "error response ({label})");

                    if let Some(backoff) =
                        self.retry
                            .backoff_for(attempt, Some(status.as_u16()), false)
                    {
                        tracing::warn!(
                            target: "recursive::llm",
                            attempt,
                            backoff_ms = backoff.as_millis(),
                            status = status.as_u16(),
                            "transient HTTP error, retrying ({label})"
                        );
                        tokio::time::sleep(backoff).await;
                        attempt += 1;
                        continue;
                    }

                    return Err(self.make_err(format!("HTTP {status}: {text}")));
                }
                Err(e) => {
                    if let Some(backoff) = self.retry.backoff_for(attempt, None, true) {
                        tracing::warn!(
                            target: "recursive::llm",
                            attempt,
                            backoff_ms = backoff.as_millis(),
                            error = %e,
                            "network error, retrying ({label})"
                        );
                        tokio::time::sleep(backoff).await;
                        attempt += 1;
                        continue;
                    }
                    return Err(self.make_err(format!("request failed: {e}")));
                }
            }
        }
    }
}

#[async_trait]
impl ChatProvider for OpenAiProvider {
    #[tracing::instrument(skip(self, messages, tools), fields(
        provider = %self.base_url.split('/').next_back().unwrap_or("unknown"),
        model = %self.model
    ))]
    async fn complete(&self, messages: &[Message], tools: &[ToolSpec]) -> Result<Completion> {
        let body = build_request(
            &self.model,
            self.temperature,
            self.max_tokens,
            messages,
            tools,
        );
        let url = format!("{}/chat/completions", self.base_url);
        let text = self.post_json_with_retry(&url, &body, "complete").await?;
        let parsed: ChatResponse = serde_json::from_str(&text)
            .map_err(|e| self.make_err(format!("failed to parse response: {e}; body: {text}")))?;
        let choice = parsed
            .choices
            .into_iter()
            .next()
            .ok_or_else(|| self.make_err("response had no choices"))?;
        Ok(parse_completion(choice, parsed.usage))
    }

    async fn complete_structured(&self, req: StructuredRequest) -> Result<Value> {
        let mut body = build_request(
            &self.model,
            self.temperature,
            self.max_tokens,
            &req.messages,
            &[],
        );
        body["response_format"] = serde_json::json!({
            "type": "json_schema",
            "json_schema": {
                "name": req.schema_name,
                "strict": true,
                "schema": req.schema,
            }
        });
        let url = format!("{}/chat/completions", self.base_url);
        let text = self.post_json_with_retry(&url, &body, "structured").await?;
        let parsed: ChatResponse = serde_json::from_str(&text)
            .map_err(|e| self.make_err(format!("failed to parse response: {e}; body: {text}")))?;
        let choice = parsed
            .choices
            .into_iter()
            .next()
            .ok_or_else(|| self.make_err("response had no choices"))?;
        let completion = parse_completion(choice, parsed.usage);
        if completion.content.trim().is_empty() {
            return Err(self.make_err("structured response had empty content"));
        }
        serde_json::from_str(&completion.content).map_err(|e| {
            self.make_err(format!(
                "failed to parse structured response as JSON: {e}; content: {}",
                completion.content
            ))
        })
    }

    async fn complete_with_search(
        &self,
        messages: &[Message],
        eager_tools: &[SpecWithHint],
        deferred_tools: &[SpecWithHint],
    ) -> Result<Completion> {
        if deferred_tools.is_empty() {
            let specs: Vec<ToolSpec> = eager_tools.iter().map(|(s, _)| s.clone()).collect();
            return self.complete(messages, &specs).await;
        }
        self.run_search_loop(messages, eager_tools, deferred_tools, &[], 0)
            .await
    }

    async fn stream_with_search(
        &self,
        messages: &[Message],
        eager_tools: &[SpecWithHint],
        deferred_tools: &[SpecWithHint],
        stream_tx: Option<StreamSender>,
        cancel_token: Option<tokio_util::sync::CancellationToken>,
    ) -> Result<Completion> {
        if deferred_tools.is_empty() {
            let specs: Vec<ToolSpec> = eager_tools.iter().map(|(s, _)| s.clone()).collect();
            let tx = stream_tx.or_else(|| self.stream_tx.clone());
            return self.stream_inner(messages, &specs, tx, cancel_token).await;
        }
        // run_stream_search_loop 内部递归调 stream()，cancel_token 自然向下传
        self.run_stream_search_loop(
            messages,
            eager_tools,
            deferred_tools,
            &[],
            0,
            stream_tx,
            cancel_token,
        )
        .await
    }

    async fn stream(
        &self,
        messages: &[Message],
        tools: &[ToolSpec],
        stream_tx: Option<StreamSender>,
        cancel_token: Option<tokio_util::sync::CancellationToken>,
    ) -> Result<Completion> {
        // If no stream_tx provided, fall back to the instance-level one
        let tx = stream_tx.or_else(|| self.stream_tx.clone());
        self.stream_inner(messages, tools, tx, cancel_token).await
    }
}

impl OpenAiProvider {
    fn tool_search_spec() -> SpecWithHint {
        let spec = ToolSpec {
            name: TOOL_SEARCH_TOOL_NAME.to_string(),
            description: "Fetches full schema definitions for deferred tools so they can be \
                called. Until fetched, only the name is known — there is no parameter schema, so \
                the tool cannot be invoked. Use `select:<tool_name>` for direct selection, or \
                keywords to search."
                .to_string(),
            parameters: serde_json::json!({
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Query to find deferred tools."
                    },
                    "max_results": {
                        "type": "integer",
                        "description": "Maximum number of results to return (default: 5)",
                        "default": 5
                    }
                },
                "required": ["query"]
            }),
        };
        (spec, Some("find search discover tools lookup".to_string()))
    }

    /// Software-layer ToolSearch loop for `complete_with_search`.
    ///
    /// Sends only `eager_tools` + `ToolSearchTool` in round 0. If the model
    /// calls `ToolSearchTool`, resolves the query, appends the matched schemas
    /// as a plain JSON `tool_result`, and recurses with the matched tools
    /// added to `loaded_tools` (which are appended to eager for the next call).
    async fn run_search_loop(
        &self,
        messages: &[Message],
        eager_tools: &[SpecWithHint],
        deferred_tools: &[SpecWithHint],
        loaded_tools: &[ToolSpec],
        round: usize,
    ) -> Result<Completion> {
        // Build tool list: ToolSearchTool + caller eager + already-loaded deferred
        let mut call_tools: Vec<ToolSpec> = Vec::new();
        call_tools.push(Self::tool_search_spec().0);
        call_tools.extend(eager_tools.iter().map(|(s, _)| s.clone()));
        call_tools.extend_from_slice(loaded_tools);

        let completion = self.complete(messages, &call_tools).await?;

        let search_call = completion
            .tool_calls
            .iter()
            .find(|c| c.name == TOOL_SEARCH_TOOL_NAME);

        let search_call = match search_call {
            None => return Ok(completion),
            Some(c) => c.clone(),
        };

        if round >= self.max_search_rounds {
            tracing::warn!(
                target: "recursive::llm",
                round,
                max = self.max_search_rounds,
                "ToolSearchTool (openai): hit max_search_rounds, returning current completion"
            );
            return Ok(completion);
        }

        let query = search_call
            .arguments
            .get("query")
            .and_then(|v| v.as_str())
            .unwrap_or("");
        let names = self.search_engine.resolve(query, deferred_tools);
        let matched: Vec<ToolSpec> = deferred_tools
            .iter()
            .filter(|(s, _)| names.contains(&s.name))
            .map(|(s, _)| s.clone())
            .collect();
        let result_json = serde_json::to_string(&matched).unwrap_or_else(|_| "[]".into());

        // Append the ToolSearch exchange to the transcript
        let mut next_messages: Vec<Message> = messages.to_vec();
        next_messages.push(Message {
            role: Role::Assistant,
            content: completion.content.clone(),
            tool_calls: completion.tool_calls.clone(),
            tool_call_id: None,
            reasoning_content: completion.reasoning_content.clone(),
            is_compaction_summary: false,
        });
        next_messages.push(Message {
            role: Role::Tool,
            content: result_json,
            tool_calls: vec![],
            tool_call_id: Some(search_call.id.clone()),
            reasoning_content: None,
            is_compaction_summary: false,
        });

        let mut next_loaded = loaded_tools.to_vec();
        next_loaded.extend_from_slice(&matched);

        Box::pin(self.run_search_loop(
            &next_messages,
            eager_tools,
            deferred_tools,
            &next_loaded,
            round + 1,
        ))
        .await
    }

    /// Software-layer ToolSearch loop for `stream_with_search`.
    #[allow(clippy::too_many_arguments)]
    async fn run_stream_search_loop(
        &self,
        messages: &[Message],
        eager_tools: &[SpecWithHint],
        deferred_tools: &[SpecWithHint],
        loaded_tools: &[ToolSpec],
        round: usize,
        stream_tx: Option<StreamSender>,
        cancel_token: Option<tokio_util::sync::CancellationToken>,
    ) -> Result<Completion> {
        let mut call_tools: Vec<ToolSpec> = Vec::new();
        call_tools.push(Self::tool_search_spec().0);
        call_tools.extend(eager_tools.iter().map(|(s, _)| s.clone()));
        call_tools.extend_from_slice(loaded_tools);

        let tx = stream_tx.clone().or_else(|| self.stream_tx.clone());
        let completion = self
            .stream_inner(messages, &call_tools, tx, cancel_token.clone())
            .await?;

        let search_call = completion
            .tool_calls
            .iter()
            .find(|c| c.name == TOOL_SEARCH_TOOL_NAME);

        let search_call = match search_call {
            None => return Ok(completion),
            Some(c) => c.clone(),
        };

        if round >= self.max_search_rounds {
            tracing::warn!(
                target: "recursive::llm",
                round,
                max = self.max_search_rounds,
                "ToolSearchTool (openai/stream): hit max_search_rounds"
            );
            return Ok(completion);
        }

        let query = search_call
            .arguments
            .get("query")
            .and_then(|v| v.as_str())
            .unwrap_or("");
        let names = self.search_engine.resolve(query, deferred_tools);
        let matched: Vec<ToolSpec> = deferred_tools
            .iter()
            .filter(|(s, _)| names.contains(&s.name))
            .map(|(s, _)| s.clone())
            .collect();
        let result_json = serde_json::to_string(&matched).unwrap_or_else(|_| "[]".into());

        let mut next_messages: Vec<Message> = messages.to_vec();
        next_messages.push(Message {
            role: Role::Assistant,
            content: completion.content.clone(),
            tool_calls: completion.tool_calls.clone(),
            tool_call_id: None,
            reasoning_content: completion.reasoning_content.clone(),
            is_compaction_summary: false,
        });
        next_messages.push(Message {
            role: Role::Tool,
            content: result_json,
            tool_calls: vec![],
            tool_call_id: Some(search_call.id.clone()),
            reasoning_content: None,
            is_compaction_summary: false,
        });

        let mut next_loaded = loaded_tools.to_vec();
        next_loaded.extend_from_slice(&matched);

        Box::pin(self.run_stream_search_loop(
            &next_messages,
            eager_tools,
            deferred_tools,
            &next_loaded,
            round + 1,
            stream_tx,
            cancel_token,
        ))
        .await
    }

    async fn stream_inner(
        &self,
        messages: &[Message],
        tools: &[ToolSpec],
        stream_tx: Option<StreamSender>,
        cancel_token: Option<tokio_util::sync::CancellationToken>,
    ) -> Result<Completion> {
        let mut body = build_request(
            &self.model,
            self.temperature,
            self.max_tokens,
            messages,
            tools,
        );
        body["stream"] = Value::Bool(true);
        // Ask providers that gate usage emission behind a flag (MiniMax,
        // OpenAI native) to include the final usage chunk. Providers that
        // always emit usage (DeepSeek) ignore the field harmlessly. Without
        // this the streaming path never sees `usage.prompt_tokens_details`
        // and the TUI cache-rate badge stays invisible on MiniMax-M3.
        body["stream_options"] = serde_json::json!({ "include_usage": true });
        let url = format!("{}/chat/completions", self.base_url);

        // Stream requests need the raw Response object for SSE parsing, so we
        // can't use post_json_with_retry (which returns text). Retry only on
        // non-2xx and network errors; a successful 2xx hands off to parse_sse_stream.
        let mut attempt = 0;
        loop {
            tracing::debug!(target: "recursive::llm", request = %body, "POST {url} (stream)");
            let result = self
                .client
                .post(&url)
                .bearer_auth(&self.api_key)
                .json(&body)
                .send()
                .await;

            match result {
                Ok(resp) => {
                    let status = resp.status();
                    if status.is_success() {
                        return self
                            .parse_sse_stream(resp, stream_tx.clone(), cancel_token.clone())
                            .await;
                    }
                    let text = resp.text().await?;
                    tracing::debug!(target: "recursive::llm", body = %text, "error response (stream)");
                    if let Some(backoff) =
                        self.retry
                            .backoff_for(attempt, Some(status.as_u16()), false)
                    {
                        tracing::warn!(
                            target: "recursive::llm",
                            attempt,
                            backoff_ms = backoff.as_millis(),
                            status = status.as_u16(),
                            "transient HTTP error, retrying (stream)"
                        );
                        tokio::time::sleep(backoff).await;
                        attempt += 1;
                        continue;
                    }
                    return Err(self.make_err(format!("HTTP {status}: {text}")));
                }
                Err(e) => {
                    if let Some(backoff) = self.retry.backoff_for(attempt, None, true) {
                        tracing::warn!(
                            target: "recursive::llm",
                            attempt,
                            backoff_ms = backoff.as_millis(),
                            error = %e,
                            "network error, retrying (stream)"
                        );
                        tokio::time::sleep(backoff).await;
                        attempt += 1;
                        continue;
                    }
                    return Err(self.make_err(format!("request failed: {e}")));
                }
            }
        }
    }

    /// Parse an SSE stream from a successful HTTP response.
    ///
    /// Reads `data: {...}\n\n` chunks line-by-line, extracts
    /// `choices[0].delta.content` deltas, accumulates them, and emits
    /// each delta through `stream_tx` if configured. Returns the final
    /// `Completion` matching the non-streaming shape.
    async fn parse_sse_stream(
        &self,
        resp: reqwest::Response,
        stream_tx: Option<StreamSender>,
        cancel_token: Option<tokio_util::sync::CancellationToken>,
    ) -> Result<Completion> {
        let mut content = String::new();
        let mut reasoning_content = String::new();
        // key: index → (id, name, accumulated_arguments)
        let mut tool_call_builders: HashMap<usize, (String, String, String)> = HashMap::new();
        let mut finish_reason: Option<String> = None;
        let mut usage: Option<TokenUsage> = None;

        // Process the byte stream incrementally line by line.
        //
        // We maintain a raw byte buffer (`incomplete`) to handle multi-byte
        // UTF-8 sequences that may be split across HTTP chunk boundaries.
        // String::from_utf8_lossy() would silently replace the incomplete
        // tail bytes with U+FFFD, permanently corrupting non-ASCII content
        // (e.g. Chinese/Japanese text in tool arguments or assistant output).
        let mut byte_stream = resp.bytes_stream();
        let mut incomplete: Vec<u8> = Vec::new();
        let mut line_buf = String::new();

        // 决策 4b：把 SSE 循环改成 tokio::select!，让 cancel_token 能在两个 chunk 之间立刻
        // 触发。触发后函数返回 Err(Error::Cancelled)；调用方 stream() 把 reqwest::Response
        // drop（连接关闭）；run_inner 的 dispatch_llm_step 调用点会把这个错误翻译成
        // FinishReason::Cancelled（Invariant #7 不破）。
        loop {
            let chunk_result = if let Some(ct) = cancel_token.as_ref() {
                tokio::select! {
                    biased;
                    _ = ct.cancelled() => {
                        // Goal 382: a user-initiated cancel mid-stream should
                        // still persist whatever content already arrived, so
                        // the next "继续" turn doesn't see a gap where the
                        // reply was. If anything accumulated, return it as a
                        // partial Completion (finish_reason "interrupted"
                        // sentinel); run_inner routes it to
                        // FinishReason::Cancelled. With nothing accumulated
                        // there is nothing to save — keep the original
                        // Err(Error::Cancelled) so run_inner's Cancelled arm
                        // handles it.
                        let have_partial = !content.is_empty()
                            || !reasoning_content.is_empty()
                            || !tool_call_builders.is_empty();
                        if have_partial {
                            tracing::warn!(
                                target: "recursive::llm",
                                content_len = content.len(),
                                "SSE stream cancelled; returning partial completion"
                            );
                            return Ok(Self::build_completion(
                                content,
                                reasoning_content,
                                HashMap::new(), // drop half-received tool calls
                                Some("interrupted".to_string()),
                                usage.take(),
                            ));
                        }
                        // reqwest::Response 在函数返回时被 drop，HTTP 连接关闭
                        return Err(Error::Cancelled);
                    }
                    next = byte_stream.next() => next,
                }
            } else {
                byte_stream.next().await
            };
            let chunk = match chunk_result {
                Some(Ok(c)) => c,
                Some(Err(e)) => {
                    // Goal 382: a network drop mid-stream currently discards
                    // everything accumulated in `content` /
                    // `reasoning_content` / `tool_call_builders`. If any
                    // content arrived, return it as a partial Completion so
                    // the caller persists it (the "interrupted" sentinel
                    // routes to FinishReason::Cancelled). A drop before the
                    // first token has nothing to save — keep the Err.
                    let have_partial = !content.is_empty()
                        || !reasoning_content.is_empty()
                        || !tool_call_builders.is_empty();
                    if have_partial {
                        tracing::warn!(
                            target: "recursive::llm",
                            error = %e,
                            content_len = content.len(),
                            "SSE stream interrupted; returning partial completion"
                        );
                        return Ok(Self::build_completion(
                            content,
                            reasoning_content,
                            HashMap::new(), // drop half-received tool calls
                            Some("interrupted".to_string()),
                            usage.take(),
                        ));
                    }
                    return Err(self.make_err(format!("SSE stream read error: {e}")));
                }
                None => break,
            };
            // Prepend any leftover bytes from the previous chunk.
            let combined: Vec<u8> = if incomplete.is_empty() {
                chunk.to_vec()
            } else {
                let mut v = std::mem::take(&mut incomplete);
                v.extend_from_slice(&chunk);
                v
            };
            // Decode as much valid UTF-8 as possible; keep the remainder.
            let valid_up_to = match std::str::from_utf8(&combined) {
                Ok(_) => combined.len(),
                Err(e) => e.valid_up_to(),
            };
            incomplete = combined[valid_up_to..].to_vec();
            // SAFETY: valid_up_to is guaranteed by from_utf8 to be a valid
            // UTF-8 boundary within `combined`.
            let text = unsafe { std::str::from_utf8_unchecked(&combined[..valid_up_to]) };
            for ch in text.chars() {
                if ch == '\n' {
                    let line = std::mem::take(&mut line_buf);
                    Self::process_sse_line(
                        &line,
                        &self.model,
                        &mut content,
                        &mut reasoning_content,
                        &mut tool_call_builders,
                        &mut finish_reason,
                        &mut usage,
                        &stream_tx,
                    )?;
                } else if ch != '\r' {
                    line_buf.push(ch);
                }
            }
        }
        // Flush any remaining bytes (shouldn't happen with a well-formed stream,
        // but handle gracefully to avoid silent truncation).
        Self::warn_incomplete_utf8_tail(&incomplete);
        // Flush any trailing line without a terminating newline.
        if !line_buf.is_empty() {
            Self::process_sse_line(
                &line_buf,
                &self.model,
                &mut content,
                &mut reasoning_content,
                &mut tool_call_builders,
                &mut finish_reason,
                &mut usage,
                &stream_tx,
            )?;
        }

        // Normal end of stream: assemble the Completion from the accumulated
        // state (shared with the Goal-382 partial-return paths above).
        Ok(Self::build_completion(
            content,
            reasoning_content,
            tool_call_builders,
            finish_reason,
            usage,
        ))
    }

    /// Assemble a [`Completion`] from accumulated SSE state. Shared by the
    /// normal end-of-stream path and the Goal-382 partial-return paths
    /// (stream read error / user-initiated cancel mid-stream).
    fn build_completion(
        content: String,
        reasoning_content: String,
        mut tool_call_builders: HashMap<usize, (String, String, String)>,
        finish_reason: Option<String>,
        usage: Option<TokenUsage>,
    ) -> Completion {
        // Convert accumulated builders into ToolCall objects, sorted by index.
        let mut sorted_indices: Vec<usize> = tool_call_builders.keys().copied().collect();
        sorted_indices.sort_unstable();
        let tool_calls: Vec<ToolCall> = sorted_indices
            .into_iter()
            .filter_map(|i| {
                let (id, name, partial_json) = tool_call_builders.remove(&i)?;
                if id.is_empty() && name.is_empty() {
                    return None;
                }
                let arguments = if partial_json.trim().is_empty() {
                    Value::Object(Default::default())
                } else {
                    serde_json::from_str(&partial_json).unwrap_or(Value::String(partial_json))
                };
                Some(ToolCall {
                    id,
                    name,
                    arguments,
                })
            })
            .collect();

        Completion {
            content,
            tool_calls,
            finish_reason,
            usage,
            reasoning_content: if reasoning_content.is_empty() {
                None
            } else {
                Some(reasoning_content)
            },
        }
    }

    /// Log (and discard) a trailing incomplete UTF-8 sequence at end of stream.
    ///
    /// `#[mutants::skip]`: this helper only emits a warning, so mutating the
    /// `!is_empty` check is behavior-equivalent and untestable without
    /// asserting on log output.
    #[cfg_attr(test, mutants::skip)]
    fn warn_incomplete_utf8_tail(incomplete: &[u8]) {
        if !incomplete.is_empty() {
            tracing::warn!(
                target: "recursive::llm",
                bytes = incomplete.len(),
                "SSE stream ended with incomplete UTF-8 sequence; discarding tail bytes"
            );
        }
    }

    // provider is needed to produce accurate error messages when the OpenAI adapter
    // is used by non-OpenAI backends (DeepSeek, GLM, Moonshot, …).
    #[allow(clippy::too_many_arguments)]
    fn process_sse_line(
        line: &str,
        provider: &str,
        content: &mut String,
        reasoning_content: &mut String,
        tool_call_builders: &mut HashMap<usize, (String, String, String)>,
        finish_reason: &mut Option<String>,
        usage: &mut Option<TokenUsage>,
        stream_tx: &Option<StreamSender>,
    ) -> Result<()> {
        let data = match line.strip_prefix("data: ") {
            Some(d) => d,
            None => return Ok(()),
        };
        if data.trim() == "[DONE]" {
            return Ok(());
        }

        let chunk: Value = serde_json::from_str(data).map_err(|e| Error::Llm {
            provider: provider.to_string(),
            message: format!("SSE parse error: {e}; data: {data}"),
        })?;

        if let Some(choices) = chunk.get("choices").and_then(|c| c.as_array()) {
            if let Some(choice) = choices.first() {
                if let Some(delta) = choice.get("delta") {
                    if let Some(delta_content) = delta.get("content").and_then(|c| c.as_str()) {
                        if !delta_content.is_empty() {
                            content.push_str(delta_content);
                            if let Some(tx) = stream_tx {
                                let _ = tx.send(StreamChunk::Text(delta_content.to_string()));
                            }
                        }
                    }
                    if let Some(delta_reasoning) =
                        delta.get("reasoning_content").and_then(|c| c.as_str())
                    {
                        if !delta_reasoning.is_empty() {
                            reasoning_content.push_str(delta_reasoning);
                            if let Some(tx) = stream_tx {
                                let _ =
                                    tx.send(StreamChunk::Reasoning(delta_reasoning.to_string()));
                            }
                        }
                    }
                    // Accumulate tool_calls deltas
                    if let Some(tcs) = delta.get("tool_calls").and_then(|v| v.as_array()) {
                        for tc in tcs {
                            let idx =
                                tc.get("index").and_then(|v| v.as_u64()).unwrap_or(0) as usize;
                            let entry = tool_call_builders
                                .entry(idx)
                                .or_insert_with(|| (String::new(), String::new(), String::new()));
                            if let Some(id) = tc.get("id").and_then(|v| v.as_str()) {
                                if !id.is_empty() {
                                    entry.0 = id.to_string();
                                }
                            }
                            if let Some(func) = tc.get("function") {
                                if let Some(name) = func.get("name").and_then(|v| v.as_str()) {
                                    if !name.is_empty() {
                                        entry.1 = name.to_string();
                                    }
                                }
                                if let Some(args) = func.get("arguments").and_then(|v| v.as_str()) {
                                    entry.2.push_str(args);
                                }
                            }
                        }
                    }
                }
                if let Some(fr) = choice.get("finish_reason").and_then(|f| f.as_str()) {
                    if !fr.is_empty() {
                        *finish_reason = Some(fr.to_string());
                    }
                }
            }
        }

        if let Some(u) = chunk.get("usage") {
            let prompt = u.get("prompt_tokens").and_then(|v| v.as_u64()).unwrap_or(0) as u32;
            let completion = u
                .get("completion_tokens")
                .and_then(|v| v.as_u64())
                .unwrap_or(0) as u32;
            let total = u.get("total_tokens").and_then(|v| v.as_u64()).unwrap_or(0) as u32;
            // Cache hit: prefer top-level field (DeepSeek / OpenAI native),
            // fall back to `prompt_tokens_details.cached_tokens` (MiniMax-M3).
            let cache_hit = u
                .get("prompt_cache_hit_tokens")
                .and_then(|v| v.as_u64())
                .or_else(|| {
                    u.get("prompt_tokens_details")
                        .and_then(|d| d.get("cached_tokens"))
                        .and_then(|v| v.as_u64())
                })
                .unwrap_or(0) as u32;
            let explicit_miss = u
                .get("prompt_cache_miss_tokens")
                .and_then(|v| v.as_u64())
                .unwrap_or(0) as u32;
            // Cache miss: prefer the explicit value when it is non-zero;
            // an explicit 0 is ambiguous (OpenAI often omits the field or
            // sends 0 while still reporting `cached_tokens`), so derive
            // from `prompt_tokens - cache_hit` instead to honour the
            // TokenUsage invariant `hit + miss == prompt_tokens` and keep
            // the TUI percentage accurate instead of pegging at 100%.
            let cache_miss = if explicit_miss > 0 {
                explicit_miss
            } else if cache_hit > 0 {
                prompt.saturating_sub(cache_hit)
            } else {
                0
            };
            let reasoning = u
                .get("completion_tokens_details")
                .and_then(|d| d.get("reasoning_tokens"))
                .and_then(|v| v.as_u64())
                .unwrap_or(0) as u32;
            *usage = Some(TokenUsage {
                prompt_tokens: prompt,
                completion_tokens: completion,
                total_tokens: total,
                cache_hit_tokens: cache_hit,
                cache_miss_tokens: cache_miss,
                reasoning_tokens: reasoning,
            });
        }
        Ok(())
    }
}

fn build_request(
    model: &str,
    temperature: f64,
    max_tokens: u32,
    messages: &[Message],
    tools: &[ToolSpec],
) -> Value {
    let mut req = serde_json::json!({
        "model": model,
        "temperature": temperature,
        "max_tokens": max_tokens,
        "messages": messages.iter().map(serialize_message).collect::<Vec<_>>(),
    });
    if !tools.is_empty() {
        let tools_json: Vec<Value> = tools
            .iter()
            .map(|t| {
                serde_json::json!({
                    "type": "function",
                    "function": {
                        "name": t.name,
                        "description": t.description,
                        "parameters": t.parameters,
                    }
                })
            })
            .collect();
        req["tools"] = Value::Array(tools_json);
        req["tool_choice"] = Value::String("auto".into());
    }
    // MiniMax-M3 (and M2.x) default to embedding chain-of-thought inside
    // `content` as `<think>…</think>`. That makes the TUI stream thinking
    // as a normal assistant bullet, then hoist it into a Reasoning block
    // only at finalise. `reasoning_split: true` asks MiniMax to emit
    // thinking on the dedicated `reasoning_content` SSE field instead,
    // which our stream parser already forwards as StreamChunk::Reasoning.
    // Other OpenAI-compatible providers ignore unknown fields or reject
    // them — keep this MiniMax-only.
    if model_wants_reasoning_split(model) {
        req["reasoning_split"] = Value::Bool(true);
    }
    req
}

/// MiniMax OpenAI-compatible models that honour `reasoning_split`.
fn model_wants_reasoning_split(model: &str) -> bool {
    model
        .to_ascii_lowercase()
        .split(|c: char| !c.is_ascii_alphanumeric())
        .any(|part| part == "minimax")
}

fn serialize_message(m: &Message) -> Value {
    let role = match m.role {
        Role::System => "system",
        Role::User => "user",
        Role::Assistant => "assistant",
        Role::Tool => "tool",
    };
    let mut obj = serde_json::Map::new();
    obj.insert("role".into(), Value::String(role.into()));
    obj.insert("content".into(), Value::String(m.content.clone()));
    if let Some(id) = &m.tool_call_id {
        obj.insert("tool_call_id".into(), Value::String(id.clone()));
    }
    // Echo reasoning_content back to the API (required by DeepSeek thinking mode).
    // Only valid on Assistant messages — other roles have no reasoning field.
    if matches!(m.role, Role::Assistant) {
        if let Some(ref reasoning) = m.reasoning_content {
            obj.insert("reasoning_content".into(), Value::String(reasoning.clone()));
        }
    }
    if !m.tool_calls.is_empty() {
        let calls: Vec<Value> = m
            .tool_calls
            .iter()
            .map(|c| {
                serde_json::json!({
                    "id": c.id,
                    "type": "function",
                    "function": {
                        "name": c.name,
                        "arguments": serde_json::to_string(&c.arguments).unwrap_or_else(|_| "{}".into()),
                    }
                })
            })
            .collect();
        obj.insert("tool_calls".into(), Value::Array(calls));
    }
    Value::Object(obj)
}

#[derive(Debug, Deserialize)]
struct ChatResponse {
    choices: Vec<ChatChoice>,
    #[serde(default)]
    usage: Option<ResponseUsage>,
}

#[derive(Debug, Deserialize)]
struct ResponseUsage {
    #[serde(default)]
    prompt_tokens: Option<u32>,
    #[serde(default)]
    completion_tokens: Option<u32>,
    #[serde(default)]
    total_tokens: Option<u32>,
    /// OpenAI standard. Some compatible providers (DeepSeek) report cache
    /// usage here directly. Others (MiniMax-M3) nest it under
    /// `prompt_tokens_details.cached_tokens` instead — see below.
    #[serde(default)]
    prompt_cache_hit_tokens: Option<u32>,
    #[serde(default)]
    prompt_cache_miss_tokens: Option<u32>,
    /// OpenAI-standard nested details. MiniMax-M3 reports prompt cache hits
    /// here as `cached_tokens` rather than at the top level. The TUI cache
    /// rate depends on this being read.
    #[serde(default)]
    prompt_tokens_details: Option<PromptTokensDetails>,
    /// OpenAI-standard nested details. o1/o3 and thinking-capable MiniMax
    /// models (e.g. MiniMax-M3) report reasoning tokens here. Previously
    /// only the streaming path captured this; non-streaming calls left
    /// `reasoning_tokens` at 0, undercharging the cost total. (Goal 273
    /// follow-up.)
    #[serde(default)]
    completion_tokens_details: Option<CompletionTokensDetails>,
}

#[derive(Debug, Default, Deserialize)]
struct PromptTokensDetails {
    #[serde(default)]
    cached_tokens: Option<u32>,
}

#[derive(Debug, Default, Deserialize)]
struct CompletionTokensDetails {
    #[serde(default)]
    reasoning_tokens: Option<u32>,
}

impl ResponseUsage {
    fn to_token_usage(&self) -> TokenUsage {
        // Cache hit: top-level field first (DeepSeek, OpenAI native), then
        // fall back to `prompt_tokens_details.cached_tokens` (MiniMax-M3).
        let cache_hit = self
            .prompt_cache_hit_tokens
            .or_else(|| {
                self.prompt_tokens_details
                    .as_ref()
                    .and_then(|d| d.cached_tokens)
            })
            .unwrap_or(0);

        // Cache miss: providers that report hit explicitly usually also
        // report miss, so prefer the explicit value — BUT a `0` miss is
        // ambiguous: OpenAI's official API frequently omits
        // `prompt_cache_miss_tokens` (or sends it as 0) while still
        // reporting `cached_tokens`, which would falsely peg the TUI at
        // 100%. Treat an explicit `0` the same as "not provided" and
        // derive miss from `prompt_tokens - hit` so the documented
        // invariant `cache_hit + cache_miss == prompt_tokens` holds. This
        // keeps the TUI cache-hit percentage accurate instead of reading
        // 100% on turns that actually had cache misses.
        let cache_miss = match self.prompt_cache_miss_tokens {
            Some(m) if m > 0 => m,
            _ if cache_hit > 0 => self.prompt_tokens.unwrap_or(0).saturating_sub(cache_hit),
            _ => 0,
        };

        let reasoning_tokens = self
            .completion_tokens_details
            .as_ref()
            .and_then(|d| d.reasoning_tokens)
            .unwrap_or(0);

        TokenUsage {
            reasoning_tokens,
            prompt_tokens: self.prompt_tokens.unwrap_or(0),
            completion_tokens: self.completion_tokens.unwrap_or(0),
            total_tokens: self.total_tokens.unwrap_or(0),
            cache_hit_tokens: cache_hit,
            cache_miss_tokens: cache_miss,
        }
    }
}

#[derive(Debug, Deserialize)]
struct ChatChoice {
    message: ChatChoiceMessage,
    #[serde(default)]
    finish_reason: Option<String>,
}

#[derive(Debug, Deserialize)]
struct ChatChoiceMessage {
    #[serde(default)]
    content: Option<String>,
    #[serde(default)]
    reasoning_content: Option<String>,
    #[serde(default)]
    tool_calls: Vec<RawToolCall>,
}

#[derive(Debug, Deserialize, Serialize)]
struct RawToolCall {
    id: String,
    #[serde(default)]
    function: RawFunction,
}

#[derive(Debug, Default, Deserialize, Serialize)]
struct RawFunction {
    #[serde(default)]
    name: String,
    #[serde(default)]
    arguments: String,
}

fn parse_completion(choice: ChatChoice, usage: Option<ResponseUsage>) -> Completion {
    let content = choice.message.content.unwrap_or_default();
    let reasoning_content = choice.message.reasoning_content;
    let tool_calls = choice
        .message
        .tool_calls
        .into_iter()
        .map(|c| {
            let args: Value = if c.function.arguments.trim().is_empty() {
                Value::Object(Default::default())
            } else {
                serde_json::from_str(&c.function.arguments)
                    .unwrap_or_else(|_| Value::String(c.function.arguments.clone()))
            };
            ToolCall {
                id: c.id,
                name: c.function.name,
                arguments: args,
            }
        })
        .collect();
    Completion {
        content,
        tool_calls,
        finish_reason: choice.finish_reason,
        usage: usage.map(|u| u.to_token_usage()),
        reasoning_content,
    }
}

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

    #[test]
    fn parses_plain_text_choice() {
        let raw = r#"{"choices":[{"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}]}"#;
        let parsed: ChatResponse = serde_json::from_str(raw).unwrap();
        let c = parse_completion(parsed.choices.into_iter().next().unwrap(), parsed.usage);
        assert_eq!(c.content, "hi");
        assert!(c.tool_calls.is_empty());
        assert_eq!(c.finish_reason.as_deref(), Some("stop"));
        assert!(c.usage.is_none());
    }

    #[test]
    fn parses_tool_call_choice() {
        let raw = r#"{
            "choices":[{
                "message":{
                    "role":"assistant",
                    "content":null,
                    "tool_calls":[{
                        "id":"call_1",
                        "type":"function",
                        "function":{"name":"Read","arguments":"{\"path\":\"src/lib.rs\"}"}
                    }]
                },
                "finish_reason":"tool_calls"
            }]
        }"#;
        let parsed: ChatResponse = serde_json::from_str(raw).unwrap();
        let c = parse_completion(parsed.choices.into_iter().next().unwrap(), parsed.usage);
        assert_eq!(c.tool_calls.len(), 1);
        assert_eq!(c.tool_calls[0].name, "Read");
        assert_eq!(c.tool_calls[0].arguments["path"], "src/lib.rs");
    }

    #[test]
    fn parses_usage_from_response() {
        let raw = r#"{
            "choices":[{"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}],
            "usage":{"prompt_tokens":41,"completion_tokens":7,"total_tokens":48}
        }"#;
        let parsed: ChatResponse = serde_json::from_str(raw).unwrap();
        let c = parse_completion(parsed.choices.into_iter().next().unwrap(), parsed.usage);
        assert!(c.usage.is_some());
        let u = c.usage.unwrap();
        assert_eq!(u.prompt_tokens, 41);
        assert_eq!(u.completion_tokens, 7);
        assert_eq!(u.total_tokens, 48);
    }

    #[test]
    fn parses_missing_usage_as_none() {
        let raw = r#"{"choices":[{"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}]}"#;
        let parsed: ChatResponse = serde_json::from_str(raw).unwrap();
        let c = parse_completion(parsed.choices.into_iter().next().unwrap(), parsed.usage);
        assert!(c.usage.is_none());
    }

    #[test]
    fn parses_partial_usage_fills_zeros() {
        let raw = r#"{
            "choices":[{"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}],
            "usage":{"total_tokens":50}
        }"#;
        let parsed: ChatResponse = serde_json::from_str(raw).unwrap();
        let c = parse_completion(parsed.choices.into_iter().next().unwrap(), parsed.usage);
        assert!(c.usage.is_some());
        let u = c.usage.unwrap();
        assert_eq!(u.prompt_tokens, 0);
        assert_eq!(u.completion_tokens, 0);
        assert_eq!(u.total_tokens, 50);
        assert_eq!(u.cache_hit_tokens, 0);
        assert_eq!(u.cache_miss_tokens, 0);
    }

    #[test]
    fn parses_cache_fields_from_deepseek_usage() {
        let raw = r#"{
            "choices":[{"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}],
            "usage":{
                "prompt_tokens":100,
                "completion_tokens":50,
                "total_tokens":150,
                "prompt_cache_hit_tokens":60,
                "prompt_cache_miss_tokens":40
            }
        }"#;
        let parsed: ChatResponse = serde_json::from_str(raw).unwrap();
        let c = parse_completion(parsed.choices.into_iter().next().unwrap(), parsed.usage);
        assert!(c.usage.is_some());
        let u = c.usage.unwrap();
        assert_eq!(u.prompt_tokens, 100);
        assert_eq!(u.completion_tokens, 50);
        assert_eq!(u.total_tokens, 150);
        assert_eq!(u.cache_hit_tokens, 60);
        assert_eq!(u.cache_miss_tokens, 40);
    }

    #[test]
    fn parses_cache_fields_as_zero_when_absent() {
        let raw = r#"{
            "choices":[{"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}],
            "usage":{"prompt_tokens":10,"completion_tokens":5,"total_tokens":15}
        }"#;
        let parsed: ChatResponse = serde_json::from_str(raw).unwrap();
        let c = parse_completion(parsed.choices.into_iter().next().unwrap(), parsed.usage);
        assert!(c.usage.is_some());
        let u = c.usage.unwrap();
        assert_eq!(u.cache_hit_tokens, 0);
        assert_eq!(u.cache_miss_tokens, 0);
    }

    #[test]
    fn parses_minimax_cached_tokens_nested_under_prompt_tokens_details() {
        // MiniMax-M3 returns cache hits as
        // `usage.prompt_tokens_details.cached_tokens` rather than at the
        // top level. Without the nested read the TUI cache rate stays
        // invisible because both hit and miss collapse to 0.
        let raw = r#"{
            "choices":[{"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}],
            "usage":{
                "prompt_tokens":780,
                "completion_tokens":20,
                "total_tokens":800,
                "completion_tokens_details":{"reasoning_tokens":19},
                "prompt_tokens_details":{"cached_tokens":768}
            }
        }"#;
        let parsed: ChatResponse = serde_json::from_str(raw).unwrap();
        let c = parse_completion(parsed.choices.into_iter().next().unwrap(), parsed.usage);
        let u = c.usage.expect("usage should be parsed");
        assert_eq!(u.prompt_tokens, 780);
        assert_eq!(u.completion_tokens, 20);
        assert_eq!(u.cache_hit_tokens, 768, "cached_tokens must populate hit");
        // hit + miss should equal prompt_tokens (invariant).
        assert_eq!(
            u.cache_hit_tokens + u.cache_miss_tokens,
            u.prompt_tokens,
            "cache_hit + cache_miss must equal prompt_tokens"
        );
        assert_eq!(u.reasoning_tokens, 19);
    }

    #[test]
    fn derives_cache_miss_when_only_hit_is_reported() {
        // MiniMax-style usage: only `cached_tokens` reported, no explicit
        // `prompt_cache_miss_tokens`. Derive miss = prompt - hit so the
        // TUI cache-hit percentage reflects reality (here ~98.5%) instead
        // of pegging at 100%.
        let raw = r#"{
            "choices":[{"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}],
            "usage":{
                "prompt_tokens":780,
                "completion_tokens":20,
                "total_tokens":800,
                "prompt_tokens_details":{"cached_tokens":768}
            }
        }"#;
        let parsed: ChatResponse = serde_json::from_str(raw).unwrap();
        let c = parse_completion(parsed.choices.into_iter().next().unwrap(), parsed.usage);
        let u = c.usage.expect("usage should be parsed");
        assert_eq!(u.cache_hit_tokens, 768);
        assert_eq!(u.cache_miss_tokens, 12);
    }

    #[test]
    fn explicit_zero_cache_miss_is_derived_not_trusted() {
        // OpenAI's official API often sends `prompt_cache_miss_tokens: 0`
        // (or omits it) while still reporting a non-zero `cached_tokens`.
        // A literal 0 must NOT be treated as "no misses" — that would peg
        // the TUI at 100% on turns that actually had misses. Derive
        // miss = prompt - hit instead, so the rate is ~98.5% here.
        let raw = r#"{
            "choices":[{"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}],
            "usage":{
                "prompt_tokens":780,
                "completion_tokens":20,
                "total_tokens":800,
                "prompt_cache_hit_tokens":768,
                "prompt_cache_miss_tokens":0
            }
        }"#;
        let parsed: ChatResponse = serde_json::from_str(raw).unwrap();
        let c = parse_completion(parsed.choices.into_iter().next().unwrap(), parsed.usage);
        let u = c.usage.expect("usage should be parsed");
        assert_eq!(u.cache_hit_tokens, 768);
        assert_eq!(
            u.cache_miss_tokens, 12,
            "explicit 0 miss must derive, not peg at 100%"
        );
    }

    #[test]
    fn explicit_cache_miss_takes_precedence_over_derived() {
        // When the provider reports both `prompt_cache_miss_tokens` and
        // `cached_tokens`, the explicit miss wins — even if it would
        // disagree with `prompt - hit`. This protects against providers
        // that round or batch differently (e.g. DeepSeek reports the
        // split directly and that's the source of truth).
        let raw = r#"{
            "choices":[{"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}],
            "usage":{
                "prompt_tokens":100,
                "completion_tokens":50,
                "total_tokens":150,
                "prompt_cache_hit_tokens":60,
                "prompt_cache_miss_tokens":40,
                "prompt_tokens_details":{"cached_tokens":999}
            }
        }"#;
        let parsed: ChatResponse = serde_json::from_str(raw).unwrap();
        let c = parse_completion(parsed.choices.into_iter().next().unwrap(), parsed.usage);
        let u = c.usage.expect("usage should be parsed");
        // Top-level hit field wins over the nested value.
        assert_eq!(u.cache_hit_tokens, 60);
        // Explicit miss wins over derived (100 - 60 = 40, agrees here,
        // but the test guards against future regressions).
        assert_eq!(u.cache_miss_tokens, 40);
    }

    #[test]
    fn parses_reasoning_tokens_in_non_streaming_path() {
        // o1 / o3 / MiniMax thinking-mode models report reasoning tokens
        // under `usage.completion_tokens_details.reasoning_tokens`. The
        // non-streaming path used to drop this on the floor, undercharging
        // the cost total.
        let raw = r#"{
            "choices":[{"message":{"role":"assistant","content":"answer"},"finish_reason":"stop"}],
            "usage":{
                "prompt_tokens":50,
                "completion_tokens":200,
                "total_tokens":250,
                "completion_tokens_details":{"reasoning_tokens":175}
            }
        }"#;
        let parsed: ChatResponse = serde_json::from_str(raw).unwrap();
        let c = parse_completion(parsed.choices.into_iter().next().unwrap(), parsed.usage);
        let u = c.usage.expect("usage should be parsed");
        assert_eq!(u.completion_tokens, 200);
        assert_eq!(u.reasoning_tokens, 175);
    }

    #[test]
    fn stream_usage_parses_minimax_nested_cache_and_reasoning() {
        // Drive process_sse_line directly with a single chunk whose
        // usage shape matches MiniMax-M3. Assert both `cached_tokens`
        // and `reasoning_tokens` are extracted.
        let mut usage: Option<TokenUsage> = None;
        let mut content = String::new();
        let mut reasoning_content = String::new();
        let mut tool_call_builders: std::collections::HashMap<usize, (String, String, String)> =
            std::collections::HashMap::new();
        let mut finish_reason: Option<String> = None;
        let data = r#"data: {"choices":[{"delta":{"content":"ok"},"finish_reason":"stop"}],"usage":{"prompt_tokens":780,"completion_tokens":20,"total_tokens":800,"completion_tokens_details":{"reasoning_tokens":19},"prompt_tokens_details":{"cached_tokens":768}}}"#;
        OpenAiProvider::process_sse_line(
            data,
            "MiniMax-M3",
            &mut content,
            &mut reasoning_content,
            &mut tool_call_builders,
            &mut finish_reason,
            &mut usage,
            &None,
        )
        .unwrap();
        let u = usage.expect("usage should be set");
        assert_eq!(u.cache_hit_tokens, 768);
        assert_eq!(u.cache_miss_tokens, 12);
        assert_eq!(u.reasoning_tokens, 19);
        assert_eq!(content, "ok");
    }

    #[test]
    fn serialises_assistant_with_tool_calls() {
        let msg = Message::assistant_with_tool_calls(
            "",
            vec![ToolCall {
                id: "abc".into(),
                name: "Write".into(),
                arguments: serde_json::json!({"path":"a","contents":"b"}),
            }],
        );
        let v = serialize_message(&msg);
        assert_eq!(v["tool_calls"][0]["function"]["name"], "Write");
        let args = v["tool_calls"][0]["function"]["arguments"]
            .as_str()
            .unwrap();
        let decoded: Value = serde_json::from_str(args).unwrap();
        assert_eq!(decoded["contents"], "b");
    }

    // ---- ToolSearch loop tests ----

    #[derive(Debug)]
    struct AlwaysReturnsEngine(Vec<String>);
    impl ToolSearchEngine for AlwaysReturnsEngine {
        fn resolve(&self, _query: &str, _candidates: &[SpecWithHint]) -> Vec<String> {
            self.0.clone()
        }
    }

    fn make_tool_spec(name: &str) -> ToolSpec {
        ToolSpec {
            name: name.to_string(),
            description: format!("{name} tool"),
            parameters: serde_json::json!({"type":"object","properties":{},"required":[]}),
        }
    }

    #[tokio::test]
    async fn openai_search_loop_skips_when_no_deferred() {
        // With no deferred tools, run_search_loop should make exactly one
        // request (no ToolSearch round-trip). We use a mock server that
        // returns a plain text response and assert it's called exactly once.
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        let call_count = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let call_count_clone = call_count.clone();

        std::thread::spawn(move || {
            use std::io::{Read, Write};
            loop {
                if let Ok((mut stream, _)) = listener.accept() {
                    call_count_clone.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                    let mut buf = [0u8; 8192];
                    let _ = stream.read(&mut buf);
                    let body = r#"{"choices":[{"message":{"role":"assistant","content":"done"},"finish_reason":"stop"}]}"#;
                    write!(
                        stream,
                        "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                        body.len(),
                        body
                    )
                    .unwrap();
                    stream.flush().unwrap();
                }
            }
        });
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        let provider = OpenAiProvider::new(format!("http://{addr}"), "sk-noop", "m").unwrap();
        let eager = vec![(make_tool_spec("Read"), None)];
        let result = provider
            .complete_with_search(&[Message::user("hi")], &eager, &[])
            .await
            .unwrap();
        assert_eq!(result.content, "done");
        assert_eq!(call_count.load(std::sync::atomic::Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn openai_search_loop_resolves_deferred_tool() {
        // Two mock responses:
        // Round 0: model calls ToolSearchTool with query "MyTool"
        // Round 1: model returns normal content
        // After round 1 the tool list should include MyTool's schema.
        use std::sync::Mutex;
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        let captured_tools: std::sync::Arc<Mutex<Vec<String>>> =
            std::sync::Arc::new(Mutex::new(vec![]));
        let captured_clone = captured_tools.clone();
        let call_idx = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let call_idx_clone = call_idx.clone();

        std::thread::spawn(move || {
            use std::io::{Read, Write};
            loop {
                if let Ok((mut stream, _)) = listener.accept() {
                    let idx = call_idx_clone.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                    let mut buf = [0u8; 65536];
                    let n = stream.read(&mut buf).unwrap_or(0);
                    let req = String::from_utf8_lossy(&buf[..n]);
                    // Extract JSON body and record tool names
                    if let Some(start) = req.find("\r\n\r\n") {
                        let body_str = req[start + 4..].trim();
                        if let Ok(body) = serde_json::from_str::<serde_json::Value>(body_str) {
                            if let Some(tools) = body["tools"].as_array() {
                                let names: Vec<String> = tools
                                    .iter()
                                    .filter_map(|t| {
                                        t["function"]["name"].as_str().map(String::from)
                                    })
                                    .collect();
                                *captured_clone.lock().unwrap() = names;
                            }
                        }
                    }
                    let resp = if idx == 0 {
                        // First call: return a ToolSearchTool call
                        r#"{"choices":[{"message":{"role":"assistant","content":"","tool_calls":[{"id":"ts1","type":"function","function":{"name":"ToolSearchTool","arguments":"{\"query\":\"MyTool\"}"}}]},"finish_reason":"tool_calls"}]}"#
                    } else {
                        // Second call: return normal content
                        r#"{"choices":[{"message":{"role":"assistant","content":"found it"},"finish_reason":"stop"}]}"#
                    };
                    write!(
                        stream,
                        "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                        resp.len(),
                        resp
                    )
                    .unwrap();
                    stream.flush().unwrap();
                }
            }
        });
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        let my_tool = make_tool_spec("MyTool");
        let deferred = vec![(my_tool.clone(), None)];
        let engine = Arc::new(AlwaysReturnsEngine(vec!["MyTool".to_string()]));
        let provider = OpenAiProvider::new(format!("http://{addr}"), "sk-noop", "m")
            .unwrap()
            .with_search_engine(engine);

        let result = provider
            .complete_with_search(&[Message::user("hi")], &[], &deferred)
            .await
            .unwrap();
        assert_eq!(result.content, "found it");
        // The second request should include MyTool's schema
        let tools = captured_tools.lock().unwrap().clone();
        assert!(
            tools.contains(&"MyTool".to_string()),
            "second request should include MyTool, got: {tools:?}"
        );
        assert_eq!(call_idx.load(std::sync::atomic::Ordering::SeqCst), 2);
    }

    #[tokio::test]
    async fn openai_search_loop_caps_at_max_rounds() {
        // Mock server always returns a ToolSearchTool call.
        // Assert we stop after max_search_rounds and return.
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        let call_count = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let call_count_clone = call_count.clone();

        std::thread::spawn(move || {
            use std::io::{Read, Write};
            loop {
                if let Ok((mut stream, _)) = listener.accept() {
                    call_count_clone.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                    let mut buf = [0u8; 65536];
                    let _ = stream.read(&mut buf);
                    // Always respond with a ToolSearchTool call
                    let resp = r#"{"choices":[{"message":{"role":"assistant","content":"","tool_calls":[{"id":"ts1","type":"function","function":{"name":"ToolSearchTool","arguments":"{\"query\":\"anything\"}"}}]},"finish_reason":"tool_calls"}]}"#;
                    write!(
                        stream,
                        "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                        resp.len(),
                        resp
                    )
                    .unwrap();
                    stream.flush().unwrap();
                }
            }
        });
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        let deferred = vec![(make_tool_spec("SomeTool"), None)];
        let engine = Arc::new(AlwaysReturnsEngine(vec![])); // returns nothing
        let provider = OpenAiProvider::new(format!("http://{addr}"), "sk-noop", "m")
            .unwrap()
            .with_search_engine(engine);

        let result = provider
            .complete_with_search(&[Message::user("hi")], &[], &deferred)
            .await
            .unwrap();
        // Should have made max_search_rounds + 1 calls and stopped
        let calls = call_count.load(std::sync::atomic::Ordering::SeqCst);
        assert_eq!(calls, 3 + 1, "expected {} calls, got {calls}", 3 + 1);
        // The returned completion should be the last ToolSearchTool response
        assert!(!result.tool_calls.is_empty());
    }

    #[test]
    fn builds_request_without_tools_omits_field() {
        let req = build_request("m", 0.2, 16384, &[Message::user("hi")], &[]);
        assert!(req.get("tools").is_none());
        assert_eq!(req["messages"][0]["role"], "user");
        assert_eq!(req["max_tokens"], 16384);
        assert!(
            req.get("reasoning_split").is_none(),
            "non-MiniMax models must not get reasoning_split"
        );
    }

    #[test]
    fn builds_request_includes_max_tokens() {
        let req = build_request("m", 0.2, 1024, &[Message::user("hi")], &[]);
        assert_eq!(req["max_tokens"], 1024);
    }

    #[test]
    fn builds_request_minimax_sets_reasoning_split() {
        for model in [
            "MiniMax-M3",
            "MiniMax-M2.7",
            "minimax-m3",
            "ab-MiniMax-test",
        ] {
            let req = build_request(model, 0.2, 1024, &[Message::user("hi")], &[]);
            assert_eq!(
                req.get("reasoning_split"),
                Some(&Value::Bool(true)),
                "model {model} should request reasoning_split"
            );
        }
    }

    #[test]
    fn model_wants_reasoning_split_detects_minimax_token() {
        assert!(model_wants_reasoning_split("MiniMax-M3"));
        assert!(model_wants_reasoning_split("minimax-m2.7"));
        assert!(!model_wants_reasoning_split("deepseek-chat"));
        assert!(!model_wants_reasoning_split("gpt-4o"));
        // Must not false-positive on substrings that aren't a MiniMax token.
        assert!(!model_wants_reasoning_split("notminimax"));
    }

    #[tokio::test]
    async fn stream_request_includes_include_usage_flag() {
        // Mock a streaming endpoint and capture the request body. Assert
        // `stream_options.include_usage == true` is set, otherwise providers
        // like MiniMax-M3 won't emit the final usage chunk and the TUI
        // cache-rate badge stays invisible (regression for the bug where
        // MiniMax cache hit rate never showed up after a parser fix).
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        let captured = std::sync::Arc::new(std::sync::Mutex::new(String::new()));
        let captured_clone = captured.clone();

        std::thread::spawn(move || {
            let (mut stream, _) = listener.accept().unwrap();
            use std::io::{Read, Write};
            let mut buf = [0u8; 8192];
            let n = stream.read(&mut buf).unwrap();
            let request = String::from_utf8_lossy(&buf[..n]).to_string();
            if let Some(body_start) = request.find("\r\n\r\n") {
                let body = request[body_start + 4..].trim().to_string();
                *captured_clone.lock().unwrap() = body.clone();
            }
            let body = "data: {\"choices\":[{\"delta\":{\"content\":\"ok\"},\"finish_reason\":\"stop\"}]}\n\n\
data: [DONE]\n\n";
            write!(
                stream,
                "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                body.len(),
                body,
            )
            .unwrap();
            stream.flush().unwrap();
        });

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        let provider =
            OpenAiProvider::new(format!("http://{addr}"), "sk-noop", "MiniMax-M3").unwrap();
        let _ = provider
            .stream(&[Message::user("hi")], &[], None, None)
            .await
            .unwrap();

        let body_str = captured.lock().unwrap().clone();
        let body: serde_json::Value =
            serde_json::from_str(&body_str).expect("captured body should be valid JSON");
        assert_eq!(body["stream"], true, "stream flag must be set");
        assert_eq!(
            body["stream_options"]["include_usage"], true,
            "stream_options.include_usage must be true so providers like MiniMax emit the usage chunk"
        );
    }

    #[tokio::test]
    async fn error_includes_model_name_on_network_failure() {
        // Bind a listener on an ephemeral port, then drop it so the port is freed.
        // The next connect attempt gets ECONNREFUSED (a network error).
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        drop(listener);

        // Small sleep so the OS releases the port.
        tokio::time::sleep(std::time::Duration::from_millis(20)).await;

        let provider =
            OpenAiProvider::new(format!("http://{addr}"), "sk-noop", "test-model").unwrap();
        let err = provider
            .complete(&[Message::user("hi")], &[])
            .await
            .unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("test-model"),
            "error should contain model name: {msg}"
        );
    }

    #[tokio::test]
    async fn error_includes_model_name_and_status_on_http_error() {
        // Spawn a one-shot TCP listener that sends a 400 response.
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();

        // Use a dedicated thread with a blocking server to avoid tokio issues.
        std::thread::spawn(move || {
            let (mut stream, _) = listener.accept().unwrap();
            use std::io::{Read, Write};
            let mut buf = [0u8; 4096];
            let _ = stream.read(&mut buf);
            write!(
                stream,
                "HTTP/1.1 400 Bad Request\r\nContent-Length: 2\r\nConnection: close\r\n\r\n{{}}"
            )
            .unwrap();
            stream.flush().unwrap();
        });

        // Give the server a moment to start
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        let provider =
            OpenAiProvider::new(format!("http://{addr}"), "sk-noop", "test-model-http").unwrap();
        let err = provider
            .complete(&[Message::user("hi")], &[])
            .await
            .unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("test-model-http"),
            "error should contain model name: {msg}"
        );
        assert!(
            msg.contains("400"),
            "error should contain HTTP status: {msg}"
        );
    }

    #[tokio::test]
    async fn stream_concatenates_sse_chunks() {
        // Spawn a one-shot TCP server that serves a canned SSE response
        // with 3 chunks. Assert the returned Completion's content is the
        // concatenation of the deltas.
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();

        std::thread::spawn(move || {
            let (mut stream, _) = listener.accept().unwrap();
            use std::io::{Read, Write};
            let mut buf = [0u8; 4096];
            let _ = stream.read(&mut buf);
            let body = "\
data: {\"choices\":[{\"delta\":{\"content\":\"Hello\"},\"finish_reason\":null}]}\n\n\
data: {\"choices\":[{\"delta\":{\"content\":\" \"},\"finish_reason\":null}]}\n\n\
data: {\"choices\":[{\"delta\":{\"content\":\"World\"},\"finish_reason\":\"stop\"}]}\n\n\
data: [DONE]\n\n";
            write!(
                stream,
                "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                body.len(),
                body,
            )
            .unwrap();
            stream.flush().unwrap();
        });

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        let provider =
            OpenAiProvider::new(format!("http://{addr}"), "sk-noop", "test-stream-model").unwrap();
        let completion = provider
            .stream(&[Message::user("hi")], &[], None, None)
            .await
            .unwrap();
        assert_eq!(completion.content, "Hello World");
        assert_eq!(completion.finish_reason.as_deref(), Some("stop"));
    }

    #[tokio::test]
    async fn parse_sse_stream_returns_error_on_malformed_chunk() {
        // A flaky proxy can truncate/garbble a chunk mid-stream. The parser
        // must surface that as a diagnosable `Error::Llm` whose message
        // contains "SSE parse error" (from `process_sse_line`'s
        // `serde_json::from_str(...).map_err(...)`), NOT a silent panic or
        // an infinite hang. The malformed `data:` line comes FIRST so the
        // error fires before any valid chunk is processed; a subsequent
        // valid `[DONE]` proves the parser stops at the bad line rather
        // than skipping it and completing the stream.
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();

        std::thread::spawn(move || {
            let (mut stream, _) = listener.accept().unwrap();
            use std::io::{Read, Write};
            let mut buf = [0u8; 4096];
            let _ = stream.read(&mut buf);
            let body = "\
data: {not valid json

data: {\"choices\":[{\"delta\":{\"content\":\"ok\"},\"finish_reason\":\"stop\"}]}

data: [DONE]

";
            write!(
                stream,
                "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                body.len(),
                body,
            )
            .unwrap();
            stream.flush().unwrap();
        });

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        let provider =
            OpenAiProvider::new(format!("http://{addr}"), "sk-noop", "test-malformed-model")
                .unwrap();
        // Timeout guards against a parser bug that loops on malformed input
        // — the test should fail on timeout, not hang the suite.
        let result = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            provider.stream(&[Message::user("hi")], &[], None, None),
        )
        .await
        .expect("SSE parser must not hang on malformed input");

        let err = match result {
            Ok(_) => panic!("expected Err on malformed SSE chunk, got Ok"),
            Err(e) => e,
        };
        match err {
            Error::Llm { provider, message } => {
                assert!(
                    message.contains("SSE parse error"),
                    "error message should be diagnosable, got: {message}"
                );
                assert!(
                    provider.contains("test-malformed-model"),
                    "error should carry the provider/model name, got: {provider}"
                );
            }
            other => panic!("expected Error::Llm, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn parse_sse_stream_returns_partial_on_mid_stream_error() {
        // Goal 382: a network drop mid-stream must NOT discard the content
        // that already arrived — it must be returned as a partial Completion
        // with finish_reason "interrupted" so run_inner persists it via the
        // Cancelled path. Serve one valid SSE chunk, then close the
        // connection early with a Content-Length LARGER than the body: the
        // client's next bytes_stream().next() sees a truncated-body read
        // error (exactly the `Some(Err(e))` arm this goal touches).
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();

        std::thread::spawn(move || {
            let (mut stream, _) = listener.accept().unwrap();
            use std::io::{Read, Write};
            let mut buf = [0u8; 4096];
            let _ = stream.read(&mut buf);
            let body = "data: {\"choices\":[{\"delta\":{\"content\":\"partial\"},\"finish_reason\":null}]}\n\n";
            write!(
                stream,
                "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                9999,
                body,
            )
            .unwrap();
            stream.flush().unwrap();
            // Normal FIN close: the truncated body (9999 declared vs body.len()
            // sent) makes the client's next read error, deterministically.
        });

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        let provider =
            OpenAiProvider::new(format!("http://{addr}"), "sk-noop", "test-stream-model").unwrap();
        let completion = provider
            .stream(&[Message::user("hi")], &[], None, None)
            .await
            .expect("partial content must be returned, not Err");
        assert_eq!(
            completion.content, "partial",
            "accumulated delta must survive the mid-stream read error",
        );
        assert_eq!(
            completion.finish_reason.as_deref(),
            Some("interrupted"),
            "partial completion must carry the interrupted sentinel",
        );
        assert!(
            completion.tool_calls.is_empty(),
            "half-received tool calls must be dropped from a partial completion",
        );
    }

    #[tokio::test]
    async fn stream_fallback_delegates_to_complete() {
        // MockProvider doesn't override stream, so it falls back to complete.
        // Verify the fallback path works by using a MockProvider.
        use crate::llm::MockProvider;
        let provider = MockProvider::new(vec![super::Completion {
            content: "fallback works".to_string(),
            tool_calls: vec![],
            finish_reason: Some("stop".to_string()),
            usage: None,
            reasoning_content: None,
        }]);
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let completion = provider
            .stream(&[Message::user("hi")], &[], Some(tx), None)
            .await
            .unwrap();
        assert_eq!(completion.content, "fallback works");
        // Should have received the full content as a single delta
        let delta = rx.recv().await.unwrap();
        assert_eq!(delta, StreamChunk::Text("fallback works".to_string()));
    }

    #[test]
    fn policy_retries_5xx_with_exponential_backoff() {
        let policy = RetryPolicy::default(); // max_retries = 2
                                             // Attempt 0: should return initial_backoff (1s)
        assert_eq!(
            policy.backoff_for(0, Some(503), false),
            Some(Duration::from_secs(1))
        );
        // Attempt 1: should return 2s (1s * 2^1)
        assert_eq!(
            policy.backoff_for(1, Some(500), false),
            Some(Duration::from_secs(2))
        );
        // Attempt 2: should return None (exceeds max_retries=2)
        assert_eq!(policy.backoff_for(2, Some(500), false), None);
    }

    #[test]
    fn policy_retries_network_errors() {
        let policy = RetryPolicy::default();
        // Network error at attempt 0 should return initial_backoff
        assert_eq!(
            policy.backoff_for(0, None, true),
            Some(Duration::from_secs(1))
        );
    }

    #[test]
    fn policy_does_not_retry_4xx() {
        let policy = RetryPolicy::default();
        // Non-rate-limit 4xx errors should not be retried
        assert_eq!(policy.backoff_for(0, Some(400), false), None);
        assert_eq!(policy.backoff_for(0, Some(401), false), None);
        assert_eq!(policy.backoff_for(0, Some(404), false), None);
        // 429 (rate limit) IS retried with backoff
        assert!(policy.backoff_for(0, Some(429), false).is_some());
    }

    #[test]
    fn policy_caps_backoff_at_max() {
        let policy = RetryPolicy {
            max_retries: 10,
            initial_backoff: Duration::from_secs(1),
            max_backoff: Duration::from_secs(3),
        };
        // At attempt 5, exponential backoff would be 32s but capped to 3s
        assert_eq!(
            policy.backoff_for(5, Some(500), false),
            Some(Duration::from_secs(3))
        );
    }

    #[tokio::test]
    async fn openai_structured_includes_schema_in_request_body() {
        // Spawn a mock server that captures the request body and returns
        // a valid JSON response. Assert the request body contains the
        // response_format block with the schema.
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        let captured = std::sync::Arc::new(std::sync::Mutex::new(String::new()));
        let captured_clone = captured.clone();

        std::thread::spawn(move || {
            let (mut stream, _) = listener.accept().unwrap();
            use std::io::{Read, Write};
            let mut buf = [0u8; 8192];
            let n = stream.read(&mut buf).unwrap();
            let request = String::from_utf8_lossy(&buf[..n]).to_string();
            // Extract the JSON body (after the blank line)
            if let Some(body_start) = request.find("\r\n\r\n") {
                let body = request[body_start + 4..].trim().to_string();
                *captured_clone.lock().unwrap() = body.clone();
            }
            let response_body = r#"{"choices":[{"message":{"role":"assistant","content":"{\"answer\":42}"},"finish_reason":"stop"}],"usage":null}"#;
            write!(
                stream,
                "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                response_body.len(),
                response_body,
            )
            .unwrap();
            stream.flush().unwrap();
        });

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        let provider =
            OpenAiProvider::new(format!("http://{addr}"), "sk-noop", "test-structured-model")
                .unwrap();

        let schema = serde_json::json!({
            "type": "object",
            "properties": {
                "answer": {"type": "integer"}
            },
            "required": ["answer"]
        });

        let req = StructuredRequest {
            messages: vec![Message::user("what is 6 times 7?".to_string())],
            schema: schema.clone(),
            schema_name: "math_answer".to_string(),
        };

        let _ = provider.complete_structured(req).await;

        // Check the captured request body
        let body_str = captured.lock().unwrap().clone();
        let body: serde_json::Value = serde_json::from_str(&body_str).unwrap();

        // Assert response_format is present with the right shape
        let rf = body.get("response_format").unwrap();
        assert_eq!(rf["type"], "json_schema");
        assert_eq!(rf["json_schema"]["name"], "math_answer");
        assert_eq!(rf["json_schema"]["strict"], true);
        assert_eq!(rf["json_schema"]["schema"], schema);
    }

    #[tokio::test]
    async fn openai_structured_parses_response_json() {
        // Spawn a mock server that returns a known JSON response.
        // Assert the parsed value matches.
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();

        std::thread::spawn(move || {
            let (mut stream, _) = listener.accept().unwrap();
            use std::io::{Read, Write};
            let mut buf = [0u8; 8192];
            let _ = stream.read(&mut buf);
            // Return a JSON object with summary and kept_facts
            let response_body = r#"{"choices":[{"message":{"role":"assistant","content":"{\"summary\":\"test\",\"kept_facts\":[\"a\",\"b\"]}"},"finish_reason":"stop"}],"usage":null}"#;
            write!(
                stream,
                "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                response_body.len(),
                response_body,
            )
            .unwrap();
            stream.flush().unwrap();
        });

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        let provider =
            OpenAiProvider::new(format!("http://{addr}"), "sk-noop", "test-structured-model")
                .unwrap();

        let schema = serde_json::json!({
            "type": "object",
            "properties": {
                "summary": {"type": "string"},
                "kept_facts": {"type": "array", "items": {"type": "string"}}
            },
            "required": ["summary", "kept_facts"]
        });

        let req = StructuredRequest {
            messages: vec![Message::user("summarize".to_string())],
            schema,
            schema_name: "summary".to_string(),
        };

        let result = provider.complete_structured(req).await;
        assert!(result.is_ok(), "got error: {:?}", result.err());
        let value = result.unwrap();
        assert_eq!(value["summary"], "test");
        assert_eq!(value["kept_facts"][0], "a");
        assert_eq!(value["kept_facts"][1], "b");
    }

    // ── serialize_message targeted tests ──────────────────────────────────────

    #[test]
    fn serialize_message_system_role() {
        // kills Role::System → "user" swap mutations
        let msg = Message::system("You are helpful.".to_string());
        let v = serialize_message(&msg);
        assert_eq!(v["role"].as_str(), Some("system"));
        assert_eq!(v["content"].as_str(), Some("You are helpful."));
    }

    #[test]
    fn serialize_message_user_role() {
        // kills Role::User → "assistant" swap mutations
        let msg = Message::user("Hello".to_string());
        let v = serialize_message(&msg);
        assert_eq!(v["role"].as_str(), Some("user"));
    }

    #[test]
    fn serialize_message_tool_role_includes_tool_call_id() {
        // kills `if let Some(id) = &m.tool_call_id` guard removal
        let msg = Message::tool_result("tc-abc".to_string(), "result text".to_string());
        let v = serialize_message(&msg);
        assert_eq!(v["role"].as_str(), Some("tool"));
        assert_eq!(v["tool_call_id"].as_str(), Some("tc-abc"));
    }

    #[test]
    fn serialize_message_assistant_includes_reasoning_content() {
        // kills `if matches!(m.role, Role::Assistant)` guard removal
        // and the reasoning_content insertion
        let mut msg = Message::assistant("answer".to_string());
        msg.reasoning_content = Some("thinking step".to_string());
        let v = serialize_message(&msg);
        assert_eq!(v["reasoning_content"].as_str(), Some("thinking step"));
    }

    #[test]
    fn serialize_message_non_assistant_omits_reasoning_content() {
        // kills removal of `if matches!(m.role, Role::Assistant)` guard —
        // without the guard, user messages would also get reasoning_content
        let mut msg = Message::user("hello".to_string());
        msg.reasoning_content = Some("should not appear".to_string());
        let v = serialize_message(&msg);
        assert!(
            v.get("reasoning_content").is_none()
                || v["reasoning_content"] == serde_json::Value::Null,
            "non-assistant messages must not include reasoning_content"
        );
    }

    #[tokio::test]
    async fn post_json_with_retry_gives_up_after_max_empty_body_retries() {
        // Kills `attempt += 1 -> -= 1` (usize underflow panic) and
        // `+= 1 -> *= 1` (attempt stuck → infinite retry → test-timeout)
        // at the HTTP-200-empty-body retry site (line 168). An always-empty
        // 200 body makes real code give up after max_retries with Err.
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        std::thread::spawn(move || {
            use std::io::{Read, Write};
            loop {
                if let Ok((mut stream, _)) = listener.accept() {
                    let mut buf = [0u8; 4096];
                    let _ = stream.read(&mut buf);
                    write!(
                        stream,
                        "HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
                    )
                    .unwrap();
                    stream.flush().unwrap();
                }
            }
        });
        tokio::time::sleep(Duration::from_millis(50)).await;

        let provider = OpenAiProvider::new(format!("http://{addr}"), "sk-noop", "m")
            .unwrap()
            .with_retry_policy(RetryPolicy {
                max_retries: 2,
                initial_backoff: Duration::from_millis(5),
                max_backoff: Duration::from_millis(10),
            });
        let result = provider
            .post_json_with_retry(
                &format!("http://{addr}/chat"),
                &serde_json::json!({}),
                "test",
            )
            .await;
        assert!(
            result.is_err(),
            "must give up after max_retries empty-body responses"
        );
    }

    #[tokio::test]
    async fn post_json_with_retry_gives_up_after_max_transient_http_retries() {
        // Kills `attempt += 1` mutants at the transient-HTTP retry site
        // (line 191): always-503 server → real code gives up with Err.
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        std::thread::spawn(move || {
            use std::io::{Read, Write};
            loop {
                if let Ok((mut stream, _)) = listener.accept() {
                    let mut buf = [0u8; 4096];
                    let _ = stream.read(&mut buf);
                    let body = r#"{"error":"boom"}"#;
                    write!(
                        stream,
                        "HTTP/1.1 503 Service Unavailable\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                        body.len(),
                        body
                    )
                    .unwrap();
                    stream.flush().unwrap();
                }
            }
        });
        tokio::time::sleep(Duration::from_millis(50)).await;

        let provider = OpenAiProvider::new(format!("http://{addr}"), "sk-noop", "m")
            .unwrap()
            .with_retry_policy(RetryPolicy {
                max_retries: 2,
                initial_backoff: Duration::from_millis(5),
                max_backoff: Duration::from_millis(10),
            });
        let result = provider
            .post_json_with_retry(
                &format!("http://{addr}/chat"),
                &serde_json::json!({}),
                "test",
            )
            .await;
        assert!(
            result.is_err(),
            "must give up after max_retries transient HTTP failures"
        );
    }

    #[tokio::test]
    async fn post_json_with_retry_gives_up_after_max_network_retries() {
        // Kills `attempt += 1 -> *= 1` at the network-error retry site
        // (line 207): server accepts and drops 3 times → real code gives up.
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        std::thread::spawn(move || {
            for _ in 0..3 {
                if let Ok((stream, _)) = listener.accept() {
                    drop(stream);
                }
            }
        });
        tokio::time::sleep(Duration::from_millis(50)).await;

        let provider = OpenAiProvider::new(format!("http://{addr}"), "sk-noop", "m")
            .unwrap()
            .with_retry_policy(RetryPolicy {
                max_retries: 2,
                initial_backoff: Duration::from_millis(5),
                max_backoff: Duration::from_millis(10),
            });
        let result = provider
            .post_json_with_retry(
                &format!("http://{addr}/chat"),
                &serde_json::json!({}),
                "test",
            )
            .await;
        assert!(
            result.is_err(),
            "must give up after max_retries network failures"
        );
    }

    #[tokio::test]
    async fn stream_inner_gives_up_after_max_transient_http_retries() {
        // Kills `attempt += 1` mutants at the stream_inner transient-HTTP
        // retry site (line 595): always-503 server → real code gives up.
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        std::thread::spawn(move || {
            use std::io::{Read, Write};
            loop {
                if let Ok((mut stream, _)) = listener.accept() {
                    let mut buf = [0u8; 4096];
                    let _ = stream.read(&mut buf);
                    let body = r#"{"error":"boom"}"#;
                    write!(
                        stream,
                        "HTTP/1.1 503 Service Unavailable\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                        body.len(),
                        body
                    )
                    .unwrap();
                    stream.flush().unwrap();
                }
            }
        });
        tokio::time::sleep(Duration::from_millis(50)).await;

        let provider = OpenAiProvider::new(format!("http://{addr}"), "sk-noop", "m")
            .unwrap()
            .with_retry_policy(RetryPolicy {
                max_retries: 2,
                initial_backoff: Duration::from_millis(5),
                max_backoff: Duration::from_millis(10),
            });
        let result = provider
            .stream_inner(&[Message::user("hi")], &[], None, None)
            .await;
        assert!(
            result.is_err(),
            "stream_inner must give up after max_retries transient HTTP failures"
        );
    }

    #[tokio::test]
    async fn stream_inner_gives_up_after_max_network_retries() {
        // Kills `attempt += 1` mutants at the stream_inner network-error
        // retry site (line 610): server accepts and drops 3 times.
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        std::thread::spawn(move || {
            for _ in 0..3 {
                if let Ok((stream, _)) = listener.accept() {
                    drop(stream);
                }
            }
        });
        tokio::time::sleep(Duration::from_millis(50)).await;

        let provider = OpenAiProvider::new(format!("http://{addr}"), "sk-noop", "m")
            .unwrap()
            .with_retry_policy(RetryPolicy {
                max_retries: 2,
                initial_backoff: Duration::from_millis(5),
                max_backoff: Duration::from_millis(10),
            });
        let result = provider
            .stream_inner(&[Message::user("hi")], &[], None, None)
            .await;
        assert!(
            result.is_err(),
            "stream_inner must give up after max_retries network failures"
        );
    }

    #[tokio::test]
    async fn stream_with_search_without_deferred_delegates_to_stream() {
        // Kills `-> Ok(Default::default())` (302:9) in stream_with_search:
        // with no deferred tools it must delegate to stream_inner and return
        // the streamed content (not an empty default).
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        std::thread::spawn(move || {
            let (mut stream, _) = listener.accept().unwrap();
            use std::io::{Read, Write};
            let mut buf = [0u8; 4096];
            let _ = stream.read(&mut buf);
            let body = "data: {\"choices\":[{\"delta\":{\"content\":\"hello search\"},\"finish_reason\":\"stop\"}]}\n\ndata: [DONE]\n\n";
            write!(
                stream,
                "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                body.len(),
                body,
            )
            .unwrap();
            stream.flush().unwrap();
        });
        tokio::time::sleep(Duration::from_millis(50)).await;

        let provider = OpenAiProvider::new(format!("http://{addr}"), "sk-noop", "m").unwrap();
        let completion = provider
            .stream_with_search(&[Message::user("hi")], &[], &[], None, None)
            .await
            .expect("stream_with_search without deferred tools should succeed");
        assert_eq!(completion.content, "hello search");
    }

    #[tokio::test]
    async fn stream_search_loop_resolves_round_trip_and_stops() {
        // Kills `-> Ok(Default::default())` (460:9), `== -> !=` (473:30) and
        // `+ -> -` / `+ -> *` (529:19) in run_stream_search_loop: a round-0
        // ToolSearchTool call must be resolved via a tool result and the
        // round-1 (non-search) completion returned. The `-` mutant underflows
        // (panic), `*` keeps round at 0 → infinite recursion (test-timeout).
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        std::thread::spawn(move || {
            use std::io::{Read, Write};
            loop {
                if let Ok((mut stream, _)) = listener.accept() {
                    let mut buf = [0u8; 65536];
                    let _ = stream.read(&mut buf);
                    let req = String::from_utf8_lossy(&buf);
                    let body = if req.contains("\"role\":\"tool\"") {
                        // Round 1+: search resolved → final text
                        "data: {\"choices\":[{\"delta\":{\"content\":\"final answer\"},\"finish_reason\":\"stop\"}]}\n\ndata: [DONE]\n\n"
                    } else {
                        // Round 0: ask for a ToolSearchTool call
                        "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"ts1\",\"function\":{\"name\":\"ToolSearchTool\",\"arguments\":\"{\\\"query\\\":\\\"x\\\"}\"}}]},\"finish_reason\":\"tool_calls\"}]}\n\ndata: [DONE]\n\n"
                    };
                    write!(
                        stream,
                        "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                        body.len(),
                        body,
                    )
                    .unwrap();
                    stream.flush().unwrap();
                }
            }
        });
        tokio::time::sleep(Duration::from_millis(50)).await;

        let deferred = vec![(make_tool_spec("MyTool"), None)];
        let provider = OpenAiProvider::new(format!("http://{addr}"), "sk-noop", "m")
            .unwrap()
            .with_search_engine(Arc::new(AlwaysReturnsEngine(vec!["MyTool".to_string()])));
        let completion = provider
            .stream_with_search(&[Message::user("hi")], &[], &deferred, None, None)
            .await
            .expect("stream search loop should complete");
        assert_eq!(completion.content, "final answer");
        assert!(
            completion.tool_calls.is_empty(),
            "round-1 completion must not carry a search call"
        );
    }

    #[tokio::test]
    async fn stream_search_loop_honors_zero_max_rounds() {
        // Kills `>= -> <` (480:18): with max_search_rounds=0 the FIRST search
        // call must be returned as-is; the `<` mutant would recurse forever
        // (round < 0 is never true → test-timeout).
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        std::thread::spawn(move || {
            use std::io::{Read, Write};
            loop {
                if let Ok((mut stream, _)) = listener.accept() {
                    let mut buf = [0u8; 65536];
                    let _ = stream.read(&mut buf);
                    let body = "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"ts1\",\"function\":{\"name\":\"ToolSearchTool\",\"arguments\":\"{\\\"query\\\":\\\"x\\\"}\"}}]},\"finish_reason\":\"tool_calls\"}]}\n\ndata: [DONE]\n\n";
                    write!(
                        stream,
                        "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                        body.len(),
                        body,
                    )
                    .unwrap();
                    stream.flush().unwrap();
                }
            }
        });
        tokio::time::sleep(Duration::from_millis(50)).await;

        let deferred = vec![(make_tool_spec("MyTool"), None)];
        let provider = OpenAiProvider::new(format!("http://{addr}"), "sk-noop", "m")
            .unwrap()
            .with_search_engine(Arc::new(AlwaysReturnsEngine(vec![])))
            .with_max_search_rounds(0);
        let completion = provider
            .stream_with_search(&[Message::user("hi")], &[], &deferred, None, None)
            .await
            .expect("round-0 search call must be returned as-is");
        assert!(
            !completion.tool_calls.is_empty(),
            "the round-0 search-call completion must be returned when max rounds is 0"
        );
    }

    #[tokio::test]
    async fn stream_search_loop_stops_after_max_rounds_with_persistent_search_calls() {
        // Kills `+ -> *` (529:19): `round * 1` keeps round at 0, so a server
        // that ALWAYS returns a ToolSearchTool call makes the mutant recurse
        // forever (test-timeout), while real code stops at max_search_rounds
        // (default 3 → 4 stream calls) and returns the last completion.
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        let call_count = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let call_count_clone = call_count.clone();
        std::thread::spawn(move || {
            use std::io::{Read, Write};
            loop {
                if let Ok((mut stream, _)) = listener.accept() {
                    call_count_clone.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                    let mut buf = [0u8; 65536];
                    let _ = stream.read(&mut buf);
                    let body = "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"ts1\",\"function\":{\"name\":\"ToolSearchTool\",\"arguments\":\"{\\\"query\\\":\\\"x\\\"}\"}}]},\"finish_reason\":\"tool_calls\"}]}\n\ndata: [DONE]\n\n";
                    write!(
                        stream,
                        "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                        body.len(),
                        body,
                    )
                    .unwrap();
                    stream.flush().unwrap();
                }
            }
        });
        tokio::time::sleep(Duration::from_millis(50)).await;

        let deferred = vec![(make_tool_spec("MyTool"), None)];
        let provider = OpenAiProvider::new(format!("http://{addr}"), "sk-noop", "m")
            .unwrap()
            .with_search_engine(Arc::new(AlwaysReturnsEngine(vec![])));
        let completion = provider
            .stream_with_search(&[Message::user("hi")], &[], &deferred, None, None)
            .await
            .expect("loop must terminate at max_search_rounds");
        assert!(
            !completion.tool_calls.is_empty(),
            "the last search-call completion must be returned when rounds are exhausted"
        );
        assert_eq!(
            call_count.load(std::sync::atomic::Ordering::SeqCst),
            4,
            "rounds 0..=3 = 4 stream calls with default max_search_rounds=3"
        );
    }

    #[tokio::test]
    async fn parse_sse_stream_returns_partial_when_only_reasoning_accumulated() {
        // Kills `have_partial` reasoning-condition mutants (delete `!` on
        // `!reasoning_content.is_empty()`, `|| -> &&` at 705): a stream that
        // carried ONLY reasoning must still return a partial Completion on a
        // mid-stream read error.
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        std::thread::spawn(move || {
            let (mut stream, _) = listener.accept().unwrap();
            use std::io::{Read, Write};
            let mut buf = [0u8; 4096];
            let _ = stream.read(&mut buf);
            let body = "data: {\"choices\":[{\"delta\":{\"reasoning_content\":\"Let me think\"},\"finish_reason\":null}]}\n\n";
            write!(
                stream,
                "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                9999,
                body,
            )
            .unwrap();
            stream.flush().unwrap();
        });
        tokio::time::sleep(Duration::from_millis(50)).await;

        let provider = OpenAiProvider::new(format!("http://{addr}"), "sk-noop", "m").unwrap();
        let completion = provider
            .stream(&[Message::user("hi")], &[], None, None)
            .await
            .expect("reasoning-only partial must be returned, not Err");
        assert_eq!(completion.finish_reason.as_deref(), Some("interrupted"));
        assert_eq!(completion.content, "");
        assert_eq!(
            completion.reasoning_content.as_deref(),
            Some("Let me think"),
            "accumulated reasoning must survive the mid-stream error"
        );
    }

    #[tokio::test]
    async fn parse_sse_stream_returns_partial_when_only_tool_call_started() {
        // Kills `have_partial` tool-call-condition mutants (delete `!` on
        // `!tool_call_builders.is_empty()`, `|| -> &&` at 705): a stream that
        // started a tool call (but no text) must still return a partial
        // Completion on a mid-stream read error, with the half-received tool
        // call dropped (orphan risk).
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        std::thread::spawn(move || {
            let (mut stream, _) = listener.accept().unwrap();
            use std::io::{Read, Write};
            let mut buf = [0u8; 4096];
            let _ = stream.read(&mut buf);
            let body = "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_1\",\"function\":{\"name\":\"Read\",\"arguments\":\"{\\\"path\\\":\\\"x\\\"}\"}}]},\"finish_reason\":null}]}\n\n";
            write!(
                stream,
                "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                9999,
                body,
            )
            .unwrap();
            stream.flush().unwrap();
        });
        tokio::time::sleep(Duration::from_millis(50)).await;

        let provider = OpenAiProvider::new(format!("http://{addr}"), "sk-noop", "m").unwrap();
        let completion = provider
            .stream(&[Message::user("hi")], &[], None, None)
            .await
            .expect("tool-call-only partial must be returned, not Err");
        assert_eq!(completion.finish_reason.as_deref(), Some("interrupted"));
        assert_eq!(completion.content, "");
        assert!(
            completion.tool_calls.is_empty(),
            "half-received tool calls must be dropped from a partial completion"
        );
    }

    #[tokio::test]
    async fn parse_sse_stream_processes_trailing_line_without_newline() {
        // Kills `delete !` on `if !line_buf.is_empty()` (769): a final SSE
        // `data:` line without a terminating newline must still be processed.
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        std::thread::spawn(move || {
            let (mut stream, _) = listener.accept().unwrap();
            use std::io::{Read, Write};
            let mut buf = [0u8; 4096];
            let _ = stream.read(&mut buf);
            // NOTE: the last delta line has NO trailing newline.
            let body = "data: {\"choices\":[{\"delta\":{\"content\":\"Hello\"},\"finish_reason\":null}]}\n\ndata: {\"choices\":[{\"delta\":{\"content\":\" world\"},\"finish_reason\":null}]}";
            write!(
                stream,
                "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                body.len(),
                body,
            )
            .unwrap();
            stream.flush().unwrap();
        });
        tokio::time::sleep(Duration::from_millis(50)).await;

        let provider = OpenAiProvider::new(format!("http://{addr}"), "sk-noop", "m").unwrap();
        let completion = provider
            .stream(&[Message::user("hi")], &[], None, None)
            .await
            .expect("stream should complete");
        assert_eq!(
            completion.content, "Hello world",
            "trailing line without newline must still be processed"
        );
    }

    #[tokio::test]
    async fn build_completion_keeps_tool_call_with_empty_id_but_name() {
        // Kills `&& -> ||` at 810 in build_completion: a tool call is only
        // DROPPED when BOTH id and name are empty. A call with an empty id
        // but a real name must be kept.
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        std::thread::spawn(move || {
            let (mut stream, _) = listener.accept().unwrap();
            use std::io::{Read, Write};
            let mut buf = [0u8; 4096];
            let _ = stream.read(&mut buf);
            // No `id` field → id stays ""; name "Read" is set.
            let body = "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"name\":\"Read\",\"arguments\":\"{\\\"path\\\":\\\"x\\\"}\"}}]},\"finish_reason\":\"tool_calls\"}]}\n\ndata: [DONE]\n\n";
            write!(
                stream,
                "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                body.len(),
                body,
            )
            .unwrap();
            stream.flush().unwrap();
        });
        tokio::time::sleep(Duration::from_millis(50)).await;

        let provider = OpenAiProvider::new(format!("http://{addr}"), "sk-noop", "m").unwrap();
        let completion = provider
            .stream(&[Message::user("hi")], &[], None, None)
            .await
            .expect("stream should complete");
        assert_eq!(
            completion.tool_calls.len(),
            1,
            "tool call with empty id but real name must be kept"
        );
        assert_eq!(completion.tool_calls[0].name, "Read");
    }

    #[test]
    fn process_sse_line_accumulates_non_empty_reasoning() {
        // Kills `delete !` on `!delta_reasoning.is_empty()` (879): non-empty
        // reasoning_content deltas must be accumulated.
        let mut content = String::new();
        let mut reasoning = String::new();
        let mut tcb = std::collections::HashMap::new();
        let mut fr = None;
        let mut usage = None;
        OpenAiProvider::process_sse_line(
            "data: {\"choices\":[{\"delta\":{\"content\":\"\",\"reasoning_content\":\"abc\"},\"finish_reason\":null}]}",
            "m",
            &mut content,
            &mut reasoning,
            &mut tcb,
            &mut fr,
            &mut usage,
            &None,
        )
        .unwrap();
        assert_eq!(reasoning, "abc");
    }

    #[test]
    fn process_sse_line_sets_tool_call_id_and_name() {
        // Kills `delete !` on `!id.is_empty()` (896) and `!name.is_empty()`
        // (902): tool_calls deltas must fill the builder's id and name.
        let mut content = String::new();
        let mut reasoning = String::new();
        let mut tcb = std::collections::HashMap::new();
        let mut fr = None;
        let mut usage = None;
        OpenAiProvider::process_sse_line(
            "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_9\",\"function\":{\"name\":\"Read\",\"arguments\":\"{}\"}}]},\"finish_reason\":null}]}",
            "m",
            &mut content,
            &mut reasoning,
            &mut tcb,
            &mut fr,
            &mut usage,
            &None,
        )
        .unwrap();
        let entry = tcb.get(&0).expect("tool call builder entry must exist");
        assert_eq!(entry.0, "call_9");
        assert_eq!(entry.1, "Read");
    }

    #[test]
    fn process_sse_line_cache_miss_prefers_explicit() {
        // Kills `> -> <` on `explicit_miss > 0` (949): an explicit non-zero
        // miss must win over the derived prompt - hit value.
        let mut content = String::new();
        let mut reasoning = String::new();
        let mut tcb = std::collections::HashMap::new();
        let mut fr = None;
        let mut usage = None;
        OpenAiProvider::process_sse_line(
            "data: {\"choices\":[{\"delta\":{\"content\":\"hi\"},\"finish_reason\":\"stop\"}],\"usage\":{\"prompt_tokens\":10,\"completion_tokens\":2,\"total_tokens\":12,\"prompt_cache_hit_tokens\":3,\"prompt_cache_miss_tokens\":5}}",
            "m",
            &mut content,
            &mut reasoning,
            &mut tcb,
            &mut fr,
            &mut usage,
            &None,
        )
        .unwrap();
        let u = usage.expect("usage must be set");
        assert_eq!(u.cache_miss_tokens, 5);
    }

    #[test]
    fn process_sse_line_cache_miss_zero_when_no_hit() {
        // Kills `> -> >=` on `cache_hit > 0` (951): with no hit and no
        // explicit miss, the miss must be 0 (not prompt - 0).
        let mut content = String::new();
        let mut reasoning = String::new();
        let mut tcb = std::collections::HashMap::new();
        let mut fr = None;
        let mut usage = None;
        OpenAiProvider::process_sse_line(
            "data: {\"choices\":[{\"delta\":{\"content\":\"hi\"},\"finish_reason\":\"stop\"}],\"usage\":{\"prompt_tokens\":10,\"completion_tokens\":2,\"total_tokens\":12,\"prompt_cache_hit_tokens\":0,\"prompt_cache_miss_tokens\":0}}",
            "m",
            &mut content,
            &mut reasoning,
            &mut tcb,
            &mut fr,
            &mut usage,
            &None,
        )
        .unwrap();
        let u = usage.expect("usage must be set");
        assert_eq!(u.cache_miss_tokens, 0);
    }

    #[test]
    fn to_token_usage_cache_miss_prefers_explicit_positive() {
        // Kills `match guard m > 0 -> false` (1138:24) and `> -> <`
        // (1138:26) in ResponseUsage::to_token_usage: an explicit positive
        // miss must be honored over deriving from prompt - hit.
        let ru = ResponseUsage {
            prompt_tokens: Some(10),
            completion_tokens: Some(2),
            total_tokens: Some(12),
            prompt_cache_hit_tokens: Some(0),
            prompt_cache_miss_tokens: Some(5),
            prompt_tokens_details: None,
            completion_tokens_details: None,
        };
        let tu = ru.to_token_usage();
        assert_eq!(tu.cache_miss_tokens, 5);
    }
}