synaps-engine 0.9.0

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"}),
    }
}

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(_) => 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.to_string(),
        outcome,
    }
}

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
    }

    /// 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.
        fn block_is_valid(block: &Value) -> bool {
            match block["type"].as_str() {
                Some("thinking") => block["thinking"]
                    .as_str()
                    .map(|s| !s.is_empty())
                    .unwrap_or(false),
                Some("redacted_thinking") => block["data"]
                    .as_str()
                    .map(|s| !s.is_empty())
                    .unwrap_or(false),
                Some("text") => block["text"]
                    .as_str()
                    .map(|s| !s.is_empty())
                    .unwrap_or(false),
                _ => true,
            }
        }

        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| block_is_valid(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(block_is_valid);
                        }
                    }
                }
            }

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

    /// 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.
    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 never touched.
        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}]);
        }

        if let Some(block) = last["content"].as_array_mut().and_then(|c| {
            c.iter_mut()
                .rev()
                .find(|b| !super::stream::is_ephemeral_turn_context_block(b))
        }) {
            block["cache_control"] = cache_control_value(ttl, MarkerSite::MessageTail);
        }
    }

    /// 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" {
            // Anthropic OAuth routes on this first system block. It is a wire
            // protocol contract, not Synaps branding or user-configurable identity.
            const CLAUDE_CODE_TRANSPORT_IDENTITY: &str =
                "You are Claude Code, Anthropic's official CLI for Claude.";
            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": "You are a helpful AI assistant with access to tools. Use them when needed."}),
            ];
            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".
    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
        }
    }

    /// 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 {
    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();
    }

    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"
        );
    }

    #[test]
    fn sanitize_drops_empty_thinking_blocks() {
        let mut msgs = vec![json!({
            "role": "assistant",
            "content": [
                {"type": "thinking", "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_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": "sig1"},
                ]
            }),
            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.";
        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);
            assert_eq!(blocks[1]["text"], crate::core::config::get_identity());
            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"),
            }
        }
    }

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