agent_engine/runtime/openai/

stream.rs

//! Streaming path for OpenAI-compatible providers.
//!
//! Mirrors `ApiMethods::call_api_stream_inner` but speaks OpenAI chat/completions
//! and translates back to Anthropic-shaped events for the rest of the runtime.

use super::translate;
use super::types::{
    ChatContentPart, ChatMessage, OaiEvent, ProviderConfig, StreamOptions, ToolCall,
};
use super::wire::StreamDecoder;
use crate::runtime::trace::openai as tr;
use crate::runtime::types::StreamEvent;
use futures::StreamExt;
use serde_json::{json, Value};
use tokio::sync::mpsc;

/// Trace capture for one batch of decoded Chat Completions events: marks the
/// first parsed model event and records provider-reported usage. Metadata
/// only — token counts, no content.
fn capture_oai_trace_signals(
    events: &[OaiEvent],
    attempt: &mut tr::StreamAttempt,
    usage: &mut Option<crate::runtime::trace::UsageMeta>,
) {
    for ev in events {
        if !matches!(ev, OaiEvent::Warning(_)) {
            attempt.mark_first_model_event();
        }
        if let OaiEvent::Usage {
            prompt_tokens,
            completion_tokens,
            cached_tokens,
        } = ev
        {
            *usage = Some(tr::provider_usage(
                u64::from(*prompt_tokens),
                u64::from(*completion_tokens),
                u64::from(*cached_tokens),
            ));
        }
    }
}

/// Persistent retry posture for the Codex transport, mirroring the Anthropic
/// OAuth overload budget (`OAUTH_OVERLOAD_RETRIES` in `runtime/api.rs`).
///
/// chatgpt.com edge turbulence — 503 upstream connect errors, Cloudflare
/// 520s, raw TCP connect/read timeouts (incident: 2026-07-16 log bursts at
/// 14:19, 18:52, 20:21, 20:27, 20:55, 21:47) — clears on its own within a
/// burst; a three-attempt budget risks aborting an autonomous turn mid-burst.
pub(crate) const CODEX_PERSISTENT_RETRIES: u32 = 10;

/// Exponential-backoff exponent cap shared with the Anthropic stream retry
/// path (`stream_retry_policy`): delay = base·2^min(n−1, CAP). Without the
/// cap a 10-deep budget would sleep 1s·2⁹ = 512s on its final attempt.
const RETRY_BACKOFF_EXP_CAP: u32 = 6;

/// Effective Codex transport retry budget: the persistent floor, unless the
/// user configured something even larger. Pure — decided at the dispatch
/// seam (`try_route`), never inside `send_with_retries`, so tests can still
/// inject exact budgets.
pub(crate) fn codex_retry_budget(configured_retries: u32) -> u32 {
    configured_retries.max(CODEX_PERSISTENT_RETRIES)
}

/// Backoff for retry attempt *n* (1-based): 1s·2^(n−1), capped at 64s.
fn retry_delay(attempt: u32) -> std::time::Duration {
    std::time::Duration::from_millis(
        1000 * 2u64.pow(attempt.saturating_sub(1).min(RETRY_BACKOFF_EXP_CAP)),
    )
}

/// Retrying a response never replays tools: only a complete Ok response reaches
/// the agent tool loop. Reset uncommitted previews, not prior rounds or usage.
async fn wait_stream_retry(
    tx: &mpsc::UnboundedSender<StreamEvent>,
    cancel: &tokio_util::sync::CancellationToken,
    retry: u32,
    max_retries: u32,
    suppress_stream_deltas: bool,
) -> Result<(), super::net::BoxedProviderError> {
    use crate::{LlmEvent, SessionEvent};
    if cancel.is_cancelled() {
        return Err("request canceled".into());
    }
    if !suppress_stream_deltas {
        let _ = tx.send(StreamEvent::Llm(LlmEvent::ResponseReset));
    }
    let _ = tx.send(StreamEvent::Session(SessionEvent::Retry(format!(
        "Response incomplete — retrying response ({retry}/{max_retries}); incomplete output discarded, no tools replayed."
    ))));
    tokio::select! {
        biased;
        _ = cancel.cancelled() => Err("request canceled".into()),
        _ = tokio::time::sleep(retry_delay(retry)) => Ok(()),
    }
}

pub(crate) const STREAM_INTERRUPTED: &str = "openai request failed: connection interrupted before response completed (stream retry budget exhausted)";

/// The endpoint one `send_with_retries` call talks to: its log label, the
/// URL (for the localhost-refusal fast-fail) and whether its 429 responses
/// carry Codex quota semantics worth probing. Static per call site; the
/// retry budget and trace clock are threaded separately because they mutate.
#[derive(Clone, Copy)]
struct SendTarget<'a> {
    label: &'a str,
    url: &'a str,
    /// Codex only: read a bounded slice of a 429 body to distinguish proven
    /// quota exhaustion (fail fast → account failover) from a generic 429
    /// (normal transport retry). Every other endpoint drops the body unread.
    quota_probe: bool,
}

/// Send a provider streaming request, retrying transient failures.
///
/// Parity fix: the Anthropic path retries transient errors with backoff
/// (`call_api_stream_inner`) but the provider routes were single-shot — one
/// transport blip against e.g. `chatgpt.com` aborted an entire autonomous
/// turn (incident: session 20260714-025948-3dab). Attempt *n* sleeps
/// 1s·2^(n−1) capped at 2^`RETRY_BACKOFF_EXP_CAP`, mirroring the Anthropic
/// budget semantics.
///
/// Retryable: transport-level send failures (timeout / connect / request)
/// and 408 / 429 / 5xx responses. Deterministic client errors (other 4xx)
/// and localhost connection refusals (Ollama/LM Studio not running — a
/// setup problem, not a blip) fail fast. Backoff sleeps are cancel-aware.
///
/// Trace (Task 10A): one record per actual HTTP send — every retried
/// failure is recorded via `attempt_failed` (status + retry class, never
/// provider text) and the per-attempt clock restarts right before each
/// re-send. Terminal failures emit their final record here; on success the
/// caller finishes the attempt after consuming the stream.
async fn send_with_retries(
    target: SendTarget<'_>,
    build: impl Fn() -> reqwest::RequestBuilder,
    cancel: &tokio_util::sync::CancellationToken,
    max_retries: u32,
    retries_used: &mut u32,
    trace_attempt: &mut tr::StreamAttempt,
) -> Result<reqwest::Response, Box<dyn std::error::Error + Send + Sync>> {
    use super::account_routing::{
        classify_codex_429, read_body_for_classification, Codex429, CodexQuotaExhausted,
        QUOTA_PROBE_BODY_CAP,
    };
    let SendTarget {
        label,
        url,
        quota_probe,
    } = target;
    loop {
        if cancel.is_cancelled() {
            return Err("request canceled".into());
        }
        let sent = tokio::select! {
            biased;
            _ = cancel.cancelled() => { trace_attempt.finish_canceled(None, None); return Err("request canceled".into()); }
            sent = build().send() => sent,
        };
        match sent {
            Ok(resp) if resp.status().is_success() => {
                trace_attempt.mark_headers();
                return Ok(resp);
            }
            Ok(resp) => {
                // A complete HTTP response was observed even when its status
                // is non-success. Preserve that timing for retried and
                // terminal failure records just as the success branch does.
                trace_attempt.mark_headers();
                let status = resp.status();
                // Provider-assigned request id from validated headers only.
                let trace_rid = tr::provider_request_id_from_headers(resp.headers());
                let mut retryable =
                    status.as_u16() == 408 || status.as_u16() == 429 || status.is_server_error();
                if quota_probe && status.as_u16() == 429 {
                    // Codex only: a 429 that PROVES the account's usage
                    // window is spent must not burn the transport retry
                    // budget (minutes of backoff that cannot succeed). The
                    // body is read into a bounded buffer purely to produce
                    // the classification enum and is dropped unread beyond
                    // that — never stored, surfaced, or logged.
                    let headers = resp.headers().clone();
                    let body = tokio::select! {
                        biased;
                        _ = cancel.cancelled() => {
                            trace_attempt.finish_canceled(Some(429), None);
                            return Err("request canceled".into());
                        }
                        body = read_body_for_classification(resp, QUOTA_PROBE_BODY_CAP) => body,
                    };
                    match classify_codex_429(&headers, &body) {
                        Codex429::QuotaExhausted(evidence) => {
                            tracing::warn!(
                                signal = ?evidence.signal,
                                reset_after_secs = ?evidence.reset_after_secs,
                                "{label} 429 is provider-declared quota exhaustion (no transport retry)"
                            );
                            return Err(Box::new(CodexQuotaExhausted {
                                evidence,
                                request_id: trace_rid,
                            }));
                        }
                        Codex429::UsageNotIncluded => {
                            // Deterministic for this account: retrying never helps.
                            tracing::warn!("{label} 429: account has no Codex usage entitlement");
                            retryable = false;
                        }
                        Codex429::Generic => {}
                    }
                } else if quota_probe && !retryable && status.is_client_error() {
                    // Codex only: a terminal 4xx is undiagnosable without its
                    // body. Read a bounded slice for classification; keep only
                    // identifier-shaped type/code/param plus Synaps-authored
                    // text (codex_wire). Provider prose is never kept (§5.1).
                    let body = tokio::select! {
                        biased;
                        _ = cancel.cancelled() => {
                            trace_attempt.finish_canceled(Some(status.as_u16()), None);
                            return Err("request canceled".into());
                        }
                        body = read_body_for_classification(resp, super::codex_wire::CODEX_ERROR_BODY_CAP) => body,
                    };
                    let detail = super::codex_wire::CodexHttpErrorDetail::from_body(&body);
                    let message = detail.user_message(label, status);
                    tracing::warn!(error = %message, "{label} request rejected (identifiers only)");
                    trace_attempt.finish_failed(
                        &detail.trace_code(status.as_u16()),
                        Some(status.as_u16()),
                        trace_rid,
                    );
                    return Err(Box::new(super::codex_wire::CodexRequestRejected {
                        message,
                        detail,
                    }));
                } else {
                    // Privacy (spec §5.1): the response body is provider-controlled
                    // and may echo the full request (prompts, system text, tool
                    // schemas, credentials). Drop it unread — it must never be
                    // stored, surfaced, or logged at any level. `status` Display
                    // uses the canonical reason phrase, never server bytes. (The
                    // Codex 4xx arm above reads it for identifiers only.)
                    drop(resp);
                }
                let code = format!("http_{}", status.as_u16());
                if !retryable || *retries_used >= max_retries {
                    trace_attempt.finish_failed(&code, Some(status.as_u16()), trace_rid);
                    return Err(format!("{label} request failed: {status}").into());
                }
                *retries_used += 1;
                let delay = retry_delay(*retries_used);
                trace_attempt.attempt_failed(
                    tr::retry_class_for_status(status.as_u16()),
                    delay,
                    Some(status.as_u16()),
                    trace_rid,
                    &code,
                );
                tracing::warn!(
                    "{label} API retry {retries_used}/{max_retries} after {delay:?}: {status}"
                );
                tokio::select! {
                    _ = tokio::time::sleep(delay) => {}
                    _ = cancel.cancelled() => {
                        trace_attempt.finish_canceled(None, None);
                        return Err("request canceled".into());
                    }
                }
                // Retry clocks reset per attempt, right before the re-send.
                trace_attempt.restart_clock();
            }
            Err(e) => {
                let localhost_refusal = e.is_connect() && url.contains("localhost");
                let transient = e.is_timeout() || e.is_connect() || e.is_request();
                if localhost_refusal || !transient || *retries_used >= max_retries {
                    tracing::warn!(
                        "{label} request failed (no retry): {}",
                        crate::core::error::error_chain_string(&e)
                    );
                    trace_attempt.finish_failed("transport_error", None, None);
                    return Err(e.into());
                }
                *retries_used += 1;
                let delay = retry_delay(*retries_used);
                trace_attempt.attempt_failed(
                    if e.is_timeout() {
                        crate::runtime::trace::RetryClass::Timeout
                    } else {
                        crate::runtime::trace::RetryClass::Other
                    },
                    delay,
                    None,
                    None,
                    "transport_error",
                );
                tracing::warn!(
                    "{label} transport retry {retries_used}/{max_retries} after {delay:?}: {}",
                    crate::core::error::error_chain_string(&e)
                );
                tokio::select! {
                    _ = tokio::time::sleep(delay) => {}
                    _ = cancel.cancelled() => {
                        trace_attempt.finish_canceled(None, None);
                        return Err("request canceled".into());
                    }
                }
                // Retry clocks reset per attempt, right before the re-send.
                trace_attempt.restart_clock();
            }
        }
    }
}

/// Run a single streaming request against an OpenAI-compatible endpoint.
///
/// Returns the final assistant response as an Anthropic-shaped content Value
/// (`{"content": [..text.., ..tool_use..]}`) so the outer agent loop can keep
/// using the same handling as the native Anthropic path.
#[allow(clippy::too_many_arguments)]
pub(crate) async fn call_oai_stream_inner(
    cfg: &ProviderConfig,
    broker: &std::sync::Arc<dyn crate::auth::CredentialBroker>,
    tools_schema: &[Value],
    system_prompt: &Option<String>,
    messages: &[crate::SharedMessage],
    tx: &mpsc::UnboundedSender<StreamEvent>,
    temperature: Option<f32>,
    max_tokens: Option<u32>,
    thinking_budget: u32,
    reasoning_level: agent_core::reasoning::ReasoningLevel,
    cancel: &tokio_util::sync::CancellationToken,
    trace: &crate::runtime::trace::TraceContext,
    exact_wire_bytes: bool,
    suppress_stream_deltas: bool,
    max_retries: u32,
) -> Result<Value, Box<dyn std::error::Error + Send + Sync>> {
    let (oai_tools, name_map) = translate::tools_to_oai(tools_schema);
    let oai_messages = translate::messages_to_oai(messages, system_prompt, &name_map);
    let tools_opt = if oai_tools.is_empty() {
        None
    } else {
        Some(oai_tools)
    };

    // Google's OpenAI-compat endpoint rejects stream_options
    let stream_options = if cfg.base_url.contains("googleapis.com") {
        None
    } else {
        Some(StreamOptions {
            include_usage: true,
        })
    };

    let mut body = serde_json::Map::new();
    body.insert("model".to_string(), json!(cfg.model.clone()));
    body.insert("messages".to_string(), serde_json::to_value(oai_messages)?);
    body.insert("stream".to_string(), json!(true));
    // ZDR: OpenAI's Zero Data Retention requires `store:false` on the wire.
    // Chat Completions already defaults to false, but we assert it explicitly so the
    // guarantee is provable in request logs (parity with the Codex Responses path,
    // build_codex_body). Gated to api.openai.com (covers eu.api.openai.com) so we
    // never send an unknown field to OpenAI-compatible providers that strict-validate
    // the body (Groq/xAI/Google), mirroring the googleapis stream_options carve-out.
    if cfg.base_url.contains("api.openai.com") {
        body.insert("store".to_string(), json!(false));
    }
    if let Some(stream_options) = stream_options {
        body.insert(
            "stream_options".to_string(),
            serde_json::to_value(stream_options)?,
        );
    }
    if let Some(max_tokens) = max_tokens {
        body.insert("max_tokens".to_string(), json!(max_tokens));
    }
    if let Some(temperature) = temperature {
        body.insert("temperature".to_string(), json!(temperature));
    }
    if let Some(tools) = tools_opt {
        body.insert("tools".to_string(), serde_json::to_value(tools)?);
    }
    super::reasoning::apply_openai_reasoning_params(
        &mut body,
        super::reasoning::provider_for_key(&cfg.provider),
        &cfg.model,
        thinking_budget,
        reasoning_level,
    );
    let body = Value::Object(body);

    tracing::debug!(provider=%cfg.provider, model=%cfg.model, "openai stream request via broker proxy");

    // Serialize ONCE via the sanctioned constructor: the returned bytes are
    // both digested for the trace and the very buffer stored on the request
    // (`body_bytes`), which `LocalBroker` sends verbatim — the digest is
    // never computed over a re-serialization, and `body`/`body_bytes`
    // cannot diverge.
    let (proxy_request, body_bytes) = crate::auth::ProxyRequest::post_json_exact(
        cfg.provider.clone(),
        "/chat/completions",
        body,
        true,
    )?;
    // ═══ TRACE (Task 10A): one record per actual attempt ═══════════════════
    // A remote broker re-serializes the JSON out of process: honest kind
    // `CloudProxy`, no wire-byte claim. Rule documented in `trace::emit`.
    let tracer = tr::begin_openai_tracer(
        trace,
        &cfg.provider,
        &cfg.model,
        if exact_wire_bytes {
            crate::runtime::trace::TransportKind::OpenAiChatCompletions
        } else {
            crate::runtime::trace::TransportKind::CloudProxy
        },
        &format!("{}/chat/completions", cfg.base_url.trim_end_matches('/')),
        exact_wire_bytes.then_some(body_bytes.as_ref()),
        messages,
        system_prompt.as_deref(),
        tools_schema,
        tr::renamed_tool_losses(&name_map),
    )
    .await;
    // One-shot explicit content capture (`/trace next content`): a no-op in
    // every context without the arm. `body_bytes` is the serialized request
    // body this process built pre-send (exact wire bytes on the local
    // broker; the same body a remote broker re-serializes) — body only,
    // headers and credentials structurally never reach this seam.
    if let Some(tracer) = &tracer {
        tr::capture_request_content(trace, tracer.request_id(), body_bytes.as_ref(), messages);
    }
    let mut attempt = tr::StreamAttempt::new(tracer);

    let mut stream_retry = 0u32;
    loop {
        if cancel.is_cancelled() {
            return Err("request canceled".into());
        }
        // The broker owns the API key and executes/signs the request; this path
        // never resolves or attaches a credential.
        let stream = crate::runtime::api::await_or_cancel(
            cancel,
            broker.proxy_stream(proxy_request.clone()),
        )
        .await
        .map_err(|_| "request canceled")?
        // Privacy (spec §5.1): a broker proxy error may carry an upstream
        // response-body snippet; redact it to status-only before surfacing.
        .map_err(|e| {
            format!(
                "openai request failed: {}",
                super::net::redact_provider_proxy_error(&e.to_string())
            )
        });
        let mut stream = match stream {
            Ok(stream) => {
                attempt.mark_headers();
                stream
            }
            Err(msg) => {
                // Status parsed from the redacted static-prefix message only;
                // codes are `http_<status>` or the static `broker_error`.
                let status = tr::broker_error_status(&msg);
                let code =
                    status.map_or_else(|| "broker_error".to_string(), |s| format!("http_{s}"));
                attempt.finish_failed(&code, status, None);
                return Err(msg.into());
            }
        };

        if !suppress_stream_deltas {
            let _ = tx.send(StreamEvent::Llm(crate::LlmEvent::ResponseStart));
        }
        let mut interrupted = false;
        let mut saw_done = false;
        let mut decoder = StreamDecoder::new();
        let mut accumulated_text = String::new();
        let mut tool_use_blocks: Vec<Value> = Vec::new();
        let mut buf = bytes::BytesMut::with_capacity(8 * 1024);
        let mut sink: Vec<OaiEvent> = Vec::with_capacity(4);
        let mut trace_usage: Option<crate::runtime::trace::UsageMeta> = None;

        while let Some(chunk) = tokio::select! {
            biased;
            _ = cancel.cancelled() => {
                attempt.finish_canceled(None, trace_usage);
                return Err("request canceled".into());
            }
            chunk = stream.next() => chunk,
        } {
            let chunk = match chunk {
                Ok(chunk) => {
                    attempt.mark_first_byte();
                    chunk
                }
                Err(e) => {
                    if !matches!(e, crate::auth::BrokerError::Transport(_)) {
                        attempt.finish_failed("broker_error", None, None);
                        return Err(e.into());
                    }
                    interrupted = !saw_done && decoder.finish_reason.is_none();
                    break;
                }
            };
            buf.extend_from_slice(&chunk);

            // Scan for newline-delimited SSE lines (SIMD-accelerated via memchr)
            while let Some(nl) = memchr::memchr(b'\n', &buf) {
                let line_bytes = buf.split_to(nl + 1); // O(1) — ref-counted split
                let line = std::str::from_utf8(&line_bytes[..nl]).unwrap_or("");

                sink.clear();
                decoder.push_line(line, &mut sink);
                saw_done |= sink.iter().any(|e| matches!(e, OaiEvent::Done));
                capture_oai_trace_signals(&sink, &mut attempt, &mut trace_usage);
                handle_events(
                    &sink,
                    tx,
                    &mut accumulated_text,
                    &mut tool_use_blocks,
                    &name_map,
                    suppress_stream_deltas,
                );
            }
        }

        // Flush any remaining buffered line + final Done
        if !interrupted && !buf.is_empty() {
            let line = std::str::from_utf8(&buf).unwrap_or("");
            sink.clear();
            decoder.push_line(line, &mut sink);
            saw_done |= sink.iter().any(|e| matches!(e, OaiEvent::Done));
            capture_oai_trace_signals(&sink, &mut attempt, &mut trace_usage);
            handle_events(
                &sink,
                tx,
                &mut accumulated_text,
                &mut tool_use_blocks,
                &name_map,
                suppress_stream_deltas,
            );
        }
        if interrupted || (!saw_done && decoder.finish_reason.is_none()) {
            if stream_retry >= max_retries {
                attempt.finish_failed("stream_error", None, None);
                return Err(STREAM_INTERRUPTED.into());
            }
            stream_retry += 1;
            attempt.attempt_failed(
                crate::runtime::trace::RetryClass::Network,
                retry_delay(stream_retry),
                None,
                None,
                "stream_error",
            );
            wait_stream_retry(
                tx,
                cancel,
                stream_retry,
                max_retries,
                suppress_stream_deltas,
            )
            .await
            .inspect_err(|_| {
                attempt.finish_canceled(None, None);
            })?;
            attempt.restart_clock();
            continue;
        }
        sink.clear();
        decoder.finish(&mut sink);
        capture_oai_trace_signals(&sink, &mut attempt, &mut trace_usage);
        handle_events(
            &sink,
            tx,
            &mut accumulated_text,
            &mut tool_use_blocks,
            &name_map,
            suppress_stream_deltas,
        );

        // Normalized stop reason: only when a finish_reason was actually
        // observed on the wire — never inferred.
        let stop_reason = decoder
            .finish_reason
            .as_deref()
            .map(tr::stop_reason_from_finish_reason);
        // Broker paths never observe the upstream HTTP status: honest `None`.
        attempt.finish_success(None, None, stop_reason, trace_usage);

        // Build Anthropic-shaped final response
        let mut content: Vec<Value> = Vec::new();
        if !accumulated_text.is_empty() {
            content.push(json!({"type": "text", "text": accumulated_text}));
        }
        content.extend(tool_use_blocks);

        return Ok(json!({
            "role": "assistant",
            "content": content,
        }));
    }
}

#[allow(clippy::too_many_arguments)]
pub(crate) async fn call_codex_stream_inner(
    cfg: &ProviderConfig,
    client: &reqwest::Client,
    broker: &std::sync::Arc<dyn crate::auth::CredentialBroker>,
    tools_schema: &[Value],
    system_prompt: &Option<String>,
    messages: &[crate::SharedMessage],
    tx: &mpsc::UnboundedSender<StreamEvent>,
    temperature: Option<f32>,
    max_tokens: Option<u32>,
    reasoning_level: agent_core::reasoning::ReasoningLevel,
    codex_request_role: crate::runtime::openai::catalog::CodexRequestRole,
    cancel: &tokio_util::sync::CancellationToken,
    max_retries: u32,
    trace: &crate::runtime::trace::TraceContext,
) -> Result<Value, Box<dyn std::error::Error + Send + Sync>> {
    // Account selection crosses the broker boundary through the router seam:
    // the broker vends one (credential, access token) pair, never a refresh
    // token, and this path never opens auth.json.
    let router = super::account_routing::BrokerCodexRouter::new(std::sync::Arc::clone(broker));
    call_codex_stream_with_router(
        cfg,
        client,
        &router,
        tools_schema,
        system_prompt,
        messages,
        tx,
        temperature,
        max_tokens,
        reasoning_level,
        codex_request_role,
        cancel,
        max_retries,
        trace,
    )
    .await
}

/// What the failover gate judges: the recognized exhaustion on the current
/// attempt and the two runtime facts that can veto a switch. Built at each
/// of the two detection sites (429 probe, in-stream terminal error) so both
/// go through the identical invariant check.
#[derive(Clone, Copy)]
struct FailoverRequest<'a> {
    evidence: &'a super::account_routing::CodexQuotaEvidence,
    /// Any content-bearing model event reached the decoder on this attempt
    /// (text, tool, reasoning — forwarded or not). Always false on the 429
    /// path: no response body was streamed.
    output_started: bool,
    /// Account switches already spent on this logical request.
    prior_failovers: u32,
}

/// Attempt a single bounded account failover after recognized quota
/// exhaustion on `failed`. Always reports the exhaustion to the router
/// (cooldown sink, best effort) — even when the runtime then refuses to
/// switch — so the broker stops advertising the seat. Returns the new pinned
/// account only when every runtime invariant holds: `Auto` selection, no
/// output produced, budget unspent, a candidate the router deems fresh and
/// eligible, and that candidate is a DIFFERENT account than the one that
/// failed. Anything else → `Ok(None)` and the caller ends the turn with the
/// static quota message. Never loops.
async fn try_codex_account_failover(
    router: &dyn super::account_routing::CodexAccountRouter,
    model: &str,
    failed: &super::account_routing::PinnedAccount,
    request: FailoverRequest<'_>,
    tx: &mpsc::UnboundedSender<StreamEvent>,
    cancel: &tokio_util::sync::CancellationToken,
) -> Result<Option<super::account_routing::PinnedAccount>, super::net::BoxedProviderError> {
    use super::account_routing::failover_gate;
    use crate::runtime::api::await_or_cancel;
    let FailoverRequest {
        evidence,
        output_started,
        prior_failovers,
    } = request;
    await_or_cancel(cancel, router.report_exhausted(failed, evidence))
        .await
        .map_err(|_| "request canceled")?;
    if let Err(blocked) = failover_gate(failed.auto, output_started, prior_failovers) {
        tracing::info!(
            account = %failed.credential,
            reason = blocked.as_str(),
            "codex quota exhausted; account failover not permitted"
        );
        return Ok(None);
    }
    let candidate = await_or_cancel(
        cancel,
        router.failover(model, failed, evidence, prior_failovers),
    )
    .await
    .map_err(|_| "request canceled")?;
    match candidate {
        Ok(Some(next)) if next.credential != failed.credential => {
            tracing::warn!(
                from = %failed.credential,
                to = %next.credential,
                "codex quota exhausted; switching account (one-time failover, no output replayed)"
            );
            let _ = tx.send(StreamEvent::Session(crate::SessionEvent::Notice(format!(
                "Codex usage quota exhausted on account '{}' — switched to account '{}' (one-time failover; no output was replayed).",
                failed.label(),
                next.label()
            ))));
            Ok(Some(next))
        }
        Ok(Some(_same)) => {
            tracing::warn!(
                account = %failed.credential,
                "codex failover candidate is the exhausted account itself; refusing to retry"
            );
            Ok(None)
        }
        Ok(None) => {
            tracing::info!(
                account = %failed.credential,
                "codex quota exhausted; no distinct account with fresh capacity"
            );
            Ok(None)
        }
        Err(e) => {
            tracing::warn!(
                account = %failed.credential,
                error = %e,
                "codex failover candidate resolution failed; not retrying"
            );
            Ok(None)
        }
    }
}

#[allow(clippy::too_many_arguments)]
pub(crate) async fn call_codex_stream_with_router(
    cfg: &ProviderConfig,
    client: &reqwest::Client,
    router: &dyn super::account_routing::CodexAccountRouter,
    tools_schema: &[Value],
    system_prompt: &Option<String>,
    messages: &[crate::SharedMessage],
    tx: &mpsc::UnboundedSender<StreamEvent>,
    temperature: Option<f32>,
    max_tokens: Option<u32>,
    reasoning_level: agent_core::reasoning::ReasoningLevel,
    codex_request_role: crate::runtime::openai::catalog::CodexRequestRole,
    cancel: &tokio_util::sync::CancellationToken,
    max_retries: u32,
    trace: &crate::runtime::trace::TraceContext,
) -> Result<Value, Box<dyn std::error::Error + Send + Sync>> {
    // Build the exact provider-qualified plan before any credential or network
    // access. Logical Ultra is lowered here, never in the generic level enum.
    use crate::runtime::openai::catalog::plan_codex_execution;
    let qualified_model = format!("{}/{}", cfg.provider, cfg.model);
    let plan =
        match plan_codex_execution(&qualified_model, reasoning_level, codex_request_role, None) {
            Ok(plan) => plan,
            Err(error) => {
                tracing::debug!(
                    event = "codex_mode_plan",
                    qualified_model = %qualified_model,
                    requested_level = %reasoning_level,
                    runtime_role = codex_request_role.as_str(),
                    decision = "deny",
                    deny_code = error.code().as_str(),
                    network_attempted = false,
                    "Codex execution plan denied"
                );
                return Err(Box::new(error));
            }
        };
    if plan.automatic_delegation() && !codex_has_required_delegation_tools(tools_schema) {
        tracing::debug!(
            event = "codex_mode_plan",
            qualified_model = %qualified_model,
            requested_level = %reasoning_level,
            runtime_role = codex_request_role.as_str(),
            decision = "deny",
            deny_code = "ultra_requires_subagent_tools",
            network_attempted = false,
            "Codex execution plan denied"
        );
        return Err(
            "Ultra requires subagent_start, subagent_status, and subagent_collect tools".into(),
        );
    }
    tracing::debug!(
        event = "codex_mode_plan",
        qualified_model = %plan.qualified_model,
        requested_level = %plan.selected_level,
        execution_mode = plan.mode.as_str(),
        wire_effort = plan.wire_effort_label(),
        capability_source = plan.capability_source.map_or("none", |source| source.as_str()),
        multi_agent_version = plan.multi_agent_version_label(),
        runtime_role = plan.request_role.as_str(),
        multi_agent_mode = plan.multi_agent_mode_label(),
        automatic_delegation = plan.automatic_delegation(),
        decision = "allow",
        network_attempted = false,
        "Codex execution plan allowed"
    );
    // Exactly one pinned (account, token) pair per attempt. The
    // `chatgpt-account-id` header is derived from THIS token's JWT claims, so
    // bearer and header always name the same credential — a second lookup
    // could race a policy change and pair account A's header with account
    // B's token.
    let mut pinned = crate::runtime::api::await_or_cancel(cancel, router.pin(&cfg.model))
        .await
        .map_err(|_| "request canceled")??;
    let mut account_id = codex_account_id_for(&pinned)?;

    let (tools, name_map) = translate::tools_to_responses(tools_schema);
    // Opt-in OpenAI-hosted web search (SYNAPS_CODEX_WEB_SEARCH, read from THIS
    // process's env — the daemon's when attached). Runs server-side, so hooks
    // can't gate it; gated by role and model (codex_wire). The session actor
    // announces the state; each call is announced by the decoder.
    let mut tools = tools;
    let web_search = super::codex_wire::codex_web_search_for_request(
        &cfg.model,
        codex_request_role,
        super::codex_wire::engine_env,
    );
    // web_search is always the last tool. Cleared on the one resend without
    // it, so that resend can't loop; the rejection is learned only if the
    // resend then succeeds (`web_search_pending`).
    let mut web_search_index = web_search.as_ref().map(|_| tools.len());
    let mut web_search_pending = false;
    tools.extend(web_search);
    // Codex already receives the effective system prompt in `instructions`.
    // Do not also prepend it to `input`: that doubles prompt tokens and repeats
    // context/checkpoint guidance. Pass history to the translator unchanged;
    // only the synthetic prepend is suppressed. Other providers keep their placement.
    let oai_messages = translate::messages_to_oai(messages, &None, &name_map);

    // Use the shared pure helper so production and unit tests exercise identical
    // body construction — no duplication between call_codex_stream_inner and tests.
    let instructions = codex_instructions(system_prompt);
    let mut input_items = codex_input_messages(oai_messages);
    // Key on the pre-insertion head: the mode item varies with the execution
    // plan (level/role switches) and must not churn the routing key.
    let prompt_cache_key = codex_prompt_cache_key(&instructions, input_items.first());
    insert_codex_multi_agent_mode(&mut input_items, &plan);
    let input = serde_json::Value::Array(input_items);
    let mut body = build_codex_body(
        &cfg.model,
        &plan,
        input,
        instructions,
        tools,
        temperature,
        max_tokens,
        &prompt_cache_key,
    );

    let url = format!(
        "{}/codex/responses",
        cfg.base_url
            .trim_end_matches('/')
            .trim_end_matches("/codex")
    );
    tracing::debug!(url=%url, model=%cfg.model, "codex stream request");

    // Serialize ONCE. These exact bytes are digested for the trace AND are
    // the request body of every attempt — retries resend the identical
    // buffer, so one wire digest is honest for all attempt records. The one
    // exception (hosted web_search rejected) re-serializes and opens a new
    // trace record for the new bytes.
    let mut body_bytes = bytes::Bytes::from(serde_json::to_vec(&body)?);
    // ═══ TRACE (Task 10A): one record per actual attempt ═══════════════════
    let tracer = tr::begin_openai_tracer(
        trace,
        &cfg.provider,
        &cfg.model,
        crate::runtime::trace::TransportKind::OpenAiResponses,
        &url,
        Some(body_bytes.as_ref()),
        messages,
        system_prompt.as_deref(),
        tools_schema,
        tr::renamed_tool_losses(&name_map),
    )
    .await;
    // One-shot explicit content capture (`/trace next content`): a no-op in
    // every context without the arm. `body_bytes` is the serialized request
    // body this process built pre-send (exact wire bytes on the local
    // broker; the same body a remote broker re-serializes) — body only,
    // headers and credentials structurally never reach this seam.
    if let Some(tracer) = &tracer {
        tr::capture_request_content(trace, tracer.request_id(), body_bytes.as_ref(), messages);
    }
    // Whether this request holds a tracer (and, on a `/trace next` context,
    // the one-shot gate): only then may a web_search resend reopen it.
    let traced = tracer.is_some();
    let mut attempt = tr::StreamAttempt::new(tracer);
    let mut stream_retry = 0u32;
    // Account switches spent on this logical request. Bounded by
    // `MAX_CODEX_ACCOUNT_FAILOVERS`; independent of the transport retry
    // budget (a failover neither consumes nor resets `stream_retry`).
    let mut failovers = 0u32;

    loop {
        if cancel.is_cancelled() {
            return Err("request canceled".into());
        }
        let sent = send_with_retries(
            SendTarget {
                label: "codex",
                url: &url,
                quota_probe: true,
            },
            || {
                client
                    .post(&url)
                    .bearer_auth(&pinned.token)
                    .header("chatgpt-account-id", account_id.as_str())
                    .header("originator", "synaps")
                    .header("OpenAI-Beta", "responses=experimental")
                    // Cache affinity: the ChatGPT backend routes prompt-cache
                    // lookups by the `session-id` header (Codex CLI sends its
                    // conversation id as both headers and as
                    // `prompt_cache_key`). Without it, every request reported
                    // cached_tokens = 0, even with a byte-identical prefix.
                    .header("session-id", prompt_cache_key.as_str())
                    .header("thread-id", prompt_cache_key.as_str())
                    .header("content-type", "application/json")
                    .header("accept", "text/event-stream")
                    .body(body_bytes.clone())
            },
            cancel,
            max_retries,
            &mut stream_retry,
            &mut attempt,
        )
        .await;
        let resp = match sent {
            Ok(resp) => {
                if std::mem::take(&mut web_search_pending) {
                    super::codex_wire::confirm_web_search_rejected(&cfg.model, tx);
                }
                resp
            }
            Err(error) => match error.downcast::<super::account_routing::CodexQuotaExhausted>() {
                // A 429 that proved exhaustion arrives before any response
                // body was streamed: nothing was produced on this attempt, so
                // the pre-output invariant holds by construction.
                Ok(quota) => {
                    match try_codex_account_failover(
                        router,
                        &cfg.model,
                        &pinned,
                        FailoverRequest {
                            evidence: &quota.evidence,
                            output_started: false,
                            prior_failovers: failovers,
                        },
                        tx,
                        cancel,
                    )
                    .await?
                    {
                        Some(next) => {
                            attempt.attempt_failed(
                                crate::runtime::trace::RetryClass::RateLimited,
                                std::time::Duration::ZERO,
                                Some(429),
                                quota.request_id.clone(),
                                "account_failover",
                            );
                            account_id = codex_account_id_for(&next)?;
                            pinned = next;
                            failovers += 1;
                            attempt.restart_clock();
                            continue;
                        }
                        None => {
                            let message = quota.to_string();
                            attempt.finish_failed("quota_exhausted", Some(429), quota.request_id);
                            return Err(message.into());
                        }
                    }
                }
                Err(other) => {
                    // A pre-inference 400 pointing at hosted web_search: drop
                    // it, re-serialize, and resend once under a fresh trace
                    // record (one-shot `/trace next` gates are reopened for
                    // it; content capture is not redone).
                    super::codex_wire::hosted_search_retry(
                        other,
                        &mut web_search_index,
                        &cfg.model,
                    )?;
                    web_search_pending = true;
                    super::codex_wire::strip_web_search_tool(&mut body);
                    body_bytes = bytes::Bytes::from(serde_json::to_vec(&body)?);
                    if traced {
                        trace.reopen_one_shot_for_resend();
                    }
                    attempt = tr::StreamAttempt::new(
                        tr::begin_openai_tracer(
                            trace,
                            &cfg.provider,
                            &cfg.model,
                            crate::runtime::trace::TransportKind::OpenAiResponses,
                            &url,
                            Some(body_bytes.as_ref()),
                            messages,
                            system_prompt.as_deref(),
                            tools_schema,
                            tr::renamed_tool_losses(&name_map),
                        )
                        .await,
                    );
                    continue;
                }
            },
        };
        // Direct HTTP: upstream status and provider request id are observed.
        let http_status = Some(resp.status().as_u16());
        let trace_rid = tr::provider_request_id_from_headers(resp.headers());

        let _ = tx.send(StreamEvent::Llm(crate::LlmEvent::ResponseStart));
        let mut interrupted = false;
        let mut accumulated_text = String::new();
        let mut parser = CodexSseDecoder::default();
        let mut buf = bytes::BytesMut::with_capacity(8 * 1024);
        let mut stream = resp.bytes_stream();

        while let Some(chunk) = tokio::select! {
            biased;
            _ = cancel.cancelled() => {
                attempt.finish_canceled(http_status, parser.trace_usage());
                return Err("request canceled".into());
            }
            chunk = stream.next() => chunk,
        } {
            let chunk = match chunk {
                Ok(chunk) => {
                    attempt.mark_first_byte();
                    chunk
                }
                Err(_) => {
                    interrupted = !parser.terminal_success && parser.terminal_failure.is_none();
                    break;
                }
            };
            buf.extend_from_slice(&chunk);
            while let Some(nl) = memchr::memchr(b'\n', &buf) {
                let line_bytes = buf.split_to(nl + 1);
                let line = std::str::from_utf8(&line_bytes[..nl]).unwrap_or("");
                parser.push_line(line, tx, &mut accumulated_text);
            }
            if parser.saw_model_event {
                attempt.mark_first_model_event();
            }
        }
        if !interrupted && !buf.is_empty() {
            let line = std::str::from_utf8(&buf).unwrap_or("");
            parser.push_line(line, tx, &mut accumulated_text);
        }
        if interrupted || (!parser.terminal_success && parser.terminal_failure.is_none()) {
            if stream_retry >= max_retries {
                attempt.finish_failed("stream_error", http_status, trace_rid);
                return Err(format!("Codex{RESPONSES_MISSING_TERMINAL_SUFFIX}").into());
            }
            stream_retry += 1;
            attempt.attempt_failed(
                crate::runtime::trace::RetryClass::Network,
                retry_delay(stream_retry),
                http_status,
                trace_rid,
                "stream_error",
            );
            wait_stream_retry(tx, cancel, stream_retry, max_retries, false)
                .await
                .inspect_err(|_| {
                    attempt.finish_canceled(None, None);
                })?;
            attempt.restart_clock();
            continue;
        }
        parser.finish();
        // Trailing-buffer flush and finish() can surface the first (or only)
        // model event — e.g. a payload dispatched by the tail line. Mark it so
        // first_model_event_ms is honest even for tail-only streams.
        if parser.saw_model_event {
            attempt.mark_first_model_event();
        }
        if let Err(failure) = parser.terminal_result() {
            if failure.code == "responses_quota" {
                // In-stream quota exhaustion (HTTP 200 + terminal quota
                // error). Failover only if the model produced nothing on this
                // attempt — text, tool, reasoning or any content-bearing
                // event makes the failure terminal (no cross-account replay).
                let output_started =
                    !accumulated_text.is_empty() || parser.model_produced_output();
                let evidence = super::account_routing::CodexQuotaEvidence {
                    signal: super::account_routing::QuotaSignal::StreamQuotaEvent,
                    reset_after_secs: None,
                };
                match try_codex_account_failover(
                    router,
                    &cfg.model,
                    &pinned,
                    FailoverRequest {
                        evidence: &evidence,
                        output_started,
                        prior_failovers: failovers,
                    },
                    tx,
                    cancel,
                )
                .await?
                {
                    Some(next) => {
                        attempt.attempt_failed(
                            crate::runtime::trace::RetryClass::RateLimited,
                            std::time::Duration::ZERO,
                            http_status,
                            trace_rid,
                            "account_failover",
                        );
                        account_id = codex_account_id_for(&next)?;
                        pinned = next;
                        failovers += 1;
                        attempt.restart_clock();
                        continue;
                    }
                    None => {
                        attempt.finish_failed(failure.code, http_status, trace_rid);
                        return Err(failure.message.into());
                    }
                }
            }
            if failure.code == "responses_empty" && stream_retry < max_retries {
                stream_retry += 1;
                let delay = retry_delay(stream_retry);
                attempt.attempt_failed(
                    crate::runtime::trace::RetryClass::Other,
                    delay,
                    http_status,
                    trace_rid,
                    failure.code,
                );
                tracing::warn!(
                    "codex stream retry {stream_retry}/{max_retries} after {delay:?}: completed without usable output"
                );
                wait_stream_retry(tx, cancel, stream_retry, max_retries, false)
                    .await
                    .inspect_err(|_| {
                        attempt.finish_canceled(None, None);
                    })?;
                attempt.restart_clock();
                continue;
            }
            attempt.finish_failed(failure.code, http_status, trace_rid);
            return Err(failure.message.into());
        }
        attempt.finish_success(http_status, trace_rid, None, parser.trace_usage());

        let mut content: Vec<Value> = Vec::new();
        if !accumulated_text.is_empty() {
            content.push(json!({"type": "text", "text": accumulated_text}));
        }
        content.extend(translate::tool_calls_to_content_blocks(
            &parser.completed_tools,
            &name_map,
        ));

        return Ok(json!({
            "role": "assistant",
            "content": content,
        }));
    }
}

/// The `chatgpt-account-id` header value for a pinned account: provider-owned
/// metadata carried inside THAT token's JWT. Deriving it from the same bearer
/// the request will present is what keeps header and token paired.
fn codex_account_id_for(
    pinned: &super::account_routing::PinnedAccount,
) -> Result<String, super::net::BoxedProviderError> {
    crate::auth::extract_codex_account_id(&pinned.token)
        .ok_or_else(|| {
            "Failed to extract ChatGPT account id from Codex token — run `synaps login --provider openai-codex`".into()
        })
}

/// Stable prompt-cache routing key for this conversation.
///
/// Sent as `prompt_cache_key` in the body AND as the `session-id` /
/// `thread-id` headers: the ChatGPT backend derives cache affinity from the
/// `session-id` header (upstream codex-rs sends its conversation UUID in all
/// three places). The body key alone yielded zero cache hits. Synaps
/// doesn't thread a session id down to the transport, so derive a
/// deterministic key from the stable head of the prompt: the instructions
/// plus the first input item. Identical heads hash to the same key — which
/// is exactly right, since those requests share the very prefix the cache
/// stores — and a /compact rewrite changes the key together with the prefix
/// it invalidates. Without this key, cache lookups are at the mercy of
/// load-balancer routing and long conversations re-pay their full prefix.
pub(crate) fn codex_prompt_cache_key(instructions: &str, first_item: Option<&Value>) -> String {
    use sha2::{Digest, Sha256};
    let mut hasher = Sha256::new();
    hasher.update(instructions.as_bytes());
    if let Some(item) = first_item {
        hasher.update(item.to_string().as_bytes());
    }
    format!("synaps-{:x}", hasher.finalize())
}

/// Pure body construction for Codex Responses-API requests.
///
/// Separated from the async function so it can be unit-tested without
/// any credential access or network I/O.
///
/// # Arguments
/// - `model`: the Codex model id (already validated by the caller)
/// - `plan`: reasoning plan (must already be validated against model capability)
/// - `input`: pre-built input items (from `codex_input_messages`)
/// - `instructions`: pre-built instructions string (from `codex_instructions`)
/// - `tools`: pre-built tool array
/// - `temperature`: optional temperature override
/// - `max_tokens`: optional max_output_tokens override
/// - `prompt_cache_key`: pre-computed routing key; computed on the
///   pre-insertion input head by the caller, so it cannot be derived here
#[allow(clippy::too_many_arguments)] // pure builder mirroring the wire body; grouping would obscure the 1:1 field mapping
pub(crate) fn build_codex_body(
    model: &str,
    plan: &crate::runtime::openai::catalog::CodexExecutionPlan,
    input: serde_json::Value,
    instructions: String,
    tools: Vec<serde_json::Value>,
    temperature: Option<f32>,
    max_tokens: Option<u32>,
    prompt_cache_key: &str,
) -> serde_json::Value {
    let mut body = json!({
        "model": model,
        "store": false,
        "stream": true,
        "instructions": instructions,
        "input": input,
        "tool_choice": "auto",
        "parallel_tool_calls": true,
        "include": ["reasoning.encrypted_content"],
        "text": { "verbosity": "medium" },
        "prompt_cache_key": prompt_cache_key,
    });
    if let Some(effort) = plan.wire_effort {
        body["reasoning"] = json!({ "effort": effort.as_str() });
    }
    if !tools.is_empty() {
        body["tools"] = serde_json::Value::Array(tools);
    }
    if let Some(temp) = temperature {
        body["temperature"] = json!(temp);
    }
    if let Some(max) = max_tokens {
        body["max_output_tokens"] = json!(max);
    }
    body
}

const CODEX_MULTI_AGENT_MODE_OPEN: &str = "<multi_agent_mode>";
const CODEX_MULTI_AGENT_MODE_CLOSE: &str = "</multi_agent_mode>";
const CODEX_EXPLICIT_REQUEST_ONLY_TEXT: &str = "Do not spawn sub-agents unless the user or applicable AGENTS.md/skill instructions explicitly ask for sub-agents, delegation, or parallel agent work.";
const CODEX_PROACTIVE_MULTI_AGENT_TEXT: &str = "Proactive multi-agent delegation is active. Any earlier instruction requiring an explicit user request before spawning sub-agents no longer applies. Use sub-agents when parallel work would materially improve speed or quality. This mode remains active until a later multi-agent mode developer message changes it.";

/// Environment switch for the OpenAI-hosted `web_search` tool on Codex
/// Responses requests.
pub(crate) const CODEX_WEB_SEARCH_ENV: &str = "SYNAPS_CODEX_WEB_SEARCH";

/// Map the `SYNAPS_CODEX_WEB_SEARCH` value to the hosted tool spec:
/// - `live` / `on` / `true` / `1` → `{"type":"web_search"}` (live web access)
/// - `cached` → `{"type":"web_search","external_web_access":false}`
/// - anything else (unset, `off`, typos) → no hosted tool.
///
/// The hosted tool's `web_search_call` output items are not function calls:
/// they never become tool uses; each completed one is announced as a notice
/// with its query (`codex_wire::announce_web_search_call`).
pub(crate) fn codex_web_search_tool(value: Option<&str>) -> Option<Value> {
    match value.map(|v| v.trim().to_ascii_lowercase()).as_deref() {
        Some("live" | "on" | "true" | "1" | "yes") => Some(json!({ "type": "web_search" })),
        Some("cached") => Some(json!({ "type": "web_search", "external_web_access": false })),
        _ => None,
    }
}

fn codex_multi_agent_mode_item(
    plan: &crate::runtime::openai::catalog::CodexExecutionPlan,
) -> Option<Value> {
    use crate::runtime::openai::catalog::CodexMultiAgentMode;
    let body = match plan.multi_agent_mode? {
        CodexMultiAgentMode::ExplicitRequestOnly => CODEX_EXPLICIT_REQUEST_ONLY_TEXT,
        CodexMultiAgentMode::Proactive => CODEX_PROACTIVE_MULTI_AGENT_TEXT,
    };
    Some(json!({
        "type": "message",
        "role": "developer",
        "content": [{
            "type": "input_text",
            "text": format!("{CODEX_MULTI_AGENT_MODE_OPEN}{body}{CODEX_MULTI_AGENT_MODE_CLOSE}"),
        }],
    }))
}

fn is_codex_multi_agent_mode_item(item: &Value) -> bool {
    item.pointer("/content/0/text")
        .and_then(Value::as_str)
        .is_some_and(|text| {
            text.trim_start().starts_with(CODEX_MULTI_AGENT_MODE_OPEN)
                && text.trim_end().ends_with(CODEX_MULTI_AGENT_MODE_CLOSE)
        })
}

fn insert_codex_multi_agent_mode(
    input: &mut Vec<Value>,
    plan: &crate::runtime::openai::catalog::CodexExecutionPlan,
) {
    input.retain(|item| !is_codex_multi_agent_mode_item(item));
    let Some(item) = codex_multi_agent_mode_item(plan) else {
        return;
    };
    // Prefix-cache stability: the mode item goes at a FIXED position (head of
    // input), never relative to the latest user message. The previous
    // placement (before the last user item) moved every user turn, so the
    // prompt prefix diverged at the prior turn's user message and the entire
    // preceding agentic turn — often the bulk of the context — was re-billed
    // uncached (incident: 2026-07-16 codex cache investigation). At index 0
    // the item is byte-stable across turns; it only changes when the
    // execution plan's mode changes, which legitimately invalidates the
    // prefix once.
    input.insert(0, item);
}

fn codex_has_required_delegation_tools(tools_schema: &[Value]) -> bool {
    ["subagent_start", "subagent_status", "subagent_collect"]
        .into_iter()
        .all(|required| {
            tools_schema.iter().any(|tool| {
                tool.get("name").and_then(Value::as_str) == Some(required)
                    || tool.pointer("/function/name").and_then(Value::as_str) == Some(required)
            })
        })
}

const CODEX_AUTONOMOUS_LOOP_POLICY: &str = "\n\n[Synaps autonomous harness policy]\nThis harness is non-interactive after the user has provided the task/spec. Do not stop at phase boundaries, milestones, checkpoints, or after presenting a plan unless the full requested job is complete. Do not ask the user whether to continue. When a phase/checkpoint is reached, run any relevant verification and continue autonomously until the full requested job is complete, blocked by an unrecoverable error, or explicit user instructions require stopping.\n[End Synaps autonomous harness policy]";

fn codex_instructions(system_prompt: &Option<String>) -> String {
    let mut instructions = system_prompt.clone().unwrap_or_default();
    if instructions.contains("[Synaps autonomous harness policy]") {
        return instructions;
    }
    instructions.push_str(CODEX_AUTONOMOUS_LOOP_POLICY);
    instructions
}

fn codex_input_messages(messages: Vec<ChatMessage>) -> Vec<Value> {
    responses_input_messages(messages, false)
}

/// Meta Muse input: like [`codex_input_messages`], but replayed assistant
/// text carries the Codex-style message `phase` Muse itself emits.
///
/// Muse labels a preamble before tool calls `phase:"commentary"` and a
/// turn-ending answer `phase:"final_answer"` (or no phase). Replaying every
/// past assistant message as a phase-less string taught the model that its
/// "final answers" go on to call tools, and it then ended live turns on a
/// bare preamble (A/B replay of the failing history: 2/16 runs reproduced
/// with phase-less replay, 0/16 with phase restored).
///
/// Session storage keeps no phase, so it is inferred from structure, which
/// matches what Muse emits: `messages_to_oai` folds an assistant turn's text
/// into ONE message that also carries the turn's calls, so text on a message
/// WITH tool calls is commentary, text on a message without them is the
/// final answer. Tool-only messages and the whitespace separator
/// `messages_to_oai` inserts between tool results and plain user text are
/// harness wire artefacts, not model text: they keep the legacy shape.
fn muse_input_messages(messages: Vec<ChatMessage>) -> Vec<Value> {
    responses_input_messages(messages, true)
}

/// Wire phase Muse uses for a replayed assistant message's text.
fn muse_assistant_phase(has_tool_calls: bool) -> &'static str {
    if has_tool_calls {
        "commentary"
    } else {
        "final_answer"
    }
}

fn responses_input_messages(messages: Vec<ChatMessage>, assistant_phase: bool) -> Vec<Value> {
    let mut out = Vec::new();
    for msg in messages {
        if msg.role == "tool" {
            // Skip tool results with no call_id — sending an empty call_id
            // to the Codex API would cause a 400 with a confusing error.
            if let Some(call_id) = msg.tool_call_id {
                out.push(json!({
                    "type": "function_call_output",
                    "call_id": call_id,
                    "output": msg.content.unwrap_or_default(),
                }));
            }
            continue;
        }
        // A mixed assistant message carries BOTH text and calls. Emit its
        // text first rather than continuing after the calls and losing it.
        // Text-only messages retain the existing string wire representation.
        let phased_text =
            if assistant_phase && msg.role == "assistant" && msg.content_parts.is_none() {
                msg.content
                    .as_deref()
                    .filter(|text| !text.trim().is_empty())
            } else {
                None
            };
        if let Some(text) = phased_text {
            out.push(json!({
                "type": "message",
                "role": "assistant",
                "phase": muse_assistant_phase(msg.tool_calls.as_ref().is_some_and(|c| !c.is_empty())),
                "content": [{"type": "output_text", "text": text}],
            }));
        } else if msg.content_parts.is_some() || msg.content.is_some() || msg.tool_calls.is_none() {
            let content = if let Some(parts) = msg.content_parts {
                Value::Array(parts.into_iter().map(|part| match part {
                    ChatContentPart::Text { text } => json!({
                        "type": if msg.role == "assistant" { "output_text" } else { "input_text" },
                        "text": text,
                    }),
                    ChatContentPart::ImageUrl { image_url } => {
                        let mut part = json!({"type": "input_image", "image_url": image_url.url});
                        if let Some(detail) = image_url.detail {
                            part["detail"] = json!(detail);
                        }
                        part
                    }
                    ChatContentPart::File { file } => json!({
                        "type": "input_file",
                        "filename": file.filename,
                        "file_data": file.file_data,
                    }),
                }).collect())
            } else {
                Value::String(msg.content.unwrap_or_default())
            };
            out.push(json!({"role": msg.role, "content": content}));
        }
        if let Some(tool_calls) = msg.tool_calls {
            for call in tool_calls {
                // The Responses API rejects `id` values that are not the
                // original `fc_…` output-item id. We only carry the
                // `call_…` correlation id today (see types::ToolCall),
                // so emit `id` *only* when the value actually starts
                // with `fc`. `call_id` is sufficient on its own to
                // correlate the eventual `function_call_output`.
                let mut item = json!({
                    "type": "function_call",
                    "call_id": call.id,
                    "name": call.function.name,
                    "arguments": call.function.arguments,
                });
                if call.id.starts_with("fc") {
                    item["id"] = Value::from(call.id.clone());
                }
                out.push(item);
            }
        }
    }
    out
}

#[derive(Clone, Debug, Eq, PartialEq)]
struct ResponsesStreamFailure {
    code: &'static str,
    /// User-facing text built from OUR static templates plus the provider
    /// display label — never provider free-text.
    message: String,
}

impl ResponsesStreamFailure {
    /// Terminal failures that are transient by nature and safe to re-send
    /// with identical request bytes: provider capacity exhaustion, an empty
    /// completion, or a stream that ended without any terminal event.
    fn is_retryable(&self) -> bool {
        matches!(
            self.code,
            "responses_capacity" | "responses_empty" | "responses_missing_terminal"
        )
    }
}

/// Sanitized provider error codes that mean "try again later" on the
/// Responses wire. Classification-only: the code is matched against this
/// fixed set, never echoed beyond the local sanitized log line.
const CAPACITY_ERROR_CODES: &[&str] = &[
    "rate_limit_exceeded",
    "server_overloaded",
    "overloaded",
    "capacity",
    "server_error",
];

/// Fixed substrings that mark a provider free-text error message as a
/// transient capacity condition (xAI: "The model is currently at capacity
/// due to high demand. Please try again in a few minutes…"). The message is
/// consulted ONLY to produce a boolean — it is never logged, stored, or
/// surfaced (privacy invariant: provider bodies can echo request content).
const CAPACITY_MESSAGE_PATTERNS: &[&str] = &[
    "at capacity",
    "high demand",
    "try again in a few minutes",
    "overloaded",
];

/// Classification-only capacity check. `error_code` is already sanitized;
/// `message` is raw provider text used solely as a boolean signal.
fn is_capacity_failure(error_code: &str, message: Option<&str>) -> bool {
    if CAPACITY_ERROR_CODES.contains(&error_code) {
        return true;
    }
    message.is_some_and(|m| {
        let lower = m.to_ascii_lowercase();
        CAPACITY_MESSAGE_PATTERNS.iter().any(|p| lower.contains(p))
    })
}

struct CodexSseDecoder {
    /// Provider display label used in user-facing terminal messages
    /// ("Codex", "xAI"). Static, ours — never provider-supplied.
    provider_label: &'static str,
    buffer: String,
    active_tools: Vec<CodexToolAccumulator>,
    completed_tools: Vec<ToolCall>,
    /// Trace (Task 10A): set once the decoder parses any model event —
    /// feeds the first-model-event timing bucket.
    saw_model_event: bool,
    /// A terminal Responses event was observed. Failure/incomplete events
    /// take precedence over completion and are surfaced after the stream is
    /// drained; provider-controlled text is never retained.
    terminal_success: bool,
    terminal_failure: Option<ResponsesStreamFailure>,
    emitted_output: bool,
    /// The model produced ANY content on this attempt — including events the
    /// decoder does not forward to the UI (reasoning / reasoning-summary
    /// deltas, output items, content parts). Replaying such a request on a
    /// second account would double-spend, so account failover keys on this,
    /// not merely on what reached the UI. Metadata-only events
    /// (`response.created`, `response.in_progress`) never set it.
    saw_content_output: bool,
    /// Trace (Task 10A): provider-reported usage from `response.completed`
    /// (input including cached, output, cached slice). `None` until observed.
    observed_usage: Option<(u64, u64, u64)>,
    /// `phase` of the most recent `message` output item (`None` when that
    /// item had no phase, or no message item was seen). Muse marks an
    /// unfinished preamble `commentary`; see [`muse_dangling_preamble`].
    last_message_phase: Option<String>,
    /// `output_index` of the most recent `message` output item.
    last_message_index: Option<usize>,
    /// Text streamed via `output_text.delta`, per `output_index`. Bounded by
    /// what the UI already received (it duplicates `text_acc`, split by item).
    message_text: std::collections::BTreeMap<usize, String>,
    /// Muse only: when a completed message item carries text that extends
    /// what was streamed for it, emit the missing suffix as a text delta.
    /// Off for every other provider so their decoding stays unchanged.
    backfill_message_text: bool,
}

impl Default for CodexSseDecoder {
    fn default() -> Self {
        Self::for_provider("Codex")
    }
}

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

/// Parse a function-call arguments string into a JSON `Value`, mirroring
/// `runtime::api::parse_tool_input` so the chat UI's `LlmEvent::ToolUse`
/// handling sees the same shape regardless of provider.
///
/// Empty / whitespace input becomes `{}`. Invalid JSON becomes
/// `{"__parse_error": "..."}` — the agent loop already understands that
/// shape and converts it into an `is_error: true` tool_result.
fn parse_tool_arguments(raw: &str) -> Value {
    if raw.trim().is_empty() {
        return json!({});
    }
    match serde_json::from_str(raw) {
        Ok(v) => v,
        Err(e) => json!({ "__parse_error": format!("invalid tool input JSON: {}", e) }),
    }
}

/// Reduce a provider-supplied error identifier (`error.type`, `error.code`,
/// `incomplete_details.reason`) to a token safe for local logs. Lowercase
/// snake-case identifiers (`server_error`, `rate_limit_exceeded`,
/// `context_length_exceeded`, `max_output_tokens`) pass through verbatim —
/// that shape cannot carry prose or credential material. Anything with
/// uppercase, whitespace, or symbols outside the identifier charset — or
/// longer than 64 bytes — is replaced so free-text (which can echo request
/// content) and secret-shaped tokens can never leak into the log stream.
pub(crate) fn sanitize_error_identifier(raw: Option<&str>) -> &str {
    match raw {
        None => "absent",
        Some(s) => {
            let identifier_shaped = !s.is_empty()
                && s.len() <= 64
                && s.bytes()
                    .all(|b| b.is_ascii_alphanumeric() || b == b'_' || b == b'-' || b == b'.');
            if identifier_shaped
                && s.bytes()
                    .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'_')
            {
                // Lowercase snake_case enum value: safe to record verbatim.
                s
            } else if identifier_shaped {
                // Mixed-case / dashed identifiers can be secret-shaped
                // (API keys, tokens): report shape only.
                "unlisted_identifier"
            } else {
                "unsafe_or_freeform"
            }
        }
    }
}

/// Static message suffixes for Responses terminal failures. The provider
/// display label is prefixed at construction; `net.rs` classifies on these
/// label-independent suffixes so every label maps to `ApiStatus`.
pub(crate) const RESPONSES_AUTH_SUFFIX: &str =
    " authentication rejected in stream. Sign in again or switch to an available account.";
pub(crate) const RESPONSES_QUOTA_SUFFIX: &str =
    " usage quota exhausted in stream. Add credit, wait for reset, or switch models.";
pub(crate) const RESPONSES_FAILED_SUFFIX: &str =
    " response failed in stream. Provider error details withheld because they can echo request content.";
pub(crate) const RESPONSES_CONTEXT_SUFFIX: &str = " rejected the request: the conversation exceeds this model's context window. Run /compact or start a fresh session to continue.";
pub(crate) const RESPONSES_CAPACITY_SUFFIX: &str = " reports the model is at capacity (retries exhausted). Try again in a few minutes or switch models with /model.";
pub(crate) const RESPONSES_INCOMPLETE_SUFFIX: &str =
    " response was incomplete. Retry the request or reduce the requested output/context size.";
pub(crate) const RESPONSES_EMPTY_SUFFIX: &str =
    " completed without text or tool output. Retry the request.";
pub(crate) const RESPONSES_MISSING_TERMINAL_SUFFIX: &str =
    " response stream ended without a terminal event. Retry the request.";

/// Map a Responses terminal failure to a user-facing message. Only reacts to
/// the already-sanitized enum identifiers (`error.type` / `error.code`) and
/// the boolean capacity classification — never to provider free-text — so no
/// request content can leak. Known, user-actionable conditions get a specific
/// remedy; everything else keeps the neutral, details-withheld default.
fn user_message_for_terminal_failure(
    label: &str,
    error_kind: &str,
    error_code: &str,
    capacity: bool,
) -> String {
    // Context-window overflow is deterministic and user-fixable: the turn
    // cannot succeed until the conversation is smaller.
    if error_code == "context_length_exceeded" || error_kind == "context_length_exceeded" {
        return format!("{label}{RESPONSES_CONTEXT_SUFFIX}");
    }
    // Both explicit identifiers must agree on a retry class. Unknown or policy
    // classes (even with a capacity-looking message) are never retry signals.
    let classify = |value: &str| match value {
        "absent" | "error" => None,
        "invalid_api_key" | "authentication_error" | "invalid_authentication_error" => Some("auth"),
        "insufficient_quota" | "quota_exceeded" | "billing_error" | "usage_limit_reached" => {
            Some("quota")
        }
        "rate_limit_exceeded"
        | "rate_limit_error"
        | "server_overloaded"
        | "overloaded"
        | "capacity"
        | "server_error"
        | "api_error"
        | "overloaded_error" => Some("capacity"),
        _ => Some("blocked"),
    };
    let (kind, code) = (classify(error_kind), classify(error_code));
    if kind == Some("blocked")
        || code == Some("blocked")
        || (kind.is_some() && code.is_some() && kind != code)
    {
        return format!("{label}{RESPONSES_FAILED_SUFFIX}");
    }
    match code.or(kind) {
        Some("auth") => return format!("{label}{RESPONSES_AUTH_SUFFIX}"),
        Some("quota") => return format!("{label}{RESPONSES_QUOTA_SUFFIX}"),
        Some("capacity") => return format!("{label}{RESPONSES_CAPACITY_SUFFIX}"),
        // Preserve legacy flat xAI capacity handling only with no explicit
        // failure identifier to contradict it. No new free-text classifiers.
        None if capacity => return format!("{label}{RESPONSES_CAPACITY_SUFFIX}"),
        _ => {}
    }
    format!("{label}{RESPONSES_FAILED_SUFFIX}")
}

impl CodexSseDecoder {
    /// Decoder whose user-facing messages name `label` as the provider.
    fn for_provider(label: &'static str) -> Self {
        Self {
            provider_label: label,
            buffer: String::new(),
            active_tools: Vec::new(),
            completed_tools: Vec::new(),
            saw_model_event: false,
            terminal_success: false,
            terminal_failure: None,
            emitted_output: false,
            saw_content_output: false,
            observed_usage: None,
            last_message_phase: None,
            last_message_index: None,
            message_text: std::collections::BTreeMap::new(),
            backfill_message_text: false,
        }
    }

    /// Record a `message` output item's phase/index (added or done event).
    fn observe_message_item(&mut self, idx: usize, item: &Value) {
        if item.get("type").and_then(Value::as_str) != Some("message") {
            return;
        }
        self.last_message_index = Some(idx);
        self.last_message_phase = item
            .get("phase")
            .and_then(Value::as_str)
            .map(str::to_string);
    }

    /// Text of the most recent message item as streamed, if any was.
    fn last_message_text(&self) -> Option<&str> {
        self.last_message_index
            .and_then(|idx| self.message_text.get(&idx))
            .map(String::as_str)
    }

    /// Muse-only (C): a completed message item whose text extends what was
    /// streamed for it (and starts with it) gets the missing suffix emitted
    /// as a normal text delta. Never rewrites or duplicates streamed text.
    fn backfill_message_suffix(
        &mut self,
        idx: usize,
        item: &Value,
        tx: &mpsc::UnboundedSender<StreamEvent>,
        text_acc: &mut String,
    ) {
        if !self.backfill_message_text
            || item.get("type").and_then(Value::as_str) != Some("message")
        {
            return;
        }
        let Some(parts) = item.get("content").and_then(Value::as_array) else {
            return;
        };
        let full: String = parts
            .iter()
            .filter(|part| part.get("type").and_then(Value::as_str) == Some("output_text"))
            .filter_map(|part| part.get("text").and_then(Value::as_str))
            .collect();
        let streamed = self.message_text.entry(idx).or_default();
        if full.len() <= streamed.len() || !full.starts_with(streamed.as_str()) {
            return;
        }
        let suffix = full[streamed.len()..].to_string();
        tracing::debug!(
            provider = self.provider_label,
            suffix_bytes = suffix.len(),
            "backfilling message text missing from the delta stream"
        );
        streamed.push_str(&suffix);
        self.emitted_output = true;
        text_acc.push_str(&suffix);
        let _ = tx.send(StreamEvent::Llm(crate::runtime::types::LlmEvent::Text(
            suffix,
        )));
    }

    /// True once anything was streamed to the UI for this attempt (text
    /// deltas, tool starts, or completed tools). A retry after this point
    /// would duplicate output, so callers must treat the failure as terminal.
    fn emitted_any_output(&self) -> bool {
        self.emitted_output
            || !self.completed_tools.is_empty()
            || self.active_tools.iter().any(|tool| tool.started)
    }

    /// True once the model produced any content on this attempt, forwarded
    /// or not. The account-failover gate uses this (stricter than
    /// `emitted_any_output`): a request that already cost output on seat A
    /// is never replayed on seat B.
    fn model_produced_output(&self) -> bool {
        self.saw_content_output || self.emitted_any_output()
    }

    fn push_line(
        &mut self,
        line: &str,
        tx: &mpsc::UnboundedSender<StreamEvent>,
        text_acc: &mut String,
    ) {
        let line = line.trim_end_matches('\r');
        if line.is_empty() {
            if !self.buffer.is_empty() {
                let payload = std::mem::take(&mut self.buffer);
                self.push_payload(&payload, tx, text_acc);
            }
            return;
        }
        let Some(data) = line.strip_prefix("data:").map(str::trim_start) else {
            return;
        };
        if data == "[DONE]" {
            self.terminal_success = true;
            self.finish();
            return;
        }
        self.buffer.push_str(data);
    }

    fn push_payload(
        &mut self,
        payload: &str,
        tx: &mpsc::UnboundedSender<StreamEvent>,
        text_acc: &mut String,
    ) {
        let Ok(event) = serde_json::from_str::<Value>(payload) else {
            tracing::debug!(
                payload_bytes = payload.len(),
                "discarding malformed Responses SSE payload"
            );
            return;
        };
        let event_type = event
            .get("type")
            .and_then(Value::as_str)
            .unwrap_or_default();
        if matches!(
            event_type,
            "response.output_text.delta"
                | "response.output_item.added"
                | "response.function_call_arguments.delta"
                | "response.output_item.done"
                | "response.completed"
                | "response.done"
                | "response.failed"
                | "response.incomplete"
                | "error"
        ) {
            self.saw_model_event = true;
        }
        // Content-bearing event families. Anything the model emitted as
        // output — text, tool arguments, reasoning, content parts, output
        // items — counts, even when this decoder ignores the event below.
        if matches!(
            event_type,
            "response.output_text.delta"
                | "response.output_text.done"
                | "response.output_item.added"
                | "response.output_item.done"
                | "response.function_call_arguments.delta"
                | "response.function_call_arguments.done"
        ) || event_type.starts_with("response.reasoning")
            || event_type.starts_with("response.content_part")
            || event_type.starts_with("response.refusal")
        {
            self.saw_content_output = true;
        }
        match event_type {
            "response.output_text.delta" => {
                if let Some(delta) = event.get("delta").and_then(Value::as_str) {
                    if !delta.is_empty() {
                        self.emitted_output = true;
                    }
                    let idx = event
                        .get("output_index")
                        .and_then(Value::as_u64)
                        .unwrap_or(0) as usize;
                    self.message_text.entry(idx).or_default().push_str(delta);
                    text_acc.push_str(delta);
                    let _ = tx.send(StreamEvent::Llm(crate::runtime::types::LlmEvent::Text(
                        delta.to_string(),
                    )));
                }
            }
            "response.output_item.added" => {
                if let Some(item) = event.get("item") {
                    let idx = event
                        .get("output_index")
                        .and_then(Value::as_u64)
                        .unwrap_or(0) as usize;
                    self.observe_message_item(idx, item);
                    self.add_tool_from_item(idx, item, tx);
                }
            }
            "response.function_call_arguments.delta" => {
                let idx = event
                    .get("output_index")
                    .and_then(Value::as_u64)
                    .unwrap_or(0) as usize;
                let delta = event
                    .get("delta")
                    .and_then(Value::as_str)
                    .unwrap_or_default();
                if !delta.is_empty() {
                    let tool = self.ensure_tool(idx);
                    tool.arguments.push_str(delta);
                    let tool_id = tool.id.clone();
                    let _ = tx.send(StreamEvent::Llm(
                        crate::runtime::types::LlmEvent::ToolUseDelta {
                            tool_id,
                            delta: delta.to_string(),
                        },
                    ));
                }
            }
            "response.output_item.done" => {
                if let Some(item) = event.get("item") {
                    let idx = event
                        .get("output_index")
                        .and_then(Value::as_u64)
                        .unwrap_or(0) as usize;
                    self.observe_message_item(idx, item);
                    self.backfill_message_suffix(idx, item, tx, text_acc);
                    super::codex_wire::announce_web_search_call(self.provider_label, item, tx);
                    self.complete_tool_from_item(idx, item, tx);
                }
            }
            "response.completed" | "response.done" => {
                self.push_usage(&event, tx);
                self.terminal_success = true;
                self.finish();
            }
            "response.failed" | "error" => {
                self.push_usage(&event, tx);
                // Local-only diagnostics: log the provider's enum-like error
                // identifiers (`type` / `code`) after sanitizing them down to a
                // short identifier charset. The free-text `message` field is
                // never logged or surfaced — it can echo request content.
                //
                // Two envelope shapes are accepted: the nested Responses form
                // (`response.error.{type,code,message}` / `error.{…}`) and
                // xAI's flat form (`{"type":"error","code":…,"message":…}`)
                // where the event's own `type` is just "error", so only the
                // top-level `code` is a meaningful identifier there.
                let error_obj = event
                    .get("response")
                    .and_then(|r| r.get("error"))
                    .or_else(|| event.get("error"))
                    .filter(|e| e.is_object());
                let error_kind = sanitize_error_identifier(
                    error_obj
                        .and_then(|e| e.get("type"))
                        .and_then(Value::as_str),
                );
                let error_code = sanitize_error_identifier(
                    error_obj
                        .and_then(|e| e.get("code"))
                        .or_else(|| event.get("code"))
                        .and_then(Value::as_str),
                );
                // Provider free-text: classification-only boolean, never
                // retained past this expression.
                let capacity = is_capacity_failure(
                    error_code,
                    error_obj
                        .and_then(|e| e.get("message"))
                        .or_else(|| event.get("message"))
                        .and_then(Value::as_str),
                );
                tracing::warn!(
                    provider = self.provider_label,
                    event_type,
                    error_kind,
                    error_code,
                    capacity,
                    "responses stream terminal failure (provider message withheld)"
                );
                let message = user_message_for_terminal_failure(
                    self.provider_label,
                    error_kind,
                    error_code,
                    capacity,
                );
                self.terminal_failure = Some(ResponsesStreamFailure {
                    // Exact account/context failures outrank a free-text capacity
                    // hint: do not spend local retries before allowing failover.
                    code: if message
                        == format!("{}{RESPONSES_CAPACITY_SUFFIX}", self.provider_label)
                    {
                        "responses_capacity"
                    } else if super::account_routing::stream_failure_is_quota(
                        error_kind, error_code,
                    ) {
                        // Vetted quota identifiers only: the account's
                        // allowance is spent. Not retried here; the Codex
                        // path may fail over once if nothing was produced.
                        "responses_quota"
                    } else {
                        "responses_failed"
                    },
                    message,
                });
                self.finish();
            }
            "response.incomplete" => {
                self.push_usage(&event, tx);
                let reason = sanitize_error_identifier(
                    event
                        .get("response")
                        .and_then(|r| r.get("incomplete_details"))
                        .and_then(|d| d.get("reason"))
                        .and_then(Value::as_str),
                );
                tracing::warn!(
                    provider = self.provider_label,
                    reason,
                    "responses stream terminal incomplete"
                );
                self.terminal_failure = Some(ResponsesStreamFailure {
                    code: "responses_incomplete",
                    message: format!("{}{RESPONSES_INCOMPLETE_SUFFIX}", self.provider_label),
                });
                self.finish();
            }
            _ => {
                if !event_type.is_empty() {
                    tracing::debug!(event_type, "ignoring unknown Responses SSE event");
                }
            }
        }
    }

    fn terminal_result(&self) -> Result<(), ResponsesStreamFailure> {
        if let Some(failure) = &self.terminal_failure {
            return Err(failure.clone());
        }
        if self.terminal_success {
            if self.emitted_output || !self.completed_tools.is_empty() {
                Ok(())
            } else {
                Err(ResponsesStreamFailure {
                    code: "responses_empty",
                    message: format!("{}{RESPONSES_EMPTY_SUFFIX}", self.provider_label),
                })
            }
        } else {
            Err(ResponsesStreamFailure {
                code: "responses_missing_terminal",
                message: format!("{}{RESPONSES_MISSING_TERMINAL_SUFFIX}", self.provider_label),
            })
        }
    }

    fn ensure_tool(&mut self, idx: usize) -> &mut CodexToolAccumulator {
        // `idx` is the provider-supplied `output_index`; a hostile or buggy
        // provider sending a huge value would otherwise grow this Vec unboundedly
        // (OOM/DoS). No legitimate response has thousands of concurrent tool
        // calls, so clamp absurd indices to the last slot instead of allocating.
        const MAX_TOOLS: usize = 4096;
        let idx = idx.min(MAX_TOOLS - 1);
        while self.active_tools.len() <= idx {
            self.active_tools.push(CodexToolAccumulator::default());
        }
        &mut self.active_tools[idx]
    }

    fn add_tool_from_item(
        &mut self,
        idx: usize,
        item: &Value,
        tx: &mpsc::UnboundedSender<StreamEvent>,
    ) {
        if item.get("type").and_then(Value::as_str) != Some("function_call") {
            return;
        }
        let tool = self.ensure_tool(idx);
        if let Some(id) = item
            .get("call_id")
            .or_else(|| item.get("id"))
            .and_then(Value::as_str)
        {
            tool.id = id.to_string();
        }
        if let Some(name) = item.get("name").and_then(Value::as_str) {
            tool.name = name.to_string();
        }
        if !tool.started && !tool.name.is_empty() {
            tool.started = true;
            let _ = tx.send(StreamEvent::Llm(
                crate::runtime::types::LlmEvent::ToolUseStart {
                    tool_name: tool.name.clone(),
                    tool_id: tool.id.clone(),
                },
            ));
        }
    }

    fn complete_tool_from_item(
        &mut self,
        idx: usize,
        item: &Value,
        tx: &mpsc::UnboundedSender<StreamEvent>,
    ) {
        if item.get("type").and_then(Value::as_str) != Some("function_call") {
            return;
        }
        let tool = self.ensure_tool(idx);
        if let Some(id) = item
            .get("call_id")
            .or_else(|| item.get("id"))
            .and_then(Value::as_str)
        {
            tool.id = id.to_string();
        }
        if let Some(name) = item.get("name").and_then(Value::as_str) {
            tool.name = name.to_string();
        }
        if let Some(arguments) = item.get("arguments").and_then(Value::as_str) {
            tool.arguments = arguments.to_string();
        }
        if !tool.started && !tool.name.is_empty() {
            tool.started = true;
            let _ = tx.send(StreamEvent::Llm(
                crate::runtime::types::LlmEvent::ToolUseStart {
                    tool_name: tool.name.clone(),
                    tool_id: tool.id.clone(),
                },
            ));
        }
        let completed = if !tool.id.is_empty() && !tool.name.is_empty() {
            Some(ToolCall {
                id: tool.id.clone(),
                kind: "function".to_string(),
                function: super::types::FunctionCall {
                    name: tool.name.clone(),
                    arguments: tool.arguments.clone(),
                },
            })
        } else {
            None
        };
        if let Some(call) = completed {
            if self.completed_tools.iter().any(|done| done.id == call.id) {
                return;
            }
            self.emitted_output = true;
            // Emit the finalized `ToolUse` event so the chat UI can collapse
            // the streaming `ToolUseStart` (animated) into a stable
            // `ToolUse` block. Without this the bash-trace animation
            // persists forever and parallel tool blocks render as "still
            // running" even after they've completed. Mirrors the
            // Anthropic path in `runtime/api.rs` which emits the same
            // event on tool-use content_block_stop.
            let input = parse_tool_arguments(&call.function.arguments);
            let _ = tx.send(StreamEvent::Llm(crate::runtime::types::LlmEvent::ToolUse {
                tool_name: call.function.name.clone(),
                tool_id: call.id.clone(),
                input,
            }));
            self.completed_tools.push(ToolCall {
                id: call.id,
                kind: call.kind,
                function: call.function,
            });
        }
    }

    fn push_usage(&mut self, event: &Value, tx: &mpsc::UnboundedSender<StreamEvent>) {
        let usage = event
            .get("response")
            .and_then(|r| r.get("usage"))
            .or_else(|| event.get("usage"));
        let input = usage
            .and_then(|u| u.get("input_tokens"))
            .and_then(Value::as_u64)
            .unwrap_or(0);
        let output = usage
            .and_then(|u| u.get("output_tokens"))
            .and_then(Value::as_u64)
            .unwrap_or(0);
        // Responses-API usage nests cache hits under input_tokens_details;
        // input_tokens INCLUDES them. Downstream accounting uses Anthropic
        // semantics (context = input + cache_read + cache_creation, hit% =
        // cache_read/total), so report the cached slice separately and
        // subtract it from input — otherwise cache hits are invisible
        // (reported 0) and the context total double-counts.
        let cached = usage
            .and_then(|u| u.pointer("/input_tokens_details/cached_tokens"))
            .and_then(Value::as_u64)
            .unwrap_or(0)
            .min(input);
        if input > 0 || output > 0 {
            // Trace capture (metadata only): the provider-reported totals as
            // sent — input including the cached slice.
            if self.observed_usage.is_none() {
                self.observed_usage = Some((input, output, cached));
            }
            let _ = tx.send(StreamEvent::Session(
                crate::runtime::types::SessionEvent::Usage {
                    input_tokens: input - cached,
                    output_tokens: output,
                    cache_read_input_tokens: cached,
                    cache_creation_input_tokens: 0,
                    cache_creation_5m: None,
                    cache_creation_1h: None,
                    model: None,
                },
            ));
        }
    }

    /// Provider-reported usage for the trace record; `None` until observed.
    fn trace_usage(&self) -> Option<crate::runtime::trace::UsageMeta> {
        self.observed_usage
            .map(|(input, output, cached)| tr::provider_usage(input, output, cached))
    }

    fn finish(&mut self) {
        for tool in self.active_tools.drain(..) {
            if !tool.id.is_empty()
                && !tool.name.is_empty()
                && !self.completed_tools.iter().any(|done| done.id == tool.id)
            {
                self.emitted_output = true;
                self.completed_tools.push(ToolCall {
                    id: tool.id,
                    kind: "function".to_string(),
                    function: super::types::FunctionCall {
                        name: tool.name,
                        arguments: tool.arguments,
                    },
                });
            }
        }
    }
}

fn handle_events(
    events: &[OaiEvent],
    tx: &mpsc::UnboundedSender<StreamEvent>,
    text_acc: &mut String,
    tool_blocks: &mut Vec<Value>,
    name_map: &translate::ToolNameMap,
    suppress_stream_deltas: bool,
) {
    for ev in events {
        if let OaiEvent::TextDelta(t) = ev {
            text_acc.push_str(t);
        }
        if let OaiEvent::ToolCallsComplete { calls, .. } = ev {
            tool_blocks.extend(translate::tool_calls_to_content_blocks(calls, name_map));
        }
        if !suppress_stream_deltas {
            if let Some(se) = translate::oai_event_to_llm(ev) {
                let _ = tx.send(se);
            }
        }
    }
}

#[cfg(test)]
mod codex_input_messages_tests {
    //! Regression tests for the Codex Responses-API `input` shape.
    //!
    //! Background: the Responses API distinguishes two ids per tool
    //! invocation — `id` (the *output item id*, prefix `fc_…`) and
    //! `call_id` (the *function call id*, prefix `call_…`). When echoing
    //! a previous `function_call` back as an input item, supplying an
    //! `id` whose value is *not* a `fc_…` triggers
    //!
    //!   400 Bad Request: Invalid 'input[N].id': 'call_…'.
    //!   Expected an ID that begins with 'fc'.
    //!
    //! `id` is *optional* on input items — only `call_id` is required to
    //! correlate the eventual `function_call_output`. We elect not to
    //! emit `id` unless we actually have a real `fc_…` value to send.

    use super::super::types::{ChatMessage, FunctionCall, ToolCall};
    use super::*;

    fn sample_tool_call() -> ToolCall {
        ToolCall {
            id: "call_nZYquCuGUh8Qs9H51dwHMDgs".to_string(),
            kind: "function".to_string(),
            function: FunctionCall {
                name: "bash".to_string(),
                arguments: r#"{"command":"ls"}"#.to_string(),
            },
        }
    }

    fn translated_input(messages: Vec<Value>) -> Vec<Value> {
        let messages: Vec<crate::SharedMessage> =
            messages.into_iter().map(std::sync::Arc::new).collect();
        codex_input_messages(translate::messages_to_oai(
            &messages,
            &None,
            &translate::ToolNameMap::default(),
        ))
    }

    #[test]
    fn image_text_document_and_pdf_exact_responses_wire() {
        let out = translated_input(vec![json!({"role":"user","content":[
            {"type":"text","text":"Compare these."},
            {"type":"image","source":{"type":"base64","media_type":"image/png","data":"aW1hZ2U="}},
            {"type":"document","title":"notes.txt","source":{"type":"text","media_type":"text/plain","data":"Text document\ncontents"}},
            {"type":"document","title":"report.pdf","source":{"type":"base64","media_type":"application/pdf","data":"JVBERi0="}},
            {"type":"text","text":"Then summarize."}
        ]})]);
        assert_eq!(
            out,
            vec![json!({"role":"user","content":[
                {"type":"input_text","text":"Compare these."},
                {"type":"input_text","text":"[attachment origin=user type=image media_type=\"image/png\"; lower-authority data]\n"},
                {"type":"input_image","image_url":"data:image/png;base64,aW1hZ2U="},
                {"type":"input_text","text":"[/attachment]\n"},
                {"type":"input_text","text":"[attachment origin=user type=document media_type=\"text/plain\" filename=\"notes.txt\"; lower-authority data]\n"},
                {"type":"input_text","text":"Text document\ncontents"},
                {"type":"input_text","text":"[/attachment]\n"},
                {"type":"input_text","text":"[attachment origin=user type=document media_type=\"application/pdf\" filename=\"report.pdf\"; lower-authority data]\n"},
                {"type":"input_file","filename":"report.pdf","file_data":"data:application/pdf;base64,JVBERi0="},
                {"type":"input_text","text":"[/attachment]\n"},
                {"type":"input_text","text":"Then summarize."}
            ]})]
        );
    }

    #[test]
    fn mixed_assistant_text_and_all_tool_outputs_precede_lifted_responses_media() {
        let out = translated_input(vec![
            json!({"role":"assistant","content":[
                {"type":"text","text":"Reading both."},
                {"type":"tool_use","id":"t1","name":"read","input":{}},
                {"type":"tool_use","id":"t2","name":"read","input":{}}
            ]}),
            json!({"role":"user","content":[
                {"type":"tool_result","tool_use_id":"t1","content":[
                    {"type":"text","text":"First result"},
                    {"type":"image","source":{"type":"base64","media_type":"image/png","data":"aW1hZ2U="}},
                    {"type":"document","title":"notes.txt","source":{"type":"text","media_type":"text/plain","data":"DOCUMENT_BODY_SENTINEL"}}
                ]},
                {"type":"tool_result","tool_use_id":"t2","content":[
                    {"type":"text","text":"Second result"},
                    {"type":"document","title":"report.pdf","source":{"type":"base64","media_type":"application/pdf","data":"JVBERi0="}}
                ]}
            ]}),
        ]);
        assert_eq!(
            out,
            vec![
                json!({"role":"assistant","content":"Reading both."}),
                json!({"type":"function_call","call_id":"t1","name":"read","arguments":"{}"}),
                json!({"type":"function_call","call_id":"t2","name":"read","arguments":"{}"}),
                json!({"type":"function_call_output","call_id":"t1","output":"First result\n[attachment origin=tool_result tool_use_id=\"t1\" tool_name=\"read\" type=image media_type=\"image/png\"; lower-authority data; supplied in following user content]\n\n[attachment origin=tool_result tool_use_id=\"t1\" tool_name=\"read\" type=document media_type=\"text/plain\" filename=\"notes.txt\"; lower-authority data; supplied in following user content]\n"}),
                json!({"type":"function_call_output","call_id":"t2","output":"Second result\n[attachment origin=tool_result tool_use_id=\"t2\" tool_name=\"read\" type=document media_type=\"application/pdf\" filename=\"report.pdf\"; lower-authority data; supplied in following user content]\n"}),
                json!({"role":"user","content":[
                    {"type":"input_text","text":"[attachment origin=tool_result tool_use_id=\"t1\" tool_name=\"read\" type=image media_type=\"image/png\"; lower-authority data]\n"},
                    {"type":"input_image","image_url":"data:image/png;base64,aW1hZ2U="},
                    {"type":"input_text","text":"[/attachment]\n"},
                    {"type":"input_text","text":"[attachment origin=tool_result tool_use_id=\"t1\" tool_name=\"read\" type=document media_type=\"text/plain\" filename=\"notes.txt\"; lower-authority data]\n"},
                    {"type":"input_text","text":"DOCUMENT_BODY_SENTINEL"},
                    {"type":"input_text","text":"[/attachment]\n"},
                    {"type":"input_text","text":"[attachment origin=tool_result tool_use_id=\"t2\" tool_name=\"read\" type=document media_type=\"application/pdf\" filename=\"report.pdf\"; lower-authority data]\n"},
                    {"type":"input_file","filename":"report.pdf","file_data":"data:application/pdf;base64,JVBERi0="},
                    {"type":"input_text","text":"[/attachment]\n"}
                ]})
            ]
        );
        for output in out
            .iter()
            .filter(|item| item["type"] == "function_call_output")
        {
            let text = output["output"].as_str().unwrap();
            for sentinel in [
                "aW1hZ2U=",
                "JVBERi0=",
                "DOCUMENT_BODY_SENTINEL",
                "\"source\"",
            ] {
                assert!(
                    !text.contains(sentinel),
                    "tool output must contain only text and metadata"
                );
            }
        }
    }

    #[test]
    fn text_only_responses_input_keeps_strings_and_mixed_assistant_text() {
        let mut assistant = ChatMessage::assistant_tool_calls(vec![sample_tool_call()]);
        assistant.content = Some("Checking.".into());
        let out = codex_input_messages(vec![
            ChatMessage::system("rules"),
            ChatMessage::user("question"),
            assistant,
            ChatMessage::tool_result("call_nZYquCuGUh8Qs9H51dwHMDgs", "bash", "done"),
            ChatMessage::assistant("answer"),
        ]);
        assert_eq!(
            out,
            vec![
                json!({"role":"system","content":"rules"}),
                json!({"role":"user","content":"question"}),
                json!({"role":"assistant","content":"Checking."}),
                json!({"type":"function_call","call_id":"call_nZYquCuGUh8Qs9H51dwHMDgs","name":"bash","arguments":"{\"command\":\"ls\"}"}),
                json!({"type":"function_call_output","call_id":"call_nZYquCuGUh8Qs9H51dwHMDgs","output":"done"}),
                json!({"role":"assistant","content":"answer"}),
            ]
        );
    }

    /// Anthropic-shaped stored history covering every assistant shape:
    /// mixed text+tool_use, tool-only, text-only, plus a plain user message
    /// right after tool results (forces the `" "` separator).
    fn phase_history() -> Vec<Value> {
        vec![
            json!({"role":"user","content":"build it"}),
            json!({"role":"assistant","content":[
                {"type":"thinking","thinking":"hidden","signature":"s"},
                {"type":"text","text":"Mapping the repo first."},
                {"type":"tool_use","id":"call_a","name":"bash","input":{"command":"ls"}},
            ]}),
            json!({"role":"user","content":[
                {"type":"tool_result","tool_use_id":"call_a","content":"src"},
            ]}),
            json!({"role":"assistant","content":[
                {"type":"tool_use","id":"call_b","name":"bash","input":{"command":"pwd"}},
            ]}),
            json!({"role":"user","content":[
                {"type":"tool_result","tool_use_id":"call_b","content":"/w"},
            ]}),
            json!({"role":"user","content":"also check logs"}),
            json!({"role":"assistant","content":[{"type":"text","text":"Done: all good."}]}),
            json!({"role":"user","content":"thanks"}),
            json!({"role":"assistant","content":"Plain string answer."}),
        ]
    }

    fn oai(messages: Vec<Value>) -> Vec<ChatMessage> {
        let messages: Vec<crate::SharedMessage> =
            messages.into_iter().map(std::sync::Arc::new).collect();
        translate::messages_to_oai(&messages, &None, &translate::ToolNameMap::default())
    }

    #[test]
    fn muse_input_infers_phase_from_message_structure() {
        let out = muse_input_messages(oai(phase_history()));
        assert_eq!(
            out,
            vec![
                json!({"role":"user","content":"build it"}),
                // Mixed message: its text is commentary and precedes its call.
                json!({"type":"message","role":"assistant","phase":"commentary",
                       "content":[{"type":"output_text","text":"Mapping the repo first."}]}),
                json!({"type":"function_call","call_id":"call_a","name":"bash","arguments":"{\"command\":\"ls\"}"}),
                json!({"type":"function_call_output","call_id":"call_a","output":"src"}),
                // Tool-only message: unchanged, no message item invented.
                json!({"type":"function_call","call_id":"call_b","name":"bash","arguments":"{\"command\":\"pwd\"}"}),
                json!({"type":"function_call_output","call_id":"call_b","output":"/w"}),
                // Harness separator keeps the legacy shape (not model text).
                json!({"role":"assistant","content":" "}),
                json!({"role":"user","content":"also check logs"}),
                // Text-only messages (block and string storage) are final.
                json!({"type":"message","role":"assistant","phase":"final_answer",
                       "content":[{"type":"output_text","text":"Done: all good."}]}),
                json!({"role":"user","content":"thanks"}),
                json!({"type":"message","role":"assistant","phase":"final_answer",
                       "content":[{"type":"output_text","text":"Plain string answer."}]}),
            ]
        );
    }

    #[test]
    fn codex_and_xai_input_stay_phase_less_for_the_same_history() {
        // Codex and xAI both build input with `codex_input_messages`; the
        // Muse change must not alter a byte of it.
        let out = codex_input_messages(oai(phase_history()));
        assert_eq!(
            out,
            vec![
                json!({"role":"user","content":"build it"}),
                json!({"role":"assistant","content":"Mapping the repo first."}),
                json!({"type":"function_call","call_id":"call_a","name":"bash","arguments":"{\"command\":\"ls\"}"}),
                json!({"type":"function_call_output","call_id":"call_a","output":"src"}),
                json!({"type":"function_call","call_id":"call_b","name":"bash","arguments":"{\"command\":\"pwd\"}"}),
                json!({"type":"function_call_output","call_id":"call_b","output":"/w"}),
                json!({"role":"assistant","content":" "}),
                json!({"role":"user","content":"also check logs"}),
                json!({"role":"assistant","content":"Done: all good."}),
                json!({"role":"user","content":"thanks"}),
                json!({"role":"assistant","content":"Plain string answer."}),
            ]
        );
        assert!(!serde_json::to_string(&out).unwrap().contains("phase"));
    }

    #[test]
    fn muse_input_leaves_non_assistant_and_tool_only_messages_identical() {
        let messages = || {
            vec![
                ChatMessage::system("rules"),
                ChatMessage::user("question"),
                ChatMessage::assistant_tool_calls(vec![sample_tool_call()]),
                ChatMessage::tool_result("call_nZYquCuGUh8Qs9H51dwHMDgs", "bash", "done"),
            ]
        };
        assert_eq!(
            muse_input_messages(messages()),
            codex_input_messages(messages())
        );
    }

    #[test]
    fn codex_instructions_appends_autonomous_loop_policy() {
        let instructions = codex_instructions(&Some("Project-specific rules.".to_string()));
        assert!(instructions.contains("Project-specific rules."));
        assert!(instructions.contains("Do not stop at phase boundaries"));
        assert!(instructions.contains("Do not ask the user whether to continue"));
        assert!(
            instructions.contains("continue autonomously until the full requested job is complete")
        );
    }

    #[test]
    fn function_call_input_omits_non_fc_id() {
        let messages = vec![ChatMessage::assistant_tool_calls(vec![sample_tool_call()])];
        let out = codex_input_messages(messages);
        assert_eq!(out.len(), 1, "one tool_call → one input item");
        let item = &out[0];
        assert_eq!(
            item.get("type").and_then(Value::as_str),
            Some("function_call")
        );
        assert!(
            item.get("id").is_none(),
            "must not echo a non-`fc_` id back; got {:?}",
            item.get("id"),
        );
        assert_eq!(
            item.get("call_id").and_then(Value::as_str),
            Some("call_nZYquCuGUh8Qs9H51dwHMDgs"),
        );
        assert_eq!(item.get("name").and_then(Value::as_str), Some("bash"));
    }

    #[test]
    fn function_call_input_keeps_real_fc_id() {
        // If we ever do have a genuine `fc_…` id (round-tripped from the
        // Responses API), we *should* echo it.
        let mut call = sample_tool_call();
        call.id = "fc_abc123".to_string();
        let messages = vec![ChatMessage::assistant_tool_calls(vec![call])];
        let out = codex_input_messages(messages);
        let item = &out[0];
        assert_eq!(item.get("id").and_then(Value::as_str), Some("fc_abc123"));
        assert_eq!(
            item.get("call_id").and_then(Value::as_str),
            Some("fc_abc123")
        );
    }

    #[test]
    fn function_call_output_round_trips_call_id() {
        // The follow-up tool message must reference the original call_id.
        let messages = vec![ChatMessage::tool_result(
            "call_nZYquCuGUh8Qs9H51dwHMDgs",
            "bash",
            "total 0",
        )];
        let out = codex_input_messages(messages);
        let item = &out[0];
        assert_eq!(
            item.get("type").and_then(Value::as_str),
            Some("function_call_output"),
        );
        assert_eq!(
            item.get("call_id").and_then(Value::as_str),
            Some("call_nZYquCuGUh8Qs9H51dwHMDgs"),
        );
        assert_eq!(item.get("output").and_then(Value::as_str), Some("total 0"));
    }
}

#[cfg(test)]
mod codex_decoder_tests {
    //! Regression tests for `CodexSseDecoder`.
    //!
    //! The decoder is sync — we drive it via `push_line` and capture
    //! emitted `StreamEvent`s from an `unbounded_channel` using
    //! `try_recv`, no async runtime needed.

    use super::*;
    use crate::runtime::types::{LlmEvent, SessionEvent, StreamEvent};

    fn collect_events(rx: &mut mpsc::UnboundedReceiver<StreamEvent>) -> Vec<StreamEvent> {
        let mut out = Vec::new();
        while let Ok(ev) = rx.try_recv() {
            out.push(ev);
        }
        out
    }

    fn drive(lines: &[&str]) -> (CodexSseDecoder, String, Vec<StreamEvent>) {
        let (tx, mut rx) = mpsc::unbounded_channel();
        let mut decoder = CodexSseDecoder::default();
        let mut text_acc = String::new();
        for line in lines {
            decoder.push_line(line, &tx, &mut text_acc);
        }
        let events = collect_events(&mut rx);
        (decoder, text_acc, events)
    }

    fn drive_with(
        mut decoder: CodexSseDecoder,
        lines: &[&str],
    ) -> (CodexSseDecoder, String, Vec<StreamEvent>) {
        let (tx, mut rx) = mpsc::unbounded_channel();
        let mut text_acc = String::new();
        for line in lines {
            decoder.push_line(line, &tx, &mut text_acc);
        }
        let events = collect_events(&mut rx);
        (decoder, text_acc, events)
    }

    #[test]
    fn decoder_captures_phase_of_the_last_message_item() {
        // Muse shape (ask1.sse): reasoning, commentary message, function_call.
        let (decoder, text, _) = drive(&[
            r#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"reasoning","id":"rs_0"}}"#,
            "",
            r#"data: {"type":"response.output_item.added","output_index":1,"item":{"type":"message","role":"assistant","phase":"commentary","content":[]}}"#,
            "",
            r#"data: {"type":"response.output_text.delta","output_index":1,"delta":"Mapping first."}"#,
            "",
            r#"data: {"type":"response.output_item.done","output_index":1,"item":{"type":"message","role":"assistant","phase":"commentary","content":[{"type":"output_text","text":"Mapping first."}]}}"#,
            "",
            r#"data: {"type":"response.output_item.added","output_index":2,"item":{"type":"function_call","call_id":"call_1","name":"bash"}}"#,
            "",
        ]);
        assert_eq!(decoder.last_message_phase.as_deref(), Some("commentary"));
        assert_eq!(decoder.last_message_index, Some(1));
        assert_eq!(decoder.last_message_text(), Some("Mapping first."));
        assert_eq!(text, "Mapping first.");

        // A later phase-less (final) message replaces it.
        let (decoder, _, _) = drive(&[
            r#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"message","role":"assistant","phase":"commentary","content":[]}}"#,
            "",
            r#"data: {"type":"response.output_item.added","output_index":1,"item":{"type":"message","role":"assistant","content":[]}}"#,
            "",
            r#"data: {"type":"response.output_text.delta","output_index":1,"delta":"Answer"}"#,
            "",
        ]);
        assert_eq!(decoder.last_message_phase, None);
        assert_eq!(decoder.last_message_text(), Some("Answer"));
    }

    const TRUNCATED_STREAM: [&str; 6] = [
        r#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"message","role":"assistant","content":[]}}"#,
        "",
        r#"data: {"type":"response.output_text.delta","output_index":0,"delta":"Key facts:"}"#,
        "",
        r#"data: {"type":"response.output_item.done","output_index":0,"item":{"type":"message","role":"assistant","content":[{"type":"output_text","text":"Key facts:\nWhich format?"}]}}"#,
        "",
    ];

    #[test]
    fn muse_decoder_backfills_unstreamed_message_suffix_once() {
        let mut decoder = CodexSseDecoder::for_provider("Meta Muse");
        decoder.backfill_message_text = true;
        let mut lines = TRUNCATED_STREAM.to_vec();
        // A repeated done event must not emit the suffix twice.
        lines.extend_from_slice(&TRUNCATED_STREAM[4..]);
        let (decoder, text, events) = drive_with(decoder, &lines);
        assert_eq!(text, "Key facts:\nWhich format?");
        let texts: Vec<_> = events
            .iter()
            .filter_map(|e| match e {
                StreamEvent::Llm(LlmEvent::Text(t)) => Some(t.as_str()),
                _ => None,
            })
            .collect();
        assert_eq!(texts, vec!["Key facts:", "\nWhich format?"]);
        assert_eq!(
            decoder.last_message_text(),
            Some("Key facts:\nWhich format?")
        );
    }

    #[test]
    fn backfill_is_off_by_default_and_never_rewrites_divergent_text() {
        // Default (Codex) decoder: decoding unchanged.
        let (_, text, _) = drive(&TRUNCATED_STREAM);
        assert_eq!(text, "Key facts:");
        // Muse decoder: final text that does not extend the stream is ignored.
        let mut decoder = CodexSseDecoder::for_provider("Meta Muse");
        decoder.backfill_message_text = true;
        let (_, text, _) = drive_with(
            decoder,
            &[
                r#"data: {"type":"response.output_text.delta","output_index":0,"delta":"Hello"}"#,
                "",
                r#"data: {"type":"response.output_item.done","output_index":0,"item":{"type":"message","content":[{"type":"output_text","text":"Goodbye world"}]}}"#,
                "",
            ],
        );
        assert_eq!(text, "Hello");
    }

    #[test]
    fn hostile_output_index_does_not_grow_active_tools_unbounded() {
        // A provider (or wire corruption) sending an absurd `output_index` must
        // not drive an unbounded `active_tools` grow (OOM/DoS). ensure_tool clamps
        // to MAX_TOOLS. Without the clamp this test would attempt to allocate ~1e9
        // accumulators and hang/OOM.
        let lines = [
            r#"data: {"type":"response.function_call_arguments.delta","output_index":1000000000,"delta":"x"}"#,
            "",
        ];
        let (decoder, _text, _events) = drive(&lines);
        assert!(
            decoder.active_tools.len() <= 4096,
            "active_tools grew to {} — output_index not bounded",
            decoder.active_tools.len()
        );
    }

    #[test]
    fn text_delta_aggregates_into_text_acc_and_emits_text_events() {
        let lines = [
            r#"data: {"type":"response.output_text.delta","delta":"Hello, "}"#,
            "",
            r#"data: {"type":"response.output_text.delta","delta":"world!"}"#,
            "",
        ];
        let (_decoder, text_acc, events) = drive(&lines);
        assert_eq!(text_acc, "Hello, world!");
        let texts: Vec<_> = events
            .iter()
            .filter_map(|e| match e {
                StreamEvent::Llm(LlmEvent::Text(t)) => Some(t.as_str()),
                _ => None,
            })
            .collect();
        assert_eq!(texts, vec!["Hello, ", "world!"]);
    }

    #[test]
    fn hosted_web_search_items_are_not_tool_uses() {
        // OpenAI-hosted web search (SYNAPS_CODEX_WEB_SEARCH) streams
        // `web_search_call` output items. They run server-side, so the decoder
        // must not surface them as local tool uses — only the answer text and
        // real function calls come through.
        let lines = [
            r#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"web_search_call","id":"ws_1","status":"in_progress"}}"#,
            "",
            r#"data: {"type":"response.web_search_call.searching","output_index":0,"item_id":"ws_1"}"#,
            "",
            r#"data: {"type":"response.output_item.done","output_index":0,"item":{"type":"web_search_call","id":"ws_1","status":"completed","action":{"type":"search","query":"ram dass puja chat youtube"}}}"#,
            "",
            r#"data: {"type":"response.output_item.added","output_index":1,"item":{"type":"function_call","call_id":"call_1","name":"get_asset"}}"#,
            "",
            r#"data: {"type":"response.output_item.done","output_index":1,"item":{"type":"function_call","call_id":"call_1","name":"get_asset","arguments":"{}"}}"#,
            "",
            r#"data: {"type":"response.output_text.delta","output_index":2,"delta":"Found it."}"#,
            "",
            r#"data: {"type":"response.output_text.annotation.added","output_index":2,"annotation":{"type":"url_citation","url":"https://www.youtube.com/watch?v=sH-u96rpYSw"}}"#,
            "",
        ];
        let (decoder, text, events) = drive(&lines);
        assert_eq!(text, "Found it.");
        assert_eq!(decoder.completed_tools.len(), 1);
        assert_eq!(decoder.completed_tools[0].function.name, "get_asset");
        let starts = events
            .iter()
            .filter(|e| matches!(e, StreamEvent::Llm(LlmEvent::ToolUseStart { .. })))
            .count();
        assert_eq!(starts, 1, "web_search_call must not emit a ToolUseStart");
    }

    #[test]
    fn web_search_env_value_maps_to_hosted_tool() {
        use super::codex_web_search_tool;
        assert_eq!(codex_web_search_tool(None), None);
        assert_eq!(codex_web_search_tool(Some("")), None);
        assert_eq!(codex_web_search_tool(Some("off")), None);
        assert_eq!(codex_web_search_tool(Some("bogus")), None);
        for on in ["live", "LIVE", " on ", "true", "1", "yes"] {
            assert_eq!(
                codex_web_search_tool(Some(on)),
                Some(serde_json::json!({ "type": "web_search" })),
                "{on:?}"
            );
        }
        assert_eq!(
            codex_web_search_tool(Some("cached")),
            Some(serde_json::json!({ "type": "web_search", "external_web_access": false }))
        );
    }

    #[test]
    fn single_function_call_completes_via_output_item_done() {
        let lines = [
            r#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","call_id":"call_abc","name":"bash"}}"#,
            "",
            r#"data: {"type":"response.function_call_arguments.delta","output_index":0,"delta":"{\"cmd\""}"#,
            "",
            r#"data: {"type":"response.function_call_arguments.delta","output_index":0,"delta":":\"ls\"}"}"#,
            "",
            r#"data: {"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","call_id":"call_abc","name":"bash","arguments":"{\"cmd\":\"ls\"}"}}"#,
            "",
        ];
        let (decoder, _text, events) = drive(&lines);

        assert_eq!(decoder.completed_tools.len(), 1);
        let tool = &decoder.completed_tools[0];
        assert_eq!(tool.id, "call_abc");
        assert_eq!(tool.function.name, "bash");
        assert_eq!(tool.function.arguments, r#"{"cmd":"ls"}"#);

        // Exactly one ToolUseStart for the tool.
        let starts: Vec<_> = events
            .iter()
            .filter_map(|e| match e {
                StreamEvent::Llm(LlmEvent::ToolUseStart { tool_name, tool_id }) => {
                    Some((tool_name.as_str(), tool_id.as_str()))
                }
                _ => None,
            })
            .collect();
        assert_eq!(
            starts,
            vec![("bash", "call_abc")],
            "exactly one ToolUseStart with correct tool_id"
        );

        // Two argument deltas streamed (each carrying the tool_id so
        // parallel calls can be routed correctly by the chat UI).
        let deltas: Vec<_> = events
            .iter()
            .filter_map(|e| match e {
                StreamEvent::Llm(LlmEvent::ToolUseDelta { tool_id, delta }) => {
                    Some((tool_id.as_str(), delta.as_str()))
                }
                _ => None,
            })
            .collect();
        assert_eq!(
            deltas,
            vec![("call_abc", r#"{"cmd""#), ("call_abc", r#":"ls"}"#)]
        );
    }

    #[test]
    fn parallel_tool_calls_indexed_separately() {
        let lines = [
            r#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","call_id":"call_1","name":"bash"}}"#,
            "",
            r#"data: {"type":"response.output_item.added","output_index":1,"item":{"type":"function_call","call_id":"call_2","name":"read"}}"#,
            "",
            r#"data: {"type":"response.function_call_arguments.delta","output_index":1,"delta":"{\"path\":\"a\"}"}"#,
            "",
            r#"data: {"type":"response.function_call_arguments.delta","output_index":0,"delta":"{\"cmd\":\"ls\"}"}"#,
            "",
            r#"data: {"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","call_id":"call_1","name":"bash","arguments":"{\"cmd\":\"ls\"}"}}"#,
            "",
            r#"data: {"type":"response.output_item.done","output_index":1,"item":{"type":"function_call","call_id":"call_2","name":"read","arguments":"{\"path\":\"a\"}"}}"#,
            "",
        ];
        let (decoder, _text, _events) = drive(&lines);

        assert_eq!(decoder.completed_tools.len(), 2);
        let mut by_id: std::collections::BTreeMap<&str, &ToolCall> =
            std::collections::BTreeMap::new();
        for tool in &decoder.completed_tools {
            by_id.insert(tool.id.as_str(), tool);
        }
        assert_eq!(by_id["call_1"].function.name, "bash");
        assert_eq!(by_id["call_1"].function.arguments, r#"{"cmd":"ls"}"#);
        assert_eq!(by_id["call_2"].function.name, "read");
        assert_eq!(by_id["call_2"].function.arguments, r#"{"path":"a"}"#);
    }

    #[test]
    fn output_item_done_emits_tool_use_event() {
        // Regression: the codex decoder must emit `LlmEvent::ToolUse` once a
        // function_call's `output_item.done` arrives so the chat UI can
        // collapse `ChatMessage::ToolUseStart` (animated) into the finalized
        // `ChatMessage::ToolUse`. Without this the bash-trace animation
        // persists forever and parallel tool blocks render as "still
        // running" even after they've completed.
        let lines = [
            r#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","call_id":"call_abc","name":"bash"}}"#,
            "",
            r#"data: {"type":"response.function_call_arguments.delta","output_index":0,"delta":"{\"command\":\"ls\"}"}"#,
            "",
            r#"data: {"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","call_id":"call_abc","name":"bash","arguments":"{\"command\":\"ls\"}"}}"#,
            "",
        ];
        let (_decoder, _text, events) = drive(&lines);

        let tool_uses: Vec<_> = events
            .iter()
            .filter_map(|e| match e {
                StreamEvent::Llm(LlmEvent::ToolUse {
                    tool_name,
                    tool_id,
                    input,
                }) => Some((tool_name.as_str(), tool_id.as_str(), input.clone())),
                _ => None,
            })
            .collect();
        assert_eq!(
            tool_uses.len(),
            1,
            "expected exactly one ToolUse finalize event"
        );
        assert_eq!(tool_uses[0].0, "bash");
        assert_eq!(tool_uses[0].1, "call_abc");
        assert_eq!(
            tool_uses[0].2,
            serde_json::json!({"command": "ls"}),
            "input must be parsed as a JSON Value, not a string"
        );
    }

    #[test]
    fn parallel_tool_calls_emit_tool_use_per_index() {
        // Regression: parallel tool calls must each get their own ToolUse
        // finalize event with the correct tool_id, so the chat UI can route
        // their results back to the right block by id.
        let lines = [
            r#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","call_id":"call_1","name":"bash"}}"#,
            "",
            r#"data: {"type":"response.output_item.added","output_index":1,"item":{"type":"function_call","call_id":"call_2","name":"read"}}"#,
            "",
            r#"data: {"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","call_id":"call_1","name":"bash","arguments":"{\"command\":\"ls\"}"}}"#,
            "",
            r#"data: {"type":"response.output_item.done","output_index":1,"item":{"type":"function_call","call_id":"call_2","name":"read","arguments":"{\"path\":\"a\"}"}}"#,
            "",
        ];
        let (_decoder, _text, events) = drive(&lines);

        let tool_uses: Vec<_> = events
            .iter()
            .filter_map(|e| match e {
                StreamEvent::Llm(LlmEvent::ToolUse {
                    tool_name,
                    tool_id,
                    input,
                }) => Some((tool_name.clone(), tool_id.clone(), input.clone())),
                _ => None,
            })
            .collect();

        assert_eq!(tool_uses.len(), 2, "one ToolUse finalize per parallel call");
        let by_id: std::collections::BTreeMap<&str, &(String, String, serde_json::Value)> =
            tool_uses.iter().map(|t| (t.1.as_str(), t)).collect();
        assert_eq!(by_id["call_1"].0, "bash");
        assert_eq!(by_id["call_1"].2, serde_json::json!({"command": "ls"}));
        assert_eq!(by_id["call_2"].0, "read");
        assert_eq!(by_id["call_2"].2, serde_json::json!({"path": "a"}));
    }

    #[test]
    fn malformed_arguments_emit_tool_use_with_parse_error() {
        // If the model produces invalid JSON arguments, surface a structured
        // parse error in the `input` (matching how the Anthropic path
        // handles it via parse_tool_input) so the agent loop can return an
        // error tool_result instead of silently dropping the tool.
        let lines = [
            r#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","call_id":"call_bad","name":"bash"}}"#,
            "",
            r#"data: {"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","call_id":"call_bad","name":"bash","arguments":"{not json"}}"#,
            "",
        ];
        let (_decoder, _text, events) = drive(&lines);

        let tool_use = events.iter().find_map(|e| match e {
            StreamEvent::Llm(LlmEvent::ToolUse { input, .. }) => Some(input.clone()),
            _ => None,
        });
        let input = tool_use.expect("ToolUse event missing");
        assert!(
            input.get("__parse_error").and_then(Value::as_str).is_some(),
            "malformed arguments must surface __parse_error, got {input}"
        );
    }

    #[test]
    fn response_completed_emits_usage_event() {
        let lines = [
            r#"data: {"type":"response.completed","response":{"usage":{"input_tokens":42,"output_tokens":17}}}"#,
            "",
        ];
        let (_decoder, _text, events) = drive(&lines);
        let usage = events.iter().find_map(|e| match e {
            StreamEvent::Session(SessionEvent::Usage {
                input_tokens,
                output_tokens,
                ..
            }) => Some((*input_tokens, *output_tokens)),
            _ => None,
        });
        assert_eq!(usage, Some((42, 17)));
    }

    /// Responses-API usage nests cache hits under input_tokens_details, and
    /// input_tokens INCLUDES them. The decoder must surface the cached slice
    /// as cache_read (Anthropic semantics) and subtract it from input —
    /// dropping it (the old hardcoded 0) made every codex turn look 100%
    /// uncached (incident: 2026-07-16 codex cache investigation).
    #[test]
    fn response_completed_surfaces_cached_tokens_without_double_count() {
        let lines = [
            r#"data: {"type":"response.completed","response":{"usage":{"input_tokens":1000,"input_tokens_details":{"cached_tokens":900},"output_tokens":17}}}"#,
            "",
        ];
        let (_decoder, _text, events) = drive(&lines);
        let usage = events.iter().find_map(|e| match e {
            StreamEvent::Session(SessionEvent::Usage {
                input_tokens,
                output_tokens,
                cache_read_input_tokens,
                ..
            }) => Some((*input_tokens, *output_tokens, *cache_read_input_tokens)),
            _ => None,
        });
        assert_eq!(
            usage,
            Some((100, 17, 900)),
            "input must exclude the cached slice; cache_read must carry it"
        );
    }

    #[test]
    fn response_completed_with_zero_usage_emits_nothing() {
        let lines = [
            r#"data: {"type":"response.output_text.delta","delta":"ok"}"#,
            "",
            r#"data: {"type":"response.completed","response":{"usage":{"input_tokens":0,"output_tokens":0}}}"#,
            "",
        ];
        let (decoder, _text, events) = drive(&lines);
        let any_usage = events
            .iter()
            .any(|e| matches!(e, StreamEvent::Session(SessionEvent::Usage { .. })));
        assert!(!any_usage, "zero-token usage should be suppressed");
        assert_eq!(decoder.terminal_result(), Ok(()));
    }

    #[test]
    fn response_failed_is_a_terminal_failure_without_retaining_provider_text() {
        let lines = [
            r#"data: {"type":"response.failed","response":{"error":{"type":"server_error","message":"ECHOED:secret prompt"}}}"#,
            "",
        ];
        let (decoder, text, events) = drive(&lines);
        let failure = decoder.terminal_result().expect_err("must fail");
        // Exact server_error is now preserved as a transient capacity class.
        assert_eq!(failure.code, "responses_capacity");
        assert!(failure.is_retryable());
        assert_eq!(failure.message, format!("Codex{RESPONSES_CAPACITY_SUFFIX}"));
        assert!(!failure.message.contains("ECHOED"));
        assert!(text.is_empty());
        assert!(events.is_empty());
    }

    #[test]
    fn top_level_error_event_is_a_terminal_failure() {
        let lines = [
            r#"data: {"type":"error","error":{"type":"server_error","message":"private"}}"#,
            "",
        ];
        let (decoder, _text, _events) = drive(&lines);
        assert_eq!(
            decoder.terminal_result().expect_err("must fail").code,
            "responses_capacity"
        );
    }

    #[test]
    fn response_incomplete_is_not_misreported_as_empty_success() {
        let lines = [
            r#"data: {"type":"response.incomplete","response":{"incomplete_details":{"reason":"max_output_tokens"}}}"#,
            "",
        ];
        let (decoder, _text, _events) = drive(&lines);
        let failure = decoder.terminal_result().expect_err("must fail");
        assert_eq!(failure.code, "responses_incomplete");
        assert!(failure.message.contains("incomplete"));
        assert!(!failure.message.contains("max_output_tokens"));
    }

    #[test]
    fn completed_without_text_or_tools_fails_at_transport_boundary() {
        let lines = [
            r#"data: {"type":"response.completed","response":{"usage":{"input_tokens":7,"output_tokens":0}}}"#,
            "",
        ];
        let (decoder, _text, _events) = drive(&lines);
        assert_eq!(
            decoder.terminal_result().expect_err("must fail").code,
            "responses_empty"
        );
    }

    #[test]
    fn eof_without_terminal_event_fails_closed() {
        let lines = [
            r#"data: {"type":"response.created","response":{"id":"resp_123"}}"#,
            "",
        ];
        let (decoder, _text, _events) = drive(&lines);
        assert_eq!(
            decoder.terminal_result().expect_err("must fail").code,
            "responses_missing_terminal"
        );
    }

    /// The local-log sanitizer must pass lowercase snake-case identifiers
    /// verbatim, collapse secret-shaped identifiers to a shape marker, and
    /// never emit free-form provider text.
    #[test]
    fn error_identifier_sanitizer_never_passes_freeform_text() {
        assert_eq!(sanitize_error_identifier(None), "absent");
        assert_eq!(
            sanitize_error_identifier(Some("rate_limit_exceeded")),
            "rate_limit_exceeded"
        );
        assert_eq!(
            sanitize_error_identifier(Some("context_length_exceeded")),
            "context_length_exceeded"
        );
        assert_eq!(
            sanitize_error_identifier(Some("some_new_code2")),
            "some_new_code2"
        );
        // Mixed-case / dashed tokens can be secret-shaped (API keys).
        assert_eq!(
            sanitize_error_identifier(Some("sk-Abc123XYZ")),
            "unlisted_identifier"
        );
        assert_eq!(
            sanitize_error_identifier(Some("Server.Error")),
            "unlisted_identifier"
        );
        assert_eq!(
            sanitize_error_identifier(Some("ECHOED: user secret prompt")),
            "unsafe_or_freeform"
        );
        assert_eq!(sanitize_error_identifier(Some("")), "unsafe_or_freeform");
        let long = "a".repeat(65);
        assert_eq!(sanitize_error_identifier(Some(&long)), "unsafe_or_freeform");
    }

    /// Context-window overflow maps to a specific, actionable message; every
    /// other terminal failure keeps the neutral details-withheld default. The
    /// mapping only ever reads sanitized enum identifiers, never free-text.
    #[test]
    fn terminal_failure_user_message_is_specific_only_for_known_codes() {
        assert!(user_message_for_terminal_failure(
            "Codex",
            "invalid_request_error",
            "context_length_exceeded",
            false
        )
        .contains("/compact"));
        // Code carried on `type` rather than `code` still maps.
        assert!(user_message_for_terminal_failure(
            "Codex",
            "context_length_exceeded",
            "absent",
            false
        )
        .contains("context window"));
        // Unknown/other codes keep the neutral default (no remedy claim).
        let default = user_message_for_terminal_failure(
            "Codex",
            "server_error",
            "unlisted_identifier",
            false,
        );
        assert!(default.starts_with("Codex response failed in stream."));
        assert!(!default.contains("/compact"));
        // Capacity gets the specific remedy, labelled for the provider.
        let capacity = user_message_for_terminal_failure("xAI", "absent", "absent", true);
        assert_eq!(
            capacity,
            "xAI reports the model is at capacity (retries exhausted). Try again in a few minutes or switch models with /model."
        );
    }

    /// The exact flat envelope xAI emits (HTTP 200, one SSE event, close):
    /// no nested `error` object, `code` null, free-text `message`. It must
    /// classify as retryable capacity — labelled for the provider, and the
    /// provider prose must never reach the surfaced message.
    #[test]
    fn xai_flat_capacity_error_is_retryable_capacity_failure() {
        let lines = [
            r#"data: {"sequence_number":0,"type":"error","code":null,"message":"The model is currently at capacity due to high demand. Please try again in a few minutes, or use a higher service tier for priority processing: https://docs.x.ai/developers/advanced-api-usage/priority-processing","param":null}"#,
            "",
        ];
        let (tx, mut rx) = mpsc::unbounded_channel();
        let mut decoder = CodexSseDecoder::for_provider("xAI");
        let mut text = String::new();
        for line in lines {
            decoder.push_line(line, &tx, &mut text);
        }
        decoder.finish();
        let failure = decoder.terminal_result().expect_err("must fail");
        assert_eq!(failure.code, "responses_capacity");
        assert!(failure.is_retryable());
        assert_eq!(
            failure.message,
            "xAI reports the model is at capacity (retries exhausted). Try again in a few minutes or switch models with /model."
        );
        for banned in ["high demand", "docs.x.ai", "service tier", "Codex"] {
            assert!(
                !failure.message.contains(banned),
                "provider text `{banned}` leaked: {}",
                failure.message
            );
        }
        assert!(text.is_empty());
        assert!(!decoder.emitted_any_output());
        assert!(rx.try_recv().is_err(), "no events for an error-only stream");
    }

    /// Flat `code` identifiers are honoured when no nested error object is
    /// present; only vetted codes classify as capacity.
    #[test]
    fn flat_code_classifies_capacity_only_for_vetted_codes() {
        let capacity = [
            r#"data: {"type":"error","code":"rate_limit_exceeded","message":"private"}"#,
            "",
        ];
        let (decoder, _, _) = drive(&capacity);
        let failure = decoder.terminal_result().expect_err("must fail");
        assert_eq!(failure.code, "responses_capacity");
        assert!(failure.is_retryable());

        let other = [
            r#"data: {"type":"error","code":"invalid_api_key","message":"private"}"#,
            "",
        ];
        let (decoder, _, _) = drive(&other);
        let failure = decoder.terminal_result().expect_err("must fail");
        assert_eq!(failure.code, "responses_failed");
        assert!(!failure.is_retryable());
        assert!(!failure.message.contains("private"));
    }

    #[test]
    fn provider_label_is_carried_into_every_terminal_message() {
        let (tx, _rx) = mpsc::unbounded_channel();
        let mut text = String::new();
        let mut decoder = CodexSseDecoder::for_provider("xAI");
        decoder.push_line(
            r#"data: {"type":"response.completed","response":{"usage":{"input_tokens":1,"output_tokens":0}}}"#,
            &tx,
            &mut text,
        );
        decoder.push_line("", &tx, &mut text);
        let failure = decoder.terminal_result().expect_err("empty");
        assert_eq!(failure.code, "responses_empty");
        assert!(failure.is_retryable());
        assert_eq!(
            failure.message,
            "xAI completed without text or tool output. Retry the request."
        );

        let decoder = CodexSseDecoder::for_provider("xAI");
        let failure = decoder.terminal_result().expect_err("missing terminal");
        assert_eq!(failure.code, "responses_missing_terminal");
        assert!(failure.is_retryable());
        assert!(failure.message.starts_with("xAI response stream ended"));

        let mut decoder = CodexSseDecoder::for_provider("xAI");
        decoder.push_line(
            r#"data: {"type":"response.incomplete","response":{"incomplete_details":{"reason":"max_output_tokens"}}}"#,
            &tx,
            &mut text,
        );
        decoder.push_line("", &tx, &mut text);
        let failure = decoder.terminal_result().expect_err("incomplete");
        assert_eq!(failure.code, "responses_incomplete");
        assert!(!failure.is_retryable());
        assert!(failure.message.starts_with("xAI response was incomplete."));
    }

    #[test]
    fn done_marker_finishes_decoder() {
        let lines = [
            r#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","call_id":"call_x","name":"bash"}}"#,
            "",
            r#"data: {"type":"response.function_call_arguments.delta","output_index":0,"delta":"{}"}"#,
            "",
            "data: [DONE]",
            "",
        ];
        let (decoder, _text, _events) = drive(&lines);
        // active_tools promoted to completed_tools by finish().
        assert_eq!(decoder.completed_tools.len(), 1);
        assert_eq!(decoder.completed_tools[0].id, "call_x");
        assert_eq!(decoder.completed_tools[0].function.arguments, "{}");
    }

    #[test]
    fn finish_is_idempotent_no_double_emit() {
        let lines = [
            r#"data: {"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","call_id":"call_y","name":"bash","arguments":"{}"}}"#,
            "",
        ];
        let (mut decoder, _text, _events) = drive(&lines);
        assert_eq!(decoder.completed_tools.len(), 1);

        // Calling finish() again must not duplicate the tool.
        decoder.finish();
        assert_eq!(
            decoder.completed_tools.len(),
            1,
            "finish() called twice must not double-emit"
        );
    }

    #[test]
    fn finish_drains_active_tools_for_state_hygiene() {
        // After [DONE], any leftover active tool entries should have been
        // promoted *and* drained from `active_tools`. This guards against
        // future code paths that re-call finish() (or new event types
        // that would otherwise re-iterate the old buffer).
        let lines = [
            r#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","call_id":"call_z","name":"bash"}}"#,
            "",
            "data: [DONE]",
            "",
        ];
        let (decoder, _text, _events) = drive(&lines);
        assert_eq!(decoder.completed_tools.len(), 1);
        assert!(
            decoder.active_tools.is_empty(),
            "active_tools must be drained after finish()"
        );
    }

    #[test]
    fn unknown_event_types_are_ignored() {
        let lines = [
            r#"data: {"type":"response.future_unknown_event","payload":{"x":1}}"#,
            "",
            r#"data: {"type":"response.output_text.delta","delta":"hi"}"#,
            "",
        ];
        let (_decoder, text_acc, _events) = drive(&lines);
        assert_eq!(text_acc, "hi");
    }
}

#[cfg(test)]
mod broker_stream_tests {
    //! Broker-boundary streaming tests: the OpenAI-compatible stream path is
    //! driven end-to-end through an in-process `LocalBroker` against a fake
    //! upstream. The runtime side supplies NO credential — the broker applies
    //! it — and SSE deltas are forwarded to the event channel in real time.

    use super::*;
    use crate::auth::{CredentialBroker, LocalBroker};
    use std::sync::Arc;

    async fn spawn_fake_openai_sse() -> (String, Arc<std::sync::Mutex<String>>) {
        use axum::routing::post;
        let seen_auth = Arc::new(std::sync::Mutex::new(String::new()));
        let seen = seen_auth.clone();
        let app = axum::Router::new().route(
            "/chat/completions",
            post(move |headers: axum::http::HeaderMap| {
                let seen = seen.clone();
                async move {
                    *seen.lock().unwrap() = headers
                        .get("authorization")
                        .and_then(|v| v.to_str().ok())
                        .unwrap_or("")
                        .to_string();
                    let body = concat!(
                        "data: {\"choices\":[{\"delta\":{\"content\":\"Hel\"}}]}\n\n",
                        "data: {\"choices\":[{\"delta\":{\"content\":\"lo\"}}]}\n\n",
                        "data: [DONE]\n\n",
                    );
                    ([("content-type", "text/event-stream")], body)
                }
            }),
        );
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move {
            axum::serve(listener, app).await.unwrap();
        });
        (format!("http://{addr}"), seen_auth)
    }

    #[tokio::test]
    async fn oai_stream_flows_through_broker_and_forwards_deltas() {
        let (upstream, seen_auth) = spawn_fake_openai_sse().await;
        let broker: Arc<dyn CredentialBroker> = Arc::new(LocalBroker::with_local_base_url(
            reqwest::Client::new(),
            upstream,
        ));
        let cfg = ProviderConfig {
            base_url: "unused-broker-derives-the-url".to_string(),
            model: "test-model".to_string(),
            provider: "local".to_string(),
        };
        let (tx, mut rx) = mpsc::unbounded_channel::<StreamEvent>();
        let cancel = tokio_util::sync::CancellationToken::new();

        let result = call_oai_stream_inner(
            &cfg,
            &broker,
            &[],
            &None,
            &[],
            &tx,
            None,
            None,
            0,
            agent_core::reasoning::ReasoningLevel::Medium,
            &cancel,
            &crate::runtime::trace::TraceContext::disabled(),
            true,
            false,
            0,
        )
        .await
        .expect("stream must complete");

        // Final Anthropic-shaped value carries the accumulated text.
        assert_eq!(result["content"][0]["text"], "Hello");

        // Deltas were forwarded in real time on the event channel.
        drop(tx);
        let mut streamed = String::new();
        while let Ok(ev) = rx.try_recv() {
            if let StreamEvent::Llm(crate::runtime::types::LlmEvent::Text(t)) = ev {
                streamed.push_str(&t);
            }
        }
        assert_eq!(streamed, "Hello");

        // The credential was applied by the broker, not by this call site.
        assert_eq!(&*seen_auth.lock().unwrap(), "Bearer local");
    }

    /// Broker errors surface as typed failures without opening a stream and
    /// without any credential material in the message.
    #[tokio::test]
    async fn suppressed_sync_route_does_not_enqueue_display_deltas() {
        let (upstream, _seen_auth) = spawn_fake_openai_sse().await;
        let broker: Arc<dyn CredentialBroker> = Arc::new(LocalBroker::with_local_base_url(
            reqwest::Client::new(),
            upstream,
        ));
        let cfg = ProviderConfig {
            base_url: "unused-broker-derives-the-url".to_string(),
            model: "test-model".to_string(),
            provider: "local".to_string(),
        };
        let (tx, mut rx) = mpsc::unbounded_channel::<StreamEvent>();
        let result = call_oai_stream_inner(
            &cfg,
            &broker,
            &[],
            &None,
            &[],
            &tx,
            None,
            None,
            0,
            agent_core::reasoning::ReasoningLevel::Medium,
            &tokio_util::sync::CancellationToken::new(),
            &crate::runtime::trace::TraceContext::disabled(),
            true,
            true,
            0,
        )
        .await
        .expect("stream must complete");
        assert_eq!(result["content"][0]["text"], "Hello");
        assert!(
            rx.try_recv().is_err(),
            "sync route must suppress display deltas at production"
        );
    }

    #[tokio::test]
    async fn oai_stream_broker_error_fails_closed() {
        // Unconfigured static provider → NotConfigured, no upstream contact.
        let broker: Arc<dyn CredentialBroker> = Arc::new(LocalBroker::new(reqwest::Client::new()));
        let cfg = ProviderConfig {
            base_url: String::new(),
            model: "m".to_string(),
            provider: "definitely-not-a-provider".to_string(),
        };
        let (tx, _rx) = mpsc::unbounded_channel::<StreamEvent>();
        let cancel = tokio_util::sync::CancellationToken::new();
        let err = call_oai_stream_inner(
            &cfg,
            &broker,
            &[],
            &None,
            &[],
            &tx,
            None,
            None,
            0,
            agent_core::reasoning::ReasoningLevel::Medium,
            &cancel,
            &crate::runtime::trace::TraceContext::disabled(),
            true,
            false,
            0,
        )
        .await
        .unwrap_err()
        .to_string();
        assert!(err.contains("unknown provider"), "got: {err}");
        assert!(!err.to_lowercase().contains("bearer"));
    }

    /// Phase 1 privacy (spec §5.1): a hostile upstream answers the streaming
    /// open with HTTP 500 whose body echoes the full request. The broker's
    /// transport error carries a body snippet; the runtime must not surface
    /// or log it — status + provider label only.
    #[tokio::test]
    async fn oai_stream_upstream_error_body_never_surfaces() {
        use axum::response::IntoResponse;
        const SENTINEL: &str = "PH1-OAI-BROKER-SENTINEL-91d2-RAW";
        let app = axum::Router::new().route(
            "/chat/completions",
            axum::routing::post(move |body: String| async move {
                (
                    axum::http::StatusCode::INTERNAL_SERVER_ERROR,
                    [("content-type", "application/json")],
                    format!("{{\"error\":{{\"message\":\"ECHOED:{body}\"}}}}"),
                )
                    .into_response()
            }),
        );
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });

        let broker: Arc<dyn CredentialBroker> = Arc::new(LocalBroker::with_local_base_url(
            reqwest::Client::new(),
            format!("http://{addr}"),
        ));
        let cfg = ProviderConfig {
            base_url: "unused-broker-derives-the-url".to_string(),
            model: "test-model".to_string(),
            provider: "local".to_string(),
        };
        let tools = vec![serde_json::json!({
            "name": "ph1_secret_tool_zz",
            "description": "internal tool schema",
            "input_schema": {"type": "object", "properties": {}}
        })];
        let msgs: Vec<crate::SharedMessage> = vec![Arc::new(
            serde_json::json!({"role": "user", "content": SENTINEL}),
        )];
        let (tx, _rx) = mpsc::unbounded_channel::<StreamEvent>();
        let cancel = tokio_util::sync::CancellationToken::new();

        let err = call_oai_stream_inner(
            &cfg,
            &broker,
            &tools,
            &Some(format!("system secret {SENTINEL}")),
            &msgs,
            &tx,
            None,
            None,
            0,
            agent_core::reasoning::ReasoningLevel::Medium,
            &cancel,
            &crate::runtime::trace::TraceContext::disabled(),
            true,
            false,
            0,
        )
        .await
        .expect_err("500 must fail")
        .to_string();

        assert!(err.contains("500"), "status must survive: {err}");
        for banned in ["ECHOED", SENTINEL, "ph1_secret_tool_zz", "system secret"] {
            assert!(
                !err.contains(banned),
                "provider body content `{banned}` leaked into the surfaced error: {err}"
            );
        }
    }
}

/// Provider display label for the xai-auth Responses route.
const XAI_PROVIDER_LABEL: &str = "xAI";

/// Broker identity + display label for one OpenAI-Responses-wire provider
/// routed through the shared streaming loop
/// ([`stream_broker_responses`]).
#[derive(Debug, Clone, Copy)]
pub(crate) struct ResponsesProfile {
    /// Broker provider key (credential + pinned origin + path allowlist).
    pub broker_provider: &'static str,
    /// Allowlisted broker path for the Responses endpoint.
    pub path: &'static str,
    /// User-facing provider name in stream errors and logs.
    pub label: &'static str,
}

pub(crate) const XAI_RESPONSES: ResponsesProfile = ResponsesProfile {
    broker_provider: "xai-auth",
    path: "/responses",
    label: XAI_PROVIDER_LABEL,
};

/// Meta Muse: `https://api.meta.ai` origin, Responses at `/v1/responses`.
pub(crate) const MUSE_RESPONSES: ResponsesProfile = ResponsesProfile {
    broker_provider: "muse",
    path: "/v1/responses",
    label: "Meta Muse",
};

#[allow(clippy::too_many_arguments)]
pub(crate) async fn call_xai_responses_stream_inner(
    cfg: &ProviderConfig,
    broker: &std::sync::Arc<dyn crate::auth::CredentialBroker>,
    tools_schema: &[Value],
    system_prompt: &Option<String>,
    messages: &[crate::SharedMessage],
    tx: &mpsc::UnboundedSender<StreamEvent>,
    max_tokens: Option<u32>,
    reasoning_level: agent_core::reasoning::ReasoningLevel,
    cancel: &tokio_util::sync::CancellationToken,
    max_retries: u32,
    trace: &crate::runtime::trace::TraceContext,
    exact_wire_bytes: bool,
) -> Result<Value, Box<dyn std::error::Error + Send + Sync>> {
    let (tools, names) = translate::tools_to_responses(tools_schema);
    let input = codex_input_messages(translate::messages_to_oai(messages, system_prompt, &names));
    // Pure, validated body construction — rejects unsupported reasoning
    // combinations BEFORE any broker credential access or network I/O.
    let body = build_xai_body(
        &cfg.model,
        reasoning_level,
        serde_json::to_value(input)?,
        tools,
        max_tokens,
    )
    .map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { e.into() })?;
    stream_broker_responses(
        XAI_RESPONSES,
        cfg,
        broker,
        body,
        &names,
        tools_schema,
        system_prompt,
        messages,
        tx,
        cancel,
        max_retries,
        trace,
        exact_wire_bytes,
    )
    .await
}

/// Dispatch a resolved `OpenAiResponses` route to its provider body builder:
/// `muse` → [`call_muse_responses_stream_inner`], otherwise xAI.
#[allow(clippy::too_many_arguments)]
pub(crate) async fn call_responses_route_inner(
    route_provider: &str,
    cfg: &ProviderConfig,
    broker: &std::sync::Arc<dyn crate::auth::CredentialBroker>,
    tools_schema: &[Value],
    system_prompt: &Option<String>,
    messages: &[crate::SharedMessage],
    tx: &mpsc::UnboundedSender<StreamEvent>,
    max_tokens: Option<u32>,
    reasoning_level: agent_core::reasoning::ReasoningLevel,
    cancel: &tokio_util::sync::CancellationToken,
    max_retries: u32,
    trace: &crate::runtime::trace::TraceContext,
    exact_wire_bytes: bool,
) -> Result<Value, Box<dyn std::error::Error + Send + Sync>> {
    if route_provider == MUSE_RESPONSES.broker_provider {
        call_muse_responses_stream_inner(
            cfg,
            broker,
            tools_schema,
            system_prompt,
            messages,
            tx,
            max_tokens,
            reasoning_level,
            cancel,
            max_retries,
            trace,
            exact_wire_bytes,
        )
        .await
    } else {
        call_xai_responses_stream_inner(
            cfg,
            broker,
            tools_schema,
            system_prompt,
            messages,
            tx,
            max_tokens,
            reasoning_level,
            cancel,
            max_retries,
            trace,
            exact_wire_bytes,
        )
        .await
    }
}

/// Meta Muse Responses route (`muse/<id>`). Mirrors the official Muse Code
/// request shape: system prompt as `instructions`, conversation as typed
/// input items, `store:false` (stateless), and a `prompt_cache_key` so the
/// server-side prompt cache routes repeated prefixes to the same cache.
#[allow(clippy::too_many_arguments)]
pub(crate) async fn call_muse_responses_stream_inner(
    cfg: &ProviderConfig,
    broker: &std::sync::Arc<dyn crate::auth::CredentialBroker>,
    tools_schema: &[Value],
    system_prompt: &Option<String>,
    messages: &[crate::SharedMessage],
    tx: &mpsc::UnboundedSender<StreamEvent>,
    max_tokens: Option<u32>,
    reasoning_level: agent_core::reasoning::ReasoningLevel,
    cancel: &tokio_util::sync::CancellationToken,
    max_retries: u32,
    trace: &crate::runtime::trace::TraceContext,
    exact_wire_bytes: bool,
) -> Result<Value, Box<dyn std::error::Error + Send + Sync>> {
    let (tools, names) = translate::tools_to_responses(tools_schema);
    let instructions = system_prompt.clone().unwrap_or_default();
    // The system prompt travels as `instructions`, never duplicated in input.
    let input = muse_input_messages(translate::messages_to_oai(messages, &None, &names));
    let cache_key = muse_prompt_cache_key(&instructions, input.first());
    // Validated BEFORE any broker credential access or network I/O.
    let body = build_muse_body(
        &cfg.model,
        reasoning_level,
        Value::Array(input),
        &instructions,
        tools,
        max_tokens,
        &cache_key,
    )
    .map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { e.into() })?;
    stream_broker_responses(
        MUSE_RESPONSES,
        cfg,
        broker,
        body,
        &names,
        tools_schema,
        system_prompt,
        messages,
        tx,
        cancel,
        max_retries,
        trace,
        exact_wire_bytes,
    )
    .await
}

/// Deterministic Muse cache-routing key derived from the stable prompt head
/// (instructions + first input item), like the Codex key: identical heads
/// share the prefix the server caches, and a /compact rewrite changes the
/// key together with the prefix it invalidates. Namespaced so it can never
/// collide with Codex keys.
pub(crate) fn muse_prompt_cache_key(instructions: &str, first_item: Option<&Value>) -> String {
    codex_prompt_cache_key(instructions, first_item).replacen("synaps-", "synaps-muse-", 1)
}

/// Pure, validated body construction for the Meta Muse Responses API.
///
/// Shape mirrors the official Muse Code client (verified by capture):
/// `model`, `instructions`, `input`, `store:false`, `stream:true`, `tools`,
/// `max_output_tokens`, `reasoning.effort`, `prompt_cache_key`. Reasoning is
/// validated against the exact model's capability first; `Adaptive` omits
/// the field (provider default). Encrypted reasoning is not requested:
/// Synaps does not replay reasoning items across turns.
#[allow(clippy::too_many_arguments)] // pure builder mirroring the wire body
pub(crate) fn build_muse_body(
    model: &str,
    level: agent_core::reasoning::ReasoningLevel,
    input: Value,
    instructions: &str,
    tools: Vec<Value>,
    max_tokens: Option<u32>,
    prompt_cache_key: &str,
) -> Result<Value, String> {
    crate::runtime::openai::catalog::validation::validate_reasoning_mutation(
        &format!("muse/{model}"),
        level,
    )?;
    let mut body = serde_json::Map::new();
    body.insert("model".into(), json!(model));
    if !instructions.is_empty() {
        body.insert("instructions".into(), json!(instructions));
    }
    body.insert("input".into(), input);
    body.insert("store".into(), json!(false));
    body.insert("stream".into(), json!(true));
    if !tools.is_empty() {
        body.insert("tools".into(), Value::Array(tools));
    }
    if let Some(max) = max_tokens {
        body.insert("max_output_tokens".into(), json!(max));
    }
    if let Some(effort) = crate::runtime::openai::catalog::muse_effort_label(level) {
        body.insert("reasoning".into(), json!({ "effort": effort }));
    }
    body.insert("prompt_cache_key".into(), json!(prompt_cache_key));
    Ok(Value::Object(body))
}

/// Shared streaming loop for broker-proxied OpenAI Responses providers
/// (xAI, Meta Muse): one exact-bytes request, SSE decode, bounded retries
/// that never replay after output reached the UI.
#[allow(clippy::too_many_arguments)]
async fn stream_broker_responses(
    profile: ResponsesProfile,
    cfg: &ProviderConfig,
    broker: &std::sync::Arc<dyn crate::auth::CredentialBroker>,
    body: Value,
    names: &translate::ToolNameMap,
    tools_schema: &[Value],
    system_prompt: &Option<String>,
    messages: &[crate::SharedMessage],
    tx: &mpsc::UnboundedSender<StreamEvent>,
    cancel: &tokio_util::sync::CancellationToken,
    max_retries: u32,
    trace: &crate::runtime::trace::TraceContext,
    exact_wire_bytes: bool,
) -> Result<Value, Box<dyn std::error::Error + Send + Sync>> {
    // Serialize ONCE via the sanctioned constructor: the digested bytes are
    // the very bytes the broker sends upstream (`body_bytes` handoff,
    // LocalBroker verbatim) — `body`/`body_bytes` cannot diverge. On a
    // remote broker the daemon re-serializes out of process → honest
    // `CloudProxy` kind and no wire-byte claim. Every retry clones this
    // same request (identical `body` and `body_bytes`), so one wire digest
    // is honest for all attempt records.
    let (proxy_request, body_bytes) = crate::auth::ProxyRequest::post_json_exact(
        profile.broker_provider,
        profile.path,
        body,
        true,
    )?;
    let tracer = tr::begin_openai_tracer(
        trace,
        &cfg.provider,
        &cfg.model,
        if exact_wire_bytes {
            crate::runtime::trace::TransportKind::OpenAiResponses
        } else {
            crate::runtime::trace::TransportKind::CloudProxy
        },
        &format!("{}/responses", cfg.base_url.trim_end_matches('/')),
        exact_wire_bytes.then_some(body_bytes.as_ref()),
        messages,
        system_prompt.as_deref(),
        tools_schema,
        tr::renamed_tool_losses(names),
    )
    .await;
    // One-shot explicit content capture (`/trace next content`): a no-op in
    // every context without the arm. `body_bytes` is the serialized request
    // body this process built pre-send (exact wire bytes on the local
    // broker; the same body a remote broker re-serializes) — body only,
    // headers and credentials structurally never reach this seam.
    if let Some(tracer) = &tracer {
        tr::capture_request_content(trace, tracer.request_id(), body_bytes.as_ref(), messages);
    }
    let mut attempt = tr::StreamAttempt::new(tracer);
    let mut stream_retry = 0u32;

    loop {
        if cancel.is_cancelled() {
            return Err("request canceled".into());
        }
        let stream = crate::runtime::api::await_or_cancel(
            cancel,
            broker.proxy_stream(proxy_request.clone()),
        )
        .await
        .map_err(|_| "request canceled")?;
        // A credential refresh failure keeps its kind as a type (retryable /
        // action-required vs credential); its text is ours, no upstream bytes.
        let stream = match stream {
            Err(crate::auth::BrokerError::Refresh(f)) => {
                attempt.finish_failed("broker_error", None, None);
                return Err(Box::new(super::net::BrokerRefreshError::new(f)));
            }
            // Privacy (spec §5.1): a broker proxy error may carry an upstream
            // response-body snippet; redact it to status-only before surfacing.
            other => other.map_err(|e| super::net::redact_provider_proxy_error(&e.to_string())),
        };
        let mut stream = match stream {
            Ok(stream) => {
                attempt.mark_headers();
                stream
            }
            Err(msg) => {
                let status = tr::broker_error_status(&msg);
                let code =
                    status.map_or_else(|| "broker_error".to_string(), |s| format!("http_{s}"));
                attempt.finish_failed(&code, status, None);
                return Err(super::net::normalize_broker_http_error(msg).into());
            }
        };
        let _ = tx.send(StreamEvent::Llm(crate::LlmEvent::ResponseStart));
        let mut interrupted = false;
        let mut text = String::new();
        let mut parser = CodexSseDecoder::for_provider(profile.label);
        parser.backfill_message_text = profile.broker_provider == MUSE_RESPONSES.broker_provider;
        let mut buf = bytes::BytesMut::new();
        while let Some(chunk) = tokio::select! {
            biased;
            _ = cancel.cancelled() => {
                attempt.finish_canceled(None, parser.trace_usage());
                return Err("request canceled".into());
            }
            c = stream.next() => c,
        } {
            let chunk = match chunk {
                Ok(chunk) => {
                    attempt.mark_first_byte();
                    chunk
                }
                Err(e) => {
                    if !matches!(e, crate::auth::BrokerError::Transport(_)) {
                        attempt.finish_failed("broker_error", None, None);
                        return Err(e.into());
                    }
                    interrupted = !parser.terminal_success && parser.terminal_failure.is_none();
                    break;
                }
            };
            buf.extend_from_slice(&chunk);
            while let Some(n) = memchr::memchr(b'\n', &buf) {
                let line = buf.split_to(n + 1);
                parser.push_line(std::str::from_utf8(&line[..n]).unwrap_or(""), tx, &mut text);
            }
            if parser.saw_model_event {
                attempt.mark_first_model_event();
            }
        }
        if !interrupted && !buf.is_empty() {
            parser.push_line(std::str::from_utf8(&buf).unwrap_or(""), tx, &mut text);
        }
        if interrupted || (!parser.terminal_success && parser.terminal_failure.is_none()) {
            if stream_retry >= max_retries {
                attempt.finish_failed("stream_error", None, None);
                return Err(STREAM_INTERRUPTED.into());
            }
            stream_retry += 1;
            attempt.attempt_failed(
                crate::runtime::trace::RetryClass::Network,
                retry_delay(stream_retry),
                None,
                None,
                "stream_error",
            );
            wait_stream_retry(tx, cancel, stream_retry, max_retries, false)
                .await
                .inspect_err(|_| {
                    attempt.finish_canceled(None, None);
                })?;
            attempt.restart_clock();
            continue;
        }
        parser.finish();
        // Trailing-buffer flush and finish() can surface the first (or only)
        // model event; mark it so first_model_event_ms is honest for tail-only
        // streams (mirrors the Codex direct path above).
        if parser.saw_model_event {
            attempt.mark_first_model_event();
        }
        // Broker paths never observe the upstream HTTP status; the Responses
        // stream does not yet expose a normalized stop reason — both stay
        // honest `None` rather than guessed values.
        if let Err(failure) = parser.terminal_result() {
            // xAI answers HTTP 200 + a lone `{"type":"error",…}` capacity
            // event while the model is saturated; a re-send seconds later
            // usually succeeds. Retry only when NOTHING reached the UI on
            // this attempt — text or tool events already streamed would be
            // emitted twice on a re-send, so those failures stay terminal.
            let output_already_streamed = !text.is_empty() || parser.emitted_any_output();
            if failure.is_retryable() && !output_already_streamed && stream_retry < max_retries {
                stream_retry += 1;
                let delay = retry_delay(stream_retry);
                attempt.attempt_failed(
                    crate::runtime::trace::RetryClass::Other,
                    delay,
                    None,
                    None,
                    failure.code,
                );
                tracing::warn!(
                    code = failure.code,
                    provider = profile.broker_provider,
                    "responses stream retry {stream_retry}/{max_retries} after {delay:?}"
                );
                tokio::select! {
                    _ = tokio::time::sleep(delay) => {}
                    _ = cancel.cancelled() => {
                        attempt.finish_canceled(None, None);
                        return Err("request canceled".into());
                    }
                }
                attempt.restart_clock();
                continue;
            }
            attempt.finish_failed(failure.code, None, None);
            return Err(failure.message.into());
        }
        attempt.finish_success(None, None, None, parser.trace_usage());
        let mut content = Vec::new();
        if !text.is_empty() {
            content.push(json!({"type":"text","text":text}));
        }
        content.extend(translate::tool_calls_to_content_blocks(
            &parser.completed_tools,
            names,
        ));
        let mut response = json!({"role":"assistant","content":content});
        if profile.broker_provider == MUSE_RESPONSES.broker_provider {
            if let Some(reason) = muse_dangling_preamble(
                !parser.completed_tools.is_empty(),
                parser.last_message_phase.as_deref(),
                parser.last_message_text().unwrap_or(&text),
            ) {
                tracing::info!(
                    provider = profile.broker_provider,
                    reason,
                    "muse response ended on a preamble without acting"
                );
                // Top-level sidecar, never inside `content`: the turn loop
                // persists only `content`, so nothing reaches stored history.
                response[DANGLING_PREAMBLE_KEY] = json!(reason);
            }
        }
        return Ok(response);
    }
}

/// Response sidecar key (top level, outside `content`) set by the Muse route
/// when a completed response ended on a preamble without acting. Read by the
/// turn loop's one-shot continuation; never persisted.
pub(crate) const DANGLING_PREAMBLE_KEY: &str = "synaps_dangling_preamble";

/// Upper bound (chars, trimmed) for the text heuristic. The observed failures
/// were 32–36 output tokens (~70–130 chars); 300 leaves room for a long
/// sentence of preamble while excluding real answers that merely end in a
/// colon after substantive content.
pub(crate) const PREAMBLE_MAX_CHARS: usize = 300;

/// Decide whether a completed Muse response ended on an unfinished preamble.
/// Returns the trigger reason, or `None`. Never fires when the response
/// called a tool.
///
/// - `commentary_phase`: the last message item was labelled
///   `phase:"commentary"` by Muse itself, yet no tool call followed — by the
///   provider's own contract the turn is unfinished.
/// - `colon_preamble`: the last message's trimmed text is short
///   (≤ [`PREAMBLE_MAX_CHARS`]) and ends with ':' — "Blocking question
///   before I build anything:" with nothing after it.
pub(crate) fn muse_dangling_preamble(
    has_tool_calls: bool,
    last_message_phase: Option<&str>,
    last_message_text: &str,
) -> Option<&'static str> {
    if has_tool_calls {
        return None;
    }
    if last_message_phase == Some("commentary") {
        return Some("commentary_phase");
    }
    let text = last_message_text.trim();
    if (text.ends_with(':') || text.ends_with('\u{ff1a}'))
        && text.chars().count() <= PREAMBLE_MAX_CHARS
    {
        return Some("colon_preamble");
    }
    None
}

/// Harness-authored nudge appended (once per user turn) after a dangling
/// preamble. Framed like the context advisories so the model does not read
/// it as a new user request.
pub(crate) const PREAMBLE_NUDGE: &str = "[Host harness note, not a new user request] You ended your turn on a preamble without acting. Continue now: run the tool call you announced, or ask your question in full.";

/// Visible system note shown when the nudge fires.
pub(crate) const PREAMBLE_NUDGE_NOTICE: &str = "Muse ended its turn on a preamble without acting; continuing once automatically (disable: SYNAPS_MUSE_PREAMBLE_NUDGE=0).";

/// `SYNAPS_MUSE_PREAMBLE_NUDGE=0|false|off|no` disables the one-shot
/// continuation. Enabled by default.
pub(crate) fn preamble_nudge_enabled() -> bool {
    preamble_nudge_enabled_value(std::env::var("SYNAPS_MUSE_PREAMBLE_NUDGE").ok().as_deref())
}

fn preamble_nudge_enabled_value(value: Option<&str>) -> bool {
    !matches!(
        value.map(|v| v.trim().to_ascii_lowercase()).as_deref(),
        Some("0" | "false" | "off" | "no")
    )
}

/// Turn-loop gate for the one-shot preamble continuation. Returns the reason
/// and arms `already_nudged` when the response carries the Muse sidecar, the
/// round was not cancelled, the feature is enabled, and this user turn has
/// not been nudged yet. At most one nudge per user turn: it cannot loop.
pub(crate) fn take_preamble_nudge(
    response: &Value,
    already_nudged: &mut bool,
    cancelled: bool,
    enabled: bool,
) -> Option<String> {
    if *already_nudged || cancelled || !enabled {
        return None;
    }
    let reason = response.get(DANGLING_PREAMBLE_KEY)?.as_str()?.to_string();
    *already_nudged = true;
    Some(reason)
}

/// Pure, validated body construction for the xAI Responses API.
///
/// Enforces the exact-model reasoning matrix from
/// `docs/specs/anthropic-xai-reasoning-modes-spec.md` BEFORE any credential or
/// network access, via the shared per-provider validator:
/// - `Off` is rejected (never silently omitted) on models whose reasoning
///   cannot be disabled; on the non-reasoning model it is trivially omission.
/// - `Adaptive` omits the `reasoning` field → documented provider default.
/// - Named levels are emitted as exact `reasoning:{effort:"..."}` only when
///   the exact model id documents them; otherwise `Err` (fail closed).
pub(crate) fn build_xai_body(
    model: &str,
    level: agent_core::reasoning::ReasoningLevel,
    input: Value,
    tools: Vec<Value>,
    max_tokens: Option<u32>,
) -> Result<Value, String> {
    use agent_core::reasoning::ReasoningLevel;
    crate::runtime::openai::catalog::validation::validate_reasoning_mutation(
        &format!("xai-auth/{model}"),
        level,
    )?;
    let mut body = serde_json::Map::new();
    body.insert("model".into(), json!(model));
    body.insert("input".into(), input);
    body.insert("stream".into(), json!(true));
    if !tools.is_empty() {
        body.insert("tools".into(), Value::Array(tools));
    }
    if let Some(max) = max_tokens {
        body.insert("max_output_tokens".into(), json!(max));
    }
    match level {
        // Off (validated: only reachable where reasoning is absent/disableable)
        // and Adaptive: omit `reasoning` — provider default applies.
        ReasoningLevel::Off | ReasoningLevel::Adaptive => {}
        l => {
            body.insert("reasoning".into(), json!({ "effort": l.as_str() }));
        }
    }
    Ok(Value::Object(body))
}

#[cfg(test)]
mod xai_tests {
    use super::*;
    use agent_core::reasoning::ReasoningLevel;

    #[test]
    fn xai_fixture_is_public_responses_shape() {
        let fixture = serde_json::json!({"type":"response.output_text.delta","delta":"hello"});
        assert_eq!(fixture["type"], "response.output_text.delta");
        assert_eq!(fixture["delta"], "hello");
    }

    fn body_for(model: &str, level: ReasoningLevel) -> Result<Value, String> {
        build_xai_body(model, level, json!([]), Vec::new(), Some(1024))
    }

    // ── Exact Responses wire (spec: anthropic-xai-reasoning-modes) ───────────

    #[test]
    fn grok45_emits_exact_documented_efforts() {
        for (level, effort) in [
            (ReasoningLevel::Low, "low"),
            (ReasoningLevel::Medium, "medium"),
            (ReasoningLevel::High, "high"),
        ] {
            let body = body_for("grok-4.5", level).expect("supported effort");
            assert_eq!(body["reasoning"], json!({"effort": effort}), "{level}");
            assert_eq!(body["model"], "grok-4.5");
            assert_eq!(body["stream"], json!(true));
            assert_eq!(body["max_output_tokens"], json!(1024));
        }
    }

    #[test]
    fn grok47_responses_wire_uses_exact_efforts_and_preserves_tool_shape() {
        for (level, effort) in [
            (ReasoningLevel::Low, "low"),
            (ReasoningLevel::Medium, "medium"),
            (ReasoningLevel::High, "high"),
            (ReasoningLevel::XHigh, "xhigh"),
        ] {
            let input = json!([{"role": "user", "content": "hello"}]);
            let tool = json!({"type": "function", "name": "inspect", "parameters": {"type": "object"}});
            let body = build_xai_body("grok-4.7", level, input.clone(), vec![tool.clone()], Some(1024)).unwrap();
            assert_eq!(body["model"], "grok-4.7");
            assert_eq!(body["reasoning"], json!({"effort": effort}));
            assert_eq!(body["input"], input);
            assert_eq!(body["tools"], json!([tool]));
            assert_eq!(body["stream"], true);
            assert_eq!(body["max_output_tokens"], 1024);
            assert!(body.get("messages").is_none());
            assert!(body.get("reasoning_effort").is_none());
            // Default ciphertext does not require opting into include or
            // changing storage policy (official docs say ignoring it works).
            assert!(body.get("include").is_none());
            assert!(body.get("store").is_none());
        }
        assert!(body_for("grok-4.7", ReasoningLevel::Adaptive).unwrap().get("reasoning").is_none());
        for level in [ReasoningLevel::Off, ReasoningLevel::Max, ReasoningLevel::Ultra, ReasoningLevel::UltraCode] {
            assert!(body_for("grok-4.7", level).is_err(), "{level}");
        }
    }

    #[test]
    fn grok46_emits_exact_documented_efforts() {
        for (level, effort) in [
            (ReasoningLevel::Low, "low"),
            (ReasoningLevel::Medium, "medium"),
            (ReasoningLevel::High, "high"),
        ] {
            let body = body_for("grok-4.6", level).expect("supported effort");
            assert_eq!(body["reasoning"], json!({"effort": effort}), "{level}");
            assert_eq!(body["model"], "grok-4.6");
            assert_eq!(body["stream"], json!(true));
            assert_eq!(body["max_output_tokens"], json!(1024));
        }
    }

    #[test]
    fn adaptive_omits_reasoning_field_provider_default() {
        for model in [
            "grok-4.5",
            "grok-4.5-latest",
            "grok-4.6",
            "grok-4.3",
            "grok-4.20-0309-non-reasoning",
        ] {
            let body = body_for(model, ReasoningLevel::Adaptive).expect("adaptive is omission");
            assert!(body.get("reasoning").is_none(), "{model}");
        }
    }

    #[test]
    fn off_is_rejected_pre_network_when_reasoning_cannot_be_disabled() {
        for model in [
            "grok-4.5",
            "grok-4.5-latest",
            "grok-4.6",
            "grok-4.20-multi-agent-0309",
            "grok-4.3",
        ] {
            let err = body_for(model, ReasoningLevel::Off).unwrap_err();
            assert!(err.contains(model), "{model}: {err}");
        }
        // Non-reasoning model: Off is trivially satisfied by omission.
        let body = body_for("grok-4.20-0309-non-reasoning", ReasoningLevel::Off).unwrap();
        assert!(body.get("reasoning").is_none());
    }

    #[test]
    fn unsupported_named_efforts_are_rejected_pre_network() {
        // 4.5/4.6 have no documented xhigh; max/ultra never exist on xAI.
        for level in [
            ReasoningLevel::XHigh,
            ReasoningLevel::Max,
            ReasoningLevel::Ultra,
        ] {
            assert!(body_for("grok-4.5", level).is_err(), "{level}");
            assert!(body_for("grok-4.6", level).is_err(), "{level}");
        }
        // Intrinsic-reasoning models have no documented effort control.
        assert!(body_for("grok-4.3", ReasoningLevel::Medium).is_err());
        // Non-reasoning models reject named reasoning outright.
        assert!(body_for("grok-4.20-0309-non-reasoning", ReasoningLevel::Medium).is_err());
        // Unknown exact ids fail closed.
        assert!(body_for("grok-9000", ReasoningLevel::Medium).is_err());
    }

    #[test]
    fn multi_agent_xhigh_is_exact_agent_count_control() {
        let body = body_for("grok-4.20-multi-agent-0309", ReasoningLevel::XHigh).unwrap();
        assert_eq!(body["reasoning"], json!({"effort": "xhigh"}));
    }

    #[test]
    fn responses_wire_shape_tools_flat_and_no_chat_completions_fields() {
        let tools = vec![json!({
            "type": "function",
            "name": "get_weather",
            "description": "d",
            "parameters": {"type": "object"}
        })];
        let body =
            build_xai_body("grok-4.5", ReasoningLevel::High, json!([]), tools, None).unwrap();
        // Responses API: flat tool objects + `input`, never chat-completions
        // `messages`/nested function wrappers.
        assert_eq!(body["tools"][0]["name"], "get_weather");
        assert!(body["tools"][0].get("function").is_none());
        assert!(body.get("input").is_some());
        assert!(body.get("messages").is_none());
        assert!(body.get("max_output_tokens").is_none());
    }
}

#[cfg(test)]
mod xai_capacity_retry_tests {
    //! Regression tests for the xai-auth Responses route: xAI answers HTTP
    //! 200 + a lone flat `{"type":"error",…}` capacity event while the model
    //! is saturated (reproduced live: grok-4.6 failed 3 of 4 turns, a re-send
    //! seconds later succeeded). The route must retry with identical request
    //! bytes, label its messages "xAI" (not "Codex"), and never surface the
    //! provider prose. A scripted fake broker serves the SSE bodies directly
    //! through `proxy_stream` — no HTTP server involved.

    use super::*;
    use agent_core::auth::{
        AccessToken, BrokerError, ProxyByteStream, ProxyRequest, ProxyResponse,
    };
    use async_trait::async_trait;
    use std::sync::{
        atomic::{AtomicUsize, Ordering},
        Arc,
    };

    /// The exact envelope observed on the wire (flat: `type` is the event
    /// type, `code` null, `message` top-level, no nested `error` object).
    const XAI_CAPACITY_SSE: &str = "data: {\"sequence_number\":0,\"type\":\"error\",\"code\":null,\"message\":\"The model is currently at capacity due to high demand. Please try again in a few minutes, or use a higher service tier for priority processing: https://docs.x.ai/developers/advanced-api-usage/priority-processing\",\"param\":null}\n\n";

    const XAI_SUCCESS_SSE: &str = concat!(
        "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_1\"}}\n\n",
        "data: {\"type\":\"response.output_text.delta\",\"delta\":\"GROK_OK\"}\n\n",
        "data: {\"type\":\"response.completed\",\"response\":{\"usage\":{\"input_tokens\":7,\"output_tokens\":2}}}\n\n",
    );

    /// Serves `bodies[n]` for the n-th `proxy_stream` call (the last body
    /// repeats once exhausted) and records every request body it saw.
    struct ScriptedXaiBroker {
        bodies: Vec<&'static str>,
        calls: Arc<AtomicUsize>,
        seen: Arc<std::sync::Mutex<Vec<ProxyRequest>>>,
    }

    #[async_trait]
    impl crate::auth::CredentialBroker for ScriptedXaiBroker {
        async fn access_token(
            &self,
            _p: agent_core::auth::OAuthProviderId,
        ) -> Result<AccessToken, BrokerError> {
            Err(BrokerError::Denied(
                "xai route must use proxy_stream".into(),
            ))
        }
        async fn proxy(&self, _request: ProxyRequest) -> Result<ProxyResponse, BrokerError> {
            Err(BrokerError::Denied("not implemented in stub".into()))
        }
        async fn proxy_stream(
            &self,
            request: ProxyRequest,
        ) -> Result<ProxyByteStream, BrokerError> {
            request.validate()?;
            let n = self.calls.fetch_add(1, Ordering::SeqCst);
            self.seen.lock().unwrap().push(request);
            let body = self.bodies[n.min(self.bodies.len() - 1)];
            Ok(Box::pin(futures::stream::iter(vec![Ok(
                bytes::Bytes::from(body),
            )])))
        }
        async fn anthropic_usage(&self) -> Result<serde_json::Value, BrokerError> {
            Err(BrokerError::Denied("not implemented in stub".into()))
        }
        async fn capabilities(&self) -> Result<Vec<agent_core::auth::ProviderStatus>, BrokerError> {
            Ok(vec![])
        }
    }

    /// Broker whose `proxy_stream` fails with a typed refresh failure (a
    /// Muse re-mint that hit the network, a corrupt store).
    struct RefreshFailingBroker(agent_core::auth::RefreshFailureKind);

    #[async_trait]
    impl crate::auth::CredentialBroker for RefreshFailingBroker {
        async fn access_token(
            &self,
            _p: agent_core::auth::OAuthProviderId,
        ) -> Result<AccessToken, BrokerError> {
            Err(BrokerError::Denied("stub".into()))
        }
        async fn proxy(&self, _request: ProxyRequest) -> Result<ProxyResponse, BrokerError> {
            Err(BrokerError::Denied("stub".into()))
        }
        async fn proxy_stream(&self, _r: ProxyRequest) -> Result<ProxyByteStream, BrokerError> {
            Err(BrokerError::Refresh(agent_core::auth::RefreshFailure::new(
                self.0,
                "transport error",
            )))
        }
        async fn anthropic_usage(&self) -> Result<serde_json::Value, BrokerError> {
            Err(BrokerError::Denied("stub".into()))
        }
        async fn capabilities(&self) -> Result<Vec<agent_core::auth::ProviderStatus>, BrokerError> {
            Ok(vec![])
        }
    }

    /// Round-2 m-a: the retryable kind is carried as a type through the real
    /// pipeline (redaction, normalization, classification), not recovered
    /// from our hint text. Retryable and action-required refresh failures
    /// are API-side, never "Config error" and never "sign in again".
    #[tokio::test]
    async fn broker_refresh_failure_kind_survives_the_stream_pipeline() {
        use agent_core::auth::RefreshFailureKind as K;
        for (kind, api) in [
            (K::Transient, true),
            (K::LockBusy, true),
            (K::Other, true),
            (K::ActionRequired, true),
            (K::Credential, false),
        ] {
            let broker: Arc<dyn crate::auth::CredentialBroker> =
                Arc::new(RefreshFailingBroker(kind));
            let cfg = ProviderConfig {
                base_url: "https://api.meta.ai/v1".to_string(),
                model: "muse-x".to_string(),
                provider: "muse".to_string(),
            };
            let (tx, _rx) = mpsc::unbounded_channel::<StreamEvent>();
            let err = call_xai_responses_stream_inner(
                &cfg,
                &broker,
                &[],
                &None,
                &[],
                &tx,
                None,
                agent_core::reasoning::ReasoningLevel::Adaptive,
                &tokio_util::sync::CancellationToken::new(),
                0,
                &crate::runtime::trace::TraceContext::disabled(),
                true,
            )
            .await
            .unwrap_err();
            let runtime = super::super::net::provider_error_to_runtime_for("muse/muse-x", err);
            match (&runtime, api) {
                (crate::RuntimeError::ApiStatus(m), true) => {
                    assert!(m.contains("credential refresh failed"), "{kind:?}: {m}");
                    assert!(!m.contains("Sign in again"), "{kind:?}: {m}");
                }
                (crate::RuntimeError::Config(_), false) => {}
                other => panic!("{kind:?}: {other:?}"),
            }
        }
    }

    struct Run {
        result: Result<Value, Box<dyn std::error::Error + Send + Sync>>,
        calls: usize,
        seen: Vec<ProxyRequest>,
        events: Vec<StreamEvent>,
    }

    async fn run_xai(bodies: Vec<&'static str>, max_retries: u32) -> Run {
        let calls = Arc::new(AtomicUsize::new(0));
        let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
        let broker: Arc<dyn crate::auth::CredentialBroker> = Arc::new(ScriptedXaiBroker {
            bodies,
            calls: Arc::clone(&calls),
            seen: Arc::clone(&seen),
        });
        let cfg = ProviderConfig {
            base_url: "https://api.x.ai/v1".to_string(),
            model: "grok-4.5".to_string(),
            provider: "xai-auth".to_string(),
        };
        let (tx, mut rx) = mpsc::unbounded_channel::<StreamEvent>();
        let result = call_xai_responses_stream_inner(
            &cfg,
            &broker,
            &[],
            &Some("system".to_string()),
            &[],
            &tx,
            None,
            agent_core::reasoning::ReasoningLevel::Adaptive,
            &tokio_util::sync::CancellationToken::new(),
            max_retries,
            &crate::runtime::trace::TraceContext::disabled(),
            true,
        )
        .await;
        drop(tx);
        let mut events = Vec::new();
        while let Ok(event) = rx.try_recv() {
            events.push(event);
        }
        let seen = std::mem::take(&mut *seen.lock().unwrap());
        Run {
            result,
            calls: calls.load(Ordering::SeqCst),
            seen,
            events,
        }
    }

    fn text_events(events: &[StreamEvent]) -> Vec<String> {
        events
            .iter()
            .filter_map(|e| match e {
                StreamEvent::Llm(crate::runtime::types::LlmEvent::Text(t)) => Some(t.clone()),
                _ => None,
            })
            .collect()
    }

    #[tokio::test(start_paused = true)]
    async fn xai_capacity_error_is_retried_with_identical_bytes_then_succeeds() {
        let run = run_xai(vec![XAI_CAPACITY_SSE, XAI_SUCCESS_SSE], 3).await;
        let value = run.result.expect("second attempt must succeed");
        assert_eq!(value["content"][0]["text"], "GROK_OK");
        assert_eq!(run.calls, 2, "exactly one retry");
        assert_eq!(run.seen.len(), 2);
        assert_eq!(
            run.seen[0].body_bytes, run.seen[1].body_bytes,
            "retry must re-send the identical serialized body"
        );
        assert_eq!(run.seen[0].body, run.seen[1].body);
        assert_eq!(run.seen[0].provider, "xai-auth");
        assert_eq!(run.seen[0].path, "/responses");
        // No output was streamed twice: the failed attempt emitted nothing.
        assert_eq!(text_events(&run.events), vec!["GROK_OK".to_string()]);
    }

    #[tokio::test(start_paused = true)]
    async fn xai_capacity_error_with_zero_retries_surfaces_xai_capacity_message() {
        let run = run_xai(vec![XAI_CAPACITY_SSE], 0).await;
        let err = run.result.expect_err("no retry budget → terminal");
        let msg = err.to_string();
        assert_eq!(
            msg,
            "xAI reports the model is at capacity (retries exhausted). Try again in a few minutes or switch models with /model."
        );
        assert_eq!(run.calls, 1);
        for banned in ["Codex", "high demand", "docs.x.ai", "service tier"] {
            assert!(!msg.contains(banned), "`{banned}` leaked: {msg}");
        }
        // And the runtime classifies it as an API failure, not a config one.
        assert!(matches!(
            super::super::net::provider_error_to_runtime(err),
            crate::RuntimeError::ApiStatus(_)
        ));
    }

    #[tokio::test(start_paused = true)]
    async fn xai_capacity_error_exhausting_budget_counts_every_attempt() {
        let run = run_xai(vec![XAI_CAPACITY_SSE], 2).await;
        let err = run.result.expect_err("persistent capacity → terminal");
        assert!(err
            .to_string()
            .starts_with("xAI reports the model is at capacity"));
        assert_eq!(run.calls, 3, "initial attempt + 2 retries");
        assert!(text_events(&run.events).is_empty());
    }

    /// If the failed attempt already streamed text, a re-send would duplicate
    /// it in the UI — the failure must stay terminal even with budget left.
    #[tokio::test(start_paused = true)]
    async fn xai_does_not_retry_after_partial_text_was_streamed() {
        const PARTIAL_THEN_ERROR: &str = concat!(
            "data: {\"type\":\"response.output_text.delta\",\"delta\":\"partial\"}\n\n",
            "data: {\"type\":\"error\",\"code\":null,\"message\":\"at capacity\"}\n\n",
        );
        let run = run_xai(vec![PARTIAL_THEN_ERROR, XAI_SUCCESS_SSE], 3).await;
        let err = run.result.expect_err("must not retry after partial output");
        assert!(err
            .to_string()
            .starts_with("xAI reports the model is at capacity"));
        assert_eq!(run.calls, 1, "no retry once output reached the UI");
        assert_eq!(text_events(&run.events), vec!["partial".to_string()]);
    }

    /// An unfinished tool preview is not execution; reset it before retrying.
    #[tokio::test(start_paused = true)]
    async fn xai_retries_eof_after_tool_preview_with_reset() {
        const TOOL_THEN_EOF: &str = "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_1\",\"name\":\"bash\"}}\n\n";
        let run = run_xai(vec![TOOL_THEN_EOF, XAI_SUCCESS_SSE], 3).await;
        assert_eq!(run.result.unwrap()["content"][0]["text"], "GROK_OK");
        assert_eq!(run.calls, 2);
        assert_eq!(run.seen[0].body_bytes, run.seen[1].body_bytes);
        assert!(run
            .events
            .iter()
            .any(|e| matches!(e, StreamEvent::Llm(crate::LlmEvent::ResponseReset))));
    }

    #[test]
    fn responses_exact_account_errors_survive_decoder_net_and_driver() {
        for (code, suffix, expected) in [
            ("invalid_api_key", RESPONSES_AUTH_SUFFIX, "auth"),
            ("authentication_error", RESPONSES_AUTH_SUFFIX, "auth"),
            (
                "invalid_authentication_error",
                RESPONSES_AUTH_SUFFIX,
                "auth",
            ),
            ("insufficient_quota", RESPONSES_QUOTA_SUFFIX, "quota"),
            ("quota_exceeded", RESPONSES_QUOTA_SUFFIX, "quota"),
            ("billing_error", RESPONSES_QUOTA_SUFFIX, "quota"),
        ] {
            for label in ["Codex", "OpenAI", "xAI"] {
                let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
                let mut decoder = CodexSseDecoder::for_provider(label);
                let frame = format!(
                    "data: {}\n\n",
                    json!({"type":"response.failed", "response":{
                        "error":{"code":code,"message":"SECRET at capacity"}
                    }})
                );
                let mut text = String::new();
                for line in frame.lines() {
                    decoder.push_line(line, &tx, &mut text);
                }
                let failure = decoder.terminal_result().unwrap_err();
                assert_eq!(failure.message, format!("{label}{suffix}"));
                assert!(!failure.is_retryable());
                let runtime = super::super::net::provider_error_to_runtime(failure.message.into());
                assert!(matches!(runtime, crate::RuntimeError::ApiStatus(_)));
                let typed = crate::runtime::helpers::turn_error_for(&runtime, "test");
                assert_eq!(
                    crate::extensions::session_driver::classify_turn_error(&typed),
                    (
                        crate::extensions::session_driver::Outcome::ProviderError,
                        expected.into()
                    )
                );
                assert!(!typed.message.contains("SECRET"));
            }
        }
        for code in [
            "permission_denied",
            "context_length_exceeded",
            "invalid_request_error",
            "unknown",
        ] {
            let message = user_message_for_terminal_failure("Codex", "absent", code, false);
            assert_eq!(
                crate::extensions::session_driver::classify_error(&message).0,
                crate::extensions::session_driver::Outcome::Blocked
            );
        }
    }

    #[test]
    fn explicit_policy_and_conflicting_error_classes_outrank_capacity_prose() {
        for (kind, code) in [
            ("absent", "permission_denied"),
            ("permission_error", "rate_limit_exceeded"),
            ("invalid_request_error", "insufficient_quota"),
            ("authentication_error", "quota_exceeded"),
            ("server_error", "invalid_api_key"),
            ("unknown", "absent"),
        ] {
            let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
            let mut decoder = CodexSseDecoder::for_provider("Codex");
            let frame = format!(
                "data: {}",
                json!({"type":"response.failed", "response":{
                    "error":{"type":kind, "code":code, "message":"private policy at capacity"}
                }})
            );
            let mut text = String::new();
            decoder.push_line(&frame, &tx, &mut text);
            decoder.push_line("", &tx, &mut text);
            let failure = decoder.terminal_result().unwrap_err();
            assert!(!failure.is_retryable());
            assert_eq!(
                crate::extensions::session_driver::classify_error(&failure.message).0,
                crate::extensions::session_driver::Outcome::Blocked
            );
            assert!(!failure.message.contains("private"));
        }
    }

    /// Deterministic (non-capacity) failures are terminal on the first try
    /// and keep the neutral, xAI-labelled, details-withheld message.
    #[tokio::test(start_paused = true)]
    async fn xai_non_capacity_error_is_terminal_and_labelled_xai() {
        const OTHER_ERROR: &str = "data: {\"type\":\"error\",\"code\":\"invalid_api_key\",\"message\":\"ECHOED:secret prompt\"}\n\n";
        let run = run_xai(vec![OTHER_ERROR, XAI_SUCCESS_SSE], 3).await;
        let err = run.result.expect_err("must fail");
        let msg = err.to_string();
        assert_eq!(msg, format!("xAI{RESPONSES_AUTH_SUFFIX}"));
        assert!(!msg.contains("ECHOED"));
        assert_eq!(run.calls, 1);
    }
}

// ─── Codex wire body tests ────────────────────────────────────────────────────

#[cfg(test)]
mod codex_wire_tests {
    use crate::runtime::openai::catalog::{
        plan_codex_execution, CodexMultiAgentMode, CodexRequestRole,
    };
    use agent_core::reasoning::ReasoningLevel;

    fn plan(
        level: ReasoningLevel,
        role: CodexRequestRole,
    ) -> crate::runtime::openai::catalog::CodexExecutionPlan {
        plan_codex_execution("openai-codex/gpt-5.6-sol", level, role, None)
            .expect("Sol request plan")
    }

    #[allow(clippy::too_many_arguments)]
    fn body_for_level(
        model: &str,
        level: ReasoningLevel,
        input: serde_json::Value,
        instructions: String,
        tools: Vec<serde_json::Value>,
        temperature: Option<f32>,
        max_tokens: Option<u32>,
    ) -> serde_json::Value {
        let qualified = format!("openai-codex/{model}");
        let plan = plan_codex_execution(&qualified, level, CodexRequestRole::Foreground, None)
            .expect("request plan");
        super::build_codex_body(
            model,
            &plan,
            input,
            instructions,
            tools,
            temperature,
            max_tokens,
            "synaps-test-cache-key",
        )
    }

    /// Validate → emit exact reasoning.effort shape for known levels.
    fn codex_effort_for(model_id: &str, level: ReasoningLevel) -> Result<String, String> {
        let qualified = format!("openai-codex/{model_id}");
        let plan = plan_codex_execution(&qualified, level, CodexRequestRole::Foreground, None)
            .map_err(|error| error.to_string())?;
        plan.wire_effort
            .map(|effort| effort.as_str().to_string())
            .ok_or_else(|| "reasoning effort omitted".to_string())
    }

    #[test]
    fn sol_lowers_ultra_to_max_for_requests() {
        let effort = codex_effort_for("gpt-5.6-sol", ReasoningLevel::Ultra).unwrap();
        assert_eq!(effort, "max");
    }

    #[test]
    fn sol_emits_max_exact() {
        let effort = codex_effort_for("gpt-5.6-sol", ReasoningLevel::Max).unwrap();
        assert_eq!(effort, "max");
    }

    #[test]
    fn sol_emits_xhigh_exact() {
        let effort = codex_effort_for("gpt-5.6-sol", ReasoningLevel::XHigh).unwrap();
        assert_eq!(effort, "xhigh");
    }

    #[test]
    fn luna_emits_max_exact() {
        let effort = codex_effort_for("gpt-5.6-luna", ReasoningLevel::Max).unwrap();
        assert_eq!(effort, "max");
    }

    #[test]
    fn luna_rejects_ultra_before_network() {
        // Validation must fail before any credential/network access.
        let err = codex_effort_for("gpt-5.6-luna", ReasoningLevel::Ultra).unwrap_err();
        assert!(err.contains("ultra"), "{err}");
        assert!(err.contains("gpt-5.6-luna"), "{err}");
    }

    #[test]
    fn gpt55_emits_xhigh_exact() {
        let effort = codex_effort_for("gpt-5.5", ReasoningLevel::XHigh).unwrap();
        assert_eq!(effort, "xhigh");
    }

    #[test]
    fn gpt55_rejects_max_before_network() {
        assert!(codex_effort_for("gpt-5.5", ReasoningLevel::Max).is_err());
        assert!(codex_effort_for("gpt-5.5", ReasoningLevel::Ultra).is_err());
    }

    #[test]
    fn off_and_adaptive_omit_reasoning_field() {
        // Off/Adaptive: build_codex_body must NOT emit a "reasoning" key.
        for level in [ReasoningLevel::Off, ReasoningLevel::Adaptive] {
            let body = body_for_level(
                "gpt-5.6-sol",
                level,
                serde_json::json!([]),
                "sys".to_string(),
                vec![],
                None,
                None,
            );
            assert!(
                body.get("reasoning").is_none(),
                "{level:?} must not emit reasoning field, body={body}"
            );
        }
    }

    #[test]
    fn max_as_str_is_not_xhigh() {
        assert_ne!(ReasoningLevel::Max.as_str(), ReasoningLevel::XHigh.as_str());
        assert_eq!(ReasoningLevel::Max.as_str(), "max");
    }

    #[test]
    fn ultra_mode_context_is_exact_proactive_developer_item() {
        let plan = plan(ReasoningLevel::Ultra, CodexRequestRole::Foreground);
        assert_eq!(plan.multi_agent_mode, Some(CodexMultiAgentMode::Proactive));
        let item = super::codex_multi_agent_mode_item(&plan).expect("Ultra context item");
        assert_eq!(
            item.get("type").and_then(serde_json::Value::as_str),
            Some("message")
        );
        assert_eq!(
            item.get("role").and_then(serde_json::Value::as_str),
            Some("developer")
        );
        assert_eq!(
            item.pointer("/content/0/text")
                .and_then(serde_json::Value::as_str),
            Some(concat!(
                "<multi_agent_mode>",
                "Proactive multi-agent delegation is active. Any earlier instruction requiring an explicit user request before spawning sub-agents no longer applies. Use sub-agents when parallel work would materially improve speed or quality. This mode remains active until a later multi-agent mode developer message changes it.",
                "</multi_agent_mode>"
            ))
        );
    }

    #[test]
    fn max_and_xhigh_mode_context_is_exact_explicit_only_item() {
        for level in [ReasoningLevel::Max, ReasoningLevel::XHigh] {
            let plan = plan(level, CodexRequestRole::Foreground);
            assert_eq!(
                plan.multi_agent_mode,
                Some(CodexMultiAgentMode::ExplicitRequestOnly)
            );
            let item = super::codex_multi_agent_mode_item(&plan).expect("v2 context item");
            assert_eq!(
                item.pointer("/content/0/text")
                    .and_then(serde_json::Value::as_str),
                Some(concat!(
                    "<multi_agent_mode>",
                    "Do not spawn sub-agents unless the user or applicable AGENTS.md/skill instructions explicitly ask for sub-agents, delegation, or parallel agent work.",
                    "</multi_agent_mode>"
                )),
                "{level:?}"
            );
        }
    }

    #[test]
    fn mode_context_is_inserted_once_at_stable_head_position() {
        let plan = plan(ReasoningLevel::Ultra, CodexRequestRole::Foreground);
        let mut input = vec![
            serde_json::json!({"role":"user","content":"first"}),
            serde_json::json!({"role":"assistant","content":"answer"}),
            serde_json::json!({"role":"user","content":"current"}),
        ];
        super::insert_codex_multi_agent_mode(&mut input, &plan);
        let mode_positions = input
            .iter()
            .enumerate()
            .filter(|(_, item)| {
                item.pointer("/content/0/text")
                    .and_then(serde_json::Value::as_str)
                    .is_some_and(|text| text.starts_with("<multi_agent_mode>"))
            })
            .map(|(index, _)| index)
            .collect::<Vec<_>>();
        assert_eq!(mode_positions, vec![0]);
        assert_eq!(
            input[3].get("content").and_then(serde_json::Value::as_str),
            Some("current")
        );
    }

    /// Prefix-cache stability: as the conversation grows across user turns,
    /// the already-sent items must remain a byte-identical prefix of the next
    /// request's input. The old before-last-user placement moved the mode
    /// item every turn, invalidating the cached prefix at the prior user
    /// message and re-billing the entire previous agentic turn uncached.
    #[test]
    fn mode_context_placement_preserves_prompt_prefix_across_turns() {
        let plan = plan(ReasoningLevel::Ultra, CodexRequestRole::Foreground);
        let turn1 = vec![
            serde_json::json!({"role":"user","content":"first"}),
            serde_json::json!({"role":"assistant","content":"answer"}),
        ];
        let mut turn2 = turn1.clone();
        turn2.push(serde_json::json!({"role":"user","content":"follow-up"}));
        let mut input1 = turn1;
        let mut input2 = turn2;
        super::insert_codex_multi_agent_mode(&mut input1, &plan);
        super::insert_codex_multi_agent_mode(&mut input2, &plan);
        assert_eq!(
            &input2[..input1.len()],
            &input1[..],
            "turn N's input must be a strict prefix of turn N+1's input"
        );
    }

    #[test]
    fn prompt_cache_key_is_stable_for_the_same_conversation_head() {
        let first = serde_json::json!({"role":"user","content":"hello"});
        let a = super::codex_prompt_cache_key("sys", Some(&first));
        let b = super::codex_prompt_cache_key("sys", Some(&first));
        assert_eq!(a, b);
        assert!(a.starts_with("synaps-"), "{a}");
    }

    #[test]
    fn prompt_cache_key_differs_across_conversations() {
        let first = serde_json::json!({"role":"user","content":"hello"});
        let other = serde_json::json!({"role":"user","content":"different task"});
        assert_ne!(
            super::codex_prompt_cache_key("sys", Some(&first)),
            super::codex_prompt_cache_key("sys", Some(&other)),
        );
        assert_ne!(
            super::codex_prompt_cache_key("sys", Some(&first)),
            super::codex_prompt_cache_key("other-sys", Some(&first)),
        );
    }

    #[test]
    fn body_carries_prompt_cache_key() {
        let body = body_for_level(
            "gpt-5.6-sol",
            ReasoningLevel::Medium,
            serde_json::json!([]),
            "sys".to_string(),
            vec![],
            None,
            None,
        );
        assert_eq!(
            body.get("prompt_cache_key")
                .and_then(serde_json::Value::as_str),
            Some("synaps-test-cache-key")
        );
    }

    #[test]
    fn worker_ultra_has_no_mode_context_item() {
        let plan = plan(ReasoningLevel::Ultra, CodexRequestRole::Worker);
        assert!(super::codex_multi_agent_mode_item(&plan).is_none());
    }

    // ── Pure body construction tests (zero network) ──────────────────────────

    #[test]
    fn body_lowers_ultra_reasoning_effort_to_max() {
        let body = body_for_level(
            "gpt-5.6-sol",
            ReasoningLevel::Ultra,
            serde_json::json!([]),
            "sys".to_string(),
            vec![],
            None,
            None,
        );
        let effort = body
            .get("reasoning")
            .and_then(|r| r.get("effort"))
            .and_then(serde_json::Value::as_str)
            .expect("reasoning.effort must be present for Ultra");
        assert_eq!(effort, "max");
    }

    #[test]
    fn body_contains_reasoning_effort_for_max() {
        let body = body_for_level(
            "gpt-5.6-sol",
            ReasoningLevel::Max,
            serde_json::json!([]),
            "sys".to_string(),
            vec![],
            None,
            None,
        );
        let effort = body
            .get("reasoning")
            .and_then(|r| r.get("effort"))
            .and_then(serde_json::Value::as_str)
            .expect("reasoning.effort must be present for Max");
        assert_eq!(effort, "max");
        // max must NOT be xhigh (critical invariant)
        assert_ne!(effort, "xhigh");
    }

    #[test]
    fn body_contains_reasoning_effort_for_xhigh() {
        let body = body_for_level(
            "gpt-5.6-sol",
            ReasoningLevel::XHigh,
            serde_json::json!([]),
            "sys".to_string(),
            vec![],
            None,
            None,
        );
        let effort = body
            .get("reasoning")
            .and_then(|r| r.get("effort"))
            .and_then(serde_json::Value::as_str)
            .expect("reasoning.effort must be present for XHigh");
        assert_eq!(effort, "xhigh");
    }

    #[test]
    fn body_contains_reasoning_effort_for_low_medium_high() {
        for (level, expected) in [
            (ReasoningLevel::Low, "low"),
            (ReasoningLevel::Medium, "medium"),
            (ReasoningLevel::High, "high"),
        ] {
            let body = body_for_level(
                "gpt-5.6-sol",
                level,
                serde_json::json!([]),
                "sys".to_string(),
                vec![],
                None,
                None,
            );
            let effort = body
                .get("reasoning")
                .and_then(|r| r.get("effort"))
                .and_then(serde_json::Value::as_str)
                .unwrap_or_else(|| panic!("{level:?} must emit reasoning.effort"));
            assert_eq!(effort, expected, "{level:?}");
        }
    }

    #[test]
    fn body_always_sets_model_and_stream_and_include() {
        let body = body_for_level(
            "gpt-5.5",
            ReasoningLevel::XHigh,
            serde_json::json!([]),
            "instructions".to_string(),
            vec![],
            None,
            None,
        );
        assert_eq!(
            body.get("model").and_then(serde_json::Value::as_str),
            Some("gpt-5.5")
        );
        assert_eq!(
            body.get("stream").and_then(serde_json::Value::as_bool),
            Some(true)
        );
        let includes = body.get("include").expect("include must be present");
        assert!(
            includes.as_array().is_some_and(|a| {
                a.iter()
                    .any(|v| v.as_str() == Some("reasoning.encrypted_content"))
            }),
            "include must contain reasoning.encrypted_content"
        );
        assert_eq!(
            body.get("store").and_then(serde_json::Value::as_bool),
            Some(false)
        );
    }

    #[test]
    fn body_omits_temperature_and_max_tokens_when_none() {
        let body = body_for_level(
            "gpt-5.6-sol",
            ReasoningLevel::Medium,
            serde_json::json!([]),
            "sys".to_string(),
            vec![],
            None,
            None,
        );
        assert!(body.get("temperature").is_none());
        assert!(body.get("max_output_tokens").is_none());
    }

    #[test]
    fn body_includes_temperature_and_max_tokens_when_some() {
        let body = body_for_level(
            "gpt-5.6-sol",
            ReasoningLevel::High,
            serde_json::json!([]),
            "sys".to_string(),
            vec![],
            Some(0.7),
            Some(2048),
        );
        let temp = body
            .get("temperature")
            .and_then(serde_json::Value::as_f64)
            .unwrap();
        assert!((temp - 0.7f64).abs() < 0.01, "temperature={temp}");
        assert_eq!(
            body.get("max_output_tokens")
                .and_then(serde_json::Value::as_u64),
            Some(2048)
        );
    }
}

#[cfg(test)]
mod send_retry_tests {
    //! Regression tests for transient-failure retry on the codex send path
    //! (incident: session 20260714-025948-3dab — a single transport failure
    //! against chatgpt.com aborted a whole autonomous turn because the codex
    //! route was single-shot while the Anthropic route retried).
    //!
    //! Mirrors the `on401_tests` mock-server pattern in `runtime/api.rs`.
    //! Credentials cross the broker boundary: a stub broker vends a
    //! JWT-shaped token carrying the ChatGPT account-id claim.

    use super::*;
    use agent_core::auth::{
        AccessToken, BrokerError, ProxyByteStream, ProxyRequest, ProxyResponse,
    };
    use agent_core::reasoning::ReasoningLevel;
    use async_trait::async_trait;
    use axum::{http::StatusCode, response::IntoResponse, routing::post as axum_post, Router};
    use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine};
    use std::sync::{
        atomic::{AtomicUsize, Ordering},
        Arc,
    };
    use tokio::sync::mpsc;

    /// Minimal Responses-API SSE success body: one text delta + completed.
    const CODEX_SSE_SUCCESS: &str = concat!(
        "data: {\"type\":\"response.output_text.delta\",\"delta\":\"hello\"}\n\n",
        "data: {\"type\":\"response.completed\",\"response\":{\"usage\":{\"input_tokens\":1,\"output_tokens\":1}}}\n\n",
        "data: [DONE]\n\n",
    );

    /// JWT-shaped token whose payload carries the ChatGPT account-id claim
    /// `extract_codex_account_id` looks for. Signature is irrelevant — the
    /// mock server never validates it.
    fn fake_codex_token() -> String {
        let payload = serde_json::json!({
            "https://api.openai.com/auth": { "chatgpt_account_id": "acct_test" }
        });
        format!("h.{}.s", URL_SAFE_NO_PAD.encode(payload.to_string()))
    }

    /// Stub broker that vends the fake codex token. Everything else is
    /// unreachable in these tests and fails closed.
    struct TokenOnlyBroker;

    #[async_trait]
    impl crate::auth::CredentialBroker for TokenOnlyBroker {
        async fn access_token(
            &self,
            _p: agent_core::auth::OAuthProviderId,
        ) -> Result<AccessToken, BrokerError> {
            Ok(AccessToken {
                token: fake_codex_token(),
                expires: u64::MAX,
            })
        }
        async fn proxy(&self, _request: ProxyRequest) -> Result<ProxyResponse, BrokerError> {
            Err(BrokerError::Denied("not implemented in stub".into()))
        }
        async fn proxy_stream(
            &self,
            _request: ProxyRequest,
        ) -> Result<ProxyByteStream, BrokerError> {
            Err(BrokerError::Denied("not implemented in stub".into()))
        }
        async fn anthropic_usage(&self) -> Result<serde_json::Value, BrokerError> {
            Err(BrokerError::Denied("not implemented in stub".into()))
        }
        async fn capabilities(&self) -> Result<Vec<agent_core::auth::ProviderStatus>, BrokerError> {
            Ok(vec![])
        }
    }

    struct DenyingCountingBroker {
        access_calls: Arc<AtomicUsize>,
    }

    #[async_trait]
    impl crate::auth::CredentialBroker for DenyingCountingBroker {
        async fn access_token(
            &self,
            _provider: agent_core::auth::OAuthProviderId,
        ) -> Result<AccessToken, BrokerError> {
            self.access_calls.fetch_add(1, Ordering::SeqCst);
            Err(BrokerError::Denied(
                "credential access must not occur".into(),
            ))
        }

        async fn proxy(&self, _request: ProxyRequest) -> Result<ProxyResponse, BrokerError> {
            Err(BrokerError::Denied("not implemented in stub".into()))
        }

        async fn proxy_stream(
            &self,
            _request: ProxyRequest,
        ) -> Result<ProxyByteStream, BrokerError> {
            Err(BrokerError::Denied("not implemented in stub".into()))
        }

        async fn anthropic_usage(&self) -> Result<serde_json::Value, BrokerError> {
            Err(BrokerError::Denied("not implemented in stub".into()))
        }

        async fn capabilities(&self) -> Result<Vec<agent_core::auth::ProviderStatus>, BrokerError> {
            Ok(vec![])
        }
    }

    /// Spawn a mock Codex endpoint. First `fail_count` POSTs → `fail_status`;
    /// subsequent → SSE success. Returns (base_url, call_counter).
    async fn spawn_mock_codex(
        fail_count: usize,
        fail_status: StatusCode,
    ) -> (String, Arc<AtomicUsize>) {
        let counter = Arc::new(AtomicUsize::new(0));
        let counter_clone = Arc::clone(&counter);
        let app = Router::new().route(
            "/codex/responses",
            axum_post(move || {
                let counter = Arc::clone(&counter_clone);
                async move {
                    let n = counter.fetch_add(1, Ordering::SeqCst);
                    if n < fail_count {
                        (
                            fail_status,
                            [("content-type", "application/json")],
                            "{\"error\":{\"message\":\"transient upstream sadness\"}}".to_string(),
                        )
                            .into_response()
                    } else {
                        (
                            StatusCode::OK,
                            [("content-type", "text/event-stream")],
                            CODEX_SSE_SUCCESS.to_string(),
                        )
                            .into_response()
                    }
                }
            }),
        );
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
        (format!("http://{addr}"), counter)
    }

    async fn spawn_codex_sse(body: &'static str) -> String {
        let app = Router::new().route(
            "/codex/responses",
            axum_post(move || async move {
                (
                    StatusCode::OK,
                    [("content-type", "text/event-stream")],
                    body,
                )
            }),
        );
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
        format!("http://{addr}")
    }

    async fn spawn_empty_then_success_codex() -> (
        String,
        Arc<AtomicUsize>,
        Arc<std::sync::Mutex<Vec<bytes::Bytes>>>,
    ) {
        let counter = Arc::new(AtomicUsize::new(0));
        let bodies = Arc::new(std::sync::Mutex::new(Vec::new()));
        let app = Router::new().route(
            "/codex/responses",
            axum_post({
                let counter = Arc::clone(&counter);
                let bodies = Arc::clone(&bodies);
                move |body: bytes::Bytes| {
                    let counter = Arc::clone(&counter);
                    let bodies = Arc::clone(&bodies);
                    async move {
                        bodies.lock().unwrap().push(body);
                        let n = counter.fetch_add(1, Ordering::SeqCst);
                        let response = if n == 0 {
                            concat!(
                                "data: {\"type\":\"response.completed\",\"response\":{\"usage\":{\"input_tokens\":7,\"output_tokens\":0}}}\n\n",
                                "data: [DONE]\n\n",
                            )
                        } else {
                            CODEX_SSE_SUCCESS
                        };
                        (
                            StatusCode::OK,
                            [("content-type", "text/event-stream")],
                            response,
                        )
                    }
                }
            }),
        );
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
        (format!("http://{addr}"), counter, bodies)
    }

    async fn run_codex(
        base_url: &str,
        max_retries: u32,
    ) -> Result<Value, Box<dyn std::error::Error + Send + Sync>> {
        let cfg = ProviderConfig {
            base_url: base_url.to_string(),
            // The incident model — validated against the static catalog ladder.
            model: "gpt-5.6-sol".to_string(),
            provider: "openai-codex".to_string(),
        };
        let client = reqwest::Client::new();
        let broker: std::sync::Arc<dyn crate::auth::CredentialBroker> =
            std::sync::Arc::new(TokenOnlyBroker);
        let (tx, _rx) = mpsc::unbounded_channel();
        call_codex_stream_inner(
            &cfg,
            &client,
            &broker,
            &[],
            &Some("test".to_string()),
            &[],
            &tx,
            None,
            None,
            agent_core::reasoning::ReasoningLevel::Medium,
            crate::runtime::openai::catalog::CodexRequestRole::Foreground,
            &tokio_util::sync::CancellationToken::new(),
            max_retries,
            &crate::runtime::trace::TraceContext::disabled(),
        )
        .await
    }

    // Raw HTTP is intentional: axum's complete bodies cannot reproduce the
    // reported EOF in an HTTP/1.1 chunk-size line.
    async fn broken_chunk_server(
        scripts: Vec<(&'static str, bool)>,
    ) -> (
        String,
        Arc<std::sync::Mutex<Vec<Vec<u8>>>>,
        tokio::task::JoinHandle<()>,
    ) {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let address = listener.local_addr().unwrap();
        let requests = Arc::new(std::sync::Mutex::new(Vec::new()));
        let seen = requests.clone();
        let task = tokio::spawn(async move {
            let mut attempt = 0;
            loop {
                let (mut socket, _) = listener.accept().await.unwrap();
                let mut request = Vec::new();
                let end = loop {
                    let mut chunk = [0; 8192];
                    let n = socket.read(&mut chunk).await.unwrap();
                    if n == 0 {
                        return;
                    }
                    request.extend_from_slice(&chunk[..n]);
                    if let Some(end) = request.windows(4).position(|w| w == b"\r\n\r\n") {
                        break end + 4;
                    }
                };
                let length: usize = String::from_utf8_lossy(&request[..end])
                    .lines()
                    .find_map(|line| {
                        line.to_ascii_lowercase()
                            .strip_prefix("content-length: ")
                            .map(str::to_owned)
                    })
                    .unwrap()
                    .parse()
                    .unwrap();
                while request.len() < end + length {
                    let mut chunk = [0; 8192];
                    let n = socket.read(&mut chunk).await.unwrap();
                    if n == 0 {
                        return;
                    }
                    request.extend_from_slice(&chunk[..n]);
                }
                seen.lock()
                    .unwrap()
                    .push(request[end..end + length].to_vec());
                let (body, broken) = scripts[attempt.min(scripts.len() - 1)];
                attempt += 1;
                socket.write_all(b"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n").await.unwrap();
                socket
                    .write_all(format!("{:x}\r\n{}\r\n", body.len(), body).as_bytes())
                    .await
                    .unwrap();
                // Omitting the 0 chunk produces precisely the user's failure.
                if !broken {
                    socket.write_all(b"0\r\n\r\n").await.unwrap();
                }
                socket.shutdown().await.unwrap();
            }
        });
        (format!("http://{address}"), requests, task)
    }

    async fn codex_with_events(
        base: &str,
        retries: u32,
        cancel: &tokio_util::sync::CancellationToken,
        tx: &mpsc::UnboundedSender<StreamEvent>,
    ) -> Result<Value, super::super::net::BoxedProviderError> {
        let cfg = ProviderConfig {
            base_url: base.into(),
            model: "gpt-5.6-sol".into(),
            provider: "openai-codex".into(),
        };
        let broker: Arc<dyn crate::auth::CredentialBroker> = Arc::new(TokenOnlyBroker);
        call_codex_stream_inner(
            &cfg,
            &reqwest::Client::new(),
            &broker,
            &[],
            &Some("test".into()),
            &[],
            tx,
            None,
            None,
            agent_core::reasoning::ReasoningLevel::Medium,
            crate::runtime::openai::catalog::CodexRequestRole::Foreground,
            cancel,
            retries,
            &crate::runtime::trace::TraceContext::disabled(),
        )
        .await
    }

    const PARTIAL_TOOL: &str = concat!(
        "data: {\"type\":\"response.output_text.delta\",\"delta\":\"discard me\"}\n\n",
        "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"function_call\",\"call_id\":\"old_call\",\"name\":\"write\",\"arguments\":\"\"}}\n\n",
        "data: {\"type\":\"response.function_call_arguments.delta\",\"output_index\":0,\"delta\":\"{\\\"path\\\":\"}\n\n",
    );
    const COMPLETE_TOOL: &str = concat!(
        "data: {\"type\":\"response.output_text.delta\",\"delta\":\"accepted\"}\n\n",
        "data: {\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{\"type\":\"function_call\",\"call_id\":\"new_call\",\"name\":\"write\",\"arguments\":\"{\\\"path\\\":\\\"ok\\\"}\"}}\n\n",
        "data: {\"type\":\"response.completed\",\"response\":{\"usage\":{\"input_tokens\":3,\"output_tokens\":2}}}\n\n",
    );

    #[tokio::test]
    async fn interrupted_partial_tool_retries_identical_bytes_and_commits_only_success() {
        use crate::LlmEvent;
        let (url, requests, server) =
            broken_chunk_server(vec![(PARTIAL_TOOL, true), (COMPLETE_TOOL, false)]).await;
        let (tx, mut rx) = mpsc::unbounded_channel();
        let result = codex_with_events(&url, 1, &tokio_util::sync::CancellationToken::new(), &tx)
            .await
            .unwrap();
        server.abort();
        let requests = requests.lock().unwrap();
        assert_eq!(requests.len(), 2);
        assert_eq!(requests[0], requests[1]);
        let tools: Vec<_> = result["content"]
            .as_array()
            .unwrap()
            .iter()
            .filter(|v| v["type"] == "tool_use")
            .collect();
        assert_eq!(tools.len(), 1);
        assert_eq!(tools[0]["id"], "new_call");
        assert_eq!(result["content"][0]["text"], "accepted");
        let mut preview = String::from("prior completed round");
        let mut baseline = 0;
        let mut resets = 0;
        let mut input_tokens = 0;
        while let Ok(event) = rx.try_recv() {
            match event {
                StreamEvent::Llm(LlmEvent::ResponseStart) => baseline = preview.len(),
                StreamEvent::Llm(LlmEvent::ResponseReset) => {
                    preview.truncate(baseline);
                    resets += 1;
                }
                StreamEvent::Llm(LlmEvent::Text(t)) => preview.push_str(&t),
                StreamEvent::Session(crate::SessionEvent::Usage {
                    input_tokens: n, ..
                }) => input_tokens += n,
                _ => {}
            }
        }
        assert_eq!(resets, 1);
        assert_eq!(preview, "prior completed roundaccepted");
        assert_eq!(input_tokens, 3);
    }

    #[tokio::test]
    async fn completed_response_followed_by_chunk_error_is_not_replayed() {
        let (url, requests, server) = broken_chunk_server(vec![(CODEX_SSE_SUCCESS, true)]).await;
        assert!(run_codex(&url, 2).await.is_ok());
        server.abort();
        assert_eq!(requests.lock().unwrap().len(), 1);
    }

    #[tokio::test]
    async fn clean_truncated_eof_retries_but_never_accepts_partial_tools() {
        let (url, requests, server) = broken_chunk_server(vec![(PARTIAL_TOOL, false)]).await;
        assert!(run_codex(&url, 1).await.is_err());
        server.abort();
        assert_eq!(requests.lock().unwrap().len(), 2);
    }

    #[tokio::test]
    async fn broken_chunk_retry_budget_is_finite() {
        let (url, requests, server) = broken_chunk_server(vec![(PARTIAL_TOOL, true)]).await;
        assert!(run_codex(&url, 1).await.is_err());
        server.abort();
        assert_eq!(requests.lock().unwrap().len(), 2);
    }

    #[tokio::test]
    async fn cancellation_during_stream_retry_prevents_second_dispatch() {
        let (url, requests, server) = broken_chunk_server(vec![(PARTIAL_TOOL, true)]).await;
        let (tx, mut rx) = mpsc::unbounded_channel();
        let cancel = tokio_util::sync::CancellationToken::new();
        let waiter = cancel.clone();
        let task = tokio::spawn(async move {
            while let Some(event) = rx.recv().await {
                if matches!(event, StreamEvent::Session(crate::SessionEvent::Retry(_))) {
                    waiter.cancel();
                    break;
                }
            }
        });
        let error = codex_with_events(&url, 10, &cancel, &tx).await.unwrap_err();
        task.await.unwrap();
        server.abort();
        assert_eq!(error.to_string(), "request canceled");
        assert_eq!(requests.lock().unwrap().len(), 1);
    }

    #[tokio::test]
    async fn direct_ultra_tool_guard_precedes_broker_access() {
        let access_calls = Arc::new(AtomicUsize::new(0));
        let broker: Arc<dyn crate::auth::CredentialBroker> = Arc::new(DenyingCountingBroker {
            access_calls: Arc::clone(&access_calls),
        });
        let cfg = ProviderConfig {
            base_url: "http://127.0.0.1:1".to_string(),
            model: "gpt-5.6-sol".to_string(),
            provider: "openai-codex".to_string(),
        };
        let (tx, _rx) = mpsc::unbounded_channel();

        let error = call_codex_stream_inner(
            &cfg,
            &reqwest::Client::new(),
            &broker,
            &[],
            &Some("test".to_string()),
            &[],
            &tx,
            None,
            None,
            ReasoningLevel::Ultra,
            crate::runtime::openai::catalog::CodexRequestRole::Foreground,
            &tokio_util::sync::CancellationToken::new(),
            0,
            &crate::runtime::trace::TraceContext::disabled(),
        )
        .await
        .expect_err("foreground Ultra without delegation tools must fail closed");

        assert!(error.to_string().contains("subagent_start"), "{error}");
        assert_eq!(
            access_calls.load(Ordering::SeqCst),
            0,
            "preflight denial must happen before broker credential access"
        );
    }

    #[tokio::test]
    async fn direct_codex_path_rejects_provider_identity_before_broker_access() {
        let access_calls = Arc::new(AtomicUsize::new(0));
        let broker: Arc<dyn crate::auth::CredentialBroker> = Arc::new(DenyingCountingBroker {
            access_calls: Arc::clone(&access_calls),
        });
        let cfg = ProviderConfig {
            base_url: "http://127.0.0.1:1".to_string(),
            model: "gpt-5.6-sol".to_string(),
            provider: "openrouter".to_string(),
        };
        let (tx, _rx) = mpsc::unbounded_channel();

        let error = call_codex_stream_inner(
            &cfg,
            &reqwest::Client::new(),
            &broker,
            &[],
            &Some("test".to_string()),
            &[],
            &tx,
            None,
            None,
            ReasoningLevel::Medium,
            crate::runtime::openai::catalog::CodexRequestRole::Foreground,
            &tokio_util::sync::CancellationToken::new(),
            0,
            &crate::runtime::trace::TraceContext::disabled(),
        )
        .await
        .expect_err("the Codex seam must validate provider-qualified identity");

        assert!(error.to_string().contains("openrouter"), "{error}");
        assert_eq!(
            access_calls.load(Ordering::SeqCst),
            0,
            "provider identity denial must precede broker credential access"
        );
    }

    /// A stream whose terminal event is `response.failed` must surface a
    /// typed API failure at the transport boundary — never `Ok` with empty
    /// content (the "model returned an empty response" misdiagnosis,
    /// incident: sessions 20260720-022603-6c2a / turn-2629074-0). The
    /// provider-controlled error body must not leak into the message.
    #[tokio::test]
    async fn codex_response_failed_event_is_a_terminal_error_not_empty_success() {
        let base_url = spawn_codex_sse(concat!(
            "data: {\"type\":\"response.failed\",\"response\":{\"error\":{\"type\":\"server_error\",\"message\":\"ECHOED:secret prompt\"}}}\n\n",
            "data: [DONE]\n\n",
        ))
        .await;
        let err = run_codex(&base_url, 0)
            .await
            .expect_err("response.failed must fail the request");
        let msg = err.to_string();
        assert!(
            msg == format!("Codex{RESPONSES_CAPACITY_SUFFIX}"),
            "static transient failure message expected: {msg}"
        );
        assert!(
            !msg.contains("ECHOED") && !msg.contains("secret prompt"),
            "provider error body must never surface: {msg}"
        );
    }

    /// `response.incomplete` is a terminal non-success: surfacing it as an
    /// empty `Ok` poisoned the turn with the generic empty-response error.
    #[tokio::test]
    async fn codex_response_incomplete_event_is_a_terminal_error() {
        let base_url = spawn_codex_sse(concat!(
            "data: {\"type\":\"response.output_text.delta\",\"delta\":\"partial\"}\n\n",
            "data: {\"type\":\"response.incomplete\",\"response\":{\"incomplete_details\":{\"reason\":\"max_output_tokens\"}}}\n\n",
            "data: [DONE]\n\n",
        ))
        .await;
        let err = run_codex(&base_url, 0)
            .await
            .expect_err("response.incomplete must fail the request");
        assert!(
            err.to_string()
                .starts_with("Codex response was incomplete."),
            "got: {err}"
        );
    }

    /// EOF without any terminal Responses event (mid-stream connection drop
    /// after HTTP 200) must fail closed instead of returning empty content.
    #[tokio::test]
    async fn codex_eof_without_terminal_event_fails_closed() {
        let base_url = spawn_codex_sse(
            "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_1\"}}\n\n",
        )
        .await;
        let err = run_codex(&base_url, 0)
            .await
            .expect_err("EOF without terminal event must fail");
        assert!(
            err.to_string()
                .starts_with("Codex response stream ended without a terminal event."),
            "got: {err}"
        );
    }

    #[tokio::test]
    async fn codex_completed_without_output_retries_identical_bytes_then_succeeds() {
        let (base_url, counter, bodies) = spawn_empty_then_success_codex().await;
        let result = run_codex(&base_url, 1)
            .await
            .expect("one empty-success stream with retries available must not abort the turn");

        assert_eq!(counter.load(Ordering::SeqCst), 2, "expected one retry");
        let received = bodies.lock().unwrap();
        assert_eq!(received.len(), 2, "both attempts must reach the endpoint");
        assert_eq!(
            received[0], received[1],
            "the logical request must resend byte-identical content"
        );
        assert_eq!(result["content"][0]["text"], "hello");
    }

    /// Mock Codex endpoint that records every request's (session-id,
    /// thread-id, prompt_cache_key). The first `fail_count` POSTs fail 500.
    async fn spawn_header_recording_codex(
        fail_count: usize,
    ) -> (String, Arc<std::sync::Mutex<Vec<(String, String, String)>>>) {
        let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
        let counter = Arc::new(AtomicUsize::new(0));
        let app = Router::new().route(
            "/codex/responses",
            axum_post({
                let seen = Arc::clone(&seen);
                move |headers: axum::http::HeaderMap, body: bytes::Bytes| {
                    let seen = Arc::clone(&seen);
                    let counter = Arc::clone(&counter);
                    async move {
                        let get = |name: &str| {
                            headers
                                .get(name)
                                .and_then(|v| v.to_str().ok())
                                .unwrap_or("")
                                .to_string()
                        };
                        let key = serde_json::from_slice::<Value>(&body)
                            .ok()
                            .and_then(|b| b["prompt_cache_key"].as_str().map(str::to_string))
                            .unwrap_or_default();
                        seen.lock()
                            .unwrap()
                            .push((get("session-id"), get("thread-id"), key));
                        if counter.fetch_add(1, Ordering::SeqCst) < fail_count {
                            (StatusCode::INTERNAL_SERVER_ERROR, "{}".to_string()).into_response()
                        } else {
                            (
                                StatusCode::OK,
                                [("content-type", "text/event-stream")],
                                CODEX_SSE_SUCCESS.to_string(),
                            )
                                .into_response()
                        }
                    }
                }
            }),
        );
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
        (format!("http://{addr}"), seen)
    }

    /// The ChatGPT backend derives prompt-cache affinity from the `session-id`
    /// header (Codex CLI `responses_session_id`); without it every request
    /// reported `cached_tokens: 0`. Measured A/B: no header 0/0/0 cached;
    /// with `session-id`/`thread-id` = prompt_cache_key, 0 then 95%, 95%.
    #[tokio::test]
    async fn codex_requests_carry_session_and_thread_id_equal_to_the_cache_key() {
        let (base_url, seen) = spawn_header_recording_codex(1).await;
        run_codex(&base_url, 2)
            .await
            .expect("one transient 500 then success");
        let seen = seen.lock().unwrap();
        assert_eq!(seen.len(), 2, "first attempt + one retry");
        for (session_id, thread_id, key) in seen.iter() {
            assert!(
                key.starts_with("synaps-"),
                "prompt_cache_key in body: {key:?}"
            );
            assert_eq!(session_id, key, "session-id header = prompt_cache_key");
            assert_eq!(thread_id, key, "thread-id header = prompt_cache_key");
        }
        assert_eq!(seen[0], seen[1], "a retry keeps the same cache routing");
    }

    #[tokio::test]
    async fn codex_retries_transient_500_then_succeeds() {
        let (base_url, counter) = spawn_mock_codex(1, StatusCode::INTERNAL_SERVER_ERROR).await;
        let result = run_codex(&base_url, 2)
            .await
            .expect("one transient 500 with retries available must not abort the turn");
        assert_eq!(
            counter.load(Ordering::SeqCst),
            2,
            "expected exactly one retry"
        );
        let text = result["content"][0]["text"].as_str().unwrap_or_default();
        assert_eq!(text, "hello");
    }

    #[tokio::test]
    async fn codex_response_failed_surfaces_typed_provider_error_not_empty_response() {
        const BODY: &str = concat!(
            "data: {\"type\":\"response.failed\",\"response\":{\"error\":{\"type\":\"server_error\",\"message\":\"ECHOED:private prompt\"}}}\n\n",
            "data: [DONE]\n\n",
        );
        let base_url = spawn_codex_sse(BODY).await;
        let err = run_codex(&base_url, 0).await.expect_err("must fail");
        let msg = err.to_string();
        assert_eq!(msg, format!("Codex{RESPONSES_CAPACITY_SUFFIX}"));
        assert!(!msg.contains("empty response"), "{msg}");
        assert!(!msg.contains("ECHOED"), "provider text leaked: {msg}");
        assert!(
            !msg.contains("private prompt"),
            "provider text leaked: {msg}"
        );
    }

    #[tokio::test]
    async fn codex_response_incomplete_surfaces_typed_provider_error_not_empty_response() {
        const BODY: &str = concat!(
            "data: {\"type\":\"response.incomplete\",\"response\":{\"incomplete_details\":{\"reason\":\"max_output_tokens\"}}}\n\n",
            "data: [DONE]\n\n",
        );
        let base_url = spawn_codex_sse(BODY).await;
        let err = run_codex(&base_url, 0).await.expect_err("must fail");
        let msg = err.to_string();
        assert!(msg.starts_with("Codex response was incomplete."), "{msg}");
        assert!(!msg.contains("empty response"), "{msg}");
        assert!(
            !msg.contains("max_output_tokens"),
            "provider text leaked: {msg}"
        );
    }

    #[tokio::test]
    async fn codex_clean_eof_without_terminal_fails_closed() {
        let base_url = spawn_codex_sse(
            "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_123\"}}\n\n",
        )
        .await;
        let err = run_codex(&base_url, 0).await.expect_err("must fail");
        assert!(
            err.to_string()
                .starts_with("Codex response stream ended without a terminal event."),
            "{err}"
        );
    }

    #[tokio::test]
    async fn codex_retries_429_then_succeeds() {
        let (base_url, counter) = spawn_mock_codex(1, StatusCode::TOO_MANY_REQUESTS).await;
        run_codex(&base_url, 2)
            .await
            .expect("429 is transient — must retry");
        assert_eq!(counter.load(Ordering::SeqCst), 2);
    }

    #[tokio::test]
    async fn codex_zero_retries_fails_fast_with_status_error() {
        let (base_url, counter) =
            spawn_mock_codex(usize::MAX, StatusCode::INTERNAL_SERVER_ERROR).await;
        let err = run_codex(&base_url, 0).await.expect_err("must fail");
        assert_eq!(counter.load(Ordering::SeqCst), 1, "no retries budgeted");
        let msg = err.to_string();
        assert!(
            msg.starts_with("codex request failed: 500"),
            "status must survive for classification: {msg}"
        );
    }

    #[tokio::test]
    async fn codex_does_not_retry_client_errors() {
        let (base_url, counter) = spawn_mock_codex(usize::MAX, StatusCode::BAD_REQUEST).await;
        let err = run_codex(&base_url, 3).await.expect_err("must fail");
        assert_eq!(
            counter.load(Ordering::SeqCst),
            1,
            "400 is deterministic — retrying it never helps"
        );
        assert!(err.to_string().starts_with("codex request failed: 400"));
    }

    #[tokio::test]
    async fn codex_exhausted_retries_reports_last_status() {
        let (base_url, counter) =
            spawn_mock_codex(usize::MAX, StatusCode::SERVICE_UNAVAILABLE).await;
        let err = run_codex(&base_url, 1).await.expect_err("must fail");
        assert_eq!(
            counter.load(Ordering::SeqCst),
            2,
            "initial attempt + 1 retry"
        );
        assert!(err.to_string().starts_with("codex request failed: 503"));
    }

    // ─── G5: account-aware selection + bounded failover ──────────────────────

    use crate::runtime::openai::account_routing::{
        CodexAccountRouter, CodexQuotaEvidence, PinnedAccount,
    };
    use agent_core::auth::{Account, CredentialRef, OAuthProviderId};

    /// JWT-shaped token carrying `account_id` as the ChatGPT account claim.
    fn fake_codex_token_for(account_id: &str) -> String {
        let payload = serde_json::json!({
            "https://api.openai.com/auth": { "chatgpt_account_id": account_id }
        });
        format!("h.{}.s", URL_SAFE_NO_PAD.encode(payload.to_string()))
    }

    fn pinned(label: &str, account_id: &str, auto: bool) -> PinnedAccount {
        PinnedAccount {
            credential: CredentialRef::new(
                OAuthProviderId::OpenAiCodex,
                Account::parse(label).unwrap(),
            ),
            token: fake_codex_token_for(account_id),
            auto,
        }
    }

    /// Scripted router: `pin` returns `first`; `failover` returns the next
    /// scripted candidate (or `None`). Records every call.
    struct ScriptedRouter {
        first: PinnedAccount,
        candidates: std::sync::Mutex<std::collections::VecDeque<Option<PinnedAccount>>>,
        failover_calls: AtomicUsize,
        reported: std::sync::Mutex<Vec<(String, CodexQuotaEvidence)>>,
    }

    impl ScriptedRouter {
        fn new(first: PinnedAccount, candidates: Vec<Option<PinnedAccount>>) -> Self {
            Self {
                first,
                candidates: std::sync::Mutex::new(candidates.into()),
                failover_calls: AtomicUsize::new(0),
                reported: std::sync::Mutex::new(Vec::new()),
            }
        }
        fn reports(&self) -> Vec<String> {
            self.reported
                .lock()
                .unwrap()
                .iter()
                .map(|(k, _)| k.clone())
                .collect()
        }
    }

    #[async_trait]
    impl CodexAccountRouter for ScriptedRouter {
        async fn pin(&self, _model: &str) -> Result<PinnedAccount, String> {
            Ok(self.first.clone())
        }
        async fn failover(
            &self,
            _model: &str,
            _failed: &PinnedAccount,
            _evidence: &CodexQuotaEvidence,
            _failovers_so_far: u32,
        ) -> Result<Option<PinnedAccount>, String> {
            self.failover_calls.fetch_add(1, Ordering::SeqCst);
            Ok(self.candidates.lock().unwrap().pop_front().flatten())
        }
        async fn report_exhausted(&self, failed: &PinnedAccount, evidence: &CodexQuotaEvidence) {
            self.reported
                .lock()
                .unwrap()
                .push((failed.credential.storage_key(), *evidence));
        }
    }

    /// One observed request: `chatgpt-account-id` header, bearer token, body.
    #[derive(Clone, Debug)]
    struct Seen {
        account: String,
        bearer: String,
        body: bytes::Bytes,
    }

    /// Mock Codex endpoint scripted per request: each entry is
    /// `(status, headers, body)`; the last entry repeats. Records the
    /// `chatgpt-account-id` header, the bearer and the body of every request.
    type Script = Vec<(StatusCode, Vec<(&'static str, &'static str)>, &'static str)>;
    async fn spawn_scripted_codex(script: Script) -> (String, Arc<std::sync::Mutex<Vec<Seen>>>) {
        let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
        let counter = Arc::new(AtomicUsize::new(0));
        let script = Arc::new(script);
        let app = Router::new().route(
            "/codex/responses",
            axum_post({
                let seen = Arc::clone(&seen);
                let counter = Arc::clone(&counter);
                let script = Arc::clone(&script);
                move |headers: axum::http::HeaderMap, body: bytes::Bytes| {
                    let seen = Arc::clone(&seen);
                    let counter = Arc::clone(&counter);
                    let script = Arc::clone(&script);
                    async move {
                        let header = |name: &str| {
                            headers
                                .get(name)
                                .and_then(|v| v.to_str().ok())
                                .unwrap_or("")
                                .to_string()
                        };
                        let bearer = header("authorization")
                            .strip_prefix("Bearer ")
                            .unwrap_or("")
                            .to_string();
                        seen.lock().unwrap().push(Seen {
                            account: header("chatgpt-account-id"),
                            bearer,
                            body,
                        });
                        let n = counter.fetch_add(1, Ordering::SeqCst);
                        let (status, hdrs, body) = &script[n.min(script.len() - 1)];
                        let mut resp = axum::response::Response::builder().status(*status);
                        for (k, v) in hdrs {
                            resp = resp.header(*k, *v);
                        }
                        let ct = if status.is_success() {
                            "text/event-stream"
                        } else {
                            "application/json"
                        };
                        resp.header("content-type", ct)
                            .body(axum::body::Body::from(*body))
                            .unwrap()
                    }
                }
            }),
        );
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
        (format!("http://{addr}"), seen)
    }

    async fn run_codex_with_router(
        base_url: &str,
        router: &dyn CodexAccountRouter,
        max_retries: u32,
        tx: &mpsc::UnboundedSender<StreamEvent>,
    ) -> Result<Value, Box<dyn std::error::Error + Send + Sync>> {
        let cfg = ProviderConfig {
            base_url: base_url.to_string(),
            model: "gpt-5.6-sol".to_string(),
            provider: "openai-codex".to_string(),
        };
        call_codex_stream_with_router(
            &cfg,
            &reqwest::Client::new(),
            router,
            &[],
            &Some("test".to_string()),
            &[],
            tx,
            None,
            None,
            ReasoningLevel::Medium,
            crate::runtime::openai::catalog::CodexRequestRole::Foreground,
            &tokio_util::sync::CancellationToken::new(),
            max_retries,
            &crate::runtime::trace::TraceContext::disabled(),
        )
        .await
    }

    const QUOTA_429: &str = r#"{"error":{"type":"usage_limit_reached","message":"ECHOED prompt","plan_type":"plus","resets_in_seconds":3600}}"#;
    const GENERIC_429: &str = r#"{"error":{"type":"rate_limit_exceeded","message":"slow down"}}"#;

    fn notices(rx: &mut mpsc::UnboundedReceiver<StreamEvent>) -> Vec<String> {
        let mut out = Vec::new();
        while let Ok(ev) = rx.try_recv() {
            if let StreamEvent::Session(
                crate::SessionEvent::Notice(n) | crate::SessionEvent::Retry(n),
            ) = ev
            {
                out.push(n);
            }
        }
        out
    }

    /// Auto policy, exhausted 429 on seat A before any output, seat B fresh:
    /// exactly one switch, the retry carries B's header (derived from B's
    /// token), identical body bytes, and the cooldown was reported for A.
    #[tokio::test]
    async fn quota_429_fails_over_once_to_distinct_account_with_paired_header() {
        let (url, seen) = spawn_scripted_codex(vec![
            (StatusCode::TOO_MANY_REQUESTS, vec![], QUOTA_429),
            (StatusCode::OK, vec![], CODEX_SSE_SUCCESS),
        ])
        .await;
        let router = ScriptedRouter::new(
            pinned("astra1", "acct_A", true),
            vec![Some(pinned("astra2", "acct_B", true))],
        );
        let (tx, mut rx) = mpsc::unbounded_channel();
        let result = run_codex_with_router(&url, &router, 3, &tx).await.unwrap();
        assert_eq!(result["content"][0]["text"], "hello");
        let seen = seen.lock().unwrap();
        assert_eq!(seen.len(), 2, "one failed attempt on A, one success on B");
        assert_eq!(seen[0].account, "acct_A");
        assert_eq!(seen[1].account, "acct_B", "header must follow the pinned token");
        // Bearer and header are derived from the SAME credential on every
        // attempt: A's token with A's id, then B's token with B's id — never
        // a mixed pair.
        assert_eq!(seen[0].bearer, fake_codex_token_for("acct_A"));
        assert_eq!(seen[1].bearer, fake_codex_token_for("acct_B"));
        assert_ne!(seen[0].bearer, seen[1].bearer);
        assert_eq!(seen[0].body, seen[1].body, "same request bytes, no rebuild");
        assert_eq!(router.failover_calls.load(Ordering::SeqCst), 1);
        assert_eq!(router.reports(), vec!["openai-codex@astra1".to_string()]);
        let notes = notices(&mut rx);
        assert!(
            notes.iter().any(|n| n.contains("'astra1'") && n.contains("'astra2'")),
            "operator must see the switch: {notes:?}"
        );
        assert!(
            notes.iter().all(|n| !n.contains("acct_") && !n.contains("h.")),
            "notices carry labels only, never ids/tokens: {notes:?}"
        );
    }

    /// Explicitly selected account: never switched, even with an eligible
    /// alternative. The turn ends with the static quota message and no
    /// transport retries are burned on a provably exhausted seat.
    #[tokio::test]
    async fn quota_429_on_explicit_account_never_fails_over() {
        let (url, seen) = spawn_scripted_codex(vec![(
            StatusCode::TOO_MANY_REQUESTS,
            vec![],
            QUOTA_429,
        )])
        .await;
        let router = ScriptedRouter::new(
            pinned("astra1", "acct_A", false),
            vec![Some(pinned("astra2", "acct_B", true))],
        );
        let (tx, _rx) = mpsc::unbounded_channel();
        let err = run_codex_with_router(&url, &router, 5, &tx)
            .await
            .expect_err("explicit account must fail, not switch");
        assert_eq!(err.to_string(), format!("Codex{RESPONSES_QUOTA_SUFFIX}"));
        assert!(!err.to_string().contains("ECHOED"));
        assert_eq!(seen.lock().unwrap().len(), 1, "no retry on exhausted seat");
        assert_eq!(router.failover_calls.load(Ordering::SeqCst), 0);
        // The cooldown is still reported so the broker stops advertising A.
        assert_eq!(router.reports(), vec!["openai-codex@astra1".to_string()]);
    }

    /// A generic 429 (no exhaustion evidence) keeps the ordinary transport
    /// retry on the SAME account — it is not proof of a spent window.
    #[tokio::test]
    async fn generic_429_retries_same_account_and_never_switches() {
        let (url, seen) = spawn_scripted_codex(vec![
            (StatusCode::TOO_MANY_REQUESTS, vec![], GENERIC_429),
            (StatusCode::OK, vec![], CODEX_SSE_SUCCESS),
        ])
        .await;
        let router = ScriptedRouter::new(
            pinned("astra1", "acct_A", true),
            vec![Some(pinned("astra2", "acct_B", true))],
        );
        let (tx, _rx) = mpsc::unbounded_channel();
        run_codex_with_router(&url, &router, 2, &tx).await.unwrap();
        let seen = seen.lock().unwrap();
        assert_eq!(seen.len(), 2);
        assert!(seen.iter().all(|r| r.account == "acct_A"));
        assert!(seen.iter().all(|r| r.bearer == fake_codex_token_for("acct_A")));
        assert_eq!(router.failover_calls.load(Ordering::SeqCst), 0);
        assert!(router.reports().is_empty());
    }

    /// Exhausted-window headers on a 429 are provider-declared evidence too.
    #[tokio::test]
    async fn quota_429_by_headers_fails_over() {
        let (url, seen) = spawn_scripted_codex(vec![
            (
                StatusCode::TOO_MANY_REQUESTS,
                vec![
                    ("x-codex-primary-used-percent", "100"),
                    ("x-codex-primary-reset-after-seconds", "500"),
                ],
                "{}",
            ),
            (StatusCode::OK, vec![], CODEX_SSE_SUCCESS),
        ])
        .await;
        let router = ScriptedRouter::new(
            pinned("astra1", "acct_A", true),
            vec![Some(pinned("astra2", "acct_B", true))],
        );
        let (tx, _rx) = mpsc::unbounded_channel();
        run_codex_with_router(&url, &router, 2, &tx).await.unwrap();
        assert_eq!(seen.lock().unwrap()[1].account, "acct_B");
        let reported = router.reported.lock().unwrap();
        assert_eq!(reported[0].1.reset_after_secs, Some(500));
    }

    /// No eligible seat: one quota failure ends the turn — no loop, no
    /// second attempt on the exhausted account.
    #[tokio::test]
    async fn quota_429_with_no_eligible_candidate_fails_without_looping() {
        let (url, seen) = spawn_scripted_codex(vec![(
            StatusCode::TOO_MANY_REQUESTS,
            vec![],
            QUOTA_429,
        )])
        .await;
        let router = ScriptedRouter::new(pinned("astra1", "acct_A", true), vec![None]);
        let (tx, _rx) = mpsc::unbounded_channel();
        let err = run_codex_with_router(&url, &router, 10, &tx)
            .await
            .expect_err("must fail");
        assert_eq!(err.to_string(), format!("Codex{RESPONSES_QUOTA_SUFFIX}"));
        assert_eq!(seen.lock().unwrap().len(), 1);
        assert_eq!(router.failover_calls.load(Ordering::SeqCst), 1);
    }

    /// The router hands back the very account that failed (e.g. a broker
    /// whose cooldown sink is a no-op): refused, never retried.
    #[tokio::test]
    async fn failover_to_same_account_is_refused() {
        let (url, seen) = spawn_scripted_codex(vec![(
            StatusCode::TOO_MANY_REQUESTS,
            vec![],
            QUOTA_429,
        )])
        .await;
        let router = ScriptedRouter::new(
            pinned("astra1", "acct_A", true),
            vec![Some(pinned("astra1", "acct_A", true))],
        );
        let (tx, _rx) = mpsc::unbounded_channel();
        let err = run_codex_with_router(&url, &router, 10, &tx)
            .await
            .expect_err("must fail");
        assert_eq!(err.to_string(), format!("Codex{RESPONSES_QUOTA_SUFFIX}"));
        assert_eq!(seen.lock().unwrap().len(), 1);
    }

    /// Both seats exhausted: exactly one switch, then stop (bounded to
    /// `MAX_CODEX_ACCOUNT_FAILOVERS`), even with a third candidate scripted.
    #[tokio::test]
    async fn all_accounts_exhausted_stops_after_one_failover() {
        let (url, seen) = spawn_scripted_codex(vec![(
            StatusCode::TOO_MANY_REQUESTS,
            vec![],
            QUOTA_429,
        )])
        .await;
        let router = ScriptedRouter::new(
            pinned("astra1", "acct_A", true),
            vec![
                Some(pinned("astra2", "acct_B", true)),
                Some(pinned("astra3", "acct_C", true)),
            ],
        );
        let (tx, _rx) = mpsc::unbounded_channel();
        let err = run_codex_with_router(&url, &router, 10, &tx)
            .await
            .expect_err("must fail");
        assert_eq!(err.to_string(), format!("Codex{RESPONSES_QUOTA_SUFFIX}"));
        let seen = seen.lock().unwrap();
        assert_eq!(seen.len(), 2, "A then B, never C");
        assert_eq!(seen[0].account, "acct_A");
        assert_eq!(seen[1].account, "acct_B");
        assert_eq!(router.failover_calls.load(Ordering::SeqCst), 1);
        assert_eq!(
            router.reports(),
            vec![
                "openai-codex@astra1".to_string(),
                "openai-codex@astra2".to_string()
            ],
            "both exhausted seats reported for cooldown"
        );
    }

    /// In-stream quota failure (HTTP 200 + terminal `insufficient_quota`)
    /// with NO output produced: one failover.
    #[tokio::test]
    async fn in_stream_quota_before_output_fails_over() {
        const QUOTA_SSE: &str = concat!(
            "data: {\"type\":\"response.created\",\"response\":{\"id\":\"r\"}}\n\n",
            "data: {\"type\":\"response.failed\",\"response\":{\"error\":{\"type\":\"insufficient_quota\",\"message\":\"ECHOED\"}}}\n\n",
            "data: [DONE]\n\n",
        );
        let (url, seen) = spawn_scripted_codex(vec![
            (StatusCode::OK, vec![], QUOTA_SSE),
            (StatusCode::OK, vec![], CODEX_SSE_SUCCESS),
        ])
        .await;
        let router = ScriptedRouter::new(
            pinned("astra1", "acct_A", true),
            vec![Some(pinned("astra2", "acct_B", true))],
        );
        let (tx, _rx) = mpsc::unbounded_channel();
        let result = run_codex_with_router(&url, &router, 2, &tx).await.unwrap();
        assert_eq!(result["content"][0]["text"], "hello");
        let seen = seen.lock().unwrap();
        assert_eq!(seen.len(), 2);
        assert_eq!(seen[1].account, "acct_B");
        assert_eq!(seen[1].bearer, fake_codex_token_for("acct_B"));
    }

    /// In-stream quota failure AFTER the model produced content — here only
    /// a reasoning-summary delta, which this decoder does not even forward —
    /// is terminal: no replay on another account.
    #[tokio::test]
    async fn in_stream_quota_after_any_output_never_replays() {
        const QUOTA_AFTER_REASONING: &str = concat!(
            "data: {\"type\":\"response.reasoning_summary_text.delta\",\"delta\":\"thinking…\"}\n\n",
            "data: {\"type\":\"response.failed\",\"response\":{\"error\":{\"code\":\"quota_exceeded\",\"message\":\"x\"}}}\n\n",
            "data: [DONE]\n\n",
        );
        let (url, seen) = spawn_scripted_codex(vec![
            (StatusCode::OK, vec![], QUOTA_AFTER_REASONING),
            (StatusCode::OK, vec![], CODEX_SSE_SUCCESS),
        ])
        .await;
        let router = ScriptedRouter::new(
            pinned("astra1", "acct_A", true),
            vec![Some(pinned("astra2", "acct_B", true))],
        );
        let (tx, _rx) = mpsc::unbounded_channel();
        let err = run_codex_with_router(&url, &router, 5, &tx)
            .await
            .expect_err("must not replay after output");
        assert_eq!(err.to_string(), format!("Codex{RESPONSES_QUOTA_SUFFIX}"));
        assert_eq!(seen.lock().unwrap().len(), 1);
        assert_eq!(router.failover_calls.load(Ordering::SeqCst), 0);
        // Still reported: the seat IS exhausted.
        assert_eq!(router.reports(), vec!["openai-codex@astra1".to_string()]);
    }

    /// Text already streamed to the UI, then quota: terminal.
    #[tokio::test]
    async fn in_stream_quota_after_text_never_replays() {
        const QUOTA_AFTER_TEXT: &str = concat!(
            "data: {\"type\":\"response.output_text.delta\",\"delta\":\"partial\"}\n\n",
            "data: {\"type\":\"error\",\"code\":\"usage_limit_reached\",\"message\":\"x\"}\n\n",
        );
        let (url, seen) = spawn_scripted_codex(vec![
            (StatusCode::OK, vec![], QUOTA_AFTER_TEXT),
            (StatusCode::OK, vec![], CODEX_SSE_SUCCESS),
        ])
        .await;
        let router = ScriptedRouter::new(
            pinned("astra1", "acct_A", true),
            vec![Some(pinned("astra2", "acct_B", true))],
        );
        let (tx, _rx) = mpsc::unbounded_channel();
        let err = run_codex_with_router(&url, &router, 5, &tx)
            .await
            .expect_err("must not replay after output");
        assert_eq!(err.to_string(), format!("Codex{RESPONSES_QUOTA_SUFFIX}"));
        assert_eq!(seen.lock().unwrap().len(), 1);
        assert_eq!(router.failover_calls.load(Ordering::SeqCst), 0);
    }

    /// `usage_not_included` is deterministic for the account: no transport
    /// retry, no failover, status error surfaced.
    #[tokio::test]
    async fn usage_not_included_fails_fast_without_retry_or_failover() {
        let (url, seen) = spawn_scripted_codex(vec![(
            StatusCode::TOO_MANY_REQUESTS,
            vec![],
            r#"{"error":{"type":"usage_not_included","message":"no codex here"}}"#,
        )])
        .await;
        let router = ScriptedRouter::new(
            pinned("astra1", "acct_A", true),
            vec![Some(pinned("astra2", "acct_B", true))],
        );
        let (tx, _rx) = mpsc::unbounded_channel();
        let err = run_codex_with_router(&url, &router, 5, &tx)
            .await
            .expect_err("must fail");
        assert!(err.to_string().starts_with("codex request failed: 429"), "{err}");
        assert!(!err.to_string().contains("no codex here"));
        assert_eq!(seen.lock().unwrap().len(), 1);
        assert_eq!(router.failover_calls.load(Ordering::SeqCst), 0);
        assert!(router.reports().is_empty());
    }

    /// A transient 503 (transport retry on A) followed by a quota 429 on A:
    /// still pre-output, so one failover to B is allowed; B then needs its
    /// own transient retry, which the shared budget still covers — a
    /// failover neither consumes nor resets transport retries.
    #[tokio::test]
    async fn transport_retry_then_quota_then_failover_keeps_one_retry_budget() {
        let (url, seen) = spawn_scripted_codex(vec![
            (StatusCode::SERVICE_UNAVAILABLE, vec![], "{}"),
            (StatusCode::TOO_MANY_REQUESTS, vec![], QUOTA_429),
            (StatusCode::SERVICE_UNAVAILABLE, vec![], "{}"),
            (StatusCode::OK, vec![], CODEX_SSE_SUCCESS),
        ])
        .await;
        let router = ScriptedRouter::new(
            pinned("astra1", "acct_A", true),
            vec![Some(pinned("astra2", "acct_B", true))],
        );
        let (tx, _rx) = mpsc::unbounded_channel();
        run_codex_with_router(&url, &router, 2, &tx).await.unwrap();
        let seen = seen.lock().unwrap();
        let accounts: Vec<&str> = seen.iter().map(|r| r.account.as_str()).collect();
        assert_eq!(accounts, ["acct_A", "acct_A", "acct_B", "acct_B"]);
        assert_eq!(router.failover_calls.load(Ordering::SeqCst), 1);
    }

    /// Transport budget already spent on A when the quota 429 arrives: the
    /// failover is still permitted (it is not a transport retry), but B gets
    /// no further transport retries — bounded on both axes.
    #[tokio::test]
    async fn failover_does_not_reset_transport_budget() {
        let (url, seen) = spawn_scripted_codex(vec![
            (StatusCode::SERVICE_UNAVAILABLE, vec![], "{}"),
            (StatusCode::TOO_MANY_REQUESTS, vec![], QUOTA_429),
            (StatusCode::SERVICE_UNAVAILABLE, vec![], "{}"),
        ])
        .await;
        let router = ScriptedRouter::new(
            pinned("astra1", "acct_A", true),
            vec![Some(pinned("astra2", "acct_B", true))],
        );
        let (tx, _rx) = mpsc::unbounded_channel();
        let err = run_codex_with_router(&url, &router, 1, &tx)
            .await
            .expect_err("B's 503 exceeds the shared budget");
        assert!(err.to_string().starts_with("codex request failed: 503"), "{err}");
        let accounts: Vec<String> = seen.lock().unwrap().iter().map(|r| r.account.clone()).collect();
        assert_eq!(accounts, ["acct_A", "acct_A", "acct_B"]);
    }

    /// Explicit NAMED account (`openai-codex@astra1`) with an in-stream quota
    /// failure and an eligible alternative scripted: never rotates.
    #[tokio::test]
    async fn explicit_named_account_in_stream_quota_never_rotates() {
        const QUOTA_SSE: &str = concat!(
            "data: {\"type\":\"response.failed\",\"response\":{\"error\":{\"type\":\"insufficient_quota\",\"message\":\"x\"}}}\n\n",
            "data: [DONE]\n\n",
        );
        let (url, seen) = spawn_scripted_codex(vec![
            (StatusCode::OK, vec![], QUOTA_SSE),
            (StatusCode::OK, vec![], CODEX_SSE_SUCCESS),
        ])
        .await;
        let router = ScriptedRouter::new(
            pinned("astra1", "acct_A", false),
            vec![Some(pinned("astra2", "acct_B", true))],
        );
        let (tx, mut rx) = mpsc::unbounded_channel();
        let err = run_codex_with_router(&url, &router, 5, &tx)
            .await
            .expect_err("explicit named account must not rotate");
        assert_eq!(err.to_string(), format!("Codex{RESPONSES_QUOTA_SUFFIX}"));
        assert_eq!(seen.lock().unwrap().len(), 1);
        assert_eq!(router.failover_calls.load(Ordering::SeqCst), 0);
        assert!(notices(&mut rx).iter().all(|n| !n.contains("switched")));
    }

    /// Explicit DEFAULT slot behaves the same as a named one: no rotation.
    #[tokio::test]
    async fn explicit_default_account_quota_429_never_rotates() {
        let (url, seen) = spawn_scripted_codex(vec![
            (StatusCode::TOO_MANY_REQUESTS, vec![], QUOTA_429),
            (StatusCode::OK, vec![], CODEX_SSE_SUCCESS),
        ])
        .await;
        let router = ScriptedRouter::new(
            pinned("default", "acct_D", false),
            vec![Some(pinned("astra2", "acct_B", true))],
        );
        let (tx, _rx) = mpsc::unbounded_channel();
        let err = run_codex_with_router(&url, &router, 5, &tx)
            .await
            .expect_err("explicit default must not rotate");
        assert_eq!(err.to_string(), format!("Codex{RESPONSES_QUOTA_SUFFIX}"));
        assert_eq!(seen.lock().unwrap().len(), 1);
        assert_eq!(router.reports(), vec!["openai-codex".to_string()]);
    }

    /// Reasoning-only output as an output ITEM (no text delta, nothing
    /// forwarded to the UI) followed by a quota failure: terminal.
    #[tokio::test]
    async fn in_stream_quota_after_reasoning_item_never_replays() {
        const REASONING_ITEM_THEN_QUOTA: &str = concat!(
            "data: {\"type\":\"response.created\",\"response\":{\"id\":\"r\"}}\n\n",
            "data: {\"type\":\"response.in_progress\",\"response\":{\"id\":\"r\"}}\n\n",
            "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"reasoning\",\"id\":\"rs_1\",\"summary\":[]}}\n\n",
            "data: {\"type\":\"response.failed\",\"response\":{\"error\":{\"type\":\"insufficient_quota\",\"message\":\"x\"}}}\n\n",
            "data: [DONE]\n\n",
        );
        let (url, seen) = spawn_scripted_codex(vec![
            (StatusCode::OK, vec![], REASONING_ITEM_THEN_QUOTA),
            (StatusCode::OK, vec![], CODEX_SSE_SUCCESS),
        ])
        .await;
        let router = ScriptedRouter::new(
            pinned("astra1", "acct_A", true),
            vec![Some(pinned("astra2", "acct_B", true))],
        );
        let (tx, mut rx) = mpsc::unbounded_channel();
        let err = run_codex_with_router(&url, &router, 5, &tx)
            .await
            .expect_err("reasoning output already produced");
        assert_eq!(err.to_string(), format!("Codex{RESPONSES_QUOTA_SUFFIX}"));
        assert_eq!(seen.lock().unwrap().len(), 1);
        assert_eq!(router.failover_calls.load(Ordering::SeqCst), 0);
        // Nothing reached the UI — the gate is stricter than "was forwarded".
        while let Ok(ev) = rx.try_recv() {
            assert!(
                !matches!(ev, StreamEvent::Llm(crate::LlmEvent::Text(_))),
                "no text was forwarded"
            );
        }
    }

    /// A metadata-only prelude (`response.created` / `in_progress`) does not
    /// count as output: failover stays permitted.
    #[tokio::test]
    async fn metadata_only_prelude_does_not_block_failover() {
        const METADATA_THEN_QUOTA: &str = concat!(
            "data: {\"type\":\"response.created\",\"response\":{\"id\":\"r\"}}\n\n",
            "data: {\"type\":\"response.in_progress\",\"response\":{\"id\":\"r\"}}\n\n",
            "data: {\"type\":\"error\",\"code\":\"usage_limit_reached\",\"message\":\"x\"}\n\n",
        );
        let (url, seen) = spawn_scripted_codex(vec![
            (StatusCode::OK, vec![], METADATA_THEN_QUOTA),
            (StatusCode::OK, vec![], CODEX_SSE_SUCCESS),
        ])
        .await;
        let router = ScriptedRouter::new(
            pinned("astra1", "acct_A", true),
            vec![Some(pinned("astra2", "acct_B", true))],
        );
        let (tx, _rx) = mpsc::unbounded_channel();
        run_codex_with_router(&url, &router, 2, &tx).await.unwrap();
        assert_eq!(seen.lock().unwrap()[1].account, "acct_B");
    }

    /// A failover candidate whose token carries no ChatGPT account claim
    /// cannot be paired with a header: the turn fails at the runtime, and
    /// no request is ever sent with a header borrowed from the failed seat.
    #[tokio::test]
    async fn failover_candidate_without_account_claim_never_sends_mixed_headers() {
        let (url, seen) = spawn_scripted_codex(vec![
            (StatusCode::TOO_MANY_REQUESTS, vec![], QUOTA_429),
            (StatusCode::OK, vec![], CODEX_SSE_SUCCESS),
        ])
        .await;
        let claimless = PinnedAccount {
            credential: CredentialRef::new(
                OAuthProviderId::OpenAiCodex,
                Account::parse("astra2").unwrap(),
            ),
            token: format!(
                "h.{}.s",
                URL_SAFE_NO_PAD.encode(serde_json::json!({"sub": "nobody"}).to_string())
            ),
            auto: true,
        };
        let router = ScriptedRouter::new(pinned("astra1", "acct_A", true), vec![Some(claimless)]);
        let (tx, _rx) = mpsc::unbounded_channel();
        let err = run_codex_with_router(&url, &router, 5, &tx)
            .await
            .expect_err("claimless token cannot be paired");
        assert!(err.to_string().contains("account id"), "{err}");
        let seen = seen.lock().unwrap();
        assert_eq!(seen.len(), 1, "only A's attempt was sent");
        assert_eq!(seen[0].account, "acct_A");
    }

    /// The router itself failing during failover is not a loop: report,
    /// fail the turn with the quota message.
    #[tokio::test]
    async fn router_failover_error_ends_turn() {
        struct FailingRouter(PinnedAccount);
        #[async_trait]
        impl CodexAccountRouter for FailingRouter {
            async fn pin(&self, _m: &str) -> Result<PinnedAccount, String> {
                Ok(self.0.clone())
            }
            async fn failover(
                &self,
                _m: &str,
                _f: &PinnedAccount,
                _e: &CodexQuotaEvidence,
                _n: u32,
            ) -> Result<Option<PinnedAccount>, String> {
                Err("broker unreachable".into())
            }
            async fn report_exhausted(&self, _f: &PinnedAccount, _e: &CodexQuotaEvidence) {}
        }
        let (url, seen) = spawn_scripted_codex(vec![(
            StatusCode::TOO_MANY_REQUESTS,
            vec![],
            QUOTA_429,
        )])
        .await;
        let router = FailingRouter(pinned("astra1", "acct_A", true));
        let (tx, _rx) = mpsc::unbounded_channel();
        let err = run_codex_with_router(&url, &router, 5, &tx)
            .await
            .expect_err("must fail");
        assert_eq!(err.to_string(), format!("Codex{RESPONSES_QUOTA_SUFFIX}"));
        assert_eq!(seen.lock().unwrap().len(), 1);
    }

    /// The broker adapter with the default trait surface pins the policy
    /// account as an explicit selection and reports no candidates: the old
    /// `TokenOnlyBroker` mock keeps working and a quota 429 simply fails.
    #[tokio::test]
    async fn broker_adapter_default_trait_surface_is_explicit_and_fails_closed() {
        let (url, seen) = spawn_scripted_codex(vec![(
            StatusCode::TOO_MANY_REQUESTS,
            vec![],
            QUOTA_429,
        )])
        .await;
        let err = run_codex(&url, 5).await.expect_err("must fail");
        assert_eq!(err.to_string(), format!("Codex{RESPONSES_QUOTA_SUFFIX}"));
        assert_eq!(seen.lock().unwrap().len(), 1, "no retry, no failover");
    }

    /// The dispatch seam lifts the generic three-attempt budget to the
    /// persistent posture adopted for Anthropic OAuth overloads (10 retries)
    /// — incident: 2026-07-16 chatgpt.com 503/520/timeout bursts.
    #[test]
    fn codex_retry_budget_lifts_configured_default_to_persistent_floor() {
        assert_eq!(codex_retry_budget(3), CODEX_PERSISTENT_RETRIES);
        assert_eq!(codex_retry_budget(0), CODEX_PERSISTENT_RETRIES);
        assert_eq!(codex_retry_budget(10), 10);
    }

    /// A user who explicitly configured an even larger budget keeps it —
    /// the floor only ever raises, never lowers.
    #[test]
    fn codex_retry_budget_honors_larger_user_configuration() {
        assert_eq!(codex_retry_budget(15), 15);
    }

    /// Backoff doubles per attempt but caps at 2^6 — a 10-deep budget must
    /// never sleep 512s on its final attempt.
    /// 04-tui I4: the incomplete-response retry is a typed retry (the
    /// header's "retrying"), not a notice the client has to recognise.
    #[tokio::test]
    async fn a_stream_retry_is_a_typed_retry_event() {
        let (tx, mut rx) = mpsc::unbounded_channel();
        let cancel = tokio_util::sync::CancellationToken::new();
        let waiting = cancel.clone();
        let task = tokio::spawn(async move { wait_stream_retry(&tx, &waiting, 1, 3, true).await.is_ok() });
        let ev = rx.recv().await.expect("the retry is announced");
        cancel.cancel();
        assert!(!task.await.unwrap(), "cancel ends the wait");
        assert!(
            matches!(&ev, StreamEvent::Session(crate::SessionEvent::Retry(t)) if t.starts_with("Response incomplete — retrying")),
            "{ev:?}"
        );
    }

    #[test]
    fn retry_delay_is_exponential_with_a_capped_tail() {
        assert_eq!(retry_delay(1).as_secs(), 1);
        assert_eq!(retry_delay(2).as_secs(), 2);
        assert_eq!(retry_delay(4).as_secs(), 8);
        assert_eq!(retry_delay(7).as_secs(), 64);
        assert_eq!(retry_delay(10).as_secs(), 64);
    }

    // ─── Phase 1 privacy: hostile provider echoes the request ───────────────
    // spec §5.1: provider error bodies are untrusted and may echo the full
    // request (prompts, system text, tool schemas, credentials). Neither the
    // surfaced error nor ANY log line may contain response-body content —
    // status + provider label only.

    /// Unique raw-content sentinel placed in the outgoing request body.
    const ECHO_SENTINEL: &str = "PH1-OAI-SENTINEL-2b8c4d-RAW-CONTENT";

    /// `std::io::Write` sink capturing formatted tracing output in-process.
    struct CaptureWriter(Arc<std::sync::Mutex<Vec<u8>>>);

    impl std::io::Write for CaptureWriter {
        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
            self.0.lock().unwrap().extend_from_slice(buf);
            Ok(buf.len())
        }
        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }

    fn capture_subscriber(
        buf: &Arc<std::sync::Mutex<Vec<u8>>>,
    ) -> impl tracing::Subscriber + Send + Sync {
        let sink = buf.clone();
        tracing_subscriber::fmt()
            .with_max_level(tracing::Level::TRACE)
            .with_writer(move || CaptureWriter(sink.clone()))
            .with_ansi(false)
            .finish()
    }

    /// Hostile loopback provider: every request is answered with `status` and
    /// a JSON error envelope whose `message` is `"ECHOED:" + the full request
    /// body` — the preserved holdout probe shape, aimed at the OpenAI routes.
    async fn spawn_hostile_echo_provider(status: StatusCode) -> String {
        let app = Router::new().fallback(move |body: String| async move {
            let envelope = serde_json::json!({
                "error": { "message": format!("ECHOED:{body}") }
            });
            (
                status,
                [("content-type", "application/json")],
                envelope.to_string(),
            )
                .into_response()
        });
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
        format!("http://{addr}")
    }

    /// Request body shaped like a real provider request: prompt sentinel plus
    /// a distinctive tool schema. If the provider echo survives anywhere,
    /// these markers betray it.
    fn sentinel_request_body() -> Value {
        json!({
            "model": "gpt-test",
            "messages": [
                {"role": "system", "content": format!("system secret {ECHO_SENTINEL}")},
                {"role": "user", "content": ECHO_SENTINEL},
            ],
            "tools": [{
                "type": "function",
                "function": {
                    "name": "ph1_secret_tool_zz",
                    "description": "internal tool schema",
                    "parameters": {"type": "object", "properties": {}}
                }
            }],
        })
    }

    /// Markers that must never surface in errors or logs.
    const BANNED_MARKERS: &[&str] = &[
        "ECHOED",
        ECHO_SENTINEL,
        "ph1_secret_tool_zz",
        "\"messages\"",
        "system secret",
    ];

    fn assert_no_banned(haystack: &str, context: &str) {
        for banned in BANNED_MARKERS {
            assert!(
                !haystack.contains(banned),
                "provider body content `{banned}` leaked into {context}: {haystack}"
            );
        }
    }

    #[tokio::test]
    async fn send_with_retries_non_retryable_error_never_surfaces_provider_body() {
        let url = spawn_hostile_echo_provider(StatusCode::BAD_REQUEST).await;
        let client = reqwest::Client::new();
        let body = sentinel_request_body();

        let buf = Arc::new(std::sync::Mutex::new(Vec::new()));
        let _guard = tracing::subscriber::set_default(capture_subscriber(&buf));

        let err = send_with_retries(
            SendTarget {
                label: "codex",
                url: &url,
                quota_probe: true,
            },
            || client.post(&url).json(&body),
            &tokio_util::sync::CancellationToken::new(),
            3,
            &mut 0,
            &mut tr::StreamAttempt::new(None),
        )
        .await
        .expect_err("400 must fail fast");

        let msg = err.to_string();
        assert!(
            msg.starts_with("codex request failed: 400"),
            "status + label must survive for classification/usability: {msg}"
        );
        assert_no_banned(&msg, "the surfaced error");
        let logs = String::from_utf8(buf.lock().unwrap().clone()).unwrap();
        assert_no_banned(&logs, "tracing output");
    }

    /// Answer every request with `status` and a fixed body.
    async fn spawn_fixed_body_provider(status: StatusCode, body: String) -> String {
        let app = Router::new().fallback(move || {
            let body = body.clone();
            async move { (status, [("content-type", "application/json")], body).into_response() }
        });
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
        format!("http://{addr}")
    }

    /// Codex 4xx: identifiers and Synaps-authored text only. Every probe is
    /// plain prose that a quote/length/charset sanitizer lets through; none
    /// of it may reach the error string, the net.rs classification log, or
    /// any captured tracing output.
    #[tokio::test]
    async fn codex_4xx_never_surfaces_provider_prose_in_error_or_logs() {
        const PROSE: &str = "ZZPROSE";
        let probes = [
            // Apostrophe flips naive quote parsing; the echo follows it.
            "The model can't accept 'ZZPROSE echo of the user prompt, it's long' here".to_string(),
            // Short quoted span with a secret.
            format!("Invalid 'input[0]': '{PROSE} AKIAIOSFODNN7EXAMPLE pls'"),
            // Unquoted prose echo.
            format!("Your request about {PROSE} my private quarterly notes was rejected"),
            // Curly / guillemet quotes.
            format!("Bad value \u{2018}{PROSE} one\u{2019} and \u{201c}{PROSE} two\u{201d} \u{ab}x\u{bb}"),
            // Secret glued to punctuation.
            format!("{PROSE} key:sk-proj-abc123 password=hunter2 https://h/x?token=abc"),
            // Bidi override and zero-width characters.
            format!("{PROSE}\u{202e}gnirts\u{202c} s\u{200b}k-\u{2066}live\u{2069}"),
        ];
        for prose in probes {
            let body = serde_json::json!({"error": {
                "message": prose,
                "type": "invalid_request_error",
                "code": "unsupported_value",
                "param": "reasoning.effort",
            }})
            .to_string();
            let url = spawn_fixed_body_provider(StatusCode::BAD_REQUEST, body).await;
            let client = reqwest::Client::new();
            let buf = Arc::new(std::sync::Mutex::new(Vec::new()));
            let _guard = tracing::subscriber::set_default(capture_subscriber(&buf));
            let err = send_with_retries(
                SendTarget {
                    label: "codex",
                    url: &url,
                    quota_probe: true,
                },
                || client.post(&url).body("{}"),
                &tokio_util::sync::CancellationToken::new(),
                3,
                &mut 0,
                &mut tr::StreamAttempt::new(None),
            )
            .await
            .expect_err("400 must fail fast");
            // Run the error through the route classifier too: it logs.
            let runtime =
                super::super::net::provider_error_to_runtime_for("openai-codex/gpt-5.6-sol", err);
            let msg = runtime.to_string();
            let logs = String::from_utf8(buf.lock().unwrap().clone()).unwrap();
            for (where_, text) in [("error", &msg), ("logs", &logs)] {
                for banned in [
                    PROSE, "can't", "AKIA", "sk-", "hunter2", "token=", "private", "gnirts",
                    "\u{202e}", "\u{200b}", "\u{2066}", "\u{2018}", "\u{201c}",
                ] {
                    assert!(
                        !text.contains(banned),
                        "provider prose `{banned}` leaked into {where_} for probe {prose:?}: {text}"
                    );
                }
            }
            // Identifiers and our own explanation do surface.
            assert!(
                msg.contains("[invalid_request_error/unsupported_value, param reasoning.effort]"),
                "{msg}"
            );
            assert!(msg.contains("/effort"), "static explanation missing: {msg}");
        }
    }

    #[tokio::test(start_paused = true)]
    async fn send_with_retries_retry_logs_never_contain_provider_body() {
        let url = spawn_hostile_echo_provider(StatusCode::SERVICE_UNAVAILABLE).await;
        let client = reqwest::Client::new();
        let body = sentinel_request_body();

        let buf = Arc::new(std::sync::Mutex::new(Vec::new()));
        let _guard = tracing::subscriber::set_default(capture_subscriber(&buf));

        let err = send_with_retries(
            SendTarget {
                label: "codex",
                url: &url,
                quota_probe: true,
            },
            || client.post(&url).json(&body),
            &tokio_util::sync::CancellationToken::new(),
            1,
            &mut 0,
            &mut tr::StreamAttempt::new(None),
        )
        .await
        .expect_err("persistent 503 must exhaust the budget");

        let msg = err.to_string();
        assert!(
            msg.starts_with("codex request failed: 503"),
            "status + label must survive after exhaustion: {msg}"
        );
        assert_no_banned(&msg, "the surfaced error");

        let logs = String::from_utf8(buf.lock().unwrap().clone()).unwrap();
        assert!(
            logs.contains("retry 1/1"),
            "retry accounting must stay observable: {logs}"
        );
        assert!(
            logs.contains("503"),
            "retry log must keep the status for diagnosis: {logs}"
        );
        assert_no_banned(&logs, "retry tracing output");
    }
}

#[cfg(test)]
mod interrupted_broker_tests {
    use super::*;
    use crate::auth::{
        AccessToken, BrokerError, CredentialBroker, ProxyByteStream, ProxyRequest, ProxyResponse,
    };
    use std::sync::{Arc, Mutex};
    use tokio_util::sync::CancellationToken;

    struct Broker {
        bodies: Vec<(&'static str, Option<BrokerError>)>,
        seen: Mutex<Vec<ProxyRequest>>,
    }
    #[async_trait::async_trait]
    impl CredentialBroker for Broker {
        async fn access_token(
            &self,
            _: crate::auth::OAuthProviderId,
        ) -> Result<AccessToken, BrokerError> {
            unreachable!()
        }
        async fn proxy(&self, _: ProxyRequest) -> Result<ProxyResponse, BrokerError> {
            unreachable!()
        }
        async fn proxy_stream(
            &self,
            request: ProxyRequest,
        ) -> Result<ProxyByteStream, BrokerError> {
            let mut seen = self.seen.lock().unwrap();
            let (body, error) = &self.bodies[seen.len().min(self.bodies.len() - 1)];
            seen.push(request);
            let mut chunks = vec![Ok(bytes::Bytes::from_static(body.as_bytes()))];
            if let Some(error) = error {
                chunks.push(Err(error.clone()));
            }
            Ok(Box::pin(futures::stream::iter(chunks)))
        }
        async fn anthropic_usage(&self) -> Result<Value, BrokerError> {
            unreachable!()
        }
        async fn capabilities(&self) -> Result<Vec<agent_core::auth::ProviderStatus>, BrokerError> {
            Ok(vec![])
        }
    }
    const CHAT_PARTIAL: &str = concat!(
        "data: {\"choices\":[{\"delta\":{\"content\":\"discard\"}}]}\n\n",
        "data: {\"choices\":[],\"usage\":{\"prompt_tokens\":7,\"completion_tokens\":1}}\n\n"
    );
    const CHAT_OK: &str =
        "data: {\"choices\":[{\"delta\":{\"content\":\"accepted\"}}]}\n\ndata: [DONE]\n\n";
    const XAI_PARTIAL: &str =
        "data: {\"type\":\"response.output_text.delta\",\"delta\":\"discard\"}\n\n";
    const XAI_OK: &str = "data: {\"type\":\"response.output_text.delta\",\"delta\":\"accepted\"}\n\ndata: {\"type\":\"response.completed\"}\n\n";

    async fn drive(
        broker: Arc<Broker>,
        xai: bool,
        retries: u32,
        suppressed: bool,
        cancel: &CancellationToken,
    ) -> (
        Result<Value, super::super::net::BoxedProviderError>,
        Vec<StreamEvent>,
    ) {
        let cfg = ProviderConfig {
            base_url: "http://unused.invalid".into(),
            model: if xai { "grok-4.5" } else { "test" }.into(),
            provider: if xai { "xai-auth" } else { "local" }.into(),
        };
        let broker: Arc<dyn CredentialBroker> = broker;
        let (tx, mut rx) = mpsc::unbounded_channel();
        let trace = crate::runtime::trace::TraceContext::disabled();
        let result = if xai {
            call_xai_responses_stream_inner(
                &cfg,
                &broker,
                &[],
                &None,
                &[],
                &tx,
                None,
                agent_core::reasoning::ReasoningLevel::Adaptive,
                cancel,
                retries,
                &trace,
                true,
            )
            .await
        } else {
            call_oai_stream_inner(
                &cfg,
                &broker,
                &[],
                &None,
                &[],
                &tx,
                None,
                None,
                0,
                agent_core::reasoning::ReasoningLevel::Medium,
                cancel,
                &trace,
                true,
                suppressed,
                retries,
            )
            .await
        };
        let mut events = vec![];
        while let Ok(event) = rx.try_recv() {
            events.push(event);
        }
        (result, events)
    }

    #[tokio::test(start_paused = true)]
    async fn broker_transport_recovery_preserves_usage_and_identical_bytes() {
        for xai in [false, true] {
            let broker = Arc::new(Broker {
                bodies: vec![
                    (
                        if xai { XAI_PARTIAL } else { CHAT_PARTIAL },
                        Some(BrokerError::Transport(
                            "unexpected EOF during chunk size line".into(),
                        )),
                    ),
                    (if xai { XAI_OK } else { CHAT_OK }, None),
                ],
                seen: Mutex::new(vec![]),
            });
            let (result, events) =
                drive(broker.clone(), xai, 1, false, &CancellationToken::new()).await;
            assert_eq!(result.unwrap()["content"][0]["text"], "accepted");
            let seen = broker.seen.lock().unwrap();
            assert_eq!(seen.len(), 2);
            assert_eq!(seen[0].body_bytes, seen[1].body_bytes);
            assert_eq!(
                events
                    .iter()
                    .filter(|e| matches!(e, StreamEvent::Llm(crate::LlmEvent::ResponseReset)))
                    .count(),
                1
            );
            if !xai {
                assert_eq!(
                    events
                        .iter()
                        .filter_map(|e| match e {
                            StreamEvent::Session(crate::SessionEvent::Usage {
                                input_tokens,
                                ..
                            }) => Some(*input_tokens),
                            _ => None,
                        })
                        .sum::<u64>(),
                    7
                );
            }
        }
    }
    #[tokio::test(start_paused = true)]
    async fn broker_nontransport_denial_never_retries_and_precancel_never_dispatches() {
        for xai in [false, true] {
            let broker = Arc::new(Broker {
                bodies: vec![(
                    if xai { XAI_PARTIAL } else { CHAT_PARTIAL },
                    Some(BrokerError::Denied("blocked".into())),
                )],
                seen: Mutex::new(vec![]),
            });
            let (result, events) =
                drive(broker.clone(), xai, 10, false, &CancellationToken::new()).await;
            assert!(result.is_err());
            assert_eq!(broker.seen.lock().unwrap().len(), 1);
            assert!(!events
                .iter()
                .any(|e| matches!(e, StreamEvent::Llm(crate::LlmEvent::ResponseReset))));
            let canceled = CancellationToken::new();
            canceled.cancel();
            assert!(drive(broker.clone(), xai, 10, false, &canceled)
                .await
                .0
                .is_err());
            assert_eq!(broker.seen.lock().unwrap().len(), 1);
        }
    }
    #[tokio::test(start_paused = true)]
    async fn suppressed_retry_emits_no_llm_previews() {
        let broker = Arc::new(Broker {
            bodies: vec![
                (CHAT_PARTIAL, Some(BrokerError::Transport("eof".into()))),
                (CHAT_OK, None),
            ],
            seen: Mutex::new(vec![]),
        });
        let (result, events) = drive(broker, false, 1, true, &CancellationToken::new()).await;
        assert!(result.is_ok());
        assert!(!events.iter().any(|e| matches!(e, StreamEvent::Llm(_))));
    }
    #[test]
    fn exhaustion_is_transient_for_auto_driver_but_not_arbitrary_error_prose() {
        use crate::extensions::session_driver::{classify_error, Outcome};
        assert_eq!(
            classify_error(STREAM_INTERRUPTED),
            (Outcome::ProviderError, "transient".into())
        );
        assert_eq!(
            classify_error(&format!("API error: {STREAM_INTERRUPTED}")),
            (Outcome::ProviderError, "transient".into())
        );
        assert_eq!(
            classify_error(&format!("denied: {STREAM_INTERRUPTED}")).0,
            Outcome::Blocked
        );
    }
}

#[cfg(test)]
mod muse_tests {
    //! Meta Muse Responses route: exact body shape (mirrors the official
    //! Muse Code client) and an end-to-end turn through a capturing broker.
    use super::*;
    use crate::auth::{
        AccessToken, BrokerError, CredentialBroker, ProxyByteStream, ProxyMethod, ProxyRequest,
        ProxyResponse,
    };
    use agent_core::reasoning::ReasoningLevel;
    use std::sync::{Arc, Mutex};

    #[test]
    fn body_mirrors_official_client_and_adaptive_omits_reasoning() {
        let input = json!([{"role":"user","content":[{"type":"input_text","text":"hi"}]}]);
        let tool = json!({"type":"function","name":"bash","parameters":{"type":"object"}});
        let key = muse_prompt_cache_key("sys", input.as_array().unwrap().first());
        let body = build_muse_body(
            "muse-wonder-1.0",
            ReasoningLevel::Adaptive,
            input.clone(),
            "sys",
            vec![tool.clone()],
            Some(2048),
            &key,
        )
        .unwrap();
        assert_eq!(body["model"], "muse-wonder-1.0");
        assert_eq!(body["instructions"], "sys");
        assert_eq!(body["input"], input);
        assert_eq!(body["store"], false);
        assert_eq!(body["stream"], true);
        assert_eq!(body["tools"], json!([tool]));
        assert_eq!(body["max_output_tokens"], 2048);
        assert_eq!(body["prompt_cache_key"], key.as_str());
        assert!(
            body.get("reasoning").is_none(),
            "adaptive = provider default"
        );
        assert!(body.get("messages").is_none());
        assert!(body.get("temperature").is_none());

        // Empty prompt/tools/limit are omitted, not sent empty.
        let bare = build_muse_body(
            "muse-wonder-1.0",
            ReasoningLevel::Adaptive,
            json!([]),
            "",
            Vec::new(),
            None,
            &key,
        )
        .unwrap();
        for absent in ["instructions", "tools", "max_output_tokens", "reasoning"] {
            assert!(bare.get(absent).is_none(), "{absent}");
        }
    }

    #[test]
    fn named_effort_requires_exact_live_evidence_before_any_io() {
        // Held for the whole test: the seed assertions read the shared
        // process-global `muse/*` cache rows too.
        let _guard = crate::runtime::openai::catalog::muse::test_cache_guard();
        crate::runtime::openai::catalog::capability_cache::replace_provider("muse", &[]);
        // The static seed advertises no effort set: named levels are refused
        // in the pure builder, i.e. before credentials or network.
        for level in [
            ReasoningLevel::Off,
            ReasoningLevel::High,
            ReasoningLevel::Max,
        ] {
            let err = build_muse_body(
                "muse-wonder-1.0",
                level,
                json!([]),
                "",
                Vec::new(),
                None,
                "k",
            )
            .unwrap_err();
            assert!(err.contains("adaptive"), "{level}: {err}");
        }
        // Uncataloged-but-routable id: provider default only; malformed: none.
        assert!(build_muse_body(
            "never-cataloged-zz9",
            ReasoningLevel::Adaptive,
            json!([]),
            "",
            Vec::new(),
            None,
            "k"
        )
        .is_ok());
        assert!(build_muse_body(
            "Bad Id",
            ReasoningLevel::Adaptive,
            json!([]),
            "",
            Vec::new(),
            None,
            "k"
        )
        .is_err());

        let models = crate::runtime::openai::catalog::parse_muse_catalog_models(
            r#"{"models":[{"id":"muse-body-probe-zz2","reasoning_effort_variants":
                [{"effort":"none"},{"effort":"xhigh"}]}]}"#,
        )
        .unwrap();
        crate::runtime::openai::catalog::capability_cache::replace_provider("muse", &models);
        for (level, effort) in [
            (ReasoningLevel::Off, "none"),
            (ReasoningLevel::XHigh, "xhigh"),
        ] {
            let body = build_muse_body(
                "muse-body-probe-zz2",
                level,
                json!([]),
                "",
                Vec::new(),
                None,
                "k",
            )
            .unwrap();
            assert_eq!(body["reasoning"], json!({"effort": effort}));
        }
        assert!(build_muse_body(
            "muse-body-probe-zz2",
            ReasoningLevel::High,
            json!([]),
            "",
            Vec::new(),
            None,
            "k"
        )
        .is_err());
        crate::runtime::openai::catalog::capability_cache::replace_provider("muse", &[]);
    }

    #[test]
    fn cache_key_is_deterministic_namespaced_and_prefix_sensitive() {
        let first = json!({"role":"user","content":"a"});
        let k1 = muse_prompt_cache_key("sys", Some(&first));
        assert_eq!(k1, muse_prompt_cache_key("sys", Some(&first)));
        assert!(k1.starts_with("synaps-muse-"), "{k1}");
        assert_ne!(k1, codex_prompt_cache_key("sys", Some(&first)));
        assert_ne!(k1, muse_prompt_cache_key("sys2", Some(&first)));
        assert_ne!(
            k1,
            muse_prompt_cache_key("sys", Some(&json!({"role":"user","content":"b"})))
        );
    }

    struct CaptureBroker {
        sse: &'static str,
        seen: Mutex<Vec<ProxyRequest>>,
    }

    #[async_trait::async_trait]
    impl CredentialBroker for CaptureBroker {
        async fn access_token(
            &self,
            _: crate::auth::OAuthProviderId,
        ) -> Result<AccessToken, BrokerError> {
            unreachable!("the runtime never holds the Muse key")
        }
        async fn proxy(&self, _: ProxyRequest) -> Result<ProxyResponse, BrokerError> {
            unreachable!()
        }
        async fn proxy_stream(
            &self,
            request: ProxyRequest,
        ) -> Result<ProxyByteStream, BrokerError> {
            self.seen.lock().unwrap().push(request);
            Ok(Box::pin(futures::stream::iter(vec![Ok(
                bytes::Bytes::from_static(self.sse.as_bytes()),
            )])))
        }
        async fn anthropic_usage(&self) -> Result<Value, BrokerError> {
            unreachable!()
        }
        async fn capabilities(&self) -> Result<Vec<agent_core::auth::ProviderStatus>, BrokerError> {
            Ok(vec![])
        }
    }

    const SSE: &str = concat!(
        "data: {\"type\":\"response.output_text.delta\",\"delta\":\"Hel\"}\n\n",
        "data: {\"type\":\"response.output_text.delta\",\"delta\":\"lo\"}\n\n",
        "data: {\"type\":\"response.output_item.done\",\"item\":{\"type\":\"function_call\",",
        "\"call_id\":\"call_1\",\"name\":\"bash\",\"arguments\":\"{\\\"command\\\":\\\"ls\\\"}\"}}\n\n",
        "data: {\"type\":\"response.completed\",\"response\":{\"usage\":",
        "{\"input_tokens\":12,\"output_tokens\":3}}}\n\n",
    );

    #[tokio::test]
    async fn turn_streams_through_broker_with_system_prompt_as_instructions() {
        let broker = Arc::new(CaptureBroker {
            sse: SSE,
            seen: Mutex::new(Vec::new()),
        });
        let dyn_broker: Arc<dyn CredentialBroker> = broker.clone();
        let cfg = ProviderConfig {
            base_url: crate::auth::muse::API_BASE_URL.into(),
            model: "muse-wonder-1.0".into(),
            provider: "muse".into(),
        };
        let tools = vec![json!({
            "name":"bash","description":"run","input_schema":{"type":"object","properties":{}}
        })];
        let messages: Vec<crate::SharedMessage> =
            vec![Arc::new(json!({"role":"user","content":"hi muse"}))];
        let (tx, mut rx) = mpsc::unbounded_channel::<StreamEvent>();
        let cancel = tokio_util::sync::CancellationToken::new();

        let result = call_responses_route_inner(
            "muse",
            &cfg,
            &dyn_broker,
            &tools,
            &Some("You are Synaps.".into()),
            &messages,
            &tx,
            Some(4096),
            ReasoningLevel::Adaptive,
            &cancel,
            0,
            &crate::runtime::trace::TraceContext::disabled(),
            true,
        )
        .await
        .expect("muse turn completes");

        // Assistant text + translated tool call come back Anthropic-shaped.
        let content = result["content"].as_array().unwrap();
        assert_eq!(content[0]["text"], "Hello");
        let tool_use = content
            .iter()
            .find(|block| block["type"] == "tool_use")
            .expect("tool call surfaced");
        assert_eq!(tool_use["name"], "bash");
        assert_eq!(tool_use["input"]["command"], "ls");

        // Exactly one broker-proxied POST to the pinned Responses path.
        let seen = broker.seen.lock().unwrap();
        assert_eq!(seen.len(), 1);
        let request = &seen[0];
        assert_eq!(request.provider, "muse");
        assert_eq!(request.method, ProxyMethod::Post);
        assert_eq!(request.path, "/v1/responses");
        assert!(request.stream);
        let body = request.body.as_ref().unwrap();
        assert_eq!(body["model"], "muse-wonder-1.0");
        assert_eq!(body["instructions"], "You are Synaps.");
        assert_eq!(body["store"], false);
        assert_eq!(body["max_output_tokens"], 4096);
        assert!(body["prompt_cache_key"]
            .as_str()
            .unwrap()
            .starts_with("synaps-muse-"));
        // System prompt is NOT duplicated into the input items.
        let wire = body["input"].to_string();
        assert!(!wire.contains("You are Synaps."), "{wire}");
        assert!(wire.contains("hi muse"), "{wire}");
        assert_eq!(body["tools"][0]["name"], "bash");
        // Exact-bytes handoff matches the structured body.
        let bytes = request.body_bytes.as_ref().expect("exact wire bytes");
        assert_eq!(
            serde_json::from_slice::<Value>(bytes).unwrap(),
            body.clone()
        );

        drop(tx);
        let mut streamed = String::new();
        while let Ok(event) = rx.try_recv() {
            if let StreamEvent::Llm(crate::runtime::types::LlmEvent::Text(text)) = event {
                streamed.push_str(&text);
            }
        }
        assert_eq!(streamed, "Hello");
    }

    #[test]
    fn dangling_preamble_trigger_cases() {
        // (i) Muse's own commentary label with no tool call.
        assert_eq!(
            muse_dangling_preamble(false, Some("commentary"), "Now checking prod logs."),
            Some("commentary_phase")
        );
        // (ii) short final text ending in ':' (the observed failures).
        assert_eq!(
            muse_dangling_preamble(
                false,
                None,
                "Key facts assembled. Blocking question before I build anything:\n"
            ),
            Some("colon_preamble")
        );
        assert_eq!(
            muse_dangling_preamble(
                false,
                Some("final_answer"),
                "Now checking whether prod logs show the error:"
            ),
            Some("colon_preamble")
        );
        // Never when the response acted.
        assert_eq!(
            muse_dangling_preamble(true, Some("commentary"), "Running:"),
            None
        );
        // Real answers: no colon, empty, or long content that ends in ':'.
        assert_eq!(muse_dangling_preamble(false, None, "All done."), None);
        assert_eq!(muse_dangling_preamble(false, None, "   "), None);
        assert_eq!(
            muse_dangling_preamble(false, Some("final_answer"), "Which format?"),
            None
        );
        let long = format!("{}:", "x".repeat(PREAMBLE_MAX_CHARS));
        assert_eq!(muse_dangling_preamble(false, None, &long), None);
        let edge = format!("{}:", "x".repeat(PREAMBLE_MAX_CHARS - 1));
        assert_eq!(
            muse_dangling_preamble(false, None, &edge),
            Some("colon_preamble")
        );
    }

    #[test]
    fn preamble_nudge_fires_at_most_once_per_turn() {
        let flagged =
            json!({"role":"assistant","content":[],DANGLING_PREAMBLE_KEY:"colon_preamble"});
        let clean = json!({"role":"assistant","content":[]});
        let mut nudged = false;
        assert_eq!(take_preamble_nudge(&clean, &mut nudged, false, true), None);
        assert!(!nudged, "a clean response must not consume the one nudge");
        assert_eq!(
            take_preamble_nudge(&flagged, &mut nudged, false, true).as_deref(),
            Some("colon_preamble")
        );
        assert!(nudged);
        // Second dangling preamble in the same turn: the turn ends normally.
        assert_eq!(
            take_preamble_nudge(&flagged, &mut nudged, false, true),
            None
        );
        // Cancelled rounds and the disable switch never nudge.
        let mut fresh = false;
        assert_eq!(take_preamble_nudge(&flagged, &mut fresh, true, true), None);
        assert_eq!(
            take_preamble_nudge(&flagged, &mut fresh, false, false),
            None
        );
        assert!(!fresh);
    }

    #[test]
    fn preamble_nudge_env_switch_parses_off_values_only() {
        assert!(preamble_nudge_enabled_value(None));
        assert!(preamble_nudge_enabled_value(Some("1")));
        assert!(preamble_nudge_enabled_value(Some("")));
        for off in ["0", "false", "OFF", " no "] {
            assert!(!preamble_nudge_enabled_value(Some(off)), "{off}");
        }
    }

    async fn muse_turn(sse: &'static str, messages: Vec<Value>) -> (Value, Value) {
        let broker = Arc::new(CaptureBroker {
            sse,
            seen: Mutex::new(Vec::new()),
        });
        let dyn_broker: Arc<dyn CredentialBroker> = broker.clone();
        let cfg = ProviderConfig {
            base_url: crate::auth::muse::API_BASE_URL.into(),
            model: "muse-wonder-1.0".into(),
            provider: "muse".into(),
        };
        let tools = vec![json!({
            "name":"bash","description":"run","input_schema":{"type":"object","properties":{}}
        })];
        let messages: Vec<crate::SharedMessage> = messages.into_iter().map(Arc::new).collect();
        let (tx, _rx) = mpsc::unbounded_channel::<StreamEvent>();
        let result = call_responses_route_inner(
            "muse",
            &cfg,
            &dyn_broker,
            &tools,
            &None,
            &messages,
            &tx,
            None,
            ReasoningLevel::Adaptive,
            &tokio_util::sync::CancellationToken::new(),
            0,
            &crate::runtime::trace::TraceContext::disabled(),
            true,
        )
        .await
        .expect("muse turn completes");
        let body = broker.seen.lock().unwrap()[0].body.clone().unwrap();
        (result, body)
    }

    const PREAMBLE_SSE: &str = concat!(
        "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":",
        "{\"type\":\"message\",\"role\":\"assistant\",\"content\":[]}}\n\n",
        "data: {\"type\":\"response.output_text.delta\",\"output_index\":0,",
        "\"delta\":\"Blocking question before I build anything:\"}\n\n",
        "data: {\"type\":\"response.completed\",\"response\":{\"usage\":",
        "{\"input_tokens\":12,\"output_tokens\":3}}}\n\n",
    );

    #[tokio::test]
    async fn muse_route_flags_dangling_preamble_outside_content() {
        let (result, _) =
            muse_turn(PREAMBLE_SSE, vec![json!({"role":"user","content":"go"})]).await;
        assert_eq!(result[DANGLING_PREAMBLE_KEY], "colon_preamble");
        // Stored content stays Anthropic-shaped and carries no sidecar/phase.
        assert_eq!(
            result["content"],
            json!([{"type":"text","text":"Blocking question before I build anything:"}])
        );
        // A response that acted is not flagged.
        let (result, _) = muse_turn(SSE, vec![json!({"role":"user","content":"go"})]).await;
        assert!(result.get(DANGLING_PREAMBLE_KEY).is_none(), "{result}");
    }

    #[tokio::test]
    async fn muse_request_replays_assistant_text_with_phase() {
        let (_, body) = muse_turn(
            SSE,
            vec![
                json!({"role":"user","content":"go"}),
                json!({"role":"assistant","content":[
                    {"type":"text","text":"Checking:"},
                    {"type":"tool_use","id":"call_a","name":"bash","input":{}},
                ]}),
                json!({"role":"user","content":[
                    {"type":"tool_result","tool_use_id":"call_a","content":"ok"},
                ]}),
                json!({"role":"assistant","content":[{"type":"text","text":"Done."}]}),
                json!({"role":"user","content":"next"}),
            ],
        )
        .await;
        let input = body["input"].as_array().unwrap();
        assert_eq!(input[1]["phase"], "commentary");
        assert_eq!(input[1]["content"][0]["text"], "Checking:");
        assert_eq!(input[2]["type"], "function_call");
        assert_eq!(input[4]["phase"], "final_answer");
        assert_eq!(input[4]["content"][0]["type"], "output_text");
    }
}