supercode-harness 0.4.13

//! FLAGSHIP end-to-end proof (feature 3): "continue a session losslessly
//! WITH massive token reduction" — the rate-limit rescue.
//!
//! This is the single scripted, measured, re-runnable demonstration of the
//! whole reduction stack's actual point: a real session gets big enough that
//! a provider starts rejecting it (a 429/rate-limit/"too many tokens"
//! error), the session is RESCUED by entering reduced mode (installing
//! [`ReductionPolicy::default`] — the full technique stack: A7 truncation,
//! TR-2 dedup, TR-3 diff-rereads, TR-4 ANSI/redraw normalization, TR-6
//! supersession, TR-10 tool-input elision) and handed to a DIFFERENT
//! provider at drastically reduced cost, continues correctly there, and —
//! the non-negotiable other half of the claim — nothing is actually lost:
//! `verify_log`/`invert`/`expand_reduction`/`export_session` all round-trip
//! byte-exact from a full disk reload, the same guarantor-path idiom
//! `tr9_handoff_guarantor.rs`/`tr2_dedup_guarantor.rs`/
//! `tr6_supersede_guarantor.rs`/`tr10_input_guarantor.rs` already established
//! (live `Agent` -> recorder(disk sidecar) -> reload from disk -> offline
//! `verify_log`/`invert`/`expand_reduction`).
//!
//! No live network/API call anywhere: two small, deterministic, in-process
//! scripted [`Provider`]s stand in for "the rate-limited account" (provider
//! A) and "the rescue provider" (provider B); a scripted `bash` [`Tool`]
//! stands in for real `cargo test`/`docker pull` runs (reusing
//! `tests/fixtures/terminal/cargo_build.raw` and `docker_pull.raw` — GENUINE
//! captures, per `reduce_normalize.rs`'s own provenance note — as the raw
//! ANSI payloads); `read_file`/`write_file`/`list_dir`/`search` are the REAL
//! built-in tools, operating on a real temp directory, so file re-reads
//! genuinely reflect what was genuinely written moments before. The
//! reduction engine (`reduce::project_messages`/`invert`/`verify_log`/
//! `rehydrate::expand_reduction`) and the token estimator
//! (`tokens::estimate_view_tokens`) are exactly the production code paths —
//! nothing here is a stand-in for those.
//!
//! Emits an inspectable artifact at `target/e2e-rescue-demo.json`
//! (git-ignored, same convention as `reduce_summaries_demo.rs`'s TR-7 dev/06
//! demo and `rehydrate.rs`'s TR-1 dev/06 demo) and prints the headline
//! reduction ratio via `println!` so a human running the test sees the
//! proof number directly.
//!
//! # Scope, plainly stated (skeptic-panel hardening pass)
//!
//! - **`tokens_before`/`tokens_after` count MESSAGE PAYLOAD only.**
//!   `tokens::estimate_view_tokens` walks `req.messages`; it does not add the
//!   `tools` JSON-schema block a live provider request also carries on the
//!   wire. This omission is SYMMETRIC (both the "before" and "after" figures,
//!   and the fixed rate-limit quota they're compared against, are all
//!   message-payload-only), so the measured 79.4%-class reduction ratio and
//!   the quota-crossing narrative are not distorted by it either way — but
//!   the absolute token counts printed/emitted here are not the full wire
//!   request size a real provider would bill for.
//! - **The fixture is CURATED, not a statistically-average session.** It was
//!   hand-assembled to exercise every reduction kind in one pass (repeated
//!   test reruns for TR-2 dedup/TR-6 supersession, a small-file re-read for
//!   TR-3 diffing, a large one for A7 truncation, genuine ANSI captures for
//!   TR-4 normalization, oversized write payloads for TR-10 elision) — all
//!   realistic redundancy patterns an actual debugging session produces, but
//!   deliberately packed together rather than sampled from one. The measured
//!   reduction percentage is therefore an ACHIEVABLE end-to-end figure that
//!   demonstrates the mechanism working across every kind at once, not a
//!   claim about the average reduction a random session sees.
//! - **Provider B's reply is scripted**, exactly like `tr9_handoff_guarantor.rs`/
//!   `tr2_dedup_guarantor.rs`/`tr6_supersede_guarantor.rs`/
//!   `tr10_input_guarantor.rs`'s scripted providers: it proves the
//!   cross-provider CONTINUATION PLUMBING works end-to-end on a reduced
//!   context (request built, sent, reply threaded back onto the same
//!   sidecar) — it is not evidence that a live model produces a correct
//!   answer from reduced context.

use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicUsize, Ordering};

use async_trait::async_trait;
use supercode_harness::reduce::rehydrate::expand_reduction;
use supercode_harness::reduce::{self, ReductionKind, ReductionLog, ReductionPolicy};
use supercode_harness::session::{Session, SessionFormat};
use supercode_harness::sidecar::SidecarWriter;
use supercode_harness::store::SessionStore;
use supercode_harness::tokens::{estimate_view_tokens, fmt_approx_tokens};
use supercode_harness::tools::{Tool, ToolContext, ToolRegistry};
use supercode_harness::{
    Agent, ChatMessage, ChatRequest, Config, Error, FunctionCall, Provider, Role, ToolCall, Usage,
};

fn temp_dir(tag: &str) -> PathBuf {
    static N: AtomicUsize = AtomicUsize::new(0);
    let dir = std::env::temp_dir().join(format!(
        "supercode-e2e-rescue-{tag}-{}-{}",
        std::process::id(),
        N.fetch_add(1, Ordering::SeqCst)
    ));
    std::fs::create_dir_all(&dir).unwrap();
    dir
}

fn fixtures_dir() -> PathBuf {
    Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/terminal")
}

fn load_raw(name: &str) -> String {
    std::fs::read_to_string(fixtures_dir().join(format!("{name}.raw")))
        .unwrap_or_else(|e| panic!("reading {name}.raw: {e}"))
}

// ---------------------------------------------------------------------------
// Realistic fixture content builders
// ---------------------------------------------------------------------------

fn repeat_lines(prefix: &str, n: usize) -> String {
    let mut s = String::new();
    for i in 0..n {
        s.push_str(&format!(
            "{prefix} line {i:04} - lorem ipsum dolor sit amet consectetur adipiscing elit\n"
        ));
    }
    s
}

/// A synthetic (but realistically-shaped) `src/lib.rs`: ~80 lines of context
/// either side of one function whose body is the only thing that changes
/// between versions — a genuine "one-line edit in a big file" re-read, the
/// exact shape TR-3's diff-rereads targets (a unified diff of one changed
/// line is tiny relative to the ~11KB whole file).
fn lib_rs_source(fix_line: &str) -> String {
    format!(
        "//! src/lib.rs -- arithmetic helpers under active investigation\n\n{}\npub fn compute(x: i32) -> i32 {{\n    {fix_line}\n}}\n\n{}\n",
        repeat_lines("context", 80),
        repeat_lines("trailer", 80),
    )
}

/// A SMALLER synthetic file (deliberately kept under the default 8,192-byte
/// A7 `tool_output_trigger_bytes`) dedicated to proving TR-3's diff-rereads
/// in isolation: `src/lib.rs` above is deliberately large enough that A7
/// claims its re-reads FIRST (oversized-truncation candidates are claimed
/// before TR-3 ever gets a look, by design — see `reduce.rs`'s A8/TR-3
/// section doc comment: "Runs after A7 so a message already claimed as an
/// oversized-truncation candidate this run is never also elided/diffed
/// here"), so a re-read that must land on `FileReadDiffed` needs its own,
/// smaller file.
fn config_rs_source(fix_line: &str) -> String {
    format!(
        "//! src/config.rs -- tunable constants\n\n{}\npub const FACTOR: i32 = {{\n    {fix_line}\n}};\n\n{}\n",
        repeat_lines("context", 35),
        repeat_lines("trailer", 35),
    )
}

fn notes_md(n: usize) -> String {
    let mut s = String::from("# Investigation Notes\n\n");
    for i in 0..n {
        s.push_str(&format!(
            "- item {i:04}: verified behavior around compute() edge case reproduction step, all clear\n"
        ));
    }
    s.push_str("\nFINAL-NOTES-MARKER-e2e-rescue-9f21\n");
    s
}

fn cargo_output(base: &str, run_marker: &str) -> String {
    format!("$ cargo test\n{base}\n{run_marker}\n")
}

/// Provider A's quota once it has "tightened" — a FIXED, INDEPENDENT limit
/// chosen up front (a realistic account-tier ceiling), deliberately NOT
/// derived from `tokens_before`. This is what makes the rejection an organic
/// boundary crossing rather than a tautology: the fixture's full request
/// (~51.9K measured tokens) genuinely exceeds this pre-existing number, and
/// the reduced view (~10.7K measured tokens) genuinely lands back under it —
/// both with real margin, verified below rather than assumed.
const PROVIDER_A_QUOTA_TOKENS: u64 = 25_000;

// ---------------------------------------------------------------------------
// Fake tools
// ---------------------------------------------------------------------------

/// A scripted `bash` tool: never spawns a real subprocess (memory-constrained
/// box, no live network) — returns the next canned output from `outputs` on
/// each call, in call order. The ASSISTANT tool_call message's own
/// `arguments.command` (constructed by [`ScriptedTurns`] below) is what the
/// reduction layer's TR-6 supersession pass actually keys off of, not
/// anything this tool inspects.
struct ScriptedBashTool {
    outputs: Vec<String>,
    calls: AtomicUsize,
}
#[async_trait]
impl Tool for ScriptedBashTool {
    fn name(&self) -> &str {
        "bash"
    }
    fn description(&self) -> &str {
        "Execute a shell command (scripted fake for the e2e rescue demo)."
    }
    fn parameters(&self) -> serde_json::Value {
        serde_json::json!({
            "type": "object",
            "properties": {"command": {"type": "string"}},
            "required": ["command"],
        })
    }
    async fn execute(
        &self,
        _args: serde_json::Value,
        _ctx: &ToolContext,
    ) -> supercode_harness::Result<String> {
        let n = self.calls.fetch_add(1, Ordering::SeqCst);
        Ok(self
            .outputs
            .get(n)
            .cloned()
            .unwrap_or_else(|| "(no more scripted bash output)".to_string()))
    }
}

// ---------------------------------------------------------------------------
// Scripted providers
// ---------------------------------------------------------------------------

enum Step {
    /// One assistant turn issuing a single tool call.
    Call {
        id: &'static str,
        tool: &'static str,
        arguments: String,
    },
    /// A plain-text assistant reply ending the current `send()`.
    Text(&'static str),
}

/// Assert every wire-relevant field of `restored` matches `original`,
/// byte-exact. `ChatMessage`/`ToolCall`/`FunctionCall` don't derive
/// `PartialEq` in production, so this compares field-by-field (this is a
/// TEST-ONLY helper — it never touches production source). Critically, this
/// covers `tool_calls` (and therefore `tool_calls[..].function.arguments`,
/// where TR-10's `ToolInputElided` restoration actually writes the recovered
/// bytes) in addition to `content`/`content_parts`/`tool_call_id`/`name` — a
/// `.content`-only comparison would vacuously pass on every scripted
/// tool-call message here, since those all have `content: None`.
fn assert_message_byte_exact(restored: &ChatMessage, original: &ChatMessage, context: &str) {
    assert_eq!(restored.role, original.role, "{context}: role mismatch");
    assert_eq!(
        restored.content, original.content,
        "{context}: content mismatch"
    );
    assert_eq!(
        restored.content_parts, original.content_parts,
        "{context}: content_parts mismatch"
    );
    assert_eq!(
        restored.tool_call_id, original.tool_call_id,
        "{context}: tool_call_id mismatch"
    );
    assert_eq!(restored.name, original.name, "{context}: name mismatch");
    match (&restored.tool_calls, &original.tool_calls) {
        (None, None) => {}
        (Some(r), Some(o)) => {
            assert_eq!(r.len(), o.len(), "{context}: tool_calls length mismatch");
            for (rc, oc) in r.iter().zip(o.iter()) {
                assert_eq!(rc.id, oc.id, "{context}: tool_call id mismatch");
                assert_eq!(rc.kind, oc.kind, "{context}: tool_call kind mismatch");
                assert_eq!(
                    rc.function.name, oc.function.name,
                    "{context}: tool_call function name mismatch"
                );
                assert_eq!(
                    rc.function.arguments, oc.function.arguments,
                    "{context}: tool_call function arguments mismatch -- this is exactly the \
                     byte-exactness TR-10's ToolInputElided restoration into \
                     tool_calls[..].function.arguments must uphold"
                );
            }
        }
        (r, o) => panic!(
            "{context}: tool_calls presence mismatch: restored.is_some()={} \
             original.is_some()={}",
            r.is_some(),
            o.is_some()
        ),
    }
}

fn call_msg(id: &'static str, tool: &'static str, arguments: String) -> ChatMessage {
    ChatMessage {
        role: Role::Assistant,
        content: None,
        content_parts: None,
        tool_calls: Some(vec![ToolCall {
            id: id.to_string(),
            kind: "function".to_string(),
            function: FunctionCall {
                name: tool.to_string(),
                arguments,
            },
        }]),
        tool_call_id: None,
        name: None,
        metadata: Default::default(),
    }
}

/// "Provider A" while it is still healthy: walks a fixed script of tool
/// calls / text replies, one assistant turn per `complete()` call, across
/// however many `Agent::send()` calls it takes to exhaust the script (each
/// `send()` drives `run_loop` through as many tool-call turns as the script
/// has queued up until the next `Step::Text`).
struct ScriptedTurns {
    calls: AtomicUsize,
    steps: Vec<Step>,
}
#[async_trait]
impl Provider for ScriptedTurns {
    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 step = self
            .steps
            .get(n)
            .unwrap_or_else(|| panic!("ScriptedTurns ran out of steps at call {n}"));
        let msg = match step {
            Step::Call {
                id,
                tool,
                arguments,
            } => call_msg(id, tool, arguments.clone()),
            Step::Text(t) => ChatMessage::assistant(t.to_string()),
        };
        Ok((msg, Usage::default()))
    }
}

/// "Provider A" once its quota has tightened: rejects any request whose
/// estimated tokens exceed `budget_tokens` — the scripted stand-in for a
/// real 429/"too many tokens" rate-limit response. Used standalone (direct
/// `.complete()` calls, not through an `Agent`) both to demonstrate that the
/// FULL-SIZE (unreduced) request is genuinely over budget AND — replayed a
/// second time against the SAME instance — that the REDUCED request now
/// clears the identical, unmoved limit.
struct RateLimitedProviderA {
    budget_tokens: u64,
}
#[async_trait]
impl Provider for RateLimitedProviderA {
    async fn complete(
        &self,
        req: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode_harness::Result<(ChatMessage, Usage)> {
        let est = estimate_view_tokens(&req.messages);
        if est > self.budget_tokens {
            return Err(Error::Other(format!(
                "rate_limit_exceeded: request is ~{est} tokens, over this account's current \
                 quota of ~{} tokens (HTTP 429 Too Many Tokens)",
                self.budget_tokens
            )));
        }
        Ok((
            ChatMessage::assistant("(would have succeeded)"),
            Usage::default(),
        ))
    }
}

/// "Provider B" — the rescue provider the session is handed to after
/// reduction. Records the estimated token size of every request it ever
/// receives (the "at minimum cost" claim: the wire request that crosses the
/// provider boundary must be bounded by the REDUCED size, not the full one)
/// and replies with one plain-text confirmation.
struct ProviderB {
    last_request_tokens: std::sync::Mutex<Option<u64>>,
}
#[async_trait]
impl Provider for ProviderB {
    async fn complete(
        &self,
        req: &ChatRequest,
        _on_delta: &(dyn for<'a> Fn(&'a str) + Send + Sync),
    ) -> supercode_harness::Result<(ChatMessage, Usage)> {
        let est = estimate_view_tokens(&req.messages);
        *self.last_request_tokens.lock().unwrap() = Some(est);
        Ok((
            ChatMessage::assistant(
                "Confirmed: the fix in src/lib.rs is correct (compute() now returns x + 1), \
                 cargo test passes, and NOTES.md records the investigation.",
            ),
            Usage::default(),
        ))
    }
}

#[tokio::test]
async fn rate_limit_rescue_continues_losslessly_with_massive_token_reduction() {
    let dir = temp_dir("main");
    // The session STORE lives in a SIBLING directory, never inside the
    // agent's own cwd — otherwise the real `search`/`list_dir` tools would
    // recurse into the sidecar file itself (which accumulates the growing
    // transcript, including every earlier ANSI-heavy `cargo test` capture)
    // and self-referentially blow up the fixture.
    let store_dir = dir.join("store");
    let cwd = dir.join("project");
    let store = SessionStore::open(&store_dir).unwrap();
    let name = "e2e-rescue-demo";
    let sidecar_path = store.sidecar_path(name);

    // -------------------------------------------------------------------
    // Fixture setup: a small "project" on real disk, seeded BEFORE the
    // agent starts, so every read/write/list/search below reflects genuine
    // filesystem state rather than a scripted illusion.
    // -------------------------------------------------------------------
    std::fs::create_dir_all(&cwd).unwrap();
    for i in 0..30 {
        std::fs::write(cwd.join(format!("file_{i:02}.txt")), "x").unwrap();
    }
    std::fs::create_dir_all(cwd.join("src")).unwrap();
    let v0 = lib_rs_source("x * 2  // BUG: should add one, TODO fix");
    std::fs::write(cwd.join("src/lib.rs"), &v0).unwrap();
    let config_v0 = config_rs_source("2 // placeholder default");
    std::fs::write(cwd.join("src/config.rs"), &config_v0).unwrap();

    let cargo_base = load_raw("cargo_build"); // genuine ANSI capture, 14,389 bytes
    let docker_base = load_raw("docker_pull"); // genuine ANSI capture, 5,956 bytes

    let run0 = cargo_output(
        &cargo_base,
        "test result: FAILED. 3 passed; 1 failed; run=0 (initial reproduction) FAIL-MARKER-e2e-0",
    );
    let run1 = cargo_output(&cargo_base, "test result: FAILED. 3 passed; 1 failed; run=1 (after first fix attempt, still broken) FAIL-MARKER-e2e-1");
    let run2 = cargo_output(&cargo_base, "test result: FAILED. 3 passed; 1 failed; run=2 (after second fix attempt, still broken) FAIL-MARKER-e2e-2");
    let run3 = cargo_output(
        &cargo_base,
        "test result: ok. 4 passed; 0 failed; run=3 (fix confirmed) PASS-MARKER-e2e-3",
    );

    let v1 = lib_rs_source("x * 3  // WIP: adjusting multiplier");
    let v2 = lib_rs_source("x + 1  // FIXED: correct increment logic");
    let notes = notes_md(150);
    let config_v1 = config_rs_source("3 // TUNED-MARKER-e2e-rescue: raised after profiling");

    let bash_outputs = vec![
        run0.clone(),
        run1.clone(),
        docker_base.clone(),
        run2.clone(),
        run3.clone(),
    ];

    // -------------------------------------------------------------------
    // The full 21-tool-call, 3-turn script (SEND A / B / C). Reused
    // reasoning (see the module doc comment for the full walkthrough): with
    // the default `protect_last_n_tool_results = 3` / `supersede_protect_last_n
    // = 3`, the newest 3 tool RESULTS are always fully visible — three tiny
    // trailing calls (C7/C8/C9) are appended after the two genuinely large,
    // reducible tail results (C4's big search, C6's big NOTES read) so THOSE
    // land outside the protected window and get reduced too, rather than
    // artificially inflating `tokens_after` with content that's merely
    // "recent," not "irreducible."
    let steps = vec![
        // ---- SEND A: "there's a failing test" ----
        Step::Call {
            id: "a1",
            tool: "bash",
            arguments: serde_json::json!({"command": "cargo test"}).to_string(),
        },
        Step::Call {
            id: "a2",
            tool: "read_file",
            arguments: serde_json::json!({"path": "src/lib.rs"}).to_string(),
        },
        Step::Call {
            id: "a3",
            tool: "write_file",
            arguments: serde_json::json!({"path": "src/lib.rs", "content": v1}).to_string(),
        },
        Step::Call {
            id: "a4",
            tool: "bash",
            arguments: serde_json::json!({"command": "cargo test"}).to_string(),
        },
        Step::Text("Applied a first fix attempt; tests still failing, checking further."),
        // ---- SEND B: "still failing, dig deeper" ----
        // The config.rs read/write/re-read trio (b0r/b0w/b0r2) is the
        // DEDICATED TR-3 (FileReadDiffed) demo: `config.rs` is deliberately
        // sized under the A7 trigger so it never gets claimed by truncation
        // first (see `config_rs_source`'s doc comment).
        Step::Call {
            id: "b0r",
            tool: "read_file",
            arguments: serde_json::json!({"path": "src/config.rs"}).to_string(),
        },
        Step::Call {
            id: "b0w",
            tool: "write_file",
            arguments: serde_json::json!({"path": "src/config.rs", "content": config_v1})
                .to_string(),
        },
        Step::Call {
            id: "b0r2",
            tool: "read_file",
            arguments: serde_json::json!({"path": "src/config.rs"}).to_string(),
        },
        Step::Call {
            id: "b1",
            tool: "read_file",
            arguments: serde_json::json!({"path": "src/lib.rs"}).to_string(),
        },
        Step::Call {
            id: "b2",
            tool: "bash",
            arguments: serde_json::json!({"command": "docker pull myimage:latest"}).to_string(),
        },
        Step::Call {
            id: "b3",
            tool: "write_file",
            arguments: serde_json::json!({"path": "src/lib.rs", "content": v2}).to_string(),
        },
        Step::Call {
            id: "b4",
            tool: "bash",
            arguments: serde_json::json!({"command": "cargo test"}).to_string(),
        },
        Step::Call {
            id: "b5",
            tool: "list_dir",
            arguments: serde_json::json!({"path": "."}).to_string(),
        },
        Step::Text("Second attempt applied; re-running tests."),
        // ---- SEND C: "confirm the fix, clean up, summarize" ----
        Step::Call {
            id: "c1",
            tool: "read_file",
            arguments: serde_json::json!({"path": "src/lib.rs"}).to_string(),
        },
        Step::Call {
            id: "c2",
            tool: "bash",
            arguments: serde_json::json!({"command": "cargo test"}).to_string(),
        },
        Step::Call {
            id: "c3",
            tool: "list_dir",
            arguments: serde_json::json!({"path": "."}).to_string(),
        },
        Step::Call {
            id: "c4",
            tool: "search",
            arguments: serde_json::json!({"pattern": "lorem"}).to_string(),
        },
        Step::Call {
            id: "c5",
            tool: "write_file",
            arguments: serde_json::json!({"path": "NOTES.md", "content": notes}).to_string(),
        },
        Step::Call {
            id: "c6",
            tool: "read_file",
            arguments: serde_json::json!({"path": "NOTES.md"}).to_string(),
        },
        Step::Call {
            id: "c7",
            tool: "list_dir",
            arguments: serde_json::json!({"path": "src"}).to_string(),
        },
        Step::Call {
            id: "c8",
            tool: "search",
            arguments: serde_json::json!({"pattern": "FIXED"}).to_string(),
        },
        Step::Call {
            id: "c9",
            tool: "list_dir",
            arguments: serde_json::json!({"path": "."}).to_string(),
        },
        Step::Text("All done -- tests pass, notes recorded, fix confirmed."),
    ];

    // -------------------------------------------------------------------
    // STEP 1 (SEED): a real Agent + recorder drives the whole script
    // against "provider A" (still healthy at this point) and records to a
    // real sidecar file on disk.
    // -------------------------------------------------------------------
    let config = Config::builder()
        .cwd(cwd.clone())
        .system_prompt("you are a careful coding agent investigating a failing test")
        .build();
    let mut registry = ToolRegistry::with_builtins();
    registry.register(ScriptedBashTool {
        outputs: bash_outputs,
        calls: AtomicUsize::new(0),
    });
    // Real builtins for read_file/write_file/list_dir/search remain
    // registered (last-wins lookup in `ToolRegistry::get` only overrides the
    // one name we re-register, "bash") — every file read/write/list/search
    // call below runs the REAL tool against the real temp directory.

    let mut agent1 = Agent::with_parts(
        config,
        Box::new(ScriptedTurns {
            calls: AtomicUsize::new(0),
            steps,
        }),
        registry,
    );
    let empty_session = Session::from_claude_code_str("").unwrap();
    let writer = SidecarWriter::create(&sidecar_path, &empty_session).unwrap();
    agent1.set_recorder(writer);
    // No reduction policy yet: every request during seeding carries the
    // FULL, unreduced history — exactly what a real pre-rescue session looks
    // like on the wire.

    let reply_a = agent1
        .send("There's a failing test around compute(); investigate.")
        .await
        .unwrap();
    assert!(reply_a.contains("still failing"));
    let reply_b = agent1
        .send("Still failing -- dig deeper and try again.")
        .await
        .unwrap();
    assert!(reply_b.contains("Second attempt"));
    let reply_c = agent1
        .send("Confirm the fix, clean up, and summarize.")
        .await
        .unwrap();
    assert!(reply_c.contains("All done"));

    // -------------------------------------------------------------------
    // Measure FULL-context tokens over the full history (system prompt +
    // every recorded turn) -- exactly what `Agent::build_request_messages`
    // would send with no reduction policy installed (which is agent1's
    // actual state throughout seeding).
    // -------------------------------------------------------------------
    let tokens_before = estimate_view_tokens(agent1.history());
    println!(
        "[e2e-rescue] tokens_before (full, unreduced context): {}",
        fmt_approx_tokens(tokens_before)
    );

    // A pristine, pre-rescue snapshot of the sidecar bytes -- captured now,
    // before ANY reduction machinery runs, for the lossless/reversible
    // verification in step 5.
    let sidecar_before_rescue_text = std::fs::read_to_string(&sidecar_path).unwrap();
    let export_before_rescue =
        reduce::export_session(&sidecar_before_rescue_text, SessionFormat::ClaudeCode)
            .expect("export_session must succeed on the pristine pre-rescue sidecar");

    // -------------------------------------------------------------------
    // STEP 2 (HIT THE LIMIT): "provider A"'s quota has tightened to a FIXED,
    // INDEPENDENT ceiling (`PROVIDER_A_QUOTA_TOKENS`, chosen up front --
    // NOT derived from `tokens_before`) -- the FULL-SIZE (unreduced) request
    // is now rejected outright because it genuinely, organically exceeds a
    // pre-existing limit (not because the limit was defined in terms of the
    // request itself).
    // -------------------------------------------------------------------
    assert!(
        tokens_before > 10_000,
        "the fixture must be genuinely large for the rate-limit rescue narrative to be real: \
         only ~{tokens_before} tokens"
    );
    assert!(
        tokens_before > PROVIDER_A_QUOTA_TOKENS,
        "the full request must genuinely exceed the fixed, independent quota: \
         tokens_before=~{tokens_before} must be > {PROVIDER_A_QUOTA_TOKENS}"
    );
    let provider_a_tightened = RateLimitedProviderA {
        budget_tokens: PROVIDER_A_QUOTA_TOKENS,
    };
    let full_request = ChatRequest::new("test-model", agent1.history().to_vec());
    let noop = |_: &str| {};
    let rejection = provider_a_tightened.complete(&full_request, &noop).await;
    let rejection_err = rejection.expect_err(
        "provider A must reject the full-size (unreduced) request once its quota has tightened",
    );
    let rejection_msg = rejection_err.to_string();
    println!("[e2e-rescue] provider A rejected the full-size request: {rejection_msg}");
    assert!(rejection_msg.contains("rate_limit_exceeded"));

    // -------------------------------------------------------------------
    // STEP 3 (RESCUE): install ReductionPolicy::default() -- the full
    // technique stack -- and project the history. Measure the REDUCED view's
    // tokens.
    // -------------------------------------------------------------------
    let policy = ReductionPolicy::default();
    let session_before = Session::from_sidecar_str(&sidecar_before_rescue_text).unwrap();
    let (reduced_view, log_computed) =
        reduce::project_messages(&session_before.messages, &policy, &ReductionLog::default());

    let mut full_reduced_request_messages = vec![agent1.history()[0].clone()];
    full_reduced_request_messages.extend(reduced_view.clone());
    let tokens_after = estimate_view_tokens(&full_reduced_request_messages);
    let reduction_pct = 1.0 - (tokens_after as f64 / tokens_before as f64);
    println!(
        "[e2e-rescue] tokens_after (reduced view sent to provider B): {}",
        fmt_approx_tokens(tokens_after)
    );
    println!(
        "[e2e-rescue] HEADLINE: {:.1}% token reduction ({} -> {})",
        reduction_pct * 100.0,
        fmt_approx_tokens(tokens_before),
        fmt_approx_tokens(tokens_after),
    );
    assert!(
        reduction_pct >= 0.70,
        "expected a massive (>=70%) reduction, got {:.1}% ({tokens_before} -> {tokens_after})",
        reduction_pct * 100.0
    );

    // The reduced view must genuinely land back UNDER the very same fixed
    // quota that just rejected the full request -- an organic boundary
    // crossing, not merely "smaller than before."
    assert!(
        tokens_after < PROVIDER_A_QUOTA_TOKENS,
        "the reduced view must land under the fixed quota: tokens_after=~{tokens_after} must \
         be < {PROVIDER_A_QUOTA_TOKENS}"
    );

    // Replay the REDUCED request against the SAME `RateLimitedProviderA`
    // instance that just rejected the full-size request above (same quota,
    // same object -- not a fresh, more lenient stand-in): this directly
    // OBSERVES the acceptance, rather than merely inferring it from the
    // tokens_after < quota inequality.
    let reduced_request = ChatRequest::new("test-model", full_reduced_request_messages.clone());
    let acceptance = provider_a_tightened.complete(&reduced_request, &noop).await;
    let (accepted_msg, _usage) = acceptance.unwrap_or_else(|e| {
        panic!(
            "the SAME rate-limited provider A that rejected the full-size request must ACCEPT \
             the reduced request now that it is under the identical fixed quota of ~{} tokens: {e}",
            PROVIDER_A_QUOTA_TOKENS
        )
    });
    println!(
        "[e2e-rescue] provider A (the SAME instance, SAME fixed quota of {}) now ACCEPTS the \
         reduced request: {:?}",
        fmt_approx_tokens(PROVIDER_A_QUOTA_TOKENS),
        accepted_msg.content
    );

    // Entering reduced mode is a pure VIEW operation -- it must not have
    // written one byte to the sidecar (SPEC.md ground rule: "reductions
    // never touch the sidecar").
    let sidecar_after_rescue_before_continue = std::fs::read_to_string(&sidecar_path).unwrap();
    assert_eq!(
        sidecar_after_rescue_before_continue, sidecar_before_rescue_text,
        "the rescue (installing a policy + projecting) must add ZERO bytes to the sidecar"
    );

    // Kind-diversity sanity: the fixture was built to exercise several
    // distinct reduction techniques, not just one.
    let mut kinds: Vec<&'static str> = log_computed
        .reductions
        .iter()
        .map(|r| match r.kind {
            ReductionKind::ToolOutputTruncated { .. } => "ToolOutputTruncated",
            ReductionKind::FileReadElided { .. } => "FileReadElided",
            ReductionKind::ImageRedacted { .. } => "ImageRedacted",
            ReductionKind::TurnsCleared { .. } => "TurnsCleared",
            ReductionKind::ToolInputElided { .. } => "ToolInputElided",
            ReductionKind::OutputNormalized { .. } => "OutputNormalized",
            ReductionKind::FileReadDiffed { .. } => "FileReadDiffed",
            ReductionKind::DuplicateOutput { .. } => "DuplicateOutput",
            ReductionKind::Superseded { .. } => "Superseded",
        })
        .collect();
    kinds.sort_unstable();
    kinds.dedup();
    println!("[e2e-rescue] reduction kinds exercised: {kinds:?}");
    assert!(
        kinds.len() >= 4,
        "expected the fixture to exercise a rich mix of reduction techniques, only saw: {kinds:?}"
    );

    // -------------------------------------------------------------------
    // STEP 4 (CONTINUE at minimum cost): resume against "provider B" on the
    // REDUCED view. The recorder continues the SAME sidecar file (proving
    // this is the SAME session, not a fresh one).
    // -------------------------------------------------------------------
    let config_b = Config::builder().cwd(cwd.clone()).build();
    let provider_b_shared = std::sync::Arc::new(ProviderB {
        last_request_tokens: std::sync::Mutex::new(None),
    });
    let mut agent2 = Agent::with_provider_arc(
        config_b,
        provider_b_shared.clone() as std::sync::Arc<dyn Provider>,
    );
    agent2.set_recorder(SidecarWriter::open_append(&sidecar_path).unwrap());
    agent2.load_session(session_before.clone());
    agent2.set_reduction_policy(policy.clone());

    let continuation_reply = agent2
        .send("Continue: confirm everything passes and summarize the fix.")
        .await
        .unwrap();
    assert!(continuation_reply.contains("Confirmed"));

    // The live agent's own build_request_messages must have recomputed the
    // SAME projection independently (determinism/prefix-stability).
    assert_eq!(
        agent2.reduction_log(),
        &log_computed,
        "agent2's own live projection must match the standalone measurement exactly"
    );

    let wire_tokens_b = provider_b_shared
        .last_request_tokens
        .lock()
        .unwrap()
        .expect("provider B must have received a request");
    println!(
        "[e2e-rescue] wire request to provider B: {} (tokens_after was {})",
        fmt_approx_tokens(wire_tokens_b),
        fmt_approx_tokens(tokens_after)
    );
    // "At minimum cost": the wire request that crossed the provider boundary
    // is bounded by ~tokens_after (the reduced size), not tokens_before.
    // (It's slightly larger than the pre-continuation `tokens_after` measurement
    // because it also carries the new user turn.)
    assert!(
        wire_tokens_b < tokens_before / 2,
        "the request to provider B must be drastically smaller than the original: \
         {wire_tokens_b} vs tokens_before={tokens_before}"
    );
    assert!(
        wire_tokens_b <= tokens_after + 200,
        "the request to provider B should be within a small margin of tokens_after \
         (just the new turn added): {wire_tokens_b} vs tokens_after={tokens_after}"
    );

    // Persist the reduction log the way the CLI does (`sessions
    // show-reductions`/`convert`/`inspect`/`resume` all call
    // `store.save_reduction_log`).
    store
        .save_reduction_log(name, agent2.reduction_log())
        .unwrap();

    // -------------------------------------------------------------------
    // STEP 5 (PROVE LOSSLESS + REVERSIBLE): reload sidecar + reduction log
    // FROM DISK -- never the live Agent/in-memory Session from here on.
    // -------------------------------------------------------------------
    let sidecar_final_text = store
        .load_sidecar(name)
        .unwrap()
        .expect("sidecar must exist on disk");
    let sidecar_final = Session::from_sidecar_str(&sidecar_final_text).unwrap();
    let log_final = store
        .load_reduction_log(name)
        .unwrap()
        .expect("reduction log must exist on disk");
    assert_eq!(log_final, log_computed);

    let original_message_count = session_before.messages.len();
    assert!(
        sidecar_final.messages.len() > original_message_count,
        "the continuation must genuinely have appended new messages to the SAME sidecar"
    );
    let base_messages = &sidecar_final.messages[..original_message_count];

    // (a) verify_log passes clean against the reloaded-from-disk sidecar --
    // the exact primitive `cli/main.rs`'s `show-reductions`/`convert`/
    // `inspect` call before doing anything else with a reduced session.
    reduce::verify_log(&log_final, &sidecar_final)
        .expect("verify_log must pass clean against the reloaded-from-disk sidecar");
    println!("[e2e-rescue] lossless check (a) verify_log: OK");

    // (b) invert() restores the FULL original pre-reduction context
    // byte-exact, reprojected fresh from the reloaded sidecar + reloaded log.
    let (fresh_view, reprojected_log) =
        reduce::project_messages(base_messages, &policy, &ReductionLog::default());
    assert_eq!(
        reprojected_log, log_final,
        "reprojecting from the reloaded sidecar must reproduce the identical log"
    );
    let inverted = reduce::invert(&fresh_view, &log_final, &sidecar_final)
        .expect("invert must pass clean against the reloaded-from-disk sidecar");
    assert_eq!(inverted.len(), base_messages.len());
    for (i, (restored, original)) in inverted.iter().zip(base_messages).enumerate() {
        assert_message_byte_exact(
            restored,
            original,
            &format!("invert byte-exact check, message index {i}"),
        );
    }
    println!(
        "[e2e-rescue] lossless check (b) invert byte-exact: OK ({} messages)",
        inverted.len()
    );

    // Sanity: the SECRET fix content is genuinely recoverable, byte-exact,
    // from the restored original -- not merely "no panic."
    let restored_text = inverted
        .iter()
        .filter_map(|m| m.content.clone())
        .collect::<Vec<_>>()
        .join("\n---\n");
    assert!(restored_text.contains("FIXED: correct increment logic"));
    assert!(restored_text.contains("FINAL-NOTES-MARKER-e2e-rescue-9f21"));
    assert!(restored_text.contains(&v0));
    assert!(restored_text.contains(&v1));
    assert!(restored_text.contains(&v2));
    assert!(restored_text.contains(&notes));
    assert!(restored_text.contains(&run0));
    assert!(restored_text.contains(&run1));
    assert!(restored_text.contains(&run2));
    assert!(restored_text.contains(&config_v0));
    assert!(restored_text.contains(&config_v1));

    // (c) expand_reduction (TR-1) drills back EVERY ToolInputElided
    // reduction in the log, individually, resolved straight against the
    // reloaded sidecar -- each checked 1:1 against THE SPECIFIC planted
    // payload for that exact reduction's `call_id` (a direct map from the
    // scripted write_file calls above), rather than membership in a pool of
    // several known strings.
    let planted_payload_by_call_id: std::collections::HashMap<&str, &str> = [
        ("a3", v1.as_str()),
        ("b0w", config_v1.as_str()),
        ("b3", v2.as_str()),
        ("c5", notes.as_str()),
    ]
    .into_iter()
    .collect();
    let tool_input_elisions: Vec<&reduce::Reduction> = log_final
        .reductions
        .iter()
        .filter(|r| matches!(r.kind, ReductionKind::ToolInputElided { .. }))
        .collect();
    assert!(
        !tool_input_elisions.is_empty(),
        "at least one ToolInputElided reduction expected"
    );
    for r in &tool_input_elisions {
        let call_id = match &r.kind {
            ReductionKind::ToolInputElided { call_id, .. } => call_id.as_str(),
            _ => unreachable!("filtered to ToolInputElided above"),
        };
        let expected = *planted_payload_by_call_id.get(call_id).unwrap_or_else(|| {
            panic!(
                "ToolInputElided reduction {} addresses unexpected call_id {call_id} -- not one \
                 of the scripted write_file calls (a3/b0w/b3/c5)",
                r.id
            )
        });
        let outcome = expand_reduction(&log_final, &sidecar_final.messages, None, &r.id, None)
            .unwrap_or_else(|e| {
                panic!(
                    "expand_reduction({}) must resolve cleanly against the reloaded sidecar: {e}",
                    r.id
                )
            });
        // The elided write's argument value is a raw JSON string value
        // (quoted); decode it back to compare against the planted plaintext.
        let expanded_plain: String =
            serde_json::from_str(&outcome.content).unwrap_or_else(|_| outcome.content.clone());
        assert_eq!(
            expanded_plain, expected,
            "expand_reduction({}) (call_id {call_id}) must restore EXACTLY the payload planted \
             for that specific write, byte-exact",
            r.id
        );
    }
    let elided_write = tool_input_elisions[0];
    println!(
        "[e2e-rescue] lossless check (c) expand_reduction byte-exact: OK ({} ToolInputElided \
         reduction(s), each matched 1:1 to its planted payload)",
        tool_input_elisions.len()
    );

    // (d) full-fidelity export_session after the rescue is byte-identical
    // (for the pre-rescue portion) to an export of the original session
    // before the rescue. Two angles on the same invariant:
    //   1. immediately after entering reduced mode (before the continuation
    //      ever touched disk), the export is LITERALLY identical (the
    //      rescue added zero bytes to the sidecar -- already proven above,
    //      re-derived here through export_session itself);
    //   2. after the FULL rescue + continuation + disk reload, the export
    //      of the complete (now-longer) session carries the original
    //      export as an EXACT PREFIX -- nothing about the original portion
    //      was rewritten, reordered, or dropped by anything that happened
    //      since (uuid/parent-chaining in `to_claude_code_jsonl` is
    //      deterministic and only ever depends on PRIOR messages, never
    //      future ones).
    let export_after_rescue_before_continue = reduce::export_session(
        &sidecar_after_rescue_before_continue,
        SessionFormat::ClaudeCode,
    )
    .unwrap();
    assert_eq!(
        export_before_rescue, export_after_rescue_before_continue,
        "export right after the rescue (pre-continuation) must be byte-identical to the \
         pre-rescue export"
    );
    let export_final =
        reduce::export_session(&sidecar_final_text, SessionFormat::ClaudeCode).unwrap();
    assert!(
        export_final.starts_with(&export_before_rescue),
        "the full-fidelity export after the rescue+continuation must carry the ENTIRE \
         original pre-rescue export as a byte-identical prefix"
    );
    println!("[e2e-rescue] lossless check (d) export_session byte-identical: OK");

    // -------------------------------------------------------------------
    // EMIT THE INSPECTABLE ARTIFACT
    // -------------------------------------------------------------------
    let per_kind_breakdown: Vec<serde_json::Value> = log_final
        .reductions
        .iter()
        .map(|r| {
            let kind_name = match &r.kind {
                ReductionKind::ToolOutputTruncated { .. } => "ToolOutputTruncated",
                ReductionKind::FileReadElided { .. } => "FileReadElided",
                ReductionKind::ImageRedacted { .. } => "ImageRedacted",
                ReductionKind::TurnsCleared { .. } => "TurnsCleared",
                ReductionKind::ToolInputElided { .. } => "ToolInputElided",
                ReductionKind::OutputNormalized { .. } => "OutputNormalized",
                ReductionKind::FileReadDiffed { .. } => "FileReadDiffed",
                ReductionKind::DuplicateOutput { .. } => "DuplicateOutput",
                ReductionKind::Superseded { .. } => "Superseded",
            };
            serde_json::json!({
                "id": r.id,
                "kind": kind_name,
                "placeholder": r.placeholder,
            })
        })
        .collect();

    let artifact = serde_json::json!({
        "test": "rate_limit_rescue_continues_losslessly_with_massive_token_reduction",
        "description": "Feature 3 flagship proof: pick up a real session, hit a provider \
            rate-limit, rescue it by reducing tokens + switching providers at minimum cost, \
            continue it, and export/rehydrate back losslessly.",
        "measured": {
            "tokens_before": tokens_before,
            "tokens_after": tokens_after,
            "reduction_ratio": tokens_after as f64 / tokens_before as f64,
            "reduction_pct": reduction_pct,
            "wire_tokens_to_provider_b": wire_tokens_b,
        },
        "rate_limit_rejection": {
            "provider_a_quota_tokens_fixed_independent": PROVIDER_A_QUOTA_TOKENS,
            "rejection_message": rejection_msg,
            "reduced_request_replayed_against_same_provider_a_instance": true,
            "reduced_request_accepted": true,
        },
        "reduction_breakdown": per_kind_breakdown,
        "continuation_turn": {
            "user_message": "Continue: confirm everything passes and summarize the fix.",
            "assistant_reply": continuation_reply,
        },
        "lossless_verification": {
            "verify_log": "ok",
            "invert_byte_exact_full_chatmessage": true,
            "invert_messages_checked": inverted.len(),
            "expand_reduction_byte_exact_1to1_all_tool_input_elisions": true,
            "expand_reduction_tool_input_elisions_checked": tool_input_elisions.len(),
            "expand_reduction_sample_id": elided_write.id,
            "export_session_byte_identical_prefix": true,
        },
        "scope_notes": {
            "token_counts_are_message_payload_only": "tokens_before/tokens_after/the fixed \
                quota all count `messages` only via tokens::estimate_view_tokens -- none \
                include the `tools` JSON-schema block a live provider request also carries. \
                The omission is symmetric across before/after/quota, so it does not distort \
                the reduction ratio or the quota-crossing comparison, but these are not full \
                wire-request byte counts.",
            "fixture_is_curated_not_statistically_average": "hand-assembled to exercise every \
                reduction kind (repeated test reruns, file re-reads, large writes -- all \
                realistic redundancy patterns) in one pass; the measured reduction_pct is an \
                ACHIEVABLE end-to-end figure demonstrating the mechanism across all kinds, not \
                a claim about an average session's reduction.",
            "provider_b_reply_is_scripted": "proves cross-provider continuation PLUMBING on \
                reduced context (request built/sent/threaded back onto the same sidecar), not \
                that a live model produces a correct answer -- same idiom as \
                tr9_handoff_guarantor.rs/tr2_dedup_guarantor.rs/tr6_supersede_guarantor.rs/\
                tr10_input_guarantor.rs's scripted providers.",
        },
    });

    let workspace_target = Path::new(env!("CARGO_MANIFEST_DIR")).join("../../target");
    std::fs::create_dir_all(&workspace_target).ok();
    let artifact_path = workspace_target.join("e2e-rescue-demo.json");
    std::fs::write(
        &artifact_path,
        serde_json::to_string_pretty(&artifact).unwrap(),
    )
    .unwrap();
    assert!(artifact_path.exists());
    println!(
        "[e2e-rescue] artifact archived at {}",
        artifact_path.display()
    );

    println!(
        "\n=== HEADLINE ===\n{} -> {}  ({:.1}% reduction)  |  provider B saw only {} on the wire\n================\n",
        fmt_approx_tokens(tokens_before),
        fmt_approx_tokens(tokens_after),
        reduction_pct * 100.0,
        fmt_approx_tokens(wire_tokens_b),
    );

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