supercode-harness 0.4.3

//! §2 module 20 `checkpoint` (COMPOSABLE-HARNESS-DESIGN.md line 470): file
//! checkpointing / shadow-git; D4-adjacent revert; D3 turn-diff tracking.
//! Line 1504: "restores FILES, not context — pairs with, never replaces,
//! the reduction sidecar."
//!
//! **Not to be confused with** [`crate::Agent::checkpoint`]/[`crate::Agent::rewind_to`]
//! (an in-memory CONVERSATION-position marker, an unrelated pre-existing
//! facility) or [`crate::session_tree`] (module 21, tree-addressable
//! sessions). This module only ever touches files on disk.
//!
//! # The write-path interception seam (D-5, shared with `formatters`/P5-11)
//! [`crate::tools::WriteObserver`] (defined in `crate::tools`, the seam's
//! natural owner since [`crate::tools::ToolContext`] carries it) is called
//! by `write_file`/`edit_file`/`apply_patch` around their mutation.
//! [`CheckpointObserver`] is this module's implementation: `before_write`
//! captures a pre-image; `after_write` is a no-op (reserved for
//! `formatters`).
//!
//! # Storage — never the user's real git
//! [`CheckpointStore`] is a content-addressed, per-project shadow store at
//! a caller-supplied `root` (the same dependency-injection shape as
//! [`crate::store::SessionStore::open`]) — blobs keyed by
//! [`crate::reduce::content_hash`] under `<root>/objects/`, one JSON
//! manifest per checkpoint under `<root>/checkpoints/<id>.json`. This
//! module never shells out to `git` and never touches the project's own
//! `.git` — there is no `GIT_DIR` to get wrong because none is ever used.
//!
//! # C8 (design line 1819) compliance
//! A [`CheckpointId`] is an opaque `String` — the ONLY thing a caller
//! (session record, CLI) ever threads around. It is never the file
//! contents, and it never reaches the model's context window: restoring
//! files is a disk operation, never a conversation-history mutation.

use std::collections::HashSet;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Mutex;

use serde::{Deserialize, Serialize};

use crate::error::{Error, Result};
use crate::tools::WriteObserver;

/// Default number of checkpoints retained per project before the oldest are
/// pruned (bounded-disk requirement) — overridable via
/// `[capabilities.checkpoint].retain`.
pub const DEFAULT_RETAIN: usize = 50;

/// An opaque, lexically-sortable (ascending = chronological, since it's
/// zero-padded-millis-prefixed) checkpoint identifier. This is ALL a
/// session record ever carries for C8 — never file contents.
pub type CheckpointId = String;

fn now_ms() -> u128 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis())
        .unwrap_or(0)
}

/// Mint a fresh id: zero-padded millis + an 8-hex-digit salt derived from
/// (pid, millis, a monotonic instant) — collision-free in practice without
/// pulling in a UUID/random dependency (matches this crate's existing
/// `shell_sentinel` precedent in `tools/builtins.rs`).
fn mint_id() -> CheckpointId {
    use std::hash::BuildHasher;
    let millis = now_ms();
    let salt = std::collections::hash_map::RandomState::new().hash_one((
        std::process::id(),
        millis,
        std::time::Instant::now(),
    ));
    format!("{millis:020}-{:08x}", salt as u32)
}

/// One file's entry in a [`CheckpointManifest`] — the turn-diff (D3) is
/// simply this manifest's `files` list.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct CheckpointFileEntry {
    /// Path relative to the project root, `/`-separated, never absolute and
    /// never containing a `..` component in what [`CheckpointObserver`]
    /// itself writes (restore independently re-validates this — see
    /// [`CheckpointStore::restore`] — rather than trusting it).
    pub path: String,
    /// blake3 hex digest ([`crate::reduce::content_hash`]) of the pre-image
    /// content, or `None` if the file did not exist before this
    /// checkpoint's turn began (i.e. the write that follows is a create —
    /// restoring deletes it again).
    pub blob: Option<String>,
}

/// A checkpoint's full manifest — one JSON file per checkpoint.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CheckpointManifest {
    /// This checkpoint's id.
    pub id: CheckpointId,
    /// Unix-epoch milliseconds when this checkpoint was minted.
    pub created_at_ms: u128,
    /// A short, human-readable label (e.g. the turn's prompt excerpt) —
    /// display only, never re-parsed.
    pub label: String,
    /// Every file this turn captured a pre-image for — the D3 turn-diff.
    pub files: Vec<CheckpointFileEntry>,
}

/// Lightweight listing entry ([`CheckpointStore::list`]) — the manifest
/// without the (potentially long) file list.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CheckpointMeta {
    /// This checkpoint's id.
    pub id: CheckpointId,
    /// Unix-epoch milliseconds when this checkpoint was minted.
    pub created_at_ms: u128,
    /// A short, human-readable label — display only.
    pub label: String,
    /// How many files this checkpoint's manifest lists.
    pub file_count: usize,
}

/// The outcome of a [`CheckpointStore::restore`] call.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct RestoreReport {
    /// Files successfully restored (project-relative paths).
    pub restored: Vec<String>,
    /// Files refused, with the reason — e.g. an out-of-root or
    /// protected-path target. Non-empty means the restore was PARTIAL; a
    /// caller must surface this, never silently swallow it.
    pub refused: Vec<(String, String)>,
}

/// A directory-backed, content-addressed shadow store rooted at `root` —
/// see the module doc comment. Never touches anything outside `root` except
/// (during [`Self::restore`]) the project files a manifest names, which are
/// re-validated against `project_root` independently of how the manifest
/// was produced.
#[derive(Debug)]
pub struct CheckpointStore {
    root: PathBuf,
}

impl CheckpointStore {
    /// Open (creating if needed) a store at `root`. `Err` if `root` can't
    /// be created (e.g. an unwritable state dir) — callers (see
    /// [`observer_for_config`]) treat that as "disable checkpoint, warn
    /// once", never a crash.
    pub fn open(root: impl Into<PathBuf>) -> Result<Self> {
        let root = root.into();
        std::fs::create_dir_all(root.join("objects"))?;
        std::fs::create_dir_all(root.join("checkpoints"))?;
        Ok(CheckpointStore { root })
    }

    fn objects_dir(&self) -> PathBuf {
        self.root.join("objects")
    }
    fn checkpoints_dir(&self) -> PathBuf {
        self.root.join("checkpoints")
    }
    fn manifest_path(&self, id: &str) -> PathBuf {
        self.checkpoints_dir().join(format!("{id}.json"))
    }
    fn blob_path(&self, hash: &str) -> PathBuf {
        let prefix = &hash[..hash.len().min(2)];
        self.objects_dir().join(prefix).join(hash)
    }

    /// Reject an `id` that could escape the store root when joined into a
    /// path — the exact same defensive shape as
    /// `crate::store::SessionStore::validate_name`, since a `checkpoint
    /// restore <id>`/`checkpoint diff <id>` CLI argument is untrusted user
    /// input by the time it reaches here.
    fn validate_id(id: &str) -> Result<()> {
        let bad = id.is_empty()
            || id.contains('/')
            || id.contains('\\')
            || id.contains('\0')
            || id.split(['/', '\\']).any(|c| c == ".." || c == ".")
            || Path::new(id).is_absolute()
            || id.trim() != id;
        if bad {
            return Err(Error::Other(format!("invalid checkpoint id: `{id}`")));
        }
        Ok(())
    }

    /// Content-address `content`, writing it only if not already present
    /// (dedup — the same content written by two different checkpoints costs
    /// one blob). Atomic (write-to-tmp, rename) so a concurrent reader
    /// never observes a partial blob; a lost race against another writer
    /// racing the SAME hash is harmless (identical content either way).
    fn write_blob(&self, content: &[u8]) -> Result<String> {
        let hash = crate::reduce::content_hash(content);
        let dest = self.blob_path(&hash);
        if dest.exists() {
            return Ok(hash);
        }
        let Some(parent) = dest.parent() else {
            return Err(Error::Other("blob path has no parent".to_string()));
        };
        std::fs::create_dir_all(parent)?;
        let tmp = parent.join(format!(".tmp-{}-{}", std::process::id(), mint_id()));
        std::fs::write(&tmp, content)?;
        match std::fs::rename(&tmp, &dest) {
            Ok(()) => {}
            Err(e) if dest.exists() => {
                // Another writer won the race with identical content.
                let _ = std::fs::remove_file(&tmp);
                let _ = e;
            }
            Err(e) => return Err(e.into()),
        }
        Ok(hash)
    }

    fn read_blob(&self, hash: &str) -> Result<Vec<u8>> {
        std::fs::read(self.blob_path(hash)).map_err(Into::into)
    }

    fn write_manifest(&self, m: &CheckpointManifest) -> Result<()> {
        Self::validate_id(&m.id)?;
        let dest = self.manifest_path(&m.id);
        let json = serde_json::to_vec_pretty(m).map_err(|e| Error::Other(e.to_string()))?;
        let dir = self.checkpoints_dir();
        std::fs::create_dir_all(&dir)?;
        let tmp = dir.join(format!(".tmp-{}-{}", std::process::id(), mint_id()));
        std::fs::write(&tmp, &json)?;
        std::fs::rename(&tmp, &dest)?;
        Ok(())
    }

    /// Mint a fresh checkpoint (a new turn) with an empty file list. `label`
    /// is display-only.
    pub fn create_checkpoint(&self, label: &str) -> Result<CheckpointId> {
        let id = mint_id();
        let manifest = CheckpointManifest {
            id: id.clone(),
            created_at_ms: now_ms(),
            label: label.to_string(),
            files: Vec::new(),
        };
        self.write_manifest(&manifest)?;
        Ok(id)
    }

    /// Idempotently record `rel`'s pre-image under checkpoint `id` — a
    /// SECOND call for the same `(id, rel)` pair is a no-op (the manifest
    /// always keeps the EARLIEST pre-image seen this turn, which is the one
    /// a revert needs). `content: None` means the file did not exist yet.
    pub fn record_pre_image(&self, id: &str, rel: &str, content: Option<Vec<u8>>) -> Result<()> {
        let mut manifest = self.manifest(id)?;
        if manifest.files.iter().any(|f| f.path == rel) {
            return Ok(());
        }
        let blob = match content {
            Some(bytes) => Some(self.write_blob(&bytes)?),
            None => None,
        };
        manifest.files.push(CheckpointFileEntry {
            path: rel.to_string(),
            blob,
        });
        self.write_manifest(&manifest)
    }

    /// Read one checkpoint's full manifest.
    pub fn manifest(&self, id: &str) -> Result<CheckpointManifest> {
        Self::validate_id(id)?;
        let text = std::fs::read_to_string(self.manifest_path(id))
            .map_err(|e| Error::Other(format!("checkpoint `{id}` not found: {e}")))?;
        serde_json::from_str(&text)
            .map_err(|e| Error::Other(format!("checkpoint `{id}` manifest is corrupt: {e}")))
    }

    /// Every checkpoint in the store, newest first (ids are millis-prefixed
    /// so lexical descending order IS chronological descending order). A
    /// corrupt individual manifest is skipped (resilience — one bad file
    /// never hides every other checkpoint), not a hard error.
    pub fn list(&self) -> Result<Vec<CheckpointMeta>> {
        let dir = self.checkpoints_dir();
        let mut metas = Vec::new();
        if !dir.exists() {
            return Ok(metas);
        }
        for entry in std::fs::read_dir(&dir)? {
            let entry = entry?;
            let path = entry.path();
            if path.extension().and_then(|e| e.to_str()) != Some("json") {
                continue;
            }
            let Ok(text) = std::fs::read_to_string(&path) else {
                continue;
            };
            if let Ok(m) = serde_json::from_str::<CheckpointManifest>(&text) {
                metas.push(CheckpointMeta {
                    id: m.id,
                    created_at_ms: m.created_at_ms,
                    label: m.label,
                    file_count: m.files.len(),
                });
            }
        }
        metas.sort_by(|a, b| b.id.cmp(&a.id));
        Ok(metas)
    }

    /// D3 turn-diff: the set of project-relative paths this checkpoint's
    /// turn touched — exactly the manifest's file list (every entry exists
    /// BECAUSE a write-tool call captured a pre-image for it this turn).
    pub fn turn_diff(&self, id: &str) -> Result<Vec<String>> {
        Ok(self
            .manifest(id)?
            .files
            .into_iter()
            .map(|f| f.path)
            .collect())
    }

    /// D4-adjacent revert: restore `project_root`'s working files to
    /// checkpoint `id`. For each manifest entry: `Some(blob)` rewrites the
    /// file to that pre-image; `None` (didn't exist before the turn)
    /// deletes it if present now (undoing a create). SECURITY: every
    /// target is independently re-validated (never trusts the manifest was
    /// produced honestly) against `project_root` containment (no symlink
    /// escape, no `..` traversal) AND `protected_globs` (plus an
    /// unconditional `.git/**` floor) checked against BOTH the LEXICAL
    /// normalized path and the symlink-RESOLVED path — a lexical-only
    /// check would miss a manifest entry like `foo/config` where `foo` is
    /// a pre-existing symlink into `.git`: lexically it's clean, but it
    /// resolves inside `root` (so containment alone accepts it too) and
    /// lands on the real `.git/config`. A refused entry is recorded in
    /// [`RestoreReport::refused`], never silently applied AND never aborts
    /// the rest of the restore (partial-success, fully reported).
    pub fn restore(
        &self,
        id: &str,
        project_root: &Path,
        protected_globs: &[String],
    ) -> Result<RestoreReport> {
        let manifest = self.manifest(id)?;
        let mut report = RestoreReport::default();
        for entry in &manifest.files {
            // (1) Up-front rejection: a legitimately-captured entry (see
            // `CheckpointFileEntry::path`'s doc comment) is always a clean
            // relative path — `record_pre_image` never produces an absolute
            // path or a `..` component. An entry that has one is by
            // definition hostile or corrupt (traversal-injected, or a
            // corrupted/copied/shared manifest — exactly this module's
            // stated threat model) and must never reach a raw-string
            // pattern match at all.
            if let Some(reason) = reject_unsafe_manifest_path(&entry.path) {
                report.refused.push((entry.path.clone(), reason));
                continue;
            }
            let target = project_root.join(&entry.path);
            // (2) Normalize-then-protect: compute the SAME lexically-
            // normalized project-relative path `contained()` uses
            // internally, ONCE, and check `is_protected` against THAT
            // (not the raw manifest string) — so a `.git/**`-floor or
            // `protected_globs` bypass via `x/../.git/config` can no
            // longer disagree between the two checks (the root cause of
            // the bug this replaces: `is_protected` matched the raw
            // string while `contained` matched the normalized path).
            let Some(normalized_rel) = normalized_project_rel(project_root, &target) else {
                report.refused.push((
                    entry.path.clone(),
                    "refused: escapes the project root".to_string(),
                ));
                continue;
            };
            if is_protected(&normalized_rel, protected_globs) {
                report
                    .refused
                    .push((entry.path.clone(), "refused: protected path".to_string()));
                continue;
            }
            // Belt-and-suspenders: the existing symlink-safe containment
            // check (canonicalizes the longest existing ancestor, refusing
            // any symlink escape) still runs unconditionally.
            if !contained(project_root, &target) {
                report.refused.push((
                    entry.path.clone(),
                    "refused: escapes the project root".to_string(),
                ));
                continue;
            }
            // (3) Resolved-then-protect: `normalized_rel` above is a purely
            // LEXICAL collapse — it never resolves symlinks — while
            // `contained` (just above) DOES resolve symlinks when it
            // canonicalizes the longest existing ancestor. Those two views
            // of the path can disagree exactly when a component of `target`
            // is a symlink: e.g. a pre-existing `foo -> .git` inside
            // `project_root` plus a manifest entry `foo/config` lexically
            // normalizes to `foo/config` (not protected — no literal
            // `.git/` prefix) yet resolves to `<root>/.git/config` (still
            // "contained" under `project_root`, so the escape check above
            // doesn't catch it either — it never leaves the root, it just
            // lands somewhere the lexical path didn't say). Re-run
            // `is_protected` against the RESOLVED, symlink-followed
            // project-relative path too, so this can't slip through: the
            // `.git`/`protected_globs` floor now sees BOTH the lexical and
            // the resolved view, and refuses if EITHER is protected.
            if let Some(resolved_rel) = resolved_project_rel(project_root, &target) {
                if is_protected(&resolved_rel, protected_globs) {
                    report
                        .refused
                        .push((entry.path.clone(), "refused: protected path".to_string()));
                    continue;
                }
            }
            match &entry.blob {
                Some(hash) => {
                    let bytes = match self.read_blob(hash) {
                        Ok(b) => b,
                        Err(e) => {
                            report
                                .refused
                                .push((entry.path.clone(), format!("blob unreadable: {e}")));
                            continue;
                        }
                    };
                    if let Some(parent) = target.parent() {
                        let _ = std::fs::create_dir_all(parent);
                    }
                    if let Err(e) = std::fs::write(&target, &bytes) {
                        report
                            .refused
                            .push((entry.path.clone(), format!("write failed: {e}")));
                        continue;
                    }
                }
                None if target.exists() => {
                    if let Err(e) = std::fs::remove_file(&target) {
                        report
                            .refused
                            .push((entry.path.clone(), format!("delete failed: {e}")));
                        continue;
                    }
                }
                None => {}
            }
            report.restored.push(entry.path.clone());
        }
        Ok(report)
    }

    /// Bounded-disk requirement: keep only the `keep` newest checkpoints,
    /// deleting the rest, then GC any blob no longer referenced by a
    /// surviving manifest. Returns the number of checkpoints removed.
    pub fn prune(&self, keep: usize) -> Result<usize> {
        let mut metas = self.list()?; // newest first
        if metas.len() <= keep {
            return Ok(0);
        }
        let stale = metas.split_off(keep);
        let removed = stale.len();
        for m in stale {
            let _ = std::fs::remove_file(self.manifest_path(&m.id));
        }
        self.gc_unreferenced_blobs()?;
        Ok(removed)
    }

    /// Delete every blob under `objects/` not referenced by ANY surviving
    /// manifest — a full scan, not a refcount (simplest correct form; the
    /// retention bound keeps this cheap in practice — see [`Self::prune`]'s
    /// doc comment).
    fn gc_unreferenced_blobs(&self) -> Result<()> {
        let mut referenced: HashSet<String> = HashSet::new();
        for meta in self.list()? {
            if let Ok(m) = self.manifest(&meta.id) {
                for f in m.files {
                    if let Some(b) = f.blob {
                        referenced.insert(b);
                    }
                }
            }
        }
        let objects = self.objects_dir();
        if !objects.exists() {
            return Ok(());
        }
        for entry in std::fs::read_dir(&objects)? {
            let entry = entry?;
            if !entry.file_type()?.is_dir() {
                continue;
            }
            for inner in std::fs::read_dir(entry.path())? {
                let inner = inner?;
                let name = inner.file_name();
                let Some(name) = name.to_str() else {
                    continue;
                };
                if name.starts_with(".tmp-") {
                    continue; // an in-flight write, not ours to reap
                }
                if !referenced.contains(name) {
                    let _ = std::fs::remove_file(inner.path());
                }
            }
        }
        Ok(())
    }
}

// SECURITY (safe-path consolidation, CRITICAL fix): every primitive below
// used to be defined HERE, locally — this module's own P5-9 fix for the
// `.git`-clobber bug class. It is now `crate::safe_path`'s canonical
// implementation instead, with checkpoint DELEGATING to it (thin wrappers,
// same names, same signatures, same behavior) so `crate::permissions`'s
// gate and `crate::tools`'s sandbox containment reuse this exact proven
// logic rather than each re-implementing (and, in the permissions gate's
// case, getting wrong) their own. See `crate::safe_path`'s module doc
// comment for the full story and the two-views-of-a-path explanation every
// doc comment below used to carry inline.
//
// Every one of this module's 23 tests (see `tests` below) still exercises
// these names directly and is UNCHANGED — that is the proof this extraction
// is behavior-preserving.

/// Is `rel` (a project-relative, `/`-separated path) a hard floor this
/// module refuses to snapshot INTO or restore OVER, regardless of config?
/// `.git` (and everything under it) is unconditional — checkpoint must
/// never touch the user's real git repo even if `permissions.protected_paths`
/// (module 13) is off. `extra_globs` layers `Config::permissions_protected_paths`
/// on top when the caller has one (restore only — capture-time protection
/// is already covered because [`CheckpointObserver::before_write`] only
/// ever fires for a write the P5-1 permission gate already approved, which
/// already folds `protected_paths` in — see that method's own doc comment).
/// See `crate::safe_path::is_protected` for the implementation.
fn is_protected(rel: &str, extra_globs: &[String]) -> bool {
    crate::safe_path::is_protected(rel, extra_globs)
}

/// Up-front rejection for a manifest-declared path that could never have
/// come from a legitimate capture: `record_pre_image` only ever stores a
/// clean, `/`-separated, project-relative path (see
/// [`CheckpointFileEntry::path`]'s doc comment), so an absolute path or one
/// containing a `..` (`ParentDir`), root, or Windows-prefix component is by
/// definition hostile or corrupt. Returns the refusal reason, or `None` if
/// `rel` is clean. Called BEFORE any raw-string pattern match (e.g.
/// [`is_protected`]) so a traversal entry like `x/../.git/config` — which
/// does not literally string-match the `.git/` floor — is refused before
/// it can ever be compared against anything. See
/// `crate::safe_path::reject_unsafe_rel_path` for the implementation.
fn reject_unsafe_manifest_path(rel: &str) -> Option<String> {
    crate::safe_path::reject_unsafe_rel_path(rel)
}

/// Lexically normalize `path` (expected to be `root.join(rel)` for some
/// manifest-declared `rel`) via the SAME collapse [`contained`] uses, then —
/// if the normalized form is still under `root` — return its
/// project-relative, `/`-separated tail. Callers (currently only
/// [`CheckpointStore::restore`]) compute this ONCE per entry and feed the
/// single result to [`is_protected`], so the protected-path floor sees
/// EXACTLY the same normalized path [`contained`]'s containment check
/// computes. See `crate::safe_path::normalized_project_rel` for the
/// implementation.
fn normalized_project_rel(root: &Path, path: &Path) -> Option<String> {
    crate::safe_path::normalized_project_rel(root, path)
}

/// Symlink- and traversal-safe containment check: is `path` (absolute,
/// possibly not-yet-existing) confined under `root`? Fails closed (`false`)
/// on any resolution error. See `crate::safe_path::contained` for the
/// implementation.
fn contained(root: &Path, path: &Path) -> bool {
    crate::safe_path::contained(root, path)
}

/// Resolved (symlink-following) counterpart to [`normalized_project_rel`]:
/// canonicalizes `path`'s longest existing ancestor (resolving any symlink
/// along the way — see `crate::safe_path::resolve_real`) and returns the
/// resulting absolute path's tail relative to `root`'s own canonical form, as a
/// `/`-separated string — or `None` if resolution fails, or the resolved
/// path lands outside `root` entirely (that case is already refused by
/// [`contained`]; this function only needs to report a rel path when
/// there IS one). [`CheckpointStore::restore`] feeds this to
/// [`is_protected`] IN ADDITION TO the lexical [`normalized_project_rel`]
/// result, closing the gap where a symlink resolves into `.git` (or a
/// `protected_globs` match) even though its LEXICAL path never mentions
/// `.git` at all and it never leaves `root` (so `contained` alone would
/// accept it) — see the call site's doc comment for the exact repro this
/// closes. See `crate::safe_path::resolved_project_rel` for the
/// implementation.
fn resolved_project_rel(root: &Path, path: &Path) -> Option<String> {
    crate::safe_path::resolved_project_rel(root, path)
}

#[derive(Debug, Default)]
struct ObserverState {
    /// The checkpoint currently open for the in-flight turn.
    current: Option<CheckpointId>,
    /// Project-relative paths already captured under `current` — first
    /// write per turn wins the pre-image; every write after the first is a
    /// no-op capture (the manifest already holds the earliest state).
    captured: HashSet<String>,
}

/// [`WriteObserver`] implementation backing `[capabilities.checkpoint]`.
/// One instance per [`crate::Agent`] (installed on its [`crate::tools::ToolContext`]
/// AND held directly so `crate::Agent::run_loop` can call
/// [`Self::begin_turn`] once per user turn — see that method's call site's
/// doc comment).
#[derive(Debug)]
pub struct CheckpointObserver {
    store: CheckpointStore,
    project_root: PathBuf,
    retain: usize,
    protected: Vec<String>,
    state: Mutex<ObserverState>,
    /// Set once on any I/O failure — graceful degrade (§ "git-absent /
    /// unwritable state dir ⇒ warn + disable, never crash"), not a panic
    /// and not a blocked tool call. `Relaxed` throughout: this is a
    /// best-effort circuit breaker, not a correctness-critical ordering.
    disabled: AtomicBool,
}

impl CheckpointObserver {
    /// `protected` layers extra glob patterns (typically
    /// `Config::permissions_protected_paths`, when module 13 is active) on
    /// top of the unconditional `.git/**` floor — consulted by
    /// [`Self::restore`].
    pub fn new(
        store: CheckpointStore,
        project_root: PathBuf,
        retain: usize,
        protected: Vec<String>,
    ) -> Self {
        CheckpointObserver {
            store,
            project_root,
            retain: retain.max(1),
            protected,
            state: Mutex::new(ObserverState::default()),
            disabled: AtomicBool::new(false),
        }
    }

    /// Read-only access to the underlying store (e.g. so a caller can
    /// `list`/`turn_diff`/`restore` without re-deriving the root path).
    pub fn store(&self) -> &CheckpointStore {
        &self.store
    }

    /// List every checkpoint, newest first — see [`CheckpointStore::list`].
    pub fn list(&self) -> Result<Vec<CheckpointMeta>> {
        self.store.list()
    }

    /// D3 turn-diff for one checkpoint — see [`CheckpointStore::turn_diff`].
    pub fn turn_diff(&self, id: &str) -> Result<Vec<String>> {
        self.store.turn_diff(id)
    }

    /// D4-adjacent revert: restore this project's working files to
    /// checkpoint `id`, honoring THIS observer's own `project_root` and
    /// `protected` globs (the fields set at construction) — see
    /// [`CheckpointStore::restore`] for the full security contract.
    pub fn restore(&self, id: &str) -> Result<RestoreReport> {
        self.store.restore(id, &self.project_root, &self.protected)
    }

    /// Whether this observer has disabled itself after an I/O failure.
    pub fn is_disabled(&self) -> bool {
        self.disabled.load(Ordering::Relaxed)
    }

    /// Open a fresh checkpoint for a new turn — called once at the top of
    /// `crate::Agent::run_loop` (i.e. once per `Agent::send`/
    /// `send_with_files`/`send_with_images` call, cc's "per-prompt
    /// file-history-snapshot"). `label` is a short excerpt of the turn's
    /// prompt, display-only. Also prunes past the retention bound here
    /// (once per turn, not once per write) — see [`CheckpointStore::prune`].
    /// Returns `None` when disabled (config-off is never routed here at
    /// all — see [`observer_for_config`] — so `None` here specifically
    /// means an I/O failure already tripped the breaker).
    pub fn begin_turn(&self, label: &str) -> Option<CheckpointId> {
        if self.disabled.load(Ordering::Relaxed) {
            return None;
        }
        let short: String = label.chars().take(120).collect();
        match self.store.create_checkpoint(&short) {
            Ok(id) => {
                if let Ok(mut st) = self.state.lock() {
                    st.current = Some(id.clone());
                    st.captured.clear();
                }
                if let Err(e) = self.store.prune(self.retain) {
                    eprintln!("warning: checkpoint: prune failed: {e}");
                }
                Some(id)
            }
            Err(e) => {
                eprintln!(
                    "warning: checkpoint disabled for the rest of this session — \
                     failed to open a new checkpoint: {e}"
                );
                self.disabled.store(true, Ordering::Relaxed);
                None
            }
        }
    }

    /// The checkpoint currently open for the in-flight turn, if any.
    pub fn current(&self) -> Option<CheckpointId> {
        self.state.lock().ok().and_then(|s| s.current.clone())
    }
}

#[async_trait::async_trait]
impl WriteObserver for CheckpointObserver {
    async fn before_write(&self, path: &Path) {
        if self.disabled.load(Ordering::Relaxed) {
            return;
        }
        if !contained(&self.project_root, path) {
            // Outside the project root — out of this module's scope (§ "an
            // honest gap", never a crash or a wrong snapshot).
            return;
        }
        let Some(normalized) = crate::tools::normalize(path) else {
            return;
        };
        let Some(root_normalized) = crate::tools::normalize(&self.project_root) else {
            return;
        };
        let Ok(rel_path) = normalized.strip_prefix(&root_normalized) else {
            return;
        };
        let rel = rel_path.to_string_lossy().replace('\\', "/");
        if rel.is_empty() || rel == ".git" || rel.starts_with(".git/") {
            return;
        }

        // The ENTIRE check-read-record sequence runs under one lock, so two
        // concurrent writes to the SAME path (e.g. a `run_tools_concurrently`
        // batch, or a P5-6 background job racing a foreground write) can
        // never both read-and-lose a torn pre-image: the second one to
        // arrive here always sees `captured` already contains `rel` and
        // skips entirely, never re-reading a post-first-write state.
        let mut st = match self.state.lock() {
            Ok(g) => g,
            Err(poisoned) => poisoned.into_inner(),
        };
        if st.current.is_none() {
            // A write reached this seam with no open turn (e.g. a caller
            // driving the tool registry directly, outside `Agent::send`) —
            // self-heal with an ad-hoc checkpoint rather than silently
            // dropping the capture. `begin_turn` takes this same lock, so
            // it must be called with `st` released first.
            drop(st);
            self.begin_turn("untracked");
            st = match self.state.lock() {
                Ok(g) => g,
                Err(poisoned) => poisoned.into_inner(),
            };
        }
        if st.captured.contains(&rel) {
            return;
        }
        let Some(id) = st.current.clone() else {
            return; // begin_turn's own failure already warned + disabled
        };
        let content = std::fs::read(path).ok(); // None => doesn't exist yet (a create)
        match self.store.record_pre_image(&id, &rel, content) {
            Ok(()) => {
                st.captured.insert(rel);
            }
            Err(e) => {
                eprintln!(
                    "warning: checkpoint disabled for the rest of this session — \
                     failed to record a snapshot: {e}"
                );
                self.disabled.store(true, Ordering::Relaxed);
            }
        }
    }

    async fn after_write(&self, _path: &Path) -> Option<String> {
        // P5-11: `formatters`/`lsp` now occupy this hook (via
        // `crate::tools::WriteObserverChain`, installed AFTER this observer
        // in `crate::agent::build_tool_context`'s chain) — checkpoint itself
        // still has nothing to do after a write completes, and returning
        // `None` keeps the tool-result text this hook contributes
        // byte-identical to before P5-11 whenever checkpoint is the only
        // observer installed.
        None
    }
}

/// A stable per-project-directory tag — the same hash-of-canonicalized-cwd
/// idea `crates/cli/src/main.rs::cwd_tag` uses for session naming, kept
/// separately here (a `core`-crate concern, and `cli` depends on `core` not
/// the reverse) so two different projects never share one shadow store even
/// though they'd otherwise both resolve to the same `$SUPERCODE_HOME`-
/// derived parent directory.
fn project_tag(cwd: &Path) -> String {
    use std::hash::{Hash, Hasher};
    let canon = std::fs::canonicalize(cwd).unwrap_or_else(|_| cwd.to_path_buf());
    let mut h = std::collections::hash_map::DefaultHasher::new();
    canon.hash(&mut h);
    format!("{:016x}", h.finish())
}

/// The DEFAULT shadow-store root for `config.cwd` when
/// [`crate::Config::checkpoint_dir`] is unset: `$SUPERCODE_HOME/checkpoints/<project_tag>`
/// (`crate::agent::global_instructions_dir` is the same `$SUPERCODE_HOME`
/// resolver the global instruction tier and `checkpoint` both use).
fn default_shadow_root(cwd: &Path) -> PathBuf {
    crate::agent::global_instructions_dir()
        .join("checkpoints")
        .join(project_tag(cwd))
}

/// Build the [`CheckpointObserver`] a fresh [`crate::Agent`] should install,
/// given a resolved [`crate::Config`] — called once, from
/// `crate::agent::build_tool_context`. `Config::checkpoint_enabled` is the
/// ONE gate: `false` (the default) returns `None` WITHOUT touching the
/// filesystem at all (no `CheckpointStore::open`, no directory created) —
/// the default-off byte-identity guarantee. `true` opens (creating if
/// needed) the shadow store at `Config::checkpoint_dir`, or
/// `default_shadow_root` when that's unset; an I/O failure (unwritable
/// state dir) is reported via a one-time `eprintln!` warning and returns
/// `None` — graceful degrade, never a crash, never a blocked `Agent::new`.
pub fn observer_for_config(config: &crate::Config) -> Option<std::sync::Arc<CheckpointObserver>> {
    if !config.checkpoint_enabled {
        return None;
    }
    let root = config
        .checkpoint_dir
        .clone()
        .unwrap_or_else(|| default_shadow_root(&config.cwd));
    match CheckpointStore::open(&root) {
        Ok(store) => Some(std::sync::Arc::new(CheckpointObserver::new(
            store,
            config.cwd.clone(),
            config.checkpoint_retain,
            config.permissions_protected_paths.clone(),
        ))),
        Err(e) => {
            eprintln!(
                "warning: [capabilities.checkpoint] is enabled but the shadow store at \
                 {} could not be opened — checkpoint is disabled for this session: {e}",
                root.display()
            );
            None
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

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

    #[test]
    fn open_creates_objects_and_checkpoints_dirs() {
        let root = tmp("open");
        let store = CheckpointStore::open(&root).unwrap();
        assert!(root.join("objects").is_dir());
        assert!(root.join("checkpoints").is_dir());
        drop(store);
        std::fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn create_list_and_manifest_round_trip() {
        let root = tmp("list");
        let store = CheckpointStore::open(&root).unwrap();
        let id1 = store.create_checkpoint("first turn").unwrap();
        std::thread::sleep(std::time::Duration::from_millis(2));
        let id2 = store.create_checkpoint("second turn").unwrap();
        let metas = store.list().unwrap();
        assert_eq!(metas.len(), 2);
        // newest first.
        assert_eq!(metas[0].id, id2);
        assert_eq!(metas[1].id, id1);
        assert_eq!(metas[0].label, "second turn");
        let m = store.manifest(&id1).unwrap();
        assert_eq!(m.id, id1);
        assert!(m.files.is_empty());
        std::fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn record_pre_image_is_idempotent_keeping_the_earliest() {
        let root = tmp("idempotent");
        let store = CheckpointStore::open(&root).unwrap();
        let id = store.create_checkpoint("t").unwrap();
        store
            .record_pre_image(&id, "a.txt", Some(b"first".to_vec()))
            .unwrap();
        // A second capture of the SAME path (e.g. a second edit later in
        // the same turn) must NOT overwrite the first pre-image.
        store
            .record_pre_image(&id, "a.txt", Some(b"second".to_vec()))
            .unwrap();
        let m = store.manifest(&id).unwrap();
        assert_eq!(m.files.len(), 1);
        let bytes = store
            .read_blob(m.files[0].blob.as_deref().unwrap())
            .unwrap();
        assert_eq!(bytes, b"first");
        std::fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn blob_dedup_two_identical_contents_share_one_object() {
        let root = tmp("dedup");
        let store = CheckpointStore::open(&root).unwrap();
        let id = store.create_checkpoint("t").unwrap();
        store
            .record_pre_image(&id, "a.txt", Some(b"same".to_vec()))
            .unwrap();
        store
            .record_pre_image(&id, "b.txt", Some(b"same".to_vec()))
            .unwrap();
        let m = store.manifest(&id).unwrap();
        assert_eq!(m.files[0].blob, m.files[1].blob);
        // Exactly one blob file under objects/.
        let mut count = 0;
        for entry in walkdir(&root.join("objects")) {
            if entry.is_file() && !entry.to_string_lossy().contains(".tmp-") {
                count += 1;
            }
        }
        assert_eq!(count, 1);
        std::fs::remove_dir_all(&root).ok();
    }

    fn walkdir(dir: &Path) -> Vec<PathBuf> {
        let mut out = Vec::new();
        let Ok(rd) = std::fs::read_dir(dir) else {
            return out;
        };
        for entry in rd.flatten() {
            let p = entry.path();
            if p.is_dir() {
                out.extend(walkdir(&p));
            } else {
                out.push(p);
            }
        }
        out
    }

    #[test]
    fn restore_rewrites_modified_and_deletes_created_files() {
        let root = tmp("restore");
        let project = tmp("restore-project");
        std::fs::write(project.join("existing.txt"), "modified").unwrap();
        let store = CheckpointStore::open(&root).unwrap();
        let id = store.create_checkpoint("t").unwrap();
        // existing.txt existed before with "original".
        store
            .record_pre_image(&id, "existing.txt", Some(b"original".to_vec()))
            .unwrap();
        // new.txt did NOT exist before (a create this turn).
        store.record_pre_image(&id, "new.txt", None).unwrap();
        std::fs::write(project.join("new.txt"), "brand new").unwrap();

        let report = store.restore(&id, &project, &[]).unwrap();
        assert!(report.refused.is_empty(), "{:?}", report.refused);
        assert_eq!(report.restored.len(), 2);
        assert_eq!(
            std::fs::read_to_string(project.join("existing.txt")).unwrap(),
            "original"
        );
        assert!(!project.join("new.txt").exists());
        std::fs::remove_dir_all(&root).ok();
        std::fs::remove_dir_all(&project).ok();
    }

    #[test]
    fn restore_refuses_a_manifest_entry_that_traverses_outside_the_project_root() {
        // Hostile test (spec requirement): a checkpoint whose manifest
        // (however it got there) names a `..`-traversing path must be
        // refused, never applied — restore independently re-validates
        // every target, it does not trust the manifest.
        let root = tmp("hostile");
        let project = tmp("hostile-project");
        std::fs::create_dir_all(&project).unwrap();
        let store = CheckpointStore::open(&root).unwrap();
        let id = store.create_checkpoint("t").unwrap();
        // Hand-craft a manifest with a traversal path directly (bypassing
        // `record_pre_image`'s own normal, honest callers).
        let mut m = store.manifest(&id).unwrap();
        m.files.push(CheckpointFileEntry {
            path: "../../../../../../etc/passwd-supercode-test".to_string(),
            blob: None,
        });
        store.write_manifest(&m).unwrap();

        let victim = project
            .parent()
            .unwrap()
            .parent()
            .unwrap()
            .join("etc/passwd-supercode-test");
        assert!(
            !victim.exists(),
            "test precondition: victim path must not already exist"
        );

        let report = store.restore(&id, &project, &[]).unwrap();
        assert_eq!(report.restored.len(), 0);
        assert_eq!(report.refused.len(), 1);
        assert!(report.refused[0].1.contains("escapes"));
        assert!(
            !victim.exists(),
            "restore must never have written outside the project root"
        );
        std::fs::remove_dir_all(&root).ok();
        std::fs::remove_dir_all(&project).ok();
    }

    #[test]
    fn restore_refuses_dot_git_unconditionally() {
        let root = tmp("gitfloor");
        let project = tmp("gitfloor-project");
        std::fs::create_dir_all(project.join(".git")).unwrap();
        std::fs::write(project.join(".git").join("config"), "real git config").unwrap();
        let store = CheckpointStore::open(&root).unwrap();
        let id = store.create_checkpoint("t").unwrap();
        let mut m = store.manifest(&id).unwrap();
        m.files.push(CheckpointFileEntry {
            path: ".git/config".to_string(),
            blob: None,
        });
        store.write_manifest(&m).unwrap();

        let report = store.restore(&id, &project, &[]).unwrap();
        assert_eq!(report.restored.len(), 0);
        assert_eq!(report.refused.len(), 1);
        assert!(report.refused[0].1.contains("protected"));
        assert_eq!(
            std::fs::read_to_string(project.join(".git").join("config")).unwrap(),
            "real git config",
            "the real .git must be untouched"
        );
        std::fs::remove_dir_all(&root).ok();
        std::fs::remove_dir_all(&project).ok();
    }

    #[test]
    fn restore_honors_extra_protected_globs() {
        let root = tmp("protectedglob");
        let project = tmp("protectedglob-project");
        std::fs::write(project.join(".env"), "SECRET=1").unwrap();
        let store = CheckpointStore::open(&root).unwrap();
        let id = store.create_checkpoint("t").unwrap();
        let mut m = store.manifest(&id).unwrap();
        m.files.push(CheckpointFileEntry {
            path: ".env".to_string(),
            blob: Some(store.write_blob(b"OLD=1").unwrap()),
        });
        store.write_manifest(&m).unwrap();

        let report = store
            .restore(&id, &project, &[".env*".to_string()])
            .unwrap();
        assert_eq!(report.restored.len(), 0);
        assert_eq!(report.refused.len(), 1);
        assert_eq!(
            std::fs::read_to_string(project.join(".env")).unwrap(),
            "SECRET=1"
        );
        std::fs::remove_dir_all(&root).ok();
        std::fs::remove_dir_all(&project).ok();
    }

    #[test]
    fn restore_refuses_traversal_into_dot_git_real_git_config_stays_untouched() {
        // The exact reviewer repro: a manifest entry `x/../.git/config`
        // does NOT string-match the raw `.git/` floor (it literally starts
        // with `x/`), but lexically normalizes right back into
        // `<root>/.git/config`. Pre-fix, `is_protected` (raw string) said
        // "not protected" while `contained` (normalized) said "inside
        // root" — so restore wrote straight into the real `.git`. Post-fix
        // the up-front `..`-rejection refuses this before either check
        // runs, and even if that were bypassed, `is_protected` now runs on
        // the SAME normalized path `contained` uses.
        let root = tmp("traversal-git");
        let project = tmp("traversal-git-project");
        std::fs::create_dir_all(project.join(".git")).unwrap();
        std::fs::write(project.join(".git").join("config"), "real git config").unwrap();
        let store = CheckpointStore::open(&root).unwrap();
        let id = store.create_checkpoint("t").unwrap();
        let mut m = store.manifest(&id).unwrap();
        m.files.push(CheckpointFileEntry {
            path: "x/../.git/config".to_string(),
            blob: Some(store.write_blob(b"PWNED-by-traversal").unwrap()),
        });
        store.write_manifest(&m).unwrap();

        let report = store.restore(&id, &project, &[]).unwrap();
        assert_eq!(report.restored.len(), 0, "must not restore into .git");
        assert_eq!(report.refused.len(), 1);
        assert!(
            report.refused[0].1.contains("escapes"),
            "unexpected refusal reason: {}",
            report.refused[0].1
        );
        assert_eq!(
            std::fs::read_to_string(project.join(".git").join("config")).unwrap(),
            "real git config",
            "the real .git/config must be untouched by the traversal entry"
        );
        std::fs::remove_dir_all(&root).ok();
        std::fs::remove_dir_all(&project).ok();
    }

    #[test]
    fn restore_refuses_traversal_bypass_of_protected_globs_dot_env_stays_untouched() {
        // Same bypass shape as the `.git` floor, but against
        // `protected_globs` this time: `x/../.env` doesn't glob-match
        // `.env*` as a raw string, but normalizes right back into
        // `<root>/.env`.
        let root = tmp("traversal-env");
        let project = tmp("traversal-env-project");
        std::fs::create_dir_all(&project).unwrap();
        std::fs::write(project.join(".env"), "SECRET=1").unwrap();
        let store = CheckpointStore::open(&root).unwrap();
        let id = store.create_checkpoint("t").unwrap();
        let mut m = store.manifest(&id).unwrap();
        m.files.push(CheckpointFileEntry {
            path: "x/../.env".to_string(),
            blob: Some(store.write_blob(b"PWNED=1").unwrap()),
        });
        store.write_manifest(&m).unwrap();

        let report = store
            .restore(&id, &project, &[".env*".to_string()])
            .unwrap();
        assert_eq!(report.restored.len(), 0);
        assert_eq!(report.refused.len(), 1);
        assert_eq!(
            std::fs::read_to_string(project.join(".env")).unwrap(),
            "SECRET=1",
            "the real .env must be untouched by the traversal entry"
        );
        std::fs::remove_dir_all(&root).ok();
        std::fs::remove_dir_all(&project).ok();
    }

    #[test]
    fn restore_refuses_an_absolute_manifest_path() {
        // An absolute manifest entry is never something a legitimate
        // capture produces (`record_pre_image` always stores a clean
        // relative path) — even one that happens to point AT a path
        // inside the project (here, the real `.git/config`) must be
        // refused up front, not evaluated by whatever it happens to
        // resolve to. (Pre-fix: `Path::join` REPLACES the base when the
        // joined component is absolute, so this entry's raw string never
        // matched the `.git/` floor and its target — being genuinely
        // inside root — passed `contained` too: a second, independent
        // bypass of the same floor.)
        let root = tmp("absolute");
        let project = tmp("absolute-project");
        std::fs::create_dir_all(project.join(".git")).unwrap();
        std::fs::write(project.join(".git").join("config"), "real git config").unwrap();
        let store = CheckpointStore::open(&root).unwrap();
        let id = store.create_checkpoint("t").unwrap();
        let absolute_git_config = project.join(".git").join("config");
        let mut m = store.manifest(&id).unwrap();
        m.files.push(CheckpointFileEntry {
            path: absolute_git_config.to_string_lossy().to_string(),
            blob: Some(store.write_blob(b"PWNED-by-absolute-path").unwrap()),
        });
        store.write_manifest(&m).unwrap();

        let report = store.restore(&id, &project, &[]).unwrap();
        assert_eq!(report.restored.len(), 0);
        assert_eq!(report.refused.len(), 1);
        assert!(
            report.refused[0].1.contains("escapes"),
            "unexpected refusal reason: {}",
            report.refused[0].1
        );
        assert_eq!(
            std::fs::read_to_string(project.join(".git").join("config")).unwrap(),
            "real git config"
        );
        std::fs::remove_dir_all(&root).ok();
        std::fs::remove_dir_all(&project).ok();
    }

    #[test]
    fn restore_still_restores_a_clean_relative_nested_entry() {
        // Proves the fix didn't over-block honest captures: a normal
        // capture -> restore round trip — including a nested directory, so
        // the normalize step's `strip_prefix` is exercised on a
        // multi-component path — still restores.
        let root = tmp("cleanroundtrip");
        let project = tmp("cleanroundtrip-project");
        std::fs::create_dir_all(project.join("src")).unwrap();
        std::fs::write(project.join("src").join("main.rs"), "fn main() {}").unwrap();
        let store = CheckpointStore::open(&root).unwrap();
        let id = store.create_checkpoint("t").unwrap();
        store
            .record_pre_image(&id, "src/main.rs", Some(b"fn old() {}".to_vec()))
            .unwrap();

        let report = store.restore(&id, &project, &[]).unwrap();
        assert!(report.refused.is_empty(), "{:?}", report.refused);
        assert_eq!(report.restored, vec!["src/main.rs".to_string()]);
        assert_eq!(
            std::fs::read_to_string(project.join("src").join("main.rs")).unwrap(),
            "fn old() {}"
        );
        std::fs::remove_dir_all(&root).ok();
        std::fs::remove_dir_all(&project).ok();
    }

    #[test]
    #[cfg(unix)]
    fn restore_refuses_symlink_traversal_into_dot_git() {
        // The exact reviewer repro this unit fixes: `is_protected` ran on
        // the LEXICAL normalized path (never resolves symlinks) while
        // `contained` CANONICALIZES (resolves symlinks). A pre-existing
        // symlink `foo -> .git` in the working tree plus a manifest entry
        // `foo/config`:
        //  - passes `reject_unsafe_manifest_path` (no `..`/absolute);
        //  - lexically normalizes to `foo/config` — `is_protected` says
        //    NOT protected (no literal `.git/` prefix);
        //  - `contained` canonicalizes `foo`, resolving the symlink to the
        //    real `<root>/.git`, which IS inside `real_root` — containment
        //    PASSES.
        // Pre-fix, restore then wrote straight into the real `.git/config`.
        // Post-fix, `resolved_project_rel` also runs `is_protected` on the
        // symlink-RESOLVED path (`.git/config`), so this is refused.
        let root = tmp("symlink-traversal");
        let project = tmp("symlink-traversal-project");
        std::fs::create_dir_all(project.join(".git")).unwrap();
        std::fs::write(project.join(".git").join("config"), "real git config").unwrap();
        std::os::unix::fs::symlink(project.join(".git"), project.join("foo")).unwrap();

        let store = CheckpointStore::open(&root).unwrap();
        let id = store.create_checkpoint("t").unwrap();
        let mut m = store.manifest(&id).unwrap();
        m.files.push(CheckpointFileEntry {
            path: "foo/config".to_string(),
            blob: Some(store.write_blob(b"PWNED-VIA-SYMLINK").unwrap()),
        });
        store.write_manifest(&m).unwrap();

        let report = store.restore(&id, &project, &[]).unwrap();
        assert_eq!(
            report.restored.len(),
            0,
            "must not restore through the symlink into .git"
        );
        assert_eq!(report.refused.len(), 1);
        assert!(
            report.refused[0].1.contains("protected"),
            "unexpected refusal reason: {}",
            report.refused[0].1
        );
        assert_eq!(
            std::fs::read_to_string(project.join(".git").join("config")).unwrap(),
            "real git config",
            "the real .git/config must be byte-identical — untouched by the symlink entry"
        );
        std::fs::remove_dir_all(&root).ok();
        std::fs::remove_dir_all(&project).ok();
    }

    #[test]
    #[cfg(unix)]
    fn restore_refuses_symlink_traversal_into_a_protected_glob() {
        // Same shape as the `.git` floor above, but against a caller-
        // supplied `protected_globs` entry: a symlink `secrets -> real_env`
        // where `real_env/creds` is a file that, once resolved, matches
        // the `real_env/**` protected glob — even though the manifest's
        // lexical path (`secrets/creds`) does not.
        let root = tmp("symlink-glob");
        let project = tmp("symlink-glob-project");
        std::fs::create_dir_all(project.join("real_env")).unwrap();
        std::fs::write(project.join("real_env").join("creds"), "real secret").unwrap();
        std::os::unix::fs::symlink(project.join("real_env"), project.join("secrets")).unwrap();

        let store = CheckpointStore::open(&root).unwrap();
        let id = store.create_checkpoint("t").unwrap();
        let mut m = store.manifest(&id).unwrap();
        m.files.push(CheckpointFileEntry {
            path: "secrets/creds".to_string(),
            blob: Some(store.write_blob(b"PWNED-VIA-SYMLINK-GLOB").unwrap()),
        });
        store.write_manifest(&m).unwrap();

        let report = store
            .restore(&id, &project, &["real_env/**".to_string()])
            .unwrap();
        assert_eq!(report.restored.len(), 0);
        assert_eq!(report.refused.len(), 1);
        assert!(
            report.refused[0].1.contains("protected"),
            "unexpected refusal reason: {}",
            report.refused[0].1
        );
        assert_eq!(
            std::fs::read_to_string(project.join("real_env").join("creds")).unwrap(),
            "real secret",
            "the real protected file must be untouched by the symlink entry"
        );
        std::fs::remove_dir_all(&root).ok();
        std::fs::remove_dir_all(&project).ok();
    }

    #[test]
    fn prune_keeps_only_the_newest_and_gcs_unreferenced_blobs() {
        let root = tmp("prune");
        let store = CheckpointStore::open(&root).unwrap();
        for i in 0..5 {
            let id = store.create_checkpoint(&format!("turn {i}")).unwrap();
            store
                .record_pre_image(&id, "f.txt", Some(format!("content-{i}").into_bytes()))
                .unwrap();
            std::thread::sleep(std::time::Duration::from_millis(2));
        }
        assert_eq!(store.list().unwrap().len(), 5);
        let removed = store.prune(2).unwrap();
        assert_eq!(removed, 3);
        let remaining = store.list().unwrap();
        assert_eq!(remaining.len(), 2);
        // The two newest survive (turn 4, turn 3).
        assert_eq!(remaining[0].label, "turn 4");
        assert_eq!(remaining[1].label, "turn 3");
        // Every surviving blob is still readable; nothing else lingers.
        for meta in &remaining {
            let m = store.manifest(&meta.id).unwrap();
            for f in &m.files {
                if let Some(hash) = &f.blob {
                    store.read_blob(hash).unwrap();
                }
            }
        }
        std::fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn turn_diff_lists_exactly_the_captured_files() {
        let root = tmp("diff");
        let store = CheckpointStore::open(&root).unwrap();
        let id = store.create_checkpoint("t").unwrap();
        store.record_pre_image(&id, "a.rs", None).unwrap();
        store
            .record_pre_image(&id, "b.rs", Some(b"x".to_vec()))
            .unwrap();
        let mut diff = store.turn_diff(&id).unwrap();
        diff.sort();
        assert_eq!(diff, vec!["a.rs".to_string(), "b.rs".to_string()]);
        std::fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn manifest_rejects_a_path_traversing_id() {
        let root = tmp("badid");
        let store = CheckpointStore::open(&root).unwrap();
        assert!(store.manifest("../../etc/passwd").is_err());
        assert!(store.restore("../../etc/passwd", &root, &[]).is_err());
        std::fs::remove_dir_all(&root).ok();
    }

    #[tokio::test]
    async fn observer_before_write_ignores_paths_outside_the_project_root() {
        let root = tmp("obs-outside");
        let project = tmp("obs-outside-project");
        let outside = tmp("obs-outside-elsewhere");
        std::fs::write(outside.join("victim.txt"), "do not touch").unwrap();
        let store = CheckpointStore::open(&root).unwrap();
        let observer = CheckpointObserver::new(store, project.clone(), DEFAULT_RETAIN, vec![]);
        observer.begin_turn("t");
        observer.before_write(&outside.join("victim.txt")).await;
        // Nothing captured: the checkpoint's manifest stays empty.
        let id = observer.current().unwrap();
        let diff = observer.store().turn_diff(&id).unwrap();
        assert!(
            diff.is_empty(),
            "must not capture writes outside project_root"
        );
        std::fs::remove_dir_all(&root).ok();
        std::fs::remove_dir_all(&project).ok();
        std::fs::remove_dir_all(&outside).ok();
    }

    #[tokio::test]
    async fn observer_captures_only_the_first_write_to_a_path_in_a_turn() {
        let root = tmp("obs-firstwrite");
        let project = tmp("obs-firstwrite-project");
        std::fs::write(project.join("f.txt"), "v1").unwrap();
        let store = CheckpointStore::open(&root).unwrap();
        let observer = CheckpointObserver::new(store, project.clone(), DEFAULT_RETAIN, vec![]);
        observer.begin_turn("t");
        observer.before_write(&project.join("f.txt")).await;
        std::fs::write(project.join("f.txt"), "v2").unwrap();
        observer.before_write(&project.join("f.txt")).await; // second write, same turn
        let id = observer.current().unwrap();
        let m = observer.store().manifest(&id).unwrap();
        assert_eq!(m.files.len(), 1);
        let bytes = observer
            .store()
            .read_blob(m.files[0].blob.as_deref().unwrap())
            .unwrap();
        assert_eq!(bytes, b"v1", "must keep the EARLIEST pre-image, not v2");
        std::fs::remove_dir_all(&root).ok();
        std::fs::remove_dir_all(&project).ok();
    }

    #[tokio::test]
    async fn observer_begin_turn_clears_captured_set_for_a_new_turn() {
        let root = tmp("obs-newturn");
        let project = tmp("obs-newturn-project");
        std::fs::write(project.join("f.txt"), "v1").unwrap();
        let store = CheckpointStore::open(&root).unwrap();
        let observer = CheckpointObserver::new(store, project.clone(), DEFAULT_RETAIN, vec![]);
        observer.begin_turn("turn 1");
        observer.before_write(&project.join("f.txt")).await;
        std::fs::write(project.join("f.txt"), "v2").unwrap();
        observer.begin_turn("turn 2");
        observer.before_write(&project.join("f.txt")).await;
        let id2 = observer.current().unwrap();
        let m2 = observer.store().manifest(&id2).unwrap();
        assert_eq!(m2.files.len(), 1);
        let bytes = observer
            .store()
            .read_blob(m2.files[0].blob.as_deref().unwrap())
            .unwrap();
        assert_eq!(
            bytes, b"v2",
            "turn 2's checkpoint must capture v2 as ITS pre-image"
        );
        std::fs::remove_dir_all(&root).ok();
        std::fs::remove_dir_all(&project).ok();
    }

    #[test]
    fn is_protected_hard_floor_covers_dot_git_regardless_of_extra_globs() {
        assert!(is_protected(".git", &[]));
        assert!(is_protected(".git/config", &[]));
        assert!(is_protected(".git/objects/aa/bb", &[]));
        assert!(!is_protected(".gitignore", &[]));
        assert!(!is_protected("src/main.rs", &[]));
    }

    #[test]
    fn contained_rejects_symlink_escape_for_an_existing_target() {
        let project = tmp("symlink-project");
        let outside = tmp("symlink-outside");
        std::fs::write(outside.join("secret.txt"), "s").unwrap();
        #[cfg(unix)]
        {
            std::os::unix::fs::symlink(outside.join("secret.txt"), project.join("link.txt"))
                .unwrap();
            assert!(!contained(&project, &project.join("link.txt")));
        }
        std::fs::remove_dir_all(&project).ok();
        std::fs::remove_dir_all(&outside).ok();
    }

    #[test]
    fn contained_accepts_a_brand_new_file_inside_the_root() {
        let project = tmp("newfile-project");
        assert!(contained(&project, &project.join("does_not_exist_yet.txt")));
        assert!(contained(&project, &project.join("nested/dir/new.txt")));
        std::fs::remove_dir_all(&project).ok();
    }
}