synaps-engine 0.11.0

use super::helpers::HelperMethods;
use super::sse_types::{AnthropicEvent, ContentBlock, Delta};
use super::types::{AuthState, LlmEvent, SessionEvent, StreamEvent};
use crate::runtime::telemetry::{self, TelemetryLevel};
use crate::{Result, RuntimeError, ToolRegistry};
#[cfg(test)]
use futures::StreamExt;
use reqwest::Client;
use serde_json::{json, Value};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{mpsc, RwLock};
use tokio_util::sync::CancellationToken;

/// Trace the outgoing Anthropic request. Metadata only — raw payload content
/// (message text, system prompt, tool schemas/arguments) must NEVER reach any
/// log sink at any level. Guarded by `outgoing_request_trace_is_metadata_only`.
fn trace_outgoing_request(model: &str, body_bytes: &[u8], message_count: usize, tool_count: usize) {
    // Per-process monotonic sequence: correlates this trace line with any
    // later log lines for the same request without embedding content.
    static REQUEST_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
    let correlation_id = REQUEST_SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    // Count prompt-cache markers by scanning the serialized bytes — reports
    // marker placement (a count) without deserializing or echoing content.
    const NEEDLE: &[u8] = b"\"cache_control\"";
    let cache_marker_count = body_bytes
        .windows(NEEDLE.len())
        .filter(|w| *w == NEEDLE)
        .count();
    tracing::trace!(
        provider = "anthropic",
        model = %model,
        payload_bytes = body_bytes.len(),
        message_count,
        tool_count,
        cache_marker_count,
        correlation_id,
        "Outgoing API request (metadata only)"
    );
}

/// Parse accumulated tool input JSON. On failure, returns a JSON object with
/// `__parse_error` key so the tool executor can report it back to the model.
fn parse_tool_input(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) }),
    }
}

/// All mutable state for one SSE stream parse. Mutated exclusively through
/// `process_event()` + `finalize()` — single write path makes duplicate-site
/// drift structurally impossible.
struct ParseState {
    // ── Output accumulation (stays Value — outgoing message format) ──
    accumulated_content: Vec<Value>,
    current_text: String,
    // ── Tool-use block accumulation ──
    current_tool_name: String,
    current_tool_id: String,
    current_tool_input_json: String,
    in_tool_use: bool,
    // ── Thinking block accumulation ──
    current_thinking: String,
    current_thinking_signature: String,
    in_thinking: bool,
    /// Set by `finalize()` when it flushed a `tool_use` whose input was still
    /// streaming (no `content_block_stop`). Always the LAST accumulated block.
    /// Only `take_outcome_content(cancelled = true)` acts on it.
    flushed_partial_tool: bool,
    // ── Telemetry captures ──
    telem_msg_id: Option<String>,
    telem_ttft: Option<u64>,
    telem_stop_reason: Option<String>,
    telem_usage: telemetry::UsageRecord,
    first_event_seen: bool,
    // ── Usage captured from message_start (single-emission protocol) ──
    // Live API shape (probed 2024+): message_start carries the full
    // `cache_creation` sub-object; message_delta carries ONLY the aggregate
    // (but repeats input/cache_read/cache_create and brings the final
    // output count). The start arm CAPTURES ONLY — the delta arm emits THE
    // single authoritative Usage event, falling back to these values for
    // anything the delta doesn't carry. Emitting from both arms double-counts
    // input/cache in every accumulating consumer (the pre-cache-ttl
    // double-billing bug).
    msg_start_input: Option<u64>,
    msg_start_output: Option<u64>,
    msg_start_cache_read: Option<u64>,
    msg_start_cache_create: Option<u64>,
    msg_start_cache_5m: Option<u64>,
    msg_start_cache_1h: Option<u64>,
    /// Set once the authoritative Usage event has been emitted. The residual
    /// path (stream died between start and delta) checks this so dead streams
    /// still bill what message_start reported — exactly one emission per
    /// request in all paths.
    usage_emitted: bool,
    /// Captured from an in-stream Anthropic `error` event. When set, the stream
    /// produced a failure (e.g. `overloaded_error`, `request_too_large`) the API
    /// delivered inside a 200 response. `classify_stream_outcome` turns this into
    /// either a retry (transient) or a terminal `Err` so the turn never silently
    /// ends with empty content (the "stopping" bug, task #130).
    stream_error: Option<StreamError>,
    /// True once a `message_delta` carrying a stop_reason was seen — captured
    /// UNCONDITIONALLY (independent of the telemetry flag) so
    /// `classify_stream_outcome` can tell a valid empty `end_turn` from a silent
    /// stop even in the default (telemetry-off) config.
    stop_reason_seen: bool,
    /// True when the captured stop_reason is exactly "refusal" — the model
    /// declined the request. Used by `classify_stream_outcome` to gate the
    /// refusal-retry path independently of the telemetry flag.
    stop_reason_is_refusal: bool,
    // ── Trace captures (Task 8) — unconditional, metadata-only ──
    /// Offset of the first parsed model event from `ctx.request_start`
    /// (post-headers), in ms. Feeds the trace `first_model_event` bucket.
    first_event_elapsed_ms: Option<u64>,
    /// Normalized stop reason for the trace record (raw wire string is
    /// mapped immediately, never retained).
    trace_stop_reason: Option<crate::runtime::trace::StopReason>,
    /// Provider-reported usage for the trace record; `None` until observed.
    trace_usage: Option<crate::runtime::trace::UsageMeta>,
    /// Cache-diagnosis verdict from message_start (only when this request
    /// sent `diagnostics`); `None` = no divergence / not requested.
    server_diagnosis: Option<crate::runtime::trace::ServerCacheDiagnosis>,
}

/// A failure surfaced as an in-stream Anthropic `error` event under a 200
/// response. `retryable` follows Anthropic's error taxonomy: transient classes
/// (`overloaded_error`, `api_error`, `rate_limit_error`) are retried with
/// backoff; everything else (`request_too_large`, `invalid_request_error`,
/// auth/permission) is terminal — retrying a context-overflow never helps.
#[derive(Debug, Clone)]
struct StreamError {
    /// Anthropic wire error type, retained separately from display text so
    /// retry policy never depends on parsing a human-facing message.
    error_type: Option<String>,
    /// Human-facing, actionable message. Built from vetted static text only —
    /// the provider's error message is untrusted (it can echo the request)
    /// and is never copied into this field.
    message: String,
    /// True for transient classes that a backoff retry can clear.
    retryable: bool,
}

impl StreamError {
    /// Anthropic error `type`s that a retry can plausibly clear. Mirrors the
    /// HTTP-status retry set (429 rate_limit, 500 api_error, 529 overloaded).
    fn is_retryable_type(error_type: &str) -> bool {
        matches!(
            error_type,
            "overloaded_error" | "api_error" | "rate_limit_error"
        )
    }
}

/// The decided fate of a finished stream — pure, unit-tested (task #130).
#[derive(Debug)]
enum StreamOutcome {
    /// A valid turn — hand the assistant content envelope back to the caller.
    Done(Value),
    /// A transient in-stream error — caller should back off and re-send.
    Retry(StreamError),
    /// A terminal failure — surface as a visible `Err`, never a silent stop.
    Fail(String),
    /// The model set stop_reason=refusal — discard all partial content and
    /// retry the identical request (up to the refusal_retries budget).
    Refusal,
}

impl ParseState {
    fn new() -> Self {
        Self {
            accumulated_content: Vec::new(),
            current_text: String::new(),
            current_tool_name: String::new(),
            current_tool_id: String::new(),
            current_tool_input_json: String::new(),
            in_tool_use: false,
            current_thinking: String::new(),
            current_thinking_signature: String::new(),
            in_thinking: false,
            flushed_partial_tool: false,
            telem_msg_id: None,
            telem_ttft: None,
            telem_stop_reason: None,
            telem_usage: telemetry::UsageRecord::default(),
            first_event_seen: false,
            msg_start_input: None,
            msg_start_output: None,
            msg_start_cache_read: None,
            msg_start_cache_create: None,
            msg_start_cache_5m: None,
            msg_start_cache_1h: None,
            usage_emitted: false,
            stream_error: None,
            stop_reason_seen: false,
            stop_reason_is_refusal: false,
            first_event_elapsed_ms: None,
            trace_stop_reason: None,
            trace_usage: None,
            server_diagnosis: None,
        }
    }

    /// End-of-stream flush of any partial thinking/tool/text block.
    /// Idempotent: clears `in_*` and `current_text` so a second call is a no-op.
    fn finalize(&mut self) {
        if self.in_thinking {
            // Never emit an empty `thinking` field — Anthropic rejects such
            // blocks on the next turn (see content_block_stop arm).
            if !self.current_thinking.is_empty() {
                self.accumulated_content.push(json!({
                    "type": "thinking",
                    "thinking": self.current_thinking,
                    "signature": self.current_thinking_signature
                }));
            }
            self.in_thinking = false;
        } else if self.in_tool_use {
            let input = parse_tool_input(&self.current_tool_input_json);
            self.accumulated_content.push(json!({
                "type": "tool_use",
                "id": self.current_tool_id,
                "name": self.current_tool_name,
                "input": input
            }));
            self.in_tool_use = false;
            self.flushed_partial_tool = true;
        } else if !self.current_text.is_empty() {
            self.accumulated_content.push(json!({
                "type": "text",
                "text": self.current_text
            }));
        }
        self.current_text.clear();
    }

    /// The content handed to `classify_stream_outcome`.
    ///
    /// Uncancelled streams: exactly `accumulated_content`, unchanged.
    ///
    /// Cancelled streams keep only what the model verifiably produced, so the
    /// interrupted assistant turn can stay in history as a REAL message
    /// (never a user-side recap — see `session::actor::cancel_turn`):
    /// - partial text is kept verbatim (it is what the model said);
    /// - thinking without a signature is dropped — a block cut off before its
    ///   `signature_delta` cannot be replayed to the provider;
    /// - a `tool_use` whose input was still streaming is dropped — the model
    ///   never finished making that call, and its truncated JSON would be
    ///   recorded as `{"__parse_error": …}`. Completed calls stay (the tool
    ///   loop pairs them with a canceled result).
    fn take_outcome_content(&mut self, cancelled: bool) -> Vec<Value> {
        let mut content = std::mem::take(&mut self.accumulated_content);
        if cancelled {
            if self.flushed_partial_tool
                && content.last().and_then(|b| b["type"].as_str()) == Some("tool_use")
            {
                content.pop();
            }
            content.retain(|b| {
                b["type"].as_str() != Some("thinking")
                    || b["signature"].as_str().is_some_and(|s| !s.is_empty())
            });
        }
        self.flushed_partial_tool = false;
        content
    }
}

/// Immutable per-stream context — not state. `tx` deliberately lives here and
/// not in `ParseState`: read/write separation is the point.
struct EventCtx<'t> {
    model: &'t str,
    tx: &'t mpsc::UnboundedSender<StreamEvent>,
    suppress_stream_deltas: bool,
    telemetry_level: TelemetryLevel,
    request_start: std::time::Instant,
    /// Requested cache TTL — used by the silent-downgrade detector.
    cache_ttl: crate::core::config::CacheTtl,
    /// Once-per-session latch for the downgrade notice (shared via Runtime).
    ttl_downgrade_notified: std::sync::Arc<std::sync::atomic::AtomicBool>,
    /// Session-scoped latch: set once any response shows a nonzero 1h cache
    /// write. A healthy Hybrid session writes the 1h prefix on turn 1 and
    /// then only the 5m tail on later turns — without this latch that
    /// signature is indistinguishable from a genuine downgrade (spec §3.4.1).
    saw_1h_honored: std::sync::Arc<std::sync::atomic::AtomicBool>,
    /// True iff at least one 1h cache marker was actually emitted in THIS
    /// request. Degenerate case (hybrid + API key + no system prompt + zero
    /// tools): no 1h marker exists anywhere in the payload, so a 1h bucket of
    /// zero proves nothing — the detector must stay disarmed.
    request_has_1h_marker: bool,
    /// Optional Task 23 turn-budget usage counters, recorded at the single
    /// authoritative Usage emission (and its residual twin) only.
    usage_counters: Option<std::sync::Arc<crate::runtime::budget::UsageCounters>>,
    /// This request sent `diagnostics`: capture the message id (chain) and
    /// the server's verdict from message_start.
    cache_diagnosis: bool,
}

/// THE TEST SEAM. Strips SSE framing, skips non-data lines and the `[DONE]`
/// marker, parses JSON, dispatches to `process_event`. Both the main loop and
/// the tail-flush path call this — one write path for every parse site.
fn process_data_line(line: &str, state: &mut ParseState, ctx: &EventCtx) {
    let Some(data_part) = line.strip_prefix("data: ") else {
        return;
    };
    if data_part.trim() == "[DONE]" {
        return;
    }
    let event = match serde_json::from_str::<AnthropicEvent>(data_part) {
        Ok(e) => e,
        Err(_) => return, // malformed JSON: skip the line, never panic
    };
    process_event(event, data_part, state, ctx);
}

/// Handle one parsed SSE event. The TTFT capture rides along so the main and
/// tail paths are uniform. `raw` is the un-parsed data line — `#[serde(other)]`
/// discards the tag on Unknown events, so the raw line is the only forensics.
fn process_event(event: AnthropicEvent<'_>, raw: &str, state: &mut ParseState, ctx: &EventCtx) {
    // First-event capture: the trace timing bucket is unconditional
    // (Task 8); the telemetry TTFT mirror stays gated as before.
    if !state.first_event_seen {
        state.first_event_seen = true;
        state.first_event_elapsed_ms = Some(ctx.request_start.elapsed().as_millis() as u64);
        if ctx.telemetry_level.enabled() {
            state.telem_ttft = state.first_event_elapsed_ms;
        }
    }

    match event {
        AnthropicEvent::ContentBlockStart { content_block } => match content_block {
            ContentBlock::Thinking => {
                state.current_thinking.clear();
                state.current_thinking_signature.clear();
                state.in_thinking = true;
            }
            ContentBlock::ToolUse { id, name } => {
                // Start accumulating a tool_use block
                state.current_tool_name = name.into_owned();
                state.current_tool_id = id.into_owned();
                state.current_tool_input_json.clear();
                state.in_tool_use = true;
                let _ = ctx.tx.send(StreamEvent::Llm(LlmEvent::ToolUseStart {
                    tool_name: state.current_tool_name.clone(),
                    tool_id: state.current_tool_id.clone(),
                }));
            }
            ContentBlock::Text => {
                if !state.current_text.is_empty() {
                    state.accumulated_content.push(json!({
                        "type": "text",
                        "text": state.current_text
                    }));
                    state.current_text.clear();
                }
            }
            // Unknown block type: no state change, mirrors the old `_ => {}`.
            ContentBlock::Unknown => {}
        },
        AnthropicEvent::ContentBlockDelta { delta } => match delta {
            Delta::TextDelta { text } => {
                state.current_text.push_str(&text);
                if !ctx.suppress_stream_deltas {
                    let _ = ctx
                        .tx
                        .send(StreamEvent::Llm(LlmEvent::Text(text.into_owned())));
                }
            }
            Delta::ThinkingDelta { thinking } => {
                // Anthropic sends thinking text in delta.thinking
                state.current_thinking.push_str(&thinking);
                if !ctx.suppress_stream_deltas {
                    let _ = ctx
                        .tx
                        .send(StreamEvent::Llm(LlmEvent::Thinking(thinking.into_owned())));
                }
            }
            Delta::SignatureDelta { signature } => {
                state.current_thinking_signature = signature.into_owned();
            }
            Delta::InputJsonDelta { partial_json } => {
                state.current_tool_input_json.push_str(&partial_json);
                if !ctx.suppress_stream_deltas {
                    let _ = ctx.tx.send(StreamEvent::Llm(LlmEvent::ToolUseDelta {
                        tool_id: state.current_tool_id.clone(),
                        delta: partial_json.into_owned(),
                    }));
                }
            }
            // Unknown delta subtype: no state change, mirrors the old `_ => {}`.
            Delta::Unknown => {}
        },
        AnthropicEvent::ContentBlockStop => {
            if state.in_thinking {
                // Flush thinking block with signature so it's echoed back in tool loops.
                // CRITICAL: never emit an empty `thinking` field — Anthropic rejects
                // such blocks on the next turn with
                // `messages.N.content.M.thinking: each thinking block must contain thinking`.
                // Empty blocks happen when the stream produced only a signature delta
                // (or none at all) before the block_stop arrived.
                if !state.current_thinking.is_empty() {
                    state.accumulated_content.push(json!({
                        "type": "thinking",
                        "thinking": state.current_thinking,
                        "signature": state.current_thinking_signature
                    }));
                }
                state.in_thinking = false;
            } else if state.in_tool_use {
                // Parse the accumulated JSON input
                let input = parse_tool_input(&state.current_tool_input_json);

                state.accumulated_content.push(json!({
                    "type": "tool_use",
                    "id": state.current_tool_id,
                    "name": state.current_tool_name,
                    "input": input
                }));

                // Emit the tool_use to the UI as soon as it's fully parsed,
                // so the call appears during the assistant's stream — before
                // we hand off to the tool executor. Without this the call
                // only becomes visible immediately prior to its result.
                let _ = ctx.tx.send(StreamEvent::Llm(LlmEvent::ToolUse {
                    tool_name: state.current_tool_name.clone(),
                    tool_id: state.current_tool_id.clone(),
                    input: input.clone(),
                }));

                state.in_tool_use = false;
            } else if !state.current_text.is_empty() {
                // Flush text block so ordering is preserved
                state.accumulated_content.push(json!({
                    "type": "text",
                    "text": state.current_text
                }));
                state.current_text.clear();
            }
        }
        AnthropicEvent::MessageDelta { delta, usage } => {
            // Capture stop_reason UNCONDITIONALLY — classify_stream_outcome relies
            // on it to tell a valid empty end_turn from a silent stop, and that
            // decision must NOT depend on whether telemetry is enabled (default off).
            if let Some(sr) = delta.and_then(|d| d.stop_reason) {
                state.stop_reason_seen = true;
                if sr.as_ref() == "refusal" {
                    state.stop_reason_is_refusal = true;
                }
                // Trace capture: map to the normalized enum immediately —
                // the raw wire string is never retained (Task 8).
                state.trace_stop_reason = Some(
                    crate::runtime::trace::anthropic::stop_reason_from_wire(sr.as_ref()),
                );
                if ctx.telemetry_level.enabled() {
                    state.telem_stop_reason = Some(sr.into_owned());
                }
            }
            if let Some(usage) = usage {
                // Effective values: the delta's own fields where nonzero,
                // falling back to the message_start capture. Live deltas
                // repeat all aggregates (and carry the final output count),
                // but be defensive — a sparse delta must not zero out what
                // the start already reported.
                let input_t = if usage.input_tokens > 0 {
                    usage.input_tokens
                } else {
                    state.msg_start_input.unwrap_or(0)
                };
                let output_t = if usage.output_tokens > 0 {
                    usage.output_tokens
                } else {
                    state.msg_start_output.unwrap_or(0)
                };
                let cache_read = if usage.cache_read_input_tokens > 0 {
                    usage.cache_read_input_tokens
                } else {
                    state.msg_start_cache_read.unwrap_or(0)
                };
                let cache_create = if usage.cache_creation_input_tokens > 0 {
                    usage.cache_creation_input_tokens
                } else {
                    state.msg_start_cache_create.unwrap_or(0)
                };
                // TTL breakdown: prefer the delta's own cache_creation
                // sub-object (future-proof), but in live traffic message_delta
                // carries ONLY the aggregate — the split arrives on
                // message_start. Fall back to the values captured there.
                let cache_create_5m = usage
                    .cache_creation
                    .as_ref()
                    .and_then(|cc| cc.ephemeral_5m_input_tokens)
                    .or(state.msg_start_cache_5m);
                let cache_create_1h = usage
                    .cache_creation
                    .as_ref()
                    .and_then(|cc| cc.ephemeral_1h_input_tokens)
                    .or(state.msg_start_cache_1h);

                // ═══ Silent-downgrade detector (spec §3.4.1) ═══
                // The failure mode that doesn't 400: the API accepts the
                // request but quietly honors only 5m. Fire ONE notice per
                // session and keep requesting what the user configured —
                // auto-downgrade would change pricing behavior behind the
                // user's back and mask the account-level problem.
                //
                // The saw_1h_honored latch prevents the Hybrid false
                // positive: turn 2+ of a healthy Hybrid session has
                // 1h == 0 (prefix cached) and 5m > 0 (tail rewrite) — the
                // exact downgrade signature. Turn 1's prefix write sets the
                // latch (in the message_start arm, so a stream that dies
                // before its delta still latches); a genuinely downgraded
                // account never does.
                //
                // cache_read == 0 guard: a fresh process against a still-warm
                // 1h prefix sees cache_read > 0, 1h == 0 (prefix CACHED, not
                // written), 5m > 0 (tail) — healthy, not a downgrade. A
                // genuinely downgraded account goes cold within 5m, so real
                // downgrades still fire on the first cold turn.
                //
                // request_has_1h_marker: if this request carried no 1h marker
                // at all, a zero 1h bucket proves nothing — stay silent.
                if cache_create_1h.unwrap_or(0) > 0 {
                    ctx.saw_1h_honored
                        .store(true, std::sync::atomic::Ordering::Relaxed);
                }
                if ctx.cache_ttl != crate::core::config::CacheTtl::FiveMinutes
                    && ctx.request_has_1h_marker
                    && cache_create_1h.unwrap_or(0) == 0
                    && cache_create_5m.unwrap_or(0) > 0
                    && cache_read == 0
                    && !ctx
                        .saw_1h_honored
                        .load(std::sync::atomic::Ordering::Relaxed)
                    && !ctx
                        .ttl_downgrade_notified
                        .swap(true, std::sync::atomic::Ordering::Relaxed)
                {
                    tracing::warn!(
                        "1h cache TTL not honored — check account/beta support (cache_ttl config)"
                    );
                    let _ = ctx.tx.send(StreamEvent::Session(SessionEvent::Notice(
                        "⚠ 1h cache TTL not honored — check account/beta support (cache_ttl config)".to_string(),
                    )));
                }

                if input_t > 0 || output_t > 0 || cache_read > 0 || cache_create > 0 {
                    HelperMethods::log_usage(input_t, cache_read, cache_create, output_t);
                    tracing::debug!(
                        "Token Usage: {} input | {} output | {} cache_read | {} cache_create",
                        input_t,
                        output_t,
                        cache_read,
                        cache_create
                    );
                    // ═══ TELEMETRY: accumulate usage (message_delta carries final counts) ═══
                    if ctx.telemetry_level.enabled() {
                        state.telem_usage.input = input_t;
                        state.telem_usage.output = output_t;
                        state.telem_usage.cache_read = cache_read;
                        state.telem_usage.cache_write = cache_create;
                        state.telem_usage.cache_write_5m = cache_create_5m;
                        state.telem_usage.cache_write_1h = cache_create_1h;
                        state.telem_usage.compute_hit_pct();
                    }
                    // THE single authoritative Usage emission for this
                    // request. The message_start arm captures only — every
                    // downstream consumer is an accumulator, so a second
                    // emission double-bills input/cache.
                    state.usage_emitted = true;
                    state.trace_usage = Some(crate::runtime::trace::UsageMeta {
                        provenance: crate::runtime::trace::UsageProvenance::ProviderReported,
                        input_tokens: Some(input_t),
                        output_tokens: Some(output_t),
                        cache_read_tokens: Some(cache_read),
                        cache_write_tokens: Some(cache_create),
                    });
                    if let Some(counters) = &ctx.usage_counters {
                        counters.record_priced(
                            ctx.model, input_t, output_t, cache_read, cache_create,
                            cache_create_5m, cache_create_1h,
                        );
                    }
                    let _ = ctx.tx.send(StreamEvent::Session(SessionEvent::Usage {
                        input_tokens: input_t,
                        output_tokens: output_t,
                        cache_read_input_tokens: cache_read,
                        cache_creation_input_tokens: cache_create,
                        cache_creation_5m: cache_create_5m,
                        cache_creation_1h: cache_create_1h,
                        model: None,
                    }));
                }
            }
        }
        AnthropicEvent::MessageStart { message } => {
            // ═══ TELEMETRY: capture msg_id (also the diagnostics chain) ═══
            if ctx.telemetry_level.enabled() || ctx.cache_diagnosis {
                if let Some(id) = message.id {
                    state.telem_msg_id = Some(id.into_owned());
                }
            }
            if ctx.cache_diagnosis {
                state.server_diagnosis =
                    super::cache_diagnosis::parse_server_diagnosis(message.diagnostics.as_ref());
            }
            if let Some(usage) = message.usage {
                // CAPTURE ONLY — no Usage emission from this arm. Live
                // message_delta repeats all aggregates and carries the final
                // output count, so emitting here too double-counts
                // input/cache_read/cache_create in every accumulating
                // consumer (TUI, engine, server, RPC, subagents). The delta
                // arm emits the single authoritative event; the residual
                // path covers streams that die before their delta.
                state.msg_start_input = Some(usage.input_tokens);
                state.msg_start_output = Some(usage.output_tokens);
                state.msg_start_cache_read = Some(usage.cache_read_input_tokens);
                state.msg_start_cache_create = Some(usage.cache_creation_input_tokens);
                state.msg_start_cache_5m = usage
                    .cache_creation
                    .as_ref()
                    .and_then(|cc| cc.ephemeral_5m_input_tokens);
                state.msg_start_cache_1h = usage
                    .cache_creation
                    .as_ref()
                    .and_then(|cc| cc.ephemeral_1h_input_tokens);
                // Latch 1h-honored HERE, not only in the delta arm: a stream
                // that dies between start and delta on turn 1 must not lose
                // the latch — that would arm a false downgrade notice on a
                // later healthy turn (1h == 0 because the prefix is cached).
                if state.msg_start_cache_1h.unwrap_or(0) > 0 {
                    ctx.saw_1h_honored
                        .store(true, std::sync::atomic::Ordering::Relaxed);
                }
            }
        }
        AnthropicEvent::MessageStop => {}
        AnthropicEvent::Error { error } => {
            // Anthropic delivers some failures (overloaded, context-overflow)
            // as an in-stream `error` event under a 200 response. Capture +
            // CLASSIFY it so the stream either retries (transient) or surfaces
            // a visible, accurate error — never the silent "stopping" swallow
            // (task #130). First error wins — later frames don't clobber it.
            if state.stream_error.is_none() {
                let (kind, _body) = match &error {
                    Some(e) => (
                        e.error_type.as_deref().unwrap_or("error"),
                        e.message.as_deref(),
                    ),
                    None => ("error", None),
                };
                let retryable = StreamError::is_retryable_type(kind);
                // SECURITY (spec §5.1): the frame — error type AND message —
                // is untrusted; a hostile provider can echo the entire request
                // through it. Surface only a vetted static class label; never
                // the message or the raw frame (bounded truncation is not
                // redaction).
                let vetted =
                    crate::core::error::sanitize_error_type(kind).unwrap_or("unrecognized_error");
                let message = format!(
                    "API stream error ({vetted}). Provider error details withheld — they can echo request content."
                );
                tracing::warn!(
                    retryable,
                    error_type = vetted,
                    frame_bytes = raw.len(),
                    "SSE error event (payload withheld — untrusted)"
                );
                state.stream_error = Some(StreamError {
                    error_type: Some(kind.to_string()),
                    message,
                    retryable,
                });
            }
        }
        AnthropicEvent::Unknown => {
            // #[serde(other)] discarded the tag. The raw line is untrusted
            // provider output — log only its size, never its content
            // (spec §5.1: no response-derived text in logs at any level).
            tracing::trace!(
                payload_bytes = raw.len(),
                "Unknown SSE event type (payload withheld — untrusted)"
            );
        }
    }
}

/// Residual Usage emission for dead streams: if message_start reported usage
/// but the stream terminated before message_delta could emit the single
/// authoritative event, bill what the start reported. Idempotent via
/// `usage_emitted` — exactly one emission per request in all paths.
fn emit_residual_usage(state: &mut ParseState, ctx: &EventCtx) {
    if state.usage_emitted {
        return;
    }
    let input_t = state.msg_start_input.unwrap_or(0);
    let output_t = state.msg_start_output.unwrap_or(0);
    let cache_read = state.msg_start_cache_read.unwrap_or(0);
    let cache_create = state.msg_start_cache_create.unwrap_or(0);
    if input_t == 0 && output_t == 0 && cache_read == 0 && cache_create == 0 {
        return; // never saw usage (or all-zero) — nothing to bill
    }
    HelperMethods::log_usage(input_t, cache_read, cache_create, output_t);
    tracing::debug!("Token Usage (residual, dead stream): {} input | {} output | {} cache_read | {} cache_create", input_t, output_t, cache_read, cache_create);
    if ctx.telemetry_level.enabled() {
        state.telem_usage.input = input_t;
        state.telem_usage.output = output_t;
        state.telem_usage.cache_read = cache_read;
        state.telem_usage.cache_write = cache_create;
        state.telem_usage.cache_write_5m = state.msg_start_cache_5m;
        state.telem_usage.cache_write_1h = state.msg_start_cache_1h;
        state.telem_usage.compute_hit_pct();
    }
    state.usage_emitted = true;
    state.trace_usage = Some(crate::runtime::trace::UsageMeta {
        provenance: crate::runtime::trace::UsageProvenance::ProviderReported,
        input_tokens: Some(input_t),
        output_tokens: Some(output_t),
        cache_read_tokens: Some(cache_read),
        cache_write_tokens: Some(cache_create),
    });
    if let Some(counters) = &ctx.usage_counters {
        counters.record_priced(
            ctx.model, input_t, output_t, cache_read, cache_create,
            state.msg_start_cache_5m, state.msg_start_cache_1h,
        );
    }
    let _ = ctx.tx.send(StreamEvent::Session(SessionEvent::Usage {
        input_tokens: input_t,
        output_tokens: output_t,
        cache_read_input_tokens: cache_read,
        cache_creation_input_tokens: cache_create,
        cache_creation_5m: state.msg_start_cache_5m,
        cache_creation_1h: state.msg_start_cache_1h,
        model: None,
    }));
}

/// Decide the fate of a finished stream. Pure (no I/O) so it is unit tested
/// directly — the silent-stop fix (task #130) lives here.
///
///   * An in-stream `error` event → `Retry` (transient class) or `Fail`
///     (terminal class, e.g. `request_too_large` context overflow). A user
///     cancellation downgrades any error to a clean (empty) `Done`.
///   * stop_reason=refusal → `Refusal`: ALL accumulated content discarded,
///     caller retries up to `refusal_retries` budget with a short delay.
///     Refusal on a cancelled stream downgrades to `Done` (don't fight cancel).
///   * A degenerate empty stream (no content AND no stop_reason) that was NOT
///     cancelled → `Fail`. An empty streamed response with no stop reason is
///     never valid; swallowing it as an empty assistant turn is exactly what
///     made the agent silently stop mid-flow.
///   * Otherwise → `Done` with the assistant content envelope. A cancelled
///     stream legitimately yields empty content, so it passes through.
const OAUTH_OVERLOAD_RETRIES: u32 = 10;
const OAUTH_OVERLOAD_BASE_DELAY_MS: u64 = 500;
const GENERIC_STREAM_BASE_DELAY_MS: u64 = 1000;

fn stream_retry_policy(
    auth_type: &str,
    error_type: Option<&str>,
    configured_retries: u32,
) -> (u32, u64) {
    if auth_type == "oauth" && error_type == Some("overloaded_error") {
        (OAUTH_OVERLOAD_RETRIES, OAUTH_OVERLOAD_BASE_DELAY_MS)
    } else {
        (configured_retries, GENERIC_STREAM_BASE_DELAY_MS)
    }
}

fn classify_stream_outcome(
    stream_error: Option<StreamError>,
    content: Vec<Value>,
    has_stop_reason: bool,
    stop_reason_is_refusal: bool,
    cancelled: bool,
) -> StreamOutcome {
    if let Some(e) = stream_error {
        if cancelled {
            // The user pulled the plug — don't surface a scary error or burn a
            // retry. Return whatever (likely empty) content we have.
            return StreamOutcome::Done(json!({ "content": content }));
        }
        return if e.retryable {
            StreamOutcome::Retry(e)
        } else {
            StreamOutcome::Fail(e.message)
        };
    }
    // Refusal: model ended the stream cleanly but declined the request.
    // DISCARD all partial content — nothing partial may enter history.
    if stop_reason_is_refusal && !cancelled {
        return StreamOutcome::Refusal;
    }
    if !cancelled && content.is_empty() && !has_stop_reason {
        return StreamOutcome::Fail(
            "empty response from API — the model returned no content and no stop \
             reason. This usually means the context window was exceeded or the \
             API was overloaded. Try /compact or start a fresh session."
                .to_string(),
        );
    }
    StreamOutcome::Done(json!({ "content": content }))
}

/// Extensible — new flags go here instead of adding parameters to 4 signatures.
#[derive(Debug, Clone, Default)]
pub struct ApiOptions {
    /// Immutable host binding for extension-provider interior tool execution.
    /// `None` is only for internal/test callers without a runtime capability.
    pub memory_backend: Option<crate::memory_backend::MemoryBinding>,
    /// Opt into the 1M context window beta header.
    pub use_1m_context: bool,
    /// Prompt-cache TTL strategy (spec: cache-ttl). Default `FiveMinutes`
    /// emits payloads byte-identical to the pre-feature release.
    pub cache_ttl: crate::core::config::CacheTtl,
    /// One-time-per-session latch for the silent-downgrade notice (1h
    /// requested, only 5m honored). Shared via Arc so every request in the
    /// session sees the same latch; the configured mode is NEVER auto-flipped.
    pub ttl_downgrade_notified: std::sync::Arc<std::sync::atomic::AtomicBool>,
    /// Session-scoped "1h was honored at least once" latch (spec §3.4.1).
    /// Set on any response with a nonzero 1h cache-write bucket; suppresses
    /// the downgrade notice on healthy Hybrid turns where the 1h prefix is
    /// already cached (1h == 0, 5m > 0). Shared via Arc like the notice latch.
    pub saw_1h_honored: std::sync::Arc<std::sync::atomic::AtomicBool>,
    /// Credential source for provider auth (Local/Remote). Carried here so the
    /// OpenAI/codex routing path can resolve tokens through the broker on
    /// Remote, same as the Anthropic path. (#158 C4)
    pub credential_source: crate::auth::CredentialSource,
    /// Shared broker token cache (Remote only).
    pub token_cache: crate::auth::TokenCache,
    /// Override the Anthropic API base URL. In production always `None`, which
    /// causes the code to read `SYNAPS_ANTHROPIC_BASE_URL` from the environment
    /// and then fall back to `"https://api.anthropic.com"`.
    /// Set by tests to point at a local mock server without touching env vars.
    #[doc(hidden)]
    pub anthropic_base_url: Option<String>,
    /// Dedicated lazy pool: upload and body idle deadlines are independent.
    pub(super) anthropic_stream_client: Arc<std::sync::OnceLock<Client>>,
    /// Execution authority produced by Runtime preflight. Transports must not
    /// recreate authority for Max/UltraCode from assumed prerequisites.
    pub anthropic_execution_plan: Option<crate::runtime::openai::catalog::AnthropicExecutionPlan>,
    /// Foreground/worker/internal request role.
    pub codex_request_role: crate::runtime::openai::catalog::CodexRequestRole,
    /// Trace emission seam (Task 8). Default is a disabled context (no-op
    /// sink, zero request-path work); tests install a collecting sink and
    /// production installs the Task 11 writer-backed sink via the Runtime.
    pub trace: crate::runtime::trace::TraceContext,
    /// The correlated IDs reserved for this provider request. Set before the
    /// transport starts and consumed by both RequestTracer and the stream's
    /// tool lifecycle recorder.
    pub request_correlation: Option<crate::runtime::trace::RequestCorrelation>,
    /// Internal sync routes can suppress high-volume display-only deltas at
    /// the producer boundary; final text/tool results are still returned in
    /// the normal response value.
    #[doc(hidden)]
    pub suppress_stream_deltas: bool,
    /// Bounded background writer for legacy telemetry records (Task 11).
    /// `None` (default) drops records; production sets the Runtime's shared
    /// session writer whenever the telemetry level is Basic/Full. Enqueue
    /// only — never synchronous file I/O on the request path.
    pub telemetry: Option<crate::runtime::telemetry::TelemetryWriter>,
    /// Runtime-scoped tool-session identity (Task 16). Scopes the execution
    /// gate for extension-provider interior tool loops reached through
    /// `try_route`. Stream turns (and `run_single`) thread the Runtime's
    /// host session; `None` (internal helpers, tests) makes the extension
    /// route fail closed to a locally minted identity with the identical
    /// default-core policy — never to ungated execution.
    pub tool_session_id: Option<crate::tools::activation::SessionId>,
    /// The RETAINED per-stream session tool set handle (Task 17). Stream
    /// turns thread the same shared set the execution gate authorizes
    /// against, so extension-provider interior tool loops observe the
    /// identical set/generation — including exact activations. `None`
    /// (internal helpers, non-stream paths, tests) makes the extension
    /// route fall back to a locally minted default-core set; a STALE
    /// retained set is denied there, never silently replaced.
    pub session_tool_set: Option<crate::tools::activation::SharedSessionToolSet>,
    /// Optional provider-neutral Task 18 schema projection for this request.
    /// `None` preserves the legacy registry-cached Arc and therefore the
    /// flag-off request bytes exactly. Stream rounds set `Some` only when the
    /// progressive-disclosure flag is enabled.
    pub request_tools_schema: Option<std::sync::Arc<Vec<Value>>>,
    /// Anthropic deferred-loading split for this request (progressive
    /// disclosure + `tools.deferred_loading`). `None` keeps the plain tools
    /// array. Only the native Anthropic transport reads it, and only for a
    /// model with tool search; every other transport keeps using
    /// `request_tools_schema`.
    pub anthropic_tool_split: Option<std::sync::Arc<crate::tools::DeferredToolSplit>>,
    /// Optional shared turn-budget usage counters (Task 23). Updated at the
    /// single authoritative Usage emission per request; the stream loop
    /// reads them for the optional token/cost budget dimensions. `None`
    /// (default, non-stream callers) records nothing.
    pub usage_counters: Option<std::sync::Arc<crate::runtime::budget::UsageCounters>>,
    /// "Last accepted Anthropic request" slot (task #444). Stamped with the
    /// send time of every attempt that gets a 2xx — the moment the prefix's
    /// cache entries were written or refreshed. `None` (default) records
    /// nothing; the foreground turn stream and keep-warm pings set it.
    pub cache_anchor: Option<std::sync::Arc<super::keepwarm::CacheAnchorCell>>,
    /// Prompt-cache keep-warm ping (task #444). `None` on every real turn,
    /// which keeps turn requests byte-identical to the pre-feature release.
    /// When set (native Anthropic transport only — any other route fails
    /// closed before a request is built): the conversational cache marker is
    /// stamped on the LAST HISTORY message and the trailing ping message
    /// rides after it unmarked, `max_tokens` is cut to the thinking mode's
    /// minimum, the prefix is fingerprinted (and checked, see
    /// [`KeepWarmMode`]), every retry budget is zero, and the telemetry
    /// record carries `"keepwarm": true`.
    pub keepwarm: Option<KeepWarmRequest>,
    /// Anthropic cache diagnostics chain (`cache_diagnostics = true`).
    /// `None` (default) keeps the wire byte-identical: no header, no
    /// `diagnostics` field. Set on the foreground turn stream (which threads
    /// `previous_message_id`) and on keep-warm pings (header only, so the
    /// beta set matches the turns they keep warm).
    pub cache_diagnosis: Option<super::cache_diagnosis::CacheDiagnosis>,
}

/// What a keep-warm request does once its body is built.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum KeepWarmMode {
    /// Return `{"keepwarm_prefix": <fingerprint>, "keepwarm_max_tokens": n}`;
    /// send nothing.
    Probe,
    /// Send only if the built prefix still hashes to `expect`; otherwise
    /// return `{"keepwarm_prefix": <actual>}` without sending.
    Send { expect: u64 },
}

/// Wire knobs for a keep-warm request (see [`ApiOptions::keepwarm`]).
#[derive(Debug, Clone)]
pub struct KeepWarmRequest {
    pub mode: KeepWarmMode,
    /// Last HTTP status the transport saw (0 = none), so the caller can tell
    /// a 4xx (stop) from a 5xx/429 (bounded retry).
    pub status: std::sync::Arc<std::sync::atomic::AtomicU16>,
    /// Set once the request is actually dispatched: when the HTTP client,
    /// over an established connection with the head written, pulls the body
    /// (see [`keepwarm_body`]), or when `send()` returns a response. From then
    /// on the ping may be billed even if it is aborted. A refused connection
    /// never sets it.
    pub sent: std::sync::Arc<std::sync::atomic::AtomicBool>,
    /// Set when `send()` failed to connect (no HTTP status, nothing sent).
    pub connect_failed: std::sync::Arc<std::sync::atomic::AtomicBool>,
}

/// A keep-warm request body that flips `sent` when the HTTP client first
/// pulls it: hyper only polls a request body once the connection is
/// established and the request head is written, so a ping that never
/// connected is never marked sent. `Content-Length` is set explicitly so the
/// wire framing matches a buffered body's (no chunked encoding).
fn keepwarm_body(
    req: reqwest::RequestBuilder,
    body: bytes::Bytes,
    sent: &std::sync::Arc<std::sync::atomic::AtomicBool>,
) -> reqwest::RequestBuilder {
    let flag = std::sync::Arc::clone(sent);
    let len = body.len();
    let stream = futures::stream::once(async move {
        flag.store(true, std::sync::atomic::Ordering::Relaxed);
        Ok::<bytes::Bytes, std::io::Error>(body)
    });
    req.header(reqwest::header::CONTENT_LENGTH, len)
        .body(reqwest::Body::wrap_stream(stream))
}

/// Validate a provider-assigned request ID from response headers into a
/// bounded [`trace::TraceId`]. Invalid or hostile values are omitted —
/// never copied raw into a trace record.
pub(super) fn provider_request_id_from_headers(
    headers: &reqwest::header::HeaderMap,
) -> Option<crate::runtime::trace::TraceId> {
    crate::runtime::telemetry::request_id_from_headers(headers)
        .and_then(|s| crate::runtime::trace::TraceId::new(s).ok())
}

/// Humanize a terminal (not retried) HTTP failure AND log it. Without the log
/// line the only record of a 400 is the TUI banner, so a session that fails
/// every request leaves nothing in `synaps.log` to diagnose from.
///
/// SECURITY (spec §5.1): `body` is untrusted and can echo the request. Only
/// the numeric status, the provider request id (validated `TraceId`) and OUR
/// humanized text, built from vetted static strings, are logged. Never the body.
pub(super) fn terminal_http_error(
    status: u16,
    body: &str,
    reset_hint: Option<&str>,
    request_id: Option<&crate::runtime::trace::TraceId>,
) -> String {
    let message = crate::core::error::humanize_api_error_with_reset(status, body, reset_hint);
    tracing::warn!(
        status,
        request_id = request_id.map_or("-", |id| id.as_str()),
        error = %message,
        "API request failed (not retried)"
    );
    message
}

/// Begin an Anthropic request tracer (Task 8). Returns `None` when tracing
/// is disabled or any structural identity is unrepresentable — tracing can
/// never fail the request. `body_bytes` MUST be the exact buffer handed to
/// reqwest: it is the sole source of the wire length + digest.
#[allow(clippy::too_many_arguments)]
pub(super) async fn begin_anthropic_tracer(
    options: &ApiOptions,
    url: &str,
    bare_model: &str,
    body_bytes: &[u8],
    messages: &[crate::SharedMessage],
    system_prompt: Option<&str>,
    tools_schema: &[Value],
    has_tool_marker: bool,
    has_system_marker: bool,
    translation: &crate::runtime::transport::report::TranslationReport,
) -> Option<crate::runtime::trace::RequestTracer> {
    use crate::runtime::trace as tr;
    if !options.trace.enabled() {
        return None;
    }
    let parsed = reqwest::Url::parse(url).ok()?;
    let host = parsed.host_str()?;
    let host = match parsed.port() {
        Some(port) => format!("{host}:{port}"),
        None => host.to_string(),
    };
    let endpoint = tr::EndpointMeta::new(host, parsed.path()).ok()?;
    let model =
        agent_core::prompt::QualifiedModelId::parse(format!("anthropic/{bare_model}")).ok()?;
    // Lazy, spawn_blocking-backed key load; `None` degrades the record to
    // digest-free sections and is counted, never surfaced as an error.
    let key = options.trace.digest_key().await;
    let prefix_ttl = match options.cache_ttl {
        crate::core::config::CacheTtl::FiveMinutes => Some(tr::CacheTtlClass::FiveMinutes),
        crate::core::config::CacheTtl::OneHour | crate::core::config::CacheTtl::Hybrid => {
            Some(tr::CacheTtlClass::OneHour)
        }
    };
    let structure = tr::anthropic::anthropic_request_structure(
        key.as_deref(),
        body_bytes,
        messages,
        system_prompt,
        tools_schema,
        prefix_ttl,
        has_tool_marker,
        has_system_marker,
        translation.clone().into_losses(),
        Some(options.trace.cache_snapshots()),
    );
    let tracer = tr::RequestTracer::begin(
        &options.trace,
        options.request_correlation.clone(),
        model,
        tr::TransportKind::AnthropicMessages,
        endpoint,
        structure,
    );
    // One-shot explicit content capture (`/trace next content`): a no-op in
    // every context that does not carry the arm. Body bytes only — headers
    // and credentials structurally never reach this seam.
    if let Some(tracer) = &tracer {
        options
            .trace
            .capture_request_content(tracer.request_id(), body_bytes);
    }
    tracer
}

/// Cancellation-first wait for request preparation, dispatch and retry delays.
/// The future is not polled when cancellation is already observable.
pub(super) async fn await_or_cancel<F: std::future::Future>(
    cancel: &CancellationToken,
    future: F,
) -> Result<F::Output> {
    tokio::select! {
        biased;
        _ = cancel.cancelled() => Err(RuntimeError::Canceled),
        output = future => Ok(output),
    }
}

pub(super) struct ApiMethods;

impl ApiMethods {
    #[allow(dead_code, clippy::too_many_arguments)]
    pub(super) async fn call_api_stream(
        auth: &Arc<RwLock<AuthState>>,
        client: &Client,
        model: &str,
        tools: &ToolRegistry,
        system_prompt: &Option<String>,
        thinking_budget: u32,
        reasoning_level: agent_core::reasoning::ReasoningLevel,
        messages: &[crate::SharedMessage],
        tx: mpsc::UnboundedSender<StreamEvent>,
        max_retries: u32,
        refusal_retries: u32,
        options: &ApiOptions,
        telemetry_level: crate::runtime::telemetry::TelemetryLevel,
    ) -> Result<Value> {
        Self::call_api_stream_inner(
            auth,
            client,
            model,
            tools,
            system_prompt,
            thinking_budget,
            reasoning_level,
            messages,
            tx,
            &CancellationToken::new(),
            max_retries,
            refusal_retries,
            options,
            telemetry_level,
        )
        .await
    }

    /// Static inner version — used by both `call_api_stream` (instance) and
    /// `run_stream_internal` (spawned task) so there's one implementation.
    #[allow(clippy::too_many_arguments)]
    #[allow(clippy::collapsible_match)]
    pub(super) async fn call_api_stream_inner(
        auth: &Arc<RwLock<AuthState>>,
        client: &Client,
        model: &str,
        tools: &ToolRegistry,
        system_prompt: &Option<String>,
        thinking_budget: u32,
        reasoning_level: agent_core::reasoning::ReasoningLevel,
        messages: &[crate::SharedMessage],
        tx: mpsc::UnboundedSender<StreamEvent>,
        cancel: &CancellationToken,
        max_retries: u32,
        refusal_retries: u32,
        options: &ApiOptions,
        telemetry_level: crate::runtime::telemetry::TelemetryLevel,
    ) -> Result<Value> {
        // Do not route, resolve credentials or prepare a request for a revoked turn.
        if cancel.is_cancelled() {
            return Err(RuntimeError::Canceled);
        }
        // Keep-warm pings exist only for the native Anthropic transport: every
        // other route would ignore the output cap and the marker placement.
        if options.keepwarm.is_some() && !super::auth::model_is_anthropic(model) {
            return Err(RuntimeError::Config(
                "cache keep-warm ping requires the native Anthropic transport".into(),
            ));
        }
        super::attachments::validate_history_for_request(model, messages).map_err(RuntimeError::Config)?;
        let wire_messages = super::continuation::wire_messages(messages);
        let messages = wire_messages.as_deref().unwrap_or(messages);
        // One provider-neutral schema source for every transport. The opt-in
        // progressive path supplies a per-round session projection; flag-off
        // keeps cloning the registry's existing cached Arc byte-for-byte.
        let tools_schema = options
            .request_tools_schema
            .clone()
            .unwrap_or_else(|| tools.tools_schema());
        // Cloud models always dispatch through the typed credential broker.
        // Text-only pre-flight, invocation, cancellation, and Task 10B trace
        // wiring live in `runtime::cloud_invoke` (extracted for testability).
        let (cloud_model, cloud_context) = crate::auth::cloud::split_model_route(model);
        if let Some((provider_key, _)) = cloud_model.split_once('/') {
            if let Ok(provider) = provider_key.parse::<crate::auth::CloudProviderId>() {
                if cancel.is_cancelled() {
                    return Err(RuntimeError::Canceled);
                }
                let has_tools = !tools_schema.is_empty();
                return with_usage_metering(model, options.usage_counters.clone(), &tx, |tx| async move {
                    crate::runtime::cloud_invoke::cloud_invoke_stream(
                        provider,
                        model,
                        cloud_model,
                        cloud_context,
                        has_tools,
                        || {
                            crate::auth::broker_from_source(
                                &options.credential_source,
                                &options.token_cache,
                                client.clone(),
                            )
                        },
                        system_prompt,
                        messages,
                        &tx,
                        cancel,
                        &options.trace,
                    )
                    .await
                })
                .await;
            }
        }
        // Route to OpenAI-compat provider if the model id resolves to one.
        if cancel.is_cancelled() {
            return Err(RuntimeError::Canceled);
        }
        let routed = {
            let tools_schema = &tools_schema;
            // Native Anthropic models fall through to the transport below,
            // which books its own usage.
            let counters = options
                .usage_counters
                .clone()
                .filter(|_| !super::auth::model_is_anthropic(model));
            with_usage_metering(model, counters, &tx, |tx| async move {
                crate::runtime::openai::try_route_with_memory_backend(
                    model,
                    client,
                    tools_schema,
                    system_prompt,
                    messages,
                    &tx,
                    None,
                    None,
                    thinking_budget,
                    reasoning_level,
                    cancel,
                    &options.credential_source,
                    &options.token_cache,
                    max_retries,
                    options.codex_request_role,
                    options.tool_session_id.as_ref(),
                    options.session_tool_set.as_ref(),
                    options.memory_backend.as_ref(),
                    &options.trace,
                    options.suppress_stream_deltas,
                )
                .await
            })
            .await
        };
        if let Some(result) = routed {
            // Classify honestly: transport blips are API failures, not
            // config problems (incident: session 20260714-025948-3dab).
            return result
                .map_err(|e| crate::runtime::openai::net::provider_error_to_runtime_for(model, e));
        }
        // Account failover (G3) crosses the broker boundary through the
        // router seam, exactly like the Codex transport: the broker vends one
        // (credential, access token) pair, never a refresh token, and this
        // path never opens auth.json. Built lazily — nothing is constructed
        // until a 429 proves window exhaustion.
        let router = super::anthropic_quota::BrokerAnthropicRouter::lazy(
            &options.credential_source,
            &options.token_cache,
            client.clone(),
        );
        Self::anthropic_stream_with_router(
            auth,
            client,
            model,
            tools_schema,
            system_prompt,
            thinking_budget,
            reasoning_level,
            messages,
            tx,
            cancel,
            max_retries,
            refusal_retries,
            options,
            telemetry_level,
            &router,
        )
        .await
    }

    /// The Anthropic Messages transport proper: request assembly, the
    /// unified retry loop and SSE parsing. Routing (cloud, OpenAI-compatible
    /// providers) has already happened in [`Self::call_api_stream_inner`];
    /// `messages` is the wire projection and `tools_schema` the resolved
    /// schema source. `router` is the account-failover seam (production:
    /// [`super::anthropic_quota::BrokerAnthropicRouter`]; tests script it).
    #[allow(clippy::too_many_arguments)]
    #[allow(clippy::collapsible_match)]
    pub(super) async fn anthropic_stream_with_router(
        auth: &Arc<RwLock<AuthState>>,
        client: &Client,
        model: &str,
        tools_schema: Arc<Vec<Value>>,
        system_prompt: &Option<String>,
        thinking_budget: u32,
        reasoning_level: agent_core::reasoning::ReasoningLevel,
        messages: &[crate::SharedMessage],
        tx: mpsc::UnboundedSender<StreamEvent>,
        cancel: &CancellationToken,
        max_retries: u32,
        refusal_retries: u32,
        options: &ApiOptions,
        telemetry_level: crate::runtime::telemetry::TelemetryLevel,
        router: &dyn super::anthropic_quota::AnthropicAccountRouter,
    ) -> Result<Value> {
        use super::anthropic_quota::{
            body_prefix, classify_anthropic_429, describe_reset_utc, failover_permitted,
            ratelimit_header_summary, Anthropic429, AnthropicFailoverGate, QUOTA_PROBE_BODY_CAP,
        };
        // Provider qualification is application identity; Anthropic's wire API
        // still receives its native bare model id.
        let qualified_model = model;
        let execution_plan = options
            .anthropic_execution_plan
            .as_ref()
            .filter(|plan| {
                plan.qualified_model == qualified_model
                    && plan.requested_level == reasoning_level
                    && plan.role == options.codex_request_role
            })
            .cloned()
            .or_else(|| {
                crate::runtime::openai::catalog::plan_standard_anthropic_transport(
                    qualified_model,
                    reasoning_level,
                    options.codex_request_role,
                )
            })
            .ok_or_else(|| {
                RuntimeError::Config(
                    "Anthropic special mode transport requires an authorized execution plan".into(),
                )
            })?;
        let model = model.strip_prefix("anthropic/").unwrap_or(model);

        let streaming_client = super::http_phases::client(&options.anthropic_stream_client)?;

        // Read auth state for this API call
        let (mut auth_header_name, mut auth_header_value, auth_type, mut auth_binding) =
            await_or_cancel(cancel, Self::build_auth_snapshot(auth)).await?;
        // One credential recovery per logical request (local OAuth or remote).
        let mut auth_retried = false;

        // Fail early with a clear message if no Anthropic credentials
        if auth_type == "none" {
            return Err(RuntimeError::Auth(
                "No Anthropic credentials. Run `synaps login`, or switch to a provider model with `/model groq/llama-3.3-70b-versatile`.".to_string()
            ));
        }

        tracing::info!(model = %model, "Starting API request");

        // Manual cache breakpoints for optimal prompt caching.
        // Tested vs auto-cache (top-level cache_control) — manual wins: 90% vs 53% hit rate.
        // Arc-shared (#128): to_vec() on &[SharedMessage] is N refcount bumps,
        // not a deep copy; sanitize/annotate clone-on-write only what they edit.
        let mut cleaned_messages = messages.to_vec();
        let retired_count = super::image_retention::plan(messages).map_err(RuntimeError::Config)?;
        // Deferred tool loading: send the session's loaded tools normally and
        // every other activatable tool with `defer_loading`, and resolve the
        // history's `tool_reference` blocks to wire names. Without a split
        // (or for a model without tool search) references are stripped, so
        // they never reach a request that cannot accept them.
        let split = options.anthropic_tool_split.as_deref().filter(|split| {
            !split.loaded.is_empty()
                && !split.deferred.is_empty()
                && crate::core::models::model_supports_tool_search(model)
        });
        let (wire_tools, deferred_tools): (&[Value], &[Value]) = match split {
            Some(split) => (&split.loaded, &split.deferred),
            None => (&tools_schema, &[]),
        };
        // Stable per-image projection precedes cache markers, only on Anthropic.
        // Original preflight above remains authoritative; recheck output budgets.
        await_or_cancel(
            cancel,
            super::image_limits::enforce_image_limits(&mut cleaned_messages, retired_count),
        )
        .await?
        .map_err(RuntimeError::Config)?;
        super::attachments::validate_messages(model, &cleaned_messages)
            .map_err(RuntimeError::Config)?;
        // Strict checks retain original per-source batch boundaries: sanitizing
        // first could merge valid adjacent user batches past eight attachments.
        HelperMethods::sanitize_thinking_blocks(&mut cleaned_messages);
        HelperMethods::wire_tool_references(&mut cleaned_messages, &|id: &str| {
            split.and_then(|split| split.id_to_api.get(id).cloned())
        });
        match &options.keepwarm {
            // Keep-warm: the marker goes on the last HISTORY message — the
            // durable prefix the next real turn re-sends verbatim — and the
            // trailing ping message (never persisted) rides after it as an
            // uncached tail, exactly like a real turn's newest user message.
            Some(_) if cleaned_messages.len() >= 2 => {
                let history_len = cleaned_messages.len() - 1;
                HelperMethods::annotate_cache_breakpoint(
                    &mut cleaned_messages[..history_len],
                    options.cache_ttl,
                );
            }
            _ => HelperMethods::annotate_cache_breakpoint(&mut cleaned_messages, options.cache_ttl),
        }

        // Derive the thinking level from the budget for effort mapping.
        // Body assembly (#128 Slice 4): `RequestBody` BORROWS cleaned_messages
        // and the tool schemas instead of the old `json!` assembly that deep-
        // rebuilt the entire history into a `Value` tree (copy C8). Byte-
        // identical output is enforced by `runtime::body_golden`. Task 9: the
        // transport adapter wraps that same serializer and adds the
        // provider-neutral `TranslationReport` (no Value round-trip, no
        // second history copy).
        let parts = crate::runtime::transport::anthropic::build_anthropic_request(
            model,
            &cleaned_messages,
            wire_tools,
            deferred_tools,
            system_prompt,
            &auth_type,
            thinking_budget,
            reasoning_level,
            Some(&execution_plan),
            options.cache_ttl,
            true,
        );
        let mut body = parts.body;
        // Cache diagnostics (`cache_diagnostics = true`): the foreground
        // chain only — pings and the setting-off path send no field.
        let mut diagnostics = super::cache_diagnosis::diagnostics_for(options);
        if let Some(d) = &diagnostics {
            body.set_diagnostics(d.clone());
        }
        // Keep-warm: cut the output cap (not part of the cache key), then
        // fingerprint the prefix exactly as it would go on the wire. A probe —
        // or a ping whose prefix no longer matches what is cached — returns
        // here, before any trace, telemetry or network I/O.
        if let Some(keepwarm) = &options.keepwarm {
            body.apply_keepwarm_max_tokens();
            let beta = Self::build_beta_header(&auth_type, options, model);
            let actual = body
                .prefix_fingerprint(cleaned_messages.len().saturating_sub(1), beta.as_deref());
            let send = matches!(keepwarm.mode, KeepWarmMode::Send { expect } if expect == actual);
            if !send {
                return Ok(serde_json::json!({
                    "keepwarm_prefix": actual,
                    "keepwarm_max_tokens": body.max_tokens(),
                }));
            }
        }
        let translation_report = parts.report;

        // Did THIS request actually carry at least one 1h marker? Arms the
        // silent-downgrade detector. Mirrors cache_control_value(): OneHour
        // puts 1h on every site (message tail, tools, system); Hybrid puts 1h
        // only on stable-prefix sites (tools, system). Degenerate hybrid +
        // API-key + no system prompt + zero tools → no 1h marker anywhere →
        // detector must stay disarmed (zero 1h bucket proves nothing).
        let has_tool_marker = body.has_tool_marker();
        let has_system_marker = body.has_system_marker();
        let request_has_1h_marker = match options.cache_ttl {
            crate::core::config::CacheTtl::FiveMinutes => false,
            crate::core::config::CacheTtl::OneHour => {
                !cleaned_messages.is_empty() || has_tool_marker || has_system_marker
            }
            crate::core::config::CacheTtl::Hybrid => has_tool_marker || has_system_marker,
        };

        // Serialize the body once up-front; each retry attempt reuses the same
        // bytes via a cheap `Bytes::clone()` (refcount bump, no copy). Previously
        // we called `req.json(&body).send()` per attempt, which re-ran
        // `serde_json::to_vec(&body)` over the entire conversation on every
        // 429 retry — wasted work since the payload is identical.
        let mut body_bytes: bytes::Bytes = serde_json::to_vec(&body)
            .map_err(|e| {
                RuntimeError::ApiStatus(format!("failed to serialize request body: {}", e))
            })?
            .into();
        // The derived request body is dead weight for the whole network wait
        // (retries reuse `body_bytes`): release it — and its owned `system`
        // block copy — before the first send (§3.6 f3).
        drop(body);

        trace_outgoing_request(
            model,
            &body_bytes,
            cleaned_messages.len(),
            tools_schema.len(),
        );

        // ═══ UNIFIED RETRY (task #130) ════════════════════════════════════════
        // ONE generic budget governs transient failures whether they surface as
        // an HTTP status (5xx, pre-stream), a transport drop, or an in-stream
        // `api_error` / `rate_limit_error` under a 200 response. Sharing this
        // counter across send and stream phases stops the budget from multiplying
        // (the two-loop smell). The two service-specific exceptions below are
        // 429 rate limits and Anthropic OAuth overload frames.
        //
        // 429 (rate-limit) keeps its own higher budget: OAuth windows can last
        // minutes, so we honour the reset headers rather than dying after a few
        // tries. Other transients (HTTP 5xx + in-stream api/rate-limit errors)
        // share `max_retries`. Anthropic OAuth `overloaded_error` frames use a
        // separate Claude-compatible persistent budget (10 retries), because
        // these 200/SSE overload responses carry no retry headers.
        const MAX_429_RETRIES: u32 = 8;
        // Endpoint resolved once — identical across retry attempts.
        let anthropic_base = options
            .anthropic_base_url
            .clone()
            .or_else(|| std::env::var("SYNAPS_ANTHROPIC_BASE_URL").ok())
            .unwrap_or_else(|| "https://api.anthropic.com".into());
        let anthropic_url = format!("{}/v1/messages", anthropic_base.trim_end_matches('/'));

        // ═══ TRACE (Task 8): one record per actual attempt ════════════════════
        // Emission rule documented in `trace::emit`. Tracing is metadata-only
        // and can never fail or delay the request.
        let mut tracer = await_or_cancel(
            cancel,
            begin_anthropic_tracer(
                options,
                &anthropic_url,
                model,
                &body_bytes,
                &cleaned_messages,
                system_prompt.as_deref(),
                &tools_schema,
                has_tool_marker,
                has_system_marker,
                &translation_report,
            ),
        )
        .await?;
        // Placeholder value: every send re-starts the clock immediately
        // before handing bytes to reqwest (per-attempt timing).
        #[allow(unused_assignments)]
        let mut clock = crate::runtime::trace::AttemptClock::start();

        let mut last_err = String::new();
        let mut last_status: Option<u16> = None;
        let mut last_reset_hint: Option<String> = None;
        let mut non_429_attempts: u32 = 0; // shared server-error budget
        let mut overload_attempts: u32 = 0; // Claude-compatible persistent overload budget
        let mut refusal_attempts: u32 = 0; // refusal-retry budget (separate from error budget)
        let mut attempt: u32 = 0; // total attempts (for backoff calc)
        // ═══ ACCOUNT FAILOVER (G3) request-level state ════════════════════════
        // `output_started`: any SSE frame was parsed on an earlier attempt of
        // this request — the strict, fail-closed reading of "nothing has been
        // streamed" that the hop requires (no cross-account replay).
        // `tool_activity`: an earlier attempt produced a tool_use block.
        // `anthropic_failovers`: hops spent on this request (≤ 1).
        // `quota_headers_logged`: the one-time schema-capture warn fired.
        let mut output_started = false;
        let mut tool_activity = false;
        let mut anthropic_failovers: u32 = 0;
        let mut quota_headers_logged = false;

        loop {
            let response = {
                #[allow(unused_assignments)]
                let mut response = None;

                loop {
                    if attempt > 0 {
                        // Sleep was already computed and stored in last_err context;
                        // delay is recomputed here from the empty header map if we
                        // came from a network error path (no headers available).
                        // For header-aware delays we sleep in the error arm below.
                        // Nothing to do here — sleep already happened.
                    }

                    // Rebuild request (consumed on send)
                    let mut req = streaming_client
                        .post(&anthropic_url)
                        .header(auth_header_name.clone(), auth_header_value.clone())
                        .header("anthropic-version", "2023-06-01")
                        .header("content-type", "application/json");
                    // Build the anthropic-beta header. The 1M-context opt-in
                    // (`context-1m-2025-08-07`) is only added when the user
                    // explicitly requested 1M AND the model supports it. Without
                    // this opt-in, all models default to 200k mode — which is the
                    // documented "smarter" inference regime (see
                    // anthropic.com/engineering/effective-context-engineering).
                    if let Some(beta) = Self::build_beta_header(&auth_type, options, model) {
                        req = req.header("anthropic-beta", beta);
                    }

                    // Every attempt (including 401/refusal retries) rechecks at
                    // the dispatch boundary, not merely at the round's start.
                    if cancel.is_cancelled() {
                        return Err(RuntimeError::Canceled);
                    }
                    clock = crate::runtime::trace::AttemptClock::start();
                    let sent_at = super::keepwarm::CacheAnchor::now();
                    let req = match &options.keepwarm {
                        Some(keepwarm) => keepwarm_body(req, body_bytes.clone(), &keepwarm.sent),
                        None => req.body(body_bytes.clone()),
                    };
                    match await_or_cancel(
                        cancel,
                        super::http_phases::send(
                            req,
                            super::http_phases::header_deadline(body_bytes.len()),
                        ),
                    )
                    .await?
                    {
                        Ok(resp) => {
                            let status = resp.status();
                            if let Some(keepwarm) = &options.keepwarm {
                                keepwarm
                                    .sent
                                    .store(true, std::sync::atomic::Ordering::Relaxed);
                                keepwarm
                                    .status
                                    .store(status.as_u16(), std::sync::atomic::Ordering::Relaxed);
                            }
                            if status.is_success() {
                                clock.mark_headers();
                                // Accepted: the prefix's cache entries were
                                // read or written by a request that started
                                // no earlier than `sent_at`.
                                if let Some(anchor) = &options.cache_anchor {
                                    anchor.stamp(sent_at);
                                }
                                response = Some(resp);
                                break;
                            }

                            // C1: a 401 means the token we sent was rejected —
                            // revoked, or rotated elsewhere (broker, another
                            // `synaps login`, refresh cron, synced auth.json)
                            // before our in-memory expiry said so. Status-level,
                            // so nothing has streamed yet. Invalidate, re-vend,
                            // and retry ONCE. Local API-key mode is excluded:
                            // there is no managed credential to reload.
                            if status.as_u16() == 401
                                && !auth_retried
                                && (options.credential_source.is_remote() || auth_type == "oauth")
                            {
                                auth_retried = true;
                                clock.mark_headers();
                                if let Some(t) = tracer.as_mut() {
                                    let rid = provider_request_id_from_headers(resp.headers());
                                    t.attempt_failed(
                                        clock,
                                        crate::runtime::trace::RetryClass::Auth,
                                        Duration::ZERO,
                                        Some(401),
                                        rid,
                                        "http_401",
                                    );
                                }
                                // Drop the rejected response: no unbounded body drain.
                                drop(resp);
                                if !options.credential_source.is_remote() {
                                    let rejected =
                                        auth_header_value.strip_prefix("Bearer ").unwrap_or("");
                                    await_or_cancel(
                                        cancel,
                                        super::auth::AuthMethods::recover_local_rejection(
                                            auth,
                                            client,
                                            &options.credential_source,
                                            &options.token_cache,
                                            auth_binding.as_deref(),
                                            rejected,
                                        ),
                                    )
                                    .await??;
                                } else {
                                    options.token_cache.invalidate("anthropic");
                                    {
                                        let mut g = await_or_cancel(cancel, auth.write()).await?;
                                        g.auth_token.clear();
                                        g.token_expires = None;
                                    }
                                    await_or_cancel(
                                        cancel,
                                        super::auth::AuthMethods::refresh_if_needed(
                                            std::sync::Arc::clone(auth),
                                            client,
                                            &options.credential_source,
                                            &options.token_cache,
                                            Some(model),
                                            super::auth::RefreshPoint::WithinTurn,
                                        ),
                                    )
                                    .await??;
                                }
                                let (n, v, _t, binding) =
                                    await_or_cancel(cancel, Self::build_auth_snapshot(auth))
                                        .await?;
                                auth_header_name = n;
                                auth_header_value = v;
                                auth_binding = binding;
                                continue;
                            }

                            let is_429 = status.as_u16() == 429;
                            let is_retryable =
                                matches!(status.as_u16(), 429 | 500 | 502 | 503 | 529);

                            // Capture headers before consuming the body.
                            let (delay, from_hdr) =
                                telemetry::retry_delay_from_headers(resp.headers(), attempt + 1);
                            let reset_hint = if from_hdr {
                                Some(format!("{}s", delay.as_secs()))
                            } else {
                                None
                            };
                            // Headers outlive the body read only for the OAuth
                            // 429 classifier (G3); api-key auth has no seats.
                            let quota_headers = if is_429 && auth_type == "oauth" {
                                Some(resp.headers().clone())
                            } else {
                                None
                            };

                            clock.mark_headers();
                            let trace_rid = provider_request_id_from_headers(resp.headers());
                            // Error bodies may echo prompts or be unbounded.
                            // Retain only the quota probe cap, even before
                            // classification/humanization; never buffer the
                            // full upstream response just to take a prefix.
                            let error_text = await_or_cancel(cancel, async {
                                let mut stream = resp.bytes_stream();
                                let mut bytes = Vec::new();
                                while let Some(Ok(chunk)) = super::http_phases::next(
                                    &mut stream,
                                    super::http_phases::STREAM_IDLE,
                                )
                                .await
                                {
                                    let remaining = QUOTA_PROBE_BODY_CAP - bytes.len();
                                    bytes.extend_from_slice(&chunk[..chunk.len().min(remaining)]);
                                    if bytes.len() == QUOTA_PROBE_BODY_CAP {
                                        break;
                                    }
                                }
                                String::from_utf8_lossy(&bytes).into_owned()
                            })
                            .await?;

                            // ═══ ACCOUNT FAILOVER (G3): window exhaustion on a 429 ═══
                            // Recognized ONLY from provider-declared signals (see
                            // `anthropic_quota::classify_anthropic_429`); anything
                            // else keeps the retry path below untouched. The body
                            // is consulted as a bounded prefix and never logged.
                            if let Some(headers) = quota_headers.as_ref() {
                                if !quota_headers_logged {
                                    quota_headers_logged = true;
                                    tracing::warn!(
                                        headers = %ratelimit_header_summary(headers),
                                        "anthropic 429 headers (schema capture)"
                                    );
                                }
                                let now_ms = crate::epoch_millis();
                                if let Anthropic429::QuotaExhausted(evidence) = classify_anthropic_429(
                                    headers,
                                    body_prefix(&error_text, QUOTA_PROBE_BODY_CAP),
                                    now_ms,
                                ) {
                                    // The seat this request is pinned to on THIS
                                    // source. `None` = legacy/unbound state: nothing
                                    // to report or switch, but the long-reset rule
                                    // below still applies.
                                    let seat = {
                                        let g = await_or_cancel(cancel, auth.read()).await?;
                                        super::auth::turn_pinned_credential(
                                            &options.credential_source,
                                            g.bound_credential.as_deref(),
                                        )
                                    };
                                    let auto = router.auto_selected();
                                    let until_ms = evidence.cooldown_until_ms(now_ms).unwrap_or(now_ms);
                                    let until = describe_reset_utc(until_ms, now_ms);
                                    let label = seat
                                        .as_ref()
                                        .map(|s| s.account.label_str().to_string())
                                        .unwrap_or_else(|| "anthropic".to_string());
                                    // ALWAYS report first — even when the gate then
                                    // refuses to switch — so the next turn boundary's
                                    // AutoRepin selects away from this seat.
                                    match seat.as_ref() {
                                        Some(seat) => {
                                            await_or_cancel(cancel, router.report_exhausted(seat, &evidence))
                                                .await?;
                                        }
                                        None => tracing::warn!(
                                            signal = ?evidence.signal,
                                            "anthropic window exhausted but no seat is bound; cooldown not reported"
                                        ),
                                    }
                                    let gate = AnthropicFailoverGate {
                                        auto_selector: auto,
                                        output_started,
                                        tool_activity,
                                        failovers_so_far: anthropic_failovers,
                                    };
                                    let mut hop = None;
                                    if let Some(seat) = seat.as_ref() {
                                        match failover_permitted(&gate) {
                                            Ok(()) => {
                                                match await_or_cancel(cancel, router.failover(model, seat))
                                                    .await?
                                                {
                                                    Ok(Some(next)) if next.credential != *seat => {
                                                        hop = Some(next);
                                                    }
                                                    Ok(_) => tracing::info!(
                                                        account = %seat,
                                                        "anthropic window exhausted; no distinct account with capacity"
                                                    ),
                                                    Err(e) => tracing::warn!(
                                                        account = %seat,
                                                        error = %e,
                                                        "anthropic failover candidate resolution failed; not switching"
                                                    ),
                                                }
                                            }
                                            Err(reason) => tracing::info!(
                                                account = %seat,
                                                reason,
                                                "anthropic window exhausted; account failover not permitted"
                                            ),
                                        }
                                    }
                                    if let Some(next) = hop {
                                        let next_label = next.credential.account.label_str().to_string();
                                        tracing::warn!(
                                            from = %label,
                                            to = %next_label,
                                            "anthropic window exhausted; switching account (one-time failover, no output replayed)"
                                        );
                                        if let Some(t) = tracer.as_mut() {
                                            t.attempt_failed(
                                                clock,
                                                crate::runtime::trace::RetryClass::RateLimited,
                                                Duration::ZERO,
                                                Some(429),
                                                trace_rid.clone(),
                                                "account_failover",
                                            );
                                        }
                                        // The ONE sanctioned mid-turn re-pin: bind the
                                        // new seat exactly as the refresh path does, then
                                        // rebuild the header from it (same as the 401
                                        // path). The failed seat's token is still valid —
                                        // only its capacity is spent — so no cache
                                        // invalidation is needed.
                                        await_or_cancel(
                                            cancel,
                                            super::auth::AuthMethods::bind_pinned(
                                                auth,
                                                &options.credential_source,
                                                next,
                                            ),
                                        )
                                        .await?;
                                        let (n, v, _t, binding) = await_or_cancel(
                                            cancel,
                                            Self::build_auth_snapshot(auth),
                                        )
                                        .await?;
                                        auth_header_name = n;
                                        auth_header_value = v;
                                        auth_binding = binding;
                                        anthropic_failovers += 1;
                                        let _ = tx.send(StreamEvent::Session(SessionEvent::Notice(format!(
                                            "⚠ {label} exhausted until {until} — switching to {next_label}"
                                        ))));
                                        // Retry immediately on the new seat WITHOUT
                                        // consuming a 429 retry slot.
                                        continue;
                                    }
                                    // No hop. Say why the wait is what it is, then either
                                    // end the turn now (a reset hours away must not burn
                                    // the whole 429 budget a minute at a time) or keep
                                    // the existing retry path for short/unknown resets.
                                    let notice = if auto {
                                        format!("⚠ {label} exhausted until {until}; next turn will use another account")
                                    } else {
                                        format!("⚠ account limit reached until {until}")
                                    };
                                    let _ = tx.send(StreamEvent::Session(SessionEvent::Notice(notice)));
                                    if evidence.reset_is_long(now_ms) {
                                        if let Some(t) = tracer.take() {
                                            let terminal = t.failed_terminal("quota_exhausted");
                                            t.finish(clock, Some(429), trace_rid, None, None, terminal);
                                        }
                                        let advice = if auto {
                                            "The next turn will select another account."
                                        } else {
                                            "Wait for the reset, or switch accounts with `synaps auth use --provider anthropic --account <label|auto>`."
                                        };
                                        return Err(RuntimeError::ApiStatus(format!(
                                            "Anthropic usage limit reached on account '{label}' — window resets at {until}. {advice}"
                                        )));
                                    }
                                }
                            }

                            // ═══ CACHE DIAGNOSTICS fail-soft ═══
                            // The API refused the beta header or the
                            // `diagnostics` field: latch it off for the session
                            // and re-send this request without either (header
                            // is rebuilt from the latch). Costs no retry budget;
                            // a turn never fails because of diagnostics. The
                            // trace keeps the first attempt's wire metadata.
                            let diag_on_wire = diagnostics.is_some()
                                || super::cache_diagnosis::wants_beta_header(&auth_type, options);
                            if diag_on_wire
                                && super::cache_diagnosis::is_rejection(
                                    status.as_u16(),
                                    &error_text,
                                )
                            {
                                if let Some(chain) = &options.cache_diagnosis {
                                    chain.reject(status.as_u16());
                                }
                                if let Some(d) = diagnostics.take() {
                                    match super::cache_diagnosis::strip_diagnostics(&body_bytes, &d)
                                    {
                                        Some(stripped) => body_bytes = stripped.into(),
                                        // Unreachable by construction (the
                                        // field serializes first); keep the
                                        // normal error path rather than resend
                                        // a body we could not fix.
                                        None => diagnostics = Some(d),
                                    }
                                }
                                if diagnostics.is_none() {
                                    if let Some(t) = tracer.as_mut() {
                                        t.attempt_failed(
                                            clock,
                                            crate::runtime::trace::RetryClass::Other,
                                            Duration::ZERO,
                                            Some(status.as_u16()),
                                            trace_rid,
                                            "cache_diagnosis_rejected",
                                        );
                                    }
                                    continue;
                                }
                            }

                            // Decide whether we've exhausted retries for this error class.
                            // A keep-warm ping never retries: any error stops it.
                            let retry_exhausted = if options.keepwarm.is_some() {
                                true
                            } else if is_429 {
                                attempt >= MAX_429_RETRIES
                            } else {
                                non_429_attempts >= max_retries
                            };

                            if !is_retryable || retry_exhausted {
                                let log_rid = trace_rid.clone();
                                if let Some(t) = tracer.take() {
                                    let terminal =
                                        t.failed_terminal(&format!("http_{}", status.as_u16()));
                                    t.finish(
                                        clock,
                                        Some(status.as_u16()),
                                        trace_rid,
                                        None,
                                        None,
                                        terminal,
                                    );
                                }
                                let hint = reset_hint.as_deref().or(last_reset_hint.as_deref());
                                return Err(RuntimeError::ApiStatus(terminal_http_error(
                                    status.as_u16(),
                                    &error_text,
                                    hint,
                                    log_rid.as_ref(),
                                )));
                            }

                            last_status = Some(status.as_u16());
                            last_reset_hint = reset_hint.clone();
                            // SECURITY (spec §5.1): `error_text` is untrusted and can
                            // echo the entire request. Retain only the status for
                            // retry state — never the body (it reaches logs and the
                            // exhausted-fallback error text below).
                            last_err = format!("HTTP {}", status.as_u16());

                            if !is_429 {
                                non_429_attempts += 1;
                            }

                            // Emit user-visible notice with specific timing when known.
                            let budget = if is_429 { MAX_429_RETRIES } else { max_retries };
                            let retry_num = if is_429 {
                                attempt + 1
                            } else {
                                non_429_attempts
                            };
                            let notice = if is_429 {
                                if let Some(ref hint) = reset_hint {
                                    format!(
                                        "⚠ Rate limited — resuming in {} ({}/{})",
                                        hint, retry_num, budget
                                    )
                                } else {
                                    format!("⚠ Rate limited — retrying ({}/{})", retry_num, budget)
                                }
                            } else {
                                format!("⏳ API error, retrying ({}/{})…", retry_num, budget)
                            };
                            tracing::warn!(
                                "API retry after {:?}: {} — {}",
                                delay,
                                notice,
                                last_err
                            );
                            let _ = tx.send(StreamEvent::Session(SessionEvent::Retry(notice)));

                            if let Some(t) = tracer.as_mut() {
                                t.attempt_failed(
                                    clock,
                                    crate::runtime::trace::anthropic::retry_class_for_status(
                                        status.as_u16(),
                                    ),
                                    delay,
                                    Some(status.as_u16()),
                                    trace_rid,
                                    &format!("http_{}", status.as_u16()),
                                );
                            }

                            await_or_cancel(cancel, tokio::time::sleep(delay)).await?;

                            if cancel.is_cancelled() {
                                return Err(RuntimeError::Canceled);
                            }
                        }
                        Err(e) => {
                            if let Some(keepwarm) = &options.keepwarm {
                                if matches!(&e, RuntimeError::Api(error) if error.is_connect()) {
                                    keepwarm
                                        .connect_failed
                                        .store(true, std::sync::atomic::Ordering::Relaxed);
                                }
                            }
                            non_429_attempts += 1;
                            if non_429_attempts > max_retries {
                                if let Some(t) = tracer.take() {
                                    let terminal = t.failed_terminal("network");
                                    t.finish(clock, None, None, None, None, terminal);
                                }
                                tracing::warn!(error = %e.user_message(), "API network retries exhausted");
                                return Err(e);
                            }
                            last_err = e.user_message();
                            last_status = None;
                            // No headers on network error — plain exponential back-off.
                            let delay = super::telemetry::network_retry_delay(non_429_attempts);
                            tracing::warn!(
                                "API retry {}/{} after {:?}: {}",
                                non_429_attempts,
                                max_retries,
                                delay,
                                last_err
                            );
                            let _ = tx.send(StreamEvent::Session(SessionEvent::Retry(format!(
                                "⏳ API error, retrying ({}/{})…",
                                non_429_attempts, max_retries
                            ))));
                            if let Some(t) = tracer.as_mut() {
                                t.attempt_failed(
                                    clock,
                                    crate::runtime::trace::RetryClass::Network,
                                    delay,
                                    None,
                                    None,
                                    "network",
                                );
                            }
                            await_or_cancel(cancel, tokio::time::sleep(delay)).await?;
                            if cancel.is_cancelled() {
                                return Err(RuntimeError::Canceled);
                            }
                        }
                    }

                    attempt += 1;
                }

                response.ok_or_else(|| {
                    let hint = last_reset_hint.as_deref();
                    let status = last_status.unwrap_or(0);
                    if status == 429 {
                        RuntimeError::ApiStatus(crate::core::error::humanize_api_error_with_reset(
                            429, &last_err, hint,
                        ))
                    } else {
                        RuntimeError::Tool(format!("API failed after retries: {}", last_err))
                    }
                })?
            };

            // ═══ TELEMETRY: capture headers before consuming the response body ═══
            let request_start = std::time::Instant::now();
            // Trace: validated provider request ID (invalid values omitted).
            let trace_rid_stream = if tracer.is_some() {
                provider_request_id_from_headers(response.headers())
            } else {
                None
            };
            let telem_request_id = if telemetry_level.enabled() {
                telemetry::request_id_from_headers(response.headers())
            } else {
                None
            };
            let telem_ratelimit = if telemetry_level == TelemetryLevel::Full {
                let rl = telemetry::ratelimit_from_headers(response.headers());
                if rl.is_empty() {
                    None
                } else {
                    Some(rl)
                }
            } else {
                None
            };

            let mut stream = response.bytes_stream();
            let mut transport_error = None;
            tracing::debug!("Stream opened");

            if !options.suppress_stream_deltas {
                let _ = tx.send(StreamEvent::Llm(LlmEvent::ResponseStart));
            }
            let mut state = ParseState::new();
            let ctx = EventCtx {
                model,
                tx: &tx,
                suppress_stream_deltas: options.suppress_stream_deltas,
                telemetry_level,
                request_start,
                cache_ttl: options.cache_ttl,
                ttl_downgrade_notified: options.ttl_downgrade_notified.clone(),
                saw_1h_honored: options.saw_1h_honored.clone(),
                request_has_1h_marker,
                usage_counters: options.usage_counters.clone(),
                cache_diagnosis: diagnostics.is_some(),
            };

            // SSE can split across chunk boundaries (even mid-UTF-8-codepoint), so
            // buffer raw bytes and only parse complete lines. Zero-copy: lines are
            // borrowed from the buffer, parsed in place (REVIEW.md P2).
            let mut line_buffer = super::sse::SseLineBuffer::new();

            // Cancellation exits through the normal finalization below: partial
            // content, residual Usage and attempt telemetry must not be dropped.
            while let Ok(Some(chunk)) = await_or_cancel(
                cancel,
                super::http_phases::next(&mut stream, super::http_phases::STREAM_IDLE),
            )
            .await
            {
                if cancel.is_cancelled() {
                    break;
                }
                // A transport error mid-stream means connection loss. It's the same
                // transient class as an HTTP 5xx or an in-stream overloaded_error, so
                // route it through the unified retry budget instead of hard-failing
                // the turn. Bill any start-captured usage first: the API already
                // processed the input even if the stream died on us.
                let chunk = match chunk {
                    Ok(c) => {
                        clock.mark_first_byte();
                        c
                    }
                    Err(e) => {
                        if state.stop_reason_seen || state.stream_error.is_some() {
                            break;
                        }
                        emit_residual_usage(&mut state, &ctx);
                        state.stream_error = Some(StreamError {
                            error_type: None,
                            message: e.user_message(),
                            retryable: true,
                        });
                        transport_error = Some(e);
                        break;
                    }
                };
                line_buffer.extend(&chunk);

                // Process complete lines from the buffer (zero-copy borrows)
                while let Some(line) = line_buffer.next_line() {
                    process_data_line(line, &mut state, &ctx);
                }
            }

            // Process any remaining buffered data (final line without trailing
            // newline) — same seam as the main loop, so all event types in a
            // partial final line are handled.
            let remaining = line_buffer.take_remaining().unwrap_or_default();
            process_data_line(&remaining, &mut state, &ctx);

            // Flush any partial block and return accumulated content
            state.finalize();

            // G3 failover gate inputs: once any SSE frame was parsed on this
            // request, a later 429 must never hop accounts (no cross-account
            // replay); a produced tool_use block is tool activity.
            output_started |= state.first_event_seen;
            tool_activity |= state
                .accumulated_content
                .iter()
                .any(|b| b.get("type").and_then(Value::as_str) == Some("tool_use"));

            // Dead-stream billing: if the stream terminated before message_delta
            // (cancel, transport death mid-stream), emit the one Usage event from
            // the message_start capture. No-op when the delta already emitted.
            emit_residual_usage(&mut state, &ctx);

            // Capture the failure signals BEFORE the telemetry block — it moves
            // `state.telem_stop_reason` into the record when telemetry is enabled.
            let stream_error = state.stream_error.take();
            let has_stop_reason = state.stop_reason_seen;
            let stop_reason_is_refusal = state.stop_reason_is_refusal;
            let cancelled = cancel.is_cancelled();
            let outcome_content = state.take_outcome_content(cancelled);

            // ═══ TRACE: finish this attempt's monotonic clock + captures ═══
            clock.mark_stream_end();
            if let Some(offset_ms) = state.first_event_elapsed_ms {
                clock.set_first_model_event_after_headers(offset_ms);
            }
            let trace_stop_reason = state.trace_stop_reason;
            let trace_usage = state.trace_usage;
            // Structural failure code for trace terminals: a vetted static
            // error-class label or a fixed literal — never raw provider text.
            let stream_error_code: &str = match stream_error.as_ref() {
                Some(e) => e
                    .error_type
                    .as_deref()
                    .and_then(crate::core::error::sanitize_error_type)
                    .unwrap_or("stream_error"),
                None => "empty_response",
            };

            // ═══ CACHE DIAGNOSTICS: advance the chain, tag the trace ═══
            // The server fingerprints every request that carried
            // `diagnostics`, keyed by its message id — so the next request
            // compares against this one even if its stream later failed.
            if diagnostics.is_some() {
                if let (Some(chain), Some(id)) = (&options.cache_diagnosis, &state.telem_msg_id) {
                    chain.record(id);
                }
            }
            let server_diagnosis = state.server_diagnosis.take();
            if let Some(t) = tracer.as_mut() {
                t.set_server_diagnosis(server_diagnosis);
            }

            // ═══ TELEMETRY: write the record ═══
            if telemetry_level.enabled() {
                // Build context record — what we sent
                let breakpoints: Vec<usize> = cleaned_messages
                    .iter()
                    .enumerate()
                    .filter(|(_, m)| {
                        // ANY block, not just the last: the conversational
                        // marker lands on the last DURABLE block, which sits
                        // before the ephemeral injected tail when per-turn
                        // context is attached.
                        if let Some(arr) = m["content"].as_array() {
                            arr.iter().any(|b| b.get("cache_control").is_some())
                        } else {
                            false
                        }
                    })
                    .map(|(i, _)| i)
                    .collect();

                let system_bytes = system_prompt.as_ref().map(|s| s.len()).unwrap_or(0);

                let record = telemetry::TelemetryRecord {
                    ts: telemetry::TelemetryRecord::now_ms(),
                    request_id: telem_request_id,
                    msg_id: state.telem_msg_id,
                    model: model.to_string(),
                    attempt: attempt + 1, // 1-based: 0 = first try, N = (N+1)th send
                    refusal_retries_used: if refusal_attempts > 0 {
                        Some(refusal_attempts)
                    } else {
                        None
                    },
                    ttft_ms: state.telem_ttft,
                    total_ms: request_start.elapsed().as_millis() as u64,
                    stop_reason: state.telem_stop_reason,
                    usage: state.telem_usage,
                    ratelimit: telem_ratelimit,
                    cache_diag: server_diagnosis.map(|d| telemetry::CacheDiagRecord {
                        miss_reason: d.reason.as_str().to_string(),
                        missed_tokens: d.missed_input_tokens,
                    }),
                    context: telemetry::ContextRecord {
                        messages: cleaned_messages.len(),
                        tools: tools_schema.len(),
                        system_bytes,
                        breakpoints,
                    },
                    keepwarm: options.keepwarm.is_some(),
                    worker: options.codex_request_role
                        == crate::runtime::openai::catalog::CodexRequestRole::Worker,
                };
                // Non-blocking enqueue onto the session's bounded writer
                // (Task 11): overflow or a broken log path is counted by
                // the writer and can never delay or fail the request.
                if let Some(writer) = options.telemetry.as_ref() {
                    writer.enqueue_telemetry(record);
                }
            }

            match classify_stream_outcome(
                stream_error,
                outcome_content,
                has_stop_reason,
                stop_reason_is_refusal,
                cancelled,
            ) {
                StreamOutcome::Done(v) => {
                    if let Some(t) = tracer.take() {
                        let terminal = if cancelled {
                            agent_core::TurnOutcome::Canceled
                        } else {
                            agent_core::TurnOutcome::Completed
                        };
                        t.finish(
                            clock,
                            Some(200),
                            trace_rid_stream,
                            trace_stop_reason,
                            trace_usage,
                            terminal,
                        );
                    }
                    return Ok(v);
                }
                StreamOutcome::Fail(msg) => {
                    if let Some(t) = tracer.take() {
                        let terminal = t.failed_terminal(stream_error_code);
                        t.finish(
                            clock,
                            Some(200),
                            trace_rid_stream,
                            trace_stop_reason,
                            trace_usage,
                            terminal,
                        );
                    }
                    return Err(RuntimeError::ApiStatus(msg));
                }
                StreamOutcome::Retry(stream_error) => {
                    let msg = stream_error.message;
                    // Anthropic OAuth overloads need the same persistent retry
                    // posture as Claude Code (10 retries), not the generic
                    // three-attempt server-error budget. Other transient stream
                    // failures continue to use the configured shared budget.
                    let is_overloaded =
                        stream_error.error_type.as_deref() == Some("overloaded_error");
                    let (retry_budget, base_delay_ms) = stream_retry_policy(
                        &auth_type,
                        stream_error.error_type.as_deref(),
                        max_retries,
                    );
                    let retries_used = if auth_type == "oauth" && is_overloaded {
                        overload_attempts
                    } else {
                        non_429_attempts
                    };
                    if retries_used >= retry_budget
                        || cancel.is_cancelled()
                        || options.keepwarm.is_some()
                    {
                        // Budget exhausted (or user cancelled) — surface terminally
                        // rather than silently. Still loud, never the silent stop.
                        if let Some(t) = tracer.take() {
                            let terminal = t.failed_terminal(stream_error_code);
                            t.finish(
                                clock,
                                Some(200),
                                trace_rid_stream,
                                trace_stop_reason,
                                trace_usage,
                                terminal,
                            );
                        }
                        if let Some(error) = transport_error {
                            tracing::warn!(error = %error.user_message(), "API stream network retries exhausted");
                            return Err(error);
                        }
                        return Err(RuntimeError::ApiStatus(msg));
                    }
                    let retry_index = retries_used.saturating_add(1);
                    let delay = Duration::from_millis(
                        base_delay_ms * 2u64.pow(retry_index.saturating_sub(1).min(6)),
                    );
                    if auth_type == "oauth" && is_overloaded {
                        overload_attempts += 1;
                    } else {
                        non_429_attempts += 1;
                    }
                    attempt += 1;
                    last_err = msg.clone();
                    last_status = None;
                    tracing::warn!(
                        "in-stream API error, retrying {}/{} after {:?}: {}",
                        retry_index,
                        retry_budget,
                        delay,
                        msg
                    );
                    let _ = tx.send(StreamEvent::Session(SessionEvent::Retry(format!(
                        "⏳ API stream error — retrying ({}/{})…",
                        retry_index, retry_budget
                    ))));
                    if let Some(t) = tracer.as_mut() {
                        t.attempt_failed(
                            clock,
                            crate::runtime::trace::anthropic::retry_class_for_stream_error(
                                stream_error.error_type.as_deref(),
                            ),
                            delay,
                            Some(200),
                            trace_rid_stream,
                            stream_error_code,
                        );
                    }
                    if !options.suppress_stream_deltas {
                        let _ = tx.send(StreamEvent::Llm(LlmEvent::ResponseReset));
                    }
                    await_or_cancel(cancel, tokio::time::sleep(delay)).await?;
                    if cancel.is_cancelled() {
                        return Err(RuntimeError::Canceled);
                    }
                    // fall through to the outer `loop` head: rebuild request + re-stream.
                }
                StreamOutcome::Refusal => {
                    // Model declined the request (stop_reason=refusal). Partial
                    // content was already discarded by std::mem::take in classify.
                    // Check budget before sleeping so an immediate exhaustion is
                    // surfaced without a pointless delay.
                    if refusal_attempts >= refusal_retries || cancel.is_cancelled() {
                        if let Some(t) = tracer.take() {
                            let terminal = t.failed_terminal("refusal");
                            t.finish(
                                clock,
                                Some(200),
                                trace_rid_stream,
                                trace_stop_reason,
                                trace_usage,
                                terminal,
                            );
                        }
                        let msg = format!(
                            "⚠ model refused the request ({} attempt{})",
                            refusal_attempts + 1,
                            if refusal_attempts == 0 { "" } else { "s" }
                        );
                        // Surface visible inline notice to TUI (same channel as
                        // rate-limit notices — renders as a turn-end inline message).
                        let _ = tx.send(StreamEvent::Session(SessionEvent::Notice(msg.clone())));
                        // Hard-terminate so the turn never silently ends.
                        return Err(RuntimeError::ApiStatus(msg));
                    }
                    refusal_attempts += 1;
                    attempt += 1;
                    // Fixed 1s delay — refusals are not transient load, exponential
                    // backoff doesn't help. Separate budget from the error budget.
                    let delay = Duration::from_secs(1);
                    tracing::warn!(
                        "stop_reason=refusal — retrying {}/{} after {:?}",
                        refusal_attempts,
                        refusal_retries,
                        delay
                    );
                    let _ = tx.send(StreamEvent::Session(SessionEvent::Retry(format!(
                        "⏳ model refusal — retrying ({}/{})…",
                        refusal_attempts, refusal_retries
                    ))));
                    if let Some(t) = tracer.as_mut() {
                        t.attempt_failed(
                            clock,
                            crate::runtime::trace::RetryClass::Other,
                            delay,
                            Some(200),
                            trace_rid_stream,
                            "refusal",
                        );
                    }
                    if !options.suppress_stream_deltas {
                        let _ = tx.send(StreamEvent::Llm(LlmEvent::ResponseReset));
                    }
                    await_or_cancel(cancel, tokio::time::sleep(delay)).await?;
                    if cancel.is_cancelled() {
                        return Err(RuntimeError::Canceled);
                    }
                    // fall through to outer `loop` head — fresh request, fresh ParseState
                }
            }
        } // end UNIFIED RETRY LOOP (task #130)
    }
}

/// Transports other than native Anthropic (cloud brokers, OpenAI-compatible
/// providers) emit `SessionEvent::Usage` but do not book the turn's
/// [`UsageCounters`](crate::runtime::budget::UsageCounters). Book their Usage
/// events on the way through, priced for `model`, so the turn's cost and
/// token budgets bind on every provider. Events keep their order.
pub(crate) async fn with_usage_metering<T, Fut>(
    model: &str,
    counters: Option<Arc<crate::runtime::budget::UsageCounters>>,
    tx: &mpsc::UnboundedSender<StreamEvent>,
    route: impl FnOnce(mpsc::UnboundedSender<StreamEvent>) -> Fut,
) -> T
where
    Fut: std::future::Future<Output = T>,
{
    let Some(counters) = counters else {
        return route(tx.clone()).await;
    };
    let forward = move |counters: &crate::runtime::budget::UsageCounters,
                        model: &str,
                        tx: &mpsc::UnboundedSender<StreamEvent>,
                        event: StreamEvent| {
        if let StreamEvent::Session(SessionEvent::Usage {
            input_tokens,
            output_tokens,
            cache_read_input_tokens,
            cache_creation_input_tokens,
            cache_creation_5m,
            cache_creation_1h,
            ..
        }) = &event
        {
            counters.record_priced(
                model,
                *input_tokens,
                *output_tokens,
                *cache_read_input_tokens,
                *cache_creation_input_tokens,
                *cache_creation_5m,
                *cache_creation_1h,
            );
        }
        let _ = tx.send(event);
    };
    let (route_tx, mut route_rx) = mpsc::unbounded_channel();
    let routed = route(route_tx);
    tokio::pin!(routed);
    let result = loop {
        tokio::select! {
            biased;
            Some(event) = route_rx.recv() => forward(&counters, model, tx, event),
            result = &mut routed => break result,
        }
    };
    // Drain what this request sent before returning, so its usage is booked
    // before the stream loop checks the turn budget. Normally the route's
    // sender is gone and this ends at once; a sender clone still held by a
    // provider task winding down gets a short, bounded wait.
    let drained = tokio::time::timeout(Duration::from_millis(200), async {
        while let Some(event) = route_rx.recv().await {
            forward(&counters, model, tx, event);
        }
    })
    .await
    .is_ok();
    if drained {
        return result;
    }
    // A sender that outlives even that must not lose its events.
    let tx = tx.clone();
    let model = model.to_string();
    tokio::spawn(async move {
        while let Some(event) = route_rx.recv().await {
            forward(&counters, &model, &tx, event);
        }
    });
    result
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::runtime::telemetry::TelemetryLevel;

    #[tokio::test]
    async fn non_anthropic_usage_is_booked_into_the_turn_counters() {
        let counters = Arc::new(crate::runtime::budget::UsageCounters::default());
        let (tx, mut rx) = mpsc::unbounded_channel();
        let out = with_usage_metering("gpt-5", Some(Arc::clone(&counters)), &tx, |route_tx| async move {
            let _ = route_tx.send(StreamEvent::Llm(LlmEvent::Text("hi".into())));
            let _ = route_tx.send(StreamEvent::Session(SessionEvent::Usage {
                input_tokens: 1_000,
                output_tokens: 200,
                cache_read_input_tokens: 50_000,
                cache_creation_input_tokens: 0,
                cache_creation_5m: None,
                cache_creation_1h: None,
                model: None,
            }));
            7
        })
        .await;
        assert_eq!(out, 7);
        assert_eq!(counters.records(), 1);
        assert_eq!(counters.totals(), (1_000, 200, 50_000, 0));
        let expected = agent_core::pricing::calculate_cost_optional_split(
            "gpt-5", 1_000, 200, 50_000, 0, None, None,
        );
        assert!((counters.cost_usd() - expected).abs() < 1e-6);
        // Events still reach the caller, in order.
        assert!(matches!(rx.recv().await, Some(StreamEvent::Llm(LlmEvent::Text(_)))));
        assert!(matches!(
            rx.recv().await,
            Some(StreamEvent::Session(SessionEvent::Usage { .. }))
        ));
    }

    /// Test harness: fresh ParseState + unbounded channel + EventCtx (Full
    /// telemetry so capture paths run). Returns (state, rx, ctx-parts).
    fn harness() -> (
        ParseState,
        mpsc::UnboundedSender<StreamEvent>,
        mpsc::UnboundedReceiver<StreamEvent>,
    ) {
        let (tx, rx) = mpsc::unbounded_channel();
        (ParseState::new(), tx, rx)
    }

    fn make_ctx(tx: &mpsc::UnboundedSender<StreamEvent>) -> EventCtx<'_> {
        EventCtx {
            model: "claude-sonnet-4-5",
            tx,
            suppress_stream_deltas: false,
            telemetry_level: TelemetryLevel::Full,
            request_start: std::time::Instant::now(),
            cache_ttl: crate::core::config::CacheTtl::FiveMinutes,
            ttl_downgrade_notified: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
            saw_1h_honored: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
            request_has_1h_marker: true,
            usage_counters: None,
            cache_diagnosis: false,
        }
    }

    #[test]
    fn suppressed_sync_context_accumulates_text_without_emitting_deltas() {
        let (tx, mut rx) = mpsc::unbounded_channel();
        let mut state = ParseState::new();
        let mut ctx = make_ctx(&tx);
        ctx.suppress_stream_deltas = true;
        process_data_line(
            r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"large-final-text"}}"#,
            &mut state,
            &ctx,
        );
        assert_eq!(state.current_text, "large-final-text");
        assert!(rx.try_recv().is_err());
    }

    /// Telemetry-OFF ctx — the DEFAULT runtime config. Used to prove parse-time
    /// behavior doesn't silently depend on telemetry being enabled.
    fn make_ctx_telemetry_off(tx: &mpsc::UnboundedSender<StreamEvent>) -> EventCtx<'_> {
        EventCtx {
            model: "claude-sonnet-4-5",
            tx,
            suppress_stream_deltas: false,
            telemetry_level: TelemetryLevel::Off,
            request_start: std::time::Instant::now(),
            cache_ttl: crate::core::config::CacheTtl::FiveMinutes,
            ttl_downgrade_notified: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
            saw_1h_honored: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
            request_has_1h_marker: true,
            usage_counters: None,
            cache_diagnosis: false,
        }
    }

    /// Harness variant with a configured TTL + shared latches — for the
    /// silent-downgrade detector tests.
    fn make_ctx_ttl<'a>(
        tx: &'a mpsc::UnboundedSender<StreamEvent>,
        ttl: crate::core::config::CacheTtl,
        latch: &std::sync::Arc<std::sync::atomic::AtomicBool>,
        honored: &std::sync::Arc<std::sync::atomic::AtomicBool>,
    ) -> EventCtx<'a> {
        EventCtx {
            model: "claude-sonnet-4-5",
            tx,
            suppress_stream_deltas: false,
            telemetry_level: TelemetryLevel::Full,
            request_start: std::time::Instant::now(),
            cache_ttl: ttl,
            ttl_downgrade_notified: latch.clone(),
            saw_1h_honored: honored.clone(),
            // Default armed: most detector tests model a request that DID
            // carry a 1h marker. The no-marker test overrides this.
            request_has_1h_marker: true,
            usage_counters: None,
            cache_diagnosis: false,
        }
    }

    fn feed(lines: &[&str], state: &mut ParseState, ctx: &EventCtx) {
        for line in lines {
            process_data_line(line, state, ctx);
        }
    }

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

    #[test]
    fn anthropic_oauth_overload_uses_claude_code_retry_posture() {
        assert_eq!(
            stream_retry_policy("oauth", Some("overloaded_error"), 3),
            (10, 500)
        );
    }

    #[test]
    fn api_key_overload_keeps_configured_retry_posture() {
        assert_eq!(
            stream_retry_policy("api_key", Some("overloaded_error"), 3),
            (3, 1000)
        );
    }

    #[test]
    fn oauth_non_overload_keeps_configured_retry_posture() {
        assert_eq!(
            stream_retry_policy("oauth", Some("api_error"), 3),
            (3, 1000)
        );
    }

    #[test]
    fn text_deltas_accumulate_then_flush_on_block_stop() {
        let (mut state, tx, mut rx) = harness();
        let ctx = make_ctx(&tx);
        feed(
            &[
                r#"data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}"#,
                r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello, "}}"#,
                r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"world"}}"#,
                r#"data: {"type":"content_block_stop","index":0}"#,
            ],
            &mut state,
            &ctx,
        );
        assert_eq!(state.accumulated_content.len(), 1);
        assert_eq!(
            state.accumulated_content[0],
            json!({"type":"text","text":"Hello, world"})
        );
        assert!(state.current_text.is_empty());
        let events = drain(&mut rx);
        let texts: Vec<&str> = 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 second_text_block_start_flushes_prior_text() {
        let (mut state, tx, _rx) = harness();
        let ctx = make_ctx(&tx);
        feed(
            &[
                r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"first"}}"#,
                // New text block starts while current_text is non-empty —
                // the L312–320 branch flushes the prior text block.
                r#"data: {"type":"content_block_start","index":1,"content_block":{"type":"text","text":""}}"#,
                r#"data: {"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"second"}}"#,
                r#"data: {"type":"content_block_stop","index":1}"#,
            ],
            &mut state,
            &ctx,
        );
        assert_eq!(state.accumulated_content.len(), 2);
        assert_eq!(
            state.accumulated_content[0],
            json!({"type":"text","text":"first"})
        );
        assert_eq!(
            state.accumulated_content[1],
            json!({"type":"text","text":"second"})
        );
    }

    #[test]
    fn tool_use_full_lifecycle() {
        let (mut state, tx, mut rx) = harness();
        let ctx = make_ctx(&tx);
        feed(
            &[
                r#"data: {"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"toolu_01","name":"get_weather"}}"#,
                r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"{\"city\":"}}"#,
                r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"\"Tokyo\"}"}}"#,
                r#"data: {"type":"content_block_stop","index":0}"#,
            ],
            &mut state,
            &ctx,
        );
        assert!(!state.in_tool_use, "flag must clear on block_stop");
        assert_eq!(state.accumulated_content.len(), 1);
        assert_eq!(
            state.accumulated_content[0],
            json!({"type":"tool_use","id":"toolu_01","name":"get_weather","input":{"city":"Tokyo"}})
        );
        let events = drain(&mut rx);
        assert!(matches!(
            &events[0],
            StreamEvent::Llm(LlmEvent::ToolUseStart { tool_name, tool_id })
                if tool_name == "get_weather" && tool_id == "toolu_01"
        ));
        assert!(matches!(
            &events[1],
            StreamEvent::Llm(LlmEvent::ToolUseDelta { tool_id, .. }) if tool_id == "toolu_01"
        ));
        assert!(matches!(
            events.last().unwrap(),
            StreamEvent::Llm(LlmEvent::ToolUse { tool_name, input, .. })
                if tool_name == "get_weather" && input == &json!({"city":"Tokyo"})
        ));
    }

    #[test]
    fn tool_use_invalid_json_yields_parse_error_object() {
        let (mut state, tx, _rx) = harness();
        let ctx = make_ctx(&tx);
        feed(
            &[
                r#"data: {"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"toolu_02","name":"run"}}"#,
                r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"{\"cmd\": truncated"}}"#,
                r#"data: {"type":"content_block_stop","index":0}"#,
            ],
            &mut state,
            &ctx,
        );
        let input = &state.accumulated_content[0]["input"];
        let err = input["__parse_error"]
            .as_str()
            .expect("__parse_error key present");
        assert!(err.starts_with("invalid tool input JSON:"));
    }

    #[test]
    fn tool_use_empty_input_yields_empty_object() {
        let (mut state, tx, _rx) = harness();
        let ctx = make_ctx(&tx);
        feed(
            &[
                r#"data: {"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"toolu_03","name":"noop"}}"#,
                r#"data: {"type":"content_block_stop","index":0}"#,
            ],
            &mut state,
            &ctx,
        );
        assert_eq!(state.accumulated_content[0]["input"], json!({}));
    }

    #[test]
    fn thinking_lifecycle_with_signature() {
        let (mut state, tx, mut rx) = harness();
        let ctx = make_ctx(&tx);
        feed(
            &[
                r#"data: {"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":""}}"#,
                r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"pondering"}}"#,
                r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"sig_abc"}}"#,
                r#"data: {"type":"content_block_stop","index":0}"#,
            ],
            &mut state,
            &ctx,
        );
        assert!(!state.in_thinking);
        assert_eq!(
            state.accumulated_content[0],
            json!({"type":"thinking","thinking":"pondering","signature":"sig_abc"})
        );
        let events = drain(&mut rx);
        assert!(matches!(
            &events[0],
            StreamEvent::Llm(LlmEvent::Thinking(t)) if t == "pondering"
        ));
    }

    #[test]
    fn empty_thinking_block_never_emitted() {
        let (mut state, tx, _rx) = harness();
        let ctx = make_ctx(&tx);
        feed(
            &[
                r#"data: {"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":""}}"#,
                // Only a signature delta — no thinking text. Anthropic rejects
                // empty thinking blocks; the guard must suppress the push.
                r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"sig_only"}}"#,
                r#"data: {"type":"content_block_stop","index":0}"#,
            ],
            &mut state,
            &ctx,
        );
        assert!(state.accumulated_content.is_empty());
        assert!(!state.in_thinking);
    }

    #[test]
    fn message_delta_captures_usage_stop_reason_telemetry() {
        // Future-proof path: if a delta ever carries its own cache_creation
        // sub-object, it takes precedence over the message_start capture.
        // (Live deltas don't — see live_split_from_start_survives_to_delta_emission.)
        let (mut state, tx, mut rx) = harness();
        let ctx = make_ctx(&tx);
        feed(
            &[
                r#"data: {"type":"message_delta","delta":{"stop_reason":"tool_use"},"usage":{"input_tokens":100,"output_tokens":50,"cache_read_input_tokens":300,"cache_creation_input_tokens":100,"cache_creation":{"ephemeral_5m_input_tokens":60,"ephemeral_1h_input_tokens":40}}}"#,
            ],
            &mut state,
            &ctx,
        );
        assert_eq!(state.telem_stop_reason.as_deref(), Some("tool_use"));
        assert_eq!(state.telem_usage.input, 100);
        assert_eq!(state.telem_usage.output, 50);
        assert_eq!(state.telem_usage.cache_read, 300);
        assert_eq!(state.telem_usage.cache_write, 100);
        assert_eq!(state.telem_usage.cache_write_5m, Some(60));
        assert_eq!(state.telem_usage.cache_write_1h, Some(40));
        // hit_pct = 300 / (100 + 300 + 100) * 100 = 60.0
        assert_eq!(state.telem_usage.hit_pct, 60.0);
        let events = drain(&mut rx);
        assert!(matches!(
            &events[0],
            StreamEvent::Session(SessionEvent::Usage {
                input_tokens: 100,
                output_tokens: 50,
                cache_read_input_tokens: 300,
                cache_creation_input_tokens: 100,
                cache_creation_5m: Some(60),
                cache_creation_1h: Some(40),
                model: None
            })
        ));
    }

    #[test]
    fn live_split_from_start_survives_to_delta_emission() {
        // LIVE API shape (streaming probe): message_start carries the full
        // cache_creation split; message_delta carries ONLY the aggregate.
        // Regression: reading the split exclusively in the delta arm made it
        // permanently None in live traffic — telemetry lost the 5m/1h keys
        // and the downgrade detector could never latch or fire.
        let (mut state, tx, mut rx) = harness();
        let ctx = make_ctx(&tx);
        feed(
            &[
                r#"data: {"type":"message_start","message":{"id":"msg_live","usage":{"input_tokens":4,"output_tokens":1,"cache_read_input_tokens":0,"cache_creation_input_tokens":1282,"cache_creation":{"ephemeral_5m_input_tokens":5,"ephemeral_1h_input_tokens":1277}}}}"#,
                r#"data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"input_tokens":4,"output_tokens":42,"cache_read_input_tokens":0,"cache_creation_input_tokens":1282}}"#,
            ],
            &mut state,
            &ctx,
        );
        // Telemetry carries the split captured at message_start.
        assert_eq!(state.telem_usage.cache_write, 1282);
        assert_eq!(state.telem_usage.cache_write_5m, Some(5));
        assert_eq!(state.telem_usage.cache_write_1h, Some(1277));
        // The delta-arm Usage emission (final counts) carries the split too.
        let events = drain(&mut rx);
        let last_usage = events
            .iter()
            .rfind(|e| matches!(e, StreamEvent::Session(SessionEvent::Usage { .. })))
            .expect("delta must emit a Usage event");
        assert!(matches!(
            last_usage,
            StreamEvent::Session(SessionEvent::Usage {
                output_tokens: 42,
                cache_creation_input_tokens: 1282,
                cache_creation_5m: Some(5),
                cache_creation_1h: Some(1277),
                ..
            })
        ));
    }

    #[test]
    fn message_start_captures_msg_id_and_usage() {
        // CAPTURE ONLY: message_start must emit NO Usage event — live deltas
        // repeat all aggregates, so emitting from both arms double-billed
        // input/cache in every accumulating consumer (pre-cache-ttl bug).
        let (mut state, tx, mut rx) = harness();
        let ctx = make_ctx(&tx);
        feed(
            &[
                r#"data: {"type":"message_start","message":{"id":"msg_xyz","usage":{"input_tokens":10,"output_tokens":1,"cache_read_input_tokens":7,"cache_creation_input_tokens":3}}}"#,
            ],
            &mut state,
            &ctx,
        );
        assert_eq!(state.telem_msg_id.as_deref(), Some("msg_xyz"));
        assert_eq!(state.msg_start_input, Some(10));
        assert_eq!(state.msg_start_output, Some(1));
        assert_eq!(state.msg_start_cache_read, Some(7));
        assert_eq!(state.msg_start_cache_create, Some(3));
        assert!(!state.usage_emitted);
        let events = drain(&mut rx);
        assert!(
            !events
                .iter()
                .any(|e| matches!(e, StreamEvent::Session(SessionEvent::Usage { .. }))),
            "message_start must capture only — emitting here double-counts"
        );
    }

    // ── One-emission invariant (the double-billing regression pin) ──────────

    /// Exact live API shapes (probed): message_start carries the split +
    /// aggregates + output placeholder; message_delta repeats ALL aggregates
    /// with the final output and NO split sub-object.
    const PROBED_LIVE_START: &str = r#"data: {"type":"message_start","message":{"id":"msg_probe","usage":{"input_tokens":3,"cache_creation_input_tokens":1103,"cache_read_input_tokens":0,"cache_creation":{"ephemeral_5m_input_tokens":6,"ephemeral_1h_input_tokens":1097},"output_tokens":1}}}"#;
    const PROBED_LIVE_DELTA: &str = r#"data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"input_tokens":3,"cache_creation_input_tokens":1103,"cache_read_input_tokens":0,"output_tokens":11}}"#;

    #[test]
    fn one_usage_emission_per_request_live_shapes() {
        // THE invariant: exactly ONE SessionEvent::Usage per request. Every
        // downstream consumer accumulates — two emissions = ~2x billing on
        // input/cache_read/cache_creation. This bug PREDATES cache-ttl
        // (verified at ffa83a8).
        let (mut state, tx, mut rx) = harness();
        let ctx = make_ctx(&tx);
        feed(&[PROBED_LIVE_START, PROBED_LIVE_DELTA], &mut state, &ctx);
        let events = drain(&mut rx);
        let usages: Vec<_> = events
            .iter()
            .filter(|e| matches!(e, StreamEvent::Session(SessionEvent::Usage { .. })))
            .collect();
        assert_eq!(usages.len(), 1, "exactly ONE Usage event per request");
        assert!(matches!(
            usages[0],
            StreamEvent::Session(SessionEvent::Usage {
                input_tokens: 3,
                output_tokens: 11, // delta's FINAL output, not start's placeholder
                cache_read_input_tokens: 0,
                cache_creation_input_tokens: 1103,
                cache_creation_5m: Some(6),    // start-captured split
                cache_creation_1h: Some(1097), // survives to the delta emission
                model: None,
            })
        ));
        // Residual path must be a no-op after the delta emitted.
        emit_residual_usage(&mut state, &ctx);
        assert!(
            drain(&mut rx).is_empty(),
            "residual emission must be a no-op when the delta already emitted"
        );
    }

    #[test]
    fn priced_usage_books_final_or_residual_once_with_request_model_and_ttl() {
        for final_delta in [false, true] {
            for model in ["claude-opus-4.1", "claude-opus-4.8", "claude-haiku-5.5"] {
                let (mut state, tx, mut rx) = harness();
                let counters = std::sync::Arc::new(crate::runtime::budget::UsageCounters::default());
                let mut ctx = make_ctx_telemetry_off(&tx);
                ctx.model = model;
                ctx.usage_counters = Some(counters.clone());
                feed(&[PROBED_LIVE_START], &mut state, &ctx);
                assert_eq!(counters.cost_usd(), 0.0);
                if final_delta {
                    feed(&[PROBED_LIVE_DELTA], &mut state, &ctx);
                }
                emit_residual_usage(&mut state, &ctx);
                emit_residual_usage(&mut state, &ctx);
                let output = if final_delta { 11 } else { 1 };
                let expected = agent_core::pricing::calculate_cost_split(
                    model, 3, output, 0, 6, 1097,
                );
                assert!((counters.cost_usd() - expected).abs() < 2e-9);
                assert_eq!(counters.totals(), (3, output, 0, 1103));
                assert_eq!(counters.latest_context_tokens(), 1106);
                assert_eq!(drain(&mut rx).iter().filter(|event| matches!(event,
                    StreamEvent::Session(SessionEvent::Usage { .. }))).count(), 1);
            }
        }
    }

    #[test]
    fn partial_ttl_final_and_residual_usage_agree_with_display_cost_once() {
        for final_delta in [false, true] {
            for (input, aggregate, known5) in [(0, 100_001, 0), (60_000, 40_001, 10_000)] {
                let (mut state, tx, mut rx) = harness();
                let counters = std::sync::Arc::new(crate::runtime::budget::UsageCounters::default());
                let mut ctx = make_ctx_telemetry_off(&tx);
                ctx.model = "claude-haiku-5.5";
                ctx.usage_counters = Some(counters.clone());
                let start = format!("data: {}", json!({
                    "type": "message_start", "message": {"id": "msg_partial", "usage": {
                        "input_tokens": input, "output_tokens": 0,
                        "cache_creation_input_tokens": aggregate,
                        "cache_creation": {"ephemeral_5m_input_tokens": known5}
                    }}
                }));
                feed(&[start.as_str()], &mut state, &ctx);
                assert_eq!(counters.cost_usd(), 0.0);
                if final_delta {
                    feed(&[r#"data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":0}}"#], &mut state, &ctx);
                    assert!(state.usage_emitted, "final delta must book before residual fallback");
                }
                emit_residual_usage(&mut state, &ctx);
                emit_residual_usage(&mut state, &ctx);
                let expected = if input == 0 { 0.100001 } else { 0.066251 };
                assert!((counters.cost_usd() - expected).abs() < 2e-9);
                assert_eq!(counters.totals(), (input, 0, 0, aggregate));
                let events = drain(&mut rx);
                let mut usage_count = 0;
                for event in events {
                    if let StreamEvent::Session(SessionEvent::Usage {
                        input_tokens, output_tokens, cache_read_input_tokens,
                        cache_creation_input_tokens, cache_creation_5m, cache_creation_1h, ..
                    }) = event {
                        usage_count += 1;
                        assert_eq!(cache_creation_5m, Some(known5));
                        assert_eq!(cache_creation_1h, None);
                        let display_cost = agent_core::pricing::calculate_cost_optional_split(
                            ctx.model, input_tokens, output_tokens, cache_read_input_tokens,
                            cache_creation_input_tokens, cache_creation_5m, cache_creation_1h,
                        );
                        assert!((display_cost - expected).abs() < 1e-12);
                    }
                }
                assert_eq!(usage_count, 1);
            }
        }
    }

    #[test]
    fn dead_stream_residual_bills_start_capture() {
        // Stream death between message_start and message_delta: the API
        // already processed the input — bill what start reported, once.
        let (mut state, tx, mut rx) = harness();
        let ctx = make_ctx(&tx);
        feed(&[PROBED_LIVE_START], &mut state, &ctx);
        assert!(drain(&mut rx)
            .iter()
            .all(|e| !matches!(e, StreamEvent::Session(SessionEvent::Usage { .. }))));
        emit_residual_usage(&mut state, &ctx);
        let events = drain(&mut rx);
        let usages: Vec<_> = events
            .iter()
            .filter(|e| matches!(e, StreamEvent::Session(SessionEvent::Usage { .. })))
            .collect();
        assert_eq!(usages.len(), 1, "dead stream bills exactly once");
        assert!(matches!(
            usages[0],
            StreamEvent::Session(SessionEvent::Usage {
                input_tokens: 3,
                output_tokens: 1, // start's placeholder — all we ever saw
                cache_read_input_tokens: 0,
                cache_creation_input_tokens: 1103,
                cache_creation_5m: Some(6),
                cache_creation_1h: Some(1097),
                model: None,
            })
        ));
        // Idempotent: a second residual call emits nothing.
        emit_residual_usage(&mut state, &ctx);
        assert!(drain(&mut rx).is_empty(), "residual must be idempotent");
    }

    #[test]
    fn dead_stream_before_delta_still_latches_1h_honored() {
        // B3: turn 1 writes the 1h prefix, stream dies before the delta. The
        // latch must be set at message_start, or a later healthy turn
        // (1h == 0, prefix cached) triggers a false downgrade notice.
        let (mut state, tx, _rx) = harness();
        let honored = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
        let notified = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
        let ctx = make_ctx_ttl(
            &tx,
            crate::core::config::CacheTtl::Hybrid,
            &notified,
            &honored,
        );
        feed(&[HONORED_START], &mut state, &ctx); // no delta — stream died
        assert!(
            honored.load(std::sync::atomic::Ordering::Relaxed),
            "saw_1h_honored must latch at message_start, not only at the delta"
        );
    }

    #[test]
    fn all_zero_usage_emits_no_event() {
        let (mut state, tx, mut rx) = harness();
        let ctx = make_ctx(&tx);
        feed(
            &[
                r#"data: {"type":"message_start","message":{"id":"msg_zero","usage":{"input_tokens":0,"output_tokens":0,"cache_read_input_tokens":0,"cache_creation_input_tokens":0}}}"#,
                r#"data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"input_tokens":0,"output_tokens":0,"cache_read_input_tokens":0,"cache_creation_input_tokens":0}}"#,
            ],
            &mut state,
            &ctx,
        );
        let events = drain(&mut rx);
        assert!(
            !events
                .iter()
                .any(|e| matches!(e, StreamEvent::Session(SessionEvent::Usage { .. }))),
            "all-zero usage must not emit a Usage event"
        );
        // stop_reason still captured — the gate only guards the Usage emit.
        assert_eq!(state.telem_stop_reason.as_deref(), Some("end_turn"));
    }

    #[test]
    fn ttft_set_once_on_first_event() {
        let (mut state, tx, _rx) = harness();
        let ctx = make_ctx(&tx);
        assert!(state.telem_ttft.is_none());
        feed(
            &[r#"data: {"type":"message_start","message":{"id":"msg_1"}}"#],
            &mut state,
            &ctx,
        );
        let first = state.telem_ttft;
        assert!(first.is_some());
        assert!(state.first_event_seen);
        std::thread::sleep(std::time::Duration::from_millis(5));
        feed(
            &[
                r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"x"}}"#,
            ],
            &mut state,
            &ctx,
        );
        assert_eq!(
            state.telem_ttft, first,
            "TTFT must not be overwritten by later events"
        );
    }

    /// Regression test for the double-emit bug fixed in the slice-2 pre-work
    /// micro-commit: a content_block_stop arriving via the tail path (partial
    /// final line) must clear in_tool_use so finalize() cannot re-push.
    #[test]
    fn tail_path_then_finalize_no_double_emit() {
        let (mut state, tx, mut rx) = harness();
        let ctx = make_ctx(&tx);
        // Main loop: tool_use opened and input streamed.
        feed(
            &[
                r#"data: {"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"toolu_tail","name":"ls"}}"#,
                r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"{}"}}"#,
            ],
            &mut state,
            &ctx,
        );
        // Tail path: final content_block_stop arrives as a partial last line
        // (no trailing newline) — same seam, same call.
        process_data_line(
            r#"data: {"type":"content_block_stop","index":0}"#,
            &mut state,
            &ctx,
        );
        // End-of-stream flush.
        state.finalize();

        let tool_blocks: Vec<&Value> = state
            .accumulated_content
            .iter()
            .filter(|b| b["type"] == "tool_use")
            .collect();
        assert_eq!(
            tool_blocks.len(),
            1,
            "tool_use block must be emitted exactly once"
        );
        let tool_events = drain(&mut rx)
            .into_iter()
            .filter(|e| matches!(e, StreamEvent::Llm(LlmEvent::ToolUse { .. })))
            .count();
        assert_eq!(tool_events, 1);
    }

    #[test]
    fn finalize_flushes_partial_text() {
        let (mut state, tx, _rx) = harness();
        let ctx = make_ctx(&tx);
        feed(
            &[
                r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"dangling"}}"#,
            ],
            &mut state,
            &ctx,
        );
        state.finalize();
        assert_eq!(
            state.accumulated_content,
            vec![json!({"type":"text","text":"dangling"})]
        );
    }

    #[test]
    fn finalize_flushes_partial_thinking() {
        let (mut state, tx, _rx) = harness();
        let ctx = make_ctx(&tx);
        feed(
            &[
                r#"data: {"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":""}}"#,
                r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"cut off"}}"#,
            ],
            &mut state,
            &ctx,
        );
        state.finalize();
        assert_eq!(
            state.accumulated_content,
            vec![json!({"type":"thinking","thinking":"cut off","signature":""})]
        );

        // Empty-thinking suppression in finalize too: open block, no text.
        let (mut state2, tx2, _rx2) = harness();
        let ctx2 = make_ctx(&tx2);
        feed(
            &[
                r#"data: {"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":""}}"#,
            ],
            &mut state2,
            &ctx2,
        );
        state2.finalize();
        assert!(
            state2.accumulated_content.is_empty(),
            "empty thinking must be suppressed"
        );
    }

    #[test]
    fn finalize_flushes_partial_tool() {
        let (mut state, tx, _rx) = harness();
        let ctx = make_ctx(&tx);
        feed(
            &[
                r#"data: {"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"toolu_cut","name":"grep"}}"#,
                r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"{\"pattern\":\"x\"}"}}"#,
            ],
            &mut state,
            &ctx,
        );
        state.finalize();
        assert_eq!(
            state.accumulated_content,
            vec![json!({"type":"tool_use","id":"toolu_cut","name":"grep","input":{"pattern":"x"}})]
        );
    }

    // ── take_outcome_content: cancel keeps only verifiable model output ──

    const THINK_START: &str = r#"data: {"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":""}}"#;
    const THINK_DELTA: &str = r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"weighing it"}}"#;
    const THINK_SIG: &str = r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"sig_ok"}}"#;
    const STOP0: &str = r#"data: {"type":"content_block_stop","index":0}"#;
    const TEXT_START1: &str = r#"data: {"type":"content_block_start","index":1,"content_block":{"type":"text","text":""}}"#;
    const TEXT_DELTA1: &str = r#"data: {"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"Let me check"}}"#;
    const STOP1: &str = r#"data: {"type":"content_block_stop","index":1}"#;
    const TOOL_START2: &str = r#"data: {"type":"content_block_start","index":2,"content_block":{"type":"tool_use","id":"toolu_done","name":"grep"}}"#;
    const TOOL_DELTA2: &str = r#"data: {"type":"content_block_delta","index":2,"delta":{"type":"input_json_delta","partial_json":"{\"pattern\":\"x\"}"}}"#;
    const STOP2: &str = r#"data: {"type":"content_block_stop","index":2}"#;
    const TOOL_START3: &str = r#"data: {"type":"content_block_start","index":3,"content_block":{"type":"tool_use","id":"toolu_cut","name":"bash"}}"#;
    const TOOL_DELTA3: &str = r#"data: {"type":"content_block_delta","index":3,"delta":{"type":"input_json_delta","partial_json":"{\"command\":\"rm -"}}"#;

    fn outcome_after(lines: &[&str], cancelled: bool) -> Vec<Value> {
        let (mut state, tx, _rx) = harness();
        let ctx = make_ctx(&tx);
        feed(lines, &mut state, &ctx);
        state.finalize();
        state.take_outcome_content(cancelled)
    }

    #[test]
    fn cancelled_outcome_keeps_partial_text_verbatim() {
        let content = outcome_after(&[TEXT_START1, TEXT_DELTA1], true);
        assert_eq!(content, vec![json!({"type":"text","text":"Let me check"})]);
    }

    #[test]
    fn cancelled_outcome_drops_unsigned_thinking() {
        let content = outcome_after(&[THINK_START, THINK_DELTA], true);
        assert!(content.is_empty(), "unsigned thinking must not survive a cancel: {content:?}");
    }

    #[test]
    fn cancelled_outcome_keeps_signed_thinking_and_completed_blocks() {
        let content = outcome_after(
            &[THINK_START, THINK_DELTA, THINK_SIG, STOP0, TEXT_START1, TEXT_DELTA1, STOP1],
            true,
        );
        assert_eq!(
            content,
            vec![
                json!({"type":"thinking","thinking":"weighing it","signature":"sig_ok"}),
                json!({"type":"text","text":"Let me check"}),
            ]
        );
    }

    #[test]
    fn cancelled_outcome_drops_only_the_in_flight_tool_call() {
        let content = outcome_after(
            &[
                TEXT_START1, TEXT_DELTA1, STOP1,
                TOOL_START2, TOOL_DELTA2, STOP2,
                TOOL_START3, TOOL_DELTA3,
            ],
            true,
        );
        assert_eq!(
            content,
            vec![
                json!({"type":"text","text":"Let me check"}),
                json!({"type":"tool_use","id":"toolu_done","name":"grep","input":{"pattern":"x"}}),
            ],
            "the completed call stays; the half-streamed one (truncated JSON) is dropped"
        );
    }

    #[test]
    fn uncancelled_outcome_is_byte_identical_to_accumulated_content() {
        // Same inputs, not cancelled: nothing filtered (existing finalize
        // semantics and every non-cancel path stay exactly as before).
        let lines = [THINK_START, THINK_DELTA, STOP0, TOOL_START3, TOOL_DELTA3];
        let (mut state, tx, _rx) = harness();
        let ctx = make_ctx(&tx);
        feed(&lines, &mut state, &ctx);
        state.finalize();
        let expected = state.accumulated_content.clone();
        assert_eq!(expected.len(), 2, "fixture: unsigned thinking + partial tool");
        assert_eq!(state.take_outcome_content(false), expected);
    }

    #[test]
    fn finalize_is_idempotent() {
        let (mut state, tx, _rx) = harness();
        let ctx = make_ctx(&tx);
        feed(
            &[
                r#"data: {"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"toolu_i","name":"once"}}"#,
                r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"{}"}}"#,
            ],
            &mut state,
            &ctx,
        );
        state.finalize();
        let after_first = state.accumulated_content.clone();
        state.finalize();
        assert_eq!(
            state.accumulated_content, after_first,
            "second finalize must be a no-op"
        );

        // Same for partial text.
        let (mut state2, tx2, _rx2) = harness();
        let ctx2 = make_ctx(&tx2);
        feed(
            &[
                r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"t"}}"#,
            ],
            &mut state2,
            &ctx2,
        );
        state2.finalize();
        state2.finalize();
        assert_eq!(state2.accumulated_content.len(), 1);
    }

    #[test]
    fn done_marker_and_non_data_lines_skipped() {
        let (mut state, tx, mut rx) = harness();
        let ctx = make_ctx(&tx);
        feed(
            &[
                "data: [DONE]",
                ": keepalive",
                "event: foo",
                "",
                "data: not json at all",
                r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"ok"}}"#,
            ],
            &mut state,
            &ctx,
        );
        assert_eq!(state.current_text, "ok");
        let events = drain(&mut rx);
        assert_eq!(
            events.len(),
            1,
            "only the valid data line may produce events"
        );
    }

    // ───────────────────── slice 3: typed-path additions ─────────────────────

    /// Tuple of every observable ParseState field, in declaration order:
    /// (accumulated_content, current_text, current_tool_name, current_tool_id,
    /// current_tool_input_json, in_tool_use, current_thinking,
    /// current_thinking_signature, in_thinking, telem_msg_id, telem_stop_reason).
    type StateSnapshot = (
        Vec<Value>,
        String,
        String,
        String,
        String,
        bool,
        String,
        String,
        bool,
        Option<String>,
        Option<String>,
    );

    /// Snapshot of every observable ParseState field, for bit-identical
    /// no-state-change assertions.
    fn snapshot(s: &ParseState) -> StateSnapshot {
        (
            s.accumulated_content.clone(),
            s.current_text.clone(),
            s.current_tool_name.clone(),
            s.current_tool_id.clone(),
            s.current_tool_input_json.clone(),
            s.in_tool_use,
            s.current_thinking.clone(),
            s.current_thinking_signature.clone(),
            s.in_thinking,
            s.telem_msg_id.clone(),
            s.telem_stop_reason.clone(),
        )
    }

    #[test]
    fn unknown_event_type_no_state_change() {
        let (mut state, tx, mut rx) = harness();
        let ctx = make_ctx(&tx);
        // Establish some non-trivial state first.
        feed(
            &[
                r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"pre"}}"#,
            ],
            &mut state,
            &ctx,
        );
        drain(&mut rx);
        let before = snapshot(&state);
        feed(
            &[
                r#"data: {"type":"ping"}"#,
                r#"data: {"type":"fnord","payload":[1,2,3]}"#,
            ],
            &mut state,
            &ctx,
        );
        assert_eq!(
            snapshot(&state),
            before,
            "Unknown events must not mutate state"
        );
        assert!(
            drain(&mut rx).is_empty(),
            "Unknown events must emit zero events"
        );
        assert!(
            state.stream_error.is_none(),
            "ping/unknown must not set a stream error"
        );
    }

    #[test]
    fn error_event_sets_stream_error() {
        // An in-stream Anthropic `error` event (200-status soft failure:
        // overloaded, context overflow) MUST be captured — not silently
        // dropped as an Unknown event. This is the root of task #130: a
        // dropped error event left the stream empty and the turn ended
        // silently. The captured message must name the (vetted) error type so
        // the surfaced error is actionable — but never the provider's raw
        // message, which is untrusted (spec §5.1).
        let (mut state, tx, _rx) = harness();
        let ctx = make_ctx(&tx);
        feed(
            &[
                r#"data: {"type":"error","error":{"type":"overloaded_error","message":"Overloaded"}}"#,
            ],
            &mut state,
            &ctx,
        );
        let err = state
            .stream_error
            .expect("error event must set stream_error");
        assert!(
            err.message.contains("overloaded_error"),
            "must carry the error type: {}",
            err.message
        );
        assert!(
            !err.message.contains("Overloaded"),
            "must NOT carry the raw provider message (untrusted): {}",
            err.message
        );
        assert!(
            err.retryable,
            "overloaded_error is a transient/retryable class"
        );
    }

    #[test]
    fn error_event_without_payload_still_sets_stream_error() {
        // A bare `{"type":"error"}` with no inner payload must still register
        // a failure — an error event is never a no-op.
        let (mut state, tx, _rx) = harness();
        let ctx = make_ctx(&tx);
        feed(&[r#"data: {"type":"error"}"#], &mut state, &ctx);
        assert!(
            state.stream_error.is_some(),
            "error event with no payload must still set a stream error"
        );
    }

    /// Phase 1 holdout regression — RED at HEAD: the SSE `error` event arm
    /// logged the raw frame (bounded 400 bytes) at WARN and copied the
    /// provider message verbatim into `stream_error.message` (→ retry
    /// warnings and the terminal error). GREEN: neither the captured message
    /// nor any log line carries the hostile body; the vetted error type
    /// survives for retry policy.
    #[test]
    fn hostile_sse_error_event_never_leaks_request_content() {
        const HOSTILE_SENTINEL: &str = "HOLDOUT-SENTINEL-http500-77aa";

        /// `std::io::Write` sink appending to a shared buffer.
        struct Cap(std::sync::Arc<std::sync::Mutex<Vec<u8>>>);
        impl std::io::Write for Cap {
            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(())
            }
        }

        let buf = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
        let sink = buf.clone();
        let subscriber = tracing_subscriber::fmt()
            .with_max_level(tracing::Level::TRACE)
            .with_writer(move || Cap(sink.clone()))
            .with_ansi(false)
            .finish();
        let _guard = tracing::subscriber::set_default(subscriber);

        let (mut state, tx, _rx) = harness();
        let ctx = make_ctx(&tx);
        let hostile = format!(
            r#"data: {{"type":"error","error":{{"type":"overloaded_error","message":"ECHOED:{HOSTILE_SENTINEL} {{\"system\":[...],\"tools\":[{{\"input_schema\":{{}}}}]}}"}}}}"#
        );
        feed(&[hostile.as_str()], &mut state, &ctx);

        let err = state
            .stream_error
            .take()
            .expect("error event must be captured");
        {
            let (text, ctx_name) = (err.message.as_str(), "stream_error message");
            assert!(
                !text.contains(HOSTILE_SENTINEL),
                "{ctx_name}: sentinel leaked: {text}"
            );
            assert!(
                !text.contains("ECHOED") && !text.contains("input_schema"),
                "{ctx_name}: echoed request content leaked: {text}"
            );
        }
        assert_eq!(
            err.error_type.as_deref(),
            Some("overloaded_error"),
            "vetted type must survive for retry policy"
        );
        assert!(err.retryable, "overloaded_error stays retryable");

        // Hostile unknown SSE events must not leak via the forensics trace.
        let unknown = format!(r#"data: {{"type":"fnord","echo":"{HOSTILE_SENTINEL}"}}"#);
        feed(&[unknown.as_str()], &mut state, &ctx);

        let logs = String::from_utf8(buf.lock().unwrap().clone()).unwrap();
        assert!(!logs.is_empty(), "expected SSE error warn to be logged");
        assert!(
            !logs.contains(HOSTILE_SENTINEL),
            "log sink leaked the sentinel: {logs}"
        );
        assert!(
            !logs.contains("ECHOED") && !logs.contains("input_schema"),
            "log sink leaked echoed request content: {logs}"
        );
    }

    #[test]
    fn stop_reason_seen_captured_even_with_telemetry_off() {
        // Regression (board review HIGH): classify_stream_outcome's
        // `has_stop_reason` must NOT depend on the telemetry flag. With telemetry
        // OFF (the default config), a message_delta carrying a stop_reason must
        // still set `stop_reason_seen` — otherwise a valid empty `end_turn` would
        // misclassify as a silent-stop Fail.
        let (mut state, tx, _rx) = harness();
        let ctx = make_ctx_telemetry_off(&tx);
        feed(
            &[
                r#"data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":5}}"#,
            ],
            &mut state,
            &ctx,
        );
        assert!(
            state.stop_reason_seen,
            "stop_reason must be captured with telemetry OFF"
        );
        assert!(
            state.telem_stop_reason.is_none(),
            "telem_stop_reason stays None when telemetry off — only the unconditional flag is set"
        );
    }

    #[test]
    fn context_overflow_error_is_terminal_not_retryable() {
        // `request_too_large` / `invalid_request_error` is the context-overflow
        // (#130) class — retrying never helps, so it must classify terminal.
        for kind in [
            "request_too_large",
            "invalid_request_error",
            "authentication_error",
        ] {
            let (mut state, tx, _rx) = harness();
            let ctx = make_ctx(&tx);
            let line =
                format!(r#"data: {{"type":"error","error":{{"type":"{kind}","message":"nope"}}}}"#);
            feed(&[line.as_str()], &mut state, &ctx);
            let err = state.stream_error.expect("error captured");
            assert!(!err.retryable, "{kind} must be terminal, not retryable");
        }
    }

    // ── classify_stream_outcome: the silent-stop decision (task #130) ──

    #[test]
    fn outcome_retries_on_transient_error_event() {
        let r = classify_stream_outcome(
            Some(StreamError {
                error_type: None,
                message: "overloaded".to_string(),
                retryable: true,
            }),
            vec![],
            false,
            false,
            false,
        );
        assert!(
            matches!(r, StreamOutcome::Retry(_)),
            "transient error must retry, got {r:?}"
        );
    }

    #[test]
    fn outcome_fails_on_terminal_error_event() {
        let r = classify_stream_outcome(
            Some(StreamError {
                error_type: None,
                message: "request_too_large".to_string(),
                retryable: false,
            }),
            vec![],
            false,
            false,
            false,
        );
        match r {
            StreamOutcome::Fail(m) => assert!(m.contains("request_too_large")),
            other => panic!("terminal error must Fail, got {other:?}"),
        }
    }

    #[test]
    fn outcome_fails_on_degenerate_empty_stream() {
        // Empty content + no stop_reason + not cancelled = the silent stop.
        let r = classify_stream_outcome(None, vec![], false, false, false);
        match r {
            StreamOutcome::Fail(m) => assert!(
                m.contains("context window") || m.contains("overloaded"),
                "error must be actionable: {m}"
            ),
            other => panic!("empty+no-stop must Fail, got {other:?}"),
        }
    }

    #[test]
    fn outcome_done_when_empty_but_cancelled() {
        // User cancellation legitimately yields empty content — not an error,
        // and never a retry (don't fight the user's cancel).
        let r = classify_stream_outcome(None, vec![], false, false, true);
        assert!(
            matches!(r, StreamOutcome::Done(_)),
            "cancellation is a clean Done"
        );

        // A cancel during a retryable error must ALSO downgrade to Done.
        let r2 = classify_stream_outcome(
            Some(StreamError {
                error_type: None,
                message: "overloaded".to_string(),
                retryable: true,
            }),
            vec![],
            false,
            false,
            true,
        );
        assert!(
            matches!(r2, StreamOutcome::Done(_)),
            "cancel beats a retryable error"
        );
    }

    #[test]
    fn outcome_done_when_empty_but_has_stop_reason() {
        // A real end_turn with no content (rare but valid — e.g. a pause turn)
        // carries a stop_reason and must pass through.
        let r = classify_stream_outcome(None, vec![], true, false, false);
        assert!(
            matches!(r, StreamOutcome::Done(_)),
            "empty-but-stop_reason is valid"
        );
    }

    #[test]
    fn outcome_done_passes_content_through() {
        let content = vec![json!({"type":"text","text":"hi"})];
        let r = classify_stream_outcome(None, content.clone(), true, false, false);
        match r {
            StreamOutcome::Done(v) => assert_eq!(v["content"], json!(content)),
            other => panic!("non-empty content is always Done, got {other:?}"),
        }
    }

    #[test]
    fn outcome_error_event_beats_nonempty_content() {
        // If the stream errored, surface/retry it even if partial content
        // accumulated — a partial answer after an error frame is not a turn.
        let content = vec![json!({"type":"text","text":"partial"})];
        let r = classify_stream_outcome(
            Some(StreamError {
                error_type: None,
                message: "boom".to_string(),
                retryable: false,
            }),
            content,
            true,
            false,
            false,
        );
        assert!(
            matches!(r, StreamOutcome::Fail(_)),
            "error event wins over partial content"
        );
    }

    #[test]
    fn malformed_json_line_skipped() {
        let (mut state, tx, mut rx) = harness();
        let ctx = make_ctx(&tx);
        let before = snapshot(&state);
        feed(
            &[
                r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"text_de"#, // truncated mid-string
                r#"data: {"#,
                r#"data: }{"#,
                "data: \u{1f4a5}", // raw non-JSON multi-byte
            ],
            &mut state,
            &ctx,
        );
        assert_eq!(
            snapshot(&state),
            before,
            "malformed lines must be skipped without state change"
        );
        assert!(drain(&mut rx).is_empty());
    }

    #[test]
    fn multibyte_utf8_text_delta_end_to_end() {
        // Raw multi-byte (borrow fast path) and \uXXXX-escaped (owned path)
        // variants of the same text must produce byte-identical output
        // through the full seam.
        let expected = "✨ héllo";
        for data_line in [
            "data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"✨ héllo\"}}",
            r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"\u2728 h\u00e9llo"}}"#,
        ] {
            let (mut state, tx, mut rx) = harness();
            let ctx = make_ctx(&tx);
            feed(
                &[data_line, r#"data: {"type":"content_block_stop","index":0}"#],
                &mut state,
                &ctx,
            );
            assert_eq!(
                state.accumulated_content,
                vec![json!({"type":"text","text":expected})],
                "accumulated text must be byte-identical for {data_line}"
            );
            let events = drain(&mut rx);
            assert!(
                matches!(
                    &events[0],
                    StreamEvent::Llm(LlmEvent::Text(t)) if t.as_bytes() == expected.as_bytes()
                ),
                "emitted text must be byte-identical for {data_line}"
            );
        }
    }

    #[test]
    fn event_with_unknown_delta_subtype_ignored_gracefully() {
        let (mut state, tx, mut rx) = harness();
        let ctx = make_ctx(&tx);
        feed(
            &[
                r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"keep"}}"#,
            ],
            &mut state,
            &ctx,
        );
        drain(&mut rx);
        let before = snapshot(&state);
        feed(
            &[
                r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"citations_delta","citation":{"x":1}}}"#,
            ],
            &mut state,
            &ctx,
        );
        assert_eq!(
            snapshot(&state),
            before,
            "unknown delta subtype must not mutate state"
        );
        assert!(drain(&mut rx).is_empty());
    }

    #[test]
    fn tail_partial_line_typed_parse() {
        // Tail path shape: take_remaining() yields an owned String the typed
        // event borrows from — the lifetime path slice 3 had to keep sound.
        let (mut state, tx, mut rx) = harness();
        let ctx = make_ctx(&tx);
        feed(
            &[
                r#"data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}"#,
            ],
            &mut state,
            &ctx,
        );
        let remaining: String =
            r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"tail ✨"}}"#
                .to_string();
        process_data_line(&remaining, &mut state, &ctx);
        drop(remaining); // event Cow must not outlive this — compile-time proof it didn't
        state.finalize();
        assert_eq!(
            state.accumulated_content,
            vec![json!({"type":"text","text":"tail ✨"})]
        );
        let events = drain(&mut rx);
        assert!(matches!(
            events.last().unwrap(),
            StreamEvent::Llm(LlmEvent::Text(t)) if t == "tail ✨"
        ));
    }

    // ── Silent-downgrade detector (spec §3.4.1) ─────────────────────────────
    //
    // Fixtures mirror the LIVE API shape (streaming probe): message_start
    // carries the full `cache_creation` sub-object; message_delta carries
    // ONLY the aggregate. The detector runs in the delta arm and must work
    // off the split captured at message_start.

    /// Aggregate-only delta — what message_delta actually looks like live.
    const LIVE_DELTA: &str = r#"data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"input_tokens":10,"output_tokens":5,"cache_read_input_tokens":0,"cache_creation_input_tokens":100}}"#;

    /// Downgraded turn: 1h bucket = 0, all writes landed in 5m.
    const DOWNGRADE_START: &str = r#"data: {"type":"message_start","message":{"id":"msg_dg","usage":{"input_tokens":10,"output_tokens":1,"cache_read_input_tokens":0,"cache_creation_input_tokens":100,"cache_creation":{"ephemeral_5m_input_tokens":100,"ephemeral_1h_input_tokens":0}}}}"#;

    fn count_downgrade_notices(rx: &mut mpsc::UnboundedReceiver<StreamEvent>) -> usize {
        drain(rx)
            .iter()
            .filter(|e| matches!(e, StreamEvent::Session(SessionEvent::Notice(t)) if t.contains("1h cache TTL not honored")))
            .count()
    }

    /// A turn where the 1h bucket is honored (healthy turn 1: prefix write).
    const HONORED_START: &str = r#"data: {"type":"message_start","message":{"id":"msg_ok","usage":{"input_tokens":10,"output_tokens":1,"cache_read_input_tokens":0,"cache_creation_input_tokens":100,"cache_creation":{"ephemeral_5m_input_tokens":20,"ephemeral_1h_input_tokens":80}}}}"#;

    #[test]
    fn downgrade_detector_silent_for_healthy_hybrid_session() {
        // The false positive the saw_1h_honored latch exists to kill: a
        // healthy Hybrid session's turn 2+ has 1h == 0 (prefix cached) and
        // 5m > 0 (tail rewrite) — the exact downgrade signature. Turn 1's
        // prefix write must latch saw_1h_honored and keep the detector
        // silent for the rest of the session.
        let (mut state, tx, mut rx) = harness();
        let notified = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
        let honored = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
        let ctx = make_ctx_ttl(
            &tx,
            crate::core::config::CacheTtl::Hybrid,
            &notified,
            &honored,
        );
        // Turn 1: 1h prefix written (split on message_start) → latch set, no notice.
        feed(&[HONORED_START, LIVE_DELTA], &mut state, &ctx);
        assert_eq!(count_downgrade_notices(&mut rx), 0, "turn 1 (1h honored)");
        assert!(
            honored.load(std::sync::atomic::Ordering::Relaxed),
            "latch set on 1h write"
        );
        // Turn 2: prefix cached → 1h == 0, 5m > 0. Healthy. SILENCE.
        let (mut state_t2, tx_t2, mut rx_t2) = harness();
        let ctx_t2 = make_ctx_ttl(
            &tx_t2,
            crate::core::config::CacheTtl::Hybrid,
            &notified,
            &honored,
        );
        feed(&[DOWNGRADE_START, LIVE_DELTA], &mut state_t2, &ctx_t2);
        assert_eq!(
            count_downgrade_notices(&mut rx_t2),
            0,
            "turn 2 (healthy hybrid signature)"
        );
        // Later request in the same session (new ctx, same latches): still silent.
        let (mut state2, tx2, mut rx2) = harness();
        let ctx2 = make_ctx_ttl(
            &tx2,
            crate::core::config::CacheTtl::Hybrid,
            &notified,
            &honored,
        );
        feed(&[DOWNGRADE_START, LIVE_DELTA], &mut state2, &ctx2);
        assert_eq!(
            count_downgrade_notices(&mut rx2),
            0,
            "later request, same session"
        );
    }

    #[test]
    fn downgrade_detector_fires_once_when_1h_never_honored() {
        // Genuinely downgraded account: the 1h bucket never goes nonzero —
        // the notice fires on turn 1, exactly once per session, in both modes.
        for ttl in [
            crate::core::config::CacheTtl::OneHour,
            crate::core::config::CacheTtl::Hybrid,
        ] {
            let (mut state, tx, mut rx) = harness();
            let notified = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
            let honored = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
            let ctx = make_ctx_ttl(&tx, ttl, &notified, &honored);
            // First occurrence: 1h bucket = 0, 5m bucket > 0 → exactly one Notice.
            feed(&[DOWNGRADE_START, LIVE_DELTA], &mut state, &ctx);
            assert_eq!(
                count_downgrade_notices(&mut rx),
                1,
                "first occurrence under {ttl:?}"
            );
            // Second occurrence (same session/latch): nothing.
            let (mut state_b, tx_b, mut rx_b) = harness();
            let ctx_b = make_ctx_ttl(&tx_b, ttl, &notified, &honored);
            feed(&[DOWNGRADE_START, LIVE_DELTA], &mut state_b, &ctx_b);
            assert_eq!(
                count_downgrade_notices(&mut rx_b),
                0,
                "second occurrence under {ttl:?}"
            );
            // Latch persists across requests in the session (new ctx, same latches).
            let (mut state2, tx2, mut rx2) = harness();
            let ctx2 = make_ctx_ttl(&tx2, ttl, &notified, &honored);
            feed(&[DOWNGRADE_START, LIVE_DELTA], &mut state2, &ctx2);
            assert_eq!(
                count_downgrade_notices(&mut rx2),
                0,
                "next request, same session"
            );
            // Mode is never auto-flipped — ctx still carries the configured TTL.
            assert_eq!(ctx2.cache_ttl, ttl);
        }
    }

    #[test]
    fn downgrade_detector_silent_under_default_5m() {
        let (mut state, tx, mut rx) = harness();
        let ctx = make_ctx(&tx); // FiveMinutes
        feed(&[DOWNGRADE_START, LIVE_DELTA], &mut state, &ctx);
        assert_eq!(count_downgrade_notices(&mut rx), 0, "5m mode never warns");
    }

    #[test]
    fn downgrade_detector_silent_when_1h_honored() {
        let (mut state, tx, mut rx) = harness();
        let latch = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
        let honored = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
        let ctx = make_ctx_ttl(
            &tx,
            crate::core::config::CacheTtl::OneHour,
            &latch,
            &honored,
        );
        feed(&[HONORED_START, LIVE_DELTA], &mut state, &ctx);
        assert_eq!(count_downgrade_notices(&mut rx), 0);
        assert!(
            !latch.load(std::sync::atomic::Ordering::Relaxed),
            "latch untouched when honored"
        );
    }

    #[test]
    fn downgrade_detector_silent_when_split_absent() {
        // cache_creation sub-object missing entirely → no basis to judge; stay quiet.
        let (mut state, tx, mut rx) = harness();
        let latch = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
        let honored = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
        let ctx = make_ctx_ttl(
            &tx,
            crate::core::config::CacheTtl::OneHour,
            &latch,
            &honored,
        );
        feed(
            &[
                r#"data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"input_tokens":10,"output_tokens":5,"cache_read_input_tokens":0,"cache_creation_input_tokens":100}}"#,
            ],
            &mut state,
            &ctx,
        );
        assert_eq!(count_downgrade_notices(&mut rx), 0);
    }

    #[test]
    fn downgrade_detector_silent_on_warm_restart() {
        // Fresh process + still-warm 1h prefix: turn 1 has cache_read > 0
        // (prefix CACHED, not written), 1h == 0, 5m > 0 (tail) — healthy,
        // but fresh latches make it look like a downgrade. The
        // cache_read == 0 guard keeps it silent. A genuinely downgraded
        // account goes cold within 5m, so real downgrades still fire on the
        // first cold turn.
        let (mut state, tx, mut rx) = harness();
        let notified = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
        let honored = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
        let ctx = make_ctx_ttl(
            &tx,
            crate::core::config::CacheTtl::Hybrid,
            &notified,
            &honored,
        );
        feed(
            &[
                r#"data: {"type":"message_start","message":{"id":"msg_warm","usage":{"input_tokens":10,"output_tokens":1,"cache_read_input_tokens":5000,"cache_creation_input_tokens":100,"cache_creation":{"ephemeral_5m_input_tokens":100,"ephemeral_1h_input_tokens":0}}}}"#,
                r#"data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"input_tokens":10,"output_tokens":5,"cache_read_input_tokens":5000,"cache_creation_input_tokens":100}}"#,
            ],
            &mut state,
            &ctx,
        );
        assert_eq!(
            count_downgrade_notices(&mut rx),
            0,
            "warm restart is healthy — no notice"
        );
        assert!(
            !notified.load(std::sync::atomic::Ordering::Relaxed),
            "notice latch untouched"
        );
    }

    #[test]
    fn downgrade_detector_silent_without_1h_marker() {
        // Degenerate request (hybrid + API key + no system prompt + zero
        // tools): no 1h marker anywhere in the payload → a zero 1h bucket
        // proves nothing. Detector must stay disarmed — even on the exact
        // cold downgrade signature.
        let (mut state, tx, mut rx) = harness();
        let notified = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
        let honored = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
        let ctx = EventCtx {
            model: "claude-sonnet-4-5",
            tx: &tx,
            suppress_stream_deltas: false,
            telemetry_level: TelemetryLevel::Full,
            request_start: std::time::Instant::now(),
            cache_ttl: crate::core::config::CacheTtl::Hybrid,
            ttl_downgrade_notified: notified.clone(),
            saw_1h_honored: honored.clone(),
            request_has_1h_marker: false,
            usage_counters: None,
            cache_diagnosis: false,
        };
        feed(&[DOWNGRADE_START, LIVE_DELTA], &mut state, &ctx);
        assert_eq!(
            count_downgrade_notices(&mut rx),
            0,
            "no 1h marker in request → silent"
        );
        assert!(!notified.load(std::sync::atomic::Ordering::Relaxed));
    }

    // ── refusal-retry: the 8 spec tests (§8) ──

    #[test]
    fn refusal_stop_reason_detected_in_parse_state() {
        // Test 1: feed a message_delta with stop_reason="refusal"
        // → both stop_reason_seen and stop_reason_is_refusal must be true.
        let (mut state, tx, _rx) = harness();
        let ctx = make_ctx(&tx);
        feed(
            &[
                r#"data: {"type":"message_delta","delta":{"stop_reason":"refusal"},"usage":{"output_tokens":1}}"#,
            ],
            &mut state,
            &ctx,
        );
        assert!(
            state.stop_reason_seen,
            "stop_reason_seen must be true on refusal"
        );
        assert!(
            state.stop_reason_is_refusal,
            "stop_reason_is_refusal must be true when stop_reason=refusal"
        );
    }

    #[test]
    fn non_refusal_stop_reason_does_not_set_refusal_flag() {
        // Test 2: feed a message_delta with stop_reason="end_turn"
        // → stop_reason_seen true, stop_reason_is_refusal false.
        let (mut state, tx, _rx) = harness();
        let ctx = make_ctx(&tx);
        feed(
            &[
                r#"data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":5}}"#,
            ],
            &mut state,
            &ctx,
        );
        assert!(state.stop_reason_seen, "stop_reason_seen must be true");
        assert!(
            !state.stop_reason_is_refusal,
            "stop_reason_is_refusal must be false for end_turn"
        );
    }

    #[test]
    fn outcome_refusal_when_stop_reason_is_refusal() {
        // Test 3: classify returns Refusal when stop_reason_is_refusal=true and not cancelled.
        let content = vec![json!({"type": "text", "text": "some content"})];
        let r = classify_stream_outcome(None, content, true, true, false);
        assert!(
            matches!(r, StreamOutcome::Refusal),
            "classify must return Refusal when stop_reason_is_refusal=true, got {r:?}"
        );
    }

    #[test]
    fn outcome_refusal_discards_partial_content() {
        // Test 4: even with non-empty accumulated content, classify returns
        // Refusal (not Done) — content is consumed by take() and dropped.
        let content = vec![json!({"type": "text", "text": "partial answer"})];
        let r = classify_stream_outcome(None, content, true, true, false);
        assert!(
            matches!(r, StreamOutcome::Refusal),
            "partial content must not produce Done when refusal flag set, got {r:?}"
        );
    }

    #[test]
    fn outcome_refusal_cancelled_downgrades_to_done() {
        // Test 5: a cancelled stream with refusal flag must downgrade to Done,
        // not Refusal — don't burn retry budget after user cancels.
        let r = classify_stream_outcome(None, vec![], true, true, true);
        assert!(
            matches!(r, StreamOutcome::Done(_)),
            "cancelled refusal must downgrade to Done, got {r:?}"
        );
    }

    #[test]
    fn config_refusal_retries_default_is_2() {
        // Test 6: SynapsConfig default must set refusal_retries = 2.
        use crate::core::config::SynapsConfig;
        let config = SynapsConfig::default();
        assert_eq!(
            config.refusal_retries, 2,
            "default refusal_retries must be 2"
        );
    }

    #[test]
    #[serial_test::serial(synaps_base_dir)]
    fn config_refusal_retries_parsed_from_file() {
        // Test 7: load_config() parses "refusal_retries = 5" from the config file.
        // Uses SYNAPS_BASE_DIR to point load_config at a temp dir (no HOME mutation).
        use agent_core::config::load_config;
        let dir = tempfile::tempdir().expect("tempdir");
        // Config file is <base_dir>/config (no extension).
        std::fs::write(dir.path().join("config"), b"refusal_retries = 5\n").expect("write");
        let prev = std::env::var("SYNAPS_BASE_DIR").ok();
        std::env::set_var("SYNAPS_BASE_DIR", dir.path().to_str().unwrap());
        let config = load_config();
        match prev {
            Some(v) => std::env::set_var("SYNAPS_BASE_DIR", v),
            None => std::env::remove_var("SYNAPS_BASE_DIR"),
        }
        assert_eq!(
            config.refusal_retries, 5,
            "refusal_retries must parse from config file"
        );
    }

    #[test]
    fn refusal_in_stream_sets_flag_end_to_end() {
        // Test 8: full SSE fixture through process_data_line.
        // After finalize(): stop_reason_is_refusal == true.
        // classify_stream_outcome returns Refusal.
        let (mut state, tx, _rx) = harness();
        let ctx = make_ctx(&tx);
        feed(
            &[
                r#"data: {"type":"message_start","message":{"id":"msg_test","type":"message","role":"assistant","content":[],"model":"claude-opus-4-5","stop_reason":null,"usage":{"input_tokens":10,"output_tokens":0}}}"#,
                r#"data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}"#,
                r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"I cannot"}}"#,
                r#"data: {"type":"content_block_stop","index":0}"#,
                r#"data: {"type":"message_delta","delta":{"stop_reason":"refusal"},"usage":{"output_tokens":3}}"#,
            ],
            &mut state,
            &ctx,
        );
        assert!(
            state.stop_reason_is_refusal,
            "end-to-end: refusal flag must survive full event sequence"
        );
        let content = std::mem::take(&mut state.accumulated_content);
        let r = classify_stream_outcome(None, content, true, true, false);
        assert!(
            matches!(r, StreamOutcome::Refusal),
            "end-to-end: classify must return Refusal after realistic fixture, got {r:?}"
        );
    }
}

// ─────────────────────────────────────────────────────────────────────────────
// on-401 refetch integration tests  (task #157, subtask 9)
// ─────────────────────────────────────────────────────────────────────────────
//
// Strategy: spin up two tiny axum servers — one that pretends to be Anthropic,
// one that pretends to be the credential broker.  Point ApiOptions at them via
// `anthropic_base_url` (test-only seam) + a Remote CredentialSource.
//
// This module tests the C1 guard in call_api_stream_inner directly.
// ─────────────────────────────────────────────────────────────────────────────
#[cfg(test)]
mod on401_tests {
    use axum::{
        http::{HeaderMap, StatusCode},
        response::IntoResponse,
        routing::get as axum_get,
        routing::post as axum_post,
        Router,
    };
    use reqwest::Client;
    use serde_json::json;
    use std::sync::{
        atomic::{AtomicUsize, Ordering},
        Arc,
    };
    use tokio::sync::{mpsc, RwLock};

    use super::{ApiMethods, ApiOptions};
    use crate::auth::{CredentialSource, TokenCache};
    use crate::runtime::telemetry::TelemetryLevel;
    use crate::runtime::types::AuthState;
    use crate::{StreamEvent, ToolRegistry};

    // ── minimal SSE body Anthropic would return for a "hello" text turn ──────
    const SSE_SUCCESS: &str = concat!(
        "data: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_01\",\"type\":\"message\",",
        "\"role\":\"assistant\",\"content\":[],\"model\":\"claude-sonnet-4-5\",\"stop_reason\":null,",
        "\"stop_sequence\":null,\"usage\":{\"input_tokens\":10,\"output_tokens\":0,",
        "\"cache_creation_input_tokens\":0,\"cache_read_input_tokens\":0}}}\n\n",
        "data: {\"type\":\"content_block_start\",\"index\":0,",
        "\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n",
        "data: {\"type\":\"content_block_delta\",\"index\":0,",
        "\"delta\":{\"type\":\"text_delta\",\"text\":\"hi\"}}\n\n",
        "data: {\"type\":\"content_block_stop\",\"index\":0}\n\n",
        "data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\",",
        "\"stop_sequence\":null},\"usage\":{\"input_tokens\":10,\"output_tokens\":1,",
        "\"cache_creation_input_tokens\":0,\"cache_read_input_tokens\":0}}\n\n",
        "data: {\"type\":\"message_stop\"}\n\n",
    );

    /// Spawn a mock Anthropic endpoint. Returns (base_url, call_counter).
    /// Behaviour: first `fail_count` POSTs → 401; subsequent → SSE success.
    async fn spawn_mock_anthropic(fail_count: usize) -> (String, Arc<AtomicUsize>) {
        let counter = Arc::new(AtomicUsize::new(0));
        let counter_clone = Arc::clone(&counter);
        let app = Router::new().route(
            "/v1/messages",
            axum_post(move || {
                let counter = Arc::clone(&counter_clone);
                async move {
                    let n = counter.fetch_add(1, Ordering::SeqCst);
                    if n < fail_count {
                        (
                            StatusCode::UNAUTHORIZED,
                            [("content-type", "application/json")],
                            "{\"type\":\"error\",\"error\":{\"type\":\"authentication_error\"}}"
                                .to_string(),
                        )
                            .into_response()
                    } else {
                        (
                            StatusCode::OK,
                            [("content-type", "text/event-stream")],
                            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)
    }

    /// Spawn a mock broker that always vends a fresh token.
    async fn spawn_mock_broker(token: &'static str) -> String {
        let app = Router::new().route(
            "/token",
            axum_get(move |_headers: HeaderMap| async move {
                // Accept any machine-token for test simplicity
                (
                    StatusCode::OK,
                    format!("{{\"access_token\":\"{token}\",\"expires\":9999999999999}}"),
                )
            }),
        );
        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}")
    }

    /// Build an AuthState with the given token pre-loaded (simulates a Remote
    /// client that already ran refresh_if_needed and cached a token).
    fn auth_with_token(token: &str) -> Arc<RwLock<AuthState>> {
        Arc::new(RwLock::new(AuthState {
            auth_token: token.to_string(),
            auth_type: "api_key".to_string(),
            refresh_token: None,
            token_expires: Some(9_999_999_999_999),
            auth_file_mtime: None,
            bound_credential: None,
        }))
    }

    fn make_options(base_url: String, source: CredentialSource, cache: TokenCache) -> ApiOptions {
        ApiOptions {
            anthropic_base_url: Some(base_url),
            credential_source: source,
            token_cache: cache,
            ..Default::default()
        }
    }

    async fn drive_call(
        auth: Arc<RwLock<AuthState>>,
        options: ApiOptions,
    ) -> crate::error::Result<serde_json::Value> {
        let client = Client::new();
        let tools = ToolRegistry::new();
        let messages = vec![std::sync::Arc::new(
            json!({"role": "user", "content": "hi"}),
        )];
        let (tx, _rx) = mpsc::unbounded_channel::<StreamEvent>();
        ApiMethods::call_api_stream(
            &auth,
            &client,
            "claude-haiku-4-5", // fast/cheap model name; routing check just does prefix match
            &tools,
            &None, // system prompt
            0,     // thinking_budget
            agent_core::reasoning::ReasoningLevel::Adaptive,
            &messages,
            tx,
            0, // max_retries (surface errors fast)
            0, // refusal_retries
            &options,
            TelemetryLevel::Off,
        )
        .await
    }

    // ─── T1: Happy path — first call 401, second call succeeds ───────────────
    // RED: before C1 guard is restored this MUST fail (no retry → error).
    // GREEN: with C1 guard it retries once with the new token and succeeds.
    #[tokio::test]
    #[serial_test::serial(synaps_base_dir)]
    async fn on401_remote_retries_once_and_succeeds() {
        // The remote broker reads the account selector from config; keep the
        // developer's own `auth.account.anthropic` out of this test.
        let _config = crate::test_env::BaseDirGuard::new();
        let broker_url = spawn_mock_broker("sk-fresh-token").await;
        let (anthropic_url, counter) = spawn_mock_anthropic(1).await; // 1 × 401, then SSE

        let source = CredentialSource::Remote {
            endpoint: broker_url.clone(),
            machine_token: "machine-tok".into(),
        };
        // Pre-seed cache with the STALE token (the one that gets the 401)
        let cache = TokenCache::new();
        cache.put(
            "anthropic",
            crate::auth::BrokerToken {
                access_token: "sk-stale".into(),
                expires: 9_999_999_999_999,
                ttl_ms: None,
            },
        );

        let auth = auth_with_token("sk-stale");
        let options = make_options(anthropic_url, source, cache);

        let result = drive_call(auth, options).await;

        // Call must succeed
        assert!(
            result.is_ok(),
            "expected success after retry, got: {:?}",
            result.err()
        );
        // Anthropic endpoint was hit exactly twice: once 401, once 200
        assert_eq!(
            counter.load(Ordering::SeqCst),
            2,
            "expected exactly 2 calls to Anthropic (1×401 + 1×200)"
        );
    }

    // ─── T2: Persistent 401 — no infinite loop ────────────────────────────────
    // Guard fires once: 2nd 401 must surface as a terminal error.
    // Anthropic endpoint must be called exactly 2 times total (no loop).
    #[tokio::test]
    #[serial_test::serial(synaps_base_dir)]
    async fn on401_remote_persistent_401_is_terminal_not_loop() {
        // The remote broker reads the account selector from config; keep the
        // developer's own `auth.account.anthropic` out of this test.
        let _config = crate::test_env::BaseDirGuard::new();
        let broker_url = spawn_mock_broker("sk-fresh-but-useless").await;
        // All calls → 401 (fail_count=999 is effectively "always")
        let (anthropic_url, counter) = spawn_mock_anthropic(999).await;

        let source = CredentialSource::Remote {
            endpoint: broker_url,
            machine_token: "machine-tok".into(),
        };
        let cache = TokenCache::new();
        cache.put(
            "anthropic",
            crate::auth::BrokerToken {
                access_token: "sk-stale".into(),
                expires: 9_999_999_999_999,
                ttl_ms: None,
            },
        );

        let auth = auth_with_token("sk-stale");
        let options = make_options(anthropic_url, source, cache);

        let result = drive_call(auth, options).await;

        // Must be an error (not a success or an infinite retry)
        assert!(result.is_err(), "expected terminal error on persistent 401");
        // Exactly 2 upstream calls: original + single retry
        assert_eq!(
            counter.load(Ordering::SeqCst),
            2,
            "persistent 401 must cause exactly 2 upstream calls, got {}",
            counter.load(Ordering::SeqCst)
        );
    }

    // ─── T3: Local source — 401 is NOT retried (regression guard) ────────────
    // CredentialSource::Local must NOT trigger the C1 refetch path.
    // With max_retries=0 the error surfaces immediately after 1 call.
    #[tokio::test]
    async fn on401_local_api_key_does_not_retry() {
        // Always 401
        let (anthropic_url, counter) = spawn_mock_anthropic(999).await;

        let source = CredentialSource::Local;
        let cache = TokenCache::new();
        let auth = auth_with_token("sk-local-key");
        let options = make_options(anthropic_url, source, cache);

        let result = drive_call(auth, options).await;

        assert!(result.is_err(), "401 with Local source must be an error");
        assert_eq!(
            counter.load(Ordering::SeqCst),
            1,
            "Local source must never retry on 401, got {} calls",
            counter.load(Ordering::SeqCst)
        );
    }

    // ─── Local OAuth source: 401 → reload auth.json → retry once ──────────
    //
    // Models the reported bug: auth.json was rotated externally (another
    // `synaps login`, refresh-auth cron, lab-laptop sync) while this session
    // still holds the old, unexpired token in memory.

    /// Mock Anthropic that 401s any request whose bearer is not `good_bearer`,
    /// recording every Authorization header it sees.
    async fn spawn_mock_anthropic_accepting(
        good_bearer: &'static str,
    ) -> (String, Arc<std::sync::Mutex<Vec<String>>>) {
        let seen = Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
        let seen_clone = Arc::clone(&seen);
        let app = Router::new().route(
            "/v1/messages",
            axum_post(move |headers: HeaderMap| {
                let seen = Arc::clone(&seen_clone);
                async move {
                    let got = headers
                        .get("authorization")
                        .and_then(|v| v.to_str().ok())
                        .unwrap_or("")
                        .to_string();
                    let ok = got == format!("Bearer {good_bearer}");
                    seen.lock().unwrap().push(got);
                    if ok {
                        (
                            StatusCode::OK,
                            [("content-type", "text/event-stream")],
                            SSE_SUCCESS.to_string(),
                        )
                            .into_response()
                    } else {
                        (
                            StatusCode::UNAUTHORIZED,
                            [("content-type", "application/json")],
                            "{\"type\":\"error\",\"error\":{\"type\":\"authentication_error\"}}"
                                .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)
    }

    fn write_local_anthropic_creds(base: &std::path::Path, access: &str) {
        let body = json!({
            "anthropic": {
                "type": "oauth",
                "refresh": "rt-test",
                "access": access,
                "expires": 9_999_999_999_999u64,
            }
        });
        std::fs::write(base.join("auth.json"), body.to_string()).unwrap();
    }

    /// Held token bound to the default local Anthropic seat — the state a
    /// session is in after its pre-stream refresh, which is what 401
    /// recovery (`recover_local_rejection`) re-vends for.
    fn local_oauth_auth(token: &str) -> Arc<RwLock<AuthState>> {
        Arc::new(RwLock::new(AuthState {
            auth_token: token.to_string(),
            auth_type: "oauth".to_string(),
            refresh_token: None,
            token_expires: Some(9_999_999_999_999),
            auth_file_mtime: None,
            bound_credential: Some(crate::runtime::auth::anthropic_binding(
                &CredentialSource::Local,
                &crate::auth::CredentialRef::default_for(crate::auth::OAuthProviderId::Anthropic),
            )),
        }))
    }

    #[tokio::test]
    #[serial_test::serial(synaps_base_dir)]
    async fn on401_local_oauth_reloads_rotated_auth_json_and_retries_once() {
        let base = crate::test_env::BaseDirGuard::new();
        write_local_anthropic_creds(base.path(), "sk-rotated-on-disk");
        let (url, seen) = spawn_mock_anthropic_accepting("sk-rotated-on-disk").await;

        let auth = local_oauth_auth("sk-stale-in-memory");
        let options = make_options(url, CredentialSource::Local, TokenCache::new());
        let result = drive_call(Arc::clone(&auth), options).await;

        assert!(
            result.is_ok(),
            "expected success after reload: {:?}",
            result.err()
        );
        let seen = seen.lock().unwrap().clone();
        assert_eq!(
            seen,
            vec![
                "Bearer sk-stale-in-memory".to_string(),
                "Bearer sk-rotated-on-disk".to_string()
            ],
            "exactly one retry, and it must carry the reloaded token"
        );
        assert_eq!(auth.read().await.auth_token, "sk-rotated-on-disk");
    }

    #[tokio::test]
    #[serial_test::serial(synaps_base_dir)]
    async fn on401_local_oauth_persistent_401_is_terminal_not_loop() {
        let base = crate::test_env::BaseDirGuard::new();
        write_local_anthropic_creds(base.path(), "sk-also-rejected");
        let (url, seen) = spawn_mock_anthropic_accepting("sk-never-valid").await;

        let auth = local_oauth_auth("sk-stale-in-memory");
        let options = make_options(url, CredentialSource::Local, TokenCache::new());
        let result = drive_call(auth, options).await;

        let err = result.expect_err("second 401 must be terminal");
        assert_eq!(
            seen.lock().unwrap().len(),
            2,
            "persistent 401 must cause exactly 2 upstream calls"
        );
        let msg = err.to_string();
        for secret in ["sk-stale-in-memory", "sk-also-rejected", "rt-test"] {
            assert!(
                !msg.contains(secret),
                "error must not surface token values: {msg}"
            );
        }
    }

    #[tokio::test]
    #[serial_test::serial(synaps_base_dir)]
    async fn on401_local_oauth_adopts_peer_token_and_retries_only_once() {
        let _config = crate::test_env::BaseDirGuard::new();
        let credential =
            crate::auth::CredentialRef::default_for(crate::auth::OAuthProviderId::Anthropic);
        crate::auth::save_provider_auth(
            "anthropic",
            &crate::auth::OAuthCredentials {
                auth_type: "oauth".into(),
                access: "peer-token".into(),
                refresh: "never-used".into(),
                expires: crate::epoch_millis() + 3_600_000,
                account_id: None,
            },
        )
        .unwrap();
        for failures in [1, 999] {
            let (url, counter) = spawn_mock_anthropic(failures).await;
            let auth = auth_with_token("rejected-token");
            {
                let mut state = auth.write().await;
                state.auth_type = "oauth".into();
                state.bound_credential = Some(crate::runtime::auth::anthropic_binding(
                    &CredentialSource::Local,
                    &credential,
                ));
            }
            let result = drive_call(
                auth.clone(),
                make_options(url, CredentialSource::Local, TokenCache::new()),
            )
            .await;
            assert_eq!(result.is_ok(), failures == 1);
            assert_eq!(counter.load(Ordering::SeqCst), 2);
            assert_eq!(auth.read().await.auth_token, "peer-token");
        }
    }

    #[tokio::test]
    async fn exhausted_send_and_body_disconnect_are_network_errors() {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        for partial_headers in [false, true] {
            let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            let url = format!("http://{}", listener.local_addr().unwrap());
            let server = tokio::spawn(async move {
                let (mut socket, _) = listener.accept().await.unwrap();
                let mut buf = [0; 8192];
                let _ = socket.read(&mut buf).await;
                if partial_headers {
                    socket.write_all(b"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: 10000\r\n\r\ndata: ").await.unwrap();
                }
            });
            let error = drive_call(
                auth_with_token("test"),
                make_options(url, CredentialSource::Local, TokenCache::new()),
            )
            .await
            .unwrap_err();
            server.await.unwrap();
            assert!(matches!(error, crate::RuntimeError::Api(_)), "{error:?}");
            let human = crate::runtime::helpers::turn_error_for(&error, "test-turn");
            assert!(human.category_label().contains("network_error"));
            assert!(!human.message.contains("http://"));
            assert!(!human.message.is_empty());
        }
    }

    // ─── Request-lifecycle T1: outgoing-request trace is metadata-only ──────
    // A request body containing a unique sentinel must never surface in
    // tracing output at ANY level (TRACE included); only metadata fields
    // (provider, model, byte length, counts, correlation id) may appear.

    /// `std::io::Write` sink that appends to a shared buffer, for capturing
    /// formatted tracing output in-process.
    struct CaptureWriter(std::sync::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(())
        }
    }

    #[test]
    fn outgoing_request_trace_is_metadata_only() {
        let sentinel = "SENTINEL_5f2a9c_do_not_log";
        let body = serde_json::json!({
            "model": "claude-test",
            "messages": [{
                "role": "user",
                "content": [{
                    "type": "text",
                    "text": sentinel,
                    "cache_control": {"type": "ephemeral"}
                }]
            }],
            "system": sentinel,
            "tools": [],
        });
        let body_bytes = serde_json::to_vec(&body).unwrap();

        let buf = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
        let sink = buf.clone();
        let subscriber = tracing_subscriber::fmt()
            .with_max_level(tracing::Level::TRACE)
            .with_writer(move || CaptureWriter(sink.clone()))
            .with_ansi(false)
            .finish();
        tracing::subscriber::with_default(subscriber, || {
            super::trace_outgoing_request("claude-test", &body_bytes, 1, 0);
        });

        let output = String::from_utf8(buf.lock().unwrap().clone()).unwrap();
        assert!(!output.is_empty(), "expected a trace line to be emitted");
        assert!(
            !output.contains(sentinel),
            "raw payload content leaked into trace output: {output}"
        );
        for field in [
            "provider",
            "model",
            "payload_bytes",
            "message_count",
            "tool_count",
            "cache_marker_count",
            "correlation_id",
        ] {
            assert!(
                output.contains(field),
                "metadata field `{field}` missing from trace output: {output}"
            );
        }
    }

    // ─── Phase 1 holdout regression: hostile provider echoes the request ─────
    // A hostile loopback provider returns HTTP 500 whose `error.message` is
    // "ECHOED:" + the full request body (user message, system prompts, tool
    // schemas). spec §5.1 / Task 1: no raw-content sentinel may appear in
    // logs at any level, in retry notices, or in the surfaced error.

    const HOSTILE_SENTINEL: &str = "HOLDOUT-SENTINEL-http500-77aa";

    /// Stub provider that echoes the entire request body inside a 500 JSON
    /// error envelope — the exact hostile shape from the preserved probe.
    async fn spawn_echoing_500_provider() -> String {
        let app = Router::new().route(
            "/v1/messages",
            axum_post(move |body: String| async move {
                let envelope = serde_json::json!({
                    "type": "error",
                    "error": { "type": "api_error",
                               "message": format!("ECHOED:{body}") }
                });
                (
                    StatusCode::INTERNAL_SERVER_ERROR,
                    [("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}")
    }

    fn assert_no_hostile_leak(text: &str, ctx: &str) {
        assert!(
            !text.contains(HOSTILE_SENTINEL),
            "{ctx}: sentinel leaked: {text}"
        );
        assert!(
            !text.contains("ECHOED:"),
            "{ctx}: echoed request body leaked: {text}"
        );
        assert!(
            !text.contains("input_schema"),
            "{ctx}: tool schema leaked: {text}"
        );
    }

    /// RED at HEAD: the retry `tracing::warn!` prints `last_err` (status +
    /// raw body) and the exhausted path humanizes the raw body into the
    /// user-visible error — both carried the sentinel. GREEN: neither logs
    /// (any level), notices, nor the typed error contain any body-derived
    /// text.
    #[tokio::test]
    async fn hostile_500_retry_and_exhaustion_never_leak_request_content() {
        let anthropic_url = spawn_echoing_500_provider().await;

        // Capture ALL tracing output (TRACE included) on this thread.
        let buf = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
        let sink = buf.clone();
        let subscriber = tracing_subscriber::fmt()
            .with_max_level(tracing::Level::TRACE)
            .with_writer(move || CaptureWriter(sink.clone()))
            .with_ansi(false)
            .finish();
        let _guard = tracing::subscriber::set_default(subscriber);

        let options = make_options(anthropic_url, CredentialSource::Local, TokenCache::new());
        let auth = auth_with_token("sk-local-key");
        let client = Client::new();
        let tools = ToolRegistry::new();
        let messages = vec![std::sync::Arc::new(json!({
            "role": "user",
            "content": format!("{HOSTILE_SENTINEL} tell me things"),
        }))];
        let (tx, mut rx) = mpsc::unbounded_channel::<StreamEvent>();
        // max_retries=1 → exercises BOTH the retry-warning path (attempt 1)
        // and the retries-exhausted error path (attempt 2).
        let result = ApiMethods::call_api_stream(
            &auth,
            &client,
            "claude-haiku-4-5",
            &tools,
            &Some(format!("system secret {HOSTILE_SENTINEL}")),
            0,
            agent_core::reasoning::ReasoningLevel::Adaptive,
            &messages,
            tx,
            1, // max_retries
            0, // refusal_retries
            &options,
            TelemetryLevel::Off,
        )
        .await;

        // 1) Surfaced typed error must not carry any body-derived text.
        let err = result.expect_err("hostile 500 must fail").to_string();
        assert_no_hostile_leak(&err, "typed error");
        assert!(
            err.contains("500") || err.contains("server error"),
            "error must stay actionable (status class): {err}"
        );

        // 2) User-visible notices (retry banners) must not leak.
        while let Ok(ev) = rx.try_recv() {
            if let StreamEvent::Session(
                crate::SessionEvent::Notice(n) | crate::SessionEvent::Retry(n),
            ) = ev
            {
                assert_no_hostile_leak(&n, "notice");
            }
        }

        // 3) Log sink at every level must not leak.
        let logs = String::from_utf8(buf.lock().unwrap().clone()).unwrap();
        assert!(!logs.is_empty(), "expected retry warnings to be logged");
        assert_no_hostile_leak(&logs, "tracing output");
    }
}

// ─────────────────────────────────────────────────────────────────────────────
// G3 — Anthropic window exhaustion → bounded account failover (loop tests)
// ─────────────────────────────────────────────────────────────────────────────
//
// Strategy: a scripted loopback "Anthropic" whose replies are consumed in
// order and which records the `authorization` header of every call, plus a
// scripted `AnthropicAccountRouter` (no broker, no env, no config, no
// auth.json). The loop is driven through `anthropic_stream_with_router`,
// the same seam production reaches through `call_api_stream_inner`.
// ─────────────────────────────────────────────────────────────────────────────
#[cfg(test)]
mod anthropic_failover_tests {
    // Every test in this module drives the OAuth 429 path, so each one can be
    // the first to hit the "schema capture" `warn!` callsite. tracing-core
    // caches a callsite's interest globally, and while only one scoped
    // subscriber is registered a callsite first hit on ANOTHER thread is
    // cached as `never` (that thread has no subscriber). That silently hid
    // the line `schema_capture_logs_headers_once_and_never_the_body` counts
    // and failed CI once. Serializing the module removes the race.
    use axum::{
        http::{HeaderMap, StatusCode},
        response::IntoResponse,
        routing::post as axum_post,
        Router,
    };
    use reqwest::Client;
    use serde_json::json;
    use std::sync::{Arc, Mutex};
    use tokio::sync::{mpsc, RwLock};

    use super::{ApiMethods, ApiOptions};
    use crate::auth::{
        AccessToken, Account, CredentialRef, CredentialSource, OAuthProviderId, PinnedToken,
        TokenCache,
    };
    use crate::runtime::anthropic_quota::{
        AnthropicAccountRouter, AnthropicQuotaEvidence, AnthropicQuotaSignal,
    };
    use crate::runtime::auth::anthropic_binding;
    use crate::runtime::telemetry::TelemetryLevel;
    use crate::runtime::types::AuthState;
    use crate::{StreamEvent, ToolRegistry};

    /// Sentinel inside every 429 body: must never reach logs, notices or
    /// the surfaced error.
    const BODY_SENTINEL: &str = "G3-BODY-SENTINEL-9f1c";

    const SSE_SUCCESS: &str = concat!(
        "data: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_01\",\"type\":\"message\",",
        "\"role\":\"assistant\",\"content\":[],\"model\":\"claude-sonnet-4-5\",\"stop_reason\":null,",
        "\"stop_sequence\":null,\"usage\":{\"input_tokens\":10,\"output_tokens\":0,",
        "\"cache_creation_input_tokens\":0,\"cache_read_input_tokens\":0}}}\n\n",
        "data: {\"type\":\"content_block_start\",\"index\":0,",
        "\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n",
        "data: {\"type\":\"content_block_delta\",\"index\":0,",
        "\"delta\":{\"type\":\"text_delta\",\"text\":\"hi\"}}\n\n",
        "data: {\"type\":\"content_block_stop\",\"index\":0}\n\n",
        "data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\",",
        "\"stop_sequence\":null},\"usage\":{\"input_tokens\":10,\"output_tokens\":1,",
        "\"cache_creation_input_tokens\":0,\"cache_read_input_tokens\":0}}\n\n",
        "data: {\"type\":\"message_stop\"}\n\n",
    );

    /// Partial output, then an in-stream transient error: the loop retries
    /// the request, but output has now started on this request.
    const SSE_PARTIAL_THEN_OVERLOADED: &str = concat!(
        "data: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_02\",\"type\":\"message\",",
        "\"role\":\"assistant\",\"content\":[],\"model\":\"claude-sonnet-4-5\",\"stop_reason\":null,",
        "\"stop_sequence\":null,\"usage\":{\"input_tokens\":10,\"output_tokens\":0}}}\n\n",
        "data: {\"type\":\"content_block_start\",\"index\":0,",
        "\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n",
        "data: {\"type\":\"content_block_delta\",\"index\":0,",
        "\"delta\":{\"type\":\"text_delta\",\"text\":\"partial\"}}\n\n",
        "data: {\"type\":\"error\",\"error\":{\"type\":\"overloaded_error\",\"message\":\"Overloaded\"}}\n\n",
    );

    #[derive(Clone, Copy)]
    enum Reply {
        /// `anthropic-ratelimit-unified-status: rejected` + unified reset this
        /// many seconds from now, `rate_limit_error` body.
        Rejected429 { reset_in_secs: u64 },
        /// Ordinary 429: `retry-after` only, no exhaustion signal.
        Plain429 { retry_after: u64 },
        SseOk,
        SsePartialThenOverloaded,
    }

    struct MockUpstream {
        base_url: String,
        /// `authorization` header of every call, in order.
        auth_seen: Arc<Mutex<Vec<String>>>,
    }

    impl MockUpstream {
        fn calls(&self) -> usize {
            self.auth_seen.lock().unwrap().len()
        }
        fn auth_seen(&self) -> Vec<String> {
            self.auth_seen.lock().unwrap().clone()
        }
    }

    async fn spawn_upstream(script: Vec<Reply>) -> MockUpstream {
        let script = Arc::new(Mutex::new(script.into_iter()));
        let auth_seen = Arc::new(Mutex::new(Vec::new()));
        let seen = Arc::clone(&auth_seen);
        let app = Router::new().route(
            "/v1/messages",
            axum_post(move |headers: HeaderMap| {
                let script = Arc::clone(&script);
                let seen = Arc::clone(&seen);
                async move {
                    seen.lock().unwrap().push(
                        headers
                            .get("authorization")
                            .and_then(|v| v.to_str().ok())
                            .unwrap_or("")
                            .to_string(),
                    );
                    let reply = script
                        .lock()
                        .unwrap()
                        .next()
                        .expect("upstream called more times than scripted");
                    let now_secs = crate::epoch_millis() / 1000;
                    let body_429 = json!({
                        "type": "error",
                        "error": {"type": "rate_limit_error", "message": BODY_SENTINEL}
                    })
                    .to_string();
                    match reply {
                        Reply::Rejected429 { reset_in_secs } => (
                            StatusCode::TOO_MANY_REQUESTS,
                            [
                                ("content-type", "application/json".to_string()),
                                ("anthropic-ratelimit-unified-status", "rejected".to_string()),
                                (
                                    "anthropic-ratelimit-unified-reset",
                                    (now_secs + reset_in_secs).to_string(),
                                ),
                            ],
                            body_429,
                        )
                            .into_response(),
                        Reply::Plain429 { retry_after } => (
                            StatusCode::TOO_MANY_REQUESTS,
                            [
                                ("content-type", "application/json".to_string()),
                                ("retry-after", retry_after.to_string()),
                            ],
                            body_429,
                        )
                            .into_response(),
                        Reply::SseOk => (
                            StatusCode::OK,
                            [("content-type", "text/event-stream")],
                            SSE_SUCCESS.to_string(),
                        )
                            .into_response(),
                        Reply::SsePartialThenOverloaded => (
                            StatusCode::OK,
                            [("content-type", "text/event-stream")],
                            SSE_PARTIAL_THEN_OVERLOADED.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() });
        MockUpstream {
            base_url: format!("http://{addr}"),
            auth_seen,
        }
    }

    fn seat(label: &str) -> CredentialRef {
        CredentialRef::new(OAuthProviderId::Anthropic, Account::parse(label).unwrap())
    }

    fn pinned(label: &str) -> PinnedToken {
        PinnedToken {
            credential: seat(label),
            token: AccessToken {
                token: format!("tok-{label}"),
                expires: u64::MAX,
            },
        }
    }

    /// Scripted router: fixed selector mode, fixed failover answer, records
    /// every report and failover request.
    struct ScriptedRouter {
        auto: bool,
        answer: Result<Option<PinnedToken>, String>,
        reports: Mutex<Vec<(CredentialRef, AnthropicQuotaEvidence)>>,
        failovers: Mutex<Vec<(String, CredentialRef)>>,
    }

    impl ScriptedRouter {
        fn new(auto: bool, answer: Result<Option<PinnedToken>, String>) -> Self {
            Self {
                auto,
                answer,
                reports: Mutex::new(Vec::new()),
                failovers: Mutex::new(Vec::new()),
            }
        }
        fn reports(&self) -> Vec<(CredentialRef, AnthropicQuotaEvidence)> {
            self.reports.lock().unwrap().clone()
        }
        fn failovers(&self) -> Vec<(String, CredentialRef)> {
            self.failovers.lock().unwrap().clone()
        }
    }

    #[async_trait::async_trait]
    impl AnthropicAccountRouter for ScriptedRouter {
        fn auto_selected(&self) -> bool {
            self.auto
        }
        async fn report_exhausted(
            &self,
            failed: &CredentialRef,
            evidence: &AnthropicQuotaEvidence,
        ) {
            self.reports
                .lock()
                .unwrap()
                .push((failed.clone(), *evidence));
        }
        async fn failover(
            &self,
            model: &str,
            failed: &CredentialRef,
        ) -> Result<Option<PinnedToken>, String> {
            self.failovers
                .lock()
                .unwrap()
                .push((model.to_string(), failed.clone()));
            self.answer.clone()
        }
    }

    /// OAuth state pinned to `label` on the Local source — what
    /// `refresh_if_needed` leaves behind before the request loop runs.
    fn oauth_auth_pinned_to(label: &str) -> Arc<RwLock<AuthState>> {
        Arc::new(RwLock::new(AuthState {
            auth_token: format!("tok-{label}"),
            auth_type: "oauth".to_string(),
            refresh_token: None,
            token_expires: Some(u64::MAX),
            auth_file_mtime: None,
            bound_credential: Some(anthropic_binding(&CredentialSource::Local, &seat(label))),
        }))
    }

    struct Outcome {
        result: crate::error::Result<serde_json::Value>,
        notices: Vec<String>,
        /// The subset that came as typed retries (`SessionEvent::Retry`).
        retries: Vec<String>,
    }

    async fn drive(
        upstream: &MockUpstream,
        auth: Arc<RwLock<AuthState>>,
        router: &ScriptedRouter,
        max_retries: u32,
    ) -> Outcome {
        let options = ApiOptions {
            anthropic_base_url: Some(upstream.base_url.clone()),
            credential_source: CredentialSource::Local,
            token_cache: TokenCache::new(),
            ..Default::default()
        };
        let client = Client::new();
        let tools = ToolRegistry::new();
        let messages = vec![Arc::new(json!({"role": "user", "content": "hi"}))];
        let (tx, mut rx) = mpsc::unbounded_channel::<StreamEvent>();
        let cancel = tokio_util::sync::CancellationToken::new();
        let result = ApiMethods::anthropic_stream_with_router(
            &auth,
            &client,
            "claude-haiku-4-5",
            tools.tools_schema(),
            &None,
            0,
            agent_core::reasoning::ReasoningLevel::Adaptive,
            &messages,
            tx,
            &cancel,
            max_retries,
            0,
            &options,
            TelemetryLevel::Off,
            router,
        )
        .await;
        let (mut notices, mut retries) = (Vec::new(), Vec::new());
        while let Ok(ev) = rx.try_recv() {
            match ev {
                StreamEvent::Session(crate::SessionEvent::Notice(n)) => notices.push(n),
                StreamEvent::Session(crate::SessionEvent::Retry(n)) => {
                    notices.push(n.clone());
                    retries.push(n);
                }
                _ => {}
            }
        }
        Outcome { result, notices, retries }
    }

    fn assert_no_body_leak(text: &str, ctx: &str) {
        assert!(
            !text.contains(BODY_SENTINEL),
            "{ctx}: 429 body leaked: {text}"
        );
    }

    #[tokio::test]
    #[serial_test::serial(anthropic_429_log_capture)]
    async fn pre_output_rejected_429_hops_once_and_succeeds_on_seat_2() {
        let upstream =
            spawn_upstream(vec![Reply::Rejected429 { reset_in_secs: 3 * 3600 }, Reply::SseOk])
                .await;
        let router = ScriptedRouter::new(true, Ok(Some(pinned("claude2"))));
        let auth = oauth_auth_pinned_to("claude4");

        let out = drive(&upstream, Arc::clone(&auth), &router, 0).await;

        let value = out.result.expect("success on the second seat");
        assert_eq!(value["content"][0]["text"], "hi");
        assert_eq!(
            upstream.auth_seen(),
            vec!["Bearer tok-claude4".to_string(), "Bearer tok-claude2".to_string()],
            "the retry must carry the new seat's bearer"
        );
        // Exactly one cooldown report, for the failed seat, with the reset.
        let reports = router.reports();
        assert_eq!(reports.len(), 1);
        assert_eq!(reports[0].0, seat("claude4"));
        assert_eq!(
            reports[0].1.signal,
            AnthropicQuotaSignal::UnifiedStatusRejected
        );
        assert!(reports[0].1.resets_at_ms.is_some());
        // One failover request, for the bare model id, naming the failed seat.
        assert_eq!(
            router.failovers(),
            vec![("claude-haiku-4-5".to_string(), seat("claude4"))]
        );
        // The turn is now pinned to the new seat (mid-turn rounds keep it).
        let g = auth.read().await;
        assert_eq!(g.auth_token, "tok-claude2");
        assert_eq!(
            g.bound_credential.as_deref(),
            Some(anthropic_binding(&CredentialSource::Local, &seat("claude2")).as_str())
        );
        assert!(
            out.notices
                .iter()
                .any(|n| n.starts_with("⚠ claude4 exhausted until ")
                    && n.ends_with(" — switching to claude2")),
            "notices: {:?}",
            out.notices
        );
        for n in &out.notices {
            assert_no_body_leak(n, "notice");
        }
    }

    #[tokio::test]
    #[serial_test::serial(anthropic_429_log_capture)]
    async fn post_output_rejected_429_reports_cooldown_but_never_hops() {
        // Attempt 1 streams partial text then dies transiently; the retry
        // (attempt 2) is refused with a rejected window.
        let upstream = spawn_upstream(vec![
            Reply::SsePartialThenOverloaded,
            Reply::Rejected429 { reset_in_secs: 3 * 3600 },
        ])
        .await;
        let router = ScriptedRouter::new(true, Ok(Some(pinned("claude2"))));
        let auth = oauth_auth_pinned_to("claude4");

        let out = drive(&upstream, Arc::clone(&auth), &router, 1).await;

        let err = out.result.expect_err("no hop after output; long reset is terminal");
        let msg = err.to_string();
        assert!(
            msg.contains("usage limit reached on account 'claude4'"),
            "{msg}"
        );
        assert!(msg.contains("next turn will select another account"), "{msg}");
        assert_no_body_leak(&msg, "error");
        assert_eq!(upstream.calls(), 2, "no third call on another seat");
        assert_eq!(router.reports().len(), 1, "cooldown still reported");
        assert_eq!(router.reports()[0].0, seat("claude4"));
        assert!(router.failovers().is_empty(), "gate must veto the lookup");
        // Seat unchanged.
        assert_eq!(auth.read().await.auth_token, "tok-claude4");
        assert!(out
            .notices
            .iter()
            .any(|n| n.starts_with("⚠ claude4 exhausted until ")
                && n.ends_with("; next turn will use another account")));
    }

    #[tokio::test]
    #[serial_test::serial(anthropic_429_log_capture)]
    async fn two_rejected_429s_hop_once_then_end_the_turn() {
        let upstream = spawn_upstream(vec![
            Reply::Rejected429 { reset_in_secs: 3 * 3600 },
            Reply::Rejected429 { reset_in_secs: 5 * 3600 },
        ])
        .await;
        let router = ScriptedRouter::new(true, Ok(Some(pinned("claude2"))));
        let auth = oauth_auth_pinned_to("claude4");

        let out = drive(&upstream, Arc::clone(&auth), &router, 0).await;

        let msg = out.result.expect_err("second exhaustion is terminal").to_string();
        assert!(msg.contains("account 'claude2'"), "{msg}");
        assert_eq!(upstream.calls(), 2);
        assert_eq!(
            upstream.auth_seen(),
            vec!["Bearer tok-claude4".to_string(), "Bearer tok-claude2".to_string()]
        );
        // Both seats were reported; only one hop was attempted.
        let reported: Vec<CredentialRef> = router.reports().into_iter().map(|r| r.0).collect();
        assert_eq!(reported, vec![seat("claude4"), seat("claude2")]);
        assert_eq!(router.failovers().len(), 1);
    }

    #[tokio::test]
    #[serial_test::serial(anthropic_429_log_capture)]
    async fn explicit_selector_reports_cooldown_and_ends_without_lookup() {
        let upstream =
            spawn_upstream(vec![Reply::Rejected429 { reset_in_secs: 3 * 3600 }]).await;
        let router = ScriptedRouter::new(false, Ok(Some(pinned("claude2"))));
        let auth = oauth_auth_pinned_to("claude4");

        let out = drive(&upstream, Arc::clone(&auth), &router, 0).await;

        let msg = out.result.expect_err("explicit account: terminal").to_string();
        assert!(msg.contains("synaps auth use"), "{msg}");
        assert_eq!(upstream.calls(), 1);
        assert_eq!(router.reports().len(), 1);
        assert!(router.failovers().is_empty());
        assert_eq!(auth.read().await.auth_token, "tok-claude4");
        assert!(out
            .notices
            .iter()
            .any(|n| n.starts_with("⚠ account limit reached until ")));
    }

    #[tokio::test]
    #[serial_test::serial(anthropic_429_log_capture)]
    async fn same_seat_back_or_no_candidate_never_loops() {
        for answer in [Ok(None), Ok(Some(pinned("claude4"))), Err("broker down".to_string())] {
            let upstream =
                spawn_upstream(vec![Reply::Rejected429 { reset_in_secs: 3 * 3600 }]).await;
            let router = ScriptedRouter::new(true, answer);
            let auth = oauth_auth_pinned_to("claude4");

            let out = drive(&upstream, Arc::clone(&auth), &router, 0).await;

            assert!(out.result.is_err());
            assert_eq!(upstream.calls(), 1, "no retry on the exhausted seat");
            assert_eq!(router.reports().len(), 1);
            assert_eq!(router.failovers().len(), 1);
            assert_eq!(auth.read().await.auth_token, "tok-claude4");
        }
    }

    #[tokio::test]
    #[serial_test::serial(anthropic_429_log_capture)]
    async fn ordinary_429_keeps_the_existing_retry_path() {
        let upstream =
            spawn_upstream(vec![Reply::Plain429 { retry_after: 1 }, Reply::SseOk]).await;
        let router = ScriptedRouter::new(true, Ok(Some(pinned("claude2"))));
        let auth = oauth_auth_pinned_to("claude4");

        let out = drive(&upstream, Arc::clone(&auth), &router, 0).await;

        assert!(out.result.is_ok(), "{:?}", out.result.err());
        assert_eq!(
            upstream.auth_seen(),
            vec!["Bearer tok-claude4".to_string(), "Bearer tok-claude4".to_string()],
            "same seat, plain backoff"
        );
        assert!(router.reports().is_empty());
        assert!(router.failovers().is_empty());
        // 04-tui I4: a typed retry (the header's "retrying"), not just text.
        assert!(
            out.retries.iter().any(|n| n.starts_with("⚠ Rate limited — resuming in ")),
            "{:?}",
            out.notices
        );
    }

    #[tokio::test]
    #[serial_test::serial(anthropic_429_log_capture)]
    async fn short_reset_exhaustion_without_hop_falls_through_to_retry() {
        // Explicit account, window rejected but the reset is under the
        // long-reset threshold: report, notify, then today's backoff retry.
        let upstream =
            spawn_upstream(vec![Reply::Rejected429 { reset_in_secs: 30 }, Reply::SseOk]).await;
        let router = ScriptedRouter::new(false, Ok(None));
        let auth = oauth_auth_pinned_to("claude4");

        let out = drive(&upstream, Arc::clone(&auth), &router, 0).await;

        assert!(out.result.is_ok(), "{:?}", out.result.err());
        assert_eq!(upstream.calls(), 2);
        assert_eq!(router.reports().len(), 1);
        assert!(router.failovers().is_empty());
    }

    /// `std::io::Write` sink appending to a shared buffer (in-process
    /// capture of formatted tracing output).
    struct LogCapture(Arc<Mutex<Vec<u8>>>);

    impl std::io::Write for LogCapture {
        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(())
        }
    }

    #[tokio::test]
    #[serial_test::serial(anthropic_429_log_capture)]
    async fn schema_capture_logs_headers_once_and_never_the_body() {
        let upstream = spawn_upstream(vec![
            Reply::Rejected429 { reset_in_secs: 3 * 3600 },
            Reply::Rejected429 { reset_in_secs: 3 * 3600 },
        ])
        .await;
        let router = ScriptedRouter::new(true, Ok(Some(pinned("claude2"))));
        let auth = oauth_auth_pinned_to("claude4");

        let buf = Arc::new(Mutex::new(Vec::new()));
        let sink = Arc::clone(&buf);
        let subscriber = tracing_subscriber::fmt()
            .with_max_level(tracing::Level::TRACE)
            .with_writer(move || LogCapture(Arc::clone(&sink)))
            .with_ansi(false)
            .finish();
        let _guard = tracing::subscriber::set_default(subscriber);

        let out = drive(&upstream, Arc::clone(&auth), &router, 0).await;
        assert!(out.result.is_err());
        drop(_guard);

        let logs = String::from_utf8(buf.lock().unwrap().clone()).unwrap();
        assert_eq!(
            logs.matches("anthropic 429 headers (schema capture)").count(),
            1,
            "one schema-capture line per turn, even with two 429s: {logs}"
        );
        assert!(
            logs.contains("anthropic-ratelimit-unified-status=rejected"),
            "{logs}"
        );
        assert!(
            logs.contains("anthropic-ratelimit-unified-reset="),
            "{logs}"
        );
        assert_no_body_leak(&logs, "tracing output");
        assert!(!logs.contains("tok-claude"), "bearer material in logs: {logs}");
        for n in &out.notices {
            assert_no_body_leak(n, "notice");
        }
        assert_no_body_leak(&out.result.unwrap_err().to_string(), "error");
    }

    #[tokio::test]
    #[serial_test::serial(anthropic_429_log_capture)]
    async fn api_key_auth_never_classifies_or_reports() {
        // Same rejected headers, but api-key auth has no seats: the loop
        // must take today's plain backoff path and touch no seat logic.
        let upstream =
            spawn_upstream(vec![Reply::Rejected429 { reset_in_secs: 3 * 3600 }, Reply::SseOk])
                .await;
        let router = ScriptedRouter::new(true, Ok(Some(pinned("claude2"))));
        let auth = Arc::new(RwLock::new(AuthState {
            auth_token: "sk-key".to_string(),
            auth_type: "api_key".to_string(),
            refresh_token: None,
            token_expires: None,
            auth_file_mtime: None,
            bound_credential: None,
        }));

        let out = drive(&upstream, Arc::clone(&auth), &router, 0).await;

        assert!(out.result.is_ok(), "{:?}", out.result.err());
        assert_eq!(upstream.calls(), 2);
        assert!(router.reports().is_empty());
        assert!(router.failovers().is_empty());
        assert!(out
            .notices
            .iter()
            .all(|n| !n.contains("exhausted") && !n.contains("account limit")));
    }
}

// ─────────────────────────────────────────────────────────────────────────────
// Task 5 — honest cloud tool capability (spec §5.5).
//
// Cloud broker routes are text-only until full tool translation exists. A
// mode that requires tools must fail with a typed unsupported-capability
// error BEFORE any credential lookup or network access. The counting stub
// broker below receives every credential/catalog/invoke request a Remote
// source could make; the pre-flight is proven by the counter staying at 0.
// ─────────────────────────────────────────────────────────────────────────────
#[cfg(test)]
mod cloud_capability_tests {
    use axum::{http::StatusCode, Router};
    use reqwest::Client;
    use serde_json::json;
    use std::sync::{
        atomic::{AtomicUsize, Ordering},
        Arc,
    };
    use tokio::sync::{mpsc, RwLock};

    use super::{ApiMethods, ApiOptions};
    use crate::auth::{CredentialSource, TokenCache};
    use crate::runtime::telemetry::TelemetryLevel;
    use crate::runtime::types::AuthState;
    use crate::{StreamEvent, ToolRegistry};

    /// Stub broker that counts EVERY inbound request (any method, any path):
    /// token vend, cloud catalog, and cloud invoke all land here for a Remote
    /// credential source, so `hits == 0` proves zero credential use and zero
    /// network activity.
    async fn spawn_counting_stub() -> (String, Arc<AtomicUsize>) {
        let hits = Arc::new(AtomicUsize::new(0));
        let hits_clone = Arc::clone(&hits);
        let app = Router::new().fallback(move || {
            let hits = Arc::clone(&hits_clone);
            async move {
                hits.fetch_add(1, Ordering::SeqCst);
                StatusCode::INTERNAL_SERVER_ERROR
            }
        });
        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}"), hits)
    }

    async fn drive_cloud(tools: ToolRegistry) -> (crate::error::Result<serde_json::Value>, usize) {
        let (endpoint, hits) = spawn_counting_stub().await;
        let options = ApiOptions {
            credential_source: CredentialSource::Remote {
                endpoint,
                machine_token: "machine-tok".into(),
            },
            token_cache: TokenCache::new(),
            ..Default::default()
        };
        let auth = Arc::new(RwLock::new(AuthState {
            auth_token: "unused".into(),
            auth_type: "api_key".into(),
            refresh_token: None,
            token_expires: Some(9_999_999_999_999),
            auth_file_mtime: None,
            bound_credential: None,
        }));
        let client = Client::new();
        let messages = vec![Arc::new(json!({"role": "user", "content": "hi"}))];
        let (tx, _rx) = mpsc::unbounded_channel::<StreamEvent>();
        let result = ApiMethods::call_api_stream(
            &auth,
            &client,
            "aws-bedrock/anthropic.claude-3-haiku#synaps-context=ctx-test",
            &tools,
            &None,
            0,
            agent_core::reasoning::ReasoningLevel::Adaptive,
            &messages,
            tx,
            0,
            0,
            &options,
            TelemetryLevel::Off,
        )
        .await;
        (result, hits.load(Ordering::SeqCst))
    }

    /// RED at HEAD: no pre-flight exists, so the request reaches the stub
    /// broker (hits > 0) and fails with a transport-ish error instead of the
    /// typed unsupported-capability error. GREEN once the pre-flight fires
    /// before `broker_from_source`.
    #[tokio::test]
    async fn tool_requiring_cloud_route_fails_before_credentials_or_network() {
        // Default chat registry: tools are required.
        let (result, hits) = drive_cloud(ToolRegistry::new()).await;
        let msg = result
            .expect_err("tool-requiring cloud route must fail")
            .to_string();
        assert!(
            msg.contains("text-only") && msg.contains("tools"),
            "expected typed unsupported-capability error, got: {msg}"
        );
        assert_eq!(
            hits, 0,
            "cloud pre-flight must reject tools before any credential or network use"
        );
    }
}

#[cfg(test)]
mod cancellation_tests {
    use super::*;
    use axum::{body::Body, http::Response, routing::post, Router};
    use std::sync::atomic::{AtomicUsize, Ordering};

    fn test_auth() -> Arc<RwLock<AuthState>> {
        Arc::new(RwLock::new(AuthState {
            auth_token: "offline-test-key".into(),
            auth_type: "api_key".into(),
            refresh_token: None,
            token_expires: Some(9_999_999_999_999),
            auth_file_mtime: None,
            bound_credential: None,
        }))
    }

    async fn call(
        auth: &Arc<RwLock<AuthState>>,
        cancel: &CancellationToken,
        tx: mpsc::UnboundedSender<StreamEvent>,
        options: &ApiOptions,
    ) -> Result<Value> {
        ApiMethods::call_api_stream_inner(
            auth,
            &Client::builder().no_proxy().build().unwrap(),
            "claude-haiku-4-5-20251001",
            &ToolRegistry::new(),
            &None,
            0,
            agent_core::reasoning::ReasoningLevel::Adaptive,
            &[Arc::new(json!({"role":"user", "content":"offline test"}))],
            tx,
            cancel,
            3,
            1,
            options,
            TelemetryLevel::Off,
        )
        .await
    }

    #[derive(Clone, Copy)]
    enum Stall {
        Headers,
        ErrorBody,
        Retry,
        Sse,
    }

    /// Loopback only, no credentials/config/environment changes. All streams
    /// deliberately stay open so cancellation, not EOF/client timeout, wins.
    async fn stalled_server(
        stall: Stall,
    ) -> (
        ApiOptions,
        Arc<AtomicUsize>,
        mpsc::UnboundedReceiver<()>,
        tokio::task::JoinHandle<()>,
    ) {
        let hits = Arc::new(AtomicUsize::new(0));
        let counter = hits.clone();
        let (seen, rx) = mpsc::unbounded_channel();
        let app = Router::new().route(
            "/v1/messages",
            post(move || {
                let hits = counter.clone();
                let seen = seen.clone();
                async move {
                    hits.fetch_add(1, Ordering::SeqCst);
                    let _ = seen.send(());
                    if matches!(stall, Stall::Headers) {
                        return std::future::pending::<Response<Body>>().await;
                    }
                    if matches!(stall, Stall::Retry) {
                        return Response::builder()
                            .status(429)
                            .header("retry-after", "60")
                            .body(Body::from("{}"))
                            .unwrap();
                    }
                    let prefix = if matches!(stall, Stall::Sse) {
                        concat!(
                            "data: {\"type\":\"message_start\",\"message\":{\"id\":\"offline\",",
                            "\"usage\":{\"input_tokens\":12,\"output_tokens\":1,",
                            "\"cache_read_input_tokens\":5,\"cache_creation_input_tokens\":3}}}\n\n",
                            "data: {\"type\":\"content_block_start\",\"index\":0,",
                            "\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n",
                            "data: {\"type\":\"content_block_delta\",\"index\":0,",
                            "\"delta\":{\"type\":\"text_delta\",\"text\":\"partial\"}}\n\n",
                        )
                    } else {
                        "{"
                    };
                    let chunks = futures::stream::once(async move {
                        Ok::<_, std::io::Error>(bytes::Bytes::from_static(prefix.as_bytes()))
                    })
                    .chain(futures::stream::pending());
                    Response::builder()
                        .status(if matches!(stall, Stall::Sse) { 200 } else { 500 })
                        .header("content-type", "text/event-stream")
                        .body(Body::from_stream(chunks))
                        .unwrap()
                }
            }),
        );
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let url = format!("http://{}", listener.local_addr().unwrap());
        let task = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
        (
            ApiOptions {
                anthropic_base_url: Some(url),
                ..Default::default()
            },
            hits,
            rx,
            task,
        )
    }

    #[tokio::test]
    async fn pre_cancelled_call_never_waits_for_auth_or_dispatches() {
        let auth = test_auth();
        let _held = auth.write().await;
        let cancel = CancellationToken::new();
        cancel.cancel();
        let (tx, _rx) = mpsc::unbounded_channel();
        let result = tokio::time::timeout(
            Duration::from_secs(1),
            call(&auth, &cancel, tx, &ApiOptions::default()),
        )
        .await
        .expect("pre-cancel must not wait for auth");
        assert!(matches!(result, Err(RuntimeError::Canceled)));
    }

    #[tokio::test]
    async fn cancellation_interrupts_auth_lock_before_dispatch() {
        let auth = test_auth();
        let _held = auth.write().await;
        let cancel = CancellationToken::new();
        let (tx, _rx) = mpsc::unbounded_channel();
        let options = ApiOptions::default();
        let call = call(&auth, &cancel, tx, &options);
        tokio::pin!(call);
        assert!(futures::poll!(&mut call).is_pending());
        cancel.cancel();
        let result = tokio::time::timeout(Duration::from_secs(1), call)
            .await
            .expect("cancel must release the auth wait");
        assert!(matches!(result, Err(RuntimeError::Canceled)));
    }

    #[tokio::test]
    async fn cancellation_interrupts_send_and_error_body_waits() {
        for stall in [Stall::Headers, Stall::ErrorBody] {
            let (options, hits, mut seen, server) = stalled_server(stall).await;
            let auth = test_auth();
            let cancel = CancellationToken::new();
            let (tx, _rx) = mpsc::unbounded_channel();
            let result = tokio::time::timeout(Duration::from_secs(2), async {
                let (result, ()) = tokio::join!(call(&auth, &cancel, tx, &options), async {
                    seen.recv().await.unwrap();
                    // Allow the error headers to reach the client in ErrorBody mode.
                    tokio::time::sleep(Duration::from_millis(20)).await;
                    cancel.cancel();
                });
                result
            })
            .await;
            server.abort();
            assert!(matches!(
                result.expect("cancel must not wait for response"),
                Err(RuntimeError::Canceled)
            ));
            assert_eq!(hits.load(Ordering::SeqCst), 1);
        }
    }

    #[tokio::test]
    async fn cancellation_interrupts_retry_after_without_second_dispatch() {
        let (options, hits, _seen, server) = stalled_server(Stall::Retry).await;
        let auth = test_auth();
        let cancel = CancellationToken::new();
        let (tx, mut rx) = mpsc::unbounded_channel();
        let result = tokio::time::timeout(Duration::from_secs(2), async {
            let (result, ()) = tokio::join!(call(&auth, &cancel, tx, &options), async {
                while let Some(event) = rx.recv().await {
                    if matches!(event, StreamEvent::Session(SessionEvent::Retry(_))) {
                        cancel.cancel();
                        return;
                    }
                }
                panic!("expected retry notice");
            });
            result
        })
        .await;
        server.abort();
        assert!(matches!(
            result.expect("cancel must interrupt retry-after"),
            Err(RuntimeError::Canceled)
        ));
        assert_eq!(hits.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn cancelled_stalled_sse_preserves_partial_content_and_residual_usage_once() {
        let (mut options, hits, _seen, server) = stalled_server(Stall::Sse).await;
        let counters = Arc::new(crate::runtime::budget::UsageCounters::default());
        options.usage_counters = Some(counters.clone());
        let auth = test_auth();
        let cancel = CancellationToken::new();
        let (tx, mut rx) = mpsc::unbounded_channel();
        let result = tokio::time::timeout(Duration::from_secs(2), async {
            tokio::join!(call(&auth, &cancel, tx, &options), async {
                let mut usage_events = 0;
                while let Some(event) = rx.recv().await {
                    match event {
                        StreamEvent::Llm(LlmEvent::Text(text)) => {
                            assert_eq!(text, "partial");
                            cancel.cancel();
                        }
                        StreamEvent::Session(SessionEvent::Usage {
                            input_tokens,
                            output_tokens,
                            ..
                        }) => {
                            assert_eq!((input_tokens, output_tokens), (12, 1));
                            usage_events += 1;
                        }
                        _ => {}
                    }
                }
                usage_events
            })
        })
        .await;
        server.abort();
        let (response, usage_events) = result.expect("cancel must finish a stalled SSE stream");
        assert_eq!(response.unwrap()["content"][0]["text"], "partial");
        assert_eq!(usage_events, 1);
        assert_eq!(counters.totals(), (12, 1, 5, 3));
        assert_eq!(hits.load(Ordering::SeqCst), 1);
    }
}