supercode_reduce/engine/

mod.rs

//! The reduction layer's vocabulary (SPEC.md A4): addresses, sidecar
//! pointers, reduction kinds/records, the reduction log, the stub sentinel,
//! and the `sc.` metadata discipline. Everything downstream (`project`,
//! `invert`, the CLI) is built on these types; nothing here mutates a
//! session or a sidecar — this module only describes the shapes.
//!
//! Ground rules this module exists to uphold (SPEC.md §1.3): reductions
//! never touch the sidecar, every reduction is a reversible pointer, and no
//! reduction is ever silent — it always carries an in-transcript stub
//! ([`stub`]).

pub mod handoff;
pub mod normalize;
pub mod rehydrate;
pub mod stub;
pub mod summarize;
pub(crate) mod supersede;

use std::collections::{HashMap, HashSet};
use std::path::PathBuf;

use serde::{Deserialize, Serialize};

use supercode_interchange::ChatMessage;
use supercode_interchange::{estimate_view_tokens, format_commas, Role};
pub use supercode_interchange::{
    is_tool_error, mark_tool_error, mark_tool_outcome_unknown, tool_outcome, ToolOutcome,
    TOOL_ERROR_METADATA_KEY, TOOL_OUTCOME_UNKNOWN_METADATA_KEY,
};

use crate::{ReductionError as Error, Result};

/// Address in the CANONICAL full view (the `Session` reconstructed from the
/// sidecar).
///
/// NOT a raw-line index: normalization is not 1:1 with raw lines (Claude
/// `tool_result` blocks split off from the enclosing user record,
/// `session.rs:726-765`; Codex's `compacted` record clears messages while the
/// raw log keeps every line, `session.rs:464-471`). `role` rides along as an
/// integrity cross-check — if the message at `index` isn't `role` any more,
/// the addressed view has drifted out from under the pointer.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct MessageAddr {
    /// Position in the canonical full view's message list.
    pub index: usize,
    /// The role expected at that position (integrity cross-check).
    pub role: Role,
}

/// A pointer from a reduced placeholder back to its original content in the
/// sidecar.
///
/// `invert` (A6) resolves `addr` against the sidecar-reconstructed `Session`
/// and verifies `content_hash` before ever substituting content back in — a
/// stale or foreign sidecar can never silently produce the wrong content.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SidecarPtr {
    /// Where the original message lives in the canonical full view.
    pub addr: MessageAddr,
    /// Removed byte range within the original content; `None` means the
    /// whole content was removed (as opposed to a sub-span of it).
    pub span: Option<(usize, usize)>,
    /// blake3 hex digest of the full original content.
    pub content_hash: String,
}

impl SidecarPtr {
    /// Verify that `candidate` — the content this pointer is presumed to
    /// resolve to — still hashes to [`Self::content_hash`].
    ///
    /// This is the hash-verify primitive `invert` (A6) calls before
    /// substituting any original back into the full view. Returns `Err` on
    /// any mismatch (tampered/stale/foreign content) rather than ever
    /// substituting wrong content silently.
    pub fn verify(&self, candidate: &[u8]) -> Result<()> {
        self.verify_hash(&content_hash(candidate))
    }

    /// Like [`Self::verify`], but takes an already-computed hash directly —
    /// for callers (e.g. `hash_turns_range`, A10) whose hash formula isn't
    /// simply "hash these raw bytes" (it's a hash over several messages'
    /// concatenated wire bytes) but must still fail exactly the same way on
    /// mismatch.
    pub fn verify_hash(&self, actual: &str) -> Result<()> {
        if actual == self.content_hash {
            Ok(())
        } else {
            Err(Error::new(format!(
                "sidecar pointer hash mismatch: expected {}, got {actual}",
                self.content_hash
            )))
        }
    }
}

/// What kind of reduction produced a placeholder, and the data specific to
/// that kind. String forms (for the [`stub`] grammar) map 1:1 onto these
/// variants.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum ReductionKind {
    /// A7: an oversized tool result truncated in the view (sidecar keeps the
    /// full bytes).
    ToolOutputTruncated {
        /// Total size of the original content, in bytes.
        original_bytes: usize,
        /// Size of the kept prefix, in bytes.
        kept_bytes: usize,
    },
    /// A8: a read-type tool result elided because the file is unchanged on
    /// disk since it was read.
    FileReadElided {
        /// The file that was read.
        path: PathBuf,
        /// The read-log entry recording what was read and when.
        read_log: ReadLogEntry,
    },
    /// A9: an image content part redacted to a stub.
    ImageRedacted {
        /// Index of the redacted part within the message's `content_parts`.
        part_index: usize,
    },
    /// A10: a contiguous run of old turns cleared from the view.
    TurnsCleared {
        /// Address of the first cleared message (inclusive).
        first: usize,
        /// Address of the last cleared message (inclusive).
        last: usize,
        /// TR-7 (T20): present only when this span's placeholder carries an
        /// LLM-generated summary paragraph rather than the deterministic
        /// `[turns cleared]` stub — `None` whenever
        /// [`ReductionPolicy::summarize_cleared_turns`] is off (the default,
        /// SPEC.md TR-7 dev/01), or the side-call was skipped/fell back for
        /// any other reason (below the cost-guard floor, errored, timed
        /// out). Purely a view-layer/audit annotation: `invert`/`verify_log`
        /// ignore this field entirely and restore/verify byte-exact
        /// originals from `first`/`last`/[`Reduction::ptr`] alone, same as
        /// before this field existed (SPEC.md TR-7 dev/02).
        summary: Option<SpanSummary>,
    },
    /// TR-10: an already-executed, successful tool_use call's disk-persisted
    /// payload argument (e.g. `write_file`'s `content`) elided from its
    /// serialized `arguments`, leaving every other argument (e.g. `path`)
    /// verbatim. The assistant-side twin of A7 (tool RESULTS) / A8 (stale
    /// file READS): the reduced slot here is a tool_use's arguments, not a
    /// tool_result's content.
    ///
    /// Beyond the spec statement's `original_bytes`/`path`/`content_hash`,
    /// this carries `call_id`/`field` so a single reduction can address one
    /// specific tool_call's one specific payload field — necessary since a
    /// single assistant message can carry more than one `tool_calls` entry
    /// (mirrors [`Self::ImageRedacted`]'s `part_index` playing the same role
    /// for `content_parts`).
    ToolInputElided {
        /// Byte length of the elided payload field's ORIGINAL value (not the
        /// whole `arguments` string) — the stub's size figure.
        original_bytes: usize,
        /// The file path the payload was written to, when recoverable from a
        /// sibling `path` argument — `None` if the tool's schema has none.
        path: Option<PathBuf>,
        /// blake3 hex digest of the elided payload field's ORIGINAL value.
        /// Verified before ever restoring it (mirrors [`SidecarPtr::content_hash`],
        /// but hashes just the field's value — the sub-span actually
        /// removed); also what [`probe_tool_input_fresh`] compares a fresh
        /// disk read against for the freshness matrix.
        content_hash: String,
        /// The elided tool_call's stable [`supercode_interchange::ToolCall::id`]
        /// within the addressed assistant message's `tool_calls` —
        /// disambiguates when a single assistant turn issues more than one
        /// tool call.
        call_id: String,
        /// Name of the elided payload argument field (e.g. `"content"`) —
        /// which key inside `arguments` was replaced.
        field: String,
    },
    /// T30/TR-4: a bash/exec tool result whose ANSI color codes and
    /// carriage-return/erase-line/cursor-up redraws were collapsed down to
    /// their final rendered content ([`normalize::normalize`]). A VIEW
    /// normalization (SPEC.md B10: lossy presentation over a lossless
    /// sidecar) — the rendered CONTENT is fully preserved; only presentation
    /// bytes (escape sequences, superseded redraw frames) are removed.
    OutputNormalized {
        /// Total byte size of the raw captured output before normalization.
        original_bytes: usize,
        /// Byte size of the normalized (final-rendered) text, excluding the
        /// honesty trailer appended alongside it in the view.
        normalized_bytes: usize,
    },
    /// TR-3 (T26): a read-type tool result for a file already read earlier
    /// this session, whose content has since changed — replaced with a
    /// unified diff against that prior (base) read rather than shown in full,
    /// because the diff is materially smaller than the full content (see
    /// [`ReductionPolicy::diff_max_percent`]).
    ///
    /// The base is always a genuine full read, resolved straight from the
    /// canonical message slice (`project_messages`'s `msgs` parameter, never
    /// mutated) — never a previously-diffed or -elided reduction's own
    /// (reduced) content, so diffs never compound (SPEC.md TR-3 dev/04:
    /// "no diff-of-diff").
    ///
    /// `ptr` (on the containing [`Reduction`]) addresses THIS read (the
    /// re-read being replaced) and pins `new_hash` as its content hash —
    /// `invert`/`expand_reduction` restore the full re-read byte-exact
    /// through it, identically to [`ReductionKind::FileReadElided`].
    /// `base`/`base_hash` are extra provenance: which prior read the diff is
    /// against, and its content hash at diff-mint time, so a caller can tell
    /// whether that base itself has since drifted.
    FileReadDiffed {
        /// The file that was read.
        path: PathBuf,
        /// Where the base (prior full) read lives in the canonical full view.
        base: MessageAddr,
        /// Hash of the base read's content, at the time this diff was minted.
        base_hash: ContentHash,
        /// Hash of this (new) read's full content — equal to `ptr.content_hash`
        /// on the containing [`Reduction`].
        new_hash: ContentHash,
        /// Size of the full new (re-read) content, in bytes.
        original_bytes: usize,
        /// Size of the projected unified-diff text, in bytes (excludes the
        /// stub placeholder line itself).
        diff_bytes: usize,
    },
    /// TR-2 (T15): a tool result byte-identical to an earlier one still
    /// addressable in the view, replaced by a stub naming the earlier
    /// ("canonical") instance. `canonical` is informational only — display
    /// (the stub summary) and rehydration context — never part of the
    /// restore path: like every other kind, [`SidecarPtr::addr`] on this
    /// reduction's own [`Reduction::ptr`] points at THIS message's own
    /// address, so `invert`/`expand_reduction` recover it independent of
    /// whatever later happens to `canonical`'s own slot (SPEC.md TR-2
    /// dev/04: the canonical instance may itself be truncated or cleared
    /// afterward without ever affecting this pointer).
    DuplicateOutput {
        /// Where the earlier, byte-identical instance lives in the
        /// canonical full view, at the moment this reduction was minted.
        canonical: MessageAddr,
        /// Total size of the original (duplicated) content, in bytes.
        original_bytes: usize,
    },
    /// TR-6 (T16): a tool result superseded by a LATER result of the SAME
    /// tool called with the SAME canonicalized arguments
    /// (`supersede::canonical_key`) — an old failing `cargo test` run
    /// obsoleted by the newest run, a stale directory listing, an outdated
    /// `git diff`. Unlike [`Self::DuplicateOutput`] (TR-2), the two contents
    /// are NOT required to be byte-identical — a stale FAILING run and a
    /// later PASSING one of the identical command are exactly the case this
    /// exists for.
    ///
    /// `by` is informational only — display (the stub summary) and
    /// provenance — never part of the restore path: like every other kind,
    /// [`SidecarPtr::addr`] on this reduction's own [`Reduction::ptr`] points
    /// at THIS message's own address, so `invert`/`expand_reduction` recover
    /// it independent of whatever later happens to `by`'s own slot (mirrors
    /// TR-2 dev/04's guarantee for `DuplicateOutput::canonical`: the
    /// successor may itself be truncated, superseded again, or cleared
    /// afterward without ever affecting this pointer).
    Superseded {
        /// Where the newer (successor) result lives in the canonical full
        /// view, at the moment this reduction was minted.
        by: MessageAddr,
        /// Total size of the original (superseded) content, in bytes.
        original_bytes: usize,
    },
}

/// A blake3 hex digest, as produced by [`content_hash`]. A type alias only
/// (not a newtype) — matches every existing hash field in this module
/// (`SidecarPtr::content_hash`, `ReadLogEntry::content_hash`), which stayed
/// plain `String` rather than retrofit this alias in place (SPEC.md TR-3:
/// "keep enum/match additions minimal and localized").
pub type ContentHash = String;

/// TR-7 (T20) audit metadata for a [`ReductionKind::TurnsCleared`] span whose
/// placeholder carries an LLM-generated summary. Recorded on the reduction
/// itself (persisted in the `<name>.reduction.json` sidecar-family file) so
/// the audit trail survives independent of the exact placeholder rendering
/// (SPEC.md TR-7 dev/04: "reduction log records model id, prompt version,
/// and summary hash for every summarized span").
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SpanSummary {
    /// Identifier of the model that generated the summary (e.g.
    /// `"claude-haiku-4-5"`), taken verbatim from
    /// [`summarize::SpanSummarizer::model_id`].
    pub model_id: String,
    /// Version of the fixed, in-repo summarization prompt used
    /// ([`summarize::PROMPT_VERSION`] at mint time) — never recomputed
    /// later, so a prompt-wording change never rewrites history for spans
    /// already summarized under an earlier version.
    pub prompt_version: String,
    /// blake3 hex digest of the summary paragraph text (BEFORE the honesty
    /// banner/id are appended) — verifiable independent of the placeholder's
    /// exact surrounding punctuation.
    pub summary_hash: ContentHash,
}

/// A record of a file read, kept in [`ReductionLog::read_log`] so an
/// exporter or a later model can always answer "what was read" even when the
/// read result itself was elided from the view (A8).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ReadLogEntry {
    /// The file that was read.
    pub path: PathBuf,
    /// Where the full read result lives in the canonical full view.
    pub addr: MessageAddr,
    /// Hash of the read result's content, taken at read time.
    pub content_hash: String,
    /// The file's mtime as observed at projection time, if available.
    pub mtime: Option<i64>,
}

/// One applied reduction: what kind it was, where it points, its stable id,
/// and the exact placeholder text standing in for it in the reduced view.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Reduction {
    /// Stable id, e.g. `"r0042-9f3c"` — ordinal + 4-hex content-hash prefix
    /// (D2). Unique per session; stable across re-projection.
    pub id: String,
    /// What was reduced and the kind-specific data.
    pub kind: ReductionKind,
    /// Pointer back to the original content in the sidecar.
    pub ptr: SidecarPtr,
    /// The exact stub text ([`stub::format`]) standing in for the original
    /// content in the reduced view.
    pub placeholder: String,
}

/// Durable, content-free accounting for one reduction pass. This is proof
/// metadata only: inversion and projection depend exclusively on
/// [`ReductionLog::reductions`]. Keeping it with the log lets an offline
/// inspector distinguish a disabled pass, an enabled pass with no candidate,
/// and a candidate later subsumed by a higher-order pass such as A10.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct ReductionPassAttribution {
    /// Stable stub-grammar pass name.
    pub kind: String,
    /// Whether the final configured policy enabled this pass.
    pub enabled: bool,
    /// Claims produced immediately before later cardinality-changing passes.
    pub candidate_count: usize,
    /// Original bytes addressed by those claims.
    pub candidate_original_bytes: u64,
    /// Claims retained in the final persisted reduction index.
    pub applied_count: usize,
    /// Original bytes addressed by retained claims.
    pub applied_original_bytes: u64,
    /// Earlier claims subsumed by a later pass.
    pub suppressed_by_later_pass_count: usize,
    /// Original bytes addressed by those subsumed claims.
    pub suppressed_by_later_pass_bytes: u64,
    /// Bytes this pass would save in isolation.
    pub standalone_saved_bytes: u64,
    /// Estimated tokens this pass would save in isolation.
    pub standalone_saved_tokens: u64,
    /// Bytes this pass adds after earlier persisted passes.
    pub marginal_saved_bytes: u64,
    /// Estimated tokens this pass adds after earlier persisted passes.
    pub marginal_saved_tokens: u64,
    /// Standalone byte claim removed by overlap or pass order.
    pub suppressed_bytes: u64,
    /// Standalone token claim removed by overlap or pass order.
    pub suppressed_tokens: u64,
    /// Projected bytes retained after this pass in pipeline order.
    pub retained_bytes: u64,
    /// Estimated tokens retained after this pass in pipeline order.
    pub retained_tokens: u64,
}

/// Durable aggregate for [`ReductionPassAttribution`]. Byte and token
/// aggregates are marginal sums, never sums of overlapping standalone
/// claims.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct ReductionAttribution {
    /// Serialized bytes before reduction.
    pub full_bytes: u64,
    /// Serialized bytes after all persisted reductions.
    pub view_bytes: u64,
    /// Estimated tokens before reduction.
    pub full_tokens: u64,
    /// Estimated tokens after all persisted reductions.
    pub view_tokens: u64,
    /// Actual aggregate byte savings.
    pub aggregate_saved_bytes: u64,
    /// Actual aggregate estimated-token savings.
    pub aggregate_saved_tokens: u64,
    /// Sum of byte savings claimed by passes in isolation.
    pub standalone_saved_bytes: u64,
    /// Sum of estimated-token savings claimed by passes in isolation.
    pub standalone_saved_tokens: u64,
    /// Standalone byte claims excluded from the aggregate.
    pub overlap_suppressed_bytes: u64,
    /// Standalone estimated-token claims excluded from the aggregate.
    pub overlap_suppressed_tokens: u64,
    /// One row per pass in production pipeline order.
    pub passes: Vec<ReductionPassAttribution>,
}

/// The persisted index of every reduction applied to a session, plus the A8
/// read-log. This is the `<name>.reduction.json` sidecar-family file (D1);
/// `invert` needs it (with the sidecar) to reconstruct the full view.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct ReductionLog {
    /// Every reduction applied so far, in application order.
    pub reductions: Vec<Reduction>,
    /// Reductions explicitly rehydrated by the user. They remain persisted
    /// so a later projection/restart can distinguish "deliberately
    /// expanded" from "never reduced" without re-stubbing the address.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub expanded: Vec<Reduction>,
    /// Every file read observed during projection (A8), independent of
    /// whether that particular read ended up elided.
    pub read_log: Vec<ReadLogEntry>,
    /// Optional content-free pass attribution captured by a driving surface.
    /// Older logs omit it and remain fully compatible.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub attribution: Option<ReductionAttribution>,
}

/// The sentinel prefix every reduction placeholder starts with (D2).
///
/// Defined once, here. `A11`'s export leak-guard greps for this string; the
/// projection layer never writes it into genuine (non-reduced) content.
pub const REDUCTION_SENTINEL: &str = "[sc-reduced";

/// The reserved `ChatMessage.metadata` key carrying a reduced message's
/// [`Reduction::id`] (the `sc.` prefix is reserved for this reduction
/// layer's own bookkeeping).
///
/// This rides `ChatMessage`'s guarantee that `metadata` never serializes to
/// the wire (`message.rs:49-54`, `57-79`) — the pointer reaches every
/// in-process consumer (CLI inspection, `invert`) but can never enter a
/// request body or a persisted transcript.
pub const REDUCTION_METADATA_KEY: &str = "sc.reduction";

/// Stamp `msg` with the `sc.reduction` metadata key pointing at `id`. This is
/// the one place a reduction's id is attached to a message; every A/B
/// emitter should go through this rather than writing the key by hand.
pub fn set_reduction_id(msg: &mut ChatMessage, id: &str) {
    msg.metadata
        .insert(REDUCTION_METADATA_KEY.to_string(), id.to_string());
}

/// Read back a message's `sc.reduction` id, if it was reduced.
pub fn reduction_id(msg: &ChatMessage) -> Option<&str> {
    msg.metadata.get(REDUCTION_METADATA_KEY).map(String::as_str)
}

/// Hash helper used consistently across the reduction layer: blake3 hex
/// digest of `bytes`. Used both for [`SidecarPtr::content_hash`] and
/// [`ReadLogEntry::content_hash`].
pub fn content_hash(bytes: &[u8]) -> String {
    blake3::hash(bytes).to_hex().to_string()
}

/// Build a [`Reduction::id`]: a zero-padded 4-digit ordinal plus the first 4
/// hex characters of a content hash (D2), e.g. `"r0042-9f3c"`.
///
/// `ordinal` is the reduction's position among reductions applied to this
/// session; it wraps decoratively past 9999 (id uniqueness within a session
/// still holds in practice because the hash prefix disambiguates, and no
/// real session approaches that many reductions).
pub fn make_id(ordinal: usize, hash: &str) -> String {
    let ord = ordinal % 10_000;
    let prefix: String = hash.chars().take(4).collect();
    format!("r{ord:04}-{prefix}")
}

// ---------------------------------------------------------------------------
// A5 — project(): session -> reduced view + reduction log
// ---------------------------------------------------------------------------

/// Knobs controlling [`project_messages`]. Defaults match SPEC.md A5/A7, stacked per
/// D14 with the levers that don't need an external I/O probe to be safe
/// on-by-default (`redact_images`; contrast `elide_stale_reads`, below).
///
/// `elide_stale_reads` (A8) and `clear_turns_older_than` (A10) are plumbed
/// through the struct but inert by default — `elide_stale_reads` needs a
/// freshness probe ([`probe_read_freshness`]) to mean anything, and
/// `clear_turns_older_than` is populated per-agent from
/// `compact_after_messages` (`Agent::maybe_compact`), not from this default.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReductionPolicy {
    /// Bytes kept from the front of an oversized tool result (A7). Default
    /// `4096`.
    pub tool_output_keep_bytes: usize,
    /// Only tool results strictly larger than this are truncation candidates
    /// (A7). Default `8192`.
    pub tool_output_trigger_bytes: usize,
    /// Never reduce the newest `N` tool results (#1 "keep"). Default `3`.
    pub protect_last_n_tool_results: usize,
    /// A8 — gate for stale-file-read elision. Consults
    /// [`Self::read_freshness`] for the actual per-message verdicts; setting
    /// this without ever populating `read_freshness` (via
    /// [`probe_read_freshness`]) elides nothing, since the empty default
    /// freshness map treats every read as not-yet-verified.
    pub elide_stale_reads: bool,
    /// A9 — gate for `data:` URL image redaction. Default `true` (D14:
    /// reduced mode stacks every lossless lever on together) — unlike
    /// `elide_stale_reads`, this rule is a pure function of the message
    /// content already in view, so it carries none of A8's "meaningless
    /// without a probe" caveat and can safely default on.
    pub redact_images: bool,
    /// A9 — minimum byte length of a candidate `image_url` part's `url`
    /// string for it to become a redaction candidate; inert unless
    /// `redact_images` is set. Default `8192` (mirrors
    /// `tool_output_trigger_bytes`'s scale: small inline icons stay in view,
    /// real screenshots/photos get redacted).
    pub image_redact_min_bytes: usize,
    /// A10 — inert until turn-clearing lands.
    pub clear_turns_older_than: Option<usize>,
    /// A8 — the disk-probe pre-pass's output ([`probe_read_freshness`]),
    /// consulted by [`project_messages`] only when [`Self::elide_stale_reads`]
    /// is set. This is *data*, not a config knob: it is meant to be
    /// recomputed by the caller before every `project`/`project_messages`
    /// call (disk state can change turn to turn) — `project_messages` itself
    /// never performs the I/O; that happens once, up front, in
    /// `probe_read_freshness`. Default: empty (fails closed — nothing is
    /// considered fresh without an accompanying probe).
    pub read_freshness: ReadFreshness,
    /// TR-3 (T26) — gate for diff-only re-read representation
    /// (`ReductionKind::FileReadDiffed`). Like `redact_images` (and unlike
    /// `elide_stale_reads`), this rule is a pure function of the message
    /// content already in view — a re-read's content compared against the
    /// prior read of the same path recorded in `log.read_log` — with no
    /// disk-probe caveat, so it can safely default on. Default `true`.
    pub diff_rereads: bool,
    /// TR-3 — a candidate diff must be no more than this percentage of the
    /// full re-read's size to replace it; otherwise the full re-read stays
    /// untouched (SPEC.md TR-3 dev/03's "large-change guard"). An integer
    /// percentage (rather than a float) so [`ReductionPolicy`] keeps its
    /// `Eq` derive (`f64` has none). Default `50` ("diff ≤ 50% of full
    /// content").
    pub diff_max_percent: u32,
    /// B7 coordination clamp: when an imported-prefix cache plan is
    /// active, the count of leading messages (of the slice passed to
    /// [`project_messages`]) that make up the imported session prefix. A10
    /// turn-clearing must never establish a clear range that dips into them,
    /// since doing so would bust the prefix's cache breakpoint (and its
    /// fidelity). `None` (the default) applies no clamp, matching today's
    /// behavior for callers that never set a cache plan. Set by a runtime
    /// agent from its own `imported_prefix_len`, not a
    /// user-facing knob.
    pub protect_imported_prefix: Option<usize>,
    /// TR-10 — gate for tool-INPUT elision ([`ReductionKind::ToolInputElided`]).
    /// Default `true`. Unlike `elide_stale_reads`, the candidate rule
    /// (executed successfully + oversized payload + a disk-persisting tool)
    /// is a pure function of the view plus [`Self::tool_input_elidable_fields`]
    /// — no external disk probe is needed to decide elision itself (a probe
    /// only matters later, for the freshness-matrix ESCALATION decision, see
    /// [`probe_tool_input_fresh`]) — so, like `redact_images`, this can
    /// safely default on (D14: reduced mode stacks every lossless lever
    /// together).
    pub elide_tool_inputs: bool,
    /// TR-10 — only a candidate tool_call's payload field whose value
    /// exceeds this many bytes becomes an elision candidate. Default `8192`
    /// (mirrors A7/A9's scale).
    pub tool_input_trigger_bytes: usize,
    /// TR-10 — table of tool name -> its elidable (disk-persisted) payload
    /// argument field. Defaults to the built-in write-family tools
    /// (`write_file` -> `content`, see `default_tool_input_elidable_fields`);
    /// an MCP tool opts in by inserting its own `(name, field)` entry
    /// (SPEC.md TR-10: "per-MCP-tool opt-in").
    pub tool_input_elidable_fields: HashMap<String, String>,
    /// T30/TR-4 — gate for [`ReductionKind::OutputNormalized`] (ANSI/redraw
    /// collapse over terminal tool output). Default `true`: like
    /// `redact_images`, this is a pure function of already-in-view content
    /// (no I/O probe needed) and content-lossless (rendered CONTENT is fully
    /// preserved, only presentation bytes are removed), so it stacks on by
    /// default per D14.
    pub normalize_terminal_output: bool,
    /// T30/TR-4 — minimum byte savings (`original_bytes - normalized_bytes`)
    /// for a candidate to actually become an
    /// [`ReductionKind::OutputNormalized`] reduction; below this floor the
    /// output is left untouched rather than raced through the reduction
    /// machinery for a few bytes (SPEC.md TR-4's "savings floor" knob).
    /// Default [`normalize::DEFAULT_MIN_SAVINGS`].
    pub terminal_output_min_savings: usize,
    /// TR-2 — minimum byte length of a duplicate tool-result candidate's
    /// content for it to become a [`ReductionKind::DuplicateOutput`]
    /// candidate. Below this, both the canonical and the would-be duplicate
    /// are left alone: a stub's own bytes are not free, so deduping a tiny
    /// output would spend more than it saves (the "savings floor"). Default
    /// `256`.
    pub duplicate_output_min_bytes: usize,
    /// TR-2 — gate for [`ReductionKind::DuplicateOutput`]. Default `true`:
    /// duplicate detection is a pure function of the recorded tool outputs,
    /// so it stacks on in ordinary reduced mode. Composable capability
    /// profiles can disable it without changing the savings-floor knob.
    pub deduplicate_outputs: bool,
    /// TR-6 (T16) — gate for [`ReductionKind::Superseded`] (same tool + same
    /// canonicalized arguments, keep only the newest result). Default
    /// `true`: like `redact_images`/`elide_tool_inputs`, the candidate rule
    /// is a pure function of the view plus [`Self::supersede_command_fields`]
    /// — no external disk probe needed — so it stacks on by default (D14).
    pub supersede_enabled: bool,
    /// TR-6 — protected recency zone (opencode's `PRUNE_PROTECT` spirit): the
    /// newest `N` tool RESULT messages (by position, mirrors
    /// [`Self::protect_last_n_tool_results`]'s own construction) are never a
    /// *new* `Superseded` candidate, regardless of how many older same-key
    /// occurrences exist. A DEDICATED knob rather than reusing
    /// `protect_last_n_tool_results` — TR-6.md's spec calls this out as its
    /// own independently tunable "protected recency zone," and the two
    /// passes run at different points in the pipeline (Superseded runs
    /// before A7 truncation ever computes its own candidates). Default `3`
    /// (mirrors `protect_last_n_tool_results`'s own default).
    pub supersede_protect_last_n: usize,
    /// TR-6 — minimum byte length of a superseded-candidate's OWN content for
    /// it to become a [`ReductionKind::Superseded`] candidate (the "savings
    /// floor," mirrors [`Self::duplicate_output_min_bytes`]). Below this the
    /// older result is left untouched — a stub's own bytes are not free.
    /// Default `256`.
    pub supersede_min_bytes: usize,
    /// TR-6 — table of tool name -> its command-bearing argument field (e.g.
    /// `bash`/`shell`/`exec_command` -> `"command"`), consulted by
    /// `supersede::canonical_key` to canonicalize (trim + collapse internal
    /// whitespace) just that one field's value rather than the whole
    /// arguments string. A tool absent from this table still participates in
    /// supersession — its whole (trimmed-only) arguments string becomes the
    /// key — this table only controls whitespace-collapse scope, not
    /// eligibility. Defaults to `supersede::default_command_fields`.
    pub supersede_command_fields: HashMap<String, String>,
    /// TR-6 — gate for errored-call input pruning: a FAILED tool call's
    /// oversized payload argument ([`Self::tool_input_elidable_fields`],
    /// shared with TR-10) becomes a [`ReductionKind::ToolInputElided`]
    /// candidate once [`Self::errored_input_prune_after_turns`] assistant
    /// turns have elapsed since the failed call — the disjoint, FAILURE-side
    /// complement of TR-10's `elide_tool_inputs` (which only ever considers
    /// SUCCESSFUL calls; see `detect_tool_inputs`'s own doc comment for the
    /// success/failure boundary). The failed call's own error-result message
    /// is never touched by this — only the assistant-side input argument —
    /// so the error itself stays visible exactly as TR-6.md requires. Default
    /// `true` (pure function of the view + the age clock, no I/O probe
    /// needed, D14).
    pub prune_errored_inputs: bool,
    /// TR-6 — how many LATER `Role::Assistant` messages must appear after a
    /// failed call's own message before its oversized input becomes a
    /// pruning candidate (the "N turns" aging clock in TR-6.md's errored-call
    /// case) — a message-count proxy for "turns elapsed," the same
    /// convention [`Self::clear_turns_older_than`] (A10) already uses (this
    /// codebase has no other structural definition of a conversational
    /// turn). Default `3`.
    pub errored_input_prune_after_turns: usize,
    /// TR-7 (T20) — the `summaries: on|off` config knob: gate for rendering
    /// an established [`ReductionKind::TurnsCleared`] span's placeholder as
    /// an LLM-generated summary paragraph instead of the deterministic
    /// `[turns cleared]` stub. **Default `false`** — SPEC.md TR-7 dev/01:
    /// with this off, the A10 stub must stay byte-identical to pre-TR-7
    /// behavior, so `project_messages` never even looks at
    /// [`Self::cleared_turns_summary`] while this is unset, regardless of
    /// what a caller precomputed. Turning this on with no matching
    /// [`Self::cleared_turns_summary`] entry (e.g. the side-call was never
    /// run, or failed) is exactly as safe: the deterministic stub is still
    /// what gets rendered (dev/03's failure-fallback guarantee).
    pub summarize_cleared_turns: bool,
    /// TR-7 — cost-guard floor (dev/05): a candidate cleared span's ORIGINAL
    /// byte size (the same `range_bytes` the deterministic stub's own
    /// summary clause already reports) must exceed
    /// `expected_summary_bytes * summary_cost_floor_multiple` before
    /// [`prepare_cleared_turns_summary`] ever calls the injected
    /// [`summarize::SpanSummarizer`] — below the floor, a summarization
    /// side-call would be negative-ROI (the stub it replaces is already
    /// small) and is skipped outright, never attempted. Default `400`
    /// (a rough paragraph-sized estimate).
    pub expected_summary_bytes: usize,
    /// TR-7 — see [`Self::expected_summary_bytes`]; the floor multiplier.
    /// Default `4` (the span must be at least ~4 summaries' worth of bytes).
    pub summary_cost_floor_multiple: usize,
    /// TR-7 — the side-call preparer's output
    /// ([`prepare_cleared_turns_summary`]), consulted by
    /// [`project_messages`] only when [`Self::summarize_cleared_turns`] is
    /// set AND the prepared entry's `(first, last)` matches EXACTLY the
    /// range `project_messages` independently (re)computes for this call —
    /// any other prepared entry (stale, wrong range, or simply absent) is
    /// silently ignored and the deterministic stub is rendered instead. This
    /// is *data*, not a config knob (mirrors [`Self::read_freshness`]):
    /// recomputed by the caller (via `prepare_cleared_turns_summary`, the
    /// one place TR-7's side-call happens) before every
    /// `project`/`project_messages` call that might establish a NEW
    /// `TurnsCleared` range. Default: `None`.
    pub cleared_turns_summary: Option<PreparedClearSummary>,
}

impl Default for ReductionPolicy {
    fn default() -> Self {
        ReductionPolicy {
            tool_output_keep_bytes: 4096,
            tool_output_trigger_bytes: 8192,
            protect_last_n_tool_results: 3,
            elide_stale_reads: false,
            redact_images: true,
            image_redact_min_bytes: 8192,
            clear_turns_older_than: None,
            read_freshness: ReadFreshness::default(),
            diff_rereads: true,
            diff_max_percent: 50,
            protect_imported_prefix: None,
            elide_tool_inputs: true,
            tool_input_trigger_bytes: 8192,
            tool_input_elidable_fields: default_tool_input_elidable_fields(),
            normalize_terminal_output: true,
            terminal_output_min_savings: normalize::DEFAULT_MIN_SAVINGS,
            duplicate_output_min_bytes: 256,
            deduplicate_outputs: true,
            supersede_enabled: true,
            supersede_protect_last_n: 3,
            supersede_min_bytes: 256,
            supersede_command_fields: supersede::default_command_fields(),
            prune_errored_inputs: true,
            errored_input_prune_after_turns: 3,
            summarize_cleared_turns: false,
            expected_summary_bytes: 400,
            summary_cost_floor_multiple: 4,
            cleared_turns_summary: None,
        }
    }
}

/// The built-in write-family default for
/// [`ReductionPolicy::tool_input_elidable_fields`]: `write_file` -> `content`
/// (`tools/builtins.rs`'s `WriteFileTool` schema — the one built-in tool
/// whose entire payload is disk-persisted verbatim), plus Claude Code's own
/// native `Write` tool (imported sessions carry Claude's tool names verbatim,
/// `session.rs::push_claude_assistant` — never remapped to this crate's own
/// builtin names), which shares the same `content` payload field name.
fn default_tool_input_elidable_fields() -> HashMap<String, String> {
    let mut m = HashMap::new();
    m.insert("write_file".to_string(), "content".to_string());
    m.insert("Write".to_string(), "content".to_string());
    m
}

// ---------------------------------------------------------------------------
// TR-7 (T20) — LLM-written summary placeholders over cleared spans
// ---------------------------------------------------------------------------

/// The output of [`prepare_cleared_turns_summary`] — one summarized span,
/// ready for [`project_messages`] to apply IF (and only if) it independently
/// recomputes the exact same `(first, last)` range for its own
/// [`ReductionKind::TurnsCleared`] candidate this call. Threaded through
/// [`ReductionPolicy::cleared_turns_summary`]; see that field's doc comment
/// for the full data-vs-config-knob split (mirrors [`ReadFreshness`]).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PreparedClearSummary {
    /// Address of the first message the summarized span covers (must match
    /// `project_messages`'s own candidate range exactly to be applied).
    pub first: usize,
    /// Address of the last message the summarized span covers.
    pub last: usize,
    /// The summarizer's output, already sanitized into the [`stub`] grammar's
    /// one-line/no-`]` contract ([`prepare_cleared_turns_summary`] does this
    /// once, here, so `project_messages` never needs to).
    pub text: String,
    /// [`summarize::SpanSummarizer::model_id`], carried through for the
    /// audit trail ([`SpanSummary::model_id`]).
    pub model_id: String,
}

/// Compute the A10 candidate clear range `[first, last]` (inclusive) for
/// `msgs` under `policy.clear_turns_older_than`/`policy.protect_imported_prefix`,
/// or `None` if clearing doesn't trigger (below threshold, or no room once
/// the system-prefix/tool-boundary/imported-prefix guards are applied).
///
/// A pure function of `msgs.len()` and each message's `role` alone — never
/// affected by any in-place content reduction (A7/A8/A9/TR-2/TR-3/TR-4/TR-6/
/// TR-10 all preserve `role`, only ever rewriting `content`/`tool_calls`), so
/// it is safe to call directly against the pristine `msgs` slice from BOTH
/// call sites that must agree on the identical range or a prepared summary
/// could silently apply to the wrong span: [`project_messages`] itself
/// (called against `view`, whose roles are identical to `msgs`'s at the
/// point A10 runs — see its own comment on pass order) and
/// [`prepare_cleared_turns_summary`] (called against `msgs` directly, before
/// any projection has run at all). Sharing this one function is what
/// guarantees the two agree BY CONSTRUCTION, never by convention.
fn compute_clear_range(msgs: &[ChatMessage], policy: &ReductionPolicy) -> Option<(usize, usize)> {
    let threshold = policy.clear_turns_older_than?;
    if msgs.len() <= threshold {
        return None;
    }
    // Never target a leading system/developer message (A5).
    let mut first = 0;
    while first < msgs.len() && msgs[first].role == Role::System {
        first += 1;
    }
    // B7 coordination clamp (see `ReductionPolicy::protect_imported_prefix`).
    if let Some(protected) = policy.protect_imported_prefix {
        first = first.max(protected);
    }
    let keep_recent = (threshold / 2).max(2);
    let mut cut = msgs.len().saturating_sub(keep_recent);
    // Never begin the kept (surviving) window on a tool result.
    while cut < msgs.len() && msgs[cut].role == Role::Tool {
        cut += 1;
    }
    if cut > first && cut < msgs.len() {
        Some((first, cut - 1))
    } else {
        None
    }
}

/// Render a message range as the plain, human-legible text fed to
/// [`summarize::SpanSummarizer::summarize`] (via [`summarize::render_prompt`]
/// for a real implementation) — one `role: content` line per message. Content
/// only (not tool-call argument JSON): keeps the side-call's input compact,
/// matching what TR-7's spec calls "what content existed" rather than a
/// byte-exact re-serialization (the sidecar, not this rendering, is the
/// byte-exact source of truth `invert`/`expand_reduction` always use).
fn render_span_text(msgs: &[ChatMessage]) -> String {
    let mut out = String::new();
    for m in msgs {
        let role = match m.role {
            Role::System => "system",
            Role::User => "user",
            Role::Assistant => "assistant",
            Role::Tool => "tool",
        };
        out.push_str(role);
        out.push_str(": ");
        out.push_str(m.content.as_deref().unwrap_or(""));
        out.push('\n');
    }
    out
}

/// TR-7's one side-call site (SPEC.md: "an explicit, budgeted, injectable
/// side-call... never blocking the main loop"). Call this against the exact
/// same `msgs` slice about to be projected (mirrors
/// [`probe_read_freshness`]'s own calling convention), thread the result
/// through [`ReductionPolicy::cleared_turns_summary`] before calling
/// [`project_messages`]. Only ever does anything when
/// `policy.summarize_cleared_turns` is set; callers that never enable TR-7
/// can skip calling this entirely (dev/01: `project_messages` behaves
/// identically either way when the policy gate is off).
///
/// Returns `None` — meaning `project_messages` will render the deterministic
/// stub — whenever: TR-7 is off; a `TurnsCleared` range is already
/// established (A10 fires at most once per session, singleton, so no
/// side-call is ever needed for an already-decided span); no clear range
/// currently triggers; the candidate span is below the cost-guard floor
/// (dev/05, no side-call attempted at all); or the summarizer itself errors
/// or returns an empty/blank result (dev/03's fault-injection contract —
/// this NEVER propagates an error to the caller, by design).
pub fn prepare_cleared_turns_summary(
    msgs: &[ChatMessage],
    policy: &ReductionPolicy,
    prior: &ReductionLog,
    summarizer: &dyn summarize::SpanSummarizer,
) -> Option<PreparedClearSummary> {
    if !policy.summarize_cleared_turns {
        return None;
    }
    if prior
        .reductions
        .iter()
        .any(|r| matches!(r.kind, ReductionKind::TurnsCleared { .. }))
    {
        return None; // Already established; never recomputed (A10 singleton).
    }
    let (first, last) = compute_clear_range(msgs, policy)?;
    let range = &msgs[first..=last];
    let (_hash, range_bytes) = hash_turns_range(range).ok()?;
    // Cost guard (dev/05): skip the side-call outright for a span too small
    // to be positive-ROI once the summary's own stub overhead is counted —
    // never merely discard a result already paid for.
    let floor = policy
        .expected_summary_bytes
        .saturating_mul(policy.summary_cost_floor_multiple);
    if range_bytes <= floor {
        return None;
    }
    let span_text = render_span_text(range);
    let text = match summarizer.summarize(&span_text) {
        Ok(t) if !t.trim().is_empty() => t,
        _ => return None, // dev/03: error or blank result -> deterministic fallback.
    };
    // Sanitize into the `stub` grammar's one-line, no-`]` contract
    // (SPEC.md C2/D2) regardless of what the summarizer produced — a
    // formatting quirk in the model's output must never fail the pass.
    let sanitized = text
        .split_whitespace()
        .collect::<Vec<_>>()
        .join(" ")
        .replace(']', ")");
    if sanitized.is_empty() {
        return None;
    }
    Some(PreparedClearSummary {
        first,
        last,
        text: sanitized,
        model_id: summarizer.model_id().to_string(),
    })
}

// ---------------------------------------------------------------------------
// A8 — stale-file-read detection + the disk-probe pre-pass
// ---------------------------------------------------------------------------

/// Read-type tool names A8 elision applies to. A tool result is a candidate
/// only when its paired assistant `tool_calls` entry names one of these
/// (`tools/builtins.rs:39-104`'s `read_file`). `B6` must keep this in sync
/// with any built-in tool rename.
pub const READ_TOOLS: &[&str] = &["read_file"];

/// One read-type tool result found in a message slice: the index of the
/// `Role::Tool` result, the file path pulled from the paired assistant
/// call's `path` argument (the `read_file` schema's one required field), and
/// whether that call was a partial-window read (`offset` and/or `limit` set
/// — `tools/builtins.rs`'s `ReadArgs`).
#[derive(Debug, Clone)]
struct DetectedRead {
    index: usize,
    path: PathBuf,
    /// TR-3 v1 scope guard: `true` when the paired call set `offset` and/or
    /// `limit` (a partial-window read, `tools/builtins.rs`'s `ReadArgs`).
    /// TR-3.md's frozen spec excludes partial-window reads from v1 ("full-file
    /// reads only... document the exclusion in the stub logic"): two windowed
    /// reads of the same path may cover different line ranges entirely, so a
    /// unified diff between them would present a diff between two arbitrary
    /// windows as if it were a file change. Consulted ONLY by the TR-3
    /// diffing candidate rule below — A8 elision and TR-2 dedup are
    /// unaffected (A8 already fails closed on a windowed read via its
    /// hash-mismatch fallback, `probe_read_freshness`'s doc comment above).
    windowed: bool,
}

/// Find every read-type tool result in `msgs`: a `Role::Tool` message paired
/// — by `tool_call_id` — to the nearest earlier assistant message whose
/// `tool_calls` contains a matching id naming one of [`READ_TOOLS`], with a
/// string `path` argument.
///
/// Pairing rides `tool_call_id` alone. Claude imports also stamp a
/// `sourceToolAssistantUUID` metadata edge on the tool-result message
/// (`session.rs:876-888`) parallel to `parentUuid`, but that id has no
/// corresponding field recoverable on the assistant side through
/// `ChatMessage`'s stable shape — and it doesn't need one here: Claude's own
/// `tool_use_id` already becomes `tool_call_id` on import
/// (`session.rs:876-881`), so `tool_call_id` pairing alone already covers
/// both Codex and Claude Code sessions. `sourceToolAssistantUUID` is
/// therefore not consulted (SPEC.md A8: "prefer the simple route").
fn detect_reads(msgs: &[ChatMessage]) -> Vec<DetectedRead> {
    let mut out = Vec::new();
    for (i, msg) in msgs.iter().enumerate() {
        if msg.role != Role::Tool {
            continue;
        }
        let Some(call_id) = msg.tool_call_id.as_deref() else {
            continue;
        };
        let call = msgs[..i].iter().rev().find_map(|m| {
            if m.role != Role::Assistant {
                return None;
            }
            m.tool_calls().iter().find(|c| c.id == call_id).cloned()
        });
        let Some(call) = call else {
            continue;
        };
        if !READ_TOOLS.contains(&call.function.name.as_str()) {
            continue;
        }
        let Ok(args) = call.function.parsed_arguments() else {
            continue;
        };
        let Some(path) = args.get("path").and_then(|v| v.as_str()) else {
            continue;
        };
        let windowed = args.get("offset").is_some_and(|v| !v.is_null())
            || args.get("limit").is_some_and(|v| !v.is_null());
        out.push(DetectedRead {
            index: i,
            path: PathBuf::from(path),
            windowed,
        });
    }
    out
}

/// Find every `Role::Tool` message in `msgs` whose paired assistant
/// `tool_calls` entry names one of [`normalize::NORMALIZE_TOOLS`] — T30's
/// candidate rule, keyed on tool IDENTITY rather than [`ChatMessage::name`]:
/// a live [`crate::Agent`]'s own `history` populates `name` directly
/// (`Agent::run_loop` builds tool results via
/// `ChatMessage::tool_result(call.id, call.function.name, ...)`), but a
/// session reloaded from an imported Claude Code or Codex log never does —
/// `session.rs`'s `tool_message` helper always sets `name: None` there (the
/// tool identity lives only on the paired assistant `tool_calls` entry in
/// both wire formats). Same `tool_call_id` pairing [`detect_reads`] (A8)
/// uses, and for the identical reason.
fn detect_normalize_candidates(msgs: &[ChatMessage]) -> Vec<usize> {
    let mut out = Vec::new();
    for (i, msg) in msgs.iter().enumerate() {
        if msg.role != Role::Tool {
            continue;
        }
        let Some(call_id) = msg.tool_call_id.as_deref() else {
            continue;
        };
        let named = msgs[..i].iter().rev().find_map(|m| {
            if m.role != Role::Assistant {
                return None;
            }
            m.tool_calls()
                .iter()
                .find(|c| c.id == call_id)
                .map(|c| c.function.name.clone())
        });
        if named.is_some_and(|name| normalize::NORMALIZE_TOOLS.contains(&name.as_str())) {
            out.push(i);
        }
    }
    out
}

/// One message index's freshness verdict from [`probe_read_freshness`].
#[derive(Debug, Clone, PartialEq, Eq)]
struct FreshEntry {
    fresh: bool,
    mtime: Option<i64>,
}

/// The output of [`probe_read_freshness`] (A8): per-message-index freshness
/// verdicts, threaded into [`project_messages`] via
/// [`ReductionPolicy::read_freshness`]. Opaque on purpose — build it only
/// through `probe_read_freshness`; the empty [`Default`] means "nothing is
/// fresh," so a policy with `elide_stale_reads` set but no probe run against
/// it elides nothing (fails closed).
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ReadFreshness {
    entries: HashMap<usize, FreshEntry>,
}

/// The A8 disk-probe pre-pass, deliberately kept OUTSIDE [`project_messages`]
/// so the pure projection core never touches the filesystem itself (SPEC.md
/// A8's purity requirement). Call this with the exact same `msgs` slice about
/// to be projected — indices must line up — and thread the result through
/// [`ReductionPolicy::read_freshness`] before calling [`project_messages`] (or
/// [`project_messages`]); it is only ever consulted when `policy.elide_stale_reads` is
/// set, so callers that never enable A8 can skip calling this entirely.
///
/// **Staleness check.** For each `detect_reads` hit, this re-reads the file
/// and compares content_hash("hash of the current bytes, lossy-UTF8-decoded")
/// against the hash of the tool result's *recorded* content — exactly the
/// transform `read_file` itself applies for a plain whole-file read
/// (`tools/builtins.rs:70-97`, no `offset`/`limit`, file under
/// `MAX_READ_BYTES`). This one comparison also naturally covers the two
/// harder cases without duplicating `read_file`'s own decoration/slicing
/// logic (which lives in a file this change does not touch):
/// - a sliced read (`offset`/`limit` given): the recorded content is a line
///   slice, never byte-identical to a raw whole-file re-read, so the hash
///   mismatches and the read is (correctly, conservatively) never fresh;
/// - a read whose original result already carried `read_file`'s own
///   oversize-truncation notice: same reasoning, the recorded content is not
///   raw file bytes, so it never matches a raw re-read.
///
/// Both are the documented SPEC.md A8 fallback ("mtime+len only, document")
/// taken to its simplest safe form: this implementation's fallback for
/// anything it cannot cheaply verify is "treat as changed" (never elide),
/// which only ever under-elides, never over-elides — the safe direction.
///
/// Unreadable/deleted files are likewise never fresh. `mtime` is recorded for
/// [`ReadLogEntry::mtime`] whenever the file's metadata is readable, even
/// when the freshness verdict itself is `false`.
pub fn probe_read_freshness(msgs: &[ChatMessage]) -> ReadFreshness {
    let mut entries = HashMap::new();
    for d in detect_reads(msgs) {
        let recorded = msgs[d.index].content.as_deref().unwrap_or("");
        let mtime = std::fs::metadata(&d.path)
            .ok()
            .and_then(|m| m.modified().ok())
            .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
            .map(|dur| dur.as_secs() as i64);
        let fresh = match std::fs::read(&d.path) {
            Err(_) => false, // unreadable/deleted -> never elide
            Ok(bytes) => {
                let text = String::from_utf8_lossy(&bytes);
                content_hash(text.as_bytes()) == content_hash(recorded.as_bytes())
            }
        };
        entries.insert(d.index, FreshEntry { fresh, mtime });
    }
    ReadFreshness { entries }
}

/// Populate the external disk-probe input required by A8 before projecting
/// `msgs`. All production projection/preflight sites route through this
/// helper so the context guard and the eventual provider request judge the
/// same stale-read savings. Other policy fields are untouched.
pub fn prepare_read_freshness(policy: &mut ReductionPolicy, msgs: &[ChatMessage]) {
    if policy.elide_stale_reads {
        policy.read_freshness = probe_read_freshness(msgs);
    }
}

// ---------------------------------------------------------------------------
// A9 — image redaction: data: URL detection
// ---------------------------------------------------------------------------

/// Parse a `data:` URL's declared media type: `data:<mediatype>[;base64],<data>`.
/// Returns `None` for anything not starting with the `data:` scheme (e.g. an
/// `https://…` image link, which A9 never touches — only inline base64
/// payloads are a redaction candidate). An empty or missing media type falls
/// back to `application/octet-stream` rather than failing the parse.
fn parse_data_url_mime(url: &str) -> Option<String> {
    let rest = url.strip_prefix("data:")?;
    let end = rest.find([';', ',']).unwrap_or(rest.len());
    let mime = &rest[..end];
    Some(if mime.is_empty() {
        "application/octet-stream".to_string()
    } else {
        mime.to_string()
    })
}

/// One `image_url` content part found in `msgs` whose `url` is a `data:` URL —
/// a candidate for [`ReductionKind::ImageRedacted`] once compared against
/// [`ReductionPolicy::image_redact_min_bytes`]. Remote (`https://…`) image
/// URLs and non-image parts are never candidates.
#[derive(Debug, Clone)]
struct DetectedImage {
    msg_index: usize,
    part_index: usize,
    mime: String,
    url_len: usize,
}

/// Find every `data:`-URL `image_url` content part in `msgs`. Already-redacted
/// parts are structurally excluded for free: [`project_messages`] replaces a
/// redacted part's JSON with a `{"type":"text", ...}` object, which this scan
/// no longer recognizes as an `image_url` part on a later re-projection — the
/// same reason prior reductions never need a separate "already reduced" guard
/// here the way A7/A8 do.
fn detect_images(msgs: &[ChatMessage]) -> Vec<DetectedImage> {
    let mut out = Vec::new();
    for (mi, msg) in msgs.iter().enumerate() {
        let Some(parts) = msg.content_parts.as_ref() else {
            continue;
        };
        for (pi, part) in parts.iter().enumerate() {
            if part.get("type").and_then(|t| t.as_str()) != Some("image_url") {
                continue;
            }
            let Some(url) = part
                .get("image_url")
                .and_then(|iu| iu.get("url"))
                .and_then(|u| u.as_str())
            else {
                continue;
            };
            let Some(mime) = parse_data_url_mime(url) else {
                continue; // not a data: URL -- e.g. a remote https:// link.
            };
            out.push(DetectedImage {
                msg_index: mi,
                part_index: pi,
                mime,
                url_len: url.len(),
            });
        }
    }
    out
}

// ---------------------------------------------------------------------------
// TR-10 — ToolInputElided: assistant-side tool_use argument elision
// ---------------------------------------------------------------------------

/// One structurally-eligible tool-input elision CANDIDATE found in `msgs`:
/// an assistant `tool_calls` entry naming a tool in `fields` (TR-10's
/// disk-persisted, write-family-by-default table), whose designated payload
/// field is present as a string, and whose paired tool result (matched by
/// `tool_call_id`, searched FORWARD from the assistant message — the
/// opposite direction from [`detect_reads`], which searches backward from a
/// tool result to its assistant call) exists and is
/// [`ToolOutcome::KnownSuccess`]. A call with no paired result yet (still
/// pending), an error, or an unknown result never appears here at all —
/// TR-10's success/failure boundary with TR-6 is enforced at DETECTION time,
/// never by a later filter.
#[derive(Debug, Clone)]
struct DetectedToolInput {
    msg_index: usize,
    call_id: String,
    tool_name: String,
    field: String,
    path: Option<PathBuf>,
    value: String,
}

/// Find every structurally-eligible [`DetectedToolInput`] in `msgs`. Pure and
/// side-effect free — no size/protection/prior-reduction filtering happens
/// here (mirrors [`detect_reads`]'s split: detection is unconditional, the
/// caller in [`project_messages`] applies the size threshold and the
/// already-reduced/protected/cleared-range guards).
fn detect_tool_inputs(
    msgs: &[ChatMessage],
    fields: &HashMap<String, String>,
) -> Vec<DetectedToolInput> {
    let mut out = Vec::new();
    for (i, msg) in msgs.iter().enumerate() {
        if msg.role != Role::Assistant {
            continue;
        }
        for call in msg.tool_calls() {
            let Some(field) = fields.get(&call.function.name) else {
                continue;
            };
            let Ok(args) = call.function.parsed_arguments() else {
                continue;
            };
            let Some(value) = args.get(field.as_str()).and_then(|v| v.as_str()) else {
                continue;
            };
            let Some(result) = msgs[i + 1..].iter().find(|m| {
                m.role == Role::Tool && m.tool_call_id.as_deref() == Some(call.id.as_str())
            }) else {
                continue; // Still pending: never a candidate.
            };
            if tool_outcome(result) != ToolOutcome::KnownSuccess {
                continue; // Error or unknown: never eligible for TR-10.
            }
            // `path` (this crate's own `write_file`) and `file_path` (Claude
            // Code's native `Write`) are the two real-world spellings; a
            // sibling argument under neither name just means the stub's
            // `path` field stays `None` (never a hard failure).
            let path = args
                .get("path")
                .or_else(|| args.get("file_path"))
                .and_then(|v| v.as_str())
                .map(PathBuf::from);
            out.push(DetectedToolInput {
                msg_index: i,
                call_id: call.id.clone(),
                tool_name: call.function.name.clone(),
                field: field.clone(),
                path,
                value: value.to_string(),
            });
        }
    }
    out
}

/// TR-6: the FAILURE-side complement of [`detect_tool_inputs`] — find every
/// structurally-eligible errored-call oversized-input candidate: an
/// assistant `tool_calls` entry naming a tool in `fields`, whose designated
/// payload field is present as a string, and whose paired tool result
/// (matched by `tool_call_id`, searched forward, identical pairing to
/// `detect_tool_inputs`) exists and is [`ToolOutcome::KnownError`]. A call
/// with no paired result yet, a known-success result, or an unknown result
/// never appears here at all: unknown Codex v1 outcomes fail closed on BOTH
/// reduction paths rather than being guessed from free-form text.
fn detect_errored_tool_inputs(
    msgs: &[ChatMessage],
    fields: &HashMap<String, String>,
) -> Vec<DetectedToolInput> {
    let mut out = Vec::new();
    for (i, msg) in msgs.iter().enumerate() {
        if msg.role != Role::Assistant {
            continue;
        }
        for call in msg.tool_calls() {
            let Some(field) = fields.get(&call.function.name) else {
                continue;
            };
            let Ok(args) = call.function.parsed_arguments() else {
                continue;
            };
            let Some(value) = args.get(field.as_str()).and_then(|v| v.as_str()) else {
                continue;
            };
            let Some(result) = msgs[i + 1..].iter().find(|m| {
                m.role == Role::Tool && m.tool_call_id.as_deref() == Some(call.id.as_str())
            }) else {
                continue; // Still pending: never a candidate (either side).
            };
            if tool_outcome(result) != ToolOutcome::KnownError {
                continue; // Success or unknown: never eligible for TR-6.
            }
            let path = args
                .get("path")
                .or_else(|| args.get("file_path"))
                .and_then(|v| v.as_str())
                .map(PathBuf::from);
            out.push(DetectedToolInput {
                msg_index: i,
                call_id: call.id.clone(),
                tool_name: call.function.name.clone(),
                field: field.clone(),
                path,
                value: value.to_string(),
            });
        }
    }
    out
}

/// TR-6: the number of `Role::Assistant` messages appearing strictly after
/// `index` in `msgs` — the "N turns elapsed" aging clock for errored-input
/// pruning, a message-count proxy for "turns" (this codebase has no other
/// structural definition of a conversational turn; A10's own
/// `clear_turns_older_than` is likewise a message-count threshold, not a
/// literal turn counter).
fn assistant_turns_since(msgs: &[ChatMessage], index: usize) -> usize {
    msgs.get(index + 1..)
        .map(|rest| rest.iter().filter(|m| m.role == Role::Assistant).count())
        .unwrap_or(0)
}

/// A UTF-8-safe ASCII-whitespace skip, byte-indexed — the primitive
/// [`find_top_level_string_field`]/[`skip_json_value`] share.
fn skip_ws(b: &[u8], mut i: usize) -> usize {
    while i < b.len() && b[i].is_ascii_whitespace() {
        i += 1;
    }
    i
}

/// Parse one JSON string starting at `b[i] == '"'`. Returns
/// `(content_start, content_end, after)`: `content_start..content_end` bounds
/// the RAW (still `\`-escaped) string body (excluding the surrounding
/// quotes), and `after` is the index just past the closing quote. Byte-wise
/// scanning is UTF-8-safe here: JSON's only structural bytes inside a string
/// (`"` = 0x22, `\` = 0x5c) are ASCII values that can never appear as part of
/// a multi-byte UTF-8 continuation/lead byte (those are always >= 0x80), and
/// skipping exactly one byte after a `\` is always safe — every JSON escape
/// (`\"`, `\\`, `\/`, `\b`, `\f`, `\n`, `\r`, `\t`, `\uXXXX`) has an
/// unambiguous, never-`"`-or-`\` byte immediately after the backslash.
fn parse_json_string(b: &[u8], i: usize) -> Option<(usize, usize, usize)> {
    if i >= b.len() || b[i] != b'"' {
        return None;
    }
    let content_start = i + 1;
    let mut j = content_start;
    while j < b.len() {
        match b[j] {
            b'\\' => j += 2,
            b'"' => return Some((content_start, j, j + 1)),
            _ => j += 1,
        }
    }
    None // Unterminated string.
}

/// Skip over one arbitrary JSON value (string/object/array/number/bool/null)
/// starting at (possibly whitespace before) `b[i]`. Returns the index just
/// past it. Used by [`find_top_level_string_field`] to jump over sibling
/// fields it isn't looking for, however they're shaped, without needing to
/// interpret them.
fn skip_json_value(b: &[u8], i: usize) -> Option<usize> {
    let i = skip_ws(b, i);
    if i >= b.len() {
        return None;
    }
    match b[i] {
        b'"' => parse_json_string(b, i).map(|(_, _, end)| end),
        b'{' | b'[' => {
            let open = b[i];
            let close = if open == b'{' { b'}' } else { b']' };
            let mut depth = 0usize;
            let mut j = i;
            loop {
                if j >= b.len() {
                    return None;
                }
                match b[j] {
                    b'"' => {
                        let (_, _, end) = parse_json_string(b, j)?;
                        j = end;
                    }
                    c if c == open => {
                        depth += 1;
                        j += 1;
                    }
                    c if c == close => {
                        depth -= 1;
                        j += 1;
                        if depth == 0 {
                            return Some(j);
                        }
                    }
                    _ => j += 1,
                }
            }
        }
        _ => {
            // number / true / false / null: scan to the next structural byte.
            let mut j = i;
            while j < b.len() && !matches!(b[j], b',' | b'}' | b']') && !b[j].is_ascii_whitespace()
            {
                j += 1;
            }
            Some(j)
        }
    }
}

/// Locate the byte span of the JSON STRING VALUE for top-level key `field`
/// within `json` — a serialized tool-call `arguments` string, assumed (like
/// every built-in write-family tool's flat schema) to be a JSON object.
/// Returns `(value_start, value_end)`: `json[value_start..value_end]` is the
/// RAW (still-escaped) string body, excluding the surrounding quotes — so a
/// caller can replace ONLY that span, leaving every other byte (key order,
/// whitespace, sibling fields, however shaped) untouched. Returns `None` when
/// `field` is absent, its value isn't a JSON string, or `json` isn't a
/// well-formed object — always a safe "don't touch it" signal, never a
/// guess.
fn find_top_level_string_field(json: &str, field: &str) -> Option<(usize, usize)> {
    let b = json.as_bytes();
    let mut i = skip_ws(b, 0);
    if i >= b.len() || b[i] != b'{' {
        return None;
    }
    i += 1;
    loop {
        i = skip_ws(b, i);
        if i >= b.len() {
            return None;
        }
        if b[i] == b'}' {
            return None; // Field not found.
        }
        let (key_start, key_end, after_key) = parse_json_string(b, i)?;
        let key = &json[key_start..key_end];
        i = skip_ws(b, after_key);
        if i >= b.len() || b[i] != b':' {
            return None;
        }
        i = skip_ws(b, i + 1);
        if i >= b.len() {
            return None;
        }
        if key == field {
            return if b[i] == b'"' {
                let (val_start, val_end, _) = parse_json_string(b, i)?;
                Some((val_start, val_end))
            } else {
                None // The field exists but isn't a string value.
            };
        }
        i = skip_json_value(b, i)?;
        i = skip_ws(b, i);
        match b.get(i) {
            Some(b',') => {
                i += 1;
                continue;
            }
            Some(b'}') => return None, // Reached the end without a match.
            _ => return None,          // Malformed / unexpected trailing bytes.
        }
    }
}
/// JSON-escape `s` for embedding as a string VALUE (no surrounding quotes) —
/// the content half of what `serde_json::to_string` would produce for it.
fn json_escape_content(s: &str) -> String {
    let quoted = serde_json::to_string(s).unwrap_or_default();
    let len = quoted.len();
    if len >= 2 {
        quoted[1..len - 1].to_string()
    } else {
        String::new()
    }
}

/// Byte-surgical replacement of ONE top-level string field's value inside a
/// serialized JSON object: every other byte (key order, whitespace, sibling
/// fields) is preserved character-for-character (SPEC.md TR-10: "replace
/// only the payload field's value" — never a full reparse+reserialize, which
/// would reformat/reorder the rest of the arguments). Returns `None` (never
/// touching `json`) when `field` isn't present as a top-level string-valued
/// key.
fn replace_top_level_string_field(json: &str, field: &str, new_value: &str) -> Option<String> {
    let (start, end) = find_top_level_string_field(json, field)?;
    let mut out = String::with_capacity(json.len() + new_value.len());
    out.push_str(&json[..start]);
    out.push_str(&json_escape_content(new_value));
    out.push_str(&json[end..]);
    Some(out)
}

/// Resolve and hash-verify the ORIGINAL value of a
/// [`ReductionKind::ToolInputElided`] reduction's payload field: locate the
/// addressed message, the tool_call within it named by `call_id`, extract
/// `field`'s value, and verify it against `ptr.content_hash` before ever
/// handing it back. Takes a message slice rather than a [`Session`] for the
/// same reason [`resolve_original_content`] does — both `invert`'s
/// sidecar-backed callers and [`rehydrate`]'s `minted_view`-backed caller
/// share this one resolver.
fn resolve_tool_input_value(
    ptr: &SidecarPtr,
    call_id: &str,
    field: &str,
    messages: &[ChatMessage],
) -> Result<String> {
    let msg = messages.get(ptr.addr.index).ok_or_else(|| {
        Error::new(format!(
            "invert: sidecar has no message at index {} (reduction pointer unresolvable)",
            ptr.addr.index
        ))
    })?;
    if msg.role != ptr.addr.role {
        return Err(Error::new(format!(
            "invert: role mismatch at sidecar index {}: pointer expects {:?}, sidecar has {:?}",
            ptr.addr.index, ptr.addr.role, msg.role
        )));
    }
    let call = msg
        .tool_calls()
        .iter()
        .find(|c| c.id == call_id)
        .ok_or_else(|| {
            Error::new(format!(
                "invert: sidecar message at index {} has no tool_call with id {call_id}",
                ptr.addr.index
            ))
        })?;
    let parsed = call.function.parsed_arguments().map_err(|e| {
        Error::new(format!(
            "invert: tool_call {call_id} arguments are not valid JSON: {e}"
        ))
    })?;
    let value = parsed
        .get(field)
        .and_then(|v| v.as_str())
        .ok_or_else(|| {
            Error::new(format!(
                "invert: tool_call {call_id} has no string field `{field}`"
            ))
        })?
        .to_string();
    ptr.verify(value.as_bytes())?;
    Ok(value)
}

/// The decision a future escalation orchestrator (SPEC.md D13/wave-5
/// `escalate()` — not yet implemented in this codebase) should take for one
/// existing [`ReductionKind::ToolInputElided`] stub, per TR-10's freshness
/// matrix (reused from A8's `probe_read_freshness` pattern): does disk still
/// match what was written?
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EscalationAction {
    /// The stub is still faithful to disk — leave it as a stub (D13's
    /// "smallest faithful context", re-derivable at ~0 extra tokens).
    KeepStub,
    /// The stub is no longer faithful (disk has changed since, or is gone) —
    /// only the sidecar's recorded original is still faithful; rehydrate via
    /// [`invert_one_messages`].
    RehydrateFromSidecar,
}

/// TR-10's freshness-matrix decision, given whether [`probe_tool_input_fresh`]
/// found the file still matching: `fresh` -> [`EscalationAction::KeepStub`],
/// `!fresh` -> [`EscalationAction::RehydrateFromSidecar`]. Split from the
/// probe itself (which does the actual disk I/O) so this half stays a pure,
/// trivially-testable function — the same purity discipline A8's
/// `probe_read_freshness`/`project_messages` split follows.
pub fn tool_input_escalation_action(fresh: bool) -> EscalationAction {
    if fresh {
        EscalationAction::KeepStub
    } else {
        EscalationAction::RehydrateFromSidecar
    }
}

/// The A8-style disk probe behind [`tool_input_escalation_action`]: does the
/// file at `path` currently on disk still hash to `content_hash_hex`? Unlike
/// [`probe_read_freshness`] (which lossy-UTF8-decodes before hashing, to
/// mirror `read_file`'s own transform), this hashes the RAW bytes directly —
/// `write_file` writes `content.as_bytes()` with no transform, so the exact
/// bytes on disk are the fairer comparison. Unreadable/deleted files are
/// never fresh (fails closed, the same direction A8 fails in). A free
/// function (no `Reduction`/`ReductionLog` coupling) so it composes with
/// whatever wave-5 `escalate()` orchestration eventually calls it.
pub fn probe_tool_input_fresh(path: &std::path::Path, content_hash_hex: &str) -> bool {
    match std::fs::read(path) {
        Err(_) => false,
        Ok(bytes) => content_hash(&bytes) == content_hash_hex,
    }
}

/// The largest `end <= target` such that `s.is_char_boundary(end)` — a
/// UTF-8-safe truncation point. Mirrors the boundary walk in
/// `agent.rs::cap_tool_output`.
fn char_boundary_floor(s: &str, target: usize) -> usize {
    let mut end = target.min(s.len());
    while end > 0 && !s.is_char_boundary(end) {
        end -= 1;
    }
    end
}

/// Make an untrusted string fragment safe for interpolation into a stub
/// summary: [`stub::format`]'s grammar contract is "one line, no `]`", and a
/// tool name arrives verbatim from imported JSONL — attacker-shaped input. A
/// name containing `]` or a newline would trip `stub::format`'s
/// `debug_assert` (a panic in debug builds) and, in release builds, mint a
/// grammar-breaking stub that [`stub::parse`] rejects. Every `]` and every
/// control character is replaced with `_` — visible, honest damage instead
/// of a broken line.
fn sanitize_summary_fragment(s: &str) -> String {
    s.chars()
        .map(|c| if c == ']' || c.is_control() { '_' } else { c })
        .collect()
}

/// Rebuild the exact reduced content for a [`ReductionKind::ToolOutputTruncated`]
/// reduction, given the *original* (unreduced) content at its target address:
/// the kept prefix, `"\n\n"`, then the recorded placeholder line — byte-for-byte
/// the same construction `project` used the first time it created `r`.
fn rebuild_truncated_content(original: &str, r: &Reduction) -> String {
    let kept = r.ptr.span.map(|(kept, _total)| kept).unwrap_or(0);
    let kept = kept.min(original.len());
    let mut s = original[..kept].to_string();
    s.push_str("\n\n");
    s.push_str(&r.placeholder);
    s
}

/// Rebuild the exact reduced content for a [`ReductionKind::OutputNormalized`]
/// reduction, given the *original* (raw, unreduced) content at its target
/// address: [`normalize::normalize`] is a pure deterministic function, so
/// re-running it against the same original bytes reproduces byte-identical
/// normalized text every time; `"\n\n"` then the recorded placeholder
/// (carrying the honesty trailer) is appended exactly as `project_messages`
/// did the first time it created `r` — the same "recompute, don't store the
/// derived text" strategy [`rebuild_truncated_content`] uses for A7.
fn rebuild_normalized_content(original: &str, r: &Reduction) -> String {
    let mut s = normalize::normalize(original);
    s.push_str("\n\n");
    s.push_str(&r.placeholder);
    s
}

/// Build the exact reduced content for a [`ReductionKind::FileReadDiffed`]
/// reduction: the stored placeholder line, a newline, then a freshly
/// recomputed unified diff of `base_text` (the base read's full content)
/// against `new_text` (this read's full content). Recomputing the diff
/// (rather than storing it) keeps `Reduction` itself small and — since
/// `diffy::create_patch` is a pure function of its two inputs — reproduces
/// byte-identically every time, the same prefix-stability guarantee
/// [`rebuild_truncated_content`] gives A7.
fn rebuild_diffed_content(base_text: &str, new_text: &str, r: &Reduction) -> String {
    let diff = diffy::create_patch(base_text, new_text);
    let mut s = r.placeholder.clone();
    s.push('\n');
    s.push_str(&diff.to_string());
    s
}

/// Reapply one already-applied reduction from `prior` onto `view` in place,
/// exactly reproducing its placeholder — this is the prefix-stability
/// guarantee: an older reduction never churns between projections. `msgs` is
/// the pristine, never-mutated canonical slice `project_messages` was called
/// with — [`ReductionKind::FileReadDiffed`] resolves both its base and new
/// text from it (never from `view`), so a base that itself carries some
/// OTHER reduction in `view` (e.g. it was `FileReadElided` before later being
/// superseded as a diff base) never corrupts the recomputed diff.
///
/// [`ReductionKind::ToolOutputTruncated`], [`ReductionKind::FileReadElided`],
/// [`ReductionKind::ImageRedacted`], [`ReductionKind::OutputNormalized`],
/// [`ReductionKind::FileReadDiffed`], [`ReductionKind::DuplicateOutput`], and
/// [`ReductionKind::Superseded`] are all cardinality-preserving (content
/// mutated in place, `view`'s length and msgs-index alignment are untouched)
/// — `FileReadElided` simply
/// replaces the whole content with the stored placeholder verbatim
/// (whole-content elision, `ptr.span = None`), regardless of the file's
/// CURRENT on-disk state: prior-log stability means an already-elided read
/// stays elided even after the file changes again — a changed file only ever
/// blocks *new* elisions (SPEC.md A8), it never un-elides an existing one.
/// `ImageRedacted` likewise replaces the addressed content part with a
/// `{"type":"text", ...}` object carrying the stored placeholder verbatim,
/// regardless of the current part at that index. `OutputNormalized`
/// recomputes the normalized text from the *original* content at that index
/// via [`rebuild_normalized_content`] (pure/deterministic, so it reproduces
/// byte-identically). `DuplicateOutput` (TR-2) replaces the whole content
/// with the stored placeholder verbatim too (`ptr.span = None`, exactly like
/// `FileReadElided`) — irrespective of whether its `canonical` address is
/// still a plain message, itself now reduced, or has since been swallowed by
/// a `TurnsCleared` range: the duplicate's own placeholder never depends on
/// the canonical's current shape. [`ReductionKind::TurnsCleared`] (A10) is
/// NOT cardinality-preserving: it collapses a whole range `[first..=last]`
/// down to the single stored placeholder message via `Vec::splice`, so it
/// must be the LAST reapplication performed on `view` in any given
/// `project_messages` call (everything else addresses `view` by msgs-index,
/// which this invalidates for every index past `first`).
fn reapply_reduction(view: &mut Vec<ChatMessage>, r: &Reduction, msgs: &[ChatMessage]) {
    match &r.kind {
        ReductionKind::ToolOutputTruncated { .. } => {
            let idx = r.ptr.addr.index;
            let Some(msg) = view.get_mut(idx) else {
                return; // Addressed message no longer present; nothing to reapply.
            };
            if let Some(original) = msg.content.clone() {
                msg.content = Some(rebuild_truncated_content(&original, r));
            }
            set_reduction_id(msg, &r.id);
        }
        ReductionKind::OutputNormalized { .. } => {
            let idx = r.ptr.addr.index;
            let Some(msg) = view.get_mut(idx) else {
                return; // Addressed message no longer present; nothing to reapply.
            };
            if let Some(original) = msg.content.clone() {
                msg.content = Some(rebuild_normalized_content(&original, r));
            }
            set_reduction_id(msg, &r.id);
        }
        ReductionKind::FileReadElided { .. } => {
            let idx = r.ptr.addr.index;
            let Some(msg) = view.get_mut(idx) else {
                return; // Addressed message no longer present; nothing to reapply.
            };
            msg.content = Some(r.placeholder.clone());
            set_reduction_id(msg, &r.id);
        }
        ReductionKind::FileReadDiffed { base, .. } => {
            let idx = r.ptr.addr.index;
            if view.get(idx).is_none() {
                return; // Addressed message no longer present; nothing to reapply.
            }
            let (Some(base_text), Some(new_text)) = (
                msgs.get(base.index).and_then(|m| m.content.as_deref()),
                msgs.get(idx).and_then(|m| m.content.as_deref()),
            ) else {
                return; // Base or new content no longer resolvable against `msgs`.
            };
            let content = rebuild_diffed_content(base_text, new_text, r);
            let msg = &mut view[idx];
            msg.content = Some(content);
            set_reduction_id(msg, &r.id);
        }
        ReductionKind::TurnsCleared { first, last, .. } => {
            if *first > *last || *last >= view.len() {
                return; // Range no longer resolvable against this `view`; nothing to reapply.
            }
            let mut placeholder = ChatMessage::system(r.placeholder.clone());
            set_reduction_id(&mut placeholder, &r.id);
            view.splice(*first..=*last, std::iter::once(placeholder));
        }
        ReductionKind::ImageRedacted { part_index } => {
            let idx = r.ptr.addr.index;
            let Some(msg) = view.get_mut(idx) else {
                return; // Addressed message no longer present; nothing to reapply.
            };
            if let Some(parts) = msg.content_parts.as_mut() {
                if let Some(part) = parts.get_mut(*part_index) {
                    *part = serde_json::json!({"type": "text", "text": r.placeholder});
                }
            }
            set_reduction_id(msg, &r.id);
        }
        ReductionKind::ToolInputElided { call_id, field, .. } => {
            let idx = r.ptr.addr.index;
            // Compute the spliced arguments against the pristine `view[idx]`
            // (freshly derived from `msgs` at the top of `project_messages`,
            // so this always re-derives from the ORIGINAL arguments) before
            // taking a mutable borrow, mirroring the immutable-then-mutable
            // two-step `ImageRedacted` above uses.
            let Some(spliced) = view
                .get(idx)
                .and_then(|m| m.tool_calls.as_ref())
                .and_then(|calls| calls.iter().find(|c| &c.id == call_id))
                .and_then(|call| {
                    replace_top_level_string_field(&call.function.arguments, field, &r.placeholder)
                })
            else {
                return; // Addressed call no longer present, or reshaped; nothing to reapply.
            };
            let Some(msg) = view.get_mut(idx) else {
                return;
            };
            if let Some(calls) = msg.tool_calls.as_mut() {
                if let Some(call) = calls.iter_mut().find(|c| &c.id == call_id) {
                    call.function.arguments = spliced;
                }
            }
            set_reduction_id(msg, &r.id);
        }
        ReductionKind::DuplicateOutput { .. } => {
            let idx = r.ptr.addr.index;
            let Some(msg) = view.get_mut(idx) else {
                return; // Addressed message no longer present; nothing to reapply.
            };
            msg.content = Some(r.placeholder.clone());
            set_reduction_id(msg, &r.id);
        }
        ReductionKind::Superseded { .. } => {
            let idx = r.ptr.addr.index;
            let Some(msg) = view.get_mut(idx) else {
                return; // Addressed message no longer present; nothing to reapply.
            };
            msg.content = Some(r.placeholder.clone());
            set_reduction_id(msg, &r.id);
        }
    }
}

/// Count how many of `msgs` have each of the three conversational roles
/// (user/assistant/tool) — used to build the A10 `turns-cleared` stub's
/// `(N messages: A user, B assistant, C tool)` summary clause.
fn count_roles(msgs: &[ChatMessage]) -> (usize, usize, usize) {
    let mut user = 0;
    let mut assistant = 0;
    let mut tool = 0;
    for m in msgs {
        match m.role {
            Role::User => user += 1,
            Role::Assistant => assistant += 1,
            Role::Tool => tool += 1,
            Role::System => {}
        }
    }
    (user, assistant, tool)
}

/// The blake3 hash [`resolve_turns_range`] verifies for a [`ReductionKind::TurnsCleared`]
/// pointer — blake3 over each message's wire-serialized (`serde_json`) bytes
/// in `msgs`, concatenated in order — plus that concatenation's total byte
/// length (the creating side puts it in the stub summary so a reader can
/// judge the cleared range's size; the resolving side ignores it). Shared by
/// the creating side (here) and the resolving side (`resolve_turns_range`)
/// so the two hash formulas can never drift apart.
fn hash_turns_range(msgs: &[ChatMessage]) -> Result<(String, usize)> {
    let mut combined = Vec::new();
    for m in msgs {
        let bytes = serde_json::to_vec(m)
            .map_err(|e| Error::new(format!("failed to serialize message: {e}")))?;
        combined.extend_from_slice(&bytes);
    }
    Ok((content_hash(&combined), combined.len()))
}

/// Pure function of `(msgs, policy, prior)`: the reduction engine's actual
/// body (SPEC.md A5, A7, A8, A9, A10, TR-2). A session adapter can delegate
/// its canonical message slice here; a runtime agent loop calls this
/// directly against `history[1..]` so a live agent can build the projected
/// request view without needing a `Session` wrapper around its own history.
///
/// Deterministic and side-effect free: identical inputs produce byte-identical
/// output; `msgs` is never mutated; nothing here touches the filesystem —
/// including for A8: `policy.read_freshness` is precomputed data, populated by
/// [`probe_read_freshness`] (the one place disk I/O happens) before this is
/// ever called. Every reduction already recorded in `prior` reproduces
/// verbatim (same id, same placeholder, byte-identical stub) — new reductions
/// only ever target messages older than the protected tail, so the reduced
/// prefix stays cache-stable across turns.
///
/// Pass order: TR-2 ([`ReductionKind::DuplicateOutput`], content-hash dedup)
/// runs FIRST, then TR-6 ([`ReductionKind::Superseded`], same-tool/
/// canonicalized-args keep-latest), then T30/TR-4
/// ([`ReductionKind::OutputNormalized`], ANSI/redraw collapse), then
/// [`ReductionKind::ToolOutputTruncated`] (A7), then the read-family passes —
/// [`ReductionKind::FileReadElided`] (A8) for an unchanged re-read,
/// [`ReductionKind::FileReadDiffed`] (TR-3) for a changed one — then
/// [`ReductionKind::ImageRedacted`] (A9), then TR-10
/// ([`ReductionKind::ToolInputElided`], both the successful-call case and
/// TR-6's failed-call errored-input-pruning complement), then
/// [`ReductionKind::TurnsCleared`] (A10) last (the one cardinality-changing
/// pass). Each of TR-2, TR-6, T30/TR-4, and A7 claims a message's index in
/// `reduced_this_run` the moment it mints a reduction for it, and every pass
/// after the first checks that set — so a single message is claimed by
/// exactly one pass per `project_messages` call, never two.
///
/// TR-2 running before TR-6, T30/TR-4, and A7 means a byte-identical
/// duplicate is deduped — the cheapest of the four reductions — rather than
/// independently superseded, normalized, or truncated. TR-6 running before
/// T30/TR-4 and A7 follows the identical reasoning one step further: a
/// result about to be superseded down to one small stub never needs
/// normalizing or truncating first either. See the TR-2 and TR-6 passes
/// below for why this ordering must hold within a single call, not just
/// "eventually" (in short: both verdicts are pure functions of `msgs`, so
/// they are unaffected by running before or after T30/TR-4, but T30/TR-4 and
/// A7 both read from `view`, which TR-2/TR-6 may have already stubbed — so
/// TR-2 then TR-6 must go first, or their own claims could lose a race to a
/// pass that mutates `view` ahead of them).
/// T30/TR-4 running before A7 means a noisy bash/exec output is collapsed to
/// its final rendered content BEFORE A7 ever measures it against
/// `tool_output_trigger_bytes` — but T30/TR-4 only CLAIMS the message (taking
/// it out of A7's candidate pool) when its own normalized rendering already
/// fits under `tool_output_trigger_bytes`; that rendering is what rides the
/// wire, already bounded, so no truncation is needed on top of it. When the
/// normalized rendering is STILL over the trigger — genuinely large, mostly
/// distinct content, not just redraw noise — T30/TR-4 deliberately does not
/// claim the message at all (raw, untouched) and lets it fall through to A7
/// below, which truncates the RAW bytes to `tool_output_keep_bytes`. Either
/// way the wire payload for a terminal output is bounded by A7's trigger —
/// the P7 runaway-output safety net (SPEC.md/TR-12) is preserved for BOTH
/// small-after-normalization and large-after-normalization outputs. TR-2 and
/// TR-6 both never claim a read-type tool result (`detect_reads`), even a
/// byte-identical or same-args re-read: the read-family passes own that
/// address space exclusively, with strictly more information (path-aware
/// freshness, a unified diff) than either TR-2's flat "identical to msg #N"
/// or TR-6's flat "superseded by msg #N" stub could express; T30/TR-4 is
/// likewise scoped to [`normalize::NORMALIZE_TOOLS`] tool identities,
/// disjoint from `READ_TOOLS`, so it never contends with the read-family
/// passes over the same index either.
pub fn project_messages(
    msgs: &[ChatMessage],
    policy: &ReductionPolicy,
    prior: &ReductionLog,
) -> (Vec<ChatMessage>, ReductionLog) {
    let mut view: Vec<ChatMessage> = msgs.to_vec();
    let mut log = prior.clone();

    // Reproduce every already-applied ToolOutputTruncated (etc.) reduction
    // verbatim first: cardinality-preserving, so `view` stays index-parallel
    // to `msgs` while this runs. TurnsCleared (cardinality-changing) is
    // handled last, below, once every msgs-indexed operation is done.
    for r in &prior.reductions {
        if !matches!(r.kind, ReductionKind::TurnsCleared { .. }) {
            reapply_reduction(&mut view, r, msgs);
        }
    }

    // A10: old-turn clearing is a one-time context edit for the AUTO-
    // COMPACTOR (`Agent::maybe_compact`'s `clear_turns_older_than`), not a
    // repeating compaction — once ANY `TurnsCleared` record exists, the
    // auto-compactor never mints a new one (below). But a session can carry
    // MORE than one `TurnsCleared` record: TR-9 (T24) `handoff` establishes a
    // whole set of disjoint spanning clears in one shot (a handoff keep-set
    // is generally scattered — system prompt + some named early turns + last
    // K — so the non-kept middle forms several contiguous gaps, one
    // `TurnsCleared` per gap). So this collects EVERY existing record rather
    // than just the first found; each one, once established, is reapplied
    // verbatim forever (never widened, never recomputed), so no placeholder
    // ever churns between turns.
    let mut existing_clears: Vec<(usize, usize)> = prior
        .reductions
        .iter()
        .chain(prior.expanded.iter())
        .filter_map(|r| match r.kind {
            ReductionKind::TurnsCleared { first, last, .. } => Some((first, last)),
            _ => None,
        })
        .collect();
    existing_clears.sort_by_key(|&(f, _)| f);
    let in_existing_clear = |i: usize| existing_clears.iter().any(|&(f, l)| i >= f && i <= l);

    let already_reduced: HashSet<usize> = prior
        .reductions
        .iter()
        .chain(prior.expanded.iter())
        .filter(|r| !matches!(r.kind, ReductionKind::TurnsCleared { .. }))
        .map(|r| r.ptr.addr.index)
        .collect();

    // Protect the newest `protect_last_n_tool_results` tool results from ever
    // becoming a *new* candidate (prior reductions on now-recent messages are
    // left as-is above — stability wins over re-protecting them).
    let protected: HashSet<usize> = view
        .iter()
        .enumerate()
        .filter(|(_, m)| m.role == Role::Tool)
        .map(|(i, _)| i)
        .rev()
        .take(policy.protect_last_n_tool_results)
        .collect();

    // Indices reduced (of any kind) during THIS call — as opposed to
    // `already_reduced` (reduced in a PRIOR call). A message can only ever
    // carry one reduction kind at a time, so every pass below must skip
    // anything an EARLIER pass this run just claimed, in addition to
    // everything `already_reduced`/`protected`/`in_existing_clear` already
    // exclude. Declared before TR-2 (the first pass to populate it) rather
    // than before A7, so A7's own candidate filter can already respect it.
    // `ordinal` is likewise shared by every pass below and only ever consumed
    // on an actual mint (a `continue`d candidate never advances it), so ids
    // stay dense and deterministic regardless of which passes end up firing.
    let mut reduced_this_run: HashSet<usize> = HashSet::new();

    // Logs can be sparse after an older client expanded a record by simply
    // removing it. Counting records can therefore reuse a still-live
    // ordinal (and, with the same hash prefix, the exact same id). Always
    // advance past the greatest parseable persisted ordinal across both
    // active and explicitly-expanded records.
    let mut ordinal = next_reduction_ordinal(
        prior
            .reductions
            .iter()
            .chain(prior.expanded.iter())
            .map(|r| r.id.as_str()),
    );

    // ---- TR-2: DuplicateOutput -- content-hash dedup of identical tool
    // outputs ----
    //
    // Runs BEFORE both T30/TR-4 (OutputNormalized, immediately below) and A7:
    // deduping a later duplicate is strictly cheaper than either
    // independently normalizing or truncating it, and — more importantly —
    // it must claim a duplicate's index in `reduced_this_run` before either
    // of those passes' own candidate scans run, or they would claim it first
    // and that claim would stick forever (prefix stability: `already_reduced`
    // never lets a later run downgrade an existing reduction to a cheaper
    // kind). Hash comparisons are taken from `msgs` (the ORIGINAL,
    // never-mutated slice) rather than `view`, exactly like A10's
    // `hash_turns_range` below — so the canonical bytes compared are always
    // the true original content, an already-stubbed earlier occurrence's
    // placeholder text is never what gets hashed. This also makes TR-2's own
    // verdict completely insensitive to relative pass order with
    // OutputNormalized: a terminal output that is BOTH a normalize candidate
    // (raw ANSI/CR noise) AND byte-identical to an earlier tool result is
    // claimed HERE by TR-2 — a `DuplicateOutput` stub is typically far
    // smaller than even a normalized rendering — and OutputNormalized's own
    // candidate filter (below) skips it via `reduced_this_run`. Neither pass
    // ever touches the stored sidecar bytes (both keep the RAW capture
    // there, A3), so this precedence is purely about which single stub wins
    // the VIEW, never about invert-to-raw correctness for either kind.
    //
    // The first (chronologically earliest) still-addressable occurrence of a
    // content hash is the canonical; every later occurrence with the same
    // hash becomes a `DuplicateOutput` candidate, provided it isn't already
    // reduced, protected, or about to be swallowed by an existing
    // `TurnsCleared` range. `SidecarPtr::addr` on the minted reduction is the
    // DUPLICATE's own address (not the canonical's) — same self-addressing
    // convention as every other kind — so `invert`/`expand_reduction` always
    // resolve it independent of whatever later happens to the canonical's
    // own slot (dev/04: the canonical may itself be truncated or cleared in
    // a later run without ever affecting this pointer).
    //
    // ONLY same-path re-reads (`detect_reads` hits whose path already
    // appeared in an EARLIER detected read this slice) are excluded from
    // this pass entirely — canonical registration AND duplicate candidacy —
    // and left for the A8/TR-3 pass below (post-merge reconciliation: TR-2
    // landed generalizing A8's OWN prior dedup special-case to "any tool
    // output", but a re-read of a file already carries a strictly richer,
    // path-aware redundancy mechanism there — freshness-gated elision for an
    // unchanged re-read, a unified diff for a changed one — that TR-2's flat
    // "identical to msg #N" stub would otherwise pre-empt whenever a re-read
    // happens to be byte-identical to its own prior read, silently losing
    // the elision-vs-diff distinction SPEC.md TR-3 dev/05 requires).
    //
    // Deliberately narrower than "every detected read": the FIRST read of a
    // given path has no earlier same-path read for the A8/TR-3 pass to
    // diff/elide against, so it is not that pass's address space at all —
    // TR-2 must stay free to dedup it against a byte-identical output
    // anywhere else in the slice (a different path's read, or any other
    // tool result), exactly as it would for any other tool. Excluding every
    // read unconditionally (the pre-fix behavior) silently disabled TR-2 for
    // two content-identical reads of DIFFERENT paths, which A8/TR-3's
    // path-keyed matching never claims and never will.
    let read_indices: HashSet<usize> = {
        let detected = detect_reads(msgs);
        let mut seen_paths: HashSet<&std::path::Path> = HashSet::new();
        detected
            .iter()
            .filter(|d| !seen_paths.insert(d.path.as_path()))
            .map(|d| d.index)
            .collect()
    };
    let mut first_seen: HashMap<String, usize> = HashMap::new();
    for (i, m) in msgs.iter().enumerate() {
        if !policy.deduplicate_outputs || m.role != Role::Tool || read_indices.contains(&i) {
            continue;
        }
        let Some(content) = m.content.as_ref() else {
            continue;
        };
        if content.len() < policy.duplicate_output_min_bytes {
            // Below the savings floor on EITHER side of an identical pair
            // (same hash implies same length): never a dedup candidate,
            // canonical or duplicate.
            continue;
        }
        let hash = content_hash(content.as_bytes());
        let Some(&canonical_idx) = first_seen.get(&hash) else {
            // First occurrence of this hash: a candidate canonical, unless it
            // is not genuinely "still visible in the view" as ITS OWN full
            // content right now — either already reduced (of any kind, from
            // a PRIOR run: its slot shows a stub, not the bytes a new
            // duplicate should be judged identical-and-visible against) or
            // about to vanish into an existing `TurnsCleared` range. In
            // either case leave it unrecorded so the NEXT still-fully-visible
            // occurrence becomes canonical instead (or, if there is none,
            // this hash simply never gets deduped this run — never a
            // correctness issue, only a missed savings opportunity, and
            // exactly what keeps this pass from re-litigating an A8/A7
            // decision a prior run already made).
            if !already_reduced.contains(&i) && !in_existing_clear(i) {
                first_seen.insert(hash, i);
            }
            continue;
        };
        if already_reduced.contains(&i) || protected.contains(&i) || in_existing_clear(i) {
            continue;
        }

        let original_bytes = content.len();
        let id = make_id(ordinal, &hash);
        let tool_name = sanitize_summary_fragment(m.name.as_deref().unwrap_or("tool"));
        let summary = format!(
            "{tool_name} output duplicates msg #{canonical_idx} ({}B) — identical to an \
             earlier tool result, full output in session sidecar",
            format_commas(original_bytes),
        );
        let placeholder = stub::format(stub::Kind::Duplicate, &id, &summary);
        // Structural negative-savings guard: the savings floor
        // (`duplicate_output_min_bytes`) is a heuristic, not a guarantee —
        // the tool name is interpolated into the summary, so a long
        // (untrusted, imported) name can push the stub past the size of the
        // very content it replaces. Never mint a reduction that costs more
        // than it saves. `ordinal` is only consumed on an actual mint, so a
        // skip here is invisible to later ids (deterministic either way).
        if placeholder.len() >= original_bytes {
            continue;
        }
        ordinal += 1;

        let reduction = Reduction {
            id: id.clone(),
            kind: ReductionKind::DuplicateOutput {
                canonical: MessageAddr {
                    index: canonical_idx,
                    role: Role::Tool,
                },
                original_bytes,
            },
            ptr: SidecarPtr {
                addr: MessageAddr {
                    index: i,
                    role: Role::Tool,
                },
                span: None,
                content_hash: hash,
            },
            placeholder,
        };

        view[i].content = Some(reduction.placeholder.clone());
        set_reduction_id(&mut view[i], &reduction.id);

        reduced_this_run.insert(i);
        log.reductions.push(reduction);
    }

    // ---- TR-6: Superseded — keep-latest for same tool + canonicalized args ----
    //
    // Runs immediately after TR-2's dedup pass (immediately above) and BEFORE
    // every other pass (T30/TR-4, A7, the read family, A9, TR-10, A10) — the
    // TR-6.md frozen ordering ("run after TR-2 dedup ... before A7/A10").
    // Placing it ahead of T30/TR-4 too follows the exact same reasoning TR-2
    // itself is placed ahead of T30/TR-4 for: superseding a whole message
    // down to one small stub is strictly cheaper than independently
    // normalizing or truncating content that is about to be evicted anyway,
    // and it must claim its candidates' indices in `reduced_this_run` before
    // any later pass' own candidate scan runs, or that pass would claim them
    // first and the claim would stick forever (prefix stability).
    //
    // Scoped to the exact same address space TR-2 claims from (`Role::Tool`
    // results, excluding `read_indices` — the read-family passes, A8/TR-3,
    // own re-reads exclusively, with strictly richer path-aware redundancy
    // handling than a flat same-key stub could express). Two occurrences
    // sharing a `supersede::canonical_key` are NOT required to be
    // byte-identical (unlike TR-2) — an old FAILING `cargo test` run and a
    // later PASSING run of the identical command are exactly the case this
    // exists for; a byte-identical pair is TR-2's exclusive territory
    // (`recurring_hashes`, below) — this pass never mints a `Superseded` for
    // content whose hash recurs anywhere else, so it never re-litigates
    // TR-2's own decision (see `recurring_hashes`'s doc comment for why this
    // must be an explicit content-hash check, not just "TR-2 runs first").
    if policy.supersede_enabled {
        let supersede_protected: HashSet<usize> = view
            .iter()
            .enumerate()
            .filter(|(_, m)| m.role == Role::Tool)
            .map(|(i, _)| i)
            .rev()
            .take(policy.supersede_protect_last_n)
            .collect();

        let occurrences = supersede::detect(msgs, &read_indices, &policy.supersede_command_fields);
        let mut by_key: HashMap<&str, Vec<usize>> = HashMap::new();
        for c in &occurrences {
            by_key.entry(c.key.as_str()).or_default().push(c.index);
        }

        // Content whose hash recurs ANYWHERE among (non-read) tool results is
        // TR-2's exclusive territory, full stop — never a Superseded
        // candidate, regardless of the two passes' relative protection-zone
        // timing. This is not just "TR-2 runs first within one call": TR-2's
        // OWN candidate rule only ever mints a duplicate once TWO
        // occurrences of the same hash are SIMULTANEOUSLY unprotected in the
        // same `project_messages` call (its `first_seen` canonical stays
        // unrecorded, and unreduced, until then) — a live agent calls
        // `project_messages` incrementally, once per turn, so an occurrence
        // can age out of `protect_last_n_tool_results` SOLO, one turn before
        // any later identical occurrence does too. Without this guard,
        // Superseded's own (intentionally less choosy — it needs no partner,
        // just "not the newest") candidate rule would win that race and
        // permanently evict the EARLIEST copy of a byte-identical run — the
        // exact copy TR-2 means to keep forever as its canonical. Computed
        // fresh here (not reused from TR-2's own local `first_seen`, which
        // only ever covers hashes TR-2 itself has already deemed candidate-
        // eligible at ITS point in time, not "every hash that recurs").
        let mut recurring_hashes: HashSet<String> = HashSet::new();
        let mut hash_occurrences: HashMap<String, Vec<usize>> = HashMap::new();
        for (i, m) in msgs.iter().enumerate() {
            if m.role != Role::Tool || read_indices.contains(&i) {
                continue;
            }
            let Some(content) = m.content.as_ref() else {
                continue;
            };
            let h = content_hash(content.as_bytes());
            let occurrences = hash_occurrences.entry(h.clone()).or_default();
            occurrences.push(i);
            if occurrences.len() > 1 {
                recurring_hashes.insert(h);
            }
        }

        // All but the newest (highest-index) occurrence of each key are
        // candidates; every candidate names the NEWEST occurrence as its
        // successor (SPEC.md TR-6 dev/01: "first three ... naming the 4th",
        // not each other's immediate successor). Gathered into one flat list
        // and sorted by index for deterministic minting order, mirroring
        // every other pass's tie-break rule.
        let mut mint_candidates: Vec<(usize, usize)> = Vec::new(); // (index, successor_index)
        for indices in by_key.values() {
            if indices.len() < 2 {
                continue; // A lone occurrence of a key has nothing to supersede it.
            }
            let mut sorted = indices.clone();
            sorted.sort_unstable();
            let newest = *sorted.last().expect("checked len >= 2 above");
            for &idx in &sorted[..sorted.len() - 1] {
                mint_candidates.push((idx, newest));
            }
        }
        mint_candidates.sort_by_key(|&(idx, _)| idx);

        for (idx, successor_idx) in mint_candidates {
            if already_reduced.contains(&idx)
                || reduced_this_run.contains(&idx)
                || supersede_protected.contains(&idx)
                || in_existing_clear(idx)
            {
                continue;
            }
            let original = view[idx].content.clone().unwrap_or_default();
            let original_bytes = original.len();
            if original_bytes < policy.supersede_min_bytes {
                continue; // Below the savings floor: never a candidate.
            }
            let hash = content_hash(original.as_bytes());
            if recurring_hashes.contains(&hash) {
                // Byte-identical elsewhere: TR-2's territory — with one
                // carve-out (TR-15). The guard exists because an occurrence
                // can age out of protection SOLO, a turn before its identical
                // twin does, and superseding it then would permanently evict
                // the earliest copy TR-2 means to keep visible as its dedup
                // canonical. Once every OTHER occurrence of this hash is
                // already reduced (TR-2 has paired them all, this run or
                // earlier), that race no longer exists: this candidate is the
                // group's dedup ANCHOR, its `DuplicateOutput` children
                // resolve self-addressed from the sidecar (`canonical` is
                // display-only), and a genuinely newer DISTINCT result
                // supersedes the whole group — so the anchor is stubbed too
                // and only the successor stays full. A successor that is
                // itself byte-identical to this candidate (same hash) is no
                // "newer result" at all — the group would supersede itself —
                // so the guard holds unconditionally there.
                let others_all_reduced = hash_occurrences.get(&hash).is_some_and(|occurrences| {
                    occurrences.iter().all(|&j| {
                        j == idx || already_reduced.contains(&j) || reduced_this_run.contains(&j)
                    })
                });
                let successor_is_distinct = msgs
                    .get(successor_idx)
                    .and_then(|m| m.content.as_ref())
                    .is_some_and(|content| content_hash(content.as_bytes()) != hash);
                if !(others_all_reduced && successor_is_distinct) {
                    continue;
                }
            }
            let id = make_id(ordinal, &hash);
            let tool_name = sanitize_summary_fragment(
                occurrences
                    .iter()
                    .find(|c| c.index == idx)
                    .map(|c| c.tool_name.as_str())
                    .unwrap_or("tool"),
            );
            let summary = format!(
                "{tool_name} superseded by newer result at msg #{successor_idx} ({}B) — \
                 expand_reduction(\"{id}\") to restore",
                format_commas(original_bytes),
            );
            let placeholder = stub::format(stub::Kind::Superseded, &id, &summary);
            // Structural negative-savings guard, mirrors TR-2's own: never
            // mint a reduction that costs more than it saves (a long,
            // untrusted tool name interpolated into the summary can in
            // principle push the stub past the content it replaces).
            if placeholder.len() >= original_bytes {
                continue;
            }
            ordinal += 1;

            let reduction = Reduction {
                id: id.clone(),
                kind: ReductionKind::Superseded {
                    by: MessageAddr {
                        index: successor_idx,
                        role: Role::Tool,
                    },
                    original_bytes,
                },
                ptr: SidecarPtr {
                    addr: MessageAddr {
                        index: idx,
                        role: Role::Tool,
                    },
                    span: None,
                    content_hash: hash,
                },
                placeholder,
            };

            view[idx].content = Some(reduction.placeholder.clone());
            set_reduction_id(&mut view[idx], &reduction.id);

            reduced_this_run.insert(idx);
            log.reductions.push(reduction);
        }
    }

    // ---- T30/TR-4: OutputNormalized — terminal-noise normalization ----
    //
    // Runs AFTER TR-2's dedup pass (immediately above — see its comment for
    // the dedup/normalize precedence rule) but BEFORE A7's truncation
    // candidates are even computed: normalizing a noisy bash/exec output
    // BEFORE truncating it means that, when the normalized rendering already
    // fits under `tool_output_trigger_bytes`, it is claimed HERE (collapsed,
    // no ANSI/redraw garbage, already bounded) and needs no truncation at
    // all. When the normalized rendering is STILL over the trigger — real,
    // mostly-distinct content, not redraw noise — this pass does NOT claim
    // the message (see the `normalized_bytes > policy.tool_output_trigger_bytes`
    // check below); it is left raw for A7's candidate scan to pick up and
    // truncate, so the wire payload for every terminal output stays bounded
    // by the trigger either way (the P7 runaway-output safety net).
    // Deliberately does NOT check `protected`: unlike A7's truncation (which discards
    // real content the model might need next turn), collapsing redraws is
    // content-lossless — the rendered text a recent tool result carries is
    // fully preserved, only presentation bytes are removed, so there is no
    // "protect the recent tail" reason to skip it (matching A9's
    // `ImageRedacted`, which likewise never consults `protected`). DOES
    // check `reduced_this_run` (as well as `already_reduced`), so a message
    // TR-2 just claimed above is never also claimed here — each message is
    // claimed by exactly one pass per run.
    if policy.normalize_terminal_output {
        let candidates: Vec<usize> = detect_normalize_candidates(&view)
            .into_iter()
            .filter(|i| {
                !already_reduced.contains(i)
                    && !reduced_this_run.contains(i)
                    && !in_existing_clear(*i)
            })
            .collect();

        for idx in candidates {
            let original = view[idx].content.clone().unwrap_or_default();
            let original_bytes = original.len();
            let normalized = normalize::normalize(&original);
            let normalized_bytes = normalized.len();
            if original_bytes.saturating_sub(normalized_bytes) < policy.terminal_output_min_savings
            {
                continue; // Below the savings floor: leave untouched.
            }
            if normalized_bytes > policy.tool_output_trigger_bytes {
                // P7 safety net (SPEC.md/TR-12): the wire payload for ANY
                // terminal output must stay bounded by A7's trigger. Most
                // ANSI/redraw noise collapses to far less than the trigger,
                // but when the underlying content is genuinely large and
                // mostly distinct (not just redraw noise), the normalized
                // rendering can still exceed it — claiming the message here
                // would let that uncapped view ride the wire unbounded,
                // reintroducing the runaway-output incident A7 exists to
                // prevent. Leave it unclaimed (raw, untouched, `ordinal` not
                // consumed) so it falls through to A7 below, which truncates
                // the RAW bytes to `tool_output_keep_bytes`, bounded and
                // reversible exactly as it was pre-TR-4. Only a normalized
                // rendering that already fits under the trigger is claimed
                // here, as a collapsed-and-already-bounded view.
                continue;
            }

            let hash = content_hash(original.as_bytes());
            let id = make_id(ordinal, &hash);
            ordinal += 1;

            let summary = normalize::summary(original_bytes, normalized_bytes);
            let placeholder = stub::format(stub::Kind::OutputNormalized, &id, &summary);

            let reduction = Reduction {
                id: id.clone(),
                kind: ReductionKind::OutputNormalized {
                    original_bytes,
                    normalized_bytes,
                },
                ptr: SidecarPtr {
                    addr: MessageAddr {
                        index: idx,
                        role: view[idx].role,
                    },
                    span: None,
                    content_hash: hash,
                },
                placeholder,
            };

            let mut new_content = normalized;
            new_content.push_str("\n\n");
            new_content.push_str(&reduction.placeholder);
            view[idx].content = Some(new_content);
            set_reduction_id(&mut view[idx], &reduction.id);

            reduced_this_run.insert(idx);
            log.reductions.push(reduction);
        }
    }

    // ---- A8/TR-3: FileReadElided / FileReadDiffed — re-read handling ----
    //
    // Runs BEFORE A7 (TR-14): a re-read is diff-collapsed or elided here —
    // often shrinking it well under A7's trigger — instead of being blunt
    // prefix-truncated by A7's oversized-output scan, which has no read-type
    // exclusion of its own. TR-2 still runs first (byte-identical outputs are
    // handled before either pass), A10's TurnsCleared block still runs last,
    // and the P7 runaway-output safety net is preserved by an explicit
    // defer-to-A7 guard below: a changed re-read whose stub+diff view would
    // ITSELF still exceed `tool_output_trigger_bytes` is declined here (no
    // reduction minted) so A7's later scan bounds it exactly as before.
    // Every detected
    // read appends a `ReadLogEntry` to `log.read_log` regardless of whether it
    // ends up reduced (deduped by address so re-projection never duplicates
    // it) — this now runs whenever EITHER `elide_stale_reads` or
    // `diff_rereads` is set, since TR-3's read-log lookups must see every
    // read even when A8's own elision is disabled (and vice versa).
    //
    // **A8-vs-TR-3 precedence (SPEC.md TR-3 dev/05).** For a read of a path
    // already seen earlier this session (a re-read, per `log.read_log`):
    // - content hash UNCHANGED from that prior read -> TR-3 has nothing to
    //   show (a zero-hunk diff is never useful) and does not touch this
    //   index at all; A8's ordinary disk-freshness elision runs exactly as
    //   before, unaffected by TR-3 being enabled.
    // - content hash CHANGED from that prior read -> TR-3 takes EXCLUSIVE
    //   claim of this index (A8 never runs on it this call), because A8's
    //   elision message ("unchanged on disk, re-read on demand") would throw
    //   away the very fact that changed — either TR-3 diffs it (below the
    //   size guard) or, if the change is too large to compress usefully, the
    //   full re-read is left untouched in the view (dev/03's guard: showing
    //   the genuine rewrite beats hiding it behind an elision stub the model
    //   would have to spend a turn expanding).
    // A first-ever read of a path (no prior `log.read_log` entry) is never a
    // TR-3 candidate — there is nothing to diff against — and falls straight
    // through to A8, unaffected.
    if policy.elide_stale_reads || policy.diff_rereads {
        for d in detect_reads(&view) {
            let idx = d.index;
            if already_reduced.contains(&idx)
                || reduced_this_run.contains(&idx)
                || protected.contains(&idx)
                || in_existing_clear(idx)
            {
                continue;
            }

            // Always resolved from `msgs` (never `view`): the pristine
            // canonical content at this index, regardless of processing
            // order within this call or any reduction already reapplied
            // onto `view` elsewhere.
            let original = msgs[idx].content.clone().unwrap_or_default();
            let hash = content_hash(original.as_bytes());
            let mtime = policy
                .read_freshness
                .entries
                .get(&idx)
                .and_then(|e| e.mtime);
            let addr = MessageAddr {
                index: idx,
                role: view[idx].role,
            };

            // Most recent prior read of the SAME path, if any — always
            // resolved from `log.read_log`, whose `content_hash`/`addr` were
            // themselves minted from `msgs` (never from a reduced/diffed
            // view), so a diff's base is always a genuine full read, never
            // another diff (no diff-of-diff compounding, SPEC.md TR-3
            // dev/04).
            let prior_read: Option<ReadLogEntry> = log
                .read_log
                .iter()
                .filter(|e| e.path == d.path && e.addr.index < idx)
                .max_by_key(|e| e.addr.index)
                .cloned();

            if !log.read_log.iter().any(|e| e.addr.index == idx) {
                log.read_log.push(ReadLogEntry {
                    path: d.path.clone(),
                    addr,
                    content_hash: hash.clone(),
                    mtime,
                });
            }

            // ---- TR-3: diff-only re-read representation --------------
            //
            // `!d.windowed` guards TR-3.md's frozen v1 scope: "Partial-window
            // reads (offset/limit) are out of scope for v1 — full-file reads
            // only." A partial-window re-read never becomes a diff
            // candidate — its content is a slice, not a whole file, so a
            // unified diff against a prior read (whole or another slice)
            // would present a diff between two arbitrary windows as if it
            // were a genuine file change. Falls through to A8 exactly as a
            // non-windowed read would (A8 itself already fails closed on a
            // windowed read via `probe_read_freshness`'s hash-mismatch
            // fallback), and TR-2 is untouched by this check entirely (it
            // only ever consults same-path-re-read status, not windowing).
            let mut claimed_by_diff = false;
            if policy.diff_rereads && !d.windowed {
                if let Some(prior) = &prior_read {
                    if prior.content_hash != hash {
                        // Changed since the prior read of this path: from
                        // here, A8 must never touch this index (see the
                        // precedence note above) — whether or not the diff
                        // itself ends up below the size guard.
                        claimed_by_diff = true;
                        if let Some(base_text) = msgs
                            .get(prior.addr.index)
                            .and_then(|m| m.content.as_deref())
                        {
                            let diff_text = diffy::create_patch(base_text, &original).to_string();
                            let diff_bytes = diff_text.len();
                            let original_bytes = original.len();
                            let within_guard = (diff_bytes as u128).saturating_mul(100)
                                <= (original_bytes as u128) * policy.diff_max_percent as u128;
                            if within_guard {
                                // `ordinal` is only consumed on an actual
                                // mint (TR-2's guard rule), so computing the
                                // id/stub before the defer-to-A7 decision
                                // below stays deterministic either way.
                                let id = make_id(ordinal, &hash);
                                // explicit is-zero guard is clearer than checked_div here
                                #[allow(clippy::manual_checked_ops)]
                                let percent = if original_bytes == 0 {
                                    0
                                } else {
                                    diff_bytes * 100 / original_bytes
                                };
                                let summary = format!(
                                    "read {} diffed vs prior read at msg #{} — {}B diff, {}B full ({percent}%)",
                                    d.path.display(),
                                    prior.addr.index,
                                    format_commas(diff_bytes),
                                    format_commas(original_bytes),
                                );
                                let placeholder =
                                    stub::format(stub::Kind::FileReadDiffed, &id, &summary);
                                // Defer-to-A7 (TR-14): this pass now runs
                                // before A7, so a stub+diff view that would
                                // ITSELF still exceed A7's trigger must not
                                // claim the index — declining here leaves the
                                // full re-read for A7's own scan to bound
                                // (the P7 safety net), mirroring TR-4's
                                // defer-to-A7 rule.
                                if placeholder
                                    .len()
                                    .saturating_add(1)
                                    .saturating_add(diff_bytes)
                                    > policy.tool_output_trigger_bytes
                                {
                                    continue;
                                }
                                ordinal += 1;
                                let reduction = Reduction {
                                    id: id.clone(),
                                    kind: ReductionKind::FileReadDiffed {
                                        path: d.path.clone(),
                                        base: prior.addr,
                                        base_hash: prior.content_hash.clone(),
                                        new_hash: hash.clone(),
                                        original_bytes,
                                        diff_bytes,
                                    },
                                    ptr: SidecarPtr {
                                        addr,
                                        span: None,
                                        content_hash: hash.clone(),
                                    },
                                    placeholder,
                                };

                                let mut new_content = reduction.placeholder.clone();
                                new_content.push('\n');
                                new_content.push_str(&diff_text);
                                view[idx].content = Some(new_content);
                                set_reduction_id(&mut view[idx], &reduction.id);

                                reduced_this_run.insert(idx);
                                log.reductions.push(reduction);
                            }
                            // else: large-change guard tripped (dev/03) — no
                            // reduction of any kind; the full re-read stays.
                        }
                        // else: base unresolvable (should not happen against
                        // a stable `msgs`) — fail safe, leave the full
                        // re-read untouched rather than guess.
                    }
                }
            }
            if claimed_by_diff {
                continue; // Never let A8 elide a read TR-3 has claimed.
            }

            // ---- A8: stale-file-read elision --------------------------
            if !policy.elide_stale_reads {
                continue;
            }
            let fresh = policy
                .read_freshness
                .entries
                .get(&idx)
                .is_some_and(|e| e.fresh);
            if !fresh {
                continue; // Changed or unreadable: the transcript copy stays the record.
            }

            let id = make_id(ordinal, &hash);
            ordinal += 1;
            let summary = format!(
                "read {} elided — file unchanged on disk, re-read on demand",
                d.path.display()
            );
            let placeholder = stub::format(stub::Kind::FileRead, &id, &summary);
            let reduction = Reduction {
                id: id.clone(),
                kind: ReductionKind::FileReadElided {
                    path: d.path.clone(),
                    read_log: ReadLogEntry {
                        path: d.path.clone(),
                        addr,
                        content_hash: hash.clone(),
                        mtime,
                    },
                },
                ptr: SidecarPtr {
                    addr,
                    span: None,
                    content_hash: hash,
                },
                placeholder,
            };

            view[idx].content = Some(reduction.placeholder.clone());
            set_reduction_id(&mut view[idx], &reduction.id);

            reduced_this_run.insert(idx);
            log.reductions.push(reduction);
        }
    }

    // Candidates: oversized tool results (never the system prompt, which is
    // role `System` and so never matches `role == Role::Tool` anyway), not
    // already reduced, not in the protected tail, not already inside an
    // established `TurnsCleared` range (about to vanish into its one
    // placeholder regardless), and not just claimed by TR-2's dedup pass,
    // T30/TR-4's normalize pass, or the A8/TR-3 read-family pass above
    // (TR-14: a re-read the read family diffed or elided is its claim; one it
    // DECLINED — changed too much, or diff still oversized — falls through to
    // this scan and is bounded here exactly as any oversized output).
    let mut candidates: Vec<usize> = view
        .iter()
        .enumerate()
        .filter(|(i, m)| {
            m.role == Role::Tool
                && !already_reduced.contains(i)
                && !reduced_this_run.contains(i)
                && !protected.contains(i)
                && !in_existing_clear(*i)
                && m.content.as_ref().map(|c| c.len()).unwrap_or(0)
                    > policy.tool_output_trigger_bytes
        })
        .map(|(i, _)| i)
        .collect();

    // Largest-first (#6 "largest wins"); ties broken by ascending index for
    // determinism.
    let byte_len = |i: usize| view[i].content.as_ref().map(|c| c.len()).unwrap_or(0);
    candidates.sort_by(|&a, &b| byte_len(b).cmp(&byte_len(a)).then(a.cmp(&b)));

    for idx in candidates {
        let original = view[idx].content.clone().unwrap_or_default();
        let original_bytes = original.len();
        let hash = content_hash(original.as_bytes());
        let id = make_id(ordinal, &hash);
        ordinal += 1;

        let kept_bytes = char_boundary_floor(&original, policy.tool_output_keep_bytes);
        let tool_name = sanitize_summary_fragment(view[idx].name.as_deref().unwrap_or("tool"));
        let summary = format!(
            "{tool_name} output truncated {}B, kept {}B — full output in session sidecar",
            format_commas(original_bytes),
            format_commas(kept_bytes),
        );
        let placeholder = stub::format(stub::Kind::ToolOutput, &id, &summary);

        let reduction = Reduction {
            id: id.clone(),
            kind: ReductionKind::ToolOutputTruncated {
                original_bytes,
                kept_bytes,
            },
            ptr: SidecarPtr {
                addr: MessageAddr {
                    index: idx,
                    role: view[idx].role,
                },
                span: Some((kept_bytes, original_bytes)),
                content_hash: hash,
            },
            placeholder,
        };

        let mut new_content = original[..kept_bytes].to_string();
        new_content.push_str("\n\n");
        new_content.push_str(&reduction.placeholder);
        view[idx].content = Some(new_content);
        set_reduction_id(&mut view[idx], &reduction.id);

        reduced_this_run.insert(idx);
        log.reductions.push(reduction);
    }

    // ---- A9: ImageRedacted — data: URL image stripped to a stub + pointer ----
    //
    // Runs after A7/A8 (both cardinality-preserving, like this) and before
    // A10's TurnsCleared block (which must stay last — see above). Operates
    // on `content_parts`, an axis A7/A8 never touch, so there is no
    // cross-kind conflict to guard against the way A7 guards A8.
    if policy.redact_images {
        for img in detect_images(&view) {
            if img.url_len < policy.image_redact_min_bytes || in_existing_clear(img.msg_index) {
                continue;
            }
            let idx = img.msg_index;
            let part_index = img.part_index;

            let original_part = view[idx].content_parts.as_ref().unwrap()[part_index].clone();
            let serialized = serde_json::to_vec(&original_part)
                .expect("a content part is always representable as JSON");
            let hash = content_hash(&serialized);
            let id = make_id(ordinal, &hash);
            ordinal += 1;

            let size_kb = (img.url_len + 512) / 1024;
            let summary = format!("image redacted ({}, {size_kb}KB)", img.mime);
            let placeholder = stub::format(stub::Kind::Image, &id, &summary);

            let reduction = Reduction {
                id: id.clone(),
                kind: ReductionKind::ImageRedacted { part_index },
                ptr: SidecarPtr {
                    addr: MessageAddr {
                        index: idx,
                        role: view[idx].role,
                    },
                    span: None,
                    content_hash: hash,
                },
                placeholder,
            };

            let parts = view[idx].content_parts.as_mut().unwrap();
            parts[part_index] = serde_json::json!({"type": "text", "text": reduction.placeholder});
            set_reduction_id(&mut view[idx], &reduction.id);

            reduced_this_run.insert(idx);
            log.reductions.push(reduction);
        }
    }

    // ---- TR-10: ToolInputElided — the assistant-side twin of A7/A8 ----
    //
    // Runs after A7/A8/A9 (an independent axis: assistant `tool_calls`
    // arguments, never a `Role::Tool` result or a `content_parts` image, so
    // there is no cross-kind conflict to guard the way A7 guards A8) and
    // before A10's TurnsCleared block (which must stay last — see above).
    if policy.elide_tool_inputs {
        // Tracked per CALL id, not per message index (unlike `already_reduced`
        // above): a single assistant message can carry more than one
        // tool_calls entry, each independently elidable.
        let already_call_ids: HashSet<&str> = prior
            .reductions
            .iter()
            .filter_map(|r| match &r.kind {
                ReductionKind::ToolInputElided { call_id, .. } => Some(call_id.as_str()),
                _ => None,
            })
            .collect();

        let mut candidates: Vec<DetectedToolInput> =
            detect_tool_inputs(&view, &policy.tool_input_elidable_fields)
                .into_iter()
                .filter(|d| {
                    !already_call_ids.contains(d.call_id.as_str())
                        && !in_existing_clear(d.msg_index)
                        && d.value.len() > policy.tool_input_trigger_bytes
                })
                .collect();

        // Largest-first (#6 "largest wins"), ties broken by call_id for
        // determinism.
        candidates.sort_by(|a, b| {
            b.value
                .len()
                .cmp(&a.value.len())
                .then(a.call_id.cmp(&b.call_id))
        });

        for d in candidates {
            let hash = content_hash(d.value.as_bytes());
            let id = make_id(ordinal, &hash);
            ordinal += 1;
            let original_bytes = d.value.len();
            let path_clause = d
                .path
                .as_ref()
                .map(|p| format!(", on disk at {}", p.display()))
                .unwrap_or_default();
            let hash_prefix: String = hash.chars().take(8).collect();
            let summary = format!(
                "{} input elided: `{}` field, {}B{path_clause}, blake3={hash_prefix}... — full \
                 args in session sidecar",
                d.tool_name,
                d.field,
                format_commas(original_bytes),
            );
            let placeholder = stub::format(stub::Kind::ToolInput, &id, &summary);

            let Some(original_args) = view
                .get(d.msg_index)
                .and_then(|m| m.tool_calls.as_ref())
                .and_then(|calls| calls.iter().find(|c| c.id == d.call_id))
                .map(|call| call.function.arguments.clone())
            else {
                continue; // Addressed call no longer present; skip.
            };
            let Some(spliced) =
                replace_top_level_string_field(&original_args, &d.field, &placeholder)
            else {
                continue; // Field vanished/reshaped since detection; skip rather than corrupt.
            };

            let role = view[d.msg_index].role;
            let calls = view[d.msg_index]
                .tool_calls
                .as_mut()
                .expect("checked above: this message has tool_calls");
            let call = calls
                .iter_mut()
                .find(|c| c.id == d.call_id)
                .expect("checked above: this call_id is present");
            call.function.arguments = spliced;
            set_reduction_id(&mut view[d.msg_index], &id);

            let reduction = Reduction {
                id: id.clone(),
                kind: ReductionKind::ToolInputElided {
                    original_bytes,
                    path: d.path.clone(),
                    content_hash: hash.clone(),
                    call_id: d.call_id.clone(),
                    field: d.field.clone(),
                },
                ptr: SidecarPtr {
                    addr: MessageAddr {
                        index: d.msg_index,
                        role,
                    },
                    span: None,
                    content_hash: hash,
                },
                placeholder,
            };
            log.reductions.push(reduction);
        }
    }

    // ---- TR-6: errored-input pruning — the FAILURE-side complement of TR-10 ----
    //
    // Same address space as TR-10 above (assistant `tool_calls` arguments,
    // keyed by `call_id`), but structurally disjoint from it: `detect_tool_inputs`
    // only ever matches a call whose paired result is `KnownSuccess`
    // (TR-10); `detect_errored_tool_inputs` only ever matches `KnownError`
    // (this pass) — an `Unknown` result matches neither, and a call id can never
    // satisfy both, so `already_call_ids` (recomputed here rather than shared
    // with TR-10's own local binding above, since either gate may be off
    // independently of the other) is sufficient with no extra bookkeeping to
    // keep the two disjoint. Also gated by an aging clock TR-10 has no
    // equivalent of ([`assistant_turns_since`] vs. `policy.errored_input_prune_after_turns`)
    // — a freshly-failed call's input stays visible for a while (in case the
    // model wants to see exactly what it just tried) and only becomes a
    // candidate once it's aged past that. The paired error-result message
    // itself (the actual error text) is never touched here — only the
    // assistant-side input argument — so the error stays visible exactly as
    // TR-6.md requires.
    if policy.prune_errored_inputs {
        let already_call_ids: HashSet<&str> = prior
            .reductions
            .iter()
            .filter_map(|r| match &r.kind {
                ReductionKind::ToolInputElided { call_id, .. } => Some(call_id.as_str()),
                _ => None,
            })
            .collect();

        let mut candidates: Vec<DetectedToolInput> =
            detect_errored_tool_inputs(&view, &policy.tool_input_elidable_fields)
                .into_iter()
                .filter(|d| {
                    !already_call_ids.contains(d.call_id.as_str())
                        && !in_existing_clear(d.msg_index)
                        && d.value.len() > policy.tool_input_trigger_bytes
                        && assistant_turns_since(&view, d.msg_index)
                            >= policy.errored_input_prune_after_turns
                })
                .collect();

        // Largest-first (#6 "largest wins"), ties broken by call_id for
        // determinism — identical convention to TR-10's own candidate sort.
        candidates.sort_by(|a, b| {
            b.value
                .len()
                .cmp(&a.value.len())
                .then(a.call_id.cmp(&b.call_id))
        });

        for d in candidates {
            let hash = content_hash(d.value.as_bytes());
            let id = make_id(ordinal, &hash);
            ordinal += 1;
            let original_bytes = d.value.len();
            let turns = assistant_turns_since(&view, d.msg_index);
            let path_clause = d
                .path
                .as_ref()
                .map(|p| format!(", on disk at {}", p.display()))
                .unwrap_or_default();
            let hash_prefix: String = hash.chars().take(8).collect();
            let summary = format!(
                "{} input elided (errored call, {turns} turns old): `{}` field, {}B{path_clause}, \
                 blake3={hash_prefix}... — full args in session sidecar",
                d.tool_name,
                d.field,
                format_commas(original_bytes),
            );
            let placeholder = stub::format(stub::Kind::ToolInput, &id, &summary);

            let Some(original_args) = view
                .get(d.msg_index)
                .and_then(|m| m.tool_calls.as_ref())
                .and_then(|calls| calls.iter().find(|c| c.id == d.call_id))
                .map(|call| call.function.arguments.clone())
            else {
                continue; // Addressed call no longer present; skip.
            };
            let Some(spliced) =
                replace_top_level_string_field(&original_args, &d.field, &placeholder)
            else {
                continue; // Field vanished/reshaped since detection; skip rather than corrupt.
            };

            let role = view[d.msg_index].role;
            let calls = view[d.msg_index]
                .tool_calls
                .as_mut()
                .expect("checked above: this message has tool_calls");
            let call = calls
                .iter_mut()
                .find(|c| c.id == d.call_id)
                .expect("checked above: this call_id is present");
            call.function.arguments = spliced;
            set_reduction_id(&mut view[d.msg_index], &id);

            let reduction = Reduction {
                id: id.clone(),
                kind: ReductionKind::ToolInputElided {
                    original_bytes,
                    path: d.path.clone(),
                    content_hash: hash.clone(),
                    call_id: d.call_id.clone(),
                    field: d.field.clone(),
                },
                ptr: SidecarPtr {
                    addr: MessageAddr {
                        index: d.msg_index,
                        role,
                    },
                    span: None,
                    content_hash: hash,
                },
                placeholder,
            };
            log.reductions.push(reduction);
        }
    }

    // ---- A10: TurnsCleared — old-turn clearing, re-founding `maybe_compact` ----
    //
    // `view` is still fully index-parallel to `msgs` at this point (every
    // step above only mutated content in place); this is the one step that
    // changes cardinality, so it must run last.
    if !existing_clears.is_empty() {
        // Already established in a prior projection: reapply EVERY existing
        // record verbatim, never recompute or widen any range (prefix
        // stability, A5) — descending by `first` so an earlier splice's
        // cardinality shrink never invalidates a later, still-`msgs`-indexed
        // splice (mirrors TR-9 `handoff`'s own last-gap-first splice order).
        let mut to_reapply: Vec<Reduction> = log
            .reductions
            .iter()
            .filter(|r| matches!(r.kind, ReductionKind::TurnsCleared { .. }))
            .cloned()
            .collect();
        to_reapply.sort_by_key(|r| match r.kind {
            ReductionKind::TurnsCleared { first, .. } => std::cmp::Reverse(first),
            _ => unreachable!("filtered to TurnsCleared above"),
        });
        for r in &to_reapply {
            reapply_reduction(&mut view, r, msgs);
        }
    } else if let Some((first, last)) = compute_clear_range(&view, policy) {
        // `view`'s roles are identical to `msgs`'s at every index up to this
        // point (every pass above only mutates content/tool_calls in place),
        // so `compute_clear_range` — a pure function of role/length alone —
        // derives the identical range whether called against `view` (here)
        // or `msgs` directly ([`prepare_cleared_turns_summary`], called
        // BEFORE this projection even runs). This is what lets the two
        // agree by construction; see `compute_clear_range`'s own doc.
        //
        // `msgs` is `history[1..]` (via `project_messages`, called from
        // `Agent::build_request_messages`). The hash below covers the true
        // sidecar bytes — never an already-truncated copy — SOLELY because
        // `Agent::run_loop`'s D6/A7 supersession gate (TR-12) keeps
        // `cap_tool_output` off whenever a recorder + this policy are both
        // active (the only combination `project_messages`/mint ever runs
        // under with a durable sidecar behind it): `history` then holds full
        // bytes by construction, so this slice already equals
        // `sidecar.messages`. Without that gate a legacy `cap_tool_output`
        // could shrink `msgs` first, and this comment's claim would be false
        // — exactly the land-blocker TR-12 fixed (a hash minted from capped
        // bytes can never recompute the same way from the reloaded,
        // full-bytes sidecar).
        let range = &msgs[first..=last];
        let (hash, range_bytes) =
            hash_turns_range(range).expect("ChatMessage always serializes to JSON");
        let (user, assistant, tool) = count_roles(range);
        let id = make_id(ordinal, &hash);
        let deterministic_summary = format!(
            "turns {first}..{} cleared ({} messages: {user} user, {assistant} \
             assistant, {tool} tool; {}B) — full turns in session sidecar",
            last + 1,
            format_commas(range.len()),
            format_commas(range_bytes),
        );

        // TR-7 (T20): off by default (`summarize_cleared_turns: false`,
        // SPEC.md dev/01) — the placeholder below is then byte-identical to
        // pre-TR-7 behavior, full stop. When on AND a prepared summary
        // exists for EXACTLY this `(first, last)` range (computed by
        // `prepare_cleared_turns_summary`, called by the driving caller
        // BEFORE this projection — the one place TR-7's side-call happens,
        // mirroring A8's `probe_read_freshness` split so this function
        // itself stays pure/I-O-free), the placeholder instead carries the
        // LLM summary text plus an honesty banner naming this span's
        // reduction id and turn count, with `expand_reduction("<id>")`
        // spelled out as the verbatim escape hatch. Any mismatch (off,
        // absent, or a stale/wrong-range entry) falls back to the
        // deterministic stub — dev/03's failure-fallback guarantee applies
        // equally to "never prepared" and "errored while preparing".
        let prepared = policy
            .summarize_cleared_turns
            .then_some(policy.cleared_turns_summary.as_ref())
            .flatten()
            .filter(|p| p.first == first && p.last == last);

        let (summary_text, summary_audit) = match prepared {
            Some(p) => {
                let banner = format!(
                    "sc-summary of {id}, original {} messages in sidecar — \
                     expand_reduction(\"{id}\") for verbatim",
                    format_commas(range.len()),
                );
                let text = format!("{} ({banner})", p.text);
                let audit = SpanSummary {
                    model_id: p.model_id.clone(),
                    prompt_version: summarize::PROMPT_VERSION.to_string(),
                    summary_hash: content_hash(p.text.as_bytes()),
                };
                (text, Some(audit))
            }
            None => (deterministic_summary, None),
        };
        let placeholder = stub::format(stub::Kind::TurnsCleared, &id, &summary_text);

        // Any reduction (of any kind) whose address falls inside the
        // range about to be collapsed is now subsumed by this single
        // placeholder — drop it from the log; `invert` restores the
        // WHOLE range straight from the sidecar (`resolve_turns_range`),
        // so nothing recorded there is lost.
        log.reductions
            .retain(|r| !(r.ptr.addr.index >= first && r.ptr.addr.index <= last));

        let reduction = Reduction {
            id: id.clone(),
            kind: ReductionKind::TurnsCleared {
                first,
                last,
                summary: summary_audit,
            },
            ptr: SidecarPtr {
                addr: MessageAddr {
                    index: first,
                    role: range[0].role,
                },
                span: None,
                content_hash: hash,
            },
            placeholder,
        };

        let mut stub_msg = ChatMessage::system(reduction.placeholder.clone());
        set_reduction_id(&mut stub_msg, &reduction.id);
        view.splice(first..=last, std::iter::once(stub_msg));

        log.reductions.push(reduction);
    }

    (view, log)
}

fn next_reduction_ordinal<'a>(ids: impl Iterator<Item = &'a str>) -> usize {
    ids.filter_map(|id| {
        let digits = id.strip_prefix('r')?.get(..4)?;
        digits.parse::<usize>().ok()
    })
    .max()
    .map_or(0, |max| max.saturating_add(1))
}

#[cfg(test)]
mod ordinal_tests {
    use super::next_reduction_ordinal;

    #[test]
    fn sparse_legacy_ids_advance_past_the_greatest_live_ordinal() {
        let ids = ["r0001-dead", "r0007-beef"];
        assert_eq!(next_reduction_ordinal(ids.into_iter()), 8);
    }

    #[test]
    fn malformed_ids_cannot_force_reuse_of_a_valid_live_ordinal() {
        let ids = ["legacy", "r0003-cafe", "rxxxx-nope"];
        assert_eq!(next_reduction_ordinal(ids.into_iter()), 4);
    }
}

/// PARITY-18 — how many times [`reduce_to_fit`] will re-project with a
/// tighter [`ReductionPolicy`] before giving up and returning its best
/// (tightest-attempted) effort. Each level roughly halves the byte-based
/// knobs (see [`tighten`]), so 5 levels covers a ~32x tightening range —
/// past that, more aggression stops buying meaningfully more headroom and
/// the caller's preflight guard (`crates/cli`'s `resume_cmd`) should fail
/// fast instead of looping forever.
const MAX_AGGRESSIVE_LEVELS: u32 = 5;

/// PARITY-18 — a strictly tighter variant of `base` for [`reduce_to_fit`]'s
/// escalation ladder. Only scales knobs that are pure functions of in-view
/// content (byte thresholds, protected-recency windows) — exactly the set
/// [`ReductionPolicy::default`] already turns on unconditionally (D14) —
/// never touches `elide_stale_reads`/`read_freshness` or
/// `summarize_cleared_turns`/`cleared_turns_summary`, both of which need
/// data only a live disk probe or an LLM side-call can produce and so stay
/// exactly as the caller configured them at every level. `level` is
/// 1-indexed (`level=0` would be `base` itself, never called that way here).
fn tighten(base: &ReductionPolicy, level: u32) -> ReductionPolicy {
    let shift = level.min(5);
    let shrink = |n: usize, floor: usize| -> usize { (n >> shift).max(floor) };
    ReductionPolicy {
        tool_output_keep_bytes: shrink(base.tool_output_keep_bytes, 128),
        tool_output_trigger_bytes: shrink(base.tool_output_trigger_bytes, 256),
        protect_last_n_tool_results: base
            .protect_last_n_tool_results
            .saturating_sub(level as usize),
        image_redact_min_bytes: shrink(base.image_redact_min_bytes, 256),
        tool_input_trigger_bytes: shrink(base.tool_input_trigger_bytes, 256),
        duplicate_output_min_bytes: shrink(base.duplicate_output_min_bytes, 16),
        supersede_min_bytes: shrink(base.supersede_min_bytes, 16),
        supersede_protect_last_n: base.supersede_protect_last_n.saturating_sub(level as usize),
        errored_input_prune_after_turns: base
            .errored_input_prune_after_turns
            .saturating_sub(level as usize),
        ..base.clone()
    }
}

/// PARITY-18 v3 — the "aggressive reduction" half of the fix (SPEC.md
/// scaling-context-guard): apply [`project_messages`] with `base_policy`
/// first, then, if `fits` says the projected view still doesn't pass,
/// retry with progressively tighter policies (see `tighten`) up to
/// `MAX_AGGRESSIVE_LEVELS` times, then escalate to A10 turn-clearing,
/// keeping whichever attempt is smallest. Never fails and never loops
/// unboundedly — it always returns SOME projection (the caller's own
/// preflight guard is responsible for deciding whether even the tightest
/// attempt still exceeds the target model's context limit and refusing to
/// send in that case, PARITY-18 dev/01).
///
/// `fits` is a closure over the REDUCED VIEW ALONE (this function has no
/// knowledge of the system prompt, tool schemas, or a trailing user
/// prompt — those live with the caller). It exists to close a v2 defect a
/// skeptical review caught (the flagship rescue-path regression): v2 had
/// two INDEPENDENTLY-derived boundaries — this function stopped escalating
/// once `estimate_view_tokens(view) <= target_tokens` where
/// `target_tokens = context_limit - CONTEXT_RESPONSE_RESERVE_TOKENS`, while
/// `tokens::context_guard` only ACCEPTS a request when
/// `with_guard_margin(view + tools) + CONTEXT_RESPONSE_RESERVE_TOKENS <=
/// context_limit` — a materially tighter bound (no margin, no tools, no
/// system-prompt overhead counted by the reducer's stop condition at all).
/// Any session whose reduced view landed in the band between those two
/// boundaries was declared "fits" here and then refused by the guard, with
/// D6 turn-clearing never triggered because this function had already
/// stopped. Passing `tokens::context_guard` itself (wrapped with the
/// caller's real system prompt/tools/prompt) as `fits` makes that
/// impossible BY CONSTRUCTION: the reducer's stopping condition and the
/// guard's acceptance are now the same boundary, so a session this function
/// says it reduced-to-fit is a session the guard will actually send.
/// (`resume_cmd` is the sole real caller; see its `fits` closure there.)
///
/// Returns `(view, log, policy_actually_applied)` so the caller can seed
/// a runtime agent's reduction policy/log setters with the exact
/// policy that produced the returned view (subsequent turns keep re-applying
/// it, `resume_cmd`'s existing re-reduce path).
pub fn reduce_to_fit<F>(
    full_msgs: &[ChatMessage],
    base_policy: &ReductionPolicy,
    prior_log: &ReductionLog,
    fits: F,
) -> (Vec<ChatMessage>, ReductionLog, ReductionPolicy)
where
    F: Fn(&[ChatMessage]) -> bool,
{
    let (mut best_view, mut best_log) = project_messages(full_msgs, base_policy, prior_log);
    let mut best_policy = base_policy.clone();
    let mut best_tokens = estimate_view_tokens(&best_view);

    let mut level = 1;
    while !fits(&best_view) && level <= MAX_AGGRESSIVE_LEVELS {
        let candidate_policy = tighten(base_policy, level);
        let (view, log) = project_messages(full_msgs, &candidate_policy, prior_log);
        let view_tokens = estimate_view_tokens(&view);
        if view_tokens < best_tokens {
            best_view = view;
            best_log = log;
            best_policy = candidate_policy;
            best_tokens = view_tokens;
        }
        level += 1;
    }

    // PARITY-18 D6 — [`tighten`] only scales byte-based knobs (tool output,
    // images, tool input), so a TEXT-heavy session (mostly plain user/
    // assistant turns, little/no tool output to shrink) can exhaust every
    // level above and still not `fits`. Escalate to A10 turn-clearing
    // (`clear_turns_older_than`) as a last resort: stacked on top of
    // whatever byte-knob tightening already achieved (`best_policy`),
    // progressively HALVING the surviving message-count window until the
    // projection fits OR the window bottoms out at
    // [`MIN_CLEAR_TURNS_WINDOW`]. Reversible, not lossy —
    // `ReductionKind::TurnsCleared` stubs are hash-verified and rehydrate
    // byte-exact from the sidecar via `expand_reduction`/`invert`
    // (dev/03 fidelity is untouched: nothing here bypasses that path).
    // Gated on `base_policy.clear_turns_older_than.is_none()` so this never
    // overrides an EXPLICIT threshold a caller already set (e.g.
    // `Agent::maybe_compact`'s own live-session clearing, which never calls
    // `reduce_to_fit` at all, but the guard costs nothing to keep honest).
    //
    // PARITY-18 v3 — this loop previously ran at most `MAX_AGGRESSIVE_LEVELS`
    // (5) halvings starting from `full_msgs.len()`, so any session over
    // ~128 messages bottomed out well above the floor (a 3,000-message
    // session stopped at ~93, never reaching 4) — "maximal reduction"
    // wasn't actually maximal, which weakened the honesty of a genuine
    // refusal (it would refuse having never tried the smallest window).
    // The loop now keeps halving unconditionally until `threshold` reaches
    // [`MIN_CLEAR_TURNS_WINDOW`] regardless of the starting length — still
    // `O(log2(full_msgs.len()))` iterations, so it stays bounded — or until
    // `fits` succeeds, whichever comes first.
    if !fits(&best_view) && base_policy.clear_turns_older_than.is_none() {
        let mut threshold = full_msgs.len();
        loop {
            if threshold <= MIN_CLEAR_TURNS_WINDOW {
                break;
            }
            threshold = (threshold / 2).max(MIN_CLEAR_TURNS_WINDOW);
            let mut candidate_policy = best_policy.clone();
            candidate_policy.clear_turns_older_than = Some(threshold);
            let (view, log) = project_messages(full_msgs, &candidate_policy, prior_log);
            let view_tokens = estimate_view_tokens(&view);
            if view_tokens < best_tokens {
                best_view = view;
                best_log = log;
                best_policy = candidate_policy;
                best_tokens = view_tokens;
            }
            if fits(&best_view) {
                break;
            }
        }
    }

    (best_view, best_log, best_policy)
}

/// PARITY-18 D6 — the smallest surviving message-count window
/// [`reduce_to_fit`]'s A10 escalation will ever request via
/// `clear_turns_older_than`. `compute_clear_range` itself already floors
/// `keep_recent` at `2`; `4` here leaves a little more headroom (at least
/// one full user/assistant exchange) before giving up rather than shrinking
/// the window to the bare minimum every level.
const MIN_CLEAR_TURNS_WINDOW: usize = 4;

// ---------------------------------------------------------------------------
// A6 — invert() / invert_one(): reduced view + log + sidecar -> full view
// ---------------------------------------------------------------------------

/// Resolve and hash-verify the full original content standing behind `ptr`
/// (kind-agnostic: [`ReductionKind::ToolOutputTruncated`],
/// [`ReductionKind::FileReadElided`], [`ReductionKind::OutputNormalized`],
/// [`ReductionKind::FileReadDiffed`], and [`ReductionKind::DuplicateOutput`]
/// all restore by replacing the reduced message's whole content with this —
/// for `OutputNormalized` this is the RAW captured bytes, byte-exact, never
/// the normalized/rendered text: the sidecar only ever stores the raw
/// capture, so this same kind-agnostic resolve path already restores it
/// correctly with no `OutputNormalized`-specific branch needed here).
///
/// Takes the resolved-against message slice directly (rather than a
/// [`Session`]) so [`rehydrate`]'s model-invocable path can resolve against a
/// live `Agent`'s own `history[1..]` — which the sidecar invariant
/// (`Agent::resume_recorded`'s doc comment) guarantees is byte-identical to
/// `Session::from_native_str(sidecar).messages` at every instant **once a
/// recorder and a [`ReductionPolicy`] are both installed** (TR-12's D6/A7
/// supersession gate in `Agent::run_loop`, the only combination under which
/// any reduction here is ever minted with a durable sidecar behind it) —
/// without needing to round-trip through disk in that case.
/// `invert`/`invert_one`/`verify_log` (the offline, `Session`-backed callers)
/// simply pass `sidecar.messages`, so they need no such gate to already hold:
/// a hash minted under the gate recomputes identically from either slice by
/// construction; a hash minted from a pre-gate legacy sidecar instead fails
/// safe here (content_hash mismatch) rather than resolving to the wrong
/// bytes.
fn resolve_original_content(ptr: &SidecarPtr, messages: &[ChatMessage]) -> Result<String> {
    let msg = messages.get(ptr.addr.index).ok_or_else(|| {
        Error::new(format!(
            "invert: sidecar has no message at index {} (reduction pointer unresolvable)",
            ptr.addr.index
        ))
    })?;
    if msg.role != ptr.addr.role {
        return Err(Error::new(format!(
            "invert: role mismatch at sidecar index {}: pointer expects {:?}, sidecar has {:?}",
            ptr.addr.index, ptr.addr.role, msg.role
        )));
    }
    let content = msg.content.clone().unwrap_or_default();
    ptr.verify(content.as_bytes())?;
    Ok(content)
}

/// Resolve and hash-verify the original image content part standing behind an
/// [`ReductionKind::ImageRedacted`] pointer. See [`resolve_original_content`]
/// for why this takes a message slice rather than a [`Session`].
fn resolve_image_part(
    ptr: &SidecarPtr,
    part_index: usize,
    messages: &[ChatMessage],
) -> Result<serde_json::Value> {
    let msg = messages.get(ptr.addr.index).ok_or_else(|| {
        Error::new(format!(
            "invert: sidecar has no message at index {} (reduction pointer unresolvable)",
            ptr.addr.index
        ))
    })?;
    if msg.role != ptr.addr.role {
        return Err(Error::new(format!(
            "invert: role mismatch at sidecar index {}: pointer expects {:?}, sidecar has {:?}",
            ptr.addr.index, ptr.addr.role, msg.role
        )));
    }
    let part = msg
        .content_parts
        .as_ref()
        .and_then(|parts| parts.get(part_index))
        .ok_or_else(|| {
            Error::new(format!(
                "invert: sidecar message at index {} has no content part {part_index}",
                ptr.addr.index
            ))
        })?;
    let serialized = serde_json::to_vec(part)
        .map_err(|e| Error::new(format!("invert: failed to serialize image part: {e}")))?;
    ptr.verify(&serialized)?;
    Ok(part.clone())
}

/// Resolve and hash-verify the original message range `first..=last` standing
/// behind a [`ReductionKind::TurnsCleared`] pointer. The hash covers each
/// message's wire-serialized bytes (role/content/tool_calls/tool_call_id/name
/// — the same shape [`ChatMessage`]'s custom `Serialize` puts on the wire),
/// concatenated in order. See [`resolve_original_content`] for why this takes
/// a message slice rather than a [`Session`].
fn resolve_turns_range(
    ptr: &SidecarPtr,
    first: usize,
    last: usize,
    messages: &[ChatMessage],
) -> Result<Vec<ChatMessage>> {
    let mut msgs = Vec::with_capacity(last.saturating_sub(first) + 1);
    for i in first..=last {
        let msg = messages.get(i).ok_or_else(|| {
            Error::new(format!(
                "invert: sidecar has no message at index {i} (turns-cleared range unresolvable)"
            ))
        })?;
        msgs.push(msg.clone());
    }
    if let Some(first_msg) = msgs.first() {
        if first_msg.role != ptr.addr.role {
            return Err(Error::new(format!(
                "invert: role mismatch at sidecar index {first}: pointer expects {:?}, sidecar has {:?}",
                ptr.addr.role, first_msg.role
            )));
        }
    }
    // Shared with the creating side (`hash_turns_range`, A10) so the two
    // formulas can never drift apart: blake3 over each message's wire-
    // serialized bytes, concatenated in order. (The byte length only matters
    // to the creating side's stub summary — ignored here.)
    let (hash, _bytes) = hash_turns_range(&msgs)?;
    ptr.verify_hash(&hash)?;
    Ok(msgs)
}

/// Rehydrate one reduction `r` in place within `out`: locate the placeholder
/// message by its `sc.reduction` id, then restore it from `sidecar`
/// (hash-verified). `TurnsCleared` splices the original message range back in
/// place of the single placeholder message.
fn invert_one_reduction(
    out: &mut Vec<ChatMessage>,
    r: &Reduction,
    sidecar_messages: &[ChatMessage],
) -> Result<()> {
    let pos = out
        .iter()
        .position(|m| reduction_id(m) == Some(r.id.as_str()))
        .ok_or_else(|| {
            Error::new(format!(
                "invert: no message in the reduced view carries reduction id {}",
                r.id
            ))
        })?;

    match &r.kind {
        ReductionKind::ToolOutputTruncated { .. }
        | ReductionKind::FileReadElided { .. }
        | ReductionKind::OutputNormalized { .. }
        | ReductionKind::FileReadDiffed { .. }
        | ReductionKind::DuplicateOutput { .. }
        | ReductionKind::Superseded { .. } => {
            let original = resolve_original_content(&r.ptr, sidecar_messages)?;
            out[pos].content = Some(original);
            out[pos].metadata.remove(REDUCTION_METADATA_KEY);
        }
        ReductionKind::ImageRedacted { part_index } => {
            let part = resolve_image_part(&r.ptr, *part_index, sidecar_messages)?;
            let parts = out[pos].content_parts.get_or_insert_with(Vec::new);
            if *part_index < parts.len() {
                parts[*part_index] = part;
            } else {
                parts.push(part);
            }
            out[pos].metadata.remove(REDUCTION_METADATA_KEY);
        }
        ReductionKind::TurnsCleared { first, last, .. } => {
            let msgs = resolve_turns_range(&r.ptr, *first, *last, sidecar_messages)?;
            out.splice(pos..=pos, msgs);
        }
        ReductionKind::ToolInputElided { call_id, field, .. } => {
            let original_value =
                resolve_tool_input_value(&r.ptr, call_id, field, sidecar_messages)?;
            let current_args = out[pos]
                .tool_calls
                .as_ref()
                .and_then(|calls| calls.iter().find(|c| &c.id == call_id))
                .map(|call| call.function.arguments.clone())
                .ok_or_else(|| {
                    Error::new(format!(
                        "invert: reduced message at position {pos} has no tool_call with id \
                         {call_id}"
                    ))
                })?;
            // Splicing the recovered ORIGINAL value back into the CURRENT
            // (reduced) arguments string at the same field's span restores
            // the pristine bytes exactly: the reduced string differs from
            // the original ONLY in that one field's value (the forward
            // splice at project-time never touched anything else), so this
            // is byte-exact by construction — no separate "store the whole
            // original args" bookkeeping needed.
            let restored = replace_top_level_string_field(&current_args, field, &original_value)
                .ok_or_else(|| {
                    Error::new(format!(
                        "invert: reduced tool_call {call_id} arguments do not contain field \
                         `{field}` to restore"
                    ))
                })?;
            let calls = out[pos]
                .tool_calls
                .as_mut()
                .expect("checked above: tool_calls present");
            let call = calls
                .iter_mut()
                .find(|c| &c.id == call_id)
                .expect("checked above: call_id present");
            call.function.arguments = restored;
            out[pos].metadata.remove(REDUCTION_METADATA_KEY);
        }
    }
    Ok(())
}

/// Reconstruct the full view: every placeholder in `reduced` replaced by the
/// original content resolved from `sidecar_session` (the ONE canonical
/// model — a canonical message slice reconstructed from its sidecar),
/// hash-verified against [`SidecarPtr::content_hash`] before ever
/// substituting it in.
///
/// Fails loudly (`Err`) rather than silently partial: on any reduced message
/// whose `sc.reduction` id has no matching entry in `log` (an unresolvable
/// pointer — e.g. the log entry was deleted), on any pointer that no longer
/// resolves in the sidecar, or on any hash mismatch (a stale/foreign/tampered
/// sidecar).
///
/// Strips the `sc.reduction` bookkeeping key from every message's metadata —
/// it never survives inversion.
pub fn invert_messages(
    reduced: &[ChatMessage],
    log: &ReductionLog,
    sidecar_messages: &[ChatMessage],
) -> Result<Vec<ChatMessage>> {
    let mut out: Vec<ChatMessage> = reduced.to_vec();

    // Every reduced (stub-bearing) message must resolve to a log entry —
    // otherwise it is an unresolvable pointer by design (SPEC.md A6(b)).
    let by_id: HashMap<&str, &Reduction> =
        log.reductions.iter().map(|r| (r.id.as_str(), r)).collect();
    for msg in &out {
        if let Some(id) = reduction_id(msg) {
            if !by_id.contains_key(id) {
                return Err(Error::new(format!(
                    "invert: reduced message carries reduction id {id} with no matching entry \
                     in the reduction log — unresolvable pointer"
                )));
            }
        }
    }

    for r in &log.reductions {
        invert_one_reduction(&mut out, r, sidecar_messages)?;
    }

    for msg in out.iter_mut() {
        msg.metadata.remove(REDUCTION_METADATA_KEY);
    }

    Ok(out)
}

/// Verify every reduction in `log` resolves against `sidecar` — the same
/// hash-verify path `invert`/`invert_one` walk before ever substituting
/// content back in, without needing an actual reduced view to substitute
/// into. This is the user-facing detector for a broken transparency
/// invariant (C2 contract 3): `sessions show-reductions` (C4) and `convert`
/// (C7) both call this before doing anything else with a reduced session, so
/// a corrupt/tampered/stale sidecar is reported — naming the offending
/// record id — before any output is produced, rather than surfacing as a
/// confusing downstream failure (or, worse, silently substituting the wrong
/// content).
///
/// Returns the first offending record's error (which names its `id`); `Ok`
/// means every record in `log` resolves and hash-verifies cleanly.
pub fn verify_log_messages(log: &ReductionLog, sidecar_messages: &[ChatMessage]) -> Result<()> {
    for r in log.reductions.iter().chain(log.expanded.iter()) {
        let resolved = match &r.kind {
            ReductionKind::ToolOutputTruncated { .. }
            | ReductionKind::FileReadElided { .. }
            | ReductionKind::OutputNormalized { .. }
            | ReductionKind::FileReadDiffed { .. }
            | ReductionKind::DuplicateOutput { .. }
            | ReductionKind::Superseded { .. } => {
                resolve_original_content(&r.ptr, sidecar_messages).map(|_| ())
            }
            ReductionKind::ImageRedacted { part_index } => {
                resolve_image_part(&r.ptr, *part_index, sidecar_messages).map(|_| ())
            }
            ReductionKind::TurnsCleared { first, last, .. } => {
                resolve_turns_range(&r.ptr, *first, *last, sidecar_messages).map(|_| ())
            }
            ReductionKind::ToolInputElided { call_id, field, .. } => {
                resolve_tool_input_value(&r.ptr, call_id, field, sidecar_messages).map(|_| ())
            }
        };
        if let Err(e) = resolved {
            return Err(Error::new(format!(
                "reduction {} unresolvable against the sidecar: {e}",
                r.id
            )));
        }
    }
    Ok(())
}

/// Rehydrate a single reduction by id (C4 `/expand <id>`; also used for
/// per-range `TurnsCleared` expansion). Returns the updated view and a log
/// with that record removed (an expanded reduction is no longer "applied").
///
/// Other placeholders in `reduced` are untouched — their bytes remain
/// identical.
pub fn invert_one_messages(
    reduced: &[ChatMessage],
    log: &ReductionLog,
    id: &str,
    sidecar_messages: &[ChatMessage],
) -> Result<(Vec<ChatMessage>, ReductionLog)> {
    let r = log
        .reductions
        .iter()
        .find(|r| r.id == id)
        .cloned()
        .ok_or_else(|| Error::new(format!("invert_one: no reduction with id {id} in the log")))?;

    let mut out: Vec<ChatMessage> = reduced.to_vec();
    invert_one_reduction(&mut out, &r, sidecar_messages)?;

    let mut new_log = log.clone();
    new_log.reductions.retain(|x| x.id != id);
    if !new_log.expanded.iter().any(|x| x.id == r.id) {
        new_log.expanded.push(r);
    }

    Ok((out, new_log))
}

#[cfg(test)]
mod tool_input_splice_tests {
    //! Unit tests for the TR-10 byte-surgical JSON field replacement
    //! primitives (private to this module) — the gotcha these exist to
    //! satisfy: "replace only the payload field's value, do NOT reserialize
    //! the whole args." Exercised directly here since they're not part of
    //! the public API; `tests/tool_input_elision.rs` covers the
    //! `project_messages`/`invert` integration level.
    use super::*;

    #[test]
    fn finds_and_replaces_only_the_named_fields_value() {
        let json = r#"{"path":"src/foo.rs","content":"hello world","flag":true}"#;
        let (start, end) = find_top_level_string_field(json, "content").unwrap();
        assert_eq!(&json[start..end], "hello world");

        let replaced = replace_top_level_string_field(json, "content", "STUB").unwrap();
        assert_eq!(
            replaced,
            r#"{"path":"src/foo.rs","content":"STUB","flag":true}"#
        );
        // Every other byte (key order, the `path`/`flag` values, punctuation)
        // is untouched — not a full reparse+reserialize.
        assert!(replaced.contains(r#""path":"src/foo.rs""#));
        assert!(replaced.contains(r#""flag":true"#));
    }

    #[test]
    fn preserves_whitespace_and_key_order_around_the_replaced_field() {
        // Deliberately unusual formatting a naive reparse+reserialize would
        // normalize away (extra spaces, content BEFORE path).
        let json = "{ \"content\" : \"big\",   \"path\":\"a/b.rs\" }";
        let replaced = replace_top_level_string_field(json, "content", "X").unwrap();
        assert_eq!(replaced, "{ \"content\" : \"X\",   \"path\":\"a/b.rs\" }");
    }

    #[test]
    fn handles_escaped_quotes_backslashes_and_unicode_in_the_value() {
        let original_value = "line1\nline2 \"quoted\" \\ and unicode caf\u{e9}";
        let json = serde_json::json!({"path": "p", "content": original_value}).to_string();
        let (start, end) = find_top_level_string_field(&json, "content").unwrap();
        // The span is the RAW (still-escaped) body; decoding it must recover
        // the original value exactly.
        let raw = format!("\"{}\"", &json[start..end]);
        let decoded: String = serde_json::from_str(&raw).unwrap();
        assert_eq!(decoded, original_value);

        let new_value = "replacement with \"quotes\" and \\ backslash and \u{1f600}";
        let replaced = replace_top_level_string_field(&json, "content", new_value).unwrap();
        let reparsed: serde_json::Value = serde_json::from_str(&replaced).unwrap();
        assert_eq!(reparsed["content"], new_value);
        assert_eq!(reparsed["path"], "p");
    }

    #[test]
    fn skips_nested_objects_and_arrays_in_sibling_fields() {
        let json = r#"{"meta":{"a":[1,2,{"b":"}}}"}]},"content":"payload","tags":["x","y"]}"#;
        let (start, end) = find_top_level_string_field(json, "content").unwrap();
        assert_eq!(&json[start..end], "payload");
        let replaced = replace_top_level_string_field(json, "content", "NEW").unwrap();
        let reparsed: serde_json::Value = serde_json::from_str(&replaced).unwrap();
        assert_eq!(reparsed["content"], "NEW");
        assert_eq!(reparsed["tags"][0], "x");
        assert_eq!(reparsed["meta"]["a"][2]["b"], "}}}");
    }

    #[test]
    fn returns_none_when_field_absent_or_not_a_string_or_not_an_object() {
        assert_eq!(
            find_top_level_string_field(r#"{"path":"a"}"#, "content"),
            None
        );
        assert_eq!(
            find_top_level_string_field(r#"{"content":42}"#, "content"),
            None
        );
        assert_eq!(
            find_top_level_string_field(r#"["not","an","object"]"#, "content"),
            None
        );
        assert_eq!(
            find_top_level_string_field("not json at all", "content"),
            None
        );
        assert_eq!(
            replace_top_level_string_field(r#"{"path":"a"}"#, "content", "x"),
            None
        );
    }
}