rto_graph/

store.rs

//! SQLite-backed graph store.

use std::collections::BTreeSet;
use std::path::Path;

use rusqlite::{Connection, OptionalExtension, params};

use crate::findings::{self, AnalysisRun, Finding, FindingsApplied, FindingsLayer};
use crate::media::{self, MediaFilter, MediaKind, MediaRecord, MediaWrite, ProducerSummary};
use crate::memory::{
    self, CacheEntry, CacheStats, CacheSweep, CacheWrite, MemoryError, MemoryFilter,
    MemoryForgotten, MemoryListing, MemoryRecord, MemoryWrite, Recall, RecallOptions,
};
use crate::migrations;
use crate::model::{Direction, Edge, EdgeKind, FactSet, Node, NodeKind, Span};
use crate::provenance::Provenance;

/// Errors raised by the store.
#[derive(Debug, thiserror::Error)]
pub enum StoreError {
    /// Underlying `SQLite` failure.
    #[error("sqlite error: {0}")]
    Sqlite(#[from] rusqlite::Error),
    /// A node's `meta` could not be (de)serialized as JSON.
    #[error("json error: {0}")]
    Json(#[from] serde_json::Error),
    /// An edge referenced a node key that does not exist in the store.
    #[error("unknown node key: {0}")]
    UnknownNode(String),
    /// An edge violated the provenance/confidence invariant.
    #[error("invalid edge: {0}")]
    InvalidEdge(String),
    /// A stored value could not be interpreted (database corruption).
    #[error("corrupt store: {0}")]
    Corrupt(String),
}

/// A store written by a build **newer** than this one: it records migrations
/// this binary has never heard of. Produced by [`Store::schema_ahead`].
///
/// # Why this is its own type, and not a `StoreError` variant
///
/// Opening such a store is not an error and reading its *schema* is sound —
/// migrations are additive in effect, so every column an older build selects is
/// still there (each `DROP TABLE` in `crate::migrations` is a table rebuild
/// that re-selects every prior column). What is *not* sound is **rewriting** the
/// graph: this build would re-extract every file with its older extractor and
/// replace the newer build's content with worse content, silently. So the
/// refusal belongs on the write paths, and the carrier of that refusal has no
/// business in the type every `Store` call returns.
///
/// # Columns are not the whole story, since migration 14
///
/// "Additive in effect" was a claim about *schema*, and every migration up to 13
/// kept it about values too. Migration 14 does not: it widened the provenance
/// vocabulary, so an older build reading a row written at `external-authored`
/// gets [`StoreError::Corrupt`] from `Provenance::from_token`, which knows only
/// its own six-minus-three. That is a real limit of the read guarantee and is
/// recorded here rather than left to be discovered.
///
/// It is not left silent, either. Such a store necessarily records migration 14,
/// so this type reports it as written by a newer Roteiro *before* any row is
/// read, and [`StoreError::Corrupt`]'s message names the same cause. The
/// alternative — teaching `from_token` a tolerant fallback — was rejected: every
/// fallback value is one of the six, and each choice silently re-tiers a fact.
///
/// The second reason is semver, and it is the binding one. `StoreError` is a
/// public enum on a published 1.x crate and is not `#[non_exhaustive]`, so
/// adding a variant would stop a downstream exhaustive `match` from compiling —
/// a breaking change, which in this workspace means a major bump of all seven
/// crates. A new type is purely additive. Its fields are private and read
/// through accessors for the same reason: adding one later stays non-breaking.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SchemaAhead {
    store: u32,
    build: u32,
    unknown: Vec<u32>,
}

impl SchemaAhead {
    /// The highest schema version this store records — necessarily one this
    /// build does not know.
    #[must_use]
    pub fn store_version(&self) -> u32 {
        self.store
    }

    /// The highest schema version this build knows how to apply.
    #[must_use]
    pub fn build_version(&self) -> u32 {
        self.build
    }

    /// Every recorded version this build has never heard of, ascending. Never
    /// empty. More than one means the store ran several unknown migrations, not
    /// that anything is inconsistent.
    #[must_use]
    pub fn unknown_versions(&self) -> &[u32] {
        &self.unknown
    }
}

impl std::fmt::Display for SchemaAhead {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let unknown = self
            .unknown
            .iter()
            .map(u32::to_string)
            .collect::<Vec<_>>()
            .join(", ");
        // Both versions, and what to do about it. A bare "schema mismatch" would
        // leave the reader with nothing to act on: the useful facts are *which*
        // side is behind (this binary) and that the fix is upgrading it, not
        // deleting the store.
        write!(
            f,
            "this graph store was written by a newer Roteiro: it is at schema \
             version {store}, and this build knows only up to {build} \
             (unrecognised migration(s): {unknown}). Refusing to rewrite the \
             graph — this build would re-extract every file with its older \
             extractor and replace the newer build's graph with worse content, \
             silently. Upgrade the `roteiro` binary to one that knows schema \
             version {store} or later, then retry. Reading this store is \
             unaffected and needs no upgrade.",
            store = self.store,
            build = self.build,
        )
    }
}

impl std::error::Error for SchemaAhead {}

/// A summary of applying/re-applying import layers (see
/// [`Store::apply_import_layer`] and [`Store::reapply_imports`]).
#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize)]
pub struct ImportApplied {
    /// Number of import layers processed.
    pub layers: usize,
    /// Import nodes upserted (across all layers).
    pub nodes: usize,
    /// Import edges applied — both endpoints resolved. A duplicate of an
    /// already-present edge is a harmless no-op but still counted as applied.
    pub edges_applied: usize,
    /// Import edges **pruned**: an endpoint was absent (a cross-reference to code
    /// that no longer exists), so the edge was dropped from the persisted layer
    /// rather than kept as stale data.
    pub edges_pruned: usize,
    /// Import **nodes removed**: a node this `src_ref` used to contribute and no
    /// longer does, which nothing else in the store still refers to. See
    /// [`Store::apply_import_layer`] for the rule and why a withdrawn concept
    /// has to go.
    pub nodes_removed: usize,
}

/// Qualified node columns for `SELECT`s that alias the `nodes` table as `n`.
const NODE_COLS: &str = "n.key, n.kind, n.name, n.path, n.lang, n.blob_hash, n.span_start, n.span_end, n.provenance, n.meta";

/// `SELECT` prefix that yields an [`Edge`] row (endpoints resolved back to keys).
const EDGE_SELECT: &str = "SELECT ns.key AS src, nd.key AS dst, e.kind, e.provenance, \
     e.confidence, e.src_ref \
     FROM edges e JOIN nodes ns ON ns.id = e.src JOIN nodes nd ON nd.id = e.dst";

/// A Roteiro graph store backed by a single `SQLite` database.
pub struct Store {
    conn: Connection,
}

impl Store {
    /// Open (creating if absent) a store at `path` and apply pending migrations.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] if the database cannot be opened or a
    /// migration fails.
    pub fn open(path: &Path) -> Result<Self, StoreError> {
        let conn = Connection::open(path)?;
        Self::from_conn(conn)
    }

    /// Open an in-memory store (tests, previews).
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] if a migration fails.
    pub fn open_in_memory() -> Result<Self, StoreError> {
        let conn = Connection::open_in_memory()?;
        Self::from_conn(conn)
    }

    fn from_conn(mut conn: Connection) -> Result<Self, StoreError> {
        conn.execute_batch("PRAGMA foreign_keys = ON;")?;
        // Wait for a concurrent writer instead of failing with `database is
        // locked`. Matters for workspace `serve` (ADR-0008), where a long-lived
        // server reads a project's graph while that repo's own `roteiro sync`
        // commits an update to the same file — and for any two Roteiro processes
        // over one repository, which is ordinary rather than exotic: an editor's
        // MCP client and a terminal, or several clients each spawning their own
        // stdio server.
        conn.busy_timeout(std::time::Duration::from_secs(5))?;

        // **The timeout alone does not deliver that**, which is what this pair of
        // settings is for. Both are needed and neither substitutes for the other.
        //
        // `Immediate` takes the write lock when the transaction *opens*, rather
        // than upgrading to it on the first write. rusqlite's default is the lazy
        // one — `TransactionBehavior::Deferred` — and under it two writers that  roteiro:ignore
        // each read before they write, which every upsert here does, both hold a
        // shared lock and then both ask to upgrade.
        // SQLite refuses that immediately with `SQLITE_BUSY` and **does not invoke
        // the busy handler**, because waiting could only deadlock. So the five
        // second timeout above was never reached in the case it looks like it
        // covers: one of the two writers died at once. Taking the lock up front
        // turns the deadlock into a queue, which is what the timeout can wait on.
        conn.set_transaction_behavior(rusqlite::TransactionBehavior::Immediate);

        // WAL is the other half: under the default rollback journal a writer
        // excludes readers for the whole transaction, so a [`crate::sync`] blocks every
        // query in a running `serve`. In WAL, readers never block writers and
        // writers never block readers, so only writer-against-writer waits.
        //
        // Tolerated rather than required, and **silently** — which is the one place
        // silence is right here. WAL needs shared memory and is refused on most
        // network filesystems; `PRAGMA journal_mode` reports the mode actually in
        // force rather than failing. A store that cannot take WAL keeps exactly
        // the behaviour it had before this change, so there is no new failure to
        // report — only an improvement that did not apply. `:memory:` answers
        // `memory` for the same reason and is equally fine.
        //
        // The result is discarded rather than checked because nothing here can act
        // on it. `PRAGMA journal_mode` answers for anyone who needs to know which
        // of the two a given store got.
        let _: String = conn.query_row("PRAGMA journal_mode = WAL", [], |r| r.get(0))?;

        migrations::apply(&mut conn)?;
        Ok(Self { conn })
    }

    /// The schema version this store has been migrated to: the highest `v` for
    /// which every migration `1..=v` is recorded as applied.
    ///
    /// **Not the maximum recorded version.** Migrations are selected by set
    /// membership rather than `> MAX(version)` (see `crate::migrations`), so a
    /// store *can* hold a gap — one written by a build that knew a higher
    /// migration but not a lower one. For a store recorded as `1..11, 13`, the
    /// maximum is 13 while none of migration 12's schema is present; reporting 13
    /// would be a higher number than the truth. The contiguous prefix reports 11,
    /// which is the version the store can actually be relied on to provide.
    ///
    /// The gap is transient in practice — the next [`Store::open`] repairs it —
    /// but this must not answer wrongly in the window where it exists, which is
    /// exactly the window in which someone is debugging.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn schema_version(&self) -> Result<u32, StoreError> {
        Ok(migrations::store_version(&self.conn)?)
    }

    /// The highest schema version **this build** knows how to apply — the
    /// binary's half of the comparison [`Store::schema_ahead`] makes.
    #[must_use]
    pub fn build_schema_version() -> u32 {
        migrations::latest_version()
    }

    /// `Some` when this store was written by a **newer** build: it records
    /// migrations this binary has never heard of. `None` — the ordinary case —
    /// when the store is level with this build or behind it.
    ///
    /// Call this before **rewriting** a graph (sync, reconcile, rebuild). Do not
    /// call it to gate reads: an older binary's reads are sound, which is the
    /// whole reason this is not checked in [`Store::open`] (issue #342).
    ///
    /// # Which version this compares, and why it is not [`Store::schema_version`]
    ///
    /// It compares the **set** of recorded versions against the migrations this
    /// build carries, so it answers *has anything newer than me written here?*
    /// [`Store::schema_version`] answers a different question — the highest
    /// **gap-free** version, a floor describing what a reader may rely on — and
    /// using it here would let the bug through: a store recorded `1..=13, 15`,
    /// read by a build that knows 13, has a gap-free version of 13, so
    /// `schema_version() > build` is false while migration 15's schema sits in
    /// the file and a newer build plainly assembled the graph.
    ///
    /// Conflating the two situations would also misreport both, so it does not.
    /// A **gap** — a store missing a lower migration this build knows — is not
    /// a store from the future, and [`Store::open`] has already repaired it by
    /// applying the missing migration before this can be called. What is left is
    /// only ever a version no build of this vintage could have written.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn schema_ahead(&self) -> Result<Option<SchemaAhead>, StoreError> {
        let unknown = migrations::versions_ahead_of_build(&self.conn)?;
        let Some(&store) = unknown.last() else {
            return Ok(None);
        };
        Ok(Some(SchemaAhead {
            store,
            build: Self::build_schema_version(),
            unknown,
        }))
    }

    /// Number of nodes currently in the store.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn node_count(&self) -> Result<u64, StoreError> {
        let n: i64 = self
            .conn
            .query_row("SELECT COUNT(*) FROM nodes", [], |r| r.get(0))?;
        Ok(u64::try_from(n).unwrap_or(0))
    }

    /// Number of edges currently in the store.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn edge_count(&self) -> Result<u64, StoreError> {
        let n: i64 = self
            .conn
            .query_row("SELECT COUNT(*) FROM edges", [], |r| r.get(0))?;
        Ok(u64::try_from(n).unwrap_or(0))
    }

    /// Insert or update a node, keyed by its natural [`Node::key`].
    ///
    /// # Errors
    /// Returns [`StoreError::Json`] if `meta` cannot be serialized, or
    /// [`StoreError::Sqlite`] on write failure.
    pub fn upsert_node(&self, node: &Node) -> Result<(), StoreError> {
        upsert_node(&self.conn, node)
    }

    /// Insert an edge. Both endpoints must already resolve to nodes.
    ///
    /// # Errors
    /// Returns [`StoreError::InvalidEdge`] if the provenance/confidence
    /// invariant is violated, [`StoreError::UnknownNode`] if an endpoint key is
    /// absent, or [`StoreError::Sqlite`] on write failure.
    pub fn insert_edge(&self, edge: &Edge) -> Result<(), StoreError> {
        insert_edge(&self.conn, edge)
    }

    /// Apply a fact set atomically: all nodes are upserted, then all edges are
    /// inserted, in a single transaction. On any error nothing is committed.
    ///
    /// # Errors
    /// Returns the first error encountered (see [`Store::upsert_node`] and
    /// [`Store::insert_edge`]); the transaction is rolled back.
    pub fn apply_factset(&mut self, facts: &FactSet) -> Result<(), StoreError> {
        let tx = self.conn.transaction()?;
        for node in &facts.nodes {
            upsert_node(&tx, node)?;
        }
        for edge in &facts.edges {
            insert_edge(&tx, edge)?;
        }
        tx.commit()?;
        Ok(())
    }

    /// The `HEAD` tree id recorded at the last successful [`Store::rebuild`], if
    /// any. Used by the sync engine to detect an unchanged tree.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn sync_state(&self) -> Result<Option<String>, StoreError> {
        Ok(self
            .conn
            .query_row("SELECT tree FROM sync_state WHERE id = 0", [], |r| r.get(0))
            .optional()?)
    }

    /// The extractor environment recorded with the last committed [`crate::sync`],
    /// `None` if unset (a legacy row, or the last sync was a worktree/index
    /// preview). The incremental committed [`crate::sync`] compares this to the current
    /// env and falls back to a full re-extraction when they differ.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn sync_env(&self) -> Result<Option<String>, StoreError> {
        Ok(self
            .conn
            .query_row("SELECT env FROM sync_state WHERE id = 0", [], |r| r.get(0))
            .optional()?
            .flatten())
    }

    /// Record the extractor environment for the current synced tree. Called by a
    /// committed [`crate::sync`] right after it writes the tree, so a later sync can decide
    /// whether the incremental fast path is sound. A no-op if no tree is recorded.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on write failure.
    pub fn set_sync_env(&self, env: &str) -> Result<(), StoreError> {
        self.conn
            .execute("UPDATE sync_state SET env = ?1 WHERE id = 0", [env])?;
        Ok(())
    }

    /// Which working tree this graph was assembled from, or `None` when the store
    /// has never been synced or predates the column (issue #330).
    ///
    /// `graph.db` is an assembled view of **one** tree, so a store that came to
    /// describe a different tree — restored from a backup, copied along with a
    /// `.git` directory, or reached after a layout change — would otherwise
    /// answer confidently about the wrong one: [`crate::sync`] reporting "up to date"
    /// against a state id belonging to someone else's tree, `check` validating a
    /// tree nobody is looking at. `None` reads as "unknown" and is *adopted*
    /// rather than treated as a mismatch, so an existing store is never rebuilt
    /// merely for predating the stamp.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn synced_worktree(&self) -> Result<Option<String>, StoreError> {
        Ok(self
            .conn
            .query_row("SELECT worktree FROM sync_state WHERE id = 0", [], |r| {
                r.get(0)
            })
            .optional()?
            .flatten())
    }

    /// Stamp this graph as assembled from `worktree`. Called by every sync entry
    /// point right after it writes the tree state, so the two always agree about
    /// whose tree the state describes. A no-op if no tree is recorded.
    ///
    /// Unlike [`Store::set_sync_env`]'s value, this survives a tree write: it
    /// identifies the *store*, not the tree, and a new commit does not move the
    /// graph to a different working tree.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on write failure.
    pub fn set_synced_worktree(&self, worktree: &str) -> Result<(), StoreError> {
        self.conn.execute(
            "UPDATE sync_state SET worktree = ?1 WHERE id = 0",
            [worktree],
        )?;
        Ok(())
    }

    /// Atomically replace the entire graph with `facts`, recording `tree` as the
    /// synced state (or clearing it when `tree` is `None`). All existing nodes
    /// and edges are deleted first, so the store reflects exactly the given fact
    /// set.
    ///
    /// Passing `None` records *no* synced tree — distinct from an empty string —
    /// so [`Store::sync_state`] returns `None` and a later [`crate::sync`] will not
    /// spuriously short-circuit.
    ///
    /// # Errors
    /// Returns the first error encountered (see [`Store::apply_factset`]); on any
    /// error nothing is committed.
    pub fn rebuild(&mut self, facts: &FactSet, tree: Option<&str>) -> Result<(), StoreError> {
        let tx = self.conn.transaction()?;
        tx.execute("DELETE FROM edges", [])?;
        tx.execute("DELETE FROM nodes", [])?;
        for node in &facts.nodes {
            upsert_node(&tx, node)?;
        }
        for edge in &facts.edges {
            insert_edge(&tx, edge)?;
        }
        write_sync_state(&tx, tree)?;
        tx.commit()?;
        Ok(())
    }

    /// Bring the store to exactly `facts` (as [`Store::rebuild`] does) but writing
    /// only what **differs** instead of wiping and reinserting the whole graph —
    /// the git-style "write only the delta". Unchanged node rows (which carry the
    /// heavy JSON `meta`) and unchanged edge rows are left untouched; only removed
    /// rows are deleted and new/changed rows written. The final state — nodes,
    /// edges, and `sync_state` — is identical to `rebuild(facts, tree)`.
    ///
    /// Leaving unchanged edges in place means their row ids do not match a cold
    /// rebuild's — which is safe *because* every edge query is content-ordered
    /// (`(src, dst, kind, provenance)`, see [`Store::all_edges`]), never by row
    /// id. So an incrementally reconciled store and a fresh rebuild return every
    /// query identically; the delta is invisible above the storage layer.
    ///
    /// # Errors
    /// Returns [`StoreError`] on a query failure; the transaction is rolled back.
    pub fn reconcile(&mut self, facts: &FactSet, tree: Option<&str>) -> Result<(), StoreError> {
        let current_nodes = self.all_nodes()?;
        let current_edges = self.all_edges()?;
        let cur_by_key: std::collections::HashMap<&str, &Node> =
            current_nodes.iter().map(|n| (n.key.as_str(), n)).collect();
        let new_keys: std::collections::HashSet<&str> =
            facts.nodes.iter().map(|n| n.key.as_str()).collect();

        // Edge identity is the full tuple, so a changed `confidence`/`src_ref`
        // counts as remove-old + add-new — keeping the result identical to a
        // wholesale rebuild, not the store's insert-time `DO NOTHING` semantics.
        let new_edge_ids: std::collections::HashSet<EdgeId> =
            facts.edges.iter().map(edge_identity).collect();
        let cur_edge_ids: std::collections::HashSet<EdgeId> =
            current_edges.iter().map(edge_identity).collect();

        let tx = self.conn.transaction()?;
        // 1. Delete removed edges first, so any node they reference can then be
        //    dropped (edges are FK-constrained on node ids, with no cascade). A
        //    removed node's edges are all removals, so they are gone before step 2.
        for edge in &current_edges {
            if !new_edge_ids.contains(&edge_identity(edge)) {
                delete_edge(&tx, edge)?;
            }
        }
        // 2. Drop nodes that no longer exist.
        for old in &current_nodes {
            if !new_keys.contains(old.key.as_str()) {
                tx.execute("DELETE FROM nodes WHERE key = ?1", [&old.key])?;
            }
        }
        // 3. Upsert only the nodes that are new or whose content changed (an upsert
        //    keeps the row id, so unchanged edges stay valid).
        for node in &facts.nodes {
            if cur_by_key
                .get(node.key.as_str())
                .is_none_or(|cur| *cur != node)
            {
                upsert_node(&tx, node)?;
            }
        }
        // 4. Insert only the added edges (their endpoints now all exist).
        for edge in &facts.edges {
            if !cur_edge_ids.contains(&edge_identity(edge)) {
                insert_edge(&tx, edge)?;
            }
        }
        write_sync_state(&tx, tree)?;
        tx.commit()?;
        Ok(())
    }

    /// Fetch a node by its natural key.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
    /// [`StoreError::Corrupt`] if a stored value cannot be decoded.
    pub fn get_node(&self, key: &str) -> Result<Option<Node>, StoreError> {
        let sql = format!("SELECT {NODE_COLS} FROM nodes n WHERE n.key = ?1");
        let mut stmt = self.conn.prepare(&sql)?;
        let mut rows = stmt.query([key])?;
        match rows.next()? {
            Some(row) => Ok(Some(row_to_node(row)?)),
            None => Ok(None),
        }
    }

    /// Every node key in the store, ordered. Useful for whole-graph exports.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn all_keys(&self) -> Result<Vec<String>, StoreError> {
        let mut stmt = self.conn.prepare("SELECT key FROM nodes ORDER BY key")?;
        let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
        let mut out = Vec::new();
        for row in rows {
            out.push(row?);
        }
        Ok(out)
    }

    /// Dump the entire graph as a single [`FactSet`], with nodes and edges in a
    /// deterministic order — suitable for a portable, content-stable artifact.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
    /// [`StoreError::Corrupt`] on decode failure.
    pub fn export_factset(&self) -> Result<FactSet, StoreError> {
        let node_sql = format!("SELECT {NODE_COLS} FROM nodes n ORDER BY n.key");
        let mut node_stmt = self.conn.prepare(&node_sql)?;
        let mut node_rows = node_stmt.query([])?;
        let nodes = collect_nodes(&mut node_rows)?;

        // Order edges by their resolved endpoint keys (not row id) so the dump is
        // stable regardless of insertion order.
        let edge_sql = format!("{EDGE_SELECT} ORDER BY ns.key, nd.key, e.kind, e.provenance");
        let mut edge_stmt = self.conn.prepare(&edge_sql)?;
        let mut edge_rows = edge_stmt.query([])?;
        let edges = collect_edges(&mut edge_rows)?;

        Ok(FactSet { nodes, edges })
    }

    /// All nodes of a given kind.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
    /// [`StoreError::Corrupt`] on decode failure.
    pub fn nodes_by_kind(&self, kind: &NodeKind) -> Result<Vec<Node>, StoreError> {
        let sql = format!("SELECT {NODE_COLS} FROM nodes n WHERE n.kind = ?1 ORDER BY n.key");
        let mut stmt = self.conn.prepare(&sql)?;
        let mut rows = stmt.query([kind.as_str()])?;
        collect_nodes(&mut rows)
    }

    /// Nodes of a given kind whose `name` equals `name_lower` **case-insensitively**,
    /// ordered by key. Narrows a lookup at the SQL layer — using the `kind` index and
    /// filtering `name` in-query — so only matching rows are decoded, never every
    /// node of that kind. Used by the cross-repo follow bridge to fetch just the
    /// candidate struct(s) for a config section rather than scanning all structs.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
    /// [`StoreError::Corrupt`] on decode failure.
    pub fn nodes_by_kind_named(
        &self,
        kind: &NodeKind,
        name_lower: &str,
    ) -> Result<Vec<Node>, StoreError> {
        let sql = format!(
            "SELECT {NODE_COLS} FROM nodes n \
             WHERE n.kind = ?1 AND lower(n.name) = ?2 ORDER BY n.key"
        );
        let mut stmt = self.conn.prepare(&sql)?;
        let mut rows = stmt.query([kind.as_str(), name_lower])?;
        collect_nodes(&mut rows)
    }

    /// Every `config_key` node's flattened setting (ADR-0009), read back out of
    /// the graph as [`crate::ConfigKey`]s — the graph-native source the cross-repo
    /// link matcher (`roteiro links --infer`) consumes, so it never re-parses
    /// config files. Ordered by node key (deterministic). A node missing the
    /// `key`/`path` a well-formed `config_key` carries is skipped defensively.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
    /// [`StoreError::Corrupt`] on decode failure.
    pub fn config_keys(&self) -> Result<Vec<crate::ConfigKey>, StoreError> {
        let nodes = self.nodes_by_kind(&NodeKind::Other(crate::config_keys::KIND.to_owned()))?;
        let mut out = Vec::with_capacity(nodes.len());
        for n in &nodes {
            let key = n.meta.get("key").and_then(serde_json::Value::as_str);
            // A `config_key` node carries a `value` in `meta` when it has a real
            // setting (file-derived keys always do, even an empty string). A
            // struct-derived key (`meta.source = "struct"`) omits it — its value is
            // *unknown*, not empty — so record that absence explicitly rather than
            // defaulting it to `""`, which would false-match in value agreement.
            let value = n.meta.get("value").and_then(serde_json::Value::as_str);
            if let (Some(key), Some(file)) = (key, n.path.as_deref()) {
                out.push(crate::ConfigKey {
                    file: file.to_owned(),
                    key: key.to_owned(),
                    value: value.unwrap_or_default().to_owned(),
                    value_known: value.is_some(),
                });
            }
        }
        Ok(out)
    }

    /// Every node whose source `path` is `path`, ordered by key — the file node
    /// plus the symbols and markers defined in it. Used to scope a change to the
    /// graph (e.g. `roteiro review`).
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
    /// [`StoreError::Corrupt`] on decode failure.
    pub fn nodes_by_path(&self, path: &str) -> Result<Vec<Node>, StoreError> {
        let sql = format!("SELECT {NODE_COLS} FROM nodes n WHERE n.path = ?1 ORDER BY n.key");
        let mut stmt = self.conn.prepare(&sql)?;
        let mut rows = stmt.query([path])?;
        collect_nodes(&mut rows)
    }

    /// Every node produced by a given layer, ordered by key. The incremental
    /// [`crate::sync`] loads the `Derived` layer to reconstruct the extraction graph
    /// without re-reading every blob.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
    /// [`StoreError::Corrupt`] on decode failure.
    pub fn nodes_by_provenance(&self, provenance: Provenance) -> Result<Vec<Node>, StoreError> {
        let sql = format!("SELECT {NODE_COLS} FROM nodes n WHERE n.provenance = ?1 ORDER BY n.key");
        let mut stmt = self.conn.prepare(&sql)?;
        let mut rows = stmt.query([provenance.as_str()])?;
        collect_nodes(&mut rows)
    }

    /// Every node in the store, ordered by key. Unlike [`Store::export_factset`]
    /// this decodes no edges, so it is cheap for node-only scans (e.g. search).
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
    /// [`StoreError::Corrupt`] on decode failure.
    pub fn all_nodes(&self) -> Result<Vec<Node>, StoreError> {
        let sql = format!("SELECT {NODE_COLS} FROM nodes n ORDER BY n.key");
        let mut stmt = self.conn.prepare(&sql)?;
        let mut rows = stmt.query([])?;
        collect_nodes(&mut rows)
    }

    /// Every edge in the store, with endpoints resolved to their node keys. Used
    /// by [`Store::reconcile`] to diff the edge set.
    ///
    /// Ordered by the edge's **content** — `(src key, dst key, kind, provenance)`,
    /// the table's unique tuple — not by row id. This makes the order a function
    /// of the *graph*, not of insertion history, so an incrementally
    /// [`reconcile`](Store::reconcile)d store and a cold [`rebuild`](Store::rebuild)
    /// return edges identically. (The same reason node scans order by `key`.)
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] or [`StoreError::Corrupt`] on failure.
    pub fn all_edges(&self) -> Result<Vec<Edge>, StoreError> {
        let sql = format!("{EDGE_SELECT} ORDER BY ns.key, nd.key, e.kind, e.provenance");
        let mut stmt = self.conn.prepare(&sql)?;
        let mut rows = stmt.query([])?;
        collect_edges(&mut rows)
    }

    /// Edges whose source is the node with the given key, in content order
    /// (`(dst key, kind, provenance)` — `src` is fixed). Content-ordered rather
    /// than by row id so the result is history-independent; see [`Store::all_edges`].
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] or [`StoreError::Corrupt`] on failure.
    pub fn edges_from(&self, key: &str) -> Result<Vec<Edge>, StoreError> {
        let sql = format!("{EDGE_SELECT} WHERE ns.key = ?1 ORDER BY nd.key, e.kind, e.provenance");
        let mut stmt = self.conn.prepare(&sql)?;
        let mut rows = stmt.query([key])?;
        collect_edges(&mut rows)
    }

    /// Edges whose destination is the node with the given key, in content order
    /// (`(src key, kind, provenance)` — `dst` is fixed). See [`Store::all_edges`].
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] or [`StoreError::Corrupt`] on failure.
    pub fn edges_to(&self, key: &str) -> Result<Vec<Edge>, StoreError> {
        let sql = format!("{EDGE_SELECT} WHERE nd.key = ?1 ORDER BY ns.key, e.kind, e.provenance");
        let mut stmt = self.conn.prepare(&sql)?;
        let mut rows = stmt.query([key])?;
        collect_edges(&mut rows)
    }

    /// All edges with the given provenance, in content order
    /// (`(src key, dst key, kind)` — `provenance` is fixed). See [`Store::all_edges`].
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] or [`StoreError::Corrupt`] on failure.
    pub fn edges_by_provenance(&self, provenance: Provenance) -> Result<Vec<Edge>, StoreError> {
        let sql = format!("{EDGE_SELECT} WHERE e.provenance = ?1 ORDER BY ns.key, nd.key, e.kind");
        let mut stmt = self.conn.prepare(&sql)?;
        let mut rows = stmt.query([provenance.as_str()])?;
        collect_edges(&mut rows)
    }

    /// Delete all edges with the given provenance, returning how many were
    /// removed. Used to re-derive a whole provenance class authoritatively (e.g.
    /// `inferred` edges when re-running inference with different parameters).
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on write failure.
    pub fn delete_edges_by_provenance(&self, provenance: Provenance) -> Result<u64, StoreError> {
        let n = self.conn.execute(
            "DELETE FROM edges WHERE provenance = ?1",
            [provenance.as_str()],
        )?;
        Ok(u64::try_from(n).unwrap_or(0))
    }

    /// Delete all edges carrying the given `src_ref`, returning how many were
    /// removed. Lets one producer of `inferred` edges (e.g. the embedding layer,
    /// or a Graphify import) re-derive its own edges authoritatively without
    /// touching edges another producer contributed.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on write failure.
    pub fn delete_edges_by_src_ref(&self, src_ref: &str) -> Result<u64, StoreError> {
        let n = self
            .conn
            .execute("DELETE FROM edges WHERE src_ref = ?1", [src_ref])?;
        Ok(u64::try_from(n).unwrap_or(0))
    }

    /// Apply an import layer to the live graph **and** persist it durably under
    /// `src_ref`, validating as it goes: this ref's prior edges are cleared
    /// (an authoritative re-import), the layer's nodes are upserted, and each
    /// edge is applied only if both endpoints resolve. Dangling edges — cross-
    /// references to code that is not present — are dropped, and only the
    /// validated (trimmed) layer is persisted, so stale data is never stored.
    ///
    /// This is the "validate on import" half; [`Store::reapply_imports`] is the
    /// "validate on sync" half, re-checking layers against the rebuilt graph.
    ///
    /// # Removal propagates, because a re-run that only adds is not idempotent
    ///
    /// Re-importing cannot **duplicate**: nodes are `INSERT … ON CONFLICT(key) DO
    /// UPDATE`, and this ref's edges are deleted before being re-applied. What is
    /// not free is the other direction. A node the source has since *withdrawn*
    /// used to survive as an orphan carrying the layer's provenance, because only
    /// edges were pruned by `src_ref` — so a re-run **looked** idempotent while
    /// quietly accumulating deleted facts, which is harder to notice than a
    /// duplicate precisely because nothing appears twice.
    ///
    /// So a node this ref previously contributed and no longer does is removed,
    /// under a rule deliberately narrower than "this ref owned it":
    ///
    /// 1. it must not appear in **any other persisted import layer** — two layers
    ///    can name one node on purpose, and `import:links` and
    ///    `import:links/authored` both contribute the same `extref:` placeholder;
    /// 2. **nothing may still be attached to it** — a surviving edge means some
    ///    other producer still relates it to the graph, and a node's removal must
    ///    not silently take a relationship it did not own with it.
    ///
    /// Both are checked against the live store rather than assumed from the key,
    /// so this stays correct for a layer whose namespace it knows nothing about.
    /// The node's cached context row goes with it — a bundle for a node that no
    /// longer exists can only be served as a stale answer. Episodic
    /// [`crate::MemoryKind`] records anchored to it do **not**: ADR-0013 makes
    /// memory the one tier with no generating function, so dropping it is data
    /// loss, and its anchor already reports the node as gone.
    ///
    /// # Errors
    /// Returns [`StoreError::Json`] if `facts` cannot be (de)serialized,
    /// [`StoreError::InvalidEdge`] on a malformed edge, or [`StoreError::Sqlite`]
    /// on write failure.
    pub fn apply_import_layer(
        &mut self,
        src_ref: &str,
        facts: &FactSet,
    ) -> Result<ImportApplied, StoreError> {
        // Read before the row is overwritten: what this ref contributed last time
        // is the only record of what it may now be withdrawing.
        let prior = self.import_layer_node_keys(src_ref)?;
        let tx = self.conn.transaction()?;
        // Authoritative re-import: drop this ref's prior edges from the live graph.
        tx.execute("DELETE FROM edges WHERE src_ref = ?1", [src_ref])?;
        for node in &facts.nodes {
            upsert_node(&tx, node)?;
        }
        let (kept, applied) = apply_edges_pruning(&tx, &facts.edges)?;
        let trimmed = FactSet {
            nodes: facts.nodes.clone(),
            edges: kept,
        };
        put_import_row(&tx, src_ref, &trimmed)?;
        // After the edges are gone and the new nodes are in, so the "still
        // attached" test sees the graph as it will be rather than as it was.
        let nodes_removed = remove_withdrawn_nodes(&tx, src_ref, &prior, &facts.nodes)?;
        tx.commit()?;
        Ok(ImportApplied {
            layers: 1,
            nodes: facts.nodes.len(),
            nodes_removed,
            ..applied
        })
    }

    /// The node keys the persisted layer for `src_ref` currently contributes, or
    /// an empty set when there is no such layer.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure or [`StoreError::Json`] if
    /// the stored layer cannot be decoded.
    fn import_layer_node_keys(&self, src_ref: &str) -> Result<BTreeSet<String>, StoreError> {
        let json: Option<String> = self
            .conn
            .query_row(
                "SELECT facts FROM imports WHERE src_ref = ?1",
                [src_ref],
                |r| r.get(0),
            )
            .optional()?;
        let Some(json) = json else {
            return Ok(BTreeSet::new());
        };
        let facts: FactSet = serde_json::from_str(&json)?;
        Ok(facts.nodes.into_iter().map(|n| n.key).collect())
    }

    /// Remove the persisted import layer for `src_ref`, returning whether one
    /// existed. Does not remove edges already in the live graph (use
    /// [`Store::delete_edges_by_src_ref`] for that).
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on write failure.
    pub fn delete_import(&self, src_ref: &str) -> Result<bool, StoreError> {
        let n = self
            .conn
            .execute("DELETE FROM imports WHERE src_ref = ?1", [src_ref])?;
        Ok(n > 0)
    }

    /// The `src_ref`s of all persisted import layers, ordered.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn import_refs(&self) -> Result<Vec<String>, StoreError> {
        let mut stmt = self
            .conn
            .prepare("SELECT src_ref FROM imports ORDER BY src_ref")?;
        let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
        let mut out = Vec::new();
        for row in rows {
            out.push(row?);
        }
        Ok(out)
    }

    /// Re-apply every persisted import layer on top of the current graph and
    /// **re-validate** it: all import nodes are upserted first (so cross-layer
    /// and self references resolve), then each edge is applied; an edge whose
    /// endpoint is now absent — a cross-reference to code a sync removed — is
    /// pruned from the persisted layer, not merely skipped. So the durable store
    /// keeps only still-correct data. Idempotent; safe to run after each rebuild.
    ///
    /// # Errors
    /// Returns [`StoreError::Json`] if a stored layer cannot be (de)serialized,
    /// or [`StoreError::Sqlite`] on write failure.
    pub fn reapply_imports(&mut self) -> Result<ImportApplied, StoreError> {
        let layers = self.load_import_layers()?;
        let tx = self.conn.transaction()?;
        // Pass 1: upsert every layer's nodes so intra-import edges resolve
        // regardless of which layer defines the endpoint.
        for (_, facts) in &layers {
            for node in &facts.nodes {
                upsert_node(&tx, node)?;
            }
        }
        // Pass 2: apply edges, pruning (and rewriting) any that dangle.
        let mut applied = ImportApplied {
            layers: layers.len(),
            ..ImportApplied::default()
        };
        for (src_ref, facts) in &layers {
            applied.nodes += facts.nodes.len();
            let (kept, counts) = apply_edges_pruning(&tx, &facts.edges)?;
            applied.edges_applied += counts.edges_applied;
            applied.edges_pruned += counts.edges_pruned;
            if kept.len() != facts.edges.len() {
                let trimmed = FactSet {
                    nodes: facts.nodes.clone(),
                    edges: kept,
                };
                put_import_row(&tx, src_ref, &trimmed)?;
            }
        }
        tx.commit()?;
        Ok(applied)
    }

    /// Load and decode every persisted import layer as `(src_ref, FactSet)`, in
    /// `src_ref` order.
    fn load_import_layers(&self) -> Result<Vec<(String, FactSet)>, StoreError> {
        let mut stmt = self
            .conn
            .prepare("SELECT src_ref, facts FROM imports ORDER BY src_ref")?;
        let rows = stmt.query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))?;
        let mut out = Vec::new();
        for row in rows {
            let (src_ref, json) = row?;
            let mut facts: FactSet = serde_json::from_str(&json)?;
            // An import-layer node is *never* derived (derivation is [`crate::sync`]'s job),
            // so a `Derived` tag here is always wrong. It arises two ways, both
            // repaired the same: a legacy layer persisted before nodes carried
            // provenance (the field is absent → serde defaults `Derived`), or —
            // anomalously — a layer that stored an explicit `"provenance":"derived"`
            // (a producer/data bug). We deliberately repair *both* rather than only
            // the absent case: leaving an explicit-`derived` import node in place
            // would let a layer-scoped [`crate::sync`] treat it as derived and delete it —
            // the exact corruption this guards against — so repair is the safe
            // recovery, not silent masking. Idempotent, runs on every reapply
            // (old stores self-heal), and a no-op for correctly-tagged fresh imports.
            for node in &mut facts.nodes {
                if node.provenance == Provenance::Derived {
                    node.provenance = import_node_provenance(&node.key);
                }
            }
            out.push((src_ref, facts));
        }
        Ok(out)
    }

    /// Neighbouring nodes reachable from `key` in the given direction. Returns
    /// an empty vector if the node does not exist.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
    /// [`StoreError::Corrupt`] on failure.
    pub fn neighbors(&self, key: &str, dir: Direction) -> Result<Vec<Node>, StoreError> {
        let out = format!(
            "SELECT {NODE_COLS} FROM nodes n JOIN edges e ON n.id = e.dst \
             JOIN nodes s ON s.id = e.src WHERE s.key = ?1"
        );
        let inc = format!(
            "SELECT {NODE_COLS} FROM nodes n JOIN edges e ON n.id = e.src \
             JOIN nodes d ON d.id = e.dst WHERE d.key = ?1"
        );
        // Order by output column 1 (the node key) so results are deterministic
        // across SQLite versions/plans. Positional ordering avoids both the
        // ambiguity of a bare `key` (present in every joined table) and the fact
        // that a table-qualified name cannot be used after the `Both` UNION.
        let sql = match dir {
            Direction::Outgoing => format!("{out} ORDER BY 1"),
            Direction::Incoming => format!("{inc} ORDER BY 1"),
            Direction::Both => format!("{out} UNION {inc} ORDER BY 1"),
        };
        let mut stmt = self.conn.prepare(&sql)?;
        let mut rows = stmt.query([key])?;
        collect_nodes(&mut rows)
    }

    /// Fetch the cached context bundle for `key` as `(fingerprint, json)`, if
    /// present. The caller compares the fingerprint to the node's current one to
    /// decide whether the entry is fresh (see [`crate::context`]).
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn context_cache_get(&self, key: &str) -> Result<Option<(String, String)>, StoreError> {
        let row = self
            .conn
            .query_row(
                "SELECT fingerprint, json FROM node_context WHERE key = ?1",
                [key],
                |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)),
            )
            .optional()?;
        Ok(row)
    }

    /// Fetch just the cached fingerprint for `key`, without reading the (larger)
    /// JSON payload — for a cheap freshness check.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn context_cache_fingerprint(&self, key: &str) -> Result<Option<String>, StoreError> {
        let fp = self
            .conn
            .query_row(
                "SELECT fingerprint FROM node_context WHERE key = ?1",
                [key],
                |r| r.get::<_, String>(0),
            )
            .optional()?;
        Ok(fp)
    }

    /// Store (or replace) the cached context bundle for `key`.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on write failure.
    pub fn context_cache_put(
        &self,
        key: &str,
        fingerprint: &str,
        json: &str,
    ) -> Result<(), StoreError> {
        self.conn.execute(
            "INSERT INTO node_context (key, fingerprint, json) VALUES (?1, ?2, ?3)
             ON CONFLICT(key) DO UPDATE SET
                 fingerprint = excluded.fingerprint, json = excluded.json",
            [key, fingerprint, json],
        )?;
        Ok(())
    }

    /// Delete the cached context entry for `key`, returning whether one existed.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on write failure.
    pub fn context_cache_delete(&self, key: &str) -> Result<bool, StoreError> {
        let n = self
            .conn
            .execute("DELETE FROM node_context WHERE key = ?1", [key])?;
        Ok(n > 0)
    }

    /// Every key with a cached context entry, ordered. Used to prune entries for
    /// nodes that no longer exist.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn context_cache_keys(&self) -> Result<Vec<String>, StoreError> {
        let mut stmt = self
            .conn
            .prepare("SELECT key FROM node_context ORDER BY key")?;
        let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
        let mut out = Vec::new();
        for row in rows {
            out.push(row?);
        }
        Ok(out)
    }

    // --- Analyzer findings (ADR-0012). A separate artifact store: these methods
    // touch `analysis_runs`/`findings` only, never `nodes`/`edges`, so
    // `export_factset` — and the published `GraphArtifact` — stays a pure
    // function of the tree no matter what an analyzer reports. ---

    /// The recorded answer to the OKF trust question for `peer`, if there is one
    /// (#706 phase 2).
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn okf_consent(&self, peer: &str) -> Result<Option<crate::OkfConsent>, StoreError> {
        let row = self
            .conn
            .query_row(
                "SELECT peer, decision, root, screen_classes, decided_at
                 FROM okf_consent WHERE peer = ?1",
                [peer],
                |r| {
                    Ok((
                        r.get::<_, String>(0)?,
                        r.get::<_, String>(1)?,
                        r.get::<_, String>(2)?,
                        r.get::<_, String>(3)?,
                        r.get::<_, String>(4)?,
                    ))
                },
            )
            .optional()?;
        let Some((peer, decision, root, screen_classes, decided_at)) = row else {
            return Ok(None);
        };
        // A token the `CHECK` constraint permits but this build does not know is
        // a store written by a newer Roteiro. Reading it as "no answer" is the
        // safe direction: it asks again rather than acting on a decision whose
        // meaning this build cannot state.
        let Some(decision) = crate::OkfDecision::from_token(&decision) else {
            return Ok(None);
        };
        Ok(Some(crate::OkfConsent {
            peer,
            decision,
            root,
            screen_classes,
            decided_at,
        }))
    }

    /// Whether the recorded answer for `peer` still covers the bundle now at
    /// `root` screening as `classes`, and why not when it does not.
    ///
    /// The two invalidators, and the one deliberately absent, are argued in
    /// [`crate::ConsentState::Lapsed`].
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn okf_consent_holds(
        &self,
        peer: &str,
        root: &str,
        classes: &str,
    ) -> Result<crate::ConsentState, StoreError> {
        let Some(record) = self.okf_consent(peer)? else {
            return Ok(crate::ConsentState::Unasked);
        };
        if record.root != root {
            return Ok(crate::ConsentState::Moved { was: record.root });
        }
        if crate::screen_regressed(&record.screen_classes, classes) {
            return Ok(crate::ConsentState::Lapsed {
                was: record.screen_classes,
                now: classes.to_owned(),
            });
        }
        Ok(crate::ConsentState::Holds(record.decision))
    }

    /// Record (or replace) the answer for one peer.
    ///
    /// Replacing rather than appending is deliberate: this is a *current
    /// decision*, not a history. ADR-0019's ledger appends because it records
    /// acts that happened; a consent record answers "what is it now", and two
    /// rows for one peer would need a tie-break nobody would get right.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on write failure.
    pub fn put_okf_consent(&self, record: &crate::OkfConsent) -> Result<(), StoreError> {
        self.conn.execute(
            "INSERT INTO okf_consent (peer, decision, root, screen_classes, decided_at)
             VALUES (?1, ?2, ?3, ?4, ?5)
             ON CONFLICT(peer) DO UPDATE SET
                 decision = excluded.decision,
                 root = excluded.root,
                 screen_classes = excluded.screen_classes,
                 decided_at = excluded.decided_at",
            [
                record.peer.as_str(),
                record.decision.as_str(),
                record.root.as_str(),
                record.screen_classes.as_str(),
                record.decided_at.as_str(),
            ],
        )?;
        Ok(())
    }

    /// Every recorded OKF answer, in peer order — for `roteiro import --from okf`
    /// reporting and for tests.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn okf_consents(&self) -> Result<Vec<crate::OkfConsent>, StoreError> {
        let mut stmt = self.conn.prepare(
            "SELECT peer, decision, root, screen_classes, decided_at
             FROM okf_consent ORDER BY peer",
        )?;
        let rows = stmt.query_map([], |r| {
            Ok((
                r.get::<_, String>(0)?,
                r.get::<_, String>(1)?,
                r.get::<_, String>(2)?,
                r.get::<_, String>(3)?,
                r.get::<_, String>(4)?,
            ))
        })?;
        let mut out = Vec::new();
        for row in rows {
            let (peer, decision, root, screen_classes, decided_at) = row?;
            if let Some(decision) = crate::OkfDecision::from_token(&decision) {
                out.push(crate::OkfConsent {
                    peer,
                    decision,
                    root,
                    screen_classes,
                    decided_at,
                });
            }
        }
        Ok(out)
    }

    /// Replace the findings layer `run.layer` **wholesale**, atomically: the
    /// previous run for that layer and every finding row it owned are deleted,
    /// then this run and its findings are written. A finding that has since been
    /// fixed therefore *disappears* instead of lingering, and re-ingesting an
    /// unchanged report is idempotent — the store ends up with the same rows and
    /// no growth.
    ///
    /// The owned-record cleanup is explicit rather than inherited. The import
    /// path ([`Store::apply_import_layer`]) deletes a layer's edges but leaves its
    /// obsolete *nodes* behind; copying that shape here would silently orphan
    /// findings, so the previous run's rows are deleted by hand and counted in
    /// [`FindingsApplied::removed`]. The schema's `ON DELETE CASCADE` is kept as
    /// defence in depth, not as the mechanism.
    ///
    /// `findings` must carry distinct [`FindingKey`](crate::FindingKey)s: a
    /// duplicate identity is a producer bug and is rejected by the unique index
    /// *inside* the transaction, so nothing is committed. Callers that parse
    /// untrusted reports should reject duplicates earlier, with a better message.
    ///
    /// # Errors
    /// Returns [`StoreError::Json`] if the run's command policy or a finding's
    /// `meta` cannot be serialized, or [`StoreError::Sqlite`] on write failure. On
    /// any error the transaction is rolled back and the previous layer survives
    /// intact.
    pub fn replace_findings_layer(
        &mut self,
        run: &AnalysisRun,
        findings: &[Finding],
    ) -> Result<FindingsApplied, StoreError> {
        let tx = self.conn.transaction()?;
        let applied = findings::replace_layer(&tx, run, findings)?;
        tx.commit()?;
        Ok(applied)
    }

    /// Delete a findings layer and every finding row it owns, returning how many
    /// findings went with it, or `None` if the layer was not live.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on write failure; nothing is committed on
    /// error.
    pub fn delete_findings_layer(&mut self, layer: &str) -> Result<Option<usize>, StoreError> {
        let tx = self.conn.transaction()?;
        let removed = findings::delete_layer(&tx, layer)?;
        tx.commit()?;
        Ok(removed)
    }

    /// Every live findings layer with its findings, ordered by layer key and, in
    /// each layer, by finding key. Pass `analyzer` to narrow to one analyzer.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure, [`StoreError::Json`] if a
    /// stored policy or `meta` cannot be decoded, or [`StoreError::Corrupt`] on an
    /// unrecognised stored token.
    pub fn findings_layers(
        &self,
        analyzer: Option<&str>,
    ) -> Result<Vec<FindingsLayer>, StoreError> {
        findings::layers(&self.conn, analyzer)
    }

    /// Number of findings currently stored, across every layer.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn finding_count(&self) -> Result<u64, StoreError> {
        findings::count_findings(&self.conn)
    }

    /// Number of live analysis runs — one per findings layer.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn analysis_run_count(&self) -> Result<u64, StoreError> {
        findings::count_runs(&self.conn)
    }

    // --- Generated media content (ADR-0015). A separate artifact store, on the
    // same terms as findings above: these methods touch `media_content` only,
    // never `nodes`/`edges`, so `export_factset` stays a pure function of the
    // tree across a `media build`, and generated text can never reach the
    // `authored` relevance boost in `search`. ---

    /// Write one generated-content record, returning `true` if a row was written.
    ///
    /// Keyed by `(blob_id, producer)`. A record for that exact pair already
    /// present is **left alone** and `false` is returned — which is what makes
    /// `media build` incremental and a second run free. A *different* producer is
    /// a new row beside it, never an overwrite: that is the point of keying on the
    /// producer identity, so a better model's description can be compared with the
    /// one it replaces and a distrusted producer can be dropped wholesale.
    /// [`MediaWrite::replace`] (only `media build --force`) is the one path that
    /// overwrites, and only for the identical producer.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on a write failure; the transaction is
    /// rolled back.
    pub fn record_media_content(&mut self, write: &MediaWrite<'_>) -> Result<bool, StoreError> {
        let tx = self.conn.transaction()?;
        let written = media::record(&tx, write)?;
        tx.commit()?;
        Ok(written)
    }

    /// Whether a record already exists for exactly this `(blob, producer)`.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn has_media_record(&self, blob_id: &str, producer: &str) -> Result<bool, StoreError> {
        media::exists(&self.conn, blob_id, producer)
    }

    /// Stored records matching `filter`, ordered by `(producer, blob id)`.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure, or
    /// [`StoreError::Corrupt`] if a row carries an unknown modality token.
    pub fn media_records(&self, filter: &MediaFilter<'_>) -> Result<Vec<MediaRecord>, StoreError> {
        media::records(&self.conn, filter)
    }

    /// Discard records — all of them, or only those written by `producer`.
    /// Returns how many rows went.
    ///
    /// Nothing in the graph is touched: dropping a model you no longer trust must
    /// not cost you a re-sync.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on a write failure.
    pub fn clear_media_content(&mut self, producer: Option<&str>) -> Result<usize, StoreError> {
        let tx = self.conn.transaction()?;
        let removed = media::delete(&tx, producer)?;
        tx.commit()?;
        Ok(removed)
    }

    /// Number of generated-content records currently stored.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn media_content_count(&self) -> Result<u64, StoreError> {
        media::count(&self.conn)
    }

    /// One summary per producer that owns records, ordered by producer id.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure, or
    /// [`StoreError::Corrupt`] if a row carries an unknown modality token.
    pub fn media_producer_summaries(&self) -> Result<Vec<ProducerSummary>, StoreError> {
        media::producer_summaries(&self.conn)
    }

    /// The blob ids that have at least one record carrying **generated text**
    /// for `kind`. A blob the pre-generation gate refused is not described, so it
    /// is not here — see [`Store::gated_media_blobs`].
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn described_media_blobs(
        &self,
        kind: MediaKind,
    ) -> Result<std::collections::BTreeSet<String>, StoreError> {
        media::described_blobs(&self.conn, kind)
    }

    /// The blob ids the [pre-generation gate](crate::media::gate) refused for
    /// `kind` — silent clips, blank images — each of which has a record naming
    /// the value that refused it.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn gated_media_blobs(
        &self,
        kind: MediaKind,
    ) -> Result<std::collections::BTreeSet<String>, StoreError> {
        media::gated_blobs(&self.conn, kind)
    }

    /// Records whose blob is no longer anywhere in `present`. Exposed so a caller
    /// can see what a tree change has orphaned; nothing deletes them implicitly,
    /// because a record is expensive to reproduce and a blob can return.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn orphan_media_records(
        &self,
        present: &std::collections::BTreeSet<String>,
    ) -> Result<Vec<MediaRecord>, StoreError> {
        Ok(self
            .media_records(&MediaFilter::default())?
            .into_iter()
            .filter(|r| !present.contains(&r.blob_id))
            .collect())
    }

    // --- Episodic agent memory (ADR-0013). A separate artifact store, on the
    // same terms as findings and media above: these methods write `agent_memory`
    // and nothing else — they *read* `nodes` to capture and check an anchor, and
    // never write one. So `export_factset` stays a pure function of the tree
    // across every memory write, and nothing an agent remembers can reach the
    // `authored` relevance boost in `search`.
    //
    // These rows are not touched by `rebuild`, following the `imports` precedent:
    // what has no generating function must not be destroyed by a re-derivation.
    // ---

    /// Record one memory, returning its id — the monotonic generation it was
    /// written at.
    ///
    /// The anchor's blob and path, and the `sync_state` tree witness, are
    /// captured **here, from the graph**, so a caller cannot record evidence the
    /// node never carried. An anchor key naming no node is accepted and reads
    /// back as [`crate::AnchorState::Vanished`].
    ///
    /// [`MemoryWrite::supersedes`] records the supersession explicitly, in the
    /// same transaction as the successor: either both land or neither does.
    ///
    /// # Errors
    /// Returns [`MemoryError::InvalidScope`] / [`MemoryError::InvalidBody`] /
    /// [`MemoryError::InvalidConfidence`] for a record that could not be recalled,
    /// [`MemoryError::NotFound`] or [`MemoryError::AlreadySuperseded`] for a bad
    /// supersession target, or [`MemoryError::Store`] on a write failure — in
    /// every case with nothing committed.
    pub fn record_memory(&mut self, write: &MemoryWrite<'_>) -> Result<i64, MemoryError> {
        let tx = self.conn.transaction().map_err(StoreError::from)?;
        let id = memory::record(&tx, write)?;
        tx.commit().map_err(StoreError::from)?;
        Ok(id)
    }

    /// Memory records matching `filter`, newest generation first.
    ///
    /// Live records only unless [`MemoryFilter::include_superseded`] is set: a
    /// superseded record drops out immediately and regardless of age, because the
    /// test is a recorded pointer and not a clock.
    ///
    /// Each record's [`crate::AnchorState`] is computed against the **current**
    /// graph on every call and is never stored.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure, or [`StoreError::Corrupt`]
    /// if a row carries an unknown kind token.
    pub fn memory_records(
        &self,
        filter: &MemoryFilter<'_>,
    ) -> Result<Vec<MemoryRecord>, StoreError> {
        memory::records(&self.conn, filter)
    }

    /// One memory record by id, or `None` if there is no such record.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure, or [`StoreError::Corrupt`]
    /// if the row carries an unknown kind token.
    pub fn memory_record(&self, id: i64) -> Result<Option<MemoryRecord>, StoreError> {
        memory::get(&self.conn, id)
    }

    /// A [`MemoryListing`]: the records matching `filter`, plus the whole store's
    /// live and superseded counts, so an empty result is legible as *nothing
    /// matched* rather than *nothing is stored*.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure, or [`StoreError::Corrupt`]
    /// if a row carries an unknown kind token.
    pub fn memory_listing(&self, filter: &MemoryFilter<'_>) -> Result<MemoryListing, StoreError> {
        let (live, superseded) = memory::counts(&self.conn)?;
        Ok(MemoryListing {
            schema: memory::MEMORY_SCHEMA,
            records: self.memory_records(filter)?,
            live,
            superseded,
        })
    }

    /// **The only way a memory record is ever removed.** Deletes it, and returns
    /// `None` if there was no such record.
    ///
    /// Episodic memory is unbounded and never auto-evicted — no sweep, no TTL, no
    /// capacity bound reaches this table — so an explicit call is the whole
    /// reclamation story. It is also the privacy story: memory has no redaction
    /// chokepoint, so a record that captured a token or a customer name is
    /// removed by asking.
    ///
    /// Anything the deleted record had superseded becomes **live again** and is
    /// named in [`MemoryForgotten::restored`]: leaving it superseded would hide it
    /// on the authority of a record that no longer exists.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on a write failure; the transaction is
    /// rolled back.
    pub fn forget_memory(&mut self, id: i64) -> Result<Option<MemoryForgotten>, StoreError> {
        let tx = self.conn.transaction()?;
        let forgotten = memory::forget(&tx, id)?;
        tx.commit()?;
        Ok(forgotten)
    }

    /// How many memory records are stored, as `(live, superseded)`.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn memory_counts(&self) -> Result<(u64, u64), StoreError> {
        memory::counts(&self.conn)
    }

    /// **Ranked recall**: the live records that match `opts`, scored
    /// `base_confidence × anchor_penalty × decay(age)` and ordered best first.
    ///
    /// Every term is computed **here, at retrieval time, and written to no
    /// column** (ADR-0013). A stored score that decayed would rewrite the store on
    /// every read and would be wrong in between, making recall depend on when you
    /// last looked. Three consequences follow, and each is a promise:
    ///
    /// - **This call mutates nothing.** Recall over an unchanged store and an
    ///   unchanged tree is idempotent, which is what makes
    ///   [`crate::Decay::None`] byte-identical across runs.
    /// - **A superseded record is never returned**, immediately and regardless of
    ///   age: the test is a recorded pointer, not a clock.
    /// - **A record whose anchor no longer resolves is still returned**, demoted
    ///   and labelled. Drift ranks it down; nothing deletes it.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure, or [`StoreError::Corrupt`]
    /// if a row carries an unknown kind token.
    pub fn recall_memory(&self, opts: &RecallOptions<'_>) -> Result<Recall, StoreError> {
        let (live, superseded) = memory::counts(&self.conn)?;
        Ok(Recall {
            schema: memory::RECALL_SCHEMA,
            generation: memory::generation(&self.conn)?,
            decay: opts.decay,
            reproducible: opts.decay.is_reproducible(),
            results: memory::recall(&self.conn, opts)?,
            live,
            superseded,
        })
    }

    // --- The bounded cache tier (ADR-0013, Tier 2). The *opposite* rules to the
    // episodic tier above, because it holds the opposite kind of knowledge:
    // everything here is re-derivable, so eviction costs cycles and never
    // information. Nothing in this section can reach `agent_memory` — it has no
    // `bytes`, no `last_used` and no `hits` for a capacity policy to grip. ---

    /// Write (or replace) one cache entry.
    ///
    /// The payload size is computed here and the anchor's blob is captured from
    /// the graph here, so the sweep can order and total the tier without reading
    /// it, and a caller cannot record evidence a node never carried.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on write failure.
    pub fn agent_cache_put(&self, write: &CacheWrite<'_>) -> Result<(), StoreError> {
        memory::cache_put(&self.conn, write)
    }

    /// Read one cache entry back, **recording the access** — `hits` increments and
    /// `last_used` advances.
    ///
    /// This is the one read in the memory store that writes, and what it writes is
    /// the cache's own bookkeeping: those two columns exist to be moved by exactly
    /// this, and a hit counter nothing increments is a column that lies. It
    /// touches nothing outside `agent_cache`, so [`Store::recall_memory`] — the
    /// read whose reproducibility is promised — stays free of it.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn agent_cache_get(&self, key: &str) -> Result<Option<CacheEntry>, StoreError> {
        memory::cache_get(&self.conn, key)
    }

    /// Every cache entry, ordered by key, **without** recording an access:
    /// inspecting a cache is not using it.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn agent_cache_entries(&self) -> Result<Vec<CacheEntry>, StoreError> {
        memory::cache_entries(&self.conn)
    }

    /// Delete one cache entry, returning whether there was one.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on write failure.
    pub fn agent_cache_forget(&self, key: &str) -> Result<bool, StoreError> {
        memory::cache_forget(&self.conn, key)
    }

    /// What the cache tier holds, against `budget_bytes`.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn agent_cache_stats(&self, budget_bytes: u64) -> Result<CacheStats, StoreError> {
        memory::cache_stats(&self.conn, budget_bytes)
    }

    /// **Sweep the cache tier down to `budget_bytes`**, evicting oldest-first on
    /// `(anchor_valid ASC, last_used ASC)`, and advance the generation.
    ///
    /// Called at the maintenance seam (beside `refresh_contexts`) and **never on
    /// the read path**, so an ordinary query never mutates the store. Three things
    /// are never evicted: anything episodic — structurally, there is no column to
    /// grip it by; an entry written in the current generation whose anchor still
    /// applies, which is the session's own work; and the most-recently-used entry,
    /// always, even if it alone exceeds the budget.
    ///
    /// That last pair means a sweep can legitimately finish still over budget.
    /// [`CacheSweep::over_budget`] reports it rather than leaving a bound that
    /// silently failed to bind.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on failure; the transaction is rolled back.
    pub fn sweep_agent_cache(&mut self, budget_bytes: u64) -> Result<CacheSweep, StoreError> {
        let tx = self.conn.transaction()?;
        let swept = memory::cache_sweep(&tx, budget_bytes)?;
        tx.commit()?;
        Ok(swept)
    }

    /// Findings whose owning run no longer exists. Always `0` in a healthy store;
    /// exposed so layer replacement can be asserted to clean up its own records
    /// rather than orphaning them.
    ///
    /// # Errors
    /// Returns [`StoreError::Sqlite`] on query failure.
    pub fn orphan_finding_count(&self) -> Result<u64, StoreError> {
        findings::count_orphan_findings(&self.conn)
    }
}

// --- Free helpers operating on a `Connection` (a `Transaction` derefs to one) ---

fn node_row_id(conn: &Connection, key: &str) -> rusqlite::Result<Option<i64>> {
    conn.query_row("SELECT id FROM nodes WHERE key = ?1", [key], |r| r.get(0))
        .optional()
}

/// Record (or clear) the last-synced `HEAD` tree id. Shared by `rebuild` and
/// `reconcile` so both leave identical `sync_state`.
fn write_sync_state(conn: &Connection, tree: Option<&str>) -> Result<(), StoreError> {
    match tree {
        // Clear `env` on every tree write: it is only valid for the tree a
        // committed [`crate::sync`] set it against, and that sync re-records it (via
        // `set_sync_env`) immediately after. So a worktree/index sync, or any
        // path that does not re-set it, leaves `env` NULL — reading as "unknown"
        // and forcing the safe full re-extraction next time.
        Some(tree) => conn.execute(
            "INSERT INTO sync_state (id, tree) VALUES (0, ?1)
             ON CONFLICT(id) DO UPDATE SET tree = excluded.tree, env = NULL",
            [tree],
        )?,
        None => conn.execute("DELETE FROM sync_state WHERE id = 0", [])?,
    };
    Ok(())
}

fn upsert_node(conn: &Connection, node: &Node) -> Result<(), StoreError> {
    let meta = serde_json::to_string(&node.meta)?;
    let (span_start, span_end) = match node.span {
        Some(s) => (Some(i64::from(s.start)), Some(i64::from(s.end))),
        None => (None, None),
    };
    conn.execute(
        "INSERT INTO nodes (key, kind, name, path, lang, blob_hash, span_start, span_end, provenance, meta)
         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)
         ON CONFLICT(key) DO UPDATE SET
             kind = excluded.kind, name = excluded.name, path = excluded.path,
             lang = excluded.lang, blob_hash = excluded.blob_hash,
             span_start = excluded.span_start, span_end = excluded.span_end,
             provenance = excluded.provenance, meta = excluded.meta",
        params![
            node.key,
            node.kind.as_str(),
            node.name,
            node.path,
            node.lang,
            node.blob_hash,
            span_start,
            span_end,
            node.provenance.as_str(),
            meta,
        ],
    )?;
    Ok(())
}

fn insert_edge(conn: &Connection, edge: &Edge) -> Result<(), StoreError> {
    validate_edge(edge)?;
    let src_id =
        node_row_id(conn, &edge.src)?.ok_or_else(|| StoreError::UnknownNode(edge.src.clone()))?;
    let dst_id =
        node_row_id(conn, &edge.dst)?.ok_or_else(|| StoreError::UnknownNode(edge.dst.clone()))?;
    insert_edge_row(conn, edge, src_id, dst_id)
}

/// Apply `edge` only if both endpoints already resolve to nodes, returning
/// whether it was **applied** (both endpoints resolved; a duplicate of an
/// existing edge is a harmless no-op via `ON CONFLICT DO NOTHING` but still
/// reports `true`). A missing endpoint returns `false` rather than erroring —
/// the caller prunes such dangling cross-references from the import layer.
fn insert_edge_if_present(conn: &Connection, edge: &Edge) -> Result<bool, StoreError> {
    validate_edge(edge)?;
    let (Some(src_id), Some(dst_id)) =
        (node_row_id(conn, &edge.src)?, node_row_id(conn, &edge.dst)?)
    else {
        return Ok(false);
    };
    insert_edge_row(conn, edge, src_id, dst_id)?;
    Ok(true)
}

/// Apply `edges`, keeping those whose endpoints resolve and pruning the rest.
/// Returns the kept edges plus the applied/pruned counts (in an [`ImportApplied`]
/// whose `layers`/`nodes` are left zero for the caller to fill).
fn apply_edges_pruning(
    conn: &Connection,
    edges: &[Edge],
) -> Result<(Vec<Edge>, ImportApplied), StoreError> {
    let mut kept = Vec::with_capacity(edges.len());
    let mut counts = ImportApplied::default();
    for edge in edges {
        if insert_edge_if_present(conn, edge)? {
            kept.push(edge.clone());
            counts.edges_applied += 1;
        } else {
            counts.edges_pruned += 1;
        }
    }
    Ok((kept, counts))
}

/// Delete the nodes `src_ref` has withdrawn since its last apply, returning how
/// many went.
///
/// `prior` is the key set the persisted layer held before this apply; `current`
/// is what it holds now. The difference is what the source stopped asserting —
/// and the two guards below are what keep "stopped asserting" from becoming
/// "deleted somebody else's node".
///
/// Both guards read the **live store**, not the key's namespace. A namespace rule
/// would have to be re-derived for every layer that ever exists, and would be
/// wrong the first time two producers shared a prefix; a query is right by
/// construction for a layer this function has never heard of.
fn remove_withdrawn_nodes(
    conn: &Connection,
    src_ref: &str,
    prior: &BTreeSet<String>,
    current: &[Node],
) -> Result<usize, StoreError> {
    let held: BTreeSet<&str> = current.iter().map(|n| n.key.as_str()).collect();
    let withdrawn: Vec<&str> = prior
        .iter()
        .map(String::as_str)
        .filter(|k| !held.contains(k))
        .collect();
    if withdrawn.is_empty() {
        return Ok(0);
    }

    // Guard 1: every other persisted layer's node keys. Two layers naming one
    // node is a supported arrangement, not an anomaly — `import:links` and
    // `import:links/authored` share every `extref:` placeholder between them.
    let mut claimed: BTreeSet<String> = BTreeSet::new();
    {
        let mut stmt = conn.prepare("SELECT facts FROM imports WHERE src_ref <> ?1")?;
        let rows = stmt.query_map([src_ref], |r| r.get::<_, String>(0))?;
        for row in rows {
            let facts: FactSet = serde_json::from_str(&row?)?;
            claimed.extend(facts.nodes.into_iter().map(|n| n.key));
        }
    }

    let mut removed = 0usize;
    for key in withdrawn {
        if claimed.contains(key) {
            continue;
        }
        // Guard 2: anything still attached. A surviving edge belongs to some
        // other producer, and removing the node would take that relationship
        // with it — a deletion nobody asked for, disguised as a cleanup. It
        // would also violate the foreign key, so this is correctness twice over.
        //
        // `EXISTS`, not `COUNT(*) > 0`. The question is "is there one", and a
        // count says "tally them all, then ask whether the tally was zero" —
        // which is both a different sentence and more work: with the `OR dst`
        // arm, a well-connected node in this repository would scan on the order
        // of 1,600 edge rows where the probe stops at the first hit.
        let attached: bool = conn.query_row(
            "SELECT EXISTS (SELECT 1 FROM edges \
             WHERE src = (SELECT id FROM nodes WHERE key = ?1) \
                OR dst = (SELECT id FROM nodes WHERE key = ?1))",
            [key],
            |r| r.get(0),
        )?;
        if attached {
            continue;
        }
        let gone = conn.execute("DELETE FROM nodes WHERE key = ?1", [key])?;
        if gone > 0 {
            // A cached context bundle for a node that no longer exists can only
            // ever be served as a stale answer, so it goes with it.
            conn.execute("DELETE FROM node_context WHERE key = ?1", [key])?;
            removed += 1;
        }
    }
    Ok(removed)
}

/// Upsert a persisted import layer row. Free helper so it can run inside the same
/// transaction as an apply/prune pass.
fn put_import_row(conn: &Connection, src_ref: &str, facts: &FactSet) -> Result<(), StoreError> {
    let json = serde_json::to_string(facts)?;
    conn.execute(
        "INSERT INTO imports (src_ref, facts) VALUES (?1, ?2)
         ON CONFLICT(src_ref) DO UPDATE SET facts = excluded.facts, imported_at = datetime('now')",
        params![src_ref, json],
    )?;
    Ok(())
}

/// The provenance/confidence invariant guard shared by the strict and tolerant
/// edge inserts.
fn validate_edge(edge: &Edge) -> Result<(), StoreError> {
    if edge.is_valid() {
        Ok(())
    } else {
        Err(StoreError::InvalidEdge(format!(
            "confidence must be present iff provenance is inferred (src={}, dst={})",
            edge.src, edge.dst
        )))
    }
}

/// Insert an edge row given already-resolved endpoint ids. Edges are a set: a
/// duplicate `(src, dst, kind, provenance)` is a no-op via `ON CONFLICT … DO
/// NOTHING`, so re-applying a fact set never accumulates duplicates. Other
/// constraint violations (guarded in Rust above) still surface.
fn insert_edge_row(
    conn: &Connection,
    edge: &Edge,
    src_id: i64,
    dst_id: i64,
) -> Result<(), StoreError> {
    conn.execute(
        "INSERT INTO edges (src, dst, kind, provenance, confidence, src_ref)
         VALUES (?1, ?2, ?3, ?4, ?5, ?6)
         ON CONFLICT(src, dst, kind, provenance) DO NOTHING",
        params![
            src_id,
            dst_id,
            edge.kind.as_str(),
            edge.provenance.as_str(),
            edge.confidence,
            edge.src_ref,
        ],
    )?;
    Ok(())
}

/// A hashable identity for an edge over **all** its fields — used by
/// [`Store::reconcile`] to diff the edge set. A tuple (not a delimiter-joined
/// string) so no field value can be confused with a separator: node keys embed
/// git paths, which may legally contain any byte (including control characters),
/// so a joined string could collapse distinct edges to one identity and drop an
/// edge. Confidence is compared by its exact bit pattern (`f64::to_bits`, wrapped
/// in `Option` so `None` and `Some(_)` stay distinct), the only non-`Eq` field.
fn edge_identity(edge: &Edge) -> EdgeId {
    (
        edge.src.clone(),
        edge.dst.clone(),
        edge.kind.as_str().to_owned(),
        edge.provenance.as_str().to_owned(),
        edge.confidence.map(f64::to_bits),
        edge.src_ref.clone(),
    )
}

/// The tuple form of an edge's full-field identity (see [`edge_identity`]):
/// `(src, dst, kind, provenance, confidence-bits, src_ref)`.
type EdgeId = (String, String, String, String, Option<u64>, Option<String>);

/// Delete the edge row identified by `(src, dst, kind, provenance)` — the table's
/// unique key — resolving the endpoint node keys to ids. A no-op if absent.
fn delete_edge(conn: &Connection, edge: &Edge) -> Result<(), StoreError> {
    conn.execute(
        "DELETE FROM edges
         WHERE src = (SELECT id FROM nodes WHERE key = ?1)
           AND dst = (SELECT id FROM nodes WHERE key = ?2)
           AND kind = ?3 AND provenance = ?4",
        params![
            edge.src,
            edge.dst,
            edge.kind.as_str(),
            edge.provenance.as_str()
        ],
    )?;
    Ok(())
}

fn collect_nodes(rows: &mut rusqlite::Rows) -> Result<Vec<Node>, StoreError> {
    let mut out = Vec::new();
    while let Some(row) = rows.next()? {
        out.push(row_to_node(row)?);
    }
    Ok(out)
}

fn collect_edges(rows: &mut rusqlite::Rows) -> Result<Vec<Edge>, StoreError> {
    let mut out = Vec::new();
    while let Some(row) = rows.next()? {
        out.push(row_to_edge(row)?);
    }
    Ok(out)
}

fn row_to_node(row: &rusqlite::Row) -> Result<Node, StoreError> {
    let kind: String = row.get("kind")?;
    let span_start: Option<i64> = row.get("span_start")?;
    let span_end: Option<i64> = row.get("span_end")?;
    let span = match (span_start, span_end) {
        (Some(s), Some(e)) => Some(Span::new(to_u32(s)?, to_u32(e)?)),
        _ => None,
    };
    let meta: String = row.get("meta")?;
    let provenance: String = row.get("provenance")?;
    let provenance = Provenance::from_token(&provenance)
        .ok_or_else(|| StoreError::Corrupt(unknown_provenance("node", &provenance)))?;
    Ok(Node {
        key: row.get("key")?,
        kind: NodeKind::from_token(&kind),
        name: row.get("name")?,
        path: row.get("path")?,
        lang: row.get("lang")?,
        blob_hash: row.get("blob_hash")?,
        span,
        provenance,
        meta: serde_json::from_str(&meta)?,
    })
}

fn row_to_edge(row: &rusqlite::Row) -> Result<Edge, StoreError> {
    let kind: String = row.get("kind")?;
    let provenance: String = row.get("provenance")?;
    let provenance = Provenance::from_token(&provenance)
        .ok_or_else(|| StoreError::Corrupt(unknown_provenance("edge", &provenance)))?;
    Ok(Edge {
        src: row.get("src")?,
        dst: row.get("dst")?,
        kind: EdgeKind::from_token(&kind),
        provenance,
        confidence: row.get("confidence")?,
        src_ref: row.get("src_ref")?,
    })
}

fn to_u32(v: i64) -> Result<u32, StoreError> {
    u32::try_from(v).map_err(|_| StoreError::Corrupt(format!("span offset out of range: {v}")))
}

/// The message for a stored provenance token this build cannot parse.
///
/// It names **both** causes, because until migration 14 there was only one and
/// the old wording said so: "e.g. a corrupt database row". That migration widened
/// the vocabulary, so an unrecognised token now most often means the store was
/// written by a newer Roteiro — [`Store::schema_ahead`] says so first and more
/// precisely, but a reader who reaches this message deserves not to be sent
/// looking for corruption that is not there.
fn unknown_provenance(what: &str, token: &str) -> String {
    format!(
        "unknown {what} provenance `{token}`: expected one of {}. Either this row is corrupt, \
         or this store was written by a newer Roteiro that knows a provenance this build does \
         not — check `roteiro --version` against the build that last wrote the graph.",
        Provenance::tokens().join(", ")
    )
}

/// The true provenance of an import-layer node, from its key namespace: Graphify
/// nodes (`graphify:`) are [`Provenance::Inferred`]; an OKF concept (`okf:`) is
/// [`Provenance::ExternalInferred`]; every other import node (lat, …) is
/// [`Provenance::Authored`]. Import-layer nodes are never derived, so this is
/// used to repair a legacy `Derived` tag on load (see `load_import_layers`).
fn import_node_provenance(key: &str) -> Provenance {
    if key.starts_with("graphify:") {
        Provenance::Inferred
    } else if key.starts_with("okf:") {
        // An OKF concept is a peer's, and this is a *repair* path with no
        // frontmatter left to consult — so it takes the lowest claim the peer
        // could have made. Falling through to `Authored` below would assert that
        // this graph human-authored another repository's concept, which is the
        // exact laundering the `external-*` tier exists to refuse, and it would
        // arrive through an error-recovery path where nobody would look for it.
        //
        // A **filled placeholder** deliberately does not match: its key is
        // `extref:`, it is a node this repo created, and its own provenance is
        // whichever layer wrote it last (see `rto_graph::links`).
        Provenance::ExternalInferred
    } else {
        Provenance::Authored
    }
}

#[cfg(test)]
mod tests {
    use super::Store;
    use crate::model::{Direction, Edge, EdgeKind, FactSet, Node, NodeKind, Span};
    use crate::provenance::Provenance;

    /// A fresh directory for a file-backed store, cleared on the way out of this
    /// function rather than trusted to be absent.
    ///
    /// Keyed by process id **and** a monotonic counter, matching
    /// `rto-render`'s `okf_inspect` tests and the CLI tests' `bundle` helper —
    /// whose own note explains why, and which I should have followed first time:
    /// *"uniqueness must not depend on everyone remembering to pick a distinct
    /// name."*
    ///
    /// The pre-clean is the half that matters most here, and is what review
    /// caught. A store is not one file: a failed run leaves `graph.db` beside its
    /// `-wal` and `-shm`, and reusing that state would have the concurrency test
    /// counting rows a previous run inserted — failing, eventually, for a reason
    /// that has nothing to do with locking. A test that can fail for the wrong
    /// reason is worse than no test, because the next person debugs the wrong
    /// thing.
    fn scratch(tag: &str) -> std::path::PathBuf {
        static SEQ: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
        let seq = SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        let root =
            std::env::temp_dir().join(format!("rto-store-{}-{seq}-{tag}", std::process::id()));
        let _ = std::fs::remove_dir_all(&root);
        root
    }

    fn sample_node(key: &str) -> Node {
        Node {
            key: key.to_owned(),
            kind: NodeKind::Fn,
            name: "sample".to_owned(),
            path: Some("src/lib.rs".to_owned()),
            lang: Some("rust".to_owned()),
            blob_hash: Some("deadbeef".to_owned()),
            span: Some(Span::new(10, 42)),
            provenance: Provenance::Derived,
            meta: serde_json::json!({"vis": "pub"}),
        }
    }

    /// Two Roteiro processes over one repository is ordinary rather than exotic —
    /// an editor's MCP client and a terminal, a long-lived `serve` and a [`crate::sync`],
    /// or several clients each spawning their own stdio server. Both writing must
    /// **queue**, not kill one of them.
    ///
    /// Reproduces the shape every upsert in this file has: *read, then write,
    /// inside one transaction.* That is precisely the case the five-second
    /// `busy_timeout` never covered — SQLite refuses a lock upgrade outright
    /// rather than waiting on it, because waiting could deadlock, so the handler
    /// is never invoked and the second writer dies at once. See `from_conn`.
    ///
    /// Drives the connection directly rather than a public method because the
    /// first writer has to still be *holding* the lock when the second arrives,
    /// and no public API lets a caller pause mid-transaction.
    #[test]
    fn a_second_writer_waits_for_the_first_instead_of_failing() {
        let dir = scratch("lock");
        std::fs::create_dir_all(&dir).expect("temp dir");
        let path = dir.join("graph.db");

        let mut first = Store::open(&path).expect("first store opens");
        first
            .conn
            .execute_batch("CREATE TABLE IF NOT EXISTS probe(v INTEGER);")
            .expect("probe table");
        let mut second = Store::open(&path).expect("second store opens");

        // Signalled once the first writer is inside its transaction *and* holding
        // the write lock, so the second arrives during the contention rather than
        // racing to it.
        //
        // A channel rather than a `Barrier`, because `Barrier::wait` cannot fail:
        // if the writer panicked before reaching it — a failing `expect`, a
        // migration error — the main thread would wait for ever and the test would
        // hang CI instead of failing it. Dropping the `Sender` on a panicking
        // thread disconnects the channel, so `recv_timeout` returns *immediately*
        // in that case, and the timeout covers the merely-pathological one.
        // Raised in review of this change.
        let (signal, holding) = std::sync::mpsc::channel::<()>();

        let writer = std::thread::spawn(move || {
            let tx = first.conn.transaction().expect("first writer opens");
            let _: i64 = tx
                .query_row("SELECT count(*) FROM probe", [], |r| r.get(0))
                .expect("first writer reads");
            tx.execute("INSERT INTO probe(v) VALUES (1)", [])
                .expect("first writer writes");
            // Ignored deliberately: if the main thread has already gone (its own
            // assertion failed), there is nobody to tell and nothing to do.
            let _ = signal.send(());
            std::thread::sleep(std::time::Duration::from_millis(250));
            tx.commit().expect("first writer commits");
        });

        if holding
            .recv_timeout(std::time::Duration::from_secs(10))
            .is_err()
        {
            // Re-raise the writer's own panic, which says far more than "timed
            // out" would; only if it did not panic is the timeout itself the news.
            writer.join().expect("first writer reached the lock");
            panic!("first writer neither signalled nor panicked");
        }
        let tx = second
            .conn
            .transaction()
            .expect("a second writer must wait for the first, not be refused");
        let _: i64 = tx
            .query_row("SELECT count(*) FROM probe", [], |r| r.get(0))
            .expect("second writer reads");
        tx.execute("INSERT INTO probe(v) VALUES (2)", [])
            .expect("second writer writes rather than failing `database is locked`");
        tx.commit().expect("second writer commits");

        writer.join().expect("first writer finished");

        let seen: i64 = second
            .conn
            .query_row("SELECT count(*) FROM probe", [], |r| r.get(0))
            .expect("count");
        assert_eq!(seen, 2, "both writers' rows survive — neither was dropped");

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

    /// A file-backed store takes WAL wherever the filesystem allows it, so a
    /// reader is not shut out for the whole of a writer's transaction.
    ///
    /// Asserted **against what a plain connection gets on this same filesystem**
    /// rather than against the literal `wal`, because `from_conn` treats WAL as
    /// tolerated rather than required: it is refused on most network filesystems,
    /// and a hard assertion would fail there for a store behaving exactly as
    /// designed. Raised in review — the first version of this test contradicted
    /// the tolerance documented by the code it tests.
    ///
    /// Still not vacuous, which is what an "assert the two agree" test has to
    /// earn: the probe is a raw `Connection` that asks for WAL itself, so a
    /// `from_conn` that stopped asking would leave the probe reporting `wal` and
    /// the store reporting something else, and the two would disagree.
    #[test]
    fn a_file_backed_store_takes_whatever_wal_this_filesystem_allows() {
        let dir = scratch("wal");
        std::fs::create_dir_all(&dir).expect("temp dir");

        let available: String = {
            let probe = super::Connection::open(dir.join("probe.db")).expect("probe opens");
            probe
                .query_row("PRAGMA journal_mode = WAL", [], |r| r.get(0))
                .expect("probe journal mode")
        };

        let store = Store::open(&dir.join("graph.db")).expect("store opens");
        let mode: String = store
            .conn
            .query_row("PRAGMA journal_mode", [], |r| r.get(0))
            .expect("journal mode");

        assert_eq!(
            mode.to_ascii_lowercase(),
            available.to_ascii_lowercase(),
            "a store must reach the same journal mode a plain connection does here"
        );

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

    /// And an in-memory store still opens, which is the tolerance path: WAL is
    /// meaningless there, `PRAGMA journal_mode` answers `memory`, and that must
    /// not be an error — the same branch a network filesystem takes.
    #[test]
    fn an_in_memory_store_opens_without_wal() {
        let store = Store::open_in_memory().expect("in-memory store opens");
        let mode: String = store
            .conn
            .query_row("PRAGMA journal_mode", [], |r| r.get(0))
            .expect("journal mode");
        assert_ne!(mode.to_ascii_lowercase(), "wal");
    }

    #[test]
    fn reconcile_matches_a_full_rebuild() {
        // reconcile must leave the store identical to a fresh rebuild, across an
        // add, a remove, a content change, and edge churn.
        let node = |k: &str, name: &str| {
            let mut n = sample_node(k);
            n.name = name.to_owned();
            n
        };
        let edge =
            |src: &str, dst: &str| Edge::derived(src.to_owned(), dst.to_owned(), EdgeKind::Calls);
        // An inferred edge carries confidence and a src_ref — exercise both so the
        // equivalence claim covers every edge field, not just derived calls.
        let inferred = |src: &str, dst: &str, conf: f64| {
            let mut e = Edge::inferred(src.to_owned(), dst.to_owned(), EdgeKind::Related, conf);
            e.src_ref = Some("import:demo".to_owned());
            e
        };

        let facts1 = FactSet {
            nodes: vec![node("a", "A"), node("b", "B"), node("c", "C")],
            edges: vec![edge("a", "b"), edge("b", "c"), inferred("a", "c", 0.7)],
        };
        // b changes (name), c is removed, d is added; edge b->c drops, a->d added,
        // and the inferred edge's confidence changes.
        let facts2 = FactSet {
            nodes: vec![node("a", "A"), node("b", "B2"), node("d", "D")],
            edges: vec![edge("a", "b"), edge("a", "d"), inferred("a", "d", 0.9)],
        };

        // Path 1: rebuild facts1, then reconcile to facts2.
        let mut reconciled = Store::open_in_memory().expect("open");
        reconciled.rebuild(&facts1, Some("t1")).expect("rebuild");
        reconciled
            .reconcile(&facts2, Some("t2"))
            .expect("reconcile");

        // Path 2: a fresh full rebuild of facts2.
        let mut rebuilt = Store::open_in_memory().expect("open");
        rebuilt.rebuild(&facts2, Some("t2")).expect("rebuild");

        let canon = |fs: FactSet| {
            let mut nodes = fs.nodes;
            nodes.sort_by(|a, b| a.key.cmp(&b.key));
            let mut edges: Vec<String> = fs
                .edges
                .iter()
                .map(|e| {
                    format!(
                        "{}\0{}\0{}\0{}\0{:?}\0{:?}",
                        e.kind.as_str(),
                        e.src,
                        e.dst,
                        e.provenance.as_str(),
                        e.confidence,
                        e.src_ref
                    )
                })
                .collect();
            edges.sort();
            (nodes, edges)
        };
        assert_eq!(
            canon(reconciled.export_factset().expect("export")),
            canon(rebuilt.export_factset().expect("export")),
            "reconcile must match a full rebuild",
        );
        assert_eq!(
            reconciled.sync_state().expect("state").as_deref(),
            Some("t2")
        );
    }

    /// The worktree stamp (issue #330) identifies the *store*, not the tree: an
    /// unstamped store reads as "unknown" and is adopted rather than rebuilt, and
    /// a new tree state must not clear the stamp the way it clears `env`.
    #[test]
    fn the_worktree_stamp_survives_tree_writes_and_starts_unknown() {
        let mut store = Store::open_in_memory().expect("open");
        let facts = FactSet::new().with_node(sample_node("sym:a"));

        // Never synced ⇒ no stamp. A legacy store reads the same way, which is
        // why "unknown" must mean adopt, not mismatch.
        assert_eq!(store.synced_worktree().expect("stamp"), None);

        store.reconcile(&facts, Some("t1")).expect("reconcile");
        assert_eq!(
            store.synced_worktree().expect("stamp"),
            None,
            "writing a tree does not invent a stamp; the sync engine records it"
        );

        store.set_synced_worktree("/w/one").expect("stamp");
        store.set_sync_env("env-1").expect("env");
        assert_eq!(
            store.synced_worktree().expect("stamp").as_deref(),
            Some("/w/one")
        );

        // A later tree write clears `env` (it is only valid for one tree) but must
        // KEEP the stamp: a new commit does not move the graph to another tree.
        store.reconcile(&facts, Some("t2")).expect("reconcile");
        assert_eq!(store.sync_env().expect("env"), None, "env is tree-scoped");
        assert_eq!(
            store.synced_worktree().expect("stamp").as_deref(),
            Some("/w/one"),
            "the stamp identifies the store, not the tree, so it must survive"
        );

        // Re-stamping replaces it (the store was adopted by another tree).
        store.set_synced_worktree("/w/two").expect("restamp");
        assert_eq!(
            store.synced_worktree().expect("stamp").as_deref(),
            Some("/w/two")
        );

        // Clearing the synced state clears the stamp with it: a store with no
        // recorded tree describes no working tree either.
        store.rebuild(&facts, None).expect("rebuild unstated");
        assert_eq!(store.sync_state().expect("state"), None);
        assert_eq!(store.synced_worktree().expect("stamp"), None);
    }

    #[test]
    fn reconcile_writes_only_the_edge_delta() {
        // An unchanged edge must keep its row (proving reconcile does not wipe and
        // reinsert the whole edge set); a removed edge's row goes; a new edge's row
        // appears. Row identity is the SQLite `rowid` — stable unless deleted.
        let n = |k: &str| sample_node(k);
        let e =
            |src: &str, dst: &str| Edge::derived(src.to_owned(), dst.to_owned(), EdgeKind::Calls);

        let mut store = Store::open_in_memory().expect("open");
        store
            .rebuild(
                &FactSet {
                    nodes: vec![n("a"), n("b"), n("c")],
                    edges: vec![e("a", "b"), e("b", "c")],
                },
                None,
            )
            .expect("rebuild");

        // Map (src_key, dst_key) → rowid via the private connection.
        let rowids = |store: &Store| -> std::collections::HashMap<(String, String), i64> {
            let mut stmt = store
                .conn
                .prepare(
                    "SELECT ns.key, nd.key, e.rowid FROM edges e \
                     JOIN nodes ns ON ns.id = e.src JOIN nodes nd ON nd.id = e.dst",
                )
                .expect("prepare");
            stmt.query_map([], |r| {
                Ok((
                    (r.get::<_, String>(0)?, r.get::<_, String>(1)?),
                    r.get::<_, i64>(2)?,
                ))
            })
            .expect("query")
            .map(Result::unwrap)
            .collect()
        };

        let before = rowids(&store);
        let ab_rowid = before[&("a".to_owned(), "b".to_owned())];

        // Keep a->b, drop b->c, add a->c.
        store
            .reconcile(
                &FactSet {
                    nodes: vec![n("a"), n("b"), n("c")],
                    edges: vec![e("a", "b"), e("a", "c")],
                },
                None,
            )
            .expect("reconcile");

        let after = rowids(&store);
        assert_eq!(
            after.get(&("a".to_owned(), "b".to_owned())),
            Some(&ab_rowid),
            "the unchanged edge keeps its row (not rewritten)"
        );
        assert!(
            !after.contains_key(&("b".to_owned(), "c".to_owned())),
            "the removed edge's row is gone"
        );
        assert!(
            after.contains_key(&("a".to_owned(), "c".to_owned())),
            "the added edge has a new row"
        );
    }

    #[test]
    fn reconcile_is_history_independent_for_edge_queries() {
        // The whole point of the edge delta: it must be invisible above storage.
        // A store reached by rebuild(f1)+reconcile(f2) has different edge row ids
        // than a cold rebuild at f2, yet every edge query must return byte-for-byte
        // the same result — order included — because the queries are content-ordered.
        let n = |k: &str| sample_node(k);
        let d =
            |src: &str, dst: &str| Edge::derived(src.to_owned(), dst.to_owned(), EdgeKind::Calls);
        let inf = |src: &str, dst: &str, c: f64| {
            Edge::inferred(src.to_owned(), dst.to_owned(), EdgeKind::Related, c)
        };
        let f1 = FactSet {
            nodes: vec![n("a"), n("b"), n("c")],
            edges: vec![d("a", "b"), d("b", "c"), inf("a", "c", 0.7)],
        };
        let f2 = FactSet {
            nodes: vec![n("a"), n("b"), n("d")],
            edges: vec![d("a", "b"), d("a", "d"), inf("a", "d", 0.9)],
        };

        let mut incremental = Store::open_in_memory().expect("open");
        incremental.rebuild(&f1, None).expect("rebuild");
        incremental.reconcile(&f2, None).expect("reconcile");
        let mut cold = Store::open_in_memory().expect("open");
        cold.rebuild(&f2, None).expect("rebuild");

        // Project to all fields so the comparison covers order *and* content.
        let proj = |es: Vec<Edge>| -> Vec<String> {
            es.into_iter()
                .map(|e| {
                    format!(
                        "{}|{}|{}|{}|{:?}|{:?}",
                        e.src,
                        e.dst,
                        e.kind.as_str(),
                        e.provenance.as_str(),
                        e.confidence,
                        e.src_ref
                    )
                })
                .collect()
        };

        for key in ["a", "b", "d"] {
            assert_eq!(
                proj(incremental.edges_from(key).expect("from")),
                proj(cold.edges_from(key).expect("from")),
                "edges_from({key}) must match a cold rebuild"
            );
            assert_eq!(
                proj(incremental.edges_to(key).expect("to")),
                proj(cold.edges_to(key).expect("to")),
                "edges_to({key}) must match a cold rebuild"
            );
        }
        assert_eq!(
            proj(incremental.all_edges().expect("all")),
            proj(cold.all_edges().expect("all")),
            "all_edges must match a cold rebuild"
        );
        for p in [Provenance::Derived, Provenance::Inferred] {
            assert_eq!(
                proj(incremental.edges_by_provenance(p).expect("prov")),
                proj(cold.edges_by_provenance(p).expect("prov")),
                "edges_by_provenance({}) must match a cold rebuild",
                p.as_str()
            );
        }
    }

    #[test]
    fn reconcile_updates_confidence_on_an_unchanged_tuple() {
        // A change to *only* an edge's confidence — same (src, dst, kind,
        // provenance) — must still be applied. Edge identity includes confidence,
        // so it is delete+add (matching a full rebuild), not the insert-time
        // `DO NOTHING` that would leave the stale confidence in place.
        let n = |k: &str| sample_node(k);
        let inf = |c: f64| {
            let mut e = Edge::inferred("a".to_owned(), "b".to_owned(), EdgeKind::Related, c);
            e.src_ref = Some("import:demo".to_owned());
            e
        };

        let mut store = Store::open_in_memory().expect("open");
        store
            .rebuild(
                &FactSet {
                    nodes: vec![n("a"), n("b")],
                    edges: vec![inf(0.5)],
                },
                None,
            )
            .expect("rebuild");
        store
            .reconcile(
                &FactSet {
                    nodes: vec![n("a"), n("b")],
                    edges: vec![inf(0.9)],
                },
                None,
            )
            .expect("reconcile");

        let edges = store.edges_from("a").expect("edges");
        assert_eq!(edges.len(), 1);
        assert_eq!(
            edges[0].confidence,
            Some(0.9),
            "confidence updated, not left stale"
        );
    }

    #[test]
    fn edge_identity_does_not_collide_across_field_boundaries() {
        // Node keys embed git paths, which may contain any byte — including the
        // unit separator (`\x1f`). A delimiter-joined identity would map these two
        // distinct edges to the same string (`a\x1fb\x1fc\x1f…`); the tuple identity
        // must keep them apart, or reconcile would drop one edge as a "duplicate".
        let e1 = super::edge_identity(&Edge::derived(
            "a\u{1f}b".to_owned(),
            "c".to_owned(),
            EdgeKind::Calls,
        ));
        let e2 = super::edge_identity(&Edge::derived(
            "a".to_owned(),
            "b\u{1f}c".to_owned(),
            EdgeKind::Calls,
        ));
        assert_ne!(
            e1, e2,
            "control chars in a key must not collapse identities"
        );

        // A confidence-only difference (same tuple otherwise) also stays distinct,
        // and `None` (derived) never equals `Some(0.0)`.
        let derived = super::edge_identity(&Edge::derived(
            "a".to_owned(),
            "b".to_owned(),
            EdgeKind::Related,
        ));
        let inferred0 = super::edge_identity(&Edge::inferred(
            "a".to_owned(),
            "b".to_owned(),
            EdgeKind::Related,
            0.0,
        ));
        assert_ne!(derived, inferred0, "None vs Some(0.0) confidence differ");
    }

    #[test]
    fn open_in_memory_applies_schema() {
        let store = Store::open_in_memory().expect("open");
        assert_eq!(store.node_count().expect("count"), 0);
        // Written against `latest_version()` rather than the literal that stood
        // here (8 for analyzer findings, then 9, 10, 11 as the media and
        // agent-memory tables landed). The literal was defended as making someone
        // confirm a new migration is meant to apply on open — but it could not do
        // that job: `apply` applies every migration not already recorded, so
        // "applies on open" is not a per-migration choice there is anything to
        // confirm. What the literal actually asserted was the value of a shared
        // constant, which every future migration then has to come here and edit,
        // in a file it otherwise has no business in — the brittleness #329
        // replaced with a property test elsewhere. This is the idiom
        // `migrations::tests::a_later_migration_is_additive_on_a_populated_store`
        // already uses, for the same reason.
        //
        // Since migrations are selected by set membership, the property is now
        // strictly stronger than "the biggest recorded number is N":
        // `schema_version` is the highest **gap-free** version, so equality with
        // `latest_version` also rules out a store that skipped one and stamped a
        // higher one. A fresh store cannot exhibit that gap, so
        // `reopening_repairs_a_skipped_migration` exercises it directly.
        assert_eq!(
            store.schema_version().expect("version"),
            crate::migrations::latest_version(),
            "opening a store must apply every known migration, with no gaps"
        );
    }

    /// Reopening a store that is **missing a lower-numbered migration while
    /// carrying a higher one** repairs it — through the real public API, which
    /// is where it matters to callers.
    ///
    /// This is not a hypothetical: it is the shape `main` itself carries.
    /// `feat/stage25-memory-recall` merged migration **13** while migration
    /// **12** was still on this branch, so every store opened by a current-`main`
    /// build — including this repository's own `.git/roteiro/graph.db` and each
    /// linked worktree's — records `1..11, 13`. Under the old
    /// `version > MAX(recorded)` rule, `12 > 13` is false and migration 12 would
    /// never be applied to any of them, permanently: the store opens cleanly,
    /// reports a schema it does not have, and `synced_worktree()` fails with
    /// `no such column: worktree` — the issue #330 tree stamp, silently absent.
    ///
    /// The two migrations are deliberately named by number rather than derived
    /// from `latest_version()`. Which one is missing and which is present is the
    /// specific historical fact being tested; arithmetic on the newest version
    /// would quietly re-aim the test at whatever lands next and stop covering
    /// this.
    #[test]
    fn reopening_repairs_a_skipped_migration() {
        /// The migration this branch adds — absent from a current-`main` store.
        const SKIPPED: u32 = 12;
        /// The migration `main` added in parallel — present, and stamped higher.
        const AHEAD: u32 = 13;

        let path = std::env::temp_dir().join(format!(
            "roteiro-migration-gap-{}-{:?}.db",
            std::process::id(),
            std::thread::current().id()
        ));
        std::fs::remove_file(&path).ok();

        {
            let mut store = Store::open(&path).expect("open");
            // Reconcile rather than a bare upsert: it records a synced tree, so
            // `sync_state` has the row the worktree stamp updates. Without one,
            // `set_synced_worktree` is a documented no-op and would prove nothing.
            let facts = FactSet::new().with_node(sample_node("kept"));
            store
                .reconcile(&facts, Some("t1"))
                .expect("seed graph and sync state");
        }
        // Rewind to exactly what a current-`main` build leaves behind: migration
        // 12's column and record gone, 13's schema and record intact.
        {
            let conn = rusqlite::Connection::open(&path).expect("reopen raw");
            conn.execute_batch("ALTER TABLE sync_state DROP COLUMN worktree;")
                .expect("undo migration 12");
            conn.execute(
                "DELETE FROM schema_migrations WHERE version = ?1",
                [SKIPPED],
            )
            .expect("unrecord 12");
            // Sanity: the store really is in the damaged shape, 13 and all.
            let recorded: i64 = conn
                .query_row(
                    "SELECT COUNT(*) FROM schema_migrations WHERE version = ?1",
                    [AHEAD],
                    |r| r.get(0),
                )
                .expect("count 13");
            assert_eq!(recorded, 1, "migration {AHEAD} must still be recorded");
        }
        // Migration 13's own state, captured before the repair so it can be shown
        // untouched afterwards — a repair that healed the gap by disturbing the
        // other branch's tables would be no better than the bug.
        let clock_before = cache_clock(&path);

        // Before the repair the store under-reports rather than over-reports: it
        // claims the last gap-free version, never the higher stamped one.
        {
            let damaged = rusqlite::Connection::open(&path).expect("reopen raw");
            assert_eq!(
                crate::migrations::store_version(&damaged).expect("version"),
                SKIPPED - 1,
                "a gapped store must report the version it actually provides, \
                 not the maximum recorded ({AHEAD})"
            );
        }

        // Reopening through the public API repairs the gap…
        let store = Store::open(&path).expect("reopen");
        assert_eq!(
            store.schema_version().expect("version"),
            crate::migrations::latest_version(),
            "1..={AHEAD} is contiguous once the skipped migration is applied"
        );
        // …migration 12's schema is really back, which is what the caller
        // depends on — recorded-but-absent would be the same bug in a new
        // costume…
        assert!(
            store.synced_worktree().is_ok(),
            "the repaired migration's column must exist, not merely be recorded"
        );
        store
            .set_synced_worktree("/w/repaired")
            .expect("the stamp must be writable, not just readable");
        assert_eq!(
            store.synced_worktree().expect("stamp").as_deref(),
            Some("/w/repaired")
        );
        // …migration 13's tables are untouched, clock row and all…
        assert_eq!(
            cache_clock(&path),
            clock_before,
            "the other branch's seeded `agent_cache_clock` row must survive the \
             repair unchanged"
        );
        {
            let conn = rusqlite::Connection::open(&path).expect("reopen raw");
            let rows: i64 = conn
                .query_row("SELECT COUNT(*) FROM agent_cache", [], |r| r.get(0))
                .expect("agent_cache must still exist");
            assert_eq!(rows, 0, "`agent_cache` is present and empty, not recreated");
        }
        // …and no data was lost.
        assert!(store.get_node("kept").expect("get").is_some());
        assert_eq!(store.node_count().expect("count"), 1);

        std::fs::remove_file(&path).expect("cleanup");
    }

    /// Migration 13's single `agent_cache_clock` row as `(ticks, generation)`,
    /// read raw so this test does not depend on the recall API another branch
    /// owns.
    fn cache_clock(path: &std::path::Path) -> (i64, i64) {
        let conn = rusqlite::Connection::open(path).expect("open raw");
        conn.query_row(
            "SELECT ticks, generation FROM agent_cache_clock WHERE id = 0",
            [],
            |r| Ok((r.get(0)?, r.get(1)?)),
        )
        .expect("the seeded clock row must exist")
    }

    /// Record `version` in `store`'s migration table as though a build that knew
    /// that migration had run it — the only way to manufacture a store from the
    /// future without a time machine.
    ///
    /// Only the *record* is written, not any schema the migration would have
    /// added. That is exactly the position an older binary is in: it can see the
    /// stamp and cannot see, or reason about, the shape.
    fn stamp_version(store: &Store, version: u32) {
        store
            .conn
            .execute(
                "INSERT INTO schema_migrations (version) VALUES (?1)",
                [version],
            )
            .expect("stamp a version this build does not know");
    }

    #[test]
    fn a_store_this_build_wrote_is_not_ahead_of_it() {
        let store = Store::open_in_memory().expect("open");
        assert_eq!(
            store.schema_ahead().expect("schema_ahead"),
            None,
            "a store this very build just migrated cannot be ahead of it"
        );
    }

    /// A store carrying migrations this build has never heard of is reported
    /// ahead, and the report names **both** sides plus what to do about it.
    ///
    /// Versions are `latest_version() + n` rather than literals: the property is
    /// "beyond what this build knows", and a literal would silently stop testing
    /// that the day a real migration reached the same number.
    #[test]
    fn a_store_stamped_beyond_this_build_is_reported_ahead() {
        let build = Store::build_schema_version();
        let store = Store::open_in_memory().expect("open");
        stamp_version(&store, build + 1);
        stamp_version(&store, build + 2);

        let ahead = store
            .schema_ahead()
            .expect("schema_ahead")
            .expect("a store two migrations beyond this build must be reported");
        assert_eq!(ahead.build_version(), build);
        assert_eq!(
            ahead.store_version(),
            build + 2,
            "the store's version is the highest it records, not the first \
             unknown one"
        );
        assert_eq!(ahead.unknown_versions(), [build + 1, build + 2]);

        // The message has to be actionable, not merely correct: both versions by
        // number, and the fix (upgrade the binary — never "delete the store").
        let message = ahead.to_string();
        for expected in [
            &format!("{}", build + 2),
            &format!("{build}"),
            &"pgrade".to_owned(),
        ] {
            assert!(
                message.contains(expected.as_str()),
                "the refusal must name `{expected}`: {message}"
            );
        }
    }

    /// A **gap** — a store missing a lower migration while carrying a higher one
    /// this build knows — is not a store from the future, and must not be
    /// reported as one. Conflating them would name a version the store is not
    /// at, in a message telling the reader to upgrade a binary that is already
    /// new enough.
    ///
    /// This is the shape `main` itself carried (see
    /// `reopening_repairs_a_skipped_migration`, which proves `Store::open`
    /// repairs it), so it is a live case, not a hypothetical.
    ///
    /// The second half is the one that matters: a store that is *both* gapped
    /// and from the future must be described only by the half the reader can act
    /// on. Naming the gap would tell someone to upgrade a binary that is already
    /// new enough for it.
    #[test]
    fn a_gap_below_this_build_is_not_a_store_from_the_future() {
        let build = Store::build_schema_version();

        // A gap and nothing else: behind this build, so never ahead of it.
        let gapped = Store::open_in_memory().expect("open");
        gapped
            .conn
            .execute("DELETE FROM schema_migrations WHERE version = ?1", [build])
            .expect("open a gap");
        assert!(
            gapped.schema_version().expect("version") < build,
            "the gapped store must under-report, or this test proves nothing"
        );
        assert_eq!(
            gapped.schema_ahead().expect("schema_ahead"),
            None,
            "a store *missing* a migration is behind this build, never ahead"
        );

        // A gap *and* a version from the future: reported ahead, and reported
        // only in terms of the future version.
        stamp_version(&gapped, build + 1);
        let ahead = gapped
            .schema_ahead()
            .expect("schema_ahead")
            .expect("the unknown migration is still unknown, gap or no gap");
        assert_eq!(ahead.unknown_versions(), [build + 1]);
        assert_eq!(
            ahead.store_version(),
            build + 1,
            "the gap is a separate condition and must not colour this report"
        );
    }

    /// The reason the guard compares the recorded **set** and not
    /// [`Store::schema_version`].
    ///
    /// `schema_version` is the highest *gap-free* version — a floor for readers.
    /// A store recorded `1..=build, build + 2` therefore reports `build`, and a
    /// `schema_version() > build` test sees a perfectly ordinary store, while
    /// migration `build + 2`'s schema is in the file and a newer binary plainly
    /// assembled the graph. That is issue #342 surviving the check written to
    /// catch it, so it is asserted directly rather than trusted.
    #[test]
    fn the_gap_free_version_alone_would_miss_a_store_from_the_future() {
        let build = Store::build_schema_version();
        let store = Store::open_in_memory().expect("open");
        stamp_version(&store, build + 2); // note: `build + 1` deliberately absent

        assert_eq!(
            store.schema_version().expect("version"),
            build,
            "the gap-free version stops at the last contiguous migration, so it \
             cannot see the one above the hole"
        );
        let ahead = store
            .schema_ahead()
            .expect("schema_ahead")
            .expect("the set difference sees what the contiguity walk cannot");
        assert_eq!(ahead.store_version(), build + 2);
        assert_eq!(ahead.unknown_versions(), [build + 2]);
    }

    #[test]
    fn upsert_and_get_round_trips_all_fields() {
        let store = Store::open_in_memory().expect("open");
        let node = sample_node("sym:rust:src/lib.rs#sample");
        store.upsert_node(&node).expect("upsert");
        let got = store.get_node(&node.key).expect("get").expect("present");
        assert_eq!(got, node);
    }

    #[test]
    fn upsert_updates_in_place() {
        let store = Store::open_in_memory().expect("open");
        let mut node = sample_node("k");
        store.upsert_node(&node).expect("insert");
        node.name = "renamed".to_owned();
        node.kind = NodeKind::Struct;
        store.upsert_node(&node).expect("update");
        assert_eq!(store.node_count().expect("count"), 1);
        let got = store.get_node("k").expect("get").expect("present");
        assert_eq!(got.name, "renamed");
        assert_eq!(got.kind, NodeKind::Struct);
    }

    #[test]
    fn edge_with_unknown_endpoint_is_rejected() {
        let store = Store::open_in_memory().expect("open");
        store
            .upsert_node(&Node::new("a", NodeKind::Fn, "a"))
            .expect("a");
        let edge = Edge::derived("a", "missing", EdgeKind::Calls);
        let err = store.insert_edge(&edge).expect_err("should reject");
        assert!(matches!(err, super::StoreError::UnknownNode(k) if k == "missing"));
    }

    #[test]
    fn inferred_edge_requires_confidence() {
        let store = Store::open_in_memory().expect("open");
        store
            .upsert_node(&Node::new("a", NodeKind::Fn, "a"))
            .expect("a");
        store
            .upsert_node(&Node::new("b", NodeKind::Fn, "b"))
            .expect("b");
        // Hand-build an inferred edge with no confidence to violate the invariant.
        let bad = Edge {
            src: "a".to_owned(),
            dst: "b".to_owned(),
            kind: EdgeKind::References,
            provenance: Provenance::Inferred,
            confidence: None,
            src_ref: None,
        };
        assert!(matches!(
            store.insert_edge(&bad).expect_err("reject"),
            super::StoreError::InvalidEdge(_)
        ));
    }

    #[test]
    fn apply_factset_is_atomic() {
        let mut store = Store::open_in_memory().expect("open");
        // Second edge references a missing node, so the whole set must roll back.
        let facts = FactSet::new()
            .with_node(Node::new("a", NodeKind::Fn, "a"))
            .with_node(Node::new("b", NodeKind::Fn, "b"))
            .with_edge(Edge::derived("a", "b", EdgeKind::Calls))
            .with_edge(Edge::derived("a", "ghost", EdgeKind::Calls));
        assert!(store.apply_factset(&facts).is_err());
        assert_eq!(store.node_count().expect("count"), 0, "rolled back");
        assert_eq!(store.edge_count().expect("count"), 0, "rolled back");
    }

    #[test]
    fn neighbors_and_provenance_queries() {
        let mut store = Store::open_in_memory().expect("open");
        let facts = FactSet::new()
            .with_node(Node::new("a", NodeKind::Fn, "a"))
            .with_node(Node::new("b", NodeKind::Fn, "b"))
            .with_node(Node::new("c", NodeKind::Fn, "c"))
            .with_edge(Edge::derived("a", "b", EdgeKind::Calls))
            .with_edge(Edge::inferred("a", "c", EdgeKind::References, 0.5));
        store.apply_factset(&facts).expect("apply");

        let out = store.neighbors("a", Direction::Outgoing).expect("out");
        let mut keys: Vec<_> = out.iter().map(|n| n.key.clone()).collect();
        keys.sort();
        assert_eq!(keys, ["b", "c"]);

        assert!(
            store
                .neighbors("b", Direction::Outgoing)
                .expect("b out")
                .is_empty()
        );
        assert_eq!(
            store
                .neighbors("b", Direction::Incoming)
                .expect("b in")
                .len(),
            1
        );

        let inferred = store
            .edges_by_provenance(Provenance::Inferred)
            .expect("inf");
        assert_eq!(inferred.len(), 1);
        assert_eq!(inferred[0].confidence, Some(0.5));
    }

    #[test]
    fn neighbors_of_absent_node_is_empty() {
        let store = Store::open_in_memory().expect("open");
        assert!(
            store
                .neighbors("nope", Direction::Both)
                .expect("q")
                .is_empty()
        );
    }

    #[test]
    fn get_missing_node_is_none() {
        let store = Store::open_in_memory().expect("open");
        assert!(store.get_node("absent").expect("get").is_none());
    }

    #[test]
    fn nodes_by_kind_and_edges_to() {
        let mut store = Store::open_in_memory().expect("open");
        let facts = FactSet::new()
            .with_node(Node::new("f1", NodeKind::Fn, "f1"))
            .with_node(Node::new("f2", NodeKind::Fn, "f2"))
            .with_node(Node::new("s1", NodeKind::Struct, "s1"))
            .with_edge(Edge::derived("f1", "s1", EdgeKind::References))
            .with_edge(Edge::derived("f2", "s1", EdgeKind::References));
        store.apply_factset(&facts).expect("apply");

        let fns = store.nodes_by_kind(&NodeKind::Fn).expect("fns");
        assert_eq!(
            fns.iter().map(|n| n.key.as_str()).collect::<Vec<_>>(),
            ["f1", "f2"]
        );
        assert!(
            store
                .nodes_by_kind(&NodeKind::Enum)
                .expect("enums")
                .is_empty()
        );

        let into_s1 = store.edges_to("s1").expect("edges_to");
        assert_eq!(into_s1.len(), 2);
        assert!(into_s1.iter().all(|e| e.dst == "s1"));
    }

    #[test]
    fn open_persists_across_reopen() {
        let path =
            std::env::temp_dir().join(format!("roteiro-open-test-{}.db", std::process::id()));
        std::fs::remove_file(&path).ok();
        {
            let store = Store::open(&path).expect("open");
            store
                .upsert_node(&sample_node("persisted"))
                .expect("upsert");
        }
        {
            let store = Store::open(&path).expect("reopen");
            assert_eq!(store.node_count().expect("count"), 1);
            // The subject here is that a *reopen* is at the same schema as a
            // fresh open — not what number that happens to be. See
            // `open_in_memory_applies_schema` on why the literal went.
            assert_eq!(
                store.schema_version().expect("version"),
                crate::migrations::latest_version(),
                "reopening applies any migration added since the file was written"
            );
            assert!(store.get_node("persisted").expect("get").is_some());
        }
        std::fs::remove_file(&path).expect("cleanup");
    }

    fn graphify_layer(dst: &str) -> FactSet {
        FactSet::new()
            .with_node(Node::new("graphify:doc1", NodeKind::Doc, "Doc 1"))
            .with_edge({
                let mut e = Edge::inferred("graphify:doc1", dst, EdgeKind::References, 0.9);
                e.src_ref = Some("import:graphify".to_owned());
                e
            })
    }

    /// A **withdrawn** concept goes, and a node another layer also asserts stays.
    ///
    /// This is the half of idempotence that was missing. Re-importing could never
    /// duplicate — nodes upsert by key, edges are deleted by `src_ref` first — so
    /// a re-run always *looked* idempotent, while a node the source had since
    /// deleted survived for ever as an orphan carrying the layer's provenance.
    /// Nothing appeared twice, which is exactly why nobody would notice.
    ///
    /// Three nodes, and each one is a different rule:
    ///
    /// * `okf:acme/gone.md` — withdrawn, and only this layer's. It must go.
    /// * `okf:acme/kept.md` — still asserted. Untouched.
    /// * `extref:acme::shared` — withdrawn *by this layer*, and asserted by
    ///   another. It must survive, because two layers naming one node is a
    ///   supported arrangement rather than an anomaly: `import:links` and
    ///   `import:links/authored` share every placeholder between them.
    #[test]
    fn re_importing_a_smaller_bundle_removes_what_was_withdrawn_and_nothing_else() {
        let mut store = Store::open_in_memory().expect("open");

        let okf_node = |key: &str| {
            Node::new(key, NodeKind::Doc, key).with_provenance(Provenance::ExternalAuthored)
        };
        // Another producer, which also asserts the shared placeholder.
        let other = FactSet::new().with_node(okf_node("extref:acme::shared"));
        store
            .apply_import_layer("import:links", &other)
            .expect("other layer");

        let first = FactSet::new()
            .with_node(okf_node("okf:acme/gone.md"))
            .with_node(okf_node("okf:acme/kept.md"))
            .with_node(okf_node("extref:acme::shared"));
        let applied = store
            .apply_import_layer("import:okf/acme", &first)
            .expect("first import");
        assert_eq!(
            applied.nodes_removed, 0,
            "nothing is withdrawn on a first run"
        );
        assert_eq!(store.node_count().expect("count"), 3);

        // The peer deletes one concept and stops asserting the placeholder.
        let smaller = FactSet::new().with_node(okf_node("okf:acme/kept.md"));
        let applied = store
            .apply_import_layer("import:okf/acme", &smaller)
            .expect("re-import");

        // Which nodes, before how many. The count alone catches both directions
        // of failure but names only one of them: removing the shared placeholder
        // as well would fail a `nodes_removed == 1` assertion with a message
        // about a concept surviving, which is the opposite of what went wrong.
        assert!(
            store
                .get_node("extref:acme::shared")
                .expect("get")
                .is_some(),
            "a node another layer also asserts must not be removed by this one"
        );
        assert!(
            store.get_node("okf:acme/kept.md").expect("get").is_some(),
            "a concept still asserted must survive"
        );
        assert!(
            store.get_node("okf:acme/gone.md").expect("get").is_none(),
            "a concept the peer deleted must not survive a re-import"
        );
        assert_eq!(
            applied.nodes_removed, 1,
            "the withdrawn concept must be removed, and only it"
        );

        // And it is genuinely idempotent now: re-running the same smaller
        // bundle removes nothing further.
        let again = store
            .apply_import_layer("import:okf/acme", &smaller)
            .expect("third run");
        assert_eq!(again.nodes_removed, 0);
        assert_eq!(store.node_count().expect("count"), 2);
    }

    /// A withdrawn node that something is **still attached to** survives, in
    /// either direction.
    ///
    /// The layer stopped asserting the node, but another producer's edge still
    /// relates it to the graph. Removing it would take that relationship with it
    /// — a deletion nobody asked for, wearing a cleanup's clothes — and would
    /// violate the foreign key on the way out.
    ///
    /// Both ends are exercised on purpose: the probe is an `OR` over `src` and
    /// `dst`, so a fixture attaching the node only as a destination leaves the
    /// other arm free to be deleted with nothing going red.
    #[test]
    fn a_withdrawn_node_another_producer_still_points_at_is_kept() {
        let mut store = Store::open_in_memory().expect("open");
        let derived = FactSet::new().with_node(Node::new("file:a.rs", NodeKind::File, "a.rs"));
        store.rebuild(&derived, Some("tree1")).expect("rebuild");

        // Two withdrawn nodes, one for each **direction** an edge can attach.
        // The probe is an `OR` over `src` and `dst`, and a test that only ever
        // put the node on one end would leave the other arm unguarded: deleting
        // `src = …` from the query changed no test until this second node
        // existed, which is how the gap was found.
        let inbound = Node::new("okf:acme/pointed-at.md", NodeKind::Doc, "pointed at")
            .with_provenance(Provenance::ExternalDerived);
        let outbound = Node::new("okf:acme/points-from.md", NodeKind::Doc, "points from")
            .with_provenance(Provenance::ExternalDerived);
        let first = FactSet::new().with_node(inbound).with_node(outbound);
        store
            .apply_import_layer("import:okf/acme", &first)
            .expect("first");

        // A different layer relates the derived file to one imported concept,
        // and the other imported concept to the derived file.
        let linking = FactSet::new()
            .with_edge({
                let mut e = Edge::inferred(
                    "file:a.rs",
                    "okf:acme/pointed-at.md",
                    EdgeKind::References,
                    0.9,
                );
                e.src_ref = Some("import:links".to_owned());
                e
            })
            .with_edge({
                let mut e = Edge::inferred(
                    "okf:acme/points-from.md",
                    "file:a.rs",
                    EdgeKind::References,
                    0.9,
                );
                e.src_ref = Some("import:links".to_owned());
                e
            });
        store
            .apply_import_layer("import:links", &linking)
            .expect("link layer");

        let applied = store
            .apply_import_layer("import:okf/acme", &FactSet::new())
            .expect("withdraw everything");
        assert!(
            store
                .get_node("okf:acme/pointed-at.md")
                .expect("get")
                .is_some(),
            "a node another producer's edge points *at* must not be removed"
        );
        assert!(
            store
                .get_node("okf:acme/points-from.md")
                .expect("get")
                .is_some(),
            "nor one another producer's edge points *from*"
        );
        assert_eq!(applied.nodes_removed, 0);
    }

    /// Removal takes the node's cached context with it. A bundle assembled for a
    /// node that no longer exists can only ever be served as a stale answer, and
    /// `node_context` has no foreign key to clean it up for us.
    #[test]
    fn removing_a_withdrawn_node_drops_its_cached_context() {
        let mut store = Store::open_in_memory().expect("open");
        let first = FactSet::new().with_node(
            Node::new("okf:acme/gone.md", NodeKind::Doc, "gone")
                .with_provenance(Provenance::ExternalInferred),
        );
        store
            .apply_import_layer("import:okf/acme", &first)
            .expect("first");
        store
            .context_cache_put("okf:acme/gone.md", "fp1", "{}")
            .expect("cache");
        assert_eq!(
            store.context_cache_keys().expect("keys"),
            vec!["okf:acme/gone.md".to_owned()]
        );

        store
            .apply_import_layer("import:okf/acme", &FactSet::new())
            .expect("withdraw");
        assert!(
            store.context_cache_keys().expect("keys").is_empty(),
            "a cached bundle for a deleted node is a stale answer waiting to be served"
        );
    }

    /// A persisted import layer is re-applied after a `rebuild` wipes the graph,
    /// so imported facts survive a code-changing sync.
    #[test]
    fn imports_survive_rebuild() {
        let mut store = Store::open_in_memory().expect("open");
        let derived = FactSet::new().with_node(Node::new("file:a.rs", NodeKind::File, "a.rs"));
        store.rebuild(&derived, Some("tree1")).expect("rebuild");

        // apply_import_layer applies to the live graph and persists in one step.
        let applied = store
            .apply_import_layer("import:graphify", &graphify_layer("file:a.rs"))
            .expect("apply import");
        assert_eq!(applied.edges_applied, 1);
        assert_eq!(applied.edges_pruned, 0);
        assert_eq!(store.import_refs().expect("refs"), vec!["import:graphify"]);

        // Simulate a code-changing sync: the derived graph is rebuilt (still has
        // file:a.rs), which drops the imported doc + edge from the live graph.
        store.rebuild(&derived, Some("tree2")).expect("rebuild2");
        assert!(store.get_node("graphify:doc1").expect("get").is_none());

        // Re-applying imports restores them; nothing is pruned (target present).
        let applied = store.reapply_imports().expect("reapply");
        assert_eq!(applied.layers, 1);
        assert_eq!(applied.nodes, 1);
        assert_eq!(applied.edges_applied, 1);
        assert_eq!(applied.edges_pruned, 0);
        assert!(store.get_node("graphify:doc1").expect("get").is_some());
        assert_eq!(store.edges_from("graphify:doc1").expect("edges").len(), 1);
    }

    /// When a sync removes an edge's target (e.g. a deleted file), re-applying
    /// **prunes** that stale cross-reference from the persisted layer — it is not
    /// kept and retried forever. The import node itself is preserved.
    #[test]
    fn reapply_prunes_stale_cross_references() {
        let mut store = Store::open_in_memory().expect("open");
        let derived = FactSet::new().with_node(Node::new("file:gone.rs", NodeKind::File, "g"));
        store.rebuild(&derived, Some("t1")).expect("rebuild");
        store
            .apply_import_layer("import:graphify", &graphify_layer("file:gone.rs"))
            .expect("import");

        // Code-changing sync: file:gone.rs is deleted from the derived graph.
        store
            .rebuild(&FactSet::new(), Some("t2"))
            .expect("rebuild2");
        let applied = store.reapply_imports().expect("reapply");
        assert_eq!(applied.nodes, 1);
        assert_eq!(applied.edges_applied, 0);
        assert_eq!(
            applied.edges_pruned, 1,
            "the edge to the deleted file is pruned"
        );
        assert!(store.get_node("graphify:doc1").expect("get").is_some());

        // The prune is durable: a second reapply finds nothing left to prune,
        // proving the stale edge was removed from the persisted layer.
        let again = store.reapply_imports().expect("reapply2");
        assert_eq!(again.edges_applied, 0);
        assert_eq!(again.edges_pruned, 0, "already pruned; not retried");
    }

    /// `apply_import_layer` validates on import: a dangling edge in the incoming
    /// layer is dropped and never persisted.
    #[test]
    fn apply_import_layer_prunes_on_import() {
        let mut store = Store::open_in_memory().expect("open");
        let present = || FactSet::new().with_node(Node::new("file:a.rs", NodeKind::File, "a"));
        store.rebuild(&present(), Some("t")).expect("rebuild");

        let layer = graphify_layer("file:a.rs").with_edge({
            // Points at a file that does not exist → pruned on import.
            let mut e = Edge::inferred("graphify:doc1", "file:ghost.rs", EdgeKind::References, 0.9);
            e.src_ref = Some("import:graphify".to_owned());
            e
        });
        let applied = store
            .apply_import_layer("import:graphify", &layer)
            .expect("import");
        assert_eq!(applied.edges_applied, 1);
        assert_eq!(applied.edges_pruned, 1);

        // A rebuild + reapply confirms only the valid edge was persisted.
        store.rebuild(&present(), Some("t2")).expect("rebuild2");
        let re = store.reapply_imports().expect("reapply");
        assert_eq!(re.edges_applied, 1);
        assert_eq!(re.edges_pruned, 0, "ghost edge was not persisted");
    }

    #[test]
    fn legacy_import_layer_nodes_are_retagged_non_derived() {
        // A layer persisted before nodes carried provenance: its node objects have
        // no `provenance` field, so serde defaults them to Derived. On reapply the
        // store must repair them — a Graphify node to Inferred, a lat node to
        // Authored — so a later derived-only sync never mistakes them for derived.
        let mut store = Store::open_in_memory().expect("open");
        let legacy = r#"{"nodes":[
            {"key":"graphify:doc1","kind":"doc","name":"d","path":null,"lang":null,"blob_hash":null,"span":null,"meta":null},
            {"key":"lat:lat.md/a.md","kind":"doc","name":"a","path":null,"lang":null,"blob_hash":null,"span":null,"meta":null}
        ],"edges":[]}"#;
        store
            .conn
            .execute(
                "INSERT INTO imports (src_ref, facts) VALUES ('import:legacy', ?1)",
                [legacy],
            )
            .expect("seed legacy import row");

        store.reapply_imports().expect("reapply");

        let g = store
            .get_node("graphify:doc1")
            .expect("get")
            .expect("graphify node");
        assert_eq!(
            g.provenance,
            Provenance::Inferred,
            "graphify import node repaired to inferred"
        );
        let l = store
            .get_node("lat:lat.md/a.md")
            .expect("get")
            .expect("lat node");
        assert_eq!(
            l.provenance,
            Provenance::Authored,
            "lat import node repaired to authored"
        );
    }

    /// `apply_import_layer` replaces the layer for a ref; `delete_import` removes.
    #[test]
    fn apply_import_replaces_and_delete_removes() {
        let mut store = Store::open_in_memory().expect("open");
        let a = FactSet::new().with_node(Node::new("graphify:x", NodeKind::Doc, "x"));
        let b = FactSet::new().with_node(Node::new("graphify:y", NodeKind::Doc, "y"));
        store.apply_import_layer("import:graphify", &a).expect("a");
        store.apply_import_layer("import:graphify", &b).expect("b");
        assert_eq!(
            store.import_refs().expect("refs").len(),
            1,
            "same ref replaced"
        );
        assert!(store.delete_import("import:graphify").expect("del"));
        assert!(store.import_refs().expect("refs").is_empty());
        assert!(!store.delete_import("import:graphify").expect("del again"));
    }
}