lucy/

provider.rs

use std::collections::BTreeMap;
use std::io::{self, BufRead};
use std::time::Duration;

use reqwest::blocking::Client;
use reqwest::Client as AsyncClient;
use serde_json::{json, Value};

use crate::cancellation::CancellationToken;
use crate::codex_provider::CodexProvider;
use crate::config::LlmSettings;
use crate::model::{ChatMessage, ChatToolCall};
use crate::redaction::{conflicts_with_protected_literal, redact_secret, redaction_marker};

/// Maximum idle interval between provider response reads.
pub const PROVIDER_TIMEOUT: Duration = Duration::from_secs(60);
const PROVIDER_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
const PROVIDER_RETRY_COUNT: usize = 1;
const PROVIDER_RETRY_BACKOFF: Duration = Duration::from_millis(250);
const MAX_PROVIDER_CONTENT_BYTES: usize = 1024 * 1024;
const MAX_PROVIDER_REASONING_DETAILS_BYTES: usize = 1024 * 1024;
const MAX_PROVIDER_TOOL_ARGUMENT_BYTES: usize = 1024 * 1024;
const MAX_SSE_LINE_BYTES: usize = 64 * 1024;
const MAX_SSE_EVENT_BYTES: usize = 1024 * 1024;
const MAX_SSE_STREAM_BYTES: usize = 8 * 1024 * 1024;
const MAX_SSE_DATA_LINES: usize = 1024;
const MAX_PROVIDER_TOOL_CALL_ID_BYTES: usize = 16 * 1024;
const MAX_PROVIDER_TOOL_NAME_BYTES: usize = 16 * 1024;
const MAX_PROVIDER_ERROR_BYTES: usize = 16 * 1024;
const CANCELLATION_POLL_INTERVAL: Duration = Duration::from_millis(10);
const MODEL_METADATA_TIMEOUT: Duration = Duration::from_secs(2);
const MAX_MODEL_METADATA_BYTES: usize = 4 * 1024 * 1024;
const COMPACTION_MAX_SUMMARY_TOKENS: usize = 4_096;

#[derive(Debug)]
pub struct ProviderError {
    message: String,
    cancelled: bool,
    partial: Option<ProviderTurn>,
    retryable: bool,
}

impl ProviderError {
    pub(crate) fn new(message: impl Into<String>) -> Self {
        Self {
            message: message.into(),
            cancelled: false,
            partial: None,
            retryable: false,
        }
    }

    fn retryable(message: impl Into<String>) -> Self {
        Self {
            message: message.into(),
            cancelled: false,
            partial: None,
            retryable: true,
        }
    }

    pub(crate) fn cancelled(partial: ProviderTurn) -> Self {
        Self {
            message: "provider stream canceled".to_owned(),
            cancelled: true,
            partial: Some(partial),
            retryable: false,
        }
    }

    pub fn is_cancelled(&self) -> bool {
        self.cancelled
    }

    pub fn partial_turn(&self) -> Option<&ProviderTurn> {
        self.partial.as_ref()
    }

    fn is_retryable(&self) -> bool {
        self.retryable
    }
}

impl std::fmt::Display for ProviderError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str(&self.message)
    }
}

impl std::error::Error for ProviderError {}

fn transient_http_status(status: u16) -> bool {
    matches!(status, 408 | 429 | 500 | 502 | 503 | 504)
}

fn transient_reqwest_error(error: &reqwest::Error) -> bool {
    error.is_timeout()
        || error.is_connect()
        || error.is_request()
        || error.is_body()
        || error.is_decode()
}

fn reqwest_error_kind(error: &reqwest::Error) -> &'static str {
    if error.is_timeout() {
        "timeout"
    } else if error.is_connect() {
        "connection"
    } else if error.is_body() {
        "body"
    } else if error.is_decode() {
        "decode"
    } else if error.is_request() {
        "request"
    } else {
        "transport"
    }
}

fn reqwest_failure(
    prefix: &str,
    error: reqwest::Error,
    api_key: &str,
    retry_before_payload: bool,
) -> ProviderError {
    let detail = redact_secret(&error.to_string(), Some(api_key));
    let message = format!("{prefix} ({}): {detail}", reqwest_error_kind(&error));
    let mut provider_error = ProviderError::new(message);
    provider_error.retryable = retry_before_payload && transient_reqwest_error(&error);
    provider_error
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProviderTurn {
    pub content: String,
    pub tool_calls: Vec<ChatToolCall>,
    pub reasoning_details: Vec<Value>,
}

fn empty_turn() -> ProviderTurn {
    ProviderTurn {
        content: String::new(),
        tool_calls: Vec::new(),
        reasoning_details: Vec::new(),
    }
}

pub(crate) enum ProviderStreamEvent {
    ReasoningStarted,
    Text(String),
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProviderModel {
    pub id: String,
    pub efforts: Option<Vec<String>>,
}

pub struct Provider {
    client: Client,
    async_client: AsyncClient,
    endpoint: String,
    model: String,
    effort: Option<String>,
    api_key_env: String,
    api_key: String,
    codex: Option<CodexProvider>,
}

fn model_efforts(entry: &Value) -> Option<Vec<String>> {
    let values = entry
        .get("reasoning")
        .and_then(|reasoning| reasoning.get("supported_efforts"))
        .or_else(|| {
            [
                "supported_reasoning_efforts",
                "reasoning_efforts",
                "reasoning_effort",
                "efforts",
            ]
            .into_iter()
            .find_map(|key| entry.get(key))
        })
        .and_then(Value::as_array)?;
    let efforts = values
        .iter()
        .filter_map(Value::as_str)
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .fold(Vec::new(), |mut efforts, value| {
            if !efforts.iter().any(|effort| effort == value) {
                efforts.push(value.to_owned());
            }
            efforts
        });
    (!efforts.is_empty()).then_some(efforts)
}

fn context_window_from_models(payload: &Value, model: &str) -> Option<usize> {
    let models = payload.get("data").and_then(Value::as_array)?;
    let entry = models.iter().find(|entry| {
        entry.get("id").and_then(Value::as_str) == Some(model)
            || entry.get("name").and_then(Value::as_str) == Some(model)
    })?;
    [
        entry.get("context_length"),
        entry.get("context_window"),
        entry.get("max_context_length"),
        entry
            .get("top_provider")
            .and_then(|provider| provider.get("context_length")),
    ]
    .into_iter()
    .flatten()
    .find_map(Value::as_u64)
    .and_then(|value| usize::try_from(value).ok())
    .filter(|value| *value > 0)
}

fn chat_request(
    model: &str,
    messages: &[ChatMessage],
    effort: &Option<String>,
    include_tools: bool,
) -> Value {
    let mut request = json!({
        "model": model,
        "messages": messages
            .iter()
            .map(ChatMessage::to_openai_value)
            .collect::<Vec<_>>(),
        "stream": true,
    });
    if include_tools {
        let tools = vec![json!({
            "type": "function",
            "function": {
                "name": "cmd",
                "description": "Execute a shell command in the session starting directory. Set background to return immediately and receive the completed result automatically.",
                "parameters": {"type": "object", "properties": {"command": {"type": "string"}, "background": {"type": "boolean", "default": false}}, "required": ["command"], "additionalProperties": false}
            }
        })];
        request["tools"] = Value::Array(tools);
    } else {
        request["max_tokens"] = json!(COMPACTION_MAX_SUMMARY_TOKENS);
    }
    if let Some(effort) = effort {
        request["reasoning_effort"] = json!(effort);
    }
    request
}

impl Provider {
    pub fn new(settings: &LlmSettings) -> Result<Self, ProviderError> {
        let api_key = match std::env::var(&settings.api_key_env) {
            Ok(api_key) if !api_key.is_empty() => api_key,
            Ok(_) | Err(_) => return Err(ProviderError::new("missing provider API key")),
        };
        if conflicts_with_protected_literal(&api_key) {
            return Err(ProviderError::new(redact_secret(
                "API key conflicts with a required structured output literal",
                Some(&api_key),
            )));
        }
        if redaction_marker(&api_key).is_none() {
            return Err(ProviderError::new(redact_secret(
                "API key cannot be safely redacted",
                Some(&api_key),
            )));
        }
        if settings.model.trim().is_empty() {
            return Err(ProviderError::new(redact_secret(
                "missing llm.model; set a model in config.toml",
                Some(&api_key),
            )));
        }
        let effort = match &settings.effort {
            Some(value) => {
                let trimmed = value.trim();
                if trimmed.is_empty() {
                    return Err(ProviderError::new(redact_secret(
                        "llm.effort must not be empty",
                        Some(&api_key),
                    )));
                }
                Some(trimmed.to_owned())
            }
            None => None,
        };
        let endpoint = format!(
            "{}/chat/completions",
            settings.base_url.trim_end_matches('/')
        );
        let client = Client::builder()
            .connect_timeout(PROVIDER_CONNECT_TIMEOUT)
            .timeout(PROVIDER_TIMEOUT)
            .build()
            .map_err(|_| {
                ProviderError::new(redact_secret(
                    "unable to initialize HTTP client",
                    Some(&api_key),
                ))
            })?;
        let async_client = AsyncClient::builder()
            .connect_timeout(PROVIDER_CONNECT_TIMEOUT)
            .read_timeout(PROVIDER_TIMEOUT)
            .build()
            .map_err(|_| {
                ProviderError::new(redact_secret(
                    "unable to initialize HTTP client",
                    Some(&api_key),
                ))
            })?;
        Ok(Self {
            client,
            async_client,
            endpoint,
            model: settings.model.clone(),
            effort,
            api_key_env: settings.api_key_env.clone(),
            api_key,
            codex: None,
        })
    }

    pub fn new_codex(
        home: &std::path::Path,
        settings: &LlmSettings,
    ) -> Result<Self, ProviderError> {
        let codex = CodexProvider::new(home, settings)?;
        let api_key = codex.api_key().to_owned();
        let client = Client::builder()
            .connect_timeout(PROVIDER_CONNECT_TIMEOUT)
            .timeout(PROVIDER_TIMEOUT)
            .build()
            .map_err(|_| ProviderError::new("unable to initialize HTTP client"))?;
        let async_client = AsyncClient::builder()
            .connect_timeout(PROVIDER_CONNECT_TIMEOUT)
            .read_timeout(PROVIDER_TIMEOUT)
            .build()
            .map_err(|_| ProviderError::new("unable to initialize HTTP client"))?;
        Ok(Self {
            client,
            async_client,
            endpoint: crate::codex_provider::CODEX_ENDPOINT.to_owned(),
            model: settings.model.clone(),
            effort: settings.effort.clone(),
            api_key_env: codex.api_key_env().to_owned(),
            api_key,
            codex: Some(codex),
        })
    }

    pub fn api_key(&self) -> String {
        self.codex
            .as_ref()
            .map(CodexProvider::active_access)
            .unwrap_or_else(|| self.api_key.clone())
    }

    pub fn api_key_env(&self) -> &str {
        &self.api_key_env
    }

    pub(crate) fn models(&self) -> Result<Vec<ProviderModel>, ProviderError> {
        if let Some(codex) = &self.codex {
            return Ok(codex.models());
        }
        let base_url = self
            .endpoint
            .strip_suffix("/chat/completions")
            .ok_or_else(|| ProviderError::new("invalid provider endpoint"))?;
        let response = self
            .client
            .get(format!("{base_url}/models"))
            .bearer_auth(&self.api_key)
            .timeout(MODEL_METADATA_TIMEOUT)
            .send()
            .map_err(|_| ProviderError::new("unable to load provider models"))?;
        if !response.status().is_success() {
            return Err(ProviderError::new("unable to load provider models"));
        }
        let bytes = response
            .bytes()
            .map_err(|_| ProviderError::new("unable to load provider models"))?;
        if bytes.len() > MAX_MODEL_METADATA_BYTES {
            return Err(ProviderError::new(
                "provider model catalog exceeded the response limit",
            ));
        }
        let payload: Value = serde_json::from_slice(&bytes)
            .map_err(|_| ProviderError::new("invalid provider model catalog"))?;
        let models = payload
            .get("data")
            .and_then(Value::as_array)
            .ok_or_else(|| ProviderError::new("invalid provider model catalog"))?;
        let mut result = models
            .iter()
            .filter_map(|entry| {
                let id = entry
                    .get("id")
                    .or_else(|| entry.get("name"))
                    .and_then(Value::as_str)?
                    .trim();
                if id.is_empty() {
                    return None;
                }
                let efforts = model_efforts(entry);
                Some(ProviderModel {
                    id: id.to_owned(),
                    efforts,
                })
            })
            .collect::<Vec<_>>();
        result.sort_by(|left, right| left.id.cmp(&right.id));
        result.dedup_by(|left, right| left.id == right.id);
        Ok(result)
    }

    /// Query the OpenAI-compatible model catalog for the configured model's
    /// context window. Providers that do not expose context metadata simply
    /// return `None`; this lookup is only used by the interactive statusline.
    pub(crate) fn context_window(&self) -> Option<usize> {
        if let Some(codex) = &self.codex {
            return codex.context_window();
        }
        let base_url = self.endpoint.strip_suffix("/chat/completions")?;
        let response = self
            .client
            .get(format!("{base_url}/models"))
            .bearer_auth(&self.api_key)
            .timeout(MODEL_METADATA_TIMEOUT)
            .send()
            .ok()?;
        if !response.status().is_success() {
            return None;
        }
        let bytes = response.bytes().ok()?;
        if bytes.len() > MAX_MODEL_METADATA_BYTES {
            return None;
        }
        let payload: Value = serde_json::from_slice(&bytes).ok()?;
        context_window_from_models(&payload, &self.model)
    }

    pub fn stream_chat(
        &self,
        messages: &[ChatMessage],
        on_text: &mut dyn FnMut(&str) -> io::Result<()>,
    ) -> Result<ProviderTurn, ProviderError> {
        let cancellation = CancellationToken::new();
        self.stream_chat_cancellable_with_options(messages, on_text, &cancellation, true)
    }

    /// Generate an internal compaction summary without exposing `cmd` to the
    /// summarization request or emitting its text as a normal assistant delta.
    pub(crate) fn summarize(
        &self,
        messages: &[ChatMessage],
        cancellation: &CancellationToken,
    ) -> Result<String, ProviderError> {
        let mut ignored = |_text: &str| Ok(());
        let turn =
            self.stream_chat_cancellable_with_options(messages, &mut ignored, cancellation, false)?;
        if !turn.tool_calls.is_empty() {
            return Err(ProviderError::new(
                "compaction summary requested an unsupported tool",
            ));
        }
        if turn.content.trim().is_empty() {
            return Err(ProviderError::new("compaction summary was empty"));
        }
        Ok(turn.content)
    }

    /// Stream through an async response so cancellation can drop the pending
    /// socket read instead of waiting for the blocking client's timeout.
    #[allow(dead_code)]
    pub(crate) fn stream_chat_cancellable(
        &self,
        messages: &[ChatMessage],
        on_text: &mut dyn FnMut(&str) -> io::Result<()>,
        cancellation: &CancellationToken,
    ) -> Result<ProviderTurn, ProviderError> {
        self.stream_chat_cancellable_with_options(messages, on_text, cancellation, true)
    }

    pub(crate) fn stream_chat_cancellable_with_options(
        &self,
        messages: &[ChatMessage],
        on_text: &mut dyn FnMut(&str) -> io::Result<()>,
        cancellation: &CancellationToken,
        include_tools: bool,
    ) -> Result<ProviderTurn, ProviderError> {
        let mut on_event = |event| match event {
            ProviderStreamEvent::ReasoningStarted => Ok(()),
            ProviderStreamEvent::Text(text) => on_text(&text),
        };
        self.stream_chat_cancellable_with_options_and_events(
            messages,
            &mut on_event,
            cancellation,
            include_tools,
        )
    }

    pub(crate) fn stream_chat_cancellable_with_options_and_events(
        &self,
        messages: &[ChatMessage],
        on_event: &mut dyn FnMut(ProviderStreamEvent) -> io::Result<()>,
        cancellation: &CancellationToken,
        include_tools: bool,
    ) -> Result<ProviderTurn, ProviderError> {
        if let Some(codex) = &self.codex {
            let mut on_event = |event| on_event(event);
            return codex.stream_chat(messages, &mut on_event, cancellation, include_tools);
        }
        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .map_err(|_| ProviderError::new("unable to initialize provider runtime"))?;
        runtime.block_on(self.stream_chat_async_with_retries(
            messages,
            on_event,
            cancellation,
            include_tools,
        ))
    }

    async fn stream_chat_async_with_retries(
        &self,
        messages: &[ChatMessage],
        on_event: &mut dyn FnMut(ProviderStreamEvent) -> io::Result<()>,
        cancellation: &CancellationToken,
        include_tools: bool,
    ) -> Result<ProviderTurn, ProviderError> {
        for attempt in 0..=PROVIDER_RETRY_COUNT {
            match self
                .stream_chat_async_once(messages, on_event, cancellation, include_tools)
                .await
            {
                Err(error) if error.is_retryable() && attempt < PROVIDER_RETRY_COUNT => {
                    if cancellation.is_cancelled() {
                        return Err(ProviderError::cancelled(empty_turn()));
                    }
                    tokio::time::sleep(PROVIDER_RETRY_BACKOFF).await;
                }
                result => return result,
            }
        }
        unreachable!("provider retry loop must return");
    }

    async fn stream_chat_async_once(
        &self,
        messages: &[ChatMessage],
        on_event: &mut dyn FnMut(ProviderStreamEvent) -> io::Result<()>,
        cancellation: &CancellationToken,
        include_tools: bool,
    ) -> Result<ProviderTurn, ProviderError> {
        if cancellation.is_cancelled() {
            return Err(ProviderError::cancelled(ProviderTurn {
                content: String::new(),
                tool_calls: Vec::new(),
                reasoning_details: Vec::new(),
            }));
        }
        let request = chat_request(&self.model, messages, &self.effort, include_tools);
        let request = self
            .async_client
            .post(&self.endpoint)
            .bearer_auth(&self.api_key)
            .header("accept", "text/event-stream")
            .json(&request)
            .send();
        let mut request = Box::pin(request);
        let mut response = loop {
            if cancellation.is_cancelled() {
                return Err(ProviderError::cancelled(ProviderTurn {
                    content: String::new(),
                    tool_calls: Vec::new(),
                    reasoning_details: Vec::new(),
                }));
            }
            match tokio::time::timeout(CANCELLATION_POLL_INTERVAL, request.as_mut()).await {
                Ok(response) => {
                    break response.map_err(|error| {
                        reqwest_failure("provider request failed", error, &self.api_key, true)
                    })?;
                }
                Err(_) => continue,
            }
        };
        if !response.status().is_success() {
            let status = response.status().as_u16();
            let error = if transient_http_status(status) {
                ProviderError::retryable(format!("provider returned HTTP status {status}"))
            } else {
                ProviderError::new(format!("provider returned HTTP status {status}"))
            };
            return Err(error);
        }

        let mut accumulator = ProviderAccumulator::default();
        let mut decoder = SseDecoder::default();
        loop {
            if cancellation.is_cancelled() {
                return Err(ProviderError::cancelled(accumulator.partial_turn()));
            }
            let chunk =
                match tokio::time::timeout(CANCELLATION_POLL_INTERVAL, response.chunk()).await {
                    Ok(chunk) => {
                        let retry_before_payload = !decoder.result.received_payload;
                        chunk.map_err(|error| {
                            reqwest_failure(
                                "provider stream read failed",
                                error,
                                &self.api_key,
                                retry_before_payload,
                            )
                        })?
                    }
                    Err(_) => continue,
                };
            let Some(chunk) = chunk else {
                break;
            };
            let done = decoder.feed(&chunk, &mut |data| {
                accumulator.on_data(data, &self.api_key, on_event)
            })?;
            if done {
                break;
            }
        }
        if cancellation.is_cancelled() {
            return Err(ProviderError::cancelled(accumulator.partial_turn()));
        }
        let parse_result =
            decoder.finish(&mut |data| accumulator.on_data(data, &self.api_key, on_event))?;
        if !parse_result.received_payload {
            return Err(ProviderError::new(
                "provider stream contained no valid payload",
            ));
        }
        if !parse_result.received_done {
            return Err(ProviderError::new("provider stream ended before [DONE]"));
        }
        accumulator.finish()
    }
}

#[derive(Debug, Clone, Default)]
struct PartialToolCall {
    id: String,
    name: String,
    arguments: String,
}

#[derive(Debug, Default)]
struct ProviderAccumulator {
    content: String,
    tool_calls: BTreeMap<usize, PartialToolCall>,
    reasoning_details: Vec<Value>,
    reasoning_details_bytes: usize,
    tool_argument_bytes: usize,
    finish_reason: Option<String>,
    reasoning_started: bool,
}

impl ProviderAccumulator {
    fn on_data(
        &mut self,
        data: Value,
        api_key: &str,
        on_event: &mut dyn FnMut(ProviderStreamEvent) -> io::Result<()>,
    ) -> Result<(), ProviderError> {
        if let Some(message) = provider_error_message(&data) {
            return Err(ProviderError::new(format!(
                "provider stream error: {}",
                redact_secret(message, Some(api_key))
            )));
        }
        let Some(choice) = data
            .get("choices")
            .and_then(Value::as_array)
            .and_then(|choices| choices.first())
        else {
            return Ok(());
        };
        if let Some(reason) = validate_finish_reason(choice)? {
            self.finish_reason = Some(reason.to_owned());
        }
        let Some(delta) = choice.get("delta") else {
            return Ok(());
        };
        let received_reasoning = append_reasoning_details(
            &mut self.reasoning_details,
            &mut self.reasoning_details_bytes,
            delta,
        )?;
        if received_reasoning && !self.reasoning_started {
            self.reasoning_started = true;
            on_event(ProviderStreamEvent::ReasoningStarted)
                .map_err(|_| ProviderError::new("unable to emit reasoning state"))?;
        }
        if let Some(text) = delta.get("content").and_then(Value::as_str) {
            if self.content.len().saturating_add(text.len()) > MAX_PROVIDER_CONTENT_BYTES {
                return Err(ProviderError::new(
                    "provider assistant content exceeded the response limit",
                ));
            }
            self.content.push_str(text);
            on_event(ProviderStreamEvent::Text(text.to_owned()))
                .map_err(|_| ProviderError::new("unable to emit assistant delta"))?;
        }
        if let Some(calls) = delta.get("tool_calls").and_then(Value::as_array) {
            for (position, call) in calls.iter().enumerate() {
                let index = call
                    .get("index")
                    .and_then(Value::as_u64)
                    .map_or(position, |index| index as usize);
                let partial = self.tool_calls.entry(index).or_default();
                if let Some(id) = call.get("id").and_then(Value::as_str) {
                    append_provider_field(
                        &mut partial.id,
                        id,
                        MAX_PROVIDER_TOOL_CALL_ID_BYTES,
                        "provider tool-call id exceeded the response limit",
                    )?;
                }
                if let Some(function) = call.get("function") {
                    if let Some(name) = function.get("name").and_then(Value::as_str) {
                        append_provider_field(
                            &mut partial.name,
                            name,
                            MAX_PROVIDER_TOOL_NAME_BYTES,
                            "provider tool-call name exceeded the response limit",
                        )?;
                    }
                    if let Some(arguments) = function.get("arguments").and_then(Value::as_str) {
                        if self.tool_argument_bytes.saturating_add(arguments.len())
                            > MAX_PROVIDER_TOOL_ARGUMENT_BYTES
                        {
                            return Err(ProviderError::new(
                                "provider tool arguments exceeded the response limit",
                            ));
                        }
                        self.tool_argument_bytes += arguments.len();
                        partial.arguments.push_str(arguments);
                    }
                }
            }
        }
        if let Some(function_call) = delta.get("function_call") {
            let partial = self.tool_calls.entry(0).or_default();
            if let Some(name) = function_call.get("name").and_then(Value::as_str) {
                append_provider_field(
                    &mut partial.name,
                    name,
                    MAX_PROVIDER_TOOL_NAME_BYTES,
                    "provider tool-call name exceeded the response limit",
                )?;
            }
            if let Some(arguments) = function_call.get("arguments").and_then(Value::as_str) {
                if self.tool_argument_bytes.saturating_add(arguments.len())
                    > MAX_PROVIDER_TOOL_ARGUMENT_BYTES
                {
                    return Err(ProviderError::new(
                        "provider tool arguments exceeded the response limit",
                    ));
                }
                self.tool_argument_bytes += arguments.len();
                partial.arguments.push_str(arguments);
            }
        }
        Ok(())
    }

    fn partial_turn(&self) -> ProviderTurn {
        ProviderTurn {
            content: self.content.clone(),
            tool_calls: self
                .tool_calls
                .iter()
                .map(|(index, partial)| ChatToolCall {
                    id: if partial.id.is_empty() {
                        format!("call_{index}")
                    } else {
                        partial.id.clone()
                    },
                    name: partial.name.clone(),
                    arguments: partial.arguments.clone(),
                })
                .collect(),
            reasoning_details: self.reasoning_details.clone(),
        }
    }

    fn finish(self) -> Result<ProviderTurn, ProviderError> {
        let tool_calls = self
            .tool_calls
            .into_iter()
            .map(|(index, partial)| ChatToolCall {
                id: if partial.id.is_empty() {
                    format!("call_{index}")
                } else {
                    partial.id
                },
                name: partial.name,
                arguments: partial.arguments,
            })
            .collect::<Vec<_>>();
        if let Some(reason) = self.finish_reason.as_deref() {
            if !tool_calls.is_empty() && !matches!(reason, "tool_calls" | "function_call") {
                return Err(ProviderError::new(
                    "provider tool calls ended with an incompatible finish reason",
                ));
            }
            if tool_calls.is_empty() && matches!(reason, "tool_calls" | "function_call") {
                return Err(ProviderError::new(
                    "provider reported tool completion without a tool call",
                ));
            }
        }
        if self.content.is_empty() && tool_calls.is_empty() {
            return Err(ProviderError::new(
                "provider stream contained no assistant content or tool calls",
            ));
        }
        Ok(ProviderTurn {
            content: self.content,
            tool_calls,
            reasoning_details: self.reasoning_details,
        })
    }
}

fn append_reasoning_details(
    target: &mut Vec<Value>,
    serialized_bytes: &mut usize,
    delta: &Value,
) -> Result<bool, ProviderError> {
    let Some(details) = delta.get("reasoning_details").and_then(Value::as_array) else {
        return Ok(false);
    };
    if details.is_empty() {
        return Ok(false);
    }
    let serialized_delta = serde_json::to_vec(details)
        .map_err(|_| ProviderError::new("provider reasoning details could not be serialized"))?;
    let combined_bytes = if target.is_empty() {
        serialized_delta.len()
    } else {
        serialized_bytes
            .saturating_add(serialized_delta.len())
            .saturating_sub(1)
    };
    if combined_bytes > MAX_PROVIDER_REASONING_DETAILS_BYTES {
        return Err(ProviderError::new(
            "provider reasoning details exceeded the response limit",
        ));
    }
    target.extend(details.iter().cloned());
    *serialized_bytes = combined_bytes;
    Ok(true)
}

fn append_provider_field(
    target: &mut String,
    fragment: &str,
    limit: usize,
    error_message: &str,
) -> Result<(), ProviderError> {
    if target.len().saturating_add(fragment.len()) > limit {
        return Err(ProviderError::new(error_message));
    }
    target.push_str(fragment);
    Ok(())
}

fn provider_error_message(data: &Value) -> Option<&str> {
    let error = data.get("error")?;
    let message = if let Some(message) = error.get("message").and_then(Value::as_str) {
        message
    } else if let Some(message) = error.as_str() {
        message
    } else {
        return Some("provider returned an error payload");
    };
    if message.len() > MAX_PROVIDER_ERROR_BYTES {
        Some("provider error text exceeded the response limit")
    } else {
        Some(message)
    }
}

#[derive(Debug, Default)]
struct SseDecoder {
    line: Vec<u8>,
    data_lines: Vec<String>,
    data_event_bytes: usize,
    stream_bytes: usize,
    result: SseParseResult,
    done: bool,
}

impl SseDecoder {
    fn feed<F>(&mut self, bytes: &[u8], on_data: &mut F) -> Result<bool, ProviderError>
    where
        F: FnMut(Value) -> Result<(), ProviderError>,
    {
        if self.stream_bytes.saturating_add(bytes.len()) > MAX_SSE_STREAM_BYTES {
            return Err(ProviderError::new(
                "provider SSE stream exceeded the response limit",
            ));
        }
        self.stream_bytes += bytes.len();
        for byte in bytes {
            if self.done {
                break;
            }
            if *byte == b'\n' {
                let line = std::mem::take(&mut self.line);
                if self.process_line(&line, on_data)? {
                    return Ok(true);
                }
            } else {
                self.line.push(*byte);
                if self.line.len() > MAX_SSE_LINE_BYTES {
                    return Err(ProviderError::new(
                        "provider SSE line exceeded the response limit",
                    ));
                }
            }
        }
        Ok(self.done)
    }

    fn finish<F>(&mut self, on_data: &mut F) -> Result<SseParseResult, ProviderError>
    where
        F: FnMut(Value) -> Result<(), ProviderError>,
    {
        if !self.line.is_empty() && !self.done {
            let line = std::mem::take(&mut self.line);
            self.process_line(&line, on_data)?;
        }
        if !self.done {
            self.dispatch_data(on_data)?;
        }
        Ok(self.result)
    }

    fn process_line<F>(&mut self, raw_line: &[u8], on_data: &mut F) -> Result<bool, ProviderError>
    where
        F: FnMut(Value) -> Result<(), ProviderError>,
    {
        let line = std::str::from_utf8(raw_line)
            .map_err(|_| ProviderError::new("provider stream contained invalid UTF-8"))?
            .trim_end_matches('\r');
        if line.is_empty() {
            return self.dispatch_data(on_data);
        }
        if line.starts_with(':') {
            return Ok(false);
        }
        let (field, value) = line
            .split_once(':')
            .map_or((line, ""), |(field, value)| (field, value));
        if field == "data" {
            let value = value.strip_prefix(' ').unwrap_or(value);
            let separator_bytes = (!self.data_lines.is_empty()) as usize;
            let added_bytes = separator_bytes.saturating_add(value.len());
            if self.data_event_bytes.saturating_add(added_bytes) > MAX_SSE_EVENT_BYTES {
                return Err(ProviderError::new(
                    "provider SSE data event exceeded the response limit",
                ));
            }
            if self.data_lines.len() >= MAX_SSE_DATA_LINES {
                return Err(ProviderError::new(
                    "provider SSE data line count exceeded the response limit",
                ));
            }
            self.data_event_bytes += added_bytes;
            self.data_lines.push(value.to_owned());
        }
        Ok(false)
    }

    fn dispatch_data<F>(&mut self, on_data: &mut F) -> Result<bool, ProviderError>
    where
        F: FnMut(Value) -> Result<(), ProviderError>,
    {
        if self.data_lines.is_empty() {
            self.data_event_bytes = 0;
            return Ok(false);
        }
        let data = self.data_lines.join("\n");
        self.data_lines.clear();
        self.data_event_bytes = 0;
        if data.trim().is_empty() {
            return Ok(false);
        }
        if data == "[DONE]" {
            self.result.received_done = true;
            self.done = true;
            return Ok(true);
        }
        let value: Value = serde_json::from_str(&data)
            .map_err(|_| ProviderError::new("provider sent malformed SSE data"))?;
        self.result.received_payload = true;
        on_data(value)?;
        Ok(false)
    }
}

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct SseParseResult {
    pub received_payload: bool,
    pub received_done: bool,
}

pub fn parse_sse<R, F>(reader: &mut R, mut on_data: F) -> Result<SseParseResult, ProviderError>
where
    R: BufRead,
    F: FnMut(Value) -> Result<(), ProviderError>,
{
    let mut data_lines = Vec::new();
    let mut data_event_bytes = 0;
    let mut stream_bytes: usize = 0;
    let mut result = SseParseResult::default();
    let mut line = Vec::with_capacity(MAX_SSE_LINE_BYTES);
    loop {
        let (has_line, line_bytes) = match read_sse_line(reader, &mut line) {
            Ok(result) => result,
            Err(mut error) => {
                if result.received_payload {
                    error.retryable = false;
                }
                return Err(error);
            }
        };
        if stream_bytes.saturating_add(line_bytes) > MAX_SSE_STREAM_BYTES {
            return Err(ProviderError::new(
                "provider SSE stream exceeded the response limit",
            ));
        }
        stream_bytes += line_bytes;
        if !has_line {
            if !data_lines.is_empty() {
                dispatch_data(
                    &mut data_lines,
                    &mut data_event_bytes,
                    &mut on_data,
                    &mut result,
                )?;
            }
            return Ok(result);
        }

        let line = std::str::from_utf8(&line)
            .map_err(|_| ProviderError::new("provider stream contained invalid UTF-8"))?
            .trim_end_matches('\r');
        if line.is_empty() {
            if dispatch_data(
                &mut data_lines,
                &mut data_event_bytes,
                &mut on_data,
                &mut result,
            )? {
                return Ok(result);
            }
            continue;
        }
        if line.starts_with(':') {
            continue;
        }
        let (field, value) = line
            .split_once(':')
            .map_or((line, ""), |(field, value)| (field, value));
        if field == "data" {
            let value = value.strip_prefix(' ').unwrap_or(value);
            let separator_bytes = (!data_lines.is_empty()) as usize;
            let added_bytes = separator_bytes.saturating_add(value.len());
            if data_event_bytes.saturating_add(added_bytes) > MAX_SSE_EVENT_BYTES {
                return Err(ProviderError::new(
                    "provider SSE data event exceeded the response limit",
                ));
            }
            if data_lines.len() >= MAX_SSE_DATA_LINES {
                return Err(ProviderError::new(
                    "provider SSE data line count exceeded the response limit",
                ));
            }
            data_event_bytes += added_bytes;
            data_lines.push(value.to_owned());
        }
    }
}

fn read_sse_line<R: BufRead>(
    reader: &mut R,
    line: &mut Vec<u8>,
) -> Result<(bool, usize), ProviderError> {
    line.clear();
    let mut consumed_bytes = 0;
    loop {
        let buffer = reader.fill_buf().map_err(|error| {
            ProviderError::retryable(format!("provider stream read failed: {error}"))
        })?;
        if buffer.is_empty() {
            return Ok((!line.is_empty(), consumed_bytes));
        }

        let newline = buffer.iter().position(|byte| *byte == b'\n');
        let chunk_length = newline.unwrap_or(buffer.len());
        if line.len().saturating_add(chunk_length) > MAX_SSE_LINE_BYTES {
            return Err(ProviderError::new(
                "provider SSE line exceeded the response limit",
            ));
        }
        line.extend_from_slice(&buffer[..chunk_length]);
        let consumed = newline.map_or(chunk_length, |index| index + 1);
        reader.consume(consumed);
        consumed_bytes += consumed;
        if newline.is_some() {
            return Ok((true, consumed_bytes));
        }
    }
}

fn dispatch_data<F>(
    data_lines: &mut Vec<String>,
    data_event_bytes: &mut usize,
    on_data: &mut F,
    result: &mut SseParseResult,
) -> Result<bool, ProviderError>
where
    F: FnMut(Value) -> Result<(), ProviderError>,
{
    if data_lines.is_empty() {
        *data_event_bytes = 0;
        return Ok(false);
    }
    let data = data_lines.join("\n");
    data_lines.clear();
    *data_event_bytes = 0;
    if data.trim().is_empty() {
        return Ok(false);
    }
    if data == "[DONE]" {
        result.received_done = true;
        return Ok(true);
    }
    let value: Value = serde_json::from_str(&data)
        .map_err(|_| ProviderError::new("provider sent malformed SSE data"))?;
    result.received_payload = true;
    on_data(value)?;
    Ok(false)
}

fn validate_finish_reason(choice: &Value) -> Result<Option<&str>, ProviderError> {
    let Some(reason) = choice.get("finish_reason") else {
        return Ok(None);
    };
    if reason.is_null() {
        return Ok(None);
    }
    match reason.as_str() {
        Some("stop") | Some("tool_calls") | Some("function_call") => Ok(reason.as_str()),
        Some("length") | Some("content_filter") => Err(ProviderError::new(
            "provider response ended before completion",
        )),
        Some(_) | None => Err(ProviderError::new(
            "provider response has an unsupported finish reason",
        )),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::{BufRead, BufReader, Cursor, Read, Write};
    use std::net::{TcpListener, TcpStream};
    use std::sync::mpsc;
    use std::thread;
    use std::time::{Duration, Instant};

    #[test]
    fn retry_policy_only_marks_transient_http_statuses() {
        for status in [408, 429, 500, 502, 503, 504] {
            assert!(transient_http_status(status));
        }
        for status in [400, 401, 403, 404, 422] {
            assert!(!transient_http_status(status));
        }
    }

    #[test]
    fn normal_requests_expose_only_cmd() {
        let request = chat_request(
            "model",
            &[ChatMessage::user("hello".to_owned())],
            &None,
            true,
        );
        let tools = request["tools"].as_array().expect("model tools");
        assert_eq!(tools.len(), 1);
        assert_eq!(tools[0]["function"]["name"], "cmd");
        let background = &tools[0]["function"]["parameters"]["properties"]["background"];
        assert_eq!(background["type"], "boolean");
        assert_eq!(background["default"], false);
    }

    #[test]
    fn compaction_request_does_not_include_tools() {
        let normal = chat_request(
            "model",
            &[ChatMessage::user("hello".to_owned())],
            &None,
            true,
        );
        let compact = chat_request(
            "model",
            &[ChatMessage::user("hello".to_owned())],
            &None,
            false,
        );

        assert!(normal.get("tools").is_some());
        assert!(normal.get("max_tokens").is_none());
        assert!(compact.get("tools").is_none());
        assert_eq!(compact["max_tokens"], COMPACTION_MAX_SUMMARY_TOKENS);
    }

    #[test]
    fn model_catalog_reads_nested_and_compatible_effort_metadata() {
        let openrouter = serde_json::json!({
            "reasoning": {
                "supported_efforts": ["max", "xhigh", "high", "medium", "low", "none"]
            }
        });
        assert_eq!(
            model_efforts(&openrouter),
            Some(vec![
                "max".to_owned(),
                "xhigh".to_owned(),
                "high".to_owned(),
                "medium".to_owned(),
                "low".to_owned(),
                "none".to_owned(),
            ])
        );

        let compatible = serde_json::json!({
            "supported_reasoning_efforts": ["light", "medium", "max", "light", ""]
        });
        assert_eq!(
            model_efforts(&compatible),
            Some(vec![
                "light".to_owned(),
                "medium".to_owned(),
                "max".to_owned()
            ])
        );
        assert_eq!(model_efforts(&serde_json::json!({})), None);
    }

    #[test]
    fn model_catalog_context_window_matches_configured_model() {
        let payload = serde_json::json!({
            "data": [
                {"id": "other", "context_length": 8_000},
                {"id": "provider/model", "context_length": 128_000}
            ]
        });

        assert_eq!(
            context_window_from_models(&payload, "provider/model"),
            Some(128_000)
        );
        assert_eq!(context_window_from_models(&payload, "missing"), None);
    }

    #[test]
    fn model_catalog_context_window_accepts_provider_fallback_fields() {
        let payload = serde_json::json!({
            "data": [{
                "id": "provider/model",
                "top_provider": {"context_length": 64_000}
            }]
        });

        assert_eq!(
            context_window_from_models(&payload, "provider/model"),
            Some(64_000)
        );
    }

    #[test]
    fn parses_sse_comments_multiline_data_and_done() {
        let stream = b": keep-alive\n\ndata: \n\ndata: {\"choices\":[]\ndata: }\n\ndata: [DONE]\n";
        let mut values = Vec::new();
        let result = parse_sse(&mut Cursor::new(stream), |value| {
            values.push(value);
            Ok(())
        })
        .expect("SSE");
        assert!(result.received_payload);
        assert!(result.received_done);
        assert_eq!(values.len(), 1);
        assert!(values[0]["choices"].is_array());
    }

    #[test]
    fn parses_text_and_fragmented_tool_calls() {
        let first = serde_json::json!({
            "choices": [{"delta": {"content": "hi"}}]
        });
        let second = serde_json::json!({
            "choices": [{
                "delta": {
                    "tool_calls": [{
                        "index": 0,
                        "id": "c1",
                        "function": {"name": "cmd", "arguments": "{command:"}
                    }]
                }
            }]
        });
        let third = serde_json::json!({
            "choices": [{
                "delta": {
                    "tool_calls": [{
                        "index": 0,
                        "function": {"arguments": "pwd}"}
                    }]
                }
            }]
        });
        let stream =
            format!("data: {first}\n\n data: {second}\n\ndata: {third}\n\ndata: [DONE]\n\n")
                .replace(" data:", "data:");
        let mut content = String::new();
        let mut calls = BTreeMap::<usize, PartialToolCall>::new();
        let result = parse_sse(&mut Cursor::new(stream.as_bytes()), |value| {
            let choice = &value["choices"][0];
            let delta = &choice["delta"];
            if let Some(text) = delta["content"].as_str() {
                content.push_str(text);
            }
            if let Some(tool_calls) = delta["tool_calls"].as_array() {
                for call in tool_calls {
                    let index = call["index"].as_u64().expect("index") as usize;
                    let partial = calls.entry(index).or_default();
                    partial.id.push_str(call["id"].as_str().unwrap_or(""));
                    partial
                        .name
                        .push_str(call["function"]["name"].as_str().unwrap_or(""));
                    partial
                        .arguments
                        .push_str(call["function"]["arguments"].as_str().unwrap_or(""));
                }
            }
            Ok(())
        })
        .expect("SSE");
        assert!(result.received_payload);
        assert!(result.received_done);
        assert_eq!(content, "hi");
        assert_eq!(calls[&0].id, "c1");
        assert_eq!(calls[&0].name, "cmd");
        assert_eq!(calls[&0].arguments, "{command:pwd}");
    }

    #[test]
    fn cancellable_accumulator_accepts_more_than_sixty_four_tool_calls() {
        let mut accumulator = ProviderAccumulator::default();
        let tool_calls = (0..65)
            .map(|index| {
                serde_json::json!({
                    "index": index,
                    "id": format!("call-{index}"),
                    "function": {
                        "name": "cmd",
                        "arguments": "{\"command\":\"true\"}"
                    }
                })
            })
            .collect::<Vec<_>>();
        accumulator
            .on_data(
                serde_json::json!({
                    "choices": [{
                        "delta": {"tool_calls": tool_calls},
                        "finish_reason": "tool_calls"
                    }]
                }),
                "provider-secret",
                &mut |_| Ok(()),
            )
            .expect("tool-call chunk");

        let turn = accumulator.finish().expect("provider turn");
        assert_eq!(turn.tool_calls.len(), 65);
    }

    #[test]
    fn reasoning_stream_event_is_emitted_once_before_assistant_text() {
        let mut accumulator = ProviderAccumulator::default();
        let mut events = Vec::new();
        let mut on_event = |event| {
            match event {
                ProviderStreamEvent::ReasoningStarted => events.push("started".to_owned()),
                ProviderStreamEvent::Text(text) => events.push(text),
            }
            Ok(())
        };

        accumulator
            .on_data(
                serde_json::json!({
                    "choices": [{
                        "delta": {
                            "reasoning_details": [{"type": "reasoning.text", "text": "thinking"}]
                        }
                    }]
                }),
                "provider-secret",
                &mut on_event,
            )
            .expect("reasoning chunk");
        accumulator
            .on_data(
                serde_json::json!({
                    "choices": [{"delta": {"content": "answer"}}]
                }),
                "provider-secret",
                &mut on_event,
            )
            .expect("answer chunk");

        assert_eq!(events, vec!["started".to_owned(), "answer".to_owned()]);
    }

    #[test]
    fn accumulates_reasoning_details_with_fragmented_tool_calls() {
        let mut accumulator = ProviderAccumulator::default();
        accumulator
            .on_data(
                serde_json::json!({
                    "choices": [{
                        "delta": {
                            "reasoning_details": [{
                                "type": "reasoning.text",
                                "text": "part one"
                            }]
                        }
                    }]
                }),
                "provider-secret",
                &mut |_| Ok(()),
            )
            .expect("first provider chunk");
        accumulator
            .on_data(
                serde_json::json!({
                    "choices": [{
                        "delta": {
                            "reasoning_details": [{
                                "type": "reasoning.text",
                                "text": "part two"
                            }],
                            "tool_calls": [{
                                "index": 0,
                                "id": "call-1",
                                "function": {
                                    "name": "cmd",
                                    "arguments": "{\"command\":\"true\"}"
                                }
                            }]
                        },
                        "finish_reason": "tool_calls"
                    }]
                }),
                "provider-secret",
                &mut |_| Ok(()),
            )
            .expect("second provider chunk");

        let partial = accumulator.partial_turn();
        assert_eq!(partial.reasoning_details.len(), 2);
        let turn = accumulator.finish().expect("provider turn");
        assert_eq!(
            turn.reasoning_details,
            vec![
                json!({"type": "reasoning.text", "text": "part one"}),
                json!({"type": "reasoning.text", "text": "part two"}),
            ]
        );
        assert_eq!(turn.tool_calls.len(), 1);
        assert_eq!(turn.tool_calls[0].name, "cmd");
    }

    #[test]
    fn accumulates_many_small_reasoning_details_and_rejects_overflow_atomically() {
        const FRAGMENT_COUNT: usize = 4096;
        let mut details = Vec::new();
        let mut serialized_bytes = 0;
        let delta = serde_json::json!({
            "reasoning_details": [{
                "type": "reasoning.text",
                "text": "x".repeat(64)
            }]
        });
        for _ in 0..FRAGMENT_COUNT {
            append_reasoning_details(&mut details, &mut serialized_bytes, &delta)
                .expect("small reasoning detail");
        }
        assert_eq!(details.len(), FRAGMENT_COUNT);

        let first_chunk_delta = serde_json::json!({
            "reasoning_details": [{
                "type": "reasoning.text",
                "text": "x".repeat(500 * 1024)
            }]
        });
        let first_chunk_bytes = serde_json::to_vec(&first_chunk_delta["reasoning_details"])
            .expect("first reasoning detail chunk")
            .len();
        assert!(first_chunk_bytes < MAX_PROVIDER_REASONING_DETAILS_BYTES);
        append_reasoning_details(&mut details, &mut serialized_bytes, &first_chunk_delta)
            .expect("first individually bounded reasoning detail chunk");
        assert!(serialized_bytes <= MAX_PROVIDER_REASONING_DETAILS_BYTES);

        let second_chunk_delta = serde_json::json!({
            "reasoning_details": [{
                "type": "reasoning.text",
                "text": "x".repeat(200 * 1024)
            }]
        });
        let second_chunk_bytes = serde_json::to_vec(&second_chunk_delta["reasoning_details"])
            .expect("second reasoning detail chunk")
            .len();
        assert!(second_chunk_bytes < MAX_PROVIDER_REASONING_DETAILS_BYTES);
        assert!(
            serialized_bytes
                .saturating_add(second_chunk_bytes)
                .saturating_sub(1)
                > MAX_PROVIDER_REASONING_DETAILS_BYTES
        );

        let prior_details = details.clone();
        let prior_bytes = serialized_bytes;
        let error =
            append_reasoning_details(&mut details, &mut serialized_bytes, &second_chunk_delta)
                .expect_err("reasoning details limit");

        assert_eq!(
            error.to_string(),
            "provider reasoning details exceeded the response limit"
        );
        assert_eq!(details, prior_details);
        assert_eq!(serialized_bytes, prior_bytes);
        assert_eq!(
            serde_json::to_vec(&details)
                .expect("accumulated reasoning details")
                .len(),
            serialized_bytes
        );
        assert!(serialized_bytes <= MAX_PROVIDER_REASONING_DETAILS_BYTES);
    }

    #[test]
    fn rejects_reasoning_details_that_exceed_the_serialized_response_limit_before_retaining_them() {
        let mut accumulator = ProviderAccumulator::default();
        let retained = serde_json::json!({
            "choices": [{
                "delta": {
                    "reasoning_details": [{
                        "type": "reasoning.text",
                        "text": "retained"
                    }]
                }
            }]
        });
        accumulator
            .on_data(retained, "provider-secret", &mut |_| Ok(()))
            .expect("details within limit");

        let oversized = "x".repeat(MAX_PROVIDER_REASONING_DETAILS_BYTES);
        let error = accumulator
            .on_data(
                serde_json::json!({
                    "choices": [{
                        "delta": {
                            "reasoning_details": [{"text": oversized}]
                        }
                    }]
                }),
                "provider-secret",
                &mut |_| Ok(()),
            )
            .expect_err("reasoning details limit");
        assert_eq!(
            error.to_string(),
            "provider reasoning details exceeded the response limit"
        );
        assert_eq!(
            accumulator.reasoning_details,
            vec![serde_json::json!({
                "type": "reasoning.text",
                "text": "retained"
            })]
        );
    }

    #[test]
    fn accepts_compatible_finish_reasons_and_rejects_incomplete_ones() {
        for reason in [
            None,
            Some(Value::Null),
            Some(Value::String("stop".to_owned())),
        ] {
            let mut choice = serde_json::json!({"delta": {}});
            if let Some(reason) = reason {
                choice["finish_reason"] = reason;
            }
            validate_finish_reason(&choice).expect("compatible finish reason");
        }
        for reason in ["tool_calls", "function_call"] {
            validate_finish_reason(&serde_json::json!({
                "delta": {},
                "finish_reason": reason
            }))
            .expect("tool finish reason");
        }
        for reason in ["length", "content_filter", "error"] {
            assert!(validate_finish_reason(&serde_json::json!({
                "delta": {},
                "finish_reason": reason
            }))
            .is_err());
        }
    }

    #[test]
    fn rejects_api_keys_that_conflict_with_fixed_literals() {
        for (index, secret) in [
            "session",
            "tool",
            "cmd",
            "command",
            "finite",
            "0",
            ":",
            "[REDACTED]",
        ]
        .into_iter()
        .enumerate()
        {
            let environment = format!("LUCY_PROVIDER_CONFLICT_{}_{}", std::process::id(), index);
            std::env::set_var(&environment, secret);
            let settings = LlmSettings {
                base_url: "http://localhost".to_owned(),
                model: "model".to_owned(),
                api_key_env: environment.clone(),
                effort: None,
            };
            let error = match Provider::new(&settings) {
                Ok(_) => panic!("fixed literal conflict should be rejected: {secret}"),
                Err(error) => error,
            };
            assert!(error.to_string().contains("structured output"));
            assert!(!error.to_string().contains(secret));
            std::env::remove_var(environment);
        }
    }

    #[test]
    fn accepts_a_normal_long_provider_key() {
        let environment = format!("LUCY_PROVIDER_NORMAL_{}", std::process::id());
        std::env::set_var(&environment, "provider-secret");
        let settings = LlmSettings {
            base_url: "http://localhost".to_owned(),
            model: "model".to_owned(),
            api_key_env: environment.clone(),
            effort: None,
        };
        assert!(Provider::new(&settings).is_ok());
        std::env::remove_var(environment);
    }

    #[test]
    fn accepts_a_configurable_effort() {
        let environment = format!("LUCY_PROVIDER_EFFORT_OK_{}", std::process::id());
        std::env::set_var(&environment, "provider-secret");
        let settings = LlmSettings {
            base_url: "http://localhost".to_owned(),
            model: "model".to_owned(),
            api_key_env: environment.clone(),
            effort: Some("high".to_owned()),
        };
        assert!(Provider::new(&settings).is_ok());
        std::env::remove_var(environment);
    }

    #[test]
    fn empty_effort_is_rejected_without_echoing_the_key() {
        let environment = format!("LUCY_PROVIDER_EFFORT_EMPTY_{}", std::process::id());
        std::env::set_var(&environment, "provider-secret");
        for effort in ["", "   ", "\t"] {
            let settings = LlmSettings {
                base_url: "http://localhost".to_owned(),
                model: "model".to_owned(),
                api_key_env: environment.clone(),
                effort: Some(effort.to_owned()),
            };
            let error = match Provider::new(&settings) {
                Ok(_) => panic!("empty effort should be rejected: {effort:?}"),
                Err(error) => error,
            };
            assert!(error.to_string().contains("llm.effort must not be empty"));
            assert!(!error.to_string().contains("provider-secret"));
        }
        std::env::remove_var(environment);
    }

    #[test]
    fn missing_api_key_error_does_not_echo_the_environment_name() {
        let environment = format!("LUCY_MISSING_KEY_{}", std::process::id());
        std::env::remove_var(&environment);
        let settings = LlmSettings {
            base_url: "http://localhost".to_owned(),
            model: "model".to_owned(),
            api_key_env: environment.clone(),
            effort: None,
        };
        let error = match Provider::new(&settings) {
            Ok(_) => panic!("missing key should be rejected"),
            Err(error) => error,
        };
        assert_eq!(error.to_string(), "missing provider API key");
        assert!(!error.to_string().contains(&environment));
    }

    #[test]
    fn cancellable_stream_stops_a_stalled_provider_without_waiting_for_timeout() {
        let listener = TcpListener::bind(("127.0.0.1", 0)).expect("listener");
        let address = listener.local_addr().expect("address");
        let (sent, sent_receiver) = mpsc::channel();
        let server = thread::spawn(move || {
            let (mut stream, _) = listener.accept().expect("request");
            let mut request = std::io::BufReader::new(stream.try_clone().expect("clone"));
            let mut content_length = 0;
            loop {
                let mut line = String::new();
                request.read_line(&mut line).expect("header");
                if line == "\r\n" {
                    break;
                }
                if let Some(value) = line.strip_prefix("Content-Length:") {
                    content_length = value.trim().parse::<usize>().expect("length");
                }
            }
            let mut body = vec![0; content_length];
            request.read_exact(&mut body).expect("body");

            let payload = serde_json::json!({
                "choices": [{"delta": {"content": "partial"}, "finish_reason": null}]
            });
            let event = format!("data: {payload}\n\n");
            let response = format!(
                "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nTransfer-Encoding: chunked\r\nConnection: keep-alive\r\n\r\n{:x}\r\n{}\r\n",
                event.len(), event
            );
            stream.write_all(response.as_bytes()).expect("response");
            stream.flush().expect("flush");
            sent.send(()).expect("body readiness");
            thread::sleep(Duration::from_millis(500));
        });

        let environment = format!("LUCY_PROVIDER_CANCEL_{}", std::process::id());
        std::env::set_var(&environment, "provider-secret");
        let provider = Provider::new(&LlmSettings {
            base_url: format!("http://{address}/v1"),
            model: "model".to_owned(),
            api_key_env: environment.clone(),
            effort: None,
        })
        .expect("provider");
        let token = CancellationToken::new();
        let worker_token = token.clone();
        let worker = thread::spawn(move || {
            let mut received = String::new();
            let result = provider.stream_chat_cancellable(
                &[ChatMessage::user("hello".to_owned())],
                &mut |text| {
                    received.push_str(text);
                    Ok(())
                },
                &worker_token,
            );
            (result, received)
        });
        sent_receiver
            .recv_timeout(Duration::from_secs(1))
            .expect("body was sent");
        let started = Instant::now();
        assert!(token.cancel());
        let (result, received) = worker.join().expect("provider worker");
        assert!(started.elapsed() < Duration::from_millis(400));
        let error = result.expect_err("cancellation");
        assert!(error.is_cancelled());
        assert!(received.is_empty() || received == "partial");
        server.join().expect("server");
        std::env::remove_var(environment);
    }

    #[test]
    fn rejects_an_oversized_sse_line_before_json_parsing() {
        let stream = format!("data: {}\n\n", "x".repeat(MAX_SSE_LINE_BYTES));
        let error = parse_sse(&mut Cursor::new(stream.as_bytes()), |_| Ok(())).expect_err("limit");
        assert_eq!(
            error.to_string(),
            "provider SSE line exceeded the response limit"
        );
    }

    #[test]
    fn rejects_an_oversized_sse_data_event_before_json_parsing() {
        let payload = "x".repeat(MAX_SSE_LINE_BYTES - "data: ".len());
        let line_count = MAX_SSE_EVENT_BYTES / payload.len() + 2;
        let mut stream = String::new();
        for _ in 0..line_count {
            stream.push_str("data: ");
            stream.push_str(&payload);
            stream.push('\n');
        }
        stream.push('\n');

        let error = parse_sse(&mut Cursor::new(stream.as_bytes()), |_| Ok(())).expect_err("limit");
        assert_eq!(
            error.to_string(),
            "provider SSE data event exceeded the response limit"
        );
    }

    #[test]
    fn rejects_an_oversized_sse_stream_of_ignored_fields() {
        let line = format!("ignored: {}\n", "x".repeat(1024));
        let mut stream = Vec::new();
        while stream.len() <= MAX_SSE_STREAM_BYTES {
            stream.extend_from_slice(line.as_bytes());
        }

        let error = parse_sse(&mut Cursor::new(stream), |_| Ok(())).expect_err("limit");
        assert_eq!(
            error.to_string(),
            "provider SSE stream exceeded the response limit"
        );
    }

    #[test]
    fn rejects_too_many_empty_sse_data_lines() {
        let stream = format!("{}\n", "data:\n".repeat(MAX_SSE_DATA_LINES + 1));
        let error = parse_sse(&mut Cursor::new(stream.as_bytes()), |_| Ok(())).expect_err("limit");
        assert_eq!(
            error.to_string(),
            "provider SSE data line count exceeded the response limit"
        );
    }

    #[test]
    fn reports_eof_before_done_as_incomplete() {
        let stream = b"data: {\"choices\":[{\"delta\":{\"content\":\"partial\"}}]}\n\n";
        let result = parse_sse(&mut Cursor::new(stream), |_| Ok(())).expect("SSE parse");
        assert!(result.received_payload);
        assert!(!result.received_done);
    }

    #[test]
    fn reports_empty_non_sse_input_without_payload_or_done() {
        let result =
            parse_sse(&mut Cursor::new(b"not an SSE response\n"), |_| Ok(())).expect("SSE parse");
        assert_eq!(result, SseParseResult::default());
    }

    #[test]
    fn caps_accumulated_tool_call_id_and_name_fields() {
        let fragment = "x".repeat(MAX_PROVIDER_TOOL_CALL_ID_BYTES);
        let mut id = String::new();
        append_provider_field(
            &mut id,
            &fragment,
            MAX_PROVIDER_TOOL_CALL_ID_BYTES,
            "provider tool-call id exceeded the response limit",
        )
        .expect("id within limit");
        let error = append_provider_field(
            &mut id,
            "x",
            MAX_PROVIDER_TOOL_CALL_ID_BYTES,
            "provider tool-call id exceeded the response limit",
        )
        .expect_err("id limit");
        assert_eq!(
            error.to_string(),
            "provider tool-call id exceeded the response limit"
        );

        let fragment = "x".repeat(MAX_PROVIDER_TOOL_NAME_BYTES);
        let mut name = String::new();
        append_provider_field(
            &mut name,
            &fragment,
            MAX_PROVIDER_TOOL_NAME_BYTES,
            "provider tool-call name exceeded the response limit",
        )
        .expect("name within limit");
        let error = append_provider_field(
            &mut name,
            "x",
            MAX_PROVIDER_TOOL_NAME_BYTES,
            "provider tool-call name exceeded the response limit",
        )
        .expect_err("name limit");
        assert_eq!(
            error.to_string(),
            "provider tool-call name exceeded the response limit"
        );
    }

    #[test]
    fn caps_provider_error_text_without_copying_the_full_message() {
        let message = "x".repeat(MAX_PROVIDER_ERROR_BYTES + 1);
        let value = serde_json::json!({"error": {"message": message}});
        assert_eq!(
            provider_error_message(&value),
            Some("provider error text exceeded the response limit")
        );
    }

    fn read_request_headers(stream: &TcpStream) {
        let mut reader = BufReader::new(stream.try_clone().expect("clone request"));
        loop {
            let mut line = String::new();
            reader.read_line(&mut line).expect("request header");
            if line == "\r\n" || line.is_empty() {
                return;
            }
        }
    }

    fn response_body(text: &str) -> String {
        let payload = serde_json::json!({
            "choices": [{"delta": {"content": text}, "finish_reason": null}]
        });
        let finish = serde_json::json!({
            "choices": [{"delta": {}, "finish_reason": "stop"}]
        });
        format!("data: {payload}\n\ndata: {finish}\n\ndata: [DONE]\n\n")
    }

    fn provider_for(address: std::net::SocketAddr, read_timeout: Duration) -> (Provider, String) {
        let environment = format!(
            "LUCY_PROVIDER_STREAM_TEST_{}_{}",
            std::process::id(),
            address.port()
        );
        std::env::set_var(&environment, "provider-secret");
        let settings = LlmSettings {
            base_url: format!("http://{address}/v1"),
            model: "model".to_owned(),
            api_key_env: environment.clone(),
            effort: None,
        };
        let mut provider = Provider::new(&settings).expect("provider");
        provider.async_client = AsyncClient::builder()
            .connect_timeout(Duration::from_secs(1))
            .read_timeout(read_timeout)
            .build()
            .expect("test async client");
        (provider, environment)
    }

    #[test]
    fn worker_stream_can_exceed_idle_interval_without_total_deadline() {
        let listener = TcpListener::bind(("127.0.0.1", 0)).expect("listener");
        let address = listener.local_addr().expect("address");
        let parts = (0..5)
            .map(|index| {
                let payload = serde_json::json!({
                    "choices": [{
                        "delta": {"content": format!("part-{index}")},
                        "finish_reason": null
                    }]
                });
                format!("data: {payload}\n\n")
            })
            .chain([
                format!(
                    "data: {}\n\n",
                    serde_json::json!({
                        "choices": [{"delta": {}, "finish_reason": "stop"}]
                    })
                ),
                "data: [DONE]\n\n".to_owned(),
            ])
            .collect::<Vec<_>>();
        let body = parts.concat();
        let server = thread::spawn(move || {
            let (mut stream, _) = listener.accept().expect("request");
            read_request_headers(&stream);
            let header = format!(
                "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
                body.len()
            );
            stream.write_all(header.as_bytes()).expect("header");
            stream.flush().expect("header flush");
            for part in parts {
                stream.write_all(part.as_bytes()).expect("SSE part");
                stream.flush().expect("SSE flush");
                thread::sleep(Duration::from_millis(25));
            }
        });

        let (provider, environment) = provider_for(address, Duration::from_millis(80));
        let cancellation = CancellationToken::new();
        let started = Instant::now();
        let mut output = String::new();
        let turn = provider
            .stream_chat_cancellable_with_options(
                &[ChatMessage::user("worker task".to_owned())],
                &mut |text| {
                    output.push_str(text);
                    Ok(())
                },
                &cancellation,
                true,
            )
            .expect("long worker stream");

        assert!(started.elapsed() >= Duration::from_millis(80));
        assert_eq!(turn.content, output);
        assert!(output.contains("part-4"));
        server.join().expect("server");
        std::env::remove_var(environment);
    }

    #[test]
    fn retries_a_pre_payload_stream_failure_once_and_classifies_it() {
        let listener = TcpListener::bind(("127.0.0.1", 0)).expect("listener");
        listener
            .set_nonblocking(true)
            .expect("nonblocking listener");
        let address = listener.local_addr().expect("address");
        let body = response_body("retried");
        let server = thread::spawn(move || {
            let deadline = Instant::now() + Duration::from_secs(2);
            for attempt in 0..2 {
                let (mut stream, _) = loop {
                    match listener.accept() {
                        Ok(connection) => break connection,
                        Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
                            assert!(Instant::now() < deadline, "provider did not retry");
                            thread::sleep(Duration::from_millis(5));
                        }
                        Err(error) => panic!("accept: {error}"),
                    }
                };
                read_request_headers(&stream);
                if attempt == 0 {
                    let header = "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: 100\r\nConnection: close\r\n\r\n";
                    stream.write_all(header.as_bytes()).expect("failed header");
                    stream.flush().expect("failed flush");
                } else {
                    let header = format!(
                        "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
                        body.len()
                    );
                    stream.write_all(header.as_bytes()).expect("success header");
                    stream.write_all(body.as_bytes()).expect("success body");
                    stream.flush().expect("success flush");
                }
            }
        });

        let (provider, environment) = provider_for(address, Duration::from_secs(1));
        let cancellation = CancellationToken::new();
        let mut output = String::new();
        let turn = provider
            .stream_chat_cancellable_with_options(
                &[ChatMessage::user("retry".to_owned())],
                &mut |text| {
                    output.push_str(text);
                    Ok(())
                },
                &cancellation,
                true,
            )
            .expect("retry succeeds");

        assert_eq!(turn.content, "retried");
        assert_eq!(output, "retried");
        server.join().expect("server");
        std::env::remove_var(environment);
    }

    #[test]
    fn does_not_retry_after_partial_provider_output() {
        let listener = TcpListener::bind(("127.0.0.1", 0)).expect("listener");
        listener
            .set_nonblocking(true)
            .expect("nonblocking listener");
        let address = listener.local_addr().expect("address");
        let partial = format!(
            "data: {}\n\n",
            serde_json::json!({
                "choices": [{"delta": {"content": "partial"}, "finish_reason": null}]
            })
        );
        let server = thread::spawn(move || {
            let deadline = Instant::now() + Duration::from_secs(2);
            let (mut stream, _) = loop {
                match listener.accept() {
                    Ok(connection) => break connection,
                    Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
                        assert!(Instant::now() < deadline, "provider request missing");
                        thread::sleep(Duration::from_millis(5));
                    }
                    Err(error) => panic!("accept: {error}"),
                }
            };
            read_request_headers(&stream);
            let header = format!(
                "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
                partial.len() + 10
            );
            stream.write_all(header.as_bytes()).expect("partial header");
            stream.write_all(partial.as_bytes()).expect("partial body");
            stream.flush().expect("partial flush");
            thread::sleep(Duration::from_millis(150));
            assert!(matches!(
                listener.accept(),
                Err(error) if error.kind() == std::io::ErrorKind::WouldBlock
            ));
        });

        let (provider, environment) = provider_for(address, Duration::from_secs(1));
        let cancellation = CancellationToken::new();
        let mut output = String::new();
        let error = provider
            .stream_chat_cancellable_with_options(
                &[ChatMessage::user("partial".to_owned())],
                &mut |text| {
                    output.push_str(text);
                    Ok(())
                },
                &cancellation,
                true,
            )
            .expect_err("partial stream must fail");

        let message = error.to_string();
        assert!(message.contains("provider stream read failed"));
        assert!([
            "(timeout)",
            "(connection)",
            "(body)",
            "(decode)",
            "(request)",
            "(transport)",
        ]
        .iter()
        .any(|kind| message.contains(kind)));
        assert_eq!(output, "partial");
        server.join().expect("server");
        std::env::remove_var(environment);
    }

    #[test]
    fn reports_midstream_error_without_echoing_provider_body() {
        let stream = b"data: {\"error\":{\"message\":\"bad request\"}}\n\n";
        let error = parse_sse(&mut Cursor::new(stream), |value| {
            if let Some(message) = provider_error_message(&value) {
                return Err(ProviderError::new(format!(
                    "provider stream error: {message}"
                )));
            }
            Ok(())
        })
        .expect_err("error");
        assert_eq!(error.to_string(), "provider stream error: bad request");
    }
}