agentstategraph 1.4.0

//! Repository — the high-level API for AgentStateGraph.
//!
//! A Repository wraps a Storage backend and provides the primary
//! user-facing operations: get, set, delete, branch, merge, log.
//!
//! Every write operation is an atomic commit with intent metadata.
//! There is no staging area.

use std::sync::Arc;

use agentstategraph_core::{
    Authority, Commit, CommitBuilder, Conflict, DiffOp, Intent, IntentCategory, MergeResult,
    Namespace, Object, ObjectId, ObjectResolver, StatePath, ToolCall,
};
use agentstategraph_storage::{
    CommitTrace, CommitWalk, HistoryMilestoneRow, HistoryRollupRow, Storage, StorageError,
    StoreTotals,
};

use crate::speculation::{SpecComparison, SpecError, SpecHandle, SpeculationManager};
use crate::telemetry::{self, op_span, write_span};
use crate::tree::{self, TreeError};

/// Reserved state path prefix for schema metadata. Writes under this
/// prefix are only permitted on commits tagged `IntentCategory::Migrate`.
pub const META_PATH_PREFIX: &str = "/_meta";

/// Reserved sub-prefix for secret-bearing metadata. BOTH reads AND writes
/// under `/_meta/_secret/*` are gated to `IntentCategory::Migrate`. The
/// broader `/_meta/*` prefix only gates writes — the secret sub-prefix
/// tightens this for values that must not be surfaced by casual
/// `get`/`list_paths`/`search_values` callers.
///
/// See `spec/UPGRADE-PATH.md`.
pub const META_SECRET_PREFIX: &str = "/_meta/_secret";

/// Path to the schema version sentinel written by `Repository::init()`
/// and bumped by migrations in the `agentstategraph-migrate` crate.
pub const META_SCHEMA_VERSION_PATH: &str = "/_meta/schema_version";

/// Schema version stamped into new repositories by `init()`.
///
/// This is deliberately **decoupled from the crate version** — it tracks
/// the last version at which the on-disk shape changed, not the release
/// tag of the binary. A 0.4.0-beta.3 binary stamping `"0.4.0"` here
/// reflects that the schema is compatible with any 0.4.x binary. Bump
/// this constant only when you ship a migration that advances the DB
/// shape. See `spec/UPGRADE-PATH.md` decision 5.
pub const SCHEMA_VERSION: &str = "0.4.0";

fn path_is_reserved(path: &str) -> bool {
    path == META_PATH_PREFIX || path.starts_with(&format!("{}/", META_PATH_PREFIX))
}

/// True iff `path` is inside the `/_meta/_secret` sub-tree (or is the
/// prefix itself). Used to gate READS on secret metadata.
fn path_is_secret(path: &str) -> bool {
    path == META_SECRET_PREFIX || path.starts_with(&format!("{}/", META_SECRET_PREFIX))
}

/// Returns `true` when the intent category represents a taint /
/// quarantine / watch lifecycle event. These commits bypass the
/// pre-commit taint hook because they are the mechanism by which
/// taints are created and resolved — gating them on themselves
/// would deadlock the substrate.
fn is_taint_lifecycle_intent(category: &IntentCategory) -> bool {
    matches!(
        category,
        IntentCategory::Taint
            | IntentCategory::Untaint
            | IntentCategory::Quarantine
            | IntentCategory::Unquarantine
            | IntentCategory::Watch
            | IntentCategory::Unwatch
    )
}

fn check_meta_guard(path: &str, intent: &Intent) -> Result<(), RepoError> {
    if path_is_reserved(path) && intent.category != IntentCategory::Migrate {
        return Err(RepoError::ReservedPath(path.to_string()));
    }
    Ok(())
}

/// Gate reads of `/_meta/_secret/*` unless the caller's intent is
/// `IntentCategory::Migrate`. Applied to every read surface
/// (`get`, `get_json`, `list_paths`, `search_values`).
fn check_secret_read_guard(path: &str, intent: &Intent) -> Result<(), RepoError> {
    if path_is_secret(path) && intent.category != IntentCategory::Migrate {
        return Err(RepoError::ReservedPath(path.to_string()));
    }
    Ok(())
}

fn refuse_deleting_main(name: &str) -> Result<(), RepoError> {
    if name == "main" {
        return Err(RepoError::InvalidOperation(
            "refusing to delete 'main': every branch forks from it, and the next \
             init() would replace it with an empty one"
                .to_string(),
        ));
    }
    Ok(())
}

/// Walk a diff and return the first path under `/_meta/*` touched, if any.
/// Used to enforce the meta guard on speculation commits.
fn reserved_path_in_diff(diff: &[DiffOp]) -> Option<String> {
    for op in diff {
        let candidate = match op {
            DiffOp::SetValue { path, .. } => path.clone(),
            DiffOp::AddKey { path, key, .. } | DiffOp::RemoveKey { path, key, .. } => {
                if path == "/" || path.is_empty() {
                    format!("/{}", key)
                } else {
                    format!("{}/{}", path, key)
                }
            }
            DiffOp::AddElement { path, .. }
            | DiffOp::RemoveElement { path, .. }
            | DiffOp::AddToSet { path, .. }
            | DiffOp::RemoveFromSet { path, .. }
            | DiffOp::ChangeType { path, .. } => path.clone(),
        };
        if path_is_reserved(&candidate) {
            return Some(candidate);
        }
    }
    None
}

/// Environment variable that switches epoch seal enforcement from warn
/// (default) to strict (reject ref updates that would orphan sealed commits).
/// (security threat model v3+, V8)
pub const EPOCH_SEAL_STRICT_ENV: &str = "ASG_EPOCH_SEAL_STRICT";

/// A record of a sealed commit that would become unreachable from a
/// proposed new ref target.
#[derive(Debug, Clone)]
pub struct EpochViolation {
    pub epoch_id: String,
    pub unreachable_commits: Vec<ObjectId>,
}

/// The primary API for interacting with an AgentStateGraph state store.
pub struct Repository {
    storage: Arc<dyn Storage + Send + Sync>,
    specs: SpeculationManager,
    /// Unique per `Repository` value. Open speculations pin their roots in the
    /// store under keys built from it, so a GC sweep in any process keeps
    /// their not-yet-committed objects (see `pin_speculation`).
    instance: String,
    watch_mgr: crate::watch::WatchManager,
    /// Configured namespace for this repository instance. The active session's
    /// `scope_namespace` takes priority when resolving the effective namespace
    /// (see `active_namespace()`). Defaults to `Namespace::default_ns()`.
    namespace: Namespace,
    /// Active epoch id — if set, all new commits are associated with it
    /// via `storage.set_commit_epoch` on commit finalization. Set via
    /// `set_active_epoch` / cleared via `clear_active_epoch`. Not a
    /// public MCP tool yet — that's a follow-up milestone.
    active_epoch: std::sync::RwLock<Option<String>>,
    /// Active session id — same semantics as `active_epoch`.
    active_session: std::sync::RwLock<Option<String>>,
    /// When true (**the default**), ref updates that would orphan a sealed
    /// commit are rejected with `RepoError::EpochSealViolated` — the guard
    /// Plan B's GC relies on so sealed history can't be silently dropped. Opt
    /// OUT to the legacy warn-and-proceed behavior via
    /// `Repository::with_epoch_seal_strict(false)` or a falsey
    /// `ASG_EPOCH_SEAL_STRICT` (`0`/`false`/`warn`/`off`).
    epoch_seal_strict: bool,
    /// Cached namespace override from the active session's scope_namespace.
    /// Populated eagerly in `set_active_session`; cleared when session is None.
    /// Avoids a storage round-trip on every ref operation.
    active_session_namespace: std::sync::RwLock<Option<Namespace>>,
    /// Accept only branch names as ref specs. See
    /// [`Repository::with_branch_refs_only`].
    branch_refs_only: bool,
}

/// Default batch size for [`Repository::extract_history`] — how many commits
/// are folded per transaction. Large enough to amortize the per-batch overhead,
/// small enough to keep peak memory bounded on a 512k-commit store.
const DEFAULT_HISTORY_BATCH: usize = 5000;

/// How many times a ref write rebuilds on a moved head before giving up with
/// `RepoError::WriteConflict`. Each failed compare-and-swap means another
/// writer's commit landed, so the store as a whole always makes progress; the
/// bound only stops one writer starving forever under sustained contention.
const MAX_REF_CAS_ATTEMPTS: usize = 256;

/// Frontier batch size for the GC reachability mark (Plan B t-001) — how many
/// unexpanded nodes are drained per transaction. Bounds peak memory to roughly
/// this many object blobs regardless of store size.
const GC_MARK_BATCH: usize = 10_000;

/// How long an open speculation's GC pin stays honoured without a refresh.
/// Matches the storage layer's own re-read of pins under the sweep lock.
const GC_PIN_MAX_AGE: std::time::Duration = std::time::Duration::from_secs(24 * 60 * 60);

/// Map a `YYYY-MM-DD` day to its ISO-week key `YYYY-Www`, or `None` if the day
/// doesn't parse. Used to roll daily velocity up to weekly.
fn iso_week_key(day: &str) -> Option<String> {
    use chrono::Datelike;
    let d = chrono::NaiveDate::parse_from_str(day, "%Y-%m-%d").ok()?;
    let wk = d.iso_week();
    Some(format!("{}-W{:02}", wk.year(), wk.week()))
}

/// GC retention policy (Plan B t-003): which historical commits' state stays
/// materializable. Ref tips, sealed-epoch commits, and milestones are always
/// kept regardless of this policy.
#[derive(Debug, Clone, Copy)]
pub struct RetentionPolicy {
    /// Keep the `keep_recent` most-recent commits' state in full.
    pub keep_recent: usize,
    /// Among older commits, keep every `checkpoint_every`-th as a sparse
    /// checkpoint (0 disables checkpointing — only recent + always-kept remain).
    pub checkpoint_every: usize,
    /// Keep the milestone snapshots (the human-meaningful spine).
    pub keep_milestones: bool,
}

impl Default for RetentionPolicy {
    fn default() -> Self {
        Self {
            keep_recent: 100,
            checkpoint_every: 100,
            keep_milestones: true,
        }
    }
}

/// The store's GC write lock, held by a mutating sweep from its keep-set to its
/// last delete (see `CommitStore::history_gc_lock_writers`).
///
/// Rolls back on drop unless explicitly released, so an early `?` return or a
/// panic can never leave every other writer locked out of the store.
struct GcWriterLock<'a> {
    storage: &'a (dyn Storage + Send + Sync),
    released: bool,
}

impl<'a> GcWriterLock<'a> {
    fn acquire(storage: &'a (dyn Storage + Send + Sync)) -> Result<Self, RepoError> {
        storage.history_gc_lock_writers()?;
        Ok(Self {
            storage,
            released: false,
        })
    }

    fn release(mut self, commit: bool) -> Result<(), RepoError> {
        self.released = true;
        Ok(self.storage.history_gc_unlock_writers(commit)?)
    }
}

impl Drop for GcWriterLock<'_> {
    fn drop(&mut self) {
        if !self.released {
            let _ = self.storage.history_gc_unlock_writers(false);
        }
    }
}

/// Outcome of a [`Repository::extract_history`] run.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HistoryExtractReport {
    /// Commits folded into the history tables this run (0 = already current).
    pub commits_processed: usize,
    /// The `commits.rowid` cursor after the run.
    pub cursor: i64,
}

/// Options for creating a commit.
///
/// `Clone` so a write that loses a compare-and-swap race can rebuild its
/// commit on the new head with the same provenance.
#[derive(Clone)]
pub struct CommitOptions {
    pub agent_id: String,
    pub authority: Authority,
    pub intent: Intent,
    pub reasoning: Option<String>,
    pub confidence: Option<f64>,
    /// Tool calls that contributed to this change. Persisted on the commit as
    /// provenance; empty by default. Capped at [`Commit`]'s tool-call limit on
    /// build (excess is truncated).
    pub tool_calls: Vec<ToolCall>,
}

impl CommitOptions {
    /// Create minimal commit options — the simplest way to commit.
    ///
    /// The authorizing principal defaults to `agent_id`: absent delegation, the
    /// actor making the change is also the authorizer (Plan C t-002 — minimal
    /// authority capture). Previously every commit hardcoded a constant
    /// `"default"` principal, so the authority field carried no real
    /// provenance. Override with [`CommitOptions::with_principal`] when the
    /// authorizer differs from the actor, or [`CommitOptions::with_authority`]
    /// for a full scoped/delegated `Authority` (t-004).
    pub fn new(
        agent_id: impl Into<String>,
        intent_category: IntentCategory,
        description: impl Into<String>,
    ) -> Self {
        let agent_id = agent_id.into();
        Self {
            authority: Authority::simple(agent_id.clone()),
            agent_id,
            intent: Intent::new(intent_category, description),
            reasoning: None,
            confidence: None,
            tool_calls: Vec::new(),
        }
    }

    /// Set the full authority (scope, delegation chain, expiry). For the common
    /// case of just naming the authorizing principal, prefer
    /// [`CommitOptions::with_principal`].
    pub fn with_authority(mut self, authority: Authority) -> Self {
        self.authority = authority;
        self
    }

    /// Set the authorizing principal (minimal authority capture: principal +
    /// wildcard scope, no delegation). Use when the authorizer differs from the
    /// actor `agent_id`; for a full delegated `Authority`, use
    /// [`CommitOptions::with_authority`].
    pub fn with_principal(mut self, principal: impl Into<String>) -> Self {
        self.authority = Authority::simple(principal);
        self
    }

    /// Set reasoning.
    pub fn with_reasoning(mut self, reasoning: impl Into<String>) -> Self {
        self.reasoning = Some(reasoning.into());
        self
    }

    /// Set confidence.
    pub fn with_confidence(mut self, confidence: f64) -> Self {
        self.confidence = Some(confidence);
        self
    }

    /// Set the tool calls that contributed to this change.
    pub fn with_tool_calls(mut self, tool_calls: Vec<ToolCall>) -> Self {
        self.tool_calls = tool_calls;
        self
    }

    /// Set tags on the intent.
    pub fn with_tags(mut self, tags: Vec<String>) -> Self {
        self.intent.tags = tags;
        self
    }
}

/// Errors from Repository operations.
#[derive(Debug, thiserror::Error)]
pub enum RepoError {
    #[error("branch not found: {0}")]
    BranchNotFound(String),

    #[error("branch already exists: {0}")]
    BranchAlreadyExists(String),

    #[error("ref not found: {0}")]
    RefNotFound(String),

    #[error("commit not found: {0}")]
    CommitNotFound(String),

    #[error("ambiguous commit prefix: {prefix} matched {count} commits")]
    AmbiguousCommitPrefix { prefix: String, count: usize },

    #[error("repository not initialized — call init() first")]
    NotInitialized,

    #[error(
        "path {0} is reserved for schema metadata; only IntentCategory::Migrate commits may write here"
    )]
    ReservedPath(String),

    #[error("merge conflicts: {0:?}")]
    MergeConflicts(Vec<Conflict>),

    #[error("merge would delete top-level entries {0:?}; pass allow_deletions to proceed")]
    MergeWouldDelete(Vec<String>),

    #[error(
        "integrity violation: object {missing} reachable from state root {root} is missing from the store; ref not advanced"
    )]
    IntegrityViolation { root: ObjectId, missing: ObjectId },

    #[error("write conflict: ref moved before CAS could land")]
    WriteConflict,

    #[error("speculation error: {0}")]
    Speculation(#[from] SpecError),

    #[error("tree error: {0}")]
    Tree(#[from] TreeError),

    #[error("storage error: {0}")]
    Storage(#[from] StorageError),

    #[error(
        "epoch seal violated: epoch '{epoch_id}' sealed commit(s) {unreachable_commits:?} would be orphaned by this ref update"
    )]
    EpochSealViolated {
        epoch_id: String,
        unreachable_commits: Vec<ObjectId>,
    },

    /// Pre-commit taint hook (0.7.75 §4) rejected a write. `taint_id`
    /// is the storage id of the taint that caused the rejection — use
    /// `agentstategraph_policy` / MCP `check_taint` to surface the
    /// full context.
    ///
    /// `taint_id` is `Some` for all blocking rejections (NotAuthorized,
    /// Blocked, InsufficientConfidence) and `None` only for lookup
    /// failures (NotFound) where no taint id exists to report.
    #[error("invalid operation: {0}")]
    InvalidOperation(String),

    /// A commit id or prefix where this repository accepts only branch names
    /// (see [`Repository::with_branch_refs_only`]).
    #[error("'{0}' looks like a commit id; only branch names are accepted here")]
    CommitRefNotAllowed(String),

    #[error("namespace not found: '{0}' — call create_namespace first")]
    NamespaceNotFound(String),

    #[error("cross-namespace merge denied: no PolicyStore configured")]
    CrossNamespaceAccessDenied,

    #[error("taint hook rejected write{}: {source}", match .taint_id {
        Some(id) => format!(" (taint: {id})"),
        None => String::new(),
    })]
    Taint {
        #[source]
        source: agentstategraph_taint::TaintError,
        taint_id: Option<String>,
    },
}

impl RepoError {
    /// A stable, low-cardinality code for this kind of error (DESIGN.md Plan
    /// H t-004), such as `write_conflict` or `branch_not_found`. The HTTP API
    /// returns it beside the message and picks the status from it, MCP tool
    /// errors carry it, and it becomes `error.type` on spans and metrics.
    /// Clients can branch on it; messages may change.
    pub fn code(&self) -> &'static str {
        match self {
            RepoError::BranchNotFound(_) => "branch_not_found",
            RepoError::BranchAlreadyExists(_) => "branch_already_exists",
            RepoError::RefNotFound(_) => "ref_not_found",
            RepoError::CommitNotFound(_) => "commit_not_found",
            RepoError::AmbiguousCommitPrefix { .. } => "ambiguous_commit_prefix",
            RepoError::NotInitialized => "not_initialized",
            RepoError::ReservedPath(_) => "reserved_path",
            RepoError::MergeConflicts(_) => "merge_conflict",
            RepoError::MergeWouldDelete(_) => "merge_would_delete",
            RepoError::IntegrityViolation { .. } => "integrity_violation",
            RepoError::WriteConflict => "write_conflict",
            RepoError::Speculation(e) => e.code(),
            RepoError::Tree(e) => e.code(),
            RepoError::Storage(e) => e.code(),
            RepoError::EpochSealViolated { .. } => "epoch_sealed",
            RepoError::InvalidOperation(_) => "invalid_operation",
            RepoError::CommitRefNotAllowed(_) => "commit_ref_not_allowed",
            RepoError::NamespaceNotFound(_) => "namespace_not_found",
            RepoError::CrossNamespaceAccessDenied => "cross_namespace_denied",
            RepoError::Taint { .. } => "taint_blocked",
        }
    }
}

/// Summary of what merging one ref into another would do, produced by
/// [`Repository::preview_merge`] without mutating any ref. All key lists are
/// top-level entries of the state root (e.g. `plans`, `memory`).
#[derive(Debug, Clone)]
pub struct MergePreview {
    /// True if the merge resolves to a fast-forward.
    pub fast_forward: bool,
    /// Top-level keys present after the merge but not before.
    pub added: Vec<String>,
    /// Top-level keys whose subtree id changes.
    pub changed: Vec<String>,
    /// Top-level keys that would be removed (data-loss surface).
    pub removed: Vec<String>,
    /// Conflicts that block a clean merge, if any.
    pub conflicts: Vec<Conflict>,
}

impl MergePreview {
    /// Whether committing this merge would remove any top-level entry.
    pub fn has_deletions(&self) -> bool {
        !self.removed.is_empty()
    }
}

/// Intermediate result of a merge computed but not yet committed.
struct MergeComputation {
    source_commit_id: ObjectId,
    target_commit_id: ObjectId,
    source_state_root: ObjectId,
    target_state_root: ObjectId,
    result: MergeResult,
    created: Vec<Object>,
}

enum MergeResultKind {
    Success,
    Conflicts,
    FastForward,
}

/// Top-level map entries (key -> child id) of an object, or empty if it is not
/// a map node.
fn map_entries(obj: &Object) -> std::collections::BTreeMap<String, ObjectId> {
    match obj {
        Object::Node(agentstategraph_core::Node::Map(entries)) => entries.clone(),
        _ => std::collections::BTreeMap::new(),
    }
}

impl From<agentstategraph_taint::TaintError> for RepoError {
    /// Converts a bare TaintError with no associated taint id.
    /// Used only for lookup-not-found errors; blocking rejections always
    /// go through `taint_err_with_id` in taint.rs which sets `Some(id)`.
    fn from(source: agentstategraph_taint::TaintError) -> Self {
        RepoError::Taint {
            source,
            taint_id: None,
        }
    }
}

/// How far back [`Repository::query_commits_paged`] walks before filtering,
/// when the requested page is shallower than this.
///
/// Filtering is done in memory, so the walk has to be bounded somewhere. This
/// was previously a bare `10000` inside the function body, which made it look
/// like an implementation detail rather than the correctness limit it is: a
/// store with more matching commits than this cannot page past it.
pub const COMMIT_QUERY_SCAN_FLOOR: usize = 10_000;

/// How far [`Repository::query_commits_paged`] walks for a given page.
///
/// Pulled out as a function so it can be tested directly: asserting the
/// formula inside the paged query would need a store of >10k commits, which
/// no unit test is going to build.
pub fn commit_query_scan(offset: usize, limit: usize) -> usize {
    COMMIT_QUERY_SCAN_FLOOR.max(offset.saturating_add(limit))
}

impl Repository {
    /// Create a new Repository wrapping the given storage backend.
    ///
    /// Epoch-seal enforcement is **strict by default**: a ref update that would
    /// orphan a sealed commit is rejected. Opt out to the legacy
    /// warn-and-proceed behavior by setting `ASG_EPOCH_SEAL_STRICT` to a falsey
    /// value (`0`/`false`/`warn`/`off`) at construction time, or programmatically
    /// via [`Repository::with_epoch_seal_strict`]`(false)`.
    pub fn new(storage: Box<dyn Storage + Send + Sync>) -> Self {
        Self::from_shared(Arc::from(storage))
    }

    /// Like [`Repository::new`], over storage the caller keeps a handle to —
    /// for a backend with state of its own to drive, such as the queue of
    /// writes the WASM binding flushes to IndexedDB.
    pub fn from_shared(storage: Arc<dyn Storage + Send + Sync>) -> Self {
        // Strict unless explicitly disabled. An unset or unrecognized value
        // keeps the safe default; only a clearly falsey value opts out.
        let strict = std::env::var(EPOCH_SEAL_STRICT_ENV)
            .map(|v| {
                !(v == "0"
                    || v.eq_ignore_ascii_case("false")
                    || v.eq_ignore_ascii_case("warn")
                    || v.eq_ignore_ascii_case("off"))
            })
            .unwrap_or(true);
        Self {
            storage,
            specs: SpeculationManager::new(),
            instance: uuid::Uuid::new_v4().simple().to_string(),
            watch_mgr: crate::watch::WatchManager::new(),
            namespace: Namespace::default_ns(),
            active_session_namespace: std::sync::RwLock::new(None),
            active_epoch: std::sync::RwLock::new(None),
            active_session: std::sync::RwLock::new(None),
            epoch_seal_strict: strict,
            branch_refs_only: false,
        }
    }

    /// Return a Repository configured to use the given namespace for all ref
    /// operations. The active session's `scope_namespace` overrides this value
    /// when a session is active. Callers can chain this with other builders:
    /// `Repository::new(storage).with_namespace(ns).with_epoch_seal_strict(true)`.
    pub fn with_namespace(mut self, ns: Namespace) -> Self {
        self.namespace = ns;
        self
    }

    /// Create a new `Repository` that operates in a different namespace but
    /// shares the same underlying storage.  Used for per-call namespace
    /// overrides (MCP tool `namespace` param, WASM `namespace` argument)
    /// without mutating the server-wide repository configuration.
    ///
    /// The forked repository inherits `epoch_seal_strict` and the active
    /// epoch/session ids from the parent; speculation state and watch
    /// subscriptions start fresh (they are in-memory and call-scoped).
    pub fn fork_namespace(&self, ns: Namespace) -> Self {
        let epoch = self.active_epoch.read().unwrap().clone();
        let session = self.active_session.read().unwrap().clone();
        let session_ns = self.active_session_namespace.read().unwrap().clone();
        Self {
            storage: Arc::clone(&self.storage),
            specs: crate::speculation::SpeculationManager::new(),
            instance: uuid::Uuid::new_v4().simple().to_string(),
            watch_mgr: crate::watch::WatchManager::new(),
            namespace: ns,
            active_epoch: std::sync::RwLock::new(epoch),
            active_session: std::sync::RwLock::new(session),
            epoch_seal_strict: self.epoch_seal_strict,
            branch_refs_only: self.branch_refs_only,
            active_session_namespace: std::sync::RwLock::new(session_ns),
        }
    }

    /// The namespace this repository was configured with. The *effective*
    /// namespace for a ref operation may differ when a session with its own
    /// `scope_namespace` is active — use `active_namespace()` for that.
    pub fn namespace(&self) -> &Namespace {
        &self.namespace
    }

    /// Accept only branch names where a ref is named, refusing commit ids and
    /// id prefixes ([`RepoError::CommitRefNotAllowed`]).
    ///
    /// Branch names resolve in this repository's namespace, but commit ids
    /// resolve over every commit in the store, whichever namespace made it.
    /// A repository that serves one namespace to a caller who mustn't see
    /// the others (an API key bound to a namespace) is limited this way.
    /// Forks keep the limit.
    pub fn with_branch_refs_only(mut self) -> Self {
        self.branch_refs_only = true;
        self
    }

    /// Return a Repository with epoch-seal enforcement set to the given mode.
    /// Overrides the `ASG_EPOCH_SEAL_STRICT` environment variable.
    ///
    /// Strict mode (the default) rejects ref updates that would render any
    /// sealed-epoch commit unreachable from the new target. Pass `false` for
    /// the legacy warn-and-proceed behavior. (security threat model v3+, V8)
    pub fn with_epoch_seal_strict(mut self, strict: bool) -> Self {
        self.epoch_seal_strict = strict;
        self
    }

    /// Initialize the repository with an empty state tree on "main".
    /// If "main" already exists, this is a no-op that returns its head.
    ///
    /// Safe to call from several processes at once: `main` is created only if
    /// absent, so every caller ends up with the same `main`. Setting it
    /// unconditionally let a late initializer replace an earlier one's `main`
    /// — and every commit already made on it.
    ///
    /// The initial commit stamps `/_meta/schema_version` with the crate
    /// version. See `spec/UPGRADE-PATH.md`.
    pub fn init(&self) -> Result<ObjectId, RepoError> {
        let ns = self.active_namespace()?;
        // Ensure the namespace row exists. init() is idempotent so creating here
        // is safe; create_namespace absorbs NamespaceAlreadyExists.
        self.create_namespace(ns.as_str())?;
        if let Some(id) = self.storage.get_ref(&ns, "main")? {
            return Ok(id);
        }

        let empty_root = Object::empty_map();
        let empty_root_id = self.storage.put_object(&empty_root)?;

        let version_path = StatePath::parse(META_SCHEMA_VERSION_PATH)
            .map_err(|e| TreeError::PathNotFound(e.to_string()))?;
        let version_value = Object::Atom(agentstategraph_core::Atom::String(
            SCHEMA_VERSION.to_string(),
        ));
        let stamped_root_id = tree::tree_set(
            self.storage.as_ref(),
            &empty_root_id,
            &version_path,
            &version_value,
        )?;

        let commit = CommitBuilder::new(
            stamped_root_id,
            "system",
            Authority::simple("system"),
            Intent::new(IntentCategory::Checkpoint, "Initialize empty state"),
        )
        .build();

        self.storage.put_commit(&commit)?;
        if self.storage.create_ref(&ns, "main", commit.id)? {
            return Ok(commit.id);
        }
        // Another initializer got there first; theirs is `main`.
        self.storage
            .get_ref(&ns, "main")?
            .ok_or(RepoError::NotInitialized)
    }

    // -----------------------------------------------------------------------
    // State operations
    // -----------------------------------------------------------------------

    /// Get a value from state at the given ref and path.
    ///
    /// Reads under `/_meta/_secret/*` are rejected — use
    /// [`Repository::get_with_intent`] with an
    /// `IntentCategory::Migrate` intent to access them.
    pub fn get(&self, ref_name: &str, path: &str) -> Result<Object, RepoError> {
        let span = op_span!(DEBUG, "asg.repo.get", asg.ref.name = ref_name);
        telemetry::run(span, || self.get_inner(ref_name, path))
    }

    fn get_inner(&self, ref_name: &str, path: &str) -> Result<Object, RepoError> {
        if path_is_secret(path) {
            return Err(RepoError::ReservedPath(path.to_string()));
        }
        let commit_id = self.resolve_ref(ref_name)?;
        let commit = self
            .storage
            .get_commit(&commit_id)?
            .ok_or_else(|| RepoError::RefNotFound(ref_name.to_string()))?;

        let state_path =
            StatePath::parse(path).map_err(|e| TreeError::PathNotFound(e.to_string()))?;
        let obj = tree::tree_get(self.storage.as_ref(), &commit.state_root, &state_path)?;
        Ok(obj)
    }

    /// Get a value as JSON.
    ///
    /// Reads under `/_meta/_secret/*` are rejected — use
    /// [`Repository::get_json_with_intent`] with
    /// `IntentCategory::Migrate` to access them.
    pub fn get_json(&self, ref_name: &str, path: &str) -> Result<serde_json::Value, RepoError> {
        let obj = self.get(ref_name, path)?;
        let json = tree::tree_to_json(self.storage.as_ref(), &obj)?;
        Ok(json)
    }

    /// [`Repository::get_json`] for several paths of one ref at once. The ref
    /// is resolved once, each level of every path is fetched in one batch, and
    /// the subtrees below them are read together a level at a time, so a
    /// remote store answers many paths in about as many round trips as one.
    ///
    /// Returns a value per path, in order: `None` where the path doesn't exist
    /// (a missing key or index, or a path through an atom or a container of
    /// the wrong kind). Reads under `/_meta/_secret/*` are rejected, as by
    /// `get_json`.
    pub fn get_json_many(
        &self,
        ref_name: &str,
        paths: &[&str],
    ) -> Result<Vec<Option<serde_json::Value>>, RepoError> {
        let span = op_span!(DEBUG, "asg.repo.get_many", asg.ref.name = ref_name);
        telemetry::run(span, || {
            if let Some(secret) = paths.iter().find(|p| path_is_secret(p)) {
                return Err(RepoError::ReservedPath(secret.to_string()));
            }
            let parsed = paths
                .iter()
                .map(|p| StatePath::parse(p).map_err(|e| TreeError::PathNotFound(e.to_string())))
                .collect::<Result<Vec<_>, _>>()?;
            let commit_id = self.resolve_ref(ref_name)?;
            let commit = self
                .storage
                .get_commit(&commit_id)?
                .ok_or_else(|| RepoError::RefNotFound(ref_name.to_string()))?;
            let objs = tree::tree_get_many(self.storage.as_ref(), &commit.state_root, &parsed)?;
            Ok(tree::trees_to_json(self.storage.as_ref(), &objs)?)
        })
    }

    /// Like [`Repository::get_json`] but stops materializing `max_depth` levels
    /// below `path`: nodes at the cap become `{ "_truncated": true, ... }`
    /// placeholders and their subtrees are never loaded. A cheap shallow read of
    /// a large ref — avoids walking and serializing the whole tree (plan t-007).
    pub fn get_json_capped(
        &self,
        ref_name: &str,
        path: &str,
        max_depth: usize,
    ) -> Result<serde_json::Value, RepoError> {
        let obj = self.get(ref_name, path)?;
        let json = tree::tree_to_json_capped(self.storage.as_ref(), &obj, max_depth)?;
        Ok(json)
    }

    /// Get a value with an explicit intent, permitting reads of
    /// `/_meta/_secret/*` when the intent category is `Migrate`.
    pub fn get_with_intent(
        &self,
        ref_name: &str,
        path: &str,
        intent: &Intent,
    ) -> Result<Object, RepoError> {
        let span = op_span!(DEBUG, "asg.repo.get", asg.ref.name = ref_name);
        telemetry::run(span, || self.get_with_intent_inner(ref_name, path, intent))
    }

    fn get_with_intent_inner(
        &self,
        ref_name: &str,
        path: &str,
        intent: &Intent,
    ) -> Result<Object, RepoError> {
        check_secret_read_guard(path, intent)?;
        let commit_id = self.resolve_ref(ref_name)?;
        let commit = self
            .storage
            .get_commit(&commit_id)?
            .ok_or_else(|| RepoError::RefNotFound(ref_name.to_string()))?;
        let state_path =
            StatePath::parse(path).map_err(|e| TreeError::PathNotFound(e.to_string()))?;
        let obj = tree::tree_get(self.storage.as_ref(), &commit.state_root, &state_path)?;
        Ok(obj)
    }

    /// Like [`Repository::get_json`] but honors an explicit intent for
    /// reading the `/_meta/_secret/*` sub-tree.
    pub fn get_json_with_intent(
        &self,
        ref_name: &str,
        path: &str,
        intent: &Intent,
    ) -> Result<serde_json::Value, RepoError> {
        let obj = self.get_with_intent(ref_name, path, intent)?;
        let json = tree::tree_to_json(self.storage.as_ref(), &obj)?;
        Ok(json)
    }

    /// Set a value in state, creating a new commit.
    /// Returns the new commit ID.
    pub fn set(
        &self,
        ref_name: &str,
        path: &str,
        value: &Object,
        options: CommitOptions,
    ) -> Result<ObjectId, RepoError> {
        telemetry::write(
            "set",
            write_span!("asg.repo.set", ref_name, options),
            || self.set_inner(ref_name, path, value, options),
        )
    }

    fn set_inner(
        &self,
        ref_name: &str,
        path: &str,
        value: &Object,
        options: CommitOptions,
    ) -> Result<ObjectId, RepoError> {
        check_meta_guard(path, &options.intent)?;
        // 0.7.75 §4: taint pre-commit hook. Rejects writes to blocked
        // or quarantined paths; review-effect gate enforces the
        // confidence threshold. Taint/Untaint/Quarantine/... commits
        // themselves bypass the hook — they're creating the taint
        // record and must be able to reach the path.
        if !is_taint_lifecycle_intent(&options.intent.category) {
            self.pre_commit_taint_check(&[path], &options)?;
        }
        let state_path =
            StatePath::parse(path).map_err(|e| TreeError::PathNotFound(e.to_string()))?;

        // Read-modify-write on the ref, so it must be a compare-and-swap: an
        // unconditional ref move built on a head another writer has since
        // advanced silently discards that writer's commit. See `advance_ref`.
        for attempt in 0..MAX_REF_CAS_ATTEMPTS {
            telemetry::cas_attempt("set", attempt);
            let (commit_id, is_branch) = self.head_for_write(ref_name)?;
            let commit = self
                .storage
                .get_commit(&commit_id)?
                .ok_or_else(|| RepoError::RefNotFound(ref_name.to_string()))?;
            let new_root = tree::tree_set(
                self.storage.as_ref(),
                &commit.state_root,
                &state_path,
                value,
            )?;
            let new_commit = self.create_commit(new_root, vec![commit_id], options.clone())?;
            if self.move_ref(ref_name, commit_id, is_branch, new_commit.id)? {
                return Ok(new_commit.id);
            }
        }
        telemetry::cas_gave_up("set", MAX_REF_CAS_ATTEMPTS);
        Err(RepoError::WriteConflict)
    }

    /// Set a value from JSON, creating a new commit.
    pub fn set_json(
        &self,
        ref_name: &str,
        path: &str,
        value: &serde_json::Value,
        options: CommitOptions,
    ) -> Result<ObjectId, RepoError> {
        let root_id = tree::json_to_tree(self.storage.as_ref(), value)?;
        let obj = self
            .storage
            .get_object(&root_id)?
            .ok_or_else(|| RepoError::RefNotFound("value".to_string()))?;
        // Snapshot the intent category so we can decide whether to
        // run auto-escalation *after* the main commit (the taint
        // lifecycle intents bypass it).
        let is_lifecycle = is_taint_lifecycle_intent(&options.intent.category);
        let commit_id = self.set(ref_name, path, &obj, options)?;
        if !is_lifecycle {
            // 0.7.75 §5: watch auto-escalation. Runs AFTER the
            // write — threshold-crossing creates a new taint commit
            // whose intent points at the watch-create commit.
            let _ = self.auto_escalate_watches(ref_name, path, value)?;
        }
        Ok(commit_id)
    }

    /// Return the current commit ID for `ref_name`. Needed for CAS-retry
    /// callers that must snapshot the head before a read-modify-write cycle.
    pub fn head(&self, ref_name: &str) -> Result<ObjectId, RepoError> {
        self.resolve_ref(ref_name)
    }

    /// Like `set_json` but uses a compare-and-swap on the ref rather than an
    /// unconditional `set_ref`. Builds the new commit on top of
    /// `expected_head` (not the current ref head), then calls `cas_ref`.
    ///
    /// Returns `Ok(new_commit_id)` on success, `Err(WriteConflict)` when the
    /// ref has moved since `expected_head` was snapshotted (allowing the
    /// caller to retry), or another error on hard failures.
    pub fn set_json_cas(
        &self,
        ref_name: &str,
        expected_head: ObjectId,
        path: &str,
        value: &serde_json::Value,
        options: CommitOptions,
    ) -> Result<ObjectId, RepoError> {
        telemetry::write(
            "set",
            write_span!("asg.repo.set", ref_name, options),
            || self.set_json_cas_inner(ref_name, expected_head, path, value, options),
        )
    }

    fn set_json_cas_inner(
        &self,
        ref_name: &str,
        expected_head: ObjectId,
        path: &str,
        value: &serde_json::Value,
        options: CommitOptions,
    ) -> Result<ObjectId, RepoError> {
        check_meta_guard(path, &options.intent)?;
        let is_lifecycle = is_taint_lifecycle_intent(&options.intent.category);
        if !is_lifecycle {
            self.pre_commit_taint_check(&[path], &options)?;
        }

        // Build the new tree from *expected_head*, not the live ref.
        let base_commit = self
            .storage
            .get_commit(&expected_head)?
            .ok_or_else(|| RepoError::RefNotFound(ref_name.to_string()))?;

        let root_id = tree::json_to_tree(self.storage.as_ref(), value)?;
        let obj = self
            .storage
            .get_object(&root_id)?
            .ok_or_else(|| RepoError::RefNotFound("value".to_string()))?;

        let state_path =
            StatePath::parse(path).map_err(|e| TreeError::PathNotFound(e.to_string()))?;
        let new_root = tree::tree_set(
            self.storage.as_ref(),
            &base_commit.state_root,
            &state_path,
            &obj,
        )?;

        let new_commit = self.create_commit(new_root, vec![expected_head], options)?;

        match self.advance_ref(ref_name, expected_head, new_commit.id)? {
            false => Err(RepoError::WriteConflict),
            true => {
                if !is_lifecycle {
                    let _ = self.auto_escalate_watches(ref_name, path, value)?;
                }
                Ok(new_commit.id)
            }
        }
    }

    /// Delete a value from state, creating a new commit.
    pub fn delete(
        &self,
        ref_name: &str,
        path: &str,
        options: CommitOptions,
    ) -> Result<ObjectId, RepoError> {
        telemetry::write(
            "delete",
            write_span!("asg.repo.delete", ref_name, options),
            || self.delete_inner(ref_name, path, options),
        )
    }

    fn delete_inner(
        &self,
        ref_name: &str,
        path: &str,
        options: CommitOptions,
    ) -> Result<ObjectId, RepoError> {
        check_meta_guard(path, &options.intent)?;
        if !is_taint_lifecycle_intent(&options.intent.category) {
            self.pre_commit_taint_check(&[path], &options)?;
        }
        let state_path =
            StatePath::parse(path).map_err(|e| TreeError::PathNotFound(e.to_string()))?;

        for attempt in 0..MAX_REF_CAS_ATTEMPTS {
            telemetry::cas_attempt("delete", attempt);
            let (commit_id, is_branch) = self.head_for_write(ref_name)?;
            let commit = self
                .storage
                .get_commit(&commit_id)?
                .ok_or_else(|| RepoError::RefNotFound(ref_name.to_string()))?;
            let new_root =
                tree::tree_delete(self.storage.as_ref(), &commit.state_root, &state_path)?;
            let new_commit = self.create_commit(new_root, vec![commit_id], options.clone())?;
            if self.move_ref(ref_name, commit_id, is_branch, new_commit.id)? {
                return Ok(new_commit.id);
            }
        }
        telemetry::cas_gave_up("delete", MAX_REF_CAS_ATTEMPTS);
        Err(RepoError::WriteConflict)
    }

    // -----------------------------------------------------------------------
    // Branch operations
    // -----------------------------------------------------------------------

    /// Create a new branch from the given ref. Fails with
    /// `BranchAlreadyExists` if the name is taken — including by another
    /// caller creating it at the same moment: the ref is created only if
    /// absent, so exactly one creator succeeds. (Checking and then setting
    /// gave every racer `Ok` and silently repointed all but the last.)
    ///
    /// Creating a ref orphans no commit, so there are no epoch seals to
    /// enforce, and a new ref has no leaf index to update.
    pub fn branch(&self, name: &str, from: &str) -> Result<ObjectId, RepoError> {
        let ns = self.active_namespace()?;
        if self.storage.get_ref(&ns, name)?.is_some() {
            return Err(RepoError::BranchAlreadyExists(name.to_string()));
        }
        let commit_id = self.resolve_ref(from)?;
        if !self.storage.create_ref(&ns, name, commit_id)? {
            return Err(RepoError::BranchAlreadyExists(name.to_string()));
        }
        Ok(commit_id)
    }

    /// Delete a branch. Returns true if the branch existed.
    /// Does NOT delete any commits (they remain in the DAG until GC sweeps
    /// what no ref reaches).
    ///
    /// Refuses `main`: every namespace forks from it, and with it gone the
    /// next `init()` creates an empty `main`, after which a sweep deletes
    /// everything the old one reached. A commit landing on the branch while
    /// the caller decides is lost with it; [`Repository::delete_branch_if`]
    /// deletes only the head the caller looked at.
    pub fn delete_branch(&self, name: &str) -> Result<bool, RepoError> {
        refuse_deleting_main(name)?;
        let ns = self.active_namespace()?;
        Ok(self.storage.delete_ref(&ns, name)?)
    }

    /// Delete a branch only if it still points to `expected` — the head the
    /// caller decided to delete. Returns `false`, leaving the branch alone,
    /// if it is absent or has moved since. Refuses `main`, like
    /// [`Repository::delete_branch`].
    pub fn delete_branch_if(&self, name: &str, expected: ObjectId) -> Result<bool, RepoError> {
        refuse_deleting_main(name)?;
        let ns = self.active_namespace()?;
        Ok(self.storage.cas_delete_ref(&ns, name, expected)?)
    }

    /// List all branches, optionally filtered by prefix.
    pub fn list_branches(
        &self,
        prefix: Option<&str>,
    ) -> Result<Vec<(String, ObjectId)>, RepoError> {
        let ns = self.active_namespace()?;
        Ok(self.storage.list_refs(&ns, prefix.unwrap_or(""))?)
    }

    /// Merge source branch into target branch.
    /// Uses three-way merge with the common ancestor (currently: the commit
    /// where the source branch was created from target).
    ///
    /// Returns Ok(commit_id) on success, or Err with conflicts.
    pub fn merge(
        &self,
        source: &str,
        target: &str,
        options: CommitOptions,
    ) -> Result<ObjectId, RepoError> {
        self.merge_checked(source, target, options, true)
    }

    /// Like [`Repository::merge`], but refuses to advance the target ref when
    /// the merge would remove any top-level entry (e.g. an entire `/plans` or
    /// `/memory` map) unless `allow_deletions` is true. This is a data-loss
    /// guard: the merge algorithm is correct, but a mistaken source or a
    /// genuine deletion should not silently drop a whole subtree.
    pub fn merge_checked(
        &self,
        source: &str,
        target: &str,
        options: CommitOptions,
        allow_deletions: bool,
    ) -> Result<ObjectId, RepoError> {
        let span = write_span!("asg.repo.merge", target, options);
        telemetry::write("merge", span, || {
            // The merge is computed against the target's head at the time, so
            // landing it is a compare-and-swap on that head; if another writer
            // moved the target meanwhile, recompute against the new head rather
            // than overwrite what it wrote.
            for attempt in 0..MAX_REF_CAS_ATTEMPTS {
                telemetry::cas_attempt("merge", attempt);
                if let Some(id) =
                    self.try_merge_checked(source, target, &options, allow_deletions)?
                {
                    return Ok(id);
                }
            }
            telemetry::cas_gave_up("merge", MAX_REF_CAS_ATTEMPTS);
            Err(RepoError::WriteConflict)
        })
    }

    /// One attempt of [`Repository::merge_checked`]. `Ok(None)` means the
    /// target ref moved while the merge was being computed.
    fn try_merge_checked(
        &self,
        source: &str,
        target: &str,
        options: &CommitOptions,
        allow_deletions: bool,
    ) -> Result<Option<ObjectId>, RepoError> {
        let (_, target_is_branch) = self.head_for_write(target)?;
        let comp = self.compute_merge(source, target)?;

        if !allow_deletions {
            let removed = self.top_level_removals(&comp)?;
            if !removed.is_empty() {
                return Err(RepoError::MergeWouldDelete(removed));
            }
        }

        match comp.result {
            MergeResult::Success(merged_obj) => {
                // Persist every newly-created composite node AND the root in a
                // single atomic batch. The merged tree references children by
                // id; nodes the merge fabricated (key sets that existed on
                // neither branch) are not in any store yet, so a partial write
                // would leave the tree dangling with ObjectNotFound on readback.
                let merged_root = merged_obj.id();
                let mut to_store: Vec<Object> = Vec::with_capacity(comp.created.len() + 1);
                to_store.extend(comp.created.iter().cloned());
                to_store.push(merged_obj);
                self.storage.batch_put_objects(&to_store)?;

                // Integrity gate: never advance a ref to a tree that isn't
                // fully readable. This catches both a partial write above and a
                // `created` set that failed to include a node the root
                // references — the exact defect that silently corrupted a
                // `/plans` subtree and made every plan appear to vanish.
                if let Some(missing) = self.first_missing_reachable(&merged_root)? {
                    return Err(RepoError::IntegrityViolation {
                        root: merged_root,
                        missing,
                    });
                }

                let commit = self.create_commit(
                    merged_root,
                    vec![comp.target_commit_id, comp.source_commit_id],
                    options.clone(),
                )?;
                Ok(self
                    .move_ref(target, comp.target_commit_id, target_is_branch, commit.id)?
                    .then_some(commit.id))
            }
            MergeResult::FastForward(ff_id) => {
                // Find the commit that has this state root
                // In fast-forward, we just advance the target ref
                let ff_commit = if ff_id == comp.source_state_root {
                    comp.source_commit_id
                } else {
                    comp.target_commit_id
                };
                Ok(self
                    .move_ref(target, comp.target_commit_id, target_is_branch, ff_commit)?
                    .then_some(ff_commit))
            }
            MergeResult::Conflicts { conflicts, .. } => Err(RepoError::MergeConflicts(conflicts)),
        }
    }

    /// Resolve the merge base (lowest common ancestor commit) of two refs.
    /// Useful for callers that need to reason about what each side changed
    /// relative to the branch point — e.g. domain-level merge policies.
    pub fn merge_base(&self, source: &str, target: &str) -> Result<ObjectId, RepoError> {
        let source_commit_id = self.resolve_ref(source)?;
        let target_commit_id = self.resolve_ref(target)?;
        self.find_common_ancestor(&source_commit_id, &target_commit_id)
    }

    /// Compute what merging `source` into `target` WOULD do, without advancing
    /// any ref or storing a commit. Returns a summary of top-level additions,
    /// changes, and removals plus any conflicts — the basis for a `--dry-run`.
    pub fn preview_merge(&self, source: &str, target: &str) -> Result<MergePreview, RepoError> {
        let comp = self.compute_merge(source, target)?;
        let result_root = match &comp.result {
            MergeResult::Success(obj) => obj.id(),
            MergeResult::Conflicts { partial, .. } => partial.id(),
            MergeResult::FastForward(state_root) => *state_root,
        };
        let target_entries = self.top_level_entries(&comp.target_state_root)?;
        let merged_entries = self.top_level_entries_of_obj(&comp.result, result_root)?;

        let mut added = Vec::new();
        let mut changed = Vec::new();
        let mut removed = Vec::new();
        for (k, v) in &merged_entries {
            match target_entries.get(k) {
                None => added.push(k.clone()),
                Some(tv) if tv != v => changed.push(k.clone()),
                _ => {}
            }
        }
        for k in target_entries.keys() {
            if !merged_entries.contains_key(k) {
                removed.push(k.clone());
            }
        }
        added.sort();
        changed.sort();
        removed.sort();

        let fast_forward = matches!(self.result_kind(&comp), MergeResultKind::FastForward);
        let conflicts = match comp.result {
            MergeResult::Conflicts { conflicts, .. } => conflicts,
            _ => Vec::new(),
        };
        Ok(MergePreview {
            fast_forward,
            added,
            changed,
            removed,
            conflicts,
        })
    }

    fn result_kind(&self, comp: &MergeComputation) -> MergeResultKind {
        match &comp.result {
            MergeResult::Success(_) => MergeResultKind::Success,
            MergeResult::Conflicts { .. } => MergeResultKind::Conflicts,
            MergeResult::FastForward(_) => MergeResultKind::FastForward,
        }
    }

    /// The top-level entries that would disappear from `target` if this merge
    /// were committed.
    fn top_level_removals(&self, comp: &MergeComputation) -> Result<Vec<String>, RepoError> {
        let result_root = match &comp.result {
            MergeResult::Success(obj) => obj.id(),
            MergeResult::Conflicts { partial, .. } => partial.id(),
            MergeResult::FastForward(state_root) => *state_root,
        };
        let target_entries = self.top_level_entries(&comp.target_state_root)?;
        let merged_entries = self.top_level_entries_of_obj(&comp.result, result_root)?;
        let mut removed: Vec<String> = target_entries
            .keys()
            .filter(|k| !merged_entries.contains_key(*k))
            .cloned()
            .collect();
        removed.sort();
        Ok(removed)
    }

    /// Resolve, find base, and run the collecting three-way merge without
    /// committing. Shared by `merge_checked` and `preview_merge`.
    fn compute_merge(&self, source: &str, target: &str) -> Result<MergeComputation, RepoError> {
        let source_commit_id = self.resolve_ref(source)?;
        let target_commit_id = self.resolve_ref(target)?;

        let source_commit = self
            .storage
            .get_commit(&source_commit_id)?
            .ok_or_else(|| RepoError::RefNotFound(source.to_string()))?;
        let target_commit = self
            .storage
            .get_commit(&target_commit_id)?
            .ok_or_else(|| RepoError::RefNotFound(target.to_string()))?;

        let base_commit_id = self.find_common_ancestor(&source_commit_id, &target_commit_id)?;
        let base_commit = self
            .storage
            .get_commit(&base_commit_id)?
            .ok_or_else(|| RepoError::RefNotFound("base".to_string()))?;

        let resolver = StorageResolver {
            storage: self.storage.as_ref(),
        };
        let (result, created) = agentstategraph_core::merge::try_three_way_merge_collect(
            &resolver,
            &base_commit.state_root,
            &target_commit.state_root,
            &source_commit.state_root,
        )
        .map_err(|missing| RepoError::IntegrityViolation {
            root: base_commit.state_root,
            missing,
        })?;

        Ok(MergeComputation {
            source_commit_id,
            target_commit_id,
            source_state_root: source_commit.state_root,
            target_state_root: target_commit.state_root,
            result,
            created,
        })
    }

    /// Top-level map entries (key -> child ObjectId) at a state root id.
    fn top_level_entries(
        &self,
        root: &ObjectId,
    ) -> Result<std::collections::BTreeMap<String, ObjectId>, RepoError> {
        match self.storage.get_object(root)? {
            Some(obj) => Ok(map_entries(&obj)),
            None => Ok(std::collections::BTreeMap::new()),
        }
    }

    /// Top-level entries of a merge result. For Success/Conflicts the merged
    /// object is in hand (its children may not be stored yet); for FastForward
    /// only an id is available, so fall back to the store.
    fn top_level_entries_of_obj(
        &self,
        result: &MergeResult,
        fallback_root: ObjectId,
    ) -> Result<std::collections::BTreeMap<String, ObjectId>, RepoError> {
        match result {
            MergeResult::Success(obj) | MergeResult::Conflicts { partial: obj, .. } => {
                Ok(map_entries(obj))
            }
            MergeResult::FastForward(_) => self.top_level_entries(&fallback_root),
        }
    }

    // -----------------------------------------------------------------------
    // Diff operations
    // -----------------------------------------------------------------------

    /// Compute a structured diff between two refs.
    /// Returns typed DiffOps (SetValue, AddKey, RemoveKey, etc.), not text diffs.
    pub fn diff(&self, ref_a: &str, ref_b: &str) -> Result<Vec<DiffOp>, RepoError> {
        telemetry::run(op_span!(DEBUG, "asg.repo.diff"), || {
            self.diff_inner(ref_a, ref_b)
        })
    }

    fn diff_inner(&self, ref_a: &str, ref_b: &str) -> Result<Vec<DiffOp>, RepoError> {
        let commit_a = self.resolve_ref(ref_a)?;
        let commit_b = self.resolve_ref(ref_b)?;

        let ca = self
            .storage
            .get_commit(&commit_a)?
            .ok_or_else(|| RepoError::RefNotFound(ref_a.to_string()))?;
        let cb = self
            .storage
            .get_commit(&commit_b)?
            .ok_or_else(|| RepoError::RefNotFound(ref_b.to_string()))?;

        let resolver = StorageResolver {
            storage: self.storage.as_ref(),
        };
        Ok(agentstategraph_core::diff::diff(
            &resolver,
            &ca.state_root,
            &cb.state_root,
        ))
    }

    // -----------------------------------------------------------------------
    // Speculative execution
    // -----------------------------------------------------------------------

    /// Create a speculation forked from a ref. O(1) — just a pointer.
    pub fn speculate(
        &self,
        from_ref: &str,
        label: Option<String>,
    ) -> Result<SpecHandle, RepoError> {
        let span = op_span!(
            INFO,
            "asg.speculation.create",
            asg.ref.name = from_ref,
            asg.speculation.id = tracing::field::Empty,
        );
        let result = telemetry::run(span.clone(), || self.speculate_inner(from_ref, label));
        if let Ok(handle) = &result {
            span.record("asg.speculation.id", handle.id());
        }
        result
    }

    fn speculate_inner(
        &self,
        from_ref: &str,
        label: Option<String>,
    ) -> Result<SpecHandle, RepoError> {
        let commit_id = self.resolve_ref(from_ref)?;
        let commit = self
            .storage
            .get_commit(&commit_id)?
            .ok_or_else(|| RepoError::RefNotFound(from_ref.to_string()))?;

        let handle = self
            .specs
            .create(from_ref, commit.state_root, label)
            .map_err(RepoError::Speculation)?;
        if let Err(e) = self.pin_speculation(handle, Some(&commit.state_root)) {
            let _ = self.discard_speculation(handle);
            return Err(e);
        }
        Ok(handle)
    }

    /// Get a value from a speculation's state.
    pub fn spec_get(&self, handle: SpecHandle, path: &str) -> Result<Object, RepoError> {
        self.specs
            .get(handle, self.storage.as_ref(), path)
            .map_err(RepoError::Speculation)
    }

    /// Set a value in a speculation's state.
    pub fn spec_set(
        &self,
        handle: SpecHandle,
        path: &str,
        value: &Object,
    ) -> Result<(), RepoError> {
        self.specs
            .set(handle, self.storage.as_ref(), path, value)
            .map_err(RepoError::Speculation)?;
        self.pin_speculation(handle, None)
    }

    /// Set a value from JSON in a speculation's state. Convenience wrapper
    /// around `spec_set` that mirrors `set_json` — converts the JSON value
    /// into an Object tree before applying.
    pub fn spec_set_json(
        &self,
        handle: SpecHandle,
        path: &str,
        value: &serde_json::Value,
    ) -> Result<(), RepoError> {
        let root_id = tree::json_to_tree(self.storage.as_ref(), value)?;
        let obj = self
            .storage
            .get_object(&root_id)?
            .ok_or_else(|| RepoError::RefNotFound("value".to_string()))?;
        self.spec_set(handle, path, &obj)
    }

    /// Delete a value in a speculation's state.
    pub fn spec_delete(&self, handle: SpecHandle, path: &str) -> Result<(), RepoError> {
        self.specs
            .delete(handle, self.storage.as_ref(), path)
            .map_err(RepoError::Speculation)?;
        self.pin_speculation(handle, None)
    }

    /// Compare multiple speculations side-by-side.
    pub fn compare_speculations(
        &self,
        handles: &[SpecHandle],
    ) -> Result<SpecComparison, RepoError> {
        self.specs
            .compare(handles, self.storage.as_ref())
            .map_err(RepoError::Speculation)
    }

    /// Commit a speculation — promotes it to a real commit on the base branch.
    ///
    /// If the speculation touched any `/_meta/*` path and the commit's
    /// intent is not `IntentCategory::Migrate`, the commit is rejected
    /// with `RepoError::ReservedPath`. This keeps the meta namespace
    /// enforced for speculation writes the same as for direct writes.
    pub fn commit_speculation(
        &self,
        handle: SpecHandle,
        options: CommitOptions,
    ) -> Result<ObjectId, RepoError> {
        let span = write_span!("asg.speculation.commit", tracing::field::Empty, options);
        span.record("asg.speculation.id", handle.id());
        telemetry::write("commit_speculation", span, || {
            let result = self.commit_speculation_inner(handle, options);
            self.unpin_speculation(handle);
            result
        })
    }

    fn commit_speculation_inner(
        &self,
        handle: SpecHandle,
        options: CommitOptions,
    ) -> Result<ObjectId, RepoError> {
        let (spec_root, base_root, base_ref) = self
            .specs
            .commit_with_base(handle)
            .map_err(RepoError::Speculation)?;
        tracing::Span::current().record("asg.ref.name", base_ref.as_str());
        let resolver = StorageResolver {
            storage: self.storage.as_ref(),
        };

        // Gate /_meta/* writes on the commit's intent category. Judge what the
        // speculation itself changed (base → spec), not its difference from a
        // head that other writers may have moved since.
        if options.intent.category != IntentCategory::Migrate {
            let diff = agentstategraph_core::diff::diff(&resolver, &base_root, &spec_root);
            if let Some(path) = reserved_path_in_diff(&diff) {
                return Err(RepoError::ReservedPath(path));
            }
        }

        // The speculation forked at `base_root`. Committing its root as-is
        // would revert every write that landed on the base ref since the fork
        // (the ledger entries `asd index` was silently discarding). Instead,
        // three-way merge it onto the current head: the speculation wins any
        // leaf both sides changed — what it always did — and everything else
        // the head gained is kept. Landed by compare-and-swap, so a write that
        // races the merge itself is not lost either.
        for attempt in 0..MAX_REF_CAS_ATTEMPTS {
            telemetry::cas_attempt("commit_speculation", attempt);
            let (parent_id, is_branch) = self.head_for_write(&base_ref)?;
            let parent_root = self
                .storage
                .get_commit(&parent_id)?
                .ok_or_else(|| RepoError::RefNotFound(base_ref.clone()))?
                .state_root;
            let new_root = if parent_root == base_root {
                // Nothing moved since the fork: land the speculation's root
                // as-is — but only if every object it wrote still exists. A
                // GC sweep between `speculate` and here (open speculations
                // are not GC roots in other processes) used to publish a ref
                // whose tree was missing objects.
                if let Some(missing) = self.first_missing_new(&base_root, &spec_root)? {
                    return Err(RepoError::IntegrityViolation {
                        root: spec_root,
                        missing,
                    });
                }
                spec_root
            } else {
                // ours = speculation, so conflicts resolve to it.
                let (result, created) = agentstategraph_core::merge::try_three_way_merge_collect(
                    &resolver,
                    &base_root,
                    &spec_root,
                    &parent_root,
                )
                .map_err(|missing| RepoError::IntegrityViolation {
                    root: spec_root,
                    missing,
                })?;
                match result {
                    MergeResult::FastForward(id) => id,
                    MergeResult::Success(merged)
                    | MergeResult::Conflicts {
                        partial: merged, ..
                    } => {
                        let merged_root = merged.id();
                        let mut to_store = created;
                        to_store.push(merged);
                        self.storage.batch_put_objects(&to_store)?;
                        if let Some(missing) = self.first_missing_reachable(&merged_root)? {
                            return Err(RepoError::IntegrityViolation {
                                root: merged_root,
                                missing,
                            });
                        }
                        merged_root
                    }
                }
            };
            let commit = self.create_commit(new_root, vec![parent_id], options.clone())?;
            if self.move_ref(&base_ref, parent_id, is_branch, commit.id)? {
                return Ok(commit.id);
            }
        }
        telemetry::cas_gave_up("commit_speculation", MAX_REF_CAS_ATTEMPTS);
        Err(RepoError::WriteConflict)
    }

    /// Commit a speculation **as-is**, and only if `expected_head` is still
    /// the head of the speculation's base ref and the speculation forked from
    /// that head's state. Otherwise returns `RepoError::WriteConflict` and the
    /// caller re-reads, re-speculates and retries — the speculation is
    /// consumed either way.
    ///
    /// The speculation counterpart of [`Repository::set_json_cas`]. Use it when
    /// the speculation's content was computed from values read at
    /// `expected_head` (a read-modify-write across several paths):
    /// [`Repository::commit_speculation`] merges onto whatever head is current
    /// with the speculation winning conflicting leaves, which silently reverts
    /// a concurrent change to anything the caller read and wrote back.
    pub fn commit_speculation_cas(
        &self,
        handle: SpecHandle,
        expected_head: ObjectId,
        options: CommitOptions,
    ) -> Result<ObjectId, RepoError> {
        let span = write_span!("asg.speculation.commit", tracing::field::Empty, options);
        span.record("asg.speculation.id", handle.id());
        telemetry::write("commit_speculation", span, || {
            let result = self.commit_speculation_cas_inner(handle, expected_head, options);
            self.unpin_speculation(handle);
            result
        })
    }

    fn commit_speculation_cas_inner(
        &self,
        handle: SpecHandle,
        expected_head: ObjectId,
        options: CommitOptions,
    ) -> Result<ObjectId, RepoError> {
        let (spec_root, base_root, base_ref) = self
            .specs
            .commit_with_base(handle)
            .map_err(RepoError::Speculation)?;
        tracing::Span::current().record("asg.ref.name", base_ref.as_str());
        let expected_root = self
            .storage
            .get_commit(&expected_head)?
            .ok_or_else(|| RepoError::CommitNotFound(expected_head.short()))?
            .state_root;
        if base_root != expected_root {
            // Forked from a different state than the caller read from.
            return Err(RepoError::WriteConflict);
        }
        if options.intent.category != IntentCategory::Migrate {
            let resolver = StorageResolver {
                storage: self.storage.as_ref(),
            };
            let diff = agentstategraph_core::diff::diff(&resolver, &base_root, &spec_root);
            if let Some(path) = reserved_path_in_diff(&diff) {
                return Err(RepoError::ReservedPath(path));
            }
        }
        if let Some(missing) = self.first_missing_new(&base_root, &spec_root)? {
            return Err(RepoError::IntegrityViolation {
                root: spec_root,
                missing,
            });
        }
        let commit = self.create_commit(spec_root, vec![expected_head], options)?;
        if self.advance_ref(&base_ref, expected_head, commit.id)? {
            Ok(commit.id)
        } else {
            Err(RepoError::WriteConflict)
        }
    }

    /// Discard a speculation — all changes lost. Instant.
    pub fn discard_speculation(&self, handle: SpecHandle) -> Result<(), RepoError> {
        let span = op_span!(
            INFO,
            "asg.speculation.discard",
            asg.speculation.id = handle.id(),
        );
        telemetry::run(span, || {
            self.unpin_speculation(handle);
            self.specs.discard(handle).map_err(RepoError::Speculation)
        })
    }

    /// Record an open speculation's roots as GC roots. A speculation writes its
    /// objects to the store as it goes, but nothing references them until it
    /// commits, so without a pin a sweep — in this process or another sharing
    /// the store — deleted them, and the commit then either published a broken
    /// tree or (merging onto a moved head) silently dropped the speculation's
    /// writes. `base` is pinned once, at fork; the working root on every change.
    fn pin_speculation(
        &self,
        handle: SpecHandle,
        base: Option<&ObjectId>,
    ) -> Result<(), RepoError> {
        if let Some(base) = base {
            self.storage
                .gc_pin_root(&self.speculation_pin_key(handle, "base"), base)?;
        }
        let current = self
            .specs
            .current_root(handle)
            .map_err(RepoError::Speculation)?;
        self.storage
            .gc_pin_root(&self.speculation_pin_key(handle, "cur"), &current)?;
        Ok(())
    }

    /// Best-effort: a pin left behind (a crash, a failed delete) only keeps its
    /// objects until it expires.
    fn unpin_speculation(&self, handle: SpecHandle) {
        let _ = self
            .storage
            .gc_unpin_root(&self.speculation_pin_key(handle, "base"));
        let _ = self
            .storage
            .gc_unpin_root(&self.speculation_pin_key(handle, "cur"));
    }

    fn speculation_pin_key(&self, handle: SpecHandle, which: &str) -> String {
        format!("spec:{}:{}:{}", self.instance, handle, which)
    }

    /// List all active speculations.
    pub fn list_speculations(&self) -> Vec<(SpecHandle, Option<String>)> {
        self.specs.list()
    }

    // -----------------------------------------------------------------------
    // History operations
    // -----------------------------------------------------------------------

    /// Get the commit log starting from a ref.
    ///
    /// Each returned commit has `enforce_caps` applied — the same length
    /// caps `CommitBuilder::build` uses. This bounds what a malicious or
    /// legacy `.db` can replay to readers. (security threat model v2, F3)
    pub fn log(&self, ref_name: &str, limit: usize) -> Result<Vec<Commit>, RepoError> {
        let span = op_span!(DEBUG, "asg.repo.log", asg.ref.name = ref_name);
        telemetry::run(span, || {
            let commit_id = self.resolve_ref(ref_name)?;
            let mut commits = self.storage.list_commits(&commit_id, limit)?;
            for c in &mut commits {
                c.enforce_caps();
            }
            Ok(commits)
        })
    }

    /// Like [`Repository::log`], but walks **every** parent edge.
    ///
    /// `log` follows `parents[0]`, which is a first-parent line: on a store
    /// with merges it silently under-reports. Measured on the asd store, `log`
    /// reached 4,268 of 5,896 commits — the rest being merge second-parents
    /// and commits the ref head no longer reaches.
    ///
    /// Returns [`CommitWalk`], whose `truncated` flag distinguishes "that is
    /// all of them" from "there are more". Callers that need to know whether
    /// a store is bigger than the bound they asked for must read it; the
    /// commit list alone cannot tell them.
    pub fn log_dag(&self, ref_name: &str, limit: usize) -> Result<CommitWalk, RepoError> {
        let commit_id = self.resolve_ref(ref_name)?;
        let mut walk = self.storage.list_commits_dag(&commit_id, limit)?;
        // Re-cap on read, exactly as `log` does.
        for c in &mut walk.commits {
            c.enforce_caps();
        }
        Ok(walk)
    }

    /// Get a specific commit by ID.
    ///
    /// Returns the commit with `enforce_caps` applied — see `log`.
    pub fn get_commit(&self, id: &ObjectId) -> Result<Option<Commit>, RepoError> {
        Ok(self.storage.get_commit(id)?.map(|mut c| {
            c.enforce_caps();
            c
        }))
    }

    // -----------------------------------------------------------------------
    // Query operations
    // -----------------------------------------------------------------------

    /// Query commits with composable filters. Supports offset for pagination.
    pub fn query_commits(
        &self,
        ref_name: &str,
        filters: &agentstategraph_core::QueryFilters,
        limit: usize,
    ) -> Result<Vec<Commit>, RepoError> {
        self.query_commits_paged(ref_name, filters, limit, 0)
    }

    /// Query commits with pagination (offset + limit).
    pub fn query_commits_paged(
        &self,
        ref_name: &str,
        filters: &agentstategraph_core::QueryFilters,
        limit: usize,
        offset: usize,
    ) -> Result<Vec<Commit>, RepoError> {
        // Two faults lived here. It called `log`, inheriting the first-parent
        // blind spot, and it scanned a hard-coded 10_000 — so a caller asking
        // for offset 12000 got an empty page with nothing to distinguish it
        // from "no more commits".
        //
        // Filtering happens in memory, so the scan has to cover at least the
        // window being asked for. It still cannot be exact: a filter may
        // reject arbitrarily many commits, so `offset + limit` raw commits do
        // not guarantee `limit` filtered ones. The floor keeps the common
        // case cheap; the `offset + limit` term stops a deep page silently
        // returning nothing.
        let scan = commit_query_scan(offset, limit);
        let walk = self.log_dag(ref_name, scan)?;
        let filtered = agentstategraph_core::filter_commits(&walk.commits, filters);
        Ok(filtered.into_iter().skip(offset).take(limit).collect())
    }

    /// Blame — for a path, find which commit last modified it and why.
    ///
    /// The returned entry has `timestamp_anomaly` set when the commit's
    /// timestamp is `<=` at least one of its parents' timestamps — a signal
    /// of possible clock rewind. This is detection, not prevention: honest
    /// NTP skew triggers occasional flags, but persistent or concentrated
    /// flags from one agent are worth investigating.
    /// (security threat model v3+, V4)
    pub fn blame(
        &self,
        ref_name: &str,
        path: &str,
    ) -> Result<agentstategraph_core::BlameEntry, RepoError> {
        let span = op_span!(DEBUG, "asg.repo.blame", asg.ref.name = ref_name);
        telemetry::run(span, || self.blame_inner(ref_name, path)).map(|(entry, _)| entry)
    }

    /// [`Repository::blame`], with the trace the blamed commit was made in,
    /// when [`crate::link_commits_to_traces`] recorded one.
    pub fn blame_with_trace(
        &self,
        ref_name: &str,
        path: &str,
    ) -> Result<(agentstategraph_core::BlameEntry, Option<CommitTrace>), RepoError> {
        let span = op_span!(DEBUG, "asg.repo.blame", asg.ref.name = ref_name);
        let (entry, commit_id) = telemetry::run(span, || self.blame_inner(ref_name, path))?;
        Ok((entry, self.commit_trace(&commit_id)?))
    }

    fn blame_inner(
        &self,
        ref_name: &str,
        path: &str,
    ) -> Result<(agentstategraph_core::BlameEntry, ObjectId), RepoError> {
        // DAG walk for the population, then re-sorted by time.
        //
        // `log` missed changes made on a merged-in branch, attributing them to
        // the merge commit instead of the commit that made them. But swapping
        // in `log_dag` alone would have been a different bug: this loop returns
        // the FIRST match in walk order, and a DAG walk is ordered by distance
        // from the head, not by time — so blame could return a commit that is
        // nearer in hops yet older. Sorting restores the property the old
        // first-parent order gave for free.
        //
        // Timestamps are self-reported, and `detect_timestamp_anomalies` exists
        // because they can be rewound. Blame is a question about time, so it has
        // to trust them; a store failing that check should not be trusted here
        // either.
        let mut commits = self.log_dag(ref_name, 1000)?.commits;
        commits.sort_by(|a, b| b.timestamp.cmp(&a.timestamp));
        let state_path = StatePath::parse(path)
            .map_err(|e| RepoError::Tree(tree::TreeError::PathNotFound(e.to_string())))?;

        // The most recent commit that INTRODUCED this value — meaning no parent
        // of it already had that same value.
        //
        // Comparing against `parents.first()` alone was the other half of the
        // first-parent bug, and switching the walk above would not have fixed
        // it. At a merge the value exists but the first parent (the target
        // branch) lacks it, so the old `(Some, None)` arm read that as "added
        // here" and blamed the merge rather than the commit on the branch that
        // actually set it. Checking EVERY parent makes a merge transparent: the
        // side it came from already holds the value, so the merge did not
        // introduce it.
        for commit in &commits {
            let Ok(curr) = tree::tree_get(self.storage.as_ref(), &commit.state_root, &state_path)
            else {
                continue; // path absent here; cannot be where it was introduced
            };

            if commit.parents.is_empty() {
                // Initial commit — this is where everything was "set".
                return Ok((self.blame_entry_for(path, commit)?, commit.id));
            }

            let mut introduced_here = true;
            for parent_id in &commit.parents {
                if let Some(mut parent) = self.storage.get_commit(parent_id)? {
                    // Re-cap on read — see `Commit::enforce_caps`.
                    parent.enforce_caps();
                    if let Ok(prev) =
                        tree::tree_get(self.storage.as_ref(), &parent.state_root, &state_path)
                        && prev == curr
                    {
                        introduced_here = false;
                        break;
                    }
                }
            }
            if introduced_here {
                return Ok((self.blame_entry_for(path, commit)?, commit.id));
            }
        }

        Err(RepoError::RefNotFound(format!(
            "no commit found that modified {}",
            path
        )))
    }

    /// Build a `BlameEntry` from a commit, computing the `timestamp_anomaly`
    /// flag by comparing this commit's timestamp to each parent's timestamp.
    fn blame_entry_for(
        &self,
        path: &str,
        commit: &Commit,
    ) -> Result<agentstategraph_core::BlameEntry, RepoError> {
        let anomaly = self.commit_has_timestamp_anomaly(commit)?;
        Ok(agentstategraph_core::BlameEntry {
            path: path.to_string(),
            commit_id: commit.id.short(),
            agent_id: commit.agent_id.clone(),
            intent_category: format!("{:?}", commit.intent.category),
            intent_description: commit.intent.description.clone(),
            reasoning: commit.reasoning.clone(),
            timestamp: commit.timestamp,
            timestamp_anomaly: anomaly,
        })
    }

    /// True iff this commit's timestamp is `<=` at least one of its parents'
    /// timestamps. Root commits (no parents) are always monotonic by
    /// definition.
    fn commit_has_timestamp_anomaly(&self, commit: &Commit) -> Result<bool, RepoError> {
        for parent_id in &commit.parents {
            if let Some(mut parent) = self.storage.get_commit(parent_id)? {
                parent.enforce_caps();
                if commit.timestamp <= parent.timestamp {
                    return Ok(true);
                }
            }
        }
        Ok(false)
    }

    /// Detect timestamp anomalies reachable from a ref — commits whose
    /// timestamps are `<=` at least one parent's timestamp. Returns
    /// `(commit_id, reason)` pairs. An empty list means the reachable
    /// history is monotonic.
    ///
    /// This is a READ-TIME check, not a write-time guard: a single flag is
    /// consistent with clock skew or DAG-concurrent commits, but a pattern
    /// of flags from one agent or branch is a tamper signal. Audit
    /// alongside the commit id, not in place of it.
    /// (security threat model v3+, V4)
    pub fn detect_timestamp_anomalies(
        &self,
        ref_name: &str,
    ) -> Result<Vec<(ObjectId, String)>, RepoError> {
        // Walk the whole DAG, uncapped, and both choices are deliberate.
        //
        // `log` follows first parents, so a (child, parent) pair wholly INSIDE
        // a merged-in branch was never examined — those commits never appear as
        // the outer `commit`. Merge-vs-merged-tip was checked, which is why the
        // gap is easy to miss: an agent that backdated several commits on a
        // branch before merging left no flag at all.
        //
        // The cap is gone rather than raised. This returns a plain Vec, so a
        // truncated scan is indistinguishable from a clean one — and on a
        // tamper check, "I stopped looking" reported as "nothing found" is the
        // one failure mode worth spending time to avoid. The walk is bounded by
        // the store either way; `usize::MAX` only removes the arbitrary 100_000
        // that used to sit here, which the DAG walk would reach sooner than the
        // first-parent line did.
        let walk = self.log_dag(ref_name, usize::MAX)?;
        let mut findings = Vec::new();
        for commit in &walk.commits {
            for parent_id in &commit.parents {
                if let Some(mut parent) = self.storage.get_commit(parent_id)? {
                    parent.enforce_caps();
                    if commit.timestamp <= parent.timestamp {
                        findings.push((
                            commit.id,
                            format!(
                                "commit {} timestamp {} <= parent {} timestamp {}",
                                commit.id.short(),
                                commit.timestamp.to_rfc3339(),
                                parent.id.short(),
                                parent.timestamp.to_rfc3339(),
                            ),
                        ));
                        break;
                    }
                }
            }
        }
        Ok(findings)
    }

    // -----------------------------------------------------------------------
    // Session operations (sub-agent orchestration)
    // -----------------------------------------------------------------------

    /// Get a session manager for sub-agent orchestration. The manager
    /// borrows this repository's storage backend so all state is
    /// durable.
    pub fn sessions(&self) -> crate::session::SessionManager<'_> {
        crate::session::SessionManager::new(self.storage.as_ref())
    }

    /// Set (or clear) the active epoch id. When set, commits created via
    /// `commit`/`set`/`merge`/`commit_speculation` will be associated
    /// with that epoch in storage.
    pub fn set_active_epoch(&self, id: Option<String>) -> Result<(), RepoError> {
        *self
            .active_epoch
            .write()
            .map_err(|e| RepoError::RefNotFound(e.to_string()))? = id;
        Ok(())
    }

    /// Return the currently-active epoch id, if any.
    pub fn active_epoch(&self) -> Result<Option<String>, RepoError> {
        Ok(self
            .active_epoch
            .read()
            .map_err(|e| RepoError::RefNotFound(e.to_string()))?
            .clone())
    }

    /// Set (or clear) the active session id.
    /// Eagerly caches the session's `scope_namespace` to avoid per-call storage lookups.
    pub fn set_active_session(&self, id: Option<String>) -> Result<(), RepoError> {
        // Cache the session's scope_namespace to avoid per-call storage lookups.
        let ns_override = if let Some(ref session_id) = id {
            self.storage
                .get_session(session_id)?
                .and_then(|s| s.scope_namespace)
        } else {
            None
        };
        *self
            .active_session
            .write()
            .map_err(|e| RepoError::RefNotFound(e.to_string()))? = id;
        *self
            .active_session_namespace
            .write()
            .map_err(|e| RepoError::RefNotFound(e.to_string()))? = ns_override;
        Ok(())
    }

    /// Return the currently-active session id, if any.
    pub fn active_session(&self) -> Result<Option<String>, RepoError> {
        Ok(self
            .active_session
            .read()
            .map_err(|e| RepoError::RefNotFound(e.to_string()))?
            .clone())
    }

    /// Resolve the effective namespace for the current call.
    ///
    /// Priority: active session's `scope_namespace` > repository's configured
    /// `namespace`. Falls back to `Namespace::default_ns()` only when no
    /// session is active and no namespace was configured.
    fn active_namespace(&self) -> Result<Namespace, RepoError> {
        if let Some(ns) = self
            .active_session_namespace
            .read()
            .map_err(|e| RepoError::RefNotFound(e.to_string()))?
            .clone()
        {
            return Ok(ns);
        }
        Ok(self.namespace.clone())
    }

    // -----------------------------------------------------------------------
    // Namespace operations
    // -----------------------------------------------------------------------

    /// Create a namespace. Returns `Ok(())` if already exists.
    pub fn create_namespace(&self, name: &str) -> Result<(), RepoError> {
        let ns = Namespace::new(name).map_err(|e| RepoError::InvalidOperation(e.to_string()))?;
        match self.storage.create_namespace(&ns) {
            Ok(()) => Ok(()),
            Err(StorageError::NamespaceAlreadyExists(_)) => Ok(()),
            Err(e) => Err(RepoError::Storage(e)),
        }
    }

    /// List all namespaces.
    pub fn list_namespaces(&self) -> Result<Vec<Namespace>, RepoError> {
        Ok(self.storage.list_namespaces()?)
    }

    /// Delete a namespace and all its refs. The "default" namespace cannot be deleted.
    /// Returns `true` if it existed and was removed.
    pub fn delete_namespace(&self, name: &str) -> Result<bool, RepoError> {
        let ns = Namespace::new(name).map_err(|e| RepoError::InvalidOperation(e.to_string()))?;
        Ok(self.storage.delete_namespace(&ns)?)
    }

    /// Merge a branch from a different namespace into a branch in the active namespace.
    ///
    /// Cross-namespace merges are denied by default — they require a configured
    /// PolicyStore (future work). Returns `Err(CrossNamespaceAccessDenied)` until
    /// policy integration is complete.
    ///
    /// The merge is audited: the commit's intent tags include the source namespace
    /// and branch so the history remains inspectable.
    pub fn cross_namespace_merge(
        &self,
        source_namespace: &str,
        source_branch: &str,
        target_branch: &str,
        options: CommitOptions,
    ) -> Result<ObjectId, RepoError> {
        let source_ns = Namespace::new(source_namespace)
            .map_err(|e| RepoError::InvalidOperation(e.to_string()))?;
        let target_ns = self.active_namespace()?;

        if source_ns == target_ns {
            // Same namespace — just a normal merge.
            return self.merge(source_branch, target_branch, options);
        }

        // Cross-namespace: deny until PolicyStore integration lands.
        // When a PolicyStore is wired up this check will consult it.
        Err(RepoError::CrossNamespaceAccessDenied)
    }

    // -----------------------------------------------------------------------
    // Watch operations
    // -----------------------------------------------------------------------

    /// Get the watch manager for subscribing to state changes.
    pub fn watches(&self) -> &crate::watch::WatchManager {
        &self.watch_mgr
    }

    // -----------------------------------------------------------------------
    // Epoch operations
    // -----------------------------------------------------------------------

    /// Create a new epoch, persisted via the storage backend.
    pub fn create_epoch(
        &self,
        id: &str,
        description: &str,
        root_intents: Vec<String>,
    ) -> Result<agentstategraph_core::Epoch, RepoError> {
        let epoch = agentstategraph_core::Epoch::new(id, description, root_intents)
            .with_namespace(self.active_namespace()?.as_str());
        self.storage.create_epoch(&epoch)?;
        Ok(epoch)
    }

    /// Create an epoch that declares what it covers. See [`EpochScope`].
    pub fn create_epoch_scoped(
        &self,
        id: &str,
        description: &str,
        root_intents: Vec<String>,
        scope: agentstategraph_core::EpochScope,
    ) -> Result<agentstategraph_core::Epoch, RepoError> {
        let epoch = agentstategraph_core::Epoch::new(id, description, root_intents)
            .with_namespace(self.active_namespace()?.as_str())
            .with_scope(scope);
        self.storage.create_epoch(&epoch)?;
        Ok(epoch)
    }

    /// Does this epoch bind writes in `ns`?
    ///
    /// An epoch enforces only inside its own workspace. This is the fix for
    /// seals leaking across workspaces: the store is one global table, so an
    /// unfiltered check let one workspace's sealed epoch reject writes in every
    /// other workspace — completing a plan in one project could block writes in
    /// all of them.
    ///
    /// **An epoch with no recorded owner binds nothing.** v1.1.0 treated it as
    /// binding the default workspace, reasoning that before namespacing there
    /// was effectively one workspace. On a real multi-workspace store that is
    /// wrong, and it bricks the hub: epochs sealed before the column existed
    /// hold commits from many different workspaces' graphs, none of which are
    /// reachable from the default workspace's `main`. Each one becomes a
    /// permanent, unsatisfiable veto on every default-workspace write —
    /// observed on a real store where 171 such epochs blocked startup outright.
    ///
    /// The guard exists to stop a rewind orphaning commits *in the workspace
    /// that sealed them*. An epoch with no recorded workspace cannot establish
    /// that relationship, so it has nothing to enforce. Epochs created from
    /// v1.1.0 onward are stamped at creation, and
    /// [`Repository::assign_epoch_namespace`] settles older ones deliberately.
    fn epoch_binds_namespace(epoch: &agentstategraph_core::Epoch, ns: &Namespace) -> bool {
        match epoch.namespace.as_deref() {
            Some(owner) => owner == ns.as_str(),
            None => false,
        }
    }

    /// Seal an epoch, making it immutable.
    ///
    /// Captures the set of commits reachable from `main` at seal time
    /// so that subsequent ref mutations can be checked for seal
    /// violations (a rewind of main that would orphan a sealed commit).
    /// Persisted via the storage backend — survives process restart.
    /// (security threat model v3+, V8)
    pub fn seal_epoch(&self, id: &str, summary: &str) -> Result<(), RepoError> {
        let ns = self.active_namespace()?;
        let epoch = self.get_epoch(id)?;

        // What gets sealed depends on what the epoch claims to cover.
        // A Plan-scoped epoch is defined by the commits tagged to it, not by a
        // walk, so sealing must not silently widen it to the whole branch.
        let sealed_commits = match epoch.scope.seal_ref() {
            Some(ref_name) => match self.storage.get_ref(&ns, ref_name)? {
                Some(head) => self.reachable_commits_from(&head)?,
                None => Vec::new(),
            },
            None => epoch.commits.clone(),
        };

        let seal_hash = Self::compute_seal_hash(id, &sealed_commits);
        self.storage
            .seal_epoch(id, summary, chrono::Utc::now(), &sealed_commits, &seal_hash)?;
        Ok(())
    }

    /// Tamper-evident digest over an epoch's sealed set.
    ///
    /// Order-independent (the ids are sorted first) so the same set always
    /// yields the same hash regardless of walk order, and domain-separated by
    /// the epoch id so two epochs covering identical commits do not collide.
    /// Persisted at seal time rather than recomputed on read — the point is to
    /// detect the store changing underneath, which a recomputation cannot do.
    fn compute_seal_hash(id: &str, sealed_commits: &[ObjectId]) -> ObjectId {
        let mut ids: Vec<&ObjectId> = sealed_commits.iter().collect();
        ids.sort_by_key(|o| *o.as_bytes());
        let mut buf = Vec::with_capacity(32 * ids.len() + id.len() + 16);
        buf.extend_from_slice(b"asg-epoch-seal-v1\0");
        buf.extend_from_slice(id.as_bytes());
        buf.push(0);
        for oid in ids {
            buf.extend_from_slice(oid.as_bytes());
        }
        ObjectId::hash(&buf)
    }

    /// Assign a workspace to an epoch that does not have one yet.
    ///
    /// Returns `true` if this call set it, `false` if the epoch already had an
    /// owner, and an error if there is no such epoch.
    ///
    /// ASSIGN-ONLY BY DESIGN — this is not "move epoch to workspace". An epoch
    /// binds writes in its own workspace, so allowing a reassign would let a
    /// caller relabel a sealed epoch into a different workspace and change
    /// whose writes it guards. Once set, an owner is final.
    ///
    /// The job it exists for: epochs sealed before the namespace column existed
    /// carry no owner and read as belonging to the default workspace. A caller
    /// that knows where they belong — typically from its own epoch-id
    /// convention — can settle it once. Works on sealed and archived epochs,
    /// which is the point: those are exactly the ones that predate the column.
    /// The sealed commit set and seal hash are untouched, so what was sealed
    /// stays sealed.
    pub fn assign_epoch_namespace(&self, id: &str, namespace: &str) -> Result<bool, RepoError> {
        // Validate up front: a malformed name would otherwise be written and
        // then never match any real workspace on read.
        Namespace::new(namespace)
            .map_err(|e| RepoError::InvalidOperation(format!("invalid namespace: {e}")))?;
        self.storage
            .assign_epoch_namespace(id, namespace)
            .map_err(RepoError::Storage)
    }

    /// List this workspace's epochs (as lightweight index entries).
    ///
    /// Scoped to the active namespace, so a workspace's epoch count is its own
    /// rather than the hub-wide total. Use [`Repository::list_all_epochs`] for
    /// a cross-workspace view.
    pub fn list_epochs(&self) -> Result<Vec<agentstategraph_core::EpochEntry>, RepoError> {
        let ns = self.active_namespace()?;
        let epochs = self.storage.list_epochs()?;
        Ok(epochs
            .iter()
            .filter(|e| Self::epoch_binds_namespace(e, &ns))
            .map(|e| e.to_entry())
            .collect())
    }

    /// List every epoch in the store, across all workspaces. For hub-level
    /// views that deliberately span workspaces; prefer [`Repository::list_epochs`].
    pub fn list_all_epochs(&self) -> Result<Vec<agentstategraph_core::EpochEntry>, RepoError> {
        let epochs = self.storage.list_epochs()?;
        Ok(epochs.iter().map(|e| e.to_entry()).collect())
    }

    /// Get a specific epoch by ID.
    pub fn get_epoch(&self, id: &str) -> Result<agentstategraph_core::Epoch, RepoError> {
        self.storage
            .get_epoch(id)?
            .ok_or_else(|| RepoError::RefNotFound(format!("epoch not found: {}", id)))
    }

    /// Transition a sealed epoch to Archived.
    pub fn archive_epoch(&self, id: &str) -> Result<(), RepoError> {
        self.storage.archive_epoch(id).map_err(RepoError::Storage)
    }

    /// Export a sealed or archived epoch as a self-contained JSON audit bundle.
    ///
    /// The bundle includes the epoch metadata and the full Commit records for
    /// every commit associated with the epoch, making it independently
    /// verifiable without access to the live store.
    pub fn export_epoch(&self, id: &str) -> Result<serde_json::Value, RepoError> {
        let epoch = self.get_epoch(id)?;
        if epoch.status == agentstategraph_core::EpochStatus::Active {
            return Err(RepoError::InvalidOperation(
                "cannot export an active epoch; seal it first".to_string(),
            ));
        }
        // Prefer the epoch's own tagged commits — that is the precise set, and
        // it is rehydrated from the commits table on load. An epoch that was
        // never made active tags nothing, though, which is exactly how a
        // checkpoint-style epoch is minted (create + seal in one step). Its
        // bundle would otherwise ship zero commit records while still claiming
        // to be self-contained, so fall back to the set recorded at seal time.
        let commit_ids = if epoch.commits.is_empty() {
            &epoch.sealed_commits
        } else {
            &epoch.commits
        };
        let mut commits = Vec::new();
        for cid in commit_ids {
            if let Some(commit) = self.storage.get_commit(cid)? {
                commits.push(serde_json::to_value(&commit).map_err(|e| {
                    RepoError::InvalidOperation(format!("serialize commit: {}", e))
                })?);
            }
        }
        let bundle = serde_json::json!({
            "agentstategraph_export_version": 1,
            "epoch": serde_json::to_value(&epoch).map_err(|e| {
                RepoError::InvalidOperation(format!("serialize epoch: {}", e))
            })?,
            "commits": commits,
            "exported_at": chrono::Utc::now().to_rfc3339(),
        });
        Ok(bundle)
    }

    // -----------------------------------------------------------------------
    // Explorer / viewer APIs (0.4.0)
    // -----------------------------------------------------------------------

    /// List all leaf paths in the state tree under a prefix.
    /// Use "/" or "" for all paths. max_depth limits recursion (default 50).
    pub fn list_paths(
        &self,
        ref_name: &str,
        prefix: &str,
        max_depth: Option<usize>,
    ) -> Result<Vec<String>, RepoError> {
        if path_is_secret(prefix) {
            return Err(RepoError::ReservedPath(prefix.to_string()));
        }
        let commit_id = self.resolve_ref(ref_name)?;
        let commit = self
            .storage
            .get_commit(&commit_id)?
            .ok_or_else(|| RepoError::RefNotFound(ref_name.to_string()))?;
        let depth = max_depth.unwrap_or(50);
        let mut paths =
            tree::tree_list_paths(self.storage.as_ref(), &commit.state_root, prefix, depth)?;
        // Filter out any results that land under the secret sub-prefix —
        // a broader prefix like "/_meta" or "/" must not leak secret names.
        paths.retain(|p| !path_is_secret(p));
        Ok(paths)
    }

    /// Get an entire subtree as nested JSON. Batch alternative to N×get calls.
    pub fn get_tree(&self, ref_name: &str, prefix: &str) -> Result<serde_json::Value, RepoError> {
        // get_json already handles this — just delegate with the prefix as path
        let path = if prefix.is_empty() { "/" } else { prefix };
        self.get_json(ref_name, path)
    }

    /// Search state values for a query string. Returns matching (path, value) pairs.
    pub fn search_values(
        &self,
        ref_name: &str,
        query: &str,
        max_results: Option<usize>,
    ) -> Result<Vec<(String, String)>, RepoError> {
        let commit_id = self.resolve_ref(ref_name)?;
        let commit = self
            .storage
            .get_commit(&commit_id)?
            .ok_or_else(|| RepoError::RefNotFound(ref_name.to_string()))?;
        let limit = max_results.unwrap_or(50);

        // Leaf-value index fast path (plan perf-slow-endpoints t-006). The index
        // is keyed by (namespace, ref) and maintained incrementally on the write
        // path (see `guarded_set_ref`). The first search of a ref after the
        // feature is enabled backfills it once (a full walk of the current
        // tree); every search after that — and after every write — is a trigram
        // substring probe. Backends without an index return `None` and we fall
        // through to the un-indexed tree walk unchanged.
        //
        // The index over-fetches (`limit * 4`, capped) before the secret filter
        // so dropping secret rows can't starve the result below `limit`; the
        // walk applies the same filter after its own cap.
        let ns = self.active_namespace()?;
        let ns_str = ns.as_str();
        let query_lower = query.to_lowercase();
        let index_fetch = limit.saturating_mul(4).min(tree::LIST_PATHS_MAX_RESULTS);

        let current = if self.storage.leaf_index_tracks_roots() {
            self.leaf_index_catch_up(&ns, ref_name, &commit.state_root)?
        } else {
            if !self.storage.leaf_index_is_built(ns_str, ref_name)? {
                // One-time backfill. Non-fatal on error — we simply fall back to
                // the tree walk below and can retry the backfill on a later search.
                if let Ok(leaves) =
                    tree::tree_collect_leaves(self.storage.as_ref(), &commit.state_root)
                {
                    let _ = self.storage.leaf_index_build(ns_str, ref_name, &leaves);
                }
            }
            true
        };

        if current
            && let Some(mut hits) =
                self.storage
                    .leaf_index_search(ns_str, ref_name, &query_lower, index_fetch)?
        {
            hits.retain(|(path, _)| !path_is_secret(path));
            hits.truncate(limit);
            return Ok(hits);
        }

        // Fallback: un-indexed tree walk (backend has no leaf index).
        let mut results =
            tree::tree_search_values(self.storage.as_ref(), &commit.state_root, query, limit)?;
        // Never surface values from the secret sub-prefix.
        results.retain(|(path, _)| !path_is_secret(path));
        Ok(results)
    }

    /// Bring the `(namespace, ref)` leaf set to the tree at `root`, for a
    /// backend that records which tree each set reflects, and say whether the
    /// set can answer for it. An unbuilt set (or one built before roots were
    /// recorded) is built; one at another tree is moved by the diff from it,
    /// which repairs a write whose own update was skipped or lost a race. A
    /// set already at the ref's newest tree, because a write landed after
    /// `root` was read, answers as it is rather than being moved back.
    /// Best-effort like all index upkeep: on `false` the caller walks the
    /// tree instead.
    fn leaf_index_catch_up(
        &self,
        ns: &Namespace,
        ref_name: &str,
        root: &ObjectId,
    ) -> Result<bool, RepoError> {
        let store = self.storage.as_ref();
        let namespace = ns.as_str();
        match store.leaf_index_root(namespace, ref_name)? {
            Some(indexed) if indexed == *root => return Ok(true),
            Some(indexed) if self.ref_state_root(ns, ref_name)? == Some(indexed) => {
                return Ok(true);
            }
            Some(indexed) => {
                if let Ok((removed, added)) = tree::tree_diff_leaves(store, Some(&indexed), root) {
                    let _ = store
                        .leaf_index_advance(namespace, ref_name, &indexed, root, &removed, &added);
                }
            }
            None => {
                if let Ok(leaves) = tree::tree_collect_leaves(store, root) {
                    let _ = store.leaf_index_build_at(namespace, ref_name, root, &leaves);
                }
            }
        }
        // Another writer may have moved the set meanwhile; use it only if it
        // now reflects this tree.
        Ok(store.leaf_index_root(namespace, ref_name)? == Some(*root))
    }

    /// The state root `ref_name` points at now, if it is a stored ref.
    fn ref_state_root(
        &self,
        ns: &Namespace,
        ref_name: &str,
    ) -> Result<Option<ObjectId>, RepoError> {
        let Some(head) = self.storage.get_ref(ns, ref_name)? else {
            return Ok(None);
        };
        Ok(self.storage.get_commit(&head)?.map(|c| c.state_root))
    }

    /// Get summary statistics for a ref.
    pub fn stats(&self, ref_name: &str) -> Result<serde_json::Value, RepoError> {
        // Counted the first-parent line only, so every figure here undercounted
        // by whatever the merged-in branches held.
        let commits = self.log_dag(ref_name, 10000)?.commits;
        let branches = self.list_branches(None)?;
        let paths = self.list_paths(ref_name, "/", Some(100))?;
        let epochs = self.list_epochs()?;

        // Collect unique agent IDs
        let mut agents: Vec<String> = commits
            .iter()
            .map(|c| c.agent_id.clone())
            .collect::<std::collections::HashSet<_>>()
            .into_iter()
            .collect();
        agents.sort();

        // Collect unique intent categories
        let mut categories: Vec<String> = commits
            .iter()
            .map(|c| format!("{:?}", c.intent.category))
            .collect::<std::collections::HashSet<_>>()
            .into_iter()
            .collect();
        categories.sort();

        Ok(serde_json::json!({
            "commit_count": commits.len(),
            "branch_count": branches.len(),
            "path_count": paths.len(),
            "epoch_count": epochs.len(),
            "agents": agents,
            "categories": categories,
            "latest_commit": commits.first().map(|c| serde_json::json!({
                "id": c.id.short(),
                "agent": c.agent_id,
                "intent": c.intent.description,
                "timestamp": c.timestamp.to_rfc3339(),
            })),
        }))
    }

    /// Get commit graph (DAG) for visualization.
    /// Returns commit nodes with their parents and metadata.
    /// Bring the distilled project-history tables (`asg_history_*`) up to date
    /// with the commit chain (Plan A t-001). Folds every commit newer than the
    /// stored cursor into the rollup/milestone tables in `batch_size` chunks,
    /// so a 512k-commit store is processed in bounded memory. Incremental: on a
    /// caught-up store this is a cheap no-op; after new commits it processes
    /// only those. Returns how many commits were folded and the new cursor.
    pub fn extract_history(&self, batch_size: usize) -> Result<HistoryExtractReport, RepoError> {
        let span = op_span!(
            INFO,
            "asg.history.extract",
            asg.history.commits_processed = tracing::field::Empty,
        );
        let result = telemetry::run(span.clone(), || self.extract_history_inner(batch_size));
        if let Ok(report) = &result {
            span.record("asg.history.commits_processed", report.commits_processed);
        }
        result
    }

    fn extract_history_inner(&self, batch_size: usize) -> Result<HistoryExtractReport, RepoError> {
        let batch_size = batch_size.max(1);
        let mut total = 0usize;
        loop {
            let n = self.storage.history_extract_batch(batch_size)?;
            if n == 0 {
                break;
            }
            total += n;
        }
        Ok(HistoryExtractReport {
            commits_processed: total,
            cursor: self.storage.history_cursor()?,
        })
    }

    /// Read the commit-history rollup (see [`Repository::extract_history`]).
    pub fn history_rollup(&self) -> Result<Vec<HistoryRollupRow>, RepoError> {
        Ok(self.storage.history_rollup()?)
    }

    /// Read the milestone timeline, most recent first, capped at `limit`.
    pub fn history_milestones(&self, limit: usize) -> Result<Vec<HistoryMilestoneRow>, RepoError> {
        Ok(self.storage.history_milestones(limit)?)
    }

    /// Build the `asg history` report (Plan A t-002) from the distilled
    /// `asg_history_*` tables: commit velocity (by `day` or `week`), intent
    /// mix, authorship, and the milestone timeline. When `refresh` is set the
    /// extractor is brought current first, so the report always reflects the
    /// latest commits. `namespace`, when given, restricts every view to that
    /// namespace. The rollup is small (bounded by distinct day×ns×agent×category
    /// buckets), so aggregating it in memory is cheap even on a 512k-commit
    /// store. When `store` is set, a `store_shape` block (Plan A t-003:
    /// objects/commits/bytes/amplification) is attached.
    pub fn history_report(
        &self,
        namespace: Option<&str>,
        velocity_by: &str,
        milestone_limit: usize,
        refresh: bool,
        store: bool,
    ) -> Result<serde_json::Value, RepoError> {
        use std::collections::{BTreeMap, BTreeSet};

        if refresh {
            self.extract_history(DEFAULT_HISTORY_BATCH)?;
        }
        let by_week = velocity_by.eq_ignore_ascii_case("week");
        let period_label = if by_week { "week" } else { "day" };

        let mut rollup = self.history_rollup()?;
        let mut milestones = self.history_milestones(milestone_limit.max(1))?;
        if let Some(ns) = namespace {
            rollup.retain(|r| r.namespace == ns);
            milestones.retain(|m| m.namespace == ns);
        }

        let mut velocity: BTreeMap<String, i64> = BTreeMap::new();
        let mut mix: BTreeMap<String, i64> = BTreeMap::new();
        let mut authorship: BTreeMap<String, i64> = BTreeMap::new();
        let mut namespaces: BTreeSet<String> = BTreeSet::new();
        let mut total_commits = 0i64;
        for r in &rollup {
            total_commits += r.commit_count;
            let period = if by_week {
                iso_week_key(&r.day).unwrap_or_else(|| r.day.clone())
            } else {
                r.day.clone()
            };
            *velocity.entry(period).or_default() += r.commit_count;
            *mix.entry(r.intent_category.clone()).or_default() += r.commit_count;
            *authorship.entry(r.agent_id.clone()).or_default() += r.commit_count;
            namespaces.insert(r.namespace.clone());
        }

        // Velocity stays chronological (BTreeMap key order). Mix and authorship
        // are ranked by volume, ties broken by name for determinism.
        let velocity_series: Vec<serde_json::Value> = velocity
            .iter()
            .map(|(k, v)| serde_json::json!({ "period": k, "commits": v }))
            .collect();
        let rank = |m: BTreeMap<String, i64>, key: &str| -> Vec<serde_json::Value> {
            let mut v: Vec<(String, i64)> = m.into_iter().collect();
            v.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(&b.0)));
            v.into_iter()
                .map(|(k, c)| serde_json::json!({ key: k, "commits": c }))
                .collect()
        };
        let milestones_arr: Vec<serde_json::Value> = milestones
            .iter()
            .map(|m| {
                serde_json::json!({
                    "commit": m.commit_id.short(),
                    "kind": m.kind,
                    "timestamp": m.timestamp,
                    "day": m.day,
                    "namespace": m.namespace,
                    "agent": m.agent_id,
                    "description": m.description,
                    "state_root": m.state_root.as_ref().map(|r| r.short()),
                    "git_sha": m.git_sha,
                })
            })
            .collect();

        let mut report = serde_json::json!({
            "totals": {
                "commits": total_commits,
                "periods": velocity_series.len(),
                "agents": authorship.len(),
                "cursor": self.storage.history_cursor()?,
            },
            "velocity": { "by": period_label, "series": velocity_series },
            "intent_mix": rank(mix, "category"),
            "authorship": rank(authorship, "agent"),
            "namespaces": namespaces.into_iter().collect::<Vec<_>>(),
            "milestones": milestones_arr,
        });
        if store {
            if let Some(obj) = report.as_object_mut() {
                obj.insert("store_shape".to_string(), self.history_store_shape()?);
            }
        }
        Ok(report)
    }

    /// The store's object and commit counts and its size on disk, for the
    /// store gauges (DESIGN.md Plan H t-008). Counting reads every row's key,
    /// so ask on a slow interval. `None` from a backend that doesn't count.
    pub fn store_totals(&self) -> Result<Option<StoreTotals>, RepoError> {
        Ok(self.storage.store_totals()?)
    }

    /// How many commits aren't yet distilled into the history tables: the
    /// history extractor's lag.
    pub fn undistilled_commits(&self) -> Result<i64, RepoError> {
        Ok(self.storage.history_undistilled_commit_count()?)
    }

    /// Physical store-shape report (Plan A t-003): object/commit counts, total
    /// bytes, per-table bytes (when the SQLite build exposes `dbstat`), and the
    /// path-copy amplification (avg objects created per commit) — the evidence
    /// Plan B's GC uses to set retention thresholds.
    pub fn history_store_shape(&self) -> Result<serde_json::Value, RepoError> {
        let s = self.storage.history_store_shape()?;
        let amplification = if s.commits > 0 {
            s.objects as f64 / s.commits as f64
        } else {
            0.0
        };
        let tables: Vec<serde_json::Value> = s
            .tables
            .iter()
            .map(|t| serde_json::json!({ "name": t.name, "bytes": t.bytes }))
            .collect();
        Ok(serde_json::json!({
            "objects": s.objects,
            "commits": s.commits,
            "total_bytes": s.total_bytes,
            "path_copy_amplification": { "objects_per_commit": amplification },
            "dbstat_available": s.dbstat_available,
            "tables": tables,
        }))
    }

    /// Retention hook for Plan B's GC (Plan A t-005): whether `id`'s commit has
    /// been folded into the history tables. GC checks this before pruning a
    /// commit's raw snapshot — a distilled commit's signal already survives in
    /// the metric tables, so the snapshot is safe to reclaim.
    pub fn history_is_distilled(&self, id: &ObjectId) -> Result<bool, RepoError> {
        Ok(self.storage.history_is_commit_distilled(id)?)
    }

    /// State roots the GC must keep reachable (Plan A t-005): the snapshots
    /// preserved by recorded milestones — the human-meaningful spine that
    /// survives pruning.
    pub fn history_retained_state_roots(&self) -> Result<Vec<ObjectId>, RepoError> {
        Ok(self.storage.history_retained_state_roots()?)
    }

    /// Stop pinning the snapshots of milestones whose checkpoint never asked to
    /// be pinned — every row distilled before v1.2.2, when the extractor pinned
    /// all checkpoints. Returns how many rows were unpinned.
    ///
    /// Run once on a store that predates opt-in pinning; after it a sweep can
    /// reclaim those snapshots. It deletes nothing itself. Checkpoints tagged
    /// `TAG_PIN_STATE` keep their pin.
    pub fn history_unpin_legacy_milestones(&self) -> Result<usize, RepoError> {
        Ok(self.storage.history_unpin_legacy_milestones()?)
    }

    /// The default GC keep-set (Plan B t-001): the `state_root` of every live
    /// ref tip across all namespaces, plus the retained milestone snapshots.
    /// This is what "kept alive" means for the baseline reclaimable estimate —
    /// current state + the human-meaningful spine. (A retention policy that
    /// prunes historical commits, Plan B t-003, feeds a reduced root set.)
    pub fn gc_default_roots(&self) -> Result<Vec<ObjectId>, RepoError> {
        let mut set = std::collections::BTreeSet::new();
        for ns in self.storage.list_namespaces()? {
            for (_name, target) in self.storage.list_refs(&ns, "")? {
                if let Some(commit) = self.storage.get_commit(&target)? {
                    set.insert(commit.state_root);
                }
            }
        }
        set.extend(self.history_retained_state_roots()?);
        Ok(set.into_iter().collect())
    }

    /// GC reachability report over an explicit set of root state roots
    /// (Plan B t-001): marks the live object closure in bounded memory and
    /// returns live/total/reclaimable as JSON.
    pub fn gc_reachability_from(&self, roots: &[ObjectId]) -> Result<serde_json::Value, RepoError> {
        let r = self.storage.history_gc_reachability(roots, GC_MARK_BATCH)?;
        let pct = if r.total_objects > 0 {
            r.reclaimable_objects as f64 / r.total_objects as f64 * 100.0
        } else {
            0.0
        };
        Ok(serde_json::json!({
            "total_objects": r.total_objects,
            "live_objects": r.live_objects,
            "reclaimable_objects": r.reclaimable_objects,
            "reclaimable_pct": pct,
            "roots_walked": r.roots_walked,
        }))
    }

    /// GC reachability report over the default keep-set
    /// ([`Repository::gc_default_roots`]).
    pub fn gc_reachability_report(&self) -> Result<serde_json::Value, RepoError> {
        let roots = self.gc_default_roots()?;
        self.gc_reachability_from(&roots)
    }

    /// `asg gc --dry-run` (Plan B t-002): report what a GC would reclaim —
    /// object counts, estimated reclaimable bytes, estimated post-vacuum size —
    /// WITHOUT mutating anything. The safe default.
    ///
    /// Loud about safety: reclaiming objects unreachable from the keep-set
    /// makes the historical commits behind them unmaterializable, so their
    /// signal must already be distilled (Plan A). If the extractor is behind
    /// (`undistilled_commits > 0`), `safe` is false and the report says to run
    /// the extractor first — it never silently recommends dropping
    /// not-yet-captured history.
    pub fn gc_dry_run(&self) -> Result<serde_json::Value, RepoError> {
        telemetry::run(op_span!(INFO, "asg.gc.dry_run"), || self.gc_dry_run_inner())
    }

    fn gc_dry_run_inner(&self) -> Result<serde_json::Value, RepoError> {
        let roots = self.gc_default_roots()?;
        let reach = self
            .storage
            .history_gc_reachability(&roots, GC_MARK_BATCH)?;
        let shape = self.storage.history_store_shape()?;
        let undistilled = self.storage.history_undistilled_commit_count()?;

        // Estimate reclaimable bytes as the reclaimable fraction of the objects
        // table (where ~all the bytes live). Exact bytes need a post-sweep
        // vacuum; this is the dry-run estimate.
        let objects_bytes = shape
            .tables
            .iter()
            .find(|t| t.name == "objects")
            .map(|t| t.bytes)
            .unwrap_or(shape.total_bytes);
        let recl_frac = if reach.total_objects > 0 {
            reach.reclaimable_objects as f64 / reach.total_objects as f64
        } else {
            0.0
        };
        let est_reclaimable_bytes = (objects_bytes as f64 * recl_frac) as i64;
        let est_post_vacuum_bytes = (shape.total_bytes - est_reclaimable_bytes).max(0);
        let safe = undistilled == 0;

        Ok(serde_json::json!({
            "dry_run": true,
            "mutates": false,
            "objects": {
                "total": reach.total_objects,
                "live": reach.live_objects,
                "reclaimable": reach.reclaimable_objects,
                "reclaimable_pct": recl_frac * 100.0,
            },
            "commits": { "total": shape.commits },
            "bytes": {
                "current_total": shape.total_bytes,
                "estimated_reclaimable": est_reclaimable_bytes,
                "estimated_post_vacuum": est_post_vacuum_bytes,
                "dbstat_available": shape.dbstat_available,
            },
            "roots_walked": reach.roots_walked,
            "safe": safe,
            "safety": {
                "undistilled_commits": undistilled,
                "reason": if safe {
                    "all commits distilled — reclaimable objects' signal is captured in the history tables".to_string()
                } else {
                    format!(
                        "REFUSING to recommend a sweep: {undistilled} commit(s) are not yet distilled; run the history extractor first or their signal would be lost when their state is reclaimed"
                    )
                },
            },
        }))
    }

    /// The keep-set a [`RetentionPolicy`] resolves to (Plan B t-003): the
    /// `state_root`s the GC must keep reachable. Always includes every ref tip
    /// (current state) and every sealed-epoch commit (the epoch-seal invariant —
    /// sealed history must stay materializable); adds milestones and the
    /// recent/checkpoint selection per the policy.
    pub fn gc_keep_roots(&self, policy: RetentionPolicy) -> Result<Vec<ObjectId>, RepoError> {
        let mut set = std::collections::BTreeSet::new();

        // Current state — never prune a live ref.
        for ns in self.storage.list_namespaces()? {
            for (_name, target) in self.storage.list_refs(&ns, "")? {
                if let Some(c) = self.storage.get_commit(&target)? {
                    set.insert(c.state_root);
                }
            }
        }
        // Sealed-epoch commits must stay materializable (epoch-seal invariant)
        // — every namespace's, since the sweep covers the whole store. Listing
        // only this repository's namespace let a sweep run from one workspace
        // delete another's sealed history.
        for epoch in self.storage.list_epochs()? {
            if epoch.status != agentstategraph_core::EpochStatus::Sealed
                && epoch.status != agentstategraph_core::EpochStatus::Archived
            {
                continue;
            }
            for cid in &epoch.sealed_commits {
                if let Some(c) = self.storage.get_commit(cid)? {
                    set.insert(c.state_root);
                }
            }
        }
        // Each branch's fork point from its namespace's `main` — the merge base
        // a later merge needs. Once swept, `merge` could not read the base and
        // (before it learned to fail) moved the target to the source, orphaning
        // the target's own work.
        for ns in self.storage.list_namespaces()? {
            let refs = self.storage.list_refs(&ns, "")?;
            let Some(main) = refs
                .iter()
                .find(|(name, _)| name == "main")
                .map(|(_, t)| *t)
            else {
                continue;
            };
            for (name, head) in &refs {
                if name == "main" {
                    continue;
                }
                // Disjoint histories have no fork point to keep.
                if let Ok(base) = self.find_common_ancestor(head, &main)
                    && let Some(c) = self.storage.get_commit(&base)?
                {
                    set.insert(c.state_root);
                }
            }
        }
        // Open speculations' not-yet-committed trees, in any process.
        set.extend(self.storage.gc_pinned_roots(GC_PIN_MAX_AGE)?);
        // The human-meaningful spine.
        if policy.keep_milestones {
            set.extend(self.history_retained_state_roots()?);
        }
        // Recent in full + every Kth older commit as a sparse checkpoint.
        for (i, state_root) in self
            .storage
            .commit_state_roots_recent_first()?
            .into_iter()
            .enumerate()
        {
            if i < policy.keep_recent
                || (policy.checkpoint_every > 0 && i % policy.checkpoint_every == 0)
            {
                set.insert(state_root);
            }
        }
        Ok(set.into_iter().collect())
    }

    /// Compact the store, returning freed pages to the OS (Plan B t-004). Run
    /// after a sweep — the sweep frees pages to the freelist, but the file only
    /// shrinks after a `VACUUM`. Returns bytes before/after/reclaimed.
    pub fn gc_vacuum(&self) -> Result<serde_json::Value, RepoError> {
        let span = op_span!(
            INFO,
            "asg.gc.vacuum",
            asg.gc.bytes_reclaimed = tracing::field::Empty,
        );
        telemetry::run(span.clone(), || {
            let v = self.storage.history_vacuum()?;
            span.record("asg.gc.bytes_reclaimed", v.bytes_reclaimed);
            Ok(serde_json::json!({
                "bytes_before": v.bytes_before,
                "bytes_after": v.bytes_after,
                "bytes_reclaimed": v.bytes_reclaimed,
            }))
        })
    }

    /// Run (or preview) a GC sweep under a [`RetentionPolicy`] (Plan B t-003).
    ///
    /// `mutate=false` (the safe default) previews what would be reclaimed and
    /// never touches the store. `mutate=true` DELETES every object unreachable
    /// from the policy's keep-set — but only after the Plan A safety gate:
    /// if any commit is not yet distilled it refuses (returns `refused:true`,
    /// no mutation), so pruning can't lose not-yet-captured history.
    pub fn gc_sweep(
        &self,
        policy: RetentionPolicy,
        mutate: bool,
        vacuum: bool,
    ) -> Result<serde_json::Value, RepoError> {
        let span = op_span!(
            INFO,
            "asg.gc.sweep",
            asg.gc.mutate = mutate,
            asg.gc.refused = tracing::field::Empty,
            asg.gc.objects_deleted = tracing::field::Empty,
        );
        let result = telemetry::run(span.clone(), || self.gc_sweep_inner(policy, mutate, vacuum));
        if let Ok(report) = &result {
            let refused = report.get("refused").and_then(|v| v.as_bool());
            if let Some(refused) = refused {
                span.record("asg.gc.refused", refused);
            }
            if mutate {
                telemetry::gc_swept(refused == Some(true));
            }
            if let Some(deleted) = report.get("objects_deleted").and_then(|v| v.as_i64()) {
                span.record("asg.gc.objects_deleted", deleted);
            }
        }
        result
    }

    fn gc_sweep_inner(
        &self,
        policy: RetentionPolicy,
        mutate: bool,
        vacuum: bool,
    ) -> Result<serde_json::Value, RepoError> {
        let policy_json = serde_json::json!({
            "keep_recent": policy.keep_recent,
            "checkpoint_every": policy.checkpoint_every,
            "keep_milestones": policy.keep_milestones,
        });

        if !mutate {
            let keep = self.gc_keep_roots(policy)?;
            let undistilled = self.storage.history_undistilled_commit_count()?;
            let would = self.gc_reachability_from(&keep)?;
            return Ok(serde_json::json!({
                "dry_run": true,
                "mutated": false,
                "policy": policy_json,
                "keep_roots": keep.len(),
                "would_reclaim": would,
                "safe": undistilled == 0,
                "undistilled_commits": undistilled,
            }));
        }

        // Lock out other writers BEFORE computing the keep-set, and hold the
        // lock through the last delete. Otherwise a commit made on another
        // connection in between — a git hook re-indexing, a server — is absent
        // from the keep-set and its objects are deleted out from under it.
        let lock = GcWriterLock::acquire(self.storage.as_ref())?;
        let keep = self.gc_keep_roots(policy)?;
        let undistilled = self.storage.history_undistilled_commit_count()?;
        if undistilled > 0 {
            lock.release(false)?;
            return Ok(serde_json::json!({
                "dry_run": false,
                "mutated": false,
                "refused": true,
                "policy": policy_json,
                "undistilled_commits": undistilled,
                "reason": format!(
                    "REFUSING to sweep: {undistilled} commit(s) are not yet distilled; run the history extractor first"
                ),
            }));
        }
        let sweep = self.storage.history_gc_sweep(&keep, GC_MARK_BATCH)?;
        lock.release(true)?;

        let mut report = serde_json::json!({
            "dry_run": false,
            "mutated": true,
            "policy": policy_json,
            "keep_roots": keep.len(),
            "objects_before": sweep.objects_before,
            "objects_deleted": sweep.objects_deleted,
            "objects_after": sweep.objects_after,
        });
        // Only a VACUUM returns the freed pages to the OS (shrinks the file). It
        // cannot run inside a transaction, so it follows the released lock.
        if vacuum {
            let v = self.storage.history_vacuum()?;
            if let Some(obj) = report.as_object_mut() {
                obj.insert(
                    "vacuum".to_string(),
                    serde_json::json!({
                        "bytes_before": v.bytes_before,
                        "bytes_after": v.bytes_after,
                        "bytes_reclaimed": v.bytes_reclaimed,
                    }),
                );
            }
        }
        Ok(report)
    }

    pub fn commit_graph(
        &self,
        ref_name: &str,
        depth: usize,
    ) -> Result<Vec<serde_json::Value>, RepoError> {
        // Each node carries its `parents`, so a first-parent walk emitted edges
        // pointing at nodes absent from its own node set — measured, 4 nodes
        // describing a 7-node graph. A renderer then either drops those edges
        // or invents placeholder nodes. `depth` is caller-supplied, so the
        // bound stays explicit.
        let commits = self.log_dag(ref_name, depth)?.commits;
        let nodes: Vec<serde_json::Value> = commits
            .iter()
            .map(|c| {
                serde_json::json!({
                    "id": c.id.short(),
                    "full_id": c.id.to_string(),
                    "parents": c.parents.iter().map(|p| p.short()).collect::<Vec<_>>(),
                    "agent": c.agent_id,
                    "category": format!("{:?}", c.intent.category),
                    "description": c.intent.description,
                    "confidence": c.confidence,
                    "timestamp": c.timestamp.to_rfc3339(),
                    "is_merge": c.parents.len() > 1,
                })
            })
            .collect();
        Ok(nodes)
    }

    /// Build the intent-decomposition tree for a ref.
    ///
    /// Links intents by [`Intent::parent_intent`] — the sub-intent threading
    /// the RFC describes ("Sub-intent links to parent intent via
    /// `parent_intent`, forming a queryable tree"). Roots are the intents with
    /// no parent (or, when `root_commit_id` is given, that single commit's
    /// intent); a node's children are the commits whose `intent.parent_intent`
    /// equals this node's intent id.
    ///
    /// This is deliberately NOT a commit-lineage walk — for a tree structured
    /// by commit parents, use [`Repository::commit_graph`]. Until producers
    /// populate `parent_intent` (via [`Intent::with_parent`]), every intent is
    /// its own root and the result is a flat list — an honest "no threading
    /// recorded yet", not a lie about hierarchy that isn't there.
    pub fn intent_tree(
        &self,
        ref_name: &str,
        root_commit_id: Option<&str>,
    ) -> Result<serde_json::Value, RepoError> {
        let commits = self.log(ref_name, 10000)?;

        // Roots: an explicit commit's intent, else every intent with no parent.
        let roots: Vec<&Commit> = if let Some(root_id) = root_commit_id {
            commits.iter().filter(|c| c.id.short() == root_id).collect()
        } else {
            commits
                .iter()
                .filter(|c| c.intent.parent_intent.is_none())
                .collect()
        };

        // `visiting` carries the ancestor intent-ids on the current path so a
        // user-authored `parent_intent` cycle can't spin the recursion forever.
        fn build_intent_node(
            commit: &Commit,
            all_commits: &[Commit],
            visiting: &mut std::collections::HashSet<String>,
        ) -> serde_json::Value {
            let inserted = visiting.insert(commit.intent.id.clone());
            let children: Vec<serde_json::Value> = if inserted {
                all_commits
                    .iter()
                    .filter(|c| {
                        c.intent.parent_intent.as_deref() == Some(commit.intent.id.as_str())
                    })
                    .map(|child| build_intent_node(child, all_commits, visiting))
                    .collect()
            } else {
                // Cycle: this intent is already an ancestor on the path. Emit the
                // node without descending again.
                Vec::new()
            };
            if inserted {
                visiting.remove(&commit.intent.id);
            }

            serde_json::json!({
                "id": commit.id.short(),
                "intent_id": commit.intent.id,
                "agent": commit.agent_id,
                "category": format!("{:?}", commit.intent.category),
                "description": commit.intent.description,
                "reasoning": commit.reasoning,
                "confidence": commit.confidence,
                "timestamp": commit.timestamp.to_rfc3339(),
                "children": children,
            })
        }

        let tree: Vec<serde_json::Value> = roots
            .iter()
            .map(|c| {
                let mut visiting = std::collections::HashSet::new();
                build_intent_node(c, &commits, &mut visiting)
            })
            .collect();

        Ok(serde_json::json!({
            "roots": tree,
            "total_commits": commits.len(),
        }))
    }

    // -----------------------------------------------------------------------
    // Internal helpers
    // -----------------------------------------------------------------------

    /// Find the best common ancestor (merge base) of two commits.
    ///
    /// Walks the full commit DAG following EVERY parent (merge commits have
    /// two), not just the first — a first-parent-only walk misses the true
    /// lowest common ancestor once merge commits exist, and picking too old a
    /// base makes the target's own additions look like deletions to the merge
    /// engine. Among all common ancestors, returns the deepest one (closest to
    /// the two heads); ties are broken deterministically by commit id.
    fn find_common_ancestor(&self, a: &ObjectId, b: &ObjectId) -> Result<ObjectId, RepoError> {
        if a == b {
            return Ok(*a);
        }
        // Linear in the two histories: every commit is read from storage once,
        // and generation depth is computed for all of them in one pass. This
        // used to recompute depth from scratch for each common ancestor,
        // re-reading every commit on every pass of a fixed-point loop — O(n³)
        // reads. A CTX branch with 1,304 common ancestors never finished, and
        // because each read holds the storage connection lock it starved every
        // other request to the store.
        let parents = self.load_ancestry(&[*a, *b])?;
        deepest_common_ancestor(&parents, a, b).ok_or_else(|| {
            // Disjoint histories share no ancestor — refuse rather than
            // silently merging against an empty/arbitrary base.
            RepoError::RefNotFound("no common ancestor between refs".to_string())
        })
    }

    /// The parents of every commit reachable from `roots`, following every
    /// parent, with each commit read from storage exactly once.
    ///
    /// A parent whose commit is absent from the store — pruned, or never
    /// fetched — is included with no parents of its own, so it counts as a
    /// root at generation 0.
    fn load_ancestry(
        &self,
        roots: &[ObjectId],
    ) -> Result<std::collections::HashMap<ObjectId, Vec<ObjectId>>, RepoError> {
        let mut parents: std::collections::HashMap<ObjectId, Vec<ObjectId>> =
            std::collections::HashMap::new();
        let mut stack: Vec<ObjectId> = roots.to_vec();
        while let Some(id) = stack.pop() {
            if parents.contains_key(&id) {
                continue;
            }
            let ps = self
                .storage
                .get_commit(&id)?
                .map(|c| c.parents)
                .unwrap_or_default();
            stack.extend(ps.iter().filter(|p| !parents.contains_key(*p)));
            parents.insert(id, ps);
        }
        Ok(parents)
    }

    /// Walk the state tree from `root`, returning the id of the first object
    /// that is referenced but absent from the store, or `None` if the whole
    /// tree is fully readable. Iterative with a visited-set so shared subtrees
    /// and cycles (content-addressing makes true cycles impossible, but the
    /// guard is cheap) are each visited once. Used as the pre-ref-advance
    /// integrity gate for merges.
    /// Like [`Self::first_missing_reachable`], but only checks what `new_root`
    /// adds over `base_root`: a subtree whose id is unchanged from the base is
    /// committed state and is skipped. Cost tracks the size of the change, not
    /// the store — cheap enough to gate every speculation commit.
    fn first_missing_new(
        &self,
        base_root: &ObjectId,
        new_root: &ObjectId,
    ) -> Result<Option<ObjectId>, RepoError> {
        use agentstategraph_core::Node;
        let mut seen = std::collections::HashSet::new();
        let mut stack = vec![(*new_root, Some(*base_root))];
        while let Some((id, base)) = stack.pop() {
            if Some(id) == base || !seen.insert(id) {
                continue;
            }
            let Some(obj) = self.storage.get_object(&id)? else {
                return Ok(Some(id));
            };
            let Object::Node(node) = obj else { continue };
            match node {
                Node::Map(entries) => {
                    let base_entries = match base
                        .map(|b| self.storage.get_object(&b))
                        .transpose()?
                        .flatten()
                    {
                        Some(Object::Node(Node::Map(m))) => m,
                        _ => Default::default(),
                    };
                    for (k, child) in entries {
                        stack.push((child, base_entries.get(&k).copied()));
                    }
                }
                Node::List(items) | Node::Set(items) => {
                    stack.extend(items.into_iter().map(|c| (c, None)));
                }
            }
        }
        Ok(None)
    }

    fn first_missing_reachable(&self, root: &ObjectId) -> Result<Option<ObjectId>, RepoError> {
        use agentstategraph_core::Node;
        let mut seen = std::collections::HashSet::new();
        let mut stack = vec![*root];
        while let Some(id) = stack.pop() {
            if !seen.insert(id) {
                continue;
            }
            match self.storage.get_object(&id)? {
                None => return Ok(Some(id)),
                Some(Object::Node(node)) => match node {
                    Node::Map(entries) => stack.extend(entries.values().copied()),
                    Node::List(items) | Node::Set(items) => stack.extend(items.iter().copied()),
                },
                Some(Object::Atom(_)) => {}
            }
        }
        Ok(None)
    }

    /// Integrity check for tooling (e.g. `ctx db fsck`): the id of the first
    /// object referenced by `commit`'s state tree that is missing from the
    /// store, or `None` if the commit's tree is fully readable. Shares the
    /// exact walk used by the merge integrity gate.
    pub fn first_missing_object(
        &self,
        commit_id: &ObjectId,
    ) -> Result<Option<ObjectId>, RepoError> {
        let commit = self
            .storage
            .get_commit(commit_id)?
            .ok_or_else(|| RepoError::CommitNotFound(commit_id.short()))?;
        self.first_missing_reachable(&commit.state_root)
    }

    /// The nearest commit at or before `commit_id` (walking first-parent) whose
    /// state tree is fully readable — the safe rewind target when a ref points
    /// at a commit with a dangling tree. Returns `None` if no ancestor in the
    /// chain is intact (e.g. corruption at the root of history). Because no GC
    /// ever deletes objects, an intact ancestor's tree stays intact.
    pub fn nearest_readable_ancestor(
        &self,
        commit_id: &ObjectId,
    ) -> Result<Option<ObjectId>, RepoError> {
        let mut current = Some(*commit_id);
        while let Some(id) = current {
            match self.storage.get_commit(&id)? {
                None => return Ok(None),
                Some(commit) => {
                    if self.first_missing_reachable(&commit.state_root)?.is_none() {
                        return Ok(Some(id));
                    }
                    current = commit.parents.first().copied();
                }
            }
        }
        Ok(None)
    }

    /// Walk the commit DAG backwards from `head`, returning every reachable
    /// commit id (including `head` itself). Follows every parent — merge
    /// commits include both sides.
    fn reachable_commits_from(&self, head: &ObjectId) -> Result<Vec<ObjectId>, RepoError> {
        let mut seen = std::collections::HashSet::new();
        let mut stack = vec![*head];
        while let Some(id) = stack.pop() {
            if !seen.insert(id) {
                continue;
            }
            if let Some(commit) = self.storage.get_commit(&id)? {
                for p in &commit.parents {
                    if !seen.contains(p) {
                        stack.push(*p);
                    }
                }
            }
        }
        Ok(seen.into_iter().collect())
    }

    /// Given a proposed new ref target, return one `EpochViolation` per
    /// sealed epoch whose `sealed_commits` include any commit that is NOT
    /// reachable from the new target.
    ///
    /// An empty result means the proposed update preserves every seal.
    /// (security threat model v3+, V8)
    pub fn check_epoch_seal_violations(
        &self,
        new_ref_target: &ObjectId,
    ) -> Result<Vec<EpochViolation>, RepoError> {
        let ns = self.active_namespace()?;
        let binding: Vec<_> = self
            .storage
            .list_epochs()?
            .into_iter()
            .filter(|epoch| {
                (epoch.status == agentstategraph_core::EpochStatus::Sealed
                    || epoch.status == agentstategraph_core::EpochStatus::Archived)
                    // A seal binds only its own workspace — see `epoch_binds_namespace`.
                    && Self::epoch_binds_namespace(epoch, &ns)
            })
            .collect();
        // Nothing sealed here, so nothing a ref move could orphan. Checked
        // before the reachability walk, which visits the whole commit DAG:
        // on a ~1M-commit store with no epochs at all that walk made every
        // write take ~3s — and a 3s read-modify-write window is what let
        // concurrent writers lose each other's commits.
        if binding.is_empty() {
            return Ok(Vec::new());
        }
        let reachable: std::collections::HashSet<ObjectId> = self
            .reachable_commits_from(new_ref_target)?
            .into_iter()
            .collect();
        let mut violations = Vec::new();
        for epoch in binding.iter() {
            let unreachable: Vec<ObjectId> = epoch
                .sealed_commits
                .iter()
                .copied()
                .filter(|c| !reachable.contains(c))
                .collect();
            if !unreachable.is_empty() {
                violations.push(EpochViolation {
                    epoch_id: epoch.id.clone(),
                    unreachable_commits: unreachable,
                });
            }
        }
        Ok(violations)
    }

    /// Enforce epoch seals against a proposed `set_ref` to `ref_name`.
    /// In strict mode (default) returns `RepoError::EpochSealViolated`. In
    /// warn mode (opt-out) logs to stderr and proceeds.
    fn enforce_epoch_seals(&self, ref_name: &str, new_target: &ObjectId) -> Result<(), RepoError> {
        // Only `main` is the canonical seal anchor; branch/spec updates
        // don't orphan sealed commits reachable from main because main
        // itself hasn't moved. We still check on `main` mutations and on
        // explicit low-level set_ref for any ref (cheap insurance).
        let violations = self.check_epoch_seal_violations(new_target)?;
        if violations.is_empty() {
            return Ok(());
        }
        // Prefer main-only enforcement: a side-branch can legitimately sit
        // on a state that doesn't include every sealed commit. Skip unless
        // the ref being mutated is `main`.
        if ref_name != "main" {
            return Ok(());
        }
        if self.epoch_seal_strict {
            let v = violations.into_iter().next().expect("non-empty");
            return Err(RepoError::EpochSealViolated {
                epoch_id: v.epoch_id,
                unreachable_commits: v.unreachable_commits,
            });
        }
        for v in &violations {
            eprintln!(
                "[WARN] asg epoch seal: ref '{}' → {} would orphan {} commit(s) from sealed epoch '{}' (e.g. {})",
                ref_name,
                new_target.short(),
                v.unreachable_commits.len(),
                v.epoch_id,
                v.unreachable_commits
                    .first()
                    .map(|c| c.short())
                    .unwrap_or_default(),
            );
        }
        Ok(())
    }

    /// Wrapper around `storage.set_ref` that first runs the epoch-seal
    /// enforcement check. Every internal code path that moves a ref goes
    /// through this.
    fn guarded_set_ref(&self, ref_name: &str, new_target: ObjectId) -> Result<(), RepoError> {
        let ns = self.active_namespace()?;
        self.enforce_epoch_seals(ref_name, &new_target)?;
        // Capture the outgoing state root before the ref moves, so the leaf
        // index can be updated with just the delta this write introduces.
        let old_root = self
            .storage
            .get_ref(&ns, ref_name)
            .ok()
            .flatten()
            .and_then(|cid| self.storage.get_commit(&cid).ok().flatten())
            .map(|c| c.state_root);
        self.storage.set_ref(&ns, ref_name, new_target)?;
        self.update_leaf_index(&ns, ref_name, old_root.as_ref(), &new_target);
        Ok(())
    }

    /// Compare-and-swap counterpart of `guarded_set_ref`: move `ref_name`
    /// from `expected` to `new_target` only if no other writer moved it
    /// first. Returns `Ok(false)` when it had moved; the caller must rebuild
    /// on the new head rather than retry the same commit.
    ///
    /// Every read-modify-write of a ref goes through here. An unconditional
    /// move built on a stale head discards every commit that landed since —
    /// silently, because each of those writers already got `Ok`. That lost
    /// ~14% of one AgentStateDeveloper store's ledger under a handful of
    /// concurrent writer processes.
    fn advance_ref(
        &self,
        ref_name: &str,
        expected: ObjectId,
        new_target: ObjectId,
    ) -> Result<bool, RepoError> {
        let ns = self.active_namespace()?;
        self.enforce_epoch_seals(ref_name, &new_target)?;
        let old_root = self
            .storage
            .get_commit(&expected)
            .ok()
            .flatten()
            .map(|c| c.state_root);
        if !self.storage.cas_ref(&ns, ref_name, expected, new_target)? {
            return Ok(false);
        }
        self.update_leaf_index(&ns, ref_name, old_root.as_ref(), &new_target);
        self.commit_landed(&ns, ref_name, &new_target);
        Ok(true)
    }

    /// Emit `asg.commit` for a commit that landed on `ref_name`, when
    /// [`crate::emit_commit_events`] turned the events on. Reading the commit
    /// back is best-effort: an event is never worth failing a write for.
    fn commit_landed(&self, ns: &Namespace, ref_name: &str, commit_id: &ObjectId) {
        if !telemetry::commit_events_enabled() {
            return;
        }
        if let Ok(Some(commit)) = self.storage.get_commit(commit_id) {
            telemetry::commit_landed(ns.as_str(), ref_name, &commit);
        }
    }

    /// The commit a write to `ref_name` should build on, and whether
    /// `ref_name` is a stored branch. A ref-spec that resolves without being
    /// a branch (a commit hash) has nothing to compare-and-swap against, so
    /// `move_ref` keeps the historical unconditional move for it.
    fn head_for_write(&self, ref_name: &str) -> Result<(ObjectId, bool), RepoError> {
        let ns = self.active_namespace()?;
        match self.storage.get_ref(&ns, ref_name)? {
            Some(id) => Ok((id, true)),
            None => Ok((self.resolve_ref(ref_name)?, false)),
        }
    }

    /// Land a commit built on `expected` (from `head_for_write`). `Ok(false)`
    /// means another writer won the race and the caller must rebuild.
    fn move_ref(
        &self,
        ref_name: &str,
        expected: ObjectId,
        is_branch: bool,
        new_target: ObjectId,
    ) -> Result<bool, RepoError> {
        if is_branch {
            self.advance_ref(ref_name, expected, new_target)
        } else {
            self.guarded_set_ref(ref_name, new_target)?;
            self.commit_landed(&self.active_namespace()?, ref_name, &new_target);
            Ok(true)
        }
    }

    /// Incremental leaf-index maintenance (plan t-006). Best-effort: index
    /// upkeep must never fail a write, and it only runs for a set that has
    /// already been backfilled — otherwise the first search will build it
    /// from the current state anyway.
    ///
    /// A backend that records each set's tree applies the diff only if the
    /// set is still at `old_root`. Writers racing on one ref can reach here
    /// out of order, and an unconditional apply then leaves an overwritten
    /// value in the set. A skipped update leaves the set behind, never wrong;
    /// the next search catches it up.
    fn update_leaf_index(
        &self,
        ns: &Namespace,
        ref_name: &str,
        old_root: Option<&ObjectId>,
        new_target: &ObjectId,
    ) {
        let store = self.storage.as_ref();
        if !store
            .leaf_index_is_built(ns.as_str(), ref_name)
            .unwrap_or(false)
        {
            return;
        }
        let Ok(Some(new_commit)) = store.get_commit(new_target) else {
            return;
        };
        let new_root = new_commit.state_root;
        if store.leaf_index_tracks_roots() {
            if let Some(old_root) = old_root
                && let Ok((removed, added)) =
                    tree::tree_diff_leaves(store, Some(old_root), &new_root)
            {
                let _ = store.leaf_index_advance(
                    ns.as_str(),
                    ref_name,
                    old_root,
                    &new_root,
                    &removed,
                    &added,
                );
            }
        } else if let Ok((removed, added)) = tree::tree_diff_leaves(store, old_root, &new_root) {
            let _ = store.leaf_index_apply(ns.as_str(), ref_name, &removed, &added);
        }
    }

    /// Low-level: move a ref to a specific commit id, subject to epoch-seal
    /// enforcement. Used by migration tooling and by tests that need to
    /// simulate rewinds. Prefer `set`/`merge`/etc for normal writes.
    pub fn set_ref(&self, ref_name: &str, target: ObjectId) -> Result<(), RepoError> {
        self.guarded_set_ref(ref_name, target)
    }

    /// Resolve a ref name to a commit ID.
    /// Resolve a ref-spec to a commit id. Resolution order:
    ///   1. exact branch name
    ///   2. exact full commit hash (optionally `sg_`-prefixed)
    ///   3. unique `sg_`/hex commit-id prefix
    ///
    /// A value that is neither an existing branch nor hex-shaped yields
    /// `BranchNotFound`; a hex-shaped value that matches no commit yields
    /// `CommitNotFound`; a prefix matching more than one commit yields
    /// `AmbiguousCommitPrefix`.
    ///
    /// (Tags are reserved for step 2 in the spec but not yet stored, so they
    /// are skipped here.)
    fn resolve_ref(&self, ref_name: &str) -> Result<ObjectId, RepoError> {
        let ns = self.active_namespace()?;
        // 1. Exact branch name — always wins, even if it looks like hex.
        if let Some(id) = self.storage.get_ref(&ns, ref_name)? {
            return Ok(id);
        }
        if self.branch_refs_only {
            let commit_like = ObjectId::from_hex(ref_name).is_some()
                || ObjectId::normalize_hex_prefix(ref_name).is_some();
            return Err(if commit_like {
                RepoError::CommitRefNotAllowed(ref_name.to_string())
            } else {
                RepoError::BranchNotFound(ref_name.to_string())
            });
        }
        // 2 & 3. Commit hash / prefix.
        self.resolve_commit_ref(ref_name)
    }

    /// Resolve a value that is not a branch name as a commit id or id prefix.
    fn resolve_commit_ref(&self, ref_name: &str) -> Result<ObjectId, RepoError> {
        // Full 64-char hash (with or without sg_): direct existence check.
        if let Some(id) = ObjectId::from_hex(ref_name) {
            if self.storage.has_commit(&id)? {
                return Ok(id);
            }
            return Err(RepoError::CommitNotFound(ref_name.to_string()));
        }

        // Non-hex input is not a commit ref — report it as a missing branch,
        // matching the previous behaviour for ordinary bad names.
        let prefix = match ObjectId::normalize_hex_prefix(ref_name) {
            Some(p) => p,
            None => return Err(RepoError::BranchNotFound(ref_name.to_string())),
        };

        // Unique prefix over ALL commits (including orphaned/unreferenced ones,
        // which is the whole point of resolving historical ids).
        let mut matches: Vec<ObjectId> = self
            .storage
            .all_commit_ids()?
            .into_iter()
            .filter(|id| id.to_hex().starts_with(&prefix))
            .collect();
        matches.sort();
        matches.dedup();
        match matches.len() {
            0 => Err(RepoError::CommitNotFound(ref_name.to_string())),
            1 => Ok(matches[0]),
            n => Err(RepoError::AmbiguousCommitPrefix {
                prefix: ref_name.to_string(),
                count: n,
            }),
        }
    }

    /// Create a commit and store it.
    fn create_commit(
        &self,
        state_root: ObjectId,
        parents: Vec<ObjectId>,
        options: CommitOptions,
    ) -> Result<Commit, RepoError> {
        let mut builder = CommitBuilder::new(
            state_root,
            options.agent_id,
            options.authority,
            options.intent,
        )
        .parents(parents);

        if let Some(reasoning) = options.reasoning {
            builder = builder.reasoning(reasoning);
        }
        if let Some(confidence) = options.confidence {
            builder = builder.confidence(confidence);
        }
        if !options.tool_calls.is_empty() {
            builder = builder.tool_calls(options.tool_calls);
        }

        let commit = builder.build();
        // With its trace in the same write. If that fails, the commit is
        // stored on its own: losing the link mustn't fail the write.
        let trace = telemetry::current_trace();
        if let Err(e) = self.storage.put_commit_traced(&commit, trace.as_ref()) {
            if trace.is_none() {
                return Err(e.into());
            }
            tracing::warn!(error.type = e.code(), "couldn't record the commit's trace");
            self.storage.put_commit(&commit)?;
        }
        // Associate with the active epoch/session if any. Association
        // errors are surfaced — a sealed active epoch shouldn't silently
        // drop its mark. Callers can unset the active epoch before
        // retrying.
        if let Some(epoch_id) = self.active_epoch()? {
            self.storage.set_commit_epoch(&commit.id, &epoch_id)?;
        }
        if let Some(session_id) = self.active_session()? {
            self.storage.set_commit_session(&commit.id, &session_id)?;
        }
        Ok(commit)
    }

    /// The trace `commit_id` was made in, when [`crate::link_commits_to_traces`]
    /// recorded one.
    pub fn commit_trace(&self, commit_id: &ObjectId) -> Result<Option<CommitTrace>, RepoError> {
        Ok(self.storage.get_commit_trace(commit_id)?)
    }

    // -----------------------------------------------------------------------
    // Accessors used by the taint module (0.7.75 §4)
    // -----------------------------------------------------------------------

    /// Access the underlying Storage (read-only).
    pub(crate) fn taint_storage(&self) -> &dyn Storage {
        self.storage.as_ref()
    }

    /// Write a bare "intent-only" commit with no state-tree change —
    /// used by the taint / quarantine / watch family to stamp an
    /// audit commit without mutating state. Returns the new commit id.
    pub(crate) fn write_taint_intent(
        &self,
        ref_name: &str,
        category: IntentCategory,
        description: String,
        agent_id: &str,
        reasoning: Option<String>,
    ) -> Result<ObjectId, RepoError> {
        let mut options = CommitOptions::new(agent_id, category, description);
        if let Some(r) = reasoning {
            options = options.with_reasoning(r);
        }
        for attempt in 0..MAX_REF_CAS_ATTEMPTS {
            telemetry::cas_attempt("taint", attempt);
            let (parent_id, is_branch) = self.head_for_write(ref_name)?;
            let parent = self
                .storage
                .get_commit(&parent_id)?
                .ok_or_else(|| RepoError::RefNotFound(ref_name.to_string()))?;
            let commit = self.create_commit(parent.state_root, vec![parent_id], options.clone())?;
            if self.move_ref(ref_name, parent_id, is_branch, commit.id)? {
                return Ok(commit.id);
            }
        }
        telemetry::cas_gave_up("taint", MAX_REF_CAS_ATTEMPTS);
        Err(RepoError::WriteConflict)
    }
}

impl agentstategraph_reminders::ReminderStore for Repository {
    fn save(
        &self,
        reminder: &agentstategraph_reminders::Reminder,
    ) -> Result<(), agentstategraph_reminders::ReminderError> {
        self.storage.save(reminder)
    }

    fn get(
        &self,
        id: &str,
    ) -> Result<Option<agentstategraph_reminders::Reminder>, agentstategraph_reminders::ReminderError>
    {
        self.storage.get(id)
    }

    fn update(
        &self,
        reminder: &agentstategraph_reminders::Reminder,
    ) -> Result<(), agentstategraph_reminders::ReminderError> {
        self.storage.update(reminder)
    }

    fn update_if_unchanged(
        &self,
        expected: &agentstategraph_reminders::Reminder,
        new: &agentstategraph_reminders::Reminder,
    ) -> Result<bool, agentstategraph_reminders::ReminderError> {
        self.storage.update_if_unchanged(expected, new)
    }

    fn delete(&self, id: &str) -> Result<bool, agentstategraph_reminders::ReminderError> {
        self.storage.delete(id)
    }

    fn list(
        &self,
        filter: &agentstategraph_reminders::ReminderFilter,
    ) -> Result<Vec<agentstategraph_reminders::Reminder>, agentstategraph_reminders::ReminderError>
    {
        self.storage.list(filter)
    }
}

/// Bridge between storage backends and the diff engine's ObjectResolver trait.
struct StorageResolver<'a> {
    storage: &'a dyn Storage,
}

impl<'a> ObjectResolver for StorageResolver<'a> {
    fn resolve(&self, id: &ObjectId) -> Option<Object> {
        self.storage.get_object(id).ok().flatten()
    }
}

/// Among the common ancestors of `a` and `b` in `parents`, the one with the
/// greatest generation — closest to the two heads — with ties broken by the
/// larger commit id so the choice is reproducible. `None` for disjoint
/// histories.
///
/// The winner is always a *best* common ancestor (one that is not an ancestor
/// of another common ancestor): an ancestor of a commit has a strictly smaller
/// generation than that commit.
fn deepest_common_ancestor(
    parents: &std::collections::HashMap<ObjectId, Vec<ObjectId>>,
    a: &ObjectId,
    b: &ObjectId,
) -> Option<ObjectId> {
    let of_a = reachable_in(parents, a);
    let of_b = reachable_in(parents, b);
    let generation = generations(parents);
    of_a.intersection(&of_b)
        .map(|id| (generation.get(id).copied().unwrap_or(0), *id))
        .max()
        .map(|(_, id)| id)
}

/// Every commit reachable from `start` in `parents`, including `start`.
fn reachable_in(
    parents: &std::collections::HashMap<ObjectId, Vec<ObjectId>>,
    start: &ObjectId,
) -> std::collections::HashSet<ObjectId> {
    let mut seen = std::collections::HashSet::new();
    let mut stack = vec![*start];
    while let Some(id) = stack.pop() {
        if seen.insert(id)
            && let Some(ps) = parents.get(&id)
        {
            stack.extend(ps.iter().filter(|p| !seen.contains(*p)));
        }
    }
    seen
}

/// The generation of every commit in `parents`: the length of the longest
/// path back to a root. One iterative post-order pass, so it is linear and
/// cannot overflow the stack on a long history.
///
/// Commit ids hash their parents, so a genuine cycle cannot exist; an edge back
/// into a commit still being expanded is ignored all the same, so a corrupt
/// store cannot make this loop forever.
fn generations(
    parents: &std::collections::HashMap<ObjectId, Vec<ObjectId>>,
) -> std::collections::HashMap<ObjectId, u64> {
    let mut generation: std::collections::HashMap<ObjectId, u64> =
        std::collections::HashMap::with_capacity(parents.len());
    let mut expanding: std::collections::HashSet<ObjectId> = std::collections::HashSet::new();
    for &root in parents.keys() {
        if generation.contains_key(&root) {
            continue;
        }
        let mut stack = vec![(root, false)];
        while let Some((id, children_done)) = stack.pop() {
            if generation.contains_key(&id) {
                continue;
            }
            let ps = parents.get(&id).map(Vec::as_slice).unwrap_or(&[]);
            if children_done {
                let g = ps
                    .iter()
                    .filter_map(|p| generation.get(p))
                    .map(|g| g + 1)
                    .max()
                    .unwrap_or(0);
                generation.insert(id, g);
                expanding.remove(&id);
            } else if expanding.insert(id) {
                stack.push((id, true));
                for p in ps {
                    if !generation.contains_key(p) && !expanding.contains(p) {
                        stack.push((*p, false));
                    }
                }
            }
        }
    }
    generation
}

#[cfg(test)]
mod tests {
    use super::*;
    use agentstategraph_core::{TAG_GIT_REVISION, TAG_PIN_STATE};
    use agentstategraph_storage::SqliteStorage;

    fn test_repo() -> Repository {
        let storage = SqliteStorage::in_memory().expect("in-memory sqlite");
        let repo = Repository::new(Box::new(storage));
        repo.init().unwrap();
        repo
    }

    fn quick_opts(desc: &str) -> CommitOptions {
        CommitOptions::new("agent/test", IntentCategory::Checkpoint, desc)
    }

    /// A sweep's write lock used to be a transaction on the one shared
    /// connection, so another thread's write made while it was held ran inside
    /// it — and the sweep's refusal or error path rolled it back after it had
    /// returned Ok. Now the lock lives on its own connection: a concurrent
    /// write waits and, if the sweep outlasts the busy timeout, fails loudly.
    /// Either way, nothing that returned Ok may disappear.
    #[test]
    fn write_during_a_gc_lock_is_never_acknowledged_then_rolled_back() {
        let path = std::env::temp_dir().join(format!(
            "asg-gclock-{}",
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        let repo = Repository::new(Box::new(SqliteStorage::open(&path).unwrap()));
        repo.init().unwrap();
        let before = repo.head("main").unwrap();

        repo.storage.history_gc_lock_writers().unwrap(); // a sweep starts
        let during = repo.set_json(
            "main",
            "/during",
            &serde_json::json!("written"),
            quick_opts("w"),
        );
        repo.storage.history_gc_unlock_writers(false).unwrap(); // sweep refuses

        match during {
            Ok(acked) => assert_eq!(
                repo.head("main").unwrap(),
                acked,
                "a write that returned Ok was rolled back by the sweep"
            ),
            Err(_) => assert_eq!(repo.head("main").unwrap(), before),
        }
        // And the store is writable again once the sweep is done.
        repo.set_json("main", "/after", &serde_json::json!(1), quick_opts("a"))
            .unwrap();
    }

    // --- full-DAG log + paged query (plan asg-commit-dag-walk t-002) -------

    /// Build a real merge on `main` and return the commit ids of the two
    /// sides. `log` reaches only the first-parent side.
    fn repo_with_a_merge() -> (Repository, ObjectId) {
        let repo = test_repo();
        repo.set("main", "/a", &Object::int(1), quick_opts("base"))
            .unwrap();
        repo.branch("feature", "main").unwrap();
        repo.set("main", "/a", &Object::int(2), quick_opts("on main"))
            .unwrap();
        let side = repo
            .set("feature", "/b", &Object::int(3), quick_opts("on feature"))
            .unwrap();
        repo.merge(
            "feature",
            "main",
            CommitOptions::new("agent/test", IntentCategory::Merge, "merge feature"),
        )
        .unwrap();

        // Precondition: this must really be a two-parent merge. If the merge
        // ever fast-forwards instead, every test below would pass for the
        // wrong reason — `log` and `log_dag` agree on a linear chain.
        let head = repo.resolve_ref("main").unwrap();
        let merge_commit = repo.get_commit(&head).unwrap().unwrap();
        assert_eq!(
            merge_commit.parents.len(),
            2,
            "fixture must produce a real merge, got {} parent(s)",
            merge_commit.parents.len()
        );

        (repo, side)
    }

    #[test]
    fn log_dag_reaches_the_second_parent_and_log_does_not() {
        let (repo, side) = repo_with_a_merge();

        let linear: Vec<ObjectId> = repo
            .log("main", 100)
            .unwrap()
            .iter()
            .map(|c| c.id)
            .collect();
        assert!(
            !linear.contains(&side),
            "log() must still be first-parent — this test pins the DIFFERENCE \
             between the two walks, so it has to fail if they ever collapse"
        );

        let walk = repo.log_dag("main", 100).unwrap();
        let ids: Vec<ObjectId> = walk.commits.iter().map(|c| c.id).collect();
        assert!(ids.contains(&side), "log_dag missed the merged-in side");
        assert!(!walk.truncated);
    }

    #[test]
    fn log_dag_reports_truncation() {
        let (repo, _) = repo_with_a_merge();
        let all = repo.log_dag("main", 100).unwrap();
        assert!(!all.truncated, "unbounded walk is complete");

        let capped = repo.log_dag("main", 2).unwrap();
        assert_eq!(capped.commits.len(), 2);
        assert!(capped.truncated, "stopped with commits still unexplored");
    }

    /// `log_dag` must apply `enforce_caps` exactly as `log` does, or the two
    /// return differently-shaped commits for the same store.
    ///
    /// The description has to actually EXCEED the cap: an earlier version of
    /// this test used the ordinary fixture, whose commits are far below every
    /// limit, so capping was a no-op and the test passed with `enforce_caps`
    /// deleted. Mutation-checked now — removing the call fails this.
    #[test]
    fn log_dag_enforces_caps_like_log() {
        let repo = test_repo();
        // `CommitBuilder::build` caps on WRITE, so anything created through
        // the normal API is already within limits and the read-side
        // `enforce_caps` is a no-op. Reading it back therefore proves nothing.
        // Inject an over-cap commit straight into storage instead — the case
        // read-side capping actually exists for (data written by an older
        // version).
        const OVERSIZED: usize = 20_000;
        repo.set("main", "/a", &Object::int(1), quick_opts("normal"))
            .unwrap();
        let head_id = repo.resolve_ref("main").unwrap();
        let mut raw = repo.storage.get_commit(&head_id).unwrap().unwrap();
        raw.intent.description = "d".repeat(OVERSIZED);
        // A FRESH id is essential: `put_commit` is INSERT OR IGNORE, so
        // re-putting the same id silently discards the change and the test
        // quietly proves nothing. Then point the ref at it so both walks
        // start there.
        raw.id = ObjectId::hash(b"uncapped-commit-for-caps-parity-test");
        repo.storage.put_commit(&raw).unwrap();
        let ns = repo.active_namespace().unwrap();
        repo.storage.set_ref(&ns, "main", raw.id).unwrap();
        let head_id = raw.id;

        let from_log = repo.log("main", 100).unwrap();
        let capped = from_log
            .iter()
            .find(|c| c.id == head_id)
            .expect("the injected commit");
        assert!(
            capped.intent.description.chars().count() < OVERSIZED,
            "precondition: log() must cap on read, or there is nothing to compare"
        );

        let from_dag = repo.log_dag("main", 100).unwrap();
        let same = from_dag
            .commits
            .iter()
            .find(|d| d.id == head_id)
            .expect("same commit via the DAG walk");
        assert_eq!(
            same.intent.description.chars().count(),
            capped.intent.description.chars().count(),
            "log_dag returned an uncapped description where log returned a capped one"
        );
    }

    /// The deep-page bug: the old body scanned a hard-coded 10_000, so a page
    /// past that came back empty with nothing to say why.
    #[test]
    fn query_commits_paged_scans_far_enough_for_a_deep_offset() {
        let (repo, _) = repo_with_a_merge();
        let filters = agentstategraph_core::QueryFilters::default();

        // Shallow page still works.
        let first = repo.query_commits_paged("main", &filters, 2, 0).unwrap();
        assert!(!first.is_empty(), "a shallow page must return commits");

        // An offset past the end is legitimately empty — but it must be empty
        // because the history ended, not because a constant cut the scan.
        let deep = repo
            .query_commits_paged("main", &filters, 10, COMMIT_QUERY_SCAN_FLOOR + 5_000)
            .unwrap();
        assert!(deep.is_empty());
    }

    #[test]
    fn commit_query_scan_widens_for_deep_pages() {
        // Shallow pages ride the floor — the common case stays cheap.
        assert_eq!(commit_query_scan(0, 10), COMMIT_QUERY_SCAN_FLOOR);
        assert_eq!(commit_query_scan(100, 50), COMMIT_QUERY_SCAN_FLOOR);

        // Past the floor the scan must follow the page, or the old bug
        // returns: an empty page with nothing to say why.
        assert_eq!(commit_query_scan(15_000, 10), 15_010);
        assert!(commit_query_scan(COMMIT_QUERY_SCAN_FLOOR, 1) > COMMIT_QUERY_SCAN_FLOOR);

        // Absurd input must not wrap into a tiny scan.
        assert_eq!(commit_query_scan(usize::MAX, 10), usize::MAX);
    }

    /// Paged query must see merged-in commits, not just the first-parent line.
    #[test]
    fn query_commits_paged_sees_the_second_parent() {
        let (repo, side) = repo_with_a_merge();
        let filters = agentstategraph_core::QueryFilters::default();
        let page = repo.query_commits_paged("main", &filters, 100, 0).unwrap();
        let ids: Vec<ObjectId> = page.iter().map(|c| c.id).collect();
        assert!(
            ids.contains(&side),
            "paged query inherited the first-parent blind spot"
        );
    }

    /// The case the whole DAG-walk plan exists for, and the honest limit of
    /// the fix: a commit no ref reaches is STILL not returned by `log_dag`.
    ///
    /// Walking every parent edge closes the merge-second-parent gap. It does
    /// not, and cannot, close the unreachability gap — nothing links an
    /// abandoned commit to the ref head. On the asd store that is 1,628
    /// commits: `log` reached 4,268, a full DAG walk reaches no more, and the
    /// store holds 5,896. Anything reporting "what would a prune take?" has
    /// to compare the walk against the store's own total, which is why the
    /// asd metrics endpoint reports `scanned` alongside `distilled`.
    #[test]
    fn log_dag_does_not_reach_commits_no_ref_points_at() {
        let repo = test_repo();
        repo.set("main", "/a", &Object::int(1), quick_opts("on main"))
            .unwrap();
        repo.branch("abandoned", "main").unwrap();
        let orphan = repo
            .set(
                "abandoned",
                "/b",
                &Object::int(2),
                quick_opts("abandoned work"),
            )
            .unwrap();

        // Reachable while the branch exists.
        let before: Vec<ObjectId> = repo
            .log_dag("abandoned", 100)
            .unwrap()
            .commits
            .iter()
            .map(|c| c.id)
            .collect();
        assert!(
            before.contains(&orphan),
            "precondition: reachable via its own ref"
        );

        repo.delete_branch("abandoned").unwrap();

        let from_head: Vec<ObjectId> = repo
            .log_dag("main", 100)
            .unwrap()
            .commits
            .iter()
            .map(|c| c.id)
            .collect();
        assert!(
            !from_head.contains(&orphan),
            "a DAG walk from main cannot reach an abandoned commit — if this \
             ever passes, the walk is following something other than parents"
        );

        // But it is still IN the store. This is the gap a prune estimate needs.
        assert!(
            repo.storage.get_commit(&orphan).unwrap().is_some(),
            "the abandoned commit is still stored, merely unreachable"
        );
        let all = repo.storage.all_commit_ids().unwrap();
        assert!(all.contains(&orphan), "all_commit_ids must still see it");
        assert!(
            all.len() > from_head.len(),
            "store total ({}) must exceed what the head reaches ({}) — this \
             difference is exactly what a prune would reclaim",
            all.len(),
            from_head.len()
        );
    }

    /// A merge commit reachable down both sides must appear once, at the
    /// Repository level too — dedup is in the storage default, but a future
    /// backend override could lose it.
    #[test]
    fn log_dag_dedups_at_the_repository_level() {
        let (repo, _) = repo_with_a_merge();
        let walk = repo.log_dag("main", 100).unwrap();
        let mut ids: Vec<ObjectId> = walk.commits.iter().map(|c| c.id).collect();
        let total = ids.len();
        ids.sort();
        ids.dedup();
        assert_eq!(total, ids.len(), "log_dag returned a commit twice");
    }

    /// The security gap t-004 found: clock-rewind evidence INSIDE a merged-in
    /// branch was invisible, because `log` never yielded those commits as the
    /// outer `commit`. Merge-vs-merged-tip was always checked, so a test that
    /// only backdates the branch tip passes either way and proves nothing.
    ///
    /// Injection is fiddly for two reasons, both learned the hard way:
    /// `CommitBuilder::build` stamps `Utc::now()`, so a backdated commit cannot
    /// be created through the normal API; and `put_commit` is INSERT OR IGNORE,
    /// so re-putting under an existing id silently discards the write.
    #[test]
    fn timestamp_anomalies_are_found_inside_a_merged_branch() {
        let repo = test_repo();
        repo.set("main", "/a", &Object::int(1), quick_opts("base"))
            .unwrap();
        repo.branch("feature", "main").unwrap();
        let s1 = repo
            .set("feature", "/b", &Object::int(1), quick_opts("side 1"))
            .unwrap();
        let s2 = repo
            .set("feature", "/b", &Object::int(2), quick_opts("side 2"))
            .unwrap();

        // Backdate s2 so it precedes its own parent s1 — a rewind wholly inside
        // the branch. Fresh id, and the branch ref repointed at it.
        let mut rewound = repo.storage.get_commit(&s2).unwrap().unwrap();
        rewound.timestamp =
            repo.storage.get_commit(&s1).unwrap().unwrap().timestamp - chrono::Duration::hours(6);
        rewound.id = ObjectId::hash(b"backdated-interior-commit");
        repo.storage.put_commit(&rewound).unwrap();
        let ns = repo.active_namespace().unwrap();
        repo.storage.set_ref(&ns, "feature", rewound.id).unwrap();

        repo.set("main", "/a", &Object::int(2), quick_opts("main moves"))
            .unwrap();
        repo.merge(
            "feature",
            "main",
            CommitOptions::new("agent/test", IntentCategory::Merge, "merge feature"),
        )
        .unwrap();

        // Precondition: the first-parent walk cannot see the rewound commit,
        // so this test is exercising the gap rather than something incidental.
        let linear: Vec<ObjectId> = repo
            .log("main", 1000)
            .unwrap()
            .iter()
            .map(|c| c.id)
            .collect();
        assert!(
            !linear.contains(&rewound.id),
            "precondition: log() must not reach the branch interior"
        );

        let found = repo.detect_timestamp_anomalies("main").unwrap();
        assert!(
            found.iter().any(|(id, _)| *id == rewound.id),
            "the rewound commit inside the merged branch was not flagged; got {found:?}"
        );
    }

    /// A graph whose edges name nodes it does not contain is not a graph.
    #[test]
    fn commit_graph_contains_every_node_its_edges_name() {
        let (repo, side) = repo_with_a_merge();
        let nodes = repo.commit_graph("main", 100).unwrap();
        let ids: std::collections::HashSet<String> = nodes
            .iter()
            .map(|n| n["full_id"].as_str().unwrap().to_string())
            .collect();
        assert!(
            ids.contains(&side.to_string()),
            "the merged-in side is missing from the node set"
        );
        for n in &nodes {
            for p in n["parents"].as_array().unwrap() {
                let short = p.as_str().unwrap();
                assert!(
                    ids.iter().any(|id| id.starts_with(short)),
                    "edge points at {short}, which is not in the node set"
                );
            }
        }
    }

    /// `blame` must name the commit that made the change, not the merge that
    /// brought it in — and must still prefer the most RECENT change, which is
    /// why the DAG walk is re-sorted by time.
    #[test]
    fn blame_attributes_a_change_made_on_a_merged_branch() {
        let repo = test_repo();
        repo.set("main", "/a", &Object::int(1), quick_opts("base"))
            .unwrap();
        repo.branch("feature", "main").unwrap();
        let changer = repo
            .set(
                "feature",
                "/target",
                &Object::int(42),
                quick_opts("set on branch"),
            )
            .unwrap();
        repo.set("main", "/other", &Object::int(9), quick_opts("main moves"))
            .unwrap();
        repo.merge(
            "feature",
            "main",
            CommitOptions::new("agent/test", IntentCategory::Merge, "merge feature"),
        )
        .unwrap();

        let entry = repo.blame("main", "/target").unwrap();
        assert_eq!(
            entry.commit_id,
            changer.short(),
            "blame named the merge commit rather than the commit that set the value"
        );
    }

    #[test]
    fn stats_counts_merged_in_commits() {
        let (repo, _) = repo_with_a_merge();
        let linear = repo.log("main", 1000).unwrap().len();
        let stats = repo.stats("main").unwrap();
        let counted = stats["commit_count"].as_u64().unwrap() as usize;
        assert!(
            counted > linear,
            "stats counted {counted}, no more than the first-parent line's {linear}"
        );
    }

    #[test]
    fn test_init_creates_main() {
        let repo = test_repo();
        let branches = repo.list_branches(None).unwrap();
        assert_eq!(branches.len(), 1);
        assert_eq!(branches[0].0, "main");
    }

    #[test]
    fn test_init_stamps_schema_version() {
        let repo = test_repo();
        let v = repo.get("main", META_SCHEMA_VERSION_PATH).unwrap();
        match v {
            Object::Atom(agentstategraph_core::Atom::String(s)) => {
                assert_eq!(s, SCHEMA_VERSION);
            }
            other => panic!("expected string atom, got {:?}", other),
        }
    }

    #[test]
    fn test_meta_guard_rejects_non_migrate_writes() {
        let repo = test_repo();
        let value = Object::Atom(agentstategraph_core::Atom::String("hack".into()));
        let err = repo
            .set(
                "main",
                META_SCHEMA_VERSION_PATH,
                &value,
                quick_opts("tamper"),
            )
            .unwrap_err();
        assert!(matches!(err, RepoError::ReservedPath(_)), "got {:?}", err);

        let err = repo
            .set("main", "/_meta/custom", &value, quick_opts("tamper2"))
            .unwrap_err();
        assert!(matches!(err, RepoError::ReservedPath(_)), "got {:?}", err);

        let err = repo
            .delete("main", META_SCHEMA_VERSION_PATH, quick_opts("tamper3"))
            .unwrap_err();
        assert!(matches!(err, RepoError::ReservedPath(_)), "got {:?}", err);
    }

    #[test]
    fn test_meta_guard_allows_migrate_writes() {
        let repo = test_repo();
        let value = Object::Atom(agentstategraph_core::Atom::String("0.5.0".into()));
        let opts = CommitOptions::new("agent/migrate", IntentCategory::Migrate, "bump schema");
        repo.set("main", META_SCHEMA_VERSION_PATH, &value, opts)
            .expect("migrate intent should be allowed to write /_meta");

        let got = repo.get("main", META_SCHEMA_VERSION_PATH).unwrap();
        match got {
            Object::Atom(agentstategraph_core::Atom::String(s)) => assert_eq!(s, "0.5.0"),
            other => panic!("unexpected {:?}", other),
        }
    }

    #[test]
    fn test_secret_read_guard_rejects_non_migrate_reads() {
        let repo = test_repo();

        // Seed a value under /_meta/_secret via a Migrate write.
        let value = Object::Atom(agentstategraph_core::Atom::String("shh".into()));
        let migrate_opts =
            CommitOptions::new("agent/migrate", IntentCategory::Migrate, "seed secret");
        repo.set("main", "/_meta/_secret/api_key", &value, migrate_opts)
            .expect("migrate can write to /_meta/_secret");

        // Default get() must reject.
        let err = repo
            .get("main", "/_meta/_secret/api_key")
            .expect_err("non-migrate read of secret must fail");
        assert!(matches!(err, RepoError::ReservedPath(_)), "got {:?}", err);

        // list_paths over /_meta must NOT surface the secret subtree.
        let paths = repo.list_paths("main", "/_meta", None).unwrap();
        assert!(
            paths.iter().all(|p| !p.starts_with("/_meta/_secret")),
            "list_paths leaked secret paths: {:?}",
            paths
        );

        // Directly listing the secret prefix is rejected outright.
        let err = repo
            .list_paths("main", "/_meta/_secret", None)
            .expect_err("list_paths on secret prefix must fail");
        assert!(matches!(err, RepoError::ReservedPath(_)), "got {:?}", err);
    }

    #[test]
    fn test_secret_read_guard_allows_migrate() {
        let repo = test_repo();
        let value = Object::Atom(agentstategraph_core::Atom::String("shh".into()));
        let opts = CommitOptions::new("agent/migrate", IntentCategory::Migrate, "seed");
        repo.set("main", "/_meta/_secret/token", &value, opts)
            .unwrap();

        let intent = Intent::new(IntentCategory::Migrate, "read secret for migration");
        let got = repo
            .get_with_intent("main", "/_meta/_secret/token", &intent)
            .expect("migrate read should succeed");
        match got {
            Object::Atom(agentstategraph_core::Atom::String(s)) => assert_eq!(s, "shh"),
            other => panic!("unexpected {:?}", other),
        }

        // Non-migrate intent, even via the explicit-intent method, must fail.
        let bad_intent = Intent::new(IntentCategory::Checkpoint, "sneak a peek");
        let err = repo
            .get_with_intent("main", "/_meta/_secret/token", &bad_intent)
            .unwrap_err();
        assert!(matches!(err, RepoError::ReservedPath(_)), "got {:?}", err);
    }

    #[test]
    fn test_meta_guard_does_not_match_user_paths_with_meta_substring() {
        let repo = test_repo();
        let value = Object::Atom(agentstategraph_core::Atom::Int(1));
        // user-space path that happens to share the prefix substring
        repo.set("main", "/_metadata_not_reserved", &value, quick_opts("ok"))
            .expect("unrelated path must not trip the guard");
    }

    #[test]
    fn test_set_and_get() {
        let repo = test_repo();

        repo.set(
            "main",
            "/name",
            &Object::string("my-cluster"),
            quick_opts("set name"),
        )
        .unwrap();

        let obj = repo.get("main", "/name").unwrap();
        assert_eq!(obj, Object::string("my-cluster"));
    }

    #[test]
    fn test_set_json_and_get_json() {
        let repo = test_repo();

        repo.set_json(
            "main",
            "/config",
            &serde_json::json!({
                "network": { "subnet": "10.0.0.0/24" },
                "gpu": { "enabled": true }
            }),
            quick_opts("set config"),
        )
        .unwrap();

        let json = repo.get_json("main", "/config/network/subnet").unwrap();
        assert_eq!(json, serde_json::json!("10.0.0.0/24"));

        let gpu = repo.get_json("main", "/config/gpu/enabled").unwrap();
        assert_eq!(gpu, serde_json::json!(true));
    }

    #[test]
    fn test_delete() {
        let repo = test_repo();

        repo.set(
            "main",
            "/temp",
            &Object::string("temporary"),
            quick_opts("add temp"),
        )
        .unwrap();

        repo.delete("main", "/temp", quick_opts("remove temp"))
            .unwrap();

        assert!(repo.get("main", "/temp").is_err());
    }

    #[test]
    fn test_branch_and_diverge() {
        let repo = test_repo();

        // Set initial state
        repo.set("main", "/value", &Object::int(1), quick_opts("initial"))
            .unwrap();

        // Create branch
        repo.branch("feature", "main").unwrap();

        // Modify main
        repo.set("main", "/value", &Object::int(2), quick_opts("update main"))
            .unwrap();

        // Modify branch
        repo.set(
            "feature",
            "/value",
            &Object::int(3),
            quick_opts("update feature"),
        )
        .unwrap();

        // Both diverged
        assert_eq!(repo.get("main", "/value").unwrap(), Object::int(2));
        assert_eq!(repo.get("feature", "/value").unwrap(), Object::int(3));
    }

    #[test]
    fn test_branch_already_exists() {
        let repo = test_repo();
        assert!(repo.branch("main", "main").is_err());
    }

    #[test]
    fn test_delete_branch() {
        let repo = test_repo();
        repo.branch("temp-branch", "main").unwrap();
        assert!(repo.delete_branch("temp-branch").unwrap());
        assert!(!repo.delete_branch("temp-branch").unwrap()); // already deleted
    }

    #[test]
    fn test_list_branches_with_prefix() {
        let repo = test_repo();
        repo.branch("agents/planner/workspace", "main").unwrap();
        repo.branch("agents/storage/workspace", "main").unwrap();
        repo.branch("explore/nfs", "main").unwrap();

        let agent_branches = repo.list_branches(Some("agents/")).unwrap();
        assert_eq!(agent_branches.len(), 2);

        let all_branches = repo.list_branches(None).unwrap();
        assert_eq!(all_branches.len(), 4); // main + 3 new
    }

    #[test]
    fn test_commit_log() {
        let repo = test_repo();

        repo.set("main", "/a", &Object::int(1), quick_opts("first"))
            .unwrap();
        repo.set("main", "/b", &Object::int(2), quick_opts("second"))
            .unwrap();
        repo.set("main", "/c", &Object::int(3), quick_opts("third"))
            .unwrap();

        let log = repo.log("main", 10).unwrap();
        assert_eq!(log.len(), 4); // 3 + init commit

        // Most recent first
        assert_eq!(log[0].intent.description, "third");
        assert_eq!(log[1].intent.description, "second");
        assert_eq!(log[2].intent.description, "first");
    }

    #[test]
    fn test_intent_metadata_preserved() {
        let repo = test_repo();

        let opts = CommitOptions::new(
            "agent/planner-v2",
            IntentCategory::Explore,
            "try NFS storage",
        )
        .with_reasoning("NFS is simpler than Ceph for 2-node clusters")
        .with_confidence(0.8)
        .with_tags(vec!["storage".to_string(), "nfs".to_string()]);

        repo.set("main", "/storage/type", &Object::string("nfs"), opts)
            .unwrap();

        let log = repo.log("main", 1).unwrap();
        let commit = &log[0];

        assert_eq!(commit.agent_id, "agent/planner-v2");
        assert_eq!(commit.intent.category, IntentCategory::Explore);
        assert_eq!(commit.intent.description, "try NFS storage");
        assert_eq!(commit.intent.tags, vec!["storage", "nfs"]);
        assert_eq!(
            commit.reasoning,
            Some("NFS is simpler than Ceph for 2-node clusters".to_string())
        );
        assert_eq!(commit.confidence, Some(0.8));
    }

    #[test]
    fn test_nested_set_creates_intermediate_maps() {
        let repo = test_repo();

        repo.set(
            "main",
            "/config/network/dns/primary",
            &Object::string("8.8.8.8"),
            quick_opts("set DNS"),
        )
        .unwrap();

        let dns = repo.get("main", "/config/network/dns/primary").unwrap();
        assert_eq!(dns, Object::string("8.8.8.8"));
    }

    #[test]
    fn test_immutability_across_branches() {
        let repo = test_repo();

        repo.set_json(
            "main",
            "/cluster",
            &serde_json::json!({ "name": "prod", "nodes": 5 }),
            quick_opts("init cluster"),
        )
        .unwrap();

        // Branch and modify
        repo.branch("staging", "main").unwrap();
        repo.set_json(
            "staging",
            "/cluster/name",
            &serde_json::json!("staging"),
            quick_opts("rename to staging"),
        )
        .unwrap();

        // main is untouched
        let main_name = repo.get_json("main", "/cluster/name").unwrap();
        assert_eq!(main_name, serde_json::json!("prod"));

        // staging has the change
        let staging_name = repo.get_json("staging", "/cluster/name").unwrap();
        assert_eq!(staging_name, serde_json::json!("staging"));
    }

    // -----------------------------------------------------------------------
    // Diff tests
    // -----------------------------------------------------------------------

    #[test]
    fn test_diff_identical_branches() {
        let repo = test_repo();
        repo.set("main", "/x", &Object::int(1), quick_opts("set"))
            .unwrap();
        repo.branch("copy", "main").unwrap();

        let ops = repo.diff("main", "copy").unwrap();
        assert!(ops.is_empty(), "identical branches should produce no diff");
    }

    #[test]
    fn test_diff_value_change() {
        let repo = test_repo();
        repo.set(
            "main",
            "/status",
            &Object::string("healthy"),
            quick_opts("init"),
        )
        .unwrap();
        repo.branch("feature", "main").unwrap();
        repo.set(
            "feature",
            "/status",
            &Object::string("unhealthy"),
            quick_opts("break"),
        )
        .unwrap();

        let ops = repo.diff("main", "feature").unwrap();
        assert_eq!(ops.len(), 1);
        match &ops[0] {
            agentstategraph_core::DiffOp::SetValue { path, .. } => {
                assert_eq!(path, "/status");
            }
            _ => panic!("expected SetValue"),
        }
    }

    #[test]
    fn test_diff_multiple_changes() {
        let repo = test_repo();
        repo.set_json(
            "main",
            "/cluster",
            &serde_json::json!({"name": "prod", "nodes": 3, "region": "us-east"}),
            quick_opts("init cluster"),
        )
        .unwrap();

        repo.branch("feature", "main").unwrap();

        // Change name, remove region, add version
        repo.set(
            "feature",
            "/cluster/name",
            &Object::string("staging"),
            quick_opts("rename"),
        )
        .unwrap();
        repo.delete("feature", "/cluster/region", quick_opts("remove region"))
            .unwrap();
        repo.set(
            "feature",
            "/cluster/version",
            &Object::string("v2"),
            quick_opts("add version"),
        )
        .unwrap();

        let ops = repo.diff("main", "feature").unwrap();
        assert!(
            ops.len() >= 3,
            "expected at least 3 diff ops, got {}",
            ops.len()
        );

        // Verify it's JSON-serializable (MCP-ready)
        let json = serde_json::to_string_pretty(&ops).unwrap();
        assert!(json.contains("SetValue") || json.contains("AddKey") || json.contains("RemoveKey"));
    }

    // -----------------------------------------------------------------------
    // Merge tests
    // -----------------------------------------------------------------------

    #[test]
    fn test_merge_non_conflicting() {
        let repo = test_repo();

        // Set initial state with two keys
        repo.set("main", "/a", &Object::int(1), quick_opts("init a"))
            .unwrap();
        repo.set("main", "/b", &Object::int(2), quick_opts("init b"))
            .unwrap();

        // Branch
        repo.branch("feature", "main").unwrap();

        // Change different keys on each branch
        repo.set(
            "main",
            "/a",
            &Object::int(10),
            quick_opts("update a on main"),
        )
        .unwrap();
        repo.set(
            "feature",
            "/b",
            &Object::int(20),
            quick_opts("update b on feature"),
        )
        .unwrap();

        // Merge feature into main
        let merge_opts = CommitOptions::new(
            "agent/test",
            IntentCategory::Merge,
            "merge feature into main",
        );
        repo.merge("feature", "main", merge_opts).unwrap();

        // Both changes should be present
        assert_eq!(repo.get("main", "/a").unwrap(), Object::int(10)); // ours
        assert_eq!(repo.get("main", "/b").unwrap(), Object::int(20)); // theirs

        // The merged state tree must be fully readable — no dangling nodes.
        let head = repo.resolve_ref("main").unwrap();
        let root = repo.storage.get_commit(&head).unwrap().unwrap().state_root;
        assert_eq!(repo.first_missing_reachable(&root).unwrap(), None);
    }

    #[test]
    fn test_first_missing_reachable_detects_dangling_child() {
        let repo = test_repo();
        // A map whose "x" points at an object we never store — the exact shape
        // of the `/plans` corruption (parent present, interior child absent).
        let missing = Object::int(42).id();
        let mut entries = std::collections::BTreeMap::new();
        entries.insert("x".to_string(), missing);
        let root = Object::map(entries);
        let root_id = repo.storage.put_object(&root).unwrap();

        assert_eq!(
            repo.first_missing_reachable(&root_id).unwrap(),
            Some(missing),
            "a referenced-but-absent child must be reported"
        );

        // Once the child exists, the tree is clean.
        repo.storage.put_object(&Object::int(42)).unwrap();
        assert_eq!(repo.first_missing_reachable(&root_id).unwrap(), None);
    }

    #[test]
    fn test_first_missing_reachable_clean_tree() {
        let repo = test_repo();
        let head = repo.resolve_ref("main").unwrap();
        let root = repo.storage.get_commit(&head).unwrap().unwrap().state_root;
        // A freshly-initialised repo's root tree is fully present.
        assert_eq!(repo.first_missing_reachable(&root).unwrap(), None);
    }

    #[test]
    fn test_merge_with_conflict() {
        let repo = test_repo();

        repo.set("main", "/x", &Object::int(1), quick_opts("init"))
            .unwrap();
        repo.branch("feature", "main").unwrap();

        // Both change the same key to different values
        repo.set("main", "/x", &Object::int(2), quick_opts("main change"))
            .unwrap();
        repo.set(
            "feature",
            "/x",
            &Object::int(3),
            quick_opts("feature change"),
        )
        .unwrap();

        let merge_opts = CommitOptions::new("agent/test", IntentCategory::Merge, "merge");
        let result = repo.merge("feature", "main", merge_opts);

        match result {
            Err(RepoError::MergeConflicts(conflicts)) => {
                assert!(!conflicts.is_empty(), "should have conflicts");
            }
            Ok(_) => panic!("expected merge conflicts"),
            Err(e) => panic!("unexpected error: {}", e),
        }
    }

    #[test]
    fn test_merge_fast_forward() {
        let repo = test_repo();
        repo.set("main", "/x", &Object::int(1), quick_opts("init"))
            .unwrap();
        repo.branch("feature", "main").unwrap();

        // Only feature changes, main stays the same
        repo.set(
            "feature",
            "/x",
            &Object::int(2),
            quick_opts("feature change"),
        )
        .unwrap();

        let merge_opts = CommitOptions::new("agent/test", IntentCategory::Merge, "ff merge");
        repo.merge("feature", "main", merge_opts).unwrap();

        assert_eq!(repo.get("main", "/x").unwrap(), Object::int(2));
    }

    #[test]
    fn test_merge_creates_merge_commit() {
        let repo = test_repo();
        repo.set("main", "/a", &Object::int(1), quick_opts("init"))
            .unwrap();
        repo.branch("feature", "main").unwrap();

        repo.set("main", "/a", &Object::int(10), quick_opts("main"))
            .unwrap();
        repo.set("feature", "/b", &Object::int(20), quick_opts("feature"))
            .unwrap();

        let merge_opts = CommitOptions::new("agent/test", IntentCategory::Merge, "merge feature");
        let _merge_commit_id = repo.merge("feature", "main", merge_opts).unwrap();

        let log = repo.log("main", 1).unwrap();
        let merge_commit = &log[0];
        assert_eq!(merge_commit.intent.category, IntentCategory::Merge);
        assert_eq!(
            merge_commit.parents.len(),
            2,
            "merge commit should have 2 parents"
        );
    }

    #[test]
    fn test_merge_preserves_intent_metadata() {
        let repo = test_repo();
        repo.set("main", "/a", &Object::int(1), quick_opts("init"))
            .unwrap();
        repo.branch("feature", "main").unwrap();
        repo.set("main", "/c", &Object::int(99), quick_opts("main work"))
            .unwrap();
        repo.set("feature", "/b", &Object::int(2), quick_opts("feature work"))
            .unwrap();

        let merge_opts = CommitOptions::new(
            "agent/planner",
            IntentCategory::Merge,
            "integrate feature work",
        )
        .with_reasoning("Feature branch had the storage config we need")
        .with_confidence(0.9)
        .with_tags(vec!["storage".to_string(), "merge".to_string()]);

        repo.merge("feature", "main", merge_opts).unwrap();

        let log = repo.log("main", 1).unwrap();
        let commit = &log[0];
        assert_eq!(commit.agent_id, "agent/planner");
        assert_eq!(
            commit.reasoning,
            Some("Feature branch had the storage config we need".to_string())
        );
        assert_eq!(commit.confidence, Some(0.9));
        assert_eq!(commit.intent.tags, vec!["storage", "merge"]);
    }

    #[test]
    fn test_diff_is_json_serializable() {
        let repo = test_repo();
        repo.set("main", "/a", &Object::int(1), quick_opts("init"))
            .unwrap();
        repo.branch("b", "main").unwrap();
        repo.set("b", "/a", &Object::int(2), quick_opts("change"))
            .unwrap();

        let ops = repo.diff("main", "b").unwrap();
        let json = serde_json::to_value(&ops).unwrap();
        assert!(json.is_array());
        assert_eq!(json.as_array().unwrap().len(), 1);
    }

    // -----------------------------------------------------------------------
    // Speculation tests
    // -----------------------------------------------------------------------

    #[test]
    fn test_speculate_and_read() {
        let repo = test_repo();
        repo.set("main", "/value", &Object::int(42), quick_opts("init"))
            .unwrap();

        let h = repo.speculate("main", Some("test".to_string())).unwrap();
        let obj = repo.spec_get(h, "/value").unwrap();
        assert_eq!(obj, Object::int(42));
    }

    #[test]
    fn test_speculate_modify_isolation() {
        let repo = test_repo();
        repo.set("main", "/x", &Object::int(1), quick_opts("init"))
            .unwrap();

        let h = repo.speculate("main", None).unwrap();
        repo.spec_set(h, "/x", &Object::int(99)).unwrap();

        // Speculation has new value
        assert_eq!(repo.spec_get(h, "/x").unwrap(), Object::int(99));

        // Main unchanged
        assert_eq!(repo.get("main", "/x").unwrap(), Object::int(1));
    }

    #[test]
    fn test_compare_two_speculations() {
        let repo = test_repo();
        repo.set_json(
            "main",
            "/storage",
            &serde_json::json!({"type": "none"}),
            quick_opts("init"),
        )
        .unwrap();

        let nfs = repo.speculate("main", Some("NFS".to_string())).unwrap();
        let ceph = repo.speculate("main", Some("Ceph".to_string())).unwrap();

        repo.spec_set(nfs, "/storage/type", &Object::string("nfs"))
            .unwrap();
        repo.spec_set(ceph, "/storage/type", &Object::string("ceph"))
            .unwrap();

        let comparison = repo.compare_speculations(&[nfs, ceph]).unwrap();
        assert_eq!(comparison.entries.len(), 2);
        assert!(!comparison.entries[0].diff_from_base.is_empty());
        assert!(!comparison.entries[1].diff_from_base.is_empty());
    }

    #[test]
    fn test_commit_speculation() {
        let repo = test_repo();
        repo.set("main", "/x", &Object::int(1), quick_opts("init"))
            .unwrap();

        let h = repo.speculate("main", Some("winner".to_string())).unwrap();
        repo.spec_set(h, "/x", &Object::int(42)).unwrap();

        // Commit the speculation
        let opts = CommitOptions::new(
            "agent/planner",
            IntentCategory::Refine,
            "picked best approach",
        )
        .with_reasoning("Option A was better because...");
        repo.commit_speculation(h, opts).unwrap();

        // Main now has the speculated value
        assert_eq!(repo.get("main", "/x").unwrap(), Object::int(42));

        // Verify commit metadata
        let log = repo.log("main", 1).unwrap();
        assert_eq!(log[0].intent.description, "picked best approach");
        assert_eq!(log[0].agent_id, "agent/planner");
    }

    #[test]
    fn test_discard_speculation() {
        let repo = test_repo();
        repo.set("main", "/x", &Object::int(1), quick_opts("init"))
            .unwrap();

        let h = repo.speculate("main", None).unwrap();
        repo.spec_set(h, "/x", &Object::int(999)).unwrap();
        repo.discard_speculation(h).unwrap();

        // Main unchanged
        assert_eq!(repo.get("main", "/x").unwrap(), Object::int(1));

        // Handle is invalid now
        assert!(repo.spec_get(h, "/x").is_err());
    }

    #[test]
    fn test_full_agent_speculation_workflow() {
        // The complete "explore, compare, pick winner" pattern
        let repo = test_repo();
        repo.set_json(
            "main",
            "/cluster",
            &serde_json::json!({
                "name": "prod",
                "storage": {"type": "none"},
                "network": {"subnet": "10.0.0.0/24"}
            }),
            quick_opts("initial cluster state"),
        )
        .unwrap();

        // Agent creates three speculations
        let nfs = repo
            .speculate("main", Some("NFS approach".to_string()))
            .unwrap();
        let ceph = repo
            .speculate("main", Some("Ceph approach".to_string()))
            .unwrap();
        let local = repo
            .speculate("main", Some("Local SSD".to_string()))
            .unwrap();

        // Each speculation explores a different approach
        repo.spec_set(nfs, "/cluster/storage/type", &Object::string("nfs"))
            .unwrap();
        repo.spec_set(nfs, "/cluster/storage/mount", &Object::string("/shared"))
            .unwrap();

        repo.spec_set(ceph, "/cluster/storage/type", &Object::string("ceph"))
            .unwrap();
        repo.spec_set(ceph, "/cluster/storage/replicas", &Object::int(3))
            .unwrap();

        repo.spec_set(local, "/cluster/storage/type", &Object::string("local-ssd"))
            .unwrap();
        repo.spec_set(
            local,
            "/cluster/storage/path",
            &Object::string("/dev/nvme0"),
        )
        .unwrap();

        // Compare all three
        let comparison = repo.compare_speculations(&[nfs, ceph, local]).unwrap();
        assert_eq!(comparison.entries.len(), 3);

        // Agent picks NFS (Ceph needs too many nodes, local isn't shared)
        let opts = CommitOptions::new(
            "agent/storage-planner",
            IntentCategory::Refine,
            "Selected NFS — Ceph requires 3+ nodes, local SSD not shared",
        )
        .with_reasoning("NFS provides shared storage with minimal node requirements")
        .with_confidence(0.85)
        .with_tags(vec!["storage".to_string(), "nfs".to_string()]);

        repo.commit_speculation(nfs, opts).unwrap();

        // Discard losers
        repo.discard_speculation(ceph).unwrap();
        repo.discard_speculation(local).unwrap();

        // Verify final state
        let storage_type = repo.get("main", "/cluster/storage/type").unwrap();
        assert_eq!(storage_type, Object::string("nfs"));

        let mount = repo.get("main", "/cluster/storage/mount").unwrap();
        assert_eq!(mount, Object::string("/shared"));

        // Verify full commit trail
        let log = repo.log("main", 2).unwrap();
        assert_eq!(log[0].intent.category, IntentCategory::Refine);
        assert_eq!(log[0].confidence, Some(0.85));
        assert_eq!(log[0].intent.tags, vec!["storage", "nfs"]);

        // No speculations left
        assert!(repo.list_speculations().is_empty());
    }

    // ---- v3+ timestamp anomaly detection (V4) --------------------------

    #[test]
    fn test_detect_timestamp_anomalies_flat_history() {
        // A normal history: every commit is strictly after its parent.
        let repo = Repository::new(Box::new(
            SqliteStorage::in_memory().expect("in-memory sqlite"),
        ));
        repo.init().unwrap();
        repo.set("main", "/a", &Object::string("1"), quick_opts("first"))
            .unwrap();
        repo.set("main", "/a", &Object::string("2"), quick_opts("second"))
            .unwrap();
        let anomalies = repo.detect_timestamp_anomalies("main").unwrap();
        assert!(
            anomalies.is_empty(),
            "expected no anomalies on monotonic history; got {anomalies:?}"
        );
    }

    #[test]
    fn test_blame_entry_timestamp_anomaly_is_false_by_default() {
        let repo = Repository::new(Box::new(
            SqliteStorage::in_memory().expect("in-memory sqlite"),
        ));
        repo.init().unwrap();
        repo.set("main", "/a", &Object::string("v"), quick_opts("write"))
            .unwrap();
        let entry = repo.blame("main", "/a").unwrap();
        assert!(!entry.timestamp_anomaly);
    }

    // ---- v3+ epoch seal enforcement (V8) -------------------------------

    #[test]
    fn test_seal_epoch_captures_reachable_commits() {
        let repo = Repository::new(Box::new(
            SqliteStorage::in_memory().expect("in-memory sqlite"),
        ));
        repo.init().unwrap();
        repo.set("main", "/a", &Object::string("1"), quick_opts("a"))
            .unwrap();
        repo.set("main", "/b", &Object::string("2"), quick_opts("b"))
            .unwrap();

        repo.create_epoch("e1", "first epoch", vec![]).unwrap();
        repo.seal_epoch("e1", "done").unwrap();

        let epoch = repo.get_epoch("e1").unwrap();
        assert!(
            !epoch.sealed_commits.is_empty(),
            "seal_epoch must populate sealed_commits"
        );
    }

    #[test]
    fn test_epoch_seal_violation_in_warn_mode_is_logged_not_rejected() {
        // Warn mode is now opt-out (strict is the default) — a rewind past a
        // sealed commit logs but still succeeds.
        let repo = Repository::new(Box::new(
            SqliteStorage::in_memory().expect("in-memory sqlite"),
        ))
        .with_epoch_seal_strict(false);
        repo.init().unwrap();
        let _ = repo
            .set("main", "/a", &Object::string("1"), quick_opts("a"))
            .unwrap();
        let pre_seal = repo
            .set("main", "/b", &Object::string("2"), quick_opts("b"))
            .unwrap();
        let _post_seal = repo
            .set("main", "/c", &Object::string("3"), quick_opts("c"))
            .unwrap();

        repo.create_epoch("e1", "scoped work", vec![]).unwrap();
        repo.seal_epoch("e1", "ship").unwrap();

        // Rewind main to pre_seal via set_ref. In warn mode this succeeds.
        let res = repo.set_ref("main", pre_seal);
        assert!(res.is_ok(), "warn mode must accept the rewind; got {res:?}");
    }

    #[test]
    fn test_epoch_seal_strict_is_the_default() {
        // The hard guard must fire through a PLAIN `Repository::new` — no
        // `.with_epoch_seal_strict(true)`, no env var. This is the production
        // constructor every shipped surface (MCP/HTTP/FFI/CLI) goes through, so
        // this asserts the guard Plan B relies on is engaged by default.
        assert!(
            std::env::var(EPOCH_SEAL_STRICT_ENV).is_err(),
            "test assumes {EPOCH_SEAL_STRICT_ENV} is unset in the test environment"
        );
        let repo = Repository::new(Box::new(
            SqliteStorage::in_memory().expect("in-memory sqlite"),
        ));
        repo.init().unwrap();
        repo.set("main", "/a", &Object::string("1"), quick_opts("a"))
            .unwrap();
        let pre_seal = repo
            .set("main", "/b", &Object::string("2"), quick_opts("b"))
            .unwrap();
        repo.set("main", "/c", &Object::string("3"), quick_opts("c"))
            .unwrap();

        repo.create_epoch("e1", "scoped work", vec![]).unwrap();
        repo.seal_epoch("e1", "ship").unwrap();

        match repo.set_ref("main", pre_seal) {
            Err(RepoError::EpochSealViolated { .. }) => {}
            other => panic!("expected EpochSealViolated by default, got {other:?}"),
        }
    }

    #[test]
    fn test_epoch_seal_violation_in_strict_mode_is_rejected() {
        let repo = Repository::new(Box::new(
            SqliteStorage::in_memory().expect("in-memory sqlite"),
        ))
        .with_epoch_seal_strict(true);
        repo.init().unwrap();
        let _ = repo
            .set("main", "/a", &Object::string("1"), quick_opts("a"))
            .unwrap();
        let pre_seal = repo
            .set("main", "/b", &Object::string("2"), quick_opts("b"))
            .unwrap();
        repo.set("main", "/c", &Object::string("3"), quick_opts("c"))
            .unwrap();

        repo.create_epoch("e1", "scoped work", vec![]).unwrap();
        repo.seal_epoch("e1", "ship").unwrap();

        // Now the attempt to rewind past a sealed commit must fail.
        let res = repo.set_ref("main", pre_seal);
        match res {
            Err(RepoError::EpochSealViolated { .. }) => {}
            other => panic!("expected EpochSealViolated, got {other:?}"),
        }
    }

    #[test]
    fn test_get_epoch_rehydrates_commits() {
        let repo = Repository::new(Box::new(
            SqliteStorage::in_memory().expect("in-memory sqlite"),
        ));
        repo.init().unwrap();
        repo.create_epoch("e1", "first epoch", vec![]).unwrap();
        repo.set_active_epoch(Some("e1".to_string())).unwrap();

        repo.set("main", "/a", &Object::string("1"), quick_opts("a"))
            .unwrap();
        repo.set("main", "/b", &Object::string("2"), quick_opts("b"))
            .unwrap();

        repo.set_active_epoch(None).unwrap();
        repo.seal_epoch("e1", "done").unwrap();

        let epoch = repo.get_epoch("e1").unwrap();
        assert_eq!(epoch.commits.len(), 2, "get_epoch must rehydrate 2 commits");
    }

    #[test]
    fn test_archive_epoch() {
        let repo = Repository::new(Box::new(
            SqliteStorage::in_memory().expect("in-memory sqlite"),
        ));
        repo.init().unwrap();
        repo.create_epoch("e1", "epoch", vec![]).unwrap();
        repo.seal_epoch("e1", "done").unwrap();

        repo.archive_epoch("e1").unwrap();

        let epoch = repo.get_epoch("e1").unwrap();
        assert_eq!(epoch.status, agentstategraph_core::EpochStatus::Archived);
    }

    #[test]
    fn test_archive_epoch_requires_sealed() {
        let repo = Repository::new(Box::new(
            SqliteStorage::in_memory().expect("in-memory sqlite"),
        ));
        repo.init().unwrap();
        repo.create_epoch("e1", "epoch", vec![]).unwrap();

        // Cannot archive an active epoch.
        assert!(repo.archive_epoch("e1").is_err());
    }

    #[test]
    fn test_export_epoch() {
        let repo = Repository::new(Box::new(
            SqliteStorage::in_memory().expect("in-memory sqlite"),
        ));
        repo.init().unwrap();
        repo.create_epoch("e1", "export test", vec![]).unwrap();
        repo.set_active_epoch(Some("e1".to_string())).unwrap();

        repo.set("main", "/x", &Object::string("hello"), quick_opts("x"))
            .unwrap();

        repo.set_active_epoch(None).unwrap();
        repo.seal_epoch("e1", "done").unwrap();

        let bundle = repo.export_epoch("e1").unwrap();
        assert_eq!(bundle["agentstategraph_export_version"], 1);
        assert_eq!(bundle["epoch"]["id"], "e1");
        assert!(bundle["commits"].is_array());
        assert_eq!(bundle["commits"].as_array().unwrap().len(), 1);
    }

    /// The checkpoint pattern: create and seal in one step, never made active,
    /// so NO commit is tagged to the epoch. This is how a per-plan checkpoint is
    /// minted. Iterating `commits` alone shipped a bundle with zero commit
    /// records while still advertising itself as self-contained, and reported
    /// "0 commits" in every listing.
    #[test]
    fn test_export_epoch_untagged_falls_back_to_sealed_commits() {
        let repo = Repository::new(Box::new(
            SqliteStorage::in_memory().expect("in-memory sqlite"),
        ));
        repo.init().unwrap();
        repo.set("main", "/x", &Object::string("hello"), quick_opts("x"))
            .unwrap();

        // Never set active — nothing is tagged to this epoch.
        repo.create_epoch("cp1", "checkpoint", vec![]).unwrap();
        repo.seal_epoch("cp1", "done").unwrap();

        let epoch = repo.get_epoch("cp1").unwrap();
        assert!(
            epoch.commits.is_empty(),
            "precondition: a checkpoint epoch tags no commits"
        );
        assert!(
            !epoch.sealed_commits.is_empty(),
            "sealing must record the covered set"
        );

        assert_eq!(epoch.commit_count(), epoch.sealed_commits.len());

        let bundle = repo.export_epoch("cp1").unwrap();
        let commits = bundle["commits"].as_array().unwrap();
        assert_eq!(
            commits.len(),
            epoch.sealed_commits.len(),
            "bundle must carry the sealed commit RECORDS, not just their ids"
        );
        assert!(
            commits.iter().all(|c| c.get("state_root").is_some()),
            "each entry must be a full commit record"
        );
    }

    /// The seal hash had a field on `Epoch` but no column, so it was dropped on
    /// every write and read back as `None` — the API advertised a
    /// tamper-evident digest that did not exist.
    #[test]
    fn test_seal_hash_is_persisted() {
        let repo = Repository::new(Box::new(
            SqliteStorage::in_memory().expect("in-memory sqlite"),
        ));
        repo.init().unwrap();
        repo.set("main", "/x", &Object::string("hello"), quick_opts("x"))
            .unwrap();
        repo.create_epoch("e1", "hashed", vec![]).unwrap();
        repo.seal_epoch("e1", "done").unwrap();

        let sealed = repo.get_epoch("e1").unwrap();
        let hash = sealed
            .seal_hash
            .expect("seal hash must survive the round trip");
        assert_eq!(
            hash,
            Repository::compute_seal_hash("e1", &sealed.sealed_commits),
            "persisted hash must match the digest of the sealed set"
        );
        assert_eq!(repo.list_epochs().unwrap()[0].seal_hash, Some(hash));
    }

    /// Domain separation: two epochs over an identical commit set must not
    /// share a digest, or the hash cannot identify what it sealed.
    #[test]
    fn test_seal_hash_is_domain_separated_and_order_independent() {
        let a = ObjectId::hash(b"a");
        let b = ObjectId::hash(b"b");
        assert_eq!(
            Repository::compute_seal_hash("e1", &[a, b]),
            Repository::compute_seal_hash("e1", &[b, a]),
            "walk order must not change the digest"
        );
        assert_ne!(
            Repository::compute_seal_hash("e1", &[a, b]),
            Repository::compute_seal_hash("e2", &[a, b]),
            "different epochs over the same set must differ"
        );
    }

    /// commit_count was written once at create time, from a list that is empty
    /// then, and never updated on seal.
    #[test]
    fn test_seal_updates_commit_count() {
        let repo = Repository::new(Box::new(
            SqliteStorage::in_memory().expect("in-memory sqlite"),
        ));
        repo.init().unwrap();
        repo.set("main", "/x", &Object::string("hello"), quick_opts("x"))
            .unwrap();
        repo.create_epoch("e1", "counted", vec![]).unwrap();
        repo.seal_epoch("e1", "done").unwrap();

        let entry = repo.list_epochs().unwrap().into_iter().next().unwrap();
        assert!(
            entry.commit_count > 0,
            "sealed epoch must not report 0 commits"
        );
        assert_eq!(
            entry.commit_count,
            repo.get_epoch("e1").unwrap().sealed_commits.len()
        );
    }

    /// A Plan-scoped epoch is defined by the commits tagged to it. Sealing must
    /// not widen it to everything reachable from the branch, or "this plan did
    /// N things" silently becomes "the workspace contains N things".
    #[test]
    fn test_plan_scope_seals_only_tagged_commits() {
        let repo = Repository::new(Box::new(
            SqliteStorage::in_memory().expect("in-memory sqlite"),
        ));
        repo.init().unwrap();
        // Work that predates the epoch and must NOT be swept in.
        repo.set("main", "/before", &Object::string("old"), quick_opts("a"))
            .unwrap();

        let epoch = agentstategraph_core::Epoch::new("p1", "plan scope", vec![])
            .with_scope(agentstategraph_core::EpochScope::Plan("p1".to_string()));
        repo.storage.create_epoch(&epoch).unwrap();
        repo.set_active_epoch(Some("p1".to_string())).unwrap();
        repo.set("main", "/during", &Object::string("new"), quick_opts("b"))
            .unwrap();
        repo.set_active_epoch(None).unwrap();
        repo.seal_epoch("p1", "done").unwrap();

        let sealed = repo.get_epoch("p1").unwrap();
        assert_eq!(
            sealed.sealed_commits.len(),
            1,
            "plan scope seals only its own tagged commit, got {:?}",
            sealed.sealed_commits.len()
        );
        assert_eq!(
            sealed.scope,
            agentstategraph_core::EpochScope::Plan("p1".to_string())
        );
    }

    /// Scope and namespace must survive storage, or a reader cannot tell what a
    /// sealed epoch actually covered.
    #[test]
    fn test_scope_and_namespace_round_trip() {
        let repo = Repository::new(Box::new(
            SqliteStorage::in_memory().expect("in-memory sqlite"),
        ));
        repo.init().unwrap();
        let epoch = agentstategraph_core::Epoch::new("w1", "workspace scope", vec![])
            .with_scope(agentstategraph_core::EpochScope::Workspace(
                "ctxone".to_string(),
            ))
            .with_namespace("ctxone");
        repo.storage.create_epoch(&epoch).unwrap();

        let loaded = repo.get_epoch("w1").unwrap();
        assert_eq!(
            loaded.scope,
            agentstategraph_core::EpochScope::Workspace("ctxone".to_string())
        );
        assert_eq!(loaded.namespace.as_deref(), Some("ctxone"));

        // This repo is on the default namespace, so a `ctxone` epoch is not
        // its business — that is the whole point of scoping the listing.
        assert!(
            repo.list_epochs().unwrap().is_empty(),
            "another workspace's epoch must not appear in this one's listing"
        );
        let entry = repo
            .list_all_epochs()
            .unwrap()
            .into_iter()
            .next()
            .expect("cross-workspace view still sees it");
        assert_eq!(entry.namespace.as_deref(), Some("ctxone"));
    }

    /// The job this exists for: an epoch sealed before the namespace column
    /// existed carries no owner, so it reads as belonging to the default
    /// workspace. A caller that knows better can settle it once.
    #[test]
    fn test_assign_namespace_to_a_legacy_epoch() {
        let repo = Repository::new(Box::new(
            SqliteStorage::in_memory().expect("in-memory sqlite"),
        ));
        repo.init().unwrap();
        repo.set("main", "/a", &Object::string("1"), quick_opts("a"))
            .unwrap();

        // A pre-namespace epoch: sealed, and owned by nobody.
        let legacy = agentstategraph_core::Epoch::new("plan:ctxone:p1", "legacy", vec![]);
        assert!(legacy.namespace.is_none());
        repo.storage.create_epoch(&legacy).unwrap();
        repo.seal_epoch("plan:ctxone:p1", "sealed").unwrap();

        assert!(
            repo.assign_epoch_namespace("plan:ctxone:p1", "ctxone")
                .unwrap(),
            "a sealed epoch with no owner can be assigned one"
        );
        assert_eq!(
            repo.get_epoch("plan:ctxone:p1")
                .unwrap()
                .namespace
                .as_deref(),
            Some("ctxone")
        );
    }

    /// THE security property. An epoch binds writes in its own workspace, so a
    /// reassign would let a caller relabel a sealed epoch into another
    /// workspace and change whose writes it guards. Once set, an owner is final.
    #[test]
    fn test_assign_namespace_refuses_to_move_an_owned_epoch() {
        let repo = Repository::new(Box::new(
            SqliteStorage::in_memory().expect("in-memory sqlite"),
        ));
        repo.init().unwrap();
        repo.set("main", "/a", &Object::string("1"), quick_opts("a"))
            .unwrap();

        // create_epoch stamps the active namespace, so this one has an owner.
        repo.create_epoch("owned", "has an owner", vec![]).unwrap();
        let before = repo.get_epoch("owned").unwrap().namespace;
        assert!(before.is_some());

        assert!(
            !repo
                .assign_epoch_namespace("owned", "somewhere-else")
                .unwrap(),
            "an epoch that already has an owner must report no-change"
        );
        assert_eq!(
            repo.get_epoch("owned").unwrap().namespace,
            before,
            "and must be left exactly as it was"
        );
    }

    /// Assigning does not disturb what was sealed — the commit set and the
    /// tamper-evident digest must survive untouched, or backfilling ownership
    /// would silently invalidate the audit trail.
    #[test]
    fn test_assign_namespace_leaves_the_seal_intact() {
        let repo = Repository::new(Box::new(
            SqliteStorage::in_memory().expect("in-memory sqlite"),
        ));
        repo.init().unwrap();
        repo.set("main", "/a", &Object::string("1"), quick_opts("a"))
            .unwrap();
        let legacy = agentstategraph_core::Epoch::new("plan:ws:p1", "legacy", vec![]);
        repo.storage.create_epoch(&legacy).unwrap();
        repo.seal_epoch("plan:ws:p1", "sealed").unwrap();

        let before = repo.get_epoch("plan:ws:p1").unwrap();
        repo.assign_epoch_namespace("plan:ws:p1", "ws").unwrap();
        let after = repo.get_epoch("plan:ws:p1").unwrap();

        assert_eq!(
            after.seal_hash, before.seal_hash,
            "seal hash must not change"
        );
        assert_eq!(
            after.sealed_commits, before.sealed_commits,
            "sealed commit set must not change"
        );
        assert_eq!(after.status, before.status);
        assert_eq!(after.sealed_at, before.sealed_at);
    }

    /// A typo'd id must be reported, not swallowed as a silent `false` — a
    /// caller backfilling known ids needs to hear about it.
    #[test]
    fn test_assign_namespace_errors_on_unknown_epoch() {
        let repo = Repository::new(Box::new(
            SqliteStorage::in_memory().expect("in-memory sqlite"),
        ));
        repo.init().unwrap();
        assert!(repo.assign_epoch_namespace("nope", "ws").is_err());
    }

    /// A malformed name would be written and then never match a real workspace
    /// on read, so it is rejected up front rather than stored.
    #[test]
    fn test_assign_namespace_rejects_an_invalid_name() {
        let repo = Repository::new(Box::new(
            SqliteStorage::in_memory().expect("in-memory sqlite"),
        ));
        repo.init().unwrap();
        let legacy = agentstategraph_core::Epoch::new("e1", "legacy", vec![]);
        repo.storage.create_epoch(&legacy).unwrap();
        assert!(matches!(
            repo.assign_epoch_namespace("e1", "not a valid ns"),
            Err(RepoError::InvalidOperation(_))
        ));
    }

    /// An epoch with no recorded owner binds NOTHING, and assigning ownership
    /// is what activates enforcement.
    #[test]
    fn test_orphan_epoch_binds_nothing_until_assigned() {
        let repo = Repository::new(Box::new(
            SqliteStorage::in_memory().expect("in-memory sqlite"),
        ));
        repo.init().unwrap();
        let base = repo
            .set("main", "/a", &Object::string("1"), quick_opts("a"))
            .unwrap();
        repo.set("main", "/b", &Object::string("2"), quick_opts("b"))
            .unwrap();

        let legacy = agentstategraph_core::Epoch::new("plan:orphan:p1", "legacy", vec![]);
        repo.storage.create_epoch(&legacy).unwrap();
        repo.seal_epoch("plan:orphan:p1", "sealed").unwrap();

        // No owner recorded, so it cannot establish that a rewind would orphan
        // commits in the workspace that sealed them. Treating such an epoch as
        // binding the default workspace made it an unsatisfiable veto on every
        // default write — on a real store, 171 of them blocked startup.
        assert!(
            repo.check_epoch_seal_violations(&base).unwrap().is_empty(),
            "an epoch with no owner must not veto any workspace's writes"
        );

        // Assigning ownership activates enforcement. This repo is on the
        // default workspace, so assigning it here makes it bind.
        repo.assign_epoch_namespace("plan:orphan:p1", Namespace::DEFAULT)
            .unwrap();
        assert!(
            !repo.check_epoch_seal_violations(&base).unwrap().is_empty(),
            "once owned by this workspace, the seal guards it"
        );
    }

    /// The bug this scoping exists to kill: one workspace's sealed epoch used
    /// to reject ref updates in EVERY workspace, so completing a plan in one
    /// project could block writes across all of them.
    #[test]
    fn test_sealed_epoch_does_not_bind_other_workspaces() {
        let repo = Repository::new(Box::new(
            SqliteStorage::in_memory().expect("in-memory sqlite"),
        ));
        repo.init().unwrap();
        let base = repo
            .set("main", "/a", &Object::string("1"), quick_opts("a"))
            .unwrap();
        repo.set("main", "/b", &Object::string("2"), quick_opts("b"))
            .unwrap();

        // A sealed epoch owned by a DIFFERENT workspace.
        let other = agentstategraph_core::Epoch::new("other", "elsewhere", vec![])
            .with_namespace("some-other-workspace");
        repo.storage.create_epoch(&other).unwrap();
        repo.seal_epoch("other", "sealed").unwrap();

        // Rewinding main here would orphan that epoch's commits, but the epoch
        // belongs to another workspace and must not have a say.
        assert!(
            repo.check_epoch_seal_violations(&base).unwrap().is_empty(),
            "an epoch from another workspace must not veto this one's writes"
        );
    }

    /// An epoch that DOES own this workspace still binds it — scoping must not
    /// disable enforcement altogether.
    #[test]
    fn test_sealed_epoch_still_binds_its_own_workspace() {
        let repo = Repository::new(Box::new(
            SqliteStorage::in_memory().expect("in-memory sqlite"),
        ));
        repo.init().unwrap();
        let base = repo
            .set("main", "/a", &Object::string("1"), quick_opts("a"))
            .unwrap();
        repo.set("main", "/b", &Object::string("2"), quick_opts("b"))
            .unwrap();

        // create_epoch stamps the active namespace.
        repo.create_epoch("mine", "here", vec![]).unwrap();
        repo.seal_epoch("mine", "sealed").unwrap();

        assert!(
            !repo.check_epoch_seal_violations(&base).unwrap().is_empty(),
            "a rewind that orphans this workspace's own sealed commits must be caught"
        );
    }

    #[test]
    fn test_export_epoch_rejects_active() {
        let repo = Repository::new(Box::new(
            SqliteStorage::in_memory().expect("in-memory sqlite"),
        ));
        repo.init().unwrap();
        repo.create_epoch("e1", "active", vec![]).unwrap();

        assert!(matches!(
            repo.export_epoch("e1"),
            Err(RepoError::InvalidOperation(_))
        ));
    }

    #[test]
    fn test_intent_tree_threads_by_parent_intent() {
        let repo = test_repo();

        // A parent intent, then a child intent decomposed from it via
        // `Intent::with_parent` — the RFC's sub-intent threading.
        let parent_opts = CommitOptions::new("agent/test", IntentCategory::Explore, "parent goal");
        let parent_intent_id = parent_opts.intent.id.clone();
        repo.set("main", "/a", &Object::string("1"), parent_opts)
            .unwrap();

        let mut child_opts = CommitOptions::new("agent/test", IntentCategory::Refine, "child goal");
        child_opts.intent = child_opts.intent.with_parent(parent_intent_id.clone());
        repo.set("main", "/b", &Object::string("2"), child_opts)
            .unwrap();

        let tree = repo.intent_tree("main", None).unwrap();
        let roots = tree["roots"].as_array().unwrap();

        // The child threads UNDER the parent — it must not appear as a root.
        assert!(
            roots.iter().all(|r| r["description"] != "child goal"),
            "child intent must not be a root; it is decomposed from a parent"
        );

        // The parent root carries the child among its children, matched by
        // intent id (NOT by commit lineage).
        let parent_node = roots
            .iter()
            .find(|r| r["description"] == "parent goal")
            .expect("parent intent should be a root");
        let children = parent_node["children"].as_array().unwrap();
        assert_eq!(
            children.len(),
            1,
            "parent should have exactly one sub-intent"
        );
        assert_eq!(children[0]["description"], "child goal");
        assert_eq!(parent_node["intent_id"], parent_intent_id.as_str());
    }

    #[test]
    fn test_intent_tree_flat_without_threading() {
        // With no parent_intent set (today's reality), every intent is its own
        // root — an honest flat list, not a fabricated hierarchy.
        let repo = test_repo();
        repo.set("main", "/a", &Object::string("1"), quick_opts("first"))
            .unwrap();
        repo.set("main", "/b", &Object::string("2"), quick_opts("second"))
            .unwrap();

        let tree = repo.intent_tree("main", None).unwrap();
        let roots = tree["roots"].as_array().unwrap();
        assert!(
            roots
                .iter()
                .all(|r| r["children"].as_array().unwrap().is_empty()),
            "no threading recorded → no node should have children"
        );
        // Both of our writes (plus the init commit) are top-level roots.
        assert!(roots.iter().any(|r| r["description"] == "first"));
        assert!(roots.iter().any(|r| r["description"] == "second"));
    }

    #[test]
    fn test_commit_persists_tool_calls() {
        let repo = test_repo();

        let tc = ToolCall {
            tool_name: "kubectl_apply".to_string(),
            arguments: serde_json::json!({ "file": "deploy.yaml" }),
            result: Some("configured".to_string()),
            timestamp: chrono::Utc::now(),
        };
        let opts = CommitOptions::new("agent/test", IntentCategory::Fix, "apply manifest")
            .with_tool_calls(vec![tc.clone()]);
        let commit_id = repo
            .set("main", "/deploy", &Object::string("ok"), opts)
            .unwrap();

        // Round-trips out of storage on the persisted commit.
        let commit = repo.get_commit(&commit_id).unwrap().unwrap();
        assert_eq!(commit.tool_calls.len(), 1);
        assert_eq!(commit.tool_calls[0].tool_name, "kubectl_apply");
        assert_eq!(commit.tool_calls[0].result.as_deref(), Some("configured"));
    }

    #[test]
    fn test_commit_without_tool_calls_is_empty() {
        let repo = test_repo();
        let commit_id = repo
            .set("main", "/a", &Object::string("1"), quick_opts("no tools"))
            .unwrap();
        let commit = repo.get_commit(&commit_id).unwrap().unwrap();
        assert!(commit.tool_calls.is_empty());
    }

    #[test]
    fn test_commit_authority_principal_defaults_to_agent() {
        // A commit records the acting agent as its authorizing principal — not
        // the old constant "default" (Plan C t-002).
        let repo = test_repo();
        let opts = CommitOptions::new("human/alice", IntentCategory::Fix, "patch");
        let commit_id = repo.set("main", "/a", &Object::string("1"), opts).unwrap();

        let commit = repo.get_commit(&commit_id).unwrap().unwrap();
        assert_eq!(commit.agent_id, "human/alice");
        assert_eq!(
            commit.authority.principal, "human/alice",
            "principal should default to the acting agent, not a constant"
        );
        assert_ne!(commit.authority.principal, "default");
    }

    #[test]
    fn test_commit_with_principal_overrides_actor() {
        // When the authorizer differs from the actor, `with_principal` wins for
        // the authority while `agent_id` still records who acted.
        let repo = test_repo();
        let opts = CommitOptions::new("agent/bot", IntentCategory::Refine, "act")
            .with_principal("human/alice");
        let commit_id = repo.set("main", "/a", &Object::string("1"), opts).unwrap();

        let commit = repo.get_commit(&commit_id).unwrap().unwrap();
        assert_eq!(commit.agent_id, "agent/bot");
        assert_eq!(commit.authority.principal, "human/alice");
    }

    fn refine_count(rows: &[HistoryRollupRow], agent: &str) -> Option<i64> {
        rows.iter()
            .find(|r| r.agent_id == agent && r.intent_category == "Refine")
            .map(|r| r.commit_count)
    }

    #[test]
    fn test_extract_history_rollup_and_milestones() {
        let repo = test_repo();
        let mk = |agent: &str, cat: IntentCategory, path: &str, desc: &str| {
            let opts = CommitOptions::new(agent, cat, desc);
            repo.set("main", path, &Object::string("v"), opts).unwrap();
        };
        mk("alice", IntentCategory::Refine, "/a", "r1");
        mk("alice", IntentCategory::Refine, "/b", "r2");
        mk("bob", IntentCategory::Fix, "/c", "f1");
        mk("alice", IntentCategory::Checkpoint, "/d", "ship it");

        // batch_size 2 forces the extractor to loop across several batches.
        let report = repo.extract_history(2).unwrap();
        assert!(report.commits_processed >= 4, "processed all our commits");
        assert!(report.cursor > 0);

        let rollup = repo.history_rollup().unwrap();
        let bucket = |agent: &str, cat: &str| {
            rollup
                .iter()
                .find(|r| r.agent_id == agent && r.intent_category == cat)
                .map(|r| r.commit_count)
        };
        assert_eq!(bucket("alice", "Refine"), Some(2));
        assert_eq!(bucket("bob", "Fix"), Some(1));
        assert_eq!(bucket("alice", "Checkpoint"), Some(1));
        // No sessions → nothing can be attributed, and the rollup now SAYS so
        // rather than claiming "default".
        //
        // This assertion used to read `== "default"`, which encoded the old
        // fallback as intended behaviour. It was deliberate but wrong in effect:
        // "default" is itself a real namespace, so a fallback was
        // indistinguishable from a genuine attribution. On a store where nothing
        // writes sessions that filed every commit under "default" — measured on
        // the AgentStateDeveloper store, all 39,446, though ~35,000 belong to
        // another namespace.
        assert!(
            rollup
                .iter()
                .all(|r| r.namespace == agentstategraph_storage::HISTORY_NAMESPACE_UNKNOWN)
        );

        let miles = repo.history_milestones(100).unwrap();
        assert!(
            miles
                .iter()
                .any(|m| m.description == "ship it" && m.kind == "checkpoint"),
            "the Checkpoint commit is on the milestone timeline"
        );
        // Non-checkpoint commits are not milestones.
        assert!(miles.iter().all(|m| m.description != "r1"));
    }

    #[test]
    fn test_extract_history_incremental_and_idempotent() {
        let repo = test_repo();
        repo.set(
            "main",
            "/a",
            &Object::string("1"),
            CommitOptions::new("alice", IntentCategory::Refine, "r1"),
        )
        .unwrap();

        let r1 = repo.extract_history(100).unwrap();
        assert!(r1.commits_processed >= 1);
        let before = repo.history_rollup().unwrap();
        assert_eq!(refine_count(&before, "alice"), Some(1));

        // Re-running with no new commits is a no-op and doesn't double-count.
        let r2 = repo.extract_history(100).unwrap();
        assert_eq!(r2.commits_processed, 0);
        assert_eq!(r2.cursor, r1.cursor);
        assert_eq!(
            refine_count(&repo.history_rollup().unwrap(), "alice"),
            Some(1)
        );

        // A new commit is folded incrementally — only it is processed.
        repo.set(
            "main",
            "/b",
            &Object::string("2"),
            CommitOptions::new("alice", IntentCategory::Refine, "r2"),
        )
        .unwrap();
        let r3 = repo.extract_history(100).unwrap();
        assert_eq!(r3.commits_processed, 1);
        assert!(r3.cursor > r1.cursor);
        assert_eq!(
            refine_count(&repo.history_rollup().unwrap(), "alice"),
            Some(2)
        );
    }

    #[test]
    fn test_history_report_aggregates_views() {
        let repo = test_repo();
        let mk = |agent: &str, cat: IntentCategory, path: &str, desc: &str| {
            repo.set(
                "main",
                path,
                &Object::string("v"),
                CommitOptions::new(agent, cat, desc),
            )
            .unwrap();
        };
        mk("alice", IntentCategory::Refine, "/a", "r1");
        mk("alice", IntentCategory::Refine, "/b", "r2");
        mk("bob", IntentCategory::Fix, "/c", "f1");
        mk("alice", IntentCategory::Checkpoint, "/d", "ship it");

        // refresh=true runs the extractor first, so the report is current even
        // though we never called extract_history explicitly.
        let report = repo.history_report(None, "day", 50, true, false).unwrap();

        // Intent mix, looked up by name (init also writes a Checkpoint, so we
        // assert our own categories rather than a leaderboard position).
        let mix = report["intent_mix"].as_array().unwrap();
        let mix_of = |cat: &str| {
            mix.iter()
                .find(|e| e["category"] == cat)
                .and_then(|e| e["commits"].as_i64())
        };
        assert_eq!(mix_of("Refine"), Some(2));
        assert_eq!(mix_of("Fix"), Some(1));

        // Authorship: our three "alice" commits and one "bob".
        let auth = report["authorship"].as_array().unwrap();
        let auth_of = |a: &str| {
            auth.iter()
                .find(|e| e["agent"] == a)
                .and_then(|e| e["commits"].as_i64())
        };
        assert_eq!(auth_of("alice"), Some(3));
        assert_eq!(auth_of("bob"), Some(1));

        // Velocity is a single day here; all commits land in one bucket.
        assert_eq!(report["velocity"]["by"], "day");
        assert_eq!(report["velocity"]["series"].as_array().unwrap().len(), 1);

        // The Checkpoint shows on the milestone timeline.
        let miles = report["milestones"].as_array().unwrap();
        assert!(miles.iter().any(|m| m["description"] == "ship it"));

        // refresh=false is a pure read — no extraction; week rollup collapses
        // the single day into one ISO-week bucket.
        let again = repo.history_report(None, "week", 50, false, false).unwrap();
        assert_eq!(again["velocity"]["by"], "week");
        assert_eq!(again["velocity"]["series"].as_array().unwrap().len(), 1);
        let again_auth = again["authorship"].as_array().unwrap();
        assert!(
            again_auth
                .iter()
                .any(|e| e["agent"] == "alice" && e["commits"] == 3)
        );

        // store=false omits the block; store=true attaches it.
        assert!(report.get("store_shape").is_none());
        let with_store = repo.history_report(None, "day", 50, false, true).unwrap();
        assert!(with_store["store_shape"].is_object());
    }

    #[test]
    fn test_history_store_shape() {
        let repo = test_repo();
        for i in 0..5 {
            repo.set(
                "main",
                &format!("/k{i}"),
                &Object::string("v"),
                CommitOptions::new("alice", IntentCategory::Refine, "r"),
            )
            .unwrap();
        }

        let shape = repo.history_store_shape().unwrap();
        // Counts are positive and bytes reflect a non-empty DB.
        assert!(shape["commits"].as_i64().unwrap() >= 5);
        assert!(shape["objects"].as_i64().unwrap() > 0);
        assert!(shape["total_bytes"].as_i64().unwrap() > 0);
        // Path-copy amplification = objects / commits > 0.
        assert!(
            shape["path_copy_amplification"]["objects_per_commit"]
                .as_f64()
                .unwrap()
                > 0.0
        );
        // Per-table breakdown is present only when the build exposes dbstat;
        // either way `tables` is an array and the flag matches its emptiness
        // expectation.
        assert!(shape["tables"].is_array());
        if shape["dbstat_available"].as_bool().unwrap() {
            assert!(!shape["tables"].as_array().unwrap().is_empty());
        }
    }

    #[test]
    fn test_history_retention_hooks() {
        let repo = test_repo();
        // A Checkpoint that opts into pinning its snapshot, plus a couple of
        // plain edits. Pinning is opt-in: see `test_checkpoint_pins_state_only_when_tagged`.
        let cp = repo
            .set(
                "main",
                "/a",
                &Object::string("1"),
                CommitOptions::new("alice", IntentCategory::Checkpoint, "ship")
                    .with_tags(vec![TAG_PIN_STATE.to_string()]),
            )
            .unwrap();
        let refine = repo
            .set(
                "main",
                "/b",
                &Object::string("2"),
                CommitOptions::new("alice", IntentCategory::Refine, "tweak"),
            )
            .unwrap();

        // Before extraction nothing is distilled and nothing is retained.
        assert!(!repo.history_is_distilled(&refine).unwrap());
        assert!(repo.history_retained_state_roots().unwrap().is_empty());

        repo.extract_history(100).unwrap();

        // Every folded commit is now "distilled" — GC's prune predicate.
        assert!(repo.history_is_distilled(&cp).unwrap());
        assert!(repo.history_is_distilled(&refine).unwrap());
        // A commit that doesn't exist is not distilled.
        assert!(!repo.history_is_distilled(&ObjectId::hash(b"nope")).unwrap());

        // The Checkpoint's state_root is retained (GC must keep it reachable),
        // and it matches the commit's actual state_root.
        let roots = repo.history_retained_state_roots().unwrap();
        let cp_commit = repo.get_commit(&cp).unwrap().unwrap();
        assert!(
            roots.contains(&cp_commit.state_root),
            "milestone state_root must be retained for GC"
        );

        // And it surfaces on the milestone timeline in the report.
        let report = repo.history_report(None, "day", 50, false, false).unwrap();
        let ship = report["milestones"]
            .as_array()
            .unwrap()
            .iter()
            .find(|m| m["description"] == "ship")
            .unwrap();
        assert_eq!(ship["state_root"], cp_commit.state_root.short());
    }

    /// The merge base must cost time linear in the two histories. It used to
    /// recompute generation depth from scratch for every common ancestor, and
    /// each recomputation re-read every commit on every pass of a fixed-point
    /// loop — O(n³) storage reads. A CTX branch with 1,304 common ancestors
    /// never finished, and since every read holds the storage connection lock
    /// it starved every other request to the store. At 1,500 shared commits the
    /// old code was still running when killed after 120 seconds.
    ///
    /// 3,000 commits of shared history, then a short branch on each side,
    /// written straight to storage so the test stays fast. The bound is loose
    /// on purpose — the fix takes tens of milliseconds — and exists only to
    /// fail loudly if the old behaviour returns.
    #[test]
    fn test_merge_base_is_linear_on_a_long_shared_history() {
        let repo = test_repo();
        let state = repo
            .get_commit(&repo.resolve_ref("main").unwrap())
            .unwrap()
            .unwrap()
            .state_root;
        let commit = |parent: ObjectId, what: &str| {
            let c = CommitBuilder::new(
                state,
                "agent/test",
                Authority::simple("agent/test"),
                Intent::new(IntentCategory::Refine, what),
            )
            .parent(parent)
            .build();
            repo.storage.put_commit(&c).unwrap();
            c.id
        };

        let mut tip = repo.resolve_ref("main").unwrap();
        for i in 0..3_000 {
            tip = commit(tip, &format!("shared {i}"));
        }
        let branch_point = tip;
        let (mut feature, mut main) = (branch_point, branch_point);
        for i in 0..3 {
            feature = commit(feature, &format!("feature {i}"));
            main = commit(main, &format!("main {i}"));
        }

        let started = std::time::Instant::now();
        let base = repo.find_common_ancestor(&feature, &main).unwrap();
        let elapsed = started.elapsed();

        assert_eq!(base, branch_point);
        assert!(
            elapsed < std::time::Duration::from_secs(10),
            "merge base over 3,000 shared commits took {elapsed:?}; it must be linear"
        );
    }

    /// The linear merge base must choose exactly what the old definition did:
    /// among all common ancestors, the greatest generation (longest path back
    /// to a root), ties to the larger id. Checked against a deliberately naive
    /// reference over random DAGs with merge commits — which produce
    /// criss-cross histories with several best common ancestors — plus the odd
    /// extra root (disjoint histories) and dangling parent (a pruned commit).
    #[test]
    fn test_merge_base_matches_the_naive_definition_on_random_dags() {
        use std::collections::{HashMap, HashSet};

        fn naive_ancestors(
            parents: &HashMap<ObjectId, Vec<ObjectId>>,
            id: ObjectId,
            out: &mut HashSet<ObjectId>,
        ) {
            if out.insert(id) {
                for p in parents.get(&id).into_iter().flatten() {
                    naive_ancestors(parents, *p, out);
                }
            }
        }
        // The definition, verbatim: one more than the deepest parent, 0 for a
        // root. Memoized only so merge-heavy graphs stay fast to check.
        fn naive_generation(
            parents: &HashMap<ObjectId, Vec<ObjectId>>,
            id: ObjectId,
            memo: &mut HashMap<ObjectId, u64>,
        ) -> u64 {
            if let Some(g) = memo.get(&id) {
                return *g;
            }
            let g = parents
                .get(&id)
                .into_iter()
                .flatten()
                .map(|p| naive_generation(parents, *p, memo) + 1)
                .max()
                .unwrap_or(0);
            memo.insert(id, g);
            g
        }

        let mut seed: u64 = 0x9E37_79B9_7F4A_7C15;
        let mut next = move |bound: u64| {
            seed ^= seed << 13;
            seed ^= seed >> 7;
            seed ^= seed << 17;
            seed % bound.max(1)
        };
        let node = |i: u64| ObjectId::hash(&i.to_le_bytes());

        let mut compared = 0;
        for _graph in 0..60 {
            let n = 4 + next(28);
            let mut parents: HashMap<ObjectId, Vec<ObjectId>> = HashMap::new();
            parents.insert(node(0), vec![]);
            for i in 1..n {
                let ps = match next(20) {
                    0 => vec![],                // another root
                    1 => vec![node(1_000 + i)], // dangling parent
                    2..=6 => {
                        let (x, y) = (next(i), next(i));
                        if x == y {
                            vec![node(x)]
                        } else {
                            vec![node(x), node(y)]
                        } // merge
                    }
                    _ => vec![node(next(i))],
                };
                parents.insert(node(i), ps);
            }
            for a in 0..n {
                for b in 0..n {
                    let (a, b) = (node(a), node(b));
                    let (mut of_a, mut of_b) = (HashSet::new(), HashSet::new());
                    naive_ancestors(&parents, a, &mut of_a);
                    naive_ancestors(&parents, b, &mut of_b);
                    let mut memo = HashMap::new();
                    let expected = of_a
                        .intersection(&of_b)
                        .map(|id| (naive_generation(&parents, *id, &mut memo), *id))
                        .max()
                        .map(|(_, id)| id);
                    assert_eq!(
                        deepest_common_ancestor(&parents, &a, &b),
                        expected,
                        "merge base disagrees with the definition"
                    );
                    compared += 1;
                }
            }
        }
        assert!(compared > 5_000, "only {compared} pairs compared");
    }

    /// A unique on-disk path for tests that need two connections to one store;
    /// an in-memory database is private to the connection that opened it.
    fn temp_db_path(tag: &str) -> std::path::PathBuf {
        let nanos = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        std::env::temp_dir().join(format!("asg-{tag}-{}-{nanos}.db", std::process::id()))
    }

    fn remove_db(path: &std::path::Path) {
        for suffix in ["", "-wal", "-shm"] {
            let _ = std::fs::remove_file(format!("{}{suffix}", path.display()));
        }
    }

    /// The race: a sweep computes its keep-set, another connection commits, and
    /// the sweep then deletes that commit's objects — they are reachable only
    /// from a ref tip the keep-set never saw. In the field the other connection
    /// is a git hook running `asd index`, `asd-serve`, or an MCP server, and the
    /// window is the whole mark phase (minutes on a large store). The result is
    /// a ref whose state tree has dangling references.
    ///
    /// Driven deterministically at the storage boundary rather than by racing
    /// threads: the keep-set is computed, the second connection commits, and
    /// only then does the sweep run, exactly as the interleaving would have it.
    #[test]
    fn test_gc_sweep_keeps_state_committed_after_the_keep_set_was_computed() {
        let path = temp_db_path("gc-race");
        let a = Repository::new(Box::new(SqliteStorage::open(&path).unwrap()));
        a.init().unwrap();
        // History, so the sweep has something to reclaim.
        for i in 0..5 {
            a.set(
                "main",
                "/a",
                &Object::string(format!("v{i}")),
                quick_opts("edit"),
            )
            .unwrap();
        }
        a.extract_history(100).unwrap();
        let keep = a
            .gc_keep_roots(RetentionPolicy {
                keep_recent: 1,
                checkpoint_every: 0,
                keep_milestones: false,
            })
            .unwrap();

        // Another process commits new state after that keep-set was computed.
        let b = Repository::new(Box::new(SqliteStorage::open(&path).unwrap()));
        let late = b
            .set(
                "main",
                "/late",
                &Object::string("written mid-sweep"),
                quick_opts("late write"),
            )
            .unwrap();

        a.storage.history_gc_sweep(&keep, GC_MARK_BATCH).unwrap();

        let missing = b.first_missing_object(&late).unwrap();
        remove_db(&path);
        assert_eq!(
            missing, None,
            "the sweep deleted state reachable from the live ref tip — a commit \
             that landed after the keep-set was computed must survive"
        );
    }

    /// A mutating sweep must hand the write lock back on every path — having
    /// refused, and having swept and vacuumed — or every other writer to the
    /// store waits out its busy timeout and then fails.
    #[test]
    fn test_gc_sweep_releases_the_writer_lock_on_every_path() {
        let path = temp_db_path("gc-release");
        // No grace period, so the sweep below has something to delete.
        let a = Repository::new(Box::new(
            SqliteStorage::open(&path)
                .unwrap()
                .with_gc_grace(std::time::Duration::ZERO),
        ));
        a.init().unwrap();
        for i in 0..3 {
            a.set(
                "main",
                "/a",
                &Object::string(format!("v{i}")),
                quick_opts("edit"),
            )
            .unwrap();
        }
        let b = Repository::new(Box::new(SqliteStorage::open(&path).unwrap()));

        // Refused: nothing is distilled yet.
        let refused = a.gc_sweep(RetentionPolicy::default(), true, false).unwrap();
        assert_eq!(refused["refused"], true, "{refused}");
        b.set(
            "main",
            "/b",
            &Object::string("after refusal"),
            quick_opts("b1"),
        )
        .expect("the refused sweep released the lock");

        // Swept, then vacuumed — VACUUM runs after the lock is released.
        a.extract_history(100).unwrap();
        let swept = a
            .gc_sweep(
                RetentionPolicy {
                    keep_recent: 1,
                    checkpoint_every: 0,
                    keep_milestones: false,
                },
                true,
                true,
            )
            .unwrap();
        assert_eq!(swept["mutated"], true, "{swept}");
        assert!(swept["objects_deleted"].as_i64().unwrap() > 0, "{swept}");
        assert!(swept.get("vacuum").is_some(), "{swept}");
        let late = b
            .set(
                "main",
                "/c",
                &Object::string("after sweep"),
                quick_opts("b2"),
            )
            .expect("the completed sweep released the lock");

        // And nothing live was lost along the way.
        let missing = b.first_missing_object(&late).unwrap();
        let kept = b.get_json("main", "/b").unwrap();
        remove_db(&path);
        assert_eq!(missing, None);
        assert_eq!(kept, serde_json::json!("after refusal"));
    }

    /// Pinning a snapshot is opt-in. A routine checkpoint — a re-index, a
    /// sidecar hydrate — still earns its place on the milestone timeline, but
    /// it must not pin a state tree: doing that on every run is what grows a
    /// store into millions of objects with nothing reclaimable. What it carries
    /// instead is the revision it came from, so the state can be rebuilt.
    #[test]
    fn test_checkpoint_pins_state_only_when_tagged() {
        let repo = test_repo();

        let routine = repo
            .set(
                "main",
                "/a",
                &Object::string("1"),
                CommitOptions::new("asd", IntentCategory::Checkpoint, "asd index: 9 symbols")
                    .with_tags(vec![format!("{TAG_GIT_REVISION}c0ffee1234567890")]),
            )
            .unwrap();
        let pinned = repo
            .set(
                "main",
                "/b",
                &Object::string("2"),
                CommitOptions::new("alice", IntentCategory::Checkpoint, "v1.0 release")
                    .with_tags(vec![TAG_PIN_STATE.to_string()]),
            )
            .unwrap();
        repo.extract_history(100).unwrap();

        let routine_commit = repo.get_commit(&routine).unwrap().unwrap();
        let pinned_commit = repo.get_commit(&pinned).unwrap().unwrap();

        // Only the tagged checkpoint is a GC root.
        let roots = repo.history_retained_state_roots().unwrap();
        assert!(
            roots.contains(&pinned_commit.state_root),
            "a checkpoint tagged {TAG_PIN_STATE} must pin its snapshot"
        );
        assert!(
            !roots.contains(&routine_commit.state_root),
            "an untagged checkpoint must not pin a snapshot — that is the \
             retention leak this default exists to prevent"
        );

        // Both are still distilled, and both still appear on the timeline: the
        // routine one is a real milestone, it just names no snapshot.
        assert!(repo.history_is_distilled(&routine).unwrap());
        assert!(repo.history_is_distilled(&pinned).unwrap());
        let milestones = repo.history_milestones(50).unwrap();
        let routine_row = milestones
            .iter()
            .find(|m| m.commit_id == routine)
            .expect("routine checkpoint still earns a milestone row");
        let pinned_row = milestones
            .iter()
            .find(|m| m.commit_id == pinned)
            .expect("pinned checkpoint is on the timeline");
        assert!(routine_row.state_root.is_none());
        assert_eq!(pinned_row.state_root, Some(pinned_commit.state_root));

        // The unpinned milestone records how to rebuild what it dropped.
        assert_eq!(routine_row.git_sha.as_deref(), Some("c0ffee1234567890"));
        assert!(
            pinned_row.git_sha.is_none(),
            "no git tag was supplied for the pinned checkpoint"
        );

        // And the revision surfaces on the report the UI reads.
        let report = repo.history_report(None, "day", 50, false, false).unwrap();
        let routine_json = report["milestones"]
            .as_array()
            .unwrap()
            .iter()
            .find(|m| m["description"] == "asd index: 9 symbols")
            .unwrap();
        assert_eq!(routine_json["git_sha"], "c0ffee1234567890");
        assert!(routine_json["state_root"].is_null());
    }

    #[test]
    fn test_gc_reachability_marks_live_objects() {
        let repo = test_repo();
        // Overwrite /a (orphans v1's leaf from the tip) then add /b.
        repo.set("main", "/a", &Object::string("v1"), quick_opts("a1"))
            .unwrap();
        repo.set("main", "/a", &Object::string("v2"), quick_opts("a2"))
            .unwrap();
        repo.set("main", "/b", &Object::string("w"), quick_opts("b1"))
            .unwrap();

        // Keep everything ever committed vs. keep only the current tip.
        let all_roots: Vec<ObjectId> = repo
            .log("main", 1000)
            .unwrap()
            .iter()
            .map(|c| c.state_root)
            .collect();
        let keep_all = repo.gc_reachability_from(&all_roots).unwrap();
        let keep_tip = repo.gc_reachability_report().unwrap();

        let total = keep_all["total_objects"].as_i64().unwrap();
        assert!(total > 0);
        assert_eq!(keep_tip["total_objects"].as_i64().unwrap(), total);

        // Invariant: live + reclaimable == total, both keep-sets.
        let inv = |r: &serde_json::Value| {
            r["live_objects"].as_i64().unwrap() + r["reclaimable_objects"].as_i64().unwrap()
        };
        assert_eq!(inv(&keep_all), total);
        assert_eq!(inv(&keep_tip), total);

        // Keeping all history marks at least as many live objects as keeping
        // only the tip — and the tip-only set leaves the superseded v1 churn
        // reclaimable.
        assert!(
            keep_all["live_objects"].as_i64().unwrap()
                >= keep_tip["live_objects"].as_i64().unwrap()
        );
        assert!(
            keep_tip["reclaimable_objects"].as_i64().unwrap() > 0,
            "overwriting /a should orphan the v1 leaf from the tip"
        );
        assert!(keep_tip["roots_walked"].as_u64().unwrap() >= 1);
    }

    #[test]
    fn test_gc_dry_run_safety_gate() {
        let repo = test_repo();
        repo.set("main", "/a", &Object::string("v1"), quick_opts("a1"))
            .unwrap();
        repo.set("main", "/a", &Object::string("v2"), quick_opts("a2"))
            .unwrap();

        // Before extraction the commits are undistilled → dry-run refuses to
        // call a sweep safe, but still reports the estimate and never mutates.
        let before = repo.gc_dry_run().unwrap();
        assert_eq!(before["dry_run"], true);
        assert_eq!(before["mutates"], false);
        assert_eq!(before["safe"], false);
        assert!(
            before["safety"]["undistilled_commits"].as_i64().unwrap() > 0,
            "extractor is behind before extract_history"
        );
        let total = before["objects"]["total"].as_i64().unwrap();
        assert!(total > 0);
        assert_eq!(
            before["objects"]["live"].as_i64().unwrap()
                + before["objects"]["reclaimable"].as_i64().unwrap(),
            total
        );

        // Once the extractor is caught up, every commit's signal is captured →
        // the dry-run reports it as safe.
        repo.extract_history(100).unwrap();
        let after = repo.gc_dry_run().unwrap();
        assert_eq!(after["safe"], true);
        assert_eq!(after["safety"]["undistilled_commits"].as_i64().unwrap(), 0);
        assert!(after["bytes"]["current_total"].as_i64().unwrap() > 0);
        assert!(after["bytes"]["estimated_reclaimable"].as_i64().unwrap() >= 0);
        assert!(after["bytes"]["estimated_post_vacuum"].as_i64().unwrap() >= 0);
    }

    #[test]
    fn test_gc_sweep_reclaims_and_preserves_current_state() {
        // No grace period, so the sweep takes the churn it just wrote.
        let repo = Repository::new(Box::new(
            SqliteStorage::in_memory()
                .unwrap()
                .with_gc_grace(std::time::Duration::ZERO),
        ));
        repo.init().unwrap();
        // Churn: overwrite /a 20 times → old versions become orphaned from the
        // tip, then a final key.
        for i in 0..20 {
            repo.set(
                "main",
                "/a",
                &Object::string(&format!("v{i}")),
                quick_opts("x"),
            )
            .unwrap();
        }
        repo.set("main", "/keep", &Object::string("final"), quick_opts("k"))
            .unwrap();

        // Aggressive policy: keep only the current tip. (Extract first so the
        // safety gate is satisfied.)
        let policy = RetentionPolicy {
            keep_recent: 1,
            checkpoint_every: 0,
            keep_milestones: false,
        };

        // Before distilling, a mutating sweep REFUSES (nothing deleted).
        let refused = repo.gc_sweep(policy, true, false).unwrap();
        assert_eq!(refused["refused"], true);
        assert_eq!(refused["mutated"], false);

        repo.extract_history(1000).unwrap();

        // Dry-run: reports reclaimable churn, mutates nothing.
        let dry = repo.gc_sweep(policy, false, false).unwrap();
        assert_eq!(dry["mutated"], false);
        assert_eq!(dry["safe"], true);
        let total_before = dry["would_reclaim"]["total_objects"].as_i64().unwrap();
        let would = dry["would_reclaim"]["reclaimable_objects"]
            .as_i64()
            .unwrap();
        assert!(would > 0, "20 overwrites should orphan reclaimable objects");
        assert_eq!(
            repo.history_store_shape().unwrap()["objects"]
                .as_i64()
                .unwrap(),
            total_before,
            "dry-run must not delete anything"
        );

        // Real sweep deletes exactly the reclaimable set.
        let swept = repo.gc_sweep(policy, true, true).unwrap();
        assert_eq!(swept["mutated"], true);
        assert_eq!(swept["objects_deleted"].as_i64().unwrap(), would);
        // vacuum ran and never grew the file.
        let vac = &swept["vacuum"];
        assert!(vac.is_object());
        assert!(vac["bytes_after"].as_i64().unwrap() <= vac["bytes_before"].as_i64().unwrap());
        assert!(vac["bytes_reclaimed"].as_i64().unwrap() >= 0);
        assert!(
            swept["objects_after"].as_i64().unwrap() < swept["objects_before"].as_i64().unwrap()
        );

        // The current state is fully materializable after the sweep.
        assert_eq!(repo.get("main", "/a").unwrap(), Object::string("v19"));
        assert_eq!(repo.get("main", "/keep").unwrap(), Object::string("final"));
        // Integrity: nothing reachable from the tip is missing.
        let head = repo.head("main").unwrap();
        let tip = repo.get_commit(&head).unwrap().unwrap();
        assert!(
            self::Repository::first_missing_reachable(&repo, &tip.state_root)
                .unwrap()
                .is_none(),
            "tip closure must be intact"
        );
    }
}