marver 0.0.16

//! 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};

#[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("unreadable {field} in database: {value:?}")]
    Corrupt { field: &'static str, value: String },
}

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")];

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(())
    }

    /// 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> {
        let current = self.get_task(id)?;
        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)
    }

    // ---- 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"
    };
}
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(());
    }
    let tx = conn.transaction()?;
    for (index, migration) in MIGRATIONS.iter().enumerate().skip(version as usize) {
        tx.execute_batch(migration)?;
        // user_version takes no bind parameters.
        tx.execute_batch(&format!("PRAGMA user_version = {}", index + 1))?;
    }
    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(),
    })
}

/// 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)),
        },
    }))
}

#[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"
        );
    }
}