sessionwiki 0.34.9

use crate::adapters::{self, Adapter};
use crate::model::Role;
use anyhow::{Context, Result};
use rusqlite::{params, Connection, OptionalExtension};
use serde::Serialize;
use std::collections::HashMap;
use std::io::{IsTerminal, Write};
use std::path::PathBuf;

/// Where a legacy index would sit, given where this one is going.
///
/// Beside the destination, not under `dirs::data_dir()`. The migration used to
/// ask the ambient data dir wherever the index was actually headed, and then
/// RENAME what it found into that destination - so a run with
/// `SESSIONWIKI_DATA` pointed at a temp dir would move the real
/// `~/.local/share/sessiondex` into it, and the tags, notes and summaries in
/// there are not rebuildable. Eight test files set that variable.
fn legacy_candidates(dir: &std::path::Path) -> Vec<PathBuf> {
    let Some(parent) = dir.parent() else {
        return Vec::new();
    };
    ["sessiondex", "session-atlas"]
        .iter()
        .map(|n| parent.join(n))
        .collect()
}

/// The index lives outside the session stores and never touches them.
/// Default: ~/.local/share/sessionwiki/index.db (platform equivalent).
pub fn db_path() -> Result<PathBuf> {
    let dir = std::env::var_os("SESSIONWIKI_DATA")
        .map(PathBuf::from)
        .or_else(|| dirs::data_dir().map(|d| d.join("sessionwiki")))
        .context("cannot determine a data directory")?;
    // One-time migration from earlier names, newest first. This carries over
    // the existing index AND the curated tags/notes/summaries, which are not
    // rebuildable. The project was session-atlas, then sessiondex.
    if !dir.exists() {
        {
            for old in legacy_candidates(&dir) {
                if old.exists() {
                    // A failed rename must not pass silently: the user would
                    // get a fresh empty index while their curation sits
                    // stranded in the old directory. Re-check after failure,
                    // though - a concurrent first run (hook + CLI) may have
                    // migrated it already, which is success, not failure.
                    if let Err(e) = std::fs::rename(&old, &dir) {
                        if !dir.exists() && old.exists() {
                            eprintln!(
                                "warning: could not migrate {} -> {} ({e}); \
                                 starting a fresh index. Move it manually to \
                                 keep your tags, notes, and archive.",
                                old.display(),
                                dir.display()
                            );
                        }
                    }
                    break;
                }
            }
        }
    }
    std::fs::create_dir_all(&dir)?;
    Ok(dir.join("index.db"))
}

/// The index path IF it already exists - with none of the directory creation or
/// legacy migration `db_path` performs. For strictly read-only callers (`doctor`)
/// that must not mutate the filesystem just to check for the index.
pub fn existing_db_path() -> Option<PathBuf> {
    let dir = std::env::var_os("SESSIONWIKI_DATA")
        .map(PathBuf::from)
        .or_else(|| dirs::data_dir().map(|d| d.join("sessionwiki")))?;
    let db = dir.join("index.db");
    db.exists().then_some(db)
}

/// `user_version` versions the disposable cache: a mismatch drops and rebuilds
/// the derived tables (files/messages/msgs/touched) instead of migrating. The
/// durable tables (summaries, tags, notes, archive) hold what cannot be
/// re-derived - LLM output, user curation, and sessions whose originals the tool
/// deleted - and are versioned separately by `meta.durable_version` via forward,
/// additive-only migrations that never drop, so they survive every upgrade. The
/// two counters are independent and must never gate each other.
pub const SCHEMA_VERSION: i64 = 9; // 9: compact tool lines + optional full outputs

const DEFAULT_FTS_TOKENIZER: &str = "trigram";
const FTS_TOKENIZER_META_KEY: &str = "fts_tokenizer";
const DEFAULT_TOOL_OUTPUT_MODE: ToolOutputMode = ToolOutputMode::Summary;
const TOOL_OUTPUT_MODE_META_KEY: &str = "tool_output_mode";

/// Whether full tool results are retained in the index and archive.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ToolOutputMode {
    /// Retain bounded, redacted tool output separately from searchable text.
    Full,
    /// Keep compact tool summary lines in the index and archive. Failed calls
    /// include the first output line in their summary.
    Summary,
}

impl ToolOutputMode {
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Full => "full",
            Self::Summary => "summary",
        }
    }
}

impl std::fmt::Display for ToolOutputMode {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.as_str())
    }
}

impl std::str::FromStr for ToolOutputMode {
    type Err = &'static str;

    fn from_str(value: &str) -> std::result::Result<Self, Self::Err> {
        match value {
            "full" => Ok(Self::Full),
            "summary" => Ok(Self::Summary),
            _ => Err("tool output mode must be full or summary"),
        }
    }
}

/// Version of the durable schema this binary ships. The durable CREATE
/// statements are frozen at this shape; every later durable change is a
/// migration in DURABLE_MIGRATIONS. Independent of SCHEMA_VERSION (the cache).
const BASELINE_DURABLE_VERSION: i64 = 1;

/// Read meta.durable_version, seeding the baseline when absent. Absence covers
/// both a fresh DB (durables just created at baseline) and an existing
/// pre-feature index (durables already at baseline) - both correctly start at
/// BASELINE. The common path is read-only; INSERT OR IGNORE handles a concurrent
/// first-open after the initial read.
fn read_or_init_durable_version(conn: &Connection) -> Result<i64> {
    let read = || -> Result<Option<String>> {
        Ok(conn
            .query_row(
                "SELECT value FROM meta WHERE key = 'durable_version'",
                [],
                |r| r.get(0),
            )
            .optional()?)
    };
    if let Some(v) = read()? {
        return Ok(v.parse().unwrap_or(BASELINE_DURABLE_VERSION));
    }
    conn.execute(
        "INSERT OR IGNORE INTO meta(key, value) VALUES ('durable_version', ?1)",
        params![BASELINE_DURABLE_VERSION.to_string()],
    )?;
    let v = read()?.context("durable_version missing after initialization")?;
    Ok(v.parse().unwrap_or(BASELINE_DURABLE_VERSION))
}

/// One migration step: plain DDL, or Rust code for transformations SQL cannot
/// express (e.g. unicode normalization). Both run inside the same gated
/// transaction and must stay additive-and-repair-only - never drop durable data.
enum MigrationStep {
    // The natural shape for most future migrations (ALTER/CREATE); only data
    // repairs need `Fix`. Allowed while no registered migration uses it.
    #[allow(dead_code)]
    Sql(&'static str),
    Fix(fn(&Connection) -> Result<()>),
}

struct Migration {
    version: i64,
    step: MigrationStep,
}

/// Forward-only, additive-only durable migrations applied in order. ALTER ADD
/// COLUMN / CREATE / data repair only - never DROP/RENAME a durable column or
/// table.
const DURABLE_MIGRATIONS: &[Migration] = &[
    // v2: tags written by pre-0.17 binaries were lowercased but not
    // NFC-normalized, so a decomposed-form tag (macOS IME) is unreachable by
    // the normalized lookups. Re-normalize stored rows once.
    Migration {
        version: 2,
        step: MigrationStep::Fix(normalize_stored_tags),
    },
];

/// Durable migration v2: converge every stored tag on `norm_tag` form. An NFC
/// twin that already exists absorbs the row (INSERT OR IGNORE + DELETE).
fn normalize_stored_tags(conn: &Connection) -> Result<()> {
    let rows: Vec<(String, String)> = conn
        .prepare("SELECT session_id, tag FROM tags")?
        .query_map([], |r| Ok((r.get(0)?, r.get(1)?)))?
        .collect::<rusqlite::Result<_>>()?;
    for (sid, tag) in rows {
        let norm = norm_tag(&tag);
        if norm != tag {
            conn.execute(
                "INSERT OR IGNORE INTO tags(session_id, tag) VALUES (?1, ?2)",
                params![sid, norm],
            )?;
            conn.execute(
                "DELETE FROM tags WHERE session_id = ?1 AND tag = ?2",
                params![sid, tag],
            )?;
        }
    }
    Ok(())
}

/// Apply migrations whose version exceeds the stored durable_version, in one
/// IMMEDIATE transaction (re-reading the version inside it so a concurrent
/// process that already migrated makes this a no-op). DDL + version bump are
/// atomic. Version-gating is the only thing that makes re-running safe, since
/// SQLite ALTER ADD COLUMN is not idempotent.
fn run_durable_migrations(conn: &Connection, migrations: &[Migration]) -> Result<()> {
    conn.execute_batch("BEGIN IMMEDIATE")?;
    let outcome = (|| -> Result<()> {
        let current: i64 = conn
            .query_row(
                "SELECT value FROM meta WHERE key = 'durable_version'",
                [],
                |r| r.get::<_, String>(0),
            )?
            .parse()
            .unwrap_or(BASELINE_DURABLE_VERSION);
        for m in migrations.iter().filter(|m| m.version > current) {
            match m.step {
                MigrationStep::Sql(sql) => conn.execute_batch(sql)?,
                MigrationStep::Fix(f) => f(conn)?,
            }
            conn.execute(
                "UPDATE meta SET value = ?1 WHERE key = 'durable_version'",
                params![m.version.to_string()],
            )?;
        }
        Ok(())
    })();
    match outcome {
        Ok(()) => conn.execute_batch("COMMIT")?,
        Err(e) => {
            let _ = conn.execute_batch("ROLLBACK");
            return Err(e);
        }
    }
    Ok(())
}

/// Ensure durable settings exist before the cache schema reads them.
fn create_meta_schema(conn: &Connection) -> Result<()> {
    let exists: bool = conn.query_row(
        "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'meta')",
        [],
        |row| row.get(0),
    )?;
    if !exists {
        // Durable key/value settings and versioning survive cache rebuilds.
        conn.execute_batch(
            "CREATE TABLE IF NOT EXISTS meta(
                key   TEXT PRIMARY KEY,
                value TEXT NOT NULL
            );",
        )?;
    }
    Ok(())
}

/// The chosen FTS tokenizer is durable configuration. Indexes predating this
/// setting use the historical trigram tokenizer until a user changes it.
pub fn tokenizer_spec(conn: &Connection) -> Result<String> {
    Ok(conn
        .query_row(
            "SELECT value FROM meta WHERE key = ?1",
            [FTS_TOKENIZER_META_KEY],
            |row| row.get(0),
        )
        .optional()?
        .unwrap_or_else(|| DEFAULT_FTS_TOKENIZER.to_string()))
}

fn sql_string_literal(value: &str) -> String {
    value.replace('\'', "''")
}

fn validate_tokenizer_spec(conn: &Connection, spec: &str) -> Result<()> {
    let sql = format!(
        "CREATE VIRTUAL TABLE temp.sessionwiki_tokenizer_validation USING fts5(\
             text, tokenize='{}');\
         DROP TABLE temp.sessionwiki_tokenizer_validation;",
        sql_string_literal(spec)
    );
    conn.execute_batch(&sql)
        .map_err(anyhow::Error::from)
        .context("invalid FTS5 tokenizer specification")
}

fn create_msgs_table(conn: &Connection, spec: &str) -> Result<()> {
    let sql = format!(
        "CREATE VIRTUAL TABLE IF NOT EXISTS msgs USING fts5(\
             text, content='messages', content_rowid='id', tokenize='{}');",
        sql_string_literal(spec)
    );
    conn.execute_batch(&sql)?;
    Ok(())
}

/// Store a tokenizer choice and rebuild only the FTS index. Session files are
/// not parsed; the external-content table rebuilds from `messages` in SQLite.
/// Returns whether the setting changed.
pub fn set_tokenizer_spec(conn: &Connection, spec: &str) -> Result<bool> {
    validate_tokenizer_spec(conn, spec)?;
    let current = tokenizer_spec(conn)?;
    if current == spec {
        return Ok(false);
    }

    conn.execute_batch("BEGIN IMMEDIATE")?;
    let outcome = (|| -> Result<bool> {
        let current = tokenizer_spec(conn)?;
        if current == spec {
            return Ok(false);
        }
        conn.execute_batch("DROP TABLE msgs;")?;
        create_msgs_table(conn, spec)?;
        conn.execute_batch("INSERT INTO msgs(msgs) VALUES('rebuild');")?;
        conn.execute(
            "INSERT INTO meta(key, value) VALUES (?1, ?2)
             ON CONFLICT(key) DO UPDATE SET value = excluded.value",
            params![FTS_TOKENIZER_META_KEY, spec],
        )?;
        Ok(true)
    })();
    match outcome {
        Ok(changed) => {
            conn.execute_batch("COMMIT")?;
            Ok(changed)
        }
        Err(error) => {
            let _ = conn.execute_batch("ROLLBACK");
            Err(error)
        }
    }
}

/// Read whether full tool results are retained. Indexes without an explicit
/// choice use `summary`, keeping compact tool lines in the archive.
pub fn tool_output_mode(conn: &Connection) -> Result<ToolOutputMode> {
    let value = conn
        .query_row(
            "SELECT value FROM meta WHERE key = ?1",
            [TOOL_OUTPUT_MODE_META_KEY],
            |row| row.get::<_, String>(0),
        )
        .optional()?;
    match value {
        None => Ok(DEFAULT_TOOL_OUTPUT_MODE),
        Some(value) => value.parse().map_err(anyhow::Error::msg),
    }
}

/// Change tool-output retention for this index. Switching to `summary` clears
/// full output from live sessions only; archived sessions keep the contents
/// already stored in both `messages` and `archive`. Switching back to `full`
/// marks live sessions stale so the next sync re-parses their source files.
pub fn set_tool_output_mode(conn: &Connection, mode: ToolOutputMode) -> Result<bool> {
    if tool_output_mode(conn)? == mode {
        return Ok(false);
    }

    conn.execute_batch("BEGIN IMMEDIATE")?;
    let outcome = (|| -> Result<bool> {
        if tool_output_mode(conn)? == mode {
            return Ok(false);
        }
        match mode {
            ToolOutputMode::Full => {
                // Negative change tokens cannot match filesystem metadata or
                // shared-store tokens, and sync treats them as forced work
                // even for a bounded recent-session refresh.
                conn.execute(
                    "UPDATE files SET mtime = -1, size = -1 WHERE archived_at IS NULL",
                    [],
                )?;
            }
            ToolOutputMode::Summary => {
                // Archived sessions have no source file to recover output from,
                // so preserve their durable copy in both messages and archive.
                conn.execute(
                    "UPDATE messages SET full_text = NULL
                     WHERE session_id IN (
                         SELECT session_id FROM files WHERE archived_at IS NULL
                     )",
                    [],
                )?;
            }
        }
        conn.execute(
            "INSERT INTO meta(key, value) VALUES (?1, ?2)
             ON CONFLICT(key) DO UPDATE SET value = excluded.value",
            params![TOOL_OUTPUT_MODE_META_KEY, mode.as_str()],
        )?;
        Ok(true)
    })();
    match outcome {
        Ok(changed) => {
            conn.execute_batch("COMMIT")?;
            Ok(changed)
        }
        Err(error) => {
            let _ = conn.execute_batch("ROLLBACK");
            Err(error)
        }
    }
}

struct ArchiveMessage {
    role: String,
    text: String,
    full_text: Option<String>,
}

/// Read both the historical `[role, text]` archive rows and the newer
/// `[role, text, full_text]` rows. Legacy sessions naturally have no output.
fn decode_archive_transcript(transcript: &str) -> Result<Vec<ArchiveMessage>> {
    let rows: Vec<serde_json::Value> = serde_json::from_str(transcript)?;
    rows.into_iter()
        .map(|row| {
            let fields = row
                .as_array()
                .context("archive transcript message must be an array")?;
            anyhow::ensure!(
                fields.len() == 2 || fields.len() == 3,
                "archive transcript message must have two or three fields"
            );
            let role = fields[0]
                .as_str()
                .context("archive transcript role must be a string")?
                .to_owned();
            let text = fields[1]
                .as_str()
                .context("archive transcript text must be a string")?
                .to_owned();
            let full_text = match fields.get(2) {
                None | Some(serde_json::Value::Null) => None,
                Some(serde_json::Value::String(text)) => Some(text.clone()),
                Some(_) => anyhow::bail!("archive full tool output must be a string or null"),
            };
            Ok(ArchiveMessage {
                role,
                text,
                full_text,
            })
        })
        .collect()
}

#[cfg(test)]
fn create_cache_schema(conn: &Connection) -> Result<()> {
    create_meta_schema(conn)?;
    let spec = tokenizer_spec(conn)?;
    create_cache_schema_with_tokenizer(conn, &spec)
}

fn create_cache_schema_with_tokenizer(conn: &Connection, tokenizer: &str) -> Result<()> {
    let msgs_exists: bool = conn.query_row(
        "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'msgs')",
        [],
        |row| row.get(0),
    )?;
    conn.execute_batch(
        "CREATE TABLE IF NOT EXISTS files(
            path       TEXT PRIMARY KEY,
            mtime      INTEGER NOT NULL,
            size       INTEGER NOT NULL,
            session_id TEXT NOT NULL,
            tool       TEXT NOT NULL,
            project    TEXT NOT NULL DEFAULT '',
            title      TEXT NOT NULL DEFAULT '',
            started    TEXT,
            ended      TEXT,
            msg_count  INTEGER NOT NULL DEFAULT 0,
            kind       TEXT NOT NULL DEFAULT 'main',
            -- Set when the tool deleted the original session file but we kept
            -- the indexed copy (archive mode). NULL for live sessions.
            archived_at TEXT
        );
        CREATE INDEX IF NOT EXISTS idx_files_session ON files(session_id);
        -- Plain transcript rows + external-content FTS. Deleting a session is
        -- an indexed lookup here; with session_id stored UNINDEXED inside the
        -- FTS table it was a full scan per file, which made re-index runs
        -- quadratic in practice.
        CREATE TABLE IF NOT EXISTS messages(
            id         INTEGER PRIMARY KEY,
            session_id TEXT NOT NULL,
            role       TEXT NOT NULL,
            text       TEXT NOT NULL,
            full_text  TEXT
        );
        CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id);
        CREATE TABLE IF NOT EXISTS summaries(
            session_id TEXT PRIMARY KEY,
            summary    TEXT NOT NULL,
            created    TEXT NOT NULL
        );
        -- Curation layer (the editable 'wiki' part). Like summaries, these
        -- are user-authored and survive index rebuilds: only files/messages/
        -- msgs are dropped on a schema bump, never these.
        CREATE TABLE IF NOT EXISTS tags(
            session_id TEXT NOT NULL,
            tag        TEXT NOT NULL,
            PRIMARY KEY (session_id, tag)
        );
        CREATE INDEX IF NOT EXISTS idx_tags_tag ON tags(tag);
        CREATE TABLE IF NOT EXISTS notes(
            session_id TEXT PRIMARY KEY,
            note       TEXT NOT NULL,
            updated    TEXT NOT NULL
        );
        -- Archive (durable, never dropped on a schema bump). When the tool
        -- deletes a session's original file, we keep a self-contained copy
        -- here - the distilled transcript and provenance plus the metadata
        -- needed to reconstruct the files row. This is the only table that is
        -- not re-derivable from disk, so on a schema bump the cache tables are
        -- rehydrated from it. Live sessions are NOT stored here.
        CREATE TABLE IF NOT EXISTS archive(
            session_id  TEXT PRIMARY KEY,
            path        TEXT NOT NULL,
            mtime       INTEGER NOT NULL,
            size        INTEGER NOT NULL,
            tool        TEXT NOT NULL,
            project     TEXT NOT NULL DEFAULT '',
            title       TEXT NOT NULL DEFAULT '',
            started     TEXT,
            ended       TEXT,
            msg_count   INTEGER NOT NULL DEFAULT 0,
            kind        TEXT NOT NULL DEFAULT 'main',
            transcript  TEXT NOT NULL,  -- JSON [[role,text,(full_text)],...] in order
            touched     TEXT NOT NULL,  -- JSON [path,...]
            archived_at TEXT NOT NULL
        );
        -- Provenance: which files each session edited or created, from its
        -- tool calls. Rebuilt from the sessions on sync, so it is dropped on a
        -- schema bump like messages - not curated. The path index powers
        -- `trace` (sessions for a file) and shared-file relatedness.
        CREATE TABLE IF NOT EXISTS touched(
            session_id TEXT NOT NULL,
            path       TEXT NOT NULL,
            PRIMARY KEY (session_id, path)
        );
        CREATE INDEX IF NOT EXISTS idx_touched_path ON touched(path);
        -- Evidence layer over `touched`: the concrete edits (kind + a bounded
        -- snippet of the change) behind each touched path. A log - multiple rows
        -- per (session, path) - so the whole change history of a file survives.
        -- Derived from the sessions on sync, dropped on a schema bump like
        -- touched. Powers `edits_for` and the file-history page.
        CREATE TABLE IF NOT EXISTS edits(
            session_id TEXT NOT NULL,
            path       TEXT NOT NULL,
            kind       TEXT NOT NULL,
            ts         TEXT,
            snippet    TEXT NOT NULL
        );
        CREATE INDEX IF NOT EXISTS idx_edits_path ON edits(path);
        CREATE INDEX IF NOT EXISTS idx_edits_session ON edits(session_id);
        ",
    )?;
    create_msgs_table(conn, tokenizer)?;
    if !msgs_exists {
        conn.execute_batch("INSERT INTO msgs(msgs) VALUES('rebuild');")?;
    }
    Ok(())
}

fn cache_schema_needs_work(conn: &Connection) -> Result<bool> {
    let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?;
    if version != SCHEMA_VERSION {
        return Ok(true);
    }
    for table in [
        "files",
        "messages",
        "summaries",
        "tags",
        "notes",
        "archive",
        "touched",
        "edits",
        "msgs",
    ] {
        let exists: bool = conn.query_row(
            "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?1)",
            [table],
            |row| row.get(0),
        )?;
        if !exists {
            return Ok(true);
        }
    }
    Ok(false)
}

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

    fn mem() -> Connection {
        let c = Connection::open_in_memory().unwrap();
        c.execute_batch(
            "CREATE TABLE meta(key TEXT PRIMARY KEY, value TEXT NOT NULL);
             CREATE TABLE notes(session_id TEXT PRIMARY KEY, note TEXT NOT NULL, updated TEXT NOT NULL);
             INSERT INTO meta VALUES('durable_version','1');
             INSERT INTO notes VALUES('s1','keep me','t');",
        )
        .unwrap();
        c
    }

    #[test]
    fn runner_applies_gated_and_is_idempotent() {
        let c = mem();
        let migs = [Migration {
            version: 2,
            step: MigrationStep::Sql("ALTER TABLE notes ADD COLUMN pinned INTEGER"),
        }];
        run_durable_migrations(&c, &migs).unwrap();
        let cols: Vec<String> = c
            .prepare("SELECT name FROM pragma_table_info('notes')")
            .unwrap()
            .query_map([], |r| r.get(0))
            .unwrap()
            .map(|r| r.unwrap())
            .collect();
        assert!(cols.contains(&"pinned".to_string()), "column added");
        let v: String = c
            .query_row(
                "SELECT value FROM meta WHERE key='durable_version'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(v, "2", "version advanced");
        let note: String = c
            .query_row("SELECT note FROM notes WHERE session_id='s1'", [], |r| {
                r.get(0)
            })
            .unwrap();
        assert_eq!(note, "keep me", "existing durable row preserved");
        // re-run is a clean no-op (ADD COLUMN would otherwise error duplicate column)
        run_durable_migrations(&c, &migs).unwrap();
    }
}

/// A read-only handle for serving processes (the MCP server): it can never
/// create the schema, migrate, VACUUM, or write durable data. Fails if no
/// index exists yet (unlike `open`, which creates one).
pub fn open_readonly() -> Result<Connection> {
    use rusqlite::OpenFlags;
    let conn = Connection::open_with_flags(
        db_path()?,
        OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_URI,
    )?;
    conn.busy_timeout(std::time::Duration::from_millis(5000))?;
    Ok(conn)
}

pub fn open() -> Result<Connection> {
    let conn = Connection::open(db_path()?)?;
    conn.pragma_update(None, "journal_mode", "WAL")?;
    conn.pragma_update(None, "synchronous", "NORMAL")?;
    conn.busy_timeout(std::time::Duration::from_millis(5000))?;
    create_meta_schema(&conn)?;
    if cache_schema_needs_work(&conn)? {
        conn.execute_batch("BEGIN IMMEDIATE")?;
        let cache_setup = (|| -> Result<()> {
            // Another opener may have repaired the schema while this one waited
            // for the lock, so avoid DDL if the cache is current now.
            if cache_schema_needs_work(&conn)? {
                let tokenizer = tokenizer_spec(&conn)?;
                let version: i64 = conn.pragma_query_value(None, "user_version", |r| r.get(0))?;
                // A newer binary built this cache. Rebuilding it in this
                // binary's format only makes that one rebuild it back - a
                // `sessionwiki mcp` left running across an upgrade did exactly
                // that on every call.
                if version > SCHEMA_VERSION {
                    anyhow::bail!(
                        "the index was written by a newer sessionwiki (schema {version}, this \
                         binary reads {SCHEMA_VERSION}); restart this process with the \
                         installed version"
                    );
                }
                if version != SCHEMA_VERSION {
                    // Drop only the derived cache. The durable tables (summaries, tags,
                    // notes, archive) are never dropped: rebuilding the index is cheap,
                    // re-running an LLM or recovering a session the tool already deleted is
                    // not. Archived sessions are rehydrated into the cache below.
                    conn.execute_batch(
                        "DROP TABLE IF EXISTS msgs;
                         DROP TABLE IF EXISTS messages;
                         DROP TABLE IF EXISTS touched;
                         DROP TABLE IF EXISTS files;
                         DROP TABLE IF EXISTS edits;",
                    )?;
                    conn.pragma_update(None, "user_version", SCHEMA_VERSION)?;
                    // Rebuilt rows restart at rowid 1; the fix-ups start over.
                    for key in FIXUP_MARKS {
                        conn.execute("DELETE FROM meta WHERE key = ?1", [key])?;
                    }
                }
                create_cache_schema_with_tokenizer(&conn, &tokenizer)?;
            }
            Ok(())
        })();
        match cache_setup {
            Ok(()) => conn.execute_batch("COMMIT")?,
            Err(error) => {
                let _ = conn.execute_batch("ROLLBACK");
                return Err(error);
            }
        }
    }
    // Durable-table versioning, independent of user_version (the cache). `meta`
    // was opened before any cache schema changes above.
    let durable = read_or_init_durable_version(&conn)?;
    let latest = DURABLE_MIGRATIONS
        .iter()
        .map(|m| m.version)
        .max()
        .unwrap_or(BASELINE_DURABLE_VERSION);
    if durable < latest {
        // Back up the irreplaceable durable data before the first migration runs.
        let bak = db_path()?.with_extension(format!("db.bak-v{durable}"));
        let _ = std::fs::remove_file(&bak);
        conn.execute("VACUUM INTO ?1", params![bak.to_string_lossy()])?;
        run_durable_migrations(&conn, DURABLE_MIGRATIONS)?;
    }
    // Replay archived sessions into the cache whenever any are missing from it
    // - after a schema bump (which dropped the cache) or if the cache was
    // cleared some other way. Gated on a count so a normal open does nothing.
    let arch_total: i64 = conn.query_row("SELECT count(*) FROM archive", [], |r| r.get(0))?;
    let arch_live: i64 = conn.query_row(
        "SELECT count(*) FROM files WHERE archived_at IS NOT NULL",
        [],
        |r| r.get(0),
    )?;
    if arch_total > arch_live {
        rehydrate_archive(&conn)?;
    }
    respell_projects(&conn)?;
    reclassify_codex_subagents(&conn)?;
    resolve_gemini_projects(&conn)?;
    reparse_codex_for_mcp_errors(&conn)?;
    reparse_large_for_provenance(&conn)?;
    Ok(conn)
}

/// Meta keys holding how far each index fix-up has checked `files`, by rowid.
/// A rebuild of the cache restarts rowids, so `open` clears these with it.
const FIXUP_MARKS: [&str; 3] = [
    "codex_subagent_upto",
    "project_case_upto",
    "gemini_project_upto",
];

/// The `files` rowids a fix-up has not yet checked: above its mark, up to the
/// current top. Not a once-only flag: a `sessionwiki mcp` started before an
/// upgrade keeps writing rows with the old code for weeks, and each such
/// write (INSERT OR REPLACE) lands above the mark, so the next open by this
/// version fixes it.
fn unchecked_rowids(conn: &Connection, key: &str) -> Result<(i64, i64)> {
    let mark: i64 = conn
        .query_row("SELECT value FROM meta WHERE key = ?1", [key], |r| {
            r.get::<_, String>(0)
        })
        .optional()?
        .and_then(|v| v.parse().ok())
        .unwrap_or(0);
    let top: i64 = conn.query_row("SELECT coalesce(max(rowid), 0) FROM files", [], |r| {
        r.get(0)
    })?;
    Ok((mark, top))
}

/// Apply a fix-up's writes and advance its mark, but only if the write lock is
/// free right now. Opening the index must never wait on a writer (a sync can
/// hold it for minutes); a busy index is fixed by a later open instead.
fn apply_without_waiting(
    conn: &Connection,
    key: &str,
    top: i64,
    updates: &[(&str, Vec<String>)],
) -> Result<()> {
    conn.busy_timeout(std::time::Duration::ZERO)?;
    let applied = (|| -> rusqlite::Result<()> {
        let tx =
            rusqlite::Transaction::new_unchecked(conn, rusqlite::TransactionBehavior::Immediate)?;
        for (sql, args) in updates {
            tx.execute(sql, rusqlite::params_from_iter(args.iter()))?;
        }
        tx.execute(
            "INSERT OR REPLACE INTO meta(key, value) VALUES (?1, ?2)",
            params![key, top.to_string()],
        )?;
        tx.commit()
    })();
    conn.busy_timeout(std::time::Duration::from_millis(5000))?;
    match applied {
        Err(rusqlite::Error::SqliteFailure(e, _))
            if matches!(
                e.code,
                rusqlite::ErrorCode::DatabaseBusy | rusqlite::ErrorCode::DatabaseLocked
            ) =>
        {
            Ok(())
        }
        other => Ok(other?),
    }
}

/// Mark Codex sub-agent rollouts indexed as main - by a version that did not
/// read the marker - as `sub`, as a fresh parse now does. Only each file's
/// first line is read; a schema bump would re-parse every session on disk.
fn reclassify_codex_subagents(conn: &Connection) -> Result<()> {
    const KEY: &str = "codex_subagent_upto";
    let (mark, top) = unchecked_rowids(conn, KEY)?;
    if top <= mark {
        return Ok(());
    }
    let rows: Vec<(String, String)> = conn
        .prepare(
            "SELECT session_id, path FROM files
             WHERE tool = 'codex' AND kind = 'main' AND rowid > ?1 AND rowid <= ?2",
        )?
        .query_map(params![mark, top], |r| Ok((r.get(0)?, r.get(1)?)))?
        .collect::<rusqlite::Result<_>>()?;
    let mut updates = Vec::new();
    for (session_id, path) in rows {
        let Ok(file) = std::fs::File::open(&path) else {
            continue;
        };
        let mut first = String::new();
        if std::io::BufRead::read_line(&mut std::io::BufReader::new(file), &mut first).is_err() {
            continue;
        }
        let is_sub = serde_json::from_str::<serde_json::Value>(&first).is_ok_and(|v| {
            v.get("type").and_then(serde_json::Value::as_str) == Some("session_meta")
                && crate::adapters::codex_is_subagent_meta(&v)
        });
        if is_sub {
            updates.push((
                "UPDATE files SET kind = 'sub' WHERE session_id = ?1 AND tool = 'codex'",
                vec![session_id],
            ));
        }
    }
    apply_without_waiting(conn, KEY, top, &updates)
}

/// Replace the store-dir slug that Gemini rows indexed by an older version
/// carry as their project with the directory Gemini recorded for it, as a
/// fresh parse now does.
fn resolve_gemini_projects(conn: &Connection) -> Result<()> {
    const KEY: &str = "gemini_project_upto";
    let (mark, top) = unchecked_rowids(conn, KEY)?;
    if top <= mark {
        return Ok(());
    }
    let rows: Vec<(String, String)> = conn
        .prepare(
            "SELECT path, project FROM files
             WHERE tool = 'gemini' AND rowid > ?1 AND rowid <= ?2",
        )?
        .query_map(params![mark, top], |r| Ok((r.get(0)?, r.get(1)?)))?
        .collect::<rusqlite::Result<_>>()?;
    let mut updates = Vec::new();
    for (path, project) in rows {
        let Some(root) = crate::adapters::gemini_project_root(std::path::Path::new(&path)) else {
            continue;
        };
        let root = crate::util::project_key(&root);
        if root != project {
            updates.push((
                "UPDATE files SET project = ?1 WHERE path = ?2",
                vec![root, path],
            ));
        }
    }
    apply_without_waiting(conn, KEY, top, &updates)
}

/// Codex rows parsed by an older Codex reader stay as it left them, since a
/// rollout that never changes is never parsed again. Mark the live ones for
/// one re-parse whenever the reader has changed since; the key is not a
/// fix-up mark, so a cache rebuild - which re-parses everything anyway - does
/// not repeat it. 1: failed MCP calls (0.34.3). 2: MCP calls recorded only as
/// events.
fn reparse_codex_for_mcp_errors(conn: &Connection) -> Result<()> {
    const KEY: &str = "codex_mcp_reparsed";
    const READER: i64 = 2;
    let done: i64 = conn
        .query_row("SELECT value FROM meta WHERE key = ?1", [KEY], |r| {
            r.get::<_, String>(0)
        })
        .optional()?
        .and_then(|v| v.parse().ok())
        .unwrap_or(0);
    if done >= READER {
        return Ok(());
    }
    apply_without_waiting(
        conn,
        KEY,
        READER,
        &[(
            "UPDATE files SET mtime = -1, size = -1 WHERE tool = 'codex' AND archived_at IS NULL",
            vec![],
        )],
    )
}

/// Sessions over the byte cap were indexed without the files they edited
/// between their head and tail; an ended session never changes, so it is never
/// parsed again. Mark the live windowed rows for one re-parse whenever the
/// windowed reader has changed since. 1: provenance read from the whole file.
/// 2: the conversation between head and tail kept too. 3: Codex `compacted`
/// records no longer fill the middle's budget ahead of the conversation.
fn reparse_large_for_provenance(conn: &Connection) -> Result<()> {
    const KEY: &str = "large_provenance_reparsed";
    const READER: i64 = 3;
    let done: i64 = conn
        .query_row("SELECT value FROM meta WHERE key = ?1", [KEY], |r| {
            r.get::<_, String>(0)
        })
        .optional()?
        .and_then(|v| v.parse().ok())
        .unwrap_or(0);
    if done >= READER {
        return Ok(());
    }
    apply_without_waiting(
        conn,
        KEY,
        READER,
        &[(
            "UPDATE files SET mtime = -1, size = -1
             WHERE tool IN ('claude-code', 'codex') AND title LIKE '[large]%'
               AND archived_at IS NULL",
            vec![],
        )],
    )
}

/// Respell stored project paths as on disk, so case variants written by a
/// version without `project_key` merge into one project. Only the paths in
/// rows not yet checked are looked at; a schema bump would re-parse every
/// session on disk to fix a handful of strings.
fn respell_projects(conn: &Connection) -> Result<()> {
    const KEY: &str = "project_case_upto";
    let (mark, top) = unchecked_rowids(conn, KEY)?;
    if top <= mark {
        return Ok(());
    }
    let projects: Vec<String> = conn
        .prepare(
            "SELECT DISTINCT project FROM files
             WHERE project != '' AND rowid > ?1 AND rowid <= ?2",
        )?
        .query_map(params![mark, top], |r| r.get(0))?
        .collect::<rusqlite::Result<_>>()?;
    let mut updates = Vec::new();
    for project in projects {
        let key = crate::util::project_key(&project);
        if key != project {
            updates.push((
                "UPDATE files SET project = ?1 WHERE project = ?2",
                vec![key, project],
            ));
        }
    }
    apply_without_waiting(conn, KEY, top, &updates)
}

/// After a schema bump drops the cache tables, replay archived sessions back
/// into them from the durable `archive` table, so search, `trace`, and reading
/// keep working for sessions whose originals the tool deleted. This is what
/// makes archive survive a rebuild; without it a version bump would silently
/// lose exactly the data that cannot be re-derived from disk.
fn rehydrate_archive(conn: &Connection) -> Result<()> {
    let mut stmt = conn.prepare(
        "SELECT session_id, path, tool, project, title, started, ended,
                kind, transcript, touched, archived_at FROM archive",
    )?;
    let rows: Vec<ArchiveRow> = stmt
        .query_map([], |r| {
            Ok(ArchiveRow {
                session_id: r.get(0)?,
                path: r.get(1)?,
                tool: r.get(2)?,
                project: r.get(3)?,
                title: r.get(4)?,
                started: r.get(5)?,
                ended: r.get(6)?,
                kind: r.get(7)?,
                transcript: r.get(8)?,
                touched: r.get(9)?,
                archived_at: r.get(10)?,
            })
        })?
        .collect::<rusqlite::Result<_>>()?;

    for a in rows {
        // The transcript is the durable backup; if it will not deserialize,
        // skip the session rather than rehydrate an empty shell that claims to
        // have content - that would be silent data loss disguised as success.
        let msgs = match decode_archive_transcript(&a.transcript) {
            Ok(m) => m,
            Err(e) => {
                eprintln!(
                    "archive: skipping {} - unreadable transcript ({e})",
                    a.session_id
                );
                continue;
            }
        };
        let paths: Vec<String> = serde_json::from_str(&a.touched).unwrap_or_else(|e| {
            eprintln!("archive: {} has unreadable provenance ({e})", a.session_id);
            Vec::new()
        });

        // Idempotent: clear any existing cache rows for this session first, so
        // re-running rehydrate never duplicates messages/FTS rows.
        delete_session_msgs(conn, &a.session_id)?;
        delete_session_provenance(conn, &a.session_id)?;
        conn.execute(
            "DELETE FROM files WHERE session_id = ?1",
            params![a.session_id],
        )?;

        // mtime/size are forced to 0 so that if this file ever reappears on
        // disk, the next sync always sees a mismatch and re-parses it, clearing
        // archived_at. msg_count comes from the actual transcript, never the
        // stored count, so the displayed count can never outrun the content.
        conn.execute(
            "INSERT INTO files
             (path, mtime, size, session_id, tool, project, title, started, ended,
              msg_count, kind, archived_at)
             VALUES (?1,0,0,?2,?3,?4,?5,?6,?7,?8,?9,?10)",
            params![
                a.path,
                a.session_id,
                a.tool,
                crate::util::project_key(&a.project),
                a.title,
                a.started,
                a.ended,
                msgs.len() as i64,
                a.kind,
                a.archived_at,
            ],
        )?;
        {
            let mut ins_row = conn.prepare_cached(
                "INSERT INTO messages(session_id, role, text, full_text)
                     VALUES (?1,?2,?3,?4)",
            )?;
            let mut ins_fts =
                conn.prepare_cached("INSERT INTO msgs(rowid, text) VALUES (?1,?2)")?;
            for message in &msgs {
                // Re-normalize on rehydrate: pre-fix archives hold raw/NFD JSON,
                // so this is where archived Korean sessions become NFC again.
                // Also redact - a pre-redaction archive holds raw secrets.
                let text = crate::redact::redact(&crate::util::nfc(&message.text)).into_owned();
                let full_text = message
                    .full_text
                    .as_deref()
                    .map(|text| crate::redact::redact(&crate::util::nfc(text)).into_owned());
                // External-content FTS requires byte-identical message text.
                ins_row.execute(params![a.session_id, message.role, text, full_text])?;
                ins_fts.execute(params![conn.last_insert_rowid(), text])?;
            }
        }
        let mut ins_touched =
            conn.prepare_cached("INSERT OR IGNORE INTO touched(session_id, path) VALUES (?1,?2)")?;
        for p in &paths {
            ins_touched.execute(params![a.session_id, crate::util::nfc(p)])?;
        }
    }
    Ok(())
}

struct ArchiveRow {
    session_id: String,
    path: String,
    tool: String,
    project: String,
    title: String,
    started: Option<String>,
    ended: Option<String>,
    kind: String,
    transcript: String,
    touched: String,
    archived_at: String,
}

/// Bring the index up to date with what is on disk. Files whose (mtime, size)
/// changed since the last run are re-parsed; archived rows whose source has
/// returned unchanged are made live again without re-reading the transcript.
/// Insert one parsed session into the cache tables (files, messages, msgs,
/// touched) and tag it if an oh-my-* harness drove it. `key` is the stored
/// path/identity, `mtime` the change-token, `size` the byte size (0 for
/// shared-store sessions). Shared by the file-per-session and shared-store paths.
fn index_one(
    tx: &rusqlite::Transaction,
    session: &crate::model::Session,
    key: &str,
    mtime: i64,
    size: i64,
    output_mode: ToolOutputMode,
) -> Result<()> {
    delete_session_msgs(tx, &session.id)?;
    delete_session_provenance(tx, &session.id)?;
    // A path that was archived (the tool deleted it, now it is back) is live
    // again: this INSERT clears archived_at, and the durable copy is dropped.
    tx.execute(
        "DELETE FROM archive WHERE session_id = ?1",
        params![session.id],
    )?;
    tx.execute(
        "INSERT OR REPLACE INTO files
         (path, mtime, size, session_id, tool, project, title, started, ended, msg_count, kind)
         VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11)",
        params![
            key,
            mtime,
            size,
            session.id,
            session.tool,
            crate::util::project_key(&session.project),
            crate::redact::redact(&session.title).as_ref(),
            session.started.map(|t| t.to_rfc3339()),
            session.ended.map(|t| t.to_rfc3339()),
            session.messages.len() as i64,
            if session.subagent { "sub" } else { "main" },
        ],
    )?;
    fold_moved_copy(tx, session, key)?;
    // Contract: messages are inserted in transcript order within one
    // transaction, so the autoincrement messages.id is a monotonic proxy for
    // order (preview + the web transcript rely on it). Keep this sequential.
    {
        let mut ins_row = tx.prepare_cached(
            "INSERT INTO messages(session_id, role, text, full_text)
             VALUES (?1,?2,?3,?4)",
        )?;
        let mut ins_fts = tx.prepare_cached("INSERT INTO msgs(rowid, text) VALUES (?1,?2)")?;
        for m in &session.messages {
            // Normalize once and reuse for the plain row and external-content
            // FTS row: they MUST be byte-identical or FTS delete/rebuild corrupts.
            // Strip secrets before they enter the index, which outlives the
            // original session in archive mode.
            let text = crate::redact::redact(&crate::util::nfc(&m.text)).into_owned();
            let full_text = if output_mode == ToolOutputMode::Full {
                match &m.tool {
                    Some(crate::model::ToolEvent::Summary(summary)) => {
                        summary.output.as_deref().map(|output| {
                            crate::redact::redact(&crate::util::nfc(output)).into_owned()
                        })
                    }
                    _ => None,
                }
            } else {
                None
            };
            ins_row.execute(params![session.id, m.role.label(), text, full_text])?;
            ins_fts.execute(params![tx.last_insert_rowid(), text])?;
        }
        let mut ins_touched =
            tx.prepare_cached("INSERT OR IGNORE INTO touched(session_id, path) VALUES (?1,?2)")?;
        for p in &session.touched {
            ins_touched.execute(params![session.id, crate::util::nfc(p)])?;
        }
        // Evidence layer (a log, not a set - the same file edited twice keeps
        // both rows). Prior rows were cleared with `touched` above.
        let mut ins_edit = tx.prepare_cached(
            "INSERT INTO edits(session_id, path, kind, ts, snippet) VALUES (?1,?2,?3,?4,?5)",
        )?;
        for e in &session.edits {
            ins_edit.execute(params![
                session.id,
                crate::util::nfc(&e.path),
                e.kind.as_str(),
                e.ts.map(|t| t.to_rfc3339()),
                crate::redact::redact(&e.snippet).as_ref(),
            ])?;
        }
    }
    // Tag sessions an oh-my-* harness drove (it wraps Claude Code / Codex /
    // OpenCode) so they are filterable; recomputed on every reindex.
    if matches!(session.tool, "claude-code" | "codex" | "opencode") {
        if let Some(h) = crate::adapters::harness::detect(&session.project) {
            add_tag(tx, &session.id, h)?;
        }
    }
    Ok(())
}

/// The source exists again with the exact metadata we indexed before it was
/// mistakenly archived. Its searchable rows are still present, so restoring
/// the live marker needs no transcript parse or FTS rewrite.
fn restore_unchanged(conn: &mut Connection, keys: &[String]) -> Result<()> {
    if keys.is_empty() {
        return Ok(());
    }
    let tx = conn.transaction()?;
    for key in keys {
        tx.execute(
            "DELETE FROM archive WHERE session_id = (SELECT session_id FROM files WHERE path = ?1)",
            params![key],
        )?;
        tx.execute(
            "UPDATE files SET archived_at = NULL WHERE path = ?1",
            params![key],
        )?;
    }
    tx.commit()?;
    Ok(())
}

/// A Codex conversation can change path: archiving a thread moves its rollout,
/// and after the history migration Codex may delete the rollout while the
/// thread lives on in its history db. The old path's row is kept as archived,
/// so once the conversation is indexed at its new path the kept row is a second
/// listing of it. Fold it in, moving its tags and note across, unless it held
/// more messages than the new copy - folding that would lose content.
fn fold_moved_copy(
    tx: &rusqlite::Transaction,
    session: &crate::model::Session,
    key: &str,
) -> Result<()> {
    if session.tool != "codex" {
        return Ok(());
    }
    let Some(native) = native_id_of(key) else {
        return Ok(());
    };
    // Not only rows already archived: the old row can still be live when the
    // new one is indexed (its adapter reconciles later). A live row whose
    // file still exists is a second copy, not a move, and is left alone.
    let kept: Vec<(String, String, i64, bool)> = tx
        .prepare(
            "SELECT path, session_id, msg_count, archived_at IS NOT NULL FROM files
             WHERE tool = 'codex' AND path <> ?1 AND instr(lower(path), ?2) > 0",
        )?
        .query_map(params![key, native], |r| {
            Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?))
        })?
        .collect::<rusqlite::Result<_>>()?;
    for (path, sid, count, archived) in kept {
        if sid == session.id
            || native_id_of(&path).as_deref() != Some(native.as_str())
            || count > session.messages.len() as i64
            || (!archived && std::path::Path::new(&path).exists())
        {
            continue;
        }
        for table in ["tags", "notes", "summaries"] {
            tx.execute(
                &format!("UPDATE OR IGNORE {table} SET session_id = ?2 WHERE session_id = ?1"),
                params![sid, session.id],
            )?;
        }
        delete_session_msgs(tx, &sid)?;
        delete_session_provenance(tx, &sid)?;
        for table in ["files", "archive", "summaries", "tags", "notes"] {
            tx.execute(
                &format!("DELETE FROM {table} WHERE session_id = ?1"),
                params![sid],
            )?;
        }
    }
    Ok(())
}

/// The end-of-adapter sync line. Honest about failures: "indexed 12/14
/// (2 failed to parse)" rather than pretending everything landed.
fn report_indexed(tool: &str, total: usize, failed: usize) {
    if failed > 0 {
        eprintln!(
            "\r[{tool}] indexed {}/{total} ({failed} failed to parse)    ",
            total - failed
        );
    } else {
        eprintln!("\r[{tool}] indexed {total}/{total}    ");
    }
}

pub fn sync(conn: &mut Connection, only_tool: Option<&str>) -> Result<()> {
    sync_bounded(conn, only_tool, None)
}

/// Like [`sync`], but when `since` is set, only files/sessions modified at or
/// after that epoch-second are (re)parsed - a bounded FRESHNESS top-up for the
/// MCP path, so `recent_sessions` picks up a just-started sibling without paying
/// the full-corpus re-parse (the 46GB-codex trap). Deletion reconciliation still
/// sees every file, so a bounded run never archives a live session; older
/// changed files are simply left for the next full `sync`.
pub fn sync_bounded(
    conn: &mut Connection,
    only_tool: Option<&str>,
    since: Option<i64>,
) -> Result<()> {
    let adapters: Vec<Box<dyn Adapter>> = match only_tool {
        Some(t) => adapters::named(t),
        None => adapters::all(),
    };
    sync_with(conn, &adapters, since)
}

/// Like [`sync_bounded`], but over an explicit adapter list instead of the
/// built-in registry. A program that embeds this crate as a library can index
/// its own sessions by passing its own [`Adapter`] alongside `adapters::all()`.
/// Readers without that adapter use the indexed transcript; call this again
/// to make later changes to those sessions visible to the standalone binary.
pub fn sync_with(
    conn: &mut Connection,
    adapters: &[Box<dyn Adapter>],
    since: Option<i64>,
) -> Result<()> {
    // Per-session progress redraws one line with `\r`, which only means
    // anything on a terminal. Redirected into a log it becomes a single line
    // megabytes long, so decide once per sync and skip those writes when stderr
    // is not a terminal. Warnings and the per-tool summary still print.
    let progress = std::io::stderr().is_terminal();
    let output_mode = tool_output_mode(conn)?;
    let mut known: HashMap<String, (i64, i64, bool)> = HashMap::new();
    {
        let mut stmt =
            conn.prepare("SELECT path, mtime, size, archived_at IS NOT NULL FROM files")?;
        let rows = stmt.query_map([], |r| {
            Ok((
                r.get::<_, String>(0)?,
                (
                    r.get::<_, i64>(1)?,
                    r.get::<_, i64>(2)?,
                    r.get::<_, bool>(3)?,
                ),
            ))
        })?;
        for row in rows {
            let (path, ms) = row?;
            known.insert(path, ms);
        }
    }

    let mut archived_total = 0usize;
    for adapter in adapters {
        // `store_present` tells "the tool pruned some sessions" (root exists,
        // those gone) apart from "the whole store vanished" (uninstall,
        // unmounted) - we must not mass-archive on the latter.
        let tool = adapter.name();
        let store_present = adapter.root().is_some_and(|r| r.exists());

        // Shared store (e.g. OpenCode's SQLite db): enumerate sessions by key +
        // change-token and re-parse only the changed ones, bypassing the
        // file-per-session path. The token is stored in the `mtime` column
        // (size 0), so the same change comparison works.
        if let Some(store) = adapter.store() {
            let mut seen: Vec<String> = Vec::with_capacity(store.keys.len());
            let mut pending: Vec<String> = Vec::new();
            let mut restore: Vec<String> = Vec::new();
            for (key, token) in &store.keys {
                match known.get(key) {
                    Some(&(mtime, 0, true)) if mtime == *token => restore.push(key.clone()),
                    Some(&(mtime, 0, false)) if mtime == *token => {}
                    old if old.is_some_and(|row| row.2 || row.0 < 0 || row.1 < 0)
                        || since.is_none_or(|s| *token >= s) =>
                    {
                        pending.push(key.clone());
                    }
                    _ => {}
                }
                seen.push(key.clone());
            }
            restore_unchanged(conn, &restore)?;
            // Reconcile deletions only when the whole store was read this run.
            // If a backing db was present but unreadable (locked, half-written),
            // `seen` is partial - pruning off it would archive the whole corpus
            // on a transient hiccup, so skip reconciliation until a clean read.
            if store.had_error {
                // The discover path says this out loud; this one did not. It is
                // the path aider takes, whose walk is capped at two seconds over
                // the whole home - so on a large home the flag can be set every
                // run, reconciliation never happens, and nothing says why a
                // session the tool deleted is still listed.
                eprintln!(
                    "[{tool}] the store could not be read in full; \
                     skipping deletion reconciliation this run"
                );
            } else {
                archived_total += archive_or_prune(
                    conn,
                    tool,
                    &seen,
                    store_present,
                    adapter.reconcile_scope().as_deref(),
                )?;
            }

            if !pending.is_empty() {
                let token_of: HashMap<&str, i64> =
                    store.keys.iter().map(|(k, t)| (k.as_str(), *t)).collect();
                let total = pending.len();
                let mut failed = 0usize;
                let tx = conn.transaction()?;
                for (i, key) in pending.iter().enumerate() {
                    if progress {
                        eprint!("\r[{tool}] indexing {}/{total}", i + 1);
                        std::io::stderr().flush().ok();
                    }
                    // A failed parse is warned, not silently dropped: the user
                    // must know a session is missing from the corpus.
                    let session = match adapters::parse_session_key(adapter.as_ref(), key) {
                        Ok(s) => s,
                        Err(e) => {
                            failed += 1;
                            eprintln!("\r[{tool}] failed to parse {key}: {e:#}");
                            continue;
                        }
                    };
                    let token = token_of.get(key.as_str()).copied().unwrap_or(0);
                    index_one(&tx, &session, key, token, 0, output_mode)?;
                }
                tx.commit()?;
                report_indexed(tool, total, failed);
            }
            continue;
        }

        let discovered = adapter.discover();
        let mut seen: Vec<String> = Vec::with_capacity(discovered.files.len());
        let mut pending: Vec<(PathBuf, i64, i64)> = Vec::new();
        let mut restore: Vec<String> = Vec::new();

        for f in discovered.files {
            let meta = match f.metadata() {
                Ok(m) => m,
                Err(e) => {
                    // The file was just discovered, so it exists: a failed
                    // stat must keep it in `seen` (dropping it would let
                    // reconciliation archive a live session) and be warned,
                    // not swallowed.
                    eprintln!("\r[{tool}] failed to stat {}: {e}", f.display());
                    seen.push(f.to_string_lossy().into_owned());
                    continue;
                }
            };
            let mtime = meta
                .modified()
                .ok()
                .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
                .map(|d| d.as_secs() as i64)
                .unwrap_or(0);
            let size = meta.len() as i64;
            let key = f.to_string_lossy().into_owned();
            match known.get(&key) {
                Some(&(old_mtime, old_size, true)) if old_mtime == mtime && old_size == size => {
                    restore.push(key.clone());
                }
                Some(&(old_mtime, old_size, false)) if old_mtime == mtime && old_size == size => {}
                old if old.is_some_and(|row| row.2 || row.0 < 0 || row.1 < 0)
                    || since.is_none_or(|s| mtime >= s) =>
                {
                    pending.push((f, mtime, size));
                }
                _ => {}
            }
            seen.push(key);
        }
        restore_unchanged(conn, &restore)?;

        // Same guard the shared-store path has: a partial walk (an unreadable
        // directory) means `seen` is incomplete - reconciling deletions off it
        // would archive live sessions, so wait for a clean walk.
        if discovered.had_error {
            eprintln!(
                "[{tool}] some session directories could not be read; \
                 skipping deletion reconciliation this run"
            );
        } else {
            archived_total += archive_or_prune(
                conn,
                tool,
                &seen,
                store_present,
                adapter.reconcile_scope().as_deref(),
            )?;
        }

        if pending.is_empty() {
            continue;
        }
        let total = pending.len();
        let mut done = 0usize;
        let mut failed = 0usize;
        let tx = conn.transaction()?;
        for (path, mtime, size) in pending {
            done += 1;
            if progress {
                eprint!("\r[{tool}] indexing {done}/{total}");
                std::io::stderr().flush().ok();
            }

            // A failed parse is warned, not silently dropped: the user must
            // know a session is missing from the corpus.
            let session = match adapters::parse_session(adapter.as_ref(), &path) {
                Ok(s) => s,
                Err(e) => {
                    failed += 1;
                    eprintln!("\r[{tool}] failed to parse {}: {e:#}", path.display());
                    continue;
                }
            };
            let key = path.to_string_lossy();
            index_one(&tx, &session, &key, mtime, size, output_mode)?;
        }
        tx.commit()?;
        report_indexed(tool, total, failed);
    }

    // The one passive signal that archive is earning its keep: how many
    // sessions we kept this run that the tool deleted, and the running total.
    if archived_total > 0 {
        let kept: i64 = conn.query_row(
            "SELECT count(*) FROM files WHERE kind='main' AND archived_at IS NOT NULL",
            [],
            |r| r.get(0),
        )?;
        eprintln!(
            "archived {archived_total} session(s) the tool removed ({kept} kept that your tools have deleted)"
        );
    }
    Ok(())
}

/// External-content FTS5 requires handing back the old rows on delete.
fn delete_session_msgs(conn: &Connection, session_id: &str) -> Result<()> {
    conn.execute(
        "INSERT INTO msgs(msgs, rowid, text)
         SELECT 'delete', id, text FROM messages WHERE session_id = ?1",
        params![session_id],
    )?;
    conn.execute(
        "DELETE FROM messages WHERE session_id = ?1",
        params![session_id],
    )?;
    Ok(())
}

/// Clear a session's derived provenance - `touched` AND `edits` together - so a
/// delete site can never remember one and forget the other (they drifted once,
/// leaving orphaned edit rows visible through `edits_for`).
fn delete_session_provenance(conn: &Connection, session_id: &str) -> Result<()> {
    conn.execute(
        "DELETE FROM touched WHERE session_id = ?1",
        params![session_id],
    )?;
    conn.execute(
        "DELETE FROM edits WHERE session_id = ?1",
        params![session_id],
    )?;
    Ok(())
}

/// Reconcile the index with a tool's store after discovery. Sessions whose
/// original file disappeared are **archived** (kept in the durable `archive`
/// table and flagged in `files`, with messages/touched left in place so
/// search and trace keep working) instead of deleted - unless
/// `SESSIONWIKI_NO_ARCHIVE` is set or the session has no indexed content, in
/// which case they are pruned as before. Returns how many were newly archived.
///
/// Guard: if the store root is gone (uninstalled, unmounted), do not touch its
/// sessions - that is "the whole store vanished", not "the tool pruned some".
/// An existing-but-empty store is a legitimate prune-everything and proceeds.
fn archive_or_prune(
    conn: &Connection,
    tool: &str,
    seen: &[String],
    store_present: bool,
    scope: Option<&str>,
) -> Result<usize> {
    let no_archive = std::env::var_os("SESSIONWIKI_NO_ARCHIVE").is_some();
    // A scoped adapter speaks only for the keys under its prefix; everything
    // else under the same tool name belongs to another store and must be left
    // alone. Filtering happens in Rust, not with SQL `LIKE`: keys are paths and
    // `_` is a LIKE wildcard.
    let in_scope = |key: &str| scope.is_none_or(|p| key.starts_with(p));
    let seen_set: std::collections::HashSet<&str> = seen
        .iter()
        .map(String::as_str)
        .filter(|k| in_scope(k))
        .collect();

    let mut stmt =
        conn.prepare("SELECT path, session_id FROM files WHERE tool = ?1 AND archived_at IS NULL")?;
    let all_live: Vec<(String, String)> = stmt
        .query_map(params![tool], |r| Ok((r.get(0)?, r.get(1)?)))?
        .collect::<rusqlite::Result<_>>()?;
    // A session still live at a path discovery did return was not deleted; the
    // gone row is a second spelling of it, and archiving it would keep both.
    let still_live: std::collections::HashSet<String> = all_live
        .iter()
        .filter(|(p, _)| seen_set.contains(p.as_str()))
        .map(|(_, sid)| sid.clone())
        .collect();
    let live: Vec<(String, String)> = all_live.into_iter().filter(|(p, _)| in_scope(p)).collect();
    let gone: Vec<(String, String)> = live
        .into_iter()
        .filter(|(p, _)| !seen_set.contains(p.as_str()))
        .collect();
    if gone.is_empty() {
        return Ok(0);
    }
    if !store_present {
        eprintln!(
            "[{tool}] store not found - skipping ({} indexed session(s) left untouched, not archived)",
            gone.len()
        );
        return Ok(0);
    }
    // The root exists but discovery returned nothing while we still had live
    // sessions: could be a legitimate prune-everything, but also a transient
    // read failure (permissions, a half-mounted network FS). Archiving keeps
    // the data (reversible on the next good sync), but say so loudly.
    if seen_set.is_empty() {
        eprintln!(
            "[{tool}] no sessions found on disk but {} were indexed - archiving them; \
             if the store is just unreadable right now, they will un-archive on the next sync",
            gone.len()
        );
    }

    let mut archived = 0usize;
    for (path, sid) in gone {
        if still_live.contains(&sid) {
            conn.execute("DELETE FROM files WHERE path = ?1", params![path])?;
        } else if no_archive || !has_messages(conn, &sid)? {
            conn.execute("DELETE FROM files WHERE path = ?1", params![path])?;
            delete_session_msgs(conn, &sid)?;
            delete_session_provenance(conn, &sid)?;
        } else {
            archive_session(conn, &path, &sid)?;
            archived += 1;
        }
    }
    Ok(archived)
}

/// Nothing indexed means nothing to keep: an archived copy would hold a title
/// and no transcript.
fn has_messages(conn: &Connection, sid: &str) -> Result<bool> {
    Ok(conn.query_row(
        "SELECT EXISTS(SELECT 1 FROM messages WHERE session_id = ?1)",
        params![sid],
        |r| r.get(0),
    )?)
}

/// Copy a session whose original file is gone into the durable `archive` table
/// and flag its `files` row. The messages/msgs/touched rows are left in place
/// so search and `trace` keep working; the archive copy is the rebuild-survival
/// backup (replayed by `rehydrate_archive` after a schema bump).
fn archive_session(conn: &Connection, path: &str, sid: &str) -> Result<()> {
    let mut s = conn
        .prepare("SELECT role, text, full_text FROM messages WHERE session_id = ?1 ORDER BY id")?;
    let rows: Vec<(String, String, Option<String>)> = s
        .query_map(params![sid], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))?
        .collect::<rusqlite::Result<_>>()?;
    drop(s);
    let transcript: Vec<serde_json::Value> = rows
        .into_iter()
        .map(|(role, text, full_text)| match full_text {
            Some(full_text) => serde_json::json!([role, text, full_text]),
            None => serde_json::json!([role, text]),
        })
        .collect();
    let mut s = conn.prepare("SELECT path FROM touched WHERE session_id = ?1 ORDER BY rowid")?;
    let touched: Vec<String> = s
        .query_map(params![sid], |r| r.get(0))?
        .collect::<rusqlite::Result<_>>()?;
    drop(s);
    let transcript_json = serde_json::to_string(&transcript)?;
    let touched_json = serde_json::to_string(&touched)?;
    conn.execute(
        "INSERT OR REPLACE INTO archive
         (session_id, path, mtime, size, tool, project, title, started, ended,
          msg_count, kind, transcript, touched, archived_at)
         SELECT session_id, path, mtime, size, tool, project, title, started, ended,
                msg_count, kind, ?2, ?3, datetime('now')
         FROM files WHERE path = ?1",
        params![path, transcript_json, touched_json],
    )?;
    conn.execute(
        "UPDATE files SET archived_at = datetime('now') WHERE path = ?1",
        params![path],
    )?;
    Ok(())
}

/// Serializes to the agent-facing JSON contract: snake_case keys matching the
/// web API (`id`, `msgs`, tags as an array). The absolute `path` is never
/// serialized verbatim - only the tool's own `native_id` (the codex rollout /
/// claude transcript UUID extracted from the filename) is exposed, so an agent
/// can join a harness "tower" row (which knows only the native id) back to a
/// session without the local path ever leaking.
#[derive(Serialize)]
pub struct SessionRow {
    #[serde(rename = "id")]
    pub session_id: String,
    pub tool: String,
    /// The NATIVE session file path. Never serialized as-is; it is surfaced only
    /// as the extracted `native_id` UUID (or null when the filename carries no
    /// UUID) via [`ser_native_id`].
    #[serde(rename = "native_id", serialize_with = "ser_native_id")]
    pub path: String,
    pub project: String,
    pub title: String,
    pub started: Option<String>,
    /// When the session last had a message. A session resumed for weeks began
    /// long ago; this is what makes it recent. Absent where it was not read.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub last_active: Option<String>,
    #[serde(rename = "msgs")]
    pub msg_count: i64,
    pub kind: String,
    /// Tail of the conversation (last assistant message), so a list can show
    /// how the session ended without opening it.
    pub preview: Option<String>,
    /// Cached LLM synopsis, if `summarize` has been run for this session.
    pub summary: Option<String>,
    /// Comma-joined user tags, if any. Serialized as a string array (or null).
    #[serde(serialize_with = "ser_tags")]
    pub tags: Option<String>,
    /// True if the tool deleted the original and we kept the indexed copy.
    pub archived: bool,
    /// The swapdex account profile active when this session started, when a
    /// swapdex switch timeline exists on the machine. Null otherwise - a
    /// missing badge, never a guess.
    pub account: Option<String>,
}

/// Tags are stored comma-joined but the JSON contract is an array (matching the
/// web API). Null when there are no tags.
fn ser_tags<S: serde::Serializer>(tags: &Option<String>, s: S) -> Result<S::Ok, S::Error> {
    match tags {
        Some(t) => s.collect_seq(t.split(',')),
        None => s.serialize_none(),
    }
}

/// Serialize a session's stored `path` as its `native_id` only: the tool's own
/// session UUID (or null when the filename carries no UUID). The absolute path
/// is never emitted - only the extracted UUID reaches the JSON contract.
fn ser_native_id<S: serde::Serializer>(path: &str, s: S) -> Result<S::Ok, S::Error> {
    match native_id_of(path) {
        Some(id) => s.serialize_some(&id),
        None => s.serialize_none(),
    }
}

/// Extract the native session UUID embedded in a session file's path - the id
/// the originating tool (and a harness "tower") knows the session by: the Codex
/// rollout UUID (`rollout-<ts>-<uuid>.jsonl`) or the Claude Code transcript UUID
/// (`<uuid>.jsonl`, or `agent-<uuid>.jsonl` for a subagent). Returns the first
/// canonical 8-4-4-4-12 UUID found in the file NAME, lowercased, or None when the
/// filename carries no UUID (tools that key sessions differently). Scanning the
/// file name (not the whole path) keeps a codex timestamp or a parent directory
/// from being mistaken for the session's own id.
pub fn native_id_of(path: &str) -> Option<String> {
    let name = std::path::Path::new(path).file_name()?.to_string_lossy();
    find_uuid(&name)
}

/// The first canonical UUID (8-4-4-4-12 hex with dashes) appearing in `s`,
/// lowercased. None when there is no such substring.
fn find_uuid(s: &str) -> Option<String> {
    let b = s.as_bytes();
    if b.len() < 36 {
        return None;
    }
    for start in 0..=b.len() - 36 {
        if is_uuid_bytes(&b[start..start + 36]) {
            return Some(s[start..start + 36].to_ascii_lowercase());
        }
    }
    None
}

/// Whether a 36-byte window is a canonical UUID: hex everywhere except dashes at
/// positions 8, 13, 18, 23.
fn is_uuid_bytes(b: &[u8]) -> bool {
    b.len() == 36
        && b.iter().enumerate().all(|(i, &c)| match i {
            8 | 13 | 18 | 23 => c == b'-',
            _ => c.is_ascii_hexdigit(),
        })
}

/// Whether `q` could be a native-id lookup (full UUID or a UUID prefix), as
/// opposed to a sessionwiki short id (always 12 hex chars, no dashes). A valid
/// UUID prefix longer than its first 8-hex group must carry a dash (position 8
/// is always `-`), so a plain-hex string of 9+ chars can only be a short id and
/// never triggers the native scan - which keeps short-id resolution unchanged.
fn looks_like_native_prefix(q: &str) -> bool {
    let len = q.len();
    if !(4..=36).contains(&len) {
        return false;
    }
    if !q.bytes().all(|c| c.is_ascii_hexdigit() || c == b'-') {
        return false;
    }
    // Plain hex, no dash: only a first-group prefix (<= 8 chars) can be a UUID.
    q.contains('-') || len <= 8
}

/// Correlated subquery for the preview column; messages.id preserves
/// insertion order, which is message order.
const PREVIEW_SQL: &str = "(SELECT substr(m2.text, 1, 280) FROM messages m2
    WHERE m2.session_id = f.session_id AND m2.role = 'assistant'
    ORDER BY m2.id DESC LIMIT 1)";

const SUMMARY_SQL: &str = "(SELECT s.summary FROM summaries s WHERE s.session_id = f.session_id)";

const TAGS_SQL: &str =
    "(SELECT group_concat(t.tag, ',') FROM tags t WHERE t.session_id = f.session_id)";

pub fn recent(
    conn: &Connection,
    limit: usize,
    tool: Option<&str>,
    project: Option<&str>,
    tag: Option<&str>,
    include_subagents: bool,
) -> Result<Vec<SessionRow>> {
    let mut sql = format!(
        "SELECT session_id, tool, path, project, title, started, msg_count, kind, {PREVIEW_SQL}, {SUMMARY_SQL}, {TAGS_SQL}, (archived_at IS NOT NULL), COALESCE(ended, started)
         FROM files f WHERE 1=1",
    );
    let mut args: Vec<String> = Vec::new();
    // A tag filter is an explicit ask for *those* sessions; don't hide subagent
    // hits behind the main-only default (the tag cloud counts every kind, so a
    // sub-only tag would otherwise show in the cloud but return nothing here).
    if !include_subagents && tag.is_none() {
        sql.push_str(" AND kind = 'main'");
    }
    if let Some(t) = tool {
        sql.push_str(" AND tool = ?");
        args.push(t.to_string());
    }
    if let Some(p) = project {
        sql.push_str(" AND project LIKE ?");
        args.push(format!("%{}%", crate::util::nfc(p)));
    }
    if let Some(t) = tag {
        sql.push_str(
            " AND EXISTS (SELECT 1 FROM tags g WHERE g.session_id = f.session_id AND g.tag = ?)",
        );
        args.push(norm_tag(t)); // stored tags are normalized; match their form
    }
    // Recent means recently ACTIVE: a session resumed for weeks began long
    // ago, and ordering by its start buried it below everything begun since.
    // Both columns are UTC RFC 3339, so they order as text.
    sql.push_str(&format!(
        " ORDER BY COALESCE(ended, started) DESC LIMIT {limit}"
    ));

    let mut stmt = conn.prepare(&sql)?;
    let rows = stmt.query_map(rusqlite::params_from_iter(args), |r| {
        Ok(SessionRow {
            last_active: r.get(12)?,
            session_id: r.get(0)?,
            tool: r.get(1)?,
            path: r.get(2)?,
            project: r.get(3)?,
            title: r.get(4)?,
            started: r.get(5)?,
            msg_count: r.get(6)?,
            kind: r.get(7)?,
            preview: r.get(8)?,
            summary: r.get(9)?,
            tags: r.get(10)?,
            archived: r.get(11)?,
            account: None,
        })
    })?;
    let mut out = rows.collect::<rusqlite::Result<Vec<_>>>()?;
    crate::account_link::annotate(out.iter_mut());
    Ok(out)
}

/// Recent main sessions whose launch project is EXACTLY this directory (for the
/// SessionStart recall hook). Exact equality - never the substring `--project`
/// filter, which over-matches sibling/child paths. Newest first, stable.
pub fn project_brief(conn: &Connection, project: &str, limit: usize) -> Result<Vec<SessionRow>> {
    let p = crate::util::project_key(project.trim_end_matches('/'));
    let mut stmt = conn.prepare(&format!(
        "SELECT session_id, tool, path, project, title, started, msg_count, kind, {PREVIEW_SQL}, {SUMMARY_SQL}, {TAGS_SQL}, (archived_at IS NOT NULL)
         FROM files f
         WHERE f.project = ?1 AND f.kind = 'main'
         ORDER BY f.started DESC, f.session_id
         LIMIT ?2"
    ))?;
    let rows = stmt.query_map(params![p, limit as i64], |r| {
        Ok(SessionRow {
            last_active: None,
            session_id: r.get(0)?,
            tool: r.get(1)?,
            path: r.get(2)?,
            project: r.get(3)?,
            title: r.get(4)?,
            started: r.get(5)?,
            msg_count: r.get(6)?,
            kind: r.get(7)?,
            preview: r.get(8)?,
            summary: r.get(9)?,
            tags: r.get(10)?,
            archived: r.get(11)?,
            account: None,
        })
    })?;
    let mut out = rows.collect::<rusqlite::Result<Vec<_>>>()?;
    crate::account_link::annotate(out.iter_mut());
    Ok(out)
}

pub struct Hit {
    pub row: SessionRow,
    pub role: String,
    pub snippet: String,
}

#[derive(Clone)]
struct SearchTerm {
    text: String,
}

/// Split user text into whitespace-separated terms, keeping quoted text as one
/// phrase. An unmatched opening quote consumes the remainder of the query.
fn parse_search_terms(query: &str) -> Vec<SearchTerm> {
    fn push(terms: &mut Vec<SearchTerm>, text: String) {
        if !text.trim().is_empty() {
            terms.push(SearchTerm { text });
        }
    }

    let normalized = crate::util::nfc(query);
    let mut chars = normalized.chars().peekable();
    let mut terms = Vec::new();
    let mut current = String::new();
    let mut phrase = String::new();
    let mut in_phrase = false;

    while let Some(ch) = chars.next() {
        if in_phrase {
            if ch == '"' {
                if chars.peek() == Some(&'"') {
                    chars.next();
                    phrase.push('"');
                } else {
                    push(&mut terms, std::mem::take(&mut phrase));
                    in_phrase = false;
                }
            } else {
                phrase.push(ch);
            }
        } else if ch == '"' {
            if !current.is_empty() {
                push(&mut terms, std::mem::take(&mut current));
            }
            in_phrase = true;
        } else if ch.is_whitespace() {
            if !current.is_empty() {
                push(&mut terms, std::mem::take(&mut current));
            }
        } else {
            current.push(ch);
        }
    }
    if in_phrase {
        push(&mut terms, phrase);
    } else if !current.is_empty() {
        push(&mut terms, current);
    }
    terms
}

/// Return an MCP query rejection, if any. Trigram would otherwise let a remote
/// client trigger the bounded LIKE scan with a query made only of short terms;
/// word tokenizers can handle those terms directly in FTS.
pub fn mcp_search_guard(conn: &Connection, query: &str) -> Result<Option<&'static str>> {
    let terms = parse_search_terms(query);
    if terms.is_empty() {
        return Ok(Some("query must contain at least one search term"));
    }
    if is_trigram_tokenizer(&tokenizer_spec(conn)?)
        && !terms.iter().any(|term| term.text.chars().count() >= 3)
    {
        return Ok(Some(
            "trigram search requires at least one term of 3 or more characters",
        ));
    }
    Ok(None)
}

fn is_trigram_tokenizer(spec: &str) -> bool {
    spec.split_whitespace()
        .next()
        .is_some_and(|word| word.eq_ignore_ascii_case("trigram"))
}

/// Search each message for every query term, then return its best session hit.
/// Terms are ANDed in one message; double quotes keep their contents together
/// as a phrase. Trigram supports substring and CJK searches; other configured
/// tokenizers use their own word-search behavior.
pub fn search(
    conn: &Connection,
    query: &str,
    limit: usize,
    tool: Option<&str>,
    project: Option<&str>,
) -> Result<Vec<Hit>> {
    search_with_roles(conn, query, limit, tool, project, None)
}

/// Search messages and return each session's best matching hit, optionally
/// restricted to the supplied message roles. The role restriction is applied
/// before the candidate cap and before sessions are grouped.
pub fn search_with_roles(
    conn: &Connection,
    query: &str,
    limit: usize,
    tool: Option<&str>,
    project: Option<&str>,
    roles: Option<&[Role]>,
) -> Result<Vec<Hit>> {
    if roles.is_some_and(|roles| roles.is_empty()) {
        return Ok(Vec::new());
    }
    let filters = SearchFilters {
        tool,
        project,
        roles,
    };
    let terms = parse_search_terms(query);
    if terms.is_empty() {
        return Ok(Vec::new());
    }
    let trigram = is_trigram_tokenizer(&tokenizer_spec(conn)?);
    if trigram && terms.iter().all(|term| term.text.chars().count() < 3) {
        return search_like_terms(conn, &terms, limit, filters);
    }
    let fts_terms: Vec<SearchTerm> = if trigram {
        terms
            .iter()
            .filter(|term| term.text.chars().count() >= 3)
            .cloned()
            .collect()
    } else {
        // unicode61, porter, and ascii tokenize punctuation as separators. A
        // punctuation-only phrase has no FTS tokens and must not form an empty
        // MATCH phrase. Dropping it also makes an all-punctuation query empty.
        terms
            .iter()
            .filter(|term| term.text.chars().any(char::is_alphanumeric))
            .cloned()
            .collect()
    };
    if fts_terms.is_empty() {
        return Ok(Vec::new());
    }
    search_fts_terms(conn, &terms, &fts_terms, trigram, limit, filters)
}

#[derive(Clone, Copy)]
struct SearchFilters<'a> {
    tool: Option<&'a str>,
    project: Option<&'a str>,
    roles: Option<&'a [Role]>,
}

fn search_fts_terms(
    conn: &Connection,
    terms: &[SearchTerm],
    fts_terms: &[SearchTerm],
    trigram: bool,
    limit: usize,
    filters: SearchFilters<'_>,
) -> Result<Vec<Hit>> {
    let fts_query = fts_terms
        .iter()
        .map(|term| format!("\"{}\"", term.text.replace('"', "\"\"")))
        .collect::<Vec<_>>()
        .join(" ");

    // snippet()/rank only work in a plain FTS5 query context, not under joins
    // or GROUP BY, so match in a subquery and attach metadata outside.
    //
    // Tradeoff: only the top 4000 matching messages by rank reach grouping.
    // A very common query can omit sessions whose matches rank below that cap;
    // narrowing the terms surfaces the long tail.
    let mut sql = String::from(
        "SELECT f.session_id, f.tool, f.path, f.project, f.title, f.started, f.msg_count, f.kind,
                m.role, x.snip, min(x.rank) AS best, (f.archived_at IS NOT NULL)
         FROM (SELECT rowid AS mid,
                      snippet(msgs, 0, char(2), char(3), char(8230), 18) AS snip,
                      rank
               FROM msgs WHERE msgs MATCH ?",
    );
    let mut args: Vec<Box<dyn rusqlite::ToSql>> = vec![Box::new(fts_query)];
    if let Some(roles) = filters.roles {
        sql.push_str(
            " AND EXISTS (SELECT 1 FROM messages role_message
                          WHERE role_message.id = msgs.rowid AND role_message.role IN (",
        );
        for (index, role) in roles.iter().enumerate() {
            if index > 0 {
                sql.push_str(", ");
            }
            sql.push('?');
            args.push(Box::new(role.label().to_string()));
        }
        sql.push_str("))");
    }
    sql.push_str(
        " ORDER BY rank LIMIT 4000) x
         JOIN messages m ON m.id = x.mid
         JOIN files f ON f.session_id = m.session_id
         WHERE 1=1",
    );
    if trigram {
        for term in terms.iter().filter(|term| term.text.chars().count() < 3) {
            sql.push_str(" AND m.text LIKE ? ESCAPE '\\'");
            args.push(Box::new(format!("%{}%", escape_like(&term.text))));
        }
    }
    if let Some(t) = filters.tool {
        sql.push_str(" AND f.tool = ?");
        args.push(Box::new(t.to_string()));
    }
    if let Some(p) = filters.project {
        sql.push_str(" AND f.project LIKE ?");
        args.push(Box::new(format!("%{}%", crate::util::nfc(p))));
    }
    sql.push_str(&format!(
        " GROUP BY f.session_id ORDER BY best LIMIT {limit}"
    ));

    let mut stmt = conn.prepare(&sql)?;
    let rows = stmt.query_map(
        rusqlite::params_from_iter(args.iter().map(|arg| arg.as_ref())),
        |r| {
            Ok(Hit {
                row: SessionRow {
                    last_active: None,
                    session_id: r.get(0)?,
                    tool: r.get(1)?,
                    path: r.get(2)?,
                    project: r.get(3)?,
                    title: r.get(4)?,
                    started: r.get(5)?,
                    msg_count: r.get(6)?,
                    kind: r.get(7)?,
                    preview: None,
                    summary: None,
                    tags: None,
                    archived: r.get(11)?,
                    account: None,
                },
                role: r.get(8)?,
                snippet: r.get(9)?,
            })
        },
    )?;
    let mut out = rows.collect::<rusqlite::Result<Vec<_>>>()?;
    crate::account_link::annotate(out.iter_mut().map(|hit| &mut hit.row));
    Ok(out)
}

/// Substring search for terms below the trigram tokenizer's three-character
/// floor. All terms must match the same messages.text row.
///
/// Perf: SQLite scans messages to find matching rows, then the newest 50,000
/// matches reach session grouping. A session with no match among those newest
/// candidates can be missed. With trigram, the unified search dispatcher uses
/// this only when every term is short; any longer term stays on FTS. Other
/// tokenizer specs use FTS for short terms too. LIKE has no rank, so results
/// are ordered by recency.
pub fn search_like(
    conn: &Connection,
    query: &str,
    limit: usize,
    tool: Option<&str>,
    project: Option<&str>,
) -> Result<Vec<Hit>> {
    let terms = parse_search_terms(query);
    if terms.is_empty() {
        return Ok(Vec::new());
    }
    search_like_terms(
        conn,
        &terms,
        limit,
        SearchFilters {
            tool,
            project,
            roles: None,
        },
    )
}

fn search_like_terms(
    conn: &Connection,
    terms: &[SearchTerm],
    limit: usize,
    filters: SearchFilters<'_>,
) -> Result<Vec<Hit>> {
    const SCAN_CAP: i64 = 50_000;

    let mut sql = String::from(
        "SELECT f.session_id, f.tool, f.path, f.project, f.title, f.started, f.msg_count, f.kind,
                x.role, x.text, (f.archived_at IS NOT NULL)
         FROM (SELECT m.session_id AS sid, m.role AS role, m.text AS text, m.id AS mid
               FROM messages m WHERE 1=1",
    );
    let mut args: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
    if let Some(roles) = filters.roles {
        sql.push_str(" AND m.role IN (");
        for (index, role) in roles.iter().enumerate() {
            if index > 0 {
                sql.push_str(", ");
            }
            sql.push('?');
            args.push(Box::new(role.label().to_string()));
        }
        sql.push(')');
    }
    for term in terms {
        sql.push_str(" AND m.text LIKE ? ESCAPE '\\'");
        args.push(Box::new(format!("%{}%", escape_like(&term.text))));
    }
    sql.push_str(" ORDER BY m.id DESC LIMIT ?)");
    args.push(Box::new(SCAN_CAP));
    sql.push_str(
        " x JOIN files f ON f.session_id = x.sid
         WHERE 1=1",
    );
    if let Some(t) = filters.tool {
        sql.push_str(" AND f.tool = ?");
        args.push(Box::new(t.to_string()));
    }
    if let Some(p) = filters.project {
        sql.push_str(" AND f.project LIKE ?");
        args.push(Box::new(format!("%{}%", crate::util::nfc(p))));
    }
    sql.push_str(&format!(
        " GROUP BY f.session_id ORDER BY max(x.mid) DESC LIMIT {limit}"
    ));

    let mut stmt = conn.prepare(&sql)?;
    let rows = stmt.query_map(
        rusqlite::params_from_iter(args.iter().map(|arg| arg.as_ref())),
        |r| {
            let text: String = r.get(9)?;
            Ok(Hit {
                row: SessionRow {
                    last_active: None,
                    session_id: r.get(0)?,
                    tool: r.get(1)?,
                    path: r.get(2)?,
                    project: r.get(3)?,
                    title: r.get(4)?,
                    started: r.get(5)?,
                    msg_count: r.get(6)?,
                    kind: r.get(7)?,
                    preview: None,
                    summary: None,
                    tags: None,
                    archived: r.get(10)?,
                    account: None,
                },
                role: r.get(8)?,
                snippet: snippet_around(&text, terms),
            })
        },
    )?;
    let mut out = rows.collect::<rusqlite::Result<Vec<_>>>()?;
    crate::account_link::annotate(out.iter_mut().map(|hit| &mut hit.row));
    Ok(out)
}

/// Build a short raw-text window around the first LIKE term, highlighting every
/// term occurrence in that window. Matching folds case, while a character map
/// keeps highlights aligned to the unmodified transcript.
fn snippet_around(text: &str, terms: &[SearchTerm]) -> String {
    const WINDOW: usize = 36;
    let chars: Vec<char> = text.chars().collect();
    let mut folded = Vec::new();
    let mut source = Vec::new();
    for (index, ch) in chars.iter().copied().enumerate() {
        for lower in ch.to_lowercase() {
            folded.push(lower);
            source.push((index, index + 1));
        }
    }

    let mut matches = Vec::new();
    for term in terms {
        let needle: Vec<char> = term.text.chars().flat_map(char::to_lowercase).collect();
        if needle.is_empty() || needle.len() > folded.len() {
            continue;
        }
        for start in 0..=folded.len() - needle.len() {
            let end = start + needle.len();
            if folded[start..end] == needle[..] {
                matches.push((source[start].0, source[end - 1].1));
            }
        }
    }

    let Some(&(anchor_start, anchor_end)) = matches.iter().min_by_key(|range| range.0) else {
        return chars
            .iter()
            .take(WINDOW * 2)
            .collect::<String>()
            .replace('\n', " ");
    };
    let lo = anchor_start.saturating_sub(WINDOW);
    let hi = (anchor_end + WINDOW).min(chars.len());
    matches.retain(|(start, end)| *start >= lo && *end <= hi);
    matches.sort_unstable();

    let mut ranges: Vec<(usize, usize)> = Vec::new();
    for (start, end) in matches {
        if let Some(previous) = ranges.last_mut() {
            if start < previous.1 {
                previous.1 = previous.1.max(end);
                continue;
            }
        }
        ranges.push((start, end));
    }

    let mut out = String::new();
    if lo > 0 {
        out.push('\u{2026}');
    }
    let mut cursor = lo;
    for (start, end) in ranges {
        out.extend(&chars[cursor..start]);
        out.push('\u{2}');
        out.extend(&chars[start..end]);
        out.push('\u{3}');
        cursor = end;
    }
    out.extend(&chars[cursor..hi]);
    if hi < chars.len() {
        out.push('\u{2026}');
    }
    out.replace('\n', " ")
}

/// Escape LIKE metacharacters ('\' first) so an id/path fragment like "%" or
/// "_" can't turn a prefix match into a wildcard that resolves every session.
fn escape_like(s: &str) -> String {
    s.replace('\\', "\\\\")
        .replace('%', "\\%")
        .replace('_', "\\_")
}

/// The SELECT column list every id-resolution query shares, in the order
/// [`map_resolve_row`] reads.
const RESOLVE_COLS: &str = "session_id, tool, path, project, title, started, msg_count, kind";

/// Map a resolve-query row to a [`SessionRow`]. Callers must SELECT
/// [`RESOLVE_COLS`] followed by `{SUMMARY_SQL}, {TAGS_SQL}, (archived_at IS NOT NULL)`.
fn map_resolve_row(r: &rusqlite::Row) -> rusqlite::Result<SessionRow> {
    Ok(SessionRow {
        last_active: None,
        session_id: r.get(0)?,
        tool: r.get(1)?,
        path: r.get(2)?,
        project: r.get(3)?,
        title: r.get(4)?,
        started: r.get(5)?,
        msg_count: r.get(6)?,
        kind: r.get(7)?,
        preview: None,
        summary: r.get(8)?,
        tags: r.get(9)?,
        archived: r.get(10)?,
        account: None,
    })
}

/// Resolve a (possibly abbreviated) session id to its file row(s). Matches on
/// the sessionwiki short id (prefix) AND on the tool's own native id (the codex
/// rollout / claude transcript UUID, full or prefix), so a harness "tower" row -
/// which knows only the native id - can be reopened directly. Short-id behavior
/// is unchanged: the native scan only runs for native-shaped queries, and never
/// displaces an existing short-id match on a plain-hex prefix (see
/// [`looks_like_native_prefix`]).
pub fn resolve(conn: &Connection, id_prefix: &str) -> Result<Vec<SessionRow>> {
    let mut out = resolve_by_short_id(conn, id_prefix)?;

    // Native-id join. A plain-hex prefix stays short-id-only when it already
    // matched something (no new ambiguity); a dashed prefix is unambiguously a
    // UUID (short ids have no dashes) so it always broadens to the native scan.
    let native_ok =
        looks_like_native_prefix(id_prefix) && (out.is_empty() || id_prefix.contains('-'));
    if native_ok {
        for row in resolve_by_native_id(conn, id_prefix)? {
            if out.len() >= 10 {
                break;
            }
            if out.iter().all(|r| r.session_id != row.session_id) {
                out.push(row);
            }
        }
    }
    Ok(out)
}

/// Prefix match on the sessionwiki short id (the historical `resolve`).
fn resolve_by_short_id(conn: &Connection, id_prefix: &str) -> Result<Vec<SessionRow>> {
    let mut stmt = conn.prepare(&format!(
        "SELECT {RESOLVE_COLS}, {SUMMARY_SQL}, {TAGS_SQL}, (archived_at IS NOT NULL)
         FROM files f WHERE session_id LIKE ?1 ESCAPE '\\' LIMIT 10",
    ))?;
    let pattern = format!("{}%", escape_like(id_prefix));
    let rows = stmt.query_map(params![pattern], map_resolve_row)?;
    Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
}

/// Match rows whose native id (derived from the filename) starts with `prefix`.
/// The native id is a substring of the stored path, so a coarse `path LIKE
/// '%prefix%'` yields a bounded superset which we then confirm per row - a
/// planted path fragment can never satisfy the exact `native_id_of` check.
fn resolve_by_native_id(conn: &Connection, prefix: &str) -> Result<Vec<SessionRow>> {
    let mut stmt = conn.prepare(&format!(
        "SELECT {RESOLVE_COLS}, {SUMMARY_SQL}, {TAGS_SQL}, (archived_at IS NOT NULL)
         FROM files f WHERE path LIKE ?1 ESCAPE '\\' LIMIT 500",
    ))?;
    let pattern = format!("%{}%", escape_like(prefix));
    let want = prefix.to_ascii_lowercase();
    let rows = stmt.query_map(params![pattern], map_resolve_row)?;
    let mut out = Vec::new();
    for row in rows {
        let row = row?;
        if native_id_of(&row.path).is_some_and(|n| n.starts_with(&want)) {
            out.push(row);
            if out.len() >= 10 {
                break;
            }
        }
    }
    Ok(out)
}

/// Locate a session file directly on disk by its NATIVE id (codex rollout UUID
/// or claude transcript UUID), full or prefix, WITHOUT the index. This is the
/// live-session path: a session started moments ago may not be indexed yet, but
/// its file already exists under the tool's store root, so `session_window` /
/// `show` can still open it in one call. Scans only the codex and claude roots -
/// the two tools whose native id a harness tower knows - and returns the first
/// (tool, path) whose filename-derived native id matches. None if nothing on
/// disk matches (or the query is not native-shaped).
pub fn locate_by_native_id(prefix: &str) -> Option<(String, PathBuf)> {
    if !looks_like_native_prefix(prefix) {
        return None;
    }
    let want = prefix.to_ascii_lowercase();
    for name in ["claude-code", "codex"] {
        let Some(adapter) = adapters::by_name(name) else {
            continue;
        };
        let Some(root) = adapter.root() else { continue };
        if !root.exists() {
            continue;
        }
        let hit = walkdir::WalkDir::new(&root)
            .into_iter()
            .filter_map(std::result::Result::ok)
            .find(|e| {
                e.file_type().is_file()
                    && e.path().extension().is_some_and(|x| x == "jsonl")
                    && native_id_of(&e.path().to_string_lossy())
                        .is_some_and(|n| n.starts_with(&want))
            });
        if let Some(e) = hit {
            return Some((name.to_string(), e.into_path()));
        }
    }
    None
}

/// A minimal, un-indexed [`SessionRow`] for a live session located on disk by
/// [`locate_by_native_id`]. The real path and tool drive a direct parse (via
/// `load_session`); the metadata fields are placeholders the window/show render
/// path does not consult (it reads the parsed transcript). The `session_id`
/// matches the id the adapter would assign, so a later sync reconciles cleanly.
pub fn live_row(tool: String, path: PathBuf) -> SessionRow {
    let path = path.to_string_lossy().into_owned();
    SessionRow {
        last_active: None,
        session_id: crate::util::short_id(&path),
        tool,
        path,
        project: String::new(),
        title: String::new(),
        started: None,
        msg_count: 0,
        kind: "main".into(),
        preview: None,
        summary: None,
        tags: None,
        archived: false,
        account: None,
    }
}

/// Store (or replace) the cached synopsis for a session. The synopsis comes from
/// the user's own LLM over the raw transcript, so it can echo a secret - redact
/// before it lands in this durable table.
pub fn set_summary(conn: &Connection, session_id: &str, summary: &str) -> Result<()> {
    conn.execute(
        "INSERT OR REPLACE INTO summaries(session_id, summary, created)
         VALUES (?1, ?2, datetime('now'))",
        params![session_id, crate::redact::redact(summary).as_ref()],
    )?;
    Ok(())
}

/// Most recent main sessions that have no cached summary yet.
pub fn unsummarized(
    conn: &Connection,
    limit: usize,
    tool: Option<&str>,
) -> Result<Vec<SessionRow>> {
    let mut sql = format!(
        "SELECT session_id, tool, path, project, title, started, msg_count, kind, {PREVIEW_SQL}, {SUMMARY_SQL}, {TAGS_SQL}, (archived_at IS NOT NULL)
         FROM files f
         WHERE kind = 'main' AND NOT EXISTS
               (SELECT 1 FROM summaries s WHERE s.session_id = f.session_id)",
    );
    let mut args: Vec<String> = Vec::new();
    if let Some(t) = tool {
        sql.push_str(" AND tool = ?");
        args.push(t.to_string());
    }
    sql.push_str(&format!(" ORDER BY started DESC LIMIT {limit}"));
    let mut stmt = conn.prepare(&sql)?;
    let rows = stmt.query_map(rusqlite::params_from_iter(args), |r| {
        Ok(SessionRow {
            last_active: None,
            session_id: r.get(0)?,
            tool: r.get(1)?,
            path: r.get(2)?,
            project: r.get(3)?,
            title: r.get(4)?,
            started: r.get(5)?,
            msg_count: r.get(6)?,
            kind: r.get(7)?,
            preview: r.get(8)?,
            summary: r.get(9)?,
            tags: r.get(10)?,
            archived: r.get(11)?,
            account: None,
        })
    })?;
    Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
}

// --- curation (the editable wiki layer) ---

/// One canonical form per tag: trimmed, lowercased, NFC-normalized - the same
/// normalization every other indexed string gets, so a tag typed in decomposed
/// form (macOS IME, some CJK inputs) matches on add, filter, and remove alike.
fn norm_tag(tag: &str) -> String {
    crate::util::nfc(&tag.trim().to_lowercase())
}

pub fn add_tag(conn: &Connection, session_id: &str, tag: &str) -> Result<()> {
    // Tags are joined with ',' at read time and the JSON/web contract splits on
    // it, so a comma inside a tag would corrupt the array into two elements.
    // Reject it (and the empty tag) at the input boundary.
    let tag = norm_tag(tag);
    if tag.is_empty() || tag.contains(',') {
        anyhow::bail!("a tag must be non-empty and contain no commas");
    }
    conn.execute(
        "INSERT OR IGNORE INTO tags(session_id, tag) VALUES (?1, ?2)",
        params![session_id, tag],
    )?;
    Ok(())
}

pub fn remove_tag(conn: &Connection, session_id: &str, tag: &str) -> Result<usize> {
    Ok(conn.execute(
        "DELETE FROM tags WHERE session_id = ?1 AND tag = ?2",
        params![session_id, norm_tag(tag)],
    )?)
}

pub fn set_note(conn: &Connection, session_id: &str, note: &str) -> Result<()> {
    conn.execute(
        "INSERT OR REPLACE INTO notes(session_id, note, updated)
         VALUES (?1, ?2, datetime('now'))",
        params![session_id, note],
    )?;
    Ok(())
}

pub fn note_for(conn: &Connection, session_id: &str) -> Result<Option<String>> {
    Ok(conn
        .query_row(
            "SELECT note FROM notes WHERE session_id = ?1",
            params![session_id],
            |r| r.get(0),
        )
        .ok())
}

/// All tags in use, with how many sessions carry each.
pub fn tag_counts(conn: &Connection) -> Result<Vec<(String, i64)>> {
    let mut stmt =
        conn.prepare("SELECT tag, count(*) FROM tags GROUP BY tag ORDER BY count(*) DESC, tag")?;
    let rows = stmt.query_map([], |r| Ok((r.get(0)?, r.get(1)?)))?;
    Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
}

// --- provenance: sessions <-> the code they produced ---

/// Files a session edited or created, in the order it first touched them.
pub fn files_for(conn: &Connection, session_id: &str) -> Result<Vec<String>> {
    let mut stmt = conn.prepare("SELECT path FROM touched WHERE session_id = ?1 ORDER BY rowid")?;
    let rows = stmt.query_map(params![session_id], |r| r.get(0))?;
    Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
}

/// One recorded edit to a file - the evidence layer behind `touched`. `matched_path`
/// is the absolute stored path the query resolved to.
#[derive(Debug, Serialize)]
pub struct FileEdit {
    pub session_id: String,
    pub kind: String,
    pub ts: Option<String>,
    pub snippet: String,
    pub matched_path: String,
}

/// The concrete edits made to a file, newest first - the evidence for "why does
/// this file look like this". Resolves a relative path against the absolute one
/// on disk by suffix, the same match `sessions_for_file` uses, so
/// `edits_for("src/auth.rs")` finds `/home/me/proj/src/auth.rs`.
/// A `usize` limit clamped to a positive `i64` for SQLite: a value past i64::MAX
/// wraps negative, which SQLite reads as "unlimited" - defeating the memory bound.
fn sql_limit(n: usize) -> i64 {
    i64::try_from(n).unwrap_or(i64::MAX)
}

pub fn edits_for(conn: &Connection, query: &str, limit: usize) -> Result<Vec<FileEdit>> {
    let q = crate::util::nfc(query.trim().trim_start_matches("./"));
    // Escape LIKE metacharacters so a caller-supplied `%`/`_` matches literally.
    let esc = q
        .replace('\\', "\\\\")
        .replace('%', "\\%")
        .replace('_', "\\_");
    let suffix = format!("%/{esc}");
    let mut stmt = conn.prepare(
        "SELECT session_id, kind, ts, snippet, path
         FROM edits
         WHERE path = ?1 OR path LIKE ?2 ESCAPE '\\'
            OR (length(?1) > length(path)
                AND substr(?1, -length(path)) = path
                AND substr(?1, -length(path)-1, 1) = '/')
         ORDER BY ts DESC, rowid DESC LIMIT ?3",
    )?;
    let rows = stmt.query_map(params![q, suffix, sql_limit(limit)], |r| {
        Ok(FileEdit {
            session_id: r.get(0)?,
            kind: r.get(1)?,
            ts: r.get(2)?,
            snippet: r.get(3)?,
            matched_path: r.get(4)?,
        })
    })?;
    Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
}

/// A single session's edits to a file, newest first - scoped by session so no
/// session's edits can be starved by another's under a shared cap. Uses the SAME
/// suffix path match as `sessions_for_file`, so a session that edited the file
/// under more than one spelling (abs + relative) contributes ALL its edits, not
/// just the one spelling `sessions_for_file`'s GROUP BY happened to pick.
pub fn edits_for_session(
    conn: &Connection,
    session_id: &str,
    query: &str,
    limit: usize,
) -> Result<Vec<FileEdit>> {
    let q = crate::util::nfc(query.trim().trim_start_matches("./"));
    let esc = q
        .replace('\\', "\\\\")
        .replace('%', "\\%")
        .replace('_', "\\_");
    let suffix = format!("%/{esc}");
    let mut stmt = conn.prepare(
        "SELECT session_id, kind, ts, snippet, path
         FROM edits
         WHERE session_id = ?1
           AND (path = ?2 OR path LIKE ?3 ESCAPE '\\'
                OR (length(?2) > length(path)
                    AND substr(?2, -length(path)) = path
                    AND substr(?2, -length(path)-1, 1) = '/'))
         ORDER BY ts DESC, rowid DESC LIMIT ?4",
    )?;
    let rows = stmt.query_map(params![session_id, q, suffix, sql_limit(limit)], |r| {
        Ok(FileEdit {
            session_id: r.get(0)?,
            kind: r.get(1)?,
            ts: r.get(2)?,
            snippet: r.get(3)?,
            matched_path: r.get(4)?,
        })
    })?;
    Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
}

/// One session's slice of a file's history: its metadata plus the edits it made
/// to this file.
#[derive(Serialize)]
pub struct SessionEvidence {
    pub session: SessionRow,
    pub edits: Vec<FileEdit>,
}

/// A file's full evidence chain - the sessions that edited it, newest first,
/// each carrying its own edits. What the file-history page renders.
#[derive(Serialize)]
pub struct FileHistory {
    pub path: String,
    pub sessions: Vec<SessionEvidence>,
}

/// Assemble a file's evidence chain: the sessions that touched it (with metadata,
/// newest first) joined to each session's recorded edits. Sessions that touched
/// the file but carry no structured edits (other adapters, archived sessions)
/// appear with an empty `edits` list - the touch is still evidence.
pub fn evidence_for(conn: &Connection, path: &str, limit: usize) -> Result<FileHistory> {
    let sessions = sessions_for_file(conn, path, limit)?;
    // Fetch edits PER session (scoped by session, matched by the SAME query path
    // so every spelling counts), so one session's edits are never starved by
    // another's - and `limit == 0` does no edit query.
    const PER_SESSION_EDIT_CAP: usize = 500;
    let sessions = sessions
        .into_iter()
        .map(|(session, _matched)| {
            let edits = edits_for_session(conn, &session.session_id, path, PER_SESSION_EDIT_CAP)?;
            Ok(SessionEvidence { session, edits })
        })
        .collect::<Result<Vec<_>>>()?;
    Ok(FileHistory {
        path: path.to_string(),
        sessions,
    })
}

/// Sessions that touched a file, newest first - the reverse provenance link.
/// Matches either the exact stored path or any stored path ending in the
/// query, so a relative `src/auth.rs` finds `/home/me/proj/src/auth.rs`. The
/// matched stored path is returned alongside each session.
/// The file name to retry with when a full path traces to nothing.
///
/// Folders get renamed. A session recorded `~/Project/lunch/diag.py`; the folder
/// is `~/Project/slack` now, so tracing by the path the file has TODAY matched
/// nothing and reported that no session had touched it - about a file whose
/// whole history was in the index under its old directory.
///
/// The name is the part that survives a move. `None` when there is nothing to
/// fall back to: a bare name would just repeat the same miss, and a trailing
/// slash names a directory rather than a file.
pub fn basename_fallback(query: &str) -> Option<String> {
    let q = query.trim();
    if q.is_empty() || q.ends_with('/') {
        return None;
    }
    let (head, name) = q.rsplit_once('/')?;
    (!head.is_empty() && !name.is_empty()).then(|| name.to_string())
}

pub fn sessions_for_file(
    conn: &Connection,
    query: &str,
    limit: usize,
) -> Result<Vec<(SessionRow, String)>> {
    let q = crate::util::nfc(query.trim().trim_start_matches("./"));
    // Escape LIKE metacharacters: the query is a caller-supplied path (the MCP
    // trace_file arg included), so a bare `%` must match a literal `%`, not act
    // as a wildcard that enumerates the whole index.
    let esc = q
        .replace('\\', "\\\\")
        .replace('%', "\\%")
        .replace('_', "\\_");
    let suffix = format!("%/{esc}");
    let mut stmt = conn.prepare(&format!(
        "SELECT f.session_id, f.tool, f.path, f.project, f.title, f.started, f.msg_count, f.kind,
                {SUMMARY_SQL}, {TAGS_SQL}, t.path, (f.archived_at IS NOT NULL)
         FROM touched t JOIN files f ON f.session_id = t.session_id
         WHERE t.path = ?1 OR t.path LIKE ?2 ESCAPE '\\'
            OR (length(?1) > length(t.path)
                AND substr(?1, -length(t.path)) = t.path
                AND substr(?1, -length(t.path)-1, 1) = '/')
         GROUP BY f.session_id
         ORDER BY f.started DESC, f.session_id LIMIT ?3"
    ))?;
    let rows = stmt.query_map(params![q, suffix, sql_limit(limit)], |r| {
        Ok((
            SessionRow {
                last_active: None,
                session_id: r.get(0)?,
                tool: r.get(1)?,
                path: r.get(2)?,
                project: r.get(3)?,
                title: r.get(4)?,
                started: r.get(5)?,
                msg_count: r.get(6)?,
                kind: r.get(7)?,
                preview: None,
                summary: r.get(8)?,
                tags: r.get(9)?,
                archived: r.get(11)?,
                account: None,
            },
            r.get::<_, String>(10)?,
        ))
    })?;
    let mut out = rows.collect::<rusqlite::Result<Vec<_>>>()?;
    crate::account_link::annotate(out.iter_mut().map(|(r, _)| r));
    Ok(out)
}

/// Like `sessions_for_file` but returns the start/end epoch window and project
/// that blame's commit->session attribution needs. Matches both an exact stored
/// path and any stored path ending in the query suffix, so a repo-relative query
/// (e.g. `src/auth.rs`) catches Claude Code's absolute touched paths and Codex's
/// relative ones alike (NFC-normalized).
pub fn sessions_touching(
    conn: &Connection,
    query: &str,
) -> Result<Vec<crate::blame::TouchingSession>> {
    let q = crate::util::nfc(query.trim().trim_start_matches("./"));
    let suffix = format!("%/{}", escape_like(&q));
    let mut stmt = conn.prepare(
        "SELECT f.session_id, f.tool, f.title, f.project, f.started, f.ended, (f.archived_at IS NOT NULL)
         FROM touched t JOIN files f ON f.session_id = t.session_id
         WHERE t.path = ?1 OR t.path LIKE ?2 ESCAPE '\\'
         GROUP BY f.session_id",
    )?;
    let rows = stmt.query_map(params![q, suffix], |r| {
        let started: Option<String> = r.get(4)?;
        let ended: Option<String> = r.get(5)?;
        Ok(crate::blame::TouchingSession {
            session_id: r.get(0)?,
            tool: r.get(1)?,
            title: r.get(2)?,
            project: r.get(3)?,
            started: started.as_deref().and_then(to_epoch),
            ended: ended.as_deref().and_then(to_epoch),
            archived: r.get(6)?,
        })
    })?;
    Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
}

/// Parse a stored timestamp (RFC3339) to epoch seconds; None if unparseable.
fn to_epoch(s: &str) -> Option<i64> {
    chrono::DateTime::parse_from_rfc3339(s)
        .ok()
        .map(|d| d.timestamp())
}

// --- archive: serving and forgetting sessions whose originals are gone ---

/// The display name for a tool this binary's adapter registry does not know.
///
/// A program that embeds this crate can register its own adapters, so rows in
/// the index may name a tool the standalone binary has never heard of. Those
/// rows used to print as "unknown". `Session.tool` is `&'static str`, so the
/// row's own string has to outlive the call: intern it once and leak it. The
/// set of tool names is small and fixed by the tools a user actually runs, so
/// the leak is bounded by that, not by the number of sessions.
fn interned_tool(name: &str) -> &'static str {
    use std::collections::HashSet;
    use std::sync::{Mutex, OnceLock};
    static NAMES: OnceLock<Mutex<HashSet<&'static str>>> = OnceLock::new();
    let mut names = NAMES
        .get_or_init(|| Mutex::new(HashSet::new()))
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    if let Some(existing) = names.get(name) {
        return existing;
    }
    let leaked: &'static str = Box::leak(name.to_owned().into_boxed_str());
    names.insert(leaked);
    leaked
}

/// Reconstruct an archived or external-adapter session from its indexed copy.
/// Per-message timestamps are omitted; full tool output is present only when
/// the index's retention mode was `full` when that session was indexed.
pub fn session_from_index(conn: &Connection, row: &SessionRow) -> Result<crate::model::Session> {
    use crate::model::{Message, Role, ToolEvent, ToolSummary};
    let mut stmt = conn
        .prepare("SELECT role, text, full_text FROM messages WHERE session_id = ?1 ORDER BY id")?;
    let messages: Vec<Message> = stmt
        .query_map(params![row.session_id], |r| {
            let role: String = r.get(0)?;
            let text: String = r.get(1)?;
            let full_text: Option<String> = r.get(2)?;
            let role = match role.as_str() {
                "user" => Role::User,
                "assistant" => Role::Assistant,
                _ => Role::Tool,
            };
            let tool = full_text.and_then(|output| {
                (role == Role::Tool && crate::tool_summary::is_compact_summary_line(&text))
                    .then_some(ToolEvent::Summary(ToolSummary {
                        name: "tool".into(),
                        args: serde_json::Value::Null,
                        output: Some(output),
                        is_error: None,
                        lines: None,
                    }))
            });
            Ok(Message {
                role,
                text,
                ts: None,
                tool,
            })
        })?
        .collect::<rusqlite::Result<_>>()?;
    drop(stmt);

    let tool = adapters::by_name(&row.tool)
        .map(|a| a.name())
        .unwrap_or_else(|| interned_tool(&row.tool));
    let started = row
        .started
        .as_deref()
        .and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok())
        .map(|t| t.with_timezone(&chrono::Utc));
    Ok(crate::model::Session {
        id: row.session_id.clone(),
        tool,
        // A shared-store key carries a U+001F separator in the stored path;
        // render it human-readable (`<db>#<id>`) so it never leaks into show /
        // brief / web. Real file paths never contain U+001F, so this is a no-op
        // for every other adapter.
        path: std::path::PathBuf::from(row.path.replace('\u{1f}', "#")),
        project: row.project.clone(),
        started,
        ended: started,
        title: row.title.clone(),
        subagent: row.kind == "sub",
        messages,
        touched: files_for(conn, &row.session_id)?,
        edits: Vec::new(),
    })
}

/// Permanently remove a session from the index AND the archive - the only way
/// to undo archiving for a session the user genuinely wants gone. Curation for
/// it (tags/notes/summary) goes too, since the session no longer exists here.
/// All-or-nothing: a crash mid-forget must not leave the FTS index out of sync
/// with `messages`, nor an `archive` row that would resurrect it on rebuild.
pub fn forget(conn: &mut Connection, session_id: &str) -> Result<()> {
    let tx = conn.transaction()?;
    delete_session_msgs(&tx, session_id)?;
    for table in [
        "files",
        "touched",
        "edits",
        "archive",
        "summaries",
        "tags",
        "notes",
    ] {
        tx.execute(
            &format!("DELETE FROM {table} WHERE session_id = ?1"),
            params![session_id],
        )?;
    }
    tx.commit()?;
    Ok(())
}

// --- related sessions (backlinks) ---

/// Sessions related to `session_id`. A session is most usefully "related" to
/// the others about the same codebase, so same-project sessions are the spine;
/// sessions sharing a user tag are layered on as explicit links. Both are
/// indexed lookups, so this is instant even over a large store - the earlier
/// full-text-on-title approach was both slow and noisy (generic title words
/// like "session" matched everything).
pub fn related(conn: &Connection, session_id: &str, limit: usize) -> Result<Vec<SessionRow>> {
    let Some(target) = resolve(conn, session_id)?.into_iter().next() else {
        return Ok(vec![]);
    };
    let target_tags: Vec<String> = target
        .tags
        .as_deref()
        .map(|t| t.split(',').map(String::from).collect())
        .unwrap_or_default();

    let mut out: Vec<SessionRow> = Vec::new();
    let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
    seen.insert(target.session_id.clone());

    // 1. same project (exact), most recent first - the same-context spine.
    if !target.project.is_empty() {
        let sql = format!(
            "SELECT session_id, tool, path, project, title, started, msg_count, kind,
                    {PREVIEW_SQL}, {SUMMARY_SQL}, {TAGS_SQL}, (archived_at IS NOT NULL)
             FROM files f
             WHERE kind = 'main' AND project = ?1 AND session_id != ?2
             ORDER BY started DESC LIMIT ?3"
        );
        let mut stmt = conn.prepare(&sql)?;
        let rows = stmt.query_map(
            params![target.project, target.session_id, limit as i64 + 1],
            map_row,
        )?;
        for row in rows {
            let row = row?;
            if seen.insert(row.session_id.clone()) {
                out.push(row);
            }
        }
    }

    // 2. sessions that edited a file this one also edited - the strongest
    //    signal that two sessions are about the same work, and one no other
    //    session viewer has, since it comes from the provenance link.
    if out.len() < limit {
        let sql = format!(
            "SELECT DISTINCT f.session_id, f.tool, f.path, f.project, f.title, f.started,
                    f.msg_count, f.kind, {PREVIEW_SQL}, {SUMMARY_SQL}, {TAGS_SQL}, (f.archived_at IS NOT NULL)
             FROM touched a
             JOIN touched b ON a.path = b.path AND b.session_id != a.session_id
             JOIN files f ON f.session_id = b.session_id
             WHERE a.session_id = ?1 AND f.kind = 'main'
             ORDER BY f.started DESC LIMIT 50"
        );
        let mut stmt = conn.prepare(&sql)?;
        let rows = stmt.query_map(params![target.session_id], map_row)?;
        for row in rows {
            let row = row?;
            if seen.insert(row.session_id.clone()) {
                out.push(row);
                if out.len() >= limit {
                    break;
                }
            }
        }
    }

    // 3. sessions that share a tag with the target (explicit wiki links).
    if out.len() < limit && !target_tags.is_empty() {
        let placeholders = target_tags
            .iter()
            .map(|_| "?")
            .collect::<Vec<_>>()
            .join(",");
        let sql = format!(
            "SELECT DISTINCT f.session_id, f.tool, f.path, f.project, f.title, f.started,
                    f.msg_count, f.kind, {PREVIEW_SQL}, {SUMMARY_SQL}, {TAGS_SQL}, (f.archived_at IS NOT NULL)
             FROM files f JOIN tags t ON t.session_id = f.session_id
             WHERE f.kind = 'main' AND t.tag IN ({placeholders})
             ORDER BY f.started DESC LIMIT 50"
        );
        let mut stmt = conn.prepare(&sql)?;
        let rows = stmt.query_map(rusqlite::params_from_iter(&target_tags), map_row)?;
        for row in rows {
            let row = row?;
            if seen.insert(row.session_id.clone()) {
                out.push(row);
                if out.len() >= limit {
                    break;
                }
            }
        }
    }

    out.truncate(limit);
    crate::account_link::annotate(out.iter_mut());
    Ok(out)
}

/// Row mapper for the full session-list column set.
fn map_row(r: &rusqlite::Row) -> rusqlite::Result<SessionRow> {
    Ok(SessionRow {
        last_active: None,
        session_id: r.get(0)?,
        tool: r.get(1)?,
        path: r.get(2)?,
        project: r.get(3)?,
        title: r.get(4)?,
        started: r.get(5)?,
        msg_count: r.get(6)?,
        kind: r.get(7)?,
        preview: r.get(8)?,
        summary: r.get(9)?,
        tags: r.get(10)?,
        archived: r.get(11)?,
        account: None,
    })
}

// --- session engineering: management views ---

pub struct ProjectRow {
    pub project: String,
    pub sessions: i64,
    pub messages: i64,
    pub oldest: Option<String>,
    pub newest: Option<String>,
}

/// One row per project (a wiki "category" page), busiest first. `newest` is
/// the last activity, on the session list's clock: a long session begun months
/// ago but worked in yesterday makes its project recent.
pub fn projects(conn: &Connection) -> Result<Vec<ProjectRow>> {
    let mut stmt = conn.prepare(
        "SELECT project, count(*), coalesce(sum(msg_count), 0), min(started),
                max(COALESCE(ended, started))
         FROM files WHERE kind = 'main' AND project != ''
         GROUP BY project ORDER BY count(*) DESC, max(COALESCE(ended, started)) DESC",
    )?;
    let rows = stmt.query_map([], |r| {
        Ok(ProjectRow {
            project: r.get(0)?,
            sessions: r.get(1)?,
            messages: r.get(2)?,
            oldest: r.get(3)?,
            newest: r.get(4)?,
        })
    })?;
    Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
}

pub struct Stats {
    pub per_tool: Vec<(String, i64, i64)>, // tool, sessions, messages
    pub per_month: Vec<(String, i64)>,     // YYYY-MM, sessions
    pub total_sessions: i64,
    pub total_messages: i64,
    pub projects: i64,
    pub tags: i64,
    pub summarized: i64,
    /// Distinct files linked to at least one session (provenance coverage).
    pub files: i64,
    /// Sessions kept after the tool deleted their originals (archive mode).
    pub archived: i64,
}

pub fn stats(conn: &Connection) -> Result<Stats> {
    let mut per_tool_stmt = conn.prepare(
        "SELECT tool, count(*), coalesce(sum(msg_count),0) FROM files WHERE kind='main'
         GROUP BY tool ORDER BY count(*) DESC",
    )?;
    let per_tool = per_tool_stmt
        .query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))?
        .collect::<rusqlite::Result<Vec<_>>>()?;

    let mut per_month_stmt = conn.prepare(
        "SELECT substr(started,1,7) AS ym, count(*) FROM files
         WHERE kind='main' AND started IS NOT NULL
         GROUP BY ym ORDER BY ym DESC LIMIT 12",
    )?;
    let per_month = per_month_stmt
        .query_map([], |r| Ok((r.get(0)?, r.get(1)?)))?
        .collect::<rusqlite::Result<Vec<_>>>()?;

    let one = |sql: &str| -> Result<i64> { Ok(conn.query_row(sql, [], |r| r.get(0))?) };
    Ok(Stats {
        per_tool,
        per_month,
        total_sessions: one("SELECT count(*) FROM files WHERE kind='main'")?,
        total_messages: one("SELECT coalesce(sum(msg_count),0) FROM files WHERE kind='main'")?,
        projects: one(
            "SELECT count(DISTINCT project) FROM files WHERE kind='main' AND project!=''",
        )?,
        tags: one("SELECT count(DISTINCT tag) FROM tags")?,
        summarized: one("SELECT count(*) FROM summaries")?,
        files: one("SELECT count(DISTINCT path) FROM touched")?,
        archived: one("SELECT count(*) FROM files WHERE kind='main' AND archived_at IS NOT NULL")?,
    })
}

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

    // Realistic native store paths: a Codex rollout (uuid trails a timestamp) and
    // a Claude Code transcript (uuid IS the filename), plus a subagent transcript.
    const CODEX: &str = "/home/u/.codex/sessions/2025/05/13/rollout-2025-05-13T18-19-30-0a000000-0000-4000-8000-000000000001.jsonl";
    const CODEX_UUID: &str = "0a000000-0000-4000-8000-000000000001";
    const CLAUDE: &str =
        "/home/u/.claude/projects/-home-u-proj/1b111111-1111-4111-8111-111111111111.jsonl";
    const CLAUDE_UUID: &str = "1b111111-1111-4111-8111-111111111111";
    const SUBAGENT: &str = "/home/u/.claude/projects/-x/1b111111-1111-4111-8111-111111111111/subagents/agent-2c222222-2222-4222-8222-222222222222.jsonl";
    const SUBAGENT_UUID: &str = "2c222222-2222-4222-8222-222222222222";

    fn mem() -> Connection {
        let c = Connection::open_in_memory().unwrap();
        c.execute_batch(
            "CREATE TABLE files(
                 path TEXT PRIMARY KEY, mtime INTEGER NOT NULL DEFAULT 0,
                 size INTEGER NOT NULL DEFAULT 0, session_id TEXT NOT NULL,
                 tool TEXT NOT NULL, project TEXT NOT NULL DEFAULT '',
                 title TEXT NOT NULL DEFAULT '', started TEXT, ended TEXT,
                 msg_count INTEGER NOT NULL DEFAULT 0,
                 kind TEXT NOT NULL DEFAULT 'main', archived_at TEXT);
             CREATE TABLE summaries(session_id TEXT PRIMARY KEY, summary TEXT NOT NULL, created TEXT NOT NULL);
             CREATE TABLE tags(session_id TEXT NOT NULL, tag TEXT NOT NULL, PRIMARY KEY(session_id, tag));",
        )
        .unwrap();
        c
    }

    /// Insert a files row and return the sessionwiki short id it was keyed by.
    fn seed(c: &Connection, tool: &str, path: &str) -> String {
        let sid = crate::util::short_id(path);
        c.execute(
            "INSERT INTO files(path, session_id, tool, msg_count) VALUES(?1,?2,?3,3)",
            params![path, sid, tool],
        )
        .unwrap();
        sid
    }

    #[test]
    fn native_id_extracted_per_tool() {
        assert_eq!(native_id_of(CODEX).as_deref(), Some(CODEX_UUID));
        assert_eq!(native_id_of(CLAUDE).as_deref(), Some(CLAUDE_UUID));
        // A subagent transcript resolves to its OWN uuid (scanned from the file
        // name), not the parent uuid in the directory above it.
        assert_eq!(native_id_of(SUBAGENT).as_deref(), Some(SUBAGENT_UUID));
        // Files with no uuid in the name have no native id (not every tool keys
        // sessions this way).
        assert_eq!(native_id_of("/x/opencode.db#session-42"), None);
    }

    #[test]
    fn native_id_uppercase_is_normalized_to_lowercase() {
        let p = "/a/b/AB000000-0000-4000-8000-0000000000FF.jsonl";
        assert_eq!(
            native_id_of(p).as_deref(),
            Some("ab000000-0000-4000-8000-0000000000ff")
        );
    }

    #[test]
    fn resolve_by_full_native_id() {
        let c = mem();
        let sid = seed(&c, "codex", CODEX);
        let hits = resolve(&c, CODEX_UUID).unwrap();
        assert_eq!(hits.len(), 1, "full native uuid resolves");
        assert_eq!(hits[0].session_id, sid);
    }

    #[test]
    fn resolve_by_native_prefix_hex_and_dashed() {
        let c = mem();
        let sid = seed(&c, "claude-code", CLAUDE);
        // First-group (8 hex) prefix.
        let a = resolve(&c, "1b111111").unwrap();
        assert_eq!(a.len(), 1, "8-hex native prefix resolves");
        assert_eq!(a[0].session_id, sid);
        // Dashed prefix past the first group.
        let b = resolve(&c, "1b111111-1111").unwrap();
        assert_eq!(b.len(), 1, "dashed native prefix resolves");
        assert_eq!(b[0].session_id, sid);
    }

    #[test]
    fn resolve_still_matches_short_id_unchanged() {
        let c = mem();
        let sid = seed(&c, "codex", CODEX);
        // Full short id and a short-id prefix both resolve (existing behavior).
        assert_eq!(resolve(&c, &sid).unwrap().len(), 1);
        assert_eq!(resolve(&c, &sid[..6]).unwrap()[0].session_id, sid);
    }

    #[test]
    fn short_id_lookup_does_not_run_the_native_scan() {
        // A 12-hex, dash-free string is short-id-shaped and must never trigger a
        // native scan (which would be a needless path scan and could add a
        // spurious collision). Guard the gate that governs it directly.
        assert!(!looks_like_native_prefix("abcdef012345"));
        // ... while genuine native shapes do pass.
        assert!(looks_like_native_prefix("0a000000"));
        assert!(looks_like_native_prefix(
            "0a000000-0000-4000-8000-000000000001"
        ));
        assert!(looks_like_native_prefix("0a000000-0000"));
        // Non-hex, too short, or empty never look native.
        assert!(!looks_like_native_prefix("zzz"));
        assert!(!looks_like_native_prefix("a1"));
        assert!(!looks_like_native_prefix(""));
    }

    #[test]
    fn native_prefix_never_hides_an_existing_short_id_match() {
        // A plain-hex prefix that already matched a short id stays short-id-only:
        // even if some other session's native uuid also starts with those hex
        // digits, the plain-hex query keeps the established (short-id) result.
        let c = mem();
        // Seed a session whose SHORT id begins with the same 8 hex as another
        // session's native uuid, and the native one too.
        let native_path =
            "/home/u/.codex/sessions/2025/01/01/rollout-2025-01-01T00-00-00-deadbeef-0000-4000-8000-000000000009.jsonl";
        seed(&c, "codex", native_path);
        // Force a files row whose short id we control to start with "deadbeef".
        c.execute(
            "INSERT INTO files(path, session_id, tool, msg_count) VALUES('/synthetic', 'deadbeef1234', 'codex', 1)",
            [],
        )
        .unwrap();
        let hits = resolve(&c, "deadbeef").unwrap();
        // Short id matched, so the query stays short-id-only: exactly the one row.
        assert_eq!(hits.len(), 1);
        assert_eq!(hits[0].session_id, "deadbeef1234");
    }

    #[test]
    fn session_row_serializes_native_id_not_path() {
        let row = SessionRow {
            last_active: None,
            session_id: "abc123def456".into(),
            tool: "codex".into(),
            path: CODEX.into(),
            project: "proj".into(),
            title: "t".into(),
            started: None,
            msg_count: 3,
            kind: "main".into(),
            preview: None,
            summary: None,
            tags: None,
            archived: false,
            account: None,
        };
        let v = serde_json::to_value(&row).unwrap();
        assert_eq!(v["id"], "abc123def456");
        assert_eq!(v["native_id"], CODEX_UUID, "native_id present in JSON");
        assert!(
            v.get("path").is_none(),
            "the absolute path is never serialized"
        );
        // A pathless/uuid-less session serializes native_id as null, never a guess.
        let mut row2 = row;
        row2.path = "/x/opencode.db#s1".into();
        let v2 = serde_json::to_value(&row2).unwrap();
        assert!(v2["native_id"].is_null());
    }
}

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

    fn conn_with_edits() -> Connection {
        let c = Connection::open_in_memory().unwrap();
        c.execute_batch(
            "CREATE TABLE edits(session_id TEXT NOT NULL, path TEXT NOT NULL,
                 kind TEXT NOT NULL, ts TEXT, snippet TEXT NOT NULL);
             CREATE INDEX idx_edits_path ON edits(path);",
        )
        .unwrap();
        c
    }

    fn add(c: &Connection, sid: &str, path: &str, kind: &str, ts: &str, snip: &str) {
        c.execute(
            "INSERT INTO edits(session_id, path, kind, ts, snippet) VALUES(?1,?2,?3,?4,?5)",
            params![sid, path, kind, ts, snip],
        )
        .unwrap();
    }

    #[test]
    fn edits_for_returns_a_files_edits_by_suffix_newest_first() {
        let c = conn_with_edits();
        add(
            &c,
            "s1",
            "/home/me/proj/src/auth.rs",
            "edit",
            "2026-06-08T10:00:00Z",
            "let a = 1;",
        );
        add(
            &c,
            "s2",
            "/home/me/proj/src/auth.rs",
            "write",
            "2026-06-09T10:00:00Z",
            "fn main() {}",
        );
        add(
            &c,
            "s3",
            "/home/me/proj/src/other.rs",
            "edit",
            "2026-06-10T10:00:00Z",
            "nope",
        );

        // A relative path finds the absolute stored path by suffix, like `trace`.
        let hits = edits_for(&c, "src/auth.rs", 50).unwrap();

        assert_eq!(hits.len(), 2, "both auth.rs edits, not other.rs");
        assert_eq!(hits[0].kind, "write", "newest edit first");
        assert!(hits[0].snippet.contains("fn main()"));
        assert_eq!(hits[1].kind, "edit");
    }

    #[test]
    fn index_one_persists_a_sessions_edits() {
        use crate::model::{EditEvent, EditKind, Session};
        let mut c = Connection::open_in_memory().unwrap();
        create_cache_schema(&c).unwrap();

        let session = Session {
            id: "sx".into(),
            tool: "claude-code",
            path: "/store/sx.jsonl".into(),
            project: "/proj".into(),
            started: None,
            ended: None,
            title: "t".into(),
            subagent: false,
            messages: vec![],
            touched: vec!["/proj/src/auth.rs".into()],
            edits: vec![EditEvent {
                path: "/proj/src/auth.rs".into(),
                kind: EditKind::Write,
                snippet: "fn main() {}".into(),
                ts: None,
            }],
        };

        let tx = c.transaction().unwrap();
        index_one(&tx, &session, "/store/sx.jsonl", 0, 0, ToolOutputMode::Full).unwrap();
        tx.commit().unwrap();

        let hits = edits_for(&c, "src/auth.rs", 50).unwrap();
        assert_eq!(hits.len(), 1, "the session's one edit was persisted");
        assert_eq!(hits[0].session_id, "sx");
        assert_eq!(hits[0].kind, "write");
        assert!(hits[0].snippet.contains("fn main()"));
    }

    #[test]
    fn forget_removes_a_sessions_edits() {
        let mut c = Connection::open_in_memory().unwrap();
        create_cache_schema(&c).unwrap();
        c.execute(
            "INSERT INTO files(path, session_id, tool, mtime, size) VALUES('/store/s.jsonl','s1','claude-code',0,0)",
            [],
        )
        .unwrap();
        c.execute(
            "INSERT INTO edits(session_id,path,kind,ts,snippet) VALUES('s1','/proj/a.rs','write',NULL,'x')",
            [],
        )
        .unwrap();
        assert_eq!(edits_for(&c, "a.rs", 10).unwrap().len(), 1);

        forget(&mut c, "s1").unwrap();

        assert!(
            edits_for(&c, "a.rs", 10).unwrap().is_empty(),
            "forget must remove the session's edits, not orphan them"
        );
    }

    #[test]
    fn edits_for_is_deterministic_when_timestamps_tie() {
        let c = conn_with_edits();
        add(&c, "s1", "/p/a.rs", "edit", "2026-01-01T00:00:00Z", "first");
        add(
            &c,
            "s2",
            "/p/a.rs",
            "write",
            "2026-01-01T00:00:00Z",
            "second",
        );
        let hits = edits_for(&c, "a.rs", 10).unwrap();
        // Equal ts -> deterministic tie-break by rowid DESC (latest insert first).
        assert_eq!(hits[0].snippet, "second");
        assert_eq!(hits[1].snippet, "first");
    }

    #[test]
    fn edits_for_session_returns_only_that_sessions_edits() {
        let c = conn_with_edits();
        add(
            &c,
            "s1",
            "/p/a.rs",
            "edit",
            "2026-01-01T00:00:00Z",
            "s1-edit",
        );
        add(
            &c,
            "s2",
            "/p/a.rs",
            "write",
            "2026-02-01T00:00:00Z",
            "s2-edit",
        );
        // Scoped by exact session + path, so one session's edits can never be
        // starved by another's under a shared cap.
        let hits = edits_for_session(&c, "s1", "/p/a.rs", 10).unwrap();
        assert_eq!(hits.len(), 1);
        assert_eq!(hits[0].snippet, "s1-edit");
    }

    #[test]
    fn edits_for_session_matches_every_spelling_of_the_path() {
        let c = conn_with_edits();
        // One session edited the file under two path spellings (abs + relative).
        add(
            &c,
            "s1",
            "/proj/src/auth.rs",
            "edit",
            "2026-01-01T00:00:00Z",
            "abs",
        );
        add(
            &c,
            "s1",
            "src/auth.rs",
            "write",
            "2026-01-02T00:00:00Z",
            "rel",
        );
        add(
            &c,
            "s2",
            "/other/auth.rs",
            "edit",
            "2026-01-03T00:00:00Z",
            "different-file",
        );
        // Suffix match scoped to s1 must catch BOTH spellings - never miss edits
        // just because sessions_for_file's GROUP BY picked the other spelling.
        let hits = edits_for_session(&c, "s1", "src/auth.rs", 10).unwrap();
        assert_eq!(hits.len(), 2, "all of s1's edits to the file, any spelling");
    }

    #[test]
    fn index_redacts_secrets_in_messages_and_edit_snippets() {
        use crate::model::{EditEvent, EditKind, Message, Role, Session};
        let mut c = Connection::open_in_memory().unwrap();
        create_cache_schema(&c).unwrap();
        let session = Session {
            id: "sx".into(),
            tool: "claude-code",
            path: "/s.jsonl".into(),
            project: "/p".into(),
            started: None,
            ended: None,
            title: "title with AKIAIOSFODNN7EXAMPLE in it".into(),
            subagent: false,
            messages: vec![Message {
                role: Role::User,
                text: "my key is sk-abcdef012345678901234567890123 ok".into(),
                ts: None,
                tool: None,
            }],
            touched: vec!["/p/a.rs".into()],
            edits: vec![EditEvent {
                path: "/p/a.rs".into(),
                kind: EditKind::Write,
                snippet: "const T = \"ghp_016C7f9aBcDeFgHiJkLmNoPqRsTuVwXyZ012\";".into(),
                ts: None,
            }],
        };
        let tx = c.transaction().unwrap();
        index_one(&tx, &session, "/s.jsonl", 0, 0, ToolOutputMode::Full).unwrap();
        tx.commit().unwrap();

        let msg: String = c
            .query_row("SELECT text FROM messages WHERE session_id='sx'", [], |r| {
                r.get(0)
            })
            .unwrap();
        assert!(!msg.contains("sk-abcdef"), "message secret redacted: {msg}");
        assert!(msg.contains("[redacted:openai]"), "{msg}");
        let snip = edits_for(&c, "a.rs", 10).unwrap()[0].snippet.clone();
        assert!(!snip.contains("ghp_016C"), "edit secret redacted: {snip}");
        assert!(snip.contains("[redacted:github]"), "{snip}");
        // Title is durable (copied into archives) - must be redacted too.
        let title: String = c
            .query_row("SELECT title FROM files WHERE session_id='sx'", [], |r| {
                r.get(0)
            })
            .unwrap();
        assert!(
            !title.contains("AKIAIOSFODNN7EXAMPLE"),
            "title secret redacted: {title}"
        );
        // LLM synopsis can echo a secret into the durable summaries table.
        set_summary(
            &c,
            "sx",
            "we set sk-abcdef012345678901234567890123 as the key",
        )
        .unwrap();
        let sum: String = c
            .query_row(
                "SELECT summary FROM summaries WHERE session_id='sx'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert!(!sum.contains("sk-abcdef"), "summary secret redacted: {sum}");
    }

    #[test]
    fn evidence_for_assembles_sessions_with_their_edits_newest_first() {
        use crate::model::{EditEvent, EditKind, Session};
        let mut c = Connection::open_in_memory().unwrap();
        create_cache_schema(&c).unwrap();

        for (sid, store, started, kind, snip) in [
            (
                "old",
                "/store/old.jsonl",
                "2026-06-01T00:00:00Z",
                EditKind::Edit,
                "v1",
            ),
            (
                "new",
                "/store/new.jsonl",
                "2026-06-09T00:00:00Z",
                EditKind::Write,
                "v2",
            ),
        ] {
            let started = chrono::DateTime::parse_from_rfc3339(started)
                .unwrap()
                .with_timezone(&chrono::Utc);
            let session = Session {
                id: sid.into(),
                tool: "claude-code",
                path: store.into(),
                project: "/proj".into(),
                started: Some(started),
                ended: None,
                title: format!("{sid} title"),
                subagent: false,
                messages: vec![],
                touched: vec!["/proj/src/a.rs".into()],
                edits: vec![EditEvent {
                    path: "/proj/src/a.rs".into(),
                    kind,
                    snippet: snip.into(),
                    ts: None,
                }],
            };
            let tx = c.transaction().unwrap();
            index_one(&tx, &session, store, 0, 0, ToolOutputMode::Full).unwrap();
            tx.commit().unwrap();
        }

        let hist = evidence_for(&c, "src/a.rs", 50).unwrap();
        assert_eq!(hist.path, "src/a.rs");
        assert_eq!(hist.sessions.len(), 2, "both sessions that edited the file");
        assert_eq!(
            hist.sessions[0].session.session_id, "new",
            "newest session first"
        );
        assert_eq!(hist.sessions[0].edits.len(), 1);
        assert_eq!(hist.sessions[0].edits[0].snippet, "v2");
        assert_eq!(hist.sessions[1].session.session_id, "old");
    }
}

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

    /// Folders get renamed. A session recorded `~/Project/lunch/diag.py`; the
    /// folder is now `~/Project/slack`, so tracing the file by the path it has
    /// TODAY found nothing and said "no session touched a file matching" - about
    /// a file whose whole history was sitting in the index under its old name.
    ///
    /// The basename is the part that survives a move, so a full path that finds
    /// nothing falls back to it, and the caller is told the match was by name so
    /// it can say the folder has moved.
    #[test]
    fn a_renamed_folder_still_traces_by_file_name() {
        assert_eq!(
            basename_fallback("/Users/b/Project/slack/diag.py").as_deref(),
            Some("diag.py"),
            "a full path falls back to its file name"
        );
        // Already a bare name: there is nothing to fall back to, and retrying
        // the same query would just repeat the miss.
        assert_eq!(basename_fallback("diag.py"), None);
        assert_eq!(basename_fallback(""), None);
        // A trailing slash names a directory, not a file to trace.
        assert_eq!(basename_fallback("/Users/b/Project/slack/"), None);
    }
}

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

    /// The migration looked for the old directories under `dirs::data_dir()`
    /// no matter where the index was actually going, and then RENAMED what it
    /// found into that destination. With `SESSIONWIKI_DATA` pointed at a temp
    /// dir - which eight test files do - a machine still holding
    /// `~/.local/share/sessiondex` would have had its real index moved into
    /// that temp dir and deleted with it. The comment above says the tags,
    /// notes and summaries in there are not rebuildable.
    #[test]
    fn a_legacy_index_is_only_looked_for_beside_the_new_one() {
        let under_home = std::path::Path::new("/home/someone/.local/share/sessionwiki");
        let got = legacy_candidates(under_home);
        assert_eq!(
            got,
            vec![
                std::path::PathBuf::from("/home/someone/.local/share/sessiondex"),
                std::path::PathBuf::from("/home/someone/.local/share/session-atlas"),
            ],
            "the normal case must keep working"
        );

        let redirected = std::path::Path::new("/tmp/sessionwiki-test-xyz");
        for c in legacy_candidates(redirected) {
            assert!(
                c.starts_with("/tmp"),
                "a redirected run reached outside its own tree: {}",
                c.display()
            );
        }
    }

    #[test]
    fn a_destination_with_no_parent_offers_nothing_to_migrate() {
        assert!(legacy_candidates(std::path::Path::new("/")).is_empty());
    }
}

#[cfg(test)]
mod codex_history_tests;

#[cfg(test)]
mod embedder_hook_tests {
    use super::*;
    use crate::adapters::{Adapter, Discovered, Store};
    use crate::model::{Message, Role, Session};
    use std::path::Path;

    /// A shared-store adapter an embedding program could supply: it lists only
    /// the keys under its own prefix and reconciles only that prefix.
    struct FakeStore {
        keys: Vec<(String, i64)>,
        scope: Option<String>,
    }

    impl Adapter for FakeStore {
        fn name(&self) -> &'static str {
            "mjolnir"
        }
        fn root(&self) -> Option<PathBuf> {
            // Any existing directory: the reconciliation guard only asks
            // whether the store root is still there.
            Some(std::env::current_dir().unwrap())
        }
        fn discover(&self) -> Discovered {
            Discovered {
                files: Vec::new(),
                had_error: false,
            }
        }
        fn parse(&self, _path: &Path) -> Result<Session> {
            anyhow::bail!("shared store")
        }
        fn store(&self) -> Option<Store> {
            Some(Store {
                keys: self.keys.clone(),
                files: Vec::new(),
                had_error: false,
            })
        }
        fn parse_key(&self, key: &str) -> Result<Session> {
            Ok(Session {
                id: key.rsplit('/').next().unwrap().to_string(),
                tool: "mjolnir",
                path: PathBuf::from(key),
                project: "/proj".into(),
                started: None,
                ended: None,
                title: "a restored session".into(),
                subagent: false,
                messages: vec![Message {
                    role: Role::User,
                    text: "make the tests green".into(),
                    ts: None,
                    tool: None,
                }],
                touched: vec![],
                edits: vec![],
            })
        }
        fn reconcile_scope(&self) -> Option<String> {
            self.scope.clone()
        }
    }

    fn insert_live_row(c: &Connection, path: &str, sid: &str) {
        c.execute(
            "INSERT INTO files(path, session_id, tool, mtime, size, project, title, msg_count, kind)
             VALUES(?1, ?2, 'mjolnir', 0, 0, '/proj', 't', 1, 'session')",
            params![path, sid],
        )
        .unwrap();
        c.execute(
            "INSERT INTO messages(session_id, role, text) VALUES(?1,'user','hello')",
            params![sid],
        )
        .unwrap();
    }

    fn archived_at(c: &Connection, path: &str) -> Option<String> {
        c.query_row(
            "SELECT archived_at FROM files WHERE path = ?1",
            params![path],
            |r| r.get(0),
        )
        .unwrap()
    }

    /// Two installations of one tool share a tool name and one index. A sync
    /// driven by the first must not archive the second's rows just because it
    /// never lists them.
    #[test]
    fn reconcile_scope_limits_archiving_to_the_adapters_own_keys() {
        let mut c = Connection::open_in_memory().unwrap();
        create_cache_schema(&c).unwrap();
        insert_live_row(&c, "/data/one/sess-a", "sa");
        insert_live_row(&c, "/data/two/sess-b", "sb");

        let adapters: Vec<Box<dyn Adapter>> = vec![Box::new(FakeStore {
            keys: Vec::new(),
            scope: Some("/data/one/".to_string()),
        })];
        sync_with(&mut c, &adapters, None).unwrap();

        assert!(
            archived_at(&c, "/data/one/sess-a").is_some(),
            "the in-scope row the adapter no longer lists must be archived"
        );
        assert!(
            archived_at(&c, "/data/two/sess-b").is_none(),
            "the other installation's row must be left live"
        );
    }

    /// Without a scope the adapter still speaks for every row of its tool.
    #[test]
    fn an_unscoped_adapter_still_archives_every_row_of_its_tool() {
        let mut c = Connection::open_in_memory().unwrap();
        create_cache_schema(&c).unwrap();
        insert_live_row(&c, "/data/one/sess-a", "sa");
        insert_live_row(&c, "/data/two/sess-b", "sb");

        let adapters: Vec<Box<dyn Adapter>> = vec![Box::new(FakeStore {
            keys: Vec::new(),
            scope: None,
        })];
        sync_with(&mut c, &adapters, None).unwrap();

        assert!(archived_at(&c, "/data/one/sess-a").is_some());
        assert!(archived_at(&c, "/data/two/sess-b").is_some());
    }

    #[test]
    fn a_shared_store_restores_an_archived_row_without_a_changed_token() {
        let mut c = Connection::open_in_memory().unwrap();
        create_cache_schema(&c).unwrap();
        let key = "/data/one/sess-a";
        let adapters: Vec<Box<dyn Adapter>> = vec![Box::new(FakeStore {
            keys: vec![(key.to_string(), 42)],
            scope: Some("/data/one/".to_string()),
        })];
        sync_with(&mut c, &adapters, None).unwrap();
        let message_id: i64 = c
            .query_row(
                "SELECT id FROM messages WHERE session_id = 'sess-a'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        archive_session(&c, key, "sess-a").unwrap();

        sync_with(&mut c, &adapters, Some(100)).unwrap();

        assert_eq!(archived_at(&c, key), None);
        assert_eq!(
            c.query_row(
                "SELECT id FROM messages WHERE session_id = 'sess-a'",
                [],
                |row| row.get::<_, i64>(0)
            )
            .unwrap(),
            message_id,
            "unchanged transcripts need no reparse"
        );
        assert_eq!(
            c.query_row(
                "SELECT count(*) FROM archive WHERE session_id = 'sess-a'",
                [],
                |row| row.get::<_, i64>(0)
            )
            .unwrap(),
            0
        );
    }

    #[test]
    fn a_file_store_restores_an_archived_row_without_changed_file_metadata() {
        let home = tempfile::tempdir().unwrap();
        let path = home
            .path()
            .join("sessions/2026/01/01/rollout-2026-01-01T10-00-00-abc.jsonl");
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        std::fs::write(
            &path,
            "{\"type\":\"session_meta\",\"payload\":{\"cwd\":\"/repo\"}}\n{\"type\":\"event_msg\",\"payload\":{\"type\":\"user_message\",\"message\":\"restic\"}}\n",
        )
        .unwrap();
        let adapters: Vec<Box<dyn Adapter>> =
            vec![Box::new(crate::adapters::Codex::in_home(home.path()))];
        let mut c = Connection::open_in_memory().unwrap();
        create_cache_schema(&c).unwrap();
        sync_with(&mut c, &adapters, None).unwrap();
        let key = path.to_string_lossy();
        let sid: String = c
            .query_row(
                "SELECT session_id FROM files WHERE path = ?1",
                [key.as_ref()],
                |row| row.get(0),
            )
            .unwrap();
        let message_id: i64 = c
            .query_row(
                "SELECT id FROM messages WHERE session_id = ?1 LIMIT 1",
                [&sid],
                |row| row.get(0),
            )
            .unwrap();
        archive_session(&c, &key, &sid).unwrap();

        sync_with(&mut c, &adapters, Some(i64::MAX)).unwrap();

        assert_eq!(archived_at(&c, &key), None);
        assert_eq!(
            c.query_row(
                "SELECT id FROM messages WHERE session_id = ?1 LIMIT 1",
                [&sid],
                |row| row.get::<_, i64>(0)
            )
            .unwrap(),
            message_id,
            "unchanged files need no reparse"
        );

        archive_session(&c, &key, &sid).unwrap();
        std::fs::OpenOptions::new().append(true).open(&path).unwrap().write_all(b"{\"type\":\"event_msg\",\"payload\":{\"type\":\"user_message\",\"message\":\"new content\"}}\n").unwrap();
        sync_with(&mut c, &adapters, Some(i64::MAX)).unwrap();
        assert_eq!(archived_at(&c, &key), None);
        assert_eq!(
            c.query_row(
                "SELECT count(*) FROM messages WHERE session_id = ?1 AND text = 'new content'",
                [&sid],
                |row| row.get::<_, i64>(0)
            )
            .unwrap(),
            1,
            "changed archived files must be reparsed"
        );
    }

    /// "Recent" is about activity. Ordering by when a session STARTED put one
    /// resumed every day for three months below everything begun this week -
    /// on the machine this was found, eleven sessions active in the last day
    /// were buried that way, including the one asking.
    #[test]
    fn recent_orders_by_last_activity_not_by_start() {
        let c = Connection::open_in_memory().unwrap();
        create_cache_schema(&c).unwrap();
        for (id, started, ended) in [
            ("resumed", "2026-06-01T00:00:00Z", "2026-09-30T01:00:00Z"),
            ("fresh", "2026-09-20T00:00:00Z", "2026-09-20T01:00:00Z"),
        ] {
            c.execute(
                "INSERT INTO files(path, mtime, size, session_id, tool, started, ended, msg_count, kind)
                 VALUES(?1, 0, 0, ?1, 'codex', ?2, ?3, 1, 'main')",
                rusqlite::params![id, started, ended],
            )
            .unwrap();
        }
        let rows = recent(&c, 10, None, None, None, false).unwrap();
        let ids: Vec<&str> = rows.iter().map(|r| r.session_id.as_str()).collect();
        assert_eq!(ids, ["resumed", "fresh"]);
        assert_eq!(rows[0].last_active.as_deref(), Some("2026-09-30T01:00:00Z"));
        assert_eq!(rows[0].started.as_deref(), Some("2026-06-01T00:00:00Z"));
    }

    /// The point of `sync_with`: an embedding program indexes its own sessions
    /// with its own adapter, which is in no built-in registry.
    #[test]
    fn sync_with_indexes_a_session_from_a_supplied_adapter() {
        let mut c = Connection::open_in_memory().unwrap();
        create_cache_schema(&c).unwrap();

        let adapters: Vec<Box<dyn Adapter>> = vec![Box::new(FakeStore {
            keys: vec![("/data/one/sess-a".to_string(), 42)],
            scope: Some("/data/one/".to_string()),
        })];
        sync_with(&mut c, &adapters, None).unwrap();

        let rows = recent(&c, 10, Some("mjolnir"), None, None, false).unwrap();
        assert_eq!(rows.len(), 1, "the supplied adapter's session was indexed");
        assert_eq!(rows[0].session_id, "sess-a");
        assert_eq!(rows[0].title, "a restored session");
        assert_eq!(rows[0].msg_count, 1);
        assert!(!rows[0].archived, "a listed session stays live");
    }

    /// A row whose tool only an embedder's adapter knows still shows that
    /// tool's name, rather than the "unknown" the registry lookup used to give.
    #[test]
    fn a_session_keeps_its_own_tool_name_when_no_adapter_is_registered() {
        let mut c = Connection::open_in_memory().unwrap();
        create_cache_schema(&c).unwrap();

        let adapters: Vec<Box<dyn Adapter>> = vec![Box::new(FakeStore {
            keys: vec![("/data/one/sess-a".to_string(), 42)],
            scope: Some("/data/one/".to_string()),
        })];
        sync_with(&mut c, &adapters, None).unwrap();

        let rows = recent(&c, 10, Some("mjolnir"), None, None, false).unwrap();
        assert!(
            crate::adapters::by_name("mjolnir").is_none(),
            "the built-in registry must not know this tool, or the test proves nothing"
        );
        let session = session_from_index(&c, &rows[0]).unwrap();
        assert_eq!(session.tool, "mjolnir");
    }

    /// The interner hands back one leaked string per name, however often it is
    /// asked, so the leak is bounded by the number of tool names.
    #[test]
    fn interning_a_tool_name_twice_yields_the_same_string() {
        let first = interned_tool("a-tool-no-adapter-knows");
        let second = interned_tool(&String::from("a-tool-no-adapter-knows"));
        assert_eq!(first, "a-tool-no-adapter-knows");
        assert!(std::ptr::eq(first, second));
    }
}

#[cfg(test)]
mod tool_output_tests {
    use super::*;
    use crate::model::{Message, Role, Session, ToolEvent, ToolSummary};
    use std::cell::Cell;
    use std::rc::Rc;

    const OUTPUT: &str = "fulloutputneedle: private tool result\nsecond diagnostic";

    fn tool_session(id: &str, path: &str) -> Session {
        Session {
            id: id.into(),
            tool: "codex",
            path: path.into(),
            project: "/project".into(),
            started: None,
            ended: None,
            title: "tool output test".into(),
            subagent: false,
            messages: vec![Message {
                role: Role::Tool,
                text: "→ Bash(cargo test) ⇒ ok · 2 lines".into(),
                ts: None,
                tool: Some(ToolEvent::Summary(ToolSummary {
                    name: "Bash".into(),
                    args: serde_json::json!({"command": "cargo test"}),
                    output: Some(OUTPUT.into()),
                    is_error: Some(false),
                    lines: Some(2),
                })),
            }],
            touched: Vec::new(),
            edits: Vec::new(),
        }
    }

    fn insert_tool_session(conn: &mut Connection, id: &str, path: &str) {
        let session = tool_session(id, path);
        let tx = conn.transaction().unwrap();
        index_one(
            &tx,
            &session,
            path,
            123,
            456,
            tool_output_mode(&tx).unwrap(),
        )
        .unwrap();
        tx.commit().unwrap();
    }

    #[test]
    fn full_mode_archives_and_rehydrates_output_without_adding_it_to_fts() {
        let mut conn = Connection::open_in_memory().unwrap();
        create_cache_schema(&conn).unwrap();
        assert!(set_tool_output_mode(&conn, ToolOutputMode::Full).unwrap());
        assert_eq!(tool_output_mode(&conn).unwrap(), ToolOutputMode::Full);
        insert_tool_session(&mut conn, "full-session", "/gone/full-session.jsonl");

        let stored: String = conn
            .query_row(
                "SELECT full_text FROM messages WHERE session_id = 'full-session'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(stored, OUTPUT);
        let fts_hits: i64 = conn
            .query_row(
                "SELECT count(*) FROM msgs WHERE msgs MATCH 'fulloutputneedle'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(fts_hits, 0, "full output must not be indexed by FTS");

        archive_session(&conn, "/gone/full-session.jsonl", "full-session").unwrap();
        let transcript: String = conn
            .query_row(
                "SELECT transcript FROM archive WHERE session_id = 'full-session'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        let archived = decode_archive_transcript(&transcript).unwrap();
        assert_eq!(archived[0].full_text.as_deref(), Some(OUTPUT));

        // Rehydrate is the archive path used after a schema rebuild.
        rehydrate_archive(&conn).unwrap();
        let row = resolve(&conn, "full-session").unwrap().remove(0);
        let session = session_from_index(&conn, &row).unwrap();
        let stored_output = match session.messages[0].tool.as_ref().unwrap() {
            ToolEvent::Summary(summary) => summary.output.as_deref(),
            ToolEvent::Part(_) => panic!("indexed output must be a summary"),
        };
        assert_eq!(stored_output, Some(OUTPUT));
    }

    #[test]
    fn fresh_index_defaults_to_summary_mode() {
        let conn = Connection::open_in_memory().unwrap();
        create_cache_schema(&conn).unwrap();
        assert_eq!(tool_output_mode(&conn).unwrap(), ToolOutputMode::Summary);
    }

    #[test]
    fn summary_mode_preserves_archived_output_and_full_mode_marks_live_rows_stale() {
        let mut conn = Connection::open_in_memory().unwrap();
        create_cache_schema(&conn).unwrap();
        assert!(set_tool_output_mode(&conn, ToolOutputMode::Full).unwrap());
        insert_tool_session(&mut conn, "archived", "/gone/archived.jsonl");
        insert_tool_session(&mut conn, "live", "/live/live.jsonl");
        archive_session(&conn, "/gone/archived.jsonl", "archived").unwrap();

        assert!(set_tool_output_mode(&conn, ToolOutputMode::Summary).unwrap());
        assert_eq!(tool_output_mode(&conn).unwrap(), ToolOutputMode::Summary);
        let live_output: Option<String> = conn
            .query_row(
                "SELECT full_text FROM messages WHERE session_id = 'live'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(live_output, None, "live output is cleared");
        let archived_output: Option<String> = conn
            .query_row(
                "SELECT full_text FROM messages WHERE session_id = 'archived'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(archived_output.as_deref(), Some(OUTPUT));
        let transcript: String = conn
            .query_row(
                "SELECT transcript FROM archive WHERE session_id = 'archived'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert!(transcript.contains("fulloutputneedle"));
        let archived_messages = decode_archive_transcript(&transcript).unwrap();
        assert_eq!(archived_messages[0].full_text.as_deref(), Some(OUTPUT));
        let archived_token_before_rehydrate: (i64, i64) = conn
            .query_row(
                "SELECT mtime, size FROM files WHERE session_id = 'archived'",
                [],
                |row| Ok((row.get(0)?, row.get(1)?)),
            )
            .unwrap();
        assert_eq!(archived_token_before_rehydrate, (123, 456));

        // Rehydrating after a schema rebuild preserves the archived copy in
        // messages as well, so indexed readers can still return --full output.
        rehydrate_archive(&conn).unwrap();
        let rehydrated_output: Option<String> = conn
            .query_row(
                "SELECT full_text FROM messages WHERE session_id = 'archived'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(rehydrated_output.as_deref(), Some(OUTPUT));
        let archived_row = resolve(&conn, "archived").unwrap().remove(0);
        let archived_session = session_from_index(&conn, &archived_row).unwrap();
        let shown_output = match archived_session.messages[0].tool.as_ref().unwrap() {
            ToolEvent::Summary(summary) => summary.output.as_deref(),
            ToolEvent::Part(_) => panic!("indexed output must be a summary"),
        };
        assert_eq!(shown_output, Some(OUTPUT));

        // Newly indexed sessions follow the changed mode.
        insert_tool_session(&mut conn, "summary-new", "/live/summary-new.jsonl");
        let new_output: Option<String> = conn
            .query_row(
                "SELECT full_text FROM messages WHERE session_id = 'summary-new'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(new_output, None);
        archive_session(&conn, "/live/summary-new.jsonl", "summary-new").unwrap();
        let new_archive: String = conn
            .query_row(
                "SELECT transcript FROM archive WHERE session_id = 'summary-new'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        let archived_messages = decode_archive_transcript(&new_archive).unwrap();
        assert_eq!(archived_messages.len(), 1);
        assert_eq!(
            archived_messages[0].text,
            "→ Bash(cargo test) ⇒ ok · 2 lines"
        );
        assert_eq!(archived_messages[0].full_text, None);
        assert!(!new_archive.contains("fulloutputneedle"));

        assert!(set_tool_output_mode(&conn, ToolOutputMode::Full).unwrap());
        let token: (i64, i64) = conn
            .query_row(
                "SELECT mtime, size FROM files WHERE session_id = 'live'",
                [],
                |row| Ok((row.get(0)?, row.get(1)?)),
            )
            .unwrap();
        assert_eq!(token, (-1, -1), "live row must be reparsed on next sync");
        let archived_token: (i64, i64, Option<String>) = conn
            .query_row(
                "SELECT mtime, size, archived_at FROM files WHERE session_id = 'archived'",
                [],
                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
            )
            .unwrap();
        assert_eq!((archived_token.0, archived_token.1), (0, 0));
        assert!(archived_token.2.is_some(), "archived row stays archived");
        let archived_output_after_full: Option<String> = conn
            .query_row(
                "SELECT full_text FROM messages WHERE session_id = 'archived'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(
            archived_output_after_full.as_deref(),
            Some(OUTPUT),
            "archived rows keep their prior contents"
        );
    }

    struct CountedStore {
        parses: Rc<Cell<usize>>,
    }

    impl Adapter for CountedStore {
        fn name(&self) -> &'static str {
            "tool-output-sync-test"
        }

        fn root(&self) -> Option<PathBuf> {
            std::env::current_dir().ok()
        }

        fn discover(&self) -> crate::adapters::Discovered {
            Vec::new().into()
        }

        fn parse(&self, _path: &std::path::Path) -> Result<Session> {
            anyhow::bail!("shared-store adapter")
        }

        fn store(&self) -> Option<crate::adapters::Store> {
            Some(crate::adapters::Store {
                keys: vec![("/store/forced-session".into(), 42)],
                files: Vec::new(),
                had_error: false,
            })
        }

        fn parse_key(&self, key: &str) -> Result<Session> {
            self.parses.set(self.parses.get() + 1);
            Ok(tool_session("forced-session", key))
        }
    }

    #[test]
    fn enabling_full_reparses_stale_sessions_during_a_bounded_sync() {
        let mut conn = Connection::open_in_memory().unwrap();
        create_cache_schema(&conn).unwrap();
        let parses = Rc::new(Cell::new(0));
        let adapters: Vec<Box<dyn Adapter>> = vec![Box::new(CountedStore {
            parses: Rc::clone(&parses),
        })];
        sync_with(&mut conn, &adapters, None).unwrap();
        assert_eq!(parses.get(), 1);

        set_tool_output_mode(&conn, ToolOutputMode::Summary).unwrap();
        set_tool_output_mode(&conn, ToolOutputMode::Full).unwrap();
        sync_with(&mut conn, &adapters, Some(i64::MAX)).unwrap();

        assert_eq!(parses.get(), 2, "forced sync re-parses despite its cutoff");
        let full_text: Option<String> = conn
            .query_row(
                "SELECT full_text FROM messages WHERE session_id = 'forced-session'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(full_text.as_deref(), Some(OUTPUT));
    }
}