marver 0.0.6

//! `marverd`: the process that outlives the TUI.
//!
//! It owns the database, listens for Claude Code hooks, promotes queued tasks
//! when a slot frees, and notifies. Everything the TUI does is a read or a
//! request; everything that must happen while nobody is watching happens here.
//!
//! The loop has two inputs with very different shapes — hooks arrive whenever
//! an agent does something, and the scheduler needs waking on a timer. A reader
//! thread turns the socket into a channel, so the main loop can block on
//! `recv_timeout` and treat "a hook arrived" and "the timer expired" as the same
//! kind of event.
//!
//! Startup runs [`Daemon::reconcile`] before anything else. A task recorded as
//! `running` whose tmux session is gone means the machine restarted or the agent
//! died while the daemon was down; without reconciliation it would hold a
//! concurrency slot forever and never finish.

use std::os::unix::process::CommandExt;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{RecvTimeoutError, channel};
use std::time::{Duration, Instant};

use chrono::{DateTime, Utc};

use crate::domain::{Task, TaskState};
use crate::hook::{self, Delivery, Receiver};
use crate::launcher::Launcher;
use crate::notify::{Notification, Notify};
use crate::scan::Scanner;
use crate::scheduler::{Scheduler, Tick};
use crate::store::{Store, Transition};
use crate::tmux::{self, Tmux};
use crate::worktree::WorktreeManager;

#[derive(Debug, thiserror::Error)]
pub enum Error {
    #[error(transparent)]
    Store(#[from] crate::store::Error),
    #[error(transparent)]
    Hook(#[from] hook::Error),
    #[error(transparent)]
    Scheduler(#[from] crate::scheduler::Error),
    #[error(transparent)]
    Scan(#[from] crate::scan::Error),
    #[error(transparent)]
    Tmux(#[from] tmux::Error),
    #[error("could not start a daemon: {0}")]
    Spawn(#[source] std::io::Error),
    #[error("a daemon was started but never began listening; see {}", .log.display())]
    NeverListened { log: PathBuf },
}

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

/// How often the scheduler is woken when no hook arrives.
pub const DEFAULT_TICK: Duration = Duration::from_secs(2);

/// How often `~/workspace` is rescanned for repos.
pub const DEFAULT_SCAN_INTERVAL: Duration = Duration::from_secs(60);

#[derive(Debug, Clone)]
pub struct Config {
    /// Where everything below lives. Kept so an auto-started daemon can be
    /// handed the same `--data-dir` the caller is using; deriving it from `db`
    /// would work until one of these paths stopped being a sibling.
    pub data_dir: PathBuf,
    pub db: PathBuf,
    pub socket: PathBuf,
    /// Where a detached daemon's output goes, since it has no terminal.
    pub log: PathBuf,
    /// Directory task workspaces are created under.
    pub workspace_root: PathBuf,
    /// Directory scanned for repos.
    pub scan_root: PathBuf,
    /// Path to the marver binary, baked into generated hook commands.
    pub marver_bin: PathBuf,
    pub cap: usize,
    pub tick: Duration,
    pub scan_interval: Duration,
}

impl Config {
    /// Defaults rooted at `data_dir`.
    pub fn new(data_dir: impl AsRef<Path>, scan_root: impl Into<PathBuf>) -> Self {
        let data_dir = data_dir.as_ref();
        Self {
            data_dir: data_dir.to_path_buf(),
            db: data_dir.join("marver.db"),
            socket: data_dir.join("marverd.sock"),
            log: data_dir.join("daemon.log"),
            workspace_root: data_dir.join("tasks"),
            scan_root: scan_root.into(),
            marver_bin: std::env::current_exe().unwrap_or_else(|_| PathBuf::from("marver")),
            cap: crate::scheduler::DEFAULT_CAP,
            tick: DEFAULT_TICK,
            scan_interval: DEFAULT_SCAN_INTERVAL,
        }
    }

    /// `$XDG_DATA_HOME/marver` or `~/.local/share/marver`.
    pub fn default_data_dir() -> PathBuf {
        if let Ok(xdg) = std::env::var("XDG_DATA_HOME") {
            return PathBuf::from(xdg).join("marver");
        }
        let home = std::env::var("HOME").unwrap_or_else(|_| ".".to_string());
        PathBuf::from(home).join(".local/share/marver")
    }

    pub fn default_scan_root() -> PathBuf {
        let home = std::env::var("HOME").unwrap_or_else(|_| ".".to_string());
        PathBuf::from(home).join("workspace")
    }
}

/// How long to wait for an auto-started daemon to begin listening.
///
/// Generous because it covers a cold binary on a slow disk, and cheap because
/// the poll returns the instant the socket answers. The socket is bound in
/// [`Daemon::new`], before reconciliation or the first scan, so a healthy
/// daemon answers long before this.
const STARTUP_TIMEOUT: Duration = Duration::from_secs(5);
const STARTUP_POLL: Duration = Duration::from_millis(25);

/// What [`ensure_running`] found, or did about it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Startup {
    /// One was already listening. Nothing was spawned.
    AlreadyRunning,
    /// A detached daemon was started, and is now listening.
    Started { pid: u32 },
}

/// Whether a daemon is listening for this configuration.
pub fn is_running(config: &Config) -> bool {
    hook::is_listening(&config.socket)
}

/// Start a daemon unless one is already listening.
///
/// This is what lets `marver` be the only command anyone types: tmux starts its
/// own server when a client needs one, and marver is built on tmux, so it
/// inherits the shape. `marver daemon` stays for running it under a supervisor
/// or watching it in the foreground.
///
/// The child is put in **its own process group**, which is the part that has to
/// be right. Terminal-generated signals go to the foreground process group, so a
/// plain child would take `ctrl-c` in the TUI as well — killing the supervisor
/// for every agent that was running, which is the one guarantee marver makes.
/// Its stdio goes to [`Config::log`] rather than the terminal it must outlive.
///
/// Races resolve themselves: two TUIs starting at once both spawn, and the
/// socket bind means exactly one survives while the other exits with
/// [`hook::Error::AlreadyRunning`].
pub fn ensure_running(config: &Config) -> Result<Startup> {
    if is_running(config) {
        return Ok(Startup::AlreadyRunning);
    }

    if let Some(parent) = config.log.parent() {
        std::fs::create_dir_all(parent).map_err(Error::Spawn)?;
    }
    // Appended, never truncated: the log of the daemon that just died is the
    // only evidence of why, and starting a replacement must not erase it.
    let log = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(&config.log)
        .map_err(Error::Spawn)?;

    let child = std::process::Command::new(&config.marver_bin)
        .arg("daemon")
        .arg("--data-dir")
        .arg(&config.data_dir)
        .arg("--scan-root")
        .arg(&config.scan_root)
        .arg("--cap")
        .arg(config.cap.to_string())
        .stdin(std::process::Stdio::null())
        .stdout(std::process::Stdio::from(
            log.try_clone().map_err(Error::Spawn)?,
        ))
        .stderr(std::process::Stdio::from(log))
        .process_group(0)
        .spawn()
        .map_err(Error::Spawn)?;

    // Spawning proves a process started, not that it got as far as the socket.
    // A daemon that dies on startup would otherwise look exactly like a healthy
    // one that simply never schedules anything.
    let deadline = Instant::now() + STARTUP_TIMEOUT;
    while Instant::now() < deadline {
        if is_running(config) {
            return Ok(Startup::Started { pid: child.id() });
        }
        std::thread::sleep(STARTUP_POLL);
    }
    Err(Error::NeverListened {
        log: config.log.clone(),
    })
}

/// What one pass of the daemon did, for logging.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Pass {
    pub started: Vec<i64>,
    pub failed: Vec<(i64, String)>,
    pub notified: Vec<Notification>,
    /// Tasks whose agent was still running after they finished.
    pub reaped: Vec<i64>,
}

pub struct Daemon<N: Notify> {
    store: Store,
    receiver: Option<Receiver>,
    scheduler: Scheduler,
    launcher: Launcher,
    notifier: N,
    tmux: Tmux,
    scanner: Scanner,
    config: Config,
}

impl<N: Notify> Daemon<N> {
    /// Open the database, bind the socket, and prepare the moving parts.
    pub fn new(config: Config, tmux: Tmux, notifier: N) -> Result<Self> {
        let store = Store::open(&config.db)?;
        let receiver = Receiver::bind(&config.socket)?;
        let launcher = Launcher::new(
            tmux.clone(),
            WorktreeManager::new(&config.workspace_root),
            &config.marver_bin,
            &config.socket,
        );
        Ok(Self {
            store,
            receiver: Some(receiver),
            scheduler: Scheduler::new(config.cap),
            launcher,
            notifier,
            tmux,
            scanner: Scanner::new(&config.scan_root),
            config,
        })
    }

    pub fn store(&self) -> &Store {
        &self.store
    }

    pub fn store_mut(&mut self) -> &mut Store {
        &mut self.store
    }

    pub fn socket(&self) -> &Path {
        &self.config.socket
    }

    /// Fail tasks whose agent is gone.
    ///
    /// A `running` or `blocked` task whose tmux session no longer exists cannot
    /// make progress: the session held the agent, and nothing recreates it.
    /// Left alone it would occupy a concurrency slot for ever. Returns the tasks
    /// it failed.
    pub fn reconcile(&mut self, now: DateTime<Utc>) -> Result<Vec<i64>> {
        let mut orphaned = Vec::new();
        for state in [TaskState::Running, TaskState::Blocked] {
            for task in self.store.list_tasks_in_state(state)? {
                let session = task
                    .session_name
                    .clone()
                    .unwrap_or_else(|| tmux::session_name(task.id));
                // `session_exists`, not `has_session`: the latter collapses
                // "no such session" and "tmux could not be run at all" into
                // false, so a missing binary or an unset PATH under a service
                // manager failed every live task at once — abandoning agents
                // that were still working, permanently, since `failed` is
                // terminal and their later hooks are all rejected.
                if self.tmux.session_exists(&session).map_err(|err| {
                    eprintln!("marverd: cannot reconcile, tmux is unavailable: {err}");
                    err
                })? {
                    continue;
                }
                self.store.transition(
                    task.id,
                    TaskState::Failed,
                    Transition::Failed(
                        "the agent's tmux session was gone when marverd started".to_string(),
                    ),
                    now,
                )?;
                orphaned.push(task.id);
            }
        }
        Ok(orphaned)
    }

    /// Refresh the repo list.
    pub fn scan(&mut self, now: DateTime<Utc>) -> Result<usize> {
        Ok(self.scanner.sync(&self.store, now)?.present.len())
    }

    /// Apply a hook delivery and notify if it moved the task.
    pub fn handle(
        &mut self,
        delivery: &Delivery,
        now: DateTime<Utc>,
    ) -> Result<Option<Notification>> {
        let outcome = hook::apply(&mut self.store, delivery, now)?;
        match outcome {
            hook::Outcome::Moved { .. } => {
                let task = self.store.get_task(delivery.task_id)?;
                Ok(self.announce(&task))
            }
            hook::Outcome::Recorded { .. } => Ok(None),
        }
    }

    /// Kill the agent of any task that has finished.
    ///
    /// A task reaching a terminal state does not stop its agent: Claude Code
    /// stays running after `Stop`, and cancelling in the TUI only writes a row.
    /// Without this a cancelled task leaves a live agent burning tokens in a
    /// session nobody will ever look at again, and the count only grows.
    ///
    /// Only the session is killed. The worktrees stay: `teardown` uses
    /// `git worktree remove --force`, which discards uncommitted changes, and
    /// throwing away an agent's unreviewed work as a side effect of pressing
    /// `c` would be far worse than leaving a directory behind. Removing them is
    /// a decision for the user to make explicitly.
    ///
    /// Returns the tasks whose sessions it killed.
    pub fn reap(&mut self, now: DateTime<Utc>) -> Result<Vec<i64>> {
        let mut reaped = Vec::new();
        for state in [TaskState::Cancelled, TaskState::Failed] {
            for task in self.store.list_tasks_in_state(state)? {
                let Some(session) = task.session_name.clone() else {
                    continue;
                };
                if !self.tmux.has_session(&session) {
                    continue;
                }
                if self.tmux.kill_session(&session).is_ok() {
                    // Clearing the name is what stops this retrying for ever.
                    self.store.clear_session_name(task.id, now)?;
                    reaped.push(task.id);
                }
            }
        }
        Ok(reaped)
    }

    /// Start whatever the cap allows, notifying about anything that failed.
    pub fn tick(&mut self, now: DateTime<Utc>) -> Result<Pass> {
        // Before scheduling: a finished task's agent may still be running, and
        // the slot it frees is only real once it is gone.
        let reaped = self.reap(now)?;

        let Tick {
            started, failed, ..
        } = self.scheduler.tick(&mut self.store, &self.launcher, now)?;

        let mut notified = Vec::new();
        for (id, _) in &failed {
            if let Ok(task) = self.store.get_task(*id)
                && let Some(sent) = self.announce(&task)
            {
                notified.push(sent);
            }
        }
        Ok(Pass {
            started,
            failed,
            notified,
            reaped,
        })
    }

    /// Deliver a notification, swallowing backend failures.
    ///
    /// A missing `notify-send` is not a reason to stop scheduling work; the
    /// user losing a banner is far less bad than the daemon stalling.
    fn announce(&self, task: &Task) -> Option<Notification> {
        match crate::notify::for_transition(task) {
            Some(notification) => match self.notifier.send(&notification) {
                Ok(()) => Some(notification),
                Err(err) => {
                    eprintln!("marverd: could not notify: {err}");
                    None
                }
            },
            None => None,
        }
    }

    /// Run until `shutdown` is set.
    ///
    /// Hooks arrive on a reader thread and are delivered over a channel, so the
    /// timer and the socket are the same kind of event here. Every hook is
    /// followed by a tick: a task that just finished has freed a slot, and
    /// waiting up to `tick` to notice would leave the queue idle for no reason.
    pub fn run(&mut self, shutdown: Arc<AtomicBool>) -> Result<()> {
        let now = Utc::now();
        let orphaned = self.reconcile(now)?;
        if !orphaned.is_empty() {
            eprintln!("marverd: failed {} orphaned task(s)", orphaned.len());
        }
        self.scan(now)?;

        let receiver = self
            .receiver
            .take()
            .expect("run may only be called once per daemon");
        let (tx, rx) = channel();
        std::thread::spawn(move || {
            loop {
                match receiver.accept() {
                    Ok(delivery) => {
                        if tx.send(delivery).is_err() {
                            break; // the daemon has gone
                        }
                    }
                    // A broken client must not take the listener down with it.
                    // Only an io error is the listener's own, and stopping is
                    // silent — the socket file is unlinked on the way out and
                    // every later hook exits 0 with nobody reading its stderr,
                    // so tasks would simply never leave `running` again.
                    // A liveness check, not a hook. Silent on purpose: logging
                    // a line every time someone runs `marver status` or opens
                    // the TUI would train whoever reads this log to skim it.
                    Err(hook::Error::Probe) => {}
                    Err(err @ (hook::Error::Malformed(_) | hook::Error::Rejected(_))) => {
                        eprintln!("marverd: ignoring hook: {err}");
                    }
                    Err(err) => {
                        eprintln!("marverd: hook listener stopped: {err}");
                        break;
                    }
                }
            }
        });

        let mut last_scan = Instant::now();
        while !shutdown.load(Ordering::Relaxed) {
            match rx.recv_timeout(self.config.tick) {
                Ok(delivery) => {
                    let now = Utc::now();
                    if let Err(err) = self.handle(&delivery, now) {
                        eprintln!("marverd: hook for task {}: {err}", delivery.task_id);
                    }
                    let _ = self.tick(now);
                }
                Err(RecvTimeoutError::Timeout) => {
                    let _ = self.tick(Utc::now());
                }
                Err(RecvTimeoutError::Disconnected) => break,
            }

            if last_scan.elapsed() >= self.config.scan_interval {
                let _ = self.scan(Utc::now());
                last_scan = Instant::now();
            }
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::domain::{BlockedKind, Repo};
    use crate::git::testing::init_repo;
    use crate::hook::Payload;
    use crate::notify::Notify;
    use crate::tmux::testing::TestServer;
    use serde_json::json;
    use std::cell::RefCell;
    use tempfile::TempDir;

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

    #[derive(Default)]
    struct Recorder {
        sent: RefCell<Vec<Notification>>,
    }

    impl Notify for Recorder {
        fn send(&self, notification: &Notification) -> std::result::Result<(), String> {
            self.sent.borrow_mut().push(notification.clone());
            Ok(())
        }
    }

    struct Fixture {
        _tmp: TempDir,
        server: TestServer,
        repos_dir: PathBuf,
        daemon: Daemon<Recorder>,
    }

    impl Fixture {
        fn new() -> Self {
            let tmp = TempDir::new().unwrap();
            let server = TestServer::new();
            let repos_dir = tmp.path().join("repos");
            std::fs::create_dir_all(&repos_dir).unwrap();

            let mut config = Config::new(tmp.path(), &repos_dir);
            config.cap = 2;
            let mut daemon = Daemon::new(config, server.tmux.clone(), Recorder::default()).unwrap();
            // Stand in for the agent so tests need no network.
            daemon.launcher = Launcher::new(
                server.tmux.clone(),
                WorktreeManager::new(tmp.path().join("tasks")),
                PathBuf::from("/bin/marver"),
                tmp.path().join("marverd.sock"),
            )
            .agent(crate::launcher::testing::stub_agent(tmp.path()));

            Self {
                repos_dir,
                daemon,
                server,
                _tmp: tmp,
            }
        }

        fn repo(&self, name: &str) -> Repo {
            let path = self.repos_dir.join(name);
            init_repo(&path, "main");
            self.daemon.store.upsert_repo(&path, name, at(0)).unwrap()
        }

        fn queue(&mut self, title: &str, repos: &[Repo]) -> Task {
            let ids: Vec<i64> = repos.iter().map(|r| r.id).collect();
            let root = self.daemon.config.workspace_root.clone();
            self.daemon
                .store
                .create_task(title, "do it", &root, &ids, at(0))
                .unwrap()
        }

        fn notifications(&self) -> Vec<Notification> {
            self.daemon.notifier.sent.borrow().clone()
        }
    }

    fn delivery(task_id: i64, event: &str, extra: serde_json::Value) -> Delivery {
        let mut raw = json!({
            "session_id": "s1",
            "cwd": "/tmp/w",
            "hook_event_name": event,
        });
        if let (serde_json::Value::Object(base), serde_json::Value::Object(more)) =
            (&mut raw, extra)
        {
            base.extend(more);
        }
        Delivery {
            task_id,
            payload: Payload::from_json(raw),
        }
    }

    #[test]
    fn the_socket_is_bound_on_startup() {
        let fx = Fixture::new();
        assert!(fx.daemon.socket().exists());
    }

    #[test]
    fn a_second_daemon_cannot_take_a_live_socket() {
        let fx = Fixture::new();
        let config = fx.daemon.config.clone();
        assert!(
            Daemon::new(config, fx.server.tmux.clone(), Recorder::default()).is_err(),
            "two daemons on one socket would both receive half the hooks"
        );
    }

    #[test]
    fn ticking_starts_queued_tasks_up_to_the_cap() {
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let a = fx.queue("first", std::slice::from_ref(&repo));
        let b = fx.queue("second", std::slice::from_ref(&repo));
        let c = fx.queue("third", std::slice::from_ref(&repo));

        let pass = fx.daemon.tick(at(5)).unwrap();

        assert_eq!(pass.started, [a.id, b.id]);
        assert_eq!(
            fx.daemon.store.get_task(c.id).unwrap().state,
            TaskState::Queued
        );
        assert!(fx.server.tmux.has_session(&tmux::session_name(a.id)));
    }

    #[test]
    fn a_cancelled_tasks_agent_is_killed() {
        // Cancelling in the TUI writes a row; it does not stop the agent, and
        // Claude Code keeps running after Stop regardless. Nothing used to
        // reap either, so a cancelled task left a live agent burning tokens in
        // a session nobody would ever open again.
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let task = fx.queue("doomed", std::slice::from_ref(&repo));
        fx.daemon.tick(at(5)).unwrap();
        let session = tmux::session_name(task.id);
        assert!(fx.server.tmux.has_session(&session), "should be running");

        fx.daemon
            .store
            .transition(task.id, TaskState::Cancelled, Transition::Plain, at(6))
            .unwrap();
        let pass = fx.daemon.tick(at(7)).unwrap();

        assert_eq!(pass.reaped, [task.id]);
        assert!(
            !fx.server.tmux.has_session(&session),
            "the agent must not outlive the task"
        );
    }

    #[test]
    fn reaping_leaves_the_worktrees_alone() {
        // `git worktree remove --force` discards uncommitted changes. Throwing
        // an agent's unreviewed work away as a side effect of pressing `c`
        // would be far worse than leaving a directory behind.
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let task = fx.queue("doomed", std::slice::from_ref(&repo));
        fx.daemon.tick(at(5)).unwrap();
        let worktree = fx.daemon.store.list_task_repos(task.id).unwrap()[0]
            .worktree_path
            .clone()
            .expect("provisioned");

        fx.daemon
            .store
            .transition(task.id, TaskState::Cancelled, Transition::Plain, at(6))
            .unwrap();
        fx.daemon.tick(at(7)).unwrap();

        assert!(worktree.exists(), "the work must survive the cancel");
    }

    #[test]
    fn reaping_is_not_retried_for_ever() {
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let task = fx.queue("doomed", std::slice::from_ref(&repo));
        fx.daemon.tick(at(5)).unwrap();
        fx.daemon
            .store
            .transition(task.id, TaskState::Cancelled, Transition::Plain, at(6))
            .unwrap();

        assert_eq!(fx.daemon.tick(at(7)).unwrap().reaped, [task.id]);
        assert!(
            fx.daemon.tick(at(8)).unwrap().reaped.is_empty(),
            "a reaped task has nothing left to kill"
        );
    }

    #[test]
    fn a_stop_hook_moves_a_task_to_review_and_notifies() {
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let task = fx.queue("first", std::slice::from_ref(&repo));
        fx.daemon.tick(at(5)).unwrap();

        let sent = fx
            .daemon
            .handle(&delivery(task.id, "Stop", json!({})), at(6))
            .unwrap();

        assert_eq!(
            fx.daemon.store.get_task(task.id).unwrap().state,
            TaskState::AwaitingReview
        );
        assert_eq!(sent.unwrap().title, "marver — Ready for review");
    }

    #[test]
    fn a_permission_prompt_notifies_with_its_reason() {
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let task = fx.queue("first", std::slice::from_ref(&repo));
        fx.daemon.tick(at(5)).unwrap();

        fx.daemon
            .handle(
                &delivery(
                    task.id,
                    "Notification",
                    json!({"notification_type": "permission_prompt", "message": "edit main.rs"}),
                ),
                at(6),
            )
            .unwrap();

        let stored = fx.daemon.store.get_task(task.id).unwrap();
        assert_eq!(stored.state, TaskState::Blocked);
        assert_eq!(stored.blocked_kind, Some(BlockedKind::PermissionPrompt));
        assert_eq!(
            fx.notifications()[0].body,
            "first: edit main.rs",
            "the reason is what makes the banner actionable"
        );
    }

    #[test]
    fn an_informational_notification_neither_moves_nor_notifies() {
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let task = fx.queue("first", std::slice::from_ref(&repo));
        fx.daemon.tick(at(5)).unwrap();

        let sent = fx
            .daemon
            .handle(
                &delivery(
                    task.id,
                    "Notification",
                    json!({"notification_type": "auth_success"}),
                ),
                at(6),
            )
            .unwrap();

        assert!(sent.is_none());
        assert_eq!(
            fx.daemon.store.get_task(task.id).unwrap().state,
            TaskState::Running
        );
    }

    #[test]
    fn finishing_a_task_frees_its_slot_for_the_next_tick() {
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let a = fx.queue("first", std::slice::from_ref(&repo));
        let b = fx.queue("second", std::slice::from_ref(&repo));
        let c = fx.queue("third", std::slice::from_ref(&repo));
        fx.daemon.tick(at(5)).unwrap();
        assert_eq!(
            fx.daemon.store.get_task(c.id).unwrap().state,
            TaskState::Queued
        );

        // `a` finishes; awaiting-review does not hold a slot.
        fx.daemon
            .handle(&delivery(a.id, "Stop", json!({})), at(6))
            .unwrap();
        let pass = fx.daemon.tick(at(7)).unwrap();

        assert_eq!(pass.started, [c.id]);
        assert_eq!(
            fx.daemon.store.get_task(b.id).unwrap().state,
            TaskState::Running
        );
    }

    #[test]
    fn reconcile_fails_a_task_whose_session_vanished() {
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let task = fx.queue("first", std::slice::from_ref(&repo));
        fx.daemon.tick(at(5)).unwrap();

        // The machine restarted: tmux is gone, the database is not.
        fx.server
            .tmux
            .kill_session(&tmux::session_name(task.id))
            .unwrap();

        let orphaned = fx.daemon.reconcile(at(10)).unwrap();

        assert_eq!(orphaned, [task.id]);
        let stored = fx.daemon.store.get_task(task.id).unwrap();
        assert_eq!(stored.state, TaskState::Failed);
        assert!(
            stored.failure_reason.is_some(),
            "the user needs to know why it died"
        );
    }

    #[test]
    fn reconcile_does_not_fail_tasks_when_tmux_itself_is_unavailable() {
        // "No such session" and "tmux could not be run" are different answers,
        // and has_session returned false for both. A missing binary or an unset
        // PATH under a service manager therefore failed every live task at
        // once, abandoning agents that were still working -- permanently, since
        // `failed` is terminal and every later hook is rejected.
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let task = fx.queue("working", std::slice::from_ref(&repo));
        fx.daemon.tick(at(5)).unwrap();

        // Point the daemon at a tmux that cannot be spawned at all.
        fx.daemon.tmux = Tmux::with_binary("definitely-not-tmux");

        let err = fx.daemon.reconcile(at(10));

        assert!(err.is_err(), "an unusable tmux is not an answer");
        assert_eq!(
            fx.daemon.store.get_task(task.id).unwrap().state,
            TaskState::Running,
            "a live agent must not be abandoned because tmux would not run"
        );
    }

    #[test]
    fn reconcile_leaves_live_tasks_alone() {
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let task = fx.queue("first", std::slice::from_ref(&repo));
        fx.daemon.tick(at(5)).unwrap();

        assert!(fx.daemon.reconcile(at(10)).unwrap().is_empty());
        assert_eq!(
            fx.daemon.store.get_task(task.id).unwrap().state,
            TaskState::Running
        );
    }

    #[test]
    fn an_orphaned_task_stops_holding_its_slot() {
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let a = fx.queue("first", std::slice::from_ref(&repo));
        let b = fx.queue("second", std::slice::from_ref(&repo));
        let c = fx.queue("third", std::slice::from_ref(&repo));
        fx.daemon.tick(at(5)).unwrap();

        fx.server
            .tmux
            .kill_session(&tmux::session_name(a.id))
            .unwrap();
        fx.daemon.reconcile(at(10)).unwrap();
        let pass = fx.daemon.tick(at(11)).unwrap();

        assert_eq!(
            pass.started,
            [c.id],
            "without reconciliation the dead task would block the queue for ever"
        );
        assert_eq!(
            fx.daemon.store.get_task(b.id).unwrap().state,
            TaskState::Running
        );
    }

    #[test]
    fn scanning_records_repos_under_the_scan_root() {
        let mut fx = Fixture::new();
        fx.repo("api");
        fx.repo("web");
        let found = fx.daemon.scan(at(5)).unwrap();
        assert_eq!(found, 2);
        assert_eq!(fx.daemon.store.list_repos(false).unwrap().len(), 2);
    }

    #[test]
    fn a_hook_for_an_unknown_task_is_an_error_not_a_panic() {
        let mut fx = Fixture::new();
        let result = fx.daemon.handle(&delivery(9999, "Stop", json!({})), at(5));
        assert!(matches!(
            result,
            Err(Error::Hook(hook::Error::Store(
                crate::store::Error::TaskNotFound(9999)
            )))
        ));
    }

    #[test]
    fn a_task_that_cannot_launch_fails_and_notifies() {
        let mut fx = Fixture::new();
        // Selected but never a git repo, so provisioning fails.
        let broken_path = fx.repos_dir.join("broken");
        std::fs::create_dir_all(&broken_path).unwrap();
        let broken = fx
            .daemon
            .store
            .upsert_repo(&broken_path, "broken", at(0))
            .unwrap();
        let task = fx.queue("doomed", std::slice::from_ref(&broken));

        let pass = fx.daemon.tick(at(5)).unwrap();

        assert!(pass.started.is_empty());
        assert_eq!(pass.failed.len(), 1);
        assert_eq!(
            fx.daemon.store.get_task(task.id).unwrap().state,
            TaskState::Failed
        );
        assert_eq!(pass.notified[0].title, "marver — Task failed");
    }

    #[test]
    fn the_loop_delivers_a_hook_sent_over_the_real_socket() {
        use std::sync::atomic::AtomicBool;

        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let task = fx.queue("first", std::slice::from_ref(&repo));
        fx.daemon.tick(at(5)).unwrap();

        let socket = fx.daemon.socket().to_path_buf();
        let shutdown = Arc::new(AtomicBool::new(false));
        let stop = Arc::clone(&shutdown);

        // Send a Stop hook once the loop is up, then ask it to finish.
        std::thread::spawn(move || {
            for _ in 0..50 {
                if hook::send(&socket, &delivery(task.id, "Stop", json!({}))).is_ok() {
                    break;
                }
                std::thread::sleep(Duration::from_millis(20));
            }
            std::thread::sleep(Duration::from_millis(300));
            stop.store(true, Ordering::Relaxed);
        });

        fx.daemon.config.tick = Duration::from_millis(50);
        fx.daemon.run(shutdown).unwrap();

        assert_eq!(
            fx.daemon.store.get_task(task.id).unwrap().state,
            TaskState::AwaitingReview,
            "a hook sent down the real socket must reach the store"
        );
    }
}