scc_store/

lib.rs

//! SQLite persistence for the System Context Compiler.
//!
#![allow(clippy::type_complexity)]

//! Layer mapping (docs/DATA_STRATEGY.md):
//! - L0 repository snapshot: `repositories`, `snapshots`, `files`
//! - L1 evidence: `evidence`
//! - L2 reality graph: `entities`, `relationships`, `symbols`
//! - L3 system IR: `components`, `flows`, `invariants`, `tests`
//! - L4 context indexes: FTS5 (`symbols_fts`, `entities_fts`), `context_cache`
//! - L5 history: `intent_claims`, `drift_findings`

pub use rusqlite;

pub mod history;
pub mod snapshot;
// trace:exempt reason=module-facade  # pub mod declaration only; behavior traced per item in system.rs
pub mod system;

use rusqlite::{params, Connection, OptionalExtension};
use scc_core::{
    Entity, Evidence, Flow, Invariant, Provenance, Relationship, Repository, Severity, Snapshot,
};
use std::collections::{HashMap, HashSet};
use std::io::Read;
use std::path::{Path, PathBuf};
use thiserror::Error;

// trace:exempt reason=internal-detail
pub const SCHEMA_VERSION: u32 = 10;
// trace:exempt reason=internal-detail
pub const FTS_ESCAPE: &str = "\"";
// trace:exempt reason=internal-detail
const MIGRATIONS: &[&str] = &[
    MIGRATION_1,
    MIGRATION_2,
    MIGRATION_3,
    MIGRATION_4,
    MIGRATION_5,
    MIGRATION_6,
    MIGRATION_7,
    MIGRATION_8,
    MIGRATION_9,
    MIGRATION_10,
];

/// v4: model epoch. `context_cache.revision` becomes `epoch` — the cache is
/// keyed on the composite model state (source/semantic/evidence/intent/
/// runtime/derived generations), not the git revision alone, so any change
/// to system truth invalidates stale packs (docs/SYSTEM_DESIGN.md §5).
// trace:exempt reason=internal-detail
const MIGRATION_4: &str = r#"
ALTER TABLE context_cache RENAME COLUMN revision TO epoch;
"#;

/// v5: canonical causal flow graphs (Wave 3) — the behavioral truth from
/// which the `flows` projections are derived.
// trace:exempt reason=internal-detail
const MIGRATION_5: &str = r#"
CREATE TABLE IF NOT EXISTS flow_graphs (
  id TEXT PRIMARY KEY,
  kind TEXT NOT NULL,
  name TEXT NOT NULL,
  trigger TEXT,
  graph TEXT NOT NULL
);
"#;

/// v7: versioned reality graph (§XIX) — durable revision log plus full
/// member row sets per revision for introduction/removal history and
/// transactional historical views. Complements the model epoch (active
/// compiled view), it does not replace it.
// trace:exempt reason=internal-detail
const MIGRATION_7: &str = r#"
CREATE TABLE IF NOT EXISTS graph_revisions (
  rev INTEGER PRIMARY KEY,
  base_rev INTEGER NOT NULL DEFAULT 0,
  created_at TEXT NOT NULL,
  source_hash TEXT NOT NULL DEFAULT '',
  extractor_version TEXT NOT NULL DEFAULT '',
  entity_count INTEGER NOT NULL DEFAULT 0,
  rel_count INTEGER NOT NULL DEFAULT 0,
  file_count INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS revision_members (
  rev INTEGER NOT NULL,
  kind TEXT NOT NULL,
  id TEXT NOT NULL,
  row_json TEXT NOT NULL DEFAULT '{}',
  PRIMARY KEY (rev, kind, id)
);
CREATE INDEX IF NOT EXISTS idx_revision_members_rev ON revision_members(rev);
CREATE TABLE IF NOT EXISTS context_snapshots (
  id TEXT PRIMARY KEY,
  created_at TEXT NOT NULL,
  task TEXT NOT NULL DEFAULT '',
  epoch TEXT NOT NULL DEFAULT '',
  revision INTEGER NOT NULL DEFAULT 0,
  artifact TEXT NOT NULL DEFAULT '',
  entity_ids TEXT NOT NULL DEFAULT '[]',
  budget INTEGER NOT NULL DEFAULT 0,
  warnings TEXT NOT NULL DEFAULT '[]'
);
"#;

/// v8: versioned-graph V2 + snapshot V2 (§XIX/§XXII hardening). Revisions
/// gain the semantic-config and graph-content hashes so extractor, config,
/// confidence, or provenance changes produce a revision even when source
/// files did not move. Snapshots gain entity/relationship/contract/state
/// fingerprints plus the artifact hash so diffs detect modification, not
/// just presence.
/// Version-gated (runs once per database), so plain ADD COLUMN is safe.
// trace:exempt reason=internal-detail
const MIGRATION_8: &str = r#"
ALTER TABLE graph_revisions ADD COLUMN semantic_config_hash TEXT NOT NULL DEFAULT '';
ALTER TABLE graph_revisions ADD COLUMN graph_content_hash TEXT NOT NULL DEFAULT '';
ALTER TABLE context_snapshots ADD COLUMN entity_fp TEXT NOT NULL DEFAULT '{}';
ALTER TABLE context_snapshots ADD COLUMN rel_fp TEXT NOT NULL DEFAULT '{}';
ALTER TABLE context_snapshots ADD COLUMN contract_fp TEXT NOT NULL DEFAULT '{}';
ALTER TABLE context_snapshots ADD COLUMN state_fp TEXT NOT NULL DEFAULT '{}';
ALTER TABLE context_snapshots ADD COLUMN flow_names TEXT NOT NULL DEFAULT '[]';
ALTER TABLE context_snapshots ADD COLUMN artifact_hash TEXT NOT NULL DEFAULT '';
"#;

/// v9: namespaced plugin state (§23) — key/value pairs scoped to
/// `(plugin_id, key)`. Plugins never create their own tables; the host
/// owns storage, the plugin id owns the namespace.
// trace:exempt reason=internal-detail
const MIGRATION_9: &str = r#"
CREATE TABLE IF NOT EXISTS plugin_state (
  plugin_id TEXT NOT NULL,
  key TEXT NOT NULL,
  value TEXT NOT NULL DEFAULT '{}',
  updated_at TEXT NOT NULL DEFAULT '',
  PRIMARY KEY (plugin_id, key)
);
CREATE INDEX IF NOT EXISTS idx_plugin_state_plugin ON plugin_state(plugin_id);
"#;

/// v10: namespaced sidecar facts (§124 item 35) — raw analyzer facts
/// (CPG nodes, findings, slices) scoped to `(plugin_id, graph, key)`.
/// Queryable + versioned + explicitly namespaced; NOT authoritative by
/// default. Promotion into the canonical graph goes through
/// `plugins.promote` (normal contribution path), never by reading here.
// trace:exempt reason=internal-detail
const MIGRATION_10: &str = r#"
CREATE TABLE IF NOT EXISTS sidecar_facts (
  plugin_id TEXT NOT NULL,
  graph TEXT NOT NULL DEFAULT 'default',
  key TEXT NOT NULL,
  value TEXT NOT NULL DEFAULT '{}',
  updated_at TEXT NOT NULL DEFAULT '',
  PRIMARY KEY (plugin_id, graph, key)
);
CREATE INDEX IF NOT EXISTS idx_sidecar_plugin_graph ON sidecar_facts(plugin_id, graph);
"#;

/// v6: observed trace-path signatures (Wave 6) — canonical root-to-leaf
/// service paths per trace, aggregated across ingests. `count` is additive
/// per trace occurrence, `latency_ms` a count-weighted running average,
/// `errors` additive.
// trace:exempt reason=internal-detail
const MIGRATION_6: &str = r#"
CREATE TABLE IF NOT EXISTS trace_signatures (
  signature TEXT PRIMARY KEY,
  count INTEGER NOT NULL DEFAULT 1,
  latency_ms REAL NOT NULL DEFAULT 0,
  errors INTEGER NOT NULL DEFAULT 0,
  last_observed TEXT NOT NULL
);
"#;

/// v3: entity embeddings (f32 vector blobs) for the optional semantic ranker.
// trace:exempt reason=internal-detail
const MIGRATION_3: &str = r#"
CREATE TABLE IF NOT EXISTS embeddings (
  entity_id TEXT PRIMARY KEY,
  vector BLOB NOT NULL,
  model TEXT NOT NULL,
  updated_at TEXT NOT NULL
);
"#;

/// v2: runtime edge aggregation columns (latency/error aggregates).
// trace:exempt reason=internal-detail
const MIGRATION_2: &str = r#"
ALTER TABLE runtime_edges ADD COLUMN latency_ms REAL NOT NULL DEFAULT 0;
ALTER TABLE runtime_edges ADD COLUMN errors INTEGER NOT NULL DEFAULT 0;
"#;

// trace:exempt reason=internal-detail
const MIGRATION_1: &str = r#"
CREATE TABLE IF NOT EXISTS meta (
  key TEXT PRIMARY KEY,
  value TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS repositories (
  id TEXT PRIMARY KEY,
  name TEXT NOT NULL,
  url TEXT,
  root TEXT NOT NULL,
  indexed_at TEXT
);

CREATE TABLE IF NOT EXISTS snapshots (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  revision TEXT NOT NULL,
  branch TEXT,
  indexed_at TEXT NOT NULL,
  status TEXT NOT NULL DEFAULT 'active',
  file_count INTEGER NOT NULL DEFAULT 0
);

CREATE TABLE IF NOT EXISTS files (
  path TEXT PRIMARY KEY,
  hash TEXT NOT NULL,
  language TEXT NOT NULL DEFAULT 'unknown',
  kind TEXT NOT NULL DEFAULT 'other',
  size INTEGER NOT NULL DEFAULT 0,
  indexed_at TEXT
);

CREATE TABLE IF NOT EXISTS symbols (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  file TEXT NOT NULL,
  name TEXT NOT NULL,
  symbol_kind TEXT NOT NULL,
  signature TEXT,
  start_line INTEGER NOT NULL,
  end_line INTEGER NOT NULL,
  exported INTEGER NOT NULL DEFAULT 0,
  docstring TEXT
);
CREATE INDEX IF NOT EXISTS idx_symbols_file ON symbols(file);
CREATE INDEX IF NOT EXISTS idx_symbols_name ON symbols(name);

CREATE TABLE IF NOT EXISTS imports (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  file TEXT NOT NULL,
  module TEXT NOT NULL,
  names TEXT NOT NULL DEFAULT '[]',
  line INTEGER NOT NULL DEFAULT 0,
  type TEXT NOT NULL DEFAULT 'member'
);
CREATE INDEX IF NOT EXISTS idx_imports_file ON imports(file);

CREATE TABLE IF NOT EXISTS entities (
  id TEXT PRIMARY KEY,
  kind TEXT NOT NULL,
  name TEXT NOT NULL,
  attributes TEXT NOT NULL DEFAULT '{}',
  evidence TEXT NOT NULL DEFAULT '[]',
  sources TEXT NOT NULL DEFAULT '[]'
);
CREATE INDEX IF NOT EXISTS idx_entities_kind ON entities(kind);

CREATE TABLE IF NOT EXISTS relationships (
  id TEXT PRIMARY KEY,
  subject TEXT NOT NULL,
  predicate TEXT NOT NULL,
  object TEXT NOT NULL,
  provenance TEXT NOT NULL,
  confidence REAL NOT NULL,
  evidence TEXT NOT NULL DEFAULT '[]',
  verified_at TEXT NOT NULL DEFAULT '',
  source_path TEXT NOT NULL DEFAULT ''
);
CREATE INDEX IF NOT EXISTS idx_rel_subject ON relationships(subject);
CREATE INDEX IF NOT EXISTS idx_rel_object ON relationships(object);
CREATE INDEX IF NOT EXISTS idx_rel_predicate ON relationships(predicate);
CREATE INDEX IF NOT EXISTS idx_rel_source ON relationships(source_path);

CREATE TABLE IF NOT EXISTS evidence (
  id TEXT PRIMARY KEY,
  type TEXT NOT NULL,
  path TEXT,
  symbol TEXT,
  start_line INTEGER,
  end_line INTEGER,
  revision TEXT,
  content_hash TEXT,
  extractor TEXT,
  extractor_version TEXT
);
CREATE INDEX IF NOT EXISTS idx_evidence_path ON evidence(path);

CREATE TABLE IF NOT EXISTS components (
  id TEXT PRIMARY KEY,
  name TEXT NOT NULL,
  kind TEXT NOT NULL,
  responsibility TEXT NOT NULL DEFAULT '[]',
  implementation TEXT NOT NULL DEFAULT '[]',
  evidence TEXT NOT NULL DEFAULT '[]',
  attributes TEXT NOT NULL DEFAULT '{}'
);

CREATE TABLE IF NOT EXISTS flows (
  id TEXT PRIMARY KEY,
  kind TEXT NOT NULL,
  name TEXT NOT NULL,
  trigger TEXT,
  steps TEXT NOT NULL DEFAULT '[]',
  attributes TEXT NOT NULL DEFAULT '{}'
);

CREATE TABLE IF NOT EXISTS invariants (
  id TEXT PRIMARY KEY,
  statement TEXT NOT NULL,
  severity TEXT NOT NULL,
  scope TEXT NOT NULL DEFAULT '[]',
  enforced_by TEXT NOT NULL DEFAULT '[]',
  provenance TEXT NOT NULL DEFAULT 'DECLARED',
  evidence TEXT NOT NULL DEFAULT '[]'
);

CREATE TABLE IF NOT EXISTS tests (
  id TEXT PRIMARY KEY,
  name TEXT NOT NULL,
  file TEXT NOT NULL,
  kind TEXT NOT NULL DEFAULT 'unit',
  symbol TEXT
);
CREATE INDEX IF NOT EXISTS idx_tests_file ON tests(file);

CREATE TABLE IF NOT EXISTS context_cache (
  key TEXT PRIMARY KEY,
  pack TEXT NOT NULL,
  revision TEXT NOT NULL,
  created_at TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS intent_claims (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  source TEXT NOT NULL,
  claim TEXT NOT NULL,
  created_at TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS runtime_edges (
  source TEXT NOT NULL,
  target TEXT NOT NULL,
  count INTEGER NOT NULL DEFAULT 1,
  last_observed TEXT NOT NULL,
  PRIMARY KEY (source, target)
);

CREATE TABLE IF NOT EXISTS drift_findings (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  kind TEXT NOT NULL,
  severity TEXT NOT NULL,
  message TEXT NOT NULL,
  created_at TEXT NOT NULL,
  resolved INTEGER NOT NULL DEFAULT 0
);

CREATE VIRTUAL TABLE IF NOT EXISTS symbols_fts USING fts5(
  name, signature, symbol_kind UNINDEXED, file UNINDEXED
);

CREATE VIRTUAL TABLE IF NOT EXISTS entities_fts USING fts5(
  id UNINDEXED, kind UNINDEXED, name, attributes
);
"#;

#[derive(Debug, Error)]
// trace:exempt reason=internal-detail
pub enum StoreError {
    #[error("sqlite: {0}")]
    Sqlite(#[from] rusqlite::Error),
    #[error("json: {0}")]
    Json(#[from] serde_json::Error),
    #[error("repository not initialized: {0}")]
    NotInitialized(String),
    #[error("index cache is corrupt (refusing to fabricate an empty index): {0}")]
    Corrupt(String),
}

// trace:exempt reason=internal-detail
pub type Result<T> = std::result::Result<T, StoreError>;

#[derive(Debug, Clone, serde::Serialize)]
// trace:exempt reason=internal-detail
pub struct RuntimeEdgeRow {
    pub source: String,
    pub target: String,
    pub count: u64,
    pub latency_ms: f64,
    pub errors: u64,
    pub last_observed: String,
}

// trace:exempt reason=internal-detail

/// The independent truth sources whose generations compose the model epoch.
/// Any change to one of them invalidates previously cached context packs.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
// trace:exempt reason=internal-detail
pub enum ModelEpochKind {
    /// Indexed file contents (snapshot completion).
    Source,
    /// Semantic resolver promotions (LSP/SCIP edge upgrades).
    Semantic,
    /// Imported external evidence (SCIP/CCG/GitNexus/Beads/CBM/Hindsight).
    Evidence,
    /// Declared intent (`.scc/intent.yaml` claims).
    Intent,
    /// Runtime trace ingestion.
    Runtime,
    /// Derived compilation (components/flows/invariants/drift/boundaries).
    Derived,
}

// trace:exempt reason=internal-detail
impl ModelEpochKind {
    // trace:exempt reason=internal-detail
    pub fn meta_key(&self) -> &'static str {
        match self {
            ModelEpochKind::Source => "source_generation",
            ModelEpochKind::Semantic => "semantic_generation",
            ModelEpochKind::Evidence => "evidence_generation",
            ModelEpochKind::Intent => "intent_generation",
            ModelEpochKind::Runtime => "runtime_generation",
            ModelEpochKind::Derived => "derived_generation",
        }
    }
}

// trace:exempt reason=internal-detail

/// Deterministic composite fingerprint of the model state. `composite()` is
/// the canonical cache-epoch string: it changes whenever any source of
/// system truth changes, so a previously fresh context pack can never be
/// served after its evidence is stale.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
// trace:exempt reason=internal-detail
pub struct ModelEpoch {
    pub source: u64,
    pub semantic: u64,
    pub evidence: u64,
    pub intent: u64,
    pub runtime: u64,
    pub derived: u64,
}

// trace:exempt reason=internal-detail
impl ModelEpoch {
    // trace:exempt reason=internal-detail
    pub fn zero() -> ModelEpoch {
        ModelEpoch {
            source: 0,
            semantic: 0,
            evidence: 0,
            intent: 0,
            runtime: 0,
            derived: 0,
        }
    }

    /// Composite hash over every generation. Prefix identifies the scheme so
    /// a future epoch-shape change cannot collide with old keys.
    // trace:exempt reason=internal-detail
    pub fn composite(&self, revision: &str) -> String {
        let mut h = blake3::Hasher::new();
        h.update(b"scc-model-epoch-v1");
        h.update(self.source.to_le_bytes().as_slice());
        h.update(self.semantic.to_le_bytes().as_slice());
        h.update(self.evidence.to_le_bytes().as_slice());
        h.update(self.intent.to_le_bytes().as_slice());
        h.update(self.runtime.to_le_bytes().as_slice());
        h.update(self.derived.to_le_bytes().as_slice());
        h.update(revision.as_bytes());
        format!("epoch:{}", &h.finalize().to_hex()[..24])
    }
}
// trace:v1 id=impl.scc.store work=WORK-SCC-001 satisfies=REQ-SCC-DATA implements=PLAN-SCC-001

#[derive(Debug)]
// trace:exempt reason=internal-detail
pub struct Store {
    pub conn: Connection,
    pub root: PathBuf,
    /// Repository id (repo:// id component).
    pub repo_id: String,
    pub repo_name: String,
    batch_depth: std::sync::atomic::AtomicUsize,
}

const SQLITE_MAGIC: &[u8] = b"SQLite format 3\0";

/// Refuse a non-empty existing file that is not a SQLite database. Empty
/// files are a fresh index. Truncation after the header is caught by
/// `probe_existing_schema` after open (not a full `PRAGMA quick_check`).
// trace:v1 id=impl.scc.store.refuse-corrupt work=WORK-phase-12-of-scc-x-ripwire-lessons-java-unprefixed-field-as-receiver-typ satisfies=REQ-implement-phase-12-of-scc-x-ripwire-lessons-java-unprefixed-field-as,REQ-implement-fix-pr-review-comments-without-collapsing-scc-type-script-no implements=PLAN-phase-12-of-scc-x-ripwire-lessons-java-unprefixed-field-as-receiver-typ
fn refuse_corrupt_existing_db(path: &Path) -> Result<()> {
    let Ok(meta) = std::fs::metadata(path) else {
        return Ok(());
    };
    if !meta.is_file() || meta.len() == 0 {
        return Ok(());
    }
    if meta.len() < 100 {
        return Err(StoreError::Corrupt("truncated sqlite header".into()));
    }
    let mut f = std::fs::File::open(path).map_err(|e| StoreError::Corrupt(e.to_string()))?;
    let mut magic = [0u8; 16];
    let n = f.read(&mut magic).map_err(|e| StoreError::Corrupt(e.to_string()))?;
    if n < SQLITE_MAGIC.len() || magic.as_slice() != SQLITE_MAGIC {
        return Err(StoreError::Corrupt(
            "file is not a SQLite database".into(),
        ));
    }
    Ok(())
}

/// Cheap existing-DB probe: the SCC `entities` table must already exist.
/// Missing schema on a nonempty file is corrupt — do not migrate it into
/// an empty index. Not a full integrity walk.
// trace:exempt reason=internal-detail
fn probe_existing_schema(conn: &Connection) -> Result<()> {
    match conn.query_row(
        "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'entities' LIMIT 1",
        [],
        |_| Ok(()),
    ) {
        Ok(()) => Ok(()),
        Err(rusqlite::Error::QueryReturnedNoRows) => Err(StoreError::Corrupt(
            "existing database is missing SCC schema".into(),
        )),
        Err(e) => Err(StoreError::Corrupt(e.to_string())),
    }
}

// trace:exempt reason=internal-detail
struct NestGuard<'c> {
    conn: &'c Connection,
    committed: bool,
}

// trace:exempt reason=internal-detail
impl NestGuard<'_> {
    // trace:exempt reason=internal-detail
    fn commit(mut self) -> Result<()> {
        self.committed = true;
        self.conn.execute_batch("RELEASE scc_nest")?;
        Ok(())
    }
}

// trace:exempt reason=internal-detail
impl Drop for NestGuard<'_> {
    // trace:exempt reason=internal-detail
    fn drop(&mut self) {
        if !self.committed {
            let _ = self
                .conn
                .execute_batch("ROLLBACK TO scc_nest; RELEASE scc_nest;");
        }
    }
}

// trace:exempt reason=internal-detail
impl Store {
    /// Open (creating if needed) the SCC database at `path` for repository
    /// rooted at `root`. `root` must exist. A truncated or garbage existing
    /// file is refused rather than migrated into a fake empty index.
    // trace:exempt reason=internal-detail
    /// Open, quarantining a corrupt database aside and starting fresh.
    /// ONLY for the index write path: a malformed file is renamed to
    /// `<name>.corrupt-<epoch>` (evidence preserved, WAL sidecars moved
    /// with it) and indexing rebuilds from the working tree. Readers keep
    /// using [`Store::open`] and fail loudly — a freshly rebuilt index is
    /// temporarily incomplete and must never be silently presented as
    /// truth on a read path. Returns the quarantine path when recovery ran.
    // trace:v1 id=impl.scc.store.open-recovering work=WORK-SI-MMMJA4G6 satisfies=REQ-SI-503JSBGP
    pub fn open_recovering(path: &Path, root: &Path) -> Result<(Store, Option<PathBuf>)> {
        match Self::open(path, root) {
            Ok(store) => Ok((store, None)),
            Err(e) if Self::is_corruption(&e) => {
                let quarantine = Self::quarantine_db(path)?;
                let store = Self::open(path, root)?;
                Ok((store, Some(quarantine)))
            }
            Err(e) => Err(e),
        }
    }

    /// Move a corrupt database (plus WAL sidecars) aside as
    /// `<name>.db.corrupt-<epoch>`. Evidence preserved; the path is free
    /// for a fresh rebuild.
    // trace:exempt reason=internal-detail
    pub fn quarantine_db(path: &Path) -> Result<PathBuf> {
        let stamp = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_secs())
            .unwrap_or(0);
        let quarantine = path.with_extension(format!("db.corrupt-{stamp}"));
        let _ = std::fs::rename(path, &quarantine);
        for suffix in ["-wal", "-shm", "-journal"] {
            let sidecar = PathBuf::from(format!("{}{suffix}", path.display()));
            if sidecar.is_file() {
                let dest = PathBuf::from(format!("{}{suffix}", quarantine.display()));
                let _ = std::fs::rename(&sidecar, &dest);
            }
        }
        Ok(quarantine)
    }

    /// True for corruption (as opposed to e.g. missing files or permission
    /// errors): the explicit Corrupt refusal plus SQLite's own malformed /
    /// not-a-database errors, which surface lazily on first query despite a
    /// valid-looking header.
    // trace:exempt reason=internal-detail
    pub fn is_corruption(e: &StoreError) -> bool {
        match e {
            StoreError::Corrupt(_) => true,
            StoreError::Sqlite(inner) => {
                let m = inner.to_string();
                m.contains("malformed")
                    || m.contains("not a database")
                    || m.contains("database disk image")
            }
            _ => false,
        }
    }

    /// Begin a write batch: the outermost begin issues BEGIN IMMEDIATE,
    /// nested begins are depth-counted no-ops. Pair with [`Store::batch_end`]
    /// (commit at depth zero) or [`Store::batch_abort`] (full rollback).
    /// Lets one commit cover a whole file's facts instead of one fsync per
    /// row — the dominant index-time cost on fsync-bound disks.
    // trace:exempt reason=internal-detail
    pub fn batch_begin(&self) -> Result<()> {
        use std::sync::atomic::Ordering;
        if self.batch_depth.fetch_add(1, Ordering::SeqCst) == 0 {
            self.conn.execute_batch("BEGIN IMMEDIATE")?;
        }
        Ok(())
    }

    // trace:exempt reason=internal-detail
    pub fn batch_end(&self) -> Result<()> {
        use std::sync::atomic::Ordering;
        if self.batch_depth.fetch_sub(1, Ordering::SeqCst) == 1 {
            self.conn.execute_batch("COMMIT")?;
        }
        Ok(())
    }

    // trace:exempt reason=internal-detail
    pub fn batch_abort(&self) {
        use std::sync::atomic::Ordering;
        if self.batch_depth.swap(0, Ordering::SeqCst) > 0 {
            let _ = self.conn.execute_batch("ROLLBACK");
        }
    }

    /// Run `f` inside a batch: commit on success, full rollback on error.
    /// Per-file callers use this so a failed file leaves no partial facts.
    // trace:v1 id=impl.scc.store.batch-writes work=WORK-SI-MMMJA4G6 satisfies=REQ-SI-NX53P4B7
    pub fn batch_write<T, E>(&self, f: impl FnOnce() -> std::result::Result<T, E>) -> std::result::Result<T, E>
    where
        E: From<StoreError>,
    {
        self.batch_begin()?;
        match f() {
            Ok(v) => self.batch_end().map_err(E::from).map(|_| v),
            Err(e) => {
                self.batch_abort();
                Err(e)
            }
        }
    }

    /// Nest-safe unit of work inside a batch: a savepoint when a batch is
    /// open (plain statements join the batch directly), else a standalone
    /// transaction exactly as before. Replaces every `unchecked_transaction`
    /// so batched and unbatched callers share the code path.
    // trace:exempt reason=internal-detail
    fn nest(&self) -> Result<NestGuard<'_>> {
        self.conn.execute_batch("SAVEPOINT scc_nest")?;
        Ok(NestGuard { conn: &self.conn, committed: false })
    }

        // trace:v1 id=impl.scc.store.stable-identity work=WORK-SI-MMMJA4G6 satisfies=REQ-SI-503JSBGP
    pub fn open(path: &Path, root: &Path) -> Result<Store> {
        let existed_nonempty = path.is_file()
            && std::fs::metadata(path).map(|m| m.len() > 0).unwrap_or(false);
        refuse_corrupt_existing_db(path)?;
        let conn = match Connection::open(path) {
            Ok(c) => c,
            Err(e) if existed_nonempty => {
                return Err(StoreError::Corrupt(e.to_string()));
            }
            Err(e) => return Err(e.into()),
        };
        if existed_nonempty {
            probe_existing_schema(&conn)?;
        }
        conn.pragma_update(None, "journal_mode", "WAL")?;
        // 30s lock wait (issue: `scc index` failed instantly with
        // "database is locked" while a watch/daemon writer held the DB;
        // the old 5s timeout lost to multi-second index writes and the
        // bare sqlite message told the user nothing retryable). Locks
        // are transient contention, NOT corruption — never quarantine.
        conn.pragma_update(None, "busy_timeout", 30000)?;
        // FULL durability: the store is the product (agent context reads it
        // directly), not a rebuildable cache — an OS crash or power loss
        // must never leave a half-written index behind. Speed comes from
        // batching writes into one transaction per phase (see `batch_*`),
        // not from weakening the fsync contract.
        conn.pragma_update(None, "synchronous", "FULL")?;
        conn.pragma_update(None, "foreign_keys", "ON")?;
        apply_migrations(&conn)?;

        let root = root.canonicalize().unwrap_or_else(|_| root.to_path_buf());
        let name = root
            .file_name()
            .map(|s| s.to_string_lossy().to_string())
            .unwrap_or_else(|| "repository".to_string());
        // Stable repository identity (§XIV): explicit operator id wins,
        // then the persistent id already stored in this database (checkout
        // moves keep their semantic identity), then basename fallback.
        let explicit = std::env::var("SCC_REPO_ID")
            .ok()
            .filter(|s| !s.trim().is_empty());
        let mut repo_id = scc_core::sanitize_key(&name);
        let mut id_source = "basename";
        if let Some(e) = explicit {
            repo_id = scc_core::sanitize_key(e.trim());
            id_source = "explicit";
        } else if existed_nonempty {
            if let Some(persisted) = Self::existing_repo_id(&conn)? {
                repo_id = persisted;
                id_source = "persistent";
            }
        }

        let mut store = Store {
            conn,
            root,
            repo_id,
            repo_name: name,
            batch_depth: std::sync::atomic::AtomicUsize::new(0),
        };
        store.ensure_repository()?;
        if id_source == "persistent" {
            store.refresh_repository_root()?;
        }
        store.meta_set("repo_id_source", id_source)?;
        Ok(store)
    }

    // trace:exempt reason=internal-detail
    fn ensure_repository(&mut self) -> Result<()> {
        let existing: Option<String> = self
            .conn
            .query_row(
                "SELECT id FROM repositories WHERE id = ?1",
                params![self.repo_id],
                |r| r.get(0),
            )
            .optional()?;
        if existing.is_none() {
            self.conn.execute(
                "INSERT INTO repositories (id, name, root, indexed_at) VALUES (?1, ?2, ?3, ?4)",
                params![
                    self.repo_id,
                    self.repo_name,
                    self.root.to_string_lossy(),
                    scc_core::now_rfc3339()
                ],
            )?;
        }
        Ok(())
    }
    /// The repository id already persisted in this database, if any.
    /// A moved checkout keeps the id its entities were written under.
    // trace:exempt reason=internal-detail
    fn existing_repo_id(conn: &Connection) -> Result<Option<String>> {
        let id: Option<String> = conn
            .query_row(
                "SELECT id FROM repositories ORDER BY rowid LIMIT 1",
                [],
                |r| r.get(0),
            )
            .optional()?;
        Ok(id)
    }

    /// Point the persisted repository row at the current checkout path
    /// after a move. Identity (id) is untouched; only the provenance
    /// path (root) follows the checkout.
    // trace:exempt reason=internal-detail
    fn refresh_repository_root(&self) -> Result<()> {
        self.conn.execute(
            "UPDATE repositories SET root = ?1, name = ?2 WHERE id = ?3",
            params![
                self.root.to_string_lossy(),
                self.repo_name,
                self.repo_id
            ],
        )?;
        Ok(())
    }

    /// Adopt a stable identity derived from explicit config or the git
    /// remote. Only applies to a fresh database (no entities yet): once
    /// entity ids are written under an id, that id is load-bearing and
    /// must survive via the persistent rule in [`Store::open`] instead.
    /// Returns true when the identity changed.
    // trace:v1 id=impl.scc.store.adopt-stable-identity work=WORK-SI-MMMJA4G6 satisfies=REQ-SI-503JSBGP
    pub fn adopt_stable_identity(
        &mut self,
        explicit: Option<&str>,
        remote_url: Option<&str>,
    ) -> Result<bool> {
        if self.meta_get("repo_id_source")?.as_deref() == Some("explicit") {
            return Ok(false);
        }
        let n: u64 = self.conn.query_row(
            "SELECT COUNT(*) FROM entities",
            [],
            |r| r.get(0),
        )?;
        if n > 0 {
            return Ok(false);
        }
        let basename = self
            .root
            .file_name()
            .map(|s| s.to_string_lossy().to_string())
            .unwrap_or_else(|| "repository".to_string());
        let id = scc_core::identity::stable_repo_id(explicit, remote_url, &basename);
        if id == self.repo_id {
            return Ok(false);
        }
        self.conn.execute(
            "DELETE FROM repositories WHERE id = ?1",
            params![self.repo_id],
        )?;
        self.repo_id = id;
        self.ensure_repository()?;
        self.meta_set("repo_id_source", "remote")?;
        Ok(true)
    }

    // trace:exempt reason=internal-detail
    pub fn repository(&self) -> Repository {
        Repository {
            id: self.repo_id.clone(),
            name: self.repo_name.clone(),
            url: self
                .meta_get("remote_url")
                .ok()
                .flatten()
                .filter(|s| !s.is_empty()),
        }
    }

    // ------------------------------------------------------------------
    // plugin state (§23): namespaced key/value storage. The host owns the
    // table; the plugin id owns the namespace. Keys are opaque strings;
    // values are JSON text (small config/cursor blobs, not artifacts).
    // ------------------------------------------------------------------

    // trace:exempt reason=internal-detail
    pub fn plugin_state_put(&self, plugin_id: &str, key: &str, value: &str) -> Result<()> {
        self.conn.execute(
            "INSERT INTO plugin_state (plugin_id, key, value, updated_at) VALUES (?1, ?2, ?3, ?4)
             ON CONFLICT(plugin_id, key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at",
            params![plugin_id, key, value, scc_core::now_rfc3339()],
        )?;
        Ok(())
    }

    // trace:exempt reason=internal-detail
    pub fn plugin_state_get(&self, plugin_id: &str, key: &str) -> Result<Option<String>> {
        let v = self
            .conn
            .query_row(
                "SELECT value FROM plugin_state WHERE plugin_id = ?1 AND key = ?2",
                params![plugin_id, key],
                |r| r.get(0),
            )
            .optional()?;
        Ok(v)
    }

    // ------------------------------------------------------------------
    // sidecar facts (§124 item 35): namespaced raw analyzer storage.
    // (plugin_id, graph, key) -> JSON value. Same host-owns-table pattern
    // as plugin_state; graphs are plugin-declared namespaces (e.g.
    // `joern-cpg`). Never read by ranking/context — promotion only.
    // ------------------------------------------------------------------

    // trace:exempt reason=internal-detail
    pub fn sidecar_put(&self, plugin_id: &str, graph: &str, key: &str, value: &str) -> Result<()> {
        self.conn.execute(
            "INSERT INTO sidecar_facts (plugin_id, graph, key, value, updated_at) VALUES (?1, ?2, ?3, ?4, ?5)
             ON CONFLICT(plugin_id, graph, key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at",
            params![plugin_id, graph, key, value, scc_core::now_rfc3339()],
        )?;
        Ok(())
    }

    // trace:exempt reason=internal-detail
    pub fn sidecar_get(&self, plugin_id: &str, graph: &str, key: &str) -> Result<Option<String>> {
        let v = self
            .conn
            .query_row(
                "SELECT value FROM sidecar_facts WHERE plugin_id = ?1 AND graph = ?2 AND key = ?3",
                params![plugin_id, graph, key],
                |r| r.get(0),
            )
            .optional()?;
        Ok(v)
    }

    // trace:exempt reason=internal-detail
    pub fn sidecar_scan(&self, plugin_id: &str, graph: &str, prefix: &str, limit: usize) -> Result<Vec<(String, String)>> {
        let esc = prefix.replace('\\', "\\\\").replace('%', "\\%").replace('_', "\\_");
        let like = format!("{esc}%");
        let mut stmt = self.conn.prepare(
            "SELECT key, value FROM sidecar_facts WHERE plugin_id = ?1 AND graph = ?2 AND key LIKE ?3 ESCAPE '\\' ORDER BY key LIMIT ?4",
        )?;
        let rows = stmt.query_map(params![plugin_id, graph, like, limit.max(1) as i64], |r| {
            Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?))
        })?;
        rows.collect::<std::result::Result<Vec<_>, _>>().map_err(|e| e.into())
    }

    // trace:exempt reason=internal-detail
    pub fn plugin_state_delete(&self, plugin_id: &str, key: &str) -> Result<()> {
        self.conn.execute(
            "DELETE FROM plugin_state WHERE plugin_id = ?1 AND key = ?2",
            params![plugin_id, key],
        )?;
        Ok(())
    }

    // trace:exempt reason=internal-detail
    pub fn plugin_state_scan(&self, plugin_id: &str, prefix: &str, limit: usize) -> Result<Vec<(String, String)>> {
        let esc = prefix.replace('\\', "\\\\").replace('%', "\\%").replace('_', "\\_");
        let like = format!("{esc}%");
        let mut stmt = self.conn.prepare(
            "SELECT key, value FROM plugin_state WHERE plugin_id = ?1 AND key LIKE ?2 ESCAPE '\\' ORDER BY key LIMIT ?3",
        )?;
        let rows = stmt.query_map(params![plugin_id, like, limit.max(1) as i64], |r| {
            Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?))
        })?;
        rows.collect::<std::result::Result<Vec<_>, _>>().map_err(|e| e.into())
    }

    // ------------------------------------------------------------------
    // meta
    // ------------------------------------------------------------------

    // trace:exempt reason=internal-detail
    pub fn meta_set(&self, key: &str, value: &str) -> Result<()> {
        self.conn.execute(
            "INSERT INTO meta (key, value) VALUES (?1, ?2)
             ON CONFLICT(key) DO UPDATE SET value = excluded.value",
            params![key, value],
        )?;
        Ok(())
    }

    // trace:exempt reason=internal-detail
    pub fn meta_get(&self, key: &str) -> Result<Option<String>> {
        let v = self
            .conn
            .query_row("SELECT value FROM meta WHERE key = ?1", params![key], |r| {
                r.get(0)
            })
            .optional()?;
        Ok(v)
    }

    /// Bump one model-epoch generation. Called by every mutation path that
    /// changes system truth (index completion, LSP promotion, adapter
    /// import, intent load, runtime ingestion, derived recompilation).
    // trace:exempt reason=internal-detail
    pub fn bump_epoch(&self, kind: ModelEpochKind) -> Result<()> {
        let key = kind.meta_key();
        let next: u64 = self
            .conn
            .query_row("SELECT value FROM meta WHERE key = ?1", params![key], |r| {
                r.get::<_, String>(0)
            })
            .optional()?
            .and_then(|s| s.parse::<u64>().ok())
            .unwrap_or(0)
            + 1;
        self.meta_set(key, &next.to_string())
    }

    /// Current model epoch (all generations plus the latest snapshot
    /// revision). Never cached by the caller: it must reflect every change
    /// immediately.
    // trace:exempt reason=internal-detail
    pub fn model_epoch(&self) -> Result<ModelEpoch> {
        let g = |k: &str| -> Result<u64> {
            Ok(self
                .meta_get(k)?
                .and_then(|s| s.parse::<u64>().ok())
                .unwrap_or(0))
        };
        Ok(ModelEpoch {
            source: g(ModelEpochKind::Source.meta_key())?,
            semantic: g(ModelEpochKind::Semantic.meta_key())?,
            evidence: g(ModelEpochKind::Evidence.meta_key())?,
            intent: g(ModelEpochKind::Intent.meta_key())?,
            runtime: g(ModelEpochKind::Runtime.meta_key())?,
            derived: g(ModelEpochKind::Derived.meta_key())?,
        })
    }

    // ------------------------------------------------------------------
    // snapshots
    // ------------------------------------------------------------------

    // trace:exempt reason=internal-detail
    pub fn begin_snapshot(&self, revision: &str, branch: Option<&str>) -> Result<i64> {
        self.conn.execute(
            "INSERT INTO snapshots (revision, branch, indexed_at, status) VALUES (?1, ?2, ?3, 'active')",
            params![revision, branch, scc_core::now_rfc3339()],
        )?;
        Ok(self.conn.last_insert_rowid())
    }

    // trace:exempt reason=internal-detail
    pub fn finish_snapshot(&self, id: i64, file_count: usize) -> Result<()> {
        self.conn.execute(
            "UPDATE snapshots SET file_count = ?2, status = 'complete' WHERE id = ?1",
            params![id, file_count as i64],
        )?;
        // the indexed source tree changed — invalidate epoch-keyed packs
        self.bump_epoch(ModelEpochKind::Source)?;
        Ok(())
    }

    // trace:exempt reason=internal-detail
    pub fn latest_snapshot(&self) -> Result<Option<Snapshot>> {
        let row = self
            .conn
            .query_row(
                "SELECT revision, branch, indexed_at FROM snapshots
                 WHERE status = 'complete' ORDER BY id DESC LIMIT 1",
                [],
                |r| {
                    Ok((
                        r.get::<_, String>(0)?,
                        r.get::<_, Option<String>>(1)?,
                        r.get::<_, String>(2)?,
                    ))
                },
            )
            .optional()?;
        Ok(row.map(|(revision, branch, indexed_at)| Snapshot {
            revision,
            branch,
            indexed_at,
        }))
    }

    // trace:exempt reason=internal-detail
    pub fn snapshot_status(&self) -> Result<Option<(Snapshot, i64)>> {
        let row = self
            .conn
            .query_row(
                "SELECT revision, branch, indexed_at, (SELECT COUNT(*) FROM files) FROM snapshots
                 WHERE status = 'complete' ORDER BY id DESC LIMIT 1",
                [],
                |r| {
                    Ok((
                        r.get::<_, String>(0)?,
                        r.get::<_, Option<String>>(1)?,
                        r.get::<_, String>(2)?,
                        r.get::<_, i64>(3)?,
                    ))
                },
            )
            .optional()?;
        Ok(row.map(|(revision, branch, indexed_at, files)| {
            (
                Snapshot {
                    revision,
                    branch,
                    indexed_at,
                },
                files,
            )
        }))
    }

    // ------------------------------------------------------------------
    // files
    // ------------------------------------------------------------------

    // trace:exempt reason=internal-detail
    pub fn upsert_file(&self, path: &str, hash: &str, language: &str, kind: &str, size: u64) -> Result<()> {
        self.conn.execute(
            "INSERT INTO files (path, hash, language, kind, size, indexed_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6)
             ON CONFLICT(path) DO UPDATE SET hash = excluded.hash, language = excluded.language,
               kind = excluded.kind, size = excluded.size, indexed_at = excluded.indexed_at",
            params![path, hash, language, kind, size as i64, scc_core::now_rfc3339()],
        )?;
        Ok(())
    }

    // trace:exempt reason=internal-detail
    pub fn file(&self, path: &str) -> Result<Option<(String, String, String, u64)>> {
        let row = self
            .conn
            .query_row(
                "SELECT hash, language, kind, size FROM files WHERE path = ?1",
                params![path],
                |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get::<_, i64>(3)? as u64)),
            )
            .optional()?;
        Ok(row)
    }

    // trace:exempt reason=internal-detail
    pub fn all_files(&self) -> Result<Vec<(String, String, String, String, u64)>> {
        let mut stmt = self
            .conn
            .prepare("SELECT path, hash, language, kind, size FROM files")?;
        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::<_, i64>(4)? as u64,
            ))
        })?;
        let mut out = Vec::new();
        for row in rows {
            out.push(row?);
        }
        Ok(out)
    }

    // trace:exempt reason=internal-detail
    pub fn delete_file(&self, path: &str) -> Result<()> {
        self.conn.execute("DELETE FROM files WHERE path = ?1", params![path])?;
        Ok(())
    }

// trace:exempt reason=internal-detail

    /// Remove everything tied to a source path: symbols, evidence,
    /// relationships derived from it, and symbol-level entities.
    ///
    /// SCHEMA/REACTIVE concept entities are NOT deleted by path: concepts
    /// are keyed by (kind, name) and may occur in many files. Their
    /// OCCURRENCE entities (per concept/path/owner/line) are deleted by
    /// path, the concept's `sources` provenance is recomputed from the
    /// surviving occurrences, and a concept with zero remaining
    /// occurrences is deleted entirely (with its edges) — so purging one
    /// file never deletes a schema another file still uses, and the
    /// derived occurrence count naturally reflects the survivors.
// trace:exempt reason=internal-detail
    pub fn purge_path(&self, path: &str) -> Result<()> {
        let _sp = self.nest()?;
        // concepts this path's occurrences attach to — captured before any
        // deletion so their provenance can be recomputed afterwards
        let affected_concepts: Vec<String> = {
            let mut stmt = self.conn.prepare(
                "SELECT DISTINCT r.object FROM relationships r
                 JOIN entities e ON e.id = r.subject
                 WHERE r.predicate = ?1 AND e.kind = ?2 AND e.sources LIKE ?3",
            )?;
            let rows = stmt.query_map(
                params![
                    scc_core::predicates::OCCURS,
                    scc_core::kinds::OCCURRENCE,
                    format!("%\"{path}\"%")
                ],
                |r| r.get::<_, String>(0),
            )?;
            let mut v = Vec::new();
            for r in rows {
                v.push(r?);
            }
            v.sort();
            v.dedup();
            v
        };
        // this path's occurrence entities (for edge + FTS cleanup)
        let occ_ids: Vec<String> = {
            let mut stmt = self.conn.prepare(
                "SELECT id FROM entities WHERE kind = ?1 AND sources LIKE ?2",
            )?;
            let rows = stmt.query_map(
                params![scc_core::kinds::OCCURRENCE, format!("%\"{path}\"%")],
                |r| r.get::<_, String>(0),
            )?;
            let mut v = Vec::new();
            for r in rows {
                v.push(r?);
            }
            v
        };
        // symbol entities: repo://{repo}/symbol/{path}/{name}
        let ev_ids: Vec<String> = {
            let mut stmt = self.conn.prepare("SELECT id FROM evidence WHERE path = ?1")?;
            let rows = stmt.query_map(params![path], |r| r.get::<_, String>(0))?;
            let mut v = Vec::new();
            for r in rows {
                v.push(r?);
            }
            v
        };
        // collect the file's symbol names before deleting them
        let sym_names: Vec<String> = {
            let mut stmt = self.conn.prepare("SELECT name FROM symbols WHERE file = ?1")?;
            let rows = stmt.query_map(params![path], |r| r.get::<_, String>(0))?;
            let mut v = Vec::new();
            for r in rows {
                v.push(r?);
            }
            v
        };
        self.conn.execute("DELETE FROM symbols WHERE file = ?1", params![path])?;
        self.conn.execute("DELETE FROM imports WHERE file = ?1", params![path])?;
        self.conn.execute("DELETE FROM evidence WHERE path = ?1", params![path])?;
        // relationships pointing at this file's symbol entities (calls from
        // unchanged files into removed symbols must not dangle)
        let mut orphaned_evidence: Vec<String> = Vec::new();
        for name in sym_names {
            let sid = scc_core::symbol_id(&self.repo_id, path, &name);
            let mut stmt = self.conn.prepare(
                "SELECT id, evidence FROM relationships WHERE subject = ?1 OR object = ?1",
            )?;
            let rows = stmt.query_map(params![sid], |r| {
                Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?))
            })?;
            let mut to_delete: Vec<(String, String)> = Vec::new();
            for r in rows {
                to_delete.push(r?);
            }
            drop(stmt);
            for (rid, ev_json) in to_delete {
                if let Ok(ev_ids) = serde_json::from_str::<Vec<String>>(&ev_json) {
                    orphaned_evidence.extend(ev_ids);
                }
                self.conn.execute("DELETE FROM relationships WHERE id = ?1", params![rid])?;
            }
        }
        // NOTE: evidence referenced only by deleted relationships is swept
        // later by `sweep_orphan_evidence` after the derived layer rebuilds
        // (component/flow edges may still reference it until then).
        let _ = orphaned_evidence;
        self.conn.execute(
            "DELETE FROM entities WHERE id LIKE ?1",
            params![format!("%/symbol/{path}/%")],
        )?;
        // non-concept entities whose sources include the path (stores,
        // routes, contracts, ...). SCHEMA/REACTIVE concepts are keyed by
        // (kind, name) across files and handled below from their live
        // occurrences — a shared concept must never be deleted because one
        // of its files was purged.
        self.conn.execute(
            "DELETE FROM entities WHERE sources LIKE ?1 AND kind NOT IN (?2, ?3)",
            params![
                format!("%\"{path}\"%"),
                scc_core::kinds::SCHEMA,
                scc_core::kinds::REACTIVE
            ],
        )?;
        self.conn.execute(
            "DELETE FROM relationships WHERE source_path = ?1",
            params![path],
        )?;
        // relationships referencing removed evidence ids
        for ev in ev_ids {
            self.conn.execute(
                "DELETE FROM relationships WHERE evidence LIKE ?1",
                params![format!("%\"{ev}\"%")],
            )?;
        }
        self.conn.execute(
            "DELETE FROM tests WHERE file = ?1",
            params![path],
        )?;
        // occurrence entities: delete their edges + FTS rows, then the
        // entities themselves
        for oid in &occ_ids {
            self.conn.execute(
                "DELETE FROM relationships WHERE subject = ?1 OR object = ?1",
                params![oid],
            )?;
            self.conn.execute("DELETE FROM entities_fts WHERE id = ?1", params![oid])?;
        }
        self.conn.execute(
            "DELETE FROM entities WHERE kind = ?1 AND sources LIKE ?2",
            params![scc_core::kinds::OCCURRENCE, format!("%\"{path}\"%")],
        )?;
        // recompute concept provenance: a concept survives as long as any
        // occurrence remains — its sources become the surviving occurrence
        // paths (deduped, sorted). With zero occurrences the concept and
        // its edges are gone.
        for concept in &affected_concepts {
            let remaining: Vec<String> = {
                let mut stmt = self.conn.prepare(
                    "SELECT DISTINCT json_extract(e.attributes, '$.path') FROM entities e
                     JOIN relationships r ON r.subject = e.id
                     WHERE r.predicate = ?1 AND r.object = ?2 AND e.kind = ?3
                       AND json_extract(e.attributes, '$.path') IS NOT NULL
                     ORDER BY 1",
                )?;
                let rows = stmt.query_map(
                    params![
                        scc_core::predicates::OCCURS,
                        concept,
                        scc_core::kinds::OCCURRENCE
                    ],
                    |r| r.get::<_, String>(0),
                )?;
                let mut v = Vec::new();
                for r in rows {
                    v.push(r?);
                }
                v
            };
            if remaining.is_empty() {
                self.conn.execute(
                    "DELETE FROM relationships WHERE subject = ?1 OR object = ?1",
                    params![concept],
                )?;
                self.conn.execute("DELETE FROM entities_fts WHERE id = ?1", params![concept])?;
                self.conn.execute("DELETE FROM entities WHERE id = ?1", params![concept])?;
            } else {
                self.conn.execute(
                    "UPDATE entities SET sources = ?1 WHERE id = ?2",
                    params![serde_json::to_string(&remaining)?, concept],
                )?;
            }
        }
        _sp.commit()?;
        Ok(())
    }

    /// Files that import this path or CALL one of its symbols. Used to
    /// re-extract hash-unchanged dependents so incremental ≡ cold after
    /// type-narrowed CALLS are written on the caller.
    // trace:exempt reason=internal-detail
    pub fn paths_depending_on(&self, path: &str) -> Result<Vec<String>> {
        let file_id = scc_core::entity_id(&self.repo_id, scc_core::kinds::FILE, path);
        let mut stmt = self.conn.prepare(
            "SELECT DISTINCT source_path FROM relationships
             WHERE source_path != ?1 AND source_path != ''
               AND predicate IN (?4, ?5)
               AND (
                 object = ?2
                 OR object IN (
                   SELECT id FROM entities
                   WHERE kind = ?3 AND json_extract(attributes, '$.file') = ?1
                 )
               )
             ORDER BY source_path",
        )?;
        let rows = stmt.query_map(
            params![
                path,
                file_id,
                scc_core::kinds::SYMBOL,
                scc_core::predicates::IMPORTS,
                scc_core::predicates::CALLS
            ],
            |r| r.get::<_, String>(0),
        )?;
        let mut out = Vec::new();
        for r in rows {
            let p = r?;
            if !p.is_empty() {
                out.push(p);
            }
        }
        Ok(out)
    }

    /// Remove all indexed facts (used by full reindex).
    // trace:exempt reason=internal-detail
    pub fn purge_all(&self) -> Result<()> {
        let _sp = self.nest()?;
        for table in [
            "symbols",
            "entities",
            "relationships",
            "evidence",
            "components",
            "flows",
            "invariants",
            "tests",
            "context_cache",
            "drift_findings",
        ] {
            self.conn.execute(&format!("DELETE FROM {table}"), [])?;
        }
        self.conn.execute("DELETE FROM files", [])?;
        self.conn.execute("DELETE FROM snapshots", [])?;
        _sp.commit()?;
        Ok(())
    }

    // ------------------------------------------------------------------
    // symbols
    // ------------------------------------------------------------------

    #[allow(clippy::too_many_arguments)]
    // trace:exempt reason=internal-detail
    pub fn insert_symbol(
        &self,
        file: &str,
        name: &str,
        kind: &str,
        signature: Option<&str>,
        start_line: u32,
        end_line: u32,
        exported: bool,
        docstring: Option<&str>,
    ) -> Result<i64> {
        self.conn.execute(
            "INSERT INTO symbols (file, name, symbol_kind, signature, start_line, end_line, exported, docstring)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
            params![
                file,
                name,
                kind,
                signature,
                start_line as i64,
                end_line as i64,
                exported as i64,
                docstring
            ],
        )?;
        let id = self.conn.last_insert_rowid();
        self.conn.execute(
            "INSERT INTO symbols_fts (name, signature, symbol_kind, file) VALUES (?1, ?2, ?3, ?4)",
            params![name, signature.unwrap_or(""), kind, file],
        )?;
        Ok(id)
    }

    // trace:exempt reason=internal-detail
    pub fn symbols_in_file(&self, file: &str) -> Result<Vec<(i64, String, String, Option<String>, u32, u32, bool, Option<String>)>> {
        let mut stmt = self
            .conn
            .prepare("SELECT id, name, symbol_kind, signature, start_line, end_line, exported, docstring FROM symbols WHERE file = ?1 ORDER BY start_line")?;
        let rows = stmt.query_map(params![file], |r| {
            Ok((
                r.get::<_, i64>(0)?,
                r.get::<_, String>(1)?,
                r.get::<_, String>(2)?,
                r.get::<_, Option<String>>(3)?,
                r.get::<_, i64>(4)? as u32,
                r.get::<_, i64>(5)? as u32,
                r.get::<_, i64>(6)? != 0,
                r.get::<_, Option<String>>(7)?,
            ))
        })?;
        let mut out = Vec::new();
        for r in rows {
            out.push(r?);
        }
        Ok(out)
    }

    // trace:exempt reason=internal-detail
    pub fn symbols_named(&self, name: &str) -> Result<Vec<(String, String, String)>> {
        let mut stmt = self
            .conn
            .prepare("SELECT file, symbol_kind, signature FROM symbols WHERE name = ?1")?;
        let rows = stmt.query_map(params![name], |r| {
            Ok((r.get(0)?, r.get(1)?, r.get(2)?))
        })?;
        let mut out = Vec::new();
        for r in rows {
            out.push(r?);
        }
        Ok(out)
    }

    // ------------------------------------------------------------------
    // imports
    // ------------------------------------------------------------------

    // trace:exempt reason=internal-detail
    pub fn insert_imports(&self, file: &str, imports: &[(String, Vec<(String, String)>, u32, String)]) -> Result<()> {
        let _sp = self.nest()?;
        self.conn.execute("DELETE FROM imports WHERE file = ?1", params![file])?;
        for (module, names, line, typ) in imports {
            self.conn.execute(
                "INSERT INTO imports (file, module, names, line, type) VALUES (?1, ?2, ?3, ?4, ?5)",
                params![file, module, serde_json::to_string(names)?, *line as i64, typ],
            )?;
        }
        _sp.commit()?;
        Ok(())
    }

    // trace:exempt reason=internal-detail
    pub fn imports_in_file(&self, file: &str) -> Result<Vec<(String, Vec<(String, String)>, u32, String)>> {
        let mut stmt = self
            .conn
            .prepare("SELECT module, names, line, type FROM imports WHERE file = ?1 ORDER BY line")?;
        let rows = stmt.query_map(params![file], |r| {
            Ok((
                r.get::<_, String>(0)?,
                r.get::<_, String>(1)?,
                r.get::<_, i64>(2)? as u32,
                r.get::<_, String>(3)?,
            ))
        })?;
        let mut out = Vec::new();
        for r in rows {
            let (module, names, line, typ) = r?;
            out.push((
                module,
                serde_json::from_str(&names).unwrap_or_default(),
                line,
                typ,
            ));
        }
        Ok(out)
    }

    // trace:exempt reason=internal-detail
    pub fn all_imports(&self) -> Result<Vec<(String, String, Vec<(String, String)>, u32, String)>> {
        let mut stmt = self
            .conn
            .prepare("SELECT file, module, names, line, type FROM imports ORDER BY file, line")?;
        let rows = stmt.query_map([], |r| {
            Ok((
                r.get::<_, String>(0)?,
                r.get::<_, String>(1)?,
                r.get::<_, String>(2)?,
                r.get::<_, i64>(3)? as u32,
                r.get::<_, String>(4)?,
            ))
        })?;
        let mut out = Vec::new();
        for r in rows {
            let (file, module, names, line, typ) = r?;
            out.push((
                file,
                module,
                serde_json::from_str(&names).unwrap_or_default(),
                line,
                typ,
            ));
        }
        Ok(out)
    }

    // ------------------------------------------------------------------
    // entities
    // ------------------------------------------------------------------

    // trace:exempt reason=internal-detail
    pub fn insert_entity(&self, entity: &Entity, sources: &[String]) -> Result<()> {
        self.conn.execute(
            "INSERT OR REPLACE INTO entities (id, kind, name, attributes, evidence, sources)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
            params![
                entity.id,
                entity.kind,
                entity.name,
                serde_json::to_string(&entity.attributes)?,
                serde_json::to_string(&entity.evidence)?,
                serde_json::to_string(sources)?,
            ],
        )?;
        self.conn.execute(
            "INSERT OR REPLACE INTO entities_fts (id, kind, name, attributes) VALUES (?1, ?2, ?3, ?4)",
            params![
                entity.id,
                entity.kind,
                entity.name,
                serde_json::to_string(&entity.attributes)?
            ],
        )?;
        Ok(())
    }

    // trace:exempt reason=internal-detail
    pub fn get_entity(&self, id: &str) -> Result<Option<Entity>> {
        let row = self
            .conn
            .query_row(
                "SELECT id, kind, name, attributes, evidence FROM entities WHERE id = ?1",
                params![id],
                |r| {
                    Ok((
                        r.get::<_, String>(0)?,
                        r.get::<_, String>(1)?,
                        r.get::<_, String>(2)?,
                        r.get::<_, String>(3)?,
                        r.get::<_, String>(4)?,
                    ))
                },
            )
            .optional()?;
        Ok(row.map(|(id, kind, name, attributes, evidence)| Entity {
            id,
            kind,
            name,
            attributes: serde_json::from_str(&attributes).unwrap_or_default(),
            evidence: serde_json::from_str(&evidence).unwrap_or_default(),
        }))
    }

    /// The `sources` provenance list of an entity — the repository-relative
    /// file paths that produced it. Stored separately from [`Entity`]
    /// (which carries no sources field), so this is the only reader.
    // trace:v1 id=impl.scc.store.entity_sources work=WORK-SCC-001 satisfies=REQ-SCC-IR
    pub fn entity_sources(&self, id: &str) -> Result<Vec<String>> {
        let row = self
            .conn
            .query_row(
                "SELECT sources FROM entities WHERE id = ?1",
                params![id],
                |r| r.get::<_, String>(0),
            )
            .optional()?;
        Ok(row
            .map(|s| serde_json::from_str(&s).unwrap_or_default())
            .unwrap_or_default())
    }

    /// Live OCCURRENCE entities attached to a concept (OCCURS edges),
    /// sorted by id. Concept counts and `sources` provenance are always
    /// derived from these — never from a stored, write-time-mutated counter.
    // trace:v1 id=impl.scc.store.concept_occurrences work=WORK-SCC-001 satisfies=REQ-SCC-IR
    pub fn concept_occurrences(&self, concept_id: &str) -> Result<Vec<Entity>> {
        let mut stmt = self.conn.prepare(
            "SELECT e.id, e.kind, e.name, e.attributes, e.evidence FROM entities e
             JOIN relationships r ON r.subject = e.id
             WHERE r.predicate = ?1 AND r.object = ?2 AND e.kind = ?3
             ORDER BY e.id",
        )?;
        let rows = stmt.query_map(
            params![
                scc_core::predicates::OCCURS,
                concept_id,
                scc_core::kinds::OCCURRENCE
            ],
            |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 r in rows {
            let (id, kind, name, attributes, evidence) = r?;
            out.push(Entity {
                id,
                kind,
                name,
                attributes: serde_json::from_str(&attributes).unwrap_or_default(),
                evidence: serde_json::from_str(&evidence).unwrap_or_default(),
            });
        }
        Ok(out)
    }

    // trace:exempt reason=internal-detail
    pub fn entities_by_kind(&self, kind: &str) -> Result<Vec<Entity>> {
        let mut stmt = self
            .conn
            .prepare("SELECT id, kind, name, attributes, evidence FROM entities WHERE kind = ?1 ORDER BY name")?;
        let rows = stmt.query_map(params![kind], |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 r in rows {
            let (id, kind, name, attributes, evidence) = r?;
            out.push(Entity {
                id,
                kind,
                name,
                attributes: serde_json::from_str(&attributes).unwrap_or_default(),
                evidence: serde_json::from_str(&evidence).unwrap_or_default(),
            });
        }
        Ok(out)
    }

    // trace:exempt reason=internal-detail
    pub fn all_entities(&self) -> Result<Vec<Entity>> {
        self.all_entities_impl()
    }

    // trace:exempt reason=internal-detail
    fn all_entities_impl(&self) -> Result<Vec<Entity>> {
        let mut stmt = self
            .conn
            .prepare("SELECT id, kind, name, attributes, evidence FROM entities ORDER BY kind, name")?;
        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 r in rows {
            let (id, kind, name, attributes, evidence) = r?;
            out.push(Entity {
                id,
                kind,
                name,
                attributes: serde_json::from_str(&attributes).unwrap_or_default(),
                evidence: serde_json::from_str(&evidence).unwrap_or_default(),
            });
        }
        Ok(out)
    }

    // trace:exempt reason=internal-detail
    pub fn delete_entity(&self, id: &str) -> Result<()> {
        self.conn.execute("DELETE FROM entities WHERE id = ?1", params![id])?;
        self.conn
            .execute("DELETE FROM entities_fts WHERE id = ?1", params![id])?;
        Ok(())
    }

    // trace:exempt reason=internal-detail
    pub fn delete_entities(&self, ids: &[String]) -> Result<()> {
        for id in ids {
            self.delete_entity(id)?;
        }
        Ok(())
    }

    // ------------------------------------------------------------------
    // relationships
    // ------------------------------------------------------------------

    // trace:exempt reason=internal-detail
    pub fn insert_relationship(&self, rel: &Relationship, source_path: &str) -> Result<()> {
        self.conn.execute(
            "INSERT OR REPLACE INTO relationships (id, subject, predicate, object, provenance, confidence, evidence, verified_at, source_path)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
            params![
                rel.id,
                rel.subject,
                rel.predicate,
                rel.object,
                rel.provenance.as_str(),
                rel.confidence,
                serde_json::to_string(&rel.evidence)?,
                rel.verified_at,
                source_path,
            ],
        )?;
        Ok(())
    }

    // trace:exempt reason=internal-detail
    pub fn relationships_for(&self, subject: &str) -> Result<Vec<Relationship>> {
        self.query_relationships("SELECT * FROM relationships WHERE subject = ?1 ORDER BY id", params![subject])
    }

    // trace:exempt reason=internal-detail
    pub fn relationships_to(&self, object: &str) -> Result<Vec<Relationship>> {
        self.query_relationships("SELECT * FROM relationships WHERE object = ?1 ORDER BY id", params![object])
    }

    // trace:exempt reason=internal-detail
    pub fn relationships_between(&self, subject: &str, predicate: &str, object: &str) -> Result<Vec<Relationship>> {
        self.query_relationships(
            "SELECT * FROM relationships WHERE subject = ?1 AND predicate = ?2 AND object = ?3",
            params![subject, predicate, object],
        )
    }

    // trace:exempt reason=internal-detail
    pub fn all_relationships(&self) -> Result<Vec<Relationship>> {
        self.query_relationships("SELECT * FROM relationships ORDER BY id", [])
    }

    /// Relationship ids with the given source path and predicate.
    // trace:exempt reason=internal-detail
    pub fn relationship_ids_with_source(&self, path: &str, predicate: &str) -> Result<Vec<String>> {
        let mut stmt = self.conn.prepare(
            "SELECT id FROM relationships WHERE source_path = ?1 AND predicate = ?2",
        )?;
        let rows = stmt.query_map(params![path, predicate], |r| r.get::<_, String>(0))?;
        let mut out = Vec::new();
        for r in rows {
            out.push(r?);
        }
        Ok(out)
    }

    // trace:exempt reason=internal-detail
    pub fn count_relationships(&self) -> Result<u64> {
        Ok(self
            .conn
            .query_row("SELECT COUNT(*) FROM relationships", [], |r| r.get(0))?)
    }

    // trace:exempt reason=internal-detail
    pub fn delete_relationship(&self, id: &str) -> Result<()> {
        self.conn
            .execute("DELETE FROM relationships WHERE id = ?1", params![id])?;
        Ok(())
    }

    // trace:exempt reason=internal-detail
    fn query_relationships(
        &self,
        sql: &str,
        // trace:exempt reason=internal-detail
        p: impl rusqlite::Params,
    ) -> Result<Vec<Relationship>> {
        let mut stmt = self.conn.prepare(sql)?;
        let rows = stmt.query_map(p, |r| {
            Ok((
                r.get::<_, String>(0)?,
                r.get::<_, String>(1)?,
                r.get::<_, String>(2)?,
                r.get::<_, String>(3)?,
                r.get::<_, String>(4)?,
                r.get::<_, f64>(5)?,
                r.get::<_, String>(6)?,
                r.get::<_, String>(7)?,
            ))
        })?;
        let mut out = Vec::new();
        for row in rows {
            let (id, subject, predicate, object, provenance, confidence, evidence, verified_at) =
                row?;
            out.push(Relationship {
                id,
                subject,
                predicate,
                object,
                provenance: parse_provenance(&provenance),
                confidence,
                evidence: serde_json::from_str(&evidence).unwrap_or_default(),
                verified_at,
            });
        }
        Ok(out)
    }

    // ------------------------------------------------------------------
    // evidence
    // ------------------------------------------------------------------

    // trace:exempt reason=internal-detail
    pub fn insert_evidence(&self, ev: &Evidence) -> Result<()> {
        self.conn.execute(
            "INSERT OR REPLACE INTO evidence (id, type, path, symbol, start_line, end_line, revision, content_hash, extractor, extractor_version)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
            params![
                ev.id,
                evidence_type_str(&ev.r#type),
                ev.path,
                ev.symbol,
                ev.start_line.map(|l| l as i64),
                ev.end_line.map(|l| l as i64),
                ev.revision,
                ev.content_hash,
                ev.extractor,
                ev.extractor_version,
            ],
        )?;
        Ok(())
    }

    // trace:exempt reason=internal-detail
    pub fn get_evidence(&self, id: &str) -> Result<Option<Evidence>> {
        let row = self
            .conn
            .query_row(
                "SELECT id, type, path, symbol, start_line, end_line, revision, content_hash, extractor, extractor_version FROM evidence WHERE id = ?1",
                params![id],
                |r| {
                    Ok((
                        r.get::<_, String>(0)?,
                        r.get::<_, String>(1)?,
                        r.get::<_, Option<String>>(2)?,
                        r.get::<_, Option<String>>(3)?,
                        r.get::<_, Option<i64>>(4)?,
                        r.get::<_, Option<i64>>(5)?,
                        r.get::<_, Option<String>>(6)?,
                        r.get::<_, Option<String>>(7)?,
                        r.get::<_, Option<String>>(8)?,
                        r.get::<_, Option<String>>(9)?,
                    ))
                },
            )
            .optional()?;
        Ok(row.map(|(id, typ, path, symbol, sl, el, rev, hash, ext, extv)| Evidence {
            id,
            r#type: parse_evidence_type(&typ),
            path,
            symbol,
            start_line: sl.map(|v| v as u32),
            end_line: el.map(|v| v as u32),
            revision: rev,
            content_hash: hash,
            extractor: ext,
            extractor_version: extv,
        }))
    }

    // trace:exempt reason=internal-detail
    pub fn all_evidence(&self) -> Result<Vec<Evidence>> {
        let mut stmt = self
            .conn
            .prepare("SELECT id, type, path, symbol, start_line, end_line, revision, content_hash, extractor, extractor_version FROM evidence ORDER BY id")?;
        let rows = stmt.query_map([], |r| {
            Ok((
                r.get::<_, String>(0)?,
                r.get::<_, String>(1)?,
                r.get::<_, Option<String>>(2)?,
                r.get::<_, Option<String>>(3)?,
                r.get::<_, Option<i64>>(4)?,
                r.get::<_, Option<i64>>(5)?,
                r.get::<_, Option<String>>(6)?,
                r.get::<_, Option<String>>(7)?,
                r.get::<_, Option<String>>(8)?,
                r.get::<_, Option<String>>(9)?,
            ))
        })?;
        let mut out = Vec::new();
        for row in rows {
            let (id, typ, path, symbol, sl, el, rev, hash, ext, extv) = row?;
            out.push(Evidence {
                id,
                r#type: parse_evidence_type(&typ),
                path,
                symbol,
                start_line: sl.map(|v| v as u32),
                end_line: el.map(|v| v as u32),
                revision: rev,
                content_hash: hash,
                extractor: ext,
                extractor_version: extv,
            });
        }
        Ok(out)
    }

    // trace:exempt reason=internal-detail
    pub fn evidence_for_path(&self, path: &str) -> Result<Vec<Evidence>> {
        let mut stmt = self
            .conn
            .prepare("SELECT id, type, path, symbol, start_line, end_line, revision, content_hash, extractor, extractor_version FROM evidence WHERE path = ?1 ORDER BY id")?;
        let rows = stmt.query_map(params![path], |r| {
            Ok((
                r.get::<_, String>(0)?,
                r.get::<_, String>(1)?,
                r.get::<_, Option<String>>(2)?,
                r.get::<_, Option<String>>(3)?,
                r.get::<_, Option<i64>>(4)?,
                r.get::<_, Option<i64>>(5)?,
                r.get::<_, Option<String>>(6)?,
                r.get::<_, Option<String>>(7)?,
                r.get::<_, Option<String>>(8)?,
                r.get::<_, Option<String>>(9)?,
            ))
        })?;
        let mut out = Vec::new();
        for row in rows {
            let (id, typ, path, symbol, sl, el, rev, hash, ext, extv) = row?;
            out.push(Evidence {
                id,
                r#type: parse_evidence_type(&typ),
                path,
                symbol,
                start_line: sl.map(|v| v as u32),
                end_line: el.map(|v| v as u32),
                revision: rev,
                content_hash: hash,
                extractor: ext,
                extractor_version: extv,
            });
        }
        Ok(out)
    }

    // ------------------------------------------------------------------
    // components / flows / invariants / tests
    // ------------------------------------------------------------------

    // trace:exempt reason=internal-detail
    pub fn replace_components(&self, components: &[Entity]) -> Result<()> {
        let _sp = self.nest()?;
        self.conn.execute("DELETE FROM components", [])?;
        // INSERT OR REPLACE only covers ids still present. A clustering
        // topology change (merged `root+services` splitting back into
        // `root` + `services`) must drop the vanished derived entities or
        // System IR export keeps stale component nodes.
        let keep: HashSet<&str> = components.iter().map(|c| c.id.as_str()).collect();
        let stale: Vec<String> = {
            let mut stmt = self.conn.prepare("SELECT id FROM entities WHERE kind = ?1")?;
            let rows = stmt.query_map(params![scc_core::kinds::COMPONENT], |r| {
                r.get::<_, String>(0)
            })?;
            let mut v = Vec::new();
            for r in rows {
                let id = r?;
                if !keep.contains(id.as_str()) {
                    v.push(id);
                }
            }
            v
        };
        for id in &stale {
            self.conn.execute("DELETE FROM entities WHERE id = ?1", params![id])?;
            self.conn.execute("DELETE FROM entities_fts WHERE id = ?1", params![id])?;
        }
        for c in components {
            self.conn.execute(
                "INSERT INTO components (id, name, kind, responsibility, implementation, evidence, attributes)
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
                params![
                    c.id,
                    c.name,
                    c.kind,
                    serde_json::to_string(&c.attributes.get("responsibility").cloned().unwrap_or(serde_json::json!([])))?,
                    serde_json::to_string(&c.attributes.get("implementation").cloned().unwrap_or(serde_json::json!([])))?,
                    serde_json::to_string(&c.evidence)?,
                    serde_json::to_string(&c.attributes)?,
                ],
            )?;
            self.conn.execute(
                "INSERT OR REPLACE INTO entities (id, kind, name, attributes, evidence, sources) VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
                params![c.id, c.kind, c.name, serde_json::to_string(&c.attributes)?, serde_json::to_string(&c.evidence)?, "[]"],
            )?;
            self.conn.execute(
                "INSERT OR REPLACE INTO entities_fts (id, kind, name, attributes) VALUES (?1, ?2, ?3, ?4)",
                params![c.id, c.kind, c.name, serde_json::to_string(&c.attributes)?],
            )?;
        }
        _sp.commit()?;
        Ok(())
    }

    // trace:exempt reason=internal-detail
    pub fn components(&self) -> Result<Vec<Entity>> {
        let mut stmt = self
            .conn
            .prepare("SELECT id, name, kind, responsibility, implementation, evidence, attributes FROM components ORDER BY name")?;
        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)?,
                r.get::<_, String>(5)?,
                r.get::<_, String>(6)?,
            ))
        })?;
        let mut out = Vec::new();
        for row in rows {
            let (id, name, kind, resp, implm, ev, attrs) = row?;
            let mut attributes: std::collections::BTreeMap<String, serde_json::Value> =
                serde_json::from_str(&attrs).unwrap_or_default();
            if let Ok(r) = serde_json::from_str::<Vec<serde_json::Value>>(&resp) {
                attributes.insert("responsibility".into(), serde_json::Value::Array(r));
            }
            if let Ok(r) = serde_json::from_str::<Vec<serde_json::Value>>(&implm) {
                attributes.insert("implementation".into(), serde_json::Value::Array(r));
            }
            out.push(Entity {
                id,
                kind,
                name,
                attributes,
                evidence: serde_json::from_str(&ev).unwrap_or_default(),
            });
        }
        Ok(out)
    }

    // ------------------------------------------------------------------
    // canonical flow graphs (Wave 3)
    // ------------------------------------------------------------------

    // trace:exempt reason=internal-detail
    pub fn replace_flow_graphs(&self, graphs: &[scc_core::FlowGraph]) -> Result<()> {
        let _sp = self.nest()?;
        self.conn.execute("DELETE FROM flow_graphs", [])?;
        for g in graphs {
            let kind = scc_core::flow_kind_str(&g.kind);
            let trigger = g.trigger.clone().unwrap_or_default();
            self.conn.execute(
                "INSERT OR REPLACE INTO flow_graphs (id, kind, name, trigger, graph) VALUES (?1, ?2, ?3, ?4, ?5)",
                params![g.id, kind, g.name, trigger, serde_json::to_string(g)?],
            )?;
        }
        _sp.commit()?;
        Ok(())
    }

    // trace:exempt reason=internal-detail
    pub fn flow_graphs(&self) -> Result<Vec<scc_core::FlowGraph>> {
        let mut stmt = self
            .conn
            .prepare("SELECT graph FROM flow_graphs ORDER BY id")?;
        let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
        let mut out = Vec::new();
        for r in rows {
            let json = r?;
            if let Ok(g) = serde_json::from_str(&json) {
                out.push(g);
            }
        }
        Ok(out)
    }

    // trace:exempt reason=internal-detail
    pub fn replace_flows(&self, flows: &[Flow]) -> Result<()> {
        let _sp = self.nest()?;
        self.conn.execute("DELETE FROM flows", [])?;
        for f in flows {
            self.conn.execute(
                "INSERT INTO flows (id, kind, name, trigger, steps, attributes) VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
                params![
                    f.id,
                    flow_kind_str(&f.kind),
                    f.name,
                    f.trigger,
                    serde_json::to_string(&f.steps)?,
                    serde_json::to_string(&f.attributes)?
                ],
            )?;
        }
        _sp.commit()?;
        Ok(())
    }

    // trace:exempt reason=internal-detail
    pub fn flows(&self) -> Result<Vec<Flow>> {
        let mut stmt = self
            .conn
            .prepare("SELECT id, kind, name, trigger, steps, attributes FROM flows ORDER BY kind, name")?;
        let rows = stmt.query_map([], |r| {
            Ok((
                r.get::<_, String>(0)?,
                r.get::<_, String>(1)?,
                r.get::<_, String>(2)?,
                r.get::<_, Option<String>>(3)?,
                r.get::<_, String>(4)?,
                r.get::<_, String>(5)?,
            ))
        })?;
        let mut out = Vec::new();
        for row in rows {
            let (id, kind, name, trigger, steps, attrs) = row?;
            out.push(Flow {
                id,
                kind: parse_flow_kind(&kind),
                name,
                trigger,
                steps: serde_json::from_str(&steps).unwrap_or_default(),
                attributes: serde_json::from_str(&attrs).unwrap_or_default(),
            });
        }
        Ok(out)
    }

    // trace:exempt reason=internal-detail
    pub fn flow(&self, id: &str) -> Result<Option<Flow>> {
        let row = self
            .conn
            .query_row(
                "SELECT id, kind, name, trigger, steps, attributes FROM flows WHERE id = ?1",
                params![id],
                |r| {
                    Ok((
                        r.get::<_, String>(0)?,
                        r.get::<_, String>(1)?,
                        r.get::<_, String>(2)?,
                        r.get::<_, Option<String>>(3)?,
                        r.get::<_, String>(4)?,
                        r.get::<_, String>(5)?,
                    ))
                },
            )
            .optional()?;
        Ok(row.map(|(id, kind, name, trigger, steps, attrs)| Flow {
            id,
            kind: parse_flow_kind(&kind),
            name,
            trigger,
            steps: serde_json::from_str(&steps).unwrap_or_default(),
            attributes: serde_json::from_str(&attrs).unwrap_or_default(),
        }))
    }

    // trace:exempt reason=internal-detail
    pub fn replace_invariants(&self, invariants: &[Invariant]) -> Result<()> {
        let _sp = self.nest()?;
        self.conn.execute("DELETE FROM invariants", [])?;
        for inv in invariants {
            self.conn.execute(
                "INSERT INTO invariants (id, statement, severity, scope, enforced_by, provenance, evidence)
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
                params![
                    inv.id,
                    inv.statement,
                    severity_str(&inv.severity),
                    serde_json::to_string(&inv.scope)?,
                    serde_json::to_string(&inv.enforced_by)?,
                    inv.provenance.map(|p| p.as_str().to_string()).unwrap_or_else(|| "DECLARED".into()),
                    serde_json::to_string(&inv.evidence)?,
                ],
            )?;
        }
        _sp.commit()?;
        Ok(())
    }

    // trace:exempt reason=internal-detail
    pub fn invariants(&self) -> Result<Vec<Invariant>> {
        let mut stmt = self
            .conn
            .prepare("SELECT id, statement, severity, scope, enforced_by, provenance, evidence FROM invariants ORDER BY id")?;
        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)?,
                r.get::<_, String>(5)?,
                r.get::<_, String>(6)?,
            ))
        })?;
        let mut out = Vec::new();
        for row in rows {
            let (id, statement, severity, scope, enforced_by, provenance, ev) = row?;
            out.push(Invariant {
                id,
                statement,
                severity: parse_severity(&severity),
                scope: serde_json::from_str(&scope).unwrap_or_default(),
                enforced_by: serde_json::from_str(&enforced_by).unwrap_or_default(),
                provenance: Some(parse_provenance(&provenance)),
                evidence: serde_json::from_str(&ev).unwrap_or_default(),
            });
        }
        Ok(out)
    }

    // trace:exempt reason=internal-detail
    pub fn insert_test(&self, id: &str, name: &str, file: &str, kind: &str, symbol: Option<&str>) -> Result<()> {
        self.conn.execute(
            "INSERT OR REPLACE INTO tests (id, name, file, kind, symbol) VALUES (?1, ?2, ?3, ?4, ?5)",
            params![id, name, file, kind, symbol],
        )?;
        Ok(())
    }

    // trace:exempt reason=internal-detail
    pub fn tests(&self) -> Result<Vec<(String, String, String, String, Option<String>)>> {
        let mut stmt = self
            .conn
            .prepare("SELECT id, name, file, kind, symbol FROM tests ORDER BY file, name")?;
        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::<_, Option<String>>(4)?,
            ))
        })?;
        let mut out = Vec::new();
        for r in rows {
            out.push(r?);
        }
        Ok(out)
    }

    /// Delete evidence records no longer referenced by any entity,
    /// relationship, component, invariant, flow step, or flow-graph node/
    /// edge. Run after the derived layer (components/flows) rebuilds,
    /// because derived edges may briefly hold references during
    /// recompilation.
    // trace:exempt reason=internal-detail
    pub fn sweep_orphan_evidence(&self) -> Result<u64> {
        // Set-based pass: collect every referenced evidence id ONCE (exact
        // id membership in the JSON arrays, no per-row LIKE scans), then
        // delete evidence rows whose id is not referenced anywhere.
        let mut referenced: HashSet<String> = HashSet::new();

        // Columns holding JSON arrays of evidence ids.
        for (table, column) in [
            ("entities", "evidence"),
            ("relationships", "evidence"),
            ("components", "evidence"),
            ("invariants", "evidence"),
        ] {
            let sql = format!("SELECT {column} FROM {table}");
            let mut stmt = self.conn.prepare(&sql)?;
            let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
            for r in rows {
                if let Ok(ids) = serde_json::from_str::<Vec<String>>(&r?) {
                    referenced.extend(ids);
                }
            }
            drop(stmt);
        }

        // flows.steps: array of FlowStep objects, each with an evidence list.
        {
            let mut stmt = self.conn.prepare("SELECT steps FROM flows")?;
            let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
            for r in rows {
                if let Ok(steps) = serde_json::from_str::<Vec<scc_core::FlowStep>>(&r?) {
                    for step in steps {
                        referenced.extend(step.evidence);
                    }
                }
            }
            drop(stmt);
        }

        // flow_graphs.graph: nodes[].evidence and edges[].evidence arrays.
        {
            let mut stmt = self.conn.prepare("SELECT graph FROM flow_graphs")?;
            let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
            for r in rows {
                if let Ok(g) = serde_json::from_str::<scc_core::FlowGraph>(&r?) {
                    for node in &g.nodes {
                        referenced.extend(node.evidence.iter().cloned());
                    }
                    for edge in &g.edges {
                        referenced.extend(edge.evidence.iter().cloned());
                    }
                }
            }
            drop(stmt);
        }

        let mut stmt = self.conn.prepare("SELECT id FROM evidence")?;
        let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
        let mut orphans: Vec<String> = Vec::new();
        for r in rows {
            let id = r?;
            if !referenced.contains(&id) {
                orphans.push(id);
            }
        }
        drop(stmt);

        let count = orphans.len() as u64;
        // Chunked deletes avoid one giant IN list.
        for chunk in orphans.chunks(500) {
            let placeholders = vec!["?"; chunk.len()].join(",");
            let sql = format!("DELETE FROM evidence WHERE id IN ({placeholders})");
            self.conn
                .execute(&sql, rusqlite::params_from_iter(chunk.iter()))?;
        }
        Ok(count)
    }

    // ------------------------------------------------------------------
    // context cache
    // ------------------------------------------------------------------

    /// Canonical epoch string for cache keys: composite of every model
    /// generation plus the latest snapshot revision. A previously fresh
    /// pack can never be served after any source of system truth changed.
    // trace:exempt reason=internal-detail
    pub fn cache_epoch(&self) -> Result<String> {
        let revision = self
            .latest_snapshot()?
            .map(|s| s.revision)
            .unwrap_or_else(|| "not-indexed".to_string());
        Ok(self.model_epoch()?.composite(&revision))
    }

    // trace:exempt reason=internal-detail
    pub fn cache_get(&self, key: &str, epoch: &str) -> Result<Option<String>> {
        let row = self
            .conn
            .query_row(
                "SELECT pack FROM context_cache WHERE key = ?1 AND epoch = ?2",
                params![key, epoch],
                |r| r.get(0),
            )
            .optional()?;
        Ok(row)
    }

    // trace:exempt reason=internal-detail
    pub fn cache_put(&self, key: &str, pack: &str, epoch: &str) -> Result<()> {
        self.conn.execute(
            "INSERT OR REPLACE INTO context_cache (key, pack, epoch, created_at) VALUES (?1, ?2, ?3, ?4)",
            params![key, pack, epoch, scc_core::now_rfc3339()],
        )?;
        Ok(())
    }

    // trace:exempt reason=internal-detail
    pub fn cache_clear(&self) -> Result<()> {
        self.conn.execute("DELETE FROM context_cache", [])?;
        Ok(())
    }

    // ------------------------------------------------------------------
    // intent claims / drift findings
    // ------------------------------------------------------------------

    // trace:exempt reason=internal-detail
    pub fn replace_intent_claims(&self, claims: &[(String, serde_json::Value)]) -> Result<()> {
        let _sp = self.nest()?;
        self.conn.execute("DELETE FROM intent_claims", [])?;
        for (source, claim) in claims {
            self.conn.execute(
                "INSERT INTO intent_claims (source, claim, created_at) VALUES (?1, ?2, ?3)",
                params![source, serde_json::to_string(claim)?, scc_core::now_rfc3339()],
            )?;
        }
        _sp.commit()?;
        // declared intent changed — invalidate epoch-keyed packs
        self.bump_epoch(ModelEpochKind::Intent)?;
        Ok(())
    }

    // trace:exempt reason=internal-detail
    pub fn intent_claims(&self) -> Result<Vec<(String, serde_json::Value)>> {
        let mut stmt = self
            .conn
            .prepare("SELECT source, claim FROM intent_claims ORDER BY id")?;
        let rows = stmt.query_map([], |r| {
            Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?))
        })?;
        let mut out = Vec::new();
        for r in rows {
            let (source, claim) = r?;
            if let Ok(v) = serde_json::from_str(&claim) {
                out.push((source, v));
            }
        }
        Ok(out)
    }

    // ------------------------------------------------------------------
    // runtime edges
    // ------------------------------------------------------------------

    // trace:exempt reason=internal-detail
    pub fn runtime_edge_rows(&self) -> Result<Vec<RuntimeEdgeRow>> {
        let mut stmt = self.conn.prepare(
            "SELECT source, target, count, latency_ms, errors, last_observed
             FROM runtime_edges ORDER BY source, target",
        )?;
        let rows = stmt.query_map([], |r| {
            Ok(RuntimeEdgeRow {
                source: r.get(0)?,
                target: r.get(1)?,
                count: r.get::<_, i64>(2)? as u64,
                latency_ms: r.get(3)?,
                errors: r.get::<_, i64>(4)? as u64,
                last_observed: r.get(5)?,
            })
        })?;
        let mut out = Vec::new();
        for r in rows {
            out.push(r?);
        }
        Ok(out)
    }

    // ------------------------------------------------------------------
    // trace signatures (Wave 6)
    // ------------------------------------------------------------------

    /// Record one occurrence of an observed trace-path signature. `count`
    /// increments by one (one trace occurrence), `latency_ms` is merged as a
    /// count-weighted running average, `errors` is additive. Does NOT bump
    /// any model-epoch generation: ingestion callers bump the Runtime
    /// generation once per payload.
    // trace:exempt reason=internal-detail
    pub fn upsert_trace_signature(&self, signature: &str, latency_ms: f64, errors: u64) -> Result<()> {
        self.conn.execute(
            "INSERT INTO trace_signatures (signature, count, latency_ms, errors, last_observed)
             VALUES (?1, 1, ?2, ?3, ?4)
             ON CONFLICT(signature) DO UPDATE SET
               count = trace_signatures.count + 1,
               latency_ms = (trace_signatures.latency_ms * trace_signatures.count + excluded.latency_ms)
                            / (trace_signatures.count + 1),
               errors = trace_signatures.errors + excluded.errors,
               last_observed = excluded.last_observed",
            params![signature, latency_ms, errors as i64, scc_core::now_rfc3339()],
        )?;
        Ok(())
    }

    /// All observed trace signatures, ordered by (count DESC, signature).
    // trace:exempt reason=internal-detail
    pub fn trace_signatures(&self) -> Result<Vec<(String, u64, f64, u64, String)>> {
        let mut stmt = self.conn.prepare(
            "SELECT signature, count, latency_ms, errors, last_observed
             FROM trace_signatures ORDER BY count DESC, signature",
        )?;
        let rows = stmt.query_map([], |r| {
            Ok((
                r.get::<_, String>(0)?,
                r.get::<_, i64>(1)? as u64,
                r.get::<_, f64>(2)?,
                r.get::<_, i64>(3)? as u64,
                r.get::<_, String>(4)?,
            ))
        })?;
        let mut out = Vec::new();
        for r in rows {
            out.push(r?);
        }
        Ok(out)
    }

    // ------------------------------------------------------------------
    // embeddings
    // ------------------------------------------------------------------

    // trace:exempt reason=internal-detail
    pub fn put_embedding(&self, entity_id: &str, vector: &[f32], model: &str) -> Result<()> {
        let bytes: Vec<u8> = vector
            .iter()
            .flat_map(|f| f.to_le_bytes())
            .collect();
        self.conn.execute(
            "INSERT OR REPLACE INTO embeddings (entity_id, vector, model, updated_at)
             VALUES (?1, ?2, ?3, ?4)",
            params![entity_id, bytes, model, scc_core::now_rfc3339()],
        )?;
        Ok(())
    }

    // trace:exempt reason=internal-detail
    pub fn get_embedding(&self, entity_id: &str) -> Result<Option<(Vec<f32>, String)>> {
        let row = self
            .conn
            .query_row(
                "SELECT vector, model FROM embeddings WHERE entity_id = ?1",
                params![entity_id],
                |r| Ok((r.get::<_, Vec<u8>>(0)?, r.get::<_, String>(1)?)),
            )
            .optional()?;
        Ok(row.map(|(bytes, model)| {
            let v = bytes
                .as_chunks::<4>().0.iter()
                .map(|c| f32::from_le_bytes([c[0], c[1], c[2], c[3]]))
                .collect();
            (v, model)
        }))
    }

    // trace:exempt reason=internal-detail
    pub fn embedding_count(&self) -> Result<u64> {
        Ok(self
            .conn
            .query_row("SELECT COUNT(*) FROM embeddings", [], |r| r.get(0))?)
    }

    // trace:exempt reason=internal-detail
    pub fn add_drift_finding(&self, kind: &str, severity: &str, message: &str) -> Result<i64> {
        self.conn.execute(
            "INSERT INTO drift_findings (kind, severity, message, created_at, resolved) VALUES (?1, ?2, ?3, ?4, 0)",
            params![kind, severity, message, scc_core::now_rfc3339()],
        )?;
        Ok(self.conn.last_insert_rowid())
    }

    // trace:exempt reason=internal-detail
    pub fn drift_findings(&self, unresolved_only: bool) -> Result<Vec<(i64, String, String, String, String)>> {
        let sql = if unresolved_only {
            "SELECT id, kind, severity, message, created_at FROM drift_findings WHERE resolved = 0 ORDER BY id"
        } else {
            "SELECT id, kind, severity, message, created_at FROM drift_findings ORDER BY id"
        };
        let mut stmt = self.conn.prepare(sql)?;
        let rows = stmt.query_map([], |r| {
            Ok((
                r.get::<_, i64>(0)?,
                r.get::<_, String>(1)?,
                r.get::<_, String>(2)?,
                r.get::<_, String>(3)?,
                r.get::<_, String>(4)?,
            ))
        })?;
        let mut out = Vec::new();
        for r in rows {
            out.push(r?);
        }
        Ok(out)
    }

    // trace:exempt reason=internal-detail
    pub fn clear_drift_findings(&self) -> Result<()> {
        self.conn.execute("DELETE FROM drift_findings", [])?;
        Ok(())
    }

    // ------------------------------------------------------------------
    // search (FTS5)
    // ------------------------------------------------------------------

    /// Lexical search over entities (components, routes, data, stores...).
    // trace:exempt reason=internal-detail
    pub fn search_entities(&self, query: &str, limit: usize) -> Result<Vec<Entity>> {
        let q = fts_query(query);
        let mut stmt = self.conn.prepare(
            "SELECT e.id, e.kind, e.name, e.attributes, e.evidence
             FROM entities_fts f JOIN entities e ON e.id = f.id
             WHERE entities_fts MATCH ?1 ORDER BY bm25(entities_fts) LIMIT ?2",
        )?;
        let rows = stmt.query_map(params![q, limit as i64], |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 r in rows {
            let (id, kind, name, attributes, evidence) = r?;
            out.push(Entity {
                id,
                kind,
                name,
                attributes: serde_json::from_str(&attributes).unwrap_or_default(),
                evidence: serde_json::from_str(&evidence).unwrap_or_default(),
            });
        }
        Ok(out)
    }

    /// Lexical search over symbols.
    // trace:exempt reason=internal-detail
    pub fn search_symbols(&self, query: &str, limit: usize) -> Result<Vec<(String, String, String, String, u32)>> {
        let q = fts_query(query);
        let mut stmt = self.conn.prepare(
            "SELECT f.name, f.signature, f.symbol_kind, f.file,
                    COALESCE((SELECT s.start_line FROM symbols s
                              WHERE s.file = f.file AND s.name = f.name LIMIT 1), 0)
             FROM symbols_fts f
             WHERE symbols_fts MATCH ?1 ORDER BY bm25(symbols_fts) LIMIT ?2",
        )?;
        let rows = stmt.query_map(params![q, limit as i64], |r| {
            Ok((
                r.get::<_, String>(0)?,
                r.get::<_, Option<String>>(1)?.unwrap_or_default(),
                r.get::<_, String>(2)?,
                r.get::<_, String>(3)?,
                r.get::<_, i64>(4)? as u32,
            ))
        })?;
        let mut out = Vec::new();
        for r in rows {
            out.push(r?);
        }
        Ok(out)
    }

    /// Substring fallback over entity names AND attributes (docstrings,
    /// signatures, responsibilities) — case-insensitive. Used when FTS prefix
    /// matching misses morphological variants or multi-term AND queries drop
    /// otherwise-strong matches.
    // trace:exempt reason=internal-detail
    pub fn search_entities_like(&self, term: &str, limit: usize) -> Result<Vec<Entity>> {
        let pat = format!("%{}%", term.to_ascii_lowercase());
        let mut stmt = self.conn.prepare(
            "SELECT id, kind, name, attributes, evidence FROM entities
             WHERE lower(name) LIKE ?1 OR lower(attributes) LIKE ?1
             ORDER BY CASE WHEN lower(name) LIKE ?1 THEN 0 ELSE 1 END, length(name)
             LIMIT ?2",
        )?;
        let rows = stmt.query_map(params![pat, limit as i64], |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 r in rows {
            let (id, kind, name, attributes, evidence) = r?;
            out.push(Entity {
                id,
                kind,
                name,
                attributes: serde_json::from_str(&attributes).unwrap_or_default(),
                evidence: serde_json::from_str(&evidence).unwrap_or_default(),
            });
        }
        Ok(out)
    }

    /// Kind-scoped substring start-set for traversal (§16): entities of
    /// `kind` whose name matches `term` (empty term = all of the kind,
    /// capped by `limit`). LIKE keeps this dependency-free (no FTS).
    // trace:v1 id=impl.scc-store-entities-like-kind work=WORK-SI-MMMJA4G6 satisfies=REQ-SI-503JSBGP
    pub fn search_entities_like_kind(&self, kind: &str, term: &str, limit: usize) -> Result<Vec<Entity>> {
        let pat = format!("%{}%", term.to_ascii_lowercase());
        let mut stmt = self.conn.prepare(
            "SELECT id, kind, name, attributes, evidence FROM entities
             WHERE kind = ?1 AND lower(name) LIKE ?2
             ORDER BY length(name) LIMIT ?3",
        )?;
        let rows = stmt.query_map(params![kind, pat, limit as i64], |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 r in rows {
            let (id, kind, name, attributes, evidence) = r?;
            out.push(Entity {
                id,
                kind,
                name,
                attributes: serde_json::from_str(&attributes).unwrap_or_default(),
                evidence: serde_json::from_str(&evidence).unwrap_or_default(),
            });
        }
        Ok(out)
    }

    /// Substring fallback over symbols (name, signature, docstring).
    // trace:exempt reason=internal-detail
    pub fn search_symbols_like(&self, term: &str, limit: usize) -> Result<Vec<(String, String, String, String, u32)>> {
        let pat = format!("%{}%", term.to_ascii_lowercase());
        let mut stmt = self.conn.prepare(
            "SELECT name, signature, symbol_kind, file, start_line FROM symbols
             WHERE lower(name) LIKE ?1 OR lower(signature) LIKE ?1 OR lower(docstring) LIKE ?1
             ORDER BY CASE WHEN lower(name) LIKE ?1 THEN 0 ELSE 1 END, length(name)
             LIMIT ?2",
        )?;
        let rows = stmt.query_map(params![pat, limit as i64], |r| {
            Ok((
                r.get::<_, String>(0)?,
                r.get::<_, Option<String>>(1)?.unwrap_or_default(),
                r.get::<_, String>(2)?,
                r.get::<_, String>(3)?,
                r.get::<_, i64>(4)? as u32,
            ))
        })?;
        let mut out = Vec::new();
        for r in rows {
            out.push(r?);
        }
        Ok(out)
    }

    // ------------------------------------------------------------------
    // stats
    // ------------------------------------------------------------------

    // trace:exempt reason=internal-detail
    pub fn stats(&self) -> Result<HashMap<String, u64>> {
        let mut m = HashMap::new();
        for (name, sql) in [
            ("files", "SELECT COUNT(*) FROM files"),
            ("symbols", "SELECT COUNT(*) FROM symbols"),
            ("entities", "SELECT COUNT(*) FROM entities"),
            ("relationships", "SELECT COUNT(*) FROM relationships"),
            ("evidence", "SELECT COUNT(*) FROM evidence"),
            ("components", "SELECT COUNT(*) FROM components"),
            ("flows", "SELECT COUNT(*) FROM flows"),
            ("invariants", "SELECT COUNT(*) FROM invariants"),
            ("tests", "SELECT COUNT(*) FROM tests"),
        ] {
            let n: i64 = self.conn.query_row(sql, [], |r| r.get(0))?;
            m.insert(name.to_string(), n as u64);
        }
        Ok(m)
    }
}

// ---------------------------------------------------------------------------
// helpers
// ---------------------------------------------------------------------------

// trace:exempt reason=internal-detail
fn apply_migrations(conn: &Connection) -> Result<()> {
    let version: i64 = conn.query_row("PRAGMA user_version", [], |r| r.get(0))?;
    let current = SCHEMA_VERSION as i64;
    if version > current {
        return Err(StoreError::Sqlite(rusqlite::Error::InvalidParameterName(
            format!(
                "database schema v{version} is newer than supported v{SCHEMA_VERSION}"
            )
            .into_boxed_str()
            .to_string(),
        )));
    }
    // Apply every migration after the current version, in order.
    for (i, m) in MIGRATIONS.iter().enumerate() {
        let target = (i + 1) as i64;
        if version < target {
            conn.execute_batch(m)?;
        }
    }
    if version < current {
        conn.pragma_update(None, "user_version", current)?;
    }
    Ok(())
}

// trace:exempt reason=internal-detail
pub fn parse_provenance(s: &str) -> Provenance {
    match s {
        "EXTRACTED" => Provenance::Extracted,
        "RESOLVED" => Provenance::Resolved,
        "OBSERVED" => Provenance::Observed,
        "DECLARED" => Provenance::Declared,
        "INFERRED" => Provenance::Inferred,
        "STALE" => Provenance::Stale,
        _ => Provenance::Inferred,
    }
}

// trace:exempt reason=internal-detail
pub fn evidence_type_str(t: &scc_core::EvidenceType) -> &'static str {
    match t {
        scc_core::EvidenceType::Source => "source",
        scc_core::EvidenceType::Config => "config",
        scc_core::EvidenceType::Runtime => "runtime",
        scc_core::EvidenceType::Test => "test",
        scc_core::EvidenceType::Intent => "intent",
        scc_core::EvidenceType::History => "history",
    }
}

// trace:exempt reason=internal-detail
pub fn parse_evidence_type(s: &str) -> scc_core::EvidenceType {
    match s {
        "source" => scc_core::EvidenceType::Source,
        "config" => scc_core::EvidenceType::Config,
        "runtime" => scc_core::EvidenceType::Runtime,
        "test" => scc_core::EvidenceType::Test,
        "intent" => scc_core::EvidenceType::Intent,
        "history" => scc_core::EvidenceType::History,
        _ => scc_core::EvidenceType::Source,
    }
}

// trace:exempt reason=internal-detail
pub fn flow_kind_str(k: &scc_core::FlowKind) -> &'static str {
    match k {
        scc_core::FlowKind::Architecture => "architecture",
        scc_core::FlowKind::Workflow => "workflow",
        scc_core::FlowKind::Sequence => "sequence",
        scc_core::FlowKind::Dataflow => "dataflow",
        scc_core::FlowKind::Lifecycle => "lifecycle",
    }
}

// trace:exempt reason=internal-detail
pub fn parse_flow_kind(s: &str) -> scc_core::FlowKind {
    match s {
        "architecture" => scc_core::FlowKind::Architecture,
        "workflow" => scc_core::FlowKind::Workflow,
        "sequence" => scc_core::FlowKind::Sequence,
        "dataflow" => scc_core::FlowKind::Dataflow,
        "lifecycle" => scc_core::FlowKind::Lifecycle,
        _ => scc_core::FlowKind::Sequence,
    }
}

// trace:exempt reason=internal-detail
pub fn severity_str(s: &Severity) -> &'static str {
    match s {
        Severity::Info => "info",
        Severity::Low => "low",
        Severity::Medium => "medium",
        Severity::High => "high",
        Severity::Critical => "critical",
    }
}

// trace:exempt reason=internal-detail
pub fn parse_severity(s: &str) -> Severity {
    match s {
        "info" => Severity::Info,
        "low" => Severity::Low,
        "medium" => Severity::Medium,
        "high" => Severity::High,
        "critical" => Severity::Critical,
        _ => Severity::Medium,
    }
}

/// Build a safe FTS5 MATCH expression from free-form text: quoted terms with
/// prefix matching on the last term.
// trace:exempt reason=internal-detail
fn fts_query(text: &str) -> String {
    let tokens: Vec<String> = text
        .split(|c: char| !c.is_alphanumeric() && c != '_' && c != '-' && c != '.')
        .filter(|t| !t.is_empty())
        .map(|t| {
            let t = t.trim_matches('"');
            format!("\"{t}\"*")
        })
        .collect();
    if tokens.is_empty() {
        return "\"\"".to_string();
    }
    tokens.join(" ")
}

#[cfg(test)]
// trace:exempt reason=internal-detail
mod tests {
    use super::*;
    use tempfile::TempDir;

    // trace:exempt reason=internal-detail
    pub(crate) fn tmp_store() -> (Store, TempDir) {
        let dir = TempDir::new().unwrap();
        let root = dir.path().join("repo");
        std::fs::create_dir_all(&root).unwrap();
        let store = Store::open(&dir.path().join("scc.db"), &root).unwrap();
        (store, dir)
    }

    #[test]
    // trace:v1 id=test.scc.store.recovering-open-quarantines-malformed verifies=REQ-SI-503JSBGP exercises=impl.scc.store.open-recovering
    fn recovering_open_quarantines_malformed_db() {
        let dir = TempDir::new().unwrap();
        let root = dir.path().join("repo");
        std::fs::create_dir_all(&root).unwrap();
        let db = dir.path().join("scc.db");
        // garbage with a valid SQLite header prefix still fails lazily:
        // write a valid header followed by garbage pages.
        let mut bytes = b"SQLite format 3\0".to_vec();
        bytes.resize(4096, 0xFF);
        std::fs::write(&db, &bytes).unwrap();
        let (store, quarantined) = Store::open_recovering(&db, &root).unwrap();
        let q = quarantined.expect("corrupt db must be quarantined");
        assert!(q.is_file(), "quarantine evidence preserved");
        // fresh store is fully usable
        let v: i64 = store
            .conn
            .query_row("PRAGMA user_version", [], |r| r.get(0))
            .unwrap();
        assert!(v >= 0);
        // non-corrupt opens never quarantine
        let (_, q2) = Store::open_recovering(&db, &root).unwrap();
        assert!(q2.is_none(), "healthy db must not quarantine");
    }

    #[test]
    // trace:v1 id=test.scc.store.persistent-identity-survives-move verifies=REQ-SI-503JSBGP exercises=impl.scc.store.stable-identity
    fn persistent_identity_survives_checkout_move() {
        let dir = TempDir::new().unwrap();
        let root_a = dir.path().join("aaa");
        std::fs::create_dir_all(&root_a).unwrap();
        let db = dir.path().join("scc.db");
        let s1 = Store::open(&db, &root_a).unwrap();
        assert_eq!(s1.repo_id, "aaa");
        // move the checkout: same database, new directory name
        let root_b = dir.path().join("bbb");
        std::fs::create_dir_all(&root_b).unwrap();
        let s2 = Store::open(&db, &root_b).unwrap();
        assert_eq!(s2.repo_id, "aaa", "move must not fork identity");
        assert_eq!(
            s2.meta_get("repo_id_source").unwrap().as_deref(),
            Some("persistent")
        );
        // provenance path follows the checkout
        let row: String = s2.conn.query_row(
            "SELECT root FROM repositories WHERE id = 'aaa'",
            [],
            |r| r.get(0),
        ).unwrap();
        assert!(row.ends_with("bbb"), "{row}");
    }

    #[test]
    // trace:v1 id=test.scc.store.explicit-and-remote-adopt verifies=REQ-SI-503JSBGP exercises=impl.scc.store.adopt-stable-identity
    fn explicit_and_remote_identity_adopt_on_fresh_db() {
        let (mut s, _d) = tmp_store();
        assert_eq!(s.repo_id, "repo");
        assert!(s.adopt_stable_identity(Some("MyRepo"), None).unwrap());
        assert_eq!(s.repo_id, "myrepo");
        assert!(!s.adopt_stable_identity(Some("MyRepo"), None).unwrap());
        let (mut t, _e) = tmp_store();
        assert!(t.adopt_stable_identity(None, Some("git@github.com:acme/billing.git")).unwrap());
        assert_eq!(t.repo_id, "github.com/acme/billing");
        // explicit operator id is never overwritten by a later remote
        assert_eq!(t.meta_get("repo_id_source").unwrap().as_deref(), Some("remote"));
        // entities freeze identity: adoption refuses once ids are load-bearing
        t.insert_entity(&Entity::new("x", "symbol", "f"), &["a.py".into()]).unwrap();
        assert!(!t.adopt_stable_identity(Some("other"), None).unwrap());
        assert_eq!(t.repo_id, "github.com/acme/billing");
    }


    #[test]
    // trace:exempt reason=internal-detail
    fn migrations_apply_and_reopen() {
        let dir = TempDir::new().unwrap();
        let root = dir.path().join("repo");
        std::fs::create_dir_all(&root).unwrap();
        let db = dir.path().join("scc.db");
        {
            let s = Store::open(&db, &root).unwrap();
            s.insert_entity(&Entity::new("repo://t/component/a", "component", "A"), &["x.py".into()]).unwrap();
        }
        // reopen
        let s = Store::open(&db, &root).unwrap();
        assert!(s.get_entity("repo://t/component/a").unwrap().is_some());
    }

    #[test]
    // trace:v1 id=test.scc.store.refuse-corrupt verifies=REQ-implement-phase-12-of-scc-x-ripwire-lessons-java-unprefixed-field-as,REQ-implement-fix-pr-review-comments-without-collapsing-scc-type-script-no exercises=impl.scc.store.refuse-corrupt
    fn truncated_or_garbage_db_refuses_to_open() {
        let dir = TempDir::new().unwrap();
        let root = dir.path().join("repo");
        std::fs::create_dir_all(&root).unwrap();
        let db = dir.path().join("scc.db");

        std::fs::write(&db, b"not a sqlite database at all").unwrap();
        let err = match Store::open(&db, &root) {
            Err(e) => e,
            Ok(_) => panic!("garbage db must not open"),
        };
        assert!(
            matches!(err, StoreError::Corrupt(_)),
            "garbage must be Corrupt, got {err}"
        );
        assert!(
            err.to_string().contains("corrupt") || err.to_string().contains("not a SQLite"),
            "{err}"
        );

        {
            let s = Store::open(&dir.path().join("fresh.db"), &root).unwrap();
            s.meta_set("k", "v").unwrap();
        }
        let good = std::fs::read(dir.path().join("fresh.db")).unwrap();
        assert!(good.len() > 32, "expected a real sqlite file");
        std::fs::write(&db, &good[..32]).unwrap();
        let err = match Store::open(&db, &root) {
            Err(e) => e,
            Ok(_) => panic!("truncated sqlite must not open"),
        };
        assert!(
            matches!(err, StoreError::Corrupt(_)),
            "truncated sqlite must be Corrupt, got {err}"
        );

        let empty = dir.path().join("empty.db");
        std::fs::write(&empty, b"").unwrap();
        Store::open(&empty, &root).expect("empty file is a fresh index");
    }

    #[test]
    // trace:exempt reason=internal-detail
    fn model_epoch_bumps_and_composites() {
        let (s, _d) = tmp_store();
        assert_eq!(s.model_epoch().unwrap(), ModelEpoch::zero());
        let e0 = s.cache_epoch().unwrap();

        // snapshot completion bumps the source generation
        let id = s.begin_snapshot("abc", None).unwrap();
        s.finish_snapshot(id, 3).unwrap();
        let e1 = s.model_epoch().unwrap();
        assert_eq!(e1.source, 1);
        assert_ne!(s.cache_epoch().unwrap(), e0);

        // semantic promotion is an independent generation
        s.bump_epoch(ModelEpochKind::Semantic).unwrap();
        let e2 = s.model_epoch().unwrap();
        assert_eq!(e2.semantic, 1);
        assert_eq!(e2.source, 1);

        // composite differs per generation combination and includes the
        // snapshot revision
        let c1 = e1.composite("abc");
        let c2 = e2.composite("abc");
        assert_ne!(c1, c2);
        assert_ne!(c1, e1.composite("def"));
        // same state -> same composite (deterministic)
        assert_eq!(c1, e1.composite("abc"));
    }

    #[test]
    // trace:exempt reason=internal-detail
    fn cache_is_keyed_on_epoch() {
        let (s, _d) = tmp_store();
        let e0 = s.cache_epoch().unwrap();
        s.cache_put("task:1", "pack-A", &e0).unwrap();
        assert_eq!(s.cache_get("task:1", &e0).unwrap().as_deref(), Some("pack-A"));

        // any truth change invalidates the old epoch key
        s.bump_epoch(ModelEpochKind::Runtime).unwrap();
        let e1 = s.cache_epoch().unwrap();
        assert_ne!(e0, e1);
        assert_eq!(s.cache_get("task:1", &e1).unwrap(), None);
        // old packs remain addressable only by their own epoch
        assert_eq!(s.cache_get("task:1", &e0).unwrap().as_deref(), Some("pack-A"));
    }

    #[test]
    // trace:exempt reason=internal-detail
    fn intent_replacement_bumps_intent_epoch() {
        let (s, _d) = tmp_store();
        let before = s.model_epoch().unwrap().intent;
        s.replace_intent_claims(&[("component".into(), serde_json::json!({"name": "a"}))])
            .unwrap();
        let after = s.model_epoch().unwrap();
        assert_eq!(after.intent, before + 1);
    }

    #[test]
    // trace:exempt reason=internal-detail
    fn trace_signature_upsert_roundtrip_and_epoch_neutral() {
        let dir = TempDir::new().unwrap();
        let root = dir.path().join("repo");
        std::fs::create_dir_all(&root).unwrap();
        let db = dir.path().join("scc.db");
        {
            let s = Store::open(&db, &root).unwrap();
            assert!(s.trace_signatures().unwrap().is_empty());

            // upserts must not bump any model-epoch generation (the
            // ingestion layer bumps the Runtime generation once per payload)
            let epoch_before = s.model_epoch().unwrap();
            s.upsert_trace_signature("root -> api -> db", 4.5, 1).unwrap();
            s.upsert_trace_signature("root -> api -> db", 5.5, 0).unwrap();
            s.upsert_trace_signature("root -> web -> api", 2.0, 0).unwrap();
            assert_eq!(s.model_epoch().unwrap(), epoch_before);

            let sigs = s.trace_signatures().unwrap();
            assert_eq!(sigs.len(), 2);
            // count increments, latency is a count-weighted running average
            // ((4.5 + 5.5) / 2), errors are additive
            assert_eq!(sigs[0].0, "root -> api -> db");
            assert_eq!(sigs[0].1, 2);
            assert!((sigs[0].2 - 5.0).abs() < 1e-9);
            assert_eq!(sigs[0].3, 1);
            assert!(!sigs[0].4.is_empty());
            // ordering: (count DESC, signature)
            assert_eq!(sigs[1].0, "root -> web -> api");
            assert_eq!(sigs[1].1, 1);
        }
        // schema v6 + rows survive reopen
        let s = Store::open(&db, &root).unwrap();
        let sigs = s.trace_signatures().unwrap();
        assert_eq!(sigs.len(), 2);
        assert_eq!(sigs[0].0, "root -> api -> db");
        assert_eq!(sigs[0].1, 2);
    }

    #[test]
    // trace:exempt reason=internal-detail
    fn entity_roundtrip_and_fts() {
        let (s, _d) = tmp_store();
        let mut e = Entity::new("repo://r/component/transcript", "component", "transcript-normalizer");
        e.attr("responsibility", serde_json::json!(["normalize transcripts"]));
        s.insert_entity(&e, &["src/normalize.py".into()]).unwrap();
        let got = s.get_entity("repo://r/component/transcript").unwrap().unwrap();
        assert_eq!(got.name, "transcript-normalizer");
        let hits = s.search_entities("normalize", 10).unwrap();
        assert_eq!(hits.len(), 1);
        assert_eq!(hits[0].id, "repo://r/component/transcript");
    }

    #[test]
    // trace:exempt reason=internal-detail
    fn relationship_roundtrip() {
        let (s, _d) = tmp_store();
        let rel = Relationship::new(
            "rel:1",
            "repo://r/component/a",
            "calls",
            "repo://r/component/b",
            Provenance::Resolved,
        )
        .with_evidence(vec!["evidence:1".into()]);
        s.insert_relationship(&rel, "src/a.py").unwrap();
        let got = s.relationships_for("repo://r/component/a").unwrap();
        assert_eq!(got.len(), 1);
        assert_eq!(got[0].predicate, "calls");
        assert_eq!(got[0].provenance, Provenance::Resolved);
    }

    #[test]
    // trace:exempt reason=internal-detail
    fn sweep_orphan_evidence_removes_only_unreferenced() {
        let (s, _d) = tmp_store();
        // three evidence rows: two referenced (entity, relationship), one not
        s.insert_evidence(&Evidence::source("evidence:ent", "src/a.py")).unwrap();
        s.insert_evidence(&Evidence::source("evidence:rel", "src/b.py")).unwrap();
        s.insert_evidence(&Evidence::source("evidence:orphan", "src/c.py")).unwrap();

        let mut e = Entity::new("repo://r/component/keep", "component", "keep");
        e.evidence = vec!["evidence:ent".into()];
        s.insert_entity(&e, &["src/a.py".into()]).unwrap();
        let rel = Relationship::new(
            "rel:1",
            "repo://r/component/keep",
            "calls",
            "repo://r/component/other",
            Provenance::Resolved,
        )
        .with_evidence(vec!["evidence:rel".into()]);
        s.insert_relationship(&rel, "src/b.py").unwrap();

        // sweep: the unreferenced row goes, the referenced rows stay
        assert_eq!(s.sweep_orphan_evidence().unwrap(), 1);
        assert!(s.get_evidence("evidence:ent").unwrap().is_some());
        assert!(s.get_evidence("evidence:rel").unwrap().is_some());
        assert!(s.get_evidence("evidence:orphan").unwrap().is_none());

        // drop the entity's reference, sweep again: now it becomes an orphan
        let e2 = Entity::new("repo://r/component/keep", "component", "keep");
        s.insert_entity(&e2, &["src/a.py".into()]).unwrap();
        assert_eq!(s.sweep_orphan_evidence().unwrap(), 1);
        assert!(s.get_evidence("evidence:ent").unwrap().is_none());
        assert!(s.get_evidence("evidence:rel").unwrap().is_some());
    }

    #[test]
    // trace:exempt reason=internal-detail
    fn sweep_orphan_evidence_keeps_flow_and_component_references() {
        let (s, _d) = tmp_store();
        // derived-layer references (flow step + component) must protect
        // evidence even though entities/relationships no longer mention it
        s.insert_evidence(&Evidence::source("evidence:flow", "src/a.py")).unwrap();
        s.insert_evidence(&Evidence::source("evidence:comp", "src/b.py")).unwrap();
        s.insert_evidence(&Evidence::source("evidence:gone", "src/c.py")).unwrap();

        // component referencing evidence:comp
        let mut comp = Entity::new("repo://r/component/keep", "component", "keep");
        comp.evidence = vec!["evidence:comp".into()];
        s.replace_components(&[comp]).unwrap();

        // flow step referencing evidence:flow
        let flow = scc_core::Flow {
            id: "flow:1".into(),
            kind: scc_core::FlowKind::Workflow,
            name: "wf".into(),
            trigger: None,
            steps: vec![scc_core::FlowStep {
                id: "step:1".into(),
                order: 0,
                actor: "repo://r/component/keep".into(),
                operation: "run".into(),
                condition: None,
                r#async: None,
                timeout_ms: None,
                retry_policy: None,
                failure_outcome: None,
                provenance: None,
                evidence: vec!["evidence:flow".into()],
            }],
            attributes: Default::default(),
        };
        s.replace_flows(&[flow]).unwrap();

        assert_eq!(s.sweep_orphan_evidence().unwrap(), 1);
        assert!(s.get_evidence("evidence:flow").unwrap().is_some());
        assert!(s.get_evidence("evidence:comp").unwrap().is_some());
        assert!(s.get_evidence("evidence:gone").unwrap().is_none());
    }

    #[test]
    // trace:exempt reason=internal-detail
    fn purge_path_cascades() {
        let (s, _d) = tmp_store();
        s.insert_symbol("a.py", "foo", "function", None, 1, 5, true, None).unwrap();
        s.insert_evidence(&Evidence::source("evidence:1", "a.py")).unwrap();
        let rel = Relationship::new("rel:1", "repo://r/symbol/a.py/foo", "calls", "repo://r/symbol/b.py/bar", Provenance::Extracted)
            .with_evidence(vec!["evidence:1".into()]);
        s.insert_relationship(&rel, "a.py").unwrap();
        s.purge_path("a.py").unwrap();
        assert_eq!(s.symbols_in_file("a.py").unwrap().len(), 0);
        assert_eq!(s.all_relationships().unwrap().len(), 0);
        assert!(s.get_evidence("evidence:1").unwrap().is_none());
    }

    #[test]
    // trace:v1 id=test.scc.store.paths-depending verifies=REQ-implement-fix-pr-review-comments-without-collapsing-scc-type-script-no
    fn paths_depending_on_finds_importers_and_callers() {
        let (s, _d) = tmp_store();
        let repo = s.repo_id.clone();
        let file_b = scc_core::entity_id(&repo, scc_core::kinds::FILE, "b.py");
        let file_a = scc_core::entity_id(&repo, scc_core::kinds::FILE, "a.py");
        s.insert_entity(
            &Entity::new(file_b.clone(), scc_core::kinds::FILE, "b.py"),
            &["b.py".into()],
        )
        .unwrap();
        s.insert_entity(
            &Entity::new(file_a.clone(), scc_core::kinds::FILE, "a.py"),
            &["a.py".into()],
        )
        .unwrap();
        let mut meth = Entity::new(
            scc_core::symbol_id(&repo, "b.py", "Order.process"),
            scc_core::kinds::SYMBOL,
            "Order.process",
        );
        meth.attr("file", serde_json::json!("b.py"));
        s.insert_entity(&meth, &["b.py".into()]).unwrap();
        s.insert_relationship(
            &Relationship::new(
                "rel:imp",
                file_a.clone(),
                scc_core::predicates::IMPORTS,
                file_b,
                Provenance::Extracted,
            ),
            "a.py",
        )
        .unwrap();
        s.insert_relationship(
            &Relationship::new(
                "rel:call",
                scc_core::symbol_id(&repo, "a.py", "handle"),
                scc_core::predicates::CALLS,
                scc_core::symbol_id(&repo, "b.py", "Order.process"),
                Provenance::Extracted,
            ),
            "a.py",
        )
        .unwrap();
        let deps = s.paths_depending_on("b.py").unwrap();
        assert_eq!(deps, vec!["a.py".to_string()]);
        assert!(s.paths_depending_on("a.py").unwrap().is_empty());
        s.insert_relationship(
            &Relationship::new(
                "rel:tested",
                scc_core::symbol_id(&repo, "c.py", "test_it"),
                scc_core::predicates::TESTED_BY,
                scc_core::symbol_id(&repo, "b.py", "Order.process"),
                Provenance::Extracted,
            ),
            "c.py",
        )
        .unwrap();
        let deps = s.paths_depending_on("b.py").unwrap();
        assert_eq!(
            deps,
            vec!["a.py".to_string()],
            "TESTED_BY must not cascade: {deps:?}"
        );
    }

    #[test]
    // trace:exempt reason=internal-detail
    fn replace_components_drops_vanished_component_entities() {
        let (s, _d) = tmp_store();
        let merged = Entity::new(
            "repo://r/component/root-services",
            scc_core::kinds::COMPONENT,
            "root+services",
        );
        let root = Entity::new(
            "repo://r/component/root",
            scc_core::kinds::COMPONENT,
            "root",
        );
        s.replace_components(&[merged, root.clone()]).unwrap();
        assert_eq!(s.entities_by_kind(scc_core::kinds::COMPONENT).unwrap().len(), 2);
        s.replace_components(&[root]).unwrap();
        let left = s.entities_by_kind(scc_core::kinds::COMPONENT).unwrap();
        assert_eq!(left.len(), 1, "{left:?}");
        assert_eq!(left[0].name, "root");
        assert!(s.get_entity("repo://r/component/root-services").unwrap().is_none());
    }

// trace:exempt reason=internal-detail

    /// Wave 13: shared concepts survive a single-file purge. The concept's
    /// `sources` provenance is recomputed from the surviving occurrences
    /// (never a stored counter), and the concept is deleted only when its
    /// last occurrence is purged.
    #[test]
// trace:exempt reason=internal-detail
    fn purge_path_recomputes_shared_concept_provenance() {
        let (s, _d) = tmp_store();
        let repo = &s.repo_id;
        let expr = "z.object({ name: z.string() })";
        let concept = scc_core::entity_id(repo, scc_core::kinds::SCHEMA, expr);
        // one concept, two occurrences (A.ts / B.ts)
        s.insert_entity(
            &Entity::new(concept.clone(), scc_core::kinds::SCHEMA, expr),
            &["a.ts".into(), "b.ts".into()],
        )
        .unwrap();
        for (path, owner) in [("a.ts", "makeA"), ("b.ts", "makeB")] {
            let occ = scc_core::occurrence_id(repo, expr, path, owner, 3);
            s.insert_entity(
                Entity::new(occ.clone(), scc_core::kinds::OCCURRENCE, format!("{expr}@{path}@{owner}@3"))
                    .attr("concept", serde_json::json!(concept))
                    .attr("path", serde_json::json!(path))
                    .attr("owner", serde_json::json!(owner))
                    .attr("line", serde_json::json!(3)),
                &[path.to_string()],
            )
            .unwrap();
            s.insert_relationship(
                &Relationship::new(
                    format!("rel:occ:{path}"),
                    occ.clone(),
                    scc_core::predicates::OCCURS,
                    concept.clone(),
                    Provenance::Extracted,
                ),
                path,
            )
            .unwrap();
        }
        // purge A: concept survives, count 1, provenance recomputed to B
        s.purge_path("a.ts").unwrap();
        let concept_ent = s
            .get_entity(&concept)
            .unwrap()
            .expect("concept survives a single-file purge");
        assert_eq!(concept_ent.kind, scc_core::kinds::SCHEMA);
        assert_eq!(s.entity_sources(&concept).unwrap(), vec!["b.ts"]);
        let occs = s.concept_occurrences(&concept).unwrap();
        assert_eq!(occs.len(), 1, "derived count drops to 1: {occs:?}");
        assert_eq!(
            occs[0].attributes.get("path").and_then(|v| v.as_str()),
            Some("b.ts")
        );
        // purge B: last occurrence gone -> concept deleted with its edges
        s.purge_path("b.ts").unwrap();
        assert!(s.get_entity(&concept).unwrap().is_none(), "concept gone");
        assert!(s.all_relationships().unwrap().is_empty(), "no dangling edges");
        // unrelated entities with the path in sources still purge
        s.insert_entity(
            &Entity::new("repo://r/store/db", "store", "db"),
            &["a.ts".into()],
        )
        .unwrap();
        s.purge_path("a.ts").unwrap();
        assert!(s.get_entity("repo://r/store/db").unwrap().is_none());
    }

    #[test]
    // trace:exempt reason=internal-detail
    fn cache_revision_scoped() {
        let (s, _d) = tmp_store();
        s.cache_put("k", "pack-v1", "rev1").unwrap();
        assert_eq!(s.cache_get("k", "rev1").unwrap(), Some("pack-v1".into()));
        assert_eq!(s.cache_get("k", "rev2").unwrap(), None);
    }

    #[test]
    // trace:exempt reason=internal-detail
    fn fts_escapes_punctuation() {
        let (s, _d) = tmp_store();
        let mut e = Entity::new("repo://r/route/api", "route", "GET /api/v1/items");
        e.attr("path", serde_json::json!("/api/v1/items"));
        s.insert_entity(&e, &["app.py".into()]).unwrap();
        let hits = s.search_entities("GET /api/v1/items", 10).unwrap();
        assert_eq!(hits.len(), 1);
    }
}

#[cfg(test)]
// trace:exempt reason=internal-detail
mod like_tests {
    use super::*;
    use crate::tests::tmp_store;

    #[test]
    // trace:exempt reason=internal-detail
    fn entities_like_matches_attributes() {
        let (s, _d) = tmp_store();
        let mut e = Entity::new("repo://r/symbol/a.py/f", "symbol", "transcribe");
        e.attr("docstring", serde_json::json!("External ASR client with retry and fallback."));
        s.insert_entity(&e, &["a.py".into()]).unwrap();
        let hits = s.search_entities_like("retry", 6).unwrap();
        assert_eq!(hits.len(), 1, "must match docstring attributes");
        let hits2 = s.search_symbols_like("retry", 6).unwrap();
        assert_eq!(hits2.len(), 0);
    }

    #[test]
    // trace:exempt reason=internal-detail
    fn symbols_like_matches_docstring() {
        let (s, _d) = tmp_store();
        s.insert_symbol("a.py", "transcribe", "function", None, 1, 2, true,
                        Some("External ASR client with retry and fallback.")).unwrap();
        let hits = s.search_symbols_like("retry", 6).unwrap();
        assert_eq!(hits.len(), 1, "must match symbol docstrings");
        assert_eq!(hits[0].0, "transcribe");
    }
}