supercode-runtime 0.5.49

//! The model transport.
//!
//! [`OpenAiProvider`] speaks the OpenAI chat-completions wire format and
//! defaults to OpenRouter, so a single implementation reaches Claude, GPT,
//! Gemini, Llama, and anything else OpenRouter (or another OpenAI-compatible
//! gateway) exposes. Streaming is used so callers can render tokens live.

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

use async_trait::async_trait;
use futures::StreamExt;
use serde::{Deserialize, Serialize};

use supercode_interchange::{ChatMessage, FunctionCall, Role, ToolCall};

use crate::{CachePlan, ChatRequest, Result, RuntimeError as Error, ToolSchema, Usage};

/// Bounds TCP/TLS establishment for the provider HTTP client. Matches
/// `doctor`'s 10 s timeout (`crates/cli/src/main.rs`) for consistency.
const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);

/// Per-read-operation idle timeout. Resets on every received chunk, so a live
/// SSE stream emitting deltas is never killed — only a silent connection (no
/// bytes for the window, including a server that accepts but never sends
/// response headers) errors out. Generous enough for slow time-to-first-token,
/// small enough to unstick a dead connection well within one agent turn.
const READ_IDLE_TIMEOUT: Duration = Duration::from_secs(120);

/// Maximum number of retries for the *initial* request (so at most
/// `MAX_RETRIES + 1` attempts total). Only connection-level failures and 5xx
/// responses are retried; once SSE streaming has begun, errors propagate as-is.
const MAX_RETRIES: u32 = 2;

/// Base backoff between retries; the delay for attempt `n` (0-indexed) is
/// `RETRY_BACKOFF_BASE * 2^n` (no jitter — not needed at this scale).
const RETRY_BACKOFF_BASE: Duration = Duration::from_millis(500);

/// Crate-internal knobs for the provider's HTTP client and retry behavior.
/// `connect_timeout`/`read_idle_timeout` stay test-only overrides (no
/// `Config`/CLI surface); `max_retries`/`retry_backoff_base` gained one via
/// [`Self::from_retry_config`] (P4b, §1.1/§3.1 `core.retry`) — see that
/// constructor's doc comment.
#[derive(Debug, Clone, Copy)]
#[doc(hidden)]
pub struct HttpOptions {
    pub(crate) connect_timeout: Duration,
    pub(crate) read_idle_timeout: Duration,
    pub(crate) max_retries: u32,
    pub(crate) retry_backoff_base: Duration,
}

impl Default for HttpOptions {
    fn default() -> Self {
        HttpOptions {
            connect_timeout: CONNECT_TIMEOUT,
            read_idle_timeout: READ_IDLE_TIMEOUT,
            max_retries: MAX_RETRIES,
            retry_backoff_base: RETRY_BACKOFF_BASE,
        }
    }
}

impl HttpOptions {
    /// P4b (design §5.2 "P4", §1.1/§3.1 `core.retry`, pi§3 naming
    /// precedent): derive the transport's retry behavior from
    /// [`crate::Config`]'s `retry_*` fields, keeping every other
    /// [`HttpOptions`] field at its built-in default. `enabled = false`
    /// (a NEW capability — today's transport retry has no off-switch) forces
    /// `max_retries` to `0`; `enabled = true` (the [`crate::Config`] default,
    /// matching today's always-on behavior) keeps retrying, using
    /// `max_retries`/`base_delay_ms` to OVERRIDE the built-in
    /// [`MAX_RETRIES`]/[`RETRY_BACKOFF_BASE`] when `Some`, else leaving them
    /// untouched — so a `Config` that sets none of the three `retry_*`
    /// fields (today's only reachable shape, pre-P4b) produces an
    /// [`HttpOptions`] byte-identical to [`HttpOptions::default`].
    #[doc(hidden)]
    pub fn from_retry_config(
        enabled: bool,
        max_retries: Option<u32>,
        base_delay_ms: Option<u64>,
    ) -> HttpOptions {
        let base = HttpOptions::default();
        HttpOptions {
            max_retries: if enabled {
                max_retries.unwrap_or(base.max_retries)
            } else {
                0
            },
            retry_backoff_base: base_delay_ms
                .map(Duration::from_millis)
                .unwrap_or(base.retry_backoff_base),
            ..base
        }
    }
}

/// BP-7 (catalog §4a "Auto-retry on transient provider errors"): one
/// transient failure the transport retried.
///
/// The retry loop itself is unchanged and pre-existing; before BP-7 it was
/// simply SILENT — the ledger row's residue was "no retry record is
/// persisted, so a retried request is invisible to the event stream, the
/// transcript, and `--trace`". A notice is recorded the moment the loop
/// decides to sleep and try again.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RetryNotice {
    /// 0-based index of the attempt that FAILED (attempt 0 is the first try).
    pub attempt: u32,
    /// Backoff slept before the next attempt, milliseconds.
    pub delay_ms: u64,
    /// One-line reason — an HTTP status line or a transport error.
    pub reason: String,
}

/// A shared, drainable buffer of [`RetryNotice`]s.
///
/// Wired as an `Arc` the agent and the provider both hold, rather than a
/// callback on the [`Provider`] trait: the provider is built inside
/// `Agent::new`, long before an event sink is installed, and every mock
/// provider in the test suite would otherwise have to grow a method it does
/// not use. The agent drains this after each `complete()` call, so notices
/// always attach to the round-trip that produced them.
#[derive(Debug, Default)]
pub struct RetryLog {
    notices: std::sync::Mutex<Vec<RetryNotice>>,
}

impl RetryLog {
    /// Record one retry.
    pub fn record(&self, notice: RetryNotice) {
        self.notices
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .push(notice);
    }

    /// Take everything recorded so far, leaving the log empty.
    pub fn drain(&self) -> Vec<RetryNotice> {
        std::mem::take(
            &mut *self
                .notices
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner),
        )
    }
}

/// Build the JSON request body for an OpenAI-compatible chat-completions call.
/// Exposed (crate-internal) so the wire shape can be unit-tested without a
/// network round-trip.
pub(crate) fn build_request_body(req: &ChatRequest, stream: bool) -> serde_json::Value {
    use serde_json::json;
    let mut body = json!({
        "model": req.model,
        "messages": req.messages,
        "stream": stream,
    });
    let obj = body.as_object_mut().unwrap();
    if !req.tools.is_empty() {
        obj.insert(
            "tools".into(),
            serde_json::to_value(req.tools.iter().map(WireTool::from).collect::<Vec<_>>()).unwrap(),
        );
    }
    if let Some(t) = req.temperature {
        obj.insert("temperature".into(), json!(t));
    }
    if let Some(m) = req.max_tokens {
        obj.insert("max_tokens".into(), json!(m));
    }
    if let Some(e) = &req.effort {
        obj.insert("reasoning_effort".into(), json!(e));
    }
    if let Some(rf) = &req.response_format {
        obj.insert("response_format".into(), rf.clone());
    }
    // BP-13 (D9 "Fast mode / service tiers"): the priority/fast variant
    // toggle, sent as the OpenAI-compatible `service_tier` field.
    if let Some(tier) = &req.service_tier {
        obj.insert("service_tier".into(), json!(tier));
    }
    // BP-13 (D9 "Reasoning effort / thinking budgets"): the BUDGET half.
    // `reasoning_effort` above carries the LEVEL; the token cap rides the
    // unified `reasoning` object, which is how an OpenAI-compatible gateway
    // spells Anthropic's `thinking.budget_tokens` and OpenAI's reasoning
    // cap under one name. Merged into any `reasoning` object a caller
    // already placed via `extra_body` (which still wins last, below).
    if let Some(budget) = req.thinking_budget {
        let entry = obj
            .entry("reasoning".to_string())
            .or_insert_with(|| json!({}));
        if let Some(o) = entry.as_object_mut() {
            o.insert("max_tokens".into(), json!(budget));
        }
    }
    if stream {
        obj.insert("stream_options".into(), json!({"include_usage": true}));
    }
    // Provider-native passthrough wins last (lets callers override anything).
    for (k, v) in &req.extra_body {
        obj.insert(k.clone(), v.clone());
    }
    body
}

/// SPEC.md B7: annotate a CLONE of `messages` with Anthropic-style
/// `cache_control: {"type":"ephemeral"}` prompt-cache breakpoints, message-level
/// (never the top-level `extra_body` passthrough `build_request_body` supports
/// for other provider knobs — OpenRouter's Anthropic cache keys off per-message
/// `cache_control` inside the `content` array, so only this placement can say
/// where the stable prefix ends).
///
/// `imported_prefix_len` counts leading messages of `messages` (from index 0,
/// inclusive of the system message) that make up the stable, byte-identical-
/// across-turns prefix — a caller's own leading system message plus every
/// message of a previously-imported session (`Agent::load_session`). Under
/// [`CachePlan::ImportedPrefix`], two breakpoints are placed (Anthropic allows
/// up to 4): `messages[0]` (the system message) and
/// `messages[imported_prefix_len - 1]` (the LAST message of the imported
/// prefix) — deduplicated when they're the same index. Each target message's
/// `content` moves into `content_parts` form with a trailing
/// `{"type":"text","text":…,"cache_control":{"type":"ephemeral"}}` part; an
/// already-multimodal message gets the annotation on its LAST existing text
/// part instead of growing a new one.
///
/// [`CachePlan::Off`] (or a missing/zero `imported_prefix_len`) returns an
/// unannotated clone. Either way this never mutates `messages` in place — the
/// purity requirement (SPEC.md B7-AC2) that `Agent::history` and the sidecar
/// never see `cache_control` depends on this being a read-only projection over
/// a caller-owned copy, never the retained history itself.
#[doc(hidden)]
pub fn apply_cache_plan(
    messages: &[ChatMessage],
    plan: CachePlan,
    imported_prefix_len: Option<usize>,
) -> Vec<ChatMessage> {
    let mut out = messages.to_vec();
    if !matches!(plan, CachePlan::ImportedPrefix) {
        return out;
    }
    let Some(len) = imported_prefix_len.filter(|&n| n > 0) else {
        return out;
    };
    let last = len - 1;
    let mut targets = vec![0usize];
    if last != 0 {
        targets.push(last);
    }
    for idx in targets {
        if let Some(msg) = out.get_mut(idx) {
            annotate_cache_breakpoint(msg);
        }
    }
    out
}

/// Move `msg`'s text content into an ephemeral-cache-annotated
/// `content_parts` entry — see [`apply_cache_plan`].
fn annotate_cache_breakpoint(msg: &mut ChatMessage) {
    let cache_control = serde_json::json!({"type": "ephemeral"});
    if let Some(parts) = msg.content_parts.as_mut() {
        // Already multimodal: annotate the LAST existing text part.
        if let Some(text_part) = parts
            .iter_mut()
            .rev()
            .find(|p| p.get("type").and_then(serde_json::Value::as_str) == Some("text"))
        {
            if let Some(obj) = text_part.as_object_mut() {
                obj.insert("cache_control".to_string(), cache_control);
            }
        }
        return;
    }
    let text = msg.content.take().unwrap_or_default();
    msg.content_parts = Some(vec![serde_json::json!({
        "type": "text",
        "text": text,
        "cache_control": cache_control,
    })]);
}

/// TR-8 (T5): whether the advertised tool-schema tier configuration changed
/// since the last request this agent built. Under [`CachePlan::ImportedPrefix`]
/// this is a cache-bust event: the `tools` array sent alongside `messages` is
/// part of the cache key on the prompt-caching implementations this plan
/// targets, so a byte-identical imported-message prefix does not, on its
/// own, guarantee a cache hit once the advertised schema set has been
/// reshaped by a tier change.
///
/// `previous` is `None` on an agent's very first request (nothing to have
/// busted yet), so this only ever fires from the second request onward, and
/// only for the one request immediately after the change — the caller
/// (`Agent::build_request_messages`) is expected to record the new signature
/// right after consulting this, so the NEXT request (same tier) is not
/// flagged again.
#[doc(hidden)]
pub fn tier_change_is_cache_bust(previous: Option<u64>, current: u64) -> bool {
    previous.is_some_and(|p| p != current)
}

/// UX-26 (B7-warn): Anthropic's default ephemeral prompt-cache TTL, in
/// seconds. Every breakpoint supercode places
/// ([`annotate_cache_breakpoint`]) is `{"type":"ephemeral"}` — never the
/// extended 1-hour-beta `ttl` field — so 5 minutes is the correct assumption
/// for every cache-annotated request this binary sends (Anthropic's
/// documented default TTL for an ephemeral breakpoint with no `ttl` set).
pub(crate) const CACHE_TTL_SECS: i64 = 300;

/// UX-26: cache-read ratio below which a completed, reuse-expected turn is
/// treated as an unexpected miss rather than provider-side rounding/paging
/// noise. Anthropic bills cache reads as an exact token count (not an
/// estimate), so a genuine warm hit reports at or near 100% of the
/// protected prefix's tokens; anything under 10% reflects a real miss.
pub(crate) const CACHE_MISS_RATIO_THRESHOLD: f64 = 0.10;

/// UX-26 T1 (accuracy fold-in): cache-read ratio at/above which a completed
/// turn's OWN `usage` is strong enough evidence to override an
/// idle-time-based [`CacheColdReason::Stale`] verdict. `idle_secs` is a
/// cross-process, timestamp-derived signal (see `cache_cold_reason`'s doc
/// comment) that can be stale itself — e.g. a sibling process re-resumes the
/// SAME original session file (whose on-disk timestamps never advance) and
/// warms the identical prefix within the TTL; this process's `idle_secs`
/// still reads as "past the TTL" even though the provider just proved
/// otherwise. Deliberately the exact mirror of
/// [`CACHE_MISS_RATIO_THRESHOLD`] (`1.0 -` that bar) rather than reusing it
/// directly: reusing 10% (i.e. "disprove whenever it's not already a Miss")
/// would let a merely-ambiguous ratio — e.g. 50%, no stronger evidence of
/// warmth than of staleness — silently swallow a genuinely cold turn. 90%
/// demands the same "at or near 100%" standard the Miss check already uses
/// to call a hit warm, applied in the opposite direction, so a turn only
/// suppresses `Stale` when its own usage affirmatively looks warm — not
/// merely "not obviously a miss."
pub(crate) const CACHE_STALE_DISPROVE_RATIO_THRESHOLD: f64 = 1.0 - CACHE_MISS_RATIO_THRESHOLD;

/// UX-26 T2 (accuracy fold-in): whether `model` is Anthropic-family, i.e.
/// whether [`CacheColdReason::message`]'s Anthropic-shaped wording (a fixed
/// 5-minute ephemeral TTL, cache-read ratio semantics) actually describes
/// the provider this request is going to. Every resolved model slug this
/// binary sends is either an OpenRouter-style `vendor/model` slug — see
/// `userconfig::alias_table` and [`KNOWN_MODEL_CONTEXT_LIMITS`], which both
/// use the exact same `"anthropic/…"` shape as the one and only Anthropic
/// prefix — or, for a caller pointed directly at Anthropic's own API via
/// `--base-url`, a bare `claude-…` slug with no vendor prefix at all (that
/// endpoint doesn't use OpenRouter's vendor-prefixed naming). Both forms are
/// unambiguous: no other vendor slug in this codebase starts with `claude`.
///
/// This is intentionally narrower than "could plausibly be Anthropic" — an
/// unrecognized custom slug is NOT assumed Anthropic (mirrors
/// [`model_context_limit`]'s "unknown is never assumed favorable" stance) —
/// so this only ever narrows the warning, never broadens it past what T1's
/// accuracy bar already allows.
#[doc(hidden)]
pub fn is_anthropic_family_model(model: &str) -> bool {
    model.starts_with("anthropic/") || model.starts_with("claude-") || model.starts_with("claude/")
}

/// UX-26 (B7-warn): why a completed, reuse-expected turn likely paid a
/// full-price prompt-cache miss. See [`cache_cold_reason`].
#[derive(Debug, Clone, Copy, PartialEq)]
#[doc(hidden)]
pub enum CacheColdReason {
    /// This turn was sent `idle_secs` after the cache entry was last
    /// established/refreshed — at or beyond [`CACHE_TTL_SECS`], so the
    /// provider has almost certainly already evicted it. Computable
    /// pre-send (doesn't need `usage`).
    Stale {
        /// Seconds since the cache entry was last known warm.
        idle_secs: i64,
    },
    /// The provider's own usage reported `cached_tokens` out of
    /// `prompt_tokens` — below [`CACHE_MISS_RATIO_THRESHOLD`] despite reuse
    /// being expected, and NOT already explained by [`Self::Stale`] (this
    /// turn was sent inside the TTL window).
    Miss {
        /// Tokens the provider reports as served from cache.
        cached_tokens: u64,
        /// Total prompt (input) tokens for this turn.
        prompt_tokens: u64,
    },
}

impl CacheColdReason {
    /// Render as the ready-to-print stderr line (no trailing newline).
    #[doc(hidden)]
    pub fn message(&self) -> String {
        match self {
            CacheColdReason::Stale { idle_secs } => format!(
                "cache likely cold — this turn was sent {}m{:02}s after the cache was last \
                 refreshed (Anthropic's ephemeral prompt cache expires after 5m idle) — this \
                 turn likely paid full input cost for the cached prefix",
                idle_secs / 60,
                idle_secs % 60,
            ),
            CacheColdReason::Miss {
                cached_tokens,
                prompt_tokens,
            } => format!(
                "unexpected cache miss — only {cached_tokens}/{prompt_tokens} prompt tokens \
                 were served from cache this turn even though reuse was expected — this turn \
                 likely paid full input cost for the cached prefix",
            ),
        }
    }
}

/// UX-26 (B7-warn, dev/01+dev/02): whether a completed turn likely paid a
/// full-price cache miss.
///
/// Takes two INDEPENDENT preconditions rather than one combined
/// "reuse expected" flag, because they cover genuinely different turns:
///
/// - `will_annotate`: THIS request actually carries a
///   [`CachePlan::ImportedPrefix`] `cache_control` breakpoint (not a
///   same-turn tool-schema-tier bust, not `CachePlan::Off`). Gates BOTH
///   checks below — with no annotation there was never anything to reuse,
///   by construction.
/// - `cache_established`: a PRIOR request already placed that same
///   breakpoint (in THIS process, or inferred from `idle_secs` having a
///   value at all — see below). Gates ONLY the [`CacheColdReason::Miss`]
///   check: on the very FIRST annotated request for a prefix, the provider
///   legitimately reports ~0 cached tokens (it's establishing the entry,
///   not reusing it) — reporting that as a "miss" would be a false
///   positive on every resume's opening turn.
///
/// [`CacheColdReason::Stale`] deliberately does NOT require
/// `cache_established`: `idle_secs` itself is derived (by the caller,
/// `Agent::build_request_messages`) from the RESUMED SESSION's own last
/// message timestamp when this agent has never sent a request yet — a
/// cross-process signal of how long the prefix has sat untouched by ANY
/// tool. That is precisely the flagship case (`docs/jcode-ux-parity.md`
/// §6c.1): a session idle for 20 minutes, resumed, and its very first turn
/// in supercode is a foregone cold read — which is exactly when the user
/// most needs the heads-up, not only on turn 2+. `idle_secs` is `None`
/// whenever no such signal exists (a session with no parseable timestamp),
/// so this never guesses.
///
/// Checks [`CacheColdReason::Stale`] before [`CacheColdReason::Miss`] (needs
/// the completed `usage`, so only consulted once elapsed time is inside the
/// TTL window) so a genuinely stale turn is never double-reported.
///
/// UX-26 T1 (accuracy fold-in): `Stale` is nominally computable pre-send
/// (from `idle_secs` alone), but `usage` — for the very turn about to be
/// reported `Stale` — is always in hand by the time this fn actually runs
/// (the caller only has a completed `usage` to give it). When that usage
/// affirmatively PROVES the turn was warm (cache-read ratio at/above
/// [`CACHE_STALE_DISPROVE_RATIO_THRESHOLD`] — see its doc comment for why
/// that bar, not [`CACHE_MISS_RATIO_THRESHOLD`], is used here), the
/// idle-clock-based `Stale` verdict is disproven and suppressed: a stale
/// *clock* reading doesn't mean a stale *cache* when the provider's own
/// billed usage says otherwise. Usage that's absent, unparseable, or merely
/// ambiguous (below the disprove bar but not a `Miss` either) offers no such
/// disproof, so `Stale` still fires exactly as before.
#[doc(hidden)]
pub fn cache_cold_reason(
    will_annotate: bool,
    cache_established: bool,
    idle_secs: Option<i64>,
    usage: &Usage,
) -> Option<CacheColdReason> {
    if !will_annotate {
        return None;
    }
    if let Some(idle_secs) = idle_secs {
        if idle_secs >= CACHE_TTL_SECS {
            let disproven_by_usage = usage
                .prompt_tokens_details
                .filter(|_| usage.prompt_tokens > 0)
                .is_some_and(|details| {
                    details.cached_tokens as f64 / usage.prompt_tokens as f64
                        >= CACHE_STALE_DISPROVE_RATIO_THRESHOLD
                });
            if !disproven_by_usage {
                return Some(CacheColdReason::Stale { idle_secs });
            }
        }
    }
    if !cache_established {
        // First annotated request for this prefix: a legitimate cold WRITE,
        // never a "miss" — nothing to compare `usage` against.
        return None;
    }
    let details = usage.prompt_tokens_details?;
    if usage.prompt_tokens == 0 {
        // Nothing was actually read as prompt input this turn (unusual, but
        // possible for a degenerate request) — no signal either way.
        return None;
    }
    let ratio = details.cached_tokens as f64 / usage.prompt_tokens as f64;
    if ratio < CACHE_MISS_RATIO_THRESHOLD {
        return Some(CacheColdReason::Miss {
            cached_tokens: details.cached_tokens,
            prompt_tokens: usage.prompt_tokens,
        });
    }
    None
}

/// The transport abstraction. Implement this to back the agent with something
/// other than an OpenAI-compatible HTTP endpoint (a local model, a mock, etc.).
#[async_trait]
pub trait Provider: Send + Sync {
    /// Run one completion. `on_delta` is called with each text chunk as it
    /// streams in. Returns the fully assembled assistant message and usage.
    async fn complete(
        &self,
        req: &ChatRequest,
        on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> Result<(ChatMessage, Usage)>;
}

/// An OpenAI-compatible HTTP provider. The composition layer supplies its
/// endpoint, credentials, and headers from runtime configuration.
pub struct OpenAiProvider {
    client: reqwest::Client,
    base_url: String,
    api_key: String,
    extra_headers: HashMap<String, String>,
    http_options: HttpOptions,
    /// BP-7: where [`Self::send_with_retry`] reports the retries it makes.
    /// `None` (the default for every constructor caller that does not opt
    /// in) keeps the loop byte-identical to its pre-BP-7 behavior.
    retry_log: Option<std::sync::Arc<RetryLog>>,
}

impl OpenAiProvider {
    /// Construct a provider for the given endpoint and key.
    pub fn new(
        base_url: impl Into<String>,
        api_key: impl Into<String>,
        extra_headers: HashMap<String, String>,
    ) -> Self {
        Self::new_with_options(base_url, api_key, extra_headers, HttpOptions::default())
    }

    /// Same as [`Self::new`] but with crate-internal HTTP timeout/retry
    /// options — used by tests to shrink timeouts and backoff so they run
    /// fast. Not part of the public API (no `Config`/CLI surface for these
    /// knobs).
    #[doc(hidden)]
    pub fn new_with_options(
        base_url: impl Into<String>,
        api_key: impl Into<String>,
        extra_headers: HashMap<String, String>,
        http_options: HttpOptions,
    ) -> Self {
        OpenAiProvider {
            client: reqwest::Client::builder()
                .connect_timeout(http_options.connect_timeout)
                .read_timeout(http_options.read_idle_timeout)
                .build()
                .expect("static reqwest client config cannot fail"),
            base_url: base_url.into(),
            api_key: api_key.into(),
            extra_headers,
            http_options,
            retry_log: None,
        }
    }

    /// BP-7: report every retry this provider makes into `log`, which the
    /// caller drains after each completion (see [`RetryLog`]).
    pub fn with_retry_log(mut self, log: std::sync::Arc<RetryLog>) -> Self {
        self.retry_log = Some(log);
        self
    }

    fn endpoint(&self) -> String {
        format!("{}/chat/completions", self.base_url.trim_end_matches('/'))
    }

    /// Send the initial request, retrying connection-level failures and 5xx
    /// responses with backoff. 4xx (and any other non-success, non-5xx)
    /// statuses return immediately, unretried. Once a 2xx response is
    /// received it is returned as-is for the caller to stream; this loop
    /// never runs again for the lifetime of that response (no mid-stream
    /// retry/resume).
    async fn send_with_retry(&self, wire: &serde_json::Value) -> Result<reqwest::Response> {
        let mut attempt = 0u32;
        loop {
            let mut builder = self
                .client
                .post(self.endpoint())
                .bearer_auth(&self.api_key)
                .header("Content-Type", "application/json");
            for (k, v) in &self.extra_headers {
                builder = builder.header(k, v);
            }

            let sent = builder.json(wire).send().await;
            let (retryable, result): (bool, Result<reqwest::Response>) = match sent {
                Err(e) => (true, Err(Error::from(e))),
                Ok(resp) => {
                    let status = resp.status();
                    if status.is_success() {
                        (false, Ok(resp))
                    } else if status.is_server_error() {
                        let body = resp.text().await.unwrap_or_default();
                        (
                            true,
                            Err(Error::Provider {
                                status: status.as_u16(),
                                body: truncate(&body, 2000),
                            }),
                        )
                    } else {
                        let body = resp.text().await.unwrap_or_default();
                        (
                            false,
                            Err(Error::Provider {
                                status: status.as_u16(),
                                body: truncate(&body, 2000),
                            }),
                        )
                    }
                }
            };

            if !retryable || attempt >= self.http_options.max_retries {
                return result;
            }
            let backoff = self.http_options.retry_backoff_base * 2u32.pow(attempt);
            // BP-7: recorded at the decision point — after "this is
            // retryable and we have attempts left", before the sleep — so
            // the notice exists even if the process dies during the
            // backoff.
            if let Some(log) = &self.retry_log {
                log.record(RetryNotice {
                    attempt,
                    delay_ms: backoff.as_millis() as u64,
                    reason: match &result {
                        Err(e) => e.to_string(),
                        Ok(_) => String::new(),
                    },
                });
            }
            tokio::time::sleep(backoff).await;
            attempt += 1;
        }
    }
}

#[async_trait]
impl Provider for OpenAiProvider {
    async fn complete(
        &self,
        req: &ChatRequest,
        on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> Result<(ChatMessage, Usage)> {
        let wire = build_request_body(req, true);

        let resp = self.send_with_retry(&wire).await?;

        let mut acc = Accumulator::default();
        // Buffer raw bytes, not a lossy-decoded String: network chunks split at
        // arbitrary byte offsets, so decoding each chunk independently would turn
        // any multi-byte UTF-8 scalar straddling a boundary into replacement
        // characters. We only decode *complete* SSE lines (terminated by '\n',
        // an ASCII byte that can never fall inside a multi-byte sequence).
        let mut buf: Vec<u8> = Vec::new();
        let mut deltas: Vec<String> = Vec::new();
        let mut stream = resp.bytes_stream();
        while let Some(chunk) = stream.next().await {
            let bytes = chunk?;
            buf.extend_from_slice(&bytes);
            drain_sse_lines(&mut buf, &mut acc, &mut deltas)?;
            for d in deltas.drain(..) {
                on_delta(&d);
            }
        }
        // Flush any trailing buffered line (no terminating newline).
        let tail = String::from_utf8_lossy(&buf);
        if !tail.trim().is_empty() {
            handle_sse_line(tail.trim(), &mut acc, &mut deltas)?;
            for d in deltas.drain(..) {
                on_delta(&d);
            }
        }

        Ok((acc.to_message(), acc_usage(&acc)))
    }
}

// ---- streaming assembly ---------------------------------------------------

#[derive(Default)]
struct Accumulator {
    content: String,
    tool_calls: Vec<ToolCallAccum>,
    usage: Usage,
    /// BP-13 (D9 "Model-served-vs-requested provenance"): the `model` field
    /// the PROVIDER put on its own response frames — the model that actually
    /// answered, which a gateway is free to make differ from the one asked
    /// for (aliasing, routing, a silently pinned snapshot).
    served_model: Option<String>,
}

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

impl Accumulator {
    fn ensure(&mut self, index: usize) -> &mut ToolCallAccum {
        while self.tool_calls.len() <= index {
            self.tool_calls.push(ToolCallAccum::default());
        }
        &mut self.tool_calls[index]
    }

    fn to_message(&self) -> ChatMessage {
        let calls: Vec<ToolCall> = self
            .tool_calls
            .iter()
            .filter(|c| !c.id.is_empty() || !c.name.is_empty())
            .map(|c| ToolCall {
                id: c.id.clone(),
                kind: "function".to_string(),
                function: FunctionCall {
                    name: c.name.clone(),
                    arguments: c.arguments.clone(),
                },
            })
            .collect();
        ChatMessage {
            role: Role::Assistant,
            content: (!self.content.is_empty()).then(|| self.content.clone()),
            content_parts: None,
            tool_calls: (!calls.is_empty()).then_some(calls),
            tool_call_id: None,
            name: None,
            metadata: match &self.served_model {
                Some(model) => {
                    let mut m = std::collections::BTreeMap::new();
                    m.insert(SERVED_MODEL_KEY.to_string(), model.clone());
                    m
                }
                None => Default::default(),
            },
        }
    }
}

/// Metadata key carrying the model the PROVIDER said served a response —
/// distinct from `"model"`, which every caller sets to the model it
/// REQUESTED. Present only when the response actually reported one.
pub const SERVED_MODEL_KEY: &str = "served_model";

fn acc_usage(acc: &Accumulator) -> Usage {
    acc.usage.clone()
}

fn drain_sse_lines(
    buf: &mut Vec<u8>,
    acc: &mut Accumulator,
    deltas: &mut Vec<String>,
) -> Result<()> {
    while let Some(pos) = buf.iter().position(|&b| b == b'\n') {
        let line: Vec<u8> = buf.drain(..=pos).collect();
        let line = String::from_utf8_lossy(&line);
        handle_sse_line(line.trim(), acc, deltas)?;
    }
    Ok(())
}

fn handle_sse_line(line: &str, acc: &mut Accumulator, deltas: &mut Vec<String>) -> Result<()> {
    let Some(data) = line.strip_prefix("data:") else {
        return Ok(());
    };
    let data = data.trim();
    if data.is_empty() || data == "[DONE]" {
        return Ok(());
    }
    let chunk: StreamChunk = match serde_json::from_str(data) {
        Ok(c) => c,
        Err(_) => return Ok(()), // tolerate keep-alive / partial frames
    };
    if let Some(u) = chunk.usage {
        acc.usage = u;
    }
    if let Some(model) = chunk.model {
        if !model.is_empty() {
            acc.served_model = Some(model);
        }
    }
    for choice in chunk.choices {
        if let Some(text) = choice.delta.content {
            if !text.is_empty() {
                acc.content.push_str(&text);
                deltas.push(text);
            }
        }
        for tc in choice.delta.tool_calls.unwrap_or_default() {
            let slot = acc.ensure(tc.index);
            if let Some(id) = tc.id {
                slot.id = id;
            }
            if let Some(f) = tc.function {
                if let Some(name) = f.name {
                    slot.name.push_str(&name);
                }
                if let Some(args) = f.arguments {
                    slot.arguments.push_str(&args);
                }
            }
        }
    }
    Ok(())
}

fn truncate(s: &str, max: usize) -> String {
    if s.len() <= max {
        s.to_string()
    } else {
        // Walk back to a char boundary so we never slice mid-codepoint (which
        // would panic) — provider error bodies can contain non-ASCII text.
        let mut end = max;
        while end > 0 && !s.is_char_boundary(end) {
            end -= 1;
        }
        format!("{}…", &s[..end])
    }
}

// ---- wire types -----------------------------------------------------------

#[derive(Serialize)]
struct WireTool<'a> {
    #[serde(rename = "type")]
    kind: &'static str,
    function: WireFunction<'a>,
}

#[derive(Serialize)]
struct WireFunction<'a> {
    name: &'a str,
    description: &'a str,
    parameters: &'a serde_json::Value,
}

impl<'a> From<&'a ToolSchema> for WireTool<'a> {
    fn from(t: &'a ToolSchema) -> Self {
        WireTool {
            kind: "function",
            function: WireFunction {
                name: &t.name,
                description: &t.description,
                parameters: &t.parameters,
            },
        }
    }
}

#[derive(Deserialize)]
struct StreamChunk {
    #[serde(default)]
    choices: Vec<StreamChoice>,
    #[serde(default)]
    usage: Option<Usage>,
    /// The model the provider says produced this frame (BP-13 D9
    /// served-vs-requested provenance).
    #[serde(default)]
    model: Option<String>,
}

#[derive(Deserialize)]
struct StreamChoice {
    delta: Delta,
}

#[derive(Deserialize)]
struct Delta {
    #[serde(default)]
    content: Option<String>,
    #[serde(default)]
    tool_calls: Option<Vec<ToolCallDelta>>,
}

#[derive(Deserialize)]
struct ToolCallDelta {
    #[serde(default)]
    index: usize,
    #[serde(default)]
    id: Option<String>,
    #[serde(default)]
    function: Option<FnDelta>,
}

#[derive(Deserialize)]
struct FnDelta {
    #[serde(default)]
    name: Option<String>,
    #[serde(default)]
    arguments: Option<String>,
}