car-inference 0.47.0

//! Remote inference backend — HTTP client for cloud API models.
//!
//! The main dispatch uses the `ProtocolHandler` trait from `protocol.rs`.
//! Legacy per-protocol handlers are kept temporarily for the embedding path.
//!
//! Supports: OpenAI-compatible, Anthropic, Google (Gemini).

// Legacy handlers are being phased out in favor of ProtocolHandler trait.

use reqwest::Client;
use serde::Deserialize;
use tracing::{debug, instrument};

use crate::key_pool::{KeyLease, KeyPool};
use crate::protocol::ProtocolHandler;
use crate::schema::{ApiProtocol, ModelSchema, ModelSource, ProprietaryAuth};
use crate::tasks::ContentBlock;
use crate::tls_client::Degradation;
use crate::InferenceError;
use std::collections::HashMap;
use std::sync::{Mutex, OnceLock};

/// Environment override recognized by car-auth for the Parslee OAuth2 bearer.
/// Durable credentials live in car-auth's authoritative V2 transaction, so
/// synchronous availability checks must call
/// `car_auth::access_token_is_available()` rather than resolving this name as a
/// standalone keychain slot.
pub(crate) const PARSLEE_ACCESS_TOKEN_ENV: &str = "PARSLEE_ACCESS_TOKEN";

/// Bounded transient-retry budget for a single remote HTTP call inside
/// [`RemoteBackend::execute_request`]: initial attempt + 2 retries. Mirrors the
/// values the car-cli run_task loop used before retries moved into the engine.
const REMOTE_MAX_ATTEMPTS: u32 = 3;
/// Linear backoff base (seconds): waits `base * attempt` → 3s, then 6s.
const REMOTE_BACKOFF_BASE_SECS: u64 = 3;

/// Whether an HTTP status code is worth retrying. Transient server-side /
/// rate-limit failures only — 4xx validation/auth (400/401/403/404/422) are
/// NOT retried because re-sending the same request just fails again.
fn is_transient_http_status(code: u16) -> bool {
    matches!(code, 429 | 500 | 502 | 503 | 529)
}

/// Maximum rendered length of an error chain. These strings are log lines and
/// user-facing messages, not payload dumps, and the deepest link can be
/// arbitrary upstream text.
const MAX_ERROR_CHAIN_LEN: usize = 512;

/// Maximum number of `source()` links to follow, so a pathological or
/// self-referential chain can't turn one failed request into a hang.
const MAX_ERROR_CHAIN_DEPTH: usize = 8;

/// Query-parameter names whose values are credentials.
///
/// Google AI Studio authenticates with `?key=<api key>` in the URL
/// (`protocol.rs`), and `reqwest`'s `Display` appends `for url (...)` to
/// transport errors. Without this scrub, surfacing the error chain would write
/// the very API key that broke the request into the log reporting the breakage.
/// Header-valued credentials are safe by construction: `InvalidHeaderValue`
/// renders as `failed to parse header value` and never echoes the value.
const SENSITIVE_QUERY_PARAMS: &[&str] = &[
    "key",
    "api_key",
    "apikey",
    "access_token",
    "token",
    "password",
    "secret",
    "signature",
    "sig",
];

/// Replace URL userinfo and credential-bearing query-parameter values with
/// `REDACTED`.
fn redact_sensitive(text: &str) -> String {
    let lower = text.to_ascii_lowercase();
    let bytes = text.as_bytes();
    let mut out = String::with_capacity(text.len());
    let mut i = 0usize;
    while i < text.len() {
        // A URL authority begins immediately after `://`. If it contains
        // userinfo, redact everything through the final `@` delimiter while
        // preserving the host and the rest of the URL.
        let userinfo_end = if i >= 3 && &bytes[i - 3..i] == b"://" {
            let authority_end = text[i..]
                .find(|c: char| c.is_whitespace() || matches!(c, '/' | '?' | '#' | ')'))
                .map_or(text.len(), |offset| i + offset);
            text[i..authority_end]
                .rfind('@')
                .map(|offset| i + offset + 1)
        } else {
            None
        };
        if let Some(end) = userinfo_end {
            out.push_str("REDACTED@");
            i = end;
            continue;
        }

        // A parameter name starts the string or follows a `?` / `&` delimiter.
        let value_start = if i == 0 || matches!(bytes[i - 1], b'?' | b'&') {
            SENSITIVE_QUERY_PARAMS
                .iter()
                .find(|name| {
                    let end = i + name.len();
                    end < text.len() && lower[i..].starts_with(**name) && bytes[end] == b'='
                })
                .map(|name| i + name.len() + 1)
        } else {
            None
        };
        match value_start {
            Some(start) => {
                out.push_str(&text[i..start]);
                out.push_str("REDACTED");
                i = text[start..]
                    .find(|c: char| c == '&' || c == ')' || c.is_whitespace())
                    .map_or(text.len(), |off| start + off);
            }
            None => {
                // Step a whole char so multi-byte UTF-8 is never split.
                let step = text[i..].chars().next().map_or(1, char::len_utf8);
                out.push_str(&text[i..i + step]);
                i += step;
            }
        }
    }
    out
}

/// Render an error together with its full `source()` chain.
///
/// `format!("{e}")` prints only the top-level `Display`. For a builder-kind
/// `reqwest::Error` that is the bare string `builder error` — the reason
/// (`failed to parse header value`, a URL parse failure) lives one link down
/// and was discarded at every call site in this file. That is how a trailing
/// newline on an API key presented as an unexplained transport failure for
/// weeks: the diagnosis was in the error the whole time.
fn error_chain(err: &dyn std::error::Error) -> String {
    let mut rendered = err.to_string();
    let mut source = err.source();
    let mut depth = 0usize;
    while let Some(cause) = source {
        if depth >= MAX_ERROR_CHAIN_DEPTH {
            rendered.push_str(": ...");
            break;
        }
        let text = cause.to_string();
        // Skip a link that only restates its parent.
        if !rendered.ends_with(&text) {
            rendered.push_str(": ");
            rendered.push_str(&text);
        }
        source = cause.source();
        depth += 1;
    }
    let redacted = redact_sensitive(&rendered);
    if redacted.len() <= MAX_ERROR_CHAIN_LEN {
        return redacted;
    }
    let mut end = MAX_ERROR_CHAIN_LEN;
    while end > 0 && !redacted.is_char_boundary(end) {
        end -= 1;
    }
    format!("{}... (truncated)", &redacted[..end])
}

/// Whether a transport error is worth retrying, decided on the typed
/// `reqwest::Error` rather than on its rendered text.
///
/// `is_builder()` is the load-bearing case. A builder-kind error means the
/// request was never constructed — an unparseable header value, a bad URL — so
/// the bytes never left the process and every retry reproduces it byte for
/// byte. The previous version of this check matched the substring `"http
/// error"` against the already-stringified message, so `"HTTP error: builder
/// error"` was classified transient: a permanently malformed request burned the
/// full retry budget and was then relabelled [`InferenceError::Transient`],
/// which downstream reads as "infra blip, safe to re-run". A deterministic
/// config bug therefore presented as flakiness.
fn reqwest_error_is_transient(e: &reqwest::Error) -> bool {
    if e.is_builder() {
        return false;
    }
    // A redirect loop is a property of the destination, not the network, and a
    // status-carrying error is classified by `is_transient_http_status` on the
    // code itself.
    if e.is_redirect() || e.is_status() {
        return false;
    }
    // What remains touched the network — connect, timeout, a request that
    // failed in flight, or a body that stopped arriving. All are worth the
    // bounded retry budget.
    true
}

/// A failed transport attempt carrying its retry verdict alongside the error.
///
/// The verdict has to travel with the error because it can only be computed
/// while the `reqwest::Error` is still typed. Once it has been rendered to a
/// string the only thing left to classify on is prose, and prose lies:
/// `"HTTP error: builder error"` reads as a network failure and is the exact
/// opposite of one.
struct TransportAttemptError {
    error: InferenceError,
    retryable: bool,
}

impl TransportAttemptError {
    /// A `reqwest` failure: full source chain in the message, retry verdict
    /// from the typed error kind.
    #[cfg(test)]
    fn from_reqwest(context: &str, e: &reqwest::Error) -> Self {
        Self::from_reqwest_error(
            InferenceError::InferenceFailed(format!("{context}: {}", error_chain(e))),
            e,
        )
    }

    /// Keep a caller-enriched error message paired with the retry verdict
    /// derived from the still-typed reqwest error.
    fn from_reqwest_error(error: InferenceError, e: &reqwest::Error) -> Self {
        Self {
            error,
            retryable: reqwest_error_is_transient(e),
        }
    }

    /// A `tokio::time::timeout` deadline elapsing. Always retryable — the
    /// deadline says nothing about whether the request itself is well-formed.
    fn timeout(message: &str) -> Self {
        Self {
            error: InferenceError::InferenceFailed(message.to_string()),
            retryable: true,
        }
    }
}

fn openrouter_model_is_unavailable(status: reqwest::StatusCode, body: &str) -> bool {
    if status == reqwest::StatusCode::NOT_FOUND {
        return true;
    }
    if status != reqwest::StatusCode::BAD_REQUEST {
        return false;
    }
    let message = serde_json::from_str::<serde_json::Value>(body)
        .ok()
        .and_then(|value| {
            value
                .pointer("/error/message")
                .and_then(serde_json::Value::as_str)
                .map(str::to_lowercase)
        })
        .unwrap_or_default();
    [
        "no endpoints found",
        "model not found",
        "invalid model",
        "does not exist",
        "deprecated",
    ]
    .iter()
    .any(|needle| message.contains(needle))
}

/// HTTP statuses that indict the **account**, not the model: 401 (key absent or
/// rejected), 402 (out of credits / quota), 403 (forbidden).
///
/// Every model on that account fails these identically, so they must not reach
/// the per-model health EMA or circuit breaker — see
/// [`InferenceError::ProviderAccount`] (Parslee-ai/car#650).
///
/// Written as a range because the three codes happen to be contiguous and
/// clippy's `manual_range_patterns` rejects the enumerated form; the doc above
/// is what carries the meaning of each.
pub(crate) fn is_account_http_status(status: u16) -> bool {
    matches!(status, 401..=403)
}

/// How little token life left is worth interrupting someone over.
///
/// Fifteen minutes: long enough that a user can act (re-authenticate, or hold
/// off starting a sweep), short enough that it is not firing during ordinary
/// work. The reported case had a token four minutes from expiry and a
/// multi-hour job — a warning at that moment is the difference between
/// re-authenticating first and losing hours of work (Parslee-ai/car#797).
const EXPIRY_WARNING_HORIZON_SECS: u64 = 15 * 60;

/// How often the warning path is allowed to consult the credential store.
///
/// The probe reads the store, which on macOS is a keychain query — the exact
/// per-request cost that car-releases#75 and the token cache were written to
/// remove. Doing it per request would trade a mid-run failure for a permanent
/// latency regression, which is a bad deal. Throttling to one probe per five
/// minutes makes the cost unmeasurable while still catching a token that ages
/// into the horizon during a long run.
const EXPIRY_PROBE_INTERVAL_SECS: u64 = 5 * 60;

/// Warn when the Parslee token will expire during a plausibly long operation,
/// rather than letting the run discover it hours in (#797 item 4).
///
/// Warns at most once per distinct expiry, so a re-authentication mid-session
/// re-arms it but a long batch does not repeat itself every request. Silent
/// when there is nothing to say: no session, no stored expiry, a
/// `PARSLEE_ACCESS_TOKEN` override, or plenty of life left.
///
/// This is advisory only and deliberately does not fail the request. A token
/// inside the refresh skew is normally refreshed transparently on use, so a
/// small number here usually resolves itself — the warning exists because when
/// refresh *cannot* succeed, the alternative is finding out mid-sweep.
async fn warn_if_token_expires_soon() {
    static LAST_PROBE: OnceLock<Mutex<Option<std::time::Instant>>> = OnceLock::new();
    static WARNED_FOR: OnceLock<Mutex<Option<u64>>> = OnceLock::new();

    {
        let probe_gate = LAST_PROBE.get_or_init(|| Mutex::new(None));
        let mut last = match probe_gate.lock() {
            Ok(guard) => guard,
            Err(poisoned) => poisoned.into_inner(),
        };
        if last.is_some_and(|at| {
            at.elapsed() < std::time::Duration::from_secs(EXPIRY_PROBE_INTERVAL_SECS)
        }) {
            return;
        }
        *last = Some(std::time::Instant::now());
    }

    let Some(remaining) = car_auth::access_token_lifetime_remaining().await else {
        return;
    };
    if remaining > EXPIRY_WARNING_HORIZON_SECS {
        return;
    }

    let expires_at = crate::remote::epoch_seconds().saturating_add(remaining);
    {
        let warned_gate = WARNED_FOR.get_or_init(|| Mutex::new(None));
        let mut warned = match warned_gate.lock() {
            Ok(guard) => guard,
            Err(poisoned) => poisoned.into_inner(),
        };
        // Same expiry (within probe jitter) means same token — already said.
        if warned.is_some_and(|seen| seen.abs_diff(expires_at) <= EXPIRY_PROBE_INTERVAL_SECS) {
            return;
        }
        *warned = Some(expires_at);
    }

    tracing::warn!(
        target: "car_inference::auth",
        remaining_secs = remaining,
        expires_at_unix = expires_at,
        "Parslee token expires in {}m — an operation longer than that will fail \
         partway unless it can resume. Re-authenticate with `car auth login` first \
         if you are starting a long run.",
        remaining / 60
    );
}

fn epoch_seconds() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}

/// Classify a gateway error as a **content refusal**, from the `type`/`code`
/// tags #804 forwards.
///
/// Matched on the classification fields, never the prose. Deliberately NARROW:
/// over-classifying is the more dangerous direction, because a real inference
/// failure mislabelled as a refusal is excluded from the model's health record
/// and silently stops being retried — a crash that looks like a policy decision
/// is harder to find than a policy decision that looks like a crash.
///
/// The matched set is what a filter calls itself across the providers CAR has
/// seen: `content_policy_violation` (OpenAI-family, and the value in this
/// module's own fixtures), plus the `content_filter` / `moderation` / `safety`
/// families. #796's own gateway values are NOT yet known — the issue asks for a
/// re-run to discover them — so this list is expected to grow, and the test
/// pins the over-classification boundary rather than the exact membership.
fn content_refusal_tags(detail: &str) -> Option<(Option<String>, Option<String>)> {
    let (kind, code) = crate::stream::error_tags(detail);
    let refused = |v: &str| {
        let v = v.to_ascii_lowercase();
        v.contains("content_policy")
            || v.contains("content_filter")
            || v.contains("moderation")
            || v.contains("safety")
    };
    (code.is_some_and(refused) || kind.is_some_and(refused))
        .then(|| (kind.map(str::to_string), code.map(str::to_string)))
}

/// Classify a non-success Parslee gateway body as an *environment* condition:
/// the deployment has no upstream configured for a whole namespace of models.
///
/// The gateway says so in a typed error object — the same `type`/`code` fields
/// #796 made legible on the SSE path:
///
/// ```json
/// {"error":{"code":"openrouter_not_configured","type":"gateway_error",
///           "message":"OpenRouter inference is not configured on this Parslee environment."}}
/// ```
///
/// Matched on `code`, not on the prose. The message is operator-facing text
/// that can be reworded at any deploy; the code is the contract. `*_not_configured`
/// is matched as a suffix so a future `bedrock_not_configured` classifies
/// correctly on arrival rather than silently regressing to a generic failure —
/// this whole class is "the gateway has nothing to proxy to", and the namespace
/// is the only part that differs.
///
/// Returns the message for the caller to surface. `None` means "not this
/// condition" and the body flows on as an ordinary failure.
fn gateway_unconfigured_detail(status: reqwest::StatusCode, body: &str) -> Option<String> {
    // 503 is what the live gateway returns, but do not require it: the
    // condition is defined by the typed code, and pinning the status would
    // reclassify the identical error as a generic failure if the gateway ever
    // moved it to 501/502.
    if !status.is_server_error() && status != reqwest::StatusCode::NOT_IMPLEMENTED {
        return None;
    }
    let json: serde_json::Value = serde_json::from_str(body).ok()?;
    let error = json.get("error")?;
    let code = error.get("code").and_then(|c| c.as_str())?;
    if !code.ends_with("_not_configured") {
        return None;
    }
    Some(
        error
            .get("message")
            .and_then(|m| m.as_str())
            .map(str::trim)
            .filter(|m| !m.is_empty())
            .unwrap_or(code)
            .to_string(),
    )
}

/// Turn a classified gateway condition into the error the dispatch loop acts
/// on, and record the observation so the registry stops advertising the
/// namespace (Parslee-ai/car#786).
///
/// Recording here rather than at the call sites keeps the two in lockstep: a
/// future caller that classifies but forgets to record would quietly restore
/// the "catalog lies" behavior this exists to fix.
fn gateway_unconfigured_error(
    status: reqwest::StatusCode,
    message: String,
    model: &str,
) -> InferenceError {
    let namespace = if crate::openrouter::is_curated_managed_gateway_alias(model) {
        crate::openrouter::note_gateway_unconfigured();
        "parslee/openrouter/"
    } else {
        // Same condition on some other managed namespace. Classify it — the
        // health and fallback consequences are identical — but do not touch the
        // OpenRouter suppression flag, which is specifically about those rows.
        "parslee/"
    };
    InferenceError::GatewayUnconfigured {
        provider: "parslee".to_string(),
        namespace: namespace.to_string(),
        status: status.as_u16(),
        message,
    }
}

/// Remediation text for an account rejection. Provider-specific where we can
/// actually name the fix, generic otherwise — an account error is only useful
/// if it tells the user which account and what to do about it.
fn account_error_message(protocol: ApiProtocol, status: u16) -> String {
    match (protocol, status) {
        (ApiProtocol::OpenRouter, 402) => {
            "OpenRouter account is out of credits — add credits in OpenRouter and retry".to_string()
        }
        (ApiProtocol::OpenRouter, _) => {
            "OpenRouter key rejected — re-add it with `car keys set openrouter`".to_string()
        }
        (_, 402) => "provider account is out of credits or over quota".to_string(),
        _ => "provider rejected the API key — check the configured credential".to_string(),
    }
}

/// Non-account OpenRouter failures. Account statuses (401/402/403) are handled
/// upstream as [`InferenceError::ProviderAccount`] and never reach here.
fn openrouter_http_error(status: reqwest::StatusCode, body: &str, model: &str) -> InferenceError {
    if openrouter_model_is_unavailable(status, body) {
        return InferenceError::InferenceFailed(format!(
            "OpenRouter model '{model}' is no longer available on OpenRouter"
        ));
    }
    InferenceError::InferenceFailed(format!("OpenRouter request failed with HTTP {status}"))
}

/// Interpret Gemini's typed `finishReason` without reflecting arbitrary
/// provider text across CAR's user-facing surfaces.
///
/// `STOP` is a deliberate completion and `MAX_TOKENS` is a deliberate,
/// observable truncation. Every other reason is abnormal and must terminate
/// with a sanitized Error rather than being upgraded to Done at transport EOF.
fn google_finish_outcome(reason: &str) -> Result<(), &'static str> {
    match reason {
        "STOP" | "MAX_TOKENS" => Ok(()),
        "SAFETY" => Err("Google blocked the response for safety"),
        "RECITATION" => Err("Google blocked the response for recitation"),
        "BLOCKLIST" => Err("Google blocked the response because of a blocklist match"),
        "PROHIBITED_CONTENT" => Err("Google blocked prohibited response content"),
        "SPII" => Err("Google blocked the response because it may contain sensitive information"),
        "MALFORMED_FUNCTION_CALL" => Err("Google returned a malformed function call"),
        _ => Err("Google inference ended with an abnormal finish reason"),
    }
}

/// Whether a Parslee-path error is an authentication rejection (401/403) —
/// i.e. the bearer is bad/expired. These reach us as
/// [`InferenceError::InferenceFailed`] strings of the form
/// `"...: HTTP 401: ..."` (org lookup, `/connect/session`, or the chat
/// endpoint). A match drives the reactive `force_refresh`-and-retry-once
/// path. Anchored on `"HTTP 40x"` so a 4xx body that merely mentions the
/// number doesn't false-positive.
fn is_auth_rejection(e: &InferenceError) -> bool {
    let m = e.to_string();
    m.contains("HTTP 401") || m.contains("HTTP 403")
}

/// Find the first SSE event-boundary in a byte buffer, returning
/// `(offset, separator_len)`. Servers delimit events with a blank line,
/// which may arrive as `"\n\n"` (2 bytes) or — per the SSE spec — as
/// `"\r\n\r\n"` (4 bytes, CRLF). Splitting on `"\n\n"` alone silently
/// never frames a CRLF stream. Operating on bytes (not a lossily-decoded
/// `String`) is what lets the caller keep a partial multi-byte codepoint
/// buffered across chunk boundaries instead of corrupting it into `�`.
fn find_sse_separator(buf: &[u8]) -> Option<(usize, usize)> {
    let lf = buf.windows(2).position(|w| w == b"\n\n").map(|p| (p, 2));
    let crlf = buf
        .windows(4)
        .position(|w| w == b"\r\n\r\n")
        .map(|p| (p, 4));
    match (lf, crlf) {
        (Some(a), Some(b)) => Some(if a.0 <= b.0 { a } else { b }),
        (Some(a), None) => Some(a),
        (None, Some(b)) => Some(b),
        (None, None) => None,
    }
}

/// Process-wide cache of `(org_id, user_id)` resolved from a Parslee
/// bearer, keyed by the token so a re-login re-resolves. The Parslee
/// managed-inference path (`ModelSource::Proprietary`) talks to the
/// shipped inference gateway endpoint
/// (`POST /api/v1/orgs/{org}/inference/responses`), which needs the
/// caller's org + a user identifier — neither is in the token in a form
/// we parse, so we resolve them once per token via `/organizations/me`
/// and `/connect/session` (mirrors how Parslee Hydra does it).
type ParsleeIdentityCache = HashMap<(String, String), (String, String)>;
static PARSLEE_IDENTITY: OnceLock<tokio::sync::Mutex<ParsleeIdentityCache>> = OnceLock::new();

/// Remote inference client. Reuses a single HTTP client for connection pooling.
/// Integrates with KeyPool for multi-key load balancing.
pub struct RemoteBackend {
    pub(crate) client: Client,
    /// `Some` when [`RemoteBackend::new`] could not load the OS trust store
    /// and fell back to a lesser one. Kept beside the client — rather than
    /// changing `client`'s type — so the two speech call sites in `lib.rs`
    /// that reach through `client` stay as they are, and so a later request
    /// failure can name the cause instead of leaving the operator with a bare
    /// "unknown issuer". `None` on an ordinary machine.
    pub(crate) tls_degradation: Option<Degradation>,
    pub key_pool: KeyPool,
}

/// Estimate token count from a string. Uses the standard ~4 chars/token heuristic
/// for remote models where we don't have a tokenizer.
pub fn estimate_tokens(text: &str) -> usize {
    std::cmp::max(1, text.len() / 4)
}

/// Collect the assistant text from a buffered Parslee chat SSE body.
///
/// Parslee frames events with blank-line separators. The on-wire
/// payload (per Parslee Hydra) is a leading `{ "conversationId": … }`
/// (start, ignored), zero or more `{ "content": … }` deltas
/// (concatenated), and a final event carrying a `timestamp` (stop).
/// An OpenAI-style `[DONE]` sentinel and unparseable frames are
/// skipped. Pure + side-effect-free so the parsing contract is unit-
/// testable without a live backend.
/// Parse a buffered Responses-API SSE body into stream events, reusing the
/// shared per-line parser. Each block is `event: <type>\ndata: <json>`, as
/// written by the Parslee inference endpoint (`SseHelper.WriteSseEventAsync`).
/// Multi-line `data:` payloads are joined with `\n` per the SSE spec.
fn parse_parslee_responses_sse(raw: &str) -> Vec<crate::stream::StreamEvent> {
    let mut events = Vec::new();
    for block in raw.split("\n\n") {
        let block = block.trim();
        if block.is_empty() {
            continue;
        }
        let mut event_type = "";
        let mut data = String::new();
        for line in block.lines() {
            if let Some(t) = line.strip_prefix("event:") {
                event_type = t.trim();
            } else if let Some(d) = line.strip_prefix("data:") {
                if !data.is_empty() {
                    data.push('\n');
                }
                data.push_str(d.trim());
            }
        }
        if event_type.is_empty() || data.is_empty() {
            continue;
        }
        events.extend(crate::stream::parse_openai_responses_sse_line(
            event_type, &data,
        ));
    }
    events
}

/// Build the Parslee inference request body (`POST …/inference/responses`).
///
/// The gateway speaks the **OpenAI Responses** contract, so the token cap field
/// is `max_output_tokens`. Sending `maxTokens` (an old camelCase guess) is
/// silently ignored and the request then FAILS inference: the gateway returns
/// HTTP 200 with an opaque `event: error` / `{"error":"inference failed"}`,
/// which CAR reads as "no content" and falls back to a local model — the bug
/// that made every `parslee/*` call quietly serve on-device instead of gpt-5.x.
/// Verified against the live gateway. `temperature` is likewise omitted:
/// reasoning models (gpt-5.x) reject `temperature != 1` with the same opaque
/// `event: error`, so we use the model default (1.0).
fn parslee_request_body(
    user_id: &str,
    input: Vec<serde_json::Value>,
    max_tokens: usize,
    tools: Option<&[serde_json::Value]>,
    model: Option<&str>,
) -> serde_json::Value {
    let mut body = serde_json::json!({
        "userId": user_id,
        "input": input,
        "max_output_tokens": max_tokens,
        "store": false,
        "include": ["reasoning.encrypted_content"],
    });
    if let Some(tools) = tools {
        if !tools.is_empty() {
            body["tools"] = serde_json::json!(tools);
        }
    }
    // Existing Parslee aliases deliberately retain their historic request
    // shape. Curated gateway entries must carry their allow-listed id so the
    // backend can select the requested upstream OpenRouter model.
    if let Some(model) = model.filter(|model| model.starts_with("parslee/openrouter/")) {
        body["model"] = serde_json::json!(model);
    }
    body
}

/// Translate the protocol-neutral (OpenAI Chat-Completions-shaped) message list
/// into OpenAI **Responses** `input` items.
///
/// The critical case is the multi-turn TOOL conversation. In Chat-Completions
/// shape an assistant tool call is `{role:"assistant", tool_calls:[{id, function:
/// {name, arguments}}]}` (no text `content`) and a tool result is `{role:"tool",
/// tool_call_id, content}`. The Responses contract represents those as distinct
/// item types — `function_call` (with `call_id`/`name`/`arguments`) and
/// `function_call_output` (with `call_id`/`output`) — NOT as `{role, content}`
/// objects. The earlier code forwarded only messages that had a text `content`,
/// so it silently **dropped every assistant tool call** (empty content) and
/// orphaned every tool result. The model therefore never saw its own tool
/// history: a coding agent would `read_file` the same files every turn, forever,
/// never progressing to an edit. Surfaced by dogfooding CAR's coder on gpt-5.5
/// (a 2-bug fix task issued 210 `read_file` calls and zero edits).
fn parslee_input_items(
    system: Option<&str>,
    messages: &[serde_json::Value],
) -> Vec<serde_json::Value> {
    let mut input: Vec<serde_json::Value> = Vec::new();
    if let Some(sys) = system {
        if !sys.trim().is_empty() {
            input.push(serde_json::json!({ "role": "system", "content": sys }));
        }
    }
    for m in messages {
        // `OpenAiResponsesHandler::build_messages` already emits typed
        // Responses items (function_call, function_call_output, reasoning)
        // without a role. Managed inference speaks that same contract, so
        // carry these items through byte-for-byte instead of interpreting
        // them as Chat Completions messages.
        if m.get("role").is_none() && m.get("type").is_some() {
            input.push(m.clone());
            continue;
        }
        let role = m.get("role").and_then(|v| v.as_str()).unwrap_or("user");
        match role {
            // Assistant turn: emit one `function_call` item per tool call
            // (the history the old loop erased), then any assistant text.
            "assistant" => {
                if let Some(tcs) = m.get("tool_calls").and_then(|v| v.as_array()) {
                    for tc in tcs {
                        let call_id = tc.get("id").and_then(|v| v.as_str()).unwrap_or("");
                        let func = tc.get("function");
                        let name = func
                            .and_then(|f| f.get("name"))
                            .and_then(|v| v.as_str())
                            .unwrap_or("");
                        // `arguments` is a JSON string in Chat-Completions shape;
                        // tolerate an inlined object by stringifying it.
                        let args = func.and_then(|f| f.get("arguments")).map_or_else(
                            || "{}".to_string(),
                            |a| {
                                a.as_str()
                                    .map(str::to_string)
                                    .unwrap_or_else(|| a.to_string())
                            },
                        );
                        input.push(serde_json::json!({
                            "type": "function_call",
                            "call_id": call_id,
                            "name": name,
                            "arguments": args,
                        }));
                    }
                }
                let text = m.get("content").and_then(|v| v.as_str()).unwrap_or("");
                if !text.is_empty() {
                    input.push(serde_json::json!({ "role": "assistant", "content": text }));
                }
            }
            // Tool result → `function_call_output` keyed by the call it answers.
            "tool" => {
                let call_id = m.get("tool_call_id").and_then(|v| v.as_str()).unwrap_or("");
                let output = m.get("content").and_then(|v| v.as_str()).unwrap_or("");
                input.push(serde_json::json!({
                    "type": "function_call_output",
                    "call_id": call_id,
                    "output": output,
                }));
            }
            // Plain user/system/other turn.
            _ => {
                let content = m.get("content").and_then(|v| v.as_str()).unwrap_or("");
                if !content.is_empty() {
                    input.push(serde_json::json!({ "role": role, "content": content }));
                }
            }
        }
    }
    if tracing::enabled!(tracing::Level::DEBUG) {
        let fc = input
            .iter()
            .filter(|i| i["type"] == "function_call")
            .count();
        let fco = input
            .iter()
            .filter(|i| i["type"] == "function_call_output")
            .count();
        tracing::debug!(
            target = "car::parslee_input",
            messages_in = messages.len(),
            items_out = input.len(),
            function_calls = fc,
            function_call_outputs = fco,
            "built parslee Responses input"
        );
    }
    input
}

/// Truncate a prompt to fit within a context window.
/// Keeps the most recent content (suffix) since that's most relevant.
/// Returns the truncated prompt.
/// Truncate a prompt to fit within a context window.
/// This is a last-resort fallback — callers should prefer compaction via
/// car-memgine which preserves semantic content. This function is only
/// used when compaction isn't available (e.g., raw API calls).
fn truncate_prompt_to_fit(
    prompt: &str,
    context: Option<&str>,
    tools_json: Option<&[serde_json::Value]>,
    max_tokens: usize,
    media_tokens: usize,
    context_window: usize,
) -> String {
    let context_tokens = context.map(estimate_tokens).unwrap_or(0);
    let tools_tokens = tools_json
        .map(|t| estimate_tokens(&serde_json::to_string(t).unwrap_or_default()))
        .unwrap_or(0);
    // Reserve space for: context + tools + media/history (media blocks
    // at provider-calibrated rates plus multi-turn history text, both
    // pre-estimated by the caller via `crate::media_tokens`) +
    // max_tokens (output) + overhead
    let overhead = 100; // message framing, special tokens
    let reserved = context_tokens + tools_tokens + media_tokens + max_tokens + overhead;
    let available = context_window.saturating_sub(reserved);

    let prompt_tokens = estimate_tokens(prompt);
    if prompt_tokens <= available {
        return prompt.to_string();
    }

    tracing::warn!(
        prompt_tokens = prompt_tokens,
        available_tokens = available,
        context_window = context_window,
        "truncating prompt to fit context window (prefer compaction via car-memgine)"
    );

    // Truncate from the beginning (keep the end, which is most recent/relevant)
    // Estimate chars to keep based on available tokens
    let chars_to_keep = available * 4;
    if chars_to_keep >= prompt.len() {
        return prompt.to_string();
    }

    let start = prompt.len().saturating_sub(chars_to_keep);
    let safe_start = prompt.ceil_char_boundary(start);
    let truncated = &prompt[safe_start..];
    // Find a clean break point (newline or space)
    let break_point = truncated
        .find('\n')
        .or_else(|| truncated.find(' '))
        .unwrap_or(0);

    format!(
        "[...truncated...]\n{}",
        truncated[break_point..].trim_start()
    )
}

impl RemoteBackend {
    /// Infallible by design. A trust-store failure degrades the client's trust
    /// posture (and says so, loudly, once) rather than propagating a `Result`
    /// up through `InferenceEngine::new` and the daemon's `OnceLock` engine
    /// accessor, which does not poison and so would turn one bad build into a
    /// permanently stuck daemon.
    pub fn new() -> Self {
        let (client, tls_degradation) = crate::tls_client::build_client_with_degradation(
            &crate::tls_client::REMOTE_BACKEND,
            || {
                Client::builder()
                    // A large completion (e.g. a multi-KB file written as one tool
                    // argument at a 32K output cap) holds a non-streamed socket silent
                    // for minutes while generating. The old 120s/90s ceilings killed
                    // those legitimate requests ("error sending request"). These are
                    // upper bounds, not waits — raising them never slows a fast call.
                    .timeout(std::time::Duration::from_secs(300))
                    .connect_timeout(std::time::Duration::from_secs(10))
                    .read_timeout(std::time::Duration::from_secs(180))
            },
        );
        Self {
            client,
            tls_degradation,
            key_pool: KeyPool::new(),
        }
    }

    /// Turn a transport error from a request made on this backend's client
    /// into an [`InferenceError`], naming the trust degradation when — and
    /// only when — that degradation could plausibly be the cause.
    pub(crate) fn request_error(&self, context: &str, err: &reqwest::Error) -> InferenceError {
        InferenceError::InferenceFailed(format!(
            "{context}: {}{}",
            error_chain(err),
            self.tls_degradation_note(err)
        ))
    }

    /// Preserve the typed retry classification while adding this backend's
    /// trust-store degradation context to the error operators will see.
    fn transport_attempt_error(
        &self,
        context: &str,
        err: &reqwest::Error,
    ) -> TransportAttemptError {
        TransportAttemptError::from_reqwest_error(self.request_error(context, err), err)
    }

    /// The trailing note appended by [`RemoteBackend::request_error`]. Empty
    /// unless this backend actually degraded *and* the error is the class a
    /// missing trust root produces.
    fn tls_degradation_note(&self, err: &reqwest::Error) -> String {
        match &self.tls_degradation {
            Some(degradation) if is_trust_related(err.is_connect(), err.is_timeout()) => {
                format!(" [{degradation}]")
            }
            _ => String::new(),
        }
    }

    /// Register all keys from a model schema into the key pool.
    pub async fn register_model_keys(&self, schema: &ModelSchema) {
        if let ModelSource::RemoteApi { ref endpoint, .. } = schema.source {
            let env_vars = schema.all_api_key_envs();
            if !env_vars.is_empty() {
                self.key_pool.register_endpoint(endpoint, env_vars).await;
            }
        }
    }

    /// Unified request execution using the protocol handler abstraction.
    /// All public generate methods delegate to this.
    #[instrument(
        name = "inference.remote_call",
        skip_all,
        fields(
            model = %schema.name,
            provider = %schema.provider,
        ),
    )]
    async fn execute_request(
        &self,
        schema: &ModelSchema,
        req: crate::protocol::ApiRequest,
    ) -> Result<crate::protocol::ApiResponse, InferenceError> {
        let (endpoint, protocol) = extract_remote_endpoint(schema)?;
        let lease = self.lease_key(schema, &endpoint).await?;
        // Parslee managed inference (`parslee/*`) is served by the shipped
        // inference gateway (`/api/v1/orgs/{org}/inference/responses`, SSE): a
        // thin passthrough to the org's chat model with no agent orchestration.
        // (This replaced the `/chat/stream` Employee-assistant path, which
        // wrapped the same model in memory/RAG/plugins + extra sub-calls —
        // ~30x slower for callers that just want tokens.) Route it through the
        // dedicated path instead of the generic ProtocolHandler pipeline. No
        // key_pool (single bearer).
        if matches!(schema.source, ModelSource::Proprietary { .. }) {
            // #319 (won't-fix): Proprietary is single-bearer, so key_pool endpoint stats (which exist only for multi-key load-balancing) are intentionally skipped — Parslee health is tracked at the model level via the outcome tracker (outcome.rs).
            // The Parslee inference endpoint speaks the OpenAI Responses API:
            // it takes structured `input` items and function `tools` and streams
            // Responses SSE, so tool-calling works here like any other remote
            // (parslee is gpt-5.5 today, which supports it). Structured output
            // (`response_format` / JSON schema via `text.format`) is not wired on
            // this endpoint yet, so reject that one clearly rather than silently
            // dropping it — mirroring the audio/video guard in
            // `generate_with_tools_multi`.
            if req.response_format.is_some() {
                return Err(InferenceError::UnsupportedMode {
                    mode: "structured-output",
                    backend: "parslee-inference",
                    reason: "the Parslee inference endpoint does not accept a \
                             response_format / JSON schema yet — route structured-output \
                             requests to a model that supports it",
                });
            }
            match self
                .parslee_assistant_request(&endpoint, &lease.api_key, &req)
                .await
            {
                Ok(resp) => return Ok(resp),
                Err(e) if is_auth_rejection(&e) => {
                    // The bearer was rejected (401/403). Reactively mint a
                    // fresh one and retry ONCE — the proactive expiry window
                    // (car_auth::access_token_refreshing) misses tokens that
                    // are revoked early or stored without an expiry, which is
                    // exactly the 401 burst that poisoned model health (#313).
                    match car_auth::force_refresh().await {
                        Some(fresh) => {
                            return self
                                .parslee_assistant_request(&endpoint, &fresh, &req)
                                .await;
                        }
                        // No refresh token, or refresh failed: surface the
                        // original auth error unchanged.
                        None => return Err(e),
                    }
                }
                Err(e) => return Err(e),
            }
        }

        // AWS Bedrock Converse — SigV4-signed, region-scoped, its own request
        // shape. Handled here (not the generic handler path) because signing
        // needs the assembled body + URL + timestamp. `endpoint` is the region.
        if matches!(protocol, ApiProtocol::Bedrock) {
            return self.bedrock_converse_request(&endpoint, &req).await;
        }

        let handler = crate::protocol::handler_for(protocol);
        let start = std::time::Instant::now();

        // Build URL
        let api_version = match &schema.source {
            ModelSource::RemoteApi { api_version, .. } => api_version.clone(),
            _ => None,
        };
        let url = if matches!(protocol, ApiProtocol::Google) {
            crate::protocol::google_url(&endpoint, &req.model, &lease.api_key)
        } else if matches!(protocol, ApiProtocol::VertexAi) {
            crate::protocol::vertex_url(&endpoint, &req.model)
        } else if matches!(protocol, ApiProtocol::AzureOpenAi) {
            let version = api_version.as_deref().unwrap_or("2024-10-21");
            format!(
                "{}/openai/deployments/{}/chat/completions?api-version={}",
                endpoint.trim_end_matches('/'),
                req.model,
                version
            )
        } else {
            format_endpoint(&endpoint, chat_path_for(schema, handler.as_ref()))
        };

        // Build headers
        let headers = handler.auth_headers(&lease.api_key);

        // Build request body
        let body = handler.build_request_body(&req);

        debug!(url = %url, model = %req.model, "protocol handler request");

        // Execute HTTP request with a bounded transient-retry loop. A genuine
        // blip (transport reset, timeout, 5xx/429/529 overloaded) must not throw
        // away a long-horizon task that has already done real work; client errors
        // (4xx validation/auth) are NOT retried — re-sending the same bad request
        // is pointless. The per-attempt tokio timeout (#26) is the safety net for
        // an OS/kernel-level socket hang reqwest's own timeout can't cancel.
        let mut attempt = 0u32;
        let resp_text = loop {
            attempt += 1;
            let send_outcome: Result<(reqwest::StatusCode, String), TransportAttemptError> =
                async {
                    let mut builder = self.client.post(&url);
                    for (name, value) in &headers {
                        builder = builder.header(name.as_str(), value.as_str());
                    }
                    let send_fut = builder.json(&body).send();
                    let resp = tokio::time::timeout(std::time::Duration::from_secs(300), send_fut)
                        .await
                        .map_err(|_| {
                            TransportAttemptError::timeout(
                                "request timed out after 300s (tokio safety timeout)",
                            )
                        })?
                        .map_err(|e| self.transport_attempt_error("HTTP error", &e))?;
                    let status = resp.status();
                    let body_fut = resp.text();
                    let txt = tokio::time::timeout(std::time::Duration::from_secs(300), body_fut)
                        .await
                        .map_err(|_| {
                            TransportAttemptError::timeout(
                                "response body read timed out after 300s",
                            )
                        })?
                        .map_err(|e| self.transport_attempt_error("read body", &e))?;
                    Ok((status, txt))
                }
                .await;

            match send_outcome {
                Ok((status, txt)) if status.is_success() => break txt,
                Ok((status, txt)) => {
                    // Non-2xx: keep the existing key_pool failure accounting on
                    // EVERY attempt so multi-key load balancing / rate-limit
                    // detection stays accurate.
                    let is_rl = txt.contains("429") || txt.contains("RESOURCE_EXHAUSTED");
                    self.key_pool
                        .report_failure(&endpoint, &lease.env_var, is_rl)
                        .await;
                    let err_msg = format!("API returned {status}: {txt}");
                    let transient = is_transient_http_status(status.as_u16());
                    if attempt < REMOTE_MAX_ATTEMPTS && transient {
                        tokio::time::sleep(std::time::Duration::from_secs(
                            REMOTE_BACKOFF_BASE_SECS * attempt as u64,
                        ))
                        .await;
                        continue;
                    }
                    // Account-level rejection (bad key / no credits) — provider-
                    // wide, not model-specific. Typed so the dispatch loop can
                    // resolve it as an unattributed receipt instead of blaming
                    // the model's health and tripping its breaker over someone's
                    // billing (Parslee-ai/car#650). Checked ahead of the
                    // per-protocol branches because it is true of all of them.
                    if is_account_http_status(status.as_u16()) {
                        return Err(InferenceError::ProviderAccount {
                            provider: schema.provider.clone(),
                            status: status.as_u16(),
                            message: account_error_message(protocol, status.as_u16()),
                        });
                    }
                    if matches!(protocol, ApiProtocol::OpenRouter) {
                        return Err(if transient {
                            InferenceError::Transient {
                                status: Some(status.as_u16()),
                                message: format!(
                                    "OpenRouter upstream failed after {REMOTE_MAX_ATTEMPTS} attempts"
                                ),
                            }
                        } else {
                            openrouter_http_error(status, &txt, &req.model)
                        });
                    }
                    // A retryable status that exhausted the budget is an infra
                    // blip (safe to re-run); a non-retryable status (4xx/auth)
                    // is a request-level error. Tag the former so callers don't
                    // have to sniff the message.
                    return Err(if transient {
                        InferenceError::Transient {
                            status: Some(status.as_u16()),
                            message: err_msg,
                        }
                    } else {
                        InferenceError::InferenceFailed(err_msg)
                    });
                }
                Err(e) => {
                    // Transport / timeout error. The verdict was decided on the
                    // typed reqwest error at the point of failure, not sniffed
                    // back out of the rendered message.
                    let transient = e.retryable;
                    if attempt < REMOTE_MAX_ATTEMPTS && transient {
                        tokio::time::sleep(std::time::Duration::from_secs(
                            REMOTE_BACKOFF_BASE_SECS * attempt as u64,
                        ))
                        .await;
                        continue;
                    }
                    // Exhausted a retryable transport class -> Transient (re-run);
                    // a non-transient transport error stays as-is.
                    return Err(if transient {
                        InferenceError::Transient {
                            status: None,
                            message: e.error.to_string(),
                        }
                    } else {
                        e.error
                    });
                }
            }
        };

        let latency_ms = start.elapsed().as_millis() as u64;

        let est_tokens = req
            .messages
            .iter()
            .filter_map(|m| m.get("content").and_then(|c| c.as_str()))
            .map(|s| s.len() / 4)
            .sum::<usize>() as u64;
        self.key_pool
            .report_success(&endpoint, &lease.env_var, latency_ms, est_tokens, 0)
            .await;

        let mut response = handler.parse_response(&resp_text)?;
        // Fill in context_window from the model schema
        if let Some(ref mut usage) = response.usage {
            usage.context_window = schema.context_length as u64;
        }
        Ok(response)
    }

    /// AWS Bedrock Converse request — SigV4-signed POST to
    /// `bedrock-runtime.{region}.amazonaws.com/model/{modelId}/converse`.
    /// `region` is the model's configured `endpoint`; `req.model` is the
    /// Bedrock model id. Credentials come from the standard AWS env vars.
    async fn bedrock_converse_request(
        &self,
        region: &str,
        req: &crate::protocol::ApiRequest,
    ) -> Result<crate::protocol::ApiResponse, InferenceError> {
        let creds = crate::aws_sigv4::AwsCredentials::from_env().ok_or_else(|| {
            InferenceError::InferenceFailed(
                "Bedrock requires AWS credentials: set AWS_ACCESS_KEY_ID and \
                 AWS_SECRET_ACCESS_KEY (plus AWS_SESSION_TOKEN for temporary creds)"
                    .to_string(),
            )
        })?;
        if region.trim().is_empty() {
            return Err(InferenceError::InferenceFailed(
                "Bedrock model endpoint must be the AWS region (e.g. \"us-east-1\")".to_string(),
            ));
        }

        let handler = crate::protocol::BedrockHandler;
        let body = handler.build_request_body(req);
        let body_bytes = serde_json::to_vec(&body)
            .map_err(|e| InferenceError::InferenceFailed(format!("serialize Bedrock body: {e}")))?;

        let host = format!("bedrock-runtime.{region}.amazonaws.com");
        // The model id contains `:` — encode it identically for the URL and the
        // signed canonical path (else the SigV4 signature won't match).
        let encoded_model = crate::aws_sigv4::uri_encode_segment(&req.model);
        let canonical_path = format!("/model/{encoded_model}/converse");
        let url = format!("https://{host}{canonical_path}");

        let now = chrono::Utc::now();
        let amz_date = now.format("%Y%m%dT%H%M%SZ").to_string();
        let date_stamp = now.format("%Y%m%d").to_string();
        let base_headers = vec![
            ("host".to_string(), host.clone()),
            ("content-type".to_string(), "application/json".to_string()),
        ];
        let signed = crate::aws_sigv4::signed_headers(
            &creds,
            region,
            "bedrock",
            "POST",
            &canonical_path,
            "",
            &base_headers,
            &body_bytes,
            &amz_date,
            &date_stamp,
        );

        let mut builder = self.client.post(&url);
        for (k, v) in &signed {
            // reqwest sets Host itself from the URL (matching what we signed).
            if k != "host" {
                builder = builder.header(k, v);
            }
        }
        // Send the EXACT bytes that were hashed for the signature (not `.json()`,
        // which would re-serialize and could differ).
        let send_fut = builder.body(body_bytes).send();
        let resp = tokio::time::timeout(std::time::Duration::from_secs(300), send_fut)
            .await
            .map_err(|_| {
                InferenceError::InferenceFailed("Bedrock request timed out after 300s".to_string())
            })?
            .map_err(|e| self.request_error("HTTP error", &e))?;
        let status = resp.status();
        let txt = resp.text().await.map_err(|e| {
            InferenceError::InferenceFailed(format!("read Bedrock body: {}", error_chain(&e)))
        })?;
        if !status.is_success() {
            return Err(InferenceError::InferenceFailed(format!(
                "Bedrock returned {status}: {txt}"
            )));
        }
        // context_window on usage is filled in by the caller from the schema.
        handler.parse_response(&txt)
    }

    /// Resolve `(org_id, user_id)` for a Parslee bearer, cached per API
    /// base and token. Mirrors Parslee Hydra: org from `/api/v1/organizations/me`,
    /// a user identifier (account email, best-effort) from
    /// `/connect/session`. The backend authorizes off the bearer; the
    /// `userId` in the chat body is informational, so a missing email
    /// degrades to `"car"` rather than failing the request.
    async fn parslee_identity(
        &self,
        endpoint: &str,
        bearer: &str,
    ) -> Result<(String, String), InferenceError> {
        let cache = PARSLEE_IDENTITY.get_or_init(|| tokio::sync::Mutex::new(HashMap::new()));
        let base = endpoint.trim_end_matches('/');
        let cache_key = (base.to_string(), bearer.to_string());
        if let Some(v) = cache.lock().await.get(&cache_key) {
            return Ok(v.clone());
        }

        let org_url = format!("{base}/api/v1/organizations/me");
        let org_resp = self
            .client
            .get(&org_url)
            .bearer_auth(bearer)
            .send()
            .await
            .map_err(|e| self.request_error("Parslee org lookup", &e))?;
        if !org_resp.status().is_success() {
            let s = org_resp.status();
            let b = org_resp.text().await.unwrap_or_default();
            return Err(InferenceError::InferenceFailed(format!(
                "Parslee org lookup failed: HTTP {s}: {b}"
            )));
        }
        let org_json: serde_json::Value = org_resp.json().await.map_err(|e| {
            InferenceError::InferenceFailed(format!("parse Parslee org: {}", error_chain(&e)))
        })?;
        let org_id = org_json
            .get("organizationId")
            .or_else(|| org_json.get("OrganizationId"))
            .and_then(|v| v.as_str())
            .ok_or_else(|| {
                InferenceError::InferenceFailed(
                    "Parslee org response has no organizationId — the signed-in \
                     account has no workspace yet (sign in via CAR Host.app or \
                     the web to finish onboarding)"
                        .to_string(),
                )
            })?
            .to_string();

        let sess_url = format!("{base}/connect/session");
        let user_id = match self.client.get(&sess_url).bearer_auth(bearer).send().await {
            Ok(r) if r.status().is_success() => r
                .json::<serde_json::Value>()
                .await
                .ok()
                .and_then(|j| {
                    j.get("account")
                        .and_then(|a| a.get("email"))
                        .and_then(|v| v.as_str())
                        .map(String::from)
                })
                .unwrap_or_else(|| "car".to_string()),
            _ => "car".to_string(),
        };

        let pair = (org_id, user_id);
        cache.lock().await.insert(cache_key, pair.clone());
        Ok(pair)
    }

    /// Parslee managed inference via the shipped inference gateway
    /// endpoint (`POST /api/v1/orgs/{org}/inference/responses`, SSE). The
    /// endpoint speaks the OpenAI Responses contract: CAR sends `input`
    /// items + function `tools` and streams Responses SSE back. `infer` is
    /// non-streaming, so we buffer the SSE body and concatenate the deltas.
    /// Zero backend deps — works against today's deployed Parslee.
    async fn parslee_assistant_request(
        &self,
        endpoint: &str,
        bearer: &str,
        req: &crate::protocol::ApiRequest,
    ) -> Result<crate::protocol::ApiResponse, InferenceError> {
        let (org_id, user_id) = self.parslee_identity(endpoint, bearer).await?;

        // Responses-API request: `input` items (system first, then the
        // conversation), function `tools`, and the caller's token cap /
        // temperature. The endpoint speaks the OpenAI Responses contract, so the
        // reply streams back as Responses SSE we parse with the shared handler.
        // Build Responses `input` items — including the multi-turn tool history
        // (function_call / function_call_output), which the model needs to see
        // to progress instead of re-issuing the same reads every turn.
        let input = parslee_input_items(req.system.as_deref(), &req.messages);
        if input.is_empty() {
            return Err(InferenceError::InferenceFailed(
                "Parslee inference: empty prompt".to_string(),
            ));
        }

        let url = format!(
            "{}/api/v1/orgs/{}/inference/responses",
            endpoint.trim_end_matches('/'),
            org_id
        );
        // Deliberately do NOT forward `temperature` (see `parslee_request_body`).
        let _ = req.temperature;
        let body = parslee_request_body(
            &user_id,
            input,
            req.max_tokens,
            req.tools.as_deref(),
            Some(&req.model),
        );
        let send_fut = self
            .client
            .post(&url)
            .bearer_auth(bearer)
            .header("content-type", "application/json")
            .header("accept", "text/event-stream")
            .json(&body)
            .send();
        let resp = tokio::time::timeout(std::time::Duration::from_secs(300), send_fut)
            .await
            .map_err(|_| {
                InferenceError::InferenceFailed("Parslee chat request timed out (150s)".to_string())
            })?
            .map_err(|e| self.request_error("Parslee chat HTTP error", &e))?;
        let status = resp.status();
        let text_fut = resp.text();
        let raw = tokio::time::timeout(std::time::Duration::from_secs(300), text_fut)
            .await
            .map_err(|_| {
                InferenceError::InferenceFailed(
                    "Parslee chat body read timed out (120s)".to_string(),
                )
            })?
            .map_err(|e| {
                InferenceError::InferenceFailed(format!(
                    "Parslee chat read body: {}",
                    error_chain(&e)
                ))
            })?;
        if !status.is_success() {
            if let Some(detail) = gateway_unconfigured_detail(status, &raw) {
                return Err(gateway_unconfigured_error(status, detail, &req.model));
            }
            return Err(InferenceError::InferenceFailed(format!(
                "Parslee chat failed: HTTP {status}: {raw}"
            )));
        }

        // Fold the Responses SSE into a final result: text, tool calls, usage,
        // and stop_reason — reusing the shared per-line parser + accumulator, so
        // parslee gets the same tool-call / usage handling as every other remote.
        let mut acc = crate::stream::StreamAccumulator::default();
        let mut stream_error = None;
        let mut saw_completed = false;
        for event in parse_parslee_responses_sse(&raw) {
            if let crate::stream::StreamEvent::Error(message) = &event {
                stream_error = Some(message.clone());
            }
            if matches!(event, crate::stream::StreamEvent::Done { .. }) {
                saw_completed = true;
            }
            acc.push(&event);
        }
        let (text, tool_calls, usage, stop_reason, provider_output_items) =
            acc.finish_with_provider_output_items();
        if let Some(message) = stream_error {
            // A refusal from a filter in front of the model is a ruling about
            // the request, not a crash — and the caller's response to each is
            // different (#796). Classified here rather than left to the caller
            // to substring-match, which is what made a benchmark unable to
            // score a refusal separately from a failure.
            if let Some((kind, code)) = content_refusal_tags(&message) {
                return Err(InferenceError::ContentRefused {
                    provider: "parslee".to_string(),
                    kind,
                    code,
                    message,
                });
            }
            return Err(InferenceError::InferenceFailed(message));
        }
        if !saw_completed {
            return Err(InferenceError::InferenceFailed(
                "managed inference stream ended without response.completed".to_string(),
            ));
        }
        if text.is_empty() && tool_calls.is_empty() {
            // The stream carried no assistant text and no tool call. The most
            // common cause is a server-side `error` SSE event (the controller
            // catches a streaming exception and emits `event: error`), which the
            // Responses parser has no content mapping for — surfacing a bare "no
            // content" hides the real reason. Include the raw body (truncated) so
            // the failure is actionable instead of opaque.
            let snippet: String = raw.trim().chars().take(600).collect();
            return Err(InferenceError::InferenceFailed(format!(
                "Parslee inference returned no content (HTTP {status}); raw response: {snippet}"
            )));
        }
        Ok(crate::protocol::ApiResponse {
            text,
            tool_calls,
            provider_output_items,
            // The Parslee Responses-SSE path carries no Anthropic thinking blocks.
            thinking: Vec::new(),
            usage,
            stop_reason,
        })
    }

    async fn parslee_assistant_stream_request(
        &self,
        endpoint: &str,
        bearer: &str,
        req: &crate::protocol::ApiRequest,
        spend_guard: Option<crate::routing_ext::MidStreamSpendGuard>,
    ) -> Result<tokio::sync::mpsc::Receiver<crate::stream::StreamEvent>, InferenceError> {
        let (org_id, user_id) = self.parslee_identity(endpoint, bearer).await?;
        let input = parslee_input_items(req.system.as_deref(), &req.messages);
        if input.is_empty() {
            return Err(InferenceError::InferenceFailed(
                "Parslee inference: empty prompt".to_string(),
            ));
        }
        let url = format!(
            "{}/api/v1/orgs/{}/inference/responses",
            endpoint.trim_end_matches('/'),
            org_id
        );
        let body = parslee_request_body(
            &user_id,
            input,
            req.max_tokens,
            req.tools.as_deref(),
            Some(&req.model),
        );
        let resp = self
            .client
            .post(url)
            .bearer_auth(bearer)
            .header("content-type", "application/json")
            .header("accept", "text/event-stream")
            .json(&body)
            .send()
            .await
            .map_err(|error| self.request_error("Parslee stream HTTP error", &error))?;
        let status = resp.status();
        if !status.is_success() {
            let body = resp.text().await.unwrap_or_default();
            // Same classification as the non-streaming path — a caller who
            // streams must not get a generic failure for a condition the
            // blocking call reports precisely.
            if let Some(detail) = gateway_unconfigured_detail(status, &body) {
                return Err(gateway_unconfigured_error(status, detail, &req.model));
            }
            return Err(InferenceError::InferenceFailed(format!(
                "Parslee stream failed: HTTP {status}: {body}"
            )));
        }

        let (tx, rx) = tokio::sync::mpsc::channel(64);
        tokio::spawn(async move {
            use futures::StreamExt;
            let mut stream = resp.bytes_stream();
            let mut buffer = Vec::<u8>::new();
            let mut spend_guard = spend_guard;
            let mut saw_completed = false;
            while let Some(chunk) = stream.next().await {
                match chunk {
                    Ok(bytes) => buffer.extend_from_slice(&bytes),
                    Err(error) => {
                        tracing::warn!(%error, "Parslee stream transport error");
                        let _ = tx
                            .send(crate::stream::StreamEvent::Error(
                                "Parslee stream transport error".to_string(),
                            ))
                            .await;
                        return;
                    }
                }
                while let Some((position, separator_len)) = find_sse_separator(&buffer) {
                    let block = String::from_utf8_lossy(&buffer[..position]).into_owned();
                    buffer.drain(..position + separator_len);
                    for event in parse_parslee_responses_sse(&block) {
                        if let Some(guard) = spend_guard.as_mut() {
                            if let crate::stream::StreamEvent::Usage { input_tokens, .. } = &event {
                                guard.note_prompt_tokens(*input_tokens);
                            }
                            let delta = match &event {
                                crate::stream::StreamEvent::TextDelta(text) => Some(text.as_str()),
                                crate::stream::StreamEvent::ToolCallDelta {
                                    arguments_delta,
                                    ..
                                } => Some(arguments_delta.as_str()),
                                _ => None,
                            };
                            if let Some(delta) = delta {
                                let tokens =
                                    crate::routing_ext::MidStreamSpendGuard::estimate_tokens(delta);
                                if let Some(exceeded) = guard.record_tokens(tokens) {
                                    let _ = tx.send(event).await;
                                    let _ = tx
                                        .send(crate::stream::StreamEvent::StopReason(format!(
                                            "spend_limit: {exceeded}"
                                        )))
                                        .await;
                                    let _ = tx
                                        .send(crate::stream::StreamEvent::Error(
                                            "stream cancelled because the spend limit was reached"
                                                .to_string(),
                                        ))
                                        .await;
                                    return;
                                }
                            }
                        }
                        if matches!(event, crate::stream::StreamEvent::Done { .. }) {
                            saw_completed = true;
                        }
                        let terminal = matches!(event, crate::stream::StreamEvent::Error(_));
                        if tx.send(event).await.is_err() || terminal {
                            return;
                        }
                    }
                }
            }
            if !buffer.is_empty() || !saw_completed {
                let _ = tx
                    .send(crate::stream::StreamEvent::Error(
                        "managed inference stream ended without response.completed".to_string(),
                    ))
                    .await;
            }
        });
        Ok(rx)
    }

    /// Generate text via a remote API, using load-balanced key selection.
    /// Auto-truncates the prompt if it exceeds the model's context window.
    pub async fn generate(
        &self,
        schema: &ModelSchema,
        prompt: &str,
        context: Option<&str>,
        temperature: f64,
        max_tokens: usize,
        images: Option<&[ContentBlock]>,
    ) -> Result<String, InferenceError> {
        let resp = self
            .generate_with_tools_multi(
                schema,
                prompt,
                context,
                temperature,
                max_tokens,
                None,
                images,
                None,
                None,
                None,
                0,
                false,
                crate::tasks::generate::CacheTtl::default(),
                None,
                None,
            )
            .await?;
        Ok(resp.0)
    }

    /// Generate with optional tool definitions and multi-turn conversation history.
    /// Auto-truncates the prompt if it exceeds the model's context window.
    ///
    /// When `messages` is provided, builds a proper multi-turn conversation
    /// instead of a single user message. This enables tool_use → tool_result flows.
    ///
    /// Post-processes `done` tool calls: if the result argument is suspiciously
    /// short (< 50 chars) and the model also produced text output, enriches
    /// the done result with the text (fixes #10).
    pub async fn generate_with_tools(
        &self,
        schema: &ModelSchema,
        prompt: &str,
        context: Option<&str>,
        temperature: f64,
        max_tokens: usize,
        tools: Option<&[serde_json::Value]>,
        images: Option<&[ContentBlock]>,
    ) -> Result<(String, Vec<crate::tasks::generate::ToolCall>), InferenceError> {
        let (text, calls, _thinking, _provider_items, _usage, _stop) = self
            .generate_with_tools_multi(
                schema,
                prompt,
                context,
                temperature,
                max_tokens,
                tools,
                images,
                None,
                None,
                None,
                0,
                false,
                crate::tasks::generate::CacheTtl::default(),
                None,
                None,
            )
            .await?;
        Ok((text, calls))
    }

    /// Generate with multi-turn conversation support and optional extended thinking.
    /// Uses the unified protocol handler abstraction.
    ///
    /// When `cache_control` is true, system prompt and tool definitions are marked
    /// with Anthropic prompt caching breakpoints for cache reuse across calls.
    pub async fn generate_with_tools_multi(
        &self,
        schema: &ModelSchema,
        prompt: &str,
        context: Option<&str>,
        temperature: f64,
        max_tokens: usize,
        tools: Option<&[serde_json::Value]>,
        images: Option<&[ContentBlock]>,
        messages: Option<&[crate::tasks::generate::Message]>,
        tool_choice: Option<&str>,
        parallel_tool_calls: Option<bool>,
        budget_tokens: usize,
        cache_control: bool,
        cache_ttl: crate::tasks::generate::CacheTtl,
        system_stable_prefix: Option<&str>,
        response_format: Option<&crate::tasks::generate::ResponseFormat>,
    ) -> Result<
        (
            String,
            Vec<crate::tasks::generate::ToolCall>,
            Vec<crate::tasks::generate::ThinkingBlock>,
            Vec<serde_json::Value>,
            Option<crate::TokenUsage>,
            Option<String>,
        ),
        InferenceError,
    > {
        let (_, protocol) = extract_remote_endpoint(schema)?;
        let handler = crate::protocol::handler_for(protocol);

        // Pre-check: reject video / audio content blocks on protocols
        // that don't accept them natively. Silent stringification to
        // `[video: <source>]` or `[audio: <source>]` is a correctness
        // trap — the model would answer confidently about content it
        // never saw. Surface as an explicit typed error instead.
        if !handler.supports_video() {
            let has_video_in_images =
                images.is_some_and(|blocks| blocks.iter().any(ContentBlock::is_video));
            let has_video_in_messages = messages.is_some_and(|msgs| {
                msgs.iter().any(|msg| match msg {
                    crate::tasks::generate::Message::UserMultimodal { content } => {
                        content.iter().any(ContentBlock::is_video)
                    }
                    _ => false,
                })
            });
            if has_video_in_images || has_video_in_messages {
                return Err(InferenceError::UnsupportedMode {
                    mode: "video-content-block",
                    backend: handler.protocol_name(),
                    reason: "this remote protocol has no native video input path; route to \
                         a provider that implements ProtocolHandler::supports_video() (Gemini)",
                });
            }
        }
        if !handler.supports_audio() {
            let has_audio_in_images =
                images.is_some_and(|blocks| blocks.iter().any(ContentBlock::is_audio));
            let has_audio_in_messages = messages.is_some_and(|msgs| {
                msgs.iter().any(|msg| match msg {
                    crate::tasks::generate::Message::UserMultimodal { content } => {
                        content.iter().any(ContentBlock::is_audio)
                    }
                    _ => false,
                })
            });
            if has_audio_in_images || has_audio_in_messages {
                return Err(InferenceError::UnsupportedMode {
                    mode: "audio-content-block",
                    backend: handler.protocol_name(),
                    reason: "this remote protocol has no native audio input path; route to \
                         a provider that implements ProtocolHandler::supports_audio() (Gemini)",
                });
            }
        }

        // Reject a response_format this protocol can't honor rather than
        // silently dropping the constraint (shared guard, mirrors the
        // supports_video/audio pre-checks). The `parslee/*` path rejects ALL
        // response_format separately in `execute_request`.
        if let Some(err) = response_format_rejection(handler.as_ref(), response_format) {
            return Err(err);
        }

        // Auto-truncate if prompt exceeds context window. Media blocks
        // (image/video/audio in `images` or the `messages` history)
        // reserve provider-calibrated token counts, and the multi-turn
        // history's *text* reserves its chars/4 estimate, so a
        // video-carrying or long multi-turn request doesn't over-fill
        // the window with prompt text (I2).
        let prompt = if schema.context_length > 0 {
            let media_tokens =
                crate::media_tokens::request_media_and_history_tokens(images, messages);
            truncate_prompt_to_fit(
                prompt,
                context,
                tools,
                max_tokens,
                media_tokens,
                schema.context_length,
            )
        } else {
            prompt.to_string()
        };

        // Build messages using protocol handler
        let (api_messages, system) = if matches!(schema.source, ModelSource::Proprietary { .. }) {
            // Managed Parslee inference speaks Responses, not Chat
            // Completions. This path is deliberately source-scoped so opaque
            // reasoning items never enter personal OpenRouter requests.
            crate::protocol::OpenAiResponsesHandler.build_messages(
                messages.unwrap_or(&[]),
                &prompt,
                context,
                images,
            )
        } else {
            handler.build_messages(messages.unwrap_or(&[]), &prompt, context, images)
        };

        // Build tools using protocol handler
        let api_tools = tools.map(|t| handler.build_tools(t));

        // Build and execute request
        let req = crate::protocol::ApiRequest {
            model: request_model_name(schema),
            messages: api_messages,
            system,
            system_stable_prefix: system_stable_prefix.map(str::to_string),
            temperature,
            max_tokens,
            tools: api_tools,
            tool_choice: tool_choice.map(str::to_string),
            parallel_tool_calls,
            stream: false,
            budget_tokens,
            cache_control,
            cache_ttl,
            response_format: response_format.cloned(),
        };

        let response = self.execute_request(schema, req).await?;

        let text = response.text;
        let mut calls = response.tool_calls;
        let thinking = response.thinking;
        let provider_output_items = response.provider_output_items;
        let usage = response.usage;
        let stop_reason = response.stop_reason;

        // Cache-effectiveness telemetry: when caching was requested but the
        // provider reported neither a cache read nor a write, the prompt
        // almost certainly fell below the model's minimum cacheable length
        // (e.g. 1024 tokens on Opus 4.8) — the API caches nothing and returns
        // NO error, so this is otherwise invisible. Surface it so a silent
        // no-cache is observable instead of a phantom "caching is on".
        if cache_control {
            if let Some(u) = usage.as_ref() {
                if u.cache_read_input_tokens == 0 && u.cache_creation_input_tokens == 0 {
                    tracing::debug!(
                        model = %request_model_name(schema),
                        prompt_tokens = u.prompt_tokens,
                        "prompt caching requested but nothing was cached (read=0, write=0) — \
                         prompt likely below the model's minimum cacheable length, or the \
                         cached prefix changed between requests",
                    );
                }
            }
        }

        // Fix #10: enrich empty "done" results with text output from the same response.
        // Models often call done({"result": "completed"}) while putting actual findings
        // in the text output block.
        if !text.is_empty() {
            for call in &mut calls {
                if call.name == "done" {
                    let result_val = call
                        .arguments
                        .get("result")
                        .and_then(|v| v.as_str())
                        .unwrap_or("");
                    if result_val.len() < 50 && text.len() > result_val.len() {
                        call.arguments.insert(
                            "result".to_string(),
                            serde_json::Value::String(text.clone()),
                        );
                    }
                }
            }
        }

        Ok((
            text,
            calls,
            thinking,
            provider_output_items,
            usage,
            stop_reason,
        ))
    }

    /// Generate embeddings via a remote API (OpenAI-compatible only for now).
    pub async fn embed(
        &self,
        schema: &ModelSchema,
        texts: &[String],
    ) -> Result<Vec<Vec<f32>>, InferenceError> {
        let (endpoint, protocol) = extract_remote_endpoint(schema)?;
        let lease = self.lease_key(schema, &endpoint).await?;
        let start = std::time::Instant::now();

        let result = match protocol {
            ApiProtocol::OpenAiCompat => {
                self.embed_openai(&endpoint, &lease.api_key, &schema.name, texts)
                    .await
            }
            _ => Err(InferenceError::InferenceFailed(format!(
                "embedding not supported for {:?} protocol",
                protocol
            ))),
        };

        let latency_ms = start.elapsed().as_millis() as u64;
        match &result {
            Ok(_) => {
                let est_tokens = texts
                    .iter()
                    .map(|t| t.split_whitespace().count() as u64)
                    .sum();
                self.key_pool
                    .report_success(&endpoint, &lease.env_var, latency_ms, est_tokens, 0)
                    .await;
            }
            Err(e) => {
                let is_rl =
                    e.to_string().contains("429") || e.to_string().contains("RESOURCE_EXHAUSTED");
                self.key_pool
                    .report_failure(&endpoint, &lease.env_var, is_rl)
                    .await;
            }
        }

        result
    }

    /// Lease a key from the pool, falling back to env var extraction.
    async fn lease_key(
        &self,
        schema: &ModelSchema,
        endpoint: &str,
    ) -> Result<KeyLease, InferenceError> {
        // Try to get the fallback env var name
        let fallback_env = match &schema.source {
            ModelSource::RemoteApi {
                protocol: ApiProtocol::OpenRouter,
                ..
            } => {
                let (api_key, source) = crate::openrouter::resolve_credential().ok_or_else(|| {
                    InferenceError::InferenceFailed(
                        "OpenRouter requires a key — run `car keys set openrouter` or connect your OpenRouter account in CarHost"
                            .to_string(),
                    )
                })?;
                return Ok(KeyLease {
                    api_key,
                    env_var: format!("{}:{}", crate::openrouter::API_KEY_ENV, source.as_str()),
                });
            }
            ModelSource::RemoteApi { api_key_env, .. } => api_key_env.as_str(),
            ModelSource::Ollama { .. } | ModelSource::VllmMlx { .. } => {
                return Ok(KeyLease {
                    api_key: String::new(),
                    env_var: String::new(),
                })
            }
            // Proprietary providers (Parslee) carry a single bearer
            // credential, not a pool of rotatable env keys, so resolve
            // it directly and return — no key_pool lease/round-robin.
            ModelSource::Proprietary {
                ref auth,
                ref provider,
                ..
            } => {
                let (token, source_label) = match auth {
                    // OAuth2 PKCE: `car auth login` publishes the
                    // durable token inside car-auth's V2 credential transaction;
                    // the same-named environment variable remains the explicit
                    // override.
                    ProprietaryAuth::OAuth2Pkce { .. } => {
                        if !provider.eq_ignore_ascii_case("parslee")
                            || !schema.provider.eq_ignore_ascii_case("parslee")
                        {
                            return Err(InferenceError::InferenceFailed(format!(
                                "OAuth2 PKCE credentials are supported only for provider \
                                 'parslee'; refusing model {} from proprietary provider \
                                 '{provider}'",
                                schema.id
                            )));
                        }
                        // Say something BEFORE the deadline rather than
                        // diagnosing it afterwards (#797 item 4). Throttled and
                        // warn-once internally, so this costs nothing per
                        // request and does not repeat through a long batch.
                        warn_if_token_expires_soon().await;
                        (
                            // Proactively refresh a lapsed token instead of letting
                            // the request 401 (which poisons 30-day model health,
                            // #313). The `PARSLEE_ACCESS_TOKEN` env override still
                            // wins and is never refreshed; without a stored expiry
                            // this degrades to the prior env-first/keychain read.
                            car_auth::access_token_refreshing().await,
                            PARSLEE_ACCESS_TOKEN_ENV,
                        )
                    }
                    ProprietaryAuth::BearerTokenEnv { env_var }
                    | ProprietaryAuth::ApiKeyEnv { env_var } => (
                        car_secrets::resolve_env_or_keychain(env_var),
                        env_var.as_str(),
                    ),
                };
                let token = match token {
                    Some(t) => t,
                    None => {
                        // Say WHICH failure this is. The old text — "no
                        // credential … run `car auth login {provider}`" — read
                        // as never-authenticated even when the real state was a
                        // token that aged out mid-run, and it suggested a
                        // command that does not parse: `car auth login` takes no
                        // positional provider (Parslee-ai/car#797).
                        // Classification and prose are produced together so they
                        // cannot disagree: `reason` is what consumers branch on,
                        // `detail` is what a human reads, and both come from the
                        // same match arm (#797 item 2).
                        let (reason, detail) = if provider.eq_ignore_ascii_case("parslee") {
                            match car_auth::credential_state().await {
                                car_auth::CredentialState::Expired { expires_at } => (
                                    crate::CredentialFailure::Expired { expires_at },
                                    format!(
                                        "the Parslee token expired at unix {expires_at} and could \
                                         not be refreshed. Re-authenticate with `car auth login`; \
                                         a job longer than the remaining token lifetime will fail \
                                         partway unless it is resumable"
                                    ),
                                ),
                                car_auth::CredentialState::Unreadable(e) => (
                                    crate::CredentialFailure::StoreUnreadable,
                                    format!(
                                        "the credential store could not be read ({e}). This is \
                                         not a sign-out — on macOS it usually means a keychain \
                                         prompt is waiting, or the helper timed out. Unlock the \
                                         login keychain and retry before re-authenticating"
                                    ),
                                ),
                                car_auth::CredentialState::SignedOut => (
                                    crate::CredentialFailure::SignedOut,
                                    "no account is signed in. Run `car auth login`".to_string(),
                                ),
                                // Active but the token read still failed: a race
                                // between the two reads. Say so rather than
                                // assert something we no longer know.
                                car_auth::CredentialState::Active => (
                                    crate::CredentialFailure::RaceRetryable,
                                    "the credential read failed but the store now reports an \
                                     active session — retry the request"
                                        .to_string(),
                                ),
                            }
                        } else {
                            (
                                crate::CredentialFailure::EnvVarMissing {
                                    env_var: source_label.to_string(),
                                },
                                format!("set ${source_label} for provider '{provider}'"),
                            )
                        };
                        // The variant's Display KEEPS the historical prefix
                        // verbatim — see `InferenceError::CredentialUnavailable`
                        // for why that is load bearing rather than cosmetic.
                        return Err(InferenceError::CredentialUnavailable {
                            provider: provider.to_string(),
                            model: schema.id.clone(),
                            reason,
                            detail,
                        });
                    }
                };
                return Ok(KeyLease {
                    api_key: token,
                    env_var: String::new(),
                });
            }
            _ => {
                return Err(InferenceError::InferenceFailed(format!(
                    "model {} is not remote",
                    schema.id
                )))
            }
        };

        // Register keys on first use (idempotent)
        self.register_model_keys(schema).await;

        self.key_pool
            .lease_or_env(endpoint, fallback_env)
            .await
            .ok_or_else(|| {
                InferenceError::InferenceFailed(format!(
                    "no API keys available for endpoint {} (checked env vars: {:?})",
                    endpoint,
                    schema.all_api_key_envs()
                ))
            })
    }

    // --- Embedding (not yet migrated to ProtocolHandler) ---

    async fn embed_openai(
        &self,
        endpoint: &str,
        api_key: &str,
        model: &str,
        texts: &[String],
    ) -> Result<Vec<Vec<f32>>, InferenceError> {
        let url = format_endpoint(endpoint, "/v1/embeddings");

        let body = serde_json::json!({
            "model": model,
            "input": texts,
        });

        let resp = self
            .client
            .post(&url)
            .header("Authorization", format!("Bearer {api_key}"))
            .header("Content-Type", "application/json")
            .json(&body)
            .send()
            .await
            .map_err(|e| self.request_error("HTTP error", &e))?;

        let status = resp.status();
        let text = resp.text().await.map_err(|e| {
            InferenceError::InferenceFailed(format!("read body: {}", error_chain(&e)))
        })?;

        if !status.is_success() {
            return Err(InferenceError::InferenceFailed(format!(
                "API returned {status}: {text}"
            )));
        }

        let parsed: OpenAiEmbedResponse = serde_json::from_str(&text)
            .map_err(|e| InferenceError::InferenceFailed(format!("parse response: {e}")))?;

        Ok(order_embeddings(parsed.data))
    }

    /// Stream a response from a remote API using Server-Sent Events.
    /// Returns a channel receiver that yields StreamEvents.
    /// Works with OpenAI-compatible and Anthropic APIs.
    pub async fn generate_stream(
        &self,
        schema: &ModelSchema,
        prompt: &str,
        messages: Option<&[crate::tasks::generate::Message]>,
        context: Option<&str>,
        temperature: f64,
        max_tokens: usize,
        tools: Option<&[serde_json::Value]>,
        images: Option<&[ContentBlock]>,
        tool_choice: Option<&str>,
        parallel_tool_calls: Option<bool>,
        response_format: Option<&crate::tasks::generate::ResponseFormat>,
        spend_guard: Option<crate::routing_ext::MidStreamSpendGuard>,
    ) -> Result<tokio::sync::mpsc::Receiver<crate::stream::StreamEvent>, InferenceError> {
        let (endpoint, protocol) = extract_remote_endpoint(schema)?;
        let handler = crate::protocol::handler_for(protocol);

        // Mirror the non-streaming audio/video guard so streaming
        // callers can't bypass the supports_* check and silently
        // stringify audio/video blocks via the handler fallback arms.
        if !handler.supports_video()
            && (images.is_some_and(|blocks| blocks.iter().any(ContentBlock::is_video))
                || messages.is_some_and(|messages| {
                    messages.iter().any(|message| match message {
                        crate::tasks::generate::Message::UserMultimodal { content } => {
                            content.iter().any(ContentBlock::is_video)
                        }
                        _ => false,
                    })
                }))
        {
            return Err(InferenceError::UnsupportedMode {
                mode: "video-content-block",
                backend: handler.protocol_name(),
                reason: "this remote protocol has no native video input path; route to \
                     a provider that implements ProtocolHandler::supports_video() (Gemini)",
            });
        }
        if !handler.supports_audio()
            && (images.is_some_and(|blocks| blocks.iter().any(ContentBlock::is_audio))
                || messages.is_some_and(|messages| {
                    messages.iter().any(|message| match message {
                        crate::tasks::generate::Message::UserMultimodal { content } => {
                            content.iter().any(ContentBlock::is_audio)
                        }
                        _ => false,
                    })
                }))
        {
            return Err(InferenceError::UnsupportedMode {
                mode: "audio-content-block",
                backend: handler.protocol_name(),
                reason: "this remote protocol has no native audio input path; route to \
                     a provider that implements ProtocolHandler::supports_audio() (Gemini)",
            });
        }

        // Mirror the non-streaming response_format guard (same shared helper) so
        // a streaming caller can't bypass it and silently run unconstrained.
        if let Some(err) = response_format_rejection(handler.as_ref(), response_format) {
            return Err(err);
        }

        // Streaming path now flows through the same protocol abstraction
        // the non-streaming path uses (#125): build an `ApiRequest`,
        // delegate body construction to `handler.build_request_body`,
        // delegate auth/content-type to `handler.auth_headers`. Anything
        // the protocol abstraction picks up — `response_format`,
        // `tool_choice`, `parallel_tool_calls`, future fields — flows
        // through to streaming for free without re-implementing per
        // provider.
        // All deterministic request-capability checks above must win over
        // credential lookup, matching the non-streaming seam. Otherwise an
        // unsupported request with a missing key reports configuration drift
        // instead of the stable UnsupportedMode callers can handle.
        let lease = self.lease_key(schema, &endpoint).await?;
        let api_key = lease.api_key;
        let model = request_model_name(schema);

        let (messages, system) = if matches!(schema.source, ModelSource::Proprietary { .. }) {
            crate::protocol::OpenAiResponsesHandler.build_messages(
                messages.unwrap_or(&[]),
                prompt,
                context,
                images,
            )
        } else {
            handler.build_messages(messages.unwrap_or(&[]), prompt, context, images)
        };
        let built_tools = tools.map(|t| handler.build_tools(t));
        let req = crate::protocol::ApiRequest {
            model: model.clone(),
            messages,
            system,
            // Raw-stream path does not assemble a memory context, so there is
            // no stable prefix to cache.
            system_stable_prefix: None,
            temperature,
            max_tokens,
            tools: built_tools,
            tool_choice: tool_choice.map(str::to_string),
            parallel_tool_calls,
            stream: true,
            budget_tokens: 0,
            cache_control: false,
            // Ignored while `cache_control` is false on the raw-stream path.
            cache_ttl: crate::tasks::generate::CacheTtl::default(),
            response_format: response_format.cloned(),
        };
        if matches!(schema.source, ModelSource::Proprietary { .. }) {
            return self
                .parslee_assistant_stream_request(&endpoint, &api_key, &req, spend_guard)
                .await;
        }
        let body = handler.build_request_body(&req);

        // Build URL — Google/Vertex stream via :streamGenerateContent?alt=sse,
        // Azure has its deployment shape, everyone else uses the handler path.
        let url = if matches!(protocol, ApiProtocol::Google) {
            crate::protocol::google_stream_url(&endpoint, &model, &api_key)
        } else if matches!(protocol, ApiProtocol::VertexAi) {
            crate::protocol::vertex_stream_url(&endpoint, &model)
        } else if matches!(protocol, ApiProtocol::AzureOpenAi) {
            let api_version = match &schema.source {
                ModelSource::RemoteApi { api_version, .. } => api_version.clone(),
                _ => None,
            };
            let version = api_version.as_deref().unwrap_or("2024-10-21");
            format!(
                "{}/openai/deployments/{}/chat/completions?api-version={}",
                endpoint.trim_end_matches('/'),
                model,
                version
            )
        } else {
            format_endpoint(&endpoint, chat_path_for(schema, handler.as_ref()))
        };

        let mut headers = reqwest::header::HeaderMap::new();
        for (name, value) in handler.auth_headers(&api_key) {
            headers.insert(
                reqwest::header::HeaderName::from_bytes(name.as_bytes()).map_err(|e| {
                    InferenceError::InferenceFailed(format!("auth header name: {e}"))
                })?,
                value.parse().map_err(|e| {
                    InferenceError::InferenceFailed(format!("auth header value: {e}"))
                })?,
            );
        }

        let send_fut = self.client.post(&url).headers(headers).json(&body).send();
        let resp = tokio::time::timeout(std::time::Duration::from_secs(300), send_fut)
            .await
            .map_err(|_| {
                InferenceError::InferenceFailed(
                    "stream request timed out after 300s (tokio safety timeout)".to_string(),
                )
            })?
            .map_err(|e| self.request_error("HTTP error", &e))?;

        let status = resp.status();
        if !status.is_success() {
            let err_text = resp.text().await.unwrap_or_default();
            // Same account-vs-model split as the non-streaming path: a stream
            // request refused with 401/402/403 never opened, and the reason is
            // the account, not the model (Parslee-ai/car#650).
            if is_account_http_status(status.as_u16()) {
                return Err(InferenceError::ProviderAccount {
                    provider: schema.provider.clone(),
                    status: status.as_u16(),
                    message: account_error_message(protocol, status.as_u16()),
                });
            }
            if matches!(protocol, ApiProtocol::OpenRouter) {
                return Err(openrouter_http_error(status, &err_text, &model));
            }
            return Err(InferenceError::InferenceFailed(format!(
                "API returned {status}: {err_text}"
            )));
        }

        let (tx, rx) = tokio::sync::mpsc::channel::<crate::stream::StreamEvent>(64);

        // Spawn a task to read the SSE stream and forward events. Parsing is
        // delegated to the protocol handler (`parse_stream_event`) so each
        // provider's SSE shape is decoded by its own handler — OpenAI chat,
        // Anthropic, and the Responses API all differ. (Previously this
        // hardcoded an is_anthropic-vs-OpenAI branch, which silently mis-parsed
        // any third shape, e.g. the Responses API's typed `event:` stream.)
        tokio::spawn(async move {
            use futures::StreamExt;
            let mut byte_stream = resp.bytes_stream();
            // Buffer raw BYTES, not a per-chunk lossily-decoded String: a
            // multi-byte UTF-8 codepoint split across two TCP chunks would
            // otherwise become a `�`, corrupting tool-argument JSON or text.
            let mut buffer: Vec<u8> = Vec::new();
            // Mid-stream spend guard (I4): pre-call SpendControl::check
            // can't see a runaway long output. Each text/tool-arg delta
            // feeds the guard; a trip emits a terminal StopReason and
            // drops the byte stream, aborting the upstream HTTP body.
            let mut spend_guard = spend_guard;
            let mut saw_positive_completion = false;
            let mut saw_chat_finish_reason = false;
            let mut saw_google_finish_reason = false;

            while let Some(chunk_result) = byte_stream.next().await {
                match chunk_result {
                    Ok(bytes) => buffer.extend_from_slice(&bytes),
                    Err(e) => {
                        // A transport break mid-stream is NOT a clean EOF.
                        // Previously this `break`'d silently, so the partial
                        // text looked like a finished answer. Log it and emit
                        // an abnormal-termination signal so the receiver can
                        // tell truncation from completion.
                        tracing::warn!(error = %e, "remote stream transport error mid-response");
                        let _ = tx
                            .send(crate::stream::StreamEvent::Error(
                                "remote stream transport error".to_string(),
                            ))
                            .await;
                        return;
                    }
                };

                // Process complete SSE events. Each event ends at a blank
                // line ("\n\n" or CRLF "\r\n\r\n"). The block always ends on
                // an ASCII boundary, so decoding it as UTF-8 is exact; any
                // trailing partial codepoint stays in `buffer` for next chunk.
                while let Some((pos, sep_len)) = find_sse_separator(&buffer) {
                    let event_block = String::from_utf8_lossy(&buffer[..pos]).into_owned();
                    buffer.drain(..pos + sep_len);

                    let sse_events = crate::stream::parse_sse_lines(&event_block);
                    for (event_type, data) in sse_events {
                        if data == "[DONE]" {
                            if matches!(
                                protocol,
                                ApiProtocol::OpenRouter
                                    | ApiProtocol::OpenAiCompat
                                    | ApiProtocol::AzureOpenAi
                            ) && saw_chat_finish_reason
                                && !saw_positive_completion
                            {
                                saw_positive_completion = true;
                                if tx
                                    .send(crate::stream::StreamEvent::Done {
                                        text: String::new(),
                                        tool_calls: Vec::new(),
                                    })
                                    .await
                                    .is_err()
                                {
                                    return;
                                }
                            }
                            continue;
                        }

                        let stream_events = handler.parse_stream_event(&event_type, &data);

                        for evt in stream_events {
                            if let Some(guard) = spend_guard.as_mut() {
                                // Provider-reported input tokens (Anthropic
                                // sends them at stream START) replace the
                                // estimated prompt half with ground truth.
                                if let crate::stream::StreamEvent::Usage { input_tokens, .. } = &evt
                                {
                                    guard.note_prompt_tokens(*input_tokens);
                                }
                                let delta_len = match &evt {
                                    crate::stream::StreamEvent::TextDelta(t) => Some(t.as_str()),
                                    crate::stream::StreamEvent::ToolCallDelta {
                                        arguments_delta,
                                        ..
                                    } => Some(arguments_delta.as_str()),
                                    _ => None,
                                };
                                if let Some(delta) = delta_len {
                                    let n =
                                        crate::routing_ext::MidStreamSpendGuard::estimate_tokens(
                                            delta,
                                        );
                                    if let Some(exceeded) = guard.record_tokens(n) {
                                        tracing::warn!(
                                            limit_usd = exceeded.limit_usd,
                                            current_usd = exceeded.current_usd,
                                            "mid-stream spend limit tripped; cancelling stream"
                                        );
                                        // Deliver the delta that crossed the
                                        // line, then terminate abnormally so
                                        // the receiver can tell truncation
                                        // from completion.
                                        let _ = tx.send(evt).await;
                                        let _ = tx
                                            .send(crate::stream::StreamEvent::StopReason(format!(
                                                "spend_limit: {exceeded}"
                                            )))
                                            .await;
                                        let _ = tx
                                            .send(crate::stream::StreamEvent::Error(
                                                "stream cancelled because the spend limit was reached"
                                                    .to_string(),
                                            ))
                                            .await;
                                        return; // drops byte_stream → aborts HTTP body
                                    }
                                }
                            }
                            if let crate::stream::StreamEvent::StopReason(reason) = &evt {
                                if matches!(protocol, ApiProtocol::Google | ApiProtocol::VertexAi) {
                                    match google_finish_outcome(reason) {
                                        Ok(()) => saw_google_finish_reason = true,
                                        Err(message) => {
                                            let _ = tx
                                                .send(crate::stream::StreamEvent::Error(
                                                    message.to_string(),
                                                ))
                                                .await;
                                            return;
                                        }
                                    }
                                } else if matches!(
                                    protocol,
                                    ApiProtocol::OpenRouter
                                        | ApiProtocol::OpenAiCompat
                                        | ApiProtocol::AzureOpenAi
                                ) {
                                    saw_chat_finish_reason = true;
                                }
                            }
                            if matches!(evt, crate::stream::StreamEvent::Done { .. }) {
                                saw_positive_completion = true;
                            }
                            let terminal_error =
                                matches!(&evt, crate::stream::StreamEvent::Error(_));
                            if tx.send(evt).await.is_err() {
                                return; // receiver dropped
                            }
                            if terminal_error {
                                return; // provider error frames are terminal
                            }
                        }

                        // Google AI Studio and Vertex share Gemini's typed
                        // `finishReason` contract. STOP/MAX_TOKENS are the
                        // only deliberate terminal reasons; abnormal reasons
                        // returned above as sanitized Error. Neither transport
                        // has a trailing OpenAI-style [DONE] sentinel.
                        if matches!(protocol, ApiProtocol::Google | ApiProtocol::VertexAi)
                            && saw_google_finish_reason
                            && !saw_positive_completion
                        {
                            saw_positive_completion = true;
                            if tx
                                .send(crate::stream::StreamEvent::Done {
                                    text: String::new(),
                                    tool_calls: Vec::new(),
                                })
                                .await
                                .is_err()
                            {
                                return;
                            }
                        }

                        // Anthropic's typed `message_stop` is its positive
                        // terminal proof. It has no OpenAI-style [DONE]
                        // sentinel, so surface the same internal Done marker.
                        if protocol == ApiProtocol::Anthropic && event_type == "message_stop" {
                            saw_positive_completion = true;
                            if tx
                                .send(crate::stream::StreamEvent::Done {
                                    text: String::new(),
                                    tool_calls: Vec::new(),
                                })
                                .await
                                .is_err()
                            {
                                return;
                            }
                        }
                    }
                }
            }

            if !buffer.is_empty() || !saw_positive_completion {
                let _ = tx
                    .send(crate::stream::StreamEvent::Error(
                        "remote inference stream ended before provider completion".to_string(),
                    ))
                    .await;
            }
        });

        Ok(rx)
    }
}

impl Default for RemoteBackend {
    fn default() -> Self {
        Self::new()
    }
}

/// Whether a transport error is the class a missing trust root produces.
///
/// Split out as a predicate over the two flags so the "don't dress other
/// failures up as certificate problems" rule is directly testable: auth and
/// rate-limit failures are HTTP statuses and never reach here at all, but
/// timeouts do — and a timeout is not a trust problem.
pub(crate) fn is_trust_related(is_connect: bool, is_timeout: bool) -> bool {
    is_connect && !is_timeout
}

// --- Helpers ---

/// Extract endpoint and protocol from a model schema (key comes from KeyPool now).
fn extract_remote_endpoint(schema: &ModelSchema) -> Result<(String, ApiProtocol), InferenceError> {
    match &schema.source {
        ModelSource::RemoteApi {
            endpoint, protocol, ..
        } => {
            let endpoint = if matches!(protocol, ApiProtocol::OpenRouter)
                && std::env::var("CAR_OPENROUTER_TEST_MODE").as_deref() == Ok("1")
            {
                std::env::var("OPENROUTER_API_BASE").unwrap_or_else(|_| endpoint.clone())
            } else {
                endpoint.clone()
            };
            Ok((endpoint, *protocol))
        }
        ModelSource::Ollama { host, .. } => Ok((host.clone(), ApiProtocol::OpenAiCompat)),
        ModelSource::VllmMlx { endpoint, .. } => Ok((endpoint.clone(), ApiProtocol::OpenAiCompat)),
        // Proprietary providers (Parslee) speak an OpenAI-compatible
        // wire shape; the base URL is the provider endpoint and the
        // (non-default) chat path is applied via `chat_path_for`.
        ModelSource::Proprietary {
            provider,
            endpoint,
            auth,
            ..
        } => Ok((
            if provider.eq_ignore_ascii_case("parslee")
                && matches!(auth, ProprietaryAuth::OAuth2Pkce { .. })
            {
                // Resolve at request time, not only when the registry snapshot
                // was constructed. Account switching persists this base beside
                // the token; inference must use the same authority immediately.
                car_auth::api_base(None)
            } else {
                endpoint.clone()
            },
            ApiProtocol::OpenAiCompat,
        )),
        _ => Err(InferenceError::InferenceFailed(format!(
            "model {} is not remote",
            schema.id
        ))),
    }
}

/// Reject a `response_format` the selected protocol can't honor, rather than
/// silently dropping it. `build_request_body` returns a `Value` and cannot fail,
/// so this decision lives at the remote seam (same pattern as the
/// `supports_video`/`supports_audio` guards). Shared by the non-streaming
/// (`generate_with_tools_multi`) and streaming (`generate_stream`) entry points
/// so ONE guard governs both and it is unit-testable without a live endpoint.
///
/// Returns `Some(UnsupportedMode)` when the handler can't honor `rf`, else
/// `None`. Anthropic rejects both variants because CAR has no provider-enforced
/// response-format path under its pinned API version; OpenAI/Google honor both
/// natively.
fn response_format_rejection(
    handler: &dyn crate::protocol::ProtocolHandler,
    response_format: Option<&crate::tasks::generate::ResponseFormat>,
) -> Option<InferenceError> {
    match response_format {
        Some(rf) if !handler.supports_response_format(rf) => {
            Some(InferenceError::UnsupportedMode {
                mode: match rf {
                    crate::tasks::generate::ResponseFormat::JsonSchema { .. } => {
                        "structured-output-json-schema"
                    }
                    crate::tasks::generate::ResponseFormat::JsonObject => "structured-output-json",
                },
                backend: handler.protocol_name(),
                reason: "this provider is not wired for a provider-enforced \
                     response_format under CAR's pinned API version; supply a \
                     tool whose input schema is your schema plus a forcing \
                     tool_choice",
            })
        }
        _ => None,
    }
}

/// Chat path to append to the base endpoint. Proprietary providers
/// (Parslee) declare their own `chat_path` in the schema; every other
/// source uses the protocol handler's default `endpoint_path()`.
fn chat_path_for<'a>(
    schema: &'a ModelSchema,
    handler: &'a dyn crate::protocol::ProtocolHandler,
) -> &'a str {
    match &schema.source {
        ModelSource::Proprietary { protocol, .. } => protocol.chat_path.as_str(),
        _ => handler.endpoint_path(),
    }
}

fn request_model_name(schema: &ModelSchema) -> String {
    if let Some(canonical_id) = crate::openrouter::canonical_managed_gateway_selector(schema) {
        return canonical_id.to_string();
    }
    match &schema.source {
        ModelSource::VllmMlx { model_name, .. } => model_name.clone(),
        _ => schema.name.clone(),
    }
}

/// Normalize endpoint URL for a given path.
fn format_endpoint(base: &str, path: &str) -> String {
    let base = base.trim_end_matches('/');
    // If the base already ends with the path, use it as-is
    if base.ends_with(path.trim_start_matches('/')) {
        base.to_string()
    } else {
        format!("{}{}", base, path)
    }
}

// --- Response types ---

#[derive(Debug, Deserialize)]
struct OpenAiEmbedResponse {
    data: Vec<OpenAiEmbedData>,
}

#[derive(Debug, Deserialize)]
struct OpenAiEmbedData {
    embedding: Vec<f32>,
    /// Position in the input batch. The OpenAI embeddings API does **not**
    /// guarantee `data[]` is returned in input order — `index` is authoritative.
    #[serde(default)]
    index: usize,
}

/// Collect embeddings in input order. The API may return `data[]` out of order,
/// so sort by `index` first — a caller that pairs `embeddings[i]` with input
/// text `i` (every similarity ranking does) would otherwise silently mis-pair.
fn order_embeddings(mut data: Vec<OpenAiEmbedData>) -> Vec<Vec<f32>> {
    data.sort_by_key(|d| d.index);
    data.into_iter().map(|d| d.embedding).collect()
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::schema::ProprietaryProtocol;
    use std::time::Duration;

    /// A client that never reads the machine's trust store.
    ///
    /// The three error-classification tests below only need *a* client — two
    /// talk plain HTTP to a dead loopback port and one fails while building a
    /// header, so none of them does TLS. `Client::new()` is
    /// `build().expect(..)`, which reads the native roots and panics when they
    /// don't parse, so using it here made those tests fail whenever a
    /// `TrustStoreScope` elsewhere in the suite had `SSL_CERT_FILE` pointed at
    /// its unparseable fixture. Dropping the built-in roots removes that
    /// coupling at no cost to what these tests assert.
    fn trust_store_independent_client() -> reqwest::Client {
        reqwest::Client::builder()
            .tls_built_in_root_certs(false)
            .build()
            .expect("a client with no built-in roots has no certificate work to fail at")
    }

    /// A builder-kind `reqwest::Error`, produced the way production produced
    /// it: an API key carrying a trailing newline, which `http` rejects as a
    /// header value. The request never leaves the process.
    fn newline_in_api_key_error() -> reqwest::Error {
        trust_store_independent_client()
            .post("https://example.invalid/v1/chat/completions")
            .header("authorization", "Bearer sk-test-key\n")
            .build()
            .expect_err("a header value containing a newline must fail to build")
    }

    /// The defect that hid the root cause: `{e}` renders only reqwest's
    /// top-level `Display`, which for a builder error is the bare string
    /// `builder error`. The reason lives in `source()`.
    #[test]
    fn builder_error_surfaces_its_underlying_cause() {
        let err = newline_in_api_key_error();

        // Falsifier: this is exactly what the old formatting produced, and it
        // is why weeks of logs said nothing useful.
        let old_rendering = format!("{err}");
        assert!(
            !old_rendering.contains("failed to parse header value"),
            "reqwest's Display is expected to omit the cause; if this now \
             includes it, the whole premise of error_chain has changed. \
             got: {old_rendering}"
        );

        let chain = error_chain(&err);
        assert!(
            chain.contains("failed to parse header value"),
            "the source chain must name the real cause, got: {chain}"
        );
    }

    /// A builder-kind error can never succeed on retry — the bytes never
    /// reached the network. Classifying it transient cost ~9s of backoff per
    /// call and mislabelled a deterministic config bug as an infra blip.
    #[test]
    fn builder_error_is_classified_permanent() {
        let err = newline_in_api_key_error();
        assert!(
            err.is_builder(),
            "precondition: this is a builder-kind error"
        );
        assert!(
            !reqwest_error_is_transient(&err),
            "a request that was never constructed must not be retried"
        );

        let attempt = TransportAttemptError::from_reqwest("HTTP error", &err);
        assert!(!attempt.retryable);
        // The message the operator actually sees carries the diagnosis.
        assert!(
            attempt
                .error
                .to_string()
                .contains("failed to parse header value"),
            "got: {}",
            attempt.error
        );
    }

    /// The other half of the classification: a real transport failure must
    /// still be retried. Loopback port 1 never listens, so this is
    /// deterministic and needs no network.
    #[tokio::test]
    async fn genuine_transport_error_is_classified_transient() {
        let err = trust_store_independent_client()
            .get("http://127.0.0.1:1/v1/chat/completions")
            .send()
            .await
            .expect_err("nothing listens on loopback port 1");

        assert!(
            !err.is_builder(),
            "precondition: this reached the transport"
        );
        assert!(
            reqwest_error_is_transient(&err),
            "a connect failure is worth the retry budget, got: {err}"
        );
        assert!(TransportAttemptError::from_reqwest("HTTP error", &err).retryable);
    }

    /// A transport error retains its request URL, so credentials placed in URL
    /// userinfo must not survive the operator-facing source-chain rendering.
    #[tokio::test]
    async fn transport_error_redacts_url_userinfo() {
        let mut err = trust_store_independent_client()
            .get("http://127.0.0.1:1/v1/chat/completions")
            .send()
            .await
            .expect_err("nothing listens on loopback port 1");

        let url = err
            .url_mut()
            .expect("transport errors retain the request URL");
        url.set_username("leaked-user")
            .expect("the test username is valid URL userinfo");
        url.set_password(Some("leaked-password"))
            .expect("the test password is valid URL userinfo");
        assert_eq!(url.username(), "leaked-user");
        assert_eq!(url.password(), Some("leaked-password"));

        let chain = error_chain(&err);
        assert!(
            !chain.contains("leaked-user"),
            "the username must not survive redaction, got: {chain}"
        );
        assert!(
            !chain.contains("leaked-password"),
            "the password must not survive redaction, got: {chain}"
        );
    }

    /// A tokio deadline elapsing says nothing about whether the request is
    /// well-formed, so it stays retryable.
    #[test]
    fn timeout_attempt_is_retryable() {
        assert!(TransportAttemptError::timeout("request timed out after 300s").retryable);
    }

    /// Google AI Studio authenticates with `?key=<api key>` and reqwest's
    /// `Display` appends the request URL — so surfacing the chain must not
    /// write the credential into the log that reports the failure.
    #[test]
    fn error_chain_redacts_credentials_in_urls() {
        let raw = "error sending request for url \
                   (https://generativelanguage.googleapis.com/v1beta/models/x:generateContent\
                   ?alt=sse&key=AIzaSyREAL_SECRET_VALUE&pretty=1)";
        let scrubbed = redact_sensitive(raw);

        assert!(
            !scrubbed.contains("AIzaSyREAL_SECRET_VALUE"),
            "the API key must not survive redaction, got: {scrubbed}"
        );
        assert!(scrubbed.contains("key=REDACTED"), "got: {scrubbed}");
        // Redaction stops at the delimiter — neighbouring params survive.
        assert!(scrubbed.contains("alt=sse"), "got: {scrubbed}");
        assert!(scrubbed.contains("pretty=1"), "got: {scrubbed}");
        // A parameter that merely ends in a sensitive name is not a match.
        assert_eq!(redact_sensitive("?monkey=banana"), "?monkey=banana");
    }

    #[derive(Debug)]
    struct Link {
        message: String,
        source: Option<Box<Link>>,
    }

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

    impl std::error::Error for Link {
        fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
            self.source
                .as_deref()
                .map(|s| s as &(dyn std::error::Error + 'static))
        }
    }

    /// Build a chain of `depth` links whose messages are all distinct, so the
    /// "restates its parent" de-duplication doesn't collapse them.
    fn distinct_chain(depth: usize, width: usize) -> Link {
        let mut err = Link {
            message: format!("cause-{depth}-{}", "z".repeat(width)),
            source: None,
        };
        for i in (0..depth).rev() {
            err = Link {
                message: format!("cause-{i}-{}", "y".repeat(width)),
                source: Some(Box::new(err)),
            };
        }
        err
    }

    /// An arbitrarily long upstream error can't turn one failed request into an
    /// unbounded log line.
    #[test]
    fn error_chain_is_length_bounded() {
        let chain = error_chain(&distinct_chain(40, 400));
        assert!(
            chain.len() <= MAX_ERROR_CHAIN_LEN + "... (truncated)".len(),
            "chain must stay bounded, got {} bytes",
            chain.len()
        );
        assert!(chain.ends_with("... (truncated)"), "got: {chain}");
    }

    /// A deep chain of short links stops at the depth cap rather than walking
    /// forever.
    #[test]
    fn error_chain_is_depth_bounded() {
        let chain = error_chain(&distinct_chain(40, 0));
        assert!(chain.ends_with(": ..."), "got: {chain}");
        assert!(
            chain.contains(&format!("cause-{}", MAX_ERROR_CHAIN_DEPTH - 1)),
            "the last link inside the cap must be rendered, got: {chain}"
        );
        assert!(
            !chain.contains(&format!("cause-{}", MAX_ERROR_CHAIN_DEPTH + 1)),
            "links past the cap must be dropped, got: {chain}"
        );
    }

    /// A wrapper whose `Display` already embeds its source — the thiserror
    /// `#[error("...: {0}")]` shape used throughout this crate — must not be
    /// rendered as `X: X`.
    #[test]
    fn error_chain_does_not_repeat_a_restated_cause() {
        let err = Link {
            message: "inference failed: connection reset".to_string(),
            source: Some(Box::new(Link {
                message: "connection reset".to_string(),
                source: None,
            })),
        };
        assert_eq!(error_chain(&err), "inference failed: connection reset");
    }

    fn openrouter_schema(endpoint: &str) -> ModelSchema {
        let mut schema = crate::openrouter::curated_schemas()
            .into_iter()
            .find(|schema| schema.id == "openrouter/deepseek/deepseek-v3.2")
            .expect("curated personal OpenRouter schema");
        let ModelSource::RemoteApi {
            endpoint: schema_endpoint,
            ..
        } = &mut schema.source
        else {
            panic!("personal OpenRouter schema must be remote");
        };
        *schema_endpoint = endpoint.to_string();
        schema
    }

    fn remote_stream_schema(
        endpoint: &str,
        protocol: ApiProtocol,
        model: &str,
        api_key_env: &str,
    ) -> ModelSchema {
        ModelSchema {
            id: format!("test/{model}"),
            name: model.to_string(),
            provider: "test".into(),
            family: "test".into(),
            version: "1".into(),
            capabilities: vec![crate::schema::ModelCapability::Generate],
            context_length: 128_000,
            max_output_tokens: Some(8_192),
            param_count: String::new(),
            quantization: None,
            performance: Default::default(),
            cost: Default::default(),
            source: ModelSource::RemoteApi {
                endpoint: endpoint.to_string(),
                api_key_env: api_key_env.to_string(),
                api_key_envs: vec![],
                api_version: None,
                protocol,
            },
            tags: vec!["test".into()],
            supported_params: vec![],
            public_benchmarks: vec![],
            trust_tier: crate::schema::TrustTier::Community,
            deprecated: false,
            available: true,
            weights_ready: true,
        }
    }

    /// #796 — a refusal must be legible as a refusal.
    ///
    /// Round-trips through the writer in `stream.rs`, so the reader cannot
    /// drift from the format that produces it — the two live together for
    /// exactly this reason.
    #[test]
    fn a_content_refusal_is_classified_from_the_tags_the_gateway_sent() {
        let events = crate::stream::parse_openai_responses_sse_line(
            "error",
            r#"{"error":{"message":"content refused","type":"invalid_request_error","code":"content_policy_violation"}}"#,
        );
        let crate::stream::StreamEvent::Error(detail) = events.first().expect("an error event")
        else {
            panic!("expected an Error event");
        };
        let (kind, code) = content_refusal_tags(detail).expect("must classify as a refusal");
        assert_eq!(code.as_deref(), Some("content_policy_violation"));
        assert_eq!(kind.as_deref(), Some("invalid_request_error"));
    }

    /// The load-bearing guard. Over-classification is the DANGEROUS direction:
    /// a real inference failure mislabelled as a refusal is excluded from model
    /// health and silently stops being retried, and a crash that looks like a
    /// policy decision is far harder to find than the reverse.
    #[test]
    fn ordinary_failures_are_not_mistaken_for_refusals() {
        for detail in [
            // The #796 symptom itself, with no tags at all — CAR's fallback for
            // an error object carrying no message. Must NOT classify: we do not
            // know it was a refusal, and guessing would be the whole bug again.
            "managed inference failed",
            // A genuine crash, tagged.
            "boom (type=server_error, code=internal_error)",
            // Rate limiting is transient and SHOULD keep being retried.
            "slow down (type=rate_limit_error, code=rate_limit_exceeded)",
            // Prose mentioning the words is not a classification.
            "the safety content policy doc explains this",
            // Malformed tail.
            "failed (type=",
        ] {
            assert!(
                content_refusal_tags(detail).is_none(),
                "must not classify as a refusal: {detail}"
            );
        }
    }

    /// A refusal that arrives with only a `type`, or only a `code`, still
    /// classifies — gateways are not consistent about sending both.
    #[test]
    fn either_tag_alone_is_enough() {
        assert!(content_refusal_tags("nope (code=content_filter)").is_some());
        assert!(content_refusal_tags("nope (type=content_policy_violation)").is_some());
        // …and the families beyond the one value we have a fixture for.
        assert!(content_refusal_tags("nope (code=moderation_blocked)").is_some());
        assert!(content_refusal_tags("nope (code=safety_block)").is_some());
    }

    /// The exact body the live gateway returns for car#786 must classify as an
    /// environment condition, not a generic failure — that classification is
    /// what keeps it off per-model health and stops the catalog advertising the
    /// namespace.
    #[test]
    fn gateway_not_configured_body_is_classified() {
        let body = r#"{"error":{"code":"openrouter_not_configured","type":"gateway_error",
                       "message":"OpenRouter inference is not configured on this Parslee environment."}}"#;
        let detail = gateway_unconfigured_detail(reqwest::StatusCode::SERVICE_UNAVAILABLE, body)
            .expect("the reported 503 body must classify");
        assert!(
            detail.contains("not configured on this Parslee environment"),
            "the operator-facing message must survive for the caller to surface: {detail}"
        );
    }

    /// Matched on `code`, not the prose, and not pinned to 503 — the message is
    /// reworded at will and the status could move. A future
    /// `<upstream>_not_configured` must classify on arrival rather than
    /// silently regressing to a generic failure.
    #[test]
    fn gateway_not_configured_matches_the_code_not_the_prose_or_status() {
        let other_upstream = r#"{"error":{"code":"bedrock_not_configured","type":"gateway_error","message":"nope"}}"#;
        assert!(
            gateway_unconfigured_detail(reqwest::StatusCode::BAD_GATEWAY, other_upstream).is_some(),
            "a different upstream, and a different 5xx, is the same condition"
        );

        // No message field: fall back to the code rather than an empty string,
        // so the error still names what happened.
        let bare = r#"{"error":{"code":"openrouter_not_configured"}}"#;
        assert_eq!(
            gateway_unconfigured_detail(reqwest::StatusCode::SERVICE_UNAVAILABLE, bare).as_deref(),
            Some("openrouter_not_configured")
        );
    }

    /// Everything else must stay a generic failure. Over-classifying would keep
    /// real model failures out of the health signal — the opposite bug, and a
    /// far quieter one.
    #[test]
    fn unrelated_gateway_failures_do_not_classify() {
        // A genuine 5xx crash.
        assert!(gateway_unconfigured_detail(
            reqwest::StatusCode::INTERNAL_SERVER_ERROR,
            r#"{"error":{"code":"internal_error","type":"gateway_error","message":"boom"}}"#
        )
        .is_none());
        // Right code shape, but a client error — an account/permission problem
        // is ProviderAccount's job, not this one.
        assert!(gateway_unconfigured_detail(
            reqwest::StatusCode::FORBIDDEN,
            r#"{"error":{"code":"openrouter_not_configured"}}"#
        )
        .is_none());
        // Prose mentioning the phrase must not be enough.
        assert!(gateway_unconfigured_detail(
            reqwest::StatusCode::SERVICE_UNAVAILABLE,
            r#"{"error":{"code":"overloaded","message":"openrouter is not configured yet"}}"#
        )
        .is_none());
        // Non-JSON body.
        assert!(gateway_unconfigured_detail(
            reqwest::StatusCode::SERVICE_UNAVAILABLE,
            "<html>502</html>"
        )
        .is_none());
    }

    #[test]
    fn parslee_body_uses_max_output_tokens_not_maxtokens() {
        // Regression guard for the silent fall-to-local bug: the gateway (OpenAI
        // Responses contract) needs `max_output_tokens`; `maxTokens` fails
        // inference (HTTP 200 + `event: error`). Verified against the live gateway.
        let input = vec![serde_json::json!({"role": "user", "content": "hi"})];
        let body = parslee_request_body("u1", input, 42, None, None);
        assert_eq!(body["max_output_tokens"], 42);
        assert!(
            body.get("maxTokens").is_none(),
            "maxTokens makes the gateway fail inference and CAR fall back to local"
        );
        assert_eq!(body["userId"], "u1");
        // temperature is never forwarded (reasoning models reject != 1).
        assert!(body.get("temperature").is_none());
        // No `tools` key when there are none...
        assert!(body.get("tools").is_none());
        // ...and it's an array when present (the coder path).
        let tools = vec![serde_json::json!({"type": "function", "name": "x"})];
        let body2 = parslee_request_body(
            "u1",
            vec![serde_json::json!({"role": "user", "content": "hi"})],
            8,
            Some(&tools),
            None,
        );
        assert!(body2["tools"].is_array());

        let gateway_model = "parslee/openrouter/open-reasoning";
        let gateway_body = parslee_request_body(
            "u1",
            vec![serde_json::json!({"role": "user", "content": "hi"})],
            8,
            None,
            Some(gateway_model),
        );
        assert_eq!(gateway_body["model"], gateway_model);

        let legacy_body = parslee_request_body(
            "u1",
            vec![serde_json::json!({"role": "user", "content": "hi"})],
            8,
            None,
            Some("parslee/gpt-5.5"),
        );
        assert!(legacy_body.get("model").is_none());

        for schema in crate::openrouter::builtin_schemas()
            .into_iter()
            .filter(|schema| schema.id.starts_with("parslee/openrouter/"))
        {
            assert_eq!(
                schema.supported_params,
                vec![crate::schema::GenerateParam::MaxTokens],
                "{} advertises a request control outside this body builder",
                schema.id
            );
            assert_eq!(
                gateway_body["max_output_tokens"], 8,
                "the advertised MaxTokens control must reach the managed request"
            );
        }
    }

    #[test]
    fn parslee_input_preserves_multiturn_tool_history() {
        // Regression guard for the coder read_file-forever bug: an assistant
        // tool call (no text content) and its tool result MUST reach the
        // Responses `input` as `function_call` / `function_call_output` items,
        // not be dropped. Dropping them erased the model's tool memory so it
        // re-read the same files every turn and never edited.
        let messages = vec![
            serde_json::json!({ "role": "user", "content": "fix the bug" }),
            serde_json::json!({
                "role": "assistant",
                "tool_calls": [{
                    "id": "call_1",
                    "type": "function",
                    "function": { "name": "read_file", "arguments": "{\"path\":\"src.py\"}" }
                }]
            }),
            serde_json::json!({
                "role": "tool",
                "tool_call_id": "call_1",
                "content": "def add(a,b): return a-b"
            }),
        ];
        let input = parslee_input_items(Some("you are a coder"), &messages);

        // system + user + function_call + function_call_output = 4 items.
        assert_eq!(input.len(), 4, "no tool-history item may be dropped");
        assert_eq!(input[0]["role"], "system");
        assert_eq!(input[1]["role"], "user");

        // The assistant tool call → function_call (was silently dropped before).
        assert_eq!(input[2]["type"], "function_call");
        assert_eq!(input[2]["call_id"], "call_1");
        assert_eq!(input[2]["name"], "read_file");
        assert_eq!(input[2]["arguments"], "{\"path\":\"src.py\"}");

        // The tool result → function_call_output keyed by the same call_id.
        assert_eq!(input[3]["type"], "function_call_output");
        assert_eq!(input[3]["call_id"], "call_1");
        assert_eq!(input[3]["output"], "def add(a,b): return a-b");
        // A bare {role:"tool"} object must NOT be emitted (invalid Responses input).
        assert!(input[3].get("role").is_none());
    }

    #[test]
    fn managed_incomplete_followed_by_done_remains_terminal_error() {
        let events = parse_parslee_responses_sse(
            concat!(
                "event: response.output_text.delta\n",
                "data: {\"delta\":\"partial\"}\n\n",
                "event: response.incomplete\n",
                "data: {\"response\":{\"status\":\"incomplete\",\"incomplete_details\":{\"reason\":\"max_output_tokens\"},\"usage\":{\"input_tokens\":4,\"output_tokens\":2}}}\n\n",
                "data: [DONE]\n\n"
            ),
        );
        assert!(events.iter().any(
            |event| matches!(event, crate::stream::StreamEvent::StopReason(reason) if reason == "max_output_tokens")
        ));
        assert!(events.iter().any(
            |event| matches!(event, crate::stream::StreamEvent::Error(message) if message == "managed inference incomplete: max_output_tokens")
        ));
        assert!(
            !events
                .iter()
                .any(|event| matches!(event, crate::stream::StreamEvent::Done { .. })),
            "[DONE] after response.incomplete must not upgrade the turn to success"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn managed_buffered_response_rejects_partial_text_followed_by_clean_eof() {
        use wiremock::matchers::{header, method, path};
        use wiremock::{Mock, MockServer, ResponseTemplate};

        let server = MockServer::start().await;
        let bearer = "buffered-partial-bearer";
        Mock::given(method("GET"))
            .and(path("/api/v1/organizations/me"))
            .and(header("authorization", format!("Bearer {bearer}")))
            .respond_with(
                ResponseTemplate::new(200)
                    .set_body_json(serde_json::json!({"organizationId": "org-partial"})),
            )
            .mount(&server)
            .await;
        Mock::given(method("GET"))
            .and(path("/connect/session"))
            .respond_with(
                ResponseTemplate::new(200).set_body_json(
                    serde_json::json!({"account": {"email": "partial@example.test"}}),
                ),
            )
            .mount(&server)
            .await;
        Mock::given(method("POST"))
            .and(path("/api/v1/orgs/org-partial/inference/responses"))
            .respond_with(ResponseTemplate::new(200).set_body_raw(
                "event: response.output_text.delta\ndata: {\"delta\":\"must not succeed\"}\n\n",
                "text/event-stream",
            ))
            .mount(&server)
            .await;

        let backend = RemoteBackend::new();
        let request = crate::protocol::ApiRequest {
            model: "managed-alias".into(),
            messages: vec![serde_json::json!({"role": "user", "content": "hello"})],
            system: None,
            system_stable_prefix: None,
            temperature: 0.0,
            max_tokens: 64,
            tools: None,
            tool_choice: None,
            parallel_tool_calls: None,
            stream: false,
            budget_tokens: 0,
            cache_control: false,
            cache_ttl: crate::tasks::generate::CacheTtl::default(),
            response_format: None,
        };
        let error = backend
            .parslee_assistant_request(&server.uri(), bearer, &request)
            .await
            .expect_err("clean EOF without response.completed must fail closed");
        assert!(
            error.to_string().contains("response.completed"),
            "terminal error must tell callers the completion event was missing: {error}"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn managed_stream_response_emits_error_after_partial_text_and_clean_eof() {
        use wiremock::matchers::{header, method, path};
        use wiremock::{Mock, MockServer, ResponseTemplate};

        let server = MockServer::start().await;
        let bearer = "stream-partial-bearer";
        Mock::given(method("GET"))
            .and(path("/api/v1/organizations/me"))
            .and(header("authorization", format!("Bearer {bearer}")))
            .respond_with(
                ResponseTemplate::new(200)
                    .set_body_json(serde_json::json!({"organizationId": "org-stream-partial"})),
            )
            .mount(&server)
            .await;
        Mock::given(method("GET"))
            .and(path("/connect/session"))
            .respond_with(
                ResponseTemplate::new(200).set_body_json(
                    serde_json::json!({"account": {"email": "partial@example.test"}}),
                ),
            )
            .mount(&server)
            .await;
        Mock::given(method("POST"))
            .and(path("/api/v1/orgs/org-stream-partial/inference/responses"))
            .respond_with(ResponseTemplate::new(200).set_body_raw(
                "event: response.output_text.delta\ndata: {\"delta\":\"partial\"}\n\n",
                "text/event-stream",
            ))
            .mount(&server)
            .await;

        let backend = RemoteBackend::new();
        let request = crate::protocol::ApiRequest {
            model: "managed-alias".into(),
            messages: vec![serde_json::json!({"role": "user", "content": "hello"})],
            system: None,
            system_stable_prefix: None,
            temperature: 0.0,
            max_tokens: 64,
            tools: None,
            tool_choice: None,
            parallel_tool_calls: None,
            stream: true,
            budget_tokens: 0,
            cache_control: false,
            cache_ttl: crate::tasks::generate::CacheTtl::default(),
            response_format: None,
        };
        let mut receiver = backend
            .parslee_assistant_stream_request(&server.uri(), bearer, &request, None)
            .await
            .expect("HTTP stream starts");
        let mut events = Vec::new();
        while let Some(event) = receiver.recv().await {
            events.push(event);
        }
        assert!(
            matches!(events.first(), Some(crate::stream::StreamEvent::TextDelta(text)) if text == "partial")
        );
        assert!(
            matches!(events.last(), Some(crate::stream::StreamEvent::Error(message)) if message.contains("response.completed")),
            "clean EOF without response.completed must append one terminal error: {events:?}"
        );
        assert!(
            !events
                .iter()
                .any(|event| matches!(event, crate::stream::StreamEvent::Done { .. })),
            "partial managed output must never be upgraded to success"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn openrouter_wire_success_stream_errors_concurrency_and_live_key_removal() {
        use crate::stream::StreamEvent;
        use wiremock::matchers::{body_partial_json, header, method, path};
        use wiremock::{Mock, MockServer, ResponseTemplate};

        let _credential_scope = crate::openrouter::test_credential_scope();
        let _provider_env = crate::openrouter::test_environment_scope_async().await;
        crate::openrouter::set_test_credential(None);
        let missing_schema = openrouter_schema("http://127.0.0.1:9");
        let missing = RemoteBackend::new()
            .generate(&missing_schema, "hello", None, 0.0, 32, None)
            .await
            .expect_err("missing OpenRouter key must fail before HTTP");
        assert!(missing.to_string().contains("car keys set openrouter"));

        crate::openrouter::set_test_credential(Some("test-openrouter-key"));
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/v1/chat/completions"))
            .and(header("authorization", "Bearer test-openrouter-key"))
            .and(header("x-openrouter-title", "CAR"))
            .and(body_partial_json(serde_json::json!({
                "model": "deepseek/deepseek-v3.2"
            })))
            .respond_with(
                ResponseTemplate::new(200)
                    .set_delay(Duration::from_millis(100))
                    .set_body_json(serde_json::json!({
                        "choices": [{
                            "message": {"role": "assistant", "content": "openrouter-ok"},
                            "finish_reason": "stop"
                        }],
                        "usage": {"prompt_tokens": 2, "completion_tokens": 3}
                    })),
            )
            .mount(&server)
            .await;
        let schema = openrouter_schema(&server.uri());
        let backend = RemoteBackend::new();
        let (first, second) = tokio::join!(
            backend.generate(&schema, "one", None, 0.0, 32, None),
            backend.generate(&schema, "two", None, 0.0, 32, None),
        );
        assert_eq!(first.unwrap(), "openrouter-ok");
        assert_eq!(second.unwrap(), "openrouter-ok");

        let in_flight = backend.generate(&schema, "in flight", None, 0.0, 32, None);
        let remove_key = async {
            tokio::time::sleep(Duration::from_millis(25)).await;
            crate::openrouter::set_test_credential(None);
        };
        let (in_flight_result, ()) = tokio::join!(in_flight, remove_key);
        assert_eq!(in_flight_result.unwrap(), "openrouter-ok");
        let next = backend
            .generate(&schema, "next", None, 0.0, 32, None)
            .await
            .expect_err("the next request must observe key removal");
        assert!(next.to_string().contains("car keys set openrouter"));

        crate::openrouter::set_test_credential(Some("test-openrouter-key"));
        let stream_server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/v1/chat/completions"))
            .respond_with(ResponseTemplate::new(200).set_body_raw(
                concat!(
                    "data: {\"choices\":[{\"delta\":{\"content\":\"open\"}}]}\n\n",
                    "data: {\"choices\":[{\"delta\":{\"content\":\"router\"},\"finish_reason\":\"stop\"}]}\n\n",
                    "data: [DONE]\n\n"
                ),
                "text/event-stream",
            ))
            .mount(&stream_server)
            .await;
        let stream_schema = openrouter_schema(&stream_server.uri());
        let mut receiver = backend
            .generate_stream(
                &stream_schema,
                "stream",
                None,
                None,
                0.0,
                32,
                None,
                None,
                None,
                None,
                None,
                None,
            )
            .await
            .unwrap();
        let mut streamed = String::new();
        let mut clean_stop = false;
        let mut completed = false;
        while let Some(event) = receiver.recv().await {
            match event {
                StreamEvent::TextDelta(delta) => streamed.push_str(&delta),
                StreamEvent::StopReason(reason) if reason == "stop" => clean_stop = true,
                StreamEvent::Done { .. } => completed = true,
                _ => {}
            }
        }
        assert_eq!(streamed, "openrouter");
        assert!(clean_stop, "stream must expose the provider's clean stop");
        assert!(
            completed,
            "a finish_reason followed by [DONE] is positive completion proof"
        );

        let stream_error_server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/v1/chat/completions"))
            .respond_with(ResponseTemplate::new(200).set_body_raw(
                concat!(
                    "data: {\"choices\":[{\"delta\":{\"content\":\"partial\"}}]}\n\n",
                    "data: {\"error\":{\"code\":402,\"message\":\"private balance details\"}}\n\n",
                    "data: {\"choices\":[{\"delta\":{\"content\":\"must-not-appear\"}}]}\n\n"
                ),
                "text/event-stream",
            ))
            .mount(&stream_error_server)
            .await;
        let stream_error_schema = openrouter_schema(&stream_error_server.uri());
        let mut receiver = backend
            .generate_stream(
                &stream_error_schema,
                "stream error",
                None,
                None,
                0.0,
                32,
                None,
                None,
                None,
                None,
                None,
                None,
            )
            .await
            .unwrap();
        let mut errors = Vec::new();
        let mut text = String::new();
        while let Some(event) = receiver.recv().await {
            match event {
                StreamEvent::Error(message) => errors.push(message),
                StreamEvent::TextDelta(delta) => text.push_str(&delta),
                _ => {}
            }
        }
        assert_eq!(errors, ["OpenRouter account is out of credits"]);
        assert_eq!(text, "partial");
        assert!(!errors[0].contains("private balance"));

        for (status, body, expected) in [
            (
                401,
                "sensitive upstream diagnostic must not escape",
                "key rejected",
            ),
            (
                402,
                "sensitive upstream diagnostic must not escape",
                "out of credits",
            ),
            (
                404,
                "sensitive upstream diagnostic must not escape",
                "no longer available on OpenRouter",
            ),
            (
                400,
                r#"{"error":{"message":"No endpoints found for this deprecated model; sensitive upstream diagnostic must not escape"}}"#,
                "no longer available on OpenRouter",
            ),
        ] {
            let error_server = MockServer::start().await;
            Mock::given(method("POST"))
                .and(path("/v1/chat/completions"))
                .respond_with(ResponseTemplate::new(status).set_body_string(body))
                .mount(&error_server)
                .await;
            let error_schema = openrouter_schema(&error_server.uri());
            let error = backend
                .generate(&error_schema, "fail", None, 0.0, 32, None)
                .await
                .expect_err("OpenRouter error should be translated");
            let message = error.to_string();
            assert!(message.contains(expected), "{status}: {message}");
            assert!(!message.contains("sensitive upstream diagnostic"));
            if status == 404 || status == 400 {
                assert!(message.contains("deepseek/deepseek-v3.2"));
            }
        }
    }

    #[tokio::test(flavor = "current_thread")]
    async fn personal_openrouter_partial_or_unframed_eof_never_emits_done() {
        use crate::stream::StreamEvent;
        use wiremock::matchers::{method, path};
        use wiremock::{Mock, MockServer, ResponseTemplate};

        let _credential_scope = crate::openrouter::test_credential_scope();
        crate::openrouter::set_test_credential(Some("terminal-key"));
        for body in [
            "data: {\"choices\":[{\"delta\":{\"content\":\"partial\"}}]}\n\n",
            "data: {\"choices\":[{\"delta\":{\"content\":\"partial\"},\"finish_reason\":\"stop\"}]}\n\n",
            concat!(
                "data: {\"choices\":[{\"delta\":{\"content\":\"partial\"}}]}\n\n",
                "data: [DONE]\n\n"
            ),
            "data: {\"choices\":[{\"delta\":{\"content\":\"trailing\"}}]}",
        ] {
            let server = MockServer::start().await;
            Mock::given(method("POST"))
                .and(path("/v1/chat/completions"))
                .respond_with(
                    ResponseTemplate::new(200).set_body_raw(body, "text/event-stream"),
                )
                .mount(&server)
                .await;
            let schema = openrouter_schema(&server.uri());
            let mut receiver = RemoteBackend::new()
                .generate_stream(
                    &schema, "stream", None, None, 0.0, 32, None, None, None, None, None, None,
                )
                .await
                .unwrap();
            let mut events = Vec::new();
            while let Some(event) = receiver.recv().await {
                events.push(event);
            }
            assert!(
                !events
                    .iter()
                    .any(|event| matches!(event, StreamEvent::Done { .. })),
                "partial sequence must not complete: {events:?}"
            );
            assert!(
                matches!(events.last(), Some(StreamEvent::Error(message)) if message.contains("before provider completion")),
                "partial sequence needs one sanitized terminal error: {events:?}"
            );
        }
    }

    #[tokio::test(flavor = "current_thread")]
    async fn google_and_vertex_stream_only_deliberate_finish_reasons_complete() {
        use crate::stream::StreamEvent;
        use wiremock::matchers::{method, path};
        use wiremock::{Mock, MockServer, ResponseTemplate};

        let _provider_env = crate::openrouter::test_environment_scope_async().await;
        unsafe { std::env::set_var("CAR_GOOGLE_STREAM_MATRIX_KEY", "matrix-key") };

        for protocol in [ApiProtocol::Google, ApiProtocol::VertexAi] {
            for (reason, should_complete) in [
                ("STOP", true),
                ("MAX_TOKENS", true),
                ("SAFETY", false),
                ("RECITATION", false),
                ("BLOCKLIST", false),
                ("PROHIBITED_CONTENT", false),
                ("SPII", false),
                ("MALFORMED_FUNCTION_CALL", false),
                ("UNKNOWN_PROVIDER_REASON", false),
            ] {
                let server = MockServer::start().await;
                let expected_path = match protocol {
                    ApiProtocol::Google => "/v1beta/models/gemini-test:streamGenerateContent",
                    ApiProtocol::VertexAi => {
                        "/publishers/google/models/gemini-test:streamGenerateContent"
                    }
                    _ => unreachable!(),
                };
                Mock::given(method("POST"))
                    .and(path(expected_path))
                    .respond_with(ResponseTemplate::new(200).set_body_raw(
                        format!(
                            "data: {{\"candidates\":[{{\"content\":{{\"parts\":[{{\"text\":\"matrix\"}}]}},\"finishReason\":\"{reason}\"}}]}}\n\n"
                        ),
                        "text/event-stream",
                    ))
                    .mount(&server)
                    .await;
                let schema = remote_stream_schema(
                    &server.uri(),
                    protocol,
                    "gemini-test",
                    "CAR_GOOGLE_STREAM_MATRIX_KEY",
                );
                let mut receiver = RemoteBackend::new()
                    .generate_stream(
                        &schema, "stream", None, None, 0.0, 32, None, None, None, None, None, None,
                    )
                    .await
                    .expect("Google and Vertex must both reach their streaming transports");
                let mut events = Vec::new();
                while let Some(event) = receiver.recv().await {
                    events.push(event);
                }
                assert_eq!(
                    events
                        .iter()
                        .any(|event| matches!(event, StreamEvent::Done { .. })),
                    should_complete,
                    "{protocol:?}/{reason}: {events:?}"
                );
                assert_eq!(
                    events
                        .iter()
                        .any(|event| matches!(event, StreamEvent::Error(_))),
                    !should_complete,
                    "{protocol:?}/{reason}: {events:?}"
                );
            }
        }

        unsafe { std::env::remove_var("CAR_GOOGLE_STREAM_MATRIX_KEY") };
    }

    #[tokio::test(flavor = "current_thread")]
    async fn azure_personal_openrouter_and_anthropic_terminal_matrix_stays_green() {
        use crate::stream::StreamEvent;
        use wiremock::matchers::{method, path};
        use wiremock::{Mock, MockServer, ResponseTemplate};

        let _credential_scope = crate::openrouter::test_credential_scope();
        let _provider_env = crate::openrouter::test_environment_scope_async().await;
        crate::openrouter::set_test_credential(Some("matrix-openrouter-key"));
        unsafe {
            std::env::set_var("CAR_AZURE_STREAM_MATRIX_KEY", "matrix-azure-key");
            std::env::set_var("CAR_ANTHROPIC_STREAM_MATRIX_KEY", "matrix-anthropic-key");
        }

        for (protocol, model, expected_path, body, key_env) in [
            (
                ApiProtocol::AzureOpenAi,
                "azure-deployment",
                "/openai/deployments/azure-deployment/chat/completions",
                concat!(
                    "data: {\"choices\":[{\"delta\":{\"content\":\"azure\"},\"finish_reason\":\"stop\"}]}\n\n",
                    "data: [DONE]\n\n"
                ),
                "CAR_AZURE_STREAM_MATRIX_KEY",
            ),
            (
                ApiProtocol::OpenRouter,
                "vendor/personal-model",
                "/v1/chat/completions",
                concat!(
                    "data: {\"choices\":[{\"delta\":{\"content\":\"personal\"},\"finish_reason\":\"stop\"}]}\n\n",
                    "data: [DONE]\n\n"
                ),
                "IGNORED_FOR_OPENROUTER",
            ),
            (
                ApiProtocol::Anthropic,
                "claude-test",
                "/v1/messages",
                concat!(
                    "event: content_block_delta\n",
                    "data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"anthropic\"}}\n\n",
                    "event: message_delta\n",
                    "data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\"},\"usage\":{\"output_tokens\":1}}\n\n",
                    "event: message_stop\n",
                    "data: {\"type\":\"message_stop\"}\n\n"
                ),
                "CAR_ANTHROPIC_STREAM_MATRIX_KEY",
            ),
        ] {
            let server = MockServer::start().await;
            Mock::given(method("POST"))
                .and(path(expected_path))
                .respond_with(
                    ResponseTemplate::new(200).set_body_raw(body, "text/event-stream"),
                )
                .mount(&server)
                .await;
            let schema = remote_stream_schema(&server.uri(), protocol, model, key_env);
            let mut receiver = RemoteBackend::new()
                .generate_stream(
                    &schema, "stream", None, None, 0.0, 32, None, None, None, None, None, None,
                )
                .await
                .unwrap();
            let mut events = Vec::new();
            while let Some(event) = receiver.recv().await {
                events.push(event);
            }
            assert!(
                events
                    .iter()
                    .any(|event| matches!(event, StreamEvent::Done { .. })),
                "{protocol:?}: {events:?}"
            );
            assert!(
                !events
                    .iter()
                    .any(|event| matches!(event, StreamEvent::Error(_))),
                "{protocol:?}: {events:?}"
            );
        }

        unsafe {
            std::env::remove_var("CAR_AZURE_STREAM_MATRIX_KEY");
            std::env::remove_var("CAR_ANTHROPIC_STREAM_MATRIX_KEY");
        }
    }

    #[tokio::test(flavor = "current_thread")]
    async fn personal_openrouter_spend_limit_is_error_not_success() {
        use crate::stream::StreamEvent;
        use wiremock::matchers::{method, path};
        use wiremock::{Mock, MockServer, ResponseTemplate};

        let _credential_scope = crate::openrouter::test_credential_scope();
        crate::openrouter::set_test_credential(Some("spend-key"));
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/v1/chat/completions"))
            .respond_with(ResponseTemplate::new(200).set_body_raw(
                concat!(
                    "data: {\"choices\":[{\"delta\":{\"content\":\"cross limit\"}}]}\n\n",
                    "data: {\"choices\":[{\"delta\":{\"content\":\"must not arrive\"},\"finish_reason\":\"stop\"}]}\n\n",
                    "data: [DONE]\n\n"
                ),
                "text/event-stream",
            ))
            .mount(&server)
            .await;
        let schema = openrouter_schema(&server.uri());
        let guard = crate::routing_ext::MidStreamSpendGuard::new(Some(0.0), 0.0, 0.0, 0.001);
        let mut receiver = RemoteBackend::new()
            .generate_stream(
                &schema,
                "stream",
                None,
                None,
                0.0,
                32,
                None,
                None,
                None,
                None,
                None,
                Some(guard),
            )
            .await
            .unwrap();
        let mut events = Vec::new();
        while let Some(event) = receiver.recv().await {
            events.push(event);
        }
        assert!(events
            .iter()
            .any(|event| matches!(event, StreamEvent::StopReason(reason) if reason.starts_with("spend_limit:"))));
        assert!(
            matches!(events.last(), Some(StreamEvent::Error(message)) if message == "stream cancelled because the spend limit was reached"),
            "spend cancellation must surface as a sanitized terminal error: {events:?}"
        );
        assert!(!events
            .iter()
            .any(|event| matches!(event, StreamEvent::Done { .. })));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn managed_openrouter_spend_limit_is_error_not_success() {
        use crate::routing_ext::MidStreamSpendGuard;
        use crate::stream::StreamEvent;
        use wiremock::matchers::{header, method, path};
        use wiremock::{Mock, MockServer, ResponseTemplate};

        let _provider_env = crate::openrouter::test_environment_scope_async().await;
        let server = MockServer::start().await;
        let bearer = "managed-spend-limit-bearer";
        unsafe {
            std::env::set_var(PARSLEE_ACCESS_TOKEN_ENV, bearer);
            std::env::set_var(car_auth::PARSLEE_API_BASE_KEY, server.uri());
        }
        Mock::given(method("GET"))
            .and(path("/api/v1/organizations/me"))
            .and(header("authorization", format!("Bearer {bearer}")))
            .respond_with(
                ResponseTemplate::new(200)
                    .set_body_json(serde_json::json!({"organizationId": "org-spend"})),
            )
            .mount(&server)
            .await;
        Mock::given(method("GET"))
            .and(path("/connect/session"))
            .respond_with(
                ResponseTemplate::new(200)
                    .set_body_json(serde_json::json!({"account": {"email": "user@example.test"}})),
            )
            .mount(&server)
            .await;
        Mock::given(method("POST"))
            .and(path("/api/v1/orgs/org-spend/inference/responses"))
            .respond_with(ResponseTemplate::new(200).set_body_raw(
                concat!(
                    "event: response.output_text.delta\n",
                    "data: {\"delta\":\"costly managed delta\"}\n\n",
                    "event: response.completed\n",
                    "data: {\"response\":{\"usage\":{\"input_tokens\":4,\"output_tokens\":4}}}\n\n"
                ),
                "text/event-stream",
            ))
            .mount(&server)
            .await;

        let schema = crate::openrouter::curated_schemas()
            .into_iter()
            .find(|schema| schema.id == "parslee/openrouter/frontier-general")
            .unwrap();
        let guard = MidStreamSpendGuard::new(Some(0.0), 0.0, 0.0, 1.0);
        let mut rx = RemoteBackend::new()
            .generate_stream(
                &schema,
                "hello",
                None,
                None,
                0.0,
                32,
                None,
                None,
                None,
                None,
                None,
                Some(guard),
            )
            .await
            .unwrap();
        let mut events = Vec::new();
        while let Some(event) = rx.recv().await {
            events.push(event);
        }
        assert!(events.iter().any(
            |event| matches!(event, StreamEvent::StopReason(reason) if reason.starts_with("spend_limit:"))
        ));
        assert!(matches!(
            events.last(),
            Some(StreamEvent::Error(message))
                if message == "stream cancelled because the spend limit was reached"
        ));
        assert!(!events
            .iter()
            .any(|event| matches!(event, StreamEvent::Done { .. })));

        unsafe {
            std::env::remove_var(PARSLEE_ACCESS_TOKEN_ENV);
            std::env::remove_var(car_auth::PARSLEE_API_BASE_KEY);
        }
    }

    #[test]
    fn sse_separator_handles_lf_crlf_and_partials() {
        // Plain "\n\n".
        assert_eq!(find_sse_separator(b"data: a\n\nrest"), Some((7, 2)));
        // CRLF "\r\n\r\n" must frame too (was silently never matched).
        assert_eq!(find_sse_separator(b"data: a\r\n\r\nrest"), Some((7, 4)));
        // No complete boundary yet (partial event) -> None, stays buffered.
        assert_eq!(find_sse_separator(b"data: a\n"), None);
        // Earliest boundary wins when both kinds are present.
        let buf = b"a\n\nb\r\n\r\nc";
        assert_eq!(find_sse_separator(buf), Some((1, 2)));
    }

    #[test]
    fn response_format_guard_rejects_only_unsupported() {
        // The shared guard BOTH remote seams call. Fails if the guard logic is
        // deleted/inverted. Anthropic rejects both variants; None is never
        // rejected. OpenAI accepts both.
        use crate::tasks::generate::ResponseFormat;
        let schema = ResponseFormat::JsonSchema {
            schema: serde_json::json!({"type": "object"}),
            strict: true,
            name: None,
        };
        let obj = ResponseFormat::JsonObject;

        let anthropic = crate::protocol::handler_for(crate::schema::ApiProtocol::Anthropic);
        for (format, expected_mode) in [
            (&schema, "structured-output-json-schema"),
            (&obj, "structured-output-json"),
        ] {
            let rej = response_format_rejection(anthropic.as_ref(), Some(format))
                .expect("Anthropic response format must be rejected");
            match rej {
                InferenceError::UnsupportedMode { mode, backend, .. } => {
                    assert_eq!(mode, expected_mode);
                    assert_eq!(backend, "anthropic");
                }
                other => panic!("expected UnsupportedMode, got {other:?}"),
            }
        }
        assert!(
            response_format_rejection(anthropic.as_ref(), None).is_none(),
            "no response_format is never rejected"
        );

        let openai = crate::protocol::handler_for(crate::schema::ApiProtocol::OpenAiCompat);
        assert!(response_format_rejection(openai.as_ref(), Some(&schema)).is_none());
        assert!(response_format_rejection(openai.as_ref(), Some(&obj)).is_none());
    }

    #[test]
    fn auth_rejection_is_anchored_on_http_status() {
        use crate::InferenceError;
        assert!(is_auth_rejection(&InferenceError::InferenceFailed(
            "Parslee chat failed: HTTP 401: {\"error\":\"expired\"}".into()
        )));
        assert!(is_auth_rejection(&InferenceError::InferenceFailed(
            "Parslee org lookup failed: HTTP 403: forbidden".into()
        )));
        // A 400 body that merely mentions 401 must NOT trigger a refresh.
        assert!(!is_auth_rejection(&InferenceError::InferenceFailed(
            "API returned 400: your last request 401'd upstream".into()
        )));
        assert!(!is_auth_rejection(&InferenceError::InferenceFailed(
            "HTTP 500: server error".into()
        )));
    }

    #[test]
    fn embeddings_resorted_to_input_order() {
        // Provider returns data[] out of order; index is authoritative.
        let data = vec![
            OpenAiEmbedData {
                embedding: vec![2.0],
                index: 2,
            },
            OpenAiEmbedData {
                embedding: vec![0.0],
                index: 0,
            },
            OpenAiEmbedData {
                embedding: vec![1.0],
                index: 1,
            },
        ];
        assert_eq!(
            order_embeddings(data),
            vec![vec![0.0], vec![1.0], vec![2.0]]
        );
    }

    #[test]
    fn format_endpoint_no_dup() {
        assert_eq!(
            format_endpoint("https://api.openai.com", "/v1/chat/completions"),
            "https://api.openai.com/v1/chat/completions"
        );
        assert_eq!(
            format_endpoint(
                "https://api.openai.com/v1/chat/completions",
                "/v1/chat/completions"
            ),
            "https://api.openai.com/v1/chat/completions"
        );
        assert_eq!(
            format_endpoint("https://api.openai.com/", "/v1/chat/completions"),
            "https://api.openai.com/v1/chat/completions"
        );
    }

    #[test]
    fn extract_endpoint_from_remote() {
        let schema = ModelSchema {
            id: "test/model:v1".into(),
            name: "Test".into(),
            provider: "test".into(),
            family: "test".into(),
            version: "1".into(),
            capabilities: vec![],
            context_length: 4096,
            max_output_tokens: None,
            param_count: String::new(),
            quantization: None,
            performance: Default::default(),
            cost: Default::default(),
            source: ModelSource::RemoteApi {
                endpoint: "https://api.test.com".into(),
                api_key_env: "NONEXISTENT_TEST_KEY_12345".into(),
                api_key_envs: vec![],
                api_version: None,
                protocol: ApiProtocol::OpenAiCompat,
            },
            tags: vec![],
            supported_params: vec![],
            public_benchmarks: vec![],
            trust_tier: crate::schema::TrustTier::Curated,
            deprecated: false,
            available: false,
            weights_ready: false,
        };
        let (endpoint, protocol) = extract_remote_endpoint(&schema).unwrap();
        assert_eq!(endpoint, "https://api.test.com");
        assert_eq!(protocol, ApiProtocol::OpenAiCompat);
    }

    #[test]
    fn extract_endpoint_non_remote_fails() {
        let schema = ModelSchema {
            id: "local/model:v1".into(),
            name: "Local".into(),
            provider: "test".into(),
            family: "test".into(),
            version: "1".into(),
            capabilities: vec![],
            context_length: 4096,
            max_output_tokens: None,
            param_count: String::new(),
            quantization: None,
            performance: Default::default(),
            cost: Default::default(),
            source: ModelSource::Local {
                hf_repo: "test".into(),
                hf_filename: "test".into(),
                tokenizer_repo: "test".into(),
            },
            tags: vec![],
            supported_params: vec![],
            public_benchmarks: vec![],
            trust_tier: crate::schema::TrustTier::Curated,
            deprecated: false,
            available: false,
            weights_ready: false,
        };
        assert!(extract_remote_endpoint(&schema).is_err());
    }

    #[test]
    fn proprietary_endpoint_is_openai_compat() {
        let _environment = crate::openrouter::test_environment_scope();
        unsafe {
            std::env::set_var(car_auth::PARSLEE_API_BASE_KEY, car_auth::DEFAULT_API_BASE);
        }
        // Parslee proprietary providers resolve to the OpenAI-compatible
        // protocol on the provider base URL. The live request path
        // (`parslee_assistant_request`) posts to the shipped Responses route
        // `/api/v1/orgs/{org}/inference/responses` directly — it does NOT
        // append the schema `chat_path` via `chat_path_for`, since the
        // `Proprietary` arm early-returns in `execute_request`.
        let schema = ModelSchema {
            id: "parslee/advisor".into(),
            name: "Parslee Advisor".into(),
            provider: "parslee".into(),
            family: "parslee".into(),
            version: "1".into(),
            capabilities: vec![],
            context_length: 128_000,
            max_output_tokens: None,
            param_count: String::new(),
            quantization: None,
            performance: Default::default(),
            cost: Default::default(),
            source: ModelSource::Proprietary {
                provider: "parslee".into(),
                endpoint: "https://api.parslee.ai".into(),
                auth: ProprietaryAuth::OAuth2Pkce {
                    authority: "https://api.parslee.ai".into(),
                    client_id: "parslee-car".into(),
                    scopes: vec!["inference:invoke".into()],
                },
                protocol: ProprietaryProtocol {
                    chat_path: "/api/v1/orgs/{orgId}/inference/responses".into(),
                    content_type: "application/json".into(),
                    streaming: false,
                    extra_headers: Default::default(),
                },
            },
            tags: vec![],
            supported_params: vec![],
            public_benchmarks: vec![],
            trust_tier: crate::schema::TrustTier::Curated,
            deprecated: false,
            available: true,
            weights_ready: true,
        };
        let (endpoint, protocol) = extract_remote_endpoint(&schema).unwrap();
        assert_eq!(endpoint, "https://api.parslee.ai");
        assert_eq!(protocol, ApiProtocol::OpenAiCompat);
        unsafe {
            std::env::remove_var(car_auth::PARSLEE_API_BASE_KEY);
        }
    }

    #[test]
    fn managed_parslee_endpoint_is_resolved_at_request_time() {
        let _environment = crate::openrouter::test_environment_scope();
        unsafe {
            std::env::set_var(
                car_auth::PARSLEE_API_BASE_KEY,
                "https://staging-api.parslee.ai/",
            );
        }
        let schema = ModelSchema {
            id: "parslee/openrouter/open-fast".into(),
            name: "Parslee OpenRouter Fast".into(),
            provider: "parslee".into(),
            family: "openrouter".into(),
            version: "1".into(),
            capabilities: vec![],
            context_length: 128_000,
            max_output_tokens: None,
            param_count: String::new(),
            quantization: None,
            performance: Default::default(),
            cost: Default::default(),
            source: ModelSource::Proprietary {
                provider: "parslee".into(),
                endpoint: car_auth::DEFAULT_API_BASE.into(),
                auth: ProprietaryAuth::OAuth2Pkce {
                    authority: car_auth::DEFAULT_API_BASE.into(),
                    client_id: "parslee-car".into(),
                    scopes: vec!["inference:invoke".into()],
                },
                protocol: ProprietaryProtocol {
                    chat_path: "/api/v1/orgs/{orgId}/inference/responses".into(),
                    content_type: "application/json".into(),
                    streaming: true,
                    extra_headers: Default::default(),
                },
            },
            tags: vec![],
            supported_params: vec![],
            public_benchmarks: vec![],
            trust_tier: crate::schema::TrustTier::Curated,
            deprecated: false,
            available: true,
            weights_ready: true,
        };

        let (endpoint, protocol) = extract_remote_endpoint(&schema).unwrap();
        assert_eq!(endpoint, "https://staging-api.parslee.ai");
        assert_eq!(protocol, ApiProtocol::OpenAiCompat);
        unsafe {
            std::env::remove_var(car_auth::PARSLEE_API_BASE_KEY);
        }
    }

    /// Reconciliation guard (U10): the shipped Parslee inference gateway
    /// serves chat on `/api/v1/orgs/{orgId}/inference/responses`
    /// (`parslee_assistant_request`, PR #3545). The three managed `parslee/*`
    /// aliases in the built-in catalog must describe that route, never the
    /// retired `/chat/stream` Employee-assistant stopgap.
    #[test]
    fn builtin_parslee_aliases_describe_shipped_responses_route() {
        let catalog = crate::registry::builtin_catalog();
        let aliases = ["parslee/fast", "parslee/reasoning", "parslee/advisor"];
        for id in aliases {
            let schema = catalog
                .iter()
                .find(|s| s.id == id)
                .unwrap_or_else(|| panic!("builtin catalog is missing managed alias {id}"));
            match &schema.source {
                ModelSource::Proprietary { protocol, .. } => {
                    assert_eq!(
                        protocol.chat_path, "/api/v1/orgs/{orgId}/inference/responses",
                        "{id} must point at the shipped inference/responses route, \
                         not the retired /chat/stream stopgap"
                    );
                }
                other => panic!("{id} should be a Proprietary managed provider, got {other:?}"),
            }
        }
    }

    #[tokio::test(flavor = "current_thread")]
    async fn managed_openrouter_stream_replays_provider_items_and_tool_history_on_second_turn() {
        use crate::stream::StreamEvent;
        use wiremock::matchers::{body_partial_json, header, method, path};
        use wiremock::{Mock, MockServer, ResponseTemplate};

        let server = MockServer::start().await;
        let _provider_env = crate::openrouter::test_environment_scope_async().await;
        let bearer = "managed-stream-bearer";
        unsafe {
            std::env::set_var(PARSLEE_ACCESS_TOKEN_ENV, bearer);
            std::env::set_var(car_auth::PARSLEE_API_BASE_KEY, server.uri());
        }
        Mock::given(method("GET"))
            .and(path("/api/v1/organizations/me"))
            .and(header("authorization", format!("Bearer {bearer}")))
            .respond_with(
                ResponseTemplate::new(200)
                    .set_body_json(serde_json::json!({"organizationId": "org-test"})),
            )
            .mount(&server)
            .await;
        Mock::given(method("GET"))
            .and(path("/connect/session"))
            .respond_with(
                ResponseTemplate::new(200)
                    .set_body_json(serde_json::json!({"account": {"email": "user@example.test"}})),
            )
            .mount(&server)
            .await;
        Mock::given(method("POST"))
            .and(path("/api/v1/orgs/org-test/inference/responses"))
            .and(body_partial_json(serde_json::json!({
                "model": "parslee/openrouter/frontier-general",
                "store": false,
                "include": ["reasoning.encrypted_content"]
            })))
            .respond_with(ResponseTemplate::new(200).set_body_raw(
                include_str!("../tests/fixtures/parslee-openrouter-reasoning-roundtrip.sse"),
                "text/event-stream",
            ))
            .expect(2)
            .mount(&server)
            .await;

        let schema = crate::openrouter::curated_schemas()
            .into_iter()
            .find(|schema| schema.id == "parslee/openrouter/frontier-general")
            .unwrap();
        let mut rx = RemoteBackend::new()
            .generate_stream(
                &schema, "hello", None, None, 0.0, 32, None, None, None, None, None, None,
            )
            .await
            .expect("managed stream should start");
        let mut text = String::new();
        let mut usage = None;
        let mut provider_items = Vec::new();
        while let Some(event) = rx.recv().await {
            match event {
                StreamEvent::TextDelta(delta) => text.push_str(&delta),
                StreamEvent::ProviderOutputItem(item) => provider_items.push(item),
                StreamEvent::Usage {
                    input_tokens,
                    output_tokens,
                    ..
                } => usage = Some((input_tokens, output_tokens)),
                _ => {}
            }
        }
        assert_eq!(text, "first answer");
        assert_eq!(usage, Some((17, 9)));
        let expected_reasoning = serde_json::json!({
            "type": "reasoning",
            "id": "rs_car_roundtrip",
            "status": "completed",
            "summary": [{"type": "summary_text", "text": "safe summary"}],
            "encrypted_content": "opaque-encrypted-reasoning",
        });
        assert_eq!(provider_items, vec![expected_reasoning.clone()]);

        let history = vec![
            crate::tasks::generate::Message::User {
                content: "first".into(),
            },
            crate::tasks::generate::Message::ProviderOutputItems {
                protocol: crate::protocol::OPENAI_RESPONSES_PROTOCOL.into(),
                items: provider_items,
            },
            crate::tasks::generate::Message::Assistant {
                content: text,
                tool_calls: vec![crate::tasks::generate::ToolCall {
                    id: Some("call-stream-history".into()),
                    name: "lookup".into(),
                    arguments: HashMap::from([(
                        "query".into(),
                        serde_json::Value::String("answer".into()),
                    )]),
                }],
                thinking: Vec::new(),
            },
            crate::tasks::generate::Message::ToolResult {
                tool_use_id: "call-stream-history".into(),
                content: "tool output".into(),
                provenance: Default::default(),
            },
            crate::tasks::generate::Message::User {
                content: "continue".into(),
            },
        ];
        let mut second = RemoteBackend::new()
            .generate_stream(
                &schema,
                "",
                Some(&history),
                None,
                0.0,
                32,
                None,
                None,
                None,
                None,
                None,
                None,
            )
            .await
            .expect("managed second streaming turn should start");
        while second.recv().await.is_some() {}

        let requests = server.received_requests().await.unwrap();
        let posts: Vec<serde_json::Value> = requests
            .iter()
            .filter(|request| {
                request.method.as_str() == "POST"
                    && request.url.path() == "/api/v1/orgs/org-test/inference/responses"
            })
            .map(|request| serde_json::from_slice(&request.body).unwrap())
            .collect();
        assert_eq!(posts.len(), 2);
        let second_input = posts[1]["input"].as_array().unwrap();
        assert!(second_input.contains(&expected_reasoning));
        assert!(second_input.iter().any(|item| {
            item["type"] == "function_call"
                && item["call_id"] == "call-stream-history"
                && item["name"] == "lookup"
        }));
        assert!(second_input.iter().any(|item| {
            item["type"] == "function_call_output"
                && item["call_id"] == "call-stream-history"
                && item["output"] == "tool output"
        }));
        assert!(second_input
            .iter()
            .any(|item| item["role"] == "user" && item["content"] == "continue"));
        unsafe {
            std::env::remove_var(PARSLEE_ACCESS_TOKEN_ENV);
            std::env::remove_var(car_auth::PARSLEE_API_BASE_KEY);
        }
    }

    #[tokio::test]
    async fn lease_key_proprietary_oauth2_resolves_env_override() {
        // resolve_env_or_keychain is env-first; setting the env var
        // exercises the OAuth2Pkce arm deterministically without
        // touching the OS keychain.
        let _provider_env = crate::openrouter::test_environment_scope_async().await;
        std::env::set_var(PARSLEE_ACCESS_TOKEN_ENV, "test-bearer-abc123");
        let schema = ModelSchema {
            id: "parslee/advisor".into(),
            name: "Parslee Advisor".into(),
            provider: "parslee".into(),
            family: "parslee".into(),
            version: "1".into(),
            capabilities: vec![],
            context_length: 128_000,
            max_output_tokens: None,
            param_count: String::new(),
            quantization: None,
            performance: Default::default(),
            cost: Default::default(),
            source: ModelSource::Proprietary {
                provider: "parslee".into(),
                endpoint: "https://api.parslee.ai".into(),
                auth: ProprietaryAuth::OAuth2Pkce {
                    authority: "https://api.parslee.ai".into(),
                    client_id: "parslee-car".into(),
                    scopes: vec!["inference:invoke".into()],
                },
                protocol: ProprietaryProtocol {
                    chat_path: "/api/v1/orgs/{orgId}/inference/responses".into(),
                    content_type: "application/json".into(),
                    streaming: false,
                    extra_headers: Default::default(),
                },
            },
            tags: vec![],
            supported_params: vec![],
            public_benchmarks: vec![],
            trust_tier: crate::schema::TrustTier::Curated,
            deprecated: false,
            available: true,
            weights_ready: true,
        };
        let lease = RemoteBackend::new()
            .lease_key(&schema, "https://api.parslee.ai")
            .await
            .expect("proprietary OAuth2 lease should resolve the env-override token");
        assert_eq!(lease.api_key, "test-bearer-abc123");
        std::env::remove_var(PARSLEE_ACCESS_TOKEN_ENV);
    }

    #[tokio::test]
    async fn non_parslee_oauth_schema_cannot_lease_or_send_parslee_bearer() {
        let _provider_env = crate::openrouter::test_environment_scope_async().await;
        let server = wiremock::MockServer::start().await;
        unsafe {
            std::env::set_var(PARSLEE_ACCESS_TOKEN_ENV, "must-not-leave-process");
        }
        let schema = ModelSchema {
            id: "community/custom-oauth".into(),
            name: "Custom OAuth".into(),
            provider: "community".into(),
            family: "community".into(),
            version: "1".into(),
            capabilities: vec![crate::schema::ModelCapability::Generate],
            context_length: 8_192,
            max_output_tokens: None,
            param_count: "api".into(),
            quantization: None,
            performance: Default::default(),
            cost: Default::default(),
            source: ModelSource::Proprietary {
                provider: "community".into(),
                endpoint: server.uri(),
                auth: ProprietaryAuth::OAuth2Pkce {
                    authority: "https://untrusted.example/authorize".into(),
                    client_id: "untrusted-client".into(),
                    scopes: vec!["inference".into()],
                },
                protocol: ProprietaryProtocol {
                    chat_path: "/custom/chat".into(),
                    content_type: "application/json".into(),
                    streaming: false,
                    extra_headers: Default::default(),
                },
            },
            tags: vec![],
            supported_params: vec![],
            public_benchmarks: vec![],
            trust_tier: crate::schema::TrustTier::Community,
            deprecated: false,
            available: true,
            weights_ready: true,
        };

        let lease_error = RemoteBackend::new()
            .lease_key(&schema, &server.uri())
            .await
            .expect_err("non-Parslee OAuth must be rejected before credential resolution");
        assert!(
            lease_error
                .to_string()
                .contains("OAuth2 PKCE credentials are supported only for provider 'parslee'"),
            "{lease_error}"
        );

        let inference_error = RemoteBackend::new()
            .generate(&schema, "must not send", None, 0.0, 16, None)
            .await
            .expect_err("explicit non-Parslee OAuth inference must fail closed");
        assert!(
            inference_error
                .to_string()
                .contains("OAuth2 PKCE credentials are supported only for provider 'parslee'"),
            "{inference_error}"
        );
        assert!(
            server.received_requests().await.unwrap().is_empty(),
            "the rejected schema must not send the Parslee bearer or make any network request"
        );
        unsafe {
            std::env::remove_var(PARSLEE_ACCESS_TOKEN_ENV);
        }
    }

    #[test]
    fn parse_openai_embed_response() {
        let json = r#"{"data":[{"embedding":[0.1,0.2,0.3]}]}"#;
        let resp: OpenAiEmbedResponse = serde_json::from_str(json).unwrap();
        assert_eq!(resp.data[0].embedding, vec![0.1, 0.2, 0.3]);
    }

    #[test]
    fn request_model_name_uses_vllm_server_model() {
        let schema = ModelSchema {
            id: "vllm-mlx/test".into(),
            name: "Display Name".into(),
            provider: "test".into(),
            family: "test".into(),
            version: "1".into(),
            capabilities: vec![],
            context_length: 4096,
            max_output_tokens: None,
            param_count: String::new(),
            quantization: None,
            performance: Default::default(),
            cost: Default::default(),
            source: ModelSource::VllmMlx {
                endpoint: "http://localhost:8000".into(),
                model_name: "mlx-community/Actual-Model".into(),
            },
            tags: vec![],
            supported_params: vec![],
            public_benchmarks: vec![],
            trust_tier: crate::schema::TrustTier::Curated,
            deprecated: false,
            available: true,
            weights_ready: true,
        };
        assert_eq!(request_model_name(&schema), "mlx-community/Actual-Model");
    }

    #[test]
    fn managed_gateway_request_body_uses_canonical_id_when_display_name_drifts() {
        let mut schema = crate::openrouter::curated_schemas()
            .into_iter()
            .find(|schema| schema.id == "parslee/openrouter/frontier-general")
            .expect("managed frontier alias");
        schema.name = "attacker-controlled-upstream-selector".into();

        let selector = request_model_name(&schema);
        let body = parslee_request_body(
            "u1",
            vec![serde_json::json!({"role": "user", "content": "hi"})],
            8,
            None,
            Some(&selector),
        );

        assert_eq!(selector, "parslee/openrouter/frontier-general");
        assert_eq!(body["model"], "parslee/openrouter/frontier-general");
    }

    #[test]
    fn truncate_prompt_fits_returns_unchanged() {
        let prompt = "short prompt";
        let result = truncate_prompt_to_fit(prompt, None, None, 16, 0, 256);
        assert_eq!(result, prompt);
    }

    #[test]
    fn truncate_prompt_cjk_mid_codepoint_does_not_panic() {
        // 200 CJK chars = 600 bytes. With max_tokens=20 and window=209:
        // reserved = 20 + 100 overhead = 120, available = 89, chars_to_keep = 356.
        // start = 600 - 356 = 244, which is NOT a char boundary for 3-byte chars.
        let prompt: String = std::iter::repeat_n('\u{4E16}', 200).collect();
        let result = truncate_prompt_to_fit(&prompt, None, None, 20, 0, 209);
        assert!(result.starts_with("[...truncated...]"));
        let kept = result.strip_prefix("[...truncated...]\n").unwrap();
        assert!(!kept.is_empty());
    }

    #[test]
    fn truncate_prompt_accounts_for_context_and_tools() {
        let prompt = "line one\nline two\nline three\n".repeat(50);
        let tools = vec![serde_json::json!({"name": "demo_tool"})];
        let result = truncate_prompt_to_fit(&prompt, Some("ctx"), Some(&tools), 20, 0, 240);
        assert!(result.starts_with("[...truncated...]"));
    }

    #[test]
    fn truncate_prompt_reserves_media_tokens() {
        // 400-char prompt = 100 tokens. Window 500, max_tokens 20,
        // overhead 100 → reserved 120, available 380: fits untouched
        // with no media…
        let prompt = "word ".repeat(80);
        let untouched = truncate_prompt_to_fit(&prompt, None, None, 20, 0, 500);
        assert_eq!(untouched, prompt);
        // …but a media block worth 300 tokens (e.g. ~1s of video)
        // shrinks the available text budget to 80 and forces truncation.
        let truncated = truncate_prompt_to_fit(&prompt, None, None, 20, 300, 500);
        assert!(truncated.starts_with("[...truncated...]"));
    }

    #[test]
    fn parslee_responses_sse_concatenates_text_and_usage() {
        // Responses text deltas terminated by response.completed with usage.
        let raw = "event: response.output_text.delta\ndata: {\"delta\":\"Hello\"}\n\n\
                   event: response.output_text.delta\ndata: {\"delta\":\", world\"}\n\n\
                   event: response.completed\ndata: {\"response\":{\"usage\":{\"input_tokens\":10,\"output_tokens\":3}}}\n\n";
        let mut acc = crate::stream::StreamAccumulator::default();
        for ev in parse_parslee_responses_sse(raw) {
            acc.push(&ev);
        }
        let (text, tool_calls, usage, _stop) = acc.finish_with_usage();
        assert_eq!(text, "Hello, world");
        assert!(tool_calls.is_empty());
        let usage = usage.expect("usage reported");
        assert_eq!(usage.prompt_tokens, 10);
        assert_eq!(usage.completion_tokens, 3);
    }

    #[test]
    fn parslee_responses_sse_parses_tool_call() {
        // A function_call: output_item.added (name + call_id) then an args delta.
        let raw = "event: response.output_item.added\n\
                   data: {\"output_index\":0,\"item\":{\"type\":\"function_call\",\"name\":\"get_weather\",\"call_id\":\"call_1\"}}\n\n\
                   event: response.function_call_arguments.delta\n\
                   data: {\"output_index\":0,\"delta\":\"{\\\"city\\\":\\\"NYC\\\"}\"}\n\n\
                   event: response.completed\ndata: {\"response\":{\"usage\":{\"input_tokens\":5,\"output_tokens\":2}}}\n\n";
        let mut acc = crate::stream::StreamAccumulator::default();
        for ev in parse_parslee_responses_sse(raw) {
            acc.push(&ev);
        }
        let (_text, tool_calls) = acc.finish();
        assert_eq!(tool_calls.len(), 1);
        assert_eq!(tool_calls[0].name, "get_weather");
        assert_eq!(tool_calls[0].id.as_deref(), Some("call_1"));
        assert_eq!(
            tool_calls[0].arguments.get("city").and_then(|v| v.as_str()),
            Some("NYC")
        );
    }

    #[test]
    fn parslee_responses_sse_ignores_unparseable_frames() {
        let raw = "event: response.output_text.delta\ndata: not-json\n\n\
                   event: response.output_text.delta\ndata: {\"delta\":\"ok\"}\n\n";
        let mut acc = crate::stream::StreamAccumulator::default();
        for ev in parse_parslee_responses_sse(raw) {
            acc.push(&ev);
        }
        assert_eq!(acc.finish().0, "ok");
    }

    // --- Parslee identity resolution over real HTTP (wiremock). This is the
    // org-lookup endpoint that returned `HTTP 401` in the live #313 incident;
    // it had no in-CI coverage. Unique bearers per test so the process-wide
    // `PARSLEE_IDENTITY` cache can't cross-contaminate. ---

    #[tokio::test]
    async fn parslee_identity_resolves_org_and_user() {
        use wiremock::matchers::{method, path};
        use wiremock::{Mock, MockServer, ResponseTemplate};
        let server = MockServer::start().await;
        Mock::given(method("GET"))
            .and(path("/api/v1/organizations/me"))
            .respond_with(
                ResponseTemplate::new(200)
                    .set_body_json(serde_json::json!({ "organizationId": "org_test" })),
            )
            .mount(&server)
            .await;
        Mock::given(method("GET"))
            .and(path("/connect/session"))
            .respond_with(
                ResponseTemplate::new(200).set_body_json(
                    serde_json::json!({ "account": { "email": "user@example.com" } }),
                ),
            )
            .mount(&server)
            .await;

        let backend = RemoteBackend::new();
        let (org, user) = backend
            .parslee_identity(&server.uri(), "ident-ok-bearer")
            .await
            .expect("identity should resolve");
        assert_eq!(org, "org_test");
        assert_eq!(user, "user@example.com");
    }

    #[tokio::test]
    async fn parslee_identity_cache_is_isolated_by_api_base() {
        use wiremock::matchers::{method, path};
        use wiremock::{Mock, MockServer, ResponseTemplate};

        let first = MockServer::start().await;
        let second = MockServer::start().await;
        for (server, org) in [(&first, "org-first"), (&second, "org-second")] {
            Mock::given(method("GET"))
                .and(path("/api/v1/organizations/me"))
                .respond_with(
                    ResponseTemplate::new(200)
                        .set_body_json(serde_json::json!({ "organizationId": org })),
                )
                .mount(server)
                .await;
            Mock::given(method("GET"))
                .and(path("/connect/session"))
                .respond_with(ResponseTemplate::new(200).set_body_json(
                    serde_json::json!({ "account": { "email": "same@example.test" } }),
                ))
                .mount(server)
                .await;
        }

        let backend = RemoteBackend::new();
        let first_identity = backend
            .parslee_identity(&first.uri(), "same-bearer-across-environments")
            .await
            .unwrap();
        let second_identity = backend
            .parslee_identity(&second.uri(), "same-bearer-across-environments")
            .await
            .unwrap();
        assert_eq!(first_identity.0, "org-first");
        assert_eq!(second_identity.0, "org-second");
    }

    #[tokio::test]
    async fn parslee_identity_surfaces_org_lookup_401() {
        use wiremock::matchers::{method, path};
        use wiremock::{Mock, MockServer, ResponseTemplate};
        let server = MockServer::start().await;
        Mock::given(method("GET"))
            .and(path("/api/v1/organizations/me"))
            .respond_with(ResponseTemplate::new(401).set_body_string("Authentication required"))
            .mount(&server)
            .await;

        let backend = RemoteBackend::new();
        let err = backend
            .parslee_identity(&server.uri(), "ident-401-bearer")
            .await
            .expect_err("a 401 org lookup must error");
        // The exact message shape the live incident surfaced — and the string
        // `is_auth_rejection` keys on to trigger the reactive refresh.
        let msg = err.to_string();
        assert!(msg.contains("org lookup failed"), "got: {msg}");
        assert!(msg.contains("HTTP 401"), "got: {msg}");
        assert!(
            is_auth_rejection(&err),
            "401 must be classified as auth rejection"
        );
    }

    // --- Trust-store failure: degrade, don't panic -------------------------
    //
    // These run against an injected `SSL_CERT_FILE` fixture, so they are
    // in-process, network-free, and independent of the certificates the host
    // machine actually has. Every one of them calls
    // `assert_breaks_client_construction()` first: if the fixture ever stops
    // breaking certificate loading, the run goes red there instead of turning
    // every assertion below it vacuously green.

    use crate::tls_client::test_seam::{
        mixed_cert_pem, TrustStoreScope, UNPARSEABLE_CERT_PEM, VALID_CERT_PEM,
    };
    use crate::tls_client::{TrustFallback, HUGGINGFACE_PROBE, REMOTE_BACKEND, VLLM_HEALTH_CHECK};

    /// A port nothing listens on, so a request to it fails at connect without
    /// touching the network.
    fn closed_loopback_url() -> String {
        format!("{}/", closed_loopback_endpoint())
    }

    /// An `https` origin on a port nothing listens on: requests fail at connect,
    /// before any handshake, and no network is touched.
    fn closed_loopback_endpoint() -> String {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind ephemeral port");
        let port = listener.local_addr().expect("local addr").port();
        drop(listener);
        format!("https://127.0.0.1:{port}")
    }

    #[tokio::test]
    async fn empty_and_mixed_trust_stores_build_without_degrading() {
        // Absent/empty store and "some certificates unreadable, some fine" are
        // both ordinary. Only `valid == 0 && invalid > 0` is the failure.
        let mut scope = TrustStoreScope::acquire_async("").await;
        crate::tls_client::reset_sites_for_test();
        let backend = RemoteBackend::new();
        assert!(
            backend.tls_degradation.is_none(),
            "an empty certificate store must not degrade anything"
        );

        scope.repoint(&mixed_cert_pem());
        crate::tls_client::reset_sites_for_test();
        let backend = RemoteBackend::new();
        assert!(
            backend.tls_degradation.is_none(),
            "one readable certificate alongside an unreadable one must not degrade"
        );
        assert!(
            crate::tls_client::last_degradation(&REMOTE_BACKEND).is_none(),
            "nothing degraded, so nothing should have been recorded or logged"
        );
    }

    #[tokio::test]
    async fn remote_backend_degrades_to_public_cas_instead_of_panicking() {
        let scope = TrustStoreScope::acquire_async(UNPARSEABLE_CERT_PEM).await;
        scope.assert_breaks_client_construction();
        crate::tls_client::reset_sites_for_test();

        // Before #692 this line panicked inside a `OnceLock` initializer that
        // does not poison, so every later inference request panicked too.
        let backend = RemoteBackend::new();
        let degradation = backend
            .tls_degradation
            .clone()
            .expect("a broken trust store must be recorded, not swallowed");
        assert_eq!(degradation.fallback, TrustFallback::PublicCaOnly);
        assert!(
            !degradation.source.is_empty(),
            "the original reqwest builder error must be kept"
        );

        // The request that needed the client fails as an ordinary error — the
        // task is not unwound — and it names the cause instead of leaving the
        // operator with a bare "unknown issuer".
        let err = backend
            .client
            .get(closed_loopback_url())
            .send()
            .await
            .expect_err("nothing listens on that port");
        assert!(
            err.is_connect(),
            "expected a connect-class error, got: {err}"
        );
        let surfaced = backend
            .request_error("remote call failed", &err)
            .to_string();
        assert!(
            surfaced.contains("the OS certificate store failed to load"),
            "got: {surfaced}"
        );
        assert!(
            surfaced.contains("restart the daemon"),
            "the operator must be told how to recover; got: {surfaced}"
        );
    }

    #[tokio::test]
    async fn degraded_backend_carries_its_reason_into_real_inference_requests() {
        // Drives the actual request paths on `RemoteBackend`'s client, not
        // `request_error` directly — calling the helper from the test body is
        // what let the three inference sites keep emitting a bare
        // "HTTP error: ..." while the speech sites were annotated.
        const KEY_ENV: &str = "CAR_TLS_DEGRADATION_TEST_KEY";
        let scope = TrustStoreScope::acquire_async(UNPARSEABLE_CERT_PEM).await;
        scope.assert_breaks_client_construction();
        crate::tls_client::reset_sites_for_test();
        std::env::set_var(KEY_ENV, "test-key");

        let backend = RemoteBackend::new();
        assert!(
            backend.tls_degradation.is_some(),
            "setup: must have degraded"
        );
        let schema = remote_stream_schema(
            &closed_loopback_endpoint(),
            ApiProtocol::OpenAiCompat,
            "test-model",
            KEY_ENV,
        );

        // Streaming path. No retry loop, so this one is fast.
        let streamed = backend
            .generate_stream(
                &schema, "hello", None, None, 0.0, 32, None, None, None, None, None, None,
            )
            .await
            .expect_err("nothing listens on that port")
            .to_string();
        assert!(
            streamed.contains("the OS certificate store failed to load"),
            "the streaming inference path dropped the degradation reason: {streamed}"
        );
        assert!(
            streamed.contains("restart the daemon"),
            "the operator must be told how inference recovers: {streamed}"
        );

        // Generic completion path — the one `get_inference_engine` drives, and
        // the reason the whole fix exists. It retries transient transport
        // errors (3 attempts, 3s + 6s backoff), so this leg takes ~9s.
        let completed = backend
            .generate(&schema, "hello", None, 0.0, 32, None)
            .await
            .expect_err("nothing listens on that port")
            .to_string();
        assert!(
            completed.contains("the OS certificate store failed to load"),
            "the completion inference path dropped the degradation reason: {completed}"
        );

        // The managed Parslee path performs an identity lookup before its
        // inference request. That lookup uses the same degraded client and is
        // part of the real inference call chain, so it must carry the reason
        // too instead of failing early with a bare transport error.
        let mut parslee_schema = remote_stream_schema(
            &closed_loopback_endpoint(),
            ApiProtocol::OpenAiCompat,
            "parslee/test-model",
            KEY_ENV,
        );
        parslee_schema.provider = "parslee".into();
        parslee_schema.source = ModelSource::Proprietary {
            provider: "parslee".into(),
            endpoint: closed_loopback_endpoint(),
            auth: ProprietaryAuth::BearerTokenEnv {
                env_var: KEY_ENV.into(),
            },
            protocol: ProprietaryProtocol::default(),
        };
        let managed = backend
            .generate(&parslee_schema, "hello", None, 0.0, 32, None)
            .await
            .expect_err("the managed identity lookup cannot connect")
            .to_string();
        assert!(
            managed.contains("the OS certificate store failed to load"),
            "the managed inference path dropped the degradation reason: {managed}"
        );

        std::env::remove_var(KEY_ENV);
    }

    #[tokio::test]
    async fn remote_backend_default_matches_new() {
        let scope = TrustStoreScope::acquire_async(UNPARSEABLE_CERT_PEM).await;
        scope.assert_breaks_client_construction();
        crate::tls_client::reset_sites_for_test();

        let explicit = RemoteBackend::new();
        let defaulted = RemoteBackend::default();
        assert_eq!(
            explicit.tls_degradation.map(|d| d.fallback),
            defaulted.tls_degradation.map(|d| d.fallback),
            "`Default` delegates to `new`, so the two must agree on degradation"
        );
    }

    #[tokio::test]
    async fn tls_floor_rung_builds_under_a_broken_trust_store() {
        // The ladder's guarantee, stated as a test: rung 2 keeps the built-in
        // public CAs, and rung 3 — every built-in root source off — always
        // builds, which is what makes construction panic-free.
        let scope = TrustStoreScope::acquire_async(UNPARSEABLE_CERT_PEM).await;
        scope.assert_breaks_client_construction();
        assert!(
            reqwest::Client::builder()
                .tls_built_in_native_certs(false)
                .build()
                .is_ok(),
            "rung 2 (built-in public CAs only) must build"
        );
        assert!(
            reqwest::Client::builder()
                .tls_built_in_root_certs(false)
                .build()
                .is_ok(),
            "rung 3 (the floor) must build — it is what keeps construction panic-free"
        );
    }

    #[tokio::test]
    async fn plain_transport_failures_are_not_reported_as_certificate_problems() {
        // The predicate, exhaustively: only a connect failure that is not a
        // timeout can plausibly be a trust problem.
        assert!(is_trust_related(true, false));
        assert!(
            !is_trust_related(true, true),
            "a connect *timeout* is not a trust problem"
        );
        assert!(!is_trust_related(false, true));
        assert!(!is_trust_related(false, false));

        // A backend that never degraded says nothing about certificates, even
        // on exactly the error class that would carry the note if it had.
        let scope = TrustStoreScope::acquire_async(VALID_CERT_PEM).await;
        let healthy = RemoteBackend::new();
        assert!(healthy.tls_degradation.is_none());
        let err = healthy
            .client
            .get(closed_loopback_url())
            .send()
            .await
            .expect_err("nothing listens on that port");
        let surfaced = healthy
            .request_error("remote call failed", &err)
            .to_string();
        assert!(
            !surfaced.contains("certificate store"),
            "a healthy backend must not blame certificates; got: {surfaced}"
        );

        // A degraded backend still returns clean results and clean HTTP-status
        // errors: auth and rate-limit failures come back as `Ok(response)` and
        // never reach the error-annotating path at all.
        drop(scope);
        let scope = TrustStoreScope::acquire_async(UNPARSEABLE_CERT_PEM).await;
        scope.assert_breaks_client_construction();
        crate::tls_client::reset_sites_for_test();
        let degraded = RemoteBackend::new();
        assert!(degraded.tls_degradation.is_some());

        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/ok"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_string("fine"))
            .mount(&server)
            .await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .and(wiremock::matchers::path("/denied"))
            .respond_with(wiremock::ResponseTemplate::new(401))
            .mount(&server)
            .await;

        let ok = degraded
            .client
            .get(format!("{}/ok", server.uri()))
            .send()
            .await
            .expect("a degraded client still works against reachable endpoints");
        assert!(ok.status().is_success());
        assert_eq!(
            ok.text().await.expect("body"),
            "fine",
            "a successful request carries no degradation text"
        );

        let denied = degraded
            .client
            .get(format!("{}/denied", server.uri()))
            .send()
            .await
            .expect("a 401 is a response, not a transport failure");
        assert_eq!(denied.status().as_u16(), 401);
    }

    #[test]
    fn trust_store_scope_restores_the_previous_override_even_on_panic() {
        const FILE_VAR: &str = "SSL_CERT_FILE";
        const DIR_VAR: &str = "SSL_CERT_DIR";
        // Both the "before" read and the comparison happen under the same lock
        // the scope itself takes. Reading outside it fails under a shared
        // `cargo test` process, where another env test can install its own
        // override in between.
        let lock = crate::openrouter::test_environment_scope();
        let before_file = std::env::var_os(FILE_VAR);
        let before_dir = std::env::var_os(DIR_VAR);
        let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            let _scope = TrustStoreScope::with_lock(lock, UNPARSEABLE_CERT_PEM);
            assert!(
                std::env::var_os(FILE_VAR).is_some(),
                "the scope must install the override"
            );
            assert!(
                std::env::var_os(DIR_VAR).is_none(),
                "the scope must suppress SSL_CERT_DIR so host roots cannot make \
                 the failure fixture pass"
            );
            panic!("simulated mid-test failure");
        }));
        assert!(outcome.is_err(), "the simulated failure must have unwound");
        let _relock = crate::openrouter::test_environment_scope();
        assert_eq!(
            std::env::var_os(FILE_VAR),
            before_file,
            "the guard must restore the previous SSL_CERT_FILE value"
        );
        assert_eq!(
            std::env::var_os(DIR_VAR),
            before_dir,
            "the guard must restore the previous SSL_CERT_DIR value so a \
             panicking test cannot poison the rest of the suite"
        );
    }

    /// A hand-rolled `tracing` subscriber — `tracing-subscriber` is not a
    /// dependency of this crate and adding one is out of scope. It records
    /// every event's level and rendered fields so a test can prove a warning
    /// was genuinely emitted at warn level, not merely recorded in the
    /// crate-internal slot the other tests read.
    struct CapturingSubscriber {
        events: std::sync::Arc<std::sync::Mutex<Vec<(tracing::Level, String)>>>,
    }

    #[derive(Default)]
    struct FieldCollector(String);

    impl tracing::field::Visit for FieldCollector {
        fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
            use std::fmt::Write as _;
            let _ = write!(self.0, " {}={:?}", field.name(), value);
        }
    }

    impl tracing::Subscriber for CapturingSubscriber {
        fn enabled(&self, _metadata: &tracing::Metadata<'_>) -> bool {
            true
        }
        fn new_span(&self, _attrs: &tracing::span::Attributes<'_>) -> tracing::span::Id {
            tracing::span::Id::from_u64(1)
        }
        fn record(&self, _span: &tracing::span::Id, _values: &tracing::span::Record<'_>) {}
        fn record_follows_from(&self, _span: &tracing::span::Id, _follows: &tracing::span::Id) {}
        fn event(&self, event: &tracing::Event<'_>) {
            let mut fields = FieldCollector::default();
            event.record(&mut fields);
            self.events
                .lock()
                .unwrap_or_else(|p| p.into_inner())
                .push((*event.metadata().level(), fields.0));
        }
        fn enter(&self, _span: &tracing::span::Id) {}
        fn exit(&self, _span: &tracing::span::Id) {}
    }

    #[test]
    fn every_site_warns_once_at_warn_level_and_keeps_its_own_record() {
        // Plain `#[test]` so the captured-log thread-local covers the async
        // health-check site too: a current-thread runtime polls on this thread.
        let scope = TrustStoreScope::acquire_blocking(UNPARSEABLE_CERT_PEM);
        scope.assert_breaks_client_construction();
        crate::tls_client::reset_sites_for_test();

        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .expect("current-thread runtime");
        let events = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
        tracing::subscriber::with_default(
            CapturingSubscriber {
                events: events.clone(),
            },
            || {
                // Twice each: the limiter is once *per site*, not per call, so
                // a health check on a timer cannot flood the log.
                let _ = RemoteBackend::new();
                let _ = RemoteBackend::new();
                let _ = crate::upgrade::HuggingFaceProbe::new();
                let _ = crate::upgrade::HuggingFaceProbe::new();
                runtime.block_on(async {
                    let _ = crate::vllm_mlx::health_check("http://127.0.0.1:1").await;
                    let _ = crate::vllm_mlx::health_check("http://127.0.0.1:1").await;
                });
            },
        );

        let captured = events.lock().unwrap_or_else(|p| p.into_inner()).clone();
        let warnings: Vec<&String> = captured
            .iter()
            .filter(|(level, _)| *level <= tracing::Level::WARN)
            .map(|(_, message)| message)
            .collect();
        assert_eq!(
            warnings.len(),
            3,
            "exactly one warning per site, logged once each; captured: {captured:?}"
        );

        for site in [&REMOTE_BACKEND, &HUGGINGFACE_PROBE, &VLLM_HEALTH_CHECK] {
            assert!(
                warnings.iter().any(|m| m.contains(site.name())),
                "{} degraded without its own warning — an earlier site's warning \
                 must not silence it; captured: {warnings:?}",
                site.name()
            );
            let record = crate::tls_client::last_degradation(site)
                .unwrap_or_else(|| panic!("{} must keep its own record", site.name()));
            assert_eq!(record.fallback, TrustFallback::PublicCaOnly);
            assert!(!record.source.is_empty());
        }

        // Every warning names the underlying certificate error, but the
        // recovery advice is site-specific and must be: only the inference
        // backend is built once per daemon. Telling an operator to restart for
        // the probe or the health check — both of which rebuild their client
        // on every call — would be wrong advice.
        let warning_for = |site: &'static crate::tls_client::Site| -> &String {
            warnings
                .iter()
                .find(|m| m.contains(site.name()))
                .unwrap_or_else(|| panic!("no warning for {}", site.name()))
        };
        for message in &warnings {
            assert!(
                message.contains("the OS certificate store failed to load"),
                "every warning must name the underlying error; got: {message}"
            );
        }
        assert!(
            warning_for(&REMOTE_BACKEND).contains("restart the daemon to recover"),
            "inference recovers only on a restart and must say so: {}",
            warning_for(&REMOTE_BACKEND)
        );
        for site in [&HUGGINGFACE_PROBE, &VLLM_HEALTH_CHECK] {
            let message = warning_for(site);
            assert!(
                message.contains("with no restart"),
                "{} rebuilds its client per call, so it must not demand a restart: {message}",
                site.name()
            );
            assert!(
                !message.contains("restart the daemon to recover"),
                "{} self-heals — telling the operator to restart is wrong advice: {message}",
                site.name()
            );
        }
    }
}