nexus-chat 0.1.0

// Casts here are on bounded values: token counts, byte sizes, and
// selection indices — never on unbounded input. JSON-derived indices in
// provider/tools go through try_from instead.
#![allow(
    clippy::cast_possible_truncation,
    clippy::cast_possible_wrap,
    clippy::cast_precision_loss,
    clippy::cast_sign_loss
)]
use anyhow::{Context, Result};
use chrono::Utc;
use rusqlite::{Connection, OptionalExtension};
use uuid::Uuid;

/// Stored per-model preferences.
#[derive(Debug, Clone)]
pub struct ModelPref {
    pub id: String,
    pub favorite: bool,
    pub last_used: Option<String>,
    pub reasoning: Option<String>,
}

/// A space: an isolated collection of sessions with its own memory/instructions
/// (stored as files on disk, see `space.rs`). `name` doubles as the directory name.
#[derive(Debug, Clone)]
pub struct Space {
    pub id: String,
    pub name: String,
    pub created_at: String,
}

/// One chat session (conversation) within a space.
#[derive(Debug, Clone)]
pub struct Session {
    pub id: String,
    pub title: String,
    pub model: String,
    /// Short human-readable id (kebab slug), generated by the model. `None` until
    /// generated — display falls back to a prefix of the uuid.
    pub slug: Option<String>,
    pub created_at: String,
    /// Caveman-compressed digest of the session's earlier messages, if it's
    /// ever been auto-compacted. `None` until the first compaction.
    pub compact_summary: Option<String>,
    /// How many of the session's raw messages (in `created_at` order) are
    /// folded into `compact_summary`. Messages after this point are still
    /// sent verbatim; 0 means nothing has been compacted yet.
    pub compact_through: i64,
    /// `/web` answer mode: force search-first, inline-cited replies.
    pub web_mode: bool,
    /// `/swarm` mode: turns replace a single reply with a multi-persona
    /// roundtable (see `swarm_personas`).
    pub swarm_mode: bool,
    /// `"chat"` or `"research"` — determines what's available.
    pub kind: String,
    /// If this is a research session spawned from an existing chat, the
    /// original session's id.
    pub research_parent_id: Option<String>,
}

/// One row of a session's `/swarm` roster: a model + a personality blurb.
#[derive(Debug, Clone)]
pub struct Persona {
    pub name: String,
    pub model: String,
    pub blurb: String,
}

/// A standing research watch: runs a topic's research on an interval.
#[derive(Debug, Clone)]
pub struct Watch {
    pub id: String,
    pub space_id: String,
    pub topic: String,
    pub interval_hours: i64,
    pub session_id: String,
    pub last_run_at: Option<String>,
}

/// A file imported into a space's fileset. `status` is "ok", "no text
/// (scanned?)", "unsupported", or "error: …"; extraction text lives in the
/// `file_chunks` FTS table, not here.
#[derive(Debug, Clone)]
pub struct FileRow {
    pub id: String,
    pub name: String,
    pub hash: String,
    pub size: i64,
    pub status: String,
    /// Unix mtime of the disk file when last indexed; lets rescans skip
    /// reading/hashing files whose (size, mtime) haven't changed.
    pub mtime: i64,
}

/// One message in a session. `model`/`reasoning`/`tokens`/`secs`/`cost` are
/// populated for assistant replies (None for user/system messages).
#[derive(Debug, Clone)]
pub struct Message {
    pub role: String,
    pub content: String,
    pub model: Option<String>,
    pub reasoning: Option<String>,
    pub tokens: Option<i64>,
    pub secs: Option<f64>,
    /// USD cost of the request that produced this reply, at catalog list
    /// price (`None` when unknown — non-OpenRouter backends, or replies
    /// logged before pricing existed).
    pub cost: Option<f64>,
    /// Past-tense flavour phrase for the completion line, e.g. "Vibed".
    pub phrase: Option<String>,
    /// Which `/swarm` persona produced this reply, if any (`None` for
    /// ordinary messages and for a swarm turn's final synthesis reply).
    pub persona: Option<String>,
    /// RFC3339 timestamp of the row (None for in-memory-only messages that
    /// were never persisted, e.g. incognito streams).
    pub created_at: Option<String>,
}

/// Name of the always-present, undeletable space that sessions default into.
pub const DEFAULT_SPACE: &str = "default";

// ponytail: rusqlite is synchronous and called inline on the UI task. Writes are
// tiny single-user local inserts, so no spawn_blocking. Move to a blocking pool
// only if the db ever lives on slow/remote storage.
pub struct Db {
    conn: Connection,
}

impl Db {
    pub fn open(path: &std::path::Path) -> Result<Self> {
        let conn =
            Connection::open(path).with_context(|| format!("opening db {}", path.display()))?;
        let db = Self { conn };
        db.migrate()?;
        Ok(db)
    }

    #[cfg(test)]
    pub fn open_in_memory() -> Result<Self> {
        let db = Db {
            conn: Connection::open_in_memory()?,
        };
        db.migrate()?;
        Ok(db)
    }

    #[cfg(test)]
    pub fn conn_for_test(&self) -> &Connection {
        &self.conn
    }

    // Long by design (schema migrations).
    #[allow(clippy::too_many_lines)]
    fn migrate(&self) -> Result<()> {
        self.conn.execute_batch(
            "CREATE TABLE IF NOT EXISTS sessions (
                id TEXT PRIMARY KEY,
                title TEXT NOT NULL,
                model TEXT NOT NULL,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS messages (
                id TEXT PRIMARY KEY,
                session_id TEXT NOT NULL REFERENCES sessions(id),
                role TEXT NOT NULL,
                content TEXT NOT NULL,
                created_at TEXT NOT NULL
            );
            CREATE INDEX IF NOT EXISTS idx_messages_session
                ON messages(session_id, created_at);
            CREATE TABLE IF NOT EXISTS model_prefs (
                id TEXT PRIMARY KEY,
                favorite INTEGER NOT NULL DEFAULT 0,
                last_used TEXT
            );
            CREATE TABLE IF NOT EXISTS app_settings (
                key TEXT PRIMARY KEY,
                value TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS spaces (
                id TEXT PRIMARY KEY,
                name TEXT NOT NULL UNIQUE,
                created_at TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS files (
                id TEXT PRIMARY KEY,
                space_id TEXT NOT NULL,
                name TEXT NOT NULL,
                hash TEXT NOT NULL,
                size INTEGER NOT NULL,
                status TEXT NOT NULL,
                created_at TEXT NOT NULL,
                UNIQUE(space_id, name)
            );
            CREATE VIRTUAL TABLE IF NOT EXISTS file_chunks USING fts5(
                file_id UNINDEXED,
                seq UNINDEXED,
                location UNINDEXED,
                text
            );
            CREATE TABLE IF NOT EXISTS chunk_embeddings (
                file_id TEXT NOT NULL,
                seq INTEGER NOT NULL,
                vec BLOB NOT NULL,
                PRIMARY KEY (file_id, seq)
            );
            CREATE TABLE IF NOT EXISTS web_cache (
                url_norm   TEXT PRIMARY KEY,
                url        TEXT NOT NULL,
                title      TEXT,
                text       TEXT NOT NULL,
                fetched_at TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS citations (
                id          INTEGER PRIMARY KEY AUTOINCREMENT,
                space_id    TEXT NOT NULL,
                report_file TEXT NOT NULL,
                url         TEXT NOT NULL,
                title       TEXT
            );
            CREATE INDEX IF NOT EXISTS idx_citations_space ON citations(space_id);
            CREATE TABLE IF NOT EXISTS session_sources (
                session_id TEXT NOT NULL,
                url_norm   TEXT NOT NULL,
                PRIMARY KEY (session_id, url_norm)
            );
            CREATE TABLE IF NOT EXISTS watches (
                id             TEXT PRIMARY KEY,
                space_id       TEXT NOT NULL,
                topic          TEXT NOT NULL,
                interval_hours INTEGER NOT NULL,
                session_id     TEXT NOT NULL,
                last_run_at    TEXT
            );
            CREATE TABLE IF NOT EXISTS swarm_personas (
                session_id TEXT NOT NULL,
                ord        INTEGER NOT NULL,
                name       TEXT NOT NULL,
                model      TEXT NOT NULL,
                persona    TEXT NOT NULL
            );
            CREATE INDEX IF NOT EXISTS idx_swarm_personas_session
                ON swarm_personas(session_id, ord);
            CREATE TABLE IF NOT EXISTS usage_log (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                created_at TEXT NOT NULL,
                session_id TEXT,
                space_id TEXT,
                backend TEXT NOT NULL,
                model TEXT NOT NULL,
                prompt_tokens INTEGER NOT NULL,
                completion_tokens INTEGER NOT NULL,
                cache_read_tokens INTEGER NOT NULL DEFAULT 0,
                cache_creation_tokens INTEGER NOT NULL DEFAULT 0,
                cost REAL
            );
            CREATE INDEX IF NOT EXISTS idx_usage_log_created ON usage_log(created_at);
            CREATE INDEX IF NOT EXISTS idx_usage_log_model ON usage_log(model);
            CREATE TABLE IF NOT EXISTS model_prices (
                model_id TEXT PRIMARY KEY,
                backend TEXT NOT NULL,
                prompt_price REAL NOT NULL,
                completion_price REAL NOT NULL,
                updated_at TEXT NOT NULL
            );",
        )?;
        // Columns added after v1; ignore "duplicate column" on existing dbs.
        for stmt in [
            "ALTER TABLE messages ADD COLUMN model TEXT",
            "ALTER TABLE messages ADD COLUMN reasoning TEXT",
            "ALTER TABLE messages ADD COLUMN tokens INTEGER",
            "ALTER TABLE messages ADD COLUMN secs REAL",
            "ALTER TABLE messages ADD COLUMN cost REAL",
            "ALTER TABLE messages ADD COLUMN phrase TEXT",
            "ALTER TABLE model_prefs ADD COLUMN reasoning TEXT",
            "ALTER TABLE sessions ADD COLUMN slug TEXT",
            "ALTER TABLE sessions ADD COLUMN space_id TEXT",
            "ALTER TABLE sessions ADD COLUMN compact_summary TEXT",
            "ALTER TABLE sessions ADD COLUMN compact_through INTEGER NOT NULL DEFAULT 0",
            "ALTER TABLE files ADD COLUMN mtime INTEGER NOT NULL DEFAULT 0",
            "ALTER TABLE sessions ADD COLUMN web_mode INTEGER NOT NULL DEFAULT 0",
            "ALTER TABLE session_sources ADD COLUMN flag TEXT",
            "ALTER TABLE sessions ADD COLUMN swarm_mode INTEGER NOT NULL DEFAULT 0",
            "ALTER TABLE messages ADD COLUMN persona TEXT",
            "ALTER TABLE sessions ADD COLUMN kind TEXT NOT NULL DEFAULT 'chat'",
            "ALTER TABLE sessions ADD COLUMN research_parent_id TEXT",
        ] {
            let _ = self.conn.execute(stmt, []);
        }
        // Migration: remove the message_images table — images are now embedded
        // as markdown `![alt](file)` in message content.
        let _ = self
            .conn
            .execute_batch("DROP TABLE IF EXISTS message_images;");
        // Ensure the default space exists, then backfill any session left
        // without a space (pre-spaces db, or a space that got deleted).
        let now = Utc::now().to_rfc3339();
        self.conn.execute(
            "INSERT OR IGNORE INTO spaces (id, name, created_at) VALUES (?1, ?2, ?3)",
            (Uuid::new_v4().to_string(), DEFAULT_SPACE, &now),
        )?;
        let default_id: String = self.conn.query_row(
            "SELECT id FROM spaces WHERE name = ?1",
            [DEFAULT_SPACE],
            |r| r.get(0),
        )?;
        self.conn.execute(
            "UPDATE sessions SET space_id = ?1 WHERE space_id IS NULL",
            [&default_id],
        )?;
        Ok(())
    }

    /// The default space's id (always present after `migrate`).
    pub fn default_space_id(&self) -> Result<String> {
        Ok(self.conn.query_row(
            "SELECT id FROM spaces WHERE name = ?1",
            [DEFAULT_SPACE],
            |r| r.get(0),
        )?)
    }

    pub fn create_space(&self, name: &str) -> Result<Space> {
        let id = Uuid::new_v4().to_string();
        let now = Utc::now().to_rfc3339();
        self.conn.execute(
            "INSERT INTO spaces (id, name, created_at) VALUES (?1, ?2, ?3)",
            (&id, name, &now),
        )?;
        Ok(Space {
            id,
            name: name.to_string(),
            created_at: now,
        })
    }

    /// Spaces oldest-first (`default` was inserted first, so it naturally leads).
    pub fn list_spaces(&self) -> Result<Vec<Space>> {
        let mut stmt = self
            .conn
            .prepare("SELECT id, name, created_at FROM spaces ORDER BY created_at ASC")?;
        let rows = stmt.query_map([], |r| {
            Ok(Space {
                id: r.get(0)?,
                name: r.get(1)?,
                created_at: r.get(2)?,
            })
        })?;
        Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
    }

    pub fn rename_space(&self, id: &str, name: &str) -> Result<()> {
        self.conn
            .execute("UPDATE spaces SET name = ?2 WHERE id = ?1", (id, name))?;
        Ok(())
    }

    /// Delete a space, reassigning its sessions to `default` rather than
    /// deleting them — only the space's own memory/instructions are lost.
    pub fn delete_space(&self, id: &str) -> Result<()> {
        let default_id = self.default_space_id()?;
        self.conn.execute(
            "UPDATE sessions SET space_id = ?1 WHERE space_id = ?2",
            (&default_id, id),
        )?;
        self.conn
            .execute("DELETE FROM spaces WHERE id = ?1", [id])?;
        Ok(())
    }

    /// Number of sessions currently in a space (shown in the space picker).
    pub fn count_sessions(&self, space_id: &str) -> Result<u64> {
        let n: i64 = self.conn.query_row(
            "SELECT COUNT(*) FROM sessions WHERE space_id = ?1",
            [space_id],
            |r| r.get(0),
        )?;
        Ok(n as u64)
    }

    /// The most recent user/assistant message of a session — the session
    /// picker's preview strip, so you can see what a session is about before
    /// opening it.
    pub fn last_message_preview(&self, session_id: &str) -> Option<String> {
        let mut stmt = self
            .conn
            .prepare(
                "SELECT content FROM messages WHERE session_id = ?1 \
                 AND role IN ('user','assistant') ORDER BY id DESC LIMIT 1",
            )
            .ok()?;
        let mut rows = stmt
            .query_map([session_id], |r| r.get::<_, String>(0))
            .ok()?;
        rows.next().and_then(Result::ok)
    }

    // --- key/value app settings ---

    pub fn set_setting(&self, key: &str, value: &str) -> Result<()> {
        self.conn.execute(
            "INSERT INTO app_settings (key, value) VALUES (?1, ?2)
             ON CONFLICT(key) DO UPDATE SET value = ?2",
            (key, value),
        )?;
        Ok(())
    }

    pub fn load_settings(&self) -> Result<Vec<(String, String)>> {
        let mut stmt = self.conn.prepare("SELECT key, value FROM app_settings")?;
        let rows = stmt.query_map([], |r| Ok((r.get(0)?, r.get(1)?)))?;
        Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
    }

    /// Set (or clear, with None) a model's reasoning effort.
    pub fn set_reasoning(&self, model_id: &str, effort: Option<&str>) -> Result<()> {
        self.conn.execute(
            "INSERT INTO model_prefs (id, reasoning) VALUES (?1, ?2)
             ON CONFLICT(id) DO UPDATE SET reasoning = ?2",
            (model_id, effort),
        )?;
        Ok(())
    }

    /// Flip a model's favorite flag; returns the new state.
    pub fn toggle_favorite(&self, model_id: &str) -> Result<bool> {
        self.conn.execute(
            "INSERT INTO model_prefs (id, favorite) VALUES (?1, 1)
             ON CONFLICT(id) DO UPDATE SET favorite = 1 - favorite",
            [model_id],
        )?;
        let fav: i64 = self.conn.query_row(
            "SELECT favorite FROM model_prefs WHERE id = ?1",
            [model_id],
            |r| r.get(0),
        )?;
        Ok(fav != 0)
    }

    /// Record a model as just used (for the recents ordering).
    pub fn mark_model_used(&self, model_id: &str) -> Result<()> {
        let now = Utc::now().to_rfc3339();
        self.conn.execute(
            "INSERT INTO model_prefs (id, favorite, last_used) VALUES (?1, 0, ?2)
             ON CONFLICT(id) DO UPDATE SET last_used = ?2",
            (model_id, &now),
        )?;
        Ok(())
    }

    /// All stored prefs: (model id, favorite, last used, reasoning effort).
    pub fn load_model_prefs(&self) -> Result<Vec<ModelPref>> {
        let mut stmt = self
            .conn
            .prepare("SELECT id, favorite, last_used, reasoning FROM model_prefs")?;
        let rows = stmt.query_map([], |r| {
            Ok(ModelPref {
                id: r.get(0)?,
                favorite: r.get::<_, i64>(1)? != 0,
                last_used: r.get(2)?,
                reasoning: r.get(3)?,
            })
        })?;
        Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
    }

    pub fn create_session(
        &self,
        title: &str,
        model: &str,
        space_id: &str,
        kind: &str,
    ) -> Result<Session> {
        let id = Uuid::new_v4().to_string();
        let now = Utc::now().to_rfc3339();
        self.conn.execute(
            "INSERT INTO sessions (id, title, model, space_id, kind, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?6)",
            (&id, title, model, space_id, kind, &now),
        )?;
        Ok(Session {
            id,
            title: title.to_string(),
            model: model.to_string(),
            slug: None,
            created_at: now,
            compact_summary: None,
            compact_through: 0,
            web_mode: false,
            swarm_mode: false,
            kind: kind.to_string(),
            research_parent_id: None,
        })
    }

    /// A single session by id, or `None` if it doesn't exist (e.g. deleted
    /// out from under a watch).
    pub fn get_session(&self, id: &str) -> Result<Option<Session>> {
        self.conn
            .query_row(
                "SELECT id, title, model, slug, created_at, compact_summary, compact_through, \
                 web_mode, swarm_mode, kind, research_parent_id
                 FROM sessions WHERE id = ?1",
                [id],
                |r| {
                    Ok(Session {
                        id: r.get(0)?,
                        title: r.get(1)?,
                        model: r.get(2)?,
                        slug: r.get(3)?,
                        created_at: r.get(4)?,
                        compact_summary: r.get(5)?,
                        compact_through: r.get(6)?,
                        web_mode: r.get::<_, i64>(7)? != 0,
                        swarm_mode: r.get::<_, i64>(8)? != 0,
                        kind: r.get(9)?,
                        research_parent_id: r.get(10)?,
                    })
                },
            )
            .optional()
            .map_err(Into::into)
    }

    /// Sessions in `space_id`, most-recently-updated first.
    pub fn list_sessions(&self, space_id: &str) -> Result<Vec<Session>> {
        let mut stmt = self.conn.prepare(
            "SELECT id, title, model, slug, created_at, compact_summary, compact_through, \
             web_mode, swarm_mode, kind, research_parent_id
             FROM sessions WHERE space_id = ?1 ORDER BY updated_at DESC",
        )?;
        let rows = stmt.query_map([space_id], |r| {
            Ok(Session {
                id: r.get(0)?,
                title: r.get(1)?,
                model: r.get(2)?,
                slug: r.get(3)?,
                created_at: r.get(4)?,
                compact_summary: r.get(5)?,
                compact_through: r.get(6)?,
                web_mode: r.get::<_, i64>(7)? != 0,
                swarm_mode: r.get::<_, i64>(8)? != 0,
                kind: r.get(9)?,
                research_parent_id: r.get(10)?,
            })
        })?;
        Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
    }

    /// Store an auto-compaction result: the digest plus how many raw messages
    /// it now covers.
    pub fn set_compaction(&self, session_id: &str, summary: &str, through: i64) -> Result<()> {
        self.conn.execute(
            "UPDATE sessions SET compact_summary = ?2, compact_through = ?3 WHERE id = ?1",
            (session_id, summary, through),
        )?;
        Ok(())
    }

    /// Persist a session's `/web` answer-mode toggle.
    pub fn set_session_web_mode(&self, session_id: &str, on: bool) -> Result<()> {
        self.conn.execute(
            "UPDATE sessions SET web_mode = ?2 WHERE id = ?1",
            (session_id, i64::from(on)),
        )?;
        Ok(())
    }

    /// Persist a session's `/swarm` mode toggle.
    pub fn set_session_swarm_mode(&self, session_id: &str, on: bool) -> Result<()> {
        self.conn.execute(
            "UPDATE sessions SET swarm_mode = ?2 WHERE id = ?1",
            (session_id, i64::from(on)),
        )?;
        Ok(())
    }

    /// A session's `/swarm` roster, in display order.
    pub fn list_swarm_personas(&self, session_id: &str) -> Result<Vec<Persona>> {
        let mut stmt = self.conn.prepare(
            "SELECT name, model, persona FROM swarm_personas
             WHERE session_id = ?1 ORDER BY ord ASC",
        )?;
        let rows = stmt.query_map([session_id], |r| {
            Ok(Persona {
                name: r.get(0)?,
                model: r.get(1)?,
                blurb: r.get(2)?,
            })
        })?;
        Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
    }

    /// Replace a session's whole `/swarm` roster with `personas`, in order.
    pub fn save_swarm_personas(&self, session_id: &str, personas: &[Persona]) -> Result<()> {
        self.conn.execute(
            "DELETE FROM swarm_personas WHERE session_id = ?1",
            [session_id],
        )?;
        for (i, p) in personas.iter().enumerate() {
            self.conn.execute(
                "INSERT INTO swarm_personas (session_id, ord, name, model, persona)
                 VALUES (?1, ?2, ?3, ?4, ?5)",
                (session_id, i as i64, &p.name, &p.model, &p.blurb),
            )?;
        }
        Ok(())
    }

    /// Set a session's `research_parent_id` after creation (e.g. when
    /// a regular chat is promoted to research and the original session is
    /// created first).
    pub fn set_research_parent(&self, id: &str, parent_id: &str) -> Result<()> {
        self.conn.execute(
            "UPDATE sessions SET research_parent_id = ?2 WHERE id = ?1",
            (id, parent_id),
        )?;
        Ok(())
    }

    /// Set a session's title and (optionally) its generated slug.
    pub fn set_session_title(&self, id: &str, title: &str, slug: Option<&str>) -> Result<()> {
        self.conn.execute(
            "UPDATE sessions SET title = ?2, slug = COALESCE(?3, slug) WHERE id = ?1",
            (id, title, slug),
        )?;
        Ok(())
    }

    /// Delete a single message row by id — used to roll back a persisted
    /// `gate_reply` whose channel delivery failed, so a retry can't
    /// duplicate it in the transcript.
    pub fn delete_message(&self, id: &str) -> Result<()> {
        self.conn
            .execute("DELETE FROM messages WHERE id = ?1", [id])?;
        Ok(())
    }

    /// Delete a session and all its messages.
    pub fn delete_session(&self, id: &str) -> Result<()> {
        self.conn
            .execute("DELETE FROM messages WHERE session_id = ?1", [id])?;
        self.conn
            .execute("DELETE FROM sessions WHERE id = ?1", [id])?;
        Ok(())
    }

    pub fn load_messages(&self, session_id: &str) -> Result<Vec<Message>> {
        let mut stmt = self.conn.prepare(
            "SELECT role, content, model, reasoning, tokens, secs, cost, phrase, persona, created_at
             FROM messages WHERE session_id = ?1 ORDER BY created_at ASC",
        )?;
        let messages = stmt
            .query_map([session_id], |r| {
                Ok(Message {
                    role: r.get(0)?,
                    content: r.get(1)?,
                    model: r.get(2)?,
                    reasoning: r.get(3)?,
                    tokens: r.get(4)?,
                    secs: r.get(5)?,
                    cost: r.get(6)?,
                    phrase: r.get(7)?,
                    persona: r.get(8)?,
                    created_at: r.get(9)?,
                })
            })?
            .collect::<rusqlite::Result<Vec<_>>>()?;
        Ok(messages)
    }

    /// Insert a user message (no model/reasoning/stats). Returns its id.
    pub fn add_user_message(&self, session_id: &str, content: &str) -> Result<String> {
        self.insert_message(
            session_id, "user", content, None, None, None, None, None, None,
        )
    }

    /// A user's reply to a survey/approval gate: rendered in the transcript
    /// like a user message but never replayed to the model (`gate_reply`
    /// role) — the survey/plan rows it answers are excluded from model
    /// history too, so bare answers ("the second option", "drop Q2") must
    /// not reach the model without their context.
    pub fn add_gate_reply_message(&self, session_id: &str, content: &str) -> Result<String> {
        self.insert_message(
            session_id,
            "gate_reply",
            content,
            None,
            None,
            None,
            None,
            None,
            None,
        )
    }

    /// Insert a tool-call transcript block: `content` is JSON
    /// `{"name","arguments","result"}`. Never sent back to the model.
    pub fn add_tool_call_message(&self, session_id: &str, content: &str) -> Result<String> {
        self.insert_message(
            session_id,
            "tool_call",
            content,
            None,
            None,
            None,
            None,
            None,
            None,
        )
    }

    /// Insert a failed-response line. It remains visible in the transcript
    /// after the status bar changes, but is never replayed to the model.
    pub fn add_error_message(&self, session_id: &str, content: &str) -> Result<String> {
        self.insert_message(
            session_id, "error", content, None, None, None, None, None, None,
        )
    }

    /// Insert a background-research stage/progress line: plain text, shown in
    /// the transcript but never sent back to the model (unlike `tool_call`
    /// rows, never replayed into `build_history` either — this is the job's
    /// own scratch work, not something the chat model did).
    pub fn add_research_stage_message(&self, session_id: &str, content: &str) -> Result<String> {
        self.insert_message(
            session_id,
            "research_stage",
            content,
            None,
            None,
            None,
            None,
            None,
            None,
        )
    }

    /// A research pipeline's plan-approval prompt: rendered like a stage row
    /// but actionable, and (like `research_stage`) never replayed to the model.
    pub fn add_research_plan_message(&self, session_id: &str, content: &str) -> Result<String> {
        self.insert_message(
            session_id,
            "research_plan",
            content,
            None,
            None,
            None,
            None,
            None,
            None,
        )
    }

    /// A research pipeline's clarifying-survey section: the scoping agent's
    /// questions awaiting a chat answer. Rendered like a stage row but
    /// actionable, and never replayed to the model.
    pub fn add_survey_message(&self, session_id: &str, content: &str) -> Result<String> {
        self.insert_message(
            session_id, "survey", content, None, None, None, None, None, None,
        )
    }

    /// Update the most recent `research_stage` row for `session_id` whose
    /// content starts with `label`, or insert one on the stage's first
    /// occurrence — keeps one transcript row per named stage instead of
    /// appending on every progress tick (e.g. every searcher finishing).
    pub fn upsert_research_stage_message(
        &self,
        session_id: &str,
        label: &str,
        detail: &str,
    ) -> Result<()> {
        let content = stage_content(label, detail);
        let existing: Option<String> = self
            .conn
            .query_row(
                "SELECT id FROM messages WHERE session_id = ?1 AND role = 'research_stage'
                   AND (content = ?2 OR content LIKE ?3)
                 ORDER BY created_at DESC LIMIT 1",
                (session_id, label, format!("{label}:%")),
                |r| r.get(0),
            )
            .ok();
        match existing {
            Some(id) => {
                let now = Utc::now().to_rfc3339();
                self.conn.execute(
                    "UPDATE messages SET content = ?2, created_at = ?3 WHERE id = ?1",
                    (&id, &content, &now),
                )?;
            }
            None => {
                self.add_research_stage_message(session_id, &content)?;
            }
        }
        Ok(())
    }

    /// See the free function of the same name. Production code (the
    /// research pipeline task) writes through its own connection; this
    /// handle exists for tests.
    #[cfg(test)]
    pub fn add_session_sources(&self, session_id: &str, url_norms: &[String]) -> Result<()> {
        add_session_sources(&self.conn, session_id, url_norms)
    }

    /// See the free function of the same name.
    #[cfg(test)]
    pub fn search_session_sources(
        &self,
        session_id: &str,
        query: &str,
    ) -> Result<Vec<(String, String)>> {
        search_session_sources(&self.conn, session_id, query)
    }

    /// Pin (`Some("pinned")`), discard (`Some("discarded")`), or clear
    /// (`None`) a session source's flag. `url_norm` must already exist in
    /// `session_sources` for this session (a no-op UPDATE otherwise — the
    /// row is created by `add_session_sources` when a source is first
    /// cited, not here).
    pub fn set_source_flag(
        &self,
        session_id: &str,
        url_norm: &str,
        flag: Option<&str>,
    ) -> Result<()> {
        self.conn.execute(
            "UPDATE session_sources SET flag = ?3 WHERE session_id = ?1 AND url_norm = ?2",
            (session_id, url_norm, flag),
        )?;
        Ok(())
    }

    /// Insert an assistant reply with its model, reasoning trace, and stats.
    /// `cost` is the request's USD total at catalog list price (`None` when
    /// the model's price is unknown).
    /// Args mirror the messages table columns; ~25 call sites pass inline
    /// `None`s for unused fields, so a struct would churn all of them.
    #[allow(clippy::too_many_arguments)]
    pub fn add_assistant_message(
        &self,
        session_id: &str,
        content: &str,
        model: Option<&str>,
        reasoning: Option<&str>,
        tokens: Option<i64>,
        secs: Option<f64>,
        cost: Option<f64>,
        phrase: Option<&str>,
    ) -> Result<String> {
        self.insert_message(
            session_id,
            "assistant",
            content,
            model,
            reasoning,
            tokens,
            secs,
            cost,
            phrase,
        )
    }

    /// Insert a `/swarm` persona's round reply: an assistant message tagged
    /// with which persona (and its own model) produced it.
    pub fn add_persona_message(
        &self,
        session_id: &str,
        content: &str,
        persona_name: &str,
        model: &str,
    ) -> Result<String> {
        let id = self.insert_message(
            session_id,
            "assistant",
            content,
            Some(model),
            None,
            None,
            None,
            None,
            None,
        )?;
        self.conn.execute(
            "UPDATE messages SET persona = ?2 WHERE id = ?1",
            (&id, persona_name),
        )?;
        Ok(id)
    }

    /// Shared message-row insert; kept flat for the same reason as
    /// `add_assistant_message` — column-shaped params, many inline callers.
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn insert_message(
        &self,
        session_id: &str,
        role: &str,
        content: &str,
        model: Option<&str>,
        reasoning: Option<&str>,
        tokens: Option<i64>,
        secs: Option<f64>,
        cost: Option<f64>,
        phrase: Option<&str>,
    ) -> Result<String> {
        let now = Utc::now().to_rfc3339();
        let id = Uuid::new_v4().to_string();
        self.conn.execute(
            "INSERT INTO messages
                (id, session_id, role, content, model, reasoning, tokens, secs, cost, phrase, created_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
            (
                &id, session_id, role, content, model, reasoning, tokens, secs, cost, phrase, &now,
            ),
        )?;
        self.conn.execute(
            "UPDATE sessions SET updated_at = ?2 WHERE id = ?1",
            (session_id, &now),
        )?;
        Ok(id)
    }

    /// `created_at` of the message at `index` (0-based, transcript order) —
    /// used to anchor a compaction row at the boundary without loading the
    /// whole session (e.g. when the job finishes after the user switched
    /// sessions). `None` when the session has fewer than `index + 1` messages.
    pub fn message_created_at(&self, session_id: &str, index: usize) -> Result<Option<String>> {
        let mut stmt = self.conn.prepare(
            "SELECT created_at FROM messages WHERE session_id = ?1
             ORDER BY created_at ASC LIMIT 1 OFFSET ?2",
        )?;
        Ok(stmt
            .query_row((session_id, index as i64), |r| r.get(0))
            .optional()?)
    }

    /// Insert a compaction-digest row at the exact `created_at` position —
    /// the timestamp of the last message the digest covers, so reloads keep
    /// the digest at the compaction boundary (right after the raw messages
    /// it summarizes) instead of at the end of the transcript. Unlike
    /// `insert_message`, this does not bump the session's `updated_at`:
    /// compacting is bookkeeping, not new activity.
    pub fn add_compaction_message(
        &self,
        session_id: &str,
        content: &str,
        at: &str,
    ) -> Result<String> {
        let id = Uuid::new_v4().to_string();
        self.conn.execute(
            "INSERT INTO messages
                (id, session_id, role, content, model, reasoning, tokens, secs, phrase, created_at)
             VALUES (?1, ?2, 'compaction', ?3, NULL, NULL, NULL, NULL, NULL, ?4)",
            (&id, session_id, content, at),
        )?;
        Ok(id)
    }

    /// Replace the session's compaction row's content in place — a later
    /// compaction folds new messages into the same digest, so there is
    /// exactly one row per session. Returns the number of rows updated
    /// (0 = the session has no compaction row yet).
    pub fn update_compaction_message(&self, session_id: &str, content: &str) -> Result<usize> {
        Ok(self.conn.execute(
            "UPDATE messages SET content = ?2
             WHERE session_id = ?1 AND role = 'compaction'",
            (session_id, content),
        )?)
    }

    pub fn set_session_model(&self, session_id: &str, model: &str) -> Result<()> {
        self.conn.execute(
            "UPDATE sessions SET model = ?2 WHERE id = ?1",
            (session_id, model),
        )?;
        Ok(())
    }

    // --- space filesets ---

    /// Insert or replace a file row (unique per space+name). Returns the row id;
    /// an existing row keeps its id, so its chunks can be replaced by `file_id`.
    pub fn upsert_file(
        &self,
        space_id: &str,
        name: &str,
        hash: &str,
        size: i64,
        status: &str,
    ) -> Result<String> {
        if let Ok(existing) = self.conn.query_row(
            "SELECT id FROM files WHERE space_id = ?1 AND name = ?2",
            (space_id, name),
            |r| r.get::<_, String>(0),
        ) {
            self.conn.execute(
                "UPDATE files SET hash = ?2, size = ?3, status = ?4 WHERE id = ?1",
                (&existing, hash, size, status),
            )?;
            return Ok(existing);
        }
        let id = Uuid::new_v4().to_string();
        let now = Utc::now().to_rfc3339();
        self.conn.execute(
            "INSERT INTO files (id, space_id, name, hash, size, status, created_at)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
            (&id, space_id, name, hash, size, status, &now),
        )?;
        Ok(id)
    }

    pub fn list_files(&self, space_id: &str) -> Result<Vec<FileRow>> {
        let mut stmt = self.conn.prepare(
            "SELECT id, name, hash, size, status, mtime FROM files
             WHERE space_id = ?1 ORDER BY name ASC",
        )?;
        let rows = stmt.query_map([space_id], |r| {
            Ok(FileRow {
                id: r.get(0)?,
                name: r.get(1)?,
                hash: r.get(2)?,
                size: r.get(3)?,
                status: r.get(4)?,
                mtime: r.get(5)?,
            })
        })?;
        Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
    }

    pub fn delete_file(&self, file_id: &str) -> Result<()> {
        self.conn
            .execute("DELETE FROM file_chunks WHERE file_id = ?1", [file_id])?;
        self.conn
            .execute("DELETE FROM files WHERE id = ?1", [file_id])?;
        Ok(())
    }

    /// Record the disk mtime a file was indexed at (see `FileRow::mtime`).
    pub fn set_file_mtime(&self, file_id: &str, mtime: i64) -> Result<()> {
        self.conn.execute(
            "UPDATE files SET mtime = ?2 WHERE id = ?1",
            (file_id, mtime),
        )?;
        Ok(())
    }

    /// Update a file's status column (e.g. "ok", "ocr…", or an error message).
    pub fn set_file_status(&self, file_id: &str, status: &str) -> Result<()> {
        self.conn.execute(
            "UPDATE files SET status = ?2 WHERE id = ?1",
            (file_id, status),
        )?;
        Ok(())
    }

    pub fn rename_file(&self, file_id: &str, new_name: &str) -> Result<()> {
        self.conn.execute(
            "UPDATE files SET name = ?2 WHERE id = ?1",
            (file_id, new_name),
        )?;
        Ok(())
    }

    /// Replace all occurrences of `old_name` with `new_name` in message content
    /// within the given space. Used when OCR renames a pasted image to a
    /// descriptive filename — updates `![alt](old_name)` → `![alt](new_name)`.
    pub fn replace_file_ref_in_messages(
        &self,
        space_id: &str,
        old_name: &str,
        new_name: &str,
    ) -> Result<()> {
        self.conn.execute(
            "UPDATE messages SET content = REPLACE(content, ?1, ?2)
             WHERE session_id IN (SELECT id FROM sessions WHERE space_id = ?3)",
            (old_name, new_name, space_id),
        )?;
        Ok(())
    }

    /// Replace a file's indexed chunks. `chunks` are `(location, text)` in
    /// order. Any stored embeddings are dropped too — they described the old
    /// chunk texts, and the embedder backfills the new ones.
    pub fn set_file_chunks(&self, file_id: &str, chunks: &[(String, String)]) -> Result<()> {
        self.conn
            .execute("DELETE FROM file_chunks WHERE file_id = ?1", [file_id])?;
        self.conn
            .execute("DELETE FROM chunk_embeddings WHERE file_id = ?1", [file_id])?;
        for (seq, (location, text)) in chunks.iter().enumerate() {
            self.conn.execute(
                "INSERT INTO file_chunks (file_id, seq, location, text) VALUES (?1, ?2, ?3, ?4)",
                (file_id, seq as i64, location, text),
            )?;
        }
        Ok(())
    }

    /// The underlying connection, for tests exercising the free query
    /// functions the toolbox reaches by opening the db path itself.
    #[cfg(test)]
    pub fn raw(&self) -> &Connection {
        &self.conn
    }

    /// A file's chunk texts as `(seq, text)`, in order — the embedder's input.
    pub fn file_chunk_texts(&self, file_id: &str) -> Result<Vec<(i64, String)>> {
        let mut stmt = self.conn.prepare(
            "SELECT CAST(seq AS INTEGER), text FROM file_chunks
             WHERE file_id = ?1 ORDER BY CAST(seq AS INTEGER) ASC",
        )?;
        let rows = stmt.query_map([file_id], |r| Ok((r.get(0)?, r.get(1)?)))?;
        Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
    }

    /// See the free function of the same name.
    pub fn files_missing_embeddings(&self, space_id: &str) -> Result<Vec<String>> {
        files_missing_embeddings(&self.conn, space_id)
    }

    /// Record a research report's cited sources for the citation index.
    pub fn add_citations(
        &self,
        space_id: &str,
        report_file: &str,
        citations: &[(String, Option<String>)],
    ) -> Result<()> {
        for (url, title) in citations {
            self.conn.execute(
                "INSERT INTO citations (space_id, report_file, url, title) VALUES (?1, ?2, ?3, ?4)",
                (space_id, report_file, url, title),
            )?;
        }
        Ok(())
    }

    /// See the free function of the same name. Most production code reads
    /// citations through the toolbox's own connection (the free function),
    /// but this handle is also used directly by the watch diff-section
    /// lookup (`previous_citations_for_watch_session`), plus tests.
    pub fn search_citations(
        &self,
        space_id: &str,
        query: Option<&str>,
    ) -> Result<Vec<(String, String, String)>> {
        search_citations(&self.conn, space_id, query)
    }

    /// Store embedding vectors for a file's chunks as `(seq, vector)` pairs.
    pub fn set_chunk_embeddings(&self, file_id: &str, vecs: &[(i64, Vec<f32>)]) -> Result<()> {
        for (seq, v) in vecs {
            self.conn.execute(
                "INSERT OR REPLACE INTO chunk_embeddings (file_id, seq, vec) VALUES (?1, ?2, ?3)",
                (file_id, seq, vec_to_blob(v)),
            )?;
        }
        Ok(())
    }

    pub fn create_watch(
        &self,
        space_id: &str,
        topic: &str,
        interval_hours: i64,
        session_id: &str,
    ) -> Result<String> {
        let id = Uuid::new_v4().to_string();
        self.conn.execute(
            "INSERT INTO watches (id, space_id, topic, interval_hours, session_id, last_run_at)
             VALUES (?1, ?2, ?3, ?4, ?5, NULL)",
            (&id, space_id, topic, interval_hours, session_id),
        )?;
        Ok(id)
    }

    pub fn list_watches(&self, space_id: &str) -> Result<Vec<Watch>> {
        let mut stmt = self.conn.prepare(
            "SELECT id, space_id, topic, interval_hours, session_id, last_run_at
             FROM watches WHERE space_id = ?1 ORDER BY topic",
        )?;
        let rows = stmt.query_map([space_id], |r| {
            Ok(Watch {
                id: r.get(0)?,
                space_id: r.get(1)?,
                topic: r.get(2)?,
                interval_hours: r.get(3)?,
                session_id: r.get(4)?,
                last_run_at: r.get(5)?,
            })
        })?;
        Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
    }

    /// Every watch across all spaces — used by the startup due-check, which
    /// runs before any space is necessarily "active".
    pub fn list_all_watches(&self) -> Result<Vec<Watch>> {
        let mut stmt = self.conn.prepare(
            "SELECT id, space_id, topic, interval_hours, session_id, last_run_at FROM watches",
        )?;
        let rows = stmt.query_map([], |r| {
            Ok(Watch {
                id: r.get(0)?,
                space_id: r.get(1)?,
                topic: r.get(2)?,
                interval_hours: r.get(3)?,
                session_id: r.get(4)?,
                last_run_at: r.get(5)?,
            })
        })?;
        Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
    }

    pub fn touch_watch(&self, id: &str, now_rfc3339: &str) -> Result<()> {
        self.conn.execute(
            "UPDATE watches SET last_run_at = ?2 WHERE id = ?1",
            (id, now_rfc3339),
        )?;
        Ok(())
    }

    /// Repoint a watch at the session its most recent re-run actually used,
    /// so the next due-check's diff-section lookup
    /// (`previous_citations_for_watch_session`) can match against it.
    pub fn set_watch_session(&self, id: &str, session_id: &str) -> Result<()> {
        self.conn.execute(
            "UPDATE watches SET session_id = ?2 WHERE id = ?1",
            (id, session_id),
        )?;
        Ok(())
    }

    pub fn delete_watch(&self, id: &str) -> Result<()> {
        self.conn
            .execute("DELETE FROM watches WHERE id = ?1", [id])?;
        Ok(())
    }
}

/// Encode an embedding as little-endian f32 bytes for a BLOB column.
pub fn vec_to_blob(v: &[f32]) -> Vec<u8> {
    v.iter().flat_map(|f| f.to_le_bytes()).collect()
}

/// Decode a BLOB back into an embedding (inverse of `vec_to_blob`).
pub fn blob_to_vec(b: &[u8]) -> Vec<f32> {
    b.chunks_exact(4)
        .map(|c| f32::from_le_bytes([c[0], c[1], c[2], c[3]]))
        .collect()
}

/// Citations in `space_id` whose `url/title/report_file` contains `query`
/// (case-insensitive substring), or every row when `query` is None — as
/// `(report_file, url, title)`, newest first. Free function so the toolbox
/// can call it over its own short-lived connection.
pub fn search_citations(
    conn: &Connection,
    space_id: &str,
    query: Option<&str>,
) -> Result<Vec<(String, String, String)>> {
    let mut stmt = conn.prepare(
        "SELECT report_file, url, COALESCE(title, '') FROM citations
         WHERE space_id = ?1
           AND (?2 IS NULL OR url LIKE ?2 OR title LIKE ?2 OR report_file LIKE ?2)
         ORDER BY id DESC",
    )?;
    let pattern = query.map(|q| format!("%{q}%"));
    let rows = stmt.query_map((space_id, pattern), |r| {
        Ok((r.get(0)?, r.get(1)?, r.get(2)?))
    })?;
    Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
}

/// One transcript line for a research stage: bare label, or `label: detail`.
pub fn stage_content(label: &str, detail: &str) -> String {
    if detail.is_empty() {
        label.to_string()
    } else {
        format!("{label}: {detail}")
    }
}

/// See `Db::add_session_sources`; free so the research pipeline task can
/// call it over its own short-lived connection.
pub fn add_session_sources(
    conn: &Connection,
    session_id: &str,
    url_norms: &[String],
) -> Result<()> {
    for u in url_norms {
        conn.execute(
            "INSERT OR IGNORE INTO session_sources (session_id, url_norm) VALUES (?1, ?2)",
            (session_id, u),
        )?;
    }
    Ok(())
}

/// Keyword-search (plain substring, case-insensitive) a session's cached
/// source bundle: `(url, text)` for every cached page whose text contains
/// `query`. Ponytail: substring, not FTS — a bundle is a handful of pages,
/// not a corpus.
pub fn search_session_sources(
    conn: &Connection,
    session_id: &str,
    query: &str,
) -> Result<Vec<(String, String)>> {
    let mut stmt = conn.prepare(
        "SELECT web_cache.url, web_cache.text FROM session_sources
         JOIN web_cache ON web_cache.url_norm = session_sources.url_norm
         WHERE session_sources.session_id = ?1",
    )?;
    let rows = stmt.query_map([session_id], |r| {
        Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?))
    })?;
    let needle = query.to_lowercase();
    Ok(rows
        .collect::<rusqlite::Result<Vec<_>>>()?
        .into_iter()
        .filter(|(_, text)| text.to_lowercase().contains(&needle))
        .collect())
}

/// URLs pinned in a session's source bundle — the Synthesizer/Writer
/// prompts list these as "prioritize these sources".
pub fn pinned_urls(conn: &Connection, session_id: &str) -> Result<Vec<String>> {
    let mut stmt = conn.prepare(
        "SELECT url_norm FROM session_sources WHERE session_id = ?1 AND flag = 'pinned'",
    )?;
    let rows = stmt.query_map([session_id], |r| r.get::<_, String>(0))?;
    Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
}

/// Distinct hostnames discarded in a session — excluded from later searcher
/// rounds the same way the global `blocked_domains` setting is.
pub fn discarded_domains(conn: &Connection, session_id: &str) -> Result<Vec<String>> {
    let mut stmt = conn.prepare(
        "SELECT url_norm FROM session_sources WHERE session_id = ?1 AND flag = 'discarded'",
    )?;
    let rows: Vec<String> = stmt
        .query_map([session_id], |r| r.get::<_, String>(0))?
        .collect::<rusqlite::Result<Vec<_>>>()?;
    let mut hosts: Vec<String> = rows
        .iter()
        .filter_map(|u| {
            reqwest::Url::parse(u)
                .ok()
                .and_then(|p| p.host_str().map(str::to_string))
        })
        .collect();
    hosts.sort();
    hosts.dedup();
    Ok(hosts)
}

/// Whether a cached fetch (`fetched_at`, rfc3339) is still usable — under
/// 24h old. An unparseable timestamp is treated as stale, not an error:
/// the caller just re-fetches live.
pub fn is_fresh(fetched_at: &str, now: chrono::DateTime<Utc>) -> bool {
    chrono::DateTime::parse_from_rfc3339(fetched_at)
        .is_ok_and(|dt| now.signed_duration_since(dt) < chrono::Duration::hours(24))
}

/// A cached fetched page: (title, text, `fetched_at` rfc3339), or None on a
/// cache miss. Free function — the toolbox opens its own short-lived
/// connection by path, same as the file-search queries.
pub fn cache_get(conn: &Connection, url_norm: &str) -> Result<Option<(String, String, String)>> {
    let row = conn.query_row(
        "SELECT COALESCE(title, ''), text, fetched_at FROM web_cache WHERE url_norm = ?1",
        [url_norm],
        |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
    );
    match row {
        Ok(v) => Ok(Some(v)),
        Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
        Err(e) => Err(e.into()),
    }
}

/// Write (or overwrite) a fetched page into the cache, stamped now.
pub fn cache_put(
    conn: &Connection,
    url_norm: &str,
    url: &str,
    title: Option<&str>,
    text: &str,
) -> Result<()> {
    let now = Utc::now().to_rfc3339();
    conn.execute(
        "INSERT INTO web_cache (url_norm, url, title, text, fetched_at) VALUES (?1, ?2, ?3, ?4, ?5)
         ON CONFLICT(url_norm) DO UPDATE SET url = ?2, title = ?3, text = ?4, fetched_at = ?5",
        (url_norm, url, title, text, &now),
    )?;
    Ok(())
}

/// Ids of files (in one space) that have chunks but not a vector per chunk —
/// the embedder's work queue, which doubles as the pre-upgrade backfill.
pub fn files_missing_embeddings(conn: &Connection, space_id: &str) -> Result<Vec<String>> {
    let mut stmt = conn.prepare(
        "SELECT files.id FROM files
         WHERE files.space_id = ?1
           AND (SELECT COUNT(*) FROM file_chunks WHERE file_chunks.file_id = files.id) >
               (SELECT COUNT(*) FROM chunk_embeddings WHERE chunk_embeddings.file_id = files.id)",
    )?;
    let rows = stmt.query_map([space_id], |r| r.get(0))?;
    Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
}

/// Cosine-ranked chunk search within one space: `(file name, location, text,
/// score)`, best first. Vectors whose dimension doesn't match the query (a
/// changed embedding model) are skipped. Brute force — thousands of chunks
/// scan in milliseconds, no ANN index needed.
pub fn semantic_chunks(
    conn: &Connection,
    space_id: &str,
    query: &[f32],
    limit: usize,
) -> Result<Vec<(String, String, String, f32)>> {
    let mut stmt = conn.prepare(
        "SELECT files.name, file_chunks.location, file_chunks.text, chunk_embeddings.vec
         FROM chunk_embeddings
         JOIN files ON files.id = chunk_embeddings.file_id
         JOIN file_chunks ON file_chunks.file_id = chunk_embeddings.file_id
                         AND CAST(file_chunks.seq AS INTEGER) = chunk_embeddings.seq
         WHERE files.space_id = ?1",
    )?;
    let rows = stmt.query_map([space_id], |r| {
        Ok((
            r.get::<_, String>(0)?,
            r.get::<_, String>(1)?,
            r.get::<_, String>(2)?,
            r.get::<_, Vec<u8>>(3)?,
        ))
    })?;
    let mut hits: Vec<(String, String, String, f32)> = Vec::new();
    for row in rows {
        let (name, loc, text, blob) = row?;
        let v = blob_to_vec(&blob);
        if v.len() != query.len() {
            continue;
        }
        let score = cosine(query, &v);
        hits.push((name, loc, text, score));
    }
    hits.sort_by(|a, b| b.3.partial_cmp(&a.3).unwrap_or(std::cmp::Ordering::Equal));
    hits.truncate(limit);
    Ok(hits)
}

fn cosine(a: &[f32], b: &[f32]) -> f32 {
    let (mut dot, mut na, mut nb) = (0.0f32, 0.0f32, 0.0f32);
    for (x, y) in a.iter().zip(b) {
        dot += x * y;
        na += x * x;
        nb += y * y;
    }
    let denom = na.sqrt() * nb.sqrt();
    if denom == 0.0 { 0.0 } else { dot / denom }
}

// --- usage analytics ---

/// Time window for the `/usage` dashboard: which logged requests count.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum UsageRange {
    Day,
    Week,
    Month,
    #[default]
    All,
}

impl UsageRange {
    /// Cycle order for the popup's range key (`←/→`).
    pub const CYCLE: [Self; 4] = [Self::Day, Self::Week, Self::Month, Self::All];

    /// Short badge label, e.g. the `24h` in the popup title.
    pub const fn label(self) -> &'static str {
        match self {
            Self::Day => "24h",
            Self::Week => "7d",
            Self::Month => "30d",
            Self::All => "all",
        }
    }

    /// Long form for titles and empty-state messages.
    pub const fn title(self) -> &'static str {
        match self {
            Self::Day => "last 24 hours",
            Self::Week => "last 7 days",
            Self::Month => "last 30 days",
            Self::All => "all time",
        }
    }

    /// Persisted `app_settings` key value.
    pub const fn key(self) -> &'static str {
        match self {
            Self::Day => "day",
            Self::Week => "week",
            Self::Month => "month",
            Self::All => "all",
        }
    }

    /// Parse a persisted `app_settings` value; unknown keys fall back to
    /// the default (all time).
    pub fn from_key(key: &str) -> Self {
        Self::CYCLE
            .iter()
            .copied()
            .find(|r| r.key() == key)
            .unwrap_or_default()
    }

    /// The next window in cycle order.
    pub const fn next(self) -> Self {
        match self {
            Self::Day => Self::Week,
            Self::Week => Self::Month,
            Self::Month => Self::All,
            Self::All => Self::Day,
        }
    }

    /// The previous window in cycle order.
    pub const fn prev(self) -> Self {
        match self {
            Self::Day => Self::All,
            Self::Week => Self::Day,
            Self::Month => Self::Week,
            Self::All => Self::Month,
        }
    }

    /// Inclusive cutoff timestamp for SQL filtering; `None` = no filter.
    pub fn since(self) -> Option<chrono::DateTime<chrono::Utc>> {
        use chrono::{Duration, Utc};
        match self {
            Self::Day => Some(Utc::now() - Duration::hours(24)),
            Self::Week => Some(Utc::now() - Duration::days(7)),
            Self::Month => Some(Utc::now() - Duration::days(30)),
            Self::All => None,
        }
    }

    /// Empty-state message for the dashboard/status line.
    pub const fn empty_message(self) -> &'static str {
        match self {
            Self::Day => "no usage in the last 24 hours — ←/→ for a wider window",
            Self::Week => "no usage in the last 7 days — ←/→ for a wider window",
            Self::Month => "no usage in the last 30 days — ←/→ for a wider window",
            Self::All => "no usage logged yet — send a message first",
        }
    }
}

/// The bare model name used by the OpenRouter catalog's `vendor/name` ids:
/// backend prefixes (`go:`, `openai:`, `codex:`, `opencode:`) and any
/// `vendor/` part are stripped (`go:deepseek-v4-flash` → `deepseek-v4-flash`,
/// `openai:gpt-5` → `gpt-5`). Empty when the id has no name left.
pub fn price_name(model: &str) -> &str {
    let stripped = ["go:", "openai:", "codex:", "opencode:"]
        .iter()
        .find_map(|p| model.strip_prefix(p))
        .unwrap_or(model);
    stripped.rsplit('/').next().unwrap_or(stripped)
}

/// Catalog price for a usage row's model: exact `model_prices` key first,
/// then the OpenRouter `vendor/name` entry matching the bare name (same
/// cross-backend fallback as `Db::model_price`, but against an in-memory
/// snapshot so a 28k-row backfill needs no per-row SQL).
fn catalog_price<'a>(
    prices: &'a std::collections::HashMap<String, (f64, f64)>,
    model: &str,
) -> Option<&'a (f64, f64)> {
    if let Some(price) = prices.get(model) {
        return Some(price);
    }
    let name = price_name(model);
    if name.is_empty() {
        return None;
    }
    prices
        .iter()
        .filter(|(id, _)| {
            id.strip_suffix(name)
                .is_some_and(|rest| rest.ends_with('/'))
        })
        .min_by_key(|(id, _)| id.len()) // shortest vendor wins, deterministic
        .map(|(_, price)| price)
}

/// Lifetime token/cost totals across every logged request.
#[derive(Default)]
pub struct UsageTotals {
    pub requests: u64,
    pub prompt_tokens: u64,
    pub completion_tokens: u64,
    pub cache_read_tokens: u64,
    pub cache_creation_tokens: u64,
    /// Total USD (0 when no model had a known price).
    pub cost: f64,
}

/// One backend's aggregate row.
#[derive(Default)]
pub struct UsageByBackend {
    pub backend: String,
    pub requests: u64,
    pub prompt_tokens: u64,
    pub completion_tokens: u64,
    pub cache_read_tokens: u64,
    pub cost: f64,
}

/// One model's aggregate row.
#[derive(Default)]
pub struct UsageByModel {
    pub model: String,
    pub requests: u64,
    pub prompt_tokens: u64,
    pub completion_tokens: u64,
    pub cache_read_tokens: u64,
    pub cost: f64,
}

/// One logged request, newest first.
#[derive(Default)]
pub struct UsageRow {
    pub created_at: String,
    pub backend: String,
    pub model: String,
    pub prompt_tokens: u64,
    pub completion_tokens: u64,
    pub cache_read_tokens: u64,
    pub cost: Option<f64>,
}

impl Db {
    /// Record one completed API request's usage. Content-free — only
    /// backend/model/tokens — so it never leaks conversation text.
    #[allow(clippy::too_many_arguments)]
    pub fn log_usage(
        &self,
        backend: &str,
        model: &str,
        prompt_tokens: u64,
        completion_tokens: u64,
        cache_read_tokens: u64,
        cache_creation_tokens: u64,
        cost: Option<f64>,
        session_id: Option<&str>,
        space_id: Option<&str>,
    ) -> Result<()> {
        self.conn.execute(
            "INSERT INTO usage_log (created_at, session_id, space_id, backend, model,
                prompt_tokens, completion_tokens, cache_read_tokens, cache_creation_tokens, cost)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
            (
                Utc::now().to_rfc3339(),
                session_id,
                space_id,
                backend,
                model,
                prompt_tokens as i64,
                completion_tokens as i64,
                cache_read_tokens as i64,
                cache_creation_tokens as i64,
                cost,
            ),
        )?;
        Ok(())
    }

    /// Cost of one completed request in USD at current catalog prices
    /// (`None` when no price is known). Non-OpenRouter backends expose no
    /// pricing API, so their models are priced via the OpenRouter catalog
    /// entry for the same model (see `model_price`). Prompt tokens are
    /// billed at the prompt rate even when served from cache; the catalog
    /// has no separate cache-read rate.
    pub fn request_cost(
        &self,
        model: &str,
        prompt_tokens: u64,
        completion_tokens: u64,
    ) -> Option<f64> {
        self.model_price(model)
            .map(|(prompt_price, completion_price)| {
                prompt_tokens as f64 / 1e6 * prompt_price
                    + completion_tokens as f64 / 1e6 * completion_price
            })
    }

    /// Reconcile every logged request's cost with the current `model_prices`
    /// catalog: NULL rows (logged before pricing existed) get the exact
    /// `tokens × price` total, and rows whose value no longer matches the
    /// catalog (legacy unit bug, price changes) are recomputed. Non-OpenRouter
    /// models are priced through their OpenRouter `vendor/name` twin, like
    /// `model_price`. Existing costs for models without any catalog price are
    /// left untouched — a model dropping out of the catalog shouldn't void
    /// its history.
    /// Idempotent — unchanged rows are not rewritten — so it can run after
    /// every catalog refresh and whenever the `/usage` popup opens.
    /// Returns how many rows were visited.
    pub fn backfill_usage_costs(&mut self) -> Result<usize> {
        // The catalog endpoint reports USD per token while every cost
        // formula here works in USD per 1M tokens. Heal a legacy
        // per-token-shaped catalog (MAX < $0.001/M is impossible for a
        // real catalog — o1-pro alone lists at $150/M) so backfills
        // computed against it produce real totals. Self-limiting: once
        // scaled (or refreshed by the fixed fetch path) it no longer
        // matches the shape.
        let max_price: f64 = self.conn.query_row(
            "SELECT COALESCE(MAX(prompt_price), 0) FROM model_prices",
            [],
            |r| r.get(0),
        )?;
        if max_price > 0.0 && max_price < 0.001 {
            self.conn.execute(
                "UPDATE model_prices
                 SET prompt_price = prompt_price * 1e6, completion_price = completion_price * 1e6",
                [],
            )?;
        }
        // Snapshot the catalog, then rewrite every usage row in one
        // transaction. 28k rows is a few ms even on a file DB.
        let mut prices: std::collections::HashMap<String, (f64, f64)> = Default::default();
        {
            let mut stmt = self
                .conn
                .prepare("SELECT model_id, prompt_price, completion_price FROM model_prices")?;
            let rows = stmt.query_map([], |r| {
                Ok((
                    r.get::<_, String>(0)?,
                    (r.get::<_, f64>(1)?, r.get::<_, f64>(2)?),
                ))
            })?;
            for row in rows {
                let (model, price) = row?;
                prices.insert(model, price);
            }
        }
        let rows: Vec<(i64, String, i64, i64, Option<f64>)> = {
            let mut stmt = self.conn.prepare(
                "SELECT id, model, prompt_tokens, completion_tokens, cost FROM usage_log",
            )?;
            let rows = stmt.query_map([], |r| {
                Ok((
                    r.get::<_, i64>(0)?,
                    r.get::<_, String>(1)?,
                    r.get::<_, i64>(2)?,
                    r.get::<_, i64>(3)?,
                    r.get::<_, Option<f64>>(4)?,
                ))
            })?;
            rows.collect::<rusqlite::Result<Vec<_>>>()?
        };
        let tx = self.conn.transaction()?;
        {
            let mut update = tx.prepare("UPDATE usage_log SET cost = ?2 WHERE id = ?1")?;
            for (id, model, prompt, completion, old) in &rows {
                let recomputed = catalog_price(&prices, model)
                    .map(|(p, c)| *prompt as f64 / 1e6 * p + *completion as f64 / 1e6 * c);
                // Fill rows logged before pricing existed, and heal values
                // computed against a stale catalog. Never erase an existing
                // cost when no price is currently known — a model dropping
                // out of the catalog shouldn't void its history.
                let write = match (recomputed, old) {
                    (Some(c), None) => Some(c),
                    (Some(c), Some(o)) if c != *o => Some(c),
                    _ => None,
                };
                if let Some(cost) = write {
                    update.execute((id, cost))?;
                }
            }
        }
        tx.commit()?;
        Ok(rows.len())
    }

    /// Batch save/refresh catalog prices in one transaction. The OpenRouter
    /// catalog is hundreds of models; per-row autocommits (each an fsync on
    /// the UI task) made the post-load pause noticeable. The `WHERE` clause
    /// also skips rows whose price didn't move, so re-fetches write nothing.
    pub fn upsert_model_prices(&mut self, prices: &[(String, String, f64, f64)]) -> Result<()> {
        if prices.is_empty() {
            return Ok(());
        }
        let tx = self.conn.transaction()?;
        {
            let mut stmt = tx.prepare(
                "INSERT INTO model_prices (model_id, backend, prompt_price, completion_price, updated_at)
                 VALUES (?1, ?2, ?3, ?4, ?5)
                 ON CONFLICT(model_id) DO UPDATE SET
                    prompt_price = excluded.prompt_price,
                    completion_price = excluded.completion_price,
                    updated_at = excluded.updated_at
                 WHERE model_prices.prompt_price != excluded.prompt_price
                    OR model_prices.completion_price != excluded.completion_price",
            )?;
            let now = Utc::now().to_rfc3339();
            for (model, backend, prompt, completion) in prices {
                stmt.execute((model, backend, prompt, completion, &now))?;
            }
        }
        tx.commit()?;
        Ok(())
    }

    /// Catalog price for a model, `(prompt, completion)` USD per 1M tokens.
    /// Tries the exact `model_prices` row first (OpenRouter ids match
    /// directly); if there is none, falls back to the OpenRouter catalog
    /// entry for the same model — backend prefixes (`go:`, `openai:`,
    /// `codex:`, `opencode:`) and the catalog's `vendor/` part are stripped
    /// (`go:deepseek-v4-flash` → `deepseek/deepseek-v4-flash`). The other
    /// backends' APIs expose no pricing, so OpenRouter's list price for the
    /// same model is the best available estimate.
    pub fn model_price(&self, model: &str) -> Option<(f64, f64)> {
        if let Ok(price) = self.conn.query_row(
            "SELECT prompt_price, completion_price FROM model_prices WHERE model_id = ?1",
            [model],
            |r| Ok((r.get::<_, f64>(0)?, r.get::<_, f64>(1)?)),
        ) {
            return Some(price);
        }
        let name = price_name(model);
        if name.is_empty() {
            return None;
        }
        // Suffix match on `vendor/name`: the shortest vendor wins when a
        // bare name appears under several vendors.
        self.conn
            .query_row(
                "SELECT prompt_price, completion_price FROM model_prices
                 WHERE backend = 'OpenRouter'
                   AND substr(model_id, -length(?1) - 1) = '/' || ?1
                 ORDER BY length(model_id) LIMIT 1",
                [name],
                |r| Ok((r.get::<_, f64>(0)?, r.get::<_, f64>(1)?)),
            )
            .ok()
    }

    /// Totals across logged requests, optionally limited to requests logged
    /// at or after `since` (RFC3339; `None` = all time). `created_at` is
    /// stored as fixed-width UTC RFC3339, so lexicographic comparison is a
    /// correct time filter.
    pub fn usage_totals(&self, since: Option<&str>) -> Result<UsageTotals> {
        let mut sql = String::from(
            "SELECT COUNT(*),
                    COALESCE(SUM(prompt_tokens), 0),
                    COALESCE(SUM(completion_tokens), 0),
                    COALESCE(SUM(cache_read_tokens), 0),
                    COALESCE(SUM(cache_creation_tokens), 0),
                    COALESCE(SUM(cost), 0)
             FROM usage_log",
        );
        if since.is_some() {
            sql.push_str(" WHERE created_at >= ?1");
        }
        let map = |r: &rusqlite::Row| {
            Ok(UsageTotals {
                requests: r.get::<_, i64>(0)? as u64,
                prompt_tokens: r.get::<_, i64>(1)? as u64,
                completion_tokens: r.get::<_, i64>(2)? as u64,
                cache_read_tokens: r.get::<_, i64>(3)? as u64,
                cache_creation_tokens: r.get::<_, i64>(4)? as u64,
                cost: r.get::<_, f64>(5)?,
            })
        };
        let totals = match since {
            Some(s) => self.conn.query_row(&sql, [s], map),
            None => self.conn.query_row(&sql, [], map),
        }?;
        Ok(totals)
    }

    /// Per-backend aggregates, most-used first. `since` filters the window
    /// (RFC3339 cutoff; `None` = all time).
    pub fn usage_by_backend(&self, since: Option<&str>) -> Result<Vec<UsageByBackend>> {
        let mut sql = String::from(
            "SELECT backend, COUNT(*),
                    COALESCE(SUM(prompt_tokens), 0),
                    COALESCE(SUM(completion_tokens), 0),
                    COALESCE(SUM(cache_read_tokens), 0),
                    COALESCE(SUM(cost), 0)
             FROM usage_log",
        );
        if since.is_some() {
            sql.push_str(" WHERE created_at >= ?1");
        }
        sql.push_str(" GROUP BY backend ORDER BY COUNT(*) DESC");
        let map = |r: &rusqlite::Row| {
            Ok(UsageByBackend {
                backend: r.get(0)?,
                requests: r.get::<_, i64>(1)? as u64,
                prompt_tokens: r.get::<_, i64>(2)? as u64,
                completion_tokens: r.get::<_, i64>(3)? as u64,
                cache_read_tokens: r.get::<_, i64>(4)? as u64,
                cost: r.get::<_, f64>(5)?,
            })
        };
        let mut stmt = self.conn.prepare(&sql)?;
        let rows = match since {
            Some(s) => stmt.query_map([s], map),
            None => stmt.query_map([], map),
        }?;
        Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
    }

    /// Per-model aggregates, most-used first. `since` filters the window
    /// (RFC3339 cutoff; `None` = all time).
    pub fn usage_by_model(&self, limit: u64, since: Option<&str>) -> Result<Vec<UsageByModel>> {
        let mut sql = String::from(
            "SELECT model, COUNT(*),
                    COALESCE(SUM(prompt_tokens), 0),
                    COALESCE(SUM(completion_tokens), 0),
                    COALESCE(SUM(cache_read_tokens), 0),
                    COALESCE(SUM(cost), 0)
             FROM usage_log",
        );
        if since.is_some() {
            sql.push_str(" WHERE created_at >= ?1");
            sql.push_str(" GROUP BY model ORDER BY COUNT(*) DESC LIMIT ?2");
        } else {
            sql.push_str(" GROUP BY model ORDER BY COUNT(*) DESC LIMIT ?1");
        }
        let map = |r: &rusqlite::Row| {
            Ok(UsageByModel {
                model: r.get(0)?,
                requests: r.get::<_, i64>(1)? as u64,
                prompt_tokens: r.get::<_, i64>(2)? as u64,
                completion_tokens: r.get::<_, i64>(3)? as u64,
                cache_read_tokens: r.get::<_, i64>(4)? as u64,
                cost: r.get::<_, f64>(5)?,
            })
        };
        let mut stmt = self.conn.prepare(&sql)?;
        let rows = match since {
            Some(s) => stmt.query_map(rusqlite::params![s, limit as i64], map),
            None => stmt.query_map(rusqlite::params![limit as i64], map),
        }?;
        Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
    }

    /// The most recent logged requests, newest first. `since` filters the
    /// window (RFC3339 cutoff; `None` = all time).
    pub fn usage_recent(&self, limit: u64, since: Option<&str>) -> Result<Vec<UsageRow>> {
        let mut sql = String::from(
            "SELECT created_at, backend, model, prompt_tokens, completion_tokens,
                    cache_read_tokens, cost
             FROM usage_log",
        );
        if since.is_some() {
            sql.push_str(" WHERE created_at >= ?1");
            sql.push_str(" ORDER BY id DESC LIMIT ?2");
        } else {
            sql.push_str(" ORDER BY id DESC LIMIT ?1");
        }
        let map = |r: &rusqlite::Row| {
            Ok(UsageRow {
                created_at: r.get(0)?,
                backend: r.get(1)?,
                model: r.get(2)?,
                prompt_tokens: r.get::<_, i64>(3)? as u64,
                completion_tokens: r.get::<_, i64>(4)? as u64,
                cache_read_tokens: r.get::<_, i64>(5)? as u64,
                cost: r.get(6)?,
            })
        };
        let mut stmt = self.conn.prepare(&sql)?;
        let rows = match since {
            Some(s) => stmt.query_map(rusqlite::params![s, limit as i64], map),
            None => stmt.query_map(rusqlite::params![limit as i64], map),
        }?;
        Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
    }
}

/// Quote a query for FTS5 MATCH: each whitespace token becomes a quoted
/// phrase (inner quotes doubled), so model-supplied text can't be an FTS
/// syntax error. Tokens are implicitly `ANDed` by FTS5.
pub fn fts_quote(query: &str) -> String {
    query
        .split_whitespace()
        .map(|t| format!("\"{}\"", t.replace('"', "\"\"")))
        .collect::<Vec<_>>()
        .join(" ")
}

/// BM25-ranked chunk search within one space: `(file name, location, snippet)`.
pub fn search_chunks(
    conn: &Connection,
    space_id: &str,
    query: &str,
    limit: usize,
) -> Result<Vec<(String, String, String)>> {
    let q = fts_quote(query);
    if q.is_empty() {
        return Ok(Vec::new());
    }
    let mut stmt = conn.prepare(
        "SELECT files.name, file_chunks.location,
                snippet(file_chunks, 3, '', '', '…', 24)
         FROM file_chunks JOIN files ON files.id = file_chunks.file_id
         WHERE file_chunks MATCH ?1 AND files.space_id = ?2
         ORDER BY bm25(file_chunks) LIMIT ?3",
    )?;
    let rows = stmt.query_map((q, space_id, limit as i64), |r| {
        Ok((r.get(0)?, r.get(1)?, r.get(2)?))
    })?;
    Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
}

/// A file's full extracted text (chunks re-joined in order), by display name.
pub fn file_text(conn: &Connection, space_id: &str, name: &str) -> Result<Option<String>> {
    let mut stmt = conn.prepare(
        "SELECT file_chunks.text
         FROM file_chunks JOIN files ON files.id = file_chunks.file_id
         WHERE files.space_id = ?1 AND files.name = ?2
         ORDER BY CAST(file_chunks.seq AS INTEGER) ASC",
    )?;
    let rows = stmt.query_map((space_id, name), |r| r.get::<_, String>(0))?;
    let parts = rows.collect::<rusqlite::Result<Vec<_>>>()?;
    Ok((!parts.is_empty()).then(|| parts.join("\n")))
}

pub fn count_files(conn: &Connection, space_id: &str) -> Result<u64> {
    let n: i64 = conn.query_row(
        "SELECT COUNT(*) FROM files WHERE space_id = ?1",
        [space_id],
        |r| r.get(0),
    )?;
    Ok(n as u64)
}

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

    #[test]
    fn is_fresh_true_under_24h_false_over() {
        let now = Utc::now();
        let recent = (now - chrono::Duration::hours(1)).to_rfc3339();
        let stale = (now - chrono::Duration::hours(25)).to_rfc3339();
        assert!(is_fresh(&recent, now));
        assert!(!is_fresh(&stale, now));
        assert!(!is_fresh("not a timestamp", now)); // unparseable = not fresh
    }

    #[test]
    fn usage_log_round_trips_and_aggregates() {
        let mut db = Db::open_in_memory().unwrap();
        db.upsert_model_prices(&[(
            "anthropic/claude-3.5-sonnet".to_string(),
            "OpenRouter".to_string(),
            3.0,
            15.0,
        )])
        .unwrap();
        assert_eq!(
            db.model_price("anthropic/claude-3.5-sonnet"),
            Some((3.0, 15.0))
        );
        // Re-upsert refreshes rather than duplicating.
        db.upsert_model_prices(&[(
            "anthropic/claude-3.5-sonnet".to_string(),
            "OpenRouter".to_string(),
            4.0,
            16.0,
        )])
        .unwrap();
        assert_eq!(
            db.model_price("anthropic/claude-3.5-sonnet"),
            Some((4.0, 16.0))
        );
        assert_eq!(db.model_price("unknown/model"), None);

        // 100 prompt @ $4/1M + 10 completion @ $16/1M = $0.0004 + $0.00016.
        db.log_usage(
            "OpenRouter",
            "anthropic/claude-3.5-sonnet",
            100,
            10,
            70,
            20,
            Some(0.00056),
            Some("s1"),
            Some("space-a"),
        )
        .unwrap();
        db.log_usage("Codex", "gpt-5.1-codex", 50, 5, 0, 0, None, None, None)
            .unwrap();

        let totals = db.usage_totals(None).unwrap();
        assert_eq!(totals.requests, 2);
        assert_eq!(totals.prompt_tokens, 150);
        assert_eq!(totals.completion_tokens, 15);
        assert_eq!(totals.cache_read_tokens, 70);
        assert_eq!(totals.cache_creation_tokens, 20);
        assert!((totals.cost - 0.00056).abs() < 1e-9);

        let by_backend = db.usage_by_backend(None).unwrap();
        assert_eq!(by_backend.len(), 2);
        assert_eq!(by_backend[0].backend, "OpenRouter"); // most-used first
        assert_eq!(by_backend[0].requests, 1);

        let by_model = db.usage_by_model(5, None).unwrap();
        assert_eq!(by_model.len(), 2);
        assert!(by_model.iter().any(|m| m.model == "gpt-5.1-codex"));

        let recent = db.usage_recent(10, None).unwrap();
        assert_eq!(recent.len(), 2);
        assert_eq!(recent[0].model, "gpt-5.1-codex"); // newest first
        assert_eq!(recent[0].cost, None);
        assert_eq!(recent[1].cache_read_tokens, 70);
    }

    #[test]
    fn usage_queries_filter_by_since_window() {
        let db = Db::open_in_memory().unwrap();
        // Rows with explicit timestamps (raw insert: log_usage stamps now).
        let insert = |created: &str| {
            db.raw().execute(
                "INSERT INTO usage_log (created_at, session_id, backend, model,
                    prompt_tokens, completion_tokens, cache_read_tokens, cache_creation_tokens, cost)
                 VALUES (?1, NULL, 'OpenRouter', 'a/model', 100, 10, 0, 0, 0.001)",
                [created],
            )
        };
        insert("2026-01-01T00:00:00+00:00").unwrap();
        insert("2026-01-02T00:00:00+00:00").unwrap();
        insert("2026-01-03T00:00:00+00:00").unwrap();

        let since = Some("2026-01-02T00:00:00+00:00");
        let totals = db.usage_totals(since).unwrap();
        assert_eq!(totals.requests, 2);
        assert_eq!(totals.prompt_tokens, 200);
        assert!((totals.cost - 0.002).abs() < 1e-12);
        assert_eq!(db.usage_totals(None).unwrap().requests, 3);
        assert_eq!(db.usage_by_backend(since).unwrap()[0].requests, 2);
        assert_eq!(db.usage_by_model(5, since).unwrap()[0].requests, 2);
        let recent = db.usage_recent(10, since).unwrap();
        assert_eq!(recent.len(), 2);
        assert_eq!(recent[0].created_at, "2026-01-03T00:00:00+00:00");
        // Boundary is inclusive: the exact-cutoff row is included.
        assert!(
            recent
                .iter()
                .any(|r| r.created_at == "2026-01-02T00:00:00+00:00")
        );
    }

    #[test]
    fn usage_range_cycles_and_persists() {
        use crate::db::UsageRange;
        assert_eq!(UsageRange::Day.next(), UsageRange::Week);
        assert_eq!(UsageRange::All.next(), UsageRange::Day);
        assert_eq!(UsageRange::Day.prev(), UsageRange::All);
        assert_eq!(UsageRange::from_key("month"), UsageRange::Month);
        assert_eq!(UsageRange::from_key("bogus"), UsageRange::All);
        assert_eq!(UsageRange::Week.key(), "week");
        assert_eq!(UsageRange::Day.title(), "last 24 hours");
        assert!(UsageRange::All.since().is_none());
        assert!(UsageRange::Day.since().is_some());
    }

    #[test]
    fn backfill_usage_costs_recomputes_history_from_catalog() {
        let mut db = Db::open_in_memory().unwrap();
        db.upsert_model_prices(&[(
            "anthropic/claude-3.5-sonnet".to_string(),
            "OpenRouter".to_string(),
            3.0,
            15.0,
        )])
        .unwrap();
        // Rows logged before pricing existed: one priced model (NULL cost),
        // one model with no catalog entry.
        db.log_usage(
            "OpenRouter",
            "anthropic/claude-3.5-sonnet",
            100,
            10,
            70,
            20,
            None,
            None,
            None,
        )
        .unwrap();
        db.log_usage("Codex", "gpt-5.1-codex", 50, 5, 0, 0, None, None, None)
            .unwrap();

        let visited = db.backfill_usage_costs().unwrap();
        assert_eq!(visited, 2);
        // 100 prompt @ $3/1M + 10 completion @ $15/1M = $0.0003 + $0.00015.
        let totals = db.usage_totals(None).unwrap();
        assert!((totals.cost - 0.00045).abs() < 1e-12);
        let recent = db.usage_recent(10, None).unwrap();
        assert!((recent[1].cost.unwrap() - 0.00045).abs() < 1e-12); // priced row filled in
        assert_eq!(recent[0].cost, None); // unknown price stays unknown

        // Idempotent: a second pass leaves the values untouched.
        db.backfill_usage_costs().unwrap();
        assert!((db.usage_totals(None).unwrap().cost - 0.00045).abs() < 1e-12);
    }

    #[test]
    fn backfill_usage_costs_heals_per_token_catalog() {
        // A legacy catalog holding the endpoint's raw per-token values
        // (deepseek-v4-flash at $0.08/M stores 8e-08) must be scaled to the
        // per-1M convention before costs are computed against it.
        let mut db = Db::open_in_memory().unwrap();
        db.upsert_model_prices(&[(
            "deepseek/deepseek-v4-flash-0731".to_string(),
            "OpenRouter".to_string(),
            8e-08,
            1.8e-07,
        )])
        .unwrap();
        db.log_usage(
            "OpenRouter",
            "deepseek/deepseek-v4-flash-0731",
            122_221,
            672,
            118_784,
            0,
            Some(9.89864e-09), // the old, 1e6×-too-small value
            None,
            None,
        )
        .unwrap();

        db.backfill_usage_costs().unwrap();

        assert_eq!(
            db.model_price("deepseek/deepseek-v4-flash-0731"),
            Some((0.08, 0.18))
        );
        // 122221/1e6 × 0.08 + 672/1e6 × 0.18 ≈ $0.00989.
        let recent = db.usage_recent(10, None).unwrap();
        let cost = recent[0].cost.unwrap();
        assert!((cost - 0.0098986).abs() < 1e-6, "cost was {cost}");
        assert!(cost > 0.009, "cost was {cost}");
    }

    #[test]
    fn request_cost_prices_tokens_against_catalog() {
        let mut db = Db::open_in_memory().unwrap();
        assert_eq!(db.request_cost("unknown/model", 100, 10), None);
        db.upsert_model_prices(&[(
            "anthropic/claude-3.5-sonnet".to_string(),
            "OpenRouter".to_string(),
            3.0,
            15.0,
        )])
        .unwrap();
        let cost = db
            .request_cost("anthropic/claude-3.5-sonnet", 100, 10)
            .unwrap();
        assert!((cost - 0.00045).abs() < 1e-12);
    }

    #[test]
    fn model_price_cross_references_openrouter_catalog_twins() {
        let mut db = Db::open_in_memory().unwrap();
        db.upsert_model_prices(&[
            (
                "deepseek/deepseek-v4-flash".to_string(),
                "OpenRouter".to_string(),
                0.08,
                0.18,
            ),
            (
                "openai/gpt-5".to_string(),
                "OpenRouter".to_string(),
                1.25,
                10.0,
            ),
        ])
        .unwrap();
        // Exact ids hit directly; other backends' prefixed/bare ids resolve
        // through the vendor/name twin.
        assert_eq!(
            db.model_price("deepseek/deepseek-v4-flash"),
            Some((0.08, 0.18))
        );
        assert_eq!(db.model_price("go:deepseek-v4-flash"), Some((0.08, 0.18)));
        assert_eq!(db.model_price("deepseek-v4-flash"), Some((0.08, 0.18)));
        assert_eq!(db.model_price("openai:gpt-5"), Some((1.25, 10.0)));
        assert_eq!(db.model_price("codex:gpt-5"), Some((1.25, 10.0)));
        // No twin anywhere: unknown.
        assert_eq!(db.model_price("no-such-model-anywhere"), None);
        // The price flows into per-request costs for the other backend.
        let cost = db.request_cost("go:deepseek-v4-flash", 100, 10).unwrap();
        assert!((cost - 0.0000098).abs() < 1e-15);
    }

    #[test]
    fn price_name_strips_backend_prefixes_and_vendors() {
        assert_eq!(price_name("go:deepseek-v4-flash"), "deepseek-v4-flash");
        assert_eq!(price_name("openai:gpt-5"), "gpt-5");
        assert_eq!(price_name("codex:gpt-5.1-codex"), "gpt-5.1-codex");
        assert_eq!(price_name("opencode:qwen3.6-plus"), "qwen3.6-plus");
        assert_eq!(
            price_name("deepseek/deepseek-v4-flash"),
            "deepseek-v4-flash"
        );
        assert_eq!(price_name("gpt-5"), "gpt-5");
    }

    #[test]
    fn backfill_prices_non_openrouter_models_via_catalog_twins() {
        let mut db = Db::open_in_memory().unwrap();
        db.upsert_model_prices(&[(
            "deepseek/deepseek-v4-flash".to_string(),
            "OpenRouter".to_string(),
            0.08,
            0.18,
        )])
        .unwrap();
        // OpenCode Go rows logged with no cost — the flat-fee backend has
        // no pricing of its own, so the twin's list price is the estimate.
        db.log_usage(
            "OpenCode Go",
            "go:deepseek-v4-flash",
            100,
            10,
            0,
            0,
            None,
            None,
            None,
        )
        .unwrap();

        db.backfill_usage_costs().unwrap();

        let recent = db.usage_recent(10, None).unwrap();
        let cost = recent[0].cost.unwrap();
        assert!((cost - 0.0000098).abs() < 1e-15, "cost was {cost}");
        assert!((db.usage_totals(None).unwrap().cost - 0.0000098).abs() < 1e-15);
    }

    #[test]
    fn web_cache_roundtrips_and_updates_on_rewrite() {
        let db = Db::open_in_memory().unwrap();
        assert!(cache_get(db.raw(), "example.com/a").unwrap().is_none());
        cache_put(
            db.raw(),
            "example.com/a",
            "https://example.com/a",
            Some("Title"),
            "body text",
        )
        .unwrap();
        let (title, text, fetched_at) = cache_get(db.raw(), "example.com/a").unwrap().unwrap();
        assert_eq!(title, "Title");
        assert_eq!(text, "body text");
        assert!(!fetched_at.is_empty());

        // Re-fetching overwrites the row, not duplicates it.
        cache_put(
            db.raw(),
            "example.com/a",
            "https://example.com/a",
            None,
            "new body",
        )
        .unwrap();
        let (title, text, _) = cache_get(db.raw(), "example.com/a").unwrap().unwrap();
        assert_eq!(title, "");
        assert_eq!(text, "new body");
    }

    #[test]
    fn web_mode_defaults_off_and_toggles() {
        let db = Db::open_in_memory().unwrap();
        let space = db.default_space_id().unwrap();
        let s = db.create_session("t", "a/b", &space, "chat").unwrap();
        assert!(!s.web_mode);
        db.set_session_web_mode(&s.id, true).unwrap();
        assert!(db.list_sessions(&space).unwrap()[0].web_mode);
    }

    #[test]
    fn swarm_mode_defaults_off_and_toggles() {
        let db = Db::open_in_memory().unwrap();
        let space = db.default_space_id().unwrap();
        let s = db.create_session("t", "a/b", &space, "chat").unwrap();
        assert!(!s.swarm_mode);
        db.set_session_swarm_mode(&s.id, true).unwrap();
        assert!(db.list_sessions(&space).unwrap()[0].swarm_mode);
        assert!(db.get_session(&s.id).unwrap().unwrap().swarm_mode);
    }

    #[test]
    fn swarm_personas_roundtrip_and_replace_all_on_save() {
        let db = Db::open_in_memory().unwrap();
        let space = db.default_space_id().unwrap();
        let s = db.create_session("t", "a/b", &space, "chat").unwrap();
        assert!(db.list_swarm_personas(&s.id).unwrap().is_empty());

        let roster = vec![
            Persona {
                name: "Skeptic".into(),
                model: "a/one".into(),
                blurb: "pokes holes".into(),
            },
            Persona {
                name: "Advocate".into(),
                model: "b/two".into(),
                blurb: "user-first".into(),
            },
        ];
        db.save_swarm_personas(&s.id, &roster).unwrap();
        let loaded = db.list_swarm_personas(&s.id).unwrap();
        assert_eq!(loaded.len(), 2);
        assert_eq!(loaded[0].name, "Skeptic");
        assert_eq!(loaded[1].name, "Advocate");

        // A second save fully replaces the roster, not appends.
        db.save_swarm_personas(&s.id, &roster[..1]).unwrap();
        assert_eq!(db.list_swarm_personas(&s.id).unwrap().len(), 1);
    }

    #[test]
    fn persona_message_tags_role_assistant_with_persona_and_model() {
        let db = Db::open_in_memory().unwrap();
        let space = db.default_space_id().unwrap();
        let s = db.create_session("t", "a/b", &space, "chat").unwrap();
        db.add_persona_message(&s.id, "reply text", "Skeptic", "a/one")
            .unwrap();
        let msgs = db.load_messages(&s.id).unwrap();
        assert_eq!(msgs.len(), 1);
        assert_eq!(msgs[0].role, "assistant");
        assert_eq!(msgs[0].persona.as_deref(), Some("Skeptic"));
        assert_eq!(msgs[0].model.as_deref(), Some("a/one"));

        // An ordinary assistant message has no persona tag.
        db.add_assistant_message(&s.id, "final answer", None, None, None, None, None, None)
            .unwrap();
        let msgs = db.load_messages(&s.id).unwrap();
        assert_eq!(msgs[1].persona, None);
    }

    #[test]
    fn session_sources_link_to_the_web_cache_and_are_keyword_searchable() {
        let db = Db::open_in_memory().unwrap();
        let space = db.default_space_id().unwrap();
        let s = db.create_session("t", "a/b", &space, "chat").unwrap();
        cache_put(
            db.raw(),
            "https://example.com/a",
            "https://example.com/a",
            Some("A"),
            "rust borrow checker deep dive",
        )
        .unwrap();
        cache_put(
            db.raw(),
            "https://example.com/b",
            "https://example.com/b",
            Some("B"),
            "cooking pasta recipes",
        )
        .unwrap();
        db.add_session_sources(
            &s.id,
            &[
                "https://example.com/a".to_string(),
                "https://example.com/b".to_string(),
            ],
        )
        .unwrap();

        let hits = db.search_session_sources(&s.id, "borrow checker").unwrap();
        assert_eq!(hits.len(), 1);
        assert!(hits[0].1.contains("borrow checker"));

        assert!(
            db.search_session_sources(&s.id, "quantum")
                .unwrap()
                .is_empty()
        );
    }

    #[test]
    fn set_source_flag_pins_and_discards_then_clears() {
        let db = Db::open_in_memory().unwrap();
        let session_id = "sess-1";
        add_session_sources(&db.conn, session_id, &["https://a.example/x".to_string()]).unwrap();
        db.set_source_flag(session_id, "https://a.example/x", Some("pinned"))
            .unwrap();
        assert_eq!(
            pinned_urls(&db.conn, session_id).unwrap(),
            vec!["https://a.example/x".to_string()]
        );
        assert!(discarded_domains(&db.conn, session_id).unwrap().is_empty());

        db.set_source_flag(session_id, "https://a.example/x", Some("discarded"))
            .unwrap();
        assert!(pinned_urls(&db.conn, session_id).unwrap().is_empty());
        assert_eq!(
            discarded_domains(&db.conn, session_id).unwrap(),
            vec!["a.example".to_string()]
        );

        db.set_source_flag(session_id, "https://a.example/x", None)
            .unwrap();
        assert!(discarded_domains(&db.conn, session_id).unwrap().is_empty());
    }

    #[test]
    fn upsert_research_stage_message_replaces_the_same_labels_row() {
        let db = Db::open_in_memory().unwrap();
        let space = db.default_space_id().unwrap();
        let s = db.create_session("t", "a/b", &space, "chat").unwrap();
        db.upsert_research_stage_message(&s.id, "searching", "round 1, 1/3")
            .unwrap();
        db.upsert_research_stage_message(&s.id, "searching", "round 1, 2/3")
            .unwrap();
        db.upsert_research_stage_message(&s.id, "planning", "")
            .unwrap();

        let msgs = db.load_messages(&s.id).unwrap();
        let searching: Vec<_> = msgs
            .iter()
            .filter(|m| m.content.starts_with("searching:"))
            .collect();
        assert_eq!(searching.len(), 1, "expected one row, updated in place");
        assert!(searching[0].content.contains("2/3"));
        assert_eq!(msgs.iter().filter(|m| m.content == "planning").count(), 1);
    }

    #[test]
    fn session_and_message_roundtrip() {
        let db = Db::open_in_memory().unwrap();
        let space = db.default_space_id().unwrap();
        let s = db
            .create_session("hello", "openai/gpt-4o", &space, "chat")
            .unwrap();

        db.add_user_message(&s.id, "hi").unwrap();
        db.add_assistant_message(
            &s.id,
            "hello there",
            Some("openai/gpt-4o"),
            Some("let me think"),
            Some(3),
            Some(1.5),
            Some(0.0042),
            Some("Vibed"),
        )
        .unwrap();

        let msgs = db.load_messages(&s.id).unwrap();
        assert_eq!(msgs.len(), 2);
        assert_eq!(msgs[0].role, "user");
        assert_eq!(msgs[1].content, "hello there");
        assert_eq!(msgs[1].model.as_deref(), Some("openai/gpt-4o"));
        assert_eq!(msgs[1].reasoning.as_deref(), Some("let me think"));
        assert_eq!(msgs[1].tokens, Some(3));
        assert_eq!(msgs[1].cost, Some(0.0042));

        let sessions = db.list_sessions(&space).unwrap();
        assert_eq!(sessions.len(), 1);
        assert_eq!(sessions[0].id, s.id);
    }

    #[test]
    fn markdown_images_in_content_roundtrip() {
        let db = Db::open_in_memory().unwrap();
        let space = db.default_space_id().unwrap();
        let s = db.create_session("t", "a/b", &space, "chat").unwrap();
        let content = "look at ![this](img.png) and ![that](other.png)";
        db.add_user_message(&s.id, content).unwrap();
        let msgs = db.load_messages(&s.id).unwrap();
        assert_eq!(msgs.len(), 1);
        assert!(msgs[0].content.contains("![this](img.png)"));
        assert!(msgs[0].content.contains("![that](other.png)"));
    }

    #[test]
    fn model_prefs_toggle_and_used() {
        let db = Db::open_in_memory().unwrap();
        assert!(db.toggle_favorite("a/one").unwrap()); // now favorite
        assert!(!db.toggle_favorite("a/one").unwrap()); // toggled off
        db.mark_model_used("a/one").unwrap();
        db.set_reasoning("a/one", Some("high")).unwrap();

        let prefs = db.load_model_prefs().unwrap();
        let p = &prefs[0];
        assert_eq!(p.id, "a/one");
        assert!(!p.favorite);
        assert!(p.last_used.is_some());
        assert_eq!(p.reasoning.as_deref(), Some("high"));
    }

    #[test]
    fn settings_roundtrip() {
        let db = Db::open_in_memory().unwrap();
        db.set_setting("temperature", "0.7").unwrap();
        db.set_setting("temperature", "0.9").unwrap(); // upsert
        let s = db.load_settings().unwrap();
        assert_eq!(s, vec![("temperature".to_string(), "0.9".to_string())]);
    }

    #[test]
    fn set_model_updates_row() {
        let db = Db::open_in_memory().unwrap();
        let space = db.default_space_id().unwrap();
        let s = db.create_session("t", "a/b", &space, "chat").unwrap();
        db.set_session_model(&s.id, "c/d").unwrap();
        assert_eq!(db.list_sessions(&space).unwrap()[0].model, "c/d");
    }

    #[test]
    fn compaction_persists_and_roundtrips() {
        let db = Db::open_in_memory().unwrap();
        let space = db.default_space_id().unwrap();
        let s = db.create_session("t", "a/b", &space, "chat").unwrap();
        assert_eq!(s.compact_summary, None);
        assert_eq!(s.compact_through, 0);

        db.set_compaction(&s.id, "digest of earlier turns", 6)
            .unwrap();
        let reloaded = &db.list_sessions(&space).unwrap()[0];
        assert_eq!(
            reloaded.compact_summary.as_deref(),
            Some("digest of earlier turns")
        );
        assert_eq!(reloaded.compact_through, 6);
    }

    #[test]
    fn spaces_crud_and_session_reassignment_on_delete() {
        let db = Db::open_in_memory().unwrap();
        let spaces = db.list_spaces().unwrap();
        assert_eq!(spaces.len(), 1);
        assert_eq!(spaces[0].name, DEFAULT_SPACE);

        let work = db.create_space("work").unwrap();
        let s = db.create_session("hi", "a/b", &work.id, "chat").unwrap();
        assert_eq!(db.count_sessions(&work.id).unwrap(), 1);

        db.rename_space(&work.id, "work-renamed").unwrap();
        assert!(
            db.list_spaces()
                .unwrap()
                .iter()
                .any(|s| s.name == "work-renamed")
        );

        db.delete_space(&work.id).unwrap();
        assert_eq!(db.list_spaces().unwrap().len(), 1); // work is gone
        let default_id = db.default_space_id().unwrap();
        let moved = db.list_sessions(&default_id).unwrap();
        assert!(moved.iter().any(|c| c.id == s.id)); // session survived, moved to default
    }

    #[test]
    fn chunk_embeddings_store_rank_and_invalidate() {
        let db = Db::open_in_memory().unwrap();
        let space = db.default_space_id().unwrap();
        let id = db.upsert_file(&space, "book.pdf", "h1", 10, "ok").unwrap();
        db.set_file_chunks(
            &id,
            &[
                ("page 1".into(), "cooking with fire".into()),
                ("page 2".into(), "quantum entanglement".into()),
            ],
        )
        .unwrap();

        // Blob codec roundtrip.
        let v = vec![0.25f32, -1.0, 3.5];
        assert_eq!(blob_to_vec(&vec_to_blob(&v)), v);

        // No vectors yet → file needs embedding.
        assert_eq!(
            files_missing_embeddings(&db.conn, &space).unwrap(),
            vec![id.clone()]
        );

        db.set_chunk_embeddings(&id, &[(0, vec![1.0, 0.0]), (1, vec![0.0, 1.0])])
            .unwrap();
        assert!(
            files_missing_embeddings(&db.conn, &space)
                .unwrap()
                .is_empty()
        );

        // Query near the second chunk's vector ranks it first.
        let hits = semantic_chunks(&db.conn, &space, &[0.1, 0.9], 5).unwrap();
        assert_eq!(hits[0].1, "page 2");
        assert!(hits[0].2.contains("quantum"));
        assert!(hits[0].3 > hits[1].3, "scores must be descending");

        // Dimension-mismatched vectors are skipped, not an error.
        let hits = semantic_chunks(&db.conn, &space, &[1.0, 0.0, 0.0], 5).unwrap();
        assert!(hits.is_empty());

        // Rewriting chunks invalidates stale vectors.
        db.set_file_chunks(&id, &[("page 1".into(), "new text".into())])
            .unwrap();
        assert_eq!(
            files_missing_embeddings(&db.conn, &space).unwrap(),
            vec![id.clone()]
        );
    }

    #[test]
    fn files_upsert_list_delete_roundtrip() {
        let db = Db::open_in_memory().unwrap();
        let space = db.default_space_id().unwrap();
        let id = db.upsert_file(&space, "notes.md", "h1", 10, "ok").unwrap();
        db.set_file_chunks(&id, &[("lines 1-40".into(), "hello fts world".into())])
            .unwrap();

        let files = db.list_files(&space).unwrap();
        assert_eq!(files.len(), 1);
        assert_eq!(files[0].name, "notes.md");
        assert_eq!(files[0].hash, "h1");
        assert_eq!(files[0].status, "ok");

        // Re-import with a new hash keeps one row (same id or replaced) and replaces chunks.
        let id2 = db.upsert_file(&space, "notes.md", "h2", 12, "ok").unwrap();
        db.set_file_chunks(&id2, &[("lines 1-40".into(), "goodbye".into())])
            .unwrap();
        let files = db.list_files(&space).unwrap();
        assert_eq!(files.len(), 1);
        assert_eq!(files[0].hash, "h2");

        db.delete_file(&files[0].id).unwrap();
        assert!(db.list_files(&space).unwrap().is_empty());
    }

    #[test]
    fn chunk_search_ranks_and_scopes_by_space() {
        let db = Db::open_in_memory().unwrap();
        let space = db.default_space_id().unwrap();
        let other = db.create_space("other").unwrap();
        let a = db.upsert_file(&space, "a.md", "h", 1, "ok").unwrap();
        let b = db.upsert_file(&other.id, "b.md", "h", 1, "ok").unwrap();
        db.set_file_chunks(&a, &[("lines 1-40".into(), "rust borrow checker".into())])
            .unwrap();
        db.set_file_chunks(&b, &[("lines 1-40".into(), "rust in other space".into())])
            .unwrap();

        let hits = search_chunks(&db.conn, &space, "rust", 8).unwrap();
        assert_eq!(hits.len(), 1); // other space's chunk is excluded
        assert_eq!(hits[0].0, "a.md");
        assert_eq!(hits[0].1, "lines 1-40");
        assert!(hits[0].2.contains("rust"));

        // Special characters must not be an FTS syntax error.
        assert!(search_chunks(&db.conn, &space, "c++ \"quoted\" -dash", 8).is_ok());
    }

    #[test]
    fn file_text_joins_chunks_in_order() {
        let db = Db::open_in_memory().unwrap();
        let space = db.default_space_id().unwrap();
        let id = db.upsert_file(&space, "doc.txt", "h", 1, "ok").unwrap();
        db.set_file_chunks(
            &id,
            &[
                ("lines 1-2".into(), "one\ntwo".into()),
                ("lines 3-4".into(), "three\nfour".into()),
            ],
        )
        .unwrap();
        let text = file_text(&db.conn, &space, "doc.txt").unwrap().unwrap();
        assert_eq!(text, "one\ntwo\nthree\nfour");
        assert!(
            file_text(&db.conn, &space, "missing.txt")
                .unwrap()
                .is_none()
        );
        assert_eq!(count_files(&db.conn, &space).unwrap(), 1);
    }

    #[test]
    fn research_stage_messages_round_trip() {
        let db = Db::open_in_memory().unwrap();
        let space = db.default_space_id().unwrap();
        let s = db.create_session("t", "a/b", &space, "chat").unwrap();
        db.add_research_stage_message(&s.id, "planning…").unwrap();
        let msgs = db.load_messages(&s.id).unwrap();
        assert_eq!(msgs.last().unwrap().role, "research_stage");
        assert_eq!(msgs.last().unwrap().content, "planning…");
    }

    #[test]
    fn survey_messages_round_trip() {
        let db = Db::open_in_memory().unwrap();
        let space = db.default_space_id().unwrap();
        let s = db.create_session("t", "a/b", &space, "chat").unwrap();
        db.add_survey_message(&s.id, "For \"topic\":\n 1. Depth or breadth?")
            .unwrap();
        let msgs = db.load_messages(&s.id).unwrap();
        let last = msgs.last().unwrap();
        assert_eq!(last.role, "survey");
        assert!(last.content.contains("Depth or breadth?"));
    }

    #[test]
    fn gate_reply_round_trip() {
        let db = Db::open_in_memory().unwrap();
        let space = db.default_space_id().unwrap();
        let s = db.create_session("t", "a/b", &space, "chat").unwrap();
        db.add_gate_reply_message(&s.id, "the second option")
            .unwrap();
        let msgs = db.load_messages(&s.id).unwrap();
        let last = msgs.last().unwrap();
        assert_eq!(last.role, "gate_reply");
        assert_eq!(last.content, "the second option");
    }

    #[test]
    fn create_list_touch_delete_watch_roundtrip() {
        let db = Db::open_in_memory().unwrap();
        let id = db
            .create_watch("space-1", "rust async runtimes", 24, "sess-1")
            .unwrap();
        let watches = db.list_watches("space-1").unwrap();
        assert_eq!(watches.len(), 1);
        assert_eq!(watches[0].topic, "rust async runtimes");
        assert_eq!(watches[0].interval_hours, 24);
        assert!(watches[0].last_run_at.is_none());

        db.touch_watch(&id, "2026-07-07T00:00:00+00:00").unwrap();
        let watches = db.list_watches("space-1").unwrap();
        assert_eq!(
            watches[0].last_run_at.as_deref(),
            Some("2026-07-07T00:00:00+00:00")
        );

        db.delete_watch(&id).unwrap();
        assert!(db.list_watches("space-1").unwrap().is_empty());
    }

    #[test]
    fn list_all_watches_returns_watches_from_all_spaces() {
        let db = Db::open_in_memory().unwrap();

        // Create watches in different spaces
        let id1 = db.create_watch("space-a", "topic-1", 24, "sess-1").unwrap();
        let id2 = db.create_watch("space-b", "topic-2", 48, "sess-2").unwrap();
        let id3 = db.create_watch("space-a", "topic-3", 12, "sess-3").unwrap();

        // list_all_watches should return watches from all spaces
        let all_watches = db.list_all_watches().unwrap();
        assert_eq!(all_watches.len(), 3);
        assert!(
            all_watches
                .iter()
                .any(|w| w.id == id1 && w.space_id == "space-a")
        );
        assert!(
            all_watches
                .iter()
                .any(|w| w.id == id2 && w.space_id == "space-b")
        );
        assert!(
            all_watches
                .iter()
                .any(|w| w.id == id3 && w.space_id == "space-a")
        );

        // list_watches for one space should only return that space's watches,
        // confirming list_all_watches is not space-scoped
        let space_a_watches = db.list_watches("space-a").unwrap();
        assert_eq!(space_a_watches.len(), 2);
        assert!(space_a_watches.iter().all(|w| w.space_id == "space-a"));

        let space_b_watches = db.list_watches("space-b").unwrap();
        assert_eq!(space_b_watches.len(), 1);
        assert!(space_b_watches.iter().all(|w| w.space_id == "space-b"));
    }
}