synaps-engine 0.11.1

use super::types::{AgentEvent, StreamEvent};
use crate::core::config::CacheTtl;
use crate::SharedMessage;
use serde_json::{json, Value};
use std::sync::Arc;
use tokio::sync::mpsc;

/// Where a cache_control marker sits in the request body.
/// The API's logical emission order is tools → system → messages, so
/// StablePrefix markers always precede the MessageTail marker — this ordering
/// is what makes the Hybrid combination legal under Anthropic's rule that
/// longer-TTL breakpoints must precede shorter-TTL ones.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum MarkerSite {
    /// Tool marker, OAuth system marker, API-key system marker.
    StablePrefix,
    /// `annotate_cache_breakpoint`'s last-message marker.
    MessageTail,
}

/// The single source of the cache_control JSON value. Bare ephemeral (5m)
/// omits `ttl` entirely — byte-identical to today's payloads, so the
/// default path cannot invalidate existing cached prefixes.
///
/// Cost trade-off (honest numbers, not marketing):
/// - 5m:     1.25× write everywhere. Wins for rapid-fire sessions.
/// - 1h:     2.0× write everywhere. Wins for SPARSE long sessions — note
///   that under Hybrid, every >5m gap forces a 5m re-write of the
///   message tail, and that tail write covers the WHOLE conversation
///   since the system breakpoint, not just the increment. Uniform 1h
///   avoids that repeated full-tail re-write.
/// - Hybrid: 2.0× write on the stable prefix (tools+system, written rarely),
///   1.25× on the message tail (written every turn). Wins for
///   bursty / medium-gap cadence: cheap per-turn writes while the
///   expensive prefix survives gaps up to 1h.
pub(super) fn cache_control_value(ttl: CacheTtl, site: MarkerSite) -> Value {
    match (ttl, site) {
        (CacheTtl::FiveMinutes, _) => json!({"type": "ephemeral"}),
        (CacheTtl::OneHour, _) => json!({"type": "ephemeral", "ttl": "1h"}),
        (CacheTtl::Hybrid, MarkerSite::StablePrefix) => json!({"type": "ephemeral", "ttl": "1h"}),
        (CacheTtl::Hybrid, MarkerSite::MessageTail) => json!({"type": "ephemeral"}),
    }
}

/// Anthropic OAuth routes on the FIRST system block. It is a wire-protocol
/// contract, not Synaps branding or user-configurable identity: an OAuth
/// request that does not lead with it is refused (observed as HTTP 429
/// `rate_limit_error`, not 401/403, so it looks like a quota problem). Every
/// Anthropic OAuth request builder, compaction included, must lead with it.
pub(super) const CLAUDE_CODE_TRANSPORT_IDENTITY: &str =
    "You are Claude Code, Anthropic's official CLI for Claude.";

/// General assistant guidance that follows the product identity on OAuth.
pub(super) const OAUTH_ASSISTANT_GUIDANCE: &str =
    "You are a helpful AI assistant with access to tools. Use them when needed.";

/// Smallest response room the window clamp leaves a request (on top of its
/// thinking budget), see [`HelperMethods::fit_max_tokens_to_window`]. The
/// clamp's input estimate is conservative (up to ~1.6x for prose), so the
/// real room is usually larger than the estimate; below this the cap stops
/// shrinking and the provider decides.
pub(super) const MIN_WINDOW_CLAMPED_RESPONSE_TOKENS: u64 = 4_096;

pub(crate) struct HelperMethods;

/// Map a terminal [`crate::RuntimeError`] to the spec §5.2 typed
/// [`agent_core::TurnError`]. This is the SINGLE derivation point: every
/// frontend receives the resulting typed outcome (variant + correlation ID)
/// through `SessionEvent::Error` and must never re-derive it from text.
pub(crate) fn turn_error_for(
    err: &crate::RuntimeError,
    correlation_id: &str,
) -> agent_core::TurnError {
    use crate::RuntimeError as E;
    let provider = |code: &str| agent_core::TurnOutcome::ProviderFailed {
        code: code.to_string(),
        correlation_id: correlation_id.to_string(),
    };
    let outcome = match err {
        E::Canceled => agent_core::TurnOutcome::Canceled,
        E::Api(_) | E::NetworkTimeout(_) => provider("network_error"),
        E::ApiStatus(_) => provider("api_status"),
        E::Auth(_) => provider("auth_error"),
        E::Config(_) => provider("config_error"),
        E::Session(_) => provider("session_error"),
        E::Timeout => provider("timeout"),
        // Tool-loop failures surfaced as stream errors today carry no tool_id
        // (they are protocol/orchestration failures, not a specific tool's
        // result); `TurnOutcome::ToolFailed` is reserved for call-site-typed
        // failures in later budget/ledger work.
        E::Tool(_) => provider("tool_error"),
    };
    agent_core::TurnError {
        message: err.user_message(),
        outcome,
    }
}

/// Whether `HelperMethods::sanitize_thinking_blocks` keeps an assistant
/// block: empty thinking, redacted thinking and text blocks are dropped (an
/// assistant message left with none is dropped whole).
fn sanitation_keeps_block(block: &Value) -> bool {
    match block["type"].as_str() {
        // Text, or an omitted-display block's signature alone (empty text,
        // the reasoning in the signature): the API takes those back. One
        // with neither is a 400 ("each thinking block must contain thinking").
        Some("thinking") => {
            block["thinking"].as_str().is_some_and(|s| !s.is_empty())
                || (block["thinking"].is_string()
                    && block["signature"].as_str().is_some_and(|s| !s.is_empty()))
        }
        Some("redacted_thinking") => block["data"].as_str().is_some_and(|s| !s.is_empty()),
        Some("text") => block["text"].as_str().is_some_and(|s| !s.is_empty()),
        _ => true,
    }
}

/// Whether the conversational cache marker may sit on `block`: durable bytes
/// the next request re-sends, on a block the API accepts `cache_control` on.
/// Never thinking/redacted_thinking, an empty text block, or request-only
/// text (per-turn extension context, a context advisory).
fn tail_markable(block: &Value) -> bool {
    match block["type"].as_str() {
        Some("thinking" | "redacted_thinking") => false,
        Some("text") => {
            block["text"].as_str().is_some_and(|text| !text.is_empty())
                && !super::stream::is_ephemeral_turn_context_block(block)
        }
        _ => true,
    }
}

impl HelperMethods {
    /// Drain all pending steering messages from the channel and inject them
    /// into the conversation as user messages. Returns true if any were injected.
    pub(super) fn drain_steering(
        steering_rx: &mut Option<mpsc::UnboundedReceiver<String>>,
        messages: &mut Vec<SharedMessage>,
        tx: &mpsc::UnboundedSender<StreamEvent>,
    ) -> bool {
        let rx = match steering_rx.as_mut() {
            Some(rx) => rx,
            None => return false,
        };

        let mut injected = false;
        while let Ok(msg) = rx.try_recv() {
            // Metadata only — steering messages are user content and must not
            // reach log sinks (request-lifecycle hardening T1).
            tracing::info!(len = msg.len(), "Steering message injected");
            let _ = tx.send(StreamEvent::Agent(AgentEvent::SteeringDelivered {
                message: msg.clone(),
            }));
            messages.push(Arc::new(json!({"role": "user", "content": msg})));
            injected = true;
        }
        injected
    }

    /// Canonical tool ids named by `tool_reference` blocks anywhere in the
    /// history's tool results (written by `activate_tools`).
    pub(crate) fn referenced_tool_ids(
        messages: &[SharedMessage],
    ) -> std::collections::HashSet<String> {
        let mut ids = std::collections::HashSet::new();
        for msg in messages {
            let Some(blocks) = msg["content"].as_array() else {
                continue;
            };
            for block in blocks {
                if block["type"] != "tool_result" {
                    continue;
                }
                let Some(inner) = block["content"].as_array() else {
                    continue;
                };
                for item in inner {
                    if item["type"] == "tool_reference" {
                        if let Some(name) = item["tool_name"].as_str() {
                            ids.insert(name.to_string());
                        }
                    }
                }
            }
        }
        ids
    }

    /// Put history `tool_reference` blocks into the only shape the Anthropic
    /// wire accepts, deterministically (same history + same tools ⇒ same
    /// bytes, so the message prefix stays cacheable):
    ///
    /// - a `tool_result` holding references may contain ONLY references (a
    ///   text block alongside them is a 400), so the text is dropped;
    /// - every reference must name a tool in this request's `tools` array (an
    ///   unknown name is a 400), so each stored canonical id is resolved
    ///   through `resolve`; unresolvable references are dropped;
    /// - with no resolvable reference left (deferred loading off, a provider
    ///   or model without tool search, a tool since removed), the result keeps
    ///   its non-reference blocks only.
    ///
    /// Only messages that actually carry references are cloned.
    pub(crate) fn wire_tool_references(
        messages: &mut [SharedMessage],
        resolve: &dyn Fn(&str) -> Option<String>,
    ) {
        for msg in messages.iter_mut() {
            let has_reference = msg["content"].as_array().is_some_and(|blocks| {
                blocks.iter().any(|b| {
                    b["type"] == "tool_result"
                        && b["content"].as_array().is_some_and(|inner| {
                            inner.iter().any(|i| i["type"] == "tool_reference")
                        })
                })
            });
            if !has_reference {
                continue;
            }
            let msg = Arc::make_mut(msg);
            let Some(blocks) = msg["content"].as_array_mut() else {
                continue;
            };
            for block in blocks.iter_mut() {
                if block["type"] != "tool_result" {
                    continue;
                }
                let Some(inner) = block["content"].as_array() else {
                    continue;
                };
                if !inner.iter().any(|i| i["type"] == "tool_reference") {
                    continue;
                }
                let mut seen = std::collections::HashSet::new();
                let refs: Vec<Value> = inner
                    .iter()
                    .filter(|i| i["type"] == "tool_reference")
                    .filter_map(|i| i["tool_name"].as_str().and_then(resolve))
                    .filter(|name| seen.insert(name.clone()))
                    .map(|name| json!({"type": "tool_reference", "tool_name": name}))
                    .collect();
                let replacement = if refs.is_empty() {
                    let rest: Vec<Value> = inner
                        .iter()
                        .filter(|i| i["type"] != "tool_reference")
                        .cloned()
                        .collect();
                    if rest.is_empty() {
                        json!("ok")
                    } else {
                        Value::Array(rest)
                    }
                } else {
                    Value::Array(refs)
                };
                block["content"] = replacement;
            }
        }
    }

    /// Strip invalid thinking blocks from assistant messages before sending to the API.
    ///
    /// Anthropic rejects any `{"type": "thinking", ...}` block whose `thinking` field
    /// is missing or empty:
    ///
    /// > messages.N.content.M.thinking: each thinking block must contain thinking
    ///
    /// And rejects empty text blocks:
    ///
    /// > messages: text content blocks must be non-empty
    ///
    /// These can sneak in from (a) older sessions persisted before the streaming
    /// accumulator was hardened, (b) redacted-thinking blocks that lost their data, or
    /// (c) any future provider quirk. We drop them defensively so a single bad block
    /// can't permanently brick a session.
    ///
    /// Algorithm:
    ///   1. For each assistant message, retain only valid (`thinking` non-empty,
    ///      `redacted_thinking` data non-empty, or any other type).
    ///   2. Also drop any text blocks that are empty/whitespace-only — those would
    ///      trigger the "text content blocks must be non-empty" error.
    ///   3. If an assistant message ends up with no content at all, mark it for
    ///      removal — it produced no real output and is not safe to ship as `[]`
    ///      (the API rejects empty content arrays too).
    ///   4. Remove the marked messages, and merge any resulting consecutive
    ///      same-role messages so we don't violate Anthropic's alternation rule.
    pub(super) fn sanitize_thinking_blocks(messages: &mut Vec<SharedMessage>) {
        // Equivalent to the original 3-pass implementation
        // (filter-empty-blocks → drop-empty-assistants → merge-adjacent-same-role),
        // fused into a single forward pass.
        //
        // Correctness vs the 3-pass original (case-by-case):
        //   * Non-assistant messages: untouched in both. Here we don't enter the
        //     retain branch and push the message through. ✓
        //   * Assistant with non-array `content`: original `as_array_mut()` is
        //     `None` → `continue` (no retain, no removal). Here `as_array()`
        //     also `None` → fall through to the merge stage unchanged. ✓
        //   * Assistant whose block filter empties `content`: original marks for
        //     removal in pass 2. Here we `continue` and skip the push. ✓
        //   * Adjacency merge: original walks the post-removal vec and merges
        //     consecutive same-role pairs. Here, each kept message is compared
        //     against `out.last()` (the previous kept message), which IS the
        //     post-removal predecessor — same pairing, same order. ✓
        //
        // Arc-sharing invariant (task #128 slice 3): messages this function does
        // NOT edit are moved through untouched — the output element is literally
        // the same `Arc` allocation as the input element (`Arc::ptr_eq`). Only a
        // message that actually needs a block filtered or a merge applied is
        // deep-cloned, via `Arc::make_mut` (bounded: that one message). In the
        // healthy path this function edits nothing, so the whole pass is free.
        let original = std::mem::take(messages);
        let mut out: Vec<SharedMessage> = Vec::with_capacity(original.len());
        for mut msg in original {
            if msg["role"].as_str() == Some("assistant") {
                if let Some(content) = msg["content"].as_array() {
                    let survivors = content.iter().filter(|b| sanitation_keeps_block(b)).count();
                    if survivors == 0 {
                        // No salvageable content. The API rejects empty content
                        // arrays and empty text placeholders alike, so drop the
                        // whole message (matches Pass 2 of the original).
                        continue;
                    }
                    if survivors < content.len() {
                        // Only clone-on-write when a block actually gets dropped.
                        let m = Arc::make_mut(&mut msg);
                        if let Some(content) = m["content"].as_array_mut() {
                            content.retain(sanitation_keeps_block);
                        }
                    }
                }
            }

            // Adjacency merge: if the previous kept message shares this role,
            // concatenate content blocks rather than push (matches Pass 3 of
            // the original; preserves Anthropic's role-alternation rule).
            if let Some(last) = out.last_mut() {
                if last["role"] == msg["role"] {
                    // Take `msg`'s content without cloning when we hold the only
                    // Arc; otherwise clone just this one message's content.
                    let next_content = match Arc::try_unwrap(msg) {
                        Ok(mut owned) => std::mem::take(&mut owned["content"]),
                        Err(shared) => shared["content"].clone(),
                    };
                    let last = Arc::make_mut(last);
                    let prev_content = std::mem::take(&mut last["content"]);
                    let mut merged = Self::coerce_content_to_blocks(prev_content);
                    merged.extend(Self::coerce_content_to_blocks(next_content));
                    last["content"] = Value::Array(merged);
                    continue;
                }
            }

            out.push(msg);
        }
        *messages = out;
    }

    /// Normalize a `content` value to a Vec of content blocks. Anthropic accepts
    /// either a string or an array; we always want an array for merge operations.
    fn coerce_content_to_blocks(content: Value) -> Vec<Value> {
        match content {
            Value::String(s) if !s.is_empty() => vec![json!({"type": "text", "text": s})],
            Value::String(_) => Vec::new(),
            Value::Array(a) => a,
            _ => Vec::new(),
        }
    }

    /// Append a keep-warm ping message AFTER marker placement, the way the
    /// sanitize adjacency merge would have placed it: a new message after a
    /// different role (the idle ping after a finished reply), else folded
    /// into the last message as trailing blocks (a tool-round replay, whose
    /// prefix ends with the round's user message). Either way it sits after
    /// every marker, so it is an uncached tail that never enters a cache
    /// entry and never moves a breakpoint.
    pub(super) fn append_keepwarm_ping(messages: &mut Vec<SharedMessage>, ping: SharedMessage) {
        let Some(last) = messages.last_mut().filter(|last| last["role"] == ping["role"]) else {
            messages.push(ping);
            return;
        };
        let last = Arc::make_mut(last);
        let mut blocks = Self::coerce_content_to_blocks(std::mem::take(&mut last["content"]));
        blocks.extend(Self::coerce_content_to_blocks(ping["content"].clone()));
        last["content"] = Value::Array(blocks);
    }

    /// Annotate a cache breakpoint on the last message (single-last strategy).
    /// Per S204 benchmarks, single-last matches sliding-4 performance (96–97% hit
    /// vs 96.7%) — a single stationary marker on the most recent message maximizes
    /// the stable cacheable prefix. With no old markers to prune, the
    /// prefix-invalidation bug class is eliminated entirely.
    ///
    /// The marker is the message-tail site: bare 5m under both `FiveMinutes`
    /// and `Hybrid`, `"ttl":"1h"` only under uniform `OneHour`.
    ///
    /// The marker is stamped on the last DURABLE block of the last message:
    /// trailing per-turn extension-context blocks (see `attach_turn_context`)
    /// are skipped, because they exist only in this one request — a cache
    /// entry terminating in them could never be matched by any later request,
    /// which would kill every conversational cache hit in tool-free chat
    /// (#297 follow-up). Mid-message `cache_control` breakpoints are legal;
    /// the ephemeral block rides after the marker as an uncached tail. The
    /// same holds for a request-only context advisory, which the sanitize
    /// adjacency merge folds into the newest user message as a trailing text
    /// block (see `stream::is_ephemeral_turn_context_block`).
    ///
    /// When the last message holds no markable block (an advisory-only user
    /// message, or an empty user text plus an advisory, which the merge
    /// reduces to the advisory alone) the marker walks back to the last
    /// markable block of an earlier message instead of being dropped. It is
    /// never stamped on a thinking block or an empty text block.
    pub(super) fn annotate_cache_breakpoint(messages: &mut [SharedMessage], ttl: CacheTtl) {
        let Some(last) = messages.last_mut() else {
            return;
        };
        // The last message is always edited (marker stamp), so the CoW clone
        // here is the one unavoidable per-request message copy — exactly one
        // message, never the history. Earlier messages are only touched by
        // the walk-back below, which a durable last message never reaches.
        let last = Arc::make_mut(last);

        // Coerce raw string content into a block array so we can attach cache_control.
        if let Some(text) = last["content"].as_str().map(str::to_owned) {
            last["content"] = json!([{"type": "text", "text": text}]);
        }

        let target = messages
            .iter()
            .enumerate()
            .rev()
            .find_map(|(index, message)| match &message["content"] {
                Value::String(text) => {
                    tail_markable(&json!({"type": "text", "text": text})).then_some((index, 0))
                }
                Value::Array(blocks) => blocks
                    .iter()
                    .rposition(tail_markable)
                    .map(|block| (index, block)),
                _ => None,
            });
        let Some((index, block)) = target else {
            return;
        };
        let message = Arc::make_mut(&mut messages[index]);
        if let Some(text) = message["content"].as_str().map(str::to_owned) {
            message["content"] = json!([{"type": "text", "text": text}]);
        }
        message["content"][block]["cache_control"] =
            cache_control_value(ttl, MarkerSite::MessageTail);
    }

    /// Annotate the retention anchor: a message-tail-TTL cache breakpoint on
    /// the last markable block of the last WIRE message strictly before the
    /// one holding `frontier`, the oldest image occurrence the previous
    /// request still sent (see `image_retention::Retention::frontier`).
    /// Nothing ahead of that message differs between the two requests, so
    /// when this request retires images the lookback from the anchor finds the
    /// entry the earlier requests wrote there, instead of every message-level
    /// entry missing (the change sits far more than 20 blocks behind the tail)
    /// and the whole conversation after the system prompt being written
    /// again. Between retirements the anchor stays put and its entry is
    /// re-read. It is never stamped on a thinking block, an empty text block
    /// or per-turn extension context.
    ///
    /// Only the frontier's MESSAGE counts, never its block. A frontier image
    /// in block 1 of a parallel tool-result batch once put the anchor on
    /// block 0 of that same message: those bytes were unchanged, but the
    /// retiring request still missed the entry and wrote the whole history
    /// again (sa_13, 2026-10-30: read 6.8k, wrote 281k). An entry that ends
    /// inside a message whose later blocks change is not one a later request
    /// can rely on, so the anchor always closes a whole, unchanged message.
    /// "Wire message": thinking sanitation merges adjacent same-role messages
    /// (and first drops an assistant message it would empty), so the walk
    /// starts ahead of every message that would merge into the frontier's.
    ///
    /// Run on the outgoing copy after image projection and before thinking
    /// sanitation merges messages (block keys survive the merge). With the
    /// tools, system and tail markers this is at most the API's four
    /// breakpoints, and the anchor never shares the tail's message.
    pub(super) fn annotate_retention_anchor(
        messages: &mut [SharedMessage],
        frontier: Option<(usize, usize)>,
        ttl: CacheTtl,
    ) {
        let Some((frontier_message, _)) = frontier else {
            return;
        };
        if frontier_message >= messages.len() {
            return;
        }
        // Durable history never carries markers (only request copies are
        // stamped). One that arrives with its own (a replayed or foreign
        // request) may already fill the API's four breakpoints with tools,
        // system and tail: no anchor rather than a rejected fifth.
        let carries_marker = |block: &Value| {
            block.get("cache_control").is_some()
                || block["content"]
                    .as_array()
                    .is_some_and(|nested| nested.iter().any(|b| b.get("cache_control").is_some()))
        };
        if messages.iter().any(|message| {
            message["content"]
                .as_array()
                .is_some_and(|blocks| blocks.iter().any(carries_marker))
        }) {
            return;
        }
        let wire_start = Self::wire_message_start(messages, frontier_message);
        let markable = |block: &Value| match block["type"].as_str() {
            Some("text") => {
                block["text"].as_str().is_some_and(|t| !t.is_empty())
                    && !super::stream::is_ephemeral_turn_context_block(block)
            }
            Some("tool_use" | "tool_result") => true,
            _ => false,
        };
        let candidate = (0..wire_start).rev().find_map(|message_index| {
            messages[message_index]["content"]
                .as_array()?
                .iter()
                .rposition(markable)
                .map(|block_index| (message_index, block_index))
        });
        let Some((message_index, block_index)) = candidate else {
            return;
        };
        let message = Arc::make_mut(&mut messages[message_index]);
        message["content"][block_index]["cache_control"] =
            cache_control_value(ttl, MarkerSite::MessageTail);
    }

    /// The first original message that thinking sanitation folds into the
    /// wire message holding `messages[index]`: same-role neighbours merge,
    /// and an assistant message whose every block sanitation drops vanishes
    /// first, letting the messages on either side of it merge.
    fn wire_message_start(messages: &[SharedMessage], index: usize) -> usize {
        let dropped = |message: &Value| {
            message["role"] == "assistant"
                && message["content"]
                    .as_array()
                    .is_some_and(|blocks| !blocks.iter().any(sanitation_keeps_block))
        };
        let role = &messages[index]["role"];
        let mut start = index;
        while start > 0 && (dropped(&messages[start - 1]) || messages[start - 1]["role"] == *role) {
            start -= 1;
        }
        start
    }

    /// Build the `system` value for a request, with the cache marker on the
    /// last block. Shared by both transports so the auth-specific block
    /// layout (and the OAuth identity-first invariant) has exactly one truth.
    ///
    /// - OAuth: Anthropic's fixed Claude Code transport-contract identity block
    ///   FIRST, then Synaps's configurable product identity, the general
    ///   assistant guidance, and optional user system prompt. The transport
    ///   identity is deliberately not configurable. Last block gets the
    ///   stable-prefix marker.
    /// - API key (or any non-oauth): single user-prompt block with the
    ///   stable-prefix marker; `None` when there is no system prompt.
    pub(super) fn build_system_blocks(
        auth_type: &str,
        system_prompt: &Option<String>,
        ttl: CacheTtl,
    ) -> Option<Value> {
        if auth_type == "oauth" {
            // See CLAUDE_CODE_TRANSPORT_IDENTITY: the first block is a wire
            // protocol contract, not Synaps branding or user-configurable identity.
            let mut system_blocks = vec![
                json!({"type": "text", "text": CLAUDE_CODE_TRANSPORT_IDENTITY}),
                json!({"type": "text", "text": crate::core::config::get_identity()}),
                json!({"type": "text", "text": OAUTH_ASSISTANT_GUIDANCE}),
            ];
            if let Some(ref prompt) = system_prompt {
                system_blocks.push(json!({"type": "text", "text": prompt}));
            }
            // Prompt caching: mark the last system block so entire system prompt is cached
            if let Some(last) = system_blocks.last_mut() {
                last["cache_control"] = cache_control_value(ttl, MarkerSite::StablePrefix);
            }
            Some(json!(system_blocks))
        } else {
            system_prompt.as_ref().map(|prompt| {
                json!([
                    {"type": "text", "text": prompt, "cache_control": cache_control_value(ttl, MarkerSite::StablePrefix)}
                ])
            })
        }
    }

    /// Truncate tool results to avoid ballooning message history.
    /// The full result is still sent to the UI — this only caps what goes into
    /// the API messages that are re-sent on every subsequent call.
    ///
    /// `max_bytes` is a BYTE budget enforced at a UTF-8 char boundary via the
    /// shared [`agent_core::BoundedText`] helper (T2; the legacy code applied
    /// a char budget, so multibyte output could exceed the byte cap). Marker
    /// format intentionally changed from "total chars" to "total bytes".
    // Rich (array) tool_result content never passes through here — see
    // stream.rs `select_tool_result_content`.
    pub(crate) fn truncate_tool_result(result: &str, max_bytes: usize) -> String {
        let bounded = agent_core::BoundedText::new(result, max_bytes);
        if !bounded.truncated {
            return bounded.text;
        }
        format!(
            "{}\n\n[truncated — {} total bytes, showing first {}]",
            bounded.text, bounded.original_bytes, bounded.retained_bytes
        )
    }

    /// Returns the max output tokens for a given model.
    /// Opus-class models support 128K, Sonnet/Haiku cap at 64K.
    pub(super) fn max_tokens_for_model(model: &str) -> u64 {
        if model.contains("opus") {
            128000
        } else {
            64000
        }
    }

    /// The `max_tokens` a request may send through a `provider_window`.
    /// Unchanged for models that accept input + `max_tokens` over the window
    /// (Claude 4.5 and later: generation stops at the window); for the
    /// earlier ones that reject it ([`crate::core::models::model_rejects_output_overflow`])
    /// clamped to the room the input leaves. `input_tokens` is evaluated only
    /// then, and must be the conservative estimate, which never understates
    /// the input, so the clamp never overstates the room. Never raised, and
    /// never below `budget_tokens + MIN_WINDOW_CLAMPED_RESPONSE_TOKENS`: the
    /// wire requires `max_tokens > budget_tokens`, and a request with less
    /// room than that is beyond saving by the cap (the provider judges it).
    pub(super) fn fit_max_tokens_to_window(
        model: &str,
        max_tokens: u64,
        provider_window: u64,
        thinking_budget_tokens: Option<u64>,
        input_tokens: impl FnOnce() -> u64,
    ) -> u64 {
        if !crate::core::models::model_rejects_output_overflow(model) {
            return max_tokens;
        }
        let room = provider_window.saturating_sub(input_tokens());
        let floor = thinking_budget_tokens
            .unwrap_or(0)
            .saturating_add(MIN_WINDOW_CLAMPED_RESPONSE_TOKENS);
        max_tokens.min(room.max(floor))
    }

    /// Append a single-line usage record to the per-call log — opt-in via the
    /// `SYNAPS_USAGE_LOG` env var. Silent no-op if unset or set to "0".
    ///
    /// Value semantics:
    /// - unset or "0" or empty → logging disabled
    /// - "1" or "true" → default path `~/.cache/synaps/usage.log`
    /// - anything else → treated as an absolute path
    ///
    /// File is created with mode 0600 to prevent co-located-user snooping
    /// (previous versions wrote to `/tmp/synaps-usage.log` world-readable —
    /// flagged by S172 security review). Errors silently dropped so a broken
    /// log path never breaks the request pipeline.
    pub(super) fn log_usage(input_t: u64, cache_read: u64, cache_create: u64, output_t: u64) {
        let setting = match std::env::var("SYNAPS_USAGE_LOG") {
            Ok(v) if !v.is_empty() && v != "0" => v,
            _ => return,
        };

        let path = if matches!(setting.as_str(), "1" | "true" | "True" | "TRUE") {
            let home = match std::env::var("HOME") {
                Ok(h) => h,
                Err(_) => return,
            };
            format!("{}/.cache/synaps/usage.log", home)
        } else {
            setting
        };

        let total = input_t + cache_read + cache_create;
        let pct = if total > 0 {
            (cache_read as f64 / total as f64 * 100.0) as u32
        } else {
            0
        };

        Self::append_usage_line(
            std::path::Path::new(&path),
            &format!(
                "uncached={} cache_read={} cache_write={} output={} hit={}%",
                input_t, cache_read, cache_create, output_t, pct
            ),
        );
    }

    /// Append one line to the usage log. Best-effort — errors silently
    /// dropped so a broken log path never breaks the request pipeline.
    ///
    /// Delegates to `private_fs`: parent dir 0700, file created 0600 with
    /// O_NOFOLLOW (CWE-59 — a co-located user planting a symlink at a custom
    /// SYNAPS_USAGE_LOG path is refused), broader pre-existing modes repaired.
    pub(super) fn append_usage_line(path: &std::path::Path, line: &str) {
        if let Some(parent) = path.parent() {
            let _ = agent_core::core::private_fs::ensure_private_dir(parent);
        }
        if let Ok(mut f) = agent_core::core::private_fs::open_private_append(path) {
            use std::io::Write;
            let _ = writeln!(f, "{}", line);
        }
    }
}

#[cfg(test)]
mod tests {

    // ── deferred tool loading: history tool_reference → Anthropic wire ──────

    fn activation_result(ids: &[&str]) -> SharedMessage {
        let mut inner = vec![json!({"type": "text", "text": "{\"activated\":1}"})];
        inner.extend(
            ids.iter()
                .map(|id| json!({"type": "tool_reference", "tool_name": id})),
        );
        Arc::new(json!({"role": "user", "content": [
            {"type": "tool_result", "tool_use_id": "toolu_a", "content": inner}
        ]}))
    }

    #[test]
    fn wire_references_resolve_to_wire_names_and_drop_text() {
        let mut msgs = vec![activation_result(&["mcp.srv:echo", "builtin:gone"])];
        let resolve = |id: &str| (id == "mcp.srv:echo").then(|| "ext__srv__echo".to_string());
        HelperMethods::wire_tool_references(&mut msgs, &resolve);
        assert_eq!(
            msgs[0]["content"][0]["content"],
            json!([{"type": "tool_reference", "tool_name": "ext__srv__echo"}]),
            "references only (text alongside is a 400); unresolvable dropped"
        );
    }

    #[test]
    fn wire_references_fall_back_to_text_when_nothing_resolves() {
        let mut msgs = vec![activation_result(&["mcp.srv:echo"])];
        HelperMethods::wire_tool_references(&mut msgs, &|_| None);
        assert_eq!(
            msgs[0]["content"][0]["content"],
            json!([{"type": "text", "text": "{\"activated\":1}"}])
        );
    }

    #[test]
    fn wire_references_are_deterministic_and_leave_other_messages_shared() {
        let plain: SharedMessage = Arc::new(json!({"role": "user", "content": "hi"}));
        let resolve = |id: &str| Some(format!("w_{id}"));
        let mut a = vec![Arc::clone(&plain), activation_result(&["x", "x", "y"])];
        let mut b = a.clone();
        HelperMethods::wire_tool_references(&mut a, &resolve);
        HelperMethods::wire_tool_references(&mut b, &resolve);
        assert_eq!(
            serde_json::to_vec(&a).unwrap(),
            serde_json::to_vec(&b).unwrap()
        );
        assert!(
            Arc::ptr_eq(&a[0], &plain),
            "messages without references are not cloned"
        );
        assert_eq!(
            a[1]["content"][0]["content"].as_array().unwrap().len(),
            2,
            "deduplicated"
        );
    }

    #[test]
    fn referenced_tool_ids_reads_only_tool_result_references() {
        let msgs = vec![
            activation_result(&["mcp.srv:echo"]),
            Arc::new(
                json!({"role": "user", "content": [{"type": "tool_reference", "tool_name": "not_in_a_result"}]}),
            ),
        ];
        let ids = HelperMethods::referenced_tool_ids(&msgs);
        assert_eq!(ids.len(), 1);
        assert!(ids.contains("mcp.srv:echo"));
    }

    #[test]
    fn marked_tools_put_deferred_after_the_marked_last_loaded_tool() {
        let loaded = vec![json!({"name": "bash"}), json!({"name": "activate_tools"})];
        let deferred = vec![json!({"name": "ext__srv__echo", "defer_loading": true})];
        let v = serde_json::to_value(crate::runtime::request::MarkedTools::with_deferred(
            &loaded,
            &deferred,
            CacheTtl::OneHour,
        ))
        .unwrap();
        assert_eq!(v.as_array().unwrap().len(), 3);
        assert!(v[0].get("cache_control").is_none());
        assert_eq!(
            v[1]["cache_control"]["ttl"], "1h",
            "marker on the last LOADED tool"
        );
        assert!(
            v[2].get("cache_control").is_none(),
            "a deferred tool may not carry cache_control"
        );
        assert_eq!(v[2]["defer_loading"], true);
        let legacy = serde_json::to_vec(&crate::runtime::request::MarkedTools::new(
            &loaded,
            CacheTtl::OneHour,
        ))
        .unwrap();
        let empty = serde_json::to_vec(&crate::runtime::request::MarkedTools::with_deferred(
            &loaded,
            &[],
            CacheTtl::OneHour,
        ))
        .unwrap();
        assert_eq!(legacy, empty, "no deferred tools ⇒ byte-identical body");
    }

    use super::*;
    use crate::core::config::CacheTtl;
    use serde_json::json;

    /// Mechanical port shims (task #128 slice 3): the helpers now operate on
    /// `[SharedMessage]`. These wrap the original `Vec<Value>` fixtures into
    /// Arcs, run the real function, and unwrap back so every assertion below
    /// stays byte-identical to the pre-Arc test suite.
    fn sanitize_vals(msgs: &mut Vec<Value>) {
        let mut shared: Vec<SharedMessage> = msgs.drain(..).map(Arc::new).collect();
        HelperMethods::sanitize_thinking_blocks(&mut shared);
        *msgs = shared
            .into_iter()
            .map(|m| Arc::try_unwrap(m).unwrap_or_else(|a| (*a).clone()))
            .collect();
    }

    // ─── Retention anchor ────────────────────────────────────────────────

    fn anchor_fixture() -> Vec<SharedMessage> {
        vec![
            Arc::new(json!({"role":"user","content":[{"type":"text","text":"go"}]})),
            Arc::new(json!({"role":"assistant","content":[
                {"type":"thinking","thinking":"t","signature":"s"},
                {"type":"tool_use","id":"a","name":"shot","input":{}}
            ]})),
            Arc::new(json!({"role":"user","content":[
                {"type":"tool_result","tool_use_id":"a","content":"ok"},
                {"type":"image","source":{"type":"base64","media_type":"image/png","data":"AA=="}}
            ]})),
            Arc::new(json!({"role":"assistant","content":[
                {"type":"text","text":""},
                {"type":"thinking","thinking":"t","signature":"s"}
            ]})),
            Arc::new(json!({"role":"user","content":[
                {"type":"image","source":{"type":"base64","media_type":"image/png","data":"AA=="}}
            ]})),
        ]
    }

    /// Every `(message, block)` carrying `cache_control`, in order.
    fn marked(messages: &[SharedMessage]) -> Vec<(usize, usize)> {
        let mut out = Vec::new();
        for (m, message) in messages.iter().enumerate() {
            for (b, block) in message["content"]
                .as_array()
                .into_iter()
                .flatten()
                .enumerate()
            {
                if block.get("cache_control").is_some() {
                    out.push((m, b));
                }
            }
        }
        out
    }

    #[test]
    fn retention_anchor_marks_the_last_markable_block_before_the_frontier() {
        let ttl = CacheTtl::FiveMinutes;
        // Frontier on a message's first block: back into the previous
        // message, onto its tool_use, never its thinking block.
        let mut out = anchor_fixture();
        HelperMethods::annotate_retention_anchor(&mut out, Some((2, 0)), ttl);
        assert_eq!(marked(&out), [(1, 1)]);
        assert_eq!(
            out[1]["content"][1]["cache_control"],
            json!({"type":"ephemeral"})
        );
        // Frontier mid-message (an image in block 1 of a parallel batch):
        // still the previous message, never an earlier block of the
        // frontier's own message, whose later bytes are about to change.
        let mut out = anchor_fixture();
        HelperMethods::annotate_retention_anchor(&mut out, Some((2, 1)), ttl);
        assert_eq!(marked(&out), [(1, 1)]);
        // An assistant message holding only an empty text and a thinking
        // block offers nothing: the scan keeps going back.
        let mut out = anchor_fixture();
        HelperMethods::annotate_retention_anchor(&mut out, Some((4, 0)), ttl);
        assert_eq!(marked(&out), [(2, 0)]);
    }

    /// The anchor closes a whole WIRE message: thinking sanitation merges a
    /// same-role neighbour (or two user messages around an assistant message
    /// it empties) into the frontier's message, so the anchor goes ahead of
    /// all of them. Checked on the sanitized body, as the API sees it.
    #[test]
    fn retention_anchor_precedes_every_message_merged_into_the_frontiers() {
        let ttl = CacheTtl::FiveMinutes;
        let image = json!({"type":"image","source":{"type":"base64","media_type":"image/png","data":"AA=="}});
        let head = || -> Vec<SharedMessage> {
            vec![
                Arc::new(json!({"role":"user","content":[{"type":"text","text":"go"}]})),
                Arc::new(json!({"role":"assistant","content":[
                    {"type":"tool_use","id":"a","name":"shot","input":{}}
                ]})),
                Arc::new(json!({"role":"user","content":[
                    {"type":"tool_result","tool_use_id":"a","content":"ok"}
                ]})),
            ]
        };
        let mut adjacent = head();
        adjacent.push(Arc::new(json!({"role":"user","content":[image.clone()]})));
        let mut emptied = head();
        emptied.push(Arc::new(
            json!({"role":"assistant","content":[{"type":"text","text":""}]}),
        ));
        emptied.push(Arc::new(json!({"role":"user","content":[image.clone()]})));
        for mut out in [adjacent, emptied] {
            let frontier = out.len() - 1;
            HelperMethods::annotate_retention_anchor(&mut out, Some((frontier, 0)), ttl);
            assert_eq!(marked(&out), [(1, 0)]);
            let mut wire = out.clone();
            HelperMethods::sanitize_thinking_blocks(&mut wire);
            let anchored = wire
                .iter()
                .position(|m| {
                    m["content"]
                        .as_array()
                        .unwrap()
                        .iter()
                        .any(|b| b.get("cache_control").is_some())
                })
                .unwrap();
            let framed = wire
                .iter()
                .position(|m| {
                    m["content"]
                        .as_array()
                        .unwrap()
                        .iter()
                        .any(|b| b["type"] == "image")
                })
                .unwrap();
            assert!(
                anchored < framed,
                "anchor {anchored} must precede the frontier's wire message {framed}"
            );
        }
    }

    #[test]
    fn retention_anchor_skips_per_turn_context_and_is_a_no_op_without_a_target() {
        let ttl = CacheTtl::FiveMinutes;
        let context = format!(
            "{}\nambient\n{}",
            super::super::stream::EXTENSION_CONTEXT_OPEN,
            super::super::stream::EXTENSION_CONTEXT_CLOSE
        );
        let mut out: Vec<SharedMessage> = vec![
            Arc::new(json!({"role":"user","content":[
                {"type":"text","text":"look"},
                {"type":"text","text":context}
            ]})),
            Arc::new(json!({"role":"assistant","content":[
                {"type":"thinking","thinking":"t","signature":"s"}
            ]})),
            Arc::new(json!({"role":"user","content":[
                {"type":"image","source":{"type":"base64","media_type":"image/png","data":"AA=="}}
            ]})),
        ];
        HelperMethods::annotate_retention_anchor(&mut out, Some((2, 0)), ttl);
        assert_eq!(marked(&out), [(0, 0)]);
        // No frontier, a frontier past the history, nothing markable ahead:
        // the messages (and their Arcs) are untouched.
        let original = anchor_fixture();
        for frontier in [None, Some((9, 0)), Some((0, 0))] {
            let mut out = original.clone();
            HelperMethods::annotate_retention_anchor(&mut out, frontier, ttl);
            assert!(marked(&out).is_empty(), "{frontier:?}");
            assert!(out.iter().zip(&original).all(|(a, b)| Arc::ptr_eq(a, b)));
        }
        // A history that arrives carrying its own marker, top level or nested
        // in a tool_result, gets no anchor: never a fifth breakpoint.
        for (message, path) in [(0, "/content/0"), (2, "/content/0")] {
            let mut foreign = anchor_fixture();
            let block = Arc::make_mut(&mut foreign[message])
                .pointer_mut(path)
                .unwrap();
            if message == 2 {
                block["content"] = json!([{"type":"text","text":"ok",
                    "cache_control":{"type":"ephemeral"}}]);
            } else {
                block["cache_control"] = json!({"type":"ephemeral"});
            }
            let before = foreign.clone();
            HelperMethods::annotate_retention_anchor(&mut foreign, Some((4, 0)), ttl);
            assert!(
                foreign.iter().zip(&before).all(|(a, b)| Arc::ptr_eq(a, b)),
                "{message}"
            );
        }
    }

    /// The anchor is a message-tail site: it carries the tail's TTL in every
    /// mode, so a hybrid body keeps every 1h marker ahead of the 5m ones.
    #[test]
    fn retention_anchor_uses_the_message_tail_ttl_and_shares_the_tail_block() {
        for (ttl, expect) in [
            (CacheTtl::FiveMinutes, json!({"type":"ephemeral"})),
            (CacheTtl::Hybrid, json!({"type":"ephemeral"})),
            (CacheTtl::OneHour, json!({"type":"ephemeral","ttl":"1h"})),
        ] {
            let mut out = anchor_fixture();
            HelperMethods::annotate_retention_anchor(&mut out, Some((2, 1)), ttl);
            assert_eq!(out[1]["content"][1]["cache_control"], expect, "{ttl:?}");
        }
        // A frontier in the only (tail) message leaves nothing ahead of it:
        // the tail marker alone, never an anchor inside the tail message.
        let mut out: Vec<SharedMessage> = vec![Arc::new(json!({"role":"user","content":[
            {"type":"text","text":"look"},
            {"type":"image","source":{"type":"base64","media_type":"image/png","data":"AA=="}}
        ]}))];
        HelperMethods::annotate_retention_anchor(&mut out, Some((0, 1)), CacheTtl::FiveMinutes);
        HelperMethods::annotate_cache_breakpoint(&mut out, CacheTtl::FiveMinutes);
        assert_eq!(marked(&out), [(0, 1)]);
    }

    fn annotate_vals(msgs: &mut Vec<Value>, ttl: CacheTtl) {
        let mut shared: Vec<SharedMessage> = msgs.drain(..).map(Arc::new).collect();
        HelperMethods::annotate_cache_breakpoint(&mut shared, ttl);
        *msgs = shared
            .into_iter()
            .map(|m| Arc::try_unwrap(m).unwrap_or_else(|a| (*a).clone()))
            .collect();
    }

    /// Arc-sharing proof (task #128 slice 3): running the cleaned-messages
    /// pipeline (share → sanitize no-op → annotate last) must NOT copy any
    /// message it didn't edit. Every untouched message stays literally the
    /// same allocation (`Arc::ptr_eq`); only the annotated last message
    /// diverges — and the input side of it remains unmodified.
    #[test]
    fn cleaned_pipeline_shares_untouched_messages_ptr_eq() {
        let input: Vec<SharedMessage> = vec![
            Arc::new(json!({"role": "user", "content": "one"})),
            Arc::new(json!({"role": "assistant", "content": [{"type": "text", "text": "two"}]})),
            Arc::new(json!({"role": "user", "content": "three"})),
            Arc::new(json!({"role": "assistant", "content": [{"type": "text", "text": "four"}]})),
            Arc::new(json!({"role": "user", "content": "five"})),
        ];

        // Mirror of the request path: cleaned = share, sanitize, annotate.
        let mut cleaned = input.to_vec();
        HelperMethods::sanitize_thinking_blocks(&mut cleaned);
        assert_eq!(cleaned.len(), input.len(), "sanitize must be a no-op here");
        for (i, (a, b)) in input.iter().zip(cleaned.iter()).enumerate() {
            assert!(
                Arc::ptr_eq(a, b),
                "message {i} must be shared (same Arc) after a no-op sanitize"
            );
        }

        HelperMethods::annotate_cache_breakpoint(&mut cleaned, CacheTtl::FiveMinutes);
        for (i, (a, b)) in input.iter().zip(cleaned.iter()).enumerate() {
            if i < input.len() - 1 {
                assert!(
                    Arc::ptr_eq(a, b),
                    "untouched message {i} must remain the same Arc allocation"
                );
            } else {
                assert!(
                    !Arc::ptr_eq(a, b),
                    "annotated last message must be a diverged copy"
                );
            }
        }

        // The edit was CoW: the input's last message is untouched...
        assert!(input[4]["content"].is_string());
        assert!(input[4]["content"].as_str() == Some("five"));
        // ...while the cleaned copy carries the marker.
        assert_eq!(
            cleaned[4]["content"][0]["cache_control"]["type"],
            "ephemeral"
        );
    }

    /// Finding-1 freeze (S241 gate review): the OLD Vec-era sanitize used
    /// `msg["content"].as_array_mut()`, whose IndexMut *inserted* a
    /// `"content": null` key into assistant messages that lacked one (bytes
    /// the API rejected anyway). The Arc port reads immutably and leaves the
    /// message untouched. This test freezes the NEW behavior so nobody
    /// "fixes" it back to the accidental insertion.
    #[test]
    fn sanitize_leaves_missing_content_key_absent() {
        let mut msgs = vec![json!({"role": "assistant"})];
        let before = serde_json::to_string(&msgs[0]).unwrap();
        sanitize_vals(&mut msgs);
        let after = serde_json::to_string(&msgs[0]).unwrap();
        assert_eq!(
            before, after,
            "message without content key must pass through byte-identical"
        );
        assert!(
            msgs[0].get("content").is_none(),
            "content key must NOT be inserted"
        );
    }

    /// Empty text and no signature: the API's 400, so dropped.
    #[test]
    fn sanitize_drops_empty_thinking_blocks() {
        let mut msgs = vec![json!({
            "role": "assistant",
            "content": [
                {"type": "thinking", "thinking": "", "signature": ""},
                {"type": "text", "text": "hello"},
            ]
        })];
        sanitize_vals(&mut msgs);
        let content = msgs[0]["content"].as_array().unwrap();
        assert_eq!(content.len(), 1);
        assert_eq!(content[0]["type"], "text");
    }

    /// `display: "omitted"`: empty text, the reasoning in the signature.
    /// Kept unchanged, the history's own `Arc` (nothing edited).
    #[test]
    fn sanitize_keeps_omitted_thinking_blocks_unchanged() {
        let block = json!({"type": "thinking", "thinking": "", "signature": "sig1"});
        let msg = Arc::new(json!({
            "role": "assistant",
            "content": [block.clone(), {"type": "tool_use", "id": "t", "name": "x", "input": {}}]
        }));
        let mut msgs = vec![Arc::clone(&msg)];
        HelperMethods::sanitize_thinking_blocks(&mut msgs);
        assert!(Arc::ptr_eq(&msgs[0], &msg), "untouched");
        assert_eq!(msgs[0]["content"][0], block);
    }

    #[test]
    fn sanitize_keeps_non_empty_thinking_blocks() {
        let mut msgs = vec![json!({
            "role": "assistant",
            "content": [
                {"type": "thinking", "thinking": "reasoning here", "signature": "sig1"},
                {"type": "text", "text": "hello"},
            ]
        })];
        sanitize_vals(&mut msgs);
        assert_eq!(msgs[0]["content"].as_array().unwrap().len(), 2);
    }

    #[test]
    fn sanitize_drops_thinking_with_missing_field() {
        let mut msgs = vec![json!({
            "role": "assistant",
            "content": [
                {"type": "thinking", "signature": "sig1"},
                {"type": "text", "text": "hello"},
            ]
        })];
        sanitize_vals(&mut msgs);
        let content = msgs[0]["content"].as_array().unwrap();
        assert_eq!(content.len(), 1);
        assert_eq!(content[0]["type"], "text");
    }

    #[test]
    fn sanitize_replaces_emptied_content_with_placeholder() {
        let mut msgs = vec![
            json!({"role": "user", "content": "first"}),
            json!({
                "role": "assistant",
                "content": [
                    {"type": "thinking", "thinking": "", "signature": ""},
                ]
            }),
            json!({"role": "user", "content": "second"}),
        ];
        sanitize_vals(&mut msgs);
        // Empty assistant message must be dropped entirely (cannot be turned into
        // an empty text block — the API rejects those too).
        // The two surrounding user messages must then be merged.
        assert_eq!(msgs.len(), 1);
        assert_eq!(msgs[0]["role"], "user");
        let content = msgs[0]["content"].as_array().unwrap();
        assert_eq!(content.len(), 2);
        assert_eq!(content[0]["text"], "first");
        assert_eq!(content[1]["text"], "second");
    }

    #[test]
    fn sanitize_drops_empty_text_blocks() {
        let mut msgs = vec![json!({
            "role": "assistant",
            "content": [
                {"type": "text", "text": ""},
                {"type": "text", "text": "real content"},
            ]
        })];
        sanitize_vals(&mut msgs);
        let content = msgs[0]["content"].as_array().unwrap();
        assert_eq!(content.len(), 1);
        assert_eq!(content[0]["text"], "real content");
    }

    #[test]
    fn sanitize_merges_consecutive_user_messages_after_drop() {
        let mut msgs = vec![
            json!({"role": "user", "content": [{"type": "text", "text": "a"}]}),
            json!({"role": "assistant", "content": [{"type": "thinking", "thinking": ""}]}),
            json!({"role": "user", "content": [{"type": "text", "text": "b"}]}),
            json!({"role": "assistant", "content": [{"type": "text", "text": "ok"}]}),
        ];
        sanitize_vals(&mut msgs);
        assert_eq!(msgs.len(), 2);
        assert_eq!(msgs[0]["role"], "user");
        assert_eq!(msgs[0]["content"].as_array().unwrap().len(), 2);
        assert_eq!(msgs[1]["role"], "assistant");
    }

    #[test]
    fn sanitize_preserves_alternation_when_no_drops_needed() {
        let mut msgs = vec![
            json!({"role": "user", "content": "a"}),
            json!({"role": "assistant", "content": [{"type": "text", "text": "b"}]}),
            json!({"role": "user", "content": "c"}),
        ];
        sanitize_vals(&mut msgs);
        assert_eq!(msgs.len(), 3);
    }

    #[test]
    fn sanitize_skips_user_messages() {
        let mut msgs = vec![json!({
            "role": "user",
            "content": [
                {"type": "thinking", "thinking": "", "signature": "sig1"},
            ]
        })];
        sanitize_vals(&mut msgs);
        // We only police assistant messages — user messages would be malformed for
        // a different reason and aren't ours to rewrite.
        assert_eq!(msgs[0]["content"].as_array().unwrap().len(), 1);
    }

    #[test]
    fn sanitize_drops_redacted_thinking_with_empty_data() {
        let mut msgs = vec![json!({
            "role": "assistant",
            "content": [
                {"type": "redacted_thinking", "data": ""},
                {"type": "text", "text": "hi"},
            ]
        })];
        sanitize_vals(&mut msgs);
        let content = msgs[0]["content"].as_array().unwrap();
        assert_eq!(content.len(), 1);
        assert_eq!(content[0]["type"], "text");
    }

    // --- annotate_cache_breakpoint (single-last strategy) ---

    fn has_marker(msg: &Value) -> bool {
        msg["content"]
            .as_array()
            .map(|c| c.iter().any(|b| b.get("cache_control").is_some()))
            .unwrap_or(false)
    }

    #[test]
    fn cache_empty_messages_is_noop() {
        let mut msgs: Vec<Value> = vec![];
        annotate_vals(&mut msgs, CacheTtl::FiveMinutes);
        assert!(msgs.is_empty());
    }

    #[test]
    fn cache_single_user_string_content_coerced_and_marked() {
        let mut msgs = vec![json!({"role": "user", "content": "hello"})];
        annotate_vals(&mut msgs, CacheTtl::FiveMinutes);
        let content = msgs[0]["content"]
            .as_array()
            .expect("coerced to block array");
        assert_eq!(content.len(), 1);
        assert_eq!(content[0]["type"], "text");
        assert_eq!(content[0]["text"], "hello");
        assert_eq!(content[0]["cache_control"]["type"], "ephemeral");
    }

    #[test]
    fn cache_only_last_message_gets_marker() {
        let mut msgs = vec![
            json!({"role": "user", "content": "one"}),
            json!({"role": "assistant", "content": [{"type": "text", "text": "two"}]}),
            json!({"role": "user", "content": "three"}),
            json!({"role": "assistant", "content": [{"type": "text", "text": "four"}]}),
            json!({"role": "user", "content": "five"}),
        ];
        annotate_vals(&mut msgs, CacheTtl::FiveMinutes);
        for msg in &msgs[..4] {
            assert!(
                !has_marker(msg),
                "earlier message must not have cache_control"
            );
        }
        assert!(has_marker(&msgs[4]));
        // Earlier string contents must remain untouched strings.
        assert!(msgs[0]["content"].is_string());
        assert!(msgs[2]["content"].is_string());
    }

    #[test]
    fn cache_marks_trailing_assistant_message() {
        let mut msgs = vec![
            json!({"role": "user", "content": "question"}),
            json!({"role": "assistant", "content": [{"type": "text", "text": "answer"}]}),
        ];
        annotate_vals(&mut msgs, CacheTtl::FiveMinutes);
        assert!(!has_marker(&msgs[0]));
        assert!(has_marker(&msgs[1]), "single-last marks ANY trailing role");
        assert_eq!(msgs[1]["content"][0]["cache_control"]["type"], "ephemeral");
    }

    #[test]
    fn cache_only_final_block_of_multi_block_content_marked() {
        let mut msgs = vec![json!({
            "role": "user",
            "content": [
                {"type": "text", "text": "first"},
                {"type": "text", "text": "second"},
                {"type": "text", "text": "third"},
            ]
        })];
        annotate_vals(&mut msgs, CacheTtl::FiveMinutes);
        let content = msgs[0]["content"].as_array().unwrap();
        assert!(content[0].get("cache_control").is_none());
        assert!(content[1].get("cache_control").is_none());
        assert_eq!(content[2]["cache_control"]["type"], "ephemeral");
    }

    // ── cache_control_value: exact-string matrix (spec §5) ─────────────────
    //
    // THE most important property of the whole cache_ttl change: the default
    // (FiveMinutes) emits exactly `{"type":"ephemeral"}` — single key, no ttl
    // field — at EVERY site, so default-mode requests are byte-identical to
    // the previous release and can never invalidate existing cached prefixes.

    #[test]
    fn ccv_five_minutes_is_bare_ephemeral_exact_bytes_at_every_site() {
        for site in [MarkerSite::StablePrefix, MarkerSite::MessageTail] {
            let v = cache_control_value(CacheTtl::FiveMinutes, site);
            assert_eq!(
                serde_json::to_string(&v).unwrap(),
                r#"{"type":"ephemeral"}"#,
                "5m must serialize byte-identically to the legacy literal at {site:?}"
            );
        }
    }

    // NOTE: serde_json (without preserve_order) sorts object keys, so the
    // exact serialized form is {"ttl":"1h","type":"ephemeral"}. JSON object
    // key order is semantically irrelevant to the API; what matters is the
    // 5m form is the single-key legacy literal and 1h carries ttl:"1h".
    #[test]
    fn ccv_one_hour_carries_ttl_at_every_site() {
        for site in [MarkerSite::StablePrefix, MarkerSite::MessageTail] {
            let v = cache_control_value(CacheTtl::OneHour, site);
            assert_eq!(
                serde_json::to_string(&v).unwrap(),
                r#"{"ttl":"1h","type":"ephemeral"}"#,
                "1h must carry ttl at {site:?}"
            );
        }
    }

    #[test]
    fn ccv_hybrid_splits_by_site() {
        assert_eq!(
            serde_json::to_string(&cache_control_value(
                CacheTtl::Hybrid,
                MarkerSite::StablePrefix
            ))
            .unwrap(),
            r#"{"ttl":"1h","type":"ephemeral"}"#,
        );
        assert_eq!(
            serde_json::to_string(&cache_control_value(
                CacheTtl::Hybrid,
                MarkerSite::MessageTail
            ))
            .unwrap(),
            r#"{"type":"ephemeral"}"#,
        );
    }

    // ── annotate_cache_breakpoint TTL behavior ──────────────────────────────

    #[test]
    fn cache_breakpoint_5m_has_no_ttl_key() {
        let mut msgs = vec![json!({"role": "user", "content": "hello"})];
        annotate_vals(&mut msgs, CacheTtl::FiveMinutes);
        let cc = &msgs[0]["content"][0]["cache_control"];
        assert_eq!(cc["type"], "ephemeral");
        assert!(
            cc.get("ttl").is_none(),
            "5m must not emit a ttl key (assert absence)"
        );
        assert_eq!(cc.as_object().unwrap().len(), 1, "exactly one key: type");
    }

    #[test]
    fn cache_breakpoint_1h_emits_ttl() {
        let mut msgs = vec![json!({"role": "user", "content": "hello"})];
        annotate_vals(&mut msgs, CacheTtl::OneHour);
        let cc = &msgs[0]["content"][0]["cache_control"];
        assert_eq!(cc["type"], "ephemeral");
        assert_eq!(cc["ttl"], "1h");
    }

    #[test]
    fn cache_breakpoint_hybrid_message_tail_has_no_ttl_key() {
        // Hybrid's message-tail marker is the 5m one — bare ephemeral.
        let mut msgs = vec![json!({"role": "user", "content": "hello"})];
        annotate_vals(&mut msgs, CacheTtl::Hybrid);
        let cc = &msgs[0]["content"][0]["cache_control"];
        assert_eq!(cc["type"], "ephemeral");
        assert!(cc.get("ttl").is_none());
    }

    #[test]
    fn cache_breakpoint_coercion_and_single_last_unchanged_under_all_modes() {
        for ttl in [CacheTtl::FiveMinutes, CacheTtl::OneHour, CacheTtl::Hybrid] {
            // String coercion still happens.
            let mut msgs = vec![
                json!({"role": "user", "content": "one"}),
                json!({"role": "user", "content": [
                    {"type": "text", "text": "a"},
                    {"type": "text", "text": "b"},
                ]}),
            ];
            annotate_vals(&mut msgs, ttl);
            assert!(
                !has_marker(&msgs[0]),
                "earlier message unmarked under {ttl:?}"
            );
            assert!(
                msgs[0]["content"].is_string(),
                "earlier string content untouched"
            );
            let content = msgs[1]["content"].as_array().unwrap();
            assert!(
                content[0].get("cache_control").is_none(),
                "only final block marked"
            );
            assert!(content[1].get("cache_control").is_some());
        }
    }

    // ── last-tool marker (stable-prefix site, both transports) ──────────────
    // Legacy `mark_last_tool` mutated the assembled body; #128 Slice 4 replaced
    // it with `request::MarkedTools`, which stamps a clone of the last tool at
    // serialization time. Same assertions, pre-assembly form.

    fn tools_list() -> Vec<Value> {
        vec![
            json!({"name": "bash", "input_schema": {}}),
            json!({"name": "read", "input_schema": {}}),
        ]
    }

    fn marked_tools_value(tools: &[Value], ttl: CacheTtl) -> Value {
        serde_json::to_value(crate::runtime::request::MarkedTools::new(tools, ttl)).unwrap()
    }

    #[test]
    fn mark_last_tool_5m_is_bare_ephemeral() {
        let tools = tools_list();
        let marked = marked_tools_value(&tools, CacheTtl::FiveMinutes);
        assert!(marked[0].get("cache_control").is_none());
        assert_eq!(
            serde_json::to_string(&marked[1]["cache_control"]).unwrap(),
            r#"{"type":"ephemeral"}"#,
        );
        // Input schemas are borrowed, never mutated.
        assert!(tools[1].get("cache_control").is_none());
    }

    #[test]
    fn mark_last_tool_1h_and_hybrid_carry_ttl() {
        for ttl in [CacheTtl::OneHour, CacheTtl::Hybrid] {
            let marked = marked_tools_value(&tools_list(), ttl);
            assert_eq!(marked[1]["cache_control"]["ttl"], "1h", "under {ttl:?}");
        }
    }

    #[test]
    fn mark_last_tool_no_tools_is_noop() {
        assert_eq!(marked_tools_value(&[], CacheTtl::OneHour), json!([]));
    }

    // ── build_system_blocks (OAuth + API-key sites, both transports) ────────

    #[test]
    fn system_blocks_oauth_transport_identity_first_and_marker_on_last() {
        const CLAUDE_CODE_IDENTITY: &str =
            "You are Claude Code, Anthropic's official CLI for Claude.";
        // Pinned, not latched: `config::IDENTITY` is process-wide and set by
        // whichever test first loads a config, so read ambiently this test
        // depended on test order and on the developer's real HOME.
        crate::core::config::with_identity_for_tests(crate::core::config::DEFAULT_IDENTITY, || {
            for ttl in [CacheTtl::FiveMinutes, CacheTtl::OneHour, CacheTtl::Hybrid] {
                let prompt = Some("custom prompt".to_string());
                let system = HelperMethods::build_system_blocks("oauth", &prompt, ttl).unwrap();
                let blocks = system.as_array().unwrap();
                assert_eq!(blocks.len(), 4);
                // OAuth's protocol identity is fixed and FIRST. Product/user identity
                // remains additive and therefore cannot replace the transport contract.
                assert_eq!(blocks[0]["text"], CLAUDE_CODE_IDENTITY);
                // The configured identity rides after it ...
                assert_eq!(blocks[1]["text"], crate::core::config::DEFAULT_IDENTITY);
                // ... and out of the box it is the product identity, not a second
                // copy of the transport line.
                assert_ne!(blocks[0]["text"], blocks[1]["text"]);
                assert!(blocks[0].get("cache_control").is_none());
                assert!(blocks[1].get("cache_control").is_none());
                assert!(blocks[2].get("cache_control").is_none());
                // Last block carries the stable-prefix marker.
                let cc = &blocks[3]["cache_control"];
                assert_eq!(cc["type"], "ephemeral");
                match ttl {
                    CacheTtl::FiveMinutes => assert!(cc.get("ttl").is_none(), "5m: no ttl key"),
                    _ => assert_eq!(cc["ttl"], "1h"),
                }
            }
        });
        // Configuring the transport line itself as the identity is valid: it
        // is sent twice, transport copy still first.
        crate::core::config::with_identity_for_tests(CLAUDE_CODE_IDENTITY, || {
            let system = HelperMethods::build_system_blocks("oauth", &None, CacheTtl::FiveMinutes).unwrap();
            let blocks = system.as_array().unwrap();
            assert_eq!(blocks[0]["text"], CLAUDE_CODE_IDENTITY);
            assert_eq!(blocks[1]["text"], CLAUDE_CODE_IDENTITY);
        });
    }

    #[test]
    fn system_blocks_oauth_5m_marker_exact_bytes() {
        let system =
            HelperMethods::build_system_blocks("oauth", &None, CacheTtl::FiveMinutes).unwrap();
        let last = system.as_array().unwrap().last().unwrap().clone();
        assert_eq!(
            serde_json::to_string(&last["cache_control"]).unwrap(),
            r#"{"type":"ephemeral"}"#,
        );
    }

    #[test]
    fn system_blocks_api_key_single_block() {
        for ttl in [CacheTtl::FiveMinutes, CacheTtl::OneHour, CacheTtl::Hybrid] {
            let prompt = Some("sys".to_string());
            let system = HelperMethods::build_system_blocks("api_key", &prompt, ttl).unwrap();
            let blocks = system.as_array().unwrap();
            assert_eq!(blocks.len(), 1);
            assert_eq!(blocks[0]["text"], "sys");
            let cc = &blocks[0]["cache_control"];
            match ttl {
                CacheTtl::FiveMinutes => assert_eq!(
                    serde_json::to_string(cc).unwrap(),
                    r#"{"type":"ephemeral"}"#,
                ),
                _ => assert_eq!(cc["ttl"], "1h"),
            }
        }
    }

    #[test]
    fn system_blocks_api_key_without_prompt_is_none() {
        assert!(HelperMethods::build_system_blocks("api_key", &None, CacheTtl::OneHour).is_none());
    }

    // ── Hybrid ordering invariant (spec §3.3 / §5) ──────────────────────────

    /// Build a request the way both transports do (tools marked, system
    /// built, message tail annotated), then walk markers in the API's
    /// logical emission order (tools → system → messages) and assert every
    /// 1h marker precedes the 5m tail marker.
    #[test]
    fn hybrid_ordering_invariant_1h_prefix_precedes_5m_tail() {
        let ttl = CacheTtl::Hybrid;
        let mut body = json!({ "tools": marked_tools_value(&tools_list(), ttl) });
        let prompt = Some("sys".to_string());
        body["system"] = HelperMethods::build_system_blocks("oauth", &prompt, ttl).unwrap();
        let mut messages = vec![json!({"role": "user", "content": "hi"})];
        annotate_vals(&mut messages, ttl);
        body["messages"] = json!(messages);

        // Collect (logical_order, ttl_str) for every marker in the body.
        let mut markers: Vec<&str> = Vec::new();
        for tool in body["tools"].as_array().unwrap() {
            if let Some(cc) = tool.get("cache_control") {
                markers.push(cc.get("ttl").and_then(|t| t.as_str()).unwrap_or("5m"));
            }
        }
        for block in body["system"].as_array().unwrap() {
            if let Some(cc) = block.get("cache_control") {
                markers.push(cc.get("ttl").and_then(|t| t.as_str()).unwrap_or("5m"));
            }
        }
        for msg in body["messages"].as_array().unwrap() {
            if let Some(arr) = msg["content"].as_array() {
                for block in arr {
                    if let Some(cc) = block.get("cache_control") {
                        markers.push(cc.get("ttl").and_then(|t| t.as_str()).unwrap_or("5m"));
                    }
                }
            }
        }

        assert_eq!(
            markers,
            vec!["1h", "1h", "5m"],
            "tool 1h, system 1h, tail 5m"
        );
        // Ordering rule: once a 5m marker appears, no 1h marker may follow.
        let first_5m = markers.iter().position(|m| *m == "5m").unwrap();
        assert!(
            markers[first_5m..].iter().all(|m| *m == "5m"),
            "every 1h marker must precede the 5m tail"
        );
        // And the hybrid tail marker carries NO ttl key at all.
        let tail_cc = &body["messages"][0]["content"][0]["cache_control"];
        assert!(tail_cc.get("ttl").is_none());
    }

    /// Full-body default-mode check: with FiveMinutes, every marker in a
    /// fully constructed request serializes to exactly the legacy literal.
    #[test]
    fn default_mode_full_body_markers_byte_identical_to_legacy() {
        let ttl = CacheTtl::FiveMinutes;
        let body = json!({ "tools": marked_tools_value(&tools_list(), ttl) });
        let prompt = Some("sys".to_string());
        for auth in ["oauth", "api_key"] {
            let system = HelperMethods::build_system_blocks(auth, &prompt, ttl).unwrap();
            for block in system.as_array().unwrap() {
                if let Some(cc) = block.get("cache_control") {
                    assert_eq!(
                        serde_json::to_string(cc).unwrap(),
                        r#"{"type":"ephemeral"}"#
                    );
                }
            }
        }
        let mut messages = vec![json!({"role": "user", "content": "hi"})];
        annotate_vals(&mut messages, ttl);
        assert_eq!(
            serde_json::to_string(&messages[0]["content"][0]["cache_control"]).unwrap(),
            r#"{"type":"ephemeral"}"#,
        );
        assert_eq!(
            serde_json::to_string(&body["tools"][1]["cache_control"]).unwrap(),
            r#"{"type":"ephemeral"}"#,
        );
    }

    /// Private-mode test for the usage log (spec §5.4). Umask isolation:
    /// `#[serial(umask)]` serializes umask-mutating tests crate-wide; the
    /// guard restores the previous process-global mask on drop (panic-safe).
    #[cfg(unix)]
    mod usage_private_modes {
        use super::*;
        use serial_test::serial;
        use std::os::unix::fs::PermissionsExt;

        struct UmaskGuard {
            old: libc::mode_t,
        }
        impl UmaskGuard {
            fn set(mask: libc::mode_t) -> Self {
                Self {
                    old: unsafe { libc::umask(mask) },
                }
            }
        }
        impl Drop for UmaskGuard {
            fn drop(&mut self) {
                unsafe {
                    libc::umask(self.old);
                }
            }
        }

        fn mode_of(path: &std::path::Path) -> u32 {
            std::fs::metadata(path).unwrap().permissions().mode() & 0o777
        }

        #[test]
        #[serial(umask)]
        fn usage_line_creates_0600_file_and_0700_dir_under_permissive_umask() {
            let _umask = UmaskGuard::set(0);
            let tmp = tempfile::TempDir::new().unwrap();
            let path = tmp.path().join("synaps").join("usage.log");
            HelperMethods::append_usage_line(&path, "uncached=1");
            assert_eq!(
                mode_of(path.parent().unwrap()),
                0o700,
                "usage dir must be 0700"
            );
            assert_eq!(mode_of(&path), 0o600, "usage file must be 0600");
        }
    }
}