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::Notice(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)";

/// 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(
    label: &str,
    url: &str,
    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>> {
    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 retryable =
                    status.as_u16() == 408 || status.as_u16() == 429 || status.is_server_error();
                // 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.
                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>> {
    // 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"
    );
    // Every Codex credential, local or remote, crosses the broker boundary:
    // the broker vends an access token + expiry only (refresh tokens are
    // broker-owned), and this path never opens auth.json.
    let access = crate::runtime::api::await_or_cancel(
        cancel,
        broker.access_token(crate::auth::OAuthProviderId::OpenAiCodex),
    )
    .await
    .map_err(|_| "request canceled")?
    .map_err(|e| e.to_string())?
    .token;
    // Account id is provider-owned metadata carried inside the Codex JWT.
    let account_id = crate::auth::extract_codex_account_id(&access)
        .ok_or("Failed to extract ChatGPT account id from Codex token — run `synaps login --provider openai-codex`")?;

    let (tools, name_map) = translate::tools_to_responses(tools_schema);
    // 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 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.
    let 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);
    }
    let mut attempt = tr::StreamAttempt::new(tracer);
    let mut stream_retry = 0u32;

    loop {
        if cancel.is_cancelled() {
            return Err("request canceled".into());
        }
        let resp = send_with_retries(
            "codex",
            &url,
            || {
                client
                    .post(&url)
                    .bearer_auth(&access)
                    .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?;
        // 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_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,
        }));
    }
}

/// 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.";

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> {
    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.
        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,
    /// Trace (Task 10A): provider-reported usage from `response.completed`
    /// (input including cached, output, cached slice). `None` until observed.
    observed_usage: Option<(u64, u64, u64)>,
}

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.
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" => 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,
            observed_usage: None,
        }
    }

    /// 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)
    }

    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;
        }
        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;
                    }
                    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.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.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 {
                        "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"}),
            ]
        );
    }

    #[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)
    }

    #[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 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";

#[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() })?;
    // 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("xai-auth", "/responses", 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")?
        // 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| 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(XAI_PROVIDER_LABEL);
        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,
                    "xai 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,
        ));
        return Ok(json!({"role":"assistant","content":content}));
    }
}

/// 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![])
        }
    }

    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::Notice(_))) {
                    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"));
    }

    /// 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.
    #[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(
            "codex",
            &url,
            || 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");
    }

    #[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(
            "codex",
            &url,
            || 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
        );
    }
}