semlith 0.46.0

//! SQLite side of the store: file bookkeeping and chunk payloads.
//!
//! The vector index only ever holds `(chunk id, quantized vector)`. Everything
//! a caller actually wants back — the text, the path, the line span — lives
//! here and is looked up by chunk id after the search returns.

use anyhow::Result;
use rusqlite::types::Value;
use rusqlite::{Connection, OptionalExtension, params};
use std::path::Path;

pub(crate) const SCHEMA: &str = r#"
CREATE TABLE IF NOT EXISTS meta (
    k TEXT PRIMARY KEY,
    v TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS files (
    id         INTEGER PRIMARY KEY,
    path       TEXT NOT NULL UNIQUE,
    hash       TEXT NOT NULL,
    bytes      INTEGER NOT NULL,
    indexed_at INTEGER NOT NULL
);

-- AUTOINCREMENT, not a bare rowid: a sharded index routes a chunk id to a
-- shard by the range its file name claims, so SQLite handing a deleted row's
-- id to a new chunk would put one id inside two shards at once.
CREATE TABLE IF NOT EXISTS chunks (
    id         INTEGER PRIMARY KEY AUTOINCREMENT,
    file_id    INTEGER NOT NULL REFERENCES files(id) ON DELETE CASCADE,
    ord        INTEGER NOT NULL,
    start_line INTEGER NOT NULL,
    end_line   INTEGER NOT NULL,
    text       TEXT NOT NULL
);

CREATE INDEX IF NOT EXISTS chunks_file_id ON chunks(file_id);
-- How many lines each file spans, for the Stores page, read from the index
-- alone: without it that is every chunk's text off disk, fourteen seconds on
-- a 600 MB store, and the portal draws nothing until it has the answer.
CREATE INDEX IF NOT EXISTS chunks_file_end ON chunks(file_id, end_line);
-- A search asks whether anything is still being embedded, and which of its
-- keyword candidates are; without this that is a scan of every file row.
CREATE INDEX IF NOT EXISTS files_hash ON files(hash);

-- Keyword half of the search. `content='chunks'` means FTS5 keeps only its
-- index, not a second copy of every chunk, so the store barely grows.
CREATE VIRTUAL TABLE IF NOT EXISTS chunks_fts USING fts5(
    text,
    content='chunks',
    content_rowid='id',
    tokenize='unicode61'
);

-- External-content FTS5 does not track its source table by itself. A delete
-- has to hand back the original text so FTS5 can find the terms it needs to
-- remove; this trigger does, and foreign-key cascades fire it too, which is
-- what keeps `delete_file` correct. An insert is indexed by `insert_chunk`
-- itself: the trigger that did it cost twice the row's own insert (#198), and
-- one an older binary recreated is dropped again before every write (see
-- `drop_insert_trigger`).
DROP TRIGGER IF EXISTS chunks_fts_insert;

CREATE TRIGGER IF NOT EXISTS chunks_fts_delete AFTER DELETE ON chunks BEGIN
    INSERT INTO chunks_fts(chunks_fts, rowid, text) VALUES('delete', old.id, old.text);
END;

-- Structure half of the store, from 0.12.0. Additive: an older store grows
-- these tables on its next open and they stay empty until something indexes
-- into it, which is why the format version does not move for them. See
-- `docs/compatibility.md`.
--
-- `chunk_id` is nullable on purpose. A symbol's defining line always falls
-- inside some chunk, but a file whose chunks were rewritten between extraction
-- and insertion would otherwise have to fail rather than record the symbol.
CREATE TABLE IF NOT EXISTS symbols (
    id         INTEGER PRIMARY KEY AUTOINCREMENT,
    file_id    INTEGER NOT NULL REFERENCES files(id) ON DELETE CASCADE,
    chunk_id   INTEGER,
    kind       TEXT NOT NULL,
    name       TEXT NOT NULL,
    qualified  TEXT NOT NULL,
    start_line INTEGER NOT NULL,
    end_line   INTEGER NOT NULL
);

CREATE INDEX IF NOT EXISTS symbols_file_id ON symbols(file_id);
CREATE INDEX IF NOT EXISTS symbols_name ON symbols(name);
-- Search seeds its graph walk from the symbols inside its top chunks. Without
-- this, each seed chunk was a scan of every symbol: on the 879k-chunk benchmark
-- corpus, 1.25M rows read at random, once per seed, on every search.
CREATE INDEX IF NOT EXISTS symbols_chunk_id ON symbols(chunk_id);

-- What a symbol used to be, from 0.23.0. A re-index no longer simply deletes
-- the definitions of a file it is about to rewrite: it copies them here first,
-- stamped with the content hash the file had while they were true, so
-- `semlith symbol` can answer "what did this look like before".
--
-- A separate table rather than a validity column on `symbols`, because
-- `symbols.file_id` cascades from `files` and a re-index deletes the file row.
-- Making the live rows outlive their file would mean loosening that foreign
-- key, which is the one thing keeping the graph from rotting; copying them out
-- keeps the live table exactly as strict as it was.
--
-- Additive and `IF NOT EXISTS`, so `format_version` does not move, for the same
-- reason the 0.12.0 graph tables did not move it: an older binary opens the
-- store, never looks in here, and answers exactly as before. A store written
-- before 0.23.0 has an empty table and starts filling it at its next index
-- pass. See `docs/compatibility.md`.
--
-- Nothing prunes this. History is kept whole in 0.23.0 -- a row per changed
-- symbol per pass -- and a retention knob is a later release's problem, when
-- growth is something measured rather than imagined.
CREATE TABLE IF NOT EXISTS symbols_past (
    id           INTEGER PRIMARY KEY AUTOINCREMENT,
    path         TEXT NOT NULL,
    kind         TEXT NOT NULL,
    name         TEXT NOT NULL,
    qualified    TEXT NOT NULL,
    start_line   INTEGER NOT NULL,
    end_line     INTEGER NOT NULL,
    -- The content hash of the file version this definition belonged to. What
    -- "invalid at" means: everything after this hash is a different file.
    content_hash TEXT NOT NULL,
    retired_at   INTEGER NOT NULL
);

CREATE INDEX IF NOT EXISTS symbols_past_name ON symbols_past(name);
-- The change brief asks for one path's oldest retirement since a moment, per
-- changed file. With only the name index that was two scans of the whole
-- table per file, and on a store with 953,601 history rows `semlith report
-- change` ran past 12 minutes without finishing (0.42.0). Additive, so an
-- older binary opens the store and never uses it.
CREATE INDEX IF NOT EXISTS symbols_past_path ON symbols_past(path, retired_at);

-- An edge is owned by the file its *source* is in: `src` is a symbol id, and
-- the cascade above deletes an edge when its source file is re-indexed or
-- forgotten.
--
-- `dst` is a symbol NAME, not an id, and that asymmetry is the point. Ids are
-- reissued every time a file is re-extracted, so an id here would mean that
-- re-indexing `b.rs` silently deleted every edge pointing into it from `a.rs`
-- — the graph would rot from the one operation this release exists to make
-- safe. A name is resolved against `symbols_name` at query time instead, so an
-- edge is always as current as both of its ends, and an edge to something not
-- indexed (a standard-library call) is still recorded and simply resolves to
-- nothing.
--
-- `confidence` is `extracted` when an import in the source file named where the
-- target came from, and `inferred` when the target was matched by bare name.
-- Nothing that displays an edge may present the second as the first.
-- `hint` is what the source text said about where `dst` lives, when it said
-- anything: the module of a scoped call, the receiver of a method call, the
-- object of a qualified one. Nullable, and NULL on every row an older binary
-- wrote, which is why the format version does not move for it. It is a lead,
-- not an address, and `edges_out` treats it as one ranking signal among
-- several — see `resolve`.
-- `line` is where the reference was written in the source file, from 0.16.0.
-- Not where either symbol is defined: "where is X invoked" is a question about
-- this column, and until it existed the closest answer was the enclosing
-- symbol's own start line. Nullable and NULL on every row an older binary
-- wrote, so the format version does not move for it either.
CREATE TABLE IF NOT EXISTS edges (
    src        INTEGER NOT NULL REFERENCES symbols(id) ON DELETE CASCADE,
    dst        TEXT NOT NULL,
    kind       TEXT NOT NULL,
    confidence TEXT NOT NULL,
    hint       TEXT,
    line       INTEGER
);

CREATE INDEX IF NOT EXISTS edges_src ON edges(src);
CREATE INDEX IF NOT EXISTS edges_dst ON edges(dst);

-- What was not indexed, and why, from 0.30.0. One row per file for a secret,
-- a credential file, a policy limit or a file no reader can take, and one row
-- per folder for a pruned generated folder or an ignore rule. Written by every
-- pass that decides not to index something -- `semlith index`, the watcher and
-- the daemon's catch-up alike -- so it outlives the run that found it.
--
-- `matches` is JSON of what the scan found, masked: provider, line, the first
-- and last few characters, confidence and its signals. Never a value.
-- Additive and `IF NOT EXISTS`, so `format_version` does not move.
CREATE TABLE IF NOT EXISTS refusals (
    path       TEXT PRIMARY KEY,
    class      TEXT NOT NULL,
    rule       TEXT NOT NULL,
    matches    TEXT,
    confidence INTEGER,
    files      INTEGER NOT NULL DEFAULT 1,
    first_seen INTEGER NOT NULL,
    last_seen  INTEGER NOT NULL
);

-- A person's decision about one refused file, from 0.30.0. `fingerprints` is
-- JSON of salted blake3 fingerprints of the matches accepted -- never the
-- values -- so a later pass can tell "the same secret the person saw" from "a
-- new one", and refuse the file again for the second. `mode` is `redacted`,
-- `as-is`, or `refused` for a dummy-only file a person chose to keep out.
CREATE TABLE IF NOT EXISTS acceptances (
    path         TEXT PRIMARY KEY,
    class        TEXT NOT NULL,
    mode         TEXT NOT NULL,
    fingerprints TEXT NOT NULL,
    confidence   INTEGER,
    at           INTEGER NOT NULL,
    source       TEXT NOT NULL
);

-- Images, from 0.13.0. One row per indexed image file, with the pixel size the
-- Search and Files pages show in place of a line range. The vector itself is in
-- the store's second index, under `images/`, at CLIP's 512 dimensions.
--
-- Additive and `IF NOT EXISTS`, so `format_version` does not move: an older
-- binary opens the store, never looks in here, and searches its text exactly as
-- before. Same reasoning `docs/compatibility.md` records for the graph tables.
--
-- `id` is AUTOINCREMENT for the same reason chunks are: the id addresses a
-- vector in a sharded index, and SQLite reissuing a deleted row's id would put
-- one id inside two shards at once.
CREATE TABLE IF NOT EXISTS images (
    id      INTEGER PRIMARY KEY AUTOINCREMENT,
    file_id INTEGER NOT NULL REFERENCES files(id) ON DELETE CASCADE,
    width   INTEGER NOT NULL,
    height  INTEGER NOT NULL
);

CREATE INDEX IF NOT EXISTS images_file_id ON images(file_id);

-- The retrieval ledger, from 0.12.0. Additive and empty unless recording is
-- switched on, which it is not by default.
--
-- `prev` is the hash of the row before it and `hash` covers this row's own
-- fields plus `prev`, so the table is a chain: editing or deleting a row
-- breaks every hash after it and `semlith ledger` can say so. That is what
-- makes this an audit record rather than a log file, and it costs one blake3
-- of a short string per query.
--
-- `excerpt_tokens` is what the agent actually read. `whole_file_tokens` is
-- what reading those files whole would have cost — the honest denominator for
-- a savings number, measured rather than claimed.
CREATE TABLE IF NOT EXISTS retrievals (
    id                INTEGER PRIMARY KEY AUTOINCREMENT,
    at                INTEGER NOT NULL,
    client            TEXT NOT NULL,
    query             TEXT NOT NULL,
    hits              INTEGER NOT NULL,
    micros            INTEGER NOT NULL,
    excerpt_tokens    INTEGER NOT NULL,
    whole_file_tokens INTEGER NOT NULL,
    prev              TEXT NOT NULL,
    hash              TEXT NOT NULL
);

CREATE INDEX IF NOT EXISTS retrievals_at ON retrievals(at);
"#;

/// A chunk row joined with its file's path — what a search hit resolves to.
#[derive(Debug, Clone)]
pub struct ChunkRow {
    pub id: i64,
    pub path: String,
    pub ord: i64,
    pub start_line: u32,
    pub end_line: u32,
    pub text: String,
}

/// Open a store's database, laying down or upgrading its schema first.
///
/// Retried when SQLite answers busy or a short read. Several connections run
/// this at once as a matter of course — the daemon's writer, its readers, the
/// portal's index route and any CLI — and the first of them on a new store is
/// switching it to WAL while the others read it. The switch answers
/// `SQLITE_BUSY` at once rather than waiting out the busy timeout, which a
/// test with six openers hits within a few rounds. A read of a WAL file that
/// another connection shrank under it comes back short
/// (`SQLITE_IOERR_SHORT_READ`, the 522 in #132), and SQLite's own WAL recovery
/// answers that by retrying; the drive's 522 was never reproduced here, so it
/// is retried on that reasoning rather than on a measurement. Every step below
/// is idempotent, so running it again is safe. Anything else is a real error
/// and is returned.
pub fn open(path: &Path) -> Result<Connection> {
    // ponytail: a fixed two-second window, measured in time rather than
    // attempts because a step that waits out the busy timeout already took
    // five. A store still busy after that is held by something that will not
    // let go, and saying so beats hanging the caller.
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
    loop {
        match open_once(path) {
            Err(e) if transient(&e) && std::time::Instant::now() < deadline => {
                std::thread::sleep(std::time::Duration::from_millis(50));
            }
            answer => return answer,
        }
    }
}

/// Whether an error from [`open_once`] is a race another connection will
/// finish by itself.
fn transient(e: &anyhow::Error) -> bool {
    matches!(
        e.downcast_ref::<rusqlite::Error>(),
        Some(rusqlite::Error::SqliteFailure(f, _))
            if f.code == rusqlite::ErrorCode::DatabaseBusy
                || f.code == rusqlite::ErrorCode::SchemaChanged
                || f.extended_code == rusqlite::ffi::SQLITE_IOERR_SHORT_READ
    )
}

fn open_once(path: &Path) -> Result<Connection> {
    // A store this process is creating is not one somebody else made.
    let fresh = std::fs::metadata(path).map_or(true, |m| m.len() == 0);
    let db = Connection::open(path)?;
    // A store this home did not create — a `.semlith` beside a corpus, which
    // a clone or a sync can carry — is checked before anything below writes
    // to it (audit S24): once, by SQLite's own quick check, and on every
    // read by its cell-size check, so a crafted file is refused rather than
    // parsed further.
    if !fresh && foreign(path) {
        defensive(&db)?;
        db.pragma_update(None, "cell_size_check", "ON")?;
        check_once(&db, path)?;
    }
    db.pragma_update(None, "journal_mode", "WAL")?;
    db.pragma_update(None, "synchronous", "NORMAL")?;
    db.pragma_update(None, "foreign_keys", "ON")?;
    // A deleted row's bytes are overwritten when its page is freed, not left
    // readable in the file until a vacuum. A store holds whatever a file said,
    // and a file refused after it was indexed -- a secret appeared in it -- is
    // exactly the one whose old text must not linger (#148).
    db.pragma_update(None, "secure_delete", "ON")?;
    // The writer's inserts run once per chunk, symbol and edge, each through
    // the statement cache (#198); the default of 16 is shared with every
    // other cached read.
    db.set_prepared_statement_cache_capacity(64);
    defensive(&db)?;
    db.execute_batch(SCHEMA)?;
    db.execute_batch(GLOB_INDEX)?;
    add_columns(&db)?;
    check_format(&db)?;
    backfill_fts(&db)?;
    // The keyword index's own secure delete (S10), written once. Best effort
    // here: a reader that opens a store while its writer is creating it must
    // not fail on a setting it does not need, and the scrub that does need it
    // sets it under the writer (`secure_delete_fts`).
    let _ = secure_delete_fts(&db);
    // Everything above is the schema this binary needs in place before the
    // store is usable at all; from here the connection writes only when a
    // writer asks.
    read_only(&db, true)?;
    Ok(db)
}

/// Whether this store file lies outside the semlith home.
fn foreign(path: &Path) -> bool {
    match crate::home::home_or_error() {
        Ok(home) => !crate::canonical(path).starts_with(crate::canonical(&home)),
        Err(_) => true,
    }
}

/// The marker a passed check leaves beside the store.
const CHECKED: &str = ".integrity-checked";

/// SQLite's quick check, the first time this store is opened here.
// ponytail: once per store, not per open, as decided for 0.41.0; a store
// replaced after its check is trusted on the marker. Hash the file into the
// marker if that ever matters.
fn check_once(db: &Connection, path: &Path) -> Result<()> {
    let Some(dir) = path.parent() else {
        return Ok(());
    };
    let marker = dir.join(CHECKED);
    if marker.exists() {
        return Ok(());
    }
    let verdict: String = db.query_row("PRAGMA quick_check", [], |r| r.get(0))?;
    if verdict != "ok" {
        anyhow::bail!(
            "{} failed SQLite's integrity check ({verdict}), so semlith will not open it: \
             it is damaged, or was made to attack whatever reads it. Index the corpus \
             again into a new store.",
            path.display()
        );
    }
    let _ = crate::home::write_private(&marker, b"quick_check ok\n");
    Ok(())
}

/// Treat the file as data rather than as a program.
///
/// A SQLite file is a schema as well as rows, and a schema can carry views,
/// triggers and generated columns that run when the database is merely opened
/// or read. semlith opens store files it did not necessarily write — a
/// `.semlith` that a user has trusted arrived from somewhere, and `--store`
/// opens anything the user names — so both settings are on for every
/// connection:
///
/// - `trusted_schema=OFF` stops a view or a trigger calling the functions
///   SQLite marks as unsafe for a schema it does not trust;
/// - `SQLITE_DBCONFIG_DEFENSIVE` refuses writes to shadow tables and to
///   `sqlite_schema`, which is how a corrupted or crafted FTS5 index turns a
///   read into something else.
///
/// Both are cheap and neither affects a query semlith itself issues, which is
/// what `tests/measure.rs` is there to keep true.
fn defensive(db: &Connection) -> Result<()> {
    db.pragma_update(None, "trusted_schema", "OFF")?;
    db.set_db_config(rusqlite::config::DbConfig::SQLITE_DBCONFIG_DEFENSIVE, true)?;
    Ok(())
}

/// Make this connection refuse writes until a writer says otherwise.
///
/// `query_only` rather than `SQLITE_OPEN_READ_ONLY`, because semlith opens one
/// connection per store and writes through the same handle it reads through:
/// a read-only open would mean two connections per store and a second answer to
/// "who holds the writer", which is the one thing the store's design keeps
/// singular. The property is the same from the file's point of view — a
/// connection in this state cannot write, whatever it is asked to do — and it
/// is lifted only by [`Writing`], which the three write paths take.
pub fn read_only(db: &Connection, on: bool) -> Result<()> {
    db.pragma_update(None, "query_only", if on { "ON" } else { "OFF" })?;
    Ok(())
}

/// Permission to write, given back when it goes out of scope.
///
/// Taken by the three methods that write — indexing, forgetting, and recording
/// a retrieval in the ledger — so every other path through the store is a path
/// SQLite itself will refuse to write from.
pub struct Writing<'a>(&'a Connection);

impl<'a> Writing<'a> {
    pub fn begin(db: &'a Connection) -> Result<Self> {
        read_only(db, false)?;
        Ok(Self(db))
    }
}

impl Drop for Writing<'_> {
    fn drop(&mut self) {
        // Best effort on purpose: a connection that cannot be put back into
        // read-only mode is one whose next statement will fail anyway, and a
        // panic in a destructor would replace a clear error with a confusing
        // one.
        let _ = read_only(self.0, true);
    }
}

/// The store layout this binary understands.
///
/// Written into a store when one is created, and absent from every store
/// written before 0.6.0 — which is what makes 1 the right reading of an absent
/// key rather than an unknown.
///
/// 1 is a single `index.tv`. 2 is a directory of shards, written by 0.7.0 and
/// later; see [`crate::index`]. 3 is a store written by 0.22.0 and later,
/// whose Markdown chunks are cut at their headings and carry the heading path
/// as embedding context. 4 is a store written by 0.25.0 and later, whose code
/// chunks carry the definition they sit inside — its signature and the first
/// line of its doc comment — as embedding context. 5 is a store 0.39.0 or
/// later has written to: its vector files may be turbovec's v8, which a binary
/// before 0.39.0 cannot load. A binary understands every format up to its own,
/// so this one reads all five and creates the newest.
pub const FORMAT_VERSION: u32 = 5;

/// The first format whose vector files may be turbovec's v8.
///
/// Nothing about the rows, the chunks or the vectors' values changes: turbovec
/// 1.1 reads a v7 file as it is and writes it back as v8 the next time that
/// file is saved, so a store moves shard by shard rather than in one pass.
/// What changes is which binaries can read it — turbovec 1.0 refuses a v8 file
/// as "not a turbovec index" — so the format number moves before the first
/// write, and an older semlith refuses the whole store with its own message
/// instead of failing on whichever shard it loads first. See
/// [`mark_vector_format`].
pub const TURBOVEC_V8: u32 = 5;

/// The first format that keeps its vectors in shards.
pub const SHARDED_FORMAT: u32 = 2;

/// The first format whose code chunks embed the definition they sit inside.
///
/// A store below this holds code chunks embedded from their own text alone.
/// They still answer — the row, the span and the bytes are identical, and only
/// what the model was shown differs — but a store half embedded one way and
/// half the other ranks its own files against each other unevenly, so the
/// first full index pass under 0.25.0 re-embeds everything it walks. `semlith
/// stats` says which rule a store is on until it has.
pub const CODE_CONTEXT: u32 = 4;

/// The first format whose Markdown chunks are cut at their headings.
///
/// A store below this holds fixed windows and still answers: a chunk is a chunk
/// whichever rule cut it, and nothing about the row or the vector changes. What
/// changes is where the boundaries fall, so a store that is half one rule and
/// half the other would rank its own files against each other unevenly — which
/// is why the first full index pass under 0.22.0 re-chunks everything it walks
/// rather than waiting for each file to be edited. `semlith stats` says which
/// rule a store is on until it has.
pub const DEFINITION_CHUNKS: u32 = 3;

/// Which chunking rule this store's chunks were cut by, in the words `stats`
/// and the portal print.
pub fn chunking(db: &Connection) -> Result<&'static str> {
    Ok(match format(db)? {
        v if v >= CODE_CONTEXT => "headings, and code with its definition",
        v if v >= DEFINITION_CHUNKS => {
            "headings; code carries no definition until it is re-indexed"
        }
        _ => "fixed windows",
    })
}

/// Which layout this store's vectors are in.
///
/// An absent key is format 1: every store written before 0.6.0 has no key, and
/// reading it as anything else would refuse the whole installed base.
pub fn format(db: &Connection) -> Result<u32> {
    Ok(get_meta(db, FORMAT_KEY)?
        .and_then(|v| v.parse().ok())
        .unwrap_or(1))
}

pub const FORMAT_KEY: &str = "format_version";

/// Move a format-4 store to [`TURBOVEC_V8`] before this binary writes to it.
///
/// Called where every write starts, because any write may save a vector file
/// and a saved file is v8. Only format 4 moves here: a store below it re-chunks
/// on its first full pass and is stamped with the current format when that
/// pass ends, and stamping it now would skip that pass. A format-1 store has no
/// key and a single `index.tv`; stamping it would make it read as sharded.
pub fn mark_vector_format(db: &Connection) -> Result<()> {
    if format(db)? == CODE_CONTEXT {
        set_meta(db, FORMAT_KEY, &TURBOVEC_V8.to_string())?;
    }
    Ok(())
}

/// Meta key present while a store's first import is under way (#210).
///
/// Set by the pass that finds the store empty and removed by the pass that
/// ends with nothing left pending, so a first import the cloud runs in slices
/// of a few hundred files is one first import, not one slice of it.
pub const FIRST_IMPORT_KEY: &str = "first_import";

/// Meta key holding the keyword index's `automerge` setting from before a
/// first import held it at 0: a number, or `default` when the store had none.
const AUTOMERGE_HELD_KEY: &str = "fts_automerge_held";

/// FTS5's own `automerge` when its config table has no row for it.
const AUTOMERGE_DEFAULT: i64 = 4;

/// Whether this pass is part of a store's first import, starting one when the
/// store is empty.
pub fn first_import(db: &Connection, empty: bool) -> Result<bool> {
    if get_meta(db, FIRST_IMPORT_KEY)?.is_some() {
        return Ok(true);
    }
    if empty {
        set_meta(db, FIRST_IMPORT_KEY, "1")?;
    }
    Ok(empty)
}

/// The first import ran to its end: later passes write as any other does.
pub fn end_first_import(db: &Connection) -> Result<()> {
    db.execute("DELETE FROM meta WHERE k = ?1", params![FIRST_IMPORT_KEY])?;
    Ok(())
}

/// Stop FTS5 merging segments while a first import writes (#210).
///
/// Every commit flushes the keyword index's pending terms into a new segment,
/// and with `automerge` on FTS5 merges segments as part of that commit: about
/// a sixth of a remote-GPU first import's commit time. Held at 0 a merge waits
/// until `crisismerge` segments pile up at one level, which still bounds what
/// a search reads. The setting persists in the store, so what it was is kept
/// in `meta` first and [`release_automerge`] puts it back — after the pass,
/// and at the start of the next write if a crash came between.
pub fn hold_automerge(db: &Connection) -> Result<()> {
    if get_meta(db, AUTOMERGE_HELD_KEY)?.is_some() {
        return Ok(());
    }
    let was: Option<i64> = db
        .query_row(
            "SELECT v FROM chunks_fts_config WHERE k = 'automerge'",
            [],
            |r| r.get(0),
        )
        .optional()?;
    set_meta(
        db,
        AUTOMERGE_HELD_KEY,
        &was.map_or_else(|| "default".to_string(), |n| n.to_string()),
    )?;
    db.execute(
        "INSERT INTO chunks_fts(chunks_fts, rank) VALUES('automerge', 0)",
        [],
    )?;
    Ok(())
}

/// Put back the `automerge` setting [`hold_automerge`] recorded, if any.
///
/// A store that had no row gets none back: the value is set through FTS5 so
/// every connection's cached copy is the default again, then the row is
/// removed with the connection's defensive mode lifted for that one statement,
/// since the config table is FTS5's and refuses ordinary writes.
pub fn release_automerge(db: &Connection) -> Result<()> {
    let Some(was) = get_meta(db, AUTOMERGE_HELD_KEY)? else {
        return Ok(());
    };
    let value = was.parse().unwrap_or(AUTOMERGE_DEFAULT);
    db.execute(
        "INSERT INTO chunks_fts(chunks_fts, rank) VALUES('automerge', ?1)",
        params![value],
    )?;
    if was == "default" {
        use rusqlite::config::DbConfig::SQLITE_DBCONFIG_DEFENSIVE;
        db.set_db_config(SQLITE_DBCONFIG_DEFENSIVE, false)?;
        let removed = db.execute("DELETE FROM chunks_fts_config WHERE k = 'automerge'", []);
        db.set_db_config(SQLITE_DBCONFIG_DEFENSIVE, true)?;
        removed?;
    }
    db.execute("DELETE FROM meta WHERE k = ?1", params![AUTOMERGE_HELD_KEY])?;
    Ok(())
}

/// Refuse a store a later semlith wrote, rather than misreading it.
///
/// Nothing is written here. A store from before this key existed is format 1
/// and stays exactly as it is: a read command that quietly rewrote the store
/// it was only asked to search would be a worse surprise than the one this
/// guards against.
fn check_format(db: &Connection) -> Result<()> {
    let Some(recorded) = get_meta(db, FORMAT_KEY)? else {
        return Ok(());
    };
    let found: u32 = recorded.parse().map_err(|_| {
        anyhow::anyhow!("store records {FORMAT_KEY} {recorded:?}, which is not a version number")
    })?;
    if found > FORMAT_VERSION {
        anyhow::bail!(
            "store is format {found}, and semlith {} understands format {FORMAT_VERSION}; \
             upgrade semlith to read this store",
            env!("CARGO_PKG_VERSION"),
        );
    }
    Ok(())
}

/// Meta key recording that the keyword index has been built for this store.
///
/// Counting rows in `chunks_fts` cannot answer this. An external-content FTS5
/// table reads through to its content table, so `COUNT(*)` returns the number
/// of chunks whether or not a single term has been indexed — a guard built on
/// it would skip the backfill on exactly the stores that need it.
const FTS_BUILT: &str = "fts_built";

/// Populate the keyword index for a store written before it existed.
///
/// A 0.1.0 store has chunks but no FTS index, and the triggers only fire on new
/// writes. Rebuilding from the text already in SQLite costs no embedding and
/// leaves the vectors untouched.
/// Columns added to tables that already exist in a store written by an older
/// binary.
///
/// `CREATE TABLE IF NOT EXISTS` is how every table here arrives, and it does
/// nothing at all to a table that is already there — so a column added to the
/// schema above reaches a fresh store and no other. This is the other half:
/// one `ALTER TABLE ADD COLUMN` per column, run on every open, skipped when
/// the column is present.
///
/// Every column here must be nullable with no default, which is what makes the
/// operation an O(1) catalogue edit rather than a table rewrite, and what lets
/// an older binary keep reading the store afterwards: it selects by name and
/// never sees them. That is the whole reason `format_version` does not move —
/// the same reasoning `docs/compatibility.md` records for the graph tables.
fn add_columns(db: &Connection) -> Result<()> {
    const ADDITIONS: [(&str, &str, &str); 24] = [
        ("edges", "hint", "TEXT"),
        // 0.16.0: the line the reference was written on.
        ("edges", "line", "INTEGER"),
        // The ledger's 0.15.0 columns. `session` groups the retrievals of one
        // agent conversation, `tool` says which tool was asked, `stale_hits`
        // counts the answers that came from a file edited since it was
        // indexed, and `tokenizer` names what counted the two token figures so
        // rows counted two different ways are never summed together.
        ("retrievals", "session", "TEXT"),
        ("retrievals", "tool", "TEXT"),
        ("retrievals", "stale_hits", "INTEGER"),
        ("retrievals", "tokenizer", "TEXT"),
        // 0.20.2: the id one retrieval shares across every store that
        // contributed to it. A search over six stores used to write six rows
        // and the Ledger page counted six retrievals, so every figure on it
        // scaled with how many stores happened to be open. The rows stay per
        // store — that is what keeps each store's chain verifiable and its
        // token counts its own — and this is what makes them one retrieval
        // again when they are counted.
        ("retrievals", "query_id", "TEXT"),
        // 0.33.0: the client's own version, beside the name `clientid` resolved.
        ("retrievals", "client_version", "TEXT"),
        // 0.34.0: the model request that made the call, read afterwards from
        // the client's own session log (`crate::usage`), and what it cost.
        // Outside the chain like `client_version`: they are filled in after
        // the row is written, which a hashed column could never allow.
        // `usage_source` is the log they came from, or why there is none;
        // NULL means not looked for yet.
        ("retrievals", "model", "TEXT"),
        ("retrievals", "input_tokens", "INTEGER"),
        ("retrievals", "output_tokens", "INTEGER"),
        ("retrievals", "cache_read_tokens", "INTEGER"),
        ("retrievals", "cache_write_tokens", "INTEGER"),
        ("retrievals", "reasoning_tokens", "INTEGER"),
        ("retrievals", "cost_usd", "REAL"),
        ("retrievals", "cost_source", "TEXT"),
        ("retrievals", "usage_source", "TEXT"),
        // 0.37.0: when ledger sync sent the row to the org's ledger, in unix
        // seconds; NULL is not sent. Outside the chain like the usage
        // columns: it is written after the row, which a hashed column could
        // never allow.
        ("retrievals", "synced_at", "INTEGER"),
        // 0.25.0: what the parser made of this file — "parsed", "timeout" or
        // "none" for a language that carries no grammar. Without it a file
        // with no definitions and a file the parser gave up on look the same
        // in the per-language coverage table, which is the one question that
        // table exists to answer.
        ("files", "graph", "TEXT"),
        // 0.34.0: the file's units — PDF pages, slides, non-empty spreadsheet
        // cells or notebook cells, by its format (`chunk::COUNTED`). NULL is
        // "not counted": a format with no unit, or a file an older binary
        // wrote, which the next index pass counts without re-embedding it.
        ("files", "units", "INTEGER"),
        // 0.40.0: the file's mtime in nanoseconds when it was read, which is
        // what freshness compares now (see `Stamp::fresh`). NULL on a row an
        // older binary wrote; an older binary ignores the column.
        ("files", "read_mtime", "INTEGER"),
        // 0.42.0: what the whole-file side of a row was counted from —
        // `bytes` (the file's size on disk, every row before 0.42.0, NULL
        // here) or `text` (the text a reader extracted, for a PDF, an Office
        // file, an image). The Ledger says which, row by row.
        ("retrievals", "whole_basis", "TEXT"),
        // 0.42.0: which chain formula hashed the row. NULL is the 0.15.0/0.41.0
        // formula (or 0.14.0's, told apart by `tool`), 2 is chain v2, which
        // covers every column a figure reads.
        ("retrievals", "chain", "INTEGER"),
        // 0.42.0: the hash over a v2 row's late columns (usage, synced_at),
        // rewritten whenever they are, so a late column cannot change without
        // the verify noticing.
        ("retrievals", "seal", "TEXT"),
    ];
    for (table, column, kind) in ADDITIONS {
        if has_column(db, table, column)? {
            continue;
        }
        // Asked again under the write lock. Two connections opening the same
        // new store both see the column missing, and the second `ALTER` then
        // fails the whole open with "duplicate column name" (#132).
        let tx =
            rusqlite::Transaction::new_unchecked(db, rusqlite::TransactionBehavior::Immediate)?;
        if !has_column(&tx, table, column)? {
            tx.execute_batch(&format!("ALTER TABLE {table} ADD COLUMN {column} {kind};"))?;
        }
        tx.commit()?;
    }
    Ok(())
}

fn has_column(db: &Connection, table: &str, column: &str) -> Result<bool> {
    let mut stmt = db.prepare(&format!("PRAGMA table_info({table})"))?;
    let mut rows = stmt.query([])?;
    while let Some(row) = rows.next()? {
        let name: String = row.get(1)?;
        if name == column {
            return Ok(true);
        }
    }
    Ok(false)
}

fn backfill_fts(db: &Connection) -> Result<()> {
    if get_meta(db, FTS_BUILT)?.is_some() {
        return Ok(());
    }
    // Asked again under the write lock, as in `add_columns`: a second
    // connection opening the store at the same moment would rebuild it again.
    let tx = rusqlite::Transaction::new_unchecked(db, rusqlite::TransactionBehavior::Immediate)?;
    if get_meta(&tx, FTS_BUILT)?.is_none() {
        // FTS5's own command for external content: discard the index and
        // rebuild it from the content table.
        tx.execute_batch("INSERT INTO chunks_fts(chunks_fts) VALUES('rebuild');")?;
        set_meta(&tx, FTS_BUILT, "1")?;
    }
    tx.commit()?;
    Ok(())
}

/// `AND`ed groups of `OR`ed `GLOB` patterns as a SQL fragment, plus the values
/// to bind. An empty group list yields `1` — a predicate that selects
/// everything, so callers need no special case.
///
/// Patterns are matched against `lower(files.path)`: filtering is
/// case-insensitive, because `README.MD` and `readme.md` are the same file to
/// anyone typing `--ext md`, and no filesystem this runs on treats a path's
/// case as load-bearing.
fn glob_predicate(groups: &[Vec<String>]) -> (String, Vec<String>) {
    let mut binds = Vec::new();
    let mut clauses = Vec::new();
    for group in groups {
        // A pattern marked with a leading `!` by `filter::anchor` excludes.
        // Exclusions apply after the inclusions of their own kind, which is
        // what keeps them inside this group rather than becoming a group of
        // their own: `--path 'src/**' --path '!src/vendor/**'` is one
        // requirement, not two intersecting ones.
        let (excluded, included): (Vec<&String>, Vec<&String>) =
            group.iter().partition(|p| p.starts_with('!'));
        let placeholder = format!("{GLOB_PATH} GLOB ?");
        // Binds are positional, so they are pushed in the order the clause
        // below writes their placeholders — inclusions first — rather than in
        // the order the user typed them.
        let ors = included
            .into_iter()
            .map(|p| match literal_dir(p) {
                // The range is what the index answers; the GLOB stays, so the
                // rows are exactly the pattern's.
                Some((from, to)) => {
                    binds.extend([from, to, p.clone()]);
                    format!("({GLOB_PATH} >= ? AND {GLOB_PATH} < ? AND {placeholder})")
                }
                None => {
                    binds.push(p.clone());
                    placeholder.clone()
                }
            })
            .collect::<Vec<_>>()
            .join(" OR ");
        let nots = vec![placeholder; excluded.len()].join(" OR ");
        binds.extend(excluded.into_iter().map(|p| p[1..].to_string()));

        clauses.push(match (ors.is_empty(), nots.is_empty()) {
            (true, true) => continue,
            (false, true) => format!("({ors})"),
            // Exclusions with nothing beside them mean everything except.
            (true, false) => format!("(NOT ({nots}))"),
            (false, false) => format!("(({ors}) AND NOT ({nots}))"),
        });
    }
    if clauses.is_empty() {
        return ("1".to_string(), binds);
    }
    (clauses.join(" AND "), binds)
}

/// Run `work` over `conns`, interrupting their queries once `limit` passes.
///
/// For the reads a person asks for on a store of any size -- the Graph page's
/// drawing over every edge of a 879k-chunk store did not answer in twelve
/// minutes and held a worker the whole time (#173). An interrupted query fails
/// with SQLite's "interrupted", which [`is_interrupted`] recognises, so the
/// caller can say the request was too large rather than that it broke.
pub fn bounded<T>(
    conns: &[&Connection],
    limit: std::time::Duration,
    work: impl FnOnce() -> Result<T>,
) -> Result<T> {
    let handles: Vec<rusqlite::InterruptHandle> =
        conns.iter().map(|c| c.get_interrupt_handle()).collect();
    let (done, wait) = std::sync::mpsc::channel::<()>();
    let timer = std::thread::spawn(move || {
        if wait.recv_timeout(limit) == Err(std::sync::mpsc::RecvTimeoutError::Timeout) {
            for handle in &handles {
                handle.interrupt();
            }
        }
    });
    let out = work();
    let _ = done.send(());
    let _ = timer.join();
    out
}

/// Whether an error is a query [`bounded`] interrupted.
pub fn is_interrupted(e: &anyhow::Error) -> bool {
    e.chain().any(|cause| {
        cause
            .downcast_ref::<rusqlite::Error>()
            .and_then(rusqlite::Error::sqlite_error_code)
            == Some(rusqlite::ErrorCode::OperationInterrupted)
    })
}

/// The highest chunk id, or 0 for none. One step down the rowid b-tree.
pub fn newest_chunk(db: &Connection) -> Result<i64> {
    Ok(db.query_row("SELECT COALESCE(MAX(id), 0) FROM chunks", [], |r| r.get(0))?)
}

/// Ids of every chunk whose file matches `groups`.
///
/// This is the single source of the subset. The vector index gets these ids as
/// an allowlist and [`keyword_search`] applies the same predicate, so the two
/// halves of a hybrid search can never rank a chunk the other was forbidden to
/// see.
pub fn filtered_chunk_ids(db: &Connection, groups: &[Vec<String>]) -> Result<Vec<u64>> {
    let (predicate, binds) = glob_predicate(groups);
    // Files first, then their chunks through `chunks_file_id`. A join planned
    // the other way scanned every chunk and tested its file's path, 250-680 ms
    // on the 879k-chunk corpus; this is 24-48 ms for the same rows (#170).
    let sql = format!(
        "SELECT id FROM chunks WHERE file_id IN (SELECT f.id FROM files f WHERE {predicate})"
    );
    let mut stmt = db.prepare(&sql)?;
    let args = binds.into_iter().map(Value::Text);
    let rows = stmt.query_map(rusqlite::params_from_iter(args), |r| r.get::<_, i64>(0))?;
    Ok(rows
        .collect::<Result<Vec<i64>, _>>()?
        .into_iter()
        .map(|i| i as u64)
        .collect())
}

/// How many indexed files `groups` selects.
///
/// Zero means the filter itself is wrong — a typo in a glob, a subsystem that
/// was never indexed — which is a different thing to tell a user than "the
/// corpus does not contain this".
pub fn matching_files(db: &Connection, groups: &[Vec<String>]) -> Result<i64> {
    let (predicate, binds) = glob_predicate(groups);
    let sql = format!("SELECT COUNT(*) FROM files f WHERE {predicate}");
    let args = binds.into_iter().map(Value::Text);
    Ok(db.query_row(&sql, rusqlite::params_from_iter(args), |r| r.get(0))?)
}

/// The keyword statement for a search with no filter.
///
/// Byte-identical to the one 0.2.0 issued: no join to `chunks` and `files`, so
/// a query that uses no filter pays nothing for the filtering feature. A test
/// pins the text, because the cheapest way to regress the common path is to
/// "tidy" the two branches into one.
const UNFILTERED_SQL: &str =
    "SELECT rowid FROM chunks_fts WHERE chunks_fts MATCH ? ORDER BY rank LIMIT ?";

/// Chunk ids matching `query` as keywords, best first, restricted to files
/// matching `groups`.
///
/// The query is reduced to bare terms before it reaches FTS5. A raw query would
/// be parsed as FTS5 syntax, where a stray quote or `*` is a syntax error and a
/// bare `AND` is an operator — a search for "index AND search" would silently
/// mean something the user did not type, and a search for `foo(` would fail
/// outright.
/// English words too common to say what a chunk is about.
const STOPWORDS: &[&str] = &[
    "a", "an", "and", "are", "as", "at", "be", "but", "by", "can", "do", "does", "for", "from",
    "has", "have", "how", "i", "if", "in", "into", "is", "it", "its", "of", "on", "or", "not",
    "no", "so", "that", "the", "their", "then", "there", "these", "this", "to", "was", "were",
    "what", "when", "where", "which", "who", "why", "will", "with", "you", "your", "we", "our",
    "they", "them", "he", "she", "my", "me", "than", "also", "any", "all",
];

pub fn keyword_search(
    db: &Connection,
    query: &str,
    limit: usize,
    groups: &[Vec<String>],
) -> Result<Vec<u64>> {
    let Some(match_expr) = match_expr(query) else {
        return Ok(Vec::new());
    };
    if groups.is_empty() {
        let mut stmt = db.prepare(UNFILTERED_SQL)?;
        let rows = stmt.query_map(params![match_expr, limit as i64], |r| r.get::<_, i64>(0))?;
        return Ok(rows
            .collect::<Result<Vec<i64>, _>>()?
            .into_iter()
            .map(|i| i as u64)
            .collect());
    }
    let allowed: std::collections::HashSet<u64> =
        filtered_chunk_ids(db, groups)?.into_iter().collect();
    keyword_search_within(db, query, limit, &allowed)
}

/// The FTS5 expression for `query`: its words, quoted, OR-ed. `None` when it
/// has no word at all.
fn match_expr(query: &str) -> Option<String> {
    let words: Vec<&str> = query
        .split(|c: char| !c.is_alphanumeric() && c != '_')
        .filter(|t| !t.is_empty())
        .collect();
    // Stopwords out of the MATCH: an OR over "the" and "of" asks FTS5 to rank
    // nearly every chunk of a large store. On the 879k-chunk corpus that took
    // the keyword list from 884 to 226 ms at the median, and the 727-question
    // benchmark scored the same either way (408 against 407 scoped, 340 and
    // 340 unscoped, development split). A query of nothing but stopwords keeps
    // them all, so it still finds something.
    let content: Vec<&str> = words
        .iter()
        .copied()
        .filter(|w| !STOPWORDS.contains(&w.to_ascii_lowercase().as_str()))
        .collect();
    let kept = if content.is_empty() { &words } else { &content };
    let terms: Vec<String> = kept.iter().map(|t| format!("\"{t}\"")).collect();
    (!terms.is_empty()).then(|| terms.join(" OR "))
}

/// The keyword list inside a set of chunk ids, for a filtered search that has
/// the set already.
///
/// Every match is read with its rank and the set is applied here. Joining the
/// matches to their files in SQL stopped FTS5 applying its own `LIMIT`, so it
/// ranked and joined every match: 81-688 ms on the 879k-chunk corpus against
/// 14 ms unfiltered, which is most of why a scoped search was slower than an
/// unscoped one (#170). Reading the matches is the 14 ms, the filter well
/// under one, and the order is the same.
pub fn keyword_search_within(
    db: &Connection,
    query: &str,
    limit: usize,
    allowed: &std::collections::HashSet<u64>,
) -> Result<Vec<u64>> {
    let Some(match_expr) = match_expr(query) else {
        return Ok(Vec::new());
    };
    let mut stmt = db.prepare("SELECT rowid, rank FROM chunks_fts WHERE chunks_fts MATCH ?")?;
    let rows = stmt.query_map(params![match_expr], |r| {
        Ok((r.get::<_, i64>(0)? as u64, r.get::<_, f64>(1)?))
    })?;
    let mut kept: Vec<(u64, f64)> = Vec::new();
    for row in rows {
        let (id, rank) = row?;
        if allowed.contains(&id) {
            kept.push((id, rank));
        }
    }
    // FTS5's rank is ascending-is-better; ties by id, as the index gives them.
    kept.sort_by(|a, b| a.1.total_cmp(&b.1).then(a.0.cmp(&b.0)));
    Ok(kept.into_iter().take(limit).map(|(id, _)| id).collect())
}

pub fn get_meta(db: &Connection, k: &str) -> Result<Option<String>> {
    Ok(db
        .query_row("SELECT v FROM meta WHERE k = ?1", params![k], |r| r.get(0))
        .optional()?)
}

pub fn set_meta(db: &Connection, k: &str, v: &str) -> Result<()> {
    db.execute(
        "INSERT INTO meta (k, v) VALUES (?1, ?2)
         ON CONFLICT(k) DO UPDATE SET v = excluded.v",
        params![k, v],
    )?;
    Ok(())
}

/// Content hash recorded for `path`, if we have indexed it before.
pub fn file_hash(db: &Connection, path: &str) -> Result<Option<String>> {
    Ok(db
        .query_row(
            "SELECT hash FROM files WHERE path = ?1",
            params![path],
            |r| r.get(0),
        )
        .optional()?)
}

/// Every stored path's recorded hash, read once for a whole index pass so its
/// prepare stage can tell an unchanged file without the database.
pub fn all_hashes(db: &Connection) -> Result<std::collections::HashMap<String, String>> {
    let mut stmt = db.prepare("SELECT path, hash FROM files")?;
    let rows = stmt.query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))?;
    Ok(rows.collect::<Result<_, _>>()?)
}

/// Drop a file and its chunks, returning the chunk ids so the caller can
/// evict them from the vector index too.
/// The ids of a file's chunks.
pub fn chunk_ids_of(db: &Connection, path: &str) -> Result<Vec<u64>> {
    let mut stmt = db.prepare_cached(
        "SELECT c.id FROM chunks c JOIN files f ON f.id = c.file_id WHERE f.path = ?1",
    )?;
    let rows = stmt.query_map(params![path], |r| r.get::<_, i64>(0))?;
    Ok(rows
        .collect::<Result<Vec<i64>, _>>()?
        .into_iter()
        .map(|i| i as u64)
        .collect())
}

pub fn delete_file(db: &Connection, path: &str, now: i64) -> Result<Vec<u64>> {
    retire_symbols(db, path, now)?;
    delete_file_unretired(db, path)
}

/// [`delete_file`] without keeping the file's definitions as history: for a
/// stopped run's undo, whose files were never the store's to remember.
pub fn delete_file_unretired(db: &Connection, path: &str) -> Result<Vec<u64>> {
    let ids: Vec<u64> = {
        let mut stmt = db.prepare_cached(
            "SELECT c.id FROM chunks c JOIN files f ON f.id = c.file_id WHERE f.path = ?1",
        )?;
        let rows = stmt.query_map(params![path], |r| r.get::<_, i64>(0))?;
        rows.collect::<Result<Vec<i64>, _>>()?
            .into_iter()
            .map(|i| i as u64)
            .collect()
    };
    db.prepare_cached("DELETE FROM files WHERE path = ?1")?
        .execute(params![path])?;
    Ok(ids)
}

/// Copy a file's current definitions into `symbols_past` before the file row
/// takes them with it.
///
/// Stamped with the hash the file has right now, which is the hash they were
/// extracted from: `delete_file` runs before the new version is inserted, so
/// `files.hash` here is still the old content's.
///
/// A file with no symbols copies nothing, and a store opened by a binary that
/// never wrote this table simply has none to copy.
fn retire_symbols(db: &Connection, path: &str, now: i64) -> Result<()> {
    db.prepare_cached(
        "INSERT INTO symbols_past
             (path, kind, name, qualified, start_line, end_line, content_hash, retired_at)
         SELECT f.path, s.kind, s.name, s.qualified, s.start_line, s.end_line, f.hash, ?2
         FROM symbols s JOIN files f ON f.id = s.file_id
         WHERE f.path = ?1",
    )?
    .execute(params![path, now])?;
    Ok(())
}

/// What a symbol used to be, newest first.
///
/// The live definitions are what `symbols_named` answers; these are the ones a
/// re-index replaced, each with the content hash of the file version it was
/// true for. An empty list means the store has never seen this name change --
/// or has never re-indexed since 0.23.0, which is the same answer from the
/// outside and is why `semlith stats` says whether the store keeps history.
pub fn symbols_past_named(db: &Connection, name: &str, limit: usize) -> Result<Vec<PastSymbol>> {
    let mut stmt = db.prepare(
        "SELECT path, kind, name, qualified, start_line, end_line, content_hash, retired_at
         FROM symbols_past WHERE name = ?1 ORDER BY retired_at DESC, id DESC LIMIT ?2",
    )?;
    let rows = stmt.query_map(params![name, limit as i64], |r| {
        Ok(PastSymbol {
            path: r.get(0)?,
            kind: r.get(1)?,
            name: r.get(2)?,
            qualified: r.get(3)?,
            start_line: r.get(4)?,
            end_line: r.get(5)?,
            content_hash: r.get(6)?,
            retired_at: r.get(7)?,
            store: None,
        })
    })?;
    Ok(rows.collect::<Result<Vec<_>, _>>()?)
}

/// How many definitions this store has retired, over all names.
pub fn symbols_past_count(db: &Connection) -> Result<i64> {
    Ok(db.query_row("SELECT COUNT(*) FROM symbols_past", [], |r| r.get(0))?)
}

/// Drop the definitions retired before `before` (unix seconds). The retention
/// is the caller's; see `compact`.
pub fn prune_history(db: &Connection, before: i64) -> Result<usize> {
    Ok(db.execute(
        "DELETE FROM symbols_past WHERE retired_at < ?1",
        params![before],
    )?)
}

/// How many definitions were retired before `before`.
pub fn history_count_before(db: &Connection, before: i64) -> Result<i64> {
    Ok(db.query_row(
        "SELECT COUNT(*) FROM symbols_past WHERE retired_at < ?1",
        params![before],
        |r| r.get(0),
    )?)
}

/// Every chunk row, durable or not: what the sidecar can hold a record for.
pub fn chunk_count(db: &Connection) -> Result<i64> {
    Ok(db.query_row("SELECT COUNT(*) FROM chunks", [], |r| r.get(0))?)
}

/// Roughly what the definitions retired before `before` take: their text plus
/// a row's fixed cost. An estimate for `stats`, not an accounting — SQLite
/// packs rows into pages and the count cannot see that.
pub fn history_bytes_before(db: &Connection, before: i64) -> Result<i64> {
    Ok(db.query_row(
        "SELECT COALESCE(SUM(LENGTH(path) + LENGTH(kind) + LENGTH(name) + LENGTH(qualified)
                             + LENGTH(content_hash) + 32), 0)
         FROM symbols_past WHERE retired_at < ?1",
        params![before],
        |r| r.get(0),
    )?)
}

/// Pages SQLite holds free inside the file, in bytes: what a vacuum gives back
/// before anything else is dropped.
pub fn free_bytes(db: &Connection) -> Result<i64> {
    let pages: i64 = db.query_row("PRAGMA freelist_count", [], |r| r.get(0))?;
    let size: i64 = db.query_row("PRAGMA page_size", [], |r| r.get(0))?;
    Ok(pages * size)
}

/// Give the database's dead pages back to the filesystem.
///
/// FTS5 keeps a deleted chunk's terms in its segments until they are merged,
/// so the keyword index is optimised first; then the file is rebuilt, and the
/// write-ahead log, which still holds the pages as they were, is checkpointed
/// and truncated. A reader still inside a snapshot keeps the log from being
/// truncated, and that is not an error: the next checkpoint does it.
///
/// The connection must be writable and outside any transaction.
pub fn reclaim(db: &Connection) -> Result<()> {
    db.execute("INSERT INTO chunks_fts(chunks_fts) VALUES('optimize')", [])?;
    db.execute_batch("VACUUM")?;
    db.query_row("PRAGMA wal_checkpoint(TRUNCATE)", [], |_| Ok(()))?;
    Ok(())
}

/// A definition a re-index replaced.
#[derive(Debug, Clone, serde::Serialize)]
pub struct PastSymbol {
    #[serde(serialize_with = "crate::serialize_plain")]
    pub path: String,
    pub kind: String,
    pub name: String,
    pub qualified: String,
    pub start_line: u32,
    pub end_line: u32,
    /// The content hash of the file version this definition belonged to.
    pub content_hash: String,
    pub retired_at: i64,
    /// Which store this came from, set only when more than one was searched.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub store: Option<String>,
}

pub fn insert_file(db: &Connection, path: &str, hash: &str, bytes: u64, now: i64) -> Result<i64> {
    db.prepare_cached("INSERT INTO files (path, hash, bytes, indexed_at) VALUES (?1, ?2, ?3, ?4)")?
        .execute(params![path, hash, bytes as i64, now])?;
    Ok(db.last_insert_rowid())
}

/// Record what the parser made of a file: `parsed`, `timeout`, or `none` for
/// a language semlith carries no grammar for.
///
/// Written beside the file row rather than derived later, because the two
/// failures it separates are invisible afterwards: a file whose parse expired
/// and a file whose language has no grammar both arrive at the store with no
/// symbols at all.
pub fn set_file_graph(db: &Connection, file_id: i64, state: &str) -> Result<()> {
    db.prepare_cached("UPDATE files SET graph = ?1 WHERE id = ?2")?
        .execute(params![state, file_id])?;
    Ok(())
}

/// Record a file's units (see `chunk::COUNTED`), by row id or, for a file
/// whose row is already there, by path.
pub fn set_file_units(db: &Connection, file_id: i64, units: i64) -> Result<()> {
    db.prepare_cached("UPDATE files SET units = ?1 WHERE id = ?2")?
        .execute(params![units, file_id])?;
    Ok(())
}

pub fn set_units_by_path(db: &Connection, path: &str, units: i64) -> Result<()> {
    db.execute(
        "UPDATE files SET units = ?1 WHERE path = ?2",
        params![units, path],
    )?;
    Ok(())
}

/// Stored files of a counted format whose units were never recorded — written
/// by a binary older than 0.34.0. The index pass reads each one again for its
/// count alone.
pub fn uncounted(db: &Connection) -> Result<std::collections::HashSet<String>> {
    let mut stmt = db.prepare("SELECT path FROM files WHERE units IS NULL")?;
    let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
    let mut out = std::collections::HashSet::new();
    for path in rows {
        let path = path?;
        if crate::chunk::is_counted(std::path::Path::new(&path)) {
            out.insert(path);
        }
    }
    Ok(out)
}

/// What the graph covers, per language, read from the rows themselves.
#[derive(Debug, Clone, Default, serde::Serialize)]
pub struct LanguageCoverage {
    pub language: String,
    /// Files of this language the store holds.
    pub files: usize,
    /// Files whose parse expired, so the store holds their text and none of
    /// their structure.
    pub parser_failed: usize,
    pub definitions: usize,
    /// Call edges whose target was found in the file that made the call.
    pub extracted: usize,
    /// Call edges whose target name has exactly one definition in the store.
    pub resolved: usize,
    /// Call edges whose target name has several, and nothing here says which.
    pub ambiguous: usize,
    /// Call edges naming something no definition in this store satisfies —
    /// the standard library, a dependency nobody indexed, a typo.
    pub unresolved: usize,
}

impl LanguageCoverage {
    /// The share of this language's answerable call edges that settled on one
    /// definition, which is the figure the README quotes for the whole store.
    #[must_use]
    pub fn settled_share(&self) -> usize {
        let answerable = self.extracted + self.resolved + self.ambiguous;
        if answerable == 0 {
            return 0;
        }
        (self.extracted + self.resolved) * 100 / answerable
    }
}

/// Per-language coverage for every language the store holds.
///
/// Sorted by files descending then by name, so the language that dominates a
/// corpus is the first line a reader sees and two runs over one store print
/// the same table.
///
/// The edge classes are the same four the resolver reports, decided the same
/// way: `extracted` is stored on the edge, and an `inferred` edge is resolved,
/// ambiguous or unresolved according to how many definitions of its target
/// name this store holds. A second rule here would be a second answer to "what
/// share resolves", and the README quotes this one.
pub fn coverage_by_language(db: &Connection) -> Result<Vec<LanguageCoverage>> {
    cached_by_graph(db, "coverage", || coverage_by_language_uncached(db))
}

fn coverage_by_language_uncached(db: &Connection) -> Result<Vec<LanguageCoverage>> {
    use std::collections::{BTreeMap, HashMap};
    let mut by_language: BTreeMap<String, LanguageCoverage> = BTreeMap::new();
    // Each file's language worked out once, by id. The edge pass below used
    // to parse a path per edge: three million of them on the 879k-chunk
    // corpus, which made the Index health report take half a minute.
    let mut language_of: HashMap<i64, String> = HashMap::new();

    let mut stmt = db.prepare("SELECT id, path, graph FROM files")?;
    let rows = stmt.query_map([], |r| {
        Ok((
            r.get::<_, i64>(0)?,
            r.get::<_, String>(1)?,
            r.get::<_, Option<String>>(2)?,
        ))
    })?;
    for row in rows {
        let (id, path, graph) = row?;
        let language = crate::graph::language_of(std::path::Path::new(&path))
            .unwrap_or("other")
            .to_string();
        let entry = by_language.entry(language.clone()).or_default();
        entry.files += 1;
        if graph.as_deref() == Some("timeout") {
            entry.parser_failed += 1;
        }
        language_of.insert(id, language);
    }
    let lang = |id: i64| {
        language_of
            .get(&id)
            .cloned()
            .unwrap_or_else(|| "other".to_string())
    };

    let mut stmt = db.prepare("SELECT file_id, COUNT(*) FROM symbols GROUP BY file_id")?;
    let rows = stmt.query_map([], |r| Ok((r.get::<_, i64>(0)?, r.get::<_, i64>(1)?)))?;
    for row in rows {
        let (file, count) = row?;
        by_language.entry(lang(file)).or_default().definitions +=
            usize::try_from(count).unwrap_or(0);
    }

    // Counted per file, whether the extractor settled the edge, and how many
    // definitions its target has -- none, one or several, which is all the
    // row needs, so the count stops at two.
    //
    // A call through a receiver (`edges.hint` starting `.`) settles only on a
    // method of the receiver's stated type, as `store::through_receiver`
    // decides, so for one of those `typed` counts the definitions owned by
    // that type: one settles it, anything else is ambiguous. `-1` is any
    // other edge.
    let mut stmt = db.prepare(
        "SELECT s.file_id, e.confidence = 'extracted',
                (SELECT COUNT(*) FROM (SELECT 1 FROM symbols d WHERE d.name = e.dst LIMIT 2)) AS defs,
                CASE WHEN substr(e.hint, 1, 1) = '.' THEN
                  CASE WHEN substr(e.hint, 2, 1) BETWEEN 'A' AND 'Z' THEN
                    (SELECT COUNT(*) FROM (SELECT 1 FROM symbols d WHERE d.name = e.dst
                       AND d.qualified = substr(e.hint, 2) || '::' || e.dst LIMIT 2))
                  ELSE 0 END
                ELSE -1 END AS typed,
                COUNT(*)
         FROM edges e JOIN symbols s ON s.id = e.src
         WHERE e.kind = 'calls'
         GROUP BY s.file_id, 2, defs, typed",
    )?;
    let rows = stmt.query_map([], |r| {
        Ok((
            r.get::<_, i64>(0)?,
            r.get::<_, bool>(1)?,
            r.get::<_, i64>(2)?,
            r.get::<_, i64>(3)?,
            r.get::<_, i64>(4)?,
        ))
    })?;
    for row in rows {
        let (file, extracted, definitions, typed, count) = row?;
        let count = usize::try_from(count).unwrap_or(0);
        let entry = by_language.entry(lang(file)).or_default();
        if typed >= 0 {
            match (definitions, typed) {
                (0, _) => entry.unresolved += count,
                (_, 1) if extracted => entry.extracted += count,
                (_, 1) => entry.resolved += count,
                _ => entry.ambiguous += count,
            }
        } else if extracted {
            entry.extracted += count;
        } else {
            match definitions {
                0 => entry.unresolved += count,
                1 => entry.resolved += count,
                _ => entry.ambiguous += count,
            }
        }
    }

    let mut coverage: Vec<LanguageCoverage> = by_language
        .into_iter()
        .map(|(language, mut row)| {
            row.language = language;
            row
        })
        .collect();
    coverage.sort_by(|a, b| {
        b.files
            .cmp(&a.files)
            .then_with(|| a.language.cmp(&b.language))
    });
    Ok(coverage)
}

pub fn insert_chunk(
    db: &Connection,
    file_id: i64,
    ord: usize,
    start_line: u32,
    end_line: u32,
    text: &str,
) -> Result<i64> {
    db.prepare_cached(
        "INSERT INTO chunks (file_id, ord, start_line, end_line, text)
         VALUES (?1, ?2, ?3, ?4, ?5)",
    )?
    .execute(params![file_id, ord as i64, start_line, end_line, text])?;
    let id = db.last_insert_rowid();
    db.prepare_cached("INSERT INTO chunks_fts(rowid, text) VALUES (?1, ?2)")?
        .execute(params![id, text])?;
    Ok(id)
}

/// Drop the insert trigger an older semlith's schema puts back when it opens
/// this store, inside the writer's transaction so it cannot come back before
/// the commit. With it in place every chunk would be indexed twice.
pub fn drop_insert_trigger(db: &Connection) -> Result<()> {
    let present: bool = db
        .prepare_cached(
            "SELECT EXISTS (SELECT 1 FROM sqlite_schema
             WHERE type = 'trigger' AND name = 'chunks_fts_insert')",
        )?
        .query_row([], |r| r.get(0))?;
    if present {
        db.execute_batch("DROP TRIGGER chunks_fts_insert")?;
    }
    Ok(())
}

/// Indexed files whose path ends with `suffix`, most specific first.
///
/// A locate answer prints a store-relative path and a user types one, while
/// `files.path` is absolute. Rather than teach every caller to rebuild the
/// absolute form — which needs the store's roots and gets it wrong for a file
/// reached through a symlink — the suffix is matched against what was actually
/// indexed. `/` is prepended so `one.rs` cannot match `alone.rs`.
pub fn files_ending_with(db: &Connection, suffix: &str, limit: usize) -> Result<Vec<String>> {
    let suffix = suffix
        .trim_start_matches(['.', '/', '\\'])
        .replace('\\', "/");
    let mut stmt = db.prepare(&format!(
        "SELECT path FROM files WHERE path = ?1 OR {PATH_AS_TYPED} LIKE '%/' || ?1
         ORDER BY LENGTH(path) LIMIT ?2",
    ))?;
    let rows = stmt.query_map(params![suffix, limit as i64], |r| r.get::<_, String>(0))?;
    Ok(rows.collect::<Result<Vec<_>, _>>()?)
}

/// `files.path` with separators in the form a person types, for the comparisons
/// that take a path from one.
///
/// A Windows store holds `C:\work\api\src\lock.rs` — or the verbatim form of
/// it — and every locator, glob and `--lang` pattern is written with `/`, which
/// is what `semlith read src/main.rs:28-40` hands back. Comparing the two
/// without this matched nothing on that platform, which is the half of #74 that
/// is not the prefix. On unix the column is used as it is: a backslash there is
/// an ordinary character in a filename and rewriting it would make
/// `a\b.rs` answer to `a/b.rs`.
#[cfg(windows)]
const PATH_AS_TYPED: &str = r#"replace(path, '\', '/')"#;
#[cfg(not(windows))]
const PATH_AS_TYPED: &str = "path";

/// The same, for the queries that alias the table as `f` and lowercase it.
#[cfg(windows)]
const GLOB_PATH: &str = r#"replace(lower(f.path), '\', '/')"#;
#[cfg(not(windows))]
const GLOB_PATH: &str = "lower(f.path)";

/// An index on exactly [`GLOB_PATH`]'s expression, so a pattern with a literal
/// directory in front -- `<root>/**`, how a scope to one repository is written
/// -- is a range scan of it rather than a GLOB over every path (#183). Additive:
/// a binary that predates it never names it and SQLite keeps it current.
#[cfg(windows)]
const GLOB_INDEX: &str =
    r#"CREATE INDEX IF NOT EXISTS files_glob_path ON files(replace(lower(path), '\', '/'));"#;
#[cfg(not(windows))]
const GLOB_INDEX: &str = "CREATE INDEX IF NOT EXISTS files_glob_path ON files(lower(path));";

/// The literal directory an absolute pattern starts with, `/` included, and the
/// bound just past every path under it: `/` is followed by `0` in byte order.
/// `None` for a pattern that starts with a wildcard or names no directory.
fn literal_dir(pattern: &str) -> Option<(String, String)> {
    let wild = pattern.find(['*', '?', '['])?;
    let dir = &pattern[..pattern[..wild].rfind('/')? + 1];
    if dir.len() < 2 || !(dir.starts_with('/') || dir.as_bytes().get(1) == Some(&b':')) {
        return None;
    }
    Some((dir.to_string(), format!("{}0", &dir[..dir.len() - 1])))
}

/// Every chunk of one file whose lines overlap `start..=end`, in file order.
///
/// The second stage of a retrieval: a locate answer says `src/store.rs:1041-1080`
/// and this is what returns those lines and nothing else. Chunks overlap by two
/// lines by design, so the caller stitches rather than concatenates — which is
/// [`crate::Span::text`]'s job, not this one's.
pub fn chunks_overlapping(
    db: &Connection,
    path: &str,
    start: u32,
    end: u32,
) -> Result<Vec<ChunkRow>> {
    let mut stmt = db.prepare(
        "SELECT c.id, f.path, c.ord, c.start_line, c.end_line, c.text
         FROM chunks c JOIN files f ON f.id = c.file_id
         WHERE f.path = ?1 AND c.start_line <= ?2 AND c.end_line >= ?3
         ORDER BY c.ord",
    )?;
    let rows = stmt.query_map(params![path, end, start], |r| {
        Ok(ChunkRow {
            id: r.get(0)?,
            path: r.get(1)?,
            ord: r.get(2)?,
            start_line: r.get(3)?,
            end_line: r.get(4)?,
            text: r.get(5)?,
        })
    })?;
    Ok(rows.collect::<Result<Vec<_>, _>>()?)
}

pub fn chunk(db: &Connection, id: u64) -> Result<Option<ChunkRow>> {
    Ok(db
        .query_row(
            "SELECT c.id, f.path, c.ord, c.start_line, c.end_line, c.text
             FROM chunks c JOIN files f ON f.id = c.file_id
             WHERE c.id = ?1",
            params![id as i64],
            |r| {
                Ok(ChunkRow {
                    id: r.get(0)?,
                    path: r.get(1)?,
                    ord: r.get(2)?,
                    start_line: r.get(3)?,
                    end_line: r.get(4)?,
                    text: r.get(5)?,
                })
            },
        )
        .optional()?)
}

/// One row per indexed file: path, bytes, chunks, the last line any chunk of
/// it covers, and when it was last indexed.
#[derive(Debug, Clone)]
pub struct FileRow {
    pub path: String,
    pub bytes: i64,
    pub chunks: i64,
    pub lines: i64,
    pub indexed_at: i64,
}

/// The column a Files listing is ordered by.
///
/// An enum rather than a string spliced into the SQL: the sort column arrives
/// from a query parameter, and the only safe way to put a caller's value in an
/// `ORDER BY` — which cannot be bound as a parameter — is to never put their
/// bytes there at all.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FileSort {
    Path,
    Bytes,
    Chunks,
    Lines,
    Indexed,
    /// Which store holds the row. Not a column of this table at all — it is
    /// which database the row came out of — so it is ordered by the merge.
    Store,
    /// The reader that parsed the file, derived from its extension.
    Reader,
    /// The language the file was recognised as, derived from its extension.
    Lang,
}

impl FileSort {
    pub fn parse(name: &str) -> Option<Self> {
        match name {
            "path" => Some(Self::Path),
            "bytes" => Some(Self::Bytes),
            "chunks" => Some(Self::Chunks),
            "lines" => Some(Self::Lines),
            "indexed" => Some(Self::Indexed),
            "store" => Some(Self::Store),
            "reader" => Some(Self::Reader),
            "lang" => Some(Self::Lang),
            _ => None,
        }
    }

    /// Whether this ordering is over a value the database does not hold.
    ///
    /// Three of the seven columns are derived by the route from the path, or
    /// from which store answered. They cannot be an `ORDER BY`, so the rows
    /// are ordered by the merge instead — which needs every matching row
    /// rather than one page of each store's.
    pub fn derived(self) -> bool {
        matches!(self, Self::Store | Self::Reader | Self::Lang)
    }

    fn sql(self) -> &'static str {
        match self {
            Self::Path => "f.path",
            Self::Bytes => "f.bytes",
            Self::Chunks => "COUNT(c.id)",
            Self::Lines => "COALESCE(MAX(c.end_line), 0)",
            Self::Indexed => "f.indexed_at",
            // The base order under a derived sort: deterministic, and the
            // tie-break the merge uses anyway.
            Self::Store | Self::Reader | Self::Lang => "f.path",
        }
    }
}

/// How many indexed files match `groups`.
///
/// Separate from the rows because the Files table pages: the page is bounded
/// and the total is not, and returning every row to count them is what the old
/// silent truncation was hiding.
pub fn file_count(db: &Connection, groups: &[Vec<String>]) -> Result<i64> {
    let (predicate, binds) = glob_predicate(groups);
    let sql = format!("SELECT COUNT(*) FROM files f WHERE {predicate}");
    let mut stmt = db.prepare(&sql)?;
    let args = binds.into_iter().map(Value::Text);
    Ok(stmt.query_row(rusqlite::params_from_iter(args), |r| r.get(0))?)
}

/// The first `take` matching files in the requested order.
///
/// The Files view's row, in one query. Asking per file would be one statement
/// per file, and a corpus is tens of thousands of them. The ordering is done
/// here rather than in the browser so that sorting orders the whole store
/// rather than the page of it that happens to be loaded — with several stores
/// open the caller takes `offset + limit` from each and merges, which is why
/// this takes a count rather than an offset.
pub fn file_rows(
    db: &Connection,
    groups: &[Vec<String>],
    sort: FileSort,
    desc: bool,
    take: i64,
) -> Result<Vec<FileRow>> {
    let (predicate, binds) = glob_predicate(groups);
    let direction = if desc { "DESC" } else { "ASC" };
    let sql = format!(
        "SELECT f.path, f.bytes, COUNT(c.id), COALESCE(MAX(c.end_line), 0), f.indexed_at \
         FROM files f LEFT JOIN chunks c ON c.file_id = f.id \
         WHERE {predicate} GROUP BY f.id ORDER BY {} {direction}, f.path ASC LIMIT ?",
        sort.sql()
    );
    let mut stmt = db.prepare(&sql)?;
    let mut args: Vec<Value> = binds.into_iter().map(Value::Text).collect();
    args.push(Value::Integer(take.max(0)));
    let rows = stmt.query_map(rusqlite::params_from_iter(args), |r| {
        Ok(FileRow {
            path: r.get(0)?,
            bytes: r.get(1)?,
            chunks: r.get(2)?,
            lines: r.get(3)?,
            indexed_at: r.get(4)?,
        })
    })?;
    Ok(rows.collect::<Result<Vec<_>, _>>()?)
}

/// What a listing spans: the distinct file extensions in it, and the total
/// number of lines across every file.
///
/// Both are things the Stores and Files pages state and neither `stats()` nor
/// a page of rows can answer — a page knows about its own rows, and a header
/// that reports "9 formats" from the fifteen rows on screen is reporting the
/// page rather than the store.
#[derive(Debug, Default, Clone)]
pub struct Facets {
    pub extensions: Vec<String>,
    pub lines: i64,
}

pub fn file_facets(db: &Connection, groups: &[Vec<String>]) -> Result<Facets> {
    let (predicate, binds) = glob_predicate(groups);

    // The extension is taken in Rust rather than in SQL. SQLite has no
    // right-hand search, and the usual `rtrim`/`replace` idiom for it reads a
    // dot in a directory name as the start of an extension — which is a wrong
    // answer in a column that exists to say what the corpus is made of.
    let sql = format!("SELECT f.path FROM files f WHERE {predicate}");
    let mut stmt = db.prepare(&sql)?;
    let args = binds.clone().into_iter().map(Value::Text);
    let paths = stmt.query_map(rusqlite::params_from_iter(args), |r| r.get::<_, String>(0))?;
    let mut extensions: Vec<String> = paths
        .collect::<Result<Vec<String>, _>>()?
        .iter()
        .filter_map(|p| {
            std::path::Path::new(p)
                .extension()
                .and_then(|e| e.to_str())
                .map(|e| e.to_ascii_lowercase())
        })
        .collect();
    extensions.sort();
    extensions.dedup();

    let sql = format!(
        "SELECT COALESCE(SUM(n), 0) FROM ( \
           SELECT COALESCE(MAX(c.end_line), 0) AS n \
           FROM files f LEFT JOIN chunks c ON c.file_id = f.id \
           WHERE {predicate} GROUP BY f.id)"
    );
    let mut stmt = db.prepare(&sql)?;
    let args = binds.into_iter().map(Value::Text);
    let lines: i64 = stmt.query_row(rusqlite::params_from_iter(args), |r| r.get(0))?;

    Ok(Facets { extensions, lines })
}

/// An indexed image: its file's path and the pixel size.
#[derive(Debug, Clone)]
pub struct ImageRow {
    pub id: i64,
    pub path: String,
    pub width: u32,
    pub height: u32,
}

pub fn insert_image(db: &Connection, file_id: i64, width: u32, height: u32) -> Result<i64> {
    db.execute(
        "INSERT INTO images (file_id, width, height) VALUES (?1, ?2, ?3)",
        params![file_id, width, height],
    )?;
    Ok(db.last_insert_rowid())
}

/// One image row, joined with its file's path.
pub fn image(db: &Connection, id: i64) -> Result<Option<ImageRow>> {
    Ok(db
        .query_row(
            "SELECT i.id, f.path, i.width, i.height FROM images i \
             JOIN files f ON f.id = i.file_id WHERE i.id = ?1",
            params![id],
            |r| {
                Ok(ImageRow {
                    id: r.get(0)?,
                    path: r.get(1)?,
                    width: r.get(2)?,
                    height: r.get(3)?,
                })
            },
        )
        .optional()?)
}

/// `(tool, excerpt_tokens)` for every ledger row that names its tool, the
/// newest `limit` of them: what the Agents page's typical answer sizes are
/// the median of.
pub fn excerpt_tokens_by_tool(db: &Connection, limit: usize) -> Result<Vec<(String, i64)>> {
    let mut stmt = db.prepare(
        "SELECT tool, excerpt_tokens FROM retrievals WHERE tool IS NOT NULL \
         ORDER BY id DESC LIMIT ?1",
    )?;
    let rows = stmt.query_map(params![limit as i64], |r| Ok((r.get(0)?, r.get(1)?)))?;
    Ok(rows.collect::<Result<Vec<_>, _>>()?)
}

/// Every indexed image, for the decisions table's "read as image" rows.
pub fn image_rows(db: &Connection) -> Result<Vec<ImageRow>> {
    let mut stmt = db.prepare(
        "SELECT i.id, f.path, i.width, i.height FROM images i \
         JOIN files f ON f.id = i.file_id ORDER BY f.path",
    )?;
    let rows = stmt.query_map([], |r| {
        Ok(ImageRow {
            id: r.get(0)?,
            path: r.get(1)?,
            width: r.get(2)?,
            height: r.get(3)?,
        })
    })?;
    Ok(rows.collect::<Result<Vec<_>, _>>()?)
}

/// The image ids belonging to `path`, before its row is deleted.
///
/// Read rather than returned by `delete_file`, because the cascade that removes
/// them is what makes the ids unreadable — and the vectors they address still
/// have to be evicted from the image index afterwards.
pub fn image_ids_of(db: &Connection, path: &str) -> Result<Vec<i64>> {
    let mut stmt =
        db.prepare("SELECT i.id FROM images i JOIN files f ON f.id = i.file_id WHERE f.path = ?1")?;
    let rows = stmt.query_map(params![path], |r| r.get::<_, i64>(0))?;
    Ok(rows.collect::<Result<Vec<_>, _>>()?)
}

/// How many images this store holds.
pub fn image_count(db: &Connection) -> Result<i64> {
    Ok(db.query_row("SELECT COUNT(*) FROM images", [], |r| r.get(0))?)
}

/// Every image id, for the ids a filtered search may consider.
pub fn image_ids(db: &Connection) -> Result<Vec<i64>> {
    let mut stmt = db.prepare("SELECT id FROM images")?;
    let rows = stmt.query_map([], |r| r.get::<_, i64>(0))?;
    Ok(rows.collect::<Result<Vec<_>, _>>()?)
}

/// The image ids whose file matches `groups`, for a filtered search.
pub fn filtered_image_ids(db: &Connection, groups: &[Vec<String>]) -> Result<Vec<i64>> {
    let (predicate, binds) = glob_predicate(groups);
    let sql =
        format!("SELECT i.id FROM images i JOIN files f ON f.id = i.file_id WHERE {predicate}");
    let mut stmt = db.prepare(&sql)?;
    let args = binds.into_iter().map(Value::Text);
    let rows = stmt.query_map(rusqlite::params_from_iter(args), |r| r.get::<_, i64>(0))?;
    Ok(rows.collect::<Result<Vec<_>, _>>()?)
}

pub fn all_paths(db: &Connection) -> Result<Vec<String>> {
    filtered_paths(db, &[])
}

/// Everything this store holds for one file, its chunks in order, joined.
///
/// What `semlith scan` reads. The store's own copy rather than the file on
/// disk: the question the scan answers is what this store is holding, and a
/// file that has since been cleaned or deleted is still a credential sitting
/// in an index an agent can search.
pub fn text_of(db: &Connection, path: &str) -> Result<String> {
    let mut stmt = db.prepare(
        "SELECT c.text FROM chunks c JOIN files f ON f.id = c.file_id
         WHERE f.path = ?1 ORDER BY c.ord",
    )?;
    let rows = stmt.query_map([path], |r| r.get::<_, String>(0))?;
    let mut out = String::new();
    for row in rows {
        out.push_str(&row?);
        out.push('\n');
    }
    Ok(out)
}

/// Indexed paths whose file matches `groups`, in path order.
///
/// The same predicate the search halves are built on, so "which files would
/// this filter reach" and "which files did this filter search" cannot drift
/// apart and give an agent two different answers about one store.
pub fn filtered_paths(db: &Connection, groups: &[Vec<String>]) -> Result<Vec<String>> {
    let (predicate, binds) = glob_predicate(groups);
    let sql = format!("SELECT f.path FROM files f WHERE {predicate} ORDER BY f.path");
    let mut stmt = db.prepare(&sql)?;
    let args = binds.into_iter().map(Value::Text);
    let rows = stmt.query_map(rusqlite::params_from_iter(args), |r| r.get::<_, String>(0))?;
    Ok(rows.collect::<Result<Vec<_>, _>>()?)
}

/// Chunks whose file has a committed hash — the ones whose vectors are on disk.
///
/// A file is recorded with the [`PENDING`](crate::PENDING) empty hash until its
/// vectors are durable, so this is exactly what the vector index holds at rest,
/// and it costs one indexed count instead of reading the index to find out.
pub fn durable_chunks(db: &Connection) -> Result<i64> {
    Ok(db.query_row(
        "SELECT COUNT(*) FROM chunks c JOIN files f ON f.id = c.file_id WHERE f.hash != ''",
        [],
        |r| r.get(0),
    )?)
}

/// Of `ids`, the chunks whose file is still being embedded, with their text:
/// what a search mid-run embeds on the query path.
pub fn pending_among(db: &Connection, ids: &[u64]) -> Result<Vec<(u64, String)>> {
    let mut stmt = db.prepare_cached(
        "SELECT c.text FROM chunks c JOIN files f ON f.id = c.file_id
         WHERE c.id = ?1 AND f.hash = ''",
    )?;
    let mut out = Vec::new();
    for id in ids {
        if let Some(text) = stmt
            .query_row(params![*id as i64], |r| r.get::<_, String>(0))
            .optional()?
        {
            out.push((*id, text));
        }
    }
    Ok(out)
}

/// The share of the store's chunks still being embedded, or `None` when
/// nothing is: one indexed probe on every search, and the counts only when
/// the probe finds a run in progress.
pub fn pending_share(db: &Connection) -> Result<Option<f64>> {
    let busy: bool = db
        .query_row(
            "SELECT 1 FROM files WHERE hash = '' LIMIT 1",
            [],
            |_| Ok(()),
        )
        .optional()?
        .is_some();
    if !busy {
        return Ok(None);
    }
    let (pending, all): (i64, i64) = db.query_row(
        "SELECT SUM(f.hash = ''), COUNT(*) FROM chunks c JOIN files f ON f.id = c.file_id",
        [],
        |r| Ok((r.get::<_, Option<i64>>(0)?.unwrap_or(0), r.get(1)?)),
    )?;
    Ok((all > 0 && pending > 0).then(|| pending as f64 / all as f64))
}

/// Apply the schema to a bare in-memory connection, for tests in other
/// modules that need somewhere to put symbols and edges.
#[cfg(test)]
pub fn prepare_for_tests(db: &Connection) {
    db.execute_batch(SCHEMA).unwrap();
    db.pragma_update(None, "foreign_keys", "ON").unwrap();
}

/// A symbol row joined with its file's path — what a graph answer resolves to.
#[derive(Debug, Clone, serde::Serialize)]
pub struct SymbolRow {
    pub id: i64,
    #[serde(serialize_with = "crate::serialize_plain")]
    pub path: String,
    pub kind: String,
    pub name: String,
    pub qualified: String,
    pub start_line: u32,
    pub end_line: u32,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub chunk_id: Option<i64>,
    /// Which store this came from, when more than one is open. Absent for a
    /// single store, so one store's output is what it would have been before
    /// stores could be combined.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub store: Option<String>,
}

/// One end of a traversal: the symbol reached, and the edge that reached it.
///
/// Serialised flat, as the symbol's fields beside the edge's, with the
/// symbol's kind as `symbol_kind`: both were `kind`, so `symbol --json` printed
/// the key twice and a parser kept the edge's (2026-10-14 audit, G7).
#[derive(Debug, Clone, serde::Serialize)]
pub struct EdgeEnd {
    #[serde(flatten, serialize_with = "reached_symbol")]
    pub symbol: SymbolRow,
    pub kind: String,
    pub confidence: String,
    /// How many definitions of this name the store holds.
    ///
    /// `1` for an edge that could only ever mean one thing. Greater than one
    /// on an `ambiguous` row, where it is the number a renderer prints instead
    /// of listing every candidate as though each were a separate call — and on
    /// a `resolved` row, where it is how many candidates the ranking had to
    /// choose between, which is the difference between "there was only one"
    /// and "there were four and the source said which".
    pub definitions: usize,
    /// The definition this edge actually leaves from.
    ///
    /// [`edges_out`] is asked about a name, and a name can have several
    /// definitions, each with its own edges. Without this the answer says only
    /// that *something* called `index` reaches here — which is how a path
    /// finder ends up walking out of one definition of a name and into
    /// another without anything on the page saying so.
    ///
    /// `None` from [`edges_in`], where the queried name is the far end of the
    /// edge and there is no single row it leaves from.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub from_path: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub from_line: Option<u32>,
    /// The line the call, import or reference was written on, in the source
    /// file named by `from_path`.
    ///
    /// `None` for an edge written before 0.16.0, and for one whose source file
    /// has not been re-indexed since. A renderer says nothing rather than
    /// guessing when it is absent.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub line: Option<u32>,
    /// For a caller of a name with several definitions, the one this call
    /// resolves to — `Fleet::search_preferring` rather than "one of two".
    /// `None` when the name has one definition or the call did not settle.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub means: Option<String>,
}

/// A [`SymbolRow`] inside an [`EdgeEnd`]: every field, `kind` as `symbol_kind`.
fn reached_symbol<S: serde::Serializer>(row: &SymbolRow, s: S) -> Result<S::Ok, S::Error> {
    use serde::ser::SerializeMap;
    let mut map = s.serialize_map(None)?;
    map.serialize_entry("id", &row.id)?;
    map.serialize_entry("path", &crate::plain(&row.path))?;
    map.serialize_entry("symbol_kind", &row.kind)?;
    map.serialize_entry("name", &row.name)?;
    map.serialize_entry("qualified", &row.qualified)?;
    map.serialize_entry("start_line", &row.start_line)?;
    map.serialize_entry("end_line", &row.end_line)?;
    if let Some(chunk) = row.chunk_id {
        map.serialize_entry("chunk_id", &chunk)?;
    }
    if let Some(store) = &row.store {
        map.serialize_entry("store", store)?;
    }
    map.end()
}

const SYMBOL_COLUMNS: &str =
    "s.id, f.path, s.kind, s.name, s.qualified, s.start_line, s.end_line, s.chunk_id";

fn symbol_row(r: &rusqlite::Row<'_>) -> rusqlite::Result<SymbolRow> {
    Ok(SymbolRow {
        id: r.get(0)?,
        path: r.get(1)?,
        kind: r.get(2)?,
        name: r.get(3)?,
        qualified: r.get(4)?,
        start_line: r.get(5)?,
        end_line: r.get(6)?,
        chunk_id: r.get(7)?,
        store: None,
    })
}

pub fn insert_symbol(
    db: &Connection,
    file_id: i64,
    chunk_id: Option<i64>,
    symbol: &crate::graph::Symbol,
) -> Result<i64> {
    db.prepare_cached(
        "INSERT INTO symbols (file_id, chunk_id, kind, name, qualified, start_line, end_line)
         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
    )?
    .execute(params![
        file_id,
        chunk_id,
        symbol.kind,
        symbol.name,
        symbol.qualified,
        symbol.start_line,
        symbol.end_line
    ])?;
    Ok(db.last_insert_rowid())
}

pub fn insert_edge(
    db: &Connection,
    src: i64,
    dst: &str,
    kind: &str,
    confidence: &str,
    hint: Option<&str>,
    line: Option<u32>,
) -> Result<()> {
    db.prepare_cached(
        "INSERT INTO edges (src, dst, kind, confidence, hint, line)
         VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
    )?
    .execute(params![src, dst, kind, confidence, hint, line])?;
    Ok(())
}

/// Write one file's extraction: its symbols, then its edges by source name.
///
/// `spans` are the file's chunks as `(start_line, end_line, chunk_id)`, so each
/// symbol points at the chunk its first line sits in. Returns `(symbols,
/// edges)` written. The one writer `index_set` and the graph tests share, so a
/// test of what a file yields is a test of what the store holds.
pub fn write_extraction(
    db: &Connection,
    extraction: &crate::graph::Extraction,
    file_id: i64,
    spans: &[(u32, u32, i64)],
) -> Result<(usize, usize)> {
    // Name to id, for the edges below. A file may define the same name
    // twice — two `new` methods on two types — so an edge leaves from the
    // narrowest of them whose lines hold the line it was written on, and from
    // the first when none does. Until 0.44.1 the first always won, and every
    // call the second made was lost.
    let mut ids: std::collections::HashMap<&str, (i64, bool)> = std::collections::HashMap::new();
    let mut extents: std::collections::HashMap<&str, Vec<(i64, u32, u32)>> =
        std::collections::HashMap::new();
    // The file's own symbol, which an edge marked `from_module` leaves from
    // even when a definition shares its name.
    let mut module_id: Option<i64> = None;
    for symbol in &extraction.symbols {
        let chunk_id = spans
            .iter()
            .find(|(start, end, _)| symbol.start_line >= *start && symbol.start_line <= *end)
            .map(|(_, _, id)| *id);
        let id = insert_symbol(db, file_id, chunk_id, symbol)?;
        // A definition outranks the file's module symbol of the same name:
        // `fn brief` in `brief.rs` is where its calls come from, and giving
        // them to the module put every caller of it at line 1.
        let module = symbol.kind == "module";
        if module && module_id.is_none() {
            module_id = Some(id);
        }
        if !module {
            extents.entry(symbol.name.as_str()).or_default().push((
                id,
                symbol.start_line,
                symbol.end_line,
            ));
        }
        match ids.get(symbol.name.as_str()) {
            None => {
                ids.insert(symbol.name.as_str(), (id, module));
            }
            Some((_, true)) if !module => {
                ids.insert(symbol.name.as_str(), (id, false));
            }
            _ => {}
        }
    }
    let ids: std::collections::HashMap<&str, i64> =
        ids.into_iter().map(|(k, (id, _))| (k, id)).collect();

    let mut written = 0;
    for edge in &extraction.edges {
        let covering = edge.line.and_then(|line| {
            extents
                .get(edge.from.as_str())?
                .iter()
                .filter(|(_, start, end)| (*start..=*end).contains(&line))
                .min_by_key(|(_, start, end)| end - start)
                .map(|(id, _, _)| *id)
        });
        let src = match module_id {
            Some(id) if edge.from_module => id,
            _ => match covering.or_else(|| ids.get(edge.from.as_str()).copied()) {
                Some(id) => id,
                None => continue,
            },
        };
        insert_edge(
            db,
            src,
            &edge.to,
            &edge.kind,
            &edge.confidence,
            edge.hint.as_deref(),
            edge.line,
        )?;
        written += 1;
    }
    Ok((extraction.symbols.len(), written))
}

/// Every definition of `name`, across the whole store.
///
/// Exact match, not a glob: `semlith symbol acquire` asking about `acquire` and
/// being handed `acquire_timeout` as well would make the answer something the
/// caller has to re-filter.
pub fn symbols_named(db: &Connection, name: &str, limit: usize) -> Result<Vec<SymbolRow>> {
    let sql = format!(
        "SELECT {SYMBOL_COLUMNS} FROM symbols s JOIN files f ON f.id = s.file_id
         WHERE s.name = ?1 ORDER BY f.path, s.start_line LIMIT ?2"
    );
    let mut stmt = db.prepare(&sql)?;
    let rows = stmt.query_map(params![name, limit as i64], symbol_row)?;
    Ok(rows.collect::<Result<Vec<_>, _>>()?)
}

/// [`symbols_named`] with code first — definitions before the headings, keys
/// and selectors that share the name — and how many there are in all.
pub fn symbols_named_counted(
    db: &Connection,
    name: &str,
    limit: usize,
) -> Result<(Vec<SymbolRow>, usize)> {
    let navigational = crate::graph::not_navigational("s.kind");
    let sql = format!(
        "SELECT {SYMBOL_COLUMNS} FROM symbols s JOIN files f ON f.id = s.file_id
         WHERE s.name = ?1 ORDER BY NOT ({navigational}), f.path, s.start_line LIMIT ?2"
    );
    let mut stmt = db.prepare(&sql)?;
    let limit = i64::try_from(limit).unwrap_or(i64::MAX);
    let rows = stmt.query_map(params![name, limit], symbol_row)?;
    let rows = rows.collect::<Result<Vec<_>, _>>()?;
    let total: i64 = db.query_row(
        "SELECT COUNT(*) FROM symbols WHERE name = ?1",
        [name],
        |r| r.get(0),
    )?;
    Ok((rows, total as usize))
}

/// Symbols to draw for a scope, newest-defined first within each file.
///
/// `prefix` narrows to a directory or a file the way the Files view does.
/// Module symbols are left out: every file has one, so they would fill the
/// budget with nodes that say only "this file exists".
pub fn symbols_scoped(
    db: &Connection,
    prefix: Option<&str>,
    limit: usize,
) -> Result<Vec<SymbolRow>> {
    // Busiest first. A scope filled with symbols that touch nothing draws a
    // field of dots: technically the graph, and of no use to anyone looking at
    // it. Ordering by how connected a symbol is puts the shape on the screen.
    let Some(p) = prefix else {
        return busiest_symbols(db, limit);
    };
    let pattern = format!("*{}*", p.to_ascii_lowercase());
    let sql = format!(
        "SELECT {SYMBOL_COLUMNS} FROM symbols s JOIN files f ON f.id = s.file_id
         WHERE lower(f.path) GLOB ?1 AND s.kind != 'module'
         ORDER BY (
           (SELECT COUNT(*) FROM edges e WHERE e.src = s.id)
           + (SELECT COUNT(*) FROM edges e WHERE e.dst = s.name)
         ) DESC, f.path, s.start_line LIMIT ?2"
    );
    let mut stmt = db.prepare(&sql)?;
    let rows = stmt.query_map(params![pattern, limit as i64], symbol_row)?;
    Ok(rows.collect::<Result<Vec<_>, _>>()?)
}

/// The busiest symbols of a whole store, for [`symbols_scoped`] with no scope.
///
/// Ordering every symbol by two correlated counts took 32 s on the 879k-chunk
/// corpus, and it is the Graph page's first fetch. Here each direction is
/// counted once, on its own index, for the most-targeted names and the busiest
/// sources, and only those are ordered: about a second on the same corpus. A
/// symbol is missed only by being busy in neither direction on its own, which
/// a symbol worth opening on is not. A name counted as a target counts for
/// every definition of it, as the full count did.
fn busiest_symbols(db: &Connection, limit: usize) -> Result<Vec<SymbolRow>> {
    let candidates = (limit * 4).max(64) as i64;
    let sql = format!(
        "WITH ins AS (SELECT dst AS name, COUNT(*) AS n FROM edges GROUP BY dst ORDER BY n DESC LIMIT ?1),
              outs AS (SELECT src AS id, COUNT(*) AS n FROM edges GROUP BY src ORDER BY n DESC LIMIT ?1),
              picked AS (
                SELECT s.id FROM symbols s JOIN ins ON s.name = ins.name
                UNION
                SELECT id FROM outs
              )
         SELECT {SYMBOL_COLUMNS} FROM picked
         JOIN symbols s ON s.id = picked.id
         JOIN files f ON f.id = s.file_id
         WHERE s.kind != 'module'
         ORDER BY (
           COALESCE((SELECT n FROM outs WHERE outs.id = s.id), (SELECT COUNT(*) FROM edges e WHERE e.src = s.id))
           + COALESCE((SELECT n FROM ins WHERE ins.name = s.name), 0)
         ) DESC, f.path, s.start_line LIMIT ?2"
    );
    let mut stmt = db.prepare(&sql)?;
    let rows = stmt.query_map(params![candidates, limit as i64], symbol_row)?;
    Ok(rows.collect::<Result<Vec<_>, _>>()?)
}

/// Symbols defined in files matching `groups` whose name is one of `names`.
///
/// The resolution step every traversal shares: an edge names its target, and
/// this is what turns that name back into rows. The filter is threaded through
/// so graph expansion obeys the same chunk-id set as the other two search
/// halves rather than reaching outside it.
pub fn symbols_by_names(
    db: &Connection,
    names: &[String],
    groups: &[Vec<String>],
) -> Result<Vec<SymbolRow>> {
    // In batches, because SQLite refuses a statement with more than 32 766
    // variables, and a walk over the 70-repository benchmark corpus reached more
    // names than that: every search there failed with "too many SQL variables".
    const BATCH: usize = 1000;
    let (predicate, filter_binds) = glob_predicate(groups);
    let mut out = Vec::new();
    for batch in names.chunks(BATCH) {
        let holes = vec!["?"; batch.len()].join(", ");
        let sql = format!(
            "SELECT {SYMBOL_COLUMNS} FROM symbols s JOIN files f ON f.id = s.file_id
             WHERE {predicate} AND s.name IN ({holes})"
        );
        let mut stmt = db.prepare(&sql)?;
        let args = filter_binds.iter().chain(batch).cloned().map(Value::Text);
        let rows = stmt.query_map(rusqlite::params_from_iter(args), symbol_row)?;
        for row in rows {
            out.push(row?);
        }
    }
    out.sort_by(|a, b| (&a.path, a.start_line, a.id).cmp(&(&b.path, b.start_line, b.id)));
    Ok(out)
}

/// What the symbols named `name` point at: callees, imports, references out.
///
/// One hop. `kinds` empty means every edge kind.
///
/// # Resolution
///
/// `edges.dst` is a name, and a name is not an address: a store of any size
/// holds four `get`s and seven `index`es. Until 0.15.0 this returned every
/// definition that shared the name, each as its own row and each looking
/// exactly like a call the source makes — which is how `semlith path` came to
/// answer "yes, connected" to questions whose honest answer is "no".
///
/// Each edge is now resolved against its candidates by [`rank`], and the
/// confidence it comes back with says how it was settled:
///
/// - `extracted` — the file named where the target came from. Stored, and
///   never overwritten here: a fact from the syntax tree outranks a ranking.
/// - `resolved` — the candidates narrowed to exactly one. Either the source
///   narrowed them (same file, the hint, an import) or the corpus did, by
///   holding only one definition of the name.
/// - `ambiguous` — several candidates survived and nothing chose between them.
///   Every candidate is returned, each marked `ambiguous` and each carrying
///   the count, so a renderer can collapse them to one row and a traversal can
///   refuse to cross them.
///
/// `inferred` is the stored value and is what [`edges_in`] and the edge census
/// still show. It does not come back from here, because by the time a target
/// has been looked up there is always something to say about it: either one
/// definition answers to the name or several do.
///
/// A resolution is against the indexed corpus, not against the world. One
/// definition of `parse` in the store does not prove the call went to it
/// rather than to a dependency that was never indexed — that is what the
/// hidden-by-default unresolved targets are about, not this.
pub fn edges_out(db: &Connection, name: &str, kinds: &[String]) -> Result<Vec<EdgeEnd>> {
    edges_out_below(db, name, kinds, None, None)
}

/// How many edges touch `name`, and how many of them are certain, counted on
/// the indexes alone: `(resolved, total)`.
///
/// For the graph page's choice of where to open, which only compares counts.
/// An edge in counts as certain when the extractor said so; an edge out when
/// the extractor said so or its target has exactly one definition, which is
/// how [`edges_out`] settles the common case. An out edge to a name nothing
/// defines is not counted, as [`edges_out`] does not return it.
pub fn edge_counts(db: &Connection, name: &str) -> Result<(usize, usize)> {
    let navigational = crate::graph::not_navigational("h.kind");
    let (in_total, in_sure): (i64, i64) = db.query_row(
        "SELECT COUNT(*), COALESCE(SUM(confidence IN ('extracted', 'resolved')), 0)
         FROM edges WHERE dst = ?1",
        [name],
        |r| Ok((r.get(0)?, r.get(1)?)),
    )?;
    let (out_total, out_sure): (i64, i64) = db.query_row(
        &format!(
            "SELECT COUNT(*), COALESCE(SUM(e.confidence = 'extracted' OR
                 (substr(COALESCE(e.hint, ''), 1, 1) != '.' AND
                  (SELECT COUNT(*) FROM (SELECT 1 FROM symbols h WHERE h.name = e.dst AND {navigational} LIMIT 2)) = 1)), 0)
             FROM edges e JOIN symbols src ON src.id = e.src
             WHERE src.name = ?1
               AND EXISTS (SELECT 1 FROM symbols h WHERE h.name = e.dst AND {navigational})"
        ),
        [name],
        |r| Ok((r.get(0)?, r.get(1)?)),
    )?;
    Ok((
        (in_sure + out_sure) as usize,
        (in_total + out_total) as usize,
    ))
}

/// [`edges_out`] restricted to edges whose target is one of `targets`, settled
/// exactly as [`edges_out`] settles them.
///
/// For the graph page, which draws only edges with both ends on the canvas: on
/// the 879k-chunk corpus `edges_out("main")` joins 706,461 candidate rows in
/// about ten seconds to keep the handful that point at a drawn name, and the
/// page asks once per drawn node.
pub fn edges_out_to(db: &Connection, name: &str, targets: &[String]) -> Result<Vec<EdgeEnd>> {
    if targets.is_empty() {
        return Ok(Vec::new());
    }
    edges_out_below(db, name, &[], None, Some(targets))
}

/// [`edges_out`] without the edges whose target has more than `hub`
/// definitions, dropped before their candidates are joined.
///
/// For search's graph walk, which crosses no ambiguous edge anyway. An edge to
/// `get` in a store holding seven hundred of them joins seven hundred candidate
/// rows and nearly always resolves to "ambiguous"; on the 879k-chunk benchmark
/// corpus those rows were most of the walk's cost. `neighbors`, `impact` and
/// `path` keep [`edges_out`], because an answer about one symbol can afford to
/// settle every edge.
pub fn edges_out_short_of_hubs(
    db: &Connection,
    name: &str,
    kinds: &[String],
    hub: usize,
) -> Result<Vec<EdgeEnd>> {
    edges_out_below(db, name, kinds, Some(hub), None)
}

fn edges_out_below(
    db: &Connection,
    name: &str,
    kinds: &[String],
    hub: Option<usize>,
    targets: Option<&[String]>,
) -> Result<Vec<EdgeEnd>> {
    let filter = kind_predicate(kinds, "e.kind");
    // An edge's target is resolved by name, so without this a configuration key
    // named `path` is a candidate definition of every `path` any code calls.
    let target = crate::graph::not_navigational("s.kind");
    // Written on `e` alone, so SQLite tests it per edge, before the join to
    // the candidates it exists to avoid. The count stops at `hub + 1`.
    let short_of_hub = match hub {
        Some(hub) => format!(
            "AND (SELECT COUNT(*) FROM (SELECT 1 FROM symbols h WHERE h.name = e.dst AND {} LIMIT {})) <= {hub}",
            crate::graph::not_navigational("h.kind"),
            hub + 1
        ),
        None => String::new(),
    };
    // Bound after the kinds, in the order the holes appear.
    let to_targets = match targets {
        Some(t) => format!("AND e.dst IN ({})", vec!["?"; t.len()].join(", ")),
        None => String::new(),
    };
    let sql = format!(
        "SELECT {SYMBOL_COLUMNS}, e.kind, e.confidence, e.hint, srcf.path, src.start_line, e.line,
                src.id, src.kind, src.qualified
         FROM symbols src
         JOIN files srcf ON srcf.id = src.file_id
         JOIN edges e ON e.src = src.id
         JOIN symbols s ON s.name = e.dst AND {target}
         JOIN files f ON f.id = s.file_id
         WHERE src.name = ?1 AND {filter} {to_targets} {short_of_hub}
         ORDER BY f.path, s.start_line"
    );
    let mut stmt = db.prepare(&sql)?;
    let mut binds: Vec<Value> = vec![Value::Text(name.to_string())];
    binds.extend(kinds.iter().map(|k| Value::Text(k.clone())));
    binds.extend(
        targets
            .unwrap_or(&[])
            .iter()
            .map(|t| Value::Text(t.clone())),
    );
    let rows = stmt.query_map(rusqlite::params_from_iter(binds), |r| {
        Ok(Reached {
            symbol: symbol_row(r)?,
            kind: r.get(8)?,
            confidence: r.get(9)?,
            hint: r.get(10)?,
            src_path: r.get(11)?,
            src_line: r.get(12)?,
            line: r.get(13)?,
            src_id: r.get(14)?,
            src_top_level: top_level(
                &r.get::<_, String>(15)?,
                &r.get::<_, String>(16)?,
                &r.get::<_, String>(11)?,
            ),
        })
    })?;
    let reached = rows.collect::<Result<Vec<_>, _>>()?;
    resolve(db, reached)
}

/// One row of the join behind [`edges_out`]: a candidate for an edge's target,
/// with everything the ranking needs about the edge itself.
struct Reached {
    symbol: SymbolRow,
    kind: String,
    confidence: String,
    hint: Option<String>,
    src_path: String,
    src_line: u32,
    line: Option<u32>,
    /// The definition the edge leaves from, and whether it sits at its file's
    /// top level (see [`top_level`]).
    src_id: i64,
    src_top_level: bool,
}

/// Turn candidate rows into resolved edges.
///
/// The rows arrive as the cross product of edges and the definitions their
/// targets could mean, so they are grouped back into edges first. Order is
/// preserved: the first time an edge is seen decides where its rows sit in the
/// answer, so a caller that used to read this list top to bottom still reads
/// it in the same order.
fn resolve(db: &Connection, reached: Vec<Reached>) -> Result<Vec<EdgeEnd>> {
    // (source definition, source file, target name, edge kind, hint) — one
    // edge in the source.
    type Key = (i64, String, String, String, Option<String>);
    let mut order: Vec<Key> = Vec::new();
    let mut groups: std::collections::HashMap<Key, Vec<Reached>> = std::collections::HashMap::new();
    for row in reached {
        let key = (
            row.src_id,
            row.src_path.clone(),
            row.symbol.name.clone(),
            row.kind.clone(),
            row.hint.clone(),
        );
        if !groups.contains_key(&key) {
            order.push(key.clone());
        }
        groups.entry(key).or_default().push(row);
    }

    // The imports of each source file, read once and only when a group
    // actually needs them: the first two tiers settle most edges, and a file's
    // imports are a query this should not pay for on every hop of a traversal.
    let mut imports: std::collections::HashMap<String, Vec<String>> =
        std::collections::HashMap::new();

    let mut out = Vec::new();
    for key in order {
        let mut candidates = groups.remove(&key).unwrap_or_default();
        let (_, src_path, _, kind, hint) = &key;
        let top = candidates.first().is_some_and(|c| c.src_top_level);
        // What the call cannot mean (see `reachable`) is not ranked, and an
        // unsettled call does not list it either.
        let rows: Vec<&SymbolRow> = candidates.iter().map(|c| &c.symbol).collect();
        let mut reach = reachable(kind, &rows, src_path, hint.as_deref(), top).into_iter();
        candidates.retain(|_| reach.next().unwrap_or(true));
        // A structural edge names a definition in its own file, on the line
        // the edge records; anything elsewhere with that name is not it.
        if structural(kind) {
            candidates.retain(|c| {
                c.symbol.path == c.src_path && c.line.is_none_or(|l| l == c.symbol.start_line)
            });
            for row in candidates {
                out.push(EdgeEnd {
                    symbol: row.symbol,
                    kind: row.kind,
                    confidence: row.confidence,
                    definitions: 1,
                    from_path: Some(row.src_path),
                    from_line: Some(row.src_line),
                    line: row.line,
                    means: None,
                });
            }
            continue;
        }
        let definitions = candidates.len();

        let rows: Vec<&SymbolRow> = candidates.iter().map(|c| &c.symbol).collect();
        let (picked, settled) = choose_for(
            db,
            kind,
            &rows,
            src_path,
            hint.as_deref(),
            top,
            &mut imports,
        )?;
        // A receiver's edge that did not settle lists only what a method call
        // can mean; any other unsettled edge lists every candidate, as before.
        let receiver = hint
            .as_deref()
            .is_some_and(|h| h.starts_with(crate::graph::RECEIVER));
        let listed = if settled || receiver {
            picked.iter().filter(|p| **p).count()
        } else {
            definitions
        };
        for (row, keep) in candidates.into_iter().zip(&picked) {
            if (settled || receiver) && !*keep {
                continue;
            }
            let confidence = if settled && row.confidence == crate::graph::EXTRACTED {
                crate::graph::EXTRACTED
            } else if settled {
                crate::graph::RESOLVED
            } else {
                crate::graph::AMBIGUOUS
            };
            out.push(EdgeEnd {
                symbol: row.symbol,
                kind: row.kind,
                confidence: confidence.to_string(),
                definitions: if settled { definitions } else { listed },
                from_path: Some(row.src_path),
                from_line: Some(row.src_line),
                line: row.line,
                means: None,
            });
        }
    }
    Ok(out)
}

/// Whether an edge kind says where a definition sits rather than what it uses.
fn structural(kind: &str) -> bool {
    matches!(kind, "defines" | "contains")
}

/// [`choose`] for an edge of `kind`, over only the candidates it can reach
/// (see [`reachable`]).
fn choose_for(
    db: &Connection,
    kind: &str,
    candidates: &[&SymbolRow],
    src_path: &str,
    hint: Option<&str>,
    caller_top_level: bool,
    imports: &mut std::collections::HashMap<String, Vec<String>>,
) -> Result<(Vec<bool>, bool)> {
    let reach = reachable(kind, candidates, src_path, hint, caller_top_level);
    if reach.iter().all(|r| *r) {
        return choose(db, candidates, src_path, hint, imports);
    }
    let callable: Vec<&SymbolRow> = candidates
        .iter()
        .zip(&reach)
        .filter(|(_, r)| **r)
        .map(|(c, _)| *c)
        .collect();
    let (picked, settled) = choose(db, &callable, src_path, hint, imports)?;
    let mut picked = picked.into_iter();
    let keep = reach
        .iter()
        .map(|r| *r && picked.next().unwrap_or(false))
        .collect();
    Ok((keep, settled))
}

/// Which of `candidates` an edge of `kind` written in `src_path` with `hint`
/// can mean at all, before anything is ranked.
///
/// A call never means a module: ripgrep's main.rs declares `mod search;`
/// beside `fn search`, and the module made the bare call `search(&args, mode)`
/// a tie. Modules are candidates only when nothing else answers to the name.
///
/// And a call with no receiver never means a method where the language has no
/// implicit receiver ([`crate::graph::no_implicit_receiver`]): `lane(id)` beside
/// `impl Enabled { fn lane(&self, ..) }` in accel.rs, or
/// `crate::accel::lane("ane")` in daemon.rs, can only be the free fn
/// `accel::lane`, and the tie with `Enabled::lane` hid both callers from the
/// cloud's impact check when semlith#245 changed its return type. The mirror
/// of [`through_receiver`], which never lets a receiver call mean a free
/// function. Only when a free function answers to the name: with none, the
/// call may leave the corpus and is ranked as before. A hint that reads as a
/// type (`Enabled::lane(..)`, `Self::new`) is a path to a method and keeps
/// every candidate. In any language, a bare call made at a file's top level
/// (`caller_top_level`, see [`top_level`]) has no `this` to imply either:
/// `helper()` in Kotlin's top-level `main` is not `class A`'s member.
fn reachable(
    kind: &str,
    candidates: &[&SymbolRow],
    src_path: &str,
    hint: Option<&str>,
    caller_top_level: bool,
) -> Vec<bool> {
    let mut keep = vec![true; candidates.len()];
    if kind != "calls" {
        return keep;
    }
    if candidates.iter().any(|c| c.kind != "module") {
        for (k, c) in keep.iter_mut().zip(candidates) {
            *k = c.kind != "module";
        }
    }
    let lang = crate::filter::language_of_path(src_path).map(|l| l.name);
    // A Rust file under a crate's `tests/`, `benches/` or `examples/` is a
    // crate of its own, reachable only from beside it: the cloud's
    // `semlith::accel::lane(..)` tied src/accel.rs's `lane` with a helper of
    // that name in tests/accel.rs. ponytail: a `tests` module directory
    // inside `src/` reads as one too; only test code calls into it.
    if lang == Some("rust") {
        let reached = |p: &str| {
            !p.ends_with(".rs") || separate_crate(p).is_none_or(|dir| src_path.starts_with(dir))
        };
        if candidates
            .iter()
            .zip(&keep)
            .any(|(c, k)| *k && reached(&c.path))
        {
            for (k, c) in keep.iter_mut().zip(candidates) {
                *k = *k && reached(&c.path);
            }
        }
    }
    // Go writes `pkg.F()` and `t.F()` alike, and JavaScript `ns.f()` and
    // `obj.f()`: only the extractor knows which leads are receivers, so
    // there only a call with no lead at all is bare.
    let bare = match lang {
        Some("go" | "javascript" | "typescript" | "svelte" | "vue") => hint.is_none(),
        _ => hint.is_none_or(|h| {
            !h.starts_with(crate::graph::RECEIVER)
                && !h.chars().next().is_some_and(char::is_uppercase)
        }),
    };
    let receiver_free = lang.is_some_and(crate::graph::no_implicit_receiver);
    let no_this = (bare && receiver_free) || (caller_top_level && hint.is_none());
    // A module path drops methods only for a free function it places: in
    // flask's drive, `d.fixtures.bulk()` named fixtures.py, which holds the
    // method `Fixtures.bulk` and no free `bulk`.
    let placed = |c: &SymbolRow| {
        hint.is_none_or(|h| names_file(h, &c.path, src_path) || owner_of(&c.qualified) == Some(h))
    };
    let free_left = candidates
        .iter()
        .zip(&keep)
        .any(|(c, k)| *k && free_function(c) && placed(c));
    if no_this && free_left {
        for (k, c) in keep.iter_mut().zip(candidates) {
            *k = *k && !owned_by_type(c);
        }
    }
    keep
}

/// Which of `candidates` an edge written in `src_path` with `hint` means, and
/// whether that settled it.
///
/// Every candidate at the best rank is kept, so one `true` is a settled edge
/// and several are an ambiguous one — except through a receiver, which
/// [`through_receiver`] decides. The imports of the calling file are read
/// only when the first pass left more than one and something unplaced, and
/// cached in `imports` across calls.
///
/// A tie between candidates that a signal did place — two `fleet.rs` files
/// both named by an import, one of them a copy under `tests/fixtures` — goes
/// to the one sharing the most directories with the caller. A tie that
/// nothing placed stays a tie: nearness is a reason to prefer one of two
/// plausible answers, not a reason to pick one of twenty-four `get`s.
fn choose(
    db: &Connection,
    candidates: &[&SymbolRow],
    src_path: &str,
    hint: Option<&str>,
    imports: &mut std::collections::HashMap<String, Vec<String>>,
) -> Result<(Vec<bool>, bool)> {
    if let Some(lead) = hint.and_then(|h| h.strip_prefix(crate::graph::RECEIVER)) {
        return Ok(through_receiver(candidates, src_path, lead));
    }
    // A module path (`std::env::current_dir()`, `os.getcwd()`) says where the
    // target lives, so one that names no candidate's file, module or import
    // went somewhere the store does not hold: the edge stays unsettled even
    // with one candidate left. ripgrep's `std::env::current_dir()` was
    // resolved onto `hiargs::current_dir` this way.
    let path_hint = hint.filter(|h| !h.chars().next().is_some_and(char::is_uppercase));
    if candidates.len() == 1 && path_hint.is_none() {
        return Ok((vec![true], true));
    }
    let mut ranks: Vec<u8> = candidates
        .iter()
        .map(|c| rank(c, src_path, hint, &[]))
        .collect();
    if (survivors(&ranks) != 1 || path_hint.is_some()) && ranks.contains(&UNRANKED) {
        if !imports.contains_key(src_path) {
            let found = file_imports(db, src_path)?;
            imports.insert(src_path.to_string(), found);
        }
        let of_file = &imports[src_path];
        ranks = candidates
            .iter()
            .map(|c| rank(c, src_path, hint, of_file))
            .collect();
    }
    let best = ranks.iter().copied().min().unwrap_or(UNRANKED);
    let mut keep: Vec<bool> = ranks.iter().map(|r| *r == best).collect();
    if path_hint.is_some() && best == UNRANKED {
        return Ok((keep, false));
    }
    if best < UNRANKED && keep.iter().filter(|k| **k).count() > 1 {
        let near: Vec<usize> = candidates
            .iter()
            .map(|c| shared_directories(&c.path, src_path))
            .collect();
        let nearest = near
            .iter()
            .zip(&keep)
            .filter(|(_, k)| **k)
            .map(|(n, _)| *n)
            .max()
            .unwrap_or(0);
        for (k, n) in keep.iter_mut().zip(&near) {
            *k = *k && *n == nearest;
        }
    }
    let settled = keep.iter().filter(|k| **k).count() == 1;
    Ok((keep, settled))
}

/// The directory, separator included, that makes a Rust file a crate of its
/// own: `semlith/tests/` for `semlith/tests/accel.rs`. `None` for a file of
/// the library or binary.
fn separate_crate(path: &str) -> Option<&str> {
    let mut start = 0;
    for (i, c) in path.char_indices() {
        if c == '/' || c == '\\' {
            if matches!(&path[start..i], "tests" | "benches" | "examples") {
                return Some(&path[..=i]);
            }
            start = i + 1;
        }
    }
    None
}

/// Whether a definition sits at its file's top level, where no `this` is in
/// scope: the file's module symbol, or a function the file's own module name
/// qualifies (`accel::lane` in accel.rs, `A::helper` in A.kt).
fn top_level(kind: &str, qualified: &str, path: &str) -> bool {
    let module = std::path::Path::new(path).file_stem();
    kind == "module"
        || (kind == "function" && owner_of(qualified).is_none_or(|o| module == Some(o.as_ref())))
}

/// Whether a definition is a function no receiver can reach: not a method,
/// and not owned by a type (`main::run`, Python's `helpers::make_response`,
/// a function in `mod tests`). A method's owner is a type, upper case first —
/// unless it is the file's own module name: `fun helper()` at the top of
/// A.kt is qualified `A::helper`, and is no member of anything.
fn free_function(c: &SymbolRow) -> bool {
    let module = std::path::Path::new(&c.path).file_stem();
    c.kind == "function"
        && owner_of(&c.qualified).is_none_or(|o| {
            !o.chars().next().is_some_and(char::is_uppercase) || module == Some(o.as_ref())
        })
}

/// What only a receiver can call: a method, or a function a type owns
/// (Python's `C::f`).
fn owned_by_type(c: &SymbolRow) -> bool {
    c.kind == "method" || (c.kind == "function" && !free_function(c))
}

/// What a call through a receiver (`x.run()`, `self.flush()`) means.
///
/// Never a free function: `.build_parallel().run(||` in ripgrep's main.rs was
/// settled on `main::run` because that `run` sat in the calling file, which put
/// a false two-hop chain into `path` (2026-10-14 audit, G1). It settles only
/// on a method of the receiver's stated type — the ranking signals that place
/// a bare call, the calling file and its imports, say nothing about which type
/// a receiver is — and a receiver nothing typed leaves the edge ambiguous
/// among the methods of that name. Typing it is a type checker's job, which
/// semlith is not. One exception, for Python only, where `x.f()` is also how a
/// module's function is called: a lower-case receiver naming the file of
/// exactly one free function, as `views.render()` names `views.py`.
fn through_receiver(candidates: &[&SymbolRow], src_path: &str, lead: &str) -> (Vec<bool>, bool) {
    // What a receiver can call: a method, or a function a type owns. With
    // none in the store the call leaves the corpus, and the edge lists what
    // shares the name, never a module (`searcher.search()` is not search.rs).
    let methods: Vec<bool> = candidates.iter().map(|c| owned_by_type(c)).collect();
    let callable: Vec<bool> = candidates.iter().map(|c| c.kind != "module").collect();
    let pool = if methods.contains(&true) {
        methods
    } else if callable.contains(&true) {
        callable
    } else {
        vec![true; candidates.len()]
    };
    let upper = lead.chars().next().is_some_and(char::is_uppercase);
    let python = |p: &str| p.ends_with(".py") || p.ends_with(".pyi");
    if python(src_path) && !lead.is_empty() && !upper {
        let module: Vec<bool> = candidates
            .iter()
            .map(|c| free_function(c) && names_file(lead, &c.path, src_path))
            .collect();
        if module.iter().filter(|m| **m).count() == 1 {
            return (module, true);
        }
    }
    if upper {
        let mut typed: Vec<bool> = candidates
            .iter()
            .zip(&pool)
            .map(|(c, p)| *p && owned_by_type(c) && owner_of(&c.qualified) == Some(lead))
            .collect();
        if typed.iter().filter(|t| **t).count() > 1 {
            let near: Vec<usize> = candidates
                .iter()
                .map(|c| shared_directories(&c.path, src_path))
                .collect();
            let nearest = near
                .iter()
                .zip(&typed)
                .filter(|(_, t)| **t)
                .map(|(n, _)| *n)
                .max()
                .unwrap_or(0);
            for (t, n) in typed.iter_mut().zip(&near) {
                *t = *t && *n == nearest;
            }
        }
        match typed.iter().filter(|t| **t).count() {
            1 => return (typed, true),
            0 => {}
            _ => return (typed, false),
        }
    }
    // `this.close()` where `this` is an object literal or a function's own
    // object, a type the extractor cannot name: the one method of that name
    // in the calling file is the one meant.
    if lead.is_empty() {
        let here: Vec<bool> = candidates
            .iter()
            .zip(&pool)
            .map(|(c, p)| *p && c.path == src_path)
            .collect();
        if here.iter().filter(|h| **h).count() == 1 {
            return (here, true);
        }
    }
    (pool, false)
}

/// How many leading directories two paths share.
fn shared_directories(a: &str, b: &str) -> usize {
    let a = std::path::Path::new(a).parent();
    let b = std::path::Path::new(b).parent();
    match (a, b) {
        (Some(a), Some(b)) => a
            .components()
            .zip(b.components())
            .take_while(|(x, y)| x == y)
            .count(),
        _ => 0,
    }
}

/// The rank of a candidate that nothing placed. Sorts last, and is what says
/// "the imports are worth reading for this group".
const UNRANKED: u8 = 4;

/// How many candidates sit at the best rank.
fn survivors(ranks: &[u8]) -> usize {
    match ranks.iter().min() {
        Some(best) => ranks.iter().filter(|r| *r == best).count(),
        None => 0,
    }
}

/// The type or module a definition's `qualified` name puts it in:
/// `Fleet` for `Fleet::search_preferring`, `store` for `store::edges_out`.
pub fn owner_of(qualified: &str) -> Option<&str> {
    let (owner, _) = qualified.rsplit_once("::")?;
    Some(owner.rsplit("::").next().unwrap_or(owner))
}

/// Where a candidate definition sits in the ranking, lower being better.
///
/// 0. The type the source says the receiver is. `self.search_preferring()`
///    inside `impl Semlith`, or `x.search_preferring()` with `x: Fleet`, is a
///    statement about which method, stronger than where the call sits. Only a
///    hint that reads as a type (upper case first) counts here: a variable
///    named `fleet` is a lead about a file, and tier 2 reads it as one.
/// 1. The same file as the call. A file that defines a name and calls it means
///    its own — unless the call went through a module path that names
///    another file.
/// 2. A file the hint names. `store::edges_out` in the presence of
///    `src/store.rs` is not a coincidence.
/// 3. A file the calling file imports. Weaker than the hint because an import
///    list is a set of possibilities rather than a statement about this call.
/// 4. Nothing placed it.
fn rank(candidate: &SymbolRow, src_path: &str, hint: Option<&str>, imports: &[String]) -> u8 {
    let typed = hint.filter(|h| h.chars().next().is_some_and(char::is_uppercase));
    // A top-level function its upper-case file names (`A::helper` in A.kt)
    // shares the qualified name of `class A`'s member, and is no member.
    let in_type = !top_level(&candidate.kind, &candidate.qualified, &candidate.path);
    if in_type && typed.is_some_and(|t| owner_of(&candidate.qualified) == Some(t)) {
        return 0;
    }
    // An inline `mod m { fn f() }` is named by `m::f()` as surely as m.rs is.
    let in_named_module = hint.is_some_and(|h| owner_of(&candidate.qualified) == Some(h));
    let candidate = candidate.path.as_str();
    // A module path (`flask.Flask`, `index::read`) names where the target
    // is, so the calling file's own namesake wins only when the path names
    // the calling file: in flask's test_config.py, `flask.Flask(...)` was
    // settled on a test's subclass defined a few lines up.
    let path_hint = hint.filter(|h| {
        !h.starts_with(crate::graph::RECEIVER) && !h.chars().next().is_some_and(char::is_uppercase)
    });
    let names_caller = |h: &str| {
        std::path::Path::new(src_path)
            .file_stem()
            .is_some_and(|stem| stem == h)
    };
    if candidate == src_path && path_hint.is_none_or(names_caller) {
        return 1;
    }
    if in_named_module || hint.is_some_and(|h| names_file(h, candidate, src_path)) {
        return 2;
    }
    // Split on anything that is not part of a name, so a grouped
    // `use super::{gdrive::{url, get_json}, oauth::TokenSet}` yields
    // `gdrive` rather than one segment of braces, commas and newlines. A
    // call that names its module counts only the imports that name it too:
    // `std::env::current_dir()` is not placed by `use crate::dir::{..}`.
    fn segments(i: &str) -> Vec<&str> {
        i.split(|c: char| !(c.is_alphanumeric() || c == '_' || c == '-'))
            .filter(|s| !s.is_empty())
            .collect()
    }
    if imports.iter().any(|i| {
        let parts = segments(i);
        path_hint.is_none_or(|h| parts.contains(&h))
            && parts.iter().any(|s| names_file(s, candidate, src_path))
    }) {
        return 3;
    }
    // A same-language tier was tried here and removed
    // (US-SEMLITH-0.17.0-I01). The reasoning was sound — semlith's own portal
    // is five thousand lines of JavaScript sharing twenty-five symbol names
    // with the Rust beside it, and a Rust `calls` edge does not mean a
    // JavaScript function — but the harness scored it two hits worse: it
    // settles an ambiguous name in favour of a same-language candidate that is
    // not always the right one, and a confidently wrong edge costs more than a
    // refused one. Ambiguity across languages is a ranking problem, and it is
    // not solved by preferring the nearest guess.
    UNRANKED
}

/// Whether `word` names the file at `path` — its stem, or one of its
/// directories below the ones it shares with `from`, the calling file.
///
/// `store` names `src/store.rs` and `src/store/mod.rs` alike, which is the
/// point: a module is a file or a directory depending on how the author felt
/// that day, and a hint knows neither. A directory both files sit in names
/// nothing: flask's tests call `flask.Flask(...)`, the repository's own folder
/// is called `flask`, and every candidate under it — a test's subclass of
/// `Flask` included — read as named by the hint, so the nearest won as
/// "extracted" (72 such edges in the 2026-10-14 audit's flask store).
fn names_file(word: &str, path: &str, from: &str) -> bool {
    let path = std::path::Path::new(path);
    if path.file_stem().is_some_and(|stem| stem == word) {
        return true;
    }
    let shared = shared_directories(&path.to_string_lossy(), from);
    path.parent()
        .into_iter()
        .flat_map(|parent| parent.components().skip(shared))
        .any(|component| component.as_os_str() == word)
}

/// Everything one file imports, as the raw strings the extractor recorded.
fn file_imports(db: &Connection, path: &str) -> Result<Vec<String>> {
    let mut stmt = db.prepare(
        "SELECT DISTINCT e.dst FROM edges e
         JOIN symbols s ON s.id = e.src
         JOIN files f ON f.id = s.file_id
         WHERE f.path = ?1 AND e.kind = 'imports'",
    )?;
    let rows = stmt.query_map(params![path], |r| r.get(0))?;
    Ok(rows.collect::<Result<Vec<_>, _>>()?)
}

/// Targets of `name`'s edges that the store holds no definition for.
///
/// [`edges_out`] joins to `symbols`, so an edge pointing at something outside
/// the corpus — a standard-library call, a crate that was never indexed —
/// simply does not appear there. That is the right default: a list of names
/// the store knows nothing about is noise in an answer about this codebase.
/// It is not the right *only* option, because "semlith shows no callees" and
/// "everything this calls lives outside the index" are different facts, and
/// only one of them means the graph is working.
pub fn unresolved_out(db: &Connection, name: &str, kinds: &[String]) -> Result<Vec<Unresolved>> {
    let filter = kind_predicate(kinds, "e.kind");
    let sql = format!(
        "SELECT DISTINCT e.dst, e.kind, e.confidence
         FROM symbols src
         JOIN edges e ON e.src = src.id
         WHERE src.name = ?1 AND {filter}
           AND NOT EXISTS (SELECT 1 FROM symbols s WHERE s.name = e.dst)
         ORDER BY e.dst"
    );
    let mut stmt = db.prepare(&sql)?;
    let mut binds: Vec<Value> = vec![Value::Text(name.to_string())];
    binds.extend(kinds.iter().map(|k| Value::Text(k.clone())));
    let rows = stmt.query_map(rusqlite::params_from_iter(binds), |r| {
        Ok(Unresolved {
            name: r.get(0)?,
            kind: r.get(1)?,
            confidence: r.get(2)?,
        })
    })?;
    Ok(rows.collect::<Result<Vec<_>, _>>()?)
}

/// An edge whose target this store has no definition for.
#[derive(Debug, Clone, serde::Serialize)]
pub struct Unresolved {
    pub name: String,
    pub kind: String,
    pub confidence: String,
}

/// What points at `name`: callers, importers, references in.
///
/// The direction a reverse walk takes, and the reason `edges_dst` exists —
/// `semlith neighbors` reads it to answer "what calls this".
pub fn edges_in(db: &Connection, name: &str, kinds: &[String]) -> Result<Vec<EdgeEnd>> {
    let filter = kind_predicate(kinds, "e.kind");
    let sql = format!(
        "SELECT {SYMBOL_COLUMNS}, e.kind, e.confidence, e.line
         FROM edges e
         JOIN symbols s ON s.id = e.src
         JOIN files f ON f.id = s.file_id
         WHERE e.dst = ?1 AND {filter}
         ORDER BY f.path, s.start_line"
    );
    let mut stmt = db.prepare(&sql)?;
    let mut binds: Vec<Value> = vec![Value::Text(name.to_string())];
    binds.extend(kinds.iter().map(|k| Value::Text(k.clone())));
    let rows = stmt.query_map(rusqlite::params_from_iter(binds), |r| {
        Ok(EdgeEnd {
            symbol: symbol_row(r)?,
            kind: r.get(8)?,
            confidence: r.get(9)?,
            // A caller is a symbol, not a name: the row came from the edge's
            // own `src` id, so there is nothing to resolve and nothing to be
            // ambiguous about. This direction is unchanged from 0.14.0.
            definitions: 1,
            from_path: None,
            from_line: None,
            line: r.get(10)?,
            means: None,
        })
    })?;
    Ok(rows.collect::<Result<Vec<_>, _>>()?)
}

/// At most `limit` of [`edges_in`], the extractor's certain edges first.
///
/// For a drawing, which has room for a few dozen callers: `new` has hundreds
/// of thousands of them on the 879k-chunk corpus, and reading them all to draw
/// sixty took most of the Graph page's twenty seconds.
pub fn edges_in_first(db: &Connection, name: &str, limit: usize) -> Result<Vec<EdgeEnd>> {
    let sql = format!(
        "SELECT {SYMBOL_COLUMNS}, e.kind, e.confidence, e.line
         FROM edges e
         JOIN symbols s ON s.id = e.src
         JOIN files f ON f.id = s.file_id
         WHERE e.dst = ?1
         ORDER BY CASE e.confidence WHEN 'extracted' THEN 0 WHEN 'resolved' THEN 0 WHEN 'ambiguous' THEN 1 ELSE 2 END,
                  f.path, s.start_line
         LIMIT ?2"
    );
    let mut stmt = db.prepare(&sql)?;
    let rows = stmt.query_map(params![name, limit as i64], |r| {
        Ok(EdgeEnd {
            symbol: symbol_row(r)?,
            kind: r.get(8)?,
            confidence: r.get(9)?,
            definitions: 1,
            from_path: None,
            from_line: None,
            line: r.get(10)?,
            means: None,
        })
    })?;
    Ok(rows.collect::<Result<Vec<_>, _>>()?)
}

/// The names with the most edges in or out, busiest first, each counted once
/// on its own index. For the graph page's choice of where to open: a name with
/// seven hundred definitions is one candidate, not seven hundred.
pub fn busiest_names(db: &Connection, limit: usize) -> Result<Vec<String>> {
    let candidates = (limit * 4).max(64) as i64;
    let mut stmt = db.prepare(
        "WITH ins AS (SELECT dst AS name, COUNT(*) AS n FROM edges GROUP BY dst ORDER BY n DESC LIMIT ?1),
              outs AS (SELECT s.name AS name, COUNT(*) AS n FROM edges e JOIN symbols s ON s.id = e.src
                       WHERE s.kind != 'module' GROUP BY s.name ORDER BY n DESC LIMIT ?1)
         SELECT name FROM (SELECT name, n FROM ins UNION ALL SELECT name, n FROM outs)
         GROUP BY name ORDER BY SUM(n) DESC, name LIMIT ?2",
    )?;
    let rows = stmt.query_map(params![candidates, limit as i64], |r| r.get::<_, String>(0))?;
    Ok(rows.collect::<Result<Vec<_>, _>>()?)
}

/// One edge into a name, with the definitions of that name it resolves to.
#[derive(Debug, Clone)]
pub struct Incoming {
    /// The caller, as [`edges_in`] returns it, with the confidence the
    /// resolution earned rather than the one stored.
    pub end: EdgeEnd,
    /// Ids of the definitions of the target name this edge means: one when it
    /// settled, several when it did not.
    pub means: Vec<i64>,
}

/// What points at `name`, each edge resolved against the name's definitions
/// the way [`edges_out`] resolves it from the other end.
///
/// [`edges_in`] answers "what calls something called `search_preferring`",
/// which on a store holding two such methods is two answers mixed together.
/// This is what lets impact walk from one definition: an edge is kept for a
/// definition only when the ranking says the edge means it.
pub fn edges_in_resolved(db: &Connection, name: &str, kinds: &[String]) -> Result<Vec<Incoming>> {
    let filter = kind_predicate(kinds, "e.kind");
    let sql = format!(
        "SELECT {SYMBOL_COLUMNS}, e.kind, e.confidence, e.line, e.hint
         FROM edges e
         JOIN symbols s ON s.id = e.src
         JOIN files f ON f.id = s.file_id
         WHERE e.dst = ?1 AND {filter}
         ORDER BY f.path, s.start_line"
    );
    let mut stmt = db.prepare(&sql)?;
    let mut binds: Vec<Value> = vec![Value::Text(name.to_string())];
    binds.extend(kinds.iter().map(|k| Value::Text(k.clone())));
    let rows = stmt.query_map(rusqlite::params_from_iter(binds), |r| {
        Ok((
            symbol_row(r)?,
            r.get::<_, String>(8)?,
            r.get::<_, String>(9)?,
            r.get::<_, Option<u32>>(10)?,
            r.get::<_, Option<String>>(11)?,
        ))
    })?;
    let rows = rows.collect::<Result<Vec<_>, _>>()?;
    if rows.is_empty() {
        return Ok(Vec::new());
    }
    let navigational = crate::graph::not_navigational("s.kind");
    let sql = format!(
        "SELECT {SYMBOL_COLUMNS} FROM symbols s JOIN files f ON f.id = s.file_id
         WHERE s.name = ?1 AND {navigational} ORDER BY f.path, s.start_line"
    );
    let mut stmt = db.prepare(&sql)?;
    let definitions = stmt
        .query_map(params![name], symbol_row)?
        .collect::<Result<Vec<_>, _>>()?;
    let refs: Vec<&SymbolRow> = definitions.iter().collect();
    let mut imports = std::collections::HashMap::new();
    let mut out = Vec::with_capacity(rows.len());
    for (caller, kind, stored, line, hint) in rows {
        // A structural edge is the file saying where its own definition sits:
        // nothing to rank, and nothing in another file it could mean. Ranked,
        // a YAML `steps` holding its `run:` key read as maybe containing one
        // of ripgrep's Rust `run`s.
        if structural(&kind) {
            let means: Vec<i64> = definitions
                .iter()
                .filter(|d| d.path == caller.path && line.is_none_or(|l| l == d.start_line))
                .map(|d| d.id)
                .collect();
            out.push(Incoming {
                end: EdgeEnd {
                    symbol: caller,
                    kind,
                    confidence: stored,
                    definitions: definitions.len(),
                    from_path: None,
                    from_line: None,
                    line,
                    means: None,
                },
                means,
            });
            continue;
        }
        let (picked, settled) = if refs.is_empty() {
            (Vec::new(), false)
        } else {
            choose_for(
                db,
                &kind,
                &refs,
                &caller.path,
                hint.as_deref(),
                top_level(&caller.kind, &caller.qualified, &caller.path),
                &mut imports,
            )?
        };
        let means: Vec<i64> = definitions
            .iter()
            .zip(&picked)
            .filter(|(_, p)| **p)
            .map(|(d, _)| d.id)
            .collect();
        let confidence = if means.is_empty() {
            // No definition at all: an edge to a name the store holds only as
            // a heading or a key. Kept as stored so nothing is invented.
            stored.as_str()
        } else if settled && stored == crate::graph::EXTRACTED {
            crate::graph::EXTRACTED
        } else if settled {
            crate::graph::RESOLVED
        } else {
            crate::graph::AMBIGUOUS
        }
        .to_string();
        out.push(Incoming {
            end: EdgeEnd {
                symbol: caller,
                kind,
                confidence,
                definitions: definitions.len(),
                from_path: None,
                from_line: None,
                line,
                means: None,
            },
            means,
        });
    }
    Ok(out)
}

fn kind_predicate(kinds: &[String], column: &str) -> String {
    if kinds.is_empty() {
        return "1".to_string();
    }
    let holes = vec!["?"; kinds.len()].join(", ");
    format!("{column} IN ({holes})")
}

/// One recorded retrieval.
#[derive(Debug, Clone, serde::Serialize)]
pub struct Retrieval {
    pub id: i64,
    pub at: i64,
    pub client: String,
    pub query: String,
    pub hits: i64,
    pub micros: i64,
    pub excerpt_tokens: i64,
    pub whole_file_tokens: i64,
    pub hash: String,
    /// Which stores answered this one retrieval, so several rows can be shown
    /// as the one search they came from. Empty for a row written before
    /// 0.20.2, which was its own retrieval.
    pub query_id: String,
    /// The model request behind the call, once a reader has looked for it.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub usage: Option<RowUsage>,
}

/// A row's usage columns as they are stored. Every figure is optional: a
/// client that logs the model and not the tokens gets the model.
#[derive(Debug, Clone, Default, PartialEq, serde::Serialize)]
pub struct RowUsage {
    pub model: Option<String>,
    pub input_tokens: Option<i64>,
    pub output_tokens: Option<i64>,
    pub cache_read_tokens: Option<i64>,
    pub cache_write_tokens: Option<i64>,
    pub reasoning_tokens: Option<i64>,
    pub cost_usd: Option<f64>,
    pub cost_source: Option<String>,
    /// The log the figures came from, or why there are none.
    pub source: String,
}

/// The usage columns, from column `at` of a row onwards, in the order
/// `USAGE_COLUMNS` lists them. `None` until a reader has looked.
pub const USAGE_COLUMNS: &str = "model, input_tokens, output_tokens, cache_read_tokens, \
     cache_write_tokens, reasoning_tokens, cost_usd, cost_source, usage_source";

pub fn usage_at(r: &rusqlite::Row<'_>, at: usize) -> rusqlite::Result<Option<RowUsage>> {
    let Some(source) = r.get::<_, Option<String>>(at + 8)? else {
        return Ok(None);
    };
    Ok(Some(RowUsage {
        model: r.get(at)?,
        input_tokens: r.get(at + 1)?,
        output_tokens: r.get(at + 2)?,
        cache_read_tokens: r.get(at + 3)?,
        cache_write_tokens: r.get(at + 4)?,
        reasoning_tokens: r.get(at + 5)?,
        cost_usd: r.get(at + 6)?,
        cost_source: r.get(at + 7)?,
        source,
    }))
}

/// Append one retrieval, chained to the row before it.
///
/// The chain is the point: a row cannot be quietly edited or removed without
/// every hash after it failing to recompute.
pub fn record_retrieval(db: &Connection, row: &NewRetrieval<'_>) -> Result<()> {
    record_counted(db, row, BASIS_BYTES)
}

/// The connections holding a run's snapshot ([`pin_snapshot`]), by address.
///
/// A reader's connection lives inside a `Semlith` inside a fleet behind its
/// reader's mutex, and does not move between the pin and the release, so its
/// address names it for exactly as long as the mark matters.
static PINNED: std::sync::Mutex<Vec<usize>> = std::sync::Mutex::new(Vec::new());

fn address(db: &Connection) -> usize {
    std::ptr::from_ref(db) as usize
}

/// Hold this reader at the store as it is now, until [`unpin_snapshot`].
///
/// A run re-indexing files the store already holds deletes a file's rows and
/// writes its new ones file by file, and embeds them after; a reader between
/// those moments answered from half a run — old chunks gone, new ones not yet
/// embedded (item 17). A WAL reader inside one read transaction sees one
/// committed state however much is committed after it, so the daemon opens
/// one on every reader before the run's first write and ends it when the run
/// commits. The cost is the WAL keeping the run's pages until then.
pub fn pin_snapshot(db: &Connection) -> Result<()> {
    if !db.is_autocommit() {
        return Ok(());
    }
    db.execute_batch("BEGIN DEFERRED")?;
    // A deferred transaction takes its snapshot at its first read.
    let _: i64 = db.query_row("SELECT COUNT(*) FROM meta", [], |r| r.get(0))?;
    PINNED
        .lock()
        .unwrap_or_else(|e| e.into_inner())
        .push(address(db));
    Ok(())
}

/// End [`pin_snapshot`]'s read transaction: the reader sees the run's commit.
pub fn unpin_snapshot(db: &Connection) -> Result<()> {
    let mut pinned = PINNED.lock().unwrap_or_else(|e| e.into_inner());
    let at = address(db);
    if let Some(i) = pinned.iter().position(|a| *a == at) {
        pinned.swap_remove(i);
        drop(pinned);
        if !db.is_autocommit() {
            db.execute_batch("COMMIT")?;
        }
    }
    Ok(())
}

/// Whether this connection is holding a run's snapshot.
pub fn is_pinned(db: &Connection) -> bool {
    PINNED
        .lock()
        .unwrap_or_else(|e| e.into_inner())
        .contains(&address(db))
}

/// A connection of its own for a write a pinned reader must make — a ledger
/// row, a usage fill — which its read transaction cannot: a write there would
/// fail as soon as the run had committed anything since the pin.
fn beside(db: &Connection) -> Result<Option<Connection>> {
    if !is_pinned(db) {
        return Ok(None);
    }
    match db.path().filter(|p| !p.is_empty()) {
        Some(path) => Ok(Some(open(Path::new(path))?)),
        None => Ok(None),
    }
}

/// What the whole-file side of a row was counted from: the file's bytes.
pub const BASIS_BYTES: &str = "bytes";
/// What the whole-file side of a row was counted from: the text a reader
/// extracted from it, counted the way the excerpt was.
pub const BASIS_TEXT: &str = "text";

/// [`record_retrieval`], saying what the whole-file side was counted from.
pub fn record_counted(db: &Connection, row: &NewRetrieval<'_>, basis: &str) -> Result<()> {
    if let Some(side) = beside(db)? {
        return record_counted(&side, row, basis);
    }
    // A key pasted as a search is text a person typed, and nothing else in the
    // store keeps one in full: every file passes the scan before it is held.
    // The same scan, the same marker, before the row is hashed, so the chain
    // covers what is stored (#147). A declared dummy stays as typed.
    let found: Vec<crate::keyscan::Match> = crate::keyscan::scan("", row.query)
        .into_iter()
        .filter(|m| m.dummy.is_none())
        .collect();
    let masked_text;
    let masked_row;
    let row = if found.is_empty() {
        row
    } else {
        masked_text = crate::keyscan::redact(row.query, &found);
        masked_row = NewRetrieval {
            query: &masked_text,
            ..row.clone()
        };
        &masked_row
    };
    let (client, query) = (row.client, row.query);
    let (hits, micros) = (row.hits, row.micros);
    let (excerpt_tokens, whole_file_tokens) = (row.excerpt_tokens, row.whole_file_tokens);
    // The ledger is the one write that happens on a read: a retrieval is
    // recorded by the search that answered it. It asks for the same permission
    // an index run does.
    let _writing = Writing::begin(db)?;
    let at = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0);
    let key = ledger_key(db, true);
    append_chained(db, |prev| {
        let hash = v2_hash(key.as_ref(), prev, at, row, basis);
        db.execute(
            "INSERT INTO retrievals
             (at, client, query, hits, micros, excerpt_tokens, whole_file_tokens, prev, hash,
              session, tool, stale_hits, tokenizer, query_id, client_version, whole_basis, chain)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, 2)",
            params![
                at,
                client,
                query,
                hits,
                micros,
                excerpt_tokens,
                whole_file_tokens,
                prev,
                hash,
                row.session,
                row.tool,
                row.stale_hits,
                row.tokenizer,
                row.query_id,
                (!row.client_version.is_empty()).then_some(row.client_version),
                basis,
            ],
        )?;
        set_head(db, key.as_ref(), db.last_insert_rowid(), &hash)?;
        Ok(())
    })?;
    // The one place a retrieval is written, so the one place the portal's
    // ledger counter moves — every surface that records goes through here.
    crate::daemon::changes::bump(crate::daemon::changes::Domain::Ledger);
    Ok(())
}

/// One retrieval, before it has a timestamp or a place in the chain.
///
/// A struct rather than thirteen positional arguments, because two adjacent
/// `i64`s that mean different things are a bug waiting for a refactor.
#[derive(Debug, Clone)]
pub struct NewRetrieval<'a> {
    /// Who asked: one of the documented clients (`Claude Code`, `Zed`, …, see
    /// `clientid::label`), or `cli`, `portal`.
    pub client: &'a str,
    /// The client's own version string. Deliberately not in the chain, for the
    /// reason `query_id` is not: a rolled-back binary recomputes the formula it
    /// knows, and a verify that fails every row written since cries wolf.
    pub client_version: &'a str,
    /// Which conversation, so one agent's session can be read as a unit.
    pub session: &'a str,
    /// Which tool: `search`, `neighbors`, `path`, `symbol`.
    pub tool: &'a str,
    pub query: &'a str,
    pub hits: i64,
    pub micros: i64,
    pub excerpt_tokens: i64,
    pub whole_file_tokens: i64,
    /// How many of those hits came from a file edited since it was indexed.
    pub stale_hits: i64,
    /// What counted the two token figures. Rows counted two different ways are
    /// never summed together, so the label travels with the row rather than
    /// being assumed from its age.
    pub tokenizer: &'a str,
    /// The id shared by every row one retrieval wrote.
    ///
    /// A cross-store search writes a row in each store that answered it, so
    /// each store's chain stays its own and its token figures describe its own
    /// hits. This is what says those rows are one retrieval rather than
    /// several, and it is what "queries recorded" counts.
    pub query_id: &'a str,
}

/// A ledger row whose usage has not been looked for, as the readers in
/// `crate::usage` need it to find the model request that made the call.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Unfilled {
    pub id: i64,
    pub at: i64,
    pub client: String,
    pub session: String,
    pub tool: String,
    pub query: String,
    pub query_id: String,
}

/// Rows written by an AI client since `since` (unix seconds) with no usage
/// yet, oldest first. `cli` and `portal` rows have no model behind them, and
/// neither has a `raw-read` row, which the steering hook writes for a read
/// the agent made without semlith.
pub fn unfilled(db: &Connection, since: i64) -> Result<Vec<Unfilled>> {
    let mut q = db.prepare(
        "SELECT id, at, client, COALESCE(session, ''), COALESCE(tool, ''), query,
                COALESCE(query_id, '')
           FROM retrievals
          WHERE usage_source IS NULL AND at >= ?1 AND client NOT IN ('cli', 'portal')
            AND COALESCE(tool, '') NOT IN ('raw-read', 'note', 'refused')
          ORDER BY at, id",
    )?;
    let rows = q.query_map(params![since], |r| {
        Ok(Unfilled {
            id: r.get(0)?,
            at: r.get(1)?,
            client: r.get(2)?,
            session: r.get(3)?,
            tool: r.get(4)?,
            query: r.get(5)?,
            query_id: r.get(6)?,
        })
    })?;
    Ok(rows.collect::<Result<_, _>>()?)
}

/// What a reader found for one row: the usage, or only why there is none.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct Filled {
    pub model: Option<String>,
    pub tokens: Option<crate::prices::Tokens>,
    pub cost_usd: Option<f64>,
    pub cost_source: Option<String>,
    /// The log it came from, or why nothing did (`not recorded by Zed`).
    pub usage_source: String,
}

/// Write a row's usage. Every row sharing its `query_id` is the same call,
/// written once per store that answered, so all of them are filled.
pub fn fill_usage(db: &Connection, row: &Unfilled, filled: &Filled) -> Result<()> {
    if let Some(side) = beside(db)? {
        return fill_usage(&side, row, filled);
    }
    let _writing = Writing::begin(db)?;
    let t = filled.tokens;
    db.execute(
        "UPDATE retrievals
            SET model = ?1, input_tokens = ?2, output_tokens = ?3, cache_read_tokens = ?4,
                cache_write_tokens = ?5, reasoning_tokens = ?6, cost_usd = ?7,
                cost_source = ?8, usage_source = ?9
          WHERE id = ?10 OR (?11 != '' AND query_id = ?11)",
        params![
            filled.model,
            t.map(|t| t.input),
            t.map(|t| t.output),
            t.map(|t| t.cache_read),
            t.map(|t| t.cache_write),
            t.map(|t| t.reasoning),
            filled.cost_usd,
            filled.cost_source,
            filled.usage_source,
            row.id,
            row.query_id,
        ],
    )?;
    let ids: Vec<i64> = {
        let mut q =
            db.prepare("SELECT id FROM retrievals WHERE id = ?1 OR (?2 != '' AND query_id = ?2)")?;
        q.query_map(params![row.id, row.query_id], |r| r.get(0))?
            .collect::<Result<_, _>>()?
    };
    reseal(db, &ids)?;
    Ok(())
}

/// One ledger row as ledger sync sends it: what was retrieved, never the
/// query text, which this struct does not carry.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SyncRow {
    pub id: i64,
    pub at: i64,
    pub client: String,
    pub session: String,
    pub tool: String,
    pub excerpt_tokens: i64,
    /// What a saving is credited against (0.43.0, cloud audit C26): without
    /// these a member's local use reached the org's ledger as net 0.
    pub hits: i64,
    pub whole_file_tokens: i64,
    /// `measured` or `modelled`, as [`SessionRow::tier`] says it per row.
    pub tier: &'static str,
    /// The id one retrieval shares across stores, so the cloud counts a
    /// cross-store call once. `None` on rows written before 0.20.2.
    pub query_id: Option<String>,
}

/// Rows written at or after `since` and not yet sent, oldest first, at most
/// `limit`.
pub fn unsynced(db: &Connection, since: i64, limit: usize) -> Result<Vec<SyncRow>> {
    let mut q = db.prepare(
        "SELECT id, at, client, COALESCE(session, ''), COALESCE(tool, ''), excerpt_tokens,
                hits, whole_file_tokens, tokenizer IS NOT NULL AND tokenizer <> 'chars4',
                query_id
           FROM retrievals
          WHERE synced_at IS NULL AND at >= ?1
          ORDER BY id LIMIT ?2",
    )?;
    let rows = q.query_map(params![since, limit as i64], |r| {
        Ok(SyncRow {
            id: r.get(0)?,
            at: r.get(1)?,
            client: r.get(2)?,
            session: r.get(3)?,
            tool: r.get(4)?,
            excerpt_tokens: r.get(5)?,
            hits: r.get(6)?,
            whole_file_tokens: r.get(7)?,
            tier: if r.get(8)? { "measured" } else { "modelled" },
            query_id: r.get(9)?,
        })
    })?;
    Ok(rows.collect::<Result<_, _>>()?)
}

/// How many rows since `since` are still to send.
pub fn unsynced_count(db: &Connection, since: i64) -> Result<i64> {
    Ok(db.query_row(
        "SELECT COUNT(*) FROM retrievals WHERE synced_at IS NULL AND at >= ?1",
        params![since],
        |r| r.get(0),
    )?)
}

/// Record that these rows reached the org's ledger.
pub fn mark_synced(db: &Connection, ids: &[i64], at: i64) -> Result<()> {
    if let Some(side) = beside(db)? {
        return mark_synced(&side, ids, at);
    }
    let _writing = Writing::begin(db)?;
    let tx = rusqlite::Transaction::new_unchecked(db, rusqlite::TransactionBehavior::Immediate)?;
    {
        let mut q = tx.prepare("UPDATE retrievals SET synced_at = ?1 WHERE id = ?2")?;
        for id in ids {
            q.execute(params![at, id])?;
        }
    }
    reseal(&tx, ids)?;
    tx.commit()?;
    Ok(())
}

/// Every row with usage, keyed as [`ledger_keys`] keys it, for totals per
/// client and per model that count a cross-store call once.
pub fn usage_rows(db: &Connection, store: &str) -> Result<Vec<(String, String, RowUsage)>> {
    let mut q = db.prepare(&format!(
        "SELECT id, COALESCE(query_id, ''), client, {USAGE_COLUMNS}
           FROM retrievals WHERE model IS NOT NULL"
    ))?;
    let rows = q.query_map([], |r| {
        let id: i64 = r.get(0)?;
        let qid: String = r.get(1)?;
        let key = if qid.is_empty() {
            format!("{store}#{id}")
        } else {
            qid
        };
        Ok((
            key,
            r.get::<_, String>(2)?,
            usage_at(r, 3)?.unwrap_or_default(),
        ))
    })?;
    Ok(rows.collect::<Result<_, _>>()?)
}

/// The hash covering one row and the one before it.
///
/// # Two formulas
///
/// A row written before 0.15.0 has four columns this one does not, and its
/// hash was computed without them. Recomputing such a row under the new
/// formula would report every 0.14.0 ledger as broken — which is the one thing
/// a verify must never do, because a verify that cries wolf is worse than no
/// verify at all.
///
/// So the formula is chosen per row, by the row itself: `tool` is NULL on
/// every row written before 0.15.0 and set on every row written since. There
/// is no version column and no migration, and a store holding rows of both
/// kinds walks end to end.
/// The file beside a store's database holding its ledger key.
const LEDGER_KEY: &str = "ledger.key";

/// What a keyed row's hash starts with, so a verify knows which formula
/// wrote it.
const KEYED: &str = "k1:";

/// This store's ledger key, made the first time a row is recorded when
/// `create`. A file of its own, 0600, rather than a meta row: whoever can
/// rewrite `store.db` could otherwise recompute every hash after an edit, and
/// a copied or synced database travels without it (audit S24). `None` for an
/// in-memory database or a key that cannot be read or made.
fn ledger_key(db: &Connection, create: bool) -> Option<[u8; 32]> {
    let file = Path::new(db.path().filter(|p| !p.is_empty())?)
        .parent()?
        .join(LEDGER_KEY);
    if let Ok(bytes) = std::fs::read(&file) {
        return bytes.try_into().ok();
    }
    if !create {
        return None;
    }
    let mut key = [0u8; 32];
    getrandom::fill(&mut key).ok()?;
    let mut options = std::fs::OpenOptions::new();
    options.write(true).create_new(true);
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt;
        options.mode(crate::home::FILE_MODE);
    }
    match options.open(&file) {
        Ok(mut out) => {
            std::io::Write::write_all(&mut out, &key).ok()?;
            Some(key)
        }
        // Another process made it first: use theirs.
        Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
            std::fs::read(&file).ok()?.try_into().ok()
        }
        Err(_) => None,
    }
}

/// [`chain_hash`]'s payload under the store's key, marked [`KEYED`].
/// From 0.41.0; an older binary reads these rows as not verifying.
fn keyed_chain_hash(key: &[u8; 32], prev: &str, at: i64, row: &NewRetrieval<'_>) -> String {
    format!(
        "{KEYED}{}",
        blake3::keyed_hash(key, chain_payload(prev, at, row).as_bytes()).to_hex()
    )
}

fn chain_hash(prev: &str, at: i64, row: &NewRetrieval<'_>) -> String {
    blake3::hash(chain_payload(prev, at, row).as_bytes())
        .to_hex()
        .to_string()
}

fn chain_payload(prev: &str, at: i64, row: &NewRetrieval<'_>) -> String {
    let NewRetrieval {
        client,
        query,
        hits,
        micros,
        excerpt_tokens: excerpt,
        whole_file_tokens: whole,
        ..
    } = row;
    let mut payload = format!(
        "{prev}\u{1f}{at}\u{1f}{client}\u{1f}{query}\u{1f}{hits}\u{1f}{micros}\u{1f}{excerpt}\u{1f}{whole}"
    );
    payload.push_str(&format!(
        "\u{1f}{}\u{1f}{}\u{1f}{}\u{1f}{}",
        row.session, row.tool, row.stale_hits, row.tokenizer
    ));
    // `query_id` is deliberately NOT in the payload.
    //
    // Adding it would be a third formula, and a 0.20.1 binary — which is this
    // release's own rollback path — recomputes the second one. Every row
    // written since would then fail to verify under the version somebody
    // rolled back to, which is the one thing a verify must never do: a verify
    // that cries wolf is worse than no verify at all.
    //
    // What that costs is stated rather than hidden: the id can be altered
    // without breaking the chain, so the grouping of rows into retrievals is
    // not tamper-evident. What a retrieval *was* — who asked, what they asked,
    // how many hits, what it cost — is covered exactly as before.
    payload
}

/// What a v2 row's hash starts with: keyed by the store's ledger key, or
/// plain where no key could be made (an in-memory database).
const V2_KEYED: &str = "k2:";
const V2_PLAIN: &str = "v2:";

/// The meta key holding the chain's head: the newest v2 row's id and hash,
/// and a MAC over both, so deleting the newest rows fails `--verify`.
const LEDGER_HEAD: &str = "ledger_head";

/// Chain v2 (0.42.0): every column a figure reads.
///
/// v1 left out `query_id`, so rewriting every id collapsed a ledger's
/// "16 retrievals, coverage 75 %" into "1 of 1, coverage 100 %" and the chain
/// still read intact. It also had no head, so deleting the newest row
/// verified. v2 covers the id, the client's version and the whole-file
/// basis, the row says it is v2 in `chain`, and [`set_head`] records the head.
/// The late columns (usage, `synced_at`) cannot be in a hash written before
/// them; [`reseal`] covers them in `seal` whenever they are written.
///
/// An older binary does not know the formula and reports the first v2 row as
/// the break; rows it wrote itself keep verifying here, under v1.
fn v2_hash(
    key: Option<&[u8; 32]>,
    prev: &str,
    at: i64,
    row: &NewRetrieval<'_>,
    basis: &str,
) -> String {
    let payload = format!(
        "{}\u{1f}{}\u{1f}{}\u{1f}{basis}\u{1f}v2",
        chain_payload(prev, at, row),
        row.query_id,
        row.client_version,
    );
    mac(key, V2_KEYED, V2_PLAIN, &payload)
}

/// A keyed blake3 under the store's key, or a plain one marked as such.
fn mac(key: Option<&[u8; 32]>, keyed: &str, plain: &str, payload: &str) -> String {
    match key {
        Some(key) => format!(
            "{keyed}{}",
            blake3::keyed_hash(key, payload.as_bytes()).to_hex()
        ),
        None => format!("{plain}{}", blake3::hash(payload.as_bytes()).to_hex()),
    }
}

fn head_mac(key: Option<&[u8; 32]>, id: i64, hash: &str) -> String {
    mac(key, "k:", "p:", &format!("head\u{1f}{id}\u{1f}{hash}"))
}

/// Record the newest v2 row as the chain's head, in the append's transaction.
fn set_head(db: &Connection, key: Option<&[u8; 32]>, id: i64, hash: &str) -> Result<()> {
    set_meta(
        db,
        LEDGER_HEAD,
        &format!("{id}:{hash}:{}", head_mac(key, id, hash)),
    )
}

/// The late columns of a v2 row, as `seal` covers them.
fn seal_payload(r: &rusqlite::Row<'_>) -> rusqlite::Result<String> {
    let mut payload: String = r.get::<_, String>(0)?;
    for at in 1..=10 {
        let value: rusqlite::types::Value = r.get(at)?;
        payload.push('\u{1f}');
        payload.push_str(&match value {
            rusqlite::types::Value::Null => String::from("\u{0}"),
            rusqlite::types::Value::Integer(n) => n.to_string(),
            rusqlite::types::Value::Real(f) => format!("{f:?}"),
            rusqlite::types::Value::Text(t) => t,
            rusqlite::types::Value::Blob(b) => blake3::hash(&b).to_hex().to_string(),
        });
    }
    Ok(payload)
}

const SEAL_COLUMNS: &str = "hash, model, input_tokens, output_tokens, cache_read_tokens, \
     cache_write_tokens, reasoning_tokens, cost_usd, cost_source, usage_source, synced_at";

/// Rewrite `seal` on the v2 rows among `ids`, after their late columns moved.
fn reseal(db: &Connection, ids: &[i64]) -> Result<()> {
    let key = ledger_key(db, true);
    let mut read = db.prepare(&format!(
        "SELECT {SEAL_COLUMNS} FROM retrievals WHERE id = ?1 AND chain = 2"
    ))?;
    let mut write = db.prepare("UPDATE retrievals SET seal = ?1 WHERE id = ?2")?;
    for id in ids {
        let Some(payload) = read.query_row([id], seal_payload).optional()? else {
            continue;
        };
        write.execute(params![mac(key.as_ref(), "k:", "p:", &payload), id])?;
    }
    Ok(())
}

/// Append a row in the shape 0.14.0 wrote, for the test that proves a store
/// holding both kinds still verifies.
///
/// Public because that test is an integration test and cannot reach a private
/// function — and because the property it proves is worth proving. A verify
/// that reported every ledger written before this release as broken would be
/// worse than no verify at all, and nothing but a real pre-0.15.0 row in a
/// real store demonstrates that it does not.
pub fn record_legacy_retrieval(
    db: &Connection,
    client: &str,
    query: &str,
    excerpt_tokens: i64,
    whole_file_tokens: i64,
) -> Result<()> {
    let _writing = Writing::begin(db)?;
    let at = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0);
    let (hits, micros) = (1, 1000);
    append_chained(db, |prev| {
        let hash = legacy_chain_hash(
            prev,
            at,
            client,
            query,
            hits,
            micros,
            excerpt_tokens,
            whole_file_tokens,
        );
        // The four 0.15.0 columns are left NULL, which is what makes this a row
        // of the older kind and what `ledger_break` reads to choose the formula.
        db.execute(
            "INSERT INTO retrievals
             (at, client, query, hits, micros, excerpt_tokens, whole_file_tokens, prev, hash)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
            params![
                at,
                client,
                query,
                hits,
                micros,
                excerpt_tokens,
                whole_file_tokens,
                prev,
                hash
            ],
        )?;
        Ok(())
    })
}

/// Append a row the way 0.15.0 to 0.41.0 wrote it — chain v1, no basis, no
/// head — for the test that proves a store holding v1 and v2 rows verifies.
pub fn record_v1_retrieval(db: &Connection, row: &NewRetrieval<'_>) -> Result<()> {
    let _writing = Writing::begin(db)?;
    let at = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0);
    let key = ledger_key(db, true);
    append_chained(db, |prev| {
        let hash = match &key {
            Some(key) => keyed_chain_hash(key, prev, at, row),
            None => chain_hash(prev, at, row),
        };
        db.execute(
            "INSERT INTO retrievals
             (at, client, query, hits, micros, excerpt_tokens, whole_file_tokens, prev, hash,
              session, tool, stale_hits, tokenizer, query_id)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)",
            params![
                at,
                row.client,
                row.query,
                row.hits,
                row.micros,
                row.excerpt_tokens,
                row.whole_file_tokens,
                prev,
                hash,
                row.session,
                row.tool,
                row.stale_hits,
                row.tokenizer,
                row.query_id
            ],
        )?;
        Ok(())
    })
}

/// Read the chain's last hash and append the row chained to it, as one write.
///
/// Two statements on their own were a race. Two connections to one store —
/// the daemon and a `semlith search` from a terminal, or the portal's fleet
/// beside the daemon's `/mcp` fleet — could each read the same last hash and
/// each insert a row chained to it, and `ledger_break` then reported the second
/// as edited or removed though nothing was. `BEGIN IMMEDIATE` takes the write
/// lock before the read, so the second writer waits out the busy timeout for
/// the first and reads its row.
///
/// A caller already inside a transaction keeps it rather than being refused a
/// nested one; its write lock is taken at the insert, as it always was.
fn append_chained(db: &Connection, append: impl FnOnce(&str) -> Result<()>) -> Result<()> {
    let tx = if db.is_autocommit() {
        Some(rusqlite::Transaction::new_unchecked(
            db,
            rusqlite::TransactionBehavior::Immediate,
        )?)
    } else {
        None
    };
    let prev: String = db
        .query_row(
            "SELECT hash FROM retrievals ORDER BY id DESC LIMIT 1",
            [],
            |r| r.get(0),
        )
        .optional()?
        .unwrap_or_default();
    append(&prev)?;
    if let Some(tx) = tx {
        tx.commit()?;
    }
    Ok(())
}

/// The 0.14.0 formula, for rows written under it.
#[allow(clippy::too_many_arguments)]
fn legacy_chain_hash(
    prev: &str,
    at: i64,
    client: &str,
    query: &str,
    hits: i64,
    micros: i64,
    excerpt: i64,
    whole: i64,
) -> String {
    let payload = format!(
        "{prev}\u{1f}{at}\u{1f}{client}\u{1f}{query}\u{1f}{hits}\u{1f}{micros}\u{1f}{excerpt}\u{1f}{whole}"
    );
    blake3::hash(payload.as_bytes()).to_hex().to_string()
}

/// The most recent `limit` retrievals, newest first.
pub fn retrievals(db: &Connection, limit: usize) -> Result<Vec<Retrieval>> {
    let mut stmt = db.prepare(&format!(
        "SELECT id, at, client, query, hits, micros, excerpt_tokens, whole_file_tokens, hash,
                query_id, {USAGE_COLUMNS}
         FROM retrievals WHERE COALESCE(tool, '') NOT IN ('note', 'refused') ORDER BY id DESC LIMIT ?1"
    ))?;
    let rows = stmt.query_map(params![limit as i64], |r| {
        Ok(Retrieval {
            id: r.get(0)?,
            at: r.get(1)?,
            client: r.get(2)?,
            query: r.get(3)?,
            hits: r.get(4)?,
            micros: r.get(5)?,
            excerpt_tokens: r.get(6)?,
            whole_file_tokens: r.get(7)?,
            hash: r.get(8)?,
            query_id: r.get::<_, Option<String>>(9)?.unwrap_or_default(),
            usage: usage_at(r, 10)?,
        })
    })?;
    Ok(rows.collect::<Result<Vec<_>, _>>()?)
}

/// The rows that are an agent's retrievals, as SQL: what `ledger::figures`
/// counts by default. Not a refund (`raw-read`), not the chain's own
/// bookkeeping (a note, a refused write, an accept or revoke), and not a
/// person at the portal or the CLI. Every per-store reader below takes it, so
/// a store's card and its report say what the Ledger page says.
pub const AGENT_RETRIEVAL: &str = "COALESCE(tool, '') NOT IN ('raw-read', 'note', 'refused', 'accept', 'revoke') \
     AND client NOT IN ('cli', 'portal')";

/// Which rows of the ledger a count is about.
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
pub enum LedgerScope {
    /// Every retrieval.
    All,
    /// Those that found something, which are the only ones credited with a
    /// saving.
    Credited,
    /// Those credited whose tokens were estimated rather than counted by the
    /// store's own tokenizer.
    Estimated,
}

/// The distinct retrievals this store holds, as keys that can be unioned
/// across stores.
///
/// One search over six stores writes a row in each that answered it, sharing a
/// query id. Summing six stores' own counts turns that back into six
/// retrievals, which is the multiplier the Ledger page was reporting — so the
/// portal unions these keys instead of adding counts. A row written before
/// 0.20.2 has no id and is keyed by its own row id in its own store, which is
/// what it was.
pub fn ledger_keys(db: &Connection, store: &str, scope: LedgerScope) -> Result<Vec<String>> {
    let sql = match scope {
        LedgerScope::All => {
            format!("SELECT DISTINCT query_id, id FROM retrievals WHERE {AGENT_RETRIEVAL}")
        }
        LedgerScope::Credited => {
            format!(
                "SELECT DISTINCT query_id, id FROM retrievals WHERE hits > 0 AND {AGENT_RETRIEVAL}"
            )
        }
        LedgerScope::Estimated => format!(
            "SELECT DISTINCT query_id, id FROM retrievals
             WHERE hits > 0 AND (tokenizer IS NULL OR tokenizer != 'model') AND {AGENT_RETRIEVAL}"
        ),
    };
    let mut stmt = db.prepare(&sql)?;
    let rows = stmt.query_map([], |r| {
        let id: Option<String> = r.get(0)?;
        let row: i64 = r.get(1)?;
        Ok(id.unwrap_or_else(|| format!("{store}:row:{row}")))
    })?;
    let mut out: Vec<String> = rows.collect::<Result<Vec<_>, _>>()?;
    out.sort_unstable();
    out.dedup();
    Ok(out)
}

/// `(queries, clients, excerpt tokens, whole-file tokens)` over the whole/// `(queries, clients, excerpt tokens, whole-file tokens)` over the whole
/// ledger — what the Ledger page shows.
pub fn ledger_totals(db: &Connection) -> Result<(i64, i64, i64, i64)> {
    // Retrievals, not rows. One search across six stores writes a row in each
    // store that answered it and they share one `query_id`, so counting rows
    // here is what made every figure on the Ledger page scale with how many
    // stores happened to be open. A row written before 0.20.2 has no id and is
    // its own retrieval, which is what it was.
    Ok(db.query_row(
        &format!(
            "SELECT COUNT(DISTINCT COALESCE(query_id, 'row:' || id)), COUNT(DISTINCT client),
                    COALESCE(SUM(CASE WHEN hits > 0 THEN excerpt_tokens ELSE 0 END), 0),
                    COALESCE(SUM(CASE WHEN hits > 0 THEN whole_file_tokens ELSE 0 END), 0)
               FROM retrievals WHERE {AGENT_RETRIEVAL}"
        ),
        [],
        |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)),
    )?)
}

/// The one savings figure, with the denominators that make it readable.
///
/// `net` is what the ledger says was not read: the whole-file cost of the
/// files an answer named, less what the answer itself cost. Rows that found
/// nothing are excluded from it and counted separately — a retrieval that
/// returned no hits saved nothing, and a ledger that quietly dropped those
/// rows would report a ratio that no honest denominator supports.
///
/// `tier` is `measured` when every credited row was counted with the store's
/// own tokenizer, and `modelled` when any of them was estimated at four
/// characters per token. The two are never summed: a mixed ledger reports
/// `modelled`, because that is what the weaker half makes the whole.
///
/// # What this figure does not do
///
/// It does not deduplicate files across the retrievals of one session. If an
/// agent searches twice and both answers name `src/lib.rs`, the denominator
/// counts that file twice, and the real alternative — one read — is cheaper
/// than the figure implies. Fixing it needs the file set stored per row, which
/// this release does not add. The number is therefore an upper bound on what
/// was saved, and `semlith stats` says so rather than presenting it as exact.
pub fn ledger_savings(db: &Connection) -> Result<Savings> {
    ledger_savings_since(db, 0)
}

/// The savings, over retrievals recorded at or after `since`.
///
/// `since` of 0 is every row, which is what [`ledger_savings`] asks for and
/// what every caller before 0.27.0 meant. The ledger has always carried `at`
/// and an index on it; what it did not have was a reader that took a bound, so
/// the Reports page drew a window control over a figure that could not honour
/// one. A control that cannot change the thing it claims to change is worse
/// than no control.
pub fn ledger_savings_since(db: &Connection, since: i64) -> Result<Savings> {
    // Retrievals rather than rows on both sides of the ratio, for the reason
    // `ledger_totals` gives: the rows of one cross-store search are one search.
    // The token sums stay sums of rows, because each row holds its own store's
    // hits and nothing is double-counted by adding them.
    let (credited, net): (i64, i64) = db.query_row(
        &format!(
            "SELECT COUNT(DISTINCT COALESCE(query_id, 'row:' || id)),
                    COALESCE(SUM(whole_file_tokens - excerpt_tokens), 0)
               FROM retrievals WHERE hits > 0 AND at >= ?1 AND {AGENT_RETRIEVAL}"
        ),
        [since],
        |r| Ok((r.get(0)?, r.get(1)?)),
    )?;
    let total: i64 = db.query_row(
        &format!(
            "SELECT COUNT(DISTINCT COALESCE(query_id, 'row:' || id)) FROM retrievals
              WHERE at >= ?1 AND {AGENT_RETRIEVAL}"
        ),
        [since],
        |r| r.get(0),
    )?;
    let estimated: i64 = db.query_row(
        &format!(
            "SELECT COUNT(DISTINCT COALESCE(query_id, 'row:' || id)) FROM retrievals
              WHERE hits > 0 AND at >= ?1 AND (tokenizer IS NULL OR tokenizer != 'model')
                AND {AGENT_RETRIEVAL}"
        ),
        [since],
        |r| r.get(0),
    )?;
    // Summed as it is, never clamped: a figure a person recomputes from the
    // rows must come out the same (the tile used to read 18 tokens above the
    // tables beside it).
    Ok(Savings {
        net,
        credited,
        total,
        measured: credited > 0 && estimated == 0,
    })
}

/// What the ledger adds up to.
#[derive(Debug, Clone, Copy, serde::Serialize)]
pub struct Savings {
    /// Whole-file tokens less excerpt tokens, over rows that found something.
    pub net: i64,
    /// How many retrievals that was over.
    pub credited: i64,
    /// How many retrievals there are in total, credited or not. The
    /// denominator, so a figure is never shown without one.
    pub total: i64,
    /// Whether every credited row was counted with the store's own tokenizer.
    pub measured: bool,
}

impl Savings {
    /// `measured` or `modelled`.
    pub fn tier(&self) -> &'static str {
        if self.measured {
            "measured"
        } else {
            "modelled"
        }
    }

    /// What share of retrievals the figure covers, as a percentage.
    pub fn coverage(&self) -> i64 {
        if self.total == 0 {
            return 0;
        }
        self.credited * 100 / self.total
    }
}

/// What the ledger says about the questions semlith did *not* answer.
///
/// `refunds` counts the whole-file reads the steering hook saw on a file this
/// store holds: semlith could have answered them and was not asked. `zero_hit`
/// counts the retrievals semlith *was* asked and found nothing for. They are
/// separate figures because they call for opposite things — a refund means the
/// agent is not reaching for semlith, a zero hit means semlith is not reaching
/// the answer — and adding them together would say neither.
///
/// `measured` is false when no hook has ever written into this store, which is
/// what makes `refunds` a floor rather than a count on such a machine.
pub fn ledger_misses(db: &Connection) -> Result<Misses> {
    let refunds: i64 = db.query_row(
        "SELECT COUNT(*) FROM retrievals WHERE tool = ?1",
        params![crate::ledger::RAW_READ],
        |r| r.get(0),
    )?;
    // A retrieval is a zero hit when none of its rows found anything: one
    // search over five stores that two answered is not three zero hits.
    let zero_hit: i64 = db.query_row(
        &format!(
            "SELECT COUNT(*) FROM (SELECT COALESCE(query_id, 'row:' || id) AS k
               FROM retrievals WHERE {AGENT_RETRIEVAL}
              GROUP BY k HAVING MAX(hits) = 0)"
        ),
        [],
        |r| r.get(0),
    )?;
    Ok(Misses {
        refunds,
        zero_hit,
        measured: refunds > 0,
    })
}

/// The two figures for what semlith did not answer.
#[derive(Debug, Clone, Copy, Default, serde::Serialize)]
pub struct Misses {
    /// Whole-file reads of a file this store holds, seen by the steering hook.
    pub refunds: i64,
    /// Retrievals semlith answered with nothing.
    pub zero_hit: i64,
    /// Whether a hook has ever written here. Without one, `refunds` is a floor.
    pub measured: bool,
}

/// How many retrievals each client made, most first.
pub fn ledger_clients(db: &Connection) -> Result<Vec<(String, i64)>> {
    let mut stmt = db.prepare(&format!(
        "SELECT client, COUNT(DISTINCT COALESCE(query_id, 'row:' || id)) AS n FROM retrievals
          WHERE {AGENT_RETRIEVAL}
          GROUP BY client ORDER BY n DESC, client"
    ))?;
    let rows = stmt.query_map([], |r| Ok((r.get(0)?, r.get(1)?)))?;
    Ok(rows.collect::<Result<Vec<_>, _>>()?)
}

/// The `tool` of a ledger note: a row the chain carries that is not a
/// retrieval, and no total counts.
pub const NOTE_TOOL: &str = "note";

/// The `tool` of a call refused because the agent key is read-only; `query`
/// names the tool. Not a retrieval, so no total counts it either.
pub const REFUSED_TOOL: &str = "refused";

/// The rows a note has re-anchored the chain after: their ids.
fn acknowledged(db: &Connection) -> Result<std::collections::HashSet<i64>> {
    let mut stmt = db.prepare("SELECT query FROM retrievals WHERE tool = ?1")?;
    let rows = stmt.query_map(params![NOTE_TOOL], |r| r.get::<_, String>(0))?;
    let mut out = std::collections::HashSet::new();
    for text in rows {
        let text = text?;
        if let Some(rest) = text.split("after row ").nth(1) {
            let digits: String = rest.chars().take_while(char::is_ascii_digit).collect();
            if let Ok(id) = digits.parse() {
                out.insert(id);
            }
        }
    }
    Ok(out)
}

/// Re-anchor the chain after a row that does not verify, by appending a
/// note row onto the current head (0.35.0's Re-verify with repair).
///
/// Nothing recorded is edited or deleted: the note says which row broke, and
/// [`ledger_break`] accepts that row as it stands from then on. The edit is
/// still visible, in the note, which is what an audit trail owes its reader.
pub fn ledger_note(db: &Connection, broken: i64) -> Result<()> {
    let query = format!("chain re-anchored after row {broken} was edited or removed");
    record_retrieval(
        db,
        &NewRetrieval {
            client_version: "",
            client: "semlith",
            session: "",
            tool: NOTE_TOOL,
            query: &query,
            hits: 0,
            micros: 0,
            excerpt_tokens: 0,
            whole_file_tokens: 0,
            stale_hits: 0,
            tokenizer: crate::ledger::CHARS4,
            query_id: "",
        },
    )
}

/// Append a note for every unacknowledged break, first to last, and say which
/// rows they were. Bounded, so a chain broken at every row is not a loop.
pub fn ledger_repair(db: &Connection) -> Result<Vec<i64>> {
    let mut noted = Vec::new();
    while let Some(id) = ledger_break(db)? {
        if noted.contains(&id) || noted.len() >= 1_000 {
            break;
        }
        ledger_note(db, id)?;
        noted.push(id);
    }
    Ok(noted)
}

/// When a ledger row was written, by id.
pub fn retrieval_at(db: &Connection, id: i64) -> Result<Option<i64>> {
    Ok(db
        .query_row(
            "SELECT at FROM retrievals WHERE id = ?1",
            params![id],
            |r| r.get(0),
        )
        .optional()?)
}

/// How many rows the chain holds, notes included: what a verify walked.
pub fn ledger_rows(db: &Connection) -> Result<i64> {
    Ok(db.query_row("SELECT COUNT(*) FROM retrievals", [], |r| r.get(0))?)
}

/// Re-walk the chain and return the id of the first row that does not verify.
///
/// `None` means the ledger is intact. A row a note re-anchored after
/// ([`ledger_note`]) is accepted as it stands, and the walk goes on from it.
pub fn ledger_break(db: &Connection) -> Result<Option<i64>> {
    let acked = acknowledged(db)?;
    let key = ledger_key(db, false);
    let mut stmt = db.prepare(
        "SELECT id, at, client, query, hits, micros, excerpt_tokens, whole_file_tokens, prev, hash,
                session, tool, stale_hits, tokenizer, chain, COALESCE(query_id, ''),
                COALESCE(client_version, ''), COALESCE(whole_basis, ''), seal
         FROM retrievals ORDER BY id",
    )?;
    let mut sealed = db.prepare(&format!(
        "SELECT {SEAL_COLUMNS} FROM retrievals WHERE id = ?1"
    ))?;
    // The head: the newest v2 row as it was written. A v2 row after it, or no
    // row at its id with its hash, is a tail somebody cut or a row somebody
    // forged; v1 rows after it are an older binary's, and allowed.
    let head: Option<(i64, String)> = get_meta(db, LEDGER_HEAD)?.and_then(|text| {
        let mut parts = text.splitn(3, ':');
        let id: i64 = parts.next()?.parse().ok()?;
        let rest = parts.next()?;
        // The hash itself holds a colon (`k2:…`), so it is two parts.
        let mut tail = parts.next()?.splitn(2, ':');
        let hash = format!("{rest}:{}", tail.next()?);
        let mac = tail.next()?;
        (head_mac(key.as_ref(), id, &hash) == mac).then_some((id, hash))
    });
    // Erasing every row is one documented `DELETE`, and an empty ledger has
    // nothing left to vouch for; a head with no rows under it is that, not a
    // cut tail.
    let empty: bool = db.query_row("SELECT NOT EXISTS (SELECT 1 FROM retrievals)", [], |r| {
        r.get(0)
    })?;
    if empty {
        return Ok(None);
    }
    let head_claimed = get_meta(db, LEDGER_HEAD)?.is_some();
    if head_claimed && head.is_none() {
        // A head that does not verify is a head somebody wrote by hand.
        return Ok(Some(db.query_row(
            "SELECT COALESCE(MAX(id), 0) FROM retrievals",
            [],
            |r| r.get(0),
        )?));
    }
    let mut head_seen = false;
    let mut rows = stmt.query([])?;
    let mut expected = String::new();
    while let Some(r) = rows.next()? {
        let (id, at): (i64, i64) = (r.get(0)?, r.get(1)?);
        let client: String = r.get(2)?;
        let query: String = r.get(3)?;
        let (hits, micros, excerpt, whole): (i64, i64, i64, i64) =
            (r.get(4)?, r.get(5)?, r.get(6)?, r.get(7)?);
        let prev: String = r.get(8)?;
        let hash: String = r.get(9)?;
        if acked.contains(&id) {
            expected = hash;
            continue;
        }
        if prev != expected {
            return Ok(Some(id));
        }
        // `tool` is NULL on every row written before 0.15.0 and set on every
        // row written since, which is what says which formula wrote this hash.
        let tool: Option<String> = r.get(11)?;
        let chain: Option<i64> = r.get(14)?;
        if let Some((head_id, head_hash)) = &head {
            if id == *head_id {
                if hash != *head_hash {
                    return Ok(Some(id));
                }
                head_seen = true;
            } else if id > *head_id && chain == Some(2) && !acked.contains(&id) {
                return Ok(Some(id));
            }
        }
        let recomputed = match tool {
            Some(tool) if chain == Some(2) => {
                let session: String = r.get(10)?;
                let stale: i64 = r.get(12)?;
                let tokenizer: String = r.get(13)?;
                let query_id: String = r.get(15)?;
                let version: String = r.get(16)?;
                let basis: String = r.get(17)?;
                let seal: Option<String> = r.get(18)?;
                let row = NewRetrieval {
                    client_version: &version,
                    client: &client,
                    session: &session,
                    tool: &tool,
                    query: &query,
                    hits,
                    micros,
                    excerpt_tokens: excerpt,
                    whole_file_tokens: whole,
                    stale_hits: stale,
                    tokenizer: &tokenizer,
                    query_id: &query_id,
                };
                if hash.starts_with(V2_KEYED) && key.is_none() {
                    return Ok(Some(id));
                }
                let k = if hash.starts_with(V2_KEYED) {
                    key.as_ref()
                } else {
                    None
                };
                // The late columns: a seal over what they say now, or none at
                // all while none of them has been written.
                let late = sealed.query_row([id], seal_payload)?;
                let blank = late.split('\u{1f}').skip(1).all(|v| v == "\u{0}");
                match seal {
                    Some(seal) if seal != mac(k, "k:", "p:", &late) => return Ok(Some(id)),
                    None if !blank => return Ok(Some(id)),
                    _ => {}
                }
                v2_hash(k, &prev, at, &row, &basis)
            }
            Some(tool) => {
                let session: String = r.get(10)?;
                let stale: i64 = r.get(12)?;
                let tokenizer: String = r.get(13)?;
                let row = NewRetrieval {
                    client_version: "",
                    client: &client,
                    session: &session,
                    tool: &tool,
                    query: &query,
                    hits,
                    micros,
                    excerpt_tokens: excerpt,
                    whole_file_tokens: whole,
                    stale_hits: stale,
                    tokenizer: &tokenizer,
                    // Not covered by the hash; see `chain_hash`.
                    query_id: "",
                };
                match (hash.starts_with(KEYED), &key) {
                    (false, _) => chain_hash(&prev, at, &row),
                    (true, Some(key)) => keyed_chain_hash(key, &prev, at, &row),
                    // A keyed row with no key beside the database cannot be
                    // verified, so it is the break.
                    (true, None) => return Ok(Some(id)),
                }
            }
            None => legacy_chain_hash(&prev, at, &client, &query, hits, micros, excerpt, whole),
        };
        if recomputed != hash {
            return Ok(Some(id));
        }
        expected = hash;
    }
    if let Some((head_id, _)) = head
        && !head_seen
        && !acked.contains(&head_id)
    {
        // The head's row is gone: the tail was cut.
        return Ok(Some(head_id));
    }
    Ok(None)
}

/// One ledger row, every column a figure is made of.
///
/// The input to `ledger::figures`, the one aggregation every surface renders
/// (0.42.0). Read whole rather than summed here, because the figures union
/// retrievals across stores (one search over five stores is five rows and one
/// query), and SQL inside one store cannot see the other four.
#[derive(Debug, Clone, serde::Serialize)]
pub struct LedgerFact {
    pub id: i64,
    pub at: i64,
    pub client: String,
    pub session: String,
    /// Empty on a row written before 0.15.0.
    pub tool: String,
    pub query: String,
    pub hits: i64,
    pub micros: i64,
    pub excerpt_tokens: i64,
    pub whole_file_tokens: i64,
    pub stale_hits: i64,
    pub tokenizer: String,
    /// Empty on a row written before 0.20.2, which was its own retrieval.
    pub query_id: String,
    pub client_version: String,
    /// `bytes`, `text`, or empty for a row written before 0.42.0 (bytes).
    pub whole_basis: String,
    /// 2 for chain v2, 1 otherwise.
    pub chain: i64,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub usage: Option<RowUsage>,
}

/// Every ledger row with `since <= at < until`, oldest first.
pub fn ledger_facts(db: &Connection, since: i64, until: Option<i64>) -> Result<Vec<LedgerFact>> {
    let mut stmt = db.prepare(&format!(
        "SELECT id, at, client, COALESCE(session, ''), COALESCE(tool, ''), query, hits, micros,
                excerpt_tokens, whole_file_tokens, COALESCE(stale_hits, 0),
                COALESCE(tokenizer, ''), COALESCE(query_id, ''), COALESCE(client_version, ''),
                COALESCE(whole_basis, ''), COALESCE(chain, 1), {USAGE_COLUMNS}
           FROM retrievals WHERE at >= ?1 AND at < ?2 ORDER BY id"
    ))?;
    let rows = stmt.query_map(params![since, until.unwrap_or(i64::MAX)], |r| {
        Ok(LedgerFact {
            id: r.get(0)?,
            at: r.get(1)?,
            client: r.get(2)?,
            session: r.get(3)?,
            tool: r.get(4)?,
            query: r.get(5)?,
            hits: r.get(6)?,
            micros: r.get(7)?,
            excerpt_tokens: r.get(8)?,
            whole_file_tokens: r.get(9)?,
            stale_hits: r.get(10)?,
            tokenizer: r.get(11)?,
            query_id: r.get(12)?,
            client_version: r.get(13)?,
            whole_basis: r.get(14)?,
            chain: r.get(15)?,
            usage: usage_at(r, 16)?,
        })
    })?;
    Ok(rows.collect::<Result<_, _>>()?)
}

/// The text the store holds for `path`, stitched by line from its chunks.
///
/// The whole-file side of a row for a file a reader extracted (a PDF, an
/// Office file): reading the file would have meant reading this text, not its
/// bytes — 2.2 MB of PDF is 45,849 characters. `None` for a file not held.
pub fn held_text(db: &Connection, path: &str) -> Result<Option<String>> {
    let mut stmt = db.prepare(
        "SELECT c.start_line, c.text FROM chunks c JOIN files f ON f.id = c.file_id
          WHERE f.path = ?1 ORDER BY c.start_line, c.id",
    )?;
    let mut lines: std::collections::BTreeMap<i64, String> = Default::default();
    let mut any = false;
    let rows = stmt.query_map(params![path], |r| {
        Ok((r.get::<_, i64>(0)?, r.get::<_, String>(1)?))
    })?;
    for row in rows {
        let (start, text) = row?;
        any = true;
        for (offset, line) in text.split('\n').enumerate() {
            lines
                .entry(start + offset as i64)
                .or_insert_with(|| line.to_string());
        }
    }
    Ok(any.then(|| lines.into_values().collect::<Vec<_>>().join("\n")))
}

/// The total bytes of every file a grep for `name` would have to read.
///
/// The honest denominator for a graph answer's saving. A search's denominator
/// is the files its hits came from; a graph answer has no hits, but it does
/// have an answer that would otherwise have been assembled by grepping for the
/// name and reading what came back. These are those files: the ones that
/// define the name, and the ones whose code points at it.
pub fn grep_cost(db: &Connection, name: &str) -> Result<i64> {
    Ok(db.query_row(
        "SELECT COALESCE(SUM(bytes), 0) FROM files WHERE id IN (
            SELECT s.file_id FROM symbols s WHERE s.name = ?1
            UNION
            SELECT s.file_id FROM symbols s JOIN edges e ON e.src = s.id WHERE e.dst = ?1
         )",
        params![name],
        |r| r.get(0),
    )?)
}

/// Whether this store has `path` indexed.
///
/// Compared the way every filter compares a path — normalised separators,
/// lowercased — so a Windows store answers the same question a unix one does
/// about the same file. The steering hook's ledger row depends on it: a raw
/// read is only a refund against the store that could have answered it.
pub fn holds_path(db: &Connection, path: &str) -> Result<bool> {
    // Canonical, because that is the form a store records and a caller may hand
    // over whatever the client said — a relative path, a symlink, or on macOS a
    // `/var` that is really `/private/var`.
    let canonical = crate::canonical(std::path::Path::new(path));
    let wanted = canonical
        .to_string_lossy()
        .to_ascii_lowercase()
        .replace('\\', "/");
    let sql = format!("SELECT 1 FROM files f WHERE {GLOB_PATH} = ?1 LIMIT 1");
    Ok(db
        .query_row(&sql, params![wanted], |_| Ok(()))
        .optional()?
        .is_some())
}

/// One definition's span and identity: `(start, end, name, kind)`.
pub type SymbolSpan = (u32, u32, String, String);

/// Every symbol defined in each of `paths`.
///
/// What turns a line range into a place a person recognises. A hit that says
/// `src/store.rs:1041-1090` makes a reader open the file to find out what is
/// there; one that says `record_retrieval` does not.
///
/// Read per file rather than per hit, because eight hits in one file are one
/// question about that file.
pub fn symbols_in_files(
    db: &Connection,
    paths: &[String],
) -> Result<std::collections::HashMap<String, Vec<SymbolSpan>>> {
    let mut out: std::collections::HashMap<String, Vec<SymbolSpan>> = Default::default();
    if paths.is_empty() {
        return Ok(out);
    }
    let holes = vec!["?"; paths.len()].join(", ");
    let sql = format!(
        "SELECT f.path, s.start_line, s.end_line, s.name, s.kind
         FROM symbols s JOIN files f ON f.id = s.file_id
         WHERE f.path IN ({holes}) AND s.kind != 'module'
         ORDER BY f.path, s.start_line"
    );
    let mut stmt = db.prepare(&sql)?;
    let args = paths.iter().cloned().map(Value::Text);
    let mut rows = stmt.query(rusqlite::params_from_iter(args))?;
    while let Some(r) = rows.next()? {
        let path: String = r.get(0)?;
        out.entry(path)
            .or_default()
            .push((r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?));
    }
    Ok(out)
}

/// `(bytes, indexed_at)` for each of `paths` the store knows.
///
/// The cheap half of a freshness check: what the store recorded about a file
/// when it read it. The other half is one `stat` of the file as it is now.
pub fn file_stamps(
    db: &Connection,
    paths: &[String],
) -> Result<std::collections::HashMap<String, Stamp>> {
    if paths.is_empty() {
        return Ok(Default::default());
    }
    let holes = vec!["?"; paths.len()].join(", ");
    let sql =
        format!("SELECT path, bytes, indexed_at, read_mtime FROM files WHERE path IN ({holes})");
    let mut stmt = db.prepare(&sql)?;
    let args = paths.iter().cloned().map(Value::Text);
    let rows = stmt.query_map(rusqlite::params_from_iter(args), |r| {
        Ok((
            r.get::<_, String>(0)?,
            Stamp {
                bytes: r.get(1)?,
                indexed_at: r.get(2)?,
                read_mtime: r.get(3)?,
            },
        ))
    })?;
    Ok(rows.collect::<Result<std::collections::HashMap<_, _>, _>>()?)
}

/// What a file looked like when it was read for the store, and the one rule
/// every surface judges "edited since it was indexed" by.
#[derive(Debug, Clone, Copy)]
pub struct Stamp {
    pub bytes: i64,
    pub indexed_at: i64,
    /// The file's modification time when it was read, in nanoseconds. `None`
    /// on a row an older binary wrote.
    pub read_mtime: Option<i64>,
}

/// A file's modification time in nanoseconds since the epoch.
pub fn mtime_ns(meta: &std::fs::Metadata) -> Option<i64> {
    meta.modified()
        .ok()?
        .duration_since(std::time::UNIX_EPOCH)
        .ok()
        .map(|d| d.as_nanos() as i64)
}

impl Stamp {
    /// Whether the file on disk is still the one the store read.
    ///
    /// With the mtime the read saw, the two must be equal. Before 0.40.0 the
    /// rule was "not modified after `indexed_at`", and `indexed_at` is when the
    /// row was written, after the read: an edit that kept the size and landed
    /// between the read and the write read as fresh, and the store's old text
    /// was served as current. A row an older binary wrote keeps that rule.
    pub fn fresh(&self, meta: &std::fs::Metadata) -> bool {
        if meta.len() as i64 != self.bytes {
            return false;
        }
        match self.read_mtime {
            Some(ns) => mtime_ns(meta) == Some(ns),
            None => meta
                .modified()
                .ok()
                .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
                .is_none_or(|d| d.as_secs() as i64 <= self.indexed_at),
        }
    }
}

/// Record the modification time a file's read saw. See [`Stamp::fresh`].
pub fn set_read_mtime(db: &Connection, file_id: i64, read_mtime: Option<i64>) -> Result<()> {
    db.prepare_cached("UPDATE files SET read_mtime = ?1 WHERE id = ?2")?
        .execute(params![read_mtime, file_id])?;
    Ok(())
}

/// The classes a not-indexed row falls into (2.3), and the dummy list.
pub mod class {
    /// (a) a secret-shaped value in the text. Reviewable.
    pub const CONTENT: &str = "content";
    /// (b) a credential file by name or folder. Never acceptable.
    pub const CREDENTIAL: &str = "credential";
    /// (c) over the size cap, or inside a pruned generated folder. Reviewable.
    pub const POLICY: &str = "policy";
    /// (d) empty, binary, no text, unreadable, not a regular file.
    pub const UNINDEXABLE: &str = "unindexable";
    /// (e) `.gitignore`, `.semlithignore`, or outside the store's roots.
    pub const EXCLUDED: &str = "excluded";
    /// Indexed: every match was a declared test dummy.
    pub const DUMMY: &str = "dummy";
    /// (f) a parser crashed, ran out of memory or out of time on it. The rule
    /// is the reason. Not reviewable — there is nothing in it to review — but
    /// accepting it tries it once more.
    pub const CRASHED: &str = "crashed";

    /// Whether a person may accept a row of this class.
    pub fn reviewable(class: &str) -> bool {
        matches!(class, CONTENT | POLICY)
    }
}

/// One file or folder that was not indexed, and why.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct Refused {
    #[serde(serialize_with = "crate::serialize_plain")]
    pub path: String,
    pub class: String,
    pub rule: String,
    /// What the scan found, masked. Empty for anything but a content row
    /// and a dummy row.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub matches: Vec<serde_json::Value>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub confidence: Option<u8>,
    /// Files the row stands for: one, or a folder's count.
    pub files: u32,
    pub first_seen: i64,
    pub last_seen: i64,
    /// Whether a person may accept it.
    pub reviewable: bool,
    /// The acceptance on this path, when there is one.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub accepted: Option<String>,
}

/// Record that `path` was not indexed. The first time it was seen is kept.
#[allow(clippy::too_many_arguments)]
pub fn refuse(
    db: &Connection,
    path: &str,
    class: &str,
    rule: &str,
    matches: &[crate::keyscan::Match],
    files: u32,
    now: i64,
) -> Result<()> {
    let confidence = matches
        .iter()
        .filter(|m| m.dummy.is_none() || class == class::DUMMY)
        .map(|m| m.confidence)
        .max();
    let json = if matches.is_empty() {
        None
    } else {
        Some(serde_json::to_string(matches)?)
    };
    db.execute(
        "INSERT INTO refusals (path, class, rule, matches, confidence, files, first_seen, last_seen)
         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?7)
         ON CONFLICT(path) DO UPDATE SET class = excluded.class, rule = excluded.rule,
           matches = excluded.matches, confidence = excluded.confidence,
           files = excluded.files, last_seen = excluded.last_seen",
        params![path, class, rule, json, confidence, files, now],
    )?;
    Ok(())
}

/// Turn on FTS5's secure delete for `chunks_fts`, if it is not on: a removed
/// row's terms then leave the index's segments at once rather than at their
/// next merge, so a file refused for a secret leaves no searchable term
/// behind. A setting the index keeps; older binaries read it unchanged.
pub fn secure_delete_fts(db: &Connection) -> Result<()> {
    let set = db
        .query_row(
            "SELECT 1 FROM chunks_fts_config WHERE k = 'secure-delete' AND v = 1",
            [],
            |_| Ok(()),
        )
        .optional()?
        .is_some();
    if !set {
        db.execute(
            "INSERT INTO chunks_fts(chunks_fts, rank) VALUES('secure-delete', 1)",
            [],
        )?;
    }
    Ok(())
}

/// Forget that `path` was not indexed: it has been, or it is gone.
pub fn unrefuse(db: &Connection, path: &str) -> Result<()> {
    db.execute("DELETE FROM refusals WHERE path = ?1", params![path])?;
    Ok(())
}

/// Every not-indexed row, reviewable first, then by class and path.
pub fn refusals(db: &Connection) -> Result<Vec<Refused>> {
    let mut stmt = db.prepare(
        "SELECT r.path, r.class, r.rule, r.matches, r.confidence, r.files, r.first_seen,
                r.last_seen, a.mode
         FROM refusals r LEFT JOIN acceptances a ON a.path = r.path
         ORDER BY CASE r.class WHEN 'content' THEN 0 WHEN 'policy' THEN 1 WHEN 'credential' THEN 2
                  WHEN 'crashed' THEN 3 WHEN 'dummy' THEN 4 WHEN 'unindexable' THEN 5 ELSE 6 END, r.path",
    )?;
    let rows = stmt.query_map([], |r| {
        let class: String = r.get(1)?;
        let matches: Option<String> = r.get(3)?;
        Ok(Refused {
            path: r.get(0)?,
            reviewable: class::reviewable(&class),
            class,
            rule: r.get(2)?,
            matches: matches
                .and_then(|m| serde_json::from_str(&m).ok())
                .unwrap_or_default(),
            confidence: r.get(4)?,
            files: r.get(5)?,
            first_seen: r.get(6)?,
            last_seen: r.get(7)?,
            accepted: r.get(8)?,
        })
    })?;
    let mut rows = rows.collect::<Result<Vec<_>, _>>()?;
    // Accepted files are indexed and off the list proper, and still listed:
    // a person who accepted one has to be able to find it again to revoke it.
    for a in acceptances(db)? {
        if rows.iter().any(|r| r.path == a.path) {
            continue;
        }
        rows.push(Refused {
            path: a.path,
            class: a.class.clone(),
            rule: format!("accepted from the {} ({})", a.source, a.mode),
            matches: Vec::new(),
            confidence: a.confidence,
            files: 1,
            first_seen: a.at,
            last_seen: a.at,
            reviewable: class::reviewable(&a.class) || a.class == class::DUMMY,
            accepted: Some(a.mode),
        });
    }
    Ok(rows)
}

/// One not-indexed row, by exact path.
pub fn refusal(db: &Connection, path: &str) -> Result<Option<Refused>> {
    Ok(refusals(db)?.into_iter().find(|r| r.path == path))
}

/// A person's decision about one file.
#[derive(Debug, Clone, serde::Serialize)]
pub struct Acceptance {
    #[serde(serialize_with = "crate::serialize_plain")]
    pub path: String,
    pub class: String,
    pub mode: String,
    #[serde(skip)]
    pub fingerprints: Vec<String>,
    pub confidence: Option<u8>,
    pub at: i64,
    pub source: String,
}

pub fn acceptance(db: &Connection, path: &str) -> Result<Option<Acceptance>> {
    Ok(db
        .query_row(
            "SELECT path, class, mode, fingerprints, confidence, at, source
             FROM acceptances WHERE path = ?1",
            params![path],
            |r| {
                let prints: String = r.get(3)?;
                Ok(Acceptance {
                    path: r.get(0)?,
                    class: r.get(1)?,
                    mode: r.get(2)?,
                    fingerprints: serde_json::from_str(&prints).unwrap_or_default(),
                    confidence: r.get(4)?,
                    at: r.get(5)?,
                    source: r.get(6)?,
                })
            },
        )
        .optional()?)
}

/// Every acceptance, for counts and for the folders the walk must not prune.
pub fn acceptances(db: &Connection) -> Result<Vec<Acceptance>> {
    let mut stmt = db.prepare("SELECT path FROM acceptances ORDER BY path")?;
    let paths: Vec<String> = stmt
        .query_map([], |r| r.get(0))?
        .collect::<Result<Vec<_>, _>>()?;
    let mut out = Vec::new();
    for path in paths {
        if let Some(a) = acceptance(db, &path)? {
            out.push(a);
        }
    }
    Ok(out)
}

pub fn accept(db: &Connection, a: &Acceptance) -> Result<()> {
    db.execute(
        "INSERT INTO acceptances (path, class, mode, fingerprints, confidence, at, source)
         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
         ON CONFLICT(path) DO UPDATE SET class = excluded.class, mode = excluded.mode,
           fingerprints = excluded.fingerprints, confidence = excluded.confidence,
           at = excluded.at, source = excluded.source",
        params![
            a.path,
            a.class,
            a.mode,
            serde_json::to_string(&a.fingerprints)?,
            a.confidence,
            a.at,
            a.source
        ],
    )?;
    Ok(())
}

pub fn revoke(db: &Connection, path: &str) -> Result<bool> {
    Ok(db.execute("DELETE FROM acceptances WHERE path = ?1", params![path])? > 0)
}

/// The per-store salt fingerprints are keyed with, made on first need.
///
/// Random, from the operating system, and never leaves the store: a
/// fingerprint without it confirms nothing about a guessed value.
pub fn salt(db: &Connection) -> Result<[u8; 32]> {
    if let Some(hex) = get_meta(db, "secret_salt")?
        && let Some(bytes) = unhex(&hex)
    {
        return Ok(bytes);
    }
    let mut bytes = [0u8; 32];
    getrandom::fill(&mut bytes).map_err(|e| anyhow::anyhow!("no random source: {e}"))?;
    let hex: String = bytes.iter().map(|b| format!("{b:02x}")).collect();
    set_meta(db, "secret_salt", &hex)?;
    Ok(bytes)
}

/// The salt, when a store has one. A read-only caller cannot make one, and
/// a store without one has no acceptances to compare against.
pub fn salt_if_any(db: &Connection) -> Result<Option<[u8; 32]>> {
    Ok(get_meta(db, "secret_salt")?.and_then(|h| unhex(&h)))
}

fn unhex(hex: &str) -> Option<[u8; 32]> {
    if hex.len() != 64 || !hex.is_ascii() {
        return None;
    }
    let bytes: Vec<u8> = (0..32)
        .map(|i| u8::from_str_radix(&hex[i * 2..i * 2 + 2], 16).ok())
        .collect::<Option<Vec<u8>>>()?;
    bytes.try_into().ok()
}

/// A chunk's first line, last line and id.
pub type ChunkSpan = (u32, u32, i64);

/// A file's id and its chunks' line spans, for writing its graph again.
pub fn graph_input(db: &Connection, path: &str) -> Result<Option<(i64, Vec<ChunkSpan>)>> {
    let Some(file_id) = db
        .query_row("SELECT id FROM files WHERE path = ?1", params![path], |r| {
            r.get::<_, i64>(0)
        })
        .optional()?
    else {
        return Ok(None);
    };
    let mut stmt = db.prepare(
        "SELECT start_line, end_line, id FROM chunks WHERE file_id = ?1 ORDER BY start_line",
    )?;
    let spans = stmt
        .query_map(params![file_id], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))?
        .collect::<Result<Vec<_>, _>>()?;
    Ok(Some((file_id, spans)))
}

/// Remove one file's symbols, and with them the edges that leave them.
pub fn delete_graph(db: &Connection, file_id: i64) -> Result<()> {
    db.execute(
        "DELETE FROM edges WHERE src IN (SELECT id FROM symbols WHERE file_id = ?1)",
        params![file_id],
    )?;
    db.execute("DELETE FROM symbols WHERE file_id = ?1", params![file_id])?;
    Ok(())
}

/// Bring an unchanged file's `bytes` and `indexed_at` up to what is on disk,
/// when the file's size or mtime has moved past the row.
///
/// Only called once the content hash has matched, so the chunks are exactly
/// as current as the file; it is the stamps that freshness is judged by that
/// were behind. A row already current is not written, so an unchanged tree's
/// pass writes nothing.
pub fn heal_stamps(
    db: &Connection,
    path: &str,
    bytes: i64,
    mtime: i64,
    read_mtime: Option<i64>,
    now: i64,
) -> Result<()> {
    db.execute(
        "UPDATE files SET bytes = ?2, indexed_at = MAX(?3, ?4), read_mtime = ?5
         WHERE path = ?1 AND (bytes != ?2 OR indexed_at < ?3 OR read_mtime IS NOT ?5)",
        params![path, bytes, mtime, now, read_mtime],
    )?;
    Ok(())
}

/// The newest `indexed_at` this store holds, or `None` when it holds nothing.
///
/// The store's own answer to "last write", which is what the Stores page's
/// column says it shows. The daemon used to answer it from a counter that
/// started at zero every time it launched, so a store with 438 files in it read
/// `never` until something wrote to it again — finding 1.7 of the 2026-09-17
/// drive.
pub fn last_write(db: &Connection) -> Result<Option<i64>> {
    let at: Option<i64> = db.query_row("SELECT MAX(indexed_at) FROM files", [], |r| r.get(0))?;
    Ok(at)
}

/// `(symbols, edges)` — the graph's size, for `stats` and the portal.
pub fn graph_stats(db: &Connection) -> Result<(i64, i64)> {
    let symbols: i64 = db.query_row("SELECT COUNT(*) FROM symbols", [], |r| r.get(0))?;
    let edges: i64 = db.query_row("SELECT COUNT(*) FROM edges", [], |r| r.get(0))?;
    Ok((symbols, edges))
}

/// `(files, chunks, indexed bytes)`
/// Whether this store can still be read, as cheaply as that can be asked.
///
/// One row out of each of the two large tables, by the rowid index. It exists
/// because the obvious probe — `stats()` below — is three full scans, and this
/// runs once per store on every aggregating read, which includes every search.
/// Measured on a 10 390-chunk store: `stats()` 2.65 ms, this 0.24 ms, and the
/// gap widens linearly because one is O(n) and the other is O(log n).
///
/// It is not weaker for being cheaper. What it has to catch is a database that
/// opened and cannot be read now — truncated by a killed index run, on a volume
/// that was unplugged — and a btree descent into a page that is no longer there
/// fails exactly as a scan would. What it deliberately does not do is read
/// `meta` and call that proof: `Semlith::open` already read `meta`, and page one
/// of a truncated file is usually the part that survived.
///
/// An empty table is not a failure: a store with nothing indexed yet is a store
/// that reads fine and holds nothing.
pub fn readable(db: &Connection) -> Result<()> {
    for table in ["files", "chunks"] {
        let sql = format!("SELECT id FROM {table} ORDER BY id DESC LIMIT 1");
        let mut stmt = db.prepare(&sql)?;
        let mut rows = stmt.query([])?;
        // Stepped rather than collected: the error this is looking for arrives
        // on the step, and `Option` is the empty-table case rather than a fault.
        let _: Option<i64> = rows.next()?.map(|r| r.get(0)).transpose()?;
    }
    Ok(())
}

pub fn stats(db: &Connection) -> Result<(i64, i64, i64)> {
    let files: i64 = db.query_row("SELECT COUNT(*) FROM files", [], |r| r.get(0))?;
    let chunks: i64 = db.query_row("SELECT COUNT(*) FROM chunks", [], |r| r.get(0))?;
    let bytes: i64 = db.query_row("SELECT COALESCE(SUM(bytes), 0) FROM files", [], |r| {
        r.get(0)
    })?;
    Ok((files, chunks, bytes))
}

/// Everything the `Inside the index` page shows about one store.
///
/// Measured, not estimated — that is the page's own claim about itself and so
/// it is the contract for this function. Where a figure cannot be measured
/// from what the store holds, it is not here; the page says what it does not
/// know rather than printing a plausible number.
///
/// One pass over `chunks` is the expensive part: the text of every chunk is
/// read to count lines, words, characters and blanks. A 130k-chunk store is
/// tens of megabytes, which is a second or so — this runs when the page is
/// opened, not on every write.
#[derive(Debug, Clone, Default, serde::Serialize)]
pub struct Corpus {
    pub files: i64,
    pub chunks: i64,
    pub bytes: i64,
    /// Lines covered by chunks. A file's lines are `MAX(end_line)` over its
    /// chunks, which is the file's length wherever it was chunked whole.
    pub lines: i64,
    pub words: i64,
    pub characters: i64,
    pub blank_lines: i64,
    /// Lines whose first non-space characters open a comment in that file's
    /// language. A heuristic over a short table of openers, named as one on
    /// the page.
    pub comment_lines: i64,
    /// Per language, by line: the mix the page draws as a stacked bar.
    pub languages: Vec<LanguageLines>,
    /// Files by the kind of thing they are, for the prose panel.
    pub kinds: Vec<FileKind>,
    pub longest_file: Option<FileLines>,
    /// The most folders any indexed path descends through.
    pub deepest_path: i64,
    /// Unix seconds of the first and last `indexed_at` in the store.
    pub first_indexed: i64,
    pub last_indexed: i64,
    /// Chunks added per month, oldest first, as `("2026-09", 9610)`.
    pub months: Vec<MonthCount>,
    /// One vector per chunk, of this many numbers each.
    pub vector_dim: i64,
    pub symbols: i64,
    pub edges: i64,
    /// Edges by tier, in the order the graph rail lists them. Decided by
    /// `coverage_by_language`, which is the one rule for this.
    pub tiers: Vec<KindCount>,
    /// Edges naming something no definition in this store satisfies, and the
    /// names most often behind that.
    pub unresolved: i64,
    pub unresolved_names: Vec<KindCount>,
    /// Names defined more than once, and the worst of them.
    pub ambiguous_names: i64,
    pub ambiguous_worst: Vec<KindCount>,
    /// Languages this store holds, and how many of them carry an edge.
    pub languages_with_edges: i64,
    /// The median milliseconds of the retrievals the ledger recorded, or zero
    /// when it has recorded none: a claim about this machine rather than a
    /// benchmark from somewhere else.
    pub median_query_ms: i64,
}

#[derive(Debug, Clone, Default, serde::Serialize)]
pub struct LanguageLines {
    pub language: String,
    pub lines: i64,
    pub files: i64,
}

#[derive(Debug, Clone, Default, serde::Serialize)]
pub struct KindCount {
    pub name: String,
    pub count: i64,
}

/// One kind of file on the prose panel: how many there are and, for a kind
/// with a unit, how many of that unit they hold.
#[derive(Debug, Clone, Default, serde::Serialize)]
pub struct FileKind {
    pub name: String,
    /// Files of this kind.
    pub count: i64,
    /// `pages`, `slides` or `cells`; `None` for a kind counted in files only.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub unit: Option<&'static str>,
    /// The unit summed over the files that have a count.
    pub units: i64,
    /// Files of a counted kind with no count yet — stored by an older binary,
    /// counted by the next index pass.
    pub uncounted: i64,
}

#[derive(Debug, Clone, Default, serde::Serialize)]
pub struct FileLines {
    pub path: String,
    pub lines: i64,
}

#[derive(Debug, Clone, Default, serde::Serialize)]
pub struct MonthCount {
    /// `YYYY-MM`, in UTC, because that is what SQLite's `strftime` gives and a
    /// month boundary in the reader's zone is not worth a time-zone database.
    pub month: String,
    pub chunks: i64,
}

/// A line's comment openers, by the language the path says it is.
///
/// Short on purpose. It is here to answer "how much of this corpus is prose
/// inside code", and getting `//` and `#` right answers that for nearly every
/// file anybody indexes. A language missing from it contributes no comment
/// lines rather than a wrong count.
fn comment_openers(language: &str) -> &'static [&'static str] {
    // Lowercased, because `filter::Language::name` is already lowercase —
    // `rust`, not `Rust`. Matching the capitalised spellings found nothing and
    // reported a corpus with no comments in it at all.
    match language.to_ascii_lowercase().as_str() {
        "rust" | "c" | "c++" | "c#" | "go" | "java" | "javascript" | "typescript" | "tsx"
        | "jsx" | "swift" | "kotlin" | "scala" | "php" | "dart" | "zig" | "css" => {
            &["//", "/*", "*"]
        }
        "python" | "ruby" | "shell" | "bash" | "perl" | "r" | "yaml" | "toml" | "makefile"
        | "dockerfile" | "elixir" | "nix" => &["#"],
        "sql" | "lua" | "haskell" => &["--"],
        "html" | "xml" | "markdown" => &["<!--"],
        "lisp" | "clojure" => &[";"],
        _ => &[],
    }
}

/// Measure one store, for the `Inside the index` page.
pub fn corpus(db: &Connection, language_of: impl Fn(&str) -> String) -> Result<Corpus> {
    let (files, chunks, bytes) = stats(db)?;
    let mut out = Corpus {
        files,
        chunks,
        bytes,
        vector_dim: 384,
        ..Default::default()
    };

    // One pass over the chunks, joined to the path so the comment rule knows
    // which language it is reading. Everything counted per file first, so
    // "lines" is the file's own length rather than the sum of its chunks'
    // spans — chunks overlap.
    let mut per_file: std::collections::HashMap<i64, (String, i64)> =
        std::collections::HashMap::new();
    {
        let mut q = db.prepare(
            "SELECT c.file_id, f.path, c.end_line, c.text
               FROM chunks c JOIN files f ON f.id = c.file_id",
        )?;
        let mut rows = q.query([])?;
        while let Some(row) = rows.next()? {
            let file_id: i64 = row.get(0)?;
            let path: String = row.get(1)?;
            let end: i64 = row.get(2)?;
            let text: String = row.get(3)?;
            let language = language_of(&path);
            let openers = comment_openers(&language);
            out.characters += text.chars().count() as i64;
            out.words += text.split_whitespace().count() as i64;
            for line in text.lines() {
                let trimmed = line.trim_start();
                if trimmed.is_empty() {
                    out.blank_lines += 1;
                } else if openers.iter().any(|o| trimmed.starts_with(o)) {
                    out.comment_lines += 1;
                }
            }
            let slot = per_file.entry(file_id).or_insert((path, 0));
            slot.1 = slot.1.max(end);
        }
    }

    let mut by_language: std::collections::HashMap<String, (i64, i64)> =
        std::collections::HashMap::new();
    for (path, lines) in per_file.values() {
        out.lines += lines;
        let language = language_of(path);
        let name = if language.is_empty() {
            "other".to_string()
        } else {
            language
        };
        let slot = by_language.entry(name).or_insert((0, 0));
        slot.0 += lines;
        slot.1 += 1;
        if out.longest_file.as_ref().is_none_or(|f| f.lines < *lines) {
            out.longest_file = Some(FileLines {
                path: path.clone(),
                lines: *lines,
            });
        }
        let depth = path.trim_matches('/').matches('/').count() as i64;
        out.deepest_path = out.deepest_path.max(depth);
    }
    out.languages = by_language
        .into_iter()
        .map(|(language, (lines, files))| LanguageLines {
            language,
            lines,
            files,
        })
        .collect();
    out.languages
        .sort_by_key(|row| std::cmp::Reverse(row.lines));

    // The kinds with a unit are exactly the formats `chunk::COUNTED` names,
    // so every file under them either has a count or is waiting for one.
    {
        let mut q = db.prepare(
            "SELECT CASE
                      WHEN lower(path) LIKE '%.pdf' THEN 'PDF'
                      WHEN lower(path) LIKE '%.pptx' OR lower(path) LIKE '%.odp' THEN 'Slide deck'
                      WHEN lower(path) LIKE '%.xlsx' OR lower(path) LIKE '%.ods'
                        OR lower(path) LIKE '%.csv' OR lower(path) LIKE '%.tsv' THEN 'Spreadsheet'
                      WHEN lower(path) LIKE '%.ipynb' THEN 'Notebook'
                      WHEN lower(path) LIKE '%.docx' OR lower(path) LIKE '%.odt' THEN 'Document'
                      WHEN lower(path) LIKE '%.png' OR lower(path) LIKE '%.jpg'
                        OR lower(path) LIKE '%.jpeg' OR lower(path) LIKE '%.webp' THEN 'Image'
                      ELSE 'Text and code'
                    END AS k,
                    COUNT(*), COALESCE(SUM(units), 0), SUM(units IS NULL)
               FROM files GROUP BY k ORDER BY COUNT(*) DESC",
        )?;
        let mut rows = q.query([])?;
        while let Some(row) = rows.next()? {
            let name: String = row.get(0)?;
            let unit = match name.as_str() {
                "PDF" => Some("pages"),
                "Slide deck" => Some("slides"),
                "Spreadsheet" | "Notebook" => Some("cells"),
                _ => None,
            };
            out.kinds.push(FileKind {
                count: row.get(1)?,
                units: row.get(2)?,
                uncounted: if unit.is_some() { row.get(3)? } else { 0 },
                unit,
                name,
            });
        }
    }

    let span: (i64, i64) = db.query_row(
        "SELECT COALESCE(MIN(indexed_at), 0), COALESCE(MAX(indexed_at), 0) FROM files",
        [],
        |r| Ok((r.get(0)?, r.get(1)?)),
    )?;
    out.first_indexed = span.0;
    out.last_indexed = span.1;

    {
        let mut q = db.prepare(
            "SELECT strftime('%Y-%m', f.indexed_at, 'unixepoch') AS m, COUNT(c.id)
               FROM files f JOIN chunks c ON c.file_id = f.id
              GROUP BY m ORDER BY m",
        )?;
        let mut rows = q.query([])?;
        while let Some(row) = rows.next()? {
            out.months.push(MonthCount {
                month: row.get(0)?,
                chunks: row.get(1)?,
            });
        }
    }

    out.symbols = db.query_row("SELECT COUNT(*) FROM symbols", [], |r| r.get(0))?;
    out.edges = db.query_row("SELECT COUNT(*) FROM edges", [], |r| r.get(0))?;

    /* The four tiers, from `coverage_by_language`, which is the one place that
     * decides them.
     *
     * Grouping `edges.confidence` stood here and is a different question: the
     * column holds `extracted` or `inferred`, and whether an inferred edge is
     * resolved, ambiguous or unresolved depends on how many definitions of its
     * target this store holds. So the page read `resolved 0 · ambiguous 0`
     * directly above a coverage table reading 2380 and 2158 -- two answers to
     * one question, on one page, from one store.
     *
     * `languages_with_edges` comes from the same pass for the same reason. */
    let coverage = coverage_by_language(db)?;
    let mut tiers = [0i64; 4];
    for row in &coverage {
        tiers[0] += row.extracted as i64;
        tiers[1] += row.resolved as i64;
        tiers[2] += row.ambiguous as i64;
        tiers[3] += row.unresolved as i64;
        if row.extracted + row.resolved + row.ambiguous > 0 {
            out.languages_with_edges += 1;
        }
    }
    out.tiers = ["extracted", "resolved", "ambiguous", "unresolved"]
        .iter()
        .zip(tiers)
        .map(|(name, count)| KindCount {
            name: (*name).to_string(),
            count,
        })
        .collect();
    out.unresolved = tiers[3];
    let (names, _) = unresolved_targets(db, 7)?;
    out.unresolved_names = names
        .into_iter()
        .map(|(name, count)| KindCount { name, count })
        .collect();
    out.ambiguous_names = db.query_row(
        "SELECT COUNT(*) FROM (SELECT name FROM symbols GROUP BY name HAVING COUNT(*) > 1)",
        [],
        |r| r.get(0),
    )?;
    out.ambiguous_worst = counted(
        db,
        "SELECT name, COUNT(*) c FROM symbols GROUP BY name HAVING c > 1
          ORDER BY c DESC, name LIMIT 5",
    )?;

    // The middle of what this machine actually measured. `ms` is recorded per
    // retrieval, so this is a fact about this store rather than a benchmark.
    out.median_query_ms = db
        .query_row(
            // `micros`, which is what the column is called. Asking for `ms`
            // did not fail loudly — the row read errored and the fallback
            // reported a store that had never been queried.
            "SELECT micros / 1000 FROM retrievals WHERE COALESCE(tool, '') NOT IN ('note', 'refused')
              ORDER BY micros LIMIT 1 OFFSET (SELECT COUNT(*) FROM retrievals
              WHERE COALESCE(tool, '') NOT IN ('note', 'refused')) / 2",
            [],
            |r| r.get(0),
        )
        .unwrap_or(0);

    Ok(out)
}

/// A two-column `name, count` query, as the list the page draws from it.
fn counted(db: &Connection, sql: &str) -> Result<Vec<KindCount>> {
    let mut q = db.prepare(sql)?;
    let mut rows = q.query([])?;
    let mut out = Vec::new();
    while let Some(row) = rows.next()? {
        out.push(KindCount {
            name: row.get(0)?,
            count: row.get(1)?,
        });
    }
    Ok(out)
}

/// One session's whole ledger, as the per-session table reads it.
#[derive(Debug, Clone, Default)]
pub struct SessionRow {
    /// The session id the client sent at handshake. Empty for rows written
    /// before 0.15.0, which is said rather than guessed at.
    pub session: String,
    pub client: String,
    /// When the session's first and last recorded retrieval happened.
    pub first: i64,
    pub last: i64,
    pub retrievals: i64,
    /// Retrievals that found nothing. Recorded and credited nothing — a
    /// ledger that remembers only its successes is a marketing document.
    pub zero_hit: i64,
    pub excerpt_tokens: i64,
    pub whole_file_tokens: i64,
    /// Whole-file less excerpt, over the rows that found something.
    pub net: i64,
    /// Rows counted with the store's own tokenizer rather than by the
    /// four-characters fallback. The tier of this row's figures.
    pub measured: i64,
    /// The model the session's calls were made by, as its client's own log
    /// recorded it — the one most of its rows name. `None` while usage is off
    /// or the client keeps no log: a session's saving is priced at the model
    /// it ran on, or not at all.
    pub model: Option<String>,
}

impl SessionRow {
    /// `measured` when every credited row in the session was counted by a
    /// model's own tokenizer, `modelled` otherwise. Never averaged: two rows
    /// counted two ways are not one number.
    pub fn tier(&self) -> &'static str {
        if self.measured == self.retrievals && self.retrievals > 0 {
            "measured"
        } else {
            "modelled"
        }
    }
}

/// Every session this store recorded, newest last-seen first.
///
/// Grouped in SQL rather than in the page, for the reason the totals are:
/// one search over six stores writes a row in each, and a browser adding
/// them up is a second opinion about a number the store can state.
pub fn ledger_sessions(db: &Connection, limit: usize) -> Result<Vec<SessionRow>> {
    let mut stmt = db.prepare(
        "SELECT COALESCE(session, ''), client,
                MIN(at), MAX(at), COUNT(*),
                SUM(CASE WHEN hits = 0 THEN 1 ELSE 0 END),
                SUM(excerpt_tokens), SUM(whole_file_tokens),
                SUM(CASE WHEN hits > 0 THEN whole_file_tokens - excerpt_tokens ELSE 0 END),
                SUM(CASE WHEN tokenizer IS NOT NULL AND tokenizer <> 'chars4' THEN 1 ELSE 0 END),
                (SELECT r2.model FROM retrievals r2
                  WHERE COALESCE(r2.session, '') = COALESCE(retrievals.session, '')
                    AND r2.client = retrievals.client AND r2.model IS NOT NULL
                  GROUP BY r2.model ORDER BY COUNT(*) DESC LIMIT 1)
         FROM retrievals
         WHERE COALESCE(tool, '') NOT IN ('note', 'refused')
         GROUP BY COALESCE(session, ''), client
         ORDER BY MAX(at) DESC
         LIMIT ?1",
    )?;
    let rows = stmt.query_map([limit as i64], |r| {
        Ok(SessionRow {
            session: r.get(0)?,
            client: r.get(1)?,
            first: r.get(2)?,
            last: r.get(3)?,
            retrievals: r.get(4)?,
            zero_hit: r.get(5)?,
            excerpt_tokens: r.get(6)?,
            whole_file_tokens: r.get(7)?,
            net: r.get(8)?,
            measured: r.get(9)?,
            model: r.get(10)?,
        })
    })?;
    Ok(rows.collect::<rusqlite::Result<_>>()?)
}

/// Warm query latency as the ledger recorded it: p50 and p95, in microseconds.
///
/// From the rows themselves rather than from a benchmark, so the figure is
/// what this machine actually served rather than what it can serve.
pub fn ledger_latency(db: &Connection) -> Result<(i64, i64)> {
    // One figure per retrieval and none for a row that measured no work: a
    // median over raw-read rows' zeros and one row per store per search is
    // how a report came to say 724 ms where the rows said ~67.
    let mut stmt = db.prepare(&format!(
        "SELECT MAX(micros) AS m FROM retrievals WHERE micros > 0 AND {AGENT_RETRIEVAL}
          GROUP BY COALESCE(query_id, 'row:' || id) ORDER BY m"
    ))?;
    let rows = stmt.query_map([], |r| r.get::<_, i64>(0))?;
    let all: Vec<i64> = rows.collect::<rusqlite::Result<_>>()?;
    if all.is_empty() {
        return Ok((0, 0));
    }
    let at = |q: f64| all[((all.len() as f64 - 1.0) * q).round() as usize];
    Ok((at(0.50), at(0.95)))
}

/// The hash of the newest ledger row: what `--verify` checks the chain to.
pub fn ledger_head(db: &Connection) -> Result<Option<String>> {
    let hash: Option<String> = db
        .query_row(
            "SELECT hash FROM retrievals ORDER BY id DESC LIMIT 1",
            [],
            |r| r.get(0),
        )
        .optional()?;
    Ok(hash)
}

/// Queries that found nothing, commonest first.
///
/// The gap list: a question an agent asked that this corpus could not answer
/// is either a missing document or a name nobody uses in the words they
/// searched with.
pub fn ledger_zero_hit_queries(db: &Connection, limit: usize) -> Result<Vec<(String, i64)>> {
    ledger_zero_hit_queries_since(db, 0, limit)
}

/// The questions that found nothing, at or after `since`.
pub fn ledger_zero_hit_queries_since(
    db: &Connection,
    since: i64,
    limit: usize,
) -> Result<Vec<(String, i64)>> {
    let mut stmt = db.prepare(&format!(
        "SELECT query, COUNT(*) AS n FROM (
            SELECT MIN(query) AS query FROM retrievals WHERE at >= ?1 AND {AGENT_RETRIEVAL}
             GROUP BY COALESCE(query_id, 'row:' || id) HAVING MAX(hits) = 0)
         GROUP BY query ORDER BY n DESC, query LIMIT ?2"
    ))?;
    let rows = stmt.query_map(rusqlite::params![since, limit as i64], |r| {
        Ok((r.get(0)?, r.get(1)?))
    })?;
    Ok(rows.collect::<rusqlite::Result<_>>()?)
}

/// One recorded retrieval, as the access report attaches them.
#[derive(Debug, Clone)]
pub struct RetrievalRow {
    pub at: i64,
    pub client: String,
    pub tool: String,
    pub query: String,
    pub hits: i64,
    pub excerpt_tokens: i64,
    pub whole_file_tokens: i64,
}

/// Every retrieval since `since`, newest first.
///
/// `ledger_sessions` above aggregates these into one row per session, which is
/// what the access report shows by default. This is the same rows unrolled, for
/// the report's `Attach retrieved excerpts` toggle: a session line says an agent
/// asked forty times, and an auditor's question is which forty.
///
/// `tool` is NULL on every row written before 0.15.0 and is reported as `—`
/// rather than guessed — those rows are real retrievals and dropping them would
/// make the attachment disagree with the session counts above it.
pub fn ledger_retrievals(db: &Connection, since: i64, limit: usize) -> Result<Vec<RetrievalRow>> {
    let mut stmt = db.prepare(
        "SELECT at, client, COALESCE(tool, ''), query, hits, excerpt_tokens, whole_file_tokens
         FROM retrievals WHERE at >= ?1 AND COALESCE(tool, '') NOT IN ('note', 'refused')
         ORDER BY at DESC, id DESC LIMIT ?2",
    )?;
    let rows = stmt.query_map(rusqlite::params![since, limit as i64], |r| {
        Ok(RetrievalRow {
            at: r.get(0)?,
            client: r.get(1)?,
            tool: r.get(2)?,
            query: r.get(3)?,
            hits: r.get(4)?,
            excerpt_tokens: r.get(5)?,
            whole_file_tokens: r.get(6)?,
        })
    })?;
    Ok(rows.collect::<rusqlite::Result<_>>()?)
}

/// Files this store read since `since`, newest first, with when it read them.
pub fn files_indexed_since(
    db: &Connection,
    since: i64,
    limit: usize,
) -> Result<Vec<(String, i64)>> {
    let mut stmt = db.prepare(
        "SELECT path, indexed_at FROM files WHERE indexed_at >= ?1
         ORDER BY indexed_at DESC LIMIT ?2",
    )?;
    let rows = stmt.query_map([since, limit as i64], |r| Ok((r.get(0)?, r.get(1)?)))?;
    Ok(rows.collect::<rusqlite::Result<_>>()?)
}

/// One settled dependency edge, by name, for the community pass.
///
/// Names rather than ids, because `edges.dst` is a name and the whole point
/// of the pass is to group names. Only edges whose target the store actually
/// defines are returned: an edge into a dependency that was never indexed
/// cannot join two communities of this corpus.
pub struct NamedEdge {
    pub from: String,
    pub to: String,
}

/// Every `calls` and `imports` edge whose target this store defines, up to
/// `limit`, with the total so a reader is told when the list was cut.
///
/// This is the one whole-graph read in the crate and it is bounded for that
/// reason. Everything else walks one hop at a time to keep peak memory flat
/// as a corpus grows; communities are a property of the whole graph and
/// cannot be computed a hop at a time, so the bound is the substitute — past
/// it the panel says `Shown N of M` rather than quietly clustering a slice.
pub fn community_edges(db: &Connection, limit: usize) -> Result<(Vec<NamedEdge>, i64)> {
    let target = crate::graph::not_navigational("s.kind");
    // `extracted` and `inferred` are the stored values; `resolved` and
    // `ambiguous` are computed per query and never written, so "settled" here
    // means an extracted edge or a bare-name match onto exactly one
    // definition. A name with several definitions is not evidence that two
    // subsystems are joined.
    let sql = format!(
        "SELECT src.name, s.name
         FROM symbols src
         JOIN edges e ON e.src = src.id
         JOIN symbols s ON s.name = e.dst AND {target}
         WHERE e.kind IN ('calls', 'imports')
           AND src.name <> s.name
           AND (SELECT COUNT(*) FROM symbols d WHERE d.name = e.dst AND {}) = 1
         LIMIT ?1",
        crate::graph::not_navigational("d.kind")
    );
    let mut stmt = db.prepare(&sql)?;
    let rows = stmt.query_map([limit as i64], |r| {
        Ok(NamedEdge {
            from: r.get(0)?,
            to: r.get(1)?,
        })
    })?;
    let edges: Vec<NamedEdge> = rows.collect::<rusqlite::Result<_>>()?;
    let total: i64 = db.query_row(
        "SELECT COUNT(*) FROM edges WHERE kind IN ('calls', 'imports')",
        [],
        |r| r.get(0),
    )?;
    Ok((edges, total))
}

/// Where each of these names is defined, for the hub lines under a community.
pub fn where_defined(db: &Connection, names: &[String]) -> Result<Vec<(String, String, u32)>> {
    if names.is_empty() {
        return Ok(Vec::new());
    }
    let holes = vec!["?"; names.len()].join(", ");
    let sql = format!(
        "SELECT s.name, f.path, s.start_line
         FROM symbols s JOIN files f ON f.id = s.file_id
         WHERE s.name IN ({holes})
         ORDER BY f.path, s.start_line"
    );
    let mut stmt = db.prepare(&sql)?;
    let binds: Vec<Value> = names.iter().map(|n| Value::Text(n.clone())).collect();
    let rows = stmt.query_map(rusqlite::params_from_iter(binds), |r| {
        Ok((r.get(0)?, r.get(1)?, r.get(2)?))
    })?;
    Ok(rows.collect::<rusqlite::Result<_>>()?)
}

/// Call targets no definition in this store satisfies, commonest first, and
/// how many distinct ones there are.
///
/// The names a reader has to recognise as "outside this corpus" before they
/// read an absent edge as a missing call. Counted rather than listed in full.
/// The language's own builtins and standard library are left out
/// ([`crate::graph::is_builtin`]) — cobra's list was `Errorf` 198 times,
/// `len`, `append` and `os` (2026-10-14 audit) — and [`builtin_targets`]
/// counts what was left out, so a report can say so.
pub fn unresolved_targets(db: &Connection, limit: usize) -> Result<(Vec<(String, i64)>, i64)> {
    let (mut all, _) = outside_targets(db)?;
    let distinct = all.len() as i64;
    all.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(&b.0)));
    all.truncate(limit);
    Ok((all, distinct))
}

/// How many distinct call and import targets with no definition here are the
/// language's own builtins or standard library, and so are not in
/// [`unresolved_targets`].
pub fn builtin_targets(db: &Connection) -> Result<i64> {
    Ok(outside_targets(db)?.1)
}

/// Targets nothing here defines, as `(name, edges)` outside the builtins, and
/// the count of distinct names that were builtins only.
fn outside_targets(db: &Connection) -> Result<(Vec<(String, i64)>, i64)> {
    // One existence test per distinct target rather than per edge, then the
    // per-file rows of only those targets, because whether a name is a
    // builtin depends on the calling file's language and the edge's hint.
    // Cached per store until its graph changes, as the report is asked for
    // again and again while it is being read.
    cached_by_graph(db, "unresolved", || {
        let target = crate::graph::not_navigational("d.kind");
        let mut stmt = db.prepare(&format!(
            "WITH t AS (SELECT dst FROM edges WHERE kind IN ('calls', 'imports') GROUP BY dst),
                  u AS (SELECT dst FROM t
                        WHERE NOT EXISTS (SELECT 1 FROM symbols d WHERE d.name = t.dst AND {target}))
             SELECT e.dst, e.kind, e.hint, f.path, COUNT(*)
             FROM u JOIN edges e ON e.dst = u.dst
             JOIN symbols s ON s.id = e.src JOIN files f ON f.id = s.file_id
             WHERE e.kind IN ('calls', 'imports')
             GROUP BY e.dst, e.kind, e.hint, s.file_id"
        ))?;
        let rows = stmt.query_map([], |r| {
            Ok((
                r.get::<_, String>(0)?,
                r.get::<_, String>(1)?,
                r.get::<_, Option<String>>(2)?,
                r.get::<_, String>(3)?,
                r.get::<_, i64>(4)?,
            ))
        })?;
        let mut language: std::collections::HashMap<String, &'static str> =
            std::collections::HashMap::new();
        let mut kept: std::collections::HashMap<String, i64> = std::collections::HashMap::new();
        let mut builtin: std::collections::HashSet<String> = std::collections::HashSet::new();
        for row in rows {
            let (name, kind, hint, path, n) = row?;
            let lang = *language.entry(path).or_insert_with_key(|p| {
                crate::graph::language_of(std::path::Path::new(p)).unwrap_or("")
            });
            if kind == "imports" && crate::graph::is_relative_import(lang, &name) {
                continue;
            }
            if crate::graph::is_builtin(lang, &name, &kind, hint.as_deref()) {
                builtin.insert(name);
            } else {
                *kept.entry(name).or_insert(0) += n;
            }
        }
        builtin.retain(|name| !kept.contains_key(name));
        Ok((kept.into_iter().collect::<Vec<_>>(), builtin.len() as i64))
    })
}

/// A figure computed over a whole store's graph, kept until the graph's
/// symbol or edge count moves. The health report's aggregates read millions
/// of rows on a large store and give the same answer until something is
/// indexed or forgotten.
fn cached_by_graph<T: Clone + Send + 'static>(
    db: &Connection,
    what: &str,
    compute: impl FnOnce() -> Result<T>,
) -> Result<T> {
    use std::any::Any;
    type Slot = (String, (i64, i64), Box<dyn Any + Send>);
    static CACHE: std::sync::Mutex<Vec<Slot>> = std::sync::Mutex::new(Vec::new());
    let key = format!("{what}|{}", db.path().unwrap_or(""));
    let stats = graph_stats(db)?;
    if let Some((_, _, value)) = CACHE
        .lock()
        .unwrap_or_else(|e| e.into_inner())
        .iter()
        .find(|(k, st, _)| *k == key && *st == stats)
        && let Some(value) = value.downcast_ref::<T>()
    {
        return Ok(value.clone());
    }
    let value = compute()?;
    let mut cache = CACHE.lock().unwrap_or_else(|e| e.into_inner());
    cache.retain(|(k, _, _)| *k != key);
    cache.push((key, stats, Box::new(value.clone())));
    Ok(value)
}

/// How many names this store defines more than once.
///
/// The population of every ambiguous edge: a walk refuses to cross one of
/// these, so the count is what a reader needs to know before reading "not
/// connected" as "nothing calls it".
pub fn names_with_several_definitions(db: &Connection) -> Result<i64> {
    let target = crate::graph::not_navigational("s.kind");
    db.query_row(
        &format!(
            "SELECT COUNT(*) FROM (SELECT s.name FROM symbols s WHERE {target}
             GROUP BY s.name HAVING COUNT(*) > 1)"
        ),
        [],
        |r| r.get(0),
    )
    .map_err(Into::into)
}

/// Chunks by the month their file was indexed, oldest first.
///
/// Indexed rather than written: the store records when it read a file, not
/// when somebody wrote it, and labelling this "chunks per month" without
/// saying which would be a claim about a repository's history that the store
/// cannot make.
pub fn chunks_by_month(db: &Connection) -> Result<Vec<(String, i64)>> {
    let mut stmt = db.prepare(
        "SELECT strftime('%Y-%m', f.indexed_at, 'unixepoch') AS month, COUNT(c.id)
         FROM chunks c JOIN files f ON f.id = c.file_id
         GROUP BY month ORDER BY month",
    )?;
    let rows = stmt.query_map([], |r| Ok((r.get(0)?, r.get(1)?)))?;
    Ok(rows.collect::<rusqlite::Result<_>>()?)
}

/// One ledger row as the reports count it (0.42.0).
#[derive(Debug, Clone)]
pub struct ReportRow {
    pub id: i64,
    pub at: i64,
    pub client: String,
    pub client_version: String,
    pub session: String,
    pub tool: String,
    pub query: String,
    /// The retrieval this row is part of: one search over six stores is six
    /// rows and one key. A row written before 0.20.2 has no id and is its own.
    pub key: String,
    pub hits: i64,
    pub micros: i64,
    pub excerpt_tokens: i64,
    pub whole_file_tokens: i64,
    pub tokenizer: String,
}

/// Every ledger row recorded at or after `since`, oldest first.
///
/// The reports take their figures from the rows rather than from the
/// aggregates above, so a window narrows every figure a report prints — the
/// 0.41 savings report narrowed net, credited and total and left whole,
/// excerpt, refunds and latency lifetime (R5). Nothing is filtered here: what
/// counts as a query is the report's rule, and a reader that filtered would be
/// a second copy of it.
pub fn report_rows_since(db: &Connection, store: &str, since: i64) -> Result<Vec<ReportRow>> {
    let mut stmt = db.prepare(
        "SELECT id, at, client, COALESCE(client_version, ''), COALESCE(session, ''),
                COALESCE(tool, ''), query, query_id, hits, micros, excerpt_tokens,
                whole_file_tokens, COALESCE(tokenizer, '')
         FROM retrievals WHERE at >= ?1 ORDER BY id",
    )?;
    let rows = stmt.query_map([since], |r| {
        let id: i64 = r.get(0)?;
        let key: Option<String> = r.get(7)?;
        Ok(ReportRow {
            id,
            at: r.get(1)?,
            client: r.get(2)?,
            client_version: r.get(3)?,
            session: r.get(4)?,
            tool: r.get(5)?,
            query: r.get(6)?,
            key: key
                .filter(|k| !k.is_empty())
                .unwrap_or_else(|| format!("{store}:row:{id}")),
            hits: r.get(8)?,
            micros: r.get(9)?,
            excerpt_tokens: r.get(10)?,
            whole_file_tokens: r.get(11)?,
            tokenizer: r.get(12)?,
        })
    })?;
    Ok(rows.collect::<rusqlite::Result<_>>()?)
}

/// A file the store read at or after `since`, and what it held before.
#[derive(Debug, Clone)]
pub struct ReadFile {
    pub path: String,
    pub indexed_at: i64,
    /// The definitions the first re-index inside the window retired — what
    /// the file held when the window opened: `(name, kind, start, end)`.
    /// Empty for a file read for the first time, and for a re-read file that
    /// held no definitions, which the store cannot tell apart from a first read.
    pub before: Vec<(String, String, u32, u32)>,
    /// Its definitions now: `(name, kind, start, end)`.
    pub now: Vec<(String, String, u32, u32)>,
}

/// The first and last indexed path in sort order. Their common directory is
/// the common directory of every path, which is the root a report prints
/// paths against when the registry names none (a store opened by `--store`).
pub fn path_bounds(db: &Connection) -> Result<Option<(String, String)>> {
    Ok(
        db.query_row("SELECT MIN(path), MAX(path) FROM files", [], |r| {
            Ok(
                match (
                    r.get::<_, Option<String>>(0)?,
                    r.get::<_, Option<String>>(1)?,
                ) {
                    (Some(a), Some(b)) => Some((a, b)),
                    _ => None,
                },
            )
        })?,
    )
}

/// Every file read at or after `since`, newest first, with no cap — the
/// change brief's "Files re-read" was 200 per store summed (R3).
pub fn files_read_since(db: &Connection, since: i64) -> Result<Vec<ReadFile>> {
    let mut stmt = db.prepare(
        "SELECT id, path, indexed_at FROM files WHERE indexed_at >= ?1
         ORDER BY indexed_at DESC, path",
    )?;
    let files: Vec<(i64, String, i64)> = stmt
        .query_map([since], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))?
        .collect::<rusqlite::Result<_>>()?;
    let mut past = db.prepare_cached(
        "SELECT name, kind, start_line, end_line FROM symbols_past
         WHERE path = ?1 AND retired_at = (SELECT MIN(retired_at) FROM symbols_past
                                           WHERE path = ?1 AND retired_at >= ?2)
           AND kind != 'module'",
    )?;
    let mut live = db.prepare_cached(
        "SELECT name, kind, start_line, end_line FROM symbols
         WHERE file_id = ?1 AND kind != 'module' ORDER BY start_line",
    )?;
    let mut out = Vec::with_capacity(files.len());
    for (id, path, indexed_at) in files {
        let span = |r: &rusqlite::Row| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?));
        let before = past
            .query_map(params![path, since], span)?
            .collect::<rusqlite::Result<_>>()?;
        let now = live
            .query_map([id], span)?
            .collect::<rusqlite::Result<_>>()?;
        out.push(ReadFile {
            path,
            indexed_at,
            before,
            now,
        });
    }
    Ok(out)
}

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

    /// The change brief's history lookup searches the path index rather than
    /// scanning symbols_past, which never finished on the owner's store.
    #[test]
    fn the_change_briefs_history_lookup_uses_the_path_index() {
        let dir = tempfile::tempdir().unwrap();
        let db = open(&dir.path().join("s.db")).unwrap();
        let plan: Vec<String> = db
            .prepare(
                "EXPLAIN QUERY PLAN
                 SELECT name, kind, start_line, end_line FROM symbols_past
                  WHERE path = ?1 AND retired_at = (SELECT MIN(retired_at) FROM symbols_past
                                                    WHERE path = ?1 AND retired_at >= ?2)
                    AND kind != 'module'",
            )
            .unwrap()
            .query_map(params!["/a.rs", 0], |r| r.get::<_, String>(3))
            .unwrap()
            .collect::<rusqlite::Result<_>>()
            .unwrap();
        assert!(
            plan.iter()
                .all(|step| !step.starts_with("SCAN symbols_past")),
            "{plan:?}"
        );
        assert!(
            plan.iter().any(|step| step.contains("symbols_past_path")),
            "{plan:?}"
        );
    }

    /// Several connections opening one new store at the same moment, as the
    /// daemon's index route, its watcher and a CLI do (#132). Before the open
    /// was made safe to race, most rounds lost one of them to "duplicate
    /// column name" or "database is locked".
    #[test]
    fn a_new_store_opened_by_several_connections_at_once_opens_for_all() {
        let root = tempfile::tempdir().unwrap();
        for round in 0..40 {
            let db = root.path().join(format!("{round}.db"));
            let start = std::sync::Arc::new(std::sync::Barrier::new(6));
            let openers: Vec<_> = (0..6)
                .map(|_| {
                    let db = db.clone();
                    let start = std::sync::Arc::clone(&start);
                    std::thread::spawn(move || {
                        start.wait();
                        open(&db).map(drop).map_err(|e| format!("{e:#}"))
                    })
                })
                .collect();
            for opener in openers {
                if let Err(e) = opener.join().unwrap() {
                    panic!("round {round}: {e}");
                }
            }
        }
    }

    fn sym(name: &str) -> crate::graph::Symbol {
        crate::graph::Symbol {
            kind: "function".to_string(),
            name: name.to_string(),
            qualified: name.to_string(),
            start_line: 1,
            end_line: 2,
        }
    }

    fn at(mut s: crate::graph::Symbol, start: u32, end: u32) -> crate::graph::Symbol {
        s.start_line = start;
        s.end_line = end;
        s
    }

    /// Build a store with one file, one chunk and one symbol, and return the
    /// connection plus the symbol's id.
    fn one_symbol(db: &Connection, path: &str, name: &str) -> i64 {
        db.execute_batch(SCHEMA).unwrap();
        db.pragma_update(None, "foreign_keys", "ON").unwrap();
        let file_id = insert_file(db, path, "h", 1, 0).unwrap();
        let chunk_id = insert_chunk(db, file_id, 0, 1, 2, name).unwrap();
        insert_symbol(db, file_id, Some(chunk_id), &sym(name)).unwrap()
    }

    /// More names than SQLite takes variables in one statement. 0.32.0 bound
    /// them all into one `IN (...)`, and on the 70-repository benchmark corpus
    /// every search failed with "too many SQL variables".
    /// Stopwords leave the keyword list's query unless they are all there is.
    #[test]
    fn stopwords_leave_the_keyword_query_unless_nothing_else_is_there() {
        let db = Connection::open_in_memory().unwrap();
        prepare_for_tests(&db);
        let file = insert_file(&db, "a.md", "h", 1, 0).unwrap();
        for (ord, text) in ["the parser recovers", "of the and the of"]
            .iter()
            .enumerate()
        {
            insert_chunk(&db, file, ord, ord as u32 + 1, ord as u32 + 1, text).unwrap();
        }
        let ids = keyword_search(&db, "how does the parser recover", 10, &[]).unwrap();
        assert_eq!(ids.len(), 1, "a chunk of stopwords matched: {ids:?}");
        let only = keyword_search(&db, "the of", 10, &[]).unwrap();
        assert_eq!(
            only.len(),
            2,
            "a query of stopwords found nothing: {only:?}"
        );
    }

    /// A read that runs past its limit is interrupted and says so; one that
    /// finishes inside it is untouched.
    #[test]
    fn a_bounded_read_is_interrupted_at_its_limit() {
        let db = Connection::open_in_memory().unwrap();
        let started = std::time::Instant::now();
        let slow = bounded(&[&db], std::time::Duration::from_millis(100), || {
            let n: i64 = db.query_row(
                "WITH RECURSIVE c(x) AS (SELECT 1 UNION ALL SELECT x + 1 FROM c) \
                 SELECT COUNT(*) FROM (SELECT x FROM c LIMIT 10000000000)",
                [],
                |r| r.get(0),
            )?;
            Ok(n)
        });
        let err = slow.expect_err("a ten-billion-row count finished");
        assert!(is_interrupted(&err), "{err:#}");
        assert!(started.elapsed() < std::time::Duration::from_secs(5));

        let quick = bounded(&[&db], std::time::Duration::from_secs(5), || {
            Ok(db.query_row("SELECT 1", [], |r| r.get::<_, i64>(0))?)
        });
        assert_eq!(quick.unwrap(), 1);
    }

    #[test]
    fn symbols_by_names_takes_more_names_than_sqlite_takes_variables() {
        let db = Connection::open_in_memory().unwrap();
        one_symbol(&db, "b.rs", "wanted");
        let a = insert_file(&db, "a.rs", "h", 1, 0).unwrap();
        insert_symbol(&db, a, None, &at(sym("also_wanted"), 5, 6)).unwrap();
        insert_symbol(&db, a, None, &at(sym("wanted"), 1, 2)).unwrap();

        let mut names: Vec<String> = (0..40_000).map(|i| format!("absent_{i}")).collect();
        names.push("wanted".to_string());
        names.insert(0, "also_wanted".to_string());
        let found = symbols_by_names(&db, &names, &[]).unwrap();

        // One order across batches, as the single statement gave: path, then line.
        let got: Vec<(&str, u32)> = found
            .iter()
            .map(|s| (s.path.as_str(), s.start_line))
            .collect();
        assert_eq!(got, [("a.rs", 1), ("a.rs", 5), ("b.rs", 1)]);
    }

    /// The walk's variant drops an edge whose target is a hub, and only that
    /// edge; `edges_out` itself still settles every edge.
    #[test]
    fn edges_out_short_of_hubs_drops_only_edges_into_hubs() {
        let db = Connection::open_in_memory().unwrap();
        let caller = one_symbol(&db, "a.rs", "caller");
        insert_edge(&db, caller, "rare", "calls", "inferred", None, None).unwrap();
        insert_edge(&db, caller, "common", "calls", "inferred", None, None).unwrap();
        let b = insert_file(&db, "b.rs", "h", 1, 0).unwrap();
        insert_symbol(&db, b, None, &sym("rare")).unwrap();
        for line in 1..=3 {
            insert_symbol(&db, b, None, &at(sym("common"), line * 10, line * 10 + 1)).unwrap();
        }

        let targets = |ends: Vec<EdgeEnd>| {
            let mut names: Vec<String> = ends.into_iter().map(|e| e.symbol.name).collect();
            names.sort();
            names.dedup();
            names
        };
        assert_eq!(
            targets(edges_out_short_of_hubs(&db, "caller", &[], 2).unwrap()),
            ["rare"]
        );
        assert_eq!(
            targets(edges_out_short_of_hubs(&db, "caller", &[], 3).unwrap()),
            ["common", "rare"]
        );
        assert_eq!(
            targets(edges_out(&db, "caller", &[]).unwrap()),
            ["common", "rare"]
        );
    }

    /// Forgetting a file takes its symbols and its outgoing edges with it.
    ///
    /// The cascade does this, not the caller — which is why it is worth a test:
    /// a missing `ON DELETE CASCADE` would leave rows pointing at a file id
    /// SQLite is free to reissue.
    #[test]
    fn deleting_a_file_deletes_its_symbols_and_its_outgoing_edges() {
        let db = Connection::open_in_memory().unwrap();
        let caller = one_symbol(&db, "a.rs", "caller");
        insert_edge(&db, caller, "callee", "calls", "inferred", None, None).unwrap();
        assert_eq!(graph_stats(&db).unwrap(), (1, 1));

        delete_file(&db, "a.rs", 0).unwrap();
        assert_eq!(
            graph_stats(&db).unwrap(),
            (0, 0),
            "the file's symbols and the edges leaving them both cascade"
        );
    }

    /// Re-indexing the file an edge points *into* must not delete the edge.
    ///
    /// This is the whole reason `edges.dst` is a name and not an id. With an id
    /// there, re-extracting `b.rs` would reissue its symbol ids and orphan
    /// every edge from `a.rs` into it — the graph would rot on exactly the
    /// operation this release exists to make safe.
    #[test]
    fn re_indexing_the_target_file_leaves_edges_into_it_intact() {
        let db = Connection::open_in_memory().unwrap();
        let caller = one_symbol(&db, "a.rs", "caller");
        insert_edge(&db, caller, "callee", "calls", "inferred", None, None).unwrap();
        let b = insert_file(&db, "b.rs", "h", 1, 0).unwrap();
        insert_symbol(&db, b, None, &sym("callee")).unwrap();
        assert_eq!(edges_out(&db, "caller", &[]).unwrap().len(), 1);

        // b.rs changes: its symbols are dropped and re-extracted with new ids.
        delete_file(&db, "b.rs", 0).unwrap();
        let b = insert_file(&db, "b.rs", "h2", 1, 0).unwrap();
        insert_symbol(&db, b, None, &at(sym("callee"), 9, 10)).unwrap();

        let out = edges_out(&db, "caller", &[]).unwrap();
        assert_eq!(
            out.len(),
            1,
            "the edge survived its target being re-indexed"
        );
        assert_eq!(
            out[0].symbol.start_line, 9,
            "and now points at the new rows"
        );
    }

    /// A store with one caller and several definitions of the name it calls.
    ///
    /// Returns the connection; `paths` are the files each definition of
    /// `callee` lives in, and the caller is always `src/caller.rs`.
    fn many_definitions(paths: &[&str]) -> Connection {
        let db = Connection::open_in_memory().unwrap();
        one_symbol(&db, "src/caller.rs", "caller");
        for (i, path) in paths.iter().enumerate() {
            // The caller's own file is already there, and a definition in it
            // is exactly the case the same-file rank is about.
            let file = db
                .query_row("SELECT id FROM files WHERE path = ?1", params![path], |r| {
                    r.get(0)
                })
                .optional()
                .unwrap()
                .unwrap_or_else(|| insert_file(&db, path, &format!("h{i}"), 1, 0).unwrap());
            insert_symbol(&db, file, None, &sym("callee")).unwrap();
        }
        db
    }

    fn call(db: &Connection, hint: Option<&str>) {
        let src: i64 = db
            .query_row("SELECT id FROM symbols WHERE name = 'caller'", [], |r| {
                r.get(0)
            })
            .unwrap();
        insert_edge(db, src, "callee", "calls", "inferred", hint, None).unwrap();
    }

    /// One definition of the name is one answer, whatever the hint says.
    #[test]
    fn a_name_with_one_definition_resolves() {
        let db = many_definitions(&["src/other.rs"]);
        call(&db, None);
        let out = edges_out(&db, "caller", &[]).unwrap();
        assert_eq!(out.len(), 1);
        assert_eq!(out[0].confidence, crate::graph::RESOLVED);
        assert_eq!(out[0].definitions, 1);
    }

    /// Four definitions and nothing to choose between them. Every candidate
    /// comes back marked `ambiguous` and carrying the count, so a renderer can
    /// say "4 definitions" instead of printing four calls the source never
    /// made.
    #[test]
    fn a_name_nothing_narrows_is_ambiguous_and_keeps_every_candidate() {
        let db = many_definitions(&["src/a.rs", "src/b.rs", "src/c.rs", "src/d.rs"]);
        call(&db, None);
        let out = edges_out(&db, "caller", &[]).unwrap();
        assert_eq!(out.len(), 4);
        assert!(out.iter().all(|e| e.confidence == crate::graph::AMBIGUOUS));
        assert!(out.iter().all(|e| e.definitions == 4));
    }

    /// The hint picks the definition whose file it names, and the other three
    /// are not returned at all.
    #[test]
    fn a_hint_that_names_a_file_resolves_to_that_definition() {
        let db = many_definitions(&["src/a.rs", "src/store.rs", "src/c.rs"]);
        call(&db, Some("store"));
        let out = edges_out(&db, "caller", &[]).unwrap();
        assert_eq!(out.len(), 1, "{out:?}");
        assert_eq!(out[0].symbol.path, "src/store.rs");
        assert_eq!(out[0].confidence, crate::graph::RESOLVED);
        assert_eq!(
            out[0].definitions, 3,
            "the count is what the ranking chose between, not what survived"
        );
    }

    /// A hint naming a directory works the same way: a module is a file or a
    /// directory depending on how the author felt that day.
    #[test]
    fn a_hint_names_a_directory_as_readily_as_a_file() {
        let db = many_definitions(&["src/a.rs", "src/store/mod.rs"]);
        call(&db, Some("store"));
        let out = edges_out(&db, "caller", &[]).unwrap();
        assert_eq!(out.len(), 1);
        assert_eq!(out[0].symbol.path, "src/store/mod.rs");
    }

    /// A module path outranks the calling file's own namesake: `store::callee()`
    /// in caller.rs is store.rs's `callee`, whatever caller.rs defines. Until
    /// 0.42.0 the calling file won, which in flask's test_config.py settled
    /// `flask.Flask(...)` on a test's own subclass (2026-10-14 audit). A
    /// call with no path, or a path naming the calling file, still means its
    /// own.
    #[test]
    fn a_module_path_outranks_the_calling_file() {
        let db = many_definitions(&["src/caller.rs", "src/store.rs"]);
        call(&db, Some("store"));
        let out = edges_out(&db, "caller", &[]).unwrap();
        assert_eq!(out.len(), 1);
        assert_eq!(out[0].symbol.path, "src/store.rs");

        let db = many_definitions(&["src/caller.rs", "src/store.rs"]);
        call(&db, None);
        let out = edges_out(&db, "caller", &[]).unwrap();
        assert_eq!(out.len(), 1);
        assert_eq!(out[0].symbol.path, "src/caller.rs");

        let db = many_definitions(&["src/caller.rs", "src/store.rs"]);
        call(&db, Some("caller"));
        let out = edges_out(&db, "caller", &[]).unwrap();
        assert_eq!(out.len(), 1);
        assert_eq!(out[0].symbol.path, "src/caller.rs");
    }

    /// With no hint, what the calling file imports breaks the tie.
    #[test]
    fn an_import_resolves_what_the_hint_cannot() {
        let db = many_definitions(&["src/a.rs", "src/b.rs"]);
        let src: i64 = db
            .query_row("SELECT id FROM symbols WHERE name = 'caller'", [], |r| {
                r.get(0)
            })
            .unwrap();
        insert_edge(
            &db,
            src,
            "crate::b::callee",
            "imports",
            "extracted",
            None,
            None,
        )
        .unwrap();
        call(&db, None);
        let out = edges_out(&db, "caller", &["calls".to_string()]).unwrap();
        assert_eq!(out.len(), 1, "{out:?}");
        assert_eq!(out[0].symbol.path, "src/b.rs");
        assert_eq!(out[0].confidence, crate::graph::RESOLVED);
    }

    /// The hint outranks the import list: an import is a set of possibilities,
    /// a hint is a statement about this call.
    #[test]
    fn the_hint_outranks_an_import() {
        let db = many_definitions(&["src/a.rs", "src/b.rs"]);
        let src: i64 = db
            .query_row("SELECT id FROM symbols WHERE name = 'caller'", [], |r| {
                r.get(0)
            })
            .unwrap();
        insert_edge(
            &db,
            src,
            "crate::b::callee",
            "imports",
            "extracted",
            None,
            None,
        )
        .unwrap();
        call(&db, Some("a"));
        let out = edges_out(&db, "caller", &["calls".to_string()]).unwrap();
        assert_eq!(out.len(), 1, "{out:?}");
        assert_eq!(out[0].symbol.path, "src/a.rs");
    }

    /// An edge the syntax tree settled is never downgraded by a ranking.
    #[test]
    fn an_extracted_edge_stays_extracted() {
        let db = many_definitions(&["src/store.rs"]);
        let src: i64 = db
            .query_row("SELECT id FROM symbols WHERE name = 'caller'", [], |r| {
                r.get(0)
            })
            .unwrap();
        insert_edge(
            &db,
            src,
            "callee",
            "calls",
            "extracted",
            Some("store"),
            None,
        )
        .unwrap();
        let out = edges_out(&db, "caller", &[]).unwrap();
        assert_eq!(out[0].confidence, crate::graph::EXTRACTED);
    }

    /// A store written by 0.14.0 has a NULL hint on every row. It must answer
    /// exactly as it did then: one definition is the answer, several are a
    /// question.
    #[test]
    fn a_null_hint_behaves_as_it_did_before_hints_existed() {
        let db = many_definitions(&["src/a.rs"]);
        call(&db, None);
        let one = edges_out(&db, "caller", &[]).unwrap();
        assert_eq!(one.len(), 1);
        assert_eq!(one[0].symbol.path, "src/a.rs");

        let db = many_definitions(&["src/a.rs", "src/b.rs"]);
        call(&db, None);
        let two = edges_out(&db, "caller", &[]).unwrap();
        assert_eq!(
            two.len(),
            2,
            "both candidates are still reachable, now labelled rather than asserted"
        );
    }

    /// Callers are unchanged: the row came from the edge's own `src` id, so
    /// there is nothing to resolve and the stored confidence stands.
    #[test]
    fn callers_keep_the_stored_confidence() {
        let db = many_definitions(&["src/a.rs", "src/b.rs"]);
        call(&db, None);
        let inbound = edges_in(&db, "callee", &[]).unwrap();
        assert_eq!(inbound.len(), 1);
        assert_eq!(inbound[0].confidence, crate::graph::INFERRED);
        assert_eq!(inbound[0].definitions, 1);
    }

    /// Both directions resolve, and the edge kind filter applies to each.
    #[test]
    fn edges_resolve_in_both_directions_and_filter_by_kind() {
        let db = Connection::open_in_memory().unwrap();
        let caller = one_symbol(&db, "a.rs", "caller");
        let b = insert_file(&db, "b.rs", "h", 1, 0).unwrap();
        insert_symbol(&db, b, None, &sym("callee")).unwrap();
        insert_edge(&db, caller, "callee", "calls", "extracted", None, None).unwrap();
        insert_edge(&db, caller, "callee", "references", "inferred", None, None).unwrap();

        assert_eq!(edges_out(&db, "caller", &[]).unwrap().len(), 2);
        let calls = edges_out(&db, "caller", &["calls".to_string()]).unwrap();
        assert_eq!(calls.len(), 1);
        assert_eq!(calls[0].confidence, "extracted");

        let inbound = edges_in(&db, "callee", &["calls".to_string()]).unwrap();
        assert_eq!(inbound.len(), 1);
        assert_eq!(
            inbound[0].symbol.name, "caller",
            "edges_in reports the source"
        );
    }

    /// An edge to something the corpus does not contain is still recorded, and
    /// simply resolves to nothing. A call into the standard library is the
    /// common case, and dropping it would lose the fact that the call is there.
    #[test]
    fn an_edge_to_an_unindexed_target_resolves_to_nothing_without_erroring() {
        let db = Connection::open_in_memory().unwrap();
        let caller = one_symbol(&db, "a.rs", "caller");
        insert_edge(&db, caller, "println", "calls", "inferred", None, None).unwrap();
        assert!(edges_out(&db, "caller", &[]).unwrap().is_empty());
        assert_eq!(graph_stats(&db).unwrap().1, 1, "but the edge row is there");
    }

    /// Every store written before 0.6.0 lacks the key entirely. Reading that as
    /// anything but format 1 would refuse the whole installed base.
    #[test]
    fn a_store_without_the_key_is_the_format_we_understand() {
        let db = Connection::open_in_memory().unwrap();
        db.execute_batch(SCHEMA).unwrap();
        assert!(get_meta(&db, FORMAT_KEY).unwrap().is_none());
        check_format(&db).expect("a pre-0.6.0 store is readable");
        // And still has no key: opening a store must not rewrite it.
        assert!(get_meta(&db, FORMAT_KEY).unwrap().is_none());
    }

    /// The point of writing the format down: a later semlith's store is refused
    /// with both numbers, instead of read as though its layout were this one.
    #[test]
    fn a_newer_format_is_refused_naming_both_numbers() {
        let db = Connection::open_in_memory().unwrap();
        db.execute_batch(SCHEMA).unwrap();
        let newer = FORMAT_VERSION + 1;
        set_meta(&db, FORMAT_KEY, &newer.to_string()).unwrap();

        let refused = check_format(&db).expect_err("a newer format must not be read");
        let said = refused.to_string();
        assert!(said.contains(&newer.to_string()), "{said}");
        assert!(
            said.contains(&FORMAT_VERSION.to_string()),
            "the error must name what this binary understands: {said}"
        );

        // The format this binary wrote is of course readable.
        set_meta(&db, FORMAT_KEY, &FORMAT_VERSION.to_string()).unwrap();
        check_format(&db).unwrap();
    }

    #[test]
    fn only_a_format_4_store_moves_to_format_5_before_a_write() {
        let db = Connection::open_in_memory().unwrap();
        db.execute_batch(SCHEMA).unwrap();
        // No key: format 1, a single index file; stamping would read it as
        // sharded.
        mark_vector_format(&db).unwrap();
        assert!(get_meta(&db, FORMAT_KEY).unwrap().is_none());
        // Format 3 re-chunks on its first full pass, which stamps it.
        set_meta(&db, FORMAT_KEY, "3").unwrap();
        mark_vector_format(&db).unwrap();
        assert_eq!(format(&db).unwrap(), 3);
        set_meta(&db, FORMAT_KEY, &CODE_CONTEXT.to_string()).unwrap();
        mark_vector_format(&db).unwrap();
        assert_eq!(format(&db).unwrap(), TURBOVEC_V8);
        assert_eq!(TURBOVEC_V8, FORMAT_VERSION);
    }

    fn automerge_row(db: &Connection) -> Option<i64> {
        db.query_row(
            "SELECT v FROM chunks_fts_config WHERE k = 'automerge'",
            [],
            |r| r.get(0),
        )
        .optional()
        .unwrap()
    }

    #[test]
    fn a_held_automerge_is_put_back_as_it_was() {
        let db = Connection::open_in_memory().unwrap();
        db.set_db_config(rusqlite::config::DbConfig::SQLITE_DBCONFIG_DEFENSIVE, true)
            .unwrap();
        db.execute_batch(SCHEMA).unwrap();

        // No row before: none after, and nothing left in meta.
        assert_eq!(automerge_row(&db), None);
        hold_automerge(&db).unwrap();
        assert_eq!(automerge_row(&db), Some(0));
        // Held twice (a slice after a crash) still remembers the first value.
        hold_automerge(&db).unwrap();
        release_automerge(&db).unwrap();
        assert_eq!(automerge_row(&db), None);
        assert!(get_meta(&db, AUTOMERGE_HELD_KEY).unwrap().is_none());
        // Released with nothing held is nothing.
        release_automerge(&db).unwrap();
        assert_eq!(automerge_row(&db), None);

        // A store's own value comes back.
        db.execute(
            "INSERT INTO chunks_fts(chunks_fts, rank) VALUES('automerge', 8)",
            [],
        )
        .unwrap();
        hold_automerge(&db).unwrap();
        assert_eq!(automerge_row(&db), Some(0));
        release_automerge(&db).unwrap();
        assert_eq!(automerge_row(&db), Some(8));
    }

    #[test]
    fn a_first_import_lasts_until_a_pass_ends_it() {
        let db = Connection::open_in_memory().unwrap();
        db.execute_batch(SCHEMA).unwrap();
        assert!(!first_import(&db, false).unwrap(), "a store with rows");
        assert!(
            first_import(&db, true).unwrap(),
            "an empty store starts one"
        );
        assert!(first_import(&db, false).unwrap(), "the next slice is in it");
        end_first_import(&db).unwrap();
        assert!(!first_import(&db, false).unwrap());
    }

    fn seeded() -> Connection {
        let db = Connection::open_in_memory().unwrap();
        db.pragma_update(None, "foreign_keys", "ON").unwrap();
        db.execute_batch(SCHEMA).unwrap();
        let f = insert_file(&db, "/a/lib.rs", "h", 10, 0).unwrap();
        insert_chunk(&db, f, 0, 1, 5, "const EMBED_BATCH: usize = 32;").unwrap();
        insert_chunk(
            &db,
            f,
            1,
            6,
            9,
            "a paragraph about retry backoff and jitter",
        )
        .unwrap();
        db
    }

    /// A corpus with the same word in four files: two under `src`, two not,
    /// two Rust, two Markdown. Enough to tell a union from an intersection.
    /// A store holds the paths its own platform produces, and `filter::anchor`
    /// builds patterns to match those — backslashes on Windows, forward
    /// slashes everywhere else. A fixture written with one separator therefore
    /// tests nothing on the other platform, so these are spelled in the
    /// separator the code under test is going to use.
    fn native(path: &str) -> String {
        path.replace('/', std::path::MAIN_SEPARATOR_STR)
    }

    fn mixed() -> Connection {
        let db = Connection::open_in_memory().unwrap();
        db.pragma_update(None, "foreign_keys", "ON").unwrap();
        db.execute_batch(SCHEMA).unwrap();
        db.execute_batch(GLOB_INDEX).unwrap();
        for path in [
            "/proj/src/lib.rs",
            "/proj/src/notes.md",
            "/proj/vendor/other.rs",
            "/proj/README.MD",
        ] {
            let f = insert_file(&db, &native(path), "h", 10, 0).unwrap();
            insert_chunk(&db, f, 0, 1, 5, "retry backoff and jitter").unwrap();
        }
        db
    }

    /// `groups()` output for a filter, so the tests exercise the same patterns
    /// the CLI produces rather than hand-written globs that could drift from it.
    fn filter(paths: &[&str], exts: &[&str], langs: &[&str]) -> Vec<Vec<String>> {
        let own = |v: &[&str]| v.iter().map(|s| s.to_string()).collect::<Vec<_>>();
        crate::filter::Filter::new(&own(paths), &own(exts), &own(langs))
            .unwrap()
            .groups()
            .to_vec()
    }

    fn paths_of(db: &Connection, ids: &[u64]) -> Vec<String> {
        let mut out: Vec<String> = ids
            .iter()
            .map(|&i| chunk(db, i).unwrap().unwrap().path)
            .collect();
        out.sort();
        out
    }

    #[test]
    fn keyword_search_finds_an_exact_identifier() {
        let db = seeded();
        let hits = keyword_search(&db, "EMBED_BATCH", 10, &[]).unwrap();
        assert_eq!(hits, vec![1], "expected the defining chunk, got {hits:?}");
    }

    #[test]
    fn a_path_glob_selects_only_what_is_under_it() {
        let db = mixed();
        let ids = filtered_chunk_ids(&db, &filter(&["src/**"], &[], &[])).unwrap();
        assert_eq!(
            paths_of(&db, &ids),
            [native("/proj/src/lib.rs"), native("/proj/src/notes.md")]
        );
    }

    /// A scope to one root is a range of the path index, not a GLOB over every
    /// path (#183): the plan searches `files_glob_path` and scans no table.
    #[test]
    fn a_root_scope_is_a_range_of_the_path_index() {
        let db = mixed();
        let groups = filter(&["/proj/src/**"], &[], &[]);
        let (predicate, binds) = glob_predicate(&groups);
        let sql = format!("EXPLAIN QUERY PLAN SELECT f.id FROM files f WHERE {predicate}");
        let plan: Vec<String> = db
            .prepare(&sql)
            .unwrap()
            .query_map(rusqlite::params_from_iter(binds.iter()), |r| r.get(3))
            .unwrap()
            .map(Result::unwrap)
            .collect();
        assert!(
            plan.iter()
                .any(|p| p.contains("INDEX files_glob_path (<expr>>? AND <expr><?)")),
            "the scope did not use the path index: {plan:?}"
        );
        assert!(!plan.iter().any(|p| p.starts_with("SCAN")), "{plan:?}");
        // And the range changes no answer.
        let ids = filtered_chunk_ids(&db, &groups).unwrap();
        assert_eq!(
            paths_of(&db, &ids),
            [native("/proj/src/lib.rs"), native("/proj/src/notes.md")]
        );
    }

    #[test]
    fn only_a_literal_directory_in_front_is_a_range() {
        assert_eq!(
            literal_dir("/proj/src/**"),
            Some(("/proj/src/".into(), "/proj/src0".into()))
        );
        assert_eq!(
            literal_dir("/proj/src/*.rs"),
            Some(("/proj/src/".into(), "/proj/src0".into()))
        );
        assert_eq!(literal_dir("*/src/**"), None, "a leading wildcard");
        assert_eq!(literal_dir("/*"), None, "the whole disk is no range");
        assert_eq!(
            literal_dir("/proj/readme.md"),
            None,
            "no wildcard, no GLOB to narrow"
        );
    }

    /// An absolute pattern means exactly itself: it must not pick up the
    /// `*/` anchoring a relative one gets.
    #[test]
    fn an_absolute_glob_matches_only_what_it_literally_covers() {
        let db = mixed();
        let ids = filtered_chunk_ids(&db, &filter(&["/proj/vendor/*"], &[], &[])).unwrap();
        assert_eq!(paths_of(&db, &ids), [native("/proj/vendor/other.rs")]);
    }

    #[test]
    fn extensions_union_and_a_language_resolves_to_the_same_set() {
        let db = mixed();
        let by_ext = filtered_chunk_ids(&db, &filter(&[], &["rs"], &[])).unwrap();
        let by_lang = filtered_chunk_ids(&db, &filter(&[], &[], &["rust"])).unwrap();
        assert_eq!(paths_of(&db, &by_ext), paths_of(&db, &by_lang));
        assert_eq!(
            paths_of(&db, &by_ext),
            [native("/proj/src/lib.rs"), native("/proj/vendor/other.rs")]
        );

        let both = filtered_chunk_ids(&db, &filter(&[], &["rs", "md"], &[])).unwrap();
        assert_eq!(both.len(), 4, "repeating --ext must union, not intersect");
    }

    /// `README.MD` is Markdown. Extension matching folds case, so a filter
    /// written the way anyone would write it does not miss it.
    #[test]
    fn extension_matching_is_case_insensitive() {
        let db = mixed();
        let ids = filtered_chunk_ids(&db, &filter(&[], &["md"], &[])).unwrap();
        assert_eq!(
            paths_of(&db, &ids),
            [native("/proj/README.MD"), native("/proj/src/notes.md")]
        );
    }

    #[test]
    fn a_path_and_an_extension_intersect() {
        let db = mixed();
        let ids = filtered_chunk_ids(&db, &filter(&["src/**"], &["md"], &[])).unwrap();
        assert_eq!(
            paths_of(&db, &ids),
            [native("/proj/src/notes.md")],
            "the README is Markdown but is not under src"
        );
    }

    /// The whole point of the release: the keyword half must see the same
    /// subset the vector half is given, or fusion ranks a chunk one side was
    /// never allowed to return.
    #[test]
    fn the_keyword_half_honours_the_same_filter() {
        let db = mixed();
        let unfiltered = keyword_search(&db, "retry backoff", 10, &[]).unwrap();
        assert_eq!(unfiltered.len(), 4);

        let scoped =
            keyword_search(&db, "retry backoff", 10, &filter(&["src/**"], &[], &[])).unwrap();
        assert_eq!(
            paths_of(&db, &scoped),
            [native("/proj/src/lib.rs"), native("/proj/src/notes.md")]
        );
    }

    #[test]
    fn a_filter_matching_nothing_selects_nothing() {
        let db = mixed();
        let groups = filter(&["nowhere/**"], &[], &[]);
        assert_eq!(matching_files(&db, &groups).unwrap(), 0);
        assert!(filtered_chunk_ids(&db, &groups).unwrap().is_empty());
        assert!(
            keyword_search(&db, "retry", 10, &groups)
                .unwrap()
                .is_empty()
        );
    }

    /// An unfiltered search must cost exactly what it cost in 0.2.0. The
    /// statement is the whole of that promise: adding the join unconditionally
    /// would tax every query that uses no filter, and would not fail any other
    /// test in this file.
    #[test]
    fn an_unfiltered_search_issues_the_0_2_0_statement_verbatim() {
        assert_eq!(
            UNFILTERED_SQL,
            "SELECT rowid FROM chunks_fts WHERE chunks_fts MATCH ? ORDER BY rank LIMIT ?"
        );
        // And an empty filter must take that branch, not the general one.
        assert_eq!(glob_predicate(&[]), ("1".to_string(), vec![]));
    }

    #[test]
    fn no_filter_counts_every_file() {
        let db = mixed();
        assert_eq!(matching_files(&db, &[]).unwrap(), 4);
    }

    /// An exclusion on its own means everything except, so it needs no
    /// inclusion beside it to be a filter.
    #[test]
    fn an_exclusion_alone_means_everything_except() {
        let db = mixed();
        let f = filter(&["!**/vendor/**"], &[], &[]);
        assert_eq!(matching_files(&db, &f).unwrap(), 3);
    }

    /// The exclusion applies after the inclusions of its own kind, so the two
    /// intersect rather than the later one replacing the earlier.
    #[test]
    fn an_exclusion_narrows_the_inclusions_of_its_kind() {
        let db = mixed();
        let f = filter(&["proj/**", "!**/vendor/**"], &[], &[]);
        assert_eq!(matching_files(&db, &f).unwrap(), 3);

        let f = filter(&["proj/src/**", "!**/notes.md"], &[], &[]);
        assert_eq!(matching_files(&db, &f).unwrap(), 1);
    }

    /// The bind values are positional, so an exclusion written before an
    /// inclusion must still bind to the `?` the clause puts it against.
    #[test]
    fn an_exclusion_written_first_still_binds_to_its_own_placeholder() {
        let db = mixed();
        let written_first = filter(&["!**/vendor/**", "proj/**"], &[], &[]);
        let written_last = filter(&["proj/**", "!**/vendor/**"], &[], &[]);
        assert_eq!(
            matching_files(&db, &written_first).unwrap(),
            matching_files(&db, &written_last).unwrap(),
        );
        assert_eq!(matching_files(&db, &written_first).unwrap(), 3);
    }

    /// Extensions negate the way paths do, and case-insensitively: a store
    /// holding `README.MD` loses it to `--ext '!md'`.
    #[test]
    fn an_excluded_extension_removes_every_spelling_of_it() {
        let db = mixed();
        let f = filter(&[], &["!md"], &[]);
        assert_eq!(matching_files(&db, &f).unwrap(), 2);
    }

    /// A group holding only exclusions still selects from everything, and the
    /// two kinds still intersect across groups.
    #[test]
    fn exclusions_of_different_kinds_intersect() {
        let db = mixed();
        let f = filter(&["!**/vendor/**"], &["!md"], &[]);
        assert_eq!(matching_files(&db, &f).unwrap(), 1);
    }

    #[test]
    fn fts_syntax_in_a_query_is_not_executed_as_syntax() {
        let db = seeded();
        // Each of these is either an FTS5 syntax error or an FTS5 operator if
        // passed through raw. All must come back as ordinary searches.
        for query in [
            "EMBED_BATCH AND retry",
            "unbalanced \" quote",
            "call_me(",
            "prefix*",
            "NEAR(a b)",
            "-negated",
        ] {
            let result = keyword_search(&db, query, 10, &[]);
            assert!(
                result.is_ok(),
                "query {query:?} errored: {:?}",
                result.err()
            );
        }
    }

    #[test]
    fn a_query_with_no_terms_matches_nothing() {
        let db = seeded();
        assert!(keyword_search(&db, "!!! ???", 10, &[]).unwrap().is_empty());
    }

    #[test]
    fn deleting_a_file_removes_its_chunks_from_the_keyword_index() {
        let db = seeded();
        assert!(
            !keyword_search(&db, "EMBED_BATCH", 10, &[])
                .unwrap()
                .is_empty()
        );
        delete_file(&db, "/a/lib.rs", 0).unwrap();
        assert!(
            keyword_search(&db, "EMBED_BATCH", 10, &[])
                .unwrap()
                .is_empty(),
            "cascade delete left the keyword index stale"
        );
    }

    /// A store written by 0.15.0 has no `edges.line`. Opening it must add the
    /// column without touching the rows, and every edge already in it must
    /// report no call site rather than a wrong one.
    #[test]
    fn a_pre_0_16_store_opens_and_its_edges_report_no_call_site() {
        let db = Connection::open_in_memory().unwrap();
        db.execute_batch(
            "CREATE TABLE files (id INTEGER PRIMARY KEY, path TEXT NOT NULL UNIQUE,
                 hash TEXT NOT NULL, bytes INTEGER NOT NULL, indexed_at INTEGER NOT NULL);
             CREATE TABLE symbols (id INTEGER PRIMARY KEY AUTOINCREMENT,
                 file_id INTEGER NOT NULL REFERENCES files(id) ON DELETE CASCADE,
                 chunk_id INTEGER, kind TEXT NOT NULL, name TEXT NOT NULL,
                 qualified TEXT NOT NULL, start_line INTEGER NOT NULL, end_line INTEGER NOT NULL);
             CREATE TABLE edges (src INTEGER NOT NULL REFERENCES symbols(id) ON DELETE CASCADE,
                 dst TEXT NOT NULL, kind TEXT NOT NULL, confidence TEXT NOT NULL, hint TEXT);
             INSERT INTO files VALUES (1, '/a/lib.rs', 'h', 10, 0);
             INSERT INTO symbols VALUES (1, 1, NULL, 'function', 'caller', 'm::caller', 1, 9);
             INSERT INTO symbols VALUES (2, 1, NULL, 'function', 'callee', 'm::callee', 20, 24);
             CREATE TABLE retrievals (id INTEGER PRIMARY KEY AUTOINCREMENT, at INTEGER NOT NULL,
                 client TEXT NOT NULL, query TEXT NOT NULL, hits INTEGER NOT NULL,
                 micros INTEGER NOT NULL, excerpt_tokens INTEGER NOT NULL,
                 whole_file_tokens INTEGER NOT NULL);
             INSERT INTO edges VALUES (1, 'callee', 'calls', 'inferred', NULL);",
        )
        .unwrap();

        add_columns(&db).unwrap();

        let out = edges_out(&db, "caller", &[]).unwrap();
        assert_eq!(out.len(), 1, "{out:?}");
        assert_eq!(
            out[0].line, None,
            "a 0.15.0 row must not invent a call site"
        );
        assert_eq!(
            crate::graph::call_site(&out[0], &crate::graph::verbatim),
            "",
            "nothing is rendered for an edge with no line"
        );

        let into = edges_in(&db, "callee", &[]).unwrap();
        assert_eq!(into.len(), 1, "{into:?}");
        assert_eq!(into[0].line, None);
    }

    /// An older binary opening the store puts the insert trigger back; the
    /// writer drops it before inserting, so a chunk is indexed once (#198).
    #[test]
    fn an_insert_trigger_an_older_binary_restored_does_not_index_twice() {
        let db = Connection::open_in_memory().unwrap();
        db.execute_batch(SCHEMA).unwrap();
        add_columns(&db).unwrap();
        db.execute_batch(
            "CREATE TRIGGER IF NOT EXISTS chunks_fts_insert AFTER INSERT ON chunks BEGIN
                 INSERT INTO chunks_fts(rowid, text) VALUES (new.id, new.text);
             END;",
        )
        .unwrap();
        db.execute_batch("BEGIN").unwrap();
        drop_insert_trigger(&db).unwrap();
        let f = insert_file(&db, "/a/lib.rs", "h", 10, 0).unwrap();
        let id = insert_chunk(&db, f, 0, 1, 5, "retry backoff and jitter").unwrap();
        db.execute_batch("COMMIT").unwrap();
        assert_eq!(
            keyword_search(&db, "jitter", 10, &[]).unwrap(),
            vec![id as u64]
        );
        // Indexed once: an integrity check of external content fails on a
        // doubled row.
        db.execute_batch("INSERT INTO chunks_fts(chunks_fts, rank) VALUES('integrity-check', 1);")
            .unwrap();
        // And a delete still takes it out of the index.
        db.execute("DELETE FROM files WHERE id = ?1", [f]).unwrap();
        assert!(keyword_search(&db, "jitter", 10, &[]).unwrap().is_empty());
    }

    #[test]
    fn an_old_store_gets_its_keyword_index_backfilled() {
        // A 0.1.0 store: chunks written with no FTS table in the schema.
        let db = Connection::open_in_memory().unwrap();
        db.execute_batch(
            "CREATE TABLE meta (k TEXT PRIMARY KEY, v TEXT NOT NULL);
             CREATE TABLE files (id INTEGER PRIMARY KEY, path TEXT NOT NULL UNIQUE,
                 hash TEXT NOT NULL, bytes INTEGER NOT NULL, indexed_at INTEGER NOT NULL);
             CREATE TABLE chunks (id INTEGER PRIMARY KEY, file_id INTEGER NOT NULL
                 REFERENCES files(id) ON DELETE CASCADE, ord INTEGER NOT NULL,
                 start_line INTEGER NOT NULL, end_line INTEGER NOT NULL, text TEXT NOT NULL);
             INSERT INTO files VALUES (1, '/a/lib.rs', 'h', 10, 0);
             INSERT INTO chunks VALUES (1, 1, 0, 1, 5, 'const EMBED_BATCH: usize = 32;');",
        )
        .unwrap();

        // Opening it with the current schema must index what is already there.
        db.execute_batch(SCHEMA).unwrap();
        backfill_fts(&db).unwrap();
        assert_eq!(
            keyword_search(&db, "EMBED_BATCH", 10, &[]).unwrap(),
            vec![1]
        );
    }

    #[test]
    fn a_key_searched_for_is_recorded_masked_and_the_chain_still_verifies() {
        let db = Connection::open_in_memory().unwrap();
        db.execute_batch(SCHEMA).unwrap();
        add_columns(&db).unwrap();
        let key = crate::keyscan::forge(0);
        let query = format!("where is {key} used");
        for q in [query.as_str(), "an ordinary question"] {
            record_retrieval(
                &db,
                &NewRetrieval {
                    client_version: "",
                    client: "cli",
                    session: "s",
                    tool: "search",
                    query: q,
                    hits: 1,
                    micros: 1,
                    excerpt_tokens: 1,
                    whole_file_tokens: 1,
                    stale_hits: 0,
                    tokenizer: "estimate",
                    query_id: "q",
                },
            )
            .unwrap();
        }
        let rows = retrievals(&db, 10).unwrap();
        assert!(
            rows.iter().all(|r| !r.query.contains(&key)),
            "the key was recorded in full"
        );
        assert!(
            rows.iter()
                .any(|r| r.query.starts_with("where is [REDACTED:"))
        );
        assert!(rows.iter().any(|r| r.query == "an ordinary question"));
        assert_eq!(ledger_break(&db).unwrap(), None);
    }

    /// Two connections to one store recording at once leave one chain.
    ///
    /// The daemon and a CLI search are two such writers. Before 0.33.0 each read
    /// the last hash and then inserted, so a writer that read while the other
    /// held an uncommitted row chained to the row before it, and the ledger
    /// reported the second row as edited though nothing had been.
    #[test]
    fn two_connections_recording_at_once_leave_a_chain_that_verifies() {
        fn row(query: &str) -> NewRetrieval<'_> {
            NewRetrieval {
                client_version: "",
                client: "cli",
                session: "s",
                tool: "search",
                query,
                hits: 1,
                micros: 1,
                excerpt_tokens: 1,
                whole_file_tokens: 1,
                stale_hits: 0,
                tokenizer: "chars4",
                query_id: "",
            }
        }
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("store.db");
        let first = open(&path).unwrap();
        let second = open(&path).unwrap();

        // `first` is mid-write: its row is in, not yet committed.
        read_only(&first, false).unwrap();
        first.execute_batch("BEGIN IMMEDIATE").unwrap();
        record_retrieval(&first, &row("first")).unwrap();

        // `second` records meanwhile, and has to end up chained to that row.
        let waiting = std::thread::spawn(move || record_retrieval(&second, &row("second")));
        // Long enough for the unguarded read to have happened; the guarded one
        // is waiting on the lock and is unaffected by how long this is.
        std::thread::sleep(std::time::Duration::from_millis(300));
        read_only(&first, false).unwrap();
        first.execute_batch("COMMIT").unwrap();
        waiting.join().unwrap().unwrap();

        assert_eq!(retrievals(&first, 10).unwrap().len(), 2);
        assert_eq!(
            ledger_break(&first).unwrap(),
            None,
            "two writers forked the chain"
        );
    }
}