mars-terminal 0.3.2

/// LLM agent integration. Most providers speak the OpenAI-compatible chat API;
/// Claude uses Anthropic's own Messages API (separate branch in `chat()`).
/// Env precedence (paid-first): MARS_LLM_* (any endpoint, e.g. local Ollama;
/// legacy ARES_LLM_* still honored) → ANTHROPIC_API_KEY → OPENAI_API_KEY →
/// GROQ_API_KEY → GEMINI_API_KEY / GOOGLE_API_KEY.

use std::sync::mpsc;

/// What the model asked the editor to do (always user-confirmed before firing).
#[derive(Clone, Debug, PartialEq)]
pub enum AgentDirective {
    /// `RUN: <ActionName>` — an editor action from the registry.
    Run(String),
    /// `TYPE: <command>` — type a shell command into the terminal pane.
    Type(String),
    /// `OPEN: path:line` — open a file at a line (cited from a stack trace).
    Open(String),
    /// `NEED: <what>` — a read-side request for more context (W4/W5). Never gated;
    /// Mars re-asks once with the extra source. Not shown to the user.
    Need(NeedKind),
}

/// What extra context the model asked for.
#[derive(Clone, Debug, PartialEq)]
pub enum NeedKind {
    /// Full terminal scrollback of the focused pane (W5).
    Scrollback,
    /// Another tab's panes, by tab name (W4).
    Tab(String),
}

pub enum AgentEvent {
    Answer {
        text: String,
        directive: Option<AgentDirective>,
    },
    /// A streamed ask reply is starting — reset any partial from a prior turn
    /// (the escalation retry starts a fresh stream over the same question).
    AnswerStart,
    /// One streamed chunk of the in-progress ask reply (reasoning-stripped;
    /// the final `Answer` still carries the complete, directive-parsed text).
    AnswerDelta { text: String },
    /// Background tab-naming reply (tab id, proposed name).
    AutoName { tab_id: usize, name: String },
    /// Background session-naming reply (proposed name).
    SessionName { name: String },
    /// W6: one-line verdict on a watched terminal (term id, verdict).
    WatchSummary { term_id: usize, verdict: String },
    /// Background mission inference over the work journal (one line: what the
    /// user is working on). Persisted for `mars ls`.
    Mission { text: String },
    /// A background agent thread finished — clears the `bg_busy` gate even if the
    /// call failed (so one failed request can't wedge all background work).
    BgDone,
    /// W3 shell translate: English → one shell command (fills the SH bar).
    ShellTranslation { command: String, call_id: u64 },
    Error(String),
}

/// Kebab-case, ≤16 chars, alnum+dash — shared by tab/session auto-naming.
fn kebab(text: &str) -> String {
    let s: String = text
        .trim()
        .to_lowercase()
        .chars()
        .map(|c| if c.is_alphanumeric() { c } else { '-' })
        .collect::<String>()
        .split('-')
        .filter(|s| !s.is_empty())
        .collect::<Vec<_>>()
        .join("-");
    s.chars().take(16).collect()
}

/// Match a single directive line, tolerating markdown noise (`- `, backticks,
/// bold) the model sometimes wraps it in.
fn match_directive(line: &str) -> Option<AgentDirective> {
    let l = line
        .trim()
        .trim_start_matches(['-', '*', '>', ' '])
        .trim_matches('`')
        .trim_matches('*')
        .trim();
    // RUN takes only the action token (models sometimes append prose); TYPE and
    // OPEN keep the full rest (commands and paths contain spaces).
    if let Some(rest) = l.strip_prefix("RUN:") {
        if let Some(name) = rest.trim().trim_matches('`').split_whitespace().next() {
            let name = name.trim_end_matches(['.', ',', ':']);
            if !name.is_empty() {
                return Some(AgentDirective::Run(name.to_string()));
            }
        }
    }
    // NEED: read-side context request (W4/W5).
    if let Some(rest) = l.strip_prefix("NEED:") {
        let arg = rest.trim().trim_matches('`').trim();
        let low = arg.to_lowercase();
        if low.starts_with("scrollback") || low.starts_with("history") {
            return Some(AgentDirective::Need(NeedKind::Scrollback));
        }
        if let Some(tab) = low.strip_prefix("tab") {
            let name = tab.trim().to_string();
            if !name.is_empty() {
                return Some(AgentDirective::Need(NeedKind::Tab(name)));
            }
        }
    }
    for (tag, make) in [
        ("TYPE:", AgentDirective::Type as fn(String) -> AgentDirective),
        ("OPEN:", AgentDirective::Open as fn(String) -> AgentDirective),
    ] {
        if let Some(rest) = l.strip_prefix(tag) {
            let arg = rest.trim().trim_matches('`').trim().to_string();
            if !arg.is_empty() {
                return Some(make(arg));
            }
        }
    }
    None
}

/// Split a model reply into display text and a trailing directive. Lenient: the
/// directive may sit on any of the last few non-empty lines (models sometimes
/// add a sign-off after it), and may be wrapped in backticks/list markers.
pub fn parse_directive(text: &str) -> (String, Option<AgentDirective>) {
    let lines: Vec<&str> = text.lines().collect();
    // Scan the last few non-empty lines from the bottom.
    let mut hit: Option<usize> = None;
    for (i, line) in lines.iter().enumerate().rev().take(4) {
        if line.trim().is_empty() {
            continue;
        }
        if match_directive(line).is_some() {
            hit = Some(i);
            break;
        }
    }
    match hit {
        Some(i) => {
            let directive = match_directive(lines[i]);
            let display = lines[..i].join("\n").trim_end().to_string();
            (display, directive)
        }
        None => (text.to_string(), None),
    }
}

#[derive(Clone)]
pub struct AgentConfig {
    pub url: String,
    pub key: String,
    pub model: String,
    pub provider: &'static str,
    pub max_tokens: u32,
    pub temperature: f64,
    /// Set on a remote box behind an `ssh -R` auth socket: the agent proxies the
    /// LLM call home instead of holding a key. `None` = call the provider directly.
    pub broker_sock: Option<String>,
}

/// Read `MARS_<name>`, falling back to the pre-rename `ARES_<name>`.
fn env_var(name: &str) -> Result<String, std::env::VarError> {
    std::env::var(format!("MARS_{name}")).or_else(|_| std::env::var(format!("ARES_{name}")))
}

/// A 429, typed, so the cascade can tell "this family is throttled" (rotate to
/// another one) from every other failure (don't).
#[derive(Debug)]
pub struct RateLimited(pub String);

impl std::fmt::Display for RateLimited {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        self.0.fmt(f)
    }
}

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

/// Every provider with a key in the ambient env, paid-first: explicit
/// MARS_LLM_KEY wins, then a set Claude/OpenAI key (you meant it), then the
/// free Groq/Gemini tiers. Position 0 is what `from_env` picks; the rest are
/// rotation targets when it rate-limits. Cheap defaults per provider
/// (right-size, don't reach for the biggest); override any with MARS_LLM_MODEL.
fn provider_chain() -> Vec<(String, &'static str, &'static str, &'static str)> {
    let mut chain = Vec::new();
    if let Ok(k) = env_var("LLM_KEY") {
        chain.push((k, "custom", "https://api.groq.com/openai/v1", "llama-3.1-8b-instant"));
    }
    if let Ok(k) = std::env::var("ANTHROPIC_API_KEY") {
        // Claude — Anthropic's own Messages API (not OpenAI-compatible; handled
        // in chat()). Haiku is the cheap/fast default.
        chain.push((k, "anthropic", "https://api.anthropic.com", "claude-haiku-4-5"));
    }
    if let Ok(k) = std::env::var("OPENAI_API_KEY") {
        chain.push((k, "openai", "https://api.openai.com/v1", "gpt-4o-mini"));
    }
    if let Ok(k) = std::env::var("GROQ_API_KEY") {
        // Qwen3-32B: strong open model on Groq's fast free tier.
        chain.push((k, "groq", "https://api.groq.com/openai/v1", "qwen/qwen3-32b"));
    }
    if let Ok(k) = std::env::var("GEMINI_API_KEY").or_else(|_| std::env::var("GOOGLE_API_KEY")) {
        // Flash-Lite: cheapest + highest free-tier limits. (Pinned dated
        // versions age out of the free tier, so track the lite line.)
        chain.push((
            k,
            "gemini",
            "https://generativelanguage.googleapis.com/v1beta/openai",
            "gemini-3.1-flash-lite",
        ));
    }
    chain
}

/// Rotation-for-limits targets: the other keyed providers, paid-first. Empty
/// when the user pinned a model/URL or keyed a custom endpoint (an explicit
/// choice is never rotated away from) — and, trivially, with one key.
pub fn rotation_candidates(current_provider: &str) -> Vec<AgentConfig> {
    if env_var("LLM_MODEL").is_ok()
        || env_var("LLM_URL").is_ok()
        || current_provider == "custom"
    {
        return Vec::new();
    }
    provider_chain()
        .into_iter()
        .filter(|(_, p, _, _)| *p != current_provider)
        .map(|(key, provider, url, model)| AgentConfig {
            url: url.to_string(),
            key,
            model: model.to_string(),
            provider,
            max_tokens: 512,
            temperature: 0.3,
            broker_sock: None,
        })
        .collect()
}

impl AgentConfig {
    pub fn from_env() -> Self {
        // Highest precedence: a forwarded auth socket (we're on a remote box).
        // Proxy the call home — the key never lands here. An explicit MARS_LLM_KEY
        // still wins over this, so a box you deliberately keyed keeps working.
        if std::env::var("MARS_LLM_KEY").is_err()
            && std::env::var("ARES_LLM_KEY").is_err()
        {
            if let Some(sock) = crate::broker::detect_broker_sock() {
                return AgentConfig {
                    url: String::new(),
                    key: String::new(),
                    // None-equivalent: empty → the broker picks its own model.
                    model: env_var("LLM_MODEL").unwrap_or_default(),
                    provider: "broker",
                    max_tokens: 512,
                    temperature: 0.3,
                    broker_sock: Some(sock),
                };
            }
        }
        let (key, provider, default_url, default_model) =
            provider_chain().into_iter().next().unwrap_or((
                String::new(),
                "none",
                "https://api.groq.com/openai/v1",
                "llama-3.1-8b-instant",
            ));

        // Explicit URL/model overrides apply to any provider.
        let url = env_var("LLM_URL").unwrap_or_else(|_| default_url.to_string());
        let model = env_var("LLM_MODEL").unwrap_or_else(|_| default_model.to_string());
        AgentConfig { url, key, model, provider, max_tokens: 512, temperature: 0.3, broker_sock: None }
    }

    pub fn is_configured(&self) -> bool {
        if self.provider == "broker" {
            // Honest on the remote: "configured" iff the tunnel is actually up.
            return self
                .broker_sock
                .as_deref()
                .map(|s| std::os::unix::net::UnixStream::connect(s).is_ok())
                .unwrap_or(false);
        }
        !self.key.is_empty()
    }
}

fn system_prompt(registry: &str, screen: &str) -> String {
    // Screen content is user-derived — substitute it last (see prompts.rs).
    crate::prompts::ASK_SYSTEM.replace("{registry}", registry).replace("{screen}", screen)
}

/// Build the chat messages: system + up to the last 12 conversation turns +
/// the new question. Extracted so tests can assert history is really sent.
pub fn build_messages(
    registry: &str,
    screen: &str,
    history: &[(String, String)],
    question: &str,
) -> Vec<serde_json::Value> {
    let mut messages = vec![serde_json::json!({
        "role": "system", "content": system_prompt(registry, screen)
    })];
    let start = history.len().saturating_sub(12);
    for (role, content) in &history[start..] {
        messages.push(serde_json::json!({ "role": role, "content": content }));
    }
    messages.push(serde_json::json!({ "role": "user", "content": question }));
    messages
}

/// Spawn a background thread, POST to OpenAI-compatible chat endpoint, send result back.
pub fn ask(
    cfg: AgentConfig,
    question: String,
    registry: String,
    screen: String,
    history: Vec<(String, String)>,
    tx: mpsc::Sender<AgentEvent>,
) {
    std::thread::spawn(move || {
        // Memory (Axis B): retrieve from Mars's OWN knowledge (docs + tunable knobs)
        // so the agent can answer "how do I…" and propose self-reconfiguration
        // accurately. The action registry is already in the base prompt.
        let mode = crate::retrieval::MemoryMode::from_env();
        let mut messages = build_messages(&registry, &screen, &history, &question);
        if let Some(ctx) = crate::retrieval::docs_context_for(&question) {
            messages.insert(1, serde_json::json!({ "role": "system", "content": ctx }));
        }
        // Stream: tokens render as they arrive; the final Answer still carries
        // the complete, directive-parsed text.
        let _ = tx.send(AgentEvent::AnswerStart);
        let txd = tx.clone();
        let mut on_delta = move |d: &str| {
            let _ = txd.send(AgentEvent::AnswerDelta { text: d.to_string() });
        };
        match chat_with_id_streaming(&cfg, messages.clone(), "ask", mode.as_str(), &mut on_delta) {
            Ok((text, _call_id)) => {
                let (display, directive) = parse_directive(&text);
                // Escalation-for-quality: a RUN: naming no real action is an
                // unambiguous cheap-model failure — retry ONCE, one tier up.
                // The retry is logged as `ask_escalated`, which is unmapped in
                // the ring, so the escalated model passes through model_for
                // unclobbered. A still-bad directive surfaces honestly at the
                // confirm gate.
                if let Some(AgentDirective::Run(name)) = &directive {
                    if crate::palette::Action::from_name(name).is_none() {
                        if let Some(up) = crate::tiers::model_above(cfg.provider, "ask") {
                            let cfg_up = AgentConfig { model: up, ..cfg.clone() };
                            let _ = tx.send(AgentEvent::AnswerStart); // fresh stream
                            if let Ok((text2, _)) = chat_with_id_streaming(
                                &cfg_up,
                                messages,
                                "ask_escalated",
                                mode.as_str(),
                                &mut on_delta,
                            ) {
                                let (display2, directive2) = parse_directive(&text2);
                                let _ = tx.send(AgentEvent::Answer {
                                    text: display2,
                                    directive: directive2,
                                });
                                return;
                            }
                        }
                    }
                }
                let _ = tx.send(AgentEvent::Answer { text: display, directive });
            }
            Err(e) => {
                let _ = tx.send(AgentEvent::Error(e.to_string()));
            }
        }
    });
}

/// Background tab-naming: tiny prompt, no registry, quiet failure.
pub fn auto_name(cfg: AgentConfig, tab_id: usize, screen: String, tx: mpsc::Sender<AgentEvent>) {
    std::thread::spawn(move || {
        let messages = vec![
            serde_json::json!({ "role": "system", "content": crate::prompts::AUTO_NAME_SYSTEM.trim_end() }),
            serde_json::json!({ "role": "user", "content": screen }),
        ];
        // Task tag matches the ring's `auto_name` key (was "auto-name", which
        // silently skipped tier routing).
        if let Ok(text) = chat(&cfg, messages, "auto_name") {
            let name = kebab(&text);
            if !name.is_empty() {
                let _ = tx.send(AgentEvent::AutoName { tab_id, name });
            }
        }
        let _ = tx.send(AgentEvent::BgDone); // release the gate even on failure
    });
}

/// W6: one-line verdict on a watched terminal that went quiet or exited. Runs on
/// a background thread (even inside the detached daemon) and delivers a Notice.
pub fn watch_summary(
    cfg: AgentConfig,
    term_id: usize,
    reason: crate::app::WatchReason,
    tail: String,
    tx: mpsc::Sender<AgentEvent>,
) {
    std::thread::spawn(move || {
        let hint = match reason {
            crate::app::WatchReason::Exit => crate::prompts::WATCH_HINT_EXIT,
            crate::app::WatchReason::Quiet => crate::prompts::WATCH_HINT_QUIET,
        };
        let messages = vec![
            serde_json::json!({ "role": "system",
                "content": crate::prompts::WATCH_SYSTEM.trim_end().replace("{hint}", hint.trim_end()) }),
            serde_json::json!({ "role": "user", "content": tail }),
        ];
        match chat(&cfg, messages, "watch") {
            Ok(text) => {
                let verdict = text.trim().lines().next().unwrap_or("").trim().to_string();
                if !verdict.is_empty() {
                    let _ = tx.send(AgentEvent::WatchSummary { term_id, verdict });
                }
            }
            // Surface the failure instead of going silent (and clear the gate).
            Err(e) => {
                let _ = tx.send(AgentEvent::WatchSummary {
                    term_id,
                    verdict: format!("⚠ watch couldn't summarize — {e}"),
                });
            }
        }
        let _ = tx.send(AgentEvent::BgDone); // always release the gate
    });
}

/// Background mission inference: read the recent work-journal snapshots and
/// name, in one line, what the user is working on. Quiet failure — a mission
/// is a nicety, never worth a notice.
pub fn infer_mission(cfg: AgentConfig, snapshots: Vec<String>, tx: mpsc::Sender<AgentEvent>) {
    std::thread::spawn(move || {
        let messages = vec![
            serde_json::json!({ "role": "system", "content": crate::prompts::MISSION_SYSTEM.trim_end() }),
            serde_json::json!({ "role": "user", "content": snapshots.join("\n") }),
        ];
        if let Ok(text) = chat(&cfg, messages, "mission") {
            let mission = text.trim().lines().next().unwrap_or("").trim().to_string();
            if !mission.is_empty() {
                let _ = tx.send(AgentEvent::Mission { text: mission });
            }
        }
        let _ = tx.send(AgentEvent::BgDone);
    });
}

/// Background session-naming — like tab naming but for the whole session.
pub fn name_session(cfg: AgentConfig, screen: String, tx: mpsc::Sender<AgentEvent>) {
    std::thread::spawn(move || {
        let messages = vec![
            serde_json::json!({ "role": "system", "content": crate::prompts::NAME_SESSION_SYSTEM.trim_end() }),
            serde_json::json!({ "role": "user", "content": screen }),
        ];
        // Tag matches the ring's `name_session` key (was "session-name").
        if let Ok(text) = chat(&cfg, messages, "name_session") {
            let name = kebab(&text);
            if !name.is_empty() {
                let _ = tx.send(AgentEvent::SessionName { name });
            }
        }
        let _ = tx.send(AgentEvent::BgDone); // release the gate even on failure
    });
}

/// W3: turn an English request into ONE shell command (no prose, no fences).
/// ALWAYS sends exactly one event so the caller's spinner can never wedge.
/// Synchronous shell translation with memory retrieval + logging. Returns the
/// extracted command and its `call_id`. Shared by the async composer and the
/// headless `mars translate` primitive the Python eval harness drives.
///
/// Memory (Axis A): when the active [`crate::retrieval::MemoryMode`] includes
/// history, the user's own past `(request → command)` pairs + recent shell history
/// are retrieved and shown as few-shot examples — the "sits at the terminal"
/// advantage a standalone translator can't have. The variant is logged.
/// True for models that emit an internal `<think>` block (Qwen3, QwQ, DeepSeek-R1,
/// OpenAI o-series). Only these get the "cap your reasoning" prompt clause — see
/// [`translate_once`] for why applying it to non-reasoning models breaks them.
fn is_reasoning_model(model: &str) -> bool {
    let m = model.to_lowercase();
    ["qwen3", "qwq", "deepseek-r1", "-r1", "o1-", "o3", "o4-mini", "thinking", "reasoning"]
        .iter()
        .any(|p| m.contains(p))
}

pub fn translate_once(cfg: &AgentConfig, request: &str, screen: &str) -> anyhow::Result<(String, u64)> {
    let mode = crate::retrieval::MemoryMode::from_env();
    let examples = crate::retrieval::fewshot_for(request);
    // Reasoning models (Qwen3, R1, o-series) burn the token budget on a <think> block,
    // so we cap their reasoning. Non-reasoning models (Gemini Flash-Lite, gpt-4o-mini,
    // Haiku) read that same instruction as a cue to reason *silently* and can return an
    // EMPTY completion — so the cap is applied ONLY to reasoning models.
    let reasoning_cap = if is_reasoning_model(&cfg.model) {
        format!(" {}", crate::prompts::TRANSLATE_REASONING_CAP.trim())
    } else {
        String::new()
    };
    let examples_block = if examples.is_empty() {
        String::new()
    } else {
        format!(
            "\n\n{}",
            crate::prompts::TRANSLATE_EXAMPLES.trim_end().replace("{examples}", &examples)
        )
    };
    let system = crate::prompts::TRANSLATE_SYSTEM
        .trim_end()
        .replace("{reasoning_cap}", &reasoning_cap)
        .replace("{examples_block}", &examples_block);
    let messages = vec![
        serde_json::json!({ "role": "system", "content": system }),
        serde_json::json!({ "role": "user", "content": format!("SCREEN:\n{screen}\n\nREQUEST: {request}") }),
    ];
    let (text, call_id) = chat_with_id(cfg, messages, "shell", mode.as_str())?;
    let command = text
        .trim()
        .trim_matches('`')
        .lines()
        .find(|l| !l.trim().is_empty())
        .unwrap_or("")
        .trim()
        .trim_start_matches("$ ")
        .to_string();
    Ok((command, call_id))
}

pub fn translate_shell(cfg: AgentConfig, request: String, screen: String, tx: mpsc::Sender<AgentEvent>) {
    std::thread::spawn(move || {
        let ev = match translate_once(&cfg, &request, &screen) {
            Ok((command, _)) if command.is_empty() => {
                AgentEvent::Error("couldn't translate that — rephrase and retry".into())
            }
            Ok((command, call_id)) => AgentEvent::ShellTranslation { command, call_id },
            Err(e) => AgentEvent::Error(e.to_string()),
        };
        let _ = tx.send(ev);
    });
}

/// Extract "retry in 14.89s"-style hints from a 429 message → whole seconds.
pub fn retry_secs(msg: &str) -> Option<u64> {
    let after = msg.split("retry in ").nth(1)?;
    let num: String = after.chars().take_while(|c| c.is_ascii_digit() || *c == '.').collect();
    num.parse::<f64>().ok().map(|s| s.ceil() as u64)
}

/// Strip `<think>…</think>` reasoning blocks (Qwen3, DeepSeek-R1, etc.) so only
/// the user-facing answer + directive remain.
fn strip_reasoning(text: &str) -> String {
    let mut out = text.to_string();
    while let (Some(a), Some(b)) = (out.find("<think>"), out.find("</think>")) {
        if a < b {
            out.replace_range(a..b + "</think>".len(), "");
        } else {
            break;
        }
    }
    // A dangling, unclosed <think> → drop everything from it.
    if let Some(a) = out.find("<think>") {
        out.truncate(a);
    }
    out.trim().to_string()
}

/// Optional streaming sink: receives each visible chunk as it arrives.
type DeltaSink<'a> = Option<&'a mut dyn FnMut(&str)>;

/// Reborrow the sink for one call without consuming it (the rotation loop may
/// hand it to several attempts in sequence).
fn reborrow<'b>(sink: &'b mut DeltaSink<'_>) -> DeltaSink<'b> {
    match sink {
        Some(s) => Some(&mut **s),
        None => None,
    }
}

/// The safely-streamable prefix of a partially-received reply: closed <think>
/// blocks removed, an unclosed one truncated, and a trailing partial "<think>"
/// tag held back — so reasoning never flashes on screen and emitted text never
/// retracts as more chunks land.
pub(crate) fn stream_visible(raw: &str) -> String {
    let s = strip_reasoning(raw);
    let tag = "<think>";
    let mut cut = s.len();
    for k in (1..tag.len()).rev() {
        if s.ends_with(&tag[..k]) {
            cut = s.len() - k;
            break;
        }
    }
    // trim_end in EVERY path: strip_reasoning trims, so an untrimmed hold-back
    // result could exceed a later trimmed one — emitted text must never retract.
    s[..cut].trim_end().to_string()
}

/// Single choke point for every LLM call. `task` tags the call; `retrieval` names
/// the memory variant that shaped the prompt ("n/a" when the path doesn't retrieve).
/// Times the call, captures real token usage, logs it under a fresh `call_id` when
/// MARS_LLM_DEBUG is on, and returns `(reasoning-stripped reply, call_id)`. The
/// call_id lets a later behavioral outcome (accept/edit/reject) be correlated.
pub fn chat_with_id(
    cfg: &AgentConfig,
    messages: Vec<serde_json::Value>,
    task: &str,
    retrieval: &str,
) -> anyhow::Result<(String, u64)> {
    chat_inner(cfg, messages, task, retrieval, None)
}

/// `chat_with_id`, streaming: `on_delta` receives each visible chunk as it
/// arrives. The returned text is the complete reply, identical to what the
/// non-streaming path produces — directive parsing still needs the whole.
pub fn chat_with_id_streaming(
    cfg: &AgentConfig,
    messages: Vec<serde_json::Value>,
    task: &str,
    retrieval: &str,
    on_delta: &mut dyn FnMut(&str),
) -> anyhow::Result<(String, u64)> {
    chat_inner(cfg, messages, task, retrieval, Some(on_delta))
}

fn chat_inner(
    cfg: &AgentConfig,
    messages: Vec<serde_json::Value>,
    task: &str,
    retrieval: &str,
    mut sink: DeltaSink,
) -> anyhow::Result<(String, u64)> {
    // Remote box: proxy the whole call home over the forwarded socket. No key,
    // no Authorization header, ever constructed here. (Logged home-side.)
    // Frame-at-a-time protocol — the remote path does not stream.
    if cfg.provider == "broker" {
        let sock = cfg
            .broker_sock
            .as_deref()
            .ok_or_else(|| anyhow::anyhow!("broker mode with no socket"))?;
        return crate::broker::chat_via_broker(sock, cfg, messages).map(|t| (t, 0));
    }

    match attempt(cfg, &messages, task, retrieval, reborrow(&mut sink)) {
        // Rotation-for-limits: each provider meters on its own counter, so a
        // throttled call can often complete elsewhere at the same tier. Any
        // alternate failure just moves on; exhaustion surfaces the original 429.
        // A 429 arrives as an HTTP status BEFORE any token streams, so no
        // partial output can precede a rotation.
        Err(e) if e.downcast_ref::<RateLimited>().is_some() => {
            for alt in rotation_candidates(cfg.provider) {
                if let Ok(ok) = attempt(&alt, &messages, task, retrieval, reborrow(&mut sink)) {
                    return Ok(ok);
                }
            }
            Err(e)
        }
        r => r,
    }
}

/// One provider attempt: tier-resolve, call, log. Split from `chat_inner` so
/// the rotation loop logs every attempt as its own call record.
fn attempt(
    cfg: &AgentConfig,
    messages: &[serde_json::Value],
    task: &str,
    retrieval: &str,
    sink: DeltaSink,
) -> anyhow::Result<(String, u64)> {
    let call_id = crate::llm_log::next_call_id();
    // Model-tier ring: route this task to its tier's model (an explicit
    // MARS_LLM_MODEL still wins — that check lives inside model_for).
    let resolved = crate::tiers::model_for(cfg.provider, task, &cfg.model);
    let cfg = &AgentConfig { model: resolved, ..cfg.clone() };
    let start = std::time::Instant::now();
    let result = {
        // Guard the caller's sink: accumulate raw chunks, re-strip, and emit
        // only the growth of the visible prefix — reasoning-model <think>
        // output never reaches the screen, even split across chunk boundaries.
        let mut raw = String::new();
        let mut emitted = 0usize;
        let mut wrapped;
        let provider_sink: DeltaSink = match sink {
            Some(on_delta) => {
                wrapped = move |d: &str| {
                    raw.push_str(d);
                    let vis = stream_visible(&raw);
                    if vis.len() > emitted {
                        on_delta(&vis[emitted..]);
                        emitted = vis.len();
                    }
                };
                Some(&mut wrapped as &mut dyn FnMut(&str))
            }
            None => None,
        };
        if cfg.provider == "anthropic" {
            chat_anthropic(cfg, messages, provider_sink)
        } else {
            chat_openai(cfg, messages, provider_sink)
        }
    };
    let latency_ms = start.elapsed().as_millis() as u64;

    match result {
        Ok((text, pt, ct)) => {
            crate::llm_log::record(&crate::llm_log::CallRecord {
                call_id, task, provider: cfg.provider, model: &cfg.model, retrieval,
                prompt_tokens: pt, completion_tokens: ct, latency_ms,
                ok: true, error: None, input: messages, output: &text,
            });
            Ok((strip_reasoning(&text), call_id))
        }
        Err(e) => {
            let msg = e.to_string();
            crate::llm_log::record(&crate::llm_log::CallRecord {
                call_id, task, provider: cfg.provider, model: &cfg.model, retrieval,
                prompt_tokens: 0, completion_tokens: 0, latency_ms,
                ok: false, error: Some(&msg), input: messages, output: "",
            });
            Err(e)
        }
    }
}

/// Convenience wrapper for callers that don't need the `call_id` correlation
/// (watch / auto-name / session-name / remote) and don't retrieve.
pub fn chat(cfg: &AgentConfig, messages: Vec<serde_json::Value>, task: &str) -> anyhow::Result<String> {
    chat_with_id(cfg, messages, task, "n/a").map(|(text, _)| text)
}

/// OpenAI-compatible providers (OpenAI, Groq, Gemini's OpenAI shim, custom,
/// Ollama). Returns (raw_text, prompt_tokens, completion_tokens). With a sink,
/// requests SSE and forwards each content delta as it arrives.
fn chat_openai(
    cfg: &AgentConfig,
    messages: &[serde_json::Value],
    sink: DeltaSink,
) -> anyhow::Result<(String, u64, u64)> {
    let url = format!("{}/chat/completions", cfg.url);
    let mut body = serde_json::json!({
        "model": cfg.model,
        "messages": messages,
        "max_tokens": cfg.max_tokens,
        "temperature": cfg.temperature
    });
    if sink.is_some() {
        body["stream"] = serde_json::json!(true);
        // Usage-in-final-chunk is an opt-in extension; only request it where
        // it's known-supported (other shims reject unknown fields).
        if matches!(cfg.provider, "openai" | "groq") {
            body["stream_options"] = serde_json::json!({ "include_usage": true });
        }
    }

    // Bound the call so a stalled connection surfaces as an error instead of
    // hanging the agent (and the spinner) forever.
    let resp = match ureq::post(&url)
        .timeout(std::time::Duration::from_secs(30))
        .set("Authorization", &format!("Bearer {}", cfg.key))
        .set("Content-Type", "application/json")
        .send_json(body)
    {
        Ok(r) => r,
        // Pull the real message out of the error body (bad key, quota, etc.).
        // Gemini wraps it in a JSON array: [{"error":{"message": …}}].
        Err(ureq::Error::Status(code, r)) => {
            let body = r.into_string().unwrap_or_default();
            let api_msg = serde_json::from_str::<serde_json::Value>(&body).ok().and_then(|j| {
                let node = if j.is_array() { j[0].clone() } else { j };
                node["error"]["message"].as_str().map(str::to_string)
            });
            let msg = match code {
                429 => match api_msg.as_deref().and_then(retry_secs) {
                    Some(s) => format!(
                        "rate limit reached — wait ~{s}s and retry (free tier). \
                         Tip: raise limits, switch model with MARS_LLM_MODEL, or use \
                         GROQ_API_KEY / a local Ollama via MARS_LLM_URL."
                    ),
                    None => "rate limit reached (free tier) — wait ~30s and retry, or \
                             switch model/provider (MARS_LLM_MODEL / GROQ_API_KEY / \
                             MARS_LLM_URL for local Ollama)."
                        .to_string(),
                },
                401 | 403 => format!(
                    "auth failed — check your API key. ({})",
                    api_msg.as_deref().unwrap_or("invalid credentials")
                ),
                _ => api_msg.unwrap_or_else(|| format!("HTTP {code}")),
            };
            if code == 429 {
                return Err(anyhow::Error::new(RateLimited(msg)));
            }
            anyhow::bail!("{msg}");
        }
        Err(e) => anyhow::bail!("{e}"),
    };

    if let Some(on_delta) = sink {
        use std::io::BufRead;
        let mut text = String::new();
        let (mut pt, mut ct) = (0u64, 0u64);
        for line in std::io::BufReader::new(resp.into_reader()).lines() {
            let line = line?;
            let Some(data) = line.strip_prefix("data:") else { continue };
            let data = data.trim();
            if data == "[DONE]" {
                break;
            }
            let Ok(j) = serde_json::from_str::<serde_json::Value>(data) else { continue };
            if let Some(msg) = j["error"]["message"].as_str() {
                anyhow::bail!("{msg}");
            }
            if let Some(d) = j["choices"][0]["delta"]["content"].as_str() {
                if !d.is_empty() {
                    text.push_str(d);
                    on_delta(d);
                }
            }
            if j["usage"].is_object() {
                pt = j["usage"]["prompt_tokens"].as_u64().unwrap_or(pt);
                ct = j["usage"]["completion_tokens"].as_u64().unwrap_or(ct);
            }
        }
        return Ok((text, pt, ct));
    }

    let json: serde_json::Value = resp.into_json()?;
    if let Some(msg) = json["error"]["message"].as_str() {
        anyhow::bail!("{msg}");
    }
    let text = json["choices"][0]["message"]["content"].as_str().unwrap_or("").to_string();
    let pt = json["usage"]["prompt_tokens"].as_u64().unwrap_or(0);
    let ct = json["usage"]["completion_tokens"].as_u64().unwrap_or(0);
    Ok((text, pt, ct))
}

/// Anthropic Messages API — NOT OpenAI-compatible: system is a top-level field
/// (not a message role), auth is `x-api-key` + `anthropic-version`, the reply is
/// an array of content blocks, and usage is input/output_tokens. With a sink,
/// requests SSE (`content_block_delta` events carry the text).
fn chat_anthropic(
    cfg: &AgentConfig,
    messages: &[serde_json::Value],
    sink: DeltaSink,
) -> anyhow::Result<(String, u64, u64)> {
    // Split the system message(s) out of the OpenAI-style array.
    let mut system = String::new();
    let mut msgs: Vec<serde_json::Value> = Vec::new();
    for m in messages {
        if m["role"].as_str() == Some("system") {
            if !system.is_empty() {
                system.push('\n');
            }
            system.push_str(m["content"].as_str().unwrap_or(""));
        } else {
            msgs.push(m.clone());
        }
    }
    let url = format!("{}/v1/messages", cfg.url);
    // No `temperature`: the newest Claude models (Sonnet/Haiku 4.5+, Opus 4.x) reject it
    // as deprecated, and all models fall back to a sane default when it is omitted.
    let mut body = serde_json::json!({
        "model": cfg.model,
        "max_tokens": cfg.max_tokens,
        "system": system,
        "messages": msgs
    });
    if sink.is_some() {
        body["stream"] = serde_json::json!(true);
    }
    let resp = match ureq::post(&url)
        .timeout(std::time::Duration::from_secs(30))
        .set("x-api-key", &cfg.key)
        .set("anthropic-version", "2023-06-01")
        .set("Content-Type", "application/json")
        .send_json(body)
    {
        Ok(r) => r,
        Err(ureq::Error::Status(code, r)) => {
            let body = r.into_string().unwrap_or_default();
            let api_msg = serde_json::from_str::<serde_json::Value>(&body)
                .ok()
                .and_then(|j| j["error"]["message"].as_str().map(str::to_string));
            let msg = match code {
                429 => "rate limit reached (Anthropic) — wait and retry, or switch model \
                        with MARS_LLM_MODEL."
                    .to_string(),
                401 | 403 => format!(
                    "auth failed — check ANTHROPIC_API_KEY. ({})",
                    api_msg.as_deref().unwrap_or("invalid credentials")
                ),
                _ => api_msg.unwrap_or_else(|| format!("HTTP {code}")),
            };
            if code == 429 {
                return Err(anyhow::Error::new(RateLimited(msg)));
            }
            anyhow::bail!("{msg}");
        }
        Err(e) => anyhow::bail!("{e}"),
    };

    if let Some(on_delta) = sink {
        use std::io::BufRead;
        let mut text = String::new();
        let (mut pt, mut ct) = (0u64, 0u64);
        for line in std::io::BufReader::new(resp.into_reader()).lines() {
            let line = line?;
            let Some(data) = line.strip_prefix("data:") else { continue };
            let Ok(j) = serde_json::from_str::<serde_json::Value>(data.trim()) else { continue };
            match j["type"].as_str().unwrap_or("") {
                "content_block_delta" => {
                    if let Some(d) = j["delta"]["text"].as_str() {
                        if !d.is_empty() {
                            text.push_str(d);
                            on_delta(d);
                        }
                    }
                }
                "message_start" => {
                    pt = j["message"]["usage"]["input_tokens"].as_u64().unwrap_or(0);
                }
                "message_delta" => {
                    ct = j["usage"]["output_tokens"].as_u64().unwrap_or(ct);
                }
                "error" => {
                    anyhow::bail!("{}", j["error"]["message"].as_str().unwrap_or("stream error"));
                }
                _ => {}
            }
        }
        return Ok((text, pt, ct));
    }

    let json: serde_json::Value = resp.into_json()?;
    if let Some(msg) = json["error"]["message"].as_str() {
        anyhow::bail!("{msg}");
    }
    // content is an array of blocks; concatenate the text ones.
    let text = json["content"]
        .as_array()
        .map(|blocks| blocks.iter().filter_map(|b| b["text"].as_str()).collect::<Vec<_>>().join(""))
        .unwrap_or_default();
    let pt = json["usage"]["input_tokens"].as_u64().unwrap_or(0);
    let ct = json["usage"]["output_tokens"].as_u64().unwrap_or(0);
    Ok((text, pt, ct))
}