supercode-harness 0.4.6

//! Acceptance tests for SPEC.md B7 (prompt caching on the imported prefix,
//! slimmed per D9): message-level `cache_control` breakpoints applied at the
//! request-build site, never touching `Agent::history` or the sidecar.
//!
//! `provider.rs`'s own unit tests cover `apply_cache_plan` and
//! `Usage::prompt_tokens_details` as pure functions (AC1, AC5); this file
//! covers the two things that only show up once an `Agent` is driving the
//! loop: (1) history/transcript purity across a real `send` (AC2), (2) prefix
//! stability across two turns through a recorder provider (AC3), and (3) the
//! coordination clamp that keeps legacy compaction out of the imported prefix
//! (AC4) — same recorder-provider idiom as `reduce_loop.rs`.

use std::collections::VecDeque;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};

use async_trait::async_trait;
use supercode_harness::session::Session;
use supercode_harness::{
    Agent, AgentEvent, CachePlan, ChatMessage, ChatRequest, Config, EventSink, PromptTokensDetails,
    Provider, Usage,
};

fn fixture(name: &str) -> PathBuf {
    Path::new(env!("CARGO_MANIFEST_DIR"))
        .join("tests/fixtures")
        .join(name)
}

fn load_codex() -> Session {
    Session::from_codex(fixture("codex_session.jsonl")).unwrap()
}

fn temp_dir(tag: &str) -> PathBuf {
    let dir =
        std::env::temp_dir().join(format!("supercode-cache-plan-{tag}-{}", std::process::id()));
    std::fs::create_dir_all(&dir).unwrap();
    dir
}

/// Answers plainly (no tool calls) every turn, recording the full request
/// message list it was sent — the same idiom `reduce_loop.rs`'s
/// `PlainAnswerCapturing` uses.
struct PlainAnswerCapturing {
    calls: AtomicUsize,
    requests: Arc<Mutex<Vec<Vec<ChatMessage>>>>,
}
#[async_trait]
impl Provider for PlainAnswerCapturing {
    async fn complete(
        &self,
        req: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode_harness::Result<(ChatMessage, Usage)> {
        let n = self.calls.fetch_add(1, Ordering::SeqCst);
        self.requests.lock().unwrap().push(req.messages.clone());
        Ok((
            ChatMessage::assistant(format!("reply {n}")),
            Usage::default(),
        ))
    }
}

/// AC2: after a `send` under `CachePlan::ImportedPrefix`, neither
/// `agent.history()` nor `save_transcript`'s output ever contain
/// `"cache_control"` — the annotation lives only in the cloned request view
/// (`provider::apply_cache_plan`), never in what's retained/persisted. A
/// sanity check confirms the feature is actually active: the REQUEST the
/// provider received does carry `cache_control`.
#[tokio::test]
async fn history_and_transcript_never_carry_cache_control() {
    let dir = temp_dir("purity");
    let session = load_codex();

    let requests = Arc::new(Mutex::new(Vec::new()));
    let config = Config::builder()
        .cwd(dir.clone())
        .cache_plan(CachePlan::ImportedPrefix)
        .build();
    let mut agent = Agent::with_provider(
        config,
        Box::new(PlainAnswerCapturing {
            calls: AtomicUsize::new(0),
            requests: requests.clone(),
        }),
    );
    agent.load_session(session);

    agent.send("continue where we left off").await.unwrap();

    let history_json = serde_json::to_string(agent.history()).unwrap();
    assert!(
        !history_json.contains("cache_control"),
        "agent.history() must never carry cache_control"
    );

    let transcript_path = dir.join("transcript.jsonl");
    agent.save_transcript(&transcript_path).unwrap();
    let transcript = std::fs::read_to_string(&transcript_path).unwrap();
    assert!(
        !transcript.contains("cache_control"),
        "save_transcript output must never carry cache_control"
    );

    // Sanity: the request the provider actually received WAS annotated —
    // otherwise this test would pass vacuously.
    let reqs = requests.lock().unwrap();
    let last_req_json = serde_json::to_string(reqs.last().unwrap()).unwrap();
    assert!(
        last_req_json.contains("cache_control"),
        "sanity: the outgoing request must carry the cache breakpoints"
    );

    std::fs::remove_dir_all(&dir).ok();
}

/// AC3: the cache-annotated prefix serializes byte-identically across two
/// separate `send` calls (the cache-hit precondition — a single differing
/// byte busts the whole cache) — catches nondeterministic annotation.
#[tokio::test]
async fn cache_annotated_prefix_is_byte_identical_across_two_sends() {
    let session = load_codex();
    let requests = Arc::new(Mutex::new(Vec::new()));
    let config = Config::builder()
        .cache_plan(CachePlan::ImportedPrefix)
        .build();
    let mut agent = Agent::with_provider(
        config,
        Box::new(PlainAnswerCapturing {
            calls: AtomicUsize::new(0),
            requests: requests.clone(),
        }),
    );
    agent.load_session(session);
    let prefix_len = agent
        .imported_prefix_len()
        .expect("load_session must set imported_prefix_len");

    agent.send("turn one").await.unwrap();
    agent.send("turn two").await.unwrap();

    let reqs = requests.lock().unwrap().clone();
    assert_eq!(reqs.len(), 2, "one request per send (no tool calls)");
    assert!(reqs[0].len() >= prefix_len);
    assert!(reqs[1].len() >= prefix_len);

    let prefix1 = serde_json::to_string(&reqs[0][..prefix_len]).unwrap();
    let prefix2 = serde_json::to_string(&reqs[1][..prefix_len]).unwrap();
    assert_eq!(
        prefix1, prefix2,
        "the annotated imported prefix must be byte-identical across turns"
    );
    assert!(
        prefix1.contains("cache_control"),
        "sanity: the compared prefix must actually carry the breakpoints"
    );
}

/// AC4: legacy (no-`ReductionPolicy`) compaction — the only other mechanism
/// that can shrink/rewrite `history` — must never cross into the imported
/// prefix once `CachePlan::ImportedPrefix` is active, however many turns are
/// sent afterward.
#[tokio::test]
async fn legacy_compaction_never_crosses_into_the_imported_prefix() {
    let session = load_codex();
    let requests = Arc::new(Mutex::new(Vec::new()));
    let config = Config::builder()
        .cache_plan(CachePlan::ImportedPrefix)
        .compact_after_messages(4)
        .build();
    let mut agent = Agent::with_provider(
        config,
        Box::new(PlainAnswerCapturing {
            calls: AtomicUsize::new(0),
            requests: requests.clone(),
        }),
    );
    agent.load_session(session);
    let prefix_len = agent
        .imported_prefix_len()
        .expect("load_session must set imported_prefix_len");
    let prefix_before = serde_json::to_string(&agent.history()[..prefix_len]).unwrap();

    // Enough turns that, without the clamp, `compact_after_messages(4)`'s
    // small `keep_recent` window would repeatedly want to collapse well past
    // the imported prefix.
    for i in 0..20 {
        agent.send(format!("turn {i}")).await.unwrap();
    }

    // The clamp actually engaged and shrank *something* (proves this isn't
    // a vacuous "no compaction ever ran" pass): far fewer messages survive
    // than the raw imported-prefix + 20*(user, assistant) total would be.
    let raw_total = prefix_len + 20 * 2;
    assert!(
        agent.history().len() < raw_total,
        "legacy compaction must have fired at least once: {} vs uncompacted {raw_total}",
        agent.history().len()
    );

    // The clamp held: the imported prefix itself is untouched, byte-for-byte.
    assert!(agent.history().len() >= prefix_len);
    let prefix_after = serde_json::to_string(&agent.history()[..prefix_len]).unwrap();
    assert_eq!(
        prefix_before, prefix_after,
        "history[0..imported_prefix_len] must be unchanged by legacy compaction"
    );
}

// ---- UX-26 (B7-warn): cache-cold warning, end-to-end through a real Agent -

/// Answers plainly and returns a caller-scripted [`Usage`] per call (queued
/// oldest-first; the last entry repeats once the queue is exhausted) — lets
/// a test control exactly what the "provider" reports for
/// `prompt_tokens_details.cached_tokens` on each turn, the one piece
/// `provider::cache_cold_reason`'s ratio check needs and that can't be
/// faked any other way (no live network call in this test suite).
struct ScriptedUsage {
    calls: AtomicUsize,
    usages: Mutex<VecDeque<Usage>>,
}
#[async_trait]
impl Provider for ScriptedUsage {
    async fn complete(
        &self,
        _req: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode_harness::Result<(ChatMessage, Usage)> {
        let n = self.calls.fetch_add(1, Ordering::SeqCst);
        let mut q = self.usages.lock().unwrap();
        let usage = if q.len() > 1 {
            q.pop_front().unwrap()
        } else {
            q.front().cloned().unwrap_or_default()
        };
        Ok((ChatMessage::assistant(format!("reply {n}")), usage))
    }
}

fn warm_usage() -> Usage {
    Usage {
        prompt_tokens: 1000,
        completion_tokens: 20,
        total_tokens: 1020,
        prompt_tokens_details: Some(PromptTokensDetails { cached_tokens: 950 }),
    }
}

fn cold_usage() -> Usage {
    Usage {
        prompt_tokens: 1000,
        completion_tokens: 20,
        total_tokens: 1020,
        prompt_tokens_details: Some(PromptTokensDetails { cached_tokens: 3 }),
    }
}

/// Attach an event sink that collects every [`AgentEvent::CacheWarning`]
/// message emitted, in order.
fn capture_cache_warnings() -> (EventSink, Arc<Mutex<Vec<String>>>) {
    let messages = Arc::new(Mutex::new(Vec::new()));
    let out = messages.clone();
    let sink: EventSink = Box::new(move |event| {
        if let AgentEvent::CacheWarning { message } = event {
            out.lock().unwrap().push(message);
        }
    });
    (sink, messages)
}

/// dev/01, flagship case: `codex_session.jsonl`'s last message is timestamped
/// months before "now" (any real clock past mid-2026) — far beyond
/// Anthropic's 5-minute ephemeral-cache TTL. Resuming it and sending the
/// VERY FIRST turn in this process (nothing established here yet) must still
/// warn: the whole point is catching "I stepped away and came back to a
/// stale session," not just turn 2+ of a live interactive session. Scripted
/// with `cold_usage` (not `warm_usage`, see the T1 test right below for
/// that half) so this stays a clean test of "no disproof available → the
/// idle-clock verdict stands."
#[tokio::test]
async fn cache_warning_fires_on_first_turn_of_a_resumed_idle_session() {
    let session = load_codex();
    let (sink, warnings) = capture_cache_warnings();
    let config = Config::builder()
        .cache_plan(CachePlan::ImportedPrefix)
        .event_sink(sink)
        .build();
    let mut agent = Agent::with_provider(
        config,
        Box::new(ScriptedUsage {
            calls: AtomicUsize::new(0),
            usages: Mutex::new(VecDeque::from([cold_usage()])),
        }),
    );
    agent.load_session(session);

    agent.send("continue where we left off").await.unwrap();

    let seen = warnings.lock().unwrap();
    assert_eq!(seen.len(), 1, "exactly one warning on the one turn sent");
    assert!(
        seen[0].contains("cache likely cold"),
        "must be the TTL/Stale reason, not a ratio miss: {}",
        seen[0]
    );
}

/// UX-26 T1 (accuracy fold-in — FAILS pre-fix): end-to-end proof of the
/// exact false positive T1 targets, through a real `Agent`/`run_loop`, not
/// just the pure `cache_cold_reason` predicate. Same months-old resumed
/// session as the test above (idle_secs derived from it is unambiguously
/// past the TTL — the "sibling process re-resumed the same session file"
/// scenario), but this time the scripted `usage` for that very turn reports
/// a near-100% cache-read ratio, exactly as if a sibling process had just
/// warmed the identical prefix. The idle-clock signal alone would fire
/// `Stale`; the turn's own usage disproves it — no warning should print.
#[tokio::test]
async fn cache_warning_suppressed_on_resumed_idle_session_when_usage_proves_warm() {
    let session = load_codex();
    let (sink, warnings) = capture_cache_warnings();
    let config = Config::builder()
        .cache_plan(CachePlan::ImportedPrefix)
        .event_sink(sink)
        .build();
    let mut agent = Agent::with_provider(
        config,
        Box::new(ScriptedUsage {
            calls: AtomicUsize::new(0),
            usages: Mutex::new(VecDeque::from([warm_usage()])),
        }),
    );
    agent.load_session(session);

    agent.send("continue where we left off").await.unwrap();

    assert!(
        warnings.lock().unwrap().is_empty(),
        "usage proving a warm cache-read ratio must suppress the idle-clock Stale warning: {:?}",
        warnings.lock().unwrap()
    );
}

/// dev/02: back-to-back turns sent moments apart (well inside the TTL) with
/// the provider reporting a warm cache-read ratio print no warning on the
/// second turn — no false positive on the common interactive case. (The
/// FIRST turn may or may not warn depending on the fixture's own age — this
/// test only asserts about the turn AFTER establishment.)
#[tokio::test]
async fn cache_warning_silent_on_warm_back_to_back_turn() {
    let session = load_codex();
    let (sink, warnings) = capture_cache_warnings();
    let config = Config::builder()
        .cache_plan(CachePlan::ImportedPrefix)
        .event_sink(sink)
        .build();
    let mut agent = Agent::with_provider(
        config,
        Box::new(ScriptedUsage {
            calls: AtomicUsize::new(0),
            usages: Mutex::new(VecDeque::from([warm_usage(), warm_usage()])),
        }),
    );
    agent.load_session(session);

    agent.send("turn one").await.unwrap(); // establishes the cache entry
    warnings.lock().unwrap().clear(); // only turn two's verdict matters here
    agent.send("turn two").await.unwrap(); // sent immediately after: idle << TTL

    assert!(
        warnings.lock().unwrap().is_empty(),
        "a warm, immediate second turn must not warn: {:?}",
        warnings.lock().unwrap()
    );
}

/// dev/01, ratio branch, end-to-end: turn two is sent immediately after turn
/// one (idle time is milliseconds, nowhere near the 5-minute TTL) but the
/// scripted provider reports a near-zero cache-read ratio on it — an
/// unexpected miss despite reuse being expected (turn one already
/// established the entry).
#[tokio::test]
async fn cache_warning_fires_unexpected_miss_inside_ttl() {
    let session = load_codex();
    let (sink, warnings) = capture_cache_warnings();
    let config = Config::builder()
        .cache_plan(CachePlan::ImportedPrefix)
        .event_sink(sink)
        .build();
    let mut agent = Agent::with_provider(
        config,
        Box::new(ScriptedUsage {
            calls: AtomicUsize::new(0),
            usages: Mutex::new(VecDeque::from([warm_usage(), cold_usage()])),
        }),
    );
    agent.load_session(session);

    agent.send("turn one").await.unwrap(); // establishes
    warnings.lock().unwrap().clear();
    agent.send("turn two").await.unwrap(); // scripted as a cold-ratio response

    let seen = warnings.lock().unwrap();
    assert_eq!(seen.len(), 1);
    assert!(
        seen[0].contains("unexpected cache miss"),
        "must be the ratio/Miss reason, not TTL/Stale: {}",
        seen[0]
    );
}

/// dev/03: `Config::cache_warnings(false)` suppresses the warning even under
/// conditions that would otherwise unambiguously fire it (the same idle,
/// months-old resumed session as the flagship test above). Scripted with
/// `cold_usage` — proven by the flagship test to fire unconditionally
/// without the flag — so suppression here is unambiguously attributable to
/// the flag, not incidentally to T1's usage-disproof branch.
#[tokio::test]
async fn cache_warning_suppressed_by_config_flag() {
    let session = load_codex();
    let (sink, warnings) = capture_cache_warnings();
    let config = Config::builder()
        .cache_plan(CachePlan::ImportedPrefix)
        .cache_warnings(false)
        .event_sink(sink)
        .build();
    let mut agent = Agent::with_provider(
        config,
        Box::new(ScriptedUsage {
            calls: AtomicUsize::new(0),
            usages: Mutex::new(VecDeque::from([cold_usage()])),
        }),
    );
    agent.load_session(session);

    agent.send("continue where we left off").await.unwrap();

    assert!(
        warnings.lock().unwrap().is_empty(),
        "cache_warnings(false) must suppress even an otherwise-firing warning"
    );
}

// ---- UX-26 T2 (accuracy fold-in): Anthropic-family scoping -----------------

/// UX-26 T2 (FAILS pre-fix): under a NON-Anthropic model, the Anthropic-TTL-
/// shaped `Stale` warning must not fire even though every other precondition
/// is identical to the flagship dev/01 test (same idle, months-old resumed
/// session, same `cold_usage` scripted response that provably fires the
/// warning under the default Anthropic model). `openai/gpt-5` is a real
/// OpenRouter slug from `KNOWN_MODEL_CONTEXT_LIMITS` — a genuinely
/// non-Anthropic model this binary can resolve and send requests for.
#[tokio::test]
async fn cache_warning_silent_for_non_anthropic_model_on_resumed_idle_session() {
    let session = load_codex();
    let (sink, warnings) = capture_cache_warnings();
    let config = Config::builder()
        .model("openai/gpt-5")
        .cache_plan(CachePlan::ImportedPrefix)
        .event_sink(sink)
        .build();
    let mut agent = Agent::with_provider(
        config,
        Box::new(ScriptedUsage {
            calls: AtomicUsize::new(0),
            usages: Mutex::new(VecDeque::from([cold_usage()])),
        }),
    );
    agent.load_session(session);

    agent.send("continue where we left off").await.unwrap();

    assert!(
        warnings.lock().unwrap().is_empty(),
        "a non-Anthropic model must never print the Anthropic-TTL-shaped warning: {:?}",
        warnings.lock().unwrap()
    );
}

/// UX-26 T2 sanity: the SAME scenario as the test above, but with an
/// explicit `anthropic/…` model, still fires — proving the gate above is
/// actually model-selective (narrows only non-Anthropic), not a blanket
/// regression of the primary, correct Anthropic case.
#[tokio::test]
async fn cache_warning_still_fires_for_explicit_anthropic_model() {
    let session = load_codex();
    let (sink, warnings) = capture_cache_warnings();
    let config = Config::builder()
        .model("anthropic/claude-sonnet-4-6")
        .cache_plan(CachePlan::ImportedPrefix)
        .event_sink(sink)
        .build();
    let mut agent = Agent::with_provider(
        config,
        Box::new(ScriptedUsage {
            calls: AtomicUsize::new(0),
            usages: Mutex::new(VecDeque::from([cold_usage()])),
        }),
    );
    agent.load_session(session);

    agent.send("continue where we left off").await.unwrap();

    let seen = warnings.lock().unwrap();
    assert_eq!(
        seen.len(),
        1,
        "explicit Anthropic-family model must still warn"
    );
    assert!(seen[0].contains("cache likely cold"));
}