marver 0.0.19

//! `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 },
    #[error("could not stop the daemon: {0}")]
    Stop(String),
    #[error("the daemon was signalled but still held the socket after {0:?}")]
    StillRunning(Duration),
}

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,
    /// Where the running daemon records which version it is.
    ///
    /// A file rather than a row, because the point is to be readable without
    /// opening the database — and a schema change to answer "what is already
    /// running" would be the wrong shape of answer anyway.
    pub version_file: PathBuf,
    /// Where the running daemon records its process id, so it can be stopped.
    ///
    /// Deliberately not used to answer "is one running" — [`is_running`]
    /// connects for that, because a pid stops meaning anything the moment the
    /// number is reused. Read only alongside a live socket, and checked against
    /// the process's own command line before anything is signalled.
    pub pid_file: 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"),
            version_file: data_dir.join("daemon.version"),
            pid_file: data_dir.join("daemon.pid"),
            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);

/// How long to wait for a signalled daemon to release the socket.
const STOP_TIMEOUT: Duration = Duration::from_secs(5);

/// How long to wait for a daemon that is already listening to record which
/// version it is. Short: the socket is bound, so this is a write away.
pub const ANNOUNCE_TIMEOUT: Duration = Duration::from_secs(2);

/// The version of this build, and of whatever daemon is already running.
pub const VERSION: &str = env!("CARGO_PKG_VERSION");

/// What [`ensure_running`] found, or did about it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Startup {
    /// One was already listening, running this same version.
    AlreadyRunning,
    /// One was already listening, but it is not this version.
    ///
    /// Not restarted automatically. The running daemon is supervising live
    /// agents, and killing it to pick up a newer build would fail every task it
    /// was watching — a worse outcome than being a version behind. Saying so and
    /// letting the user choose the moment is the honest option.
    Outdated { running: String },
    /// 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)
}

/// Which version the running daemon is, if it left a note and is still up.
///
/// Only meaningful alongside [`is_running`]: the file outlives the process that
/// wrote it, exactly as the socket file does.
pub fn running_version(config: &Config) -> Option<String> {
    let recorded = std::fs::read_to_string(&config.version_file).ok()?;
    let recorded = recorded.trim();
    (!recorded.is_empty()).then(|| recorded.to_string())
}

/// The version recorded by the daemon with pid `pid`, once it has said so.
///
/// [`ensure_running`] returns the moment the socket answers, and the version is
/// recorded just *after* that — deliberately, since writing it before the bind
/// would let a daemon that loses the race overwrite the record of the one that
/// won. So a caller reading the file the instant it is told a daemon started can
/// still find the *previous* daemon's version, which is precisely the answer it
/// must not give during an upgrade.
///
/// [`record_version`] writes the version before the pid, so a pid file naming
/// `pid` proves the version beside it belongs to that daemon too.
///
/// `None` if it says nothing within [`ANNOUNCE_TIMEOUT`]. Callers should then
/// report no version at all rather than fall back to their own constant: under
/// `marver upgrade --restart` the asking process is the binary that was just
/// replaced, so its constant names the version being retired.
pub fn announced_version(config: &Config, pid: u32) -> Option<String> {
    let deadline = Instant::now() + ANNOUNCE_TIMEOUT;
    loop {
        if running_pid(config) == Some(pid) {
            return running_version(config);
        }
        if Instant::now() >= deadline {
            return None;
        }
        std::thread::sleep(STARTUP_POLL);
    }
}

/// Record which version is now serving this data directory.
///
/// Upgrading replaces the binary on disk while the running daemon keeps
/// executing the old inode — and since it still holds the socket,
/// [`ensure_running`] finds it and declines to start a replacement. Without a
/// note somewhere, that is invisible: everything works, on last week's code.
///
/// The version goes down before the pid, and [`announced_version`] depends on
/// that order to know whose version it is reading.
fn record_version(config: &Config) {
    if let Some(parent) = config.version_file.parent() {
        let _ = std::fs::create_dir_all(parent);
    }
    // Best effort. A daemon that cannot write these should still run.
    let _ = std::fs::write(&config.version_file, format!("{VERSION}\n"));
    let _ = std::fs::write(&config.pid_file, format!("{}\n", std::process::id()));
}

/// The pid of the running daemon, if it left one and something is listening.
pub fn running_pid(config: &Config) -> Option<u32> {
    if !is_running(config) {
        return None;
    }
    std::fs::read_to_string(&config.pid_file)
        .ok()?
        .trim()
        .parse()
        .ok()
}

/// Whether `pid` is a marver daemon, rather than whatever reused the number.
///
/// A pid file is a promise about the past. Checking the command line before
/// signalling is what keeps a stale one from killing an innocent process.
fn is_marver_daemon(pid: u32) -> bool {
    let Ok(out) = std::process::Command::new("ps")
        .args(["-p", &pid.to_string(), "-o", "command="])
        .output()
    else {
        return false;
    };
    String::from_utf8_lossy(&out.stdout).contains("marver daemon")
}

/// Find a running daemon in the process table, for when there is no pid file.
///
/// Deliberately refuses to guess between several. The command line of a daemon
/// started without `--data-dir` does not say which one it serves, so picking one
/// of two would be picking at random — and the thing being picked gets killed.
fn search_for_daemon(config: &Config) -> Result<u32> {
    // An auto-started daemon is always given `--data-dir`, so this identifies
    // exactly the one serving this configuration even when several are up.
    let scoped = format!("marver daemon --data-dir {}", config.data_dir.display());
    if let [pid] = matching(&scoped)?.as_slice() {
        return Ok(*pid);
    }

    // Started by hand as a bare `marver daemon`, its command line says nothing
    // about which data directory it serves. One is unambiguous; more is a guess,
    // and the thing being guessed at gets killed.
    match matching("marver daemon")?.as_slice() {
        [pid] => Ok(*pid),
        [] => Err(Error::Stop(
            "a daemon is listening but no process matches it; stop it by hand".to_string(),
        )),
        many => Err(Error::Stop(format!(
            "{} marver daemons are running and none of them left a pid file; \
             stop the right one by hand",
            many.len()
        ))),
    }
}

/// Process ids whose command line matches `pattern`, excluding this process.
fn matching(pattern: &str) -> Result<Vec<u32>> {
    let out = std::process::Command::new("pgrep")
        .args(["-f", pattern])
        .output()
        .map_err(|err| Error::Stop(format!("could not search for the daemon: {err}")))?;
    Ok(String::from_utf8_lossy(&out.stdout)
        .lines()
        .filter_map(|line| line.trim().parse().ok())
        // `pgrep -f` matches whoever is asking, too.
        .filter(|&pid| pid != std::process::id())
        .collect())
}

/// Stop the running daemon and wait for it to let go of the socket.
///
/// Returns whether there was one to stop. `SIGTERM`, not `SIGKILL`: the default
/// disposition ends the process, and the socket file goes with it either way,
/// but a signal it could one day handle is the better thing to have sent.
pub fn stop(config: &Config) -> Result<bool> {
    if !is_running(config) {
        return Ok(false);
    }
    let pid = match running_pid(config) {
        Some(pid) => pid,
        // A daemon is listening but left no pid file: it either predates them or
        // lost it. Falling back to the process table is what stops "I cannot
        // find it" being reported as "there was nothing to find".
        None => search_for_daemon(config)?,
    };
    if !is_marver_daemon(pid) {
        return Err(Error::Stop(format!(
            "pid {pid} is not a marver daemon; refusing to signal it"
        )));
    }

    let killed = std::process::Command::new("kill")
        .arg(pid.to_string())
        .status()
        .map_err(|err| Error::Stop(err.to_string()))?;
    if !killed.success() {
        return Err(Error::Stop(format!("could not signal pid {pid}")));
    }

    let deadline = Instant::now() + STOP_TIMEOUT;
    while Instant::now() < deadline {
        if !is_running(config) {
            return Ok(true);
        }
        std::thread::sleep(STARTUP_POLL);
    }
    Err(Error::StillRunning(STOP_TIMEOUT))
}

/// 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(match running_version(config) {
            Some(running) if running != VERSION => Startup::Outdated { running },
            _ => 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, TaskState::Paused] {
            for task in self.store.list_tasks_in_state(state)? {
                // A paused task may never have launched, in which case there is
                // no session to have died and nothing to reconcile. Deriving a
                // name for it the way the others do would invent a session,
                // find it missing, and fail a task that is merely waiting its
                // turn.
                if state == TaskState::Paused && task.session_name.is_none() {
                    continue;
                }
                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)?;
        // After the transition, and never allowed to affect it. This reads a
        // file format Claude Code makes no promises about, so it is the one part
        // of handling a hook that must not be able to fail the rest.
        self.absorb_usage(delivery);
        match outcome {
            hook::Outcome::Moved { .. } => {
                let task = self.store.get_task(delivery.task_id)?;
                Ok(self.announce(&task))
            }
            hook::Outcome::Recorded { .. } => Ok(None),
        }
    }

    /// Read whatever the transcript has added since the last hook.
    ///
    /// Best effort throughout. A transcript that is missing, unreadable, or no
    /// longer shaped the way this expects costs the task its token figures and
    /// nothing else — the lifecycle is driven by hooks, which carry none of
    /// this.
    fn absorb_usage(&mut self, delivery: &Delivery) {
        let Some(path) = delivery.payload.transcript_path.as_ref() else {
            return;
        };
        let Ok(task) = self.store.get_task(delivery.task_id) else {
            return;
        };
        if task.usage.transcript_path.as_deref() != Some(path.as_path()) {
            let _ = self.store.set_transcript_path(task.id, path);
        }
        // A failure is dropped rather than logged: a hook fires several times a
        // turn, and a transcript that is absent for one task would fill the log
        // with the same line for ever.
        if let Ok(usage) = crate::usage::read_from(path, task.usage.transcript_offset) {
            let _ = self.store.record_usage(task.id, &usage);
        }
    }

    /// 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,
            abandoned,
            ..
        } = self.scheduler.tick(&mut self.store, &self.launcher, now)?;

        // Said out loud, because it is the one outcome with a live agent behind
        // a task that does not say `running` — usually a task paused in the
        // second it took to start. Rare enough to be worth a line each time.
        for (id, state) in &abandoned {
            eprintln!("marverd: task {id} became {state} while it was starting; its agent is live");
        }

        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<()> {
        // The socket is already bound by now, so anything that finds a daemon
        // listening can also find out which one it found.
        record_version(&self.config);

        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_a_task_that_was_paused_before_it_launched() {
        // A held task owns no session, so there is none to have died. Deriving
        // a name for it the way running tasks are checked would invent one,
        // find it missing, and fail a task that is only waiting its turn --
        // permanently, since `failed` is terminal.
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let task = fx.queue("not yet", std::slice::from_ref(&repo));
        fx.daemon
            .store
            .transition(task.id, TaskState::Paused, Transition::Plain, at(5))
            .unwrap();

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

        assert!(orphaned.is_empty(), "{orphaned:?}");
        assert_eq!(
            fx.daemon.store.get_task(task.id).unwrap().state,
            TaskState::Paused
        );
    }

    #[test]
    fn reconcile_fails_a_paused_task_whose_session_died() {
        // The other half: a task paused mid-work does have a session, and if
        // the machine restarted underneath it there is nothing to resume into.
        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();
        fx.daemon
            .store
            .transition(task.id, TaskState::Paused, Transition::Plain, at(6))
            .unwrap();
        fx.server
            .tmux
            .kill_session(&tmux::session_name(task.id))
            .unwrap();

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

        assert_eq!(orphaned, [task.id]);
        assert_eq!(
            fx.daemon.store.get_task(task.id).unwrap().state,
            TaskState::Failed
        );
    }

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

    use std::io::Write;

    /// A transcript in the shape Claude Code writes.
    fn transcript(dir: &std::path::Path, name: &str, turns: &[(u64, u64)]) -> std::path::PathBuf {
        let path = dir.join(name);
        let mut file = std::fs::File::create(&path).unwrap();
        for (context, output) in turns {
            writeln!(
                file,
                r#"{{"type":"assistant","message":{{"model":"claude-opus-5","usage":{{"input_tokens":{context},"cache_read_input_tokens":0,"cache_creation_input_tokens":0,"output_tokens":{output}}}}}}}"#
            )
            .unwrap();
        }
        path
    }

    #[test]
    fn a_hook_brings_the_agents_usage_with_it() {
        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();

        let dir = tempfile::TempDir::new().unwrap();
        let path = transcript(dir.path(), "s.jsonl", &[(1_000, 200)]);
        fx.daemon
            .handle(
                &delivery(
                    task.id,
                    "Notification",
                    json!({
                        "transcript_path": path.to_string_lossy(),
                        "notification_type": "idle_prompt",
                    }),
                ),
                at(6),
            )
            .unwrap();

        let usage = fx.daemon.store.get_task(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_path, Some(path));
    }

    #[test]
    fn a_later_hook_reads_only_what_is_new() {
        // A transcript reaches megabytes and hooks arrive several times a turn.
        // Re-reading it whole would double the totals and make watching a task
        // cost more the longer it runs.
        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();
        let dir = tempfile::TempDir::new().unwrap();
        let path = transcript(dir.path(), "s.jsonl", &[(1_000, 200)]);
        let hook = |p: &std::path::Path| {
            delivery(
                task.id,
                "Notification",
                json!({
                    "transcript_path": p.to_string_lossy(),
                    "notification_type": "idle_prompt",
                }),
            )
        };
        fx.daemon.handle(&hook(&path), at(6)).unwrap();

        // The same file, one turn longer.
        let path = transcript(dir.path(), "s.jsonl", &[(1_000, 200), (4_000, 50)]);
        fx.daemon.handle(&hook(&path), at(7)).unwrap();

        let usage = fx.daemon.store.get_task(task.id).unwrap().usage;
        assert_eq!(usage.output_tokens, Some(250), "counted once each");
        assert_eq!(usage.context_tokens, Some(4_000), "the latest turn's level");
    }

    #[test]
    fn a_transcript_that_is_not_there_costs_the_hook_nothing() {
        // This reads a file format Claude Code makes no promises about. It must
        // never be able to break the transition the hook actually carries.
        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();

        fx.daemon
            .handle(
                &delivery(
                    task.id,
                    "Stop",
                    json!({ "transcript_path": "/no/such/transcript.jsonl" }),
                ),
                at(6),
            )
            .unwrap();

        let task = fx.daemon.store.get_task(task.id).unwrap();
        assert_eq!(
            task.state,
            TaskState::AwaitingReview,
            "the hook still did its job"
        );
        assert!(!task.usage.is_known());
    }

    #[test]
    fn a_hook_with_no_transcript_path_is_not_a_problem() {
        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();

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

        assert_eq!(
            fx.daemon.store.get_task(task.id).unwrap().state,
            TaskState::AwaitingReview
        );
    }
}