marver 0.0.28

//! `marverd`: the process that outlives the TUI.

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::harness::Harness;
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.
    pub version_file: PathBuf,
    /// Where the running daemon records its process id, so it can be stopped.
    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,
    /// Which marver a tmux session belongs to. See [`tmux::session_name`].
    pub session_prefix: String,
    pub cap: usize,
    /// Which agent tasks are started with. See [`crate::harness`].
    pub harness: Harness,
    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")),
            session_prefix: tmux::session_prefix(data_dir),
            cap: crate::scheduler::DEFAULT_CAP,
            harness: Harness::claude(),
            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.
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 the hook listener waits out a recoverable `accept` failure, so a
/// persistent one — no descriptors left — costs a retry rather than a spin.
const ACCEPT_RETRY: Duration = Duration::from_millis(50);

/// 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.
    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.
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.
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.
fn record_version(config: &Config) {
    if let Some(parent) = config.version_file.parent() {
        let _ = std::fs::create_dir_all(parent);
    }
    // Best effort.
    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.
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.
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.
    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.
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.
        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, as a tmux client starts its
/// server.
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())
        // Passed on, or a `marver --harness codex` would open an interface and
        // start a daemon running Claude Code behind it.
        .arg("--harness")
        .arg(config.harness.spec())
        .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.
    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,
        )
        .harness(config.harness.clone())
        .session_prefix(config.session_prefix.clone());
        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
    }

    /// Take charge of a task that was still starting when the daemon stopped.
    ///
    /// `launch` provisions, opens a session, records its name, and only then
    /// does the scheduler write `running`. A daemon that dies inside that
    /// leaves a task saying `queued` with a live agent behind it — which the
    /// next launch meets as "session already exists" and answers by failing the
    /// task, killing the very agent that was working.
    ///
    /// Adopting it is both cheaper and safer than starting again: the session
    /// is there, and its hooks already carry this task's id.
    fn adopt_started(&mut self, now: DateTime<Utc>) -> Result<Vec<i64>> {
        let mut adopted = Vec::new();
        for task in self.store.list_tasks_in_state(TaskState::Queued)? {
            // A stat before a subprocess. A task that never reached a workspace
            // cannot have a session, and a long queue would otherwise cost one
            // `tmux has-session` per task per tick.
            if task.session_name.is_none() && !task.workspace_dir.exists() {
                continue;
            }
            // The name it would have been given, for the narrower window where
            // the session was opened but never written down.
            let session = task.session_name.clone().unwrap_or_else(|| {
                tmux::session_name(Some(&self.config.session_prefix), task.id)
            });
            if !self.tmux.session_exists(&session)? {
                continue;
            }
            // A tmux server is machine-wide, so a name is not proof of
            // ownership — the cwd is.
            if !crate::agent::owns_session(&self.tmux, &task, &session) {
                continue;
            }
            if task.session_name.is_none() {
                self.store.set_session_name(task.id, &session, now)?;
            }
            self.store.transition_from(
                task.id,
                TaskState::Queued,
                TaskState::Running,
                Transition::Plain,
                now,
            )?;
            adopted.push(task.id);
        }
        Ok(adopted)
    }

    /// Fail tasks whose agent is gone.
    ///
    /// Runs on every tick, not only at startup. `running` and `blocked` each
    /// hold a concurrency slot, and nothing else asks whether the agent behind
    /// one still exists — an agent that is killed, crashes, or is OOM-ed leaves
    /// its task claiming a slot for ever, and `cap` of those stop the queue for
    /// good while the daemon still reports itself healthy.
    pub fn reconcile(&mut self, now: DateTime<Utc>) -> Result<Vec<i64>> {
        self.reconcile_states(
            &[TaskState::Running, TaskState::Blocked, TaskState::Paused],
            now,
        )
    }

    /// The per-tick pass: only the states that hold a concurrency slot.
    ///
    /// `paused` is left to the startup sweep. It holds no slot, so a dead
    /// session behind one wedges nothing, and `pause::resume` asks about the
    /// session itself rather than trusting the recorded name — so the one thing
    /// that used to need this is covered where it happens. Asking every two
    /// seconds would spend a subprocess per paused task to learn nothing.
    fn supervise(&mut self, now: DateTime<Utc>) -> Result<Vec<i64>> {
        self.reconcile_states(crate::scheduler::OCCUPYING, now)
    }

    fn reconcile_states(&mut self, states: &[TaskState], now: DateTime<Utc>) -> Result<Vec<i64>> {
        let mut orphaned = Vec::new();
        for state in states.iter().copied() {
            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.
                if state == TaskState::Paused && task.session_name.is_none() {
                    continue;
                }
                // Running or blocked with nothing recorded: the launch never got
                // as far as writing the name down, so whatever it started is
                // unreachable.
                let Some(session) = task.session_name.clone() else {
                    self.store.transition(
                        task.id,
                        TaskState::Failed,
                        Transition::Failed(
                            "the task was working but no tmux session was ever recorded for it"
                                .to_string(),
                        ),
                        now,
                    )?;
                    orphaned.push(task.id);
                    continue;
                };
                // `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 is gone".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.
        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.
    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;
        };
        // A new path rewinds the offset in the same write, so the read below
        // starts at the beginning of the file it is actually about.
        let offset = if task.usage.transcript_path.as_deref() == Some(path.as_path()) {
            task.usage.transcript_offset
        } else {
            let _ = self.store.set_transcript_path(task.id, path);
            0
        };
        // 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, offset) {
            let _ = self.store.record_usage(task.id, &usage);
        }
    }

    /// Kill the agent of any task that has finished, returning the ids.
    pub fn reap(&mut self, now: DateTime<Utc>) -> Result<Vec<i64>> {
        let mut reaped = Vec::new();
        // Every terminal state, not a hand-written pair. `committed` is
        // terminal too, and its agent was left running for as long as the
        // machine was up — one `claude`, session and pty per task reviewed and
        // not archived. Worse than the leak: nothing supervises an agent behind
        // a task that can no longer say what it is doing, and the next cleanup
        // kills it mid-work and takes its worktrees.
        for state in TaskState::ALL.iter().copied().filter(|s| s.is_terminal()) {
            for task in self.store.list_tasks_in_state(state)? {
                let Some(session) = task.session_name.clone() else {
                    continue;
                };
                // Skipped rather than propagated: reaping is best effort and
                // runs again next tick, but a tmux that cannot be asked must
                // not read as a session already gone — the name would be
                // cleared and the agent left running with nothing tracking it.
                match self.tmux.session_exists(&session) {
                    Ok(true) => {}
                    Ok(false) => continue,
                    Err(err) => {
                        eprintln!("marverd: cannot reap task {}: {err}", task.id);
                        continue;
                    }
                }
                // Killing a session that is not this task's would take down
                // somebody else's agent, so the name is forgotten instead —
                // there is nothing here left to reap, and keeping it would ask
                // tmux the same question on every pass for ever.
                if !crate::agent::owns_session(&self.tmux, &task, &session) {
                    self.store.clear_session_name(task.id, now)?;
                    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> {
        // First: a task can only be believed to hold a slot for as long as the
        // agent behind it exists. Nothing but this asks, and a slot claimed by
        // a dead agent is never given back.
        //
        // An error here — tmux unreachable — abandons the whole pass rather
        // than scheduling on top of an unknown, which is the same judgement
        // `reconcile` makes internally about failing every live task at once.
        let orphaned = self.supervise(now)?;
        for id in &orphaned {
            eprintln!("marverd: task {id}: the agent is gone; failing it and freeing its slot");
        }

        // Before planning, so a task whose agent is already live is counted
        // against the cap rather than started a second time.
        for id in self.adopt_started(now)? {
            eprintln!("marverd: task {id}: its agent was already running; adopted it");
        }

        // Then: 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.
        for (id, state) in &abandoned {
            eprintln!("marverd: task {id} became {state} while it was starting; its agent is live");
        }

        // Orphans are announced with the rest: a task that failed because its
        // agent died is exactly the case the notifier exists for, and it is the
        // one nobody is sitting in front of when it happens.
        let mut notified = Vec::new();
        for id in orphaned.iter().chain(failed.iter().map(|(id, _)| id)) {
            if let Ok(task) = self.store.get_task(*id)
                && let Some(sent) = self.announce(&task)
            {
                notified.push(sent);
            }
        }
        Ok(Pass {
            started,
            failed,
            notified,
            reaped,
        })
    }

    /// A pass whose failure is said out loud rather than discarded.
    ///
    /// The loop cannot stop for a bad tick — the next one may well work, and a
    /// daemon that exits takes every agent's supervisor with it. But `let _ =`
    /// meant a store that had started refusing writes, or a tmux that could not
    /// be run, produced a daemon that quietly stopped moving tasks and a log
    /// with nothing in it — and the log is the only evidence anyone has.
    fn tick_reporting(&mut self, now: DateTime<Utc>) {
        if let Err(err) = self.tick(now) {
            eprintln!("marverd: this pass did nothing: {err}");
        }
    }

    /// Deliver a notification, swallowing backend failures.
    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.
    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.
                    Err(hook::Error::Probe) => {}
                    Err(err @ (hook::Error::Malformed(_) | hook::Error::Rejected(_))) => {
                        eprintln!("marverd: ignoring hook: {err}");
                    }
                    // `accept` failing is not the listener dying. Running out
                    // of descriptors, a signal, and a client that goes between
                    // the connection and the read all arrive here, and taking
                    // the thread down for one of them made the daemon deaf for
                    // the rest of its life — silently, since `marver hook`
                    // exits 0 by design and `marver status` only looks at the
                    // socket. Every live task then froze holding its slot.
                    Err(err @ hook::Error::Io { .. }) => {
                        eprintln!("marverd: hook listener: {err}; retrying");
                        std::thread::sleep(ACCEPT_RETRY);
                    }
                    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);
                    }
                    self.tick_reporting(now);
                }
                Err(RecvTimeoutError::Timeout) => {
                    self.tick_reporting(Utc::now());
                }
                // The listener thread gave up on something it could not retry.
                // Exiting is loud, and a supervisor or the next `marver` will
                // start a daemon that can hear; carrying on would keep the
                // socket bound while delivering nothing through it.
                Err(RecvTimeoutError::Disconnected) => {
                    eprintln!("marverd: the hook listener is gone; stopping");
                    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"),
            )
            .harness(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(None, a.id)));
    }

    #[test]
    fn an_agent_that_dies_gives_its_slot_back_on_the_next_tick() {
        // `running` holds a slot and nothing but reconcile asks whether the
        // agent behind one still exists. While that ran once per daemon
        // lifetime, `cap` dead agents stopped the queue for good — and the
        // daemon went on reporting itself healthy.
        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], "the cap is two");
        assert_eq!(
            fx.daemon.store.get_task(c.id).unwrap().state,
            TaskState::Queued
        );

        // The agent dies the way agents do: OOM, a crash, someone else's
        // `tmux kill-session`. No hook can report it — the process that would
        // have sent one is the process that died.
        fx.server
            .tmux
            .kill_session(&tmux::session_name(None, a.id))
            .unwrap();

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

        let failed = fx.daemon.store.get_task(a.id).unwrap();
        assert_eq!(failed.state, TaskState::Failed);
        assert!(
            failed.failure_reason.is_some(),
            "a failed task must say why"
        );
        assert_eq!(pass.started, [c.id], "the freed slot goes to the queue");
    }

    #[test]
    fn a_task_whose_agent_is_already_live_is_adopted_rather_than_started_again() {
        // The launch crash window: `launch` records the session name and the
        // scheduler writes `running` after it. A daemon that died in between
        // left a `queued` task with a working agent — and the next launch met
        // its own session, failed the task, and reaped the agent that was in
        // the middle of the work.
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let task = fx.queue("half-started", std::slice::from_ref(&repo));

        // Exactly as far as the daemon got before it would have said `running`.
        let launched = fx
            .daemon
            .launcher
            .launch(&mut fx.daemon.store, &task, at(1))
            .unwrap();
        assert_eq!(
            fx.daemon.store.get_task(task.id).unwrap().state,
            TaskState::Queued,
            "the state a crash would have left behind"
        );

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

        let adopted = fx.daemon.store.get_task(task.id).unwrap();
        assert_eq!(adopted.state, TaskState::Running);
        assert_eq!(
            adopted.session_name.as_deref(),
            Some(launched.session.as_str())
        );
        assert!(pass.failed.is_empty(), "adopting is not failing");
        assert!(
            pass.started.is_empty(),
            "the agent was already there; it must not be started twice"
        );
        assert!(
            fx.server.tmux.has_session(&launched.session),
            "and it must be left alone"
        );
    }

    #[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.
        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(None, 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.
        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_spares_a_session_that_only_shares_the_name() {
        // The launcher already refuses to kill a namesake on shut down; reap
        // reaches the same sessions from the daemon and must agree.
        let mut fx = Fixture::new();
        let repo = fx.repo("api");
        let task = fx.queue("never launched", std::slice::from_ref(&repo));
        let session = tmux::session_name(None, task.id);

        // Somebody else's session, sitting outside this task's workspace, whose
        // name this task has somehow come to hold.
        let elsewhere = fx.repos_dir.clone();
        fx.server
            .tmux
            .new_session(&session, &elsewhere, tmux::DEFAULT_SIZE)
            .unwrap();
        fx.daemon
            .store
            .set_session_name(task.id, &session, at(5))
            .unwrap();
        fx.daemon
            .store
            .transition(task.id, TaskState::Cancelled, Transition::Plain, at(6))
            .unwrap();

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

        assert!(pass.reaped.is_empty(), "{:?}", pass.reaped);
        assert!(
            fx.server.tmux.has_session(&session),
            "a session marver did not open must survive"
        );
        assert_eq!(
            fx.daemon.store.get_task(task.id).unwrap().session_name,
            None,
            "and the name is forgotten, or every tick asks tmux about it again"
        );
    }

    #[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(None, 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.
        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.
        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(None, 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(None, 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.
        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.
        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
        );
    }
}