marver 0.0.20

//! SQLite-backed store. Owned by the daemon; the TUI reaches it as a client.
//!
//! Two things this module is deliberately strict about:
//!
//! - **The state machine is enforced here, not by callers.** [`Store::transition`]
//!   rejects illegal moves, so no other component has to remember the diagram.
//! - **Transitions and their events are written together.** A state change that
//!   left no event, or an event describing a change that did not happen, would
//!   make the log untrustworthy.
//!
//! Timestamps are passed in rather than read from the clock, which keeps tests
//! deterministic and makes replaying history possible later.

use std::collections::HashMap;
use std::path::{Path, PathBuf};

use chrono::{DateTime, Utc};
use rusqlite::{Connection, OptionalExtension, Row, params};
use serde_json::Value;

use crate::domain::{
    BlockedKind, Event, Repo, Task, TaskRepo, TaskState, TaskUsage, Todo, TodoScope,
};

#[derive(Debug, thiserror::Error)]
pub enum Error {
    #[error(transparent)]
    Sqlite(#[from] rusqlite::Error),
    #[error(transparent)]
    Json(#[from] serde_json::Error),
    #[error("io error at {path}: {source}")]
    Io {
        path: PathBuf,
        #[source]
        source: std::io::Error,
    },
    #[error("task {0} not found")]
    TaskNotFound(i64),
    #[error("repo {0} not found")]
    RepoNotFound(i64),
    #[error("task {task_id} does not target repo {repo_id}")]
    RepoNotSelected { task_id: i64, repo_id: i64 },
    #[error("illegal transition: {from} -> {to}")]
    IllegalTransition { from: TaskState, to: TaskState },
    #[error("transition to blocked requires a BlockedKind")]
    MissingBlockedKind,
    #[error("transition to failed requires a reason")]
    MissingFailureReason,
    #[error("transition detail does not match a move to {0}")]
    MismatchedDetail(TaskState),
    #[error("a {0} task is still going, so it cannot be archived")]
    NotArchivable(TaskState),
    #[error("unreadable {field} in database: {value:?}")]
    Corrupt { field: &'static str, value: String },
    #[error("migration left {0} row(s) referencing something that is gone; rolled back")]
    MigrationBrokeReferences(i64),
    #[error("todo {0} not found")]
    TodoNotFound(i64),
    #[error("a todo cannot be empty")]
    EmptyTodo,
}

pub type Result<T> = std::result::Result<T, Error>;

/// Why a task is blocked, supplied when transitioning into
/// [`TaskState::Blocked`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BlockedInfo {
    pub kind: BlockedKind,
    pub reason: Option<String>,
}

impl BlockedInfo {
    pub fn new(kind: BlockedKind) -> Self {
        Self { kind, reason: None }
    }

    pub fn with_reason(kind: BlockedKind, reason: impl Into<String>) -> Self {
        Self {
            kind,
            reason: Some(reason.into()),
        }
    }
}

/// Extra information a state change carries.
///
/// Modelled as one value rather than several optional parameters so that
/// "blocked without a kind" and "failed without a reason" are unrepresentable
/// at the call site instead of being caught at runtime.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub enum Transition {
    /// Carries nothing. Correct for every state but `blocked` and `failed`.
    #[default]
    Plain,
    Blocked(BlockedInfo),
    Failed(String),
}

impl Transition {
    /// The state this detail is valid for, if it is only valid for one.
    fn required_state(&self) -> Option<TaskState> {
        match self {
            Self::Plain => None,
            Self::Blocked(_) => Some(TaskState::Blocked),
            Self::Failed(_) => Some(TaskState::Failed),
        }
    }

    fn blocked(&self) -> Option<&BlockedInfo> {
        match self {
            Self::Blocked(info) => Some(info),
            _ => None,
        }
    }

    fn failure(&self) -> Option<&str> {
        match self {
            Self::Failed(reason) => Some(reason),
            _ => None,
        }
    }
}

/// Applied in order; index + 1 becomes `PRAGMA user_version`.
const MIGRATIONS: &[&str] = &[
    include_str!("store/0001_initial.sql"),
    include_str!("store/0002_paused_and_todos.sql"),
    include_str!("store/0003_usage.sql"),
    include_str!("store/0004_archive.sql"),
];

pub struct Store {
    conn: Connection,
}

impl Store {
    /// Open (creating if absent) the database at `path`, applying migrations.
    pub fn open(path: impl AsRef<Path>) -> Result<Self> {
        let path = path.as_ref();
        if let Some(parent) = path.parent()
            && !parent.as_os_str().is_empty()
        {
            std::fs::create_dir_all(parent).map_err(|source| Error::Io {
                path: parent.to_path_buf(),
                source,
            })?;
        }
        Self::from_connection(Connection::open(path)?)
    }

    /// An ephemeral database. Tests only.
    pub fn open_in_memory() -> Result<Self> {
        Self::from_connection(Connection::open_in_memory()?)
    }

    fn from_connection(mut conn: Connection) -> Result<Self> {
        // WAL lets readers proceed during writes; it is a no-op in memory.
        conn.pragma_update(None, "journal_mode", "WAL")?;
        conn.pragma_update(None, "foreign_keys", "ON")?;
        conn.pragma_update(None, "busy_timeout", 5_000)?;
        migrate(&mut conn)?;
        Ok(Self { conn })
    }

    /// Current schema version. Equal to `MIGRATIONS.len()` after a successful open.
    pub fn schema_version(&self) -> Result<i64> {
        Ok(self
            .conn
            .query_row("PRAGMA user_version", [], |row| row.get(0))?)
    }

    // ---- repos ---------------------------------------------------------

    /// Record a repo found by a scan. Idempotent on `path`: an existing row has
    /// its `last_seen_at` and `name` refreshed and keeps its `ignored` flag.
    pub fn upsert_repo(&self, path: &Path, name: &str, now: DateTime<Utc>) -> Result<Repo> {
        let path_str = path_to_string(path);
        self.conn.execute(
            "INSERT INTO repos (path, name, ignored, discovered_at, last_seen_at)
             VALUES (?1, ?2, 0, ?3, ?3)
             ON CONFLICT(path) DO UPDATE SET name = ?2, last_seen_at = ?3",
            params![path_str, name, now.timestamp()],
        )?;
        self.repo_by_path(path)
    }

    pub fn repo_by_path(&self, path: &Path) -> Result<Repo> {
        self.conn
            .query_row(
                "SELECT id, path, name, ignored, discovered_at, last_seen_at
                 FROM repos WHERE path = ?1",
                params![path_to_string(path)],
                row_to_repo,
            )
            .optional()?
            .ok_or_else(|| Error::Corrupt {
                field: "repos.path",
                value: path_to_string(path),
            })
    }

    pub fn get_repo(&self, id: i64) -> Result<Repo> {
        self.conn
            .query_row(
                "SELECT id, path, name, ignored, discovered_at, last_seen_at
                 FROM repos WHERE id = ?1",
                params![id],
                row_to_repo,
            )
            .optional()?
            .ok_or(Error::RepoNotFound(id))
    }

    pub fn list_repos(&self, include_ignored: bool) -> Result<Vec<Repo>> {
        let mut stmt = self.conn.prepare(
            "SELECT id, path, name, ignored, discovered_at, last_seen_at
             FROM repos
             WHERE ?1 OR ignored = 0
             ORDER BY name, path",
        )?;
        let rows = stmt.query_map(params![include_ignored], row_to_repo)?;
        rows.collect::<rusqlite::Result<Vec<_>>>()
            .map_err(Into::into)
    }

    /// Repos not touched by a scan at or after `cutoff`.
    ///
    /// After a scan refreshes everything it found, whatever is left here has
    /// moved or been deleted. Reported rather than removed: a vanished repo may
    /// still be referenced by an existing task's worktree.
    pub fn list_repos_last_seen_before(&self, cutoff: DateTime<Utc>) -> Result<Vec<Repo>> {
        let mut stmt = self.conn.prepare(
            "SELECT id, path, name, ignored, discovered_at, last_seen_at
             FROM repos WHERE last_seen_at < ?1 ORDER BY path",
        )?;
        let rows = stmt.query_map(params![cutoff.timestamp()], row_to_repo)?;
        rows.collect::<rusqlite::Result<Vec<_>>>()
            .map_err(Into::into)
    }

    pub fn set_repo_ignored(&self, id: i64, ignored: bool) -> Result<()> {
        let changed = self.conn.execute(
            "UPDATE repos SET ignored = ?2 WHERE id = ?1",
            params![id, ignored],
        )?;
        if changed == 0 {
            return Err(Error::RepoNotFound(id));
        }
        Ok(())
    }

    // ---- tasks ---------------------------------------------------------

    /// Create a task in [`TaskState::Queued`] and log its creation.
    ///
    /// The workspace directory is `<workspace_root>/<id>`, derived here rather
    /// than supplied because the id only exists after the insert. Deriving it
    /// inside the transaction means a task is never briefly persisted with a
    /// path that does not match its id.
    /// `repo_ids` are the repos this task targets, recorded in the same
    /// transaction so a task can never exist without knowing what it works on.
    /// Their worktrees are created later, at launch.
    pub fn create_task(
        &mut self,
        title: &str,
        prompt: &str,
        workspace_root: &Path,
        repo_ids: &[i64],
        now: DateTime<Utc>,
    ) -> Result<Task> {
        let tx = self.conn.transaction()?;
        tx.execute(
            "INSERT INTO tasks (title, prompt, state, workspace_dir, created_at, updated_at)
             VALUES (?1, ?2, ?3, '', ?4, ?4)",
            params![title, prompt, TaskState::Queued.as_str(), now.timestamp(),],
        )?;
        let id = tx.last_insert_rowid();
        tx.execute(
            "UPDATE tasks SET workspace_dir = ?2 WHERE id = ?1",
            params![id, path_to_string(&workspace_root.join(id.to_string()))],
        )?;
        for repo_id in repo_ids {
            tx.execute(
                "INSERT INTO task_repos (task_id, repo_id) VALUES (?1, ?2)",
                params![id, repo_id],
            )?;
        }
        insert_event(
            &tx,
            Some(id),
            "task.created",
            &serde_json::json!({ "title": title }),
            now,
        )?;
        tx.commit()?;
        self.get_task(id)
    }

    pub fn get_task(&self, id: i64) -> Result<Task> {
        self.conn
            .query_row(TASK_SELECT, params![id], row_to_task)
            .optional()?
            .transpose()?
            .ok_or(Error::TaskNotFound(id))
    }

    pub fn list_tasks(&self) -> Result<Vec<Task>> {
        let mut stmt = self.conn.prepare(concat!(
            "SELECT ",
            task_columns!(),
            " FROM tasks ORDER BY id"
        ))?;
        let rows = stmt.query_map([], row_to_task)?;
        collect_tasks(rows)
    }

    pub fn list_tasks_in_state(&self, state: TaskState) -> Result<Vec<Task>> {
        let mut stmt = self.conn.prepare(concat!(
            "SELECT ",
            task_columns!(),
            " FROM tasks WHERE state = ?1 ORDER BY id"
        ))?;
        let rows = stmt.query_map(params![state.as_str()], row_to_task)?;
        collect_tasks(rows)
    }

    pub fn set_session_name(&self, id: i64, session_name: &str, now: DateTime<Utc>) -> Result<()> {
        let changed = self.conn.execute(
            "UPDATE tasks SET session_name = ?2, updated_at = ?3 WHERE id = ?1",
            params![id, session_name, now.timestamp()],
        )?;
        if changed == 0 {
            return Err(Error::TaskNotFound(id));
        }
        Ok(())
    }

    /// Forget a task's tmux session, once there is no longer one to point at.
    ///
    /// Recording that the session is gone is what stops the daemon trying to
    /// reap it on every pass.
    pub fn clear_session_name(&self, id: i64, now: DateTime<Utc>) -> Result<()> {
        let changed = self.conn.execute(
            "UPDATE tasks SET session_name = NULL, updated_at = ?2 WHERE id = ?1",
            params![id, now.timestamp()],
        )?;
        if changed == 0 {
            return Err(Error::TaskNotFound(id));
        }
        Ok(())
    }

    /// Put a finished task out of the way, or bring it back.
    ///
    /// Only a task that has stopped can be archived: an archived row is one the
    /// list no longer shows, and hiding work that is still happening would be a
    /// way to lose an agent. Restoring is unconditional, since a task that is in
    /// the archive is by definition one that was allowed in.
    ///
    /// Idempotent in both directions. Archiving twice keeps the first date,
    /// because when it was put away is a fact about the first time.
    pub fn set_task_archived(&self, id: i64, archived: bool, now: DateTime<Utc>) -> Result<Task> {
        let task = self.get_task(id)?;
        if archived && !task.state.is_terminal() {
            return Err(Error::NotArchivable(task.state));
        }
        if archived == task.is_archived() {
            return Ok(task);
        }
        let at = archived.then(|| now.timestamp());
        self.conn.execute(
            "UPDATE tasks SET archived_at = ?2, updated_at = ?3 WHERE id = ?1",
            params![id, at, now.timestamp()],
        )?;
        self.get_task(id)
    }

    /// Move a task to `next`, rejecting anything the lifecycle disallows.
    ///
    /// `detail` must match `next`: [`Transition::Blocked`] only when entering
    /// [`TaskState::Blocked`], [`Transition::Failed`] only when entering
    /// [`TaskState::Failed`], [`Transition::Plain`] otherwise. Leaving either
    /// state clears the stored reason, so a resumed task carries no stale
    /// explanation. The row update and the `task.transition` event are written
    /// in one transaction.
    pub fn transition(
        &mut self,
        id: i64,
        next: TaskState,
        detail: Transition,
        now: DateTime<Utc>,
    ) -> Result<Task> {
        self.move_task(id, None, next, detail, now)
    }

    /// Move a task to `next`, but only if it is still in `expected`.
    ///
    /// [`Store::transition`] asks "is this move legal from wherever the task is
    /// now", which is right for a caller reacting to the task's current state
    /// and wrong for one acting on a decision it made earlier. The scheduler is
    /// the second kind: it chooses a queued task, spends a second or two making
    /// worktrees and starting tmux, and only then records `running`. If the user
    /// paused that task in the meantime, `paused → running` is a perfectly legal
    /// move — so the plain transition happily undid the pause and started the
    /// agent anyway.
    ///
    /// Naming the state the decision was made in turns that into a refusal.
    pub fn transition_from(
        &mut self,
        id: i64,
        expected: TaskState,
        next: TaskState,
        detail: Transition,
        now: DateTime<Utc>,
    ) -> Result<Task> {
        self.move_task(id, Some(expected), next, detail, now)
    }

    fn move_task(
        &mut self,
        id: i64,
        expected: Option<TaskState>,
        next: TaskState,
        detail: Transition,
        now: DateTime<Utc>,
    ) -> Result<Task> {
        let current = self.get_task(id)?;
        if let Some(expected) = expected
            && current.state != expected
        {
            return Err(Error::IllegalTransition {
                from: current.state,
                to: next,
            });
        }
        if !current.state.can_transition_to(next) {
            return Err(Error::IllegalTransition {
                from: current.state,
                to: next,
            });
        }
        match detail.required_state() {
            Some(required) if required != next => return Err(Error::MismatchedDetail(next)),
            None if next == TaskState::Blocked => return Err(Error::MissingBlockedKind),
            None if next == TaskState::Failed => return Err(Error::MissingFailureReason),
            _ => {}
        }

        let tx = self.conn.transaction()?;
        // Compare-and-swap on the state we checked. The read above is its own
        // transaction, so between it and this write another process — the TUI
        // cancelling while the daemon commits — can have moved the task
        // already. Without the guard both writes land, the task leaves one
        // state twice, and the event log contradicts itself.
        let changed = tx.execute(
            "UPDATE tasks
             SET state = ?2, blocked_kind = ?3, blocked_reason = ?4,
                 failure_reason = ?5, updated_at = ?6
             WHERE id = ?1 AND state = ?7",
            params![
                id,
                next.as_str(),
                detail.blocked().map(|b| b.kind.as_str()),
                detail.blocked().and_then(|b| b.reason.as_deref()),
                detail.failure(),
                now.timestamp(),
                current.state.as_str(),
            ],
        )?;
        if changed == 0 {
            // Lost the race. Report the state that actually won, not the stale
            // one, so the caller's error names something real.
            drop(tx);
            let actual = self.get_task(id)?.state;
            return Err(Error::IllegalTransition {
                from: actual,
                to: next,
            });
        }
        insert_event(
            &tx,
            Some(id),
            "task.transition",
            &serde_json::json!({
                "from": current.state,
                "to": next,
                "blocked_kind": detail.blocked().map(|b| b.kind),
                "blocked_reason": detail.blocked().and_then(|b| b.reason.clone()),
                "failure_reason": detail.failure(),
            }),
            now,
        )?;
        tx.commit()?;
        self.get_task(id)
    }

    // ---- worktrees -----------------------------------------------------

    /// Add a repo to a task's selection after creation.
    pub fn select_repo(&self, task_id: i64, repo_id: i64) -> Result<()> {
        self.conn.execute(
            "INSERT INTO task_repos (task_id, repo_id) VALUES (?1, ?2)",
            params![task_id, repo_id],
        )?;
        Ok(())
    }

    /// Record the worktree provisioned for a (task, repo) pairing.
    ///
    /// Fails if the pairing was never selected: a worktree for a repo the task
    /// does not target would be orphaned the moment the task is torn down.
    pub fn record_worktree(
        &self,
        task_id: i64,
        repo_id: i64,
        worktree_path: &Path,
        branch: &str,
        base_ref: &str,
    ) -> Result<TaskRepo> {
        let changed = self.conn.execute(
            "UPDATE task_repos
             SET worktree_path = ?3, branch = ?4, base_ref = ?5
             WHERE task_id = ?1 AND repo_id = ?2",
            params![
                task_id,
                repo_id,
                path_to_string(worktree_path),
                branch,
                base_ref
            ],
        )?;
        if changed == 0 {
            return Err(Error::RepoNotSelected { task_id, repo_id });
        }
        Ok(TaskRepo {
            task_id,
            repo_id,
            worktree_path: Some(worktree_path.to_path_buf()),
            branch: Some(branch.to_string()),
            base_ref: Some(base_ref.to_string()),
        })
    }

    /// Clear the worktree details for a task, leaving the selection intact.
    ///
    /// Used after teardown so a task's repos are still known once its worktrees
    /// are gone.
    pub fn clear_worktrees(&self, task_id: i64) -> Result<()> {
        self.conn.execute(
            "UPDATE task_repos
             SET worktree_path = NULL, branch = NULL, base_ref = NULL
             WHERE task_id = ?1",
            params![task_id],
        )?;
        Ok(())
    }

    /// Repo names for every task, keyed by task id.
    ///
    /// One join rather than a query per row. The task list re-reads itself on
    /// every tick, so asking per task would turn one screen into a query for
    /// each task it shows, several times a second, for names that change only
    /// when a task is created.
    ///
    /// Names are ordered as the repos were selected, so a multi-repo task reads
    /// the same way twice running.
    pub fn repo_names_by_task(&self) -> Result<HashMap<i64, Vec<String>>> {
        let mut statement = self.conn.prepare(
            "SELECT task_repos.task_id, repos.name
             FROM task_repos
             JOIN repos ON repos.id = task_repos.repo_id
             ORDER BY task_repos.task_id, task_repos.repo_id",
        )?;
        let rows = statement.query_map([], |row| {
            Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?))
        })?;

        let mut names: HashMap<i64, Vec<String>> = HashMap::new();
        for row in rows {
            let (task_id, name) = row?;
            names.entry(task_id).or_default().push(name);
        }
        Ok(names)
    }

    pub fn list_task_repos(&self, task_id: i64) -> Result<Vec<TaskRepo>> {
        let mut stmt = self.conn.prepare(
            "SELECT task_id, repo_id, worktree_path, branch, base_ref
             FROM task_repos WHERE task_id = ?1 ORDER BY repo_id",
        )?;
        let rows = stmt.query_map(params![task_id], |row| {
            Ok(TaskRepo {
                task_id: row.get(0)?,
                repo_id: row.get(1)?,
                worktree_path: row.get::<_, Option<String>>(2)?.map(PathBuf::from),
                branch: row.get(3)?,
                base_ref: row.get(4)?,
            })
        })?;
        rows.collect::<rusqlite::Result<Vec<_>>>()
            .map_err(Into::into)
    }

    // ---- usage ---------------------------------------------------------

    /// Record what a task's agent has spent.
    ///
    /// Context and model are *replaced* — they describe the latest turn — while
    /// output tokens are *added*, since each read only sees what is new. That
    /// asymmetry is the whole reason this is one call rather than a setter per
    /// column: doing it the other way round reads as a plausible context figure
    /// several times the model's window.
    ///
    /// A read that saw no assistant turn still advances the offset, so the bytes
    /// it walked over are not walked over again, but leaves the model and context
    /// alone: nothing new to say is not the same as nothing to say.
    pub fn record_usage(&self, id: i64, usage: &crate::usage::Usage) -> Result<()> {
        let changed = self.conn.execute(
            "UPDATE tasks
             SET model = coalesce(?2, model),
                 context_tokens = coalesce(?3, context_tokens),
                 output_tokens = coalesce(output_tokens, 0) + ?4,
                 transcript_offset = ?5
             WHERE id = ?1",
            params![
                id,
                usage.model.as_deref(),
                usage.context_tokens.map(|n| n as i64),
                usage.output_tokens as i64,
                usage.offset as i64,
            ],
        )?;
        if changed == 0 {
            return Err(Error::TaskNotFound(id));
        }
        Ok(())
    }

    /// Remember where a task's transcript is, so its usage can be refreshed
    /// without waiting for another hook.
    pub fn set_transcript_path(&self, id: i64, path: &Path) -> Result<()> {
        self.conn.execute(
            "UPDATE tasks SET transcript_path = ?2 WHERE id = ?1",
            params![id, path_to_string(path)],
        )?;
        Ok(())
    }

    // ---- todos ---------------------------------------------------------

    /// Add a todo to `scope`.
    ///
    /// Empty text is refused here rather than at the screen: a todo you cannot
    /// read is not a todo, and the database says the same thing in its CHECK.
    pub fn add_todo(&self, scope: TodoScope, text: &str, now: DateTime<Utc>) -> Result<Todo> {
        let text = text.trim();
        if text.is_empty() {
            return Err(Error::EmptyTodo);
        }
        self.conn.execute(
            "INSERT INTO todos (task_id, text, done, created_at) VALUES (?1, ?2, 0, ?3)",
            params![scope.task_id(), text, now.timestamp()],
        )?;
        self.get_todo(self.conn.last_insert_rowid())
    }

    pub fn get_todo(&self, id: i64) -> Result<Todo> {
        self.conn
            .query_row(
                "SELECT id, task_id, text, done, created_at FROM todos WHERE id = ?1",
                params![id],
                row_to_todo,
            )
            .optional()?
            .ok_or(Error::TodoNotFound(id))
    }

    /// Every todo in `scope`, oldest first.
    ///
    /// Done ones are included. A list that hid them would make ticking one look
    /// like deleting it, and the two keys mean different things.
    pub fn list_todos(&self, scope: TodoScope) -> Result<Vec<Todo>> {
        let mut stmt = self.conn.prepare(
            "SELECT id, task_id, text, done, created_at
             FROM todos WHERE task_id IS ?1 ORDER BY id",
        )?;
        let rows = stmt.query_map(params![scope.task_id()], row_to_todo)?;
        rows.collect::<rusqlite::Result<Vec<_>>>()
            .map_err(Into::into)
    }

    /// How many todos are outstanding in `scope`. For badges and counts.
    pub fn open_todo_count(&self, scope: TodoScope) -> Result<usize> {
        let count: i64 = self.conn.query_row(
            "SELECT count(*) FROM todos WHERE task_id IS ?1 AND done = 0",
            params![scope.task_id()],
            |row| row.get(0),
        )?;
        Ok(count as usize)
    }

    /// Outstanding todo counts for every task that has one, by task id.
    ///
    /// One query rather than one per row: the task list draws this on every
    /// frame, and a query per task would make the badge cost grow with the
    /// board.
    pub fn open_todo_counts(&self) -> Result<HashMap<i64, usize>> {
        let mut stmt = self.conn.prepare(
            "SELECT task_id, count(*) FROM todos
             WHERE task_id IS NOT NULL AND done = 0 GROUP BY task_id",
        )?;
        let rows = stmt.query_map([], |row| {
            Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)? as usize))
        })?;
        rows.collect::<rusqlite::Result<HashMap<_, _>>>()
            .map_err(Into::into)
    }

    pub fn set_todo_done(&self, id: i64, done: bool) -> Result<Todo> {
        let changed = self.conn.execute(
            "UPDATE todos SET done = ?2 WHERE id = ?1",
            params![id, done as i64],
        )?;
        if changed == 0 {
            return Err(Error::TodoNotFound(id));
        }
        self.get_todo(id)
    }

    pub fn delete_todo(&self, id: i64) -> Result<()> {
        let changed = self
            .conn
            .execute("DELETE FROM todos WHERE id = ?1", params![id])?;
        if changed == 0 {
            return Err(Error::TodoNotFound(id));
        }
        Ok(())
    }

    // ---- events --------------------------------------------------------

    pub fn append_event(
        &self,
        task_id: Option<i64>,
        kind: &str,
        payload: &Value,
        now: DateTime<Utc>,
    ) -> Result<i64> {
        insert_event(&self.conn, task_id, kind, payload, now)
    }

    pub fn list_events(&self, task_id: i64) -> Result<Vec<Event>> {
        let mut stmt = self.conn.prepare(
            "SELECT id, task_id, kind, payload, created_at
             FROM events WHERE task_id = ?1 ORDER BY id",
        )?;
        let rows = stmt.query_map(params![task_id], row_to_event)?;
        let mut out = Vec::new();
        for row in rows {
            let (id, task_id, kind, payload, created_at) = row?;
            out.push(Event {
                id,
                task_id,
                kind,
                payload: serde_json::from_str(&payload)?,
                created_at: timestamp(created_at)?,
            });
        }
        Ok(out)
    }
}

/// The column list every task query selects, in the order [`row_to_task`] reads.
/// A macro rather than a const so callers can `concat!` it into a literal.
macro_rules! task_columns {
    () => {
        "id, title, prompt, state, blocked_kind, blocked_reason, failure_reason,
         workspace_dir, session_name, created_at, updated_at,
         model, context_tokens, output_tokens, transcript_offset, transcript_path,
         archived_at"
    };
}
use task_columns;

const TASK_SELECT: &str = concat!("SELECT ", task_columns!(), " FROM tasks WHERE id = ?1");

fn migrate(conn: &mut Connection) -> Result<()> {
    let version: i64 = conn.query_row("PRAGMA user_version", [], |row| row.get(0))?;
    if version as usize >= MIGRATIONS.len() {
        return Ok(());
    }
    // Foreign keys off for the duration, and only for it.
    //
    // SQLite cannot alter a CHECK constraint, so a migration that changes one
    // has to rebuild the table: copy, drop the original, rename. With
    // enforcement on, that DROP is not a schema operation but a delete — it
    // fires `ON DELETE CASCADE` and takes every task's repos, events, and todos
    // with it, silently, leaving a valid-looking database with nothing in it.
    //
    // The pragma is a no-op inside a transaction, so it cannot live in the
    // migration file where the rebuild is written. It has to be here, outside.
    conn.pragma_update(None, "foreign_keys", "OFF")?;
    let outcome = apply(conn, version as usize);
    // Restored whatever happened, including on the error path: the connection
    // outlives this function and the rest of the process expects enforcement.
    conn.pragma_update(None, "foreign_keys", "ON")?;
    outcome
}

/// Run the outstanding migrations in one transaction.
fn apply(conn: &mut Connection, from: usize) -> Result<()> {
    let tx = conn.transaction()?;
    for (index, migration) in MIGRATIONS.iter().enumerate().skip(from) {
        tx.execute_batch(migration)?;
        // user_version takes no bind parameters.
        tx.execute_batch(&format!("PRAGMA user_version = {}", index + 1))?;
    }
    // Enforcement was off while a table was rebuilt, so nothing was watching the
    // references. Checked before the commit, so a migration that orphans a row
    // rolls back rather than leaving the damage behind for later.
    let violations: i64 =
        tx.query_row("SELECT count(*) FROM pragma_foreign_key_check", [], |row| {
            row.get(0)
        })?;
    if violations > 0 {
        return Err(Error::MigrationBrokeReferences(violations));
    }
    tx.commit()?;
    Ok(())
}

fn insert_event(
    conn: &Connection,
    task_id: Option<i64>,
    kind: &str,
    payload: &Value,
    now: DateTime<Utc>,
) -> Result<i64> {
    conn.execute(
        "INSERT INTO events (task_id, kind, payload, created_at) VALUES (?1, ?2, ?3, ?4)",
        params![
            task_id,
            kind,
            serde_json::to_string(payload)?,
            now.timestamp()
        ],
    )?;
    Ok(conn.last_insert_rowid())
}

fn path_to_string(path: &Path) -> String {
    path.to_string_lossy().into_owned()
}

fn timestamp(secs: i64) -> Result<DateTime<Utc>> {
    DateTime::from_timestamp(secs, 0).ok_or(Error::Corrupt {
        field: "timestamp",
        value: secs.to_string(),
    })
}

fn row_to_repo(row: &Row<'_>) -> rusqlite::Result<Repo> {
    Ok(Repo {
        id: row.get(0)?,
        path: PathBuf::from(row.get::<_, String>(1)?),
        name: row.get(2)?,
        ignored: row.get(3)?,
        // Timestamps written by this module are always in range.
        discovered_at: DateTime::from_timestamp(row.get(4)?, 0).unwrap_or_default(),
        last_seen_at: DateTime::from_timestamp(row.get(5)?, 0).unwrap_or_default(),
    })
}

fn row_to_todo(row: &Row<'_>) -> rusqlite::Result<Todo> {
    Ok(Todo {
        id: row.get(0)?,
        task_id: row.get(1)?,
        text: row.get(2)?,
        done: row.get(3)?,
        created_at: DateTime::from_timestamp(row.get(4)?, 0).unwrap_or_default(),
    })
}

/// Yields a nested `Result` so a malformed enum surfaces as [`Error::Corrupt`]
/// rather than a sqlite error.
#[allow(clippy::type_complexity)]
fn row_to_task(row: &Row<'_>) -> rusqlite::Result<Result<Task>> {
    let state_raw: String = row.get(3)?;
    let blocked_raw: Option<String> = row.get(4)?;
    let created: i64 = row.get(9)?;
    let updated: i64 = row.get(10)?;

    let Some(state) = TaskState::parse(&state_raw) else {
        return Ok(Err(Error::Corrupt {
            field: "tasks.state",
            value: state_raw,
        }));
    };
    let blocked_kind = match blocked_raw {
        None => None,
        Some(raw) => match BlockedKind::parse(&raw) {
            Some(kind) => Some(kind),
            None => {
                return Ok(Err(Error::Corrupt {
                    field: "tasks.blocked_kind",
                    value: raw,
                }));
            }
        },
    };

    Ok(Ok(Task {
        id: row.get(0)?,
        title: row.get(1)?,
        prompt: row.get(2)?,
        state,
        blocked_kind,
        blocked_reason: row.get(5)?,
        failure_reason: row.get(6)?,
        workspace_dir: PathBuf::from(row.get::<_, String>(7)?),
        session_name: row.get(8)?,
        created_at: match timestamp(created) {
            Ok(ts) => ts,
            Err(err) => return Ok(Err(err)),
        },
        updated_at: match timestamp(updated) {
            Ok(ts) => ts,
            Err(err) => return Ok(Err(err)),
        },
        usage: TaskUsage {
            model: row.get(11)?,
            // Read as i64 and widened: SQLite has no unsigned type, and a
            // negative here would mean a corrupt row rather than a real count.
            context_tokens: row.get::<_, Option<i64>>(12)?.map(|n| n.max(0) as u64),
            output_tokens: row.get::<_, Option<i64>>(13)?.map(|n| n.max(0) as u64),
            transcript_offset: row.get::<_, i64>(14)?.max(0) as u64,
            transcript_path: row.get::<_, Option<String>>(15)?.map(PathBuf::from),
        },
        archived_at: match row.get::<_, Option<i64>>(16)? {
            None => None,
            Some(secs) => match timestamp(secs) {
                Ok(ts) => Some(ts),
                Err(err) => return Ok(Err(err)),
            },
        },
    }))
}

#[allow(clippy::type_complexity)]
fn row_to_event(row: &Row<'_>) -> rusqlite::Result<(i64, Option<i64>, String, String, i64)> {
    Ok((
        row.get(0)?,
        row.get(1)?,
        row.get(2)?,
        row.get(3)?,
        row.get(4)?,
    ))
}

fn collect_tasks<I>(rows: I) -> Result<Vec<Task>>
where
    I: Iterator<Item = rusqlite::Result<Result<Task>>>,
{
    let mut out = Vec::new();
    for row in rows {
        out.push(row??);
    }
    Ok(out)
}

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

    fn at(secs: i64) -> DateTime<Utc> {
        DateTime::from_timestamp(secs, 0).expect("valid timestamp")
    }

    fn store() -> Store {
        Store::open_in_memory().expect("in-memory store")
    }

    fn task(store: &mut Store) -> Task {
        store
            .create_task(
                "fix auth",
                "fix the auth flow",
                Path::new("/tmp/tasks"),
                &[],
                at(0),
            )
            .expect("create task")
    }

    #[test]
    fn migrations_apply_and_are_idempotent() {
        let store = store();
        assert_eq!(store.schema_version().unwrap(), MIGRATIONS.len() as i64);
        // Re-running against the same connection must not fail or double-apply.
        let mut conn = store.conn;
        migrate(&mut conn).expect("second migrate");
        let version: i64 = conn
            .query_row("PRAGMA user_version", [], |r| r.get(0))
            .unwrap();
        assert_eq!(version, MIGRATIONS.len() as i64);
    }

    #[test]
    fn workspace_dir_is_derived_from_the_id() {
        let mut store = store();
        let first = task(&mut store);
        let second = task(&mut store);
        assert_eq!(
            first.workspace_dir,
            Path::new("/tmp/tasks").join(first.id.to_string()),
            "the path must match the id it was assigned"
        );
        assert_ne!(first.workspace_dir, second.workspace_dir);
    }

    #[test]
    fn new_task_starts_queued_and_logs_creation() {
        let mut store = store();
        let task = task(&mut store);
        assert_eq!(task.state, TaskState::Queued);
        assert_eq!(task.blocked_kind, None);
        assert_eq!(task.session_name, None);

        let events = store.list_events(task.id).unwrap();
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].kind, "task.created");
    }

    #[test]
    fn happy_path_walks_to_committed() {
        let mut store = store();
        let task = task(&mut store);

        let task = store
            .transition(task.id, TaskState::Running, Transition::Plain, at(1))
            .unwrap();
        assert_eq!(task.state, TaskState::Running);

        let task = store
            .transition(task.id, TaskState::AwaitingReview, Transition::Plain, at(2))
            .unwrap();
        assert_eq!(task.state, TaskState::AwaitingReview);

        let task = store
            .transition(task.id, TaskState::Committed, Transition::Plain, at(3))
            .unwrap();
        assert_eq!(task.state, TaskState::Committed);
        assert_eq!(task.updated_at, at(3));
    }

    #[test]
    fn illegal_transitions_are_rejected() {
        let mut store = store();
        let task = task(&mut store);
        let err = store
            .transition(task.id, TaskState::Committed, Transition::Plain, at(1))
            .unwrap_err();
        assert!(matches!(
            err,
            Error::IllegalTransition {
                from: TaskState::Queued,
                to: TaskState::Committed
            }
        ));
        // The rejected move left nothing behind.
        assert_eq!(store.get_task(task.id).unwrap().state, TaskState::Queued);
        assert_eq!(store.list_events(task.id).unwrap().len(), 1);
    }

    #[test]
    fn blocking_requires_and_clears_its_reason() {
        let mut store = store();
        let task = task(&mut store);
        store
            .transition(task.id, TaskState::Running, Transition::Plain, at(1))
            .unwrap();

        assert!(matches!(
            store
                .transition(task.id, TaskState::Blocked, Transition::Plain, at(2))
                .unwrap_err(),
            Error::MissingBlockedKind
        ));

        let blocked = store
            .transition(
                task.id,
                TaskState::Blocked,
                Transition::Blocked(BlockedInfo::with_reason(
                    BlockedKind::PermissionPrompt,
                    "edit src/main.rs",
                )),
                at(3),
            )
            .unwrap();
        assert_eq!(blocked.blocked_kind, Some(BlockedKind::PermissionPrompt));
        assert_eq!(blocked.blocked_reason.as_deref(), Some("edit src/main.rs"));

        let resumed = store
            .transition(task.id, TaskState::Running, Transition::Plain, at(4))
            .unwrap();
        assert_eq!(resumed.state, TaskState::Running);
        assert_eq!(resumed.blocked_kind, None, "reason must be cleared");
        assert_eq!(resumed.blocked_reason, None);
    }

    #[test]
    fn blocked_details_rejected_for_other_states() {
        let mut store = store();
        let task = task(&mut store);
        let err = store
            .transition(
                task.id,
                TaskState::Running,
                Transition::Blocked(BlockedInfo::new(BlockedKind::Question)),
                at(1),
            )
            .unwrap_err();
        assert!(matches!(err, Error::MismatchedDetail(TaskState::Running)));
    }

    #[test]
    fn failing_requires_and_records_a_reason() {
        let mut store = store();
        let task = task(&mut store);
        store
            .transition(task.id, TaskState::Running, Transition::Plain, at(1))
            .unwrap();

        assert!(matches!(
            store
                .transition(task.id, TaskState::Failed, Transition::Plain, at(2))
                .unwrap_err(),
            Error::MissingFailureReason
        ));

        let failed = store
            .transition(
                task.id,
                TaskState::Failed,
                Transition::Failed("tmux session died".into()),
                at(3),
            )
            .unwrap();
        assert_eq!(failed.state, TaskState::Failed);
        assert_eq!(failed.failure_reason.as_deref(), Some("tmux session died"));
    }

    #[test]
    fn a_failed_task_cannot_be_resumed() {
        let mut store = store();
        let task = task(&mut store);
        store
            .transition(
                task.id,
                TaskState::Failed,
                Transition::Failed("worktree setup failed".into()),
                at(1),
            )
            .unwrap();
        assert!(matches!(
            store
                .transition(task.id, TaskState::Running, Transition::Plain, at(2))
                .unwrap_err(),
            Error::IllegalTransition {
                from: TaskState::Failed,
                to: TaskState::Running
            }
        ));
    }

    #[test]
    fn blocking_details_are_cleared_by_failing() {
        let mut store = store();
        let task = task(&mut store);
        store
            .transition(task.id, TaskState::Running, Transition::Plain, at(1))
            .unwrap();
        store
            .transition(
                task.id,
                TaskState::Blocked,
                Transition::Blocked(BlockedInfo::new(BlockedKind::Question)),
                at(2),
            )
            .unwrap();
        let failed = store
            .transition(
                task.id,
                TaskState::Failed,
                Transition::Failed("agent exited".into()),
                at(3),
            )
            .unwrap();
        assert_eq!(
            failed.blocked_kind, None,
            "stale blocking detail left behind"
        );
        assert_eq!(failed.blocked_reason, None);
        assert_eq!(failed.failure_reason.as_deref(), Some("agent exited"));
    }

    #[test]
    fn cancelling_works_from_every_unfinished_state() {
        for state in [
            TaskState::Queued,
            TaskState::Running,
            TaskState::Blocked,
            TaskState::AwaitingReview,
        ] {
            let mut store = store();
            let task = task(&mut store);

            // Walk to the state under test.
            match state {
                TaskState::Queued => {}
                TaskState::Running => {
                    store
                        .transition(task.id, TaskState::Running, Transition::Plain, at(1))
                        .unwrap();
                }
                TaskState::Blocked => {
                    store
                        .transition(task.id, TaskState::Running, Transition::Plain, at(1))
                        .unwrap();
                    store
                        .transition(
                            task.id,
                            TaskState::Blocked,
                            Transition::Blocked(BlockedInfo::new(BlockedKind::Silence)),
                            at(2),
                        )
                        .unwrap();
                }
                TaskState::AwaitingReview => {
                    store
                        .transition(task.id, TaskState::Running, Transition::Plain, at(1))
                        .unwrap();
                    store
                        .transition(task.id, TaskState::AwaitingReview, Transition::Plain, at(2))
                        .unwrap();
                }
                other => unreachable!("{other} is not under test"),
            }

            let cancelled = store
                .transition(task.id, TaskState::Cancelled, Transition::Plain, at(9))
                .unwrap();
            assert_eq!(cancelled.state, TaskState::Cancelled, "from {state}");
            assert_eq!(cancelled.blocked_kind, None, "from {state}");
        }
    }

    #[test]
    fn a_reviewed_task_cannot_fail() {
        let mut store = store();
        let task = task(&mut store);
        store
            .transition(task.id, TaskState::Running, Transition::Plain, at(1))
            .unwrap();
        store
            .transition(task.id, TaskState::AwaitingReview, Transition::Plain, at(2))
            .unwrap();
        assert!(matches!(
            store
                .transition(
                    task.id,
                    TaskState::Failed,
                    Transition::Failed("nope".into()),
                    at(3)
                )
                .unwrap_err(),
            Error::IllegalTransition {
                from: TaskState::AwaitingReview,
                to: TaskState::Failed
            }
        ));
    }

    #[test]
    fn rejection_returns_to_running() {
        let mut store = store();
        let task = task(&mut store);
        store
            .transition(task.id, TaskState::Running, Transition::Plain, at(1))
            .unwrap();
        store
            .transition(task.id, TaskState::AwaitingReview, Transition::Plain, at(2))
            .unwrap();
        let resumed = store
            .transition(task.id, TaskState::Running, Transition::Plain, at(3))
            .unwrap();
        assert_eq!(resumed.state, TaskState::Running);
    }

    #[test]
    fn every_transition_is_logged() {
        let mut store = store();
        let task = task(&mut store);
        store
            .transition(task.id, TaskState::Running, Transition::Plain, at(1))
            .unwrap();
        store
            .transition(task.id, TaskState::AwaitingReview, Transition::Plain, at(2))
            .unwrap();

        let events = store.list_events(task.id).unwrap();
        let kinds: Vec<_> = events.iter().map(|e| e.kind.as_str()).collect();
        assert_eq!(
            kinds,
            ["task.created", "task.transition", "task.transition"]
        );
        assert_eq!(events[2].payload["from"], "running");
        assert_eq!(events[2].payload["to"], "awaiting-review");
    }

    #[test]
    fn a_task_cannot_leave_one_state_twice_under_contention() {
        // The daemon and the TUI are two processes on one file, so the read in
        // `transition` and its write are separated by a window another writer
        // can land in. The user's version: pressing `c` to cancel at the moment
        // the review screen commits. Both used to succeed — the commit real and
        // on a branch, the row recorded as cancelled, and the task filtered out
        // of the list either way, so it just vanished with the wrong verdict.
        const TASKS: usize = 200;
        let dir = tempfile::TempDir::new().unwrap();
        let path = dir.path().join("marver.db");

        let mut store = Store::open(&path).unwrap();
        let mut ids = Vec::new();
        for _ in 0..TASKS {
            let t = task(&mut store);
            store
                .transition(t.id, TaskState::Running, Transition::Plain, at(1))
                .unwrap();
            store
                .transition(t.id, TaskState::AwaitingReview, Transition::Plain, at(2))
                .unwrap();
            ids.push(t.id);
        }
        drop(store);

        // Two writers racing to move every task out of awaiting-review.
        let racers: Vec<_> = [TaskState::Committed, TaskState::Cancelled]
            .into_iter()
            .map(|next| {
                let path = path.clone();
                let ids = ids.clone();
                std::thread::spawn(move || {
                    let mut store = Store::open(&path).unwrap();
                    for id in ids {
                        let _ = store.transition(id, next, Transition::Plain, at(3));
                    }
                })
            })
            .collect();
        for racer in racers {
            racer.join().unwrap();
        }

        let store = Store::open(&path).unwrap();
        for id in ids {
            let exits = store
                .list_events(id)
                .unwrap()
                .iter()
                .filter(|e| e.kind == "task.transition" && e.payload["from"] == "awaiting-review")
                .count();
            assert_eq!(exits, 1, "task {id} left awaiting-review {exits} times");
        }
    }

    #[test]
    fn missing_task_is_reported() {
        let store = store();
        assert!(matches!(store.get_task(404), Err(Error::TaskNotFound(404))));
    }

    #[test]
    fn repo_upsert_is_idempotent_and_preserves_ignored() {
        let store = store();
        let path = Path::new("/Users/kit/workspace/marver");
        let first = store.upsert_repo(path, "marver", at(10)).unwrap();
        store.set_repo_ignored(first.id, true).unwrap();

        let second = store.upsert_repo(path, "marver", at(20)).unwrap();
        assert_eq!(second.id, first.id, "no duplicate row");
        assert_eq!(second.discovered_at, at(10), "discovery time is kept");
        assert_eq!(second.last_seen_at, at(20), "last seen is refreshed");
        assert!(second.ignored, "ignore flag survives a rescan");

        assert_eq!(store.list_repos(false).unwrap().len(), 0);
        assert_eq!(store.list_repos(true).unwrap().len(), 1);
    }

    #[test]
    fn tasks_can_span_several_repos() {
        let mut store = store();
        let a = store
            .upsert_repo(Path::new("/w/api"), "api", at(0))
            .unwrap();
        let b = store
            .upsert_repo(Path::new("/w/web"), "web", at(0))
            .unwrap();
        let task = store
            .create_task("t", "p", Path::new("/tmp/tasks"), &[a.id, b.id], at(0))
            .unwrap();

        store
            .record_worktree(task.id, a.id, Path::new("/t/1/api"), "task/1", "main")
            .unwrap();
        store
            .record_worktree(task.id, b.id, Path::new("/t/1/web"), "task/1", "develop")
            .unwrap();

        let worktrees = store.list_task_repos(task.id).unwrap();
        assert_eq!(worktrees.len(), 2);
        assert_eq!(worktrees[0].base_ref.as_deref(), Some("main"));
        assert_eq!(worktrees[1].base_ref.as_deref(), Some("develop"));
    }

    #[test]
    fn repos_are_selected_at_creation_before_any_worktree_exists() {
        let mut store = store();
        let repo = store
            .upsert_repo(Path::new("/w/api"), "api", at(0))
            .unwrap();
        let task = store
            .create_task("t", "p", Path::new("/tmp/tasks"), &[repo.id], at(0))
            .unwrap();

        let links = store.list_task_repos(task.id).unwrap();
        assert_eq!(links.len(), 1, "the selection is recorded immediately");
        assert!(
            !links[0].is_provisioned(),
            "a queued task owns no worktree yet"
        );
    }

    #[test]
    fn a_worktree_cannot_be_recorded_for_an_unselected_repo() {
        let mut store = store();
        let repo = store
            .upsert_repo(Path::new("/w/api"), "api", at(0))
            .unwrap();
        let task = task(&mut store);
        assert!(
            matches!(
                store.record_worktree(task.id, repo.id, Path::new("/t/x"), "b", "main"),
                Err(Error::RepoNotSelected { .. })
            ),
            "a worktree for an untargeted repo would be orphaned at teardown"
        );
    }

    #[test]
    fn clearing_worktrees_keeps_the_selection() {
        let mut store = store();
        let repo = store
            .upsert_repo(Path::new("/w/api"), "api", at(0))
            .unwrap();
        let task = store
            .create_task("t", "p", Path::new("/tmp/tasks"), &[repo.id], at(0))
            .unwrap();
        store
            .record_worktree(task.id, repo.id, Path::new("/t/1/api"), "b", "main")
            .unwrap();

        store.clear_worktrees(task.id).unwrap();
        let links = store.list_task_repos(task.id).unwrap();
        assert_eq!(links.len(), 1, "the task still targets the repo");
        assert!(!links[0].is_provisioned());
    }

    #[test]
    fn a_repo_joins_a_task_only_once() {
        let mut store = store();
        let repo = store
            .upsert_repo(Path::new("/w/api"), "api", at(0))
            .unwrap();
        let task = store
            .create_task("t", "p", Path::new("/tmp/tasks"), &[repo.id], at(0))
            .unwrap();
        assert!(store.select_repo(task.id, repo.id).is_err());
    }

    #[test]
    fn worktrees_require_a_real_task() {
        let store = store();
        let repo = store
            .upsert_repo(Path::new("/w/api"), "api", at(0))
            .unwrap();
        assert!(
            store.select_repo(999, repo.id).is_err(),
            "foreign keys must be enforced"
        );
    }

    #[test]
    fn listing_by_state_partitions_tasks() {
        let mut store = store();
        let a = task(&mut store);
        let b = task(&mut store);
        store
            .transition(a.id, TaskState::Running, Transition::Plain, at(1))
            .unwrap();

        let queued = store.list_tasks_in_state(TaskState::Queued).unwrap();
        let running = store.list_tasks_in_state(TaskState::Running).unwrap();
        assert_eq!(queued.iter().map(|t| t.id).collect::<Vec<_>>(), [b.id]);
        assert_eq!(running.iter().map(|t| t.id).collect::<Vec<_>>(), [a.id]);
        assert_eq!(store.list_tasks().unwrap().len(), 2);
    }

    #[test]
    fn session_name_is_recorded() {
        let mut store = store();
        let task = task(&mut store);
        store.set_session_name(task.id, "marver-1", at(5)).unwrap();
        let task = store.get_task(task.id).unwrap();
        assert_eq!(task.session_name.as_deref(), Some("marver-1"));
        assert_eq!(task.updated_at, at(5));
    }

    #[test]
    fn database_rejects_an_unknown_state() {
        let store = store();
        let err = store.conn.execute(
            "INSERT INTO tasks (title, prompt, state, workspace_dir, created_at, updated_at)
             VALUES ('x', 'x', 'nonsense', '/tmp', 0, 0)",
            [],
        );
        assert!(err.is_err(), "CHECK constraint should reject the state");
    }

    #[test]
    fn database_rejects_a_reason_without_being_blocked() {
        let store = store();
        let err = store.conn.execute(
            "INSERT INTO tasks (title, prompt, state, blocked_kind, workspace_dir, created_at, updated_at)
             VALUES ('x', 'x', 'running', 'question', '/tmp', 0, 0)",
            [],
        );
        assert!(
            err.is_err(),
            "a blocked_kind outside the blocked state is incoherent"
        );
    }
}

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

    /// A database at schema version 1, with a row in every table that
    /// references `tasks`.
    fn version_one_database(path: &Path) {
        let conn = Connection::open(path).expect("open");
        conn.execute_batch(MIGRATIONS[0]).expect("initial schema");
        conn.execute_batch("PRAGMA user_version = 1")
            .expect("stamp");
        conn.execute_batch(
            "INSERT INTO repos (path, name, ignored, discovered_at, last_seen_at)
                 VALUES ('/r/api', 'api', 0, 0, 0);
             INSERT INTO tasks (title, prompt, state, workspace_dir, created_at, updated_at)
                 VALUES ('old task', 'do it', 'running', '/w/1', 0, 0);
             INSERT INTO task_repos (task_id, repo_id, worktree_path, branch, base_ref)
                 VALUES (1, 1, '/w/1/api', 'marver/1-old', 'main');
             INSERT INTO events (task_id, kind, payload, created_at)
                 VALUES (1, 'task.transition', '{}', 0);",
        )
        .expect("seed");
    }

    #[test]
    fn migrating_to_the_paused_schema_keeps_every_row() {
        // The tasks table is rebuilt to widen a CHECK constraint, which means
        // dropping it while task_repos and events still reference it. With
        // foreign keys enforced -- which is how every connection is opened --
        // that DROP is not a schema change but a cascading delete: it empties
        // the child tables and leaves a database that looks entirely healthy.
        let dir = TempDir::new().unwrap();
        let path = dir.path().join("v1.db");
        version_one_database(&path);

        let store = Store::open(&path).expect("migrate");

        assert_eq!(store.schema_version().unwrap(), MIGRATIONS.len() as i64);
        assert_eq!(store.get_task(1).unwrap().title, "old task");
        assert_eq!(
            store.list_task_repos(1).unwrap().len(),
            1,
            "the cascade would have taken the worktree record"
        );
        assert_eq!(
            store.list_events(1).unwrap().len(),
            1,
            "and the task's whole history with it"
        );
    }

    #[test]
    fn the_rebuilt_tasks_table_keeps_the_constraints_it_had() {
        // A rebuild is a retyping, and a retyping can quietly drop a CHECK.
        // These are the two the old table enforced, still refused.
        let dir = TempDir::new().unwrap();
        let path = dir.path().join("v1.db");
        version_one_database(&path);
        let store = Store::open(&path).expect("migrate");

        assert!(
            store
                .conn
                .execute_batch(
                    "INSERT INTO tasks (title, prompt, state, workspace_dir, created_at, updated_at)
                     VALUES ('bad', 'x', 'nonsense', '/w/2', 0, 0)",
                )
                .is_err(),
            "the state list must still be closed"
        );
        assert!(
            store
                .conn
                .execute_batch(
                    "INSERT INTO tasks (title, prompt, state, failure_reason, workspace_dir,
                                        created_at, updated_at)
                     VALUES ('bad', 'x', 'running', 'why', '/w/2', 0, 0)",
                )
                .is_err(),
            "a failure reason still belongs only to a failed task"
        );
    }

    #[test]
    fn foreign_keys_are_enforced_again_after_migrating() {
        // They are turned off around the rebuild. Leaving them off would be a
        // silent, permanent loss of every referential guarantee the schema has.
        let dir = TempDir::new().unwrap();
        let path = dir.path().join("v1.db");
        version_one_database(&path);
        let store = Store::open(&path).expect("migrate");

        let enforced: i64 = store
            .conn
            .query_row("PRAGMA foreign_keys", [], |row| row.get(0))
            .unwrap();
        assert_eq!(enforced, 1);
        assert!(
            store
                .conn
                .execute_batch(
                    "INSERT INTO todos (task_id, text, done, created_at)
                     VALUES (9999, 'orphan', 0, 0)",
                )
                .is_err(),
            "a todo pointing at no task must be refused"
        );
    }

    #[test]
    fn the_usage_columns_arrive_empty_on_a_task_that_predates_them() {
        // Added by ALTER TABLE rather than a rebuild, so the existing rows keep
        // everything they had and gain nothing but nulls -- and a null must read
        // as "not known", never as zero tokens spent.
        let dir = TempDir::new().unwrap();
        let path = dir.path().join("v1.db");
        version_one_database(&path);

        let store = Store::open(&path).expect("migrate");

        let task = store.get_task(1).unwrap();
        assert_eq!(task.title, "old task", "the row survived both migrations");
        assert!(!task.usage.is_known());
        assert_eq!(task.usage.output_tokens, None);
        assert_eq!(task.usage.transcript_offset, 0, "read it from the start");
    }

    #[test]
    fn a_fresh_database_arrives_at_the_same_place_as_a_migrated_one() {
        let dir = TempDir::new().unwrap();
        let migrated = dir.path().join("v1.db");
        version_one_database(&migrated);
        let migrated = Store::open(&migrated).expect("migrate");
        let fresh = Store::open(dir.path().join("new.db")).expect("create");

        assert_eq!(
            migrated.schema_version().unwrap(),
            fresh.schema_version().unwrap()
        );
        for store in [&migrated, &fresh] {
            let mut names: Vec<String> = store
                .conn
                .prepare("SELECT name FROM sqlite_master WHERE type = 'table' ORDER BY name")
                .unwrap()
                .query_map([], |row| row.get(0))
                .unwrap()
                .collect::<rusqlite::Result<_>>()
                .unwrap();
            names.retain(|name: &String| !name.starts_with("sqlite_"));
            assert_eq!(names, ["events", "repos", "task_repos", "tasks", "todos"]);
        }
    }
}

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

    fn at(secs: i64) -> DateTime<Utc> {
        DateTime::from_timestamp(secs, 0).expect("valid timestamp")
    }

    fn store_with_task() -> (Store, i64) {
        let mut store = Store::open_in_memory().unwrap();
        let id = store
            .create_task("a task", "do it", Path::new("/tmp/tasks"), &[], at(0))
            .unwrap()
            .id;
        (store, id)
    }

    #[test]
    fn the_two_scopes_do_not_see_each_other() {
        // The whole design rests on this: `↵` sends a task todo to an agent and
        // turns a global one into a task, so a list that mixed them would offer
        // the wrong verb for half its rows.
        let (store, task_id) = store_with_task();
        store
            .add_todo(TodoScope::Global, "upgrade ratatui", at(1))
            .unwrap();
        store
            .add_todo(TodoScope::Task(task_id), "also handle nulls", at(2))
            .unwrap();

        let global = store.list_todos(TodoScope::Global).unwrap();
        let task = store.list_todos(TodoScope::Task(task_id)).unwrap();

        assert_eq!(global.len(), 1);
        assert_eq!(global[0].text, "upgrade ratatui");
        assert!(global[0].is_global());
        assert_eq!(task.len(), 1);
        assert_eq!(task[0].text, "also handle nulls");
        assert_eq!(task[0].task_id, Some(task_id));
    }

    #[test]
    fn an_empty_todo_is_refused() {
        let (store, _) = store_with_task();
        assert!(matches!(
            store.add_todo(TodoScope::Global, "   ", at(1)),
            Err(Error::EmptyTodo)
        ));
    }

    #[test]
    fn text_is_trimmed_on_the_way_in() {
        let (store, _) = store_with_task();
        let todo = store
            .add_todo(TodoScope::Global, "  spaced  ", at(1))
            .unwrap();
        assert_eq!(todo.text, "spaced");
    }

    #[test]
    fn done_todos_stay_in_the_list() {
        // Ticking one off must not look like deleting it; `space` and `x` are
        // different keys and mean different things.
        let (store, _) = store_with_task();
        let todo = store.add_todo(TodoScope::Global, "note", at(1)).unwrap();

        let done = store.set_todo_done(todo.id, true).unwrap();

        assert!(done.done);
        assert_eq!(store.list_todos(TodoScope::Global).unwrap().len(), 1);
        assert_eq!(store.open_todo_count(TodoScope::Global).unwrap(), 0);
    }

    #[test]
    fn counts_are_per_task_and_skip_the_finished_ones() {
        let (mut store, first) = store_with_task();
        let second = store
            .create_task("another", "do it", Path::new("/tmp/tasks"), &[], at(0))
            .unwrap()
            .id;
        store
            .add_todo(TodoScope::Task(first), "one", at(1))
            .unwrap();
        let done = store
            .add_todo(TodoScope::Task(first), "two", at(2))
            .unwrap();
        store
            .add_todo(TodoScope::Task(second), "three", at(3))
            .unwrap();
        store
            .add_todo(TodoScope::Global, "not a task's", at(4))
            .unwrap();
        store.set_todo_done(done.id, true).unwrap();

        let counts = store.open_todo_counts().unwrap();

        assert_eq!(counts.get(&first), Some(&1));
        assert_eq!(counts.get(&second), Some(&1));
        assert_eq!(counts.len(), 2, "the global todo belongs to no task");
    }

    #[test]
    fn a_todo_goes_when_its_task_does() {
        let (store, task_id) = store_with_task();
        store
            .add_todo(TodoScope::Task(task_id), "note", at(1))
            .unwrap();

        store
            .conn
            .execute("DELETE FROM tasks WHERE id = ?1", params![task_id])
            .unwrap();

        assert!(
            store
                .list_todos(TodoScope::Task(task_id))
                .unwrap()
                .is_empty()
        );
    }

    #[test]
    fn deleting_something_that_is_not_there_says_so() {
        let (store, _) = store_with_task();
        assert!(matches!(
            store.delete_todo(404),
            Err(Error::TodoNotFound(404))
        ));
    }
}

#[cfg(test)]
mod usage_tests {
    use super::*;
    use crate::usage::Usage;

    fn at(secs: i64) -> DateTime<Utc> {
        DateTime::from_timestamp(secs, 0).expect("valid timestamp")
    }

    fn store_with_task() -> (Store, i64) {
        let mut store = Store::open_in_memory().unwrap();
        let id = store
            .create_task("a task", "do it", Path::new("/tmp/tasks"), &[], at(0))
            .unwrap()
            .id;
        (store, id)
    }

    #[test]
    fn a_new_task_knows_nothing_about_what_it_has_spent() {
        let (store, id) = store_with_task();
        let usage = store.get_task(id).unwrap().usage;

        assert!(!usage.is_known());
        assert_eq!(usage.transcript_offset, 0);
        assert_eq!(usage.output_tokens, None, "unknown is not zero");
    }

    #[test]
    fn context_replaces_and_output_accumulates() {
        // The two numbers mean different things: context is how full the window
        // was on the last turn, output is a total across every turn. Adding
        // context would read as hundreds of thousands of tokens over the limit;
        // replacing output would report only the most recent turn's.
        let (store, id) = store_with_task();

        store
            .record_usage(
                id,
                &Usage {
                    model: Some("claude-opus-5".into()),
                    context_tokens: Some(1_000),
                    output_tokens: 200,
                    offset: 512,
                },
            )
            .unwrap();
        store
            .record_usage(
                id,
                &Usage {
                    model: Some("claude-opus-5".into()),
                    context_tokens: Some(4_000),
                    output_tokens: 50,
                    offset: 900,
                },
            )
            .unwrap();

        let usage = store.get_task(id).unwrap().usage;
        assert_eq!(usage.context_tokens, Some(4_000), "the latest level");
        assert_eq!(usage.output_tokens, Some(250), "the running total");
        assert_eq!(usage.transcript_offset, 900);
    }

    #[test]
    fn a_read_that_found_nothing_advances_the_offset_and_keeps_the_rest() {
        // Walking over user turns is progress, but it is not a reason to forget
        // which model the agent is on.
        let (store, id) = store_with_task();
        store
            .record_usage(
                id,
                &Usage {
                    model: Some("claude-opus-5".into()),
                    context_tokens: Some(1_000),
                    output_tokens: 200,
                    offset: 512,
                },
            )
            .unwrap();

        store
            .record_usage(
                id,
                &Usage {
                    offset: 800,
                    ..Usage::default()
                },
            )
            .unwrap();

        let usage = store.get_task(id).unwrap().usage;
        assert_eq!(usage.model.as_deref(), Some("claude-opus-5"));
        assert_eq!(usage.context_tokens, Some(1_000));
        assert_eq!(usage.output_tokens, Some(200));
        assert_eq!(usage.transcript_offset, 800);
    }

    #[test]
    fn the_transcript_path_is_remembered() {
        let (store, id) = store_with_task();
        store
            .set_transcript_path(id, Path::new("/t/session.jsonl"))
            .unwrap();

        assert_eq!(
            store.get_task(id).unwrap().usage.transcript_path,
            Some(PathBuf::from("/t/session.jsonl"))
        );
    }

    #[test]
    fn recording_against_a_task_that_is_gone_says_so() {
        let (store, _) = store_with_task();
        assert!(matches!(
            store.record_usage(404, &Usage::default()),
            Err(Error::TaskNotFound(404))
        ));
    }
}