IdeA/crates/infrastructure/src/session/openai_compat.rs

//! Adapter de session HTTP OpenAI-compatible (Ollama, llama.cpp, runtime LAN).
//!
//! Contrairement aux adapters CLI, la boucle agentique est conduite par IdeA :
//! transcript local, appels HTTP `/chat/completions`, invocation in-process des
//! outils, puis rebouclage borné.

use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::Duration;

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

use domain::ports::{
    AgentSession, AgentSessionError, ReplyEvent, ReplyStream, ToolInvoker, ToolSpec,
};
use domain::profile::{HttpChatConfig, StructuredAdapter};
use domain::SessionId;

const TRANSCRIPT_FILE: &str = "chat-transcript.json";

/// Appel d'outil OpenAI-compatible normalisé pour la boucle agentique.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OpenAiToolCall {
    /// Identifiant OpenAI-compatible de l'appel d'outil.
    pub id: String,
    /// Nom de la fonction appelée.
    pub name: String,
    /// Arguments JSON bruts fournis par le modèle.
    pub arguments: String,
}

/// Résultat d'un chunk streaming OpenAI-compatible.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ParsedChatDelta {
    /// Fragment de texte assistant.
    pub text: String,
    /// Appels d'outils annoncés par ce chunk.
    pub tool_calls: Vec<OpenAiToolCall>,
}

/// Résultat d'une completion non-streaming OpenAI-compatible.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ParsedCompletion {
    /// Contenu assistant complet.
    pub content: String,
    /// Appels d'outils demandés par le modèle.
    pub tool_calls: Vec<OpenAiToolCall>,
}

#[derive(Debug, Default)]
struct StreamState {
    content: String,
    tool_calls: Vec<ToolCallBuilder>,
}

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

impl ToolCallBuilder {
    fn finish(self) -> Option<OpenAiToolCall> {
        if self.name.is_empty() {
            None
        } else {
            Some(OpenAiToolCall {
                id: if self.id.is_empty() {
                    format!("call_{}", self.name)
                } else {
                    self.id
                },
                name: self.name,
                arguments: self.arguments,
            })
        }
    }
}

/// Parse un chunk SSE `chat.completion.chunk` OpenAI-compatible.
///
/// # Errors
/// [`AgentSessionError::Decode`] si le JSON est illisible. Le payload brut ne fuit
/// jamais dans le message.
pub fn parse_chat_delta(line: &str) -> Result<ParsedChatDelta, AgentSessionError> {
    let trimmed = line.trim();
    if trimmed.is_empty() || trimmed == "[DONE]" {
        return Ok(ParsedChatDelta::default());
    }
    let value: Value = serde_json::from_str(trimmed)
        .map_err(|e| AgentSessionError::Decode(format!("chat delta JSON illisible: {e}")))?;
    let Some(delta) = value
        .get("choices")
        .and_then(Value::as_array)
        .and_then(|choices| choices.first())
        .and_then(|choice| choice.get("delta"))
    else {
        return Ok(ParsedChatDelta::default());
    };

    let text = delta
        .get("content")
        .and_then(Value::as_str)
        .unwrap_or_default()
        .to_owned();
    let tool_calls = parse_tool_calls_array(delta.get("tool_calls"));
    Ok(ParsedChatDelta { text, tool_calls })
}

/// Parse une réponse non-streaming `/chat/completions`.
///
/// # Errors
/// [`AgentSessionError::Decode`] si le JSON est illisible. Le payload brut ne fuit
/// jamais dans le message.
pub fn parse_completion(body: &str) -> Result<ParsedCompletion, AgentSessionError> {
    let value: Value = serde_json::from_str(body)
        .map_err(|e| AgentSessionError::Decode(format!("chat completion JSON illisible: {e}")))?;
    let Some(message) = value
        .get("choices")
        .and_then(Value::as_array)
        .and_then(|choices| choices.first())
        .and_then(|choice| choice.get("message"))
    else {
        return Err(AgentSessionError::Decode(
            "chat completion schema missing choices[0].message".to_owned(),
        ));
    };
    Ok(ParsedCompletion {
        content: message
            .get("content")
            .and_then(Value::as_str)
            .unwrap_or_default()
            .to_owned(),
        tool_calls: parse_tool_calls_array(message.get("tool_calls")),
    })
}

fn parse_tool_calls_array(value: Option<&Value>) -> Vec<OpenAiToolCall> {
    value
        .and_then(Value::as_array)
        .into_iter()
        .flatten()
        .filter_map(|item| {
            let function = item.get("function")?;
            let name = function
                .get("name")
                .and_then(Value::as_str)
                .unwrap_or_default();
            let arguments = function
                .get("arguments")
                .and_then(Value::as_str)
                .unwrap_or_default();
            if name.is_empty() && arguments.is_empty() {
                return None;
            }
            Some(OpenAiToolCall {
                id: item
                    .get("id")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .to_owned(),
                name: name.to_owned(),
                arguments: arguments.to_owned(),
            })
        })
        .collect()
}

/// Session structurée OpenAI-compatible, stateless côté provider et persistée côté IdeA.
pub struct OpenAiCompatibleSession {
    id: SessionId,
    client: Client,
    config: HttpChatConfig,
    endpoint: String,
    api_key: Option<String>,
    transcript_path: PathBuf,
    transcript: Mutex<Vec<Value>>,
    tool_invoker: Option<Arc<dyn ToolInvoker>>,
    tools_disabled: Mutex<bool>,
    tools_rejection_reported: Mutex<bool>,
    first_contact: Mutex<bool>,
}

impl OpenAiCompatibleSession {
    /// Construit l'adapter depuis une config domaine validée.
    ///
    /// # Errors
    /// [`AgentSessionError::Start`] si la clé d'environnement déclarée est absente
    /// ou si le client HTTP ne peut pas être construit.
    pub fn new(
        id: SessionId,
        config: HttpChatConfig,
        run_dir: impl AsRef<Path>,
        system_context: &str,
        tool_invoker: Option<Arc<dyn ToolInvoker>>,
    ) -> Result<Self, AgentSessionError> {
        let mut builder = Client::builder();
        if let Some(ms) = config.connect_timeout_ms {
            builder = builder.connect_timeout(Duration::from_millis(u64::from(ms)));
        }
        if let Some(ms) = config.request_timeout_ms {
            builder = builder.timeout(Duration::from_millis(u64::from(ms)));
        }
        let client = builder
            .build()
            .map_err(|e| AgentSessionError::Start(format!("client HTTP invalide: {e}")))?;
        let api_key = match &config.api_key_env {
            Some(var) => Some(std::env::var(var).map_err(|_| {
                AgentSessionError::Start(format!(
                    "variable d'environnement `{var}` introuvable pour la clé API"
                ))
            })?),
            None => None,
        };
        let transcript_path = run_dir.as_ref().join(TRANSCRIPT_FILE);
        let transcript = load_or_seed_transcript(&transcript_path, system_context)?;
        Ok(Self {
            id,
            endpoint: chat_completions_endpoint(&config.endpoint),
            client,
            config,
            api_key,
            transcript_path,
            transcript: Mutex::new(transcript),
            tool_invoker,
            tools_disabled: Mutex::new(false),
            tools_rejection_reported: Mutex::new(false),
            first_contact: Mutex::new(true),
        })
    }

    async fn send_inner(
        &self,
        prompt: &str,
        tap: Option<std::sync::mpsc::Sender<ReplyEvent>>,
    ) -> Result<ReplyStream, AgentSessionError> {
        {
            let mut transcript = self.transcript.lock().expect("mutex sain");
            transcript.push(json!({"role": "user", "content": prompt}));
        }

        let mut events = vec![ReplyEvent::Heartbeat];
        send_tap(&tap, &ReplyEvent::Heartbeat);
        for iteration in 0..=self.config.effective_max_tool_iterations() {
            let transcript = self.transcript.lock().expect("mutex sain").clone();
            let tools = self.effective_tools();
            let response = self.post_chat(&transcript, &tools, true).await;
            let response = match response {
                Ok(response) => response,
                Err(err) => {
                    if should_retry_without_tools(&err, !tools.is_empty()) {
                        self.disable_tools_once(&mut events, &tap);
                        let transcript = self.transcript.lock().expect("mutex sain").clone();
                        self.post_chat(&transcript, &[], true).await?
                    } else {
                        return Err(err);
                    }
                }
            };

            let turn = self.consume_response(response, &tap).await?;
            events.extend(turn.events);

            if !turn.tool_calls.is_empty() {
                if iteration >= self.config.effective_max_tool_iterations() {
                    let content = "Tool-calling stopped: max_tool_iterations reached.".to_owned();
                    self.transcript
                        .lock()
                        .expect("mutex sain")
                        .push(json!({"role": "assistant", "content": content}));
                    events.push(ReplyEvent::Final { content });
                    self.persist_transcript()?;
                    return Ok(Box::new(events.into_iter()));
                }
                self.apply_tool_calls(&turn.tool_calls, &mut events, &tap)
                    .await;
                self.persist_transcript()?;
                continue;
            }

            self.transcript
                .lock()
                .expect("mutex sain")
                .push(json!({"role": "assistant", "content": turn.content}));
            events.push(ReplyEvent::Final {
                content: turn.content,
            });
            self.persist_transcript()?;
            return Ok(Box::new(events.into_iter()));
        }
        unreachable!("loop returns at max_tool_iterations");
    }

    fn effective_tools(&self) -> Vec<ToolSpec> {
        if *self.tools_disabled.lock().expect("mutex sain") {
            return Vec::new();
        }
        self.tool_invoker
            .as_ref()
            .map_or_else(Vec::new, |invoker| invoker.tools())
    }

    async fn post_chat(
        &self,
        messages: &[Value],
        tools: &[ToolSpec],
        stream: bool,
    ) -> Result<reqwest::Response, AgentSessionError> {
        let mut body = json!({
            "model": self.config.model,
            "messages": messages,
            "stream": stream,
        });
        if !tools.is_empty() {
            body["tools"] = Value::Array(tools.iter().map(tool_spec_to_openai).collect());
        }
        let mut request = self.client.post(&self.endpoint).json(&body);
        if let Some(key) = &self.api_key {
            request = request.bearer_auth(key);
        }
        let response = request
            .send()
            .await
            .map_err(|e| self.map_reqwest_error(e))?;
        if response.status().is_success() {
            *self.first_contact.lock().expect("mutex sain") = false;
            return Ok(response);
        }
        let status = response.status();
        if status == StatusCode::BAD_REQUEST
            || status == StatusCode::NOT_FOUND
            || status == StatusCode::UNPROCESSABLE_ENTITY
        {
            return Err(AgentSessionError::Start(format!(
                "endpoint OpenAI-compatible rejected request with HTTP {status}"
            )));
        }
        Err(AgentSessionError::Io(format!(
            "endpoint OpenAI-compatible returned HTTP {status}"
        )))
    }

    async fn consume_response(
        &self,
        response: reqwest::Response,
        tap: &Option<std::sync::mpsc::Sender<ReplyEvent>>,
    ) -> Result<TurnResult, AgentSessionError> {
        let content_type = response
            .headers()
            .get(reqwest::header::CONTENT_TYPE)
            .and_then(|value| value.to_str().ok())
            .unwrap_or_default()
            .to_ascii_lowercase();
        if !content_type.contains("text/event-stream") {
            let body = response
                .text()
                .await
                .map_err(|e| AgentSessionError::Io(format!("lecture réponse HTTP: {e}")))?;
            let parsed = parse_completion(&body)?;
            return Ok(TurnResult {
                events: text_events(&parsed.content, tap),
                content: parsed.content,
                tool_calls: parsed.tool_calls,
            });
        }

        let mut state = StreamState::default();
        let mut events = Vec::new();
        let mut pending = String::new();
        let mut stream = response.bytes_stream();
        while let Some(chunk) = stream.next().await {
            let chunk =
                chunk.map_err(|e| AgentSessionError::Io(format!("lecture flux SSE: {e}")))?;
            pending.push_str(&String::from_utf8_lossy(&chunk));
            while let Some(pos) = pending.find('\n') {
                let line = pending[..pos].trim_end_matches('\r').to_owned();
                pending = pending[pos + 1..].to_owned();
                handle_sse_line(&line, &mut state, &mut events, tap)?;
            }
        }
        if !pending.trim().is_empty() {
            handle_sse_line(pending.trim_end_matches('\r'), &mut state, &mut events, tap)?;
        }
        Ok(TurnResult {
            events,
            content: state.content,
            tool_calls: state
                .tool_calls
                .into_iter()
                .filter_map(ToolCallBuilder::finish)
                .collect(),
        })
    }

    async fn apply_tool_calls(
        &self,
        tool_calls: &[OpenAiToolCall],
        events: &mut Vec<ReplyEvent>,
        tap: &Option<std::sync::mpsc::Sender<ReplyEvent>>,
    ) {
        let assistant_calls = tool_calls
            .iter()
            .map(|call| {
                json!({
                    "id": call.id,
                    "type": "function",
                    "function": { "name": call.name, "arguments": call.arguments }
                })
            })
            .collect::<Vec<_>>();
        self.transcript.lock().expect("mutex sain").push(json!({
            "role": "assistant",
            "content": Value::Null,
            "tool_calls": assistant_calls,
        }));

        for call in tool_calls {
            let event = ReplyEvent::ToolActivity {
                label: call.name.clone(),
            };
            send_tap(tap, &event);
            events.push(event);
            let result = match &self.tool_invoker {
                Some(invoker) => invoker
                    .call(&call.name, &call.arguments)
                    .await
                    .unwrap_or_else(|e| format!("Tool invocation failed: {e}")),
                None => "Tool invocation unavailable".to_owned(),
            };
            self.transcript.lock().expect("mutex sain").push(json!({
                "role": "tool",
                "tool_call_id": call.id,
                "content": result,
            }));
        }
    }

    fn disable_tools_once(
        &self,
        events: &mut Vec<ReplyEvent>,
        tap: &Option<std::sync::mpsc::Sender<ReplyEvent>>,
    ) {
        *self.tools_disabled.lock().expect("mutex sain") = true;
        let mut reported = self.tools_rejection_reported.lock().expect("mutex sain");
        if !*reported {
            let event = ReplyEvent::Announcement {
                text: "Tool calling disabled for this OpenAI-compatible endpoint; retrying chat without tools.".to_owned(),
            };
            send_tap(tap, &event);
            events.push(event);
            *reported = true;
        }
    }

    fn map_reqwest_error(&self, e: reqwest::Error) -> AgentSessionError {
        if *self.first_contact.lock().expect("mutex sain") && (e.is_connect() || e.is_request()) {
            return AgentSessionError::Start("endpoint OpenAI-compatible indisponible".to_owned());
        }
        if e.is_timeout() {
            return AgentSessionError::Io(format!("communication endpoint OpenAI-compatible: {e}"));
        }
        AgentSessionError::Io(format!("communication endpoint OpenAI-compatible: {e}"))
    }

    fn persist_transcript(&self) -> Result<(), AgentSessionError> {
        if let Some(parent) = self.transcript_path.parent() {
            std::fs::create_dir_all(parent)
                .map_err(|e| AgentSessionError::Io(format!("création run dir: {e}")))?;
        }
        let transcript = self.transcript.lock().expect("mutex sain");
        let bytes = serde_json::to_vec_pretty(&*transcript)
            .map_err(|e| AgentSessionError::Decode(format!("transcript JSON invalide: {e}")))?;
        std::fs::write(&self.transcript_path, bytes)
            .map_err(|e| AgentSessionError::Io(format!("écriture transcript: {e}")))
    }
}

#[async_trait]
impl AgentSession for OpenAiCompatibleSession {
    fn id(&self) -> SessionId {
        self.id
    }

    fn conversation_id(&self) -> Option<String> {
        None
    }

    async fn send(&self, prompt: &str) -> Result<ReplyStream, AgentSessionError> {
        self.send_inner(prompt, None).await
    }

    async fn send_with_tap(
        &self,
        prompt: &str,
        tap: std::sync::mpsc::Sender<ReplyEvent>,
    ) -> Result<ReplyStream, AgentSessionError> {
        self.send_inner(prompt, Some(tap)).await
    }

    async fn shutdown(&self) -> Result<(), AgentSessionError> {
        Ok(())
    }
}

#[derive(Debug)]
struct TurnResult {
    events: Vec<ReplyEvent>,
    content: String,
    tool_calls: Vec<OpenAiToolCall>,
}

fn handle_sse_line(
    line: &str,
    state: &mut StreamState,
    events: &mut Vec<ReplyEvent>,
    tap: &Option<std::sync::mpsc::Sender<ReplyEvent>>,
) -> Result<(), AgentSessionError> {
    let Some(data) = line.strip_prefix("data:") else {
        return Ok(());
    };
    let data = data.trim();
    if data == "[DONE]" {
        return Ok(());
    }
    let parsed = parse_chat_delta(data)?;
    if !parsed.text.is_empty() {
        state.content.push_str(&parsed.text);
        let event = ReplyEvent::TextDelta { text: parsed.text };
        send_tap(tap, &event);
        events.push(event);
    }
    merge_tool_call_deltas(state, parsed.tool_calls);
    Ok(())
}

fn merge_tool_call_deltas(state: &mut StreamState, calls: Vec<OpenAiToolCall>) {
    for (idx, call) in calls.into_iter().enumerate() {
        if state.tool_calls.len() <= idx {
            state
                .tool_calls
                .resize_with(idx + 1, ToolCallBuilder::default);
        }
        let target = &mut state.tool_calls[idx];
        if !call.id.is_empty() {
            target.id = call.id;
        }
        if !call.name.is_empty() {
            target.name = call.name;
        }
        target.arguments.push_str(&call.arguments);
    }
}

fn text_events(
    content: &str,
    tap: &Option<std::sync::mpsc::Sender<ReplyEvent>>,
) -> Vec<ReplyEvent> {
    if content.is_empty() {
        return Vec::new();
    }
    let event = ReplyEvent::TextDelta {
        text: content.to_owned(),
    };
    send_tap(tap, &event);
    vec![event]
}

fn send_tap(tap: &Option<std::sync::mpsc::Sender<ReplyEvent>>, event: &ReplyEvent) {
    if let Some(tap) = tap {
        let _ = tap.send(event.clone());
    }
}

fn should_retry_without_tools(err: &AgentSessionError, using_tools: bool) -> bool {
    using_tools && matches!(err, AgentSessionError::Start(_))
}

fn tool_spec_to_openai(spec: &ToolSpec) -> Value {
    json!({
        "type": "function",
        "function": {
            "name": spec.name,
            "description": spec.description,
            "parameters": spec.input_schema,
        }
    })
}

fn chat_completions_endpoint(endpoint: &str) -> String {
    let trimmed = endpoint.trim_end_matches('/');
    if trimmed.ends_with("/chat/completions") {
        trimmed.to_owned()
    } else {
        format!("{trimmed}/chat/completions")
    }
}

fn load_or_seed_transcript(
    path: &Path,
    system_context: &str,
) -> Result<Vec<Value>, AgentSessionError> {
    match std::fs::read(path) {
        Ok(bytes) => serde_json::from_slice(&bytes)
            .map_err(|e| AgentSessionError::Decode(format!("transcript JSON illisible: {e}"))),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(vec![json!({
            "role": "system",
            "content": system_context,
        })]),
        Err(e) => Err(AgentSessionError::Io(format!("lecture transcript: {e}"))),
    }
}

#[allow(dead_code)]
fn _assert_adapter_name(adapter: StructuredAdapter) -> &'static str {
    adapter.provider_key()
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
struct _SerdeGuard;

#[cfg(test)]
mod tests {
    use std::sync::atomic::{AtomicUsize, Ordering};

    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    use tokio::net::TcpListener;

    use super::*;
    use domain::ports::ToolInvocationError;

    #[derive(Default)]
    struct FakeInvoker {
        calls: AtomicUsize,
    }

    #[async_trait]
    impl ToolInvoker for FakeInvoker {
        fn tools(&self) -> Vec<ToolSpec> {
            vec![ToolSpec {
                name: "idea_echo".to_owned(),
                description: "Echo".to_owned(),
                input_schema: json!({"type":"object"}),
            }]
        }

        async fn call(&self, name: &str, args_json: &str) -> Result<String, ToolInvocationError> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            Ok(format!("{name}:{args_json}"))
        }
    }

    enum TestHttpResponse {
        OkJson(&'static str),
        Status { code: u16, body: &'static str },
        Hang,
    }

    impl TestHttpResponse {
        fn ok(body: &'static str) -> Self {
            Self::OkJson(body)
        }
    }

    async fn http_server(
        responses: Vec<TestHttpResponse>,
    ) -> (String, Arc<Mutex<Vec<String>>>, tokio::task::JoinHandle<()>) {
        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
        let addr = listener.local_addr().expect("addr");
        let bodies = Arc::new(Mutex::new(Vec::new()));
        let captured = Arc::clone(&bodies);
        let responses = Arc::new(Mutex::new(responses.into_iter()));
        let handle = tokio::spawn(async move {
            loop {
                let Ok((mut socket, _)) = listener.accept().await else {
                    break;
                };
                let mut buffer = Vec::new();
                let mut chunk = [0_u8; 1024];
                loop {
                    let n = socket.read(&mut chunk).await.expect("read request");
                    if n == 0 {
                        return;
                    }
                    buffer.extend_from_slice(&chunk[..n]);
                    if let Some(headers_end) = find_headers_end(&buffer) {
                        let headers = String::from_utf8_lossy(&buffer[..headers_end]);
                        let content_length = headers
                            .lines()
                            .find_map(|line| {
                                line.to_ascii_lowercase()
                                    .strip_prefix("content-length:")
                                    .and_then(|value| value.trim().parse::<usize>().ok())
                            })
                            .unwrap_or(0);
                        let total = headers_end + 4 + content_length;
                        while buffer.len() < total {
                            let n = socket.read(&mut chunk).await.expect("read body");
                            if n == 0 {
                                return;
                            }
                            buffer.extend_from_slice(&chunk[..n]);
                        }
                        let body =
                            String::from_utf8_lossy(&buffer[headers_end + 4..total]).to_string();
                        captured.lock().expect("mutex sain").push(body);
                        break;
                    }
                }
                let body = responses
                    .lock()
                    .expect("mutex sain")
                    .next()
                    .unwrap_or_else(|| {
                        TestHttpResponse::OkJson(
                            r#"{"choices":[{"message":{"role":"assistant","content":"done"}}]}"#,
                        )
                    });
                match body {
                    TestHttpResponse::OkJson(body) => {
                        let response = format!(
                            "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}",
                            body.len(),
                            body
                        );
                        socket
                            .write_all(response.as_bytes())
                            .await
                            .expect("write response");
                    }
                    TestHttpResponse::Status { code, body } => {
                        let response = format!(
                            "HTTP/1.1 {code} test\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}",
                            body.len(),
                            body
                        );
                        socket
                            .write_all(response.as_bytes())
                            .await
                            .expect("write response");
                    }
                    TestHttpResponse::Hang => {
                        tokio::time::sleep(Duration::from_secs(60)).await;
                    }
                }
            }
        });
        (format!("http://{addr}/v1"), bodies, handle)
    }

    fn find_headers_end(buffer: &[u8]) -> Option<usize> {
        buffer.windows(4).position(|window| window == b"\r\n\r\n")
    }

    fn temp_run_dir(name: &str) -> PathBuf {
        let path = std::env::temp_dir().join(format!(
            "idea-openai-compat-{name}-{}",
            uuid::Uuid::new_v4()
        ));
        std::fs::create_dir_all(&path).expect("temp run dir");
        path
    }

    fn config(endpoint: String, max_tool_iterations: Option<u16>) -> HttpChatConfig {
        config_with_timeouts(endpoint, max_tool_iterations, 10_000, 1_000)
    }

    fn config_with_timeouts(
        endpoint: String,
        max_tool_iterations: Option<u16>,
        request_timeout_ms: u32,
        connect_timeout_ms: u32,
    ) -> HttpChatConfig {
        HttpChatConfig::new(
            endpoint,
            "local-model",
            None,
            Some(request_timeout_ms),
            Some(connect_timeout_ms),
            max_tool_iterations,
        )
        .expect("valid config")
    }

    async fn drain(session: &OpenAiCompatibleSession, prompt: &str) -> Vec<ReplyEvent> {
        session
            .send(prompt)
            .await
            .expect("send")
            .collect::<Vec<_>>()
    }

    #[test]
    fn parse_openai_sse_delta_text() {
        let parsed = parse_chat_delta(
            r#"{"choices":[{"delta":{"content":"bonjour"},"finish_reason":null}]}"#,
        )
        .expect("parse ok");
        assert_eq!(parsed.text, "bonjour");
        assert!(parsed.tool_calls.is_empty());
    }

    #[test]
    fn parse_ollama_completion_message() {
        let parsed =
            parse_completion(r#"{"choices":[{"message":{"role":"assistant","content":"salut"}}]}"#)
                .expect("parse ok");
        assert_eq!(parsed.content, "salut");
    }

    #[test]
    fn parse_completion_tool_calls() {
        let parsed = parse_completion(
            r#"{"choices":[{"message":{"role":"assistant","content":null,"tool_calls":[{"id":"call_1","type":"function","function":{"name":"idea_ask_agent","arguments":"{\"target\":\"QA\"}"}}]}}]}"#,
        )
        .expect("parse ok");
        assert_eq!(
            parsed.tool_calls,
            vec![OpenAiToolCall {
                id: "call_1".to_owned(),
                name: "idea_ask_agent".to_owned(),
                arguments: "{\"target\":\"QA\"}".to_owned(),
            }]
        );
    }

    #[test]
    fn parse_broken_json_does_not_leak_payload() {
        let err = parse_chat_delta("{ not json").expect_err("decode error");
        match err {
            AgentSessionError::Decode(msg) => assert!(!msg.contains("not json")),
            other => panic!("expected Decode, got {other:?}"),
        }
    }

    #[test]
    fn endpoint_appends_chat_completions_once() {
        assert_eq!(
            chat_completions_endpoint("http://localhost:11434/v1"),
            "http://localhost:11434/v1/chat/completions"
        );
        assert_eq!(
            chat_completions_endpoint("http://x/v1/chat/completions"),
            "http://x/v1/chat/completions"
        );
    }

    #[tokio::test]
    async fn non_streaming_completion_persists_and_reloads_transcript() {
        let (endpoint, bodies, handle) = http_server(vec![
            TestHttpResponse::ok(
                r#"{"choices":[{"message":{"role":"assistant","content":"one"}}]}"#,
            ),
            TestHttpResponse::ok(
                r#"{"choices":[{"message":{"role":"assistant","content":"two"}}]}"#,
            ),
        ])
        .await;
        let run_dir = temp_run_dir("persist");
        let session = OpenAiCompatibleSession::new(
            SessionId::new_random(),
            config(endpoint.clone(), None),
            &run_dir,
            "# system",
            None,
        )
        .expect("session");
        let events = drain(&session, "hello").await;
        assert!(matches!(events.last(), Some(ReplyEvent::Final { content }) if content == "one"));
        assert!(run_dir.join(TRANSCRIPT_FILE).exists());

        let reloaded = OpenAiCompatibleSession::new(
            SessionId::new_random(),
            config(endpoint, None),
            &run_dir,
            "# ignored",
            None,
        )
        .expect("reloaded");
        let _ = drain(&reloaded, "again").await;
        let bodies = bodies.lock().expect("mutex sain");
        assert!(
            bodies[1].contains("one"),
            "second request should include persisted assistant transcript: {}",
            bodies[1]
        );
        handle.abort();
    }

    #[tokio::test]
    async fn tool_loop_invokes_tool_and_reposts_result() {
        let (endpoint, bodies, handle) = http_server(vec![
            TestHttpResponse::ok(
                r#"{"choices":[{"message":{"role":"assistant","content":null,"tool_calls":[{"id":"call_1","type":"function","function":{"name":"idea_echo","arguments":"{\"x\":1}"}}]}}]}"#,
            ),
            TestHttpResponse::ok(
                r#"{"choices":[{"message":{"role":"assistant","content":"done"}}]}"#,
            ),
        ])
        .await;
        let invoker = Arc::new(FakeInvoker::default());
        let session = OpenAiCompatibleSession::new(
            SessionId::new_random(),
            config(endpoint, Some(4)),
            temp_run_dir("tools"),
            "# system",
            Some(invoker.clone()),
        )
        .expect("session");
        let events = drain(&session, "use tool").await;
        assert_eq!(invoker.calls.load(Ordering::SeqCst), 1);
        assert!(events.iter().any(|event| matches!(
            event,
            ReplyEvent::ToolActivity { label } if label == "idea_echo"
        )));
        assert!(matches!(events.last(), Some(ReplyEvent::Final { content }) if content == "done"));
        let bodies = bodies.lock().expect("mutex sain");
        assert_eq!(bodies.len(), 2);
        assert!(bodies[1].contains("\"role\":\"tool\""));
        handle.abort();
    }

    #[tokio::test]
    async fn max_tool_iterations_returns_single_degraded_final() {
        let tool = r#"{"choices":[{"message":{"role":"assistant","content":null,"tool_calls":[{"id":"call_1","type":"function","function":{"name":"idea_echo","arguments":"{}"}}]}}]}"#;
        let (endpoint, _bodies, handle) =
            http_server(vec![TestHttpResponse::ok(tool), TestHttpResponse::ok(tool)]).await;
        let session = OpenAiCompatibleSession::new(
            SessionId::new_random(),
            config(endpoint, Some(1)),
            temp_run_dir("limit"),
            "# system",
            Some(Arc::new(FakeInvoker::default())),
        )
        .expect("session");
        let events = drain(&session, "loop").await;
        let finals = events
            .iter()
            .filter(|event| matches!(event, ReplyEvent::Final { .. }))
            .count();
        assert_eq!(finals, 1);
        assert!(matches!(
            events.last(),
            Some(ReplyEvent::Final { content }) if content.contains("max_tool_iterations")
        ));
        handle.abort();
    }

    #[tokio::test]
    async fn normal_completion_returns_exactly_one_terminal_final() {
        let (endpoint, _bodies, handle) = http_server(vec![TestHttpResponse::ok(
            r#"{"choices":[{"message":{"role":"assistant","content":"ok"}}]}"#,
        )])
        .await;
        let session = OpenAiCompatibleSession::new(
            SessionId::new_random(),
            config(endpoint, None),
            temp_run_dir("single-final"),
            "# system",
            None,
        )
        .expect("session");
        let events = drain(&session, "hello").await;
        let final_positions: Vec<usize> = events
            .iter()
            .enumerate()
            .filter_map(|(idx, event)| matches!(event, ReplyEvent::Final { .. }).then_some(idx))
            .collect();
        assert_eq!(final_positions, vec![events.len() - 1]);
        handle.abort();
    }

    #[tokio::test]
    async fn openai_compatible_conversation_id_is_none() {
        let (endpoint, _bodies, handle) = http_server(vec![TestHttpResponse::ok(
            r#"{"choices":[{"message":{"role":"assistant","content":"ok"}}]}"#,
        )])
        .await;
        let session = OpenAiCompatibleSession::new(
            SessionId::new_random(),
            config(endpoint, None),
            temp_run_dir("conversation-id-none"),
            "# system",
            None,
        )
        .expect("session");
        assert_eq!(session.conversation_id(), None);
        let _ = drain(&session, "hello").await;
        assert_eq!(session.conversation_id(), None);
        handle.abort();
    }

    #[tokio::test]
    async fn unreachable_endpoint_on_first_contact_is_start_error() {
        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
        let addr = listener.local_addr().expect("addr");
        drop(listener);
        let session = OpenAiCompatibleSession::new(
            SessionId::new_random(),
            config_with_timeouts(format!("http://{addr}/v1"), None, 500, 100),
            temp_run_dir("unreachable"),
            "# system",
            None,
        )
        .expect("session");
        match session.send("hello").await {
            Err(AgentSessionError::Start(msg)) => {
                assert!(msg.contains("indisponible"), "unexpected message: {msg}");
            }
            Err(other) => panic!("expected Start, got {other:?}"),
            Ok(_) => panic!("expected first contact failure"),
        }
    }

    #[tokio::test]
    async fn timeout_after_successful_contact_is_io_error() {
        let (endpoint, _bodies, handle) = http_server(vec![
            TestHttpResponse::ok(
                r#"{"choices":[{"message":{"role":"assistant","content":"one"}}]}"#,
            ),
            TestHttpResponse::Hang,
        ])
        .await;
        let session = OpenAiCompatibleSession::new(
            SessionId::new_random(),
            config_with_timeouts(endpoint, None, 100, 100),
            temp_run_dir("mid-turn-timeout"),
            "# system",
            None,
        )
        .expect("session");
        let _ = drain(&session, "first").await;
        match session.send("second").await {
            Err(AgentSessionError::Io(msg)) => {
                assert!(
                    msg.contains("communication endpoint OpenAI-compatible"),
                    "unexpected message: {msg}"
                );
            }
            Err(other) => panic!("expected Io, got {other:?}"),
            Ok(_) => panic!("expected timeout after first contact"),
        }
        handle.abort();
    }

    #[tokio::test]
    async fn http_400_404_422_map_to_start_without_body_leak() {
        for status in [400, 404, 422] {
            let secret_body = r#"{"error":"MODEL_SECRET_SHOULD_NOT_LEAK"}"#;
            let (endpoint, _bodies, handle) = http_server(vec![TestHttpResponse::Status {
                code: status,
                body: secret_body,
            }])
            .await;
            let session = OpenAiCompatibleSession::new(
                SessionId::new_random(),
                config(endpoint, None),
                temp_run_dir(&format!("status-{status}")),
                "# system",
                None,
            )
            .expect("session");
            match session.send("hello").await {
                Err(AgentSessionError::Start(msg)) => {
                    assert!(msg.contains(&format!("HTTP {status}")));
                    assert!(
                        !msg.contains("MODEL_SECRET_SHOULD_NOT_LEAK") && !msg.contains("error"),
                        "raw HTTP body must not leak in error message: {msg}"
                    );
                }
                Err(other) => panic!("expected Start for HTTP {status}, got {other:?}"),
                Ok(_) => panic!("expected HTTP {status} error"),
            }
            handle.abort();
        }
    }

    #[tokio::test]
    async fn tool_rejection_disables_tools_retries_once_and_keeps_conversing() {
        let (endpoint, bodies, handle) = http_server(vec![
            TestHttpResponse::Status {
                code: 400,
                body: r#"{"error":"tools unsupported"}"#,
            },
            TestHttpResponse::ok(
                r#"{"choices":[{"message":{"role":"assistant","content":"chat nu"}}]}"#,
            ),
            TestHttpResponse::ok(
                r#"{"choices":[{"message":{"role":"assistant","content":"encore"}}]}"#,
            ),
        ])
        .await;
        let session = OpenAiCompatibleSession::new(
            SessionId::new_random(),
            config(endpoint, None),
            temp_run_dir("tools-rejected"),
            "# system",
            Some(Arc::new(FakeInvoker::default())),
        )
        .expect("session");

        let first = drain(&session, "hello").await;
        assert_eq!(
            first
                .iter()
                .filter(|event| matches!(event, ReplyEvent::Announcement { .. }))
                .count(),
            1,
            "tool rejection should announce the degradation exactly once"
        );
        assert!(matches!(
            first.last(),
            Some(ReplyEvent::Final { content }) if content == "chat nu"
        ));

        let second = drain(&session, "again").await;
        assert_eq!(
            second
                .iter()
                .filter(|event| matches!(event, ReplyEvent::Announcement { .. }))
                .count(),
            0,
            "tools stay disabled; no repeated degradation announcement"
        );
        assert!(matches!(
            second.last(),
            Some(ReplyEvent::Final { content }) if content == "encore"
        ));

        let bodies = bodies.lock().expect("mutex sain");
        assert_eq!(bodies.len(), 3);
        assert!(bodies[0].contains("\"tools\""), "first request uses tools");
        assert!(
            !bodies[1].contains("\"tools\"") && !bodies[2].contains("\"tools\""),
            "retried and later requests must be plain chat: {bodies:?}"
        );
        handle.abort();
    }
}