semlith 0.35.0

//! `semlith start` — one process that owns every store.
//!
//! Before this, a developer could have a watcher keeping a store current or an
//! agent holding it open, but not both: a store has one writer, so the second
//! one was refused. The daemon removes that structurally rather than by
//! relaxing the rule. It *is* the writer, for its whole life, and everything
//! else — the portal, a forwarded `semlith_index` — is a client of it.
//!
//! Three things run at once:
//!
//! - **A watcher thread per store**, holding that store's write lock and its
//!   own [`Semlith`], re-embedding files as they are saved. It is the same
//!   loop `semlith watch` runs.
//! - **A reader [`Fleet`]** over the same stores, which every read route
//!   answers from. Opening a store twice in one process is what the store
//!   format already supports — a reader notices the writer's work through the
//!   index generation and reloads — so reads never wait behind an embed.
//! - **The HTTP server**, on 127.0.0.1 and nowhere else.
//!
//! Work that writes does not go to the reader. It goes on the store's queue,
//! and the watcher thread — the lock holder — performs it between batches.
//! That is why a portal index and a file save can never interleave.

use crate::fleet::Fleet;
use crate::home::{self, Registry};
use crate::http::{self, Refusal, Server};
use crate::lock::StoreLock;
use crate::{Semlith, watch};
use anyhow::{Context, Result, bail};
use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, VecDeque};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, RwLock, mpsc};
use std::time::{Duration, SystemTime};

/// Written beside the store's lock while a daemon holds it, so `semlith mcp`
/// can find the daemon and forward to it instead of fighting for the lock.
///
/// Beside the lock rather than in the home, because the question it answers is
/// "who is writing *this* store", which is exactly what the lock file is about.
pub const DISCOVERY_FILE: &str = "daemon.json";

/// How many watcher events the Stores view can show. A ring, not a log: this
/// is the "is it actually running" signal, and the last twenty answer it.
const EVENT_HISTORY: usize = 20;

/// How long one queued index slice may work before it yields the writer back
/// to the watcher. The same budget `semlith_index` has always had.
const SLICE: Duration = Duration::from_secs(45);

/// How many of a run's log lines the daemon keeps for a page to catch up on.
///
/// A ring rather than a log, for the same reason [`EVENT_HISTORY`] is one: the
/// question a returning tab asks is "what has happened since sequence N", and
/// five hundred lines is more than any tab that was away for a poll or two has
/// missed. A tab away for longer sees the gap in the sequence numbers rather
/// than a quietly shortened history.
const LOG_HISTORY: usize = 500;

/// How many of a store's runs are kept.
///
/// Enough that a session's cards are all there; bounded so a daemon left up
/// for a week does not accumulate them without limit. Only finished runs are
/// ever dropped.
const RUN_HISTORY: usize = 24;

/// The run id a job that is not an index run carries.
///
/// Ids are minted from 1, so 0 names nothing and folding an event into it is a
/// no-op — which is what a forget or a fetch wants.
const NO_RUN: u64 = 0;

/// How long a freshly made store's watcher waits for the run that is about to
/// index it before deciding none is coming.
const EXPECTED_RUN_GRACE_SECS: usize = 60;

/// What `semlith mcp` reads to find a running daemon.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Discovery {
    pub pid: u32,
    pub port: u16,
    /// The daemon's current token. A rotate rewrites this file, so a forwarding
    /// `semlith mcp` picks the new one up on its next call.
    pub token: String,
    pub version: String,
    /// When this daemon's `run` began, as a unix second. The file's mtime was
    /// the answer until 0.33.0, and every store open and key rotation rewrites
    /// the file, so `doctor` said a three-day-old daemon had started that
    /// evening. Zero in a file an older daemon wrote.
    #[serde(default)]
    pub started: u64,
}

impl Discovery {
    /// The daemon to forward to, if this file names one and is one semlith
    /// wrote.
    ///
    /// This file decides where `semlith mcp` sends every call, so a repository
    /// that carried one would be choosing the port an agent's questions and
    /// answers travel through. Four things have to hold, and each of them is a
    /// way the file could have come from somewhere else:
    ///
    /// - the store it sits in is one this user trusts, so a cloned `.semlith`
    ///   never gets this far;
    /// - on unix it is mode `0600` and owned by this user, so another account
    ///   on the machine did not write it;
    /// - the pid it names is alive, so a stale file from a dead daemon is
    ///   ignored rather than followed to whatever now holds that port;
    /// - the token is 64 hex characters, so what is about to be put in a header
    ///   is a token rather than whatever was in the file.
    ///
    /// Every failure falls back to opening the store directly, which is what
    /// this function's `None` has always meant.
    pub fn read(store_dir: &Path) -> Option<Self> {
        if !home::Registry::load().ok()?.trusts(store_dir) {
            return None;
        }
        let path = store_dir.join(DISCOVERY_FILE);
        if worth_warning_about(&path) {
            eprintln!(
                "semlith: ignoring {} — it is not a file this user wrote privately",
                path.display()
            );
            return None;
        }
        if !owner_only(&path) {
            return None;
        }
        let text = std::fs::read_to_string(&path).ok()?;
        let found: Self = serde_json::from_str(&text).ok()?;
        if found.token.len() != 64 || !found.token.bytes().all(|b| b.is_ascii_hexdigit()) {
            return None;
        }
        if !alive(found.pid) {
            return None;
        }
        Some(found)
    }

    fn write(&self, store_dir: &Path) -> Result<()> {
        let path = store_dir.join(DISCOVERY_FILE);
        // The mode is set as the file is created rather than afterwards. The
        // session token is in this file, and a file that is world-readable for
        // the microsecond between the two is a file that was world-readable —
        // the same reason `home::write_agent_key` opens with a mode.
        let mut options = std::fs::OpenOptions::new();
        options.write(true).create(true).truncate(true);
        #[cfg(unix)]
        {
            use std::os::unix::fs::OpenOptionsExt;
            options.mode(home::FILE_MODE);
        }
        let mut file = options
            .open(&path)
            .with_context(|| format!("writing {}", path.display()))?;
        use std::io::Write;
        file.write_all((serde_json::to_string_pretty(self)? + "\n").as_bytes())
            .with_context(|| format!("writing {}", path.display()))?;
        // An existing file keeps its own mode through `truncate`, so one left
        // loose by an older semlith is narrowed here too.
        home::tighten_file(&path);
        Ok(())
    }

    fn remove(store_dir: &Path) {
        let _ = std::fs::remove_file(store_dir.join(DISCOVERY_FILE));
    }
}

/// Whether a discovery file is one worth complaining about.
///
/// `owner_only` answers `false` for a file that is not there, because
/// `metadata` fails and there is no mode to read. Routing absence through it
/// meant a store with no daemon running — the ordinary case, and the case on
/// every one of six stores on this machine — was reported as a privacy
/// rejection. Every `semlith mcp` start printed six lines of "it is not a file
/// this user wrote privately" naming six paths that answer `No such file or
/// directory`, and those six lines were the first thing in a client's log for
/// anyone trying to work out why no server had appeared. A missing file is
/// missing; only a file that exists and fails the check is worth a line.
fn worth_warning_about(path: &Path) -> bool {
    path.exists() && !owner_only(path)
}

/// Whether a file is one this user wrote and nobody else can read.
///
/// On Windows there is no mode to read: the home sits inside the user's
/// profile, whose default ACL grants that user alone, and semlith carries no
/// API for reading an ACL. Stated rather than silently assumed, as
/// `home::check_key_mode` already does for the agent key.
#[cfg(unix)]
fn owner_only(path: &Path) -> bool {
    use std::os::unix::fs::MetadataExt;
    use std::os::unix::fs::PermissionsExt;
    let Ok(meta) = std::fs::metadata(path) else {
        return false;
    };
    // SAFETY: `getuid` reads this process's own id and cannot fail.
    let me = unsafe { libc::getuid() };
    meta.uid() == me && meta.permissions().mode() & 0o177 == 0
}

#[cfg(not(unix))]
fn owner_only(path: &Path) -> bool {
    path.exists()
}

/// Whether a process id names something running.
///
/// `kill(pid, 0)` asks the kernel without sending anything. A pid this user
/// does not own answers `EPERM`, which still means it is alive — and a daemon
/// belonging to somebody else is one this file should not have named, which
/// `owner_only` has already decided.
#[cfg(unix)]
pub(crate) fn alive(pid: u32) -> bool {
    // SAFETY: signal 0 delivers nothing; this is the documented liveness probe.
    unsafe {
        libc::kill(pid as i32, 0) == 0
            || std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM)
    }
}

#[cfg(not(unix))]
pub(crate) fn alive(_pid: u32) -> bool {
    true
}

/// Something the watcher thread should do next, on behalf of a request.
/// What an index job has left to do.
///
/// A slice yields the writer back to the watcher when its budget runs out and
/// the rest is re-queued behind whatever the watcher had waiting, so one
/// logical run is several `Index` jobs on one report channel. What a
/// continuation carries is the *remainder of the walk*, not the roots: walking
/// the roots again meant re-opening and re-hashing every file the run had
/// already done, once per slice, to be told each time that it was unchanged.
enum Work {
    /// The first slice of a run: the roots to walk.
    Roots(Vec<PathBuf>),
    /// Every later slice: exactly the files the previous one did not reach.
    Rest(Vec<PathBuf>),
}

/// What a run has done so far, across every slice of it.
///
/// Carried between slices because each slice's report is its own: the `done`
/// summary used to be whatever the last slice happened to do, so a run of 35
/// files that took three slices ended by announcing the four the third slice
/// reached. Every one of these is the run's, which is what the summary claims
/// to be.
#[derive(Default, Clone)]
struct Tally {
    indexed: u64,
    unchanged: u64,
    skipped: u64,
    removed: u64,
    chunks: u64,
    images: u64,
    /// Symbols extracted, carried for the same reason the chunks are: a
    /// slice's own count starts at zero and the page is drawing one run.
    symbols: u64,
    /// Named per file, and the skips by reason: each slice's report holds
    /// only its own, so a forwarded `semlith_index` listed one refusal of
    /// three and reasons that did not add up to the total.
    refused: Vec<(String, String)>,
    failed: Vec<(String, String)>,
    skipped_reasons: std::collections::BTreeMap<String, usize>,
    /// Where the run's time went, slice by slice added up.
    stages: Box<crate::pipeline::Stages>,
    /// Chunks the vector cache was asked for, and held.
    cache_lookups: u64,
    cache_hits: u64,
    /// Chunk rows the run wrote, embedded or not yet.
    rows: u64,
}

impl Tally {
    fn add(&mut self, report: &crate::IndexReport) {
        self.stages.add(&report.stages);
        self.cache_lookups += report.cache_lookups as u64;
        self.cache_hits += report.cache_hits as u64;
        self.rows += report.rows as u64;
        self.refused.extend(report.refused.iter().cloned());
        self.failed.extend(report.failed.iter().cloned());
        for (kind, n) in &report.skipped_reasons {
            *self.skipped_reasons.entry(kind.clone()).or_insert(0) += n;
        }
        self.indexed += report.indexed as u64;
        self.unchanged += report.unchanged as u64;
        self.skipped += report.skipped as u64;
        self.removed += report.removed as u64;
        self.chunks += report.chunks as u64;
        self.images += report.images as u64;
        self.symbols += report.symbols as u64;
    }
}

struct Indexing {
    work: Work,
    /// The first slice of this run, which announces it and records what the
    /// store held before it.
    first: bool,
    /// Every file this run has embedded, across every slice, so a stop undoes
    /// the run rather than the slice that happened to be going.
    already: Vec<String>,
    /// What the run had scanned before this slice, and how many files the walk
    /// found in total. Carried so the progress a page draws belongs to the run:
    /// a slice's own counters restart at one, which a page faithfully redrew as
    /// a run that had begun again.
    scanned: u64,
    total: u64,
    /// The same pair in bytes, which the remaining-time estimate is taken from.
    bytes: u64,
    bytes_total: u64,
    /// What the earlier slices of this run got through.
    tally: Tally,
}

enum Job {
    Index(Indexing),
    Forget(PathBuf),
    /// A person's decision about one refused file (2.5): `mode` is
    /// `redacted`, `as-is` or `refused`, and anything but the last indexes
    /// the file straight after.
    Accept {
        path: PathBuf,
        mode: String,
        source: String,
    },
    Revoke(PathBuf),
    Compact(crate::compact::CompactOptions),
}

/// A job and the channel its progress goes back on.
struct Queued {
    /// Which run this job belongs to.
    ///
    /// A store used to hold one run, so the events a job emitted could be
    /// folded into "the run" without saying which. It holds several now — a
    /// second run against the same store is its own card with its own log —
    /// and an event that does not name its run is an event the wrong card
    /// shows.
    run: u64,
    job: Job,
    report: mpsc::Sender<serde_json::Value>,
}

/// One watcher event, for the Stores view's feed.
#[derive(Debug, Clone, Serialize)]
pub struct Event {
    /// Unix seconds.
    pub at: u64,
    pub text: String,
}

/// Where an index run has got to.
///
/// `stopping` is told apart from `stopped` because undoing what a run embedded
/// takes as long as the embedding did on a large corpus, and a card that jumps
/// straight to "stopped" would be claiming the store was already back to what
/// it was.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum RunStatus {
    Queued,
    /// Scanned, and waiting for a person to review what the scan refused
    /// before it is queued at all (2.7). Holds nothing: not a slot, not the
    /// writer.
    Review,
    Running,
    /// Asked to pause; the engine has not reached its next batch yet.
    Pausing,
    Paused,
    /// Admitted once, and held again because `runs at once` was lowered below
    /// the number running. It resumes, in submission order, as slots free.
    Held,
    Stopping,
    Done,
    Stopped,
    Failed,
}

impl RunStatus {
    /// Whether this run is over. An admission is what a finished run releases,
    /// so this is the predicate the queue moves on.
    pub fn finished(self) -> bool {
        matches!(self, Self::Done | Self::Stopped | Self::Failed)
    }
}

/// What started a run.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum RunKind {
    /// Asked for, from the portal, the CLI's daemon path or an agent.
    Run,
    /// The watcher catching a store up with what changed while nothing was
    /// watching. Admitted like a run since 0.28.0: it used to bypass the
    /// limit, so "runs at once" bounded nothing on a daemon's first minute.
    CatchUp,
    /// A burst of filesystem events too large to run straight away.
    Batch,
    /// Giving a store's dead bytes back (`semlith compact`, the portal's
    /// Compact action, or the daemon on its own past the threshold). A card
    /// like any run, so it can be seen and stopped; never admitted, because
    /// it embeds nothing and holds only its own store's writer.
    Compact,
    /// The store's Re-index button over all its roots (0.35.0). Unchanged
    /// files are still skipped by hash, as the page says.
    Reindex,
    /// Re-index with `force`: every file re-embedded, hash or no hash.
    Rebuild,
    /// Exactly the files a person picked, re-embedded whatever their hash.
    Files,
}

impl RunKind {
    /// Whether this run re-embeds a file whose hash has not moved.
    fn forced(self) -> bool {
        matches!(self, Self::Rebuild | Self::Files)
    }

    /// The run's name in the history list.
    fn label(self, paths: usize) -> String {
        match self {
            Self::Run => "Index".to_string(),
            Self::Reindex | Self::Rebuild => "Re-index".to_string(),
            Self::Files => format!("Re-index {paths} files"),
            Self::CatchUp | Self::Batch => "Watcher catch-up".to_string(),
            Self::Compact => "Compact".to_string(),
        }
    }
}

/// How far back the rate a run card shows looks.
const RATE_WINDOW_MS: u64 = 10_000;

/// Active embedding time before a remaining-time estimate is offered.
const ETA_SETTLE_MS: u64 = 5_000;

/// Watcher event batches up to this many files run straight away; a larger
/// one is admitted like a run. A saved file must reach search in seconds, and
/// thirty-two files is bounded work; a `git checkout` of thousands is not.
pub const BATCH_ADMIT_OVER: usize = 32;

/// One index run, as the daemon knows it and a page reads it.
///
/// This is the whole of what 0.20.0 changes about a run's lifetime. Until now
/// a run existed only as the events travelling down one HTTP response, so the
/// tab that started it was the only thing that knew it was happening and
/// leaving the page threw that away. The run lives here instead: the same
/// `say` closure that emits every event writes this, so there is one path and
/// the snapshot cannot disagree with the stream.
pub struct RunState {
    pub id: u64,
    pub paths: Vec<PathBuf>,
    pub kind: RunKind,
    /// Files and chunks the store held when the run started, read by the
    /// daemon rather than guessed by the page. Stop's "Also delete the store"
    /// is ticked by default only when this is zero.
    pub files_before: Option<u64>,
    pub chunks_before: Option<u64>,
    /// The intra-op thread count the run's session was built with.
    pub threads: u64,
    /// `(active milliseconds, chunks, bytes)` since embedding began, trimmed
    /// to the rate window. Active time, so a paused or held stretch is not in
    /// it.
    samples: VecDeque<(u64, u64, u64)>,
    /// Each lane's chunk count at the same moments, for the rate per lane.
    lane_samples: VecDeque<(u64, BTreeMap<String, u64>)>,
    /// The first sample, which the average is taken from: queued time and a
    /// walk of unchanged files before it do not dilute the figure.
    first_sample: Option<(u64, u64, u64)>,
    /// Whether "Also delete the store" was ticked on this run's stop, and what
    /// became of the delete.
    pub delete: Option<String>,
    pub status: RunStatus,
    /// Unix seconds the run was submitted, first began, and ended. A finished
    /// card says how long the work took, from `started` to `finished`, and
    /// names any wait before it apart, rather than folding the queue into it.
    pub submitted: u64,
    pub started: Option<u64>,
    pub finished: Option<u64>,
    /// Milliseconds on the run's clock before it started: its wait in the
    /// queue. The run's clock minus this is how long the work took, to the
    /// millisecond, which two unix seconds cannot say for a short run.
    queued_ms: Option<u64>,
    /// Where this run's clock started, and what it has spent held.
    ///
    /// The clock belongs to the run, not to the slice. A run yields the writer
    /// back to the watcher every [`SLICE`] and comes back as a fresh
    /// `Job::Index`, so a duration measured inside `perform` restarted from
    /// zero every forty-five seconds — which a page faithfully redrew as the
    /// run having just begun. Measured here it spans every slice, survives the
    /// page being closed, and stops while the run is paused: held time is not
    /// time anything is happening.
    origin: std::time::Instant,
    paused_at: Option<std::time::Instant>,
    paused_total: Duration,
    /// The reading frozen at the moment the run ended, so a finished card
    /// keeps its total rather than counting on.
    ended: Option<Duration>,
    pub scanned: u64,
    pub total: u64,
    /// Bytes of the walk got through, and found. See `crate::IndexProgress`.
    pub bytes: u64,
    pub bytes_total: u64,
    pub indexed: u64,
    pub chunks: u64,
    pub symbols: u64,
    /// Chunk rows written, embedded or not yet.
    pub rows: u64,
    /// What the run is doing when it is not reading a file.
    ///
    /// Every two hundred files a run flushes its batch and rewrites the
    /// shards, which on a large corpus is twenty seconds with no counter
    /// moving and nothing said — indistinguishable from a hang. Carried on the
    /// snapshot rather than only on the log, so a page opened during one of
    /// those twenty seconds sees it too.
    pub phase: Option<String>,
    /// The scan phase's plan, as the card shows it before anything embeds.
    pub plan: Option<serde_json::Value>,
    /// The plan's own estimate, which stands until the measured rate settles:
    /// the first remaining time appears when embedding starts (2.7).
    plan_eta_ms: Option<u64>,
    /// The last estimate a poll was given, and when (ms of the run's clock).
    /// What a stall holds on to, and what a jump is measured against (#143).
    shown_eta: std::cell::Cell<Option<(u64, u64)>>,
    /// The `done` event as it was sent, with its refused, failed and
    /// skipped-by-reason lists.
    pub summary: Option<serde_json::Value>,
    /// Where the run's time has gone so far, as of its last slice.
    stages: Option<serde_json::Value>,
    /// Chunks the vector cache was asked for, and held, as of its last slice.
    cache: (u64, u64),
    /// The last [`LOG_HISTORY`] events, each carrying the sequence number a
    /// client reads after.
    log: VecDeque<serde_json::Value>,
    next_seq: u64,
}

impl RunState {
    fn new(id: u64, paths: Vec<PathBuf>, kind: RunKind) -> Self {
        Self {
            id,
            paths,
            kind,
            files_before: None,
            chunks_before: None,
            threads: 0,
            samples: VecDeque::new(),
            lane_samples: VecDeque::new(),
            first_sample: None,
            delete: None,
            status: RunStatus::Queued,
            submitted: now(),
            started: None,
            finished: None,
            queued_ms: None,
            // From submission, not from the writer taking it: the wait for a
            // writer is time the person is waiting.
            origin: std::time::Instant::now(),
            paused_at: None,
            paused_total: Duration::ZERO,
            ended: None,
            scanned: 0,
            total: 0,
            bytes: 0,
            bytes_total: 0,
            indexed: 0,
            chunks: 0,
            symbols: 0,
            rows: 0,
            phase: None,
            shown_eta: std::cell::Cell::new(None),
            plan: None,
            plan_eta_ms: None,
            summary: None,
            stages: None,
            cache: (0, 0),
            log: VecDeque::new(),
            next_seq: 0,
        }
    }

    /// How long this run has been going, not counting time it spent held.
    ///
    /// Monotonic, so it never goes backwards: a page that polls this can tick
    /// between polls and be corrected by each one without the reading ever
    /// jumping back — which is what a per-slice measurement did.
    fn elapsed(&self) -> Duration {
        if let Some(ended) = self.ended {
            return ended;
        }
        // One instant, both spans measured from it. Sampling `Instant::now()`
        // twice — once inside the held total and once for the run's own — puts
        // the gap between the two calls into the answer, so a *held* clock
        // crept upward by a few hundred nanoseconds per read and crossed a
        // millisecond boundary often enough to be visible.
        let now = std::time::Instant::now();
        let held = self.paused_total
            + self
                .paused_at
                .map(|at| now.saturating_duration_since(at))
                .unwrap_or_default();
        now.saturating_duration_since(self.origin)
            .saturating_sub(held)
    }

    /// Bytes per second from the first batch to the last, once there were two.
    fn embed_rate(&self) -> Option<f64> {
        let (t0, _, b0) = self.first_sample?;
        let &(at, _, bytes) = self.samples.back()?;
        let ms = at.checked_sub(t0).filter(|ms| *ms > 0)?;
        let moved = bytes.checked_sub(b0).filter(|b| *b > 0)?;
        Some(moved as f64 * 1000.0 / ms as f64)
    }

    /// Stop counting for as long as the run is held.
    fn hold(&mut self) {
        if self.paused_at.is_none() {
            self.paused_at = Some(std::time::Instant::now());
        }
    }

    /// Start counting again, keeping what was already counted.
    fn unhold(&mut self) {
        if let Some(at) = self.paused_at.take() {
            self.paused_total += at.elapsed();
        }
    }

    /// Chunks per second over the last [`RATE_WINDOW_MS`] of active time, and
    /// over the whole of it since embedding began. `None` before the first
    /// batch, which the card shows as a dash rather than dropping the field.
    fn rates(&self) -> (Option<f64>, Option<f64>) {
        let Some((t0, c0, _)) = self.first_sample else {
            return (None, None);
        };
        if self.chunks == 0 {
            return (None, None);
        }
        let now = self.elapsed().as_millis() as u64;
        let per_second = |from: (u64, u64, u64)| {
            let span = now.saturating_sub(from.0);
            (span > 0).then(|| {
                ((self.chunks.saturating_sub(from.1)) as f64 * 1000.0 / span as f64 * 10.0).round()
                    / 10.0
            })
        };
        // The newest sample at least a window old, or the first one while the
        // run is younger than a window: the chunks since then over the time
        // since then, measured to now, so a stall shows as a falling rate
        // rather than as the last good figure frozen in place.
        let start = now.saturating_sub(RATE_WINDOW_MS);
        let base = self
            .samples
            .iter()
            .rev()
            .find(|(at, _, _)| *at <= start)
            .copied()
            .unwrap_or((t0, c0, 0));
        (per_second(base), per_second((t0, c0, 0)))
    }

    /// Milliseconds of work left, from the bytes still to go over the bytes
    /// per second of the last [`RATE_WINDOW_MS`] of active time.
    ///
    /// `None` until at least [`ETA_SETTLE_MS`] of embedding has been seen, and
    /// whenever the run is not moving: the first seconds of a run carry the
    /// model load and whichever file happened to come first, and a figure
    /// taken from them swings by minutes between polls.
    fn eta_ms(&self) -> Option<u64> {
        if !matches!(self.status, RunStatus::Running) || self.bytes_total == 0 {
            return None;
        }
        let now = self.elapsed().as_millis() as u64;
        let from_plan = self.plan_eta_ms.map(|eta| eta.saturating_sub(now));
        let Some((t0, _, b0)) = self.first_sample else {
            return from_plan;
        };
        if now.saturating_sub(t0) < ETA_SETTLE_MS {
            return from_plan;
        }
        let start = now.saturating_sub(RATE_WINDOW_MS);
        let (at, _, was) = self
            .samples
            .iter()
            .rev()
            .find(|(at, _, _)| *at <= start)
            .copied()
            .unwrap_or((t0, 0, b0));
        let moved = self.bytes.saturating_sub(was);
        let span = now.saturating_sub(at);
        let shown = self.shown_eta.get();
        // A run writing its index moves no bytes for twenty seconds, and the
        // rate over that window says the rest will take minutes. Issue #143:
        // the card jumped to minutes, then back. While the run says it is in
        // a phase, or no bytes moved in the window, the last estimate stands.
        // Not "fewer than 1 % of the bytes": a run that takes longer than
        // about seventeen minutes never moves 1 % in ten seconds, and kept
        // its first estimate to the end — 9 h 19 min for the whole of the
        // 70-repository corpus. A spike out of a stall is the doubling cap's
        // to absorb, below.
        let stalled = self.phase.is_some() || moved == 0;
        if stalled && let Some((eta, _)) = shown {
            return Some(eta);
        }
        if moved == 0 || span == 0 {
            return None;
        }
        let left = self.bytes_total.saturating_sub(self.bytes);
        let mut eta = (left as u128 * span as u128 / moved as u128) as u64;
        // And never more than double between two polls: a real slowdown shows
        // over a few polls, a spike does not show at all.
        if let Some((last, _)) = shown {
            eta = eta.min(last.saturating_mul(2).max(1_000));
        }
        self.shown_eta.set(Some((eta, now)));
        Some(eta)
    }

    /// Chunks per second per lane over the same window as [`Self::rates`].
    fn lane_rates(&self) -> BTreeMap<String, f64> {
        let (Some((_, latest)), Some(_)) = (self.lane_samples.back(), self.first_sample) else {
            return BTreeMap::new();
        };
        let now = self.elapsed().as_millis() as u64;
        let start = now.saturating_sub(RATE_WINDOW_MS);
        let base = self
            .lane_samples
            .iter()
            .rev()
            .find(|(at, _)| *at <= start)
            .or_else(|| self.lane_samples.front());
        let Some((from, earlier)) = base else {
            return BTreeMap::new();
        };
        let span = now.saturating_sub(*from).max(1) as f64 / 1000.0;
        latest
            .iter()
            .map(|(lane, n)| {
                let was = earlier.get(lane).copied().unwrap_or(0);
                (
                    lane.clone(),
                    ((n.saturating_sub(was)) as f64 / span * 10.0).round() / 10.0,
                )
            })
            .collect()
    }

    fn sample_lanes(&mut self, lanes: BTreeMap<String, u64>) {
        let now = self.elapsed().as_millis() as u64;
        self.lane_samples.push_back((now, lanes));
        let start = now.saturating_sub(RATE_WINDOW_MS);
        while self.lane_samples.len() > 2
            && self.lane_samples.get(1).is_some_and(|(at, _)| *at <= start)
        {
            self.lane_samples.pop_front();
        }
        while self.lane_samples.len() > 1024 {
            self.lane_samples.pop_front();
        }
    }

    fn sample(&mut self) {
        let now = self.elapsed().as_millis() as u64;
        if self.first_sample.is_none() {
            self.first_sample = Some((now, self.chunks, self.bytes));
        }
        self.samples.push_back((now, self.chunks, self.bytes));
        // One sample older than the window is kept as its base.
        let start = now.saturating_sub(RATE_WINDOW_MS);
        while self.samples.len() > 2 && self.samples.get(1).is_some_and(|(at, _, _)| *at <= start) {
            self.samples.pop_front();
        }
        while self.samples.len() > 1024 {
            self.samples.pop_front();
        }
    }

    /// Fold one event into the run and put it on the log ring.
    fn absorb(&mut self, event: &serde_json::Value) {
        let num = |key: &str| event.get(key).and_then(serde_json::Value::as_u64);
        match event.get("event").and_then(serde_json::Value::as_str) {
            Some("started") => {
                self.status = RunStatus::Running;
                self.started.get_or_insert_with(now);
                let waited = self.elapsed().as_millis() as u64;
                self.queued_ms.get_or_insert(waited);
                self.files_before = num("files_before").or(self.files_before);
                self.chunks_before = num("chunks_before").or(self.chunks_before);
                self.unhold();
            }
            Some("held") => {
                self.status = RunStatus::Held;
                self.hold();
            }
            Some(kind @ ("file" | "progress")) => {
                // A card asked to pause or stop keeps saying so until the
                // engine answers; a file line is not that answer. Nor does one
                // end a pause: only `resumed` does, and a batch that was in
                // flight when the pause came may still report after it.
                if !matches!(
                    self.status,
                    RunStatus::Pausing | RunStatus::Paused | RunStatus::Stopping
                ) {
                    self.status = RunStatus::Running;
                }
                // A `writing` line is the run saying it has stopped reading
                // files for a moment; the next line of any other outcome ends
                // the phase. Progress inside a file says nothing about it.
                self.phase = if kind == "progress" {
                    self.phase.take()
                } else {
                    match event.get("outcome").and_then(serde_json::Value::as_str) {
                        Some("writing") => event
                            .get("why")
                            .and_then(serde_json::Value::as_str)
                            .map(str::to_string)
                            .or_else(|| Some("writing the index to disk".to_string())),
                        _ => None,
                    }
                };
                // Taken rather than added: every one of these is the run's
                // own running total, so summing them would count each file
                // once per event it appeared in.
                self.scanned = num("scanned").unwrap_or(self.scanned);
                self.total = num("total").unwrap_or(self.total);
                self.bytes = num("bytes").unwrap_or(self.bytes);
                self.bytes_total = num("bytes_total").unwrap_or(self.bytes_total);
                self.indexed = num("indexed").unwrap_or(self.indexed);
                self.chunks = num("chunks").unwrap_or(self.chunks);
                self.symbols = num("symbols").unwrap_or(self.symbols);
                self.rows = num("rows").unwrap_or(self.rows);
                self.threads = num("threads").filter(|n| *n > 0).unwrap_or(self.threads);
                // From the first file embedding starts on, so the walk of
                // unchanged files before it is not part of either rate.
                let embedding = self.first_sample.is_some()
                    || self.chunks > 0
                    || kind == "progress"
                    || event.get("outcome").and_then(serde_json::Value::as_str) == Some("indexing");
                if embedding {
                    self.sample();
                    if let Some(lanes) = event.get("lanes").and_then(serde_json::Value::as_object) {
                        let lanes: BTreeMap<String, u64> = lanes
                            .iter()
                            .filter_map(|(k, v)| v.as_u64().map(|n| (k.clone(), n)))
                            .collect();
                        if !lanes.is_empty() {
                            self.sample_lanes(lanes);
                        }
                    }
                }
            }
            Some("paused") => {
                self.status = RunStatus::Paused;
                self.hold();
            }
            Some("resumed" | "slice") => {
                if let Some(stages) = event.get("stages") {
                    self.stages = Some(stages.clone());
                }
                if let (Some(asked), Some(held)) = (num("cache_lookups"), num("cache_hits")) {
                    self.cache = (asked, held);
                }
                if self.status != RunStatus::Stopping {
                    self.status = RunStatus::Running;
                }
                self.unhold();
            }
            Some("deleted" | "delete-failed") => {
                self.delete = event
                    .get("text")
                    .and_then(serde_json::Value::as_str)
                    .map(str::to_string);
            }
            Some("done") => {
                if let Some(stages) = event.get("stages") {
                    self.stages = Some(stages.clone());
                }
                if let (Some(asked), Some(held)) = (num("cache_lookups"), num("cache_hits")) {
                    self.cache = (asked, held);
                }
                let stopped = event
                    .get("stopped")
                    .and_then(serde_json::Value::as_bool)
                    .unwrap_or(false);
                self.status = if stopped {
                    RunStatus::Stopped
                } else {
                    RunStatus::Done
                };
                self.phase = None;
                self.indexed = num("indexed").unwrap_or(self.indexed);
                self.chunks = num("chunks").unwrap_or(self.chunks);
                self.unhold();
                self.ended = Some(self.elapsed());
                self.finished = Some(now());
                self.summary = Some(event.clone());
            }
            Some("error") => {
                self.status = RunStatus::Failed;
                self.unhold();
                self.ended = Some(self.elapsed());
                self.finished = Some(now());
            }
            _ => {}
        }

        // Per-batch progress moves the snapshot and nothing else: a log line
        // per eight chunks would push every file line off the ring.
        if event.get("event").and_then(serde_json::Value::as_str) == Some("progress") {
            return;
        }
        let mut line = event.clone();
        if let Some(object) = line.as_object_mut() {
            object.insert("seq".into(), serde_json::json!(self.next_seq));
            // When, for the run history's log (0.35.0); additive on the line.
            object.entry("at").or_insert(serde_json::json!(now()));
        }
        self.next_seq += 1;
        if self.log.len() == LOG_HISTORY {
            self.log.pop_front();
        }
        self.log.push_back(line);
    }
}

/// How many runs a store's history keeps, and how many a route lists.
const HISTORY_KEEP: usize = 50;

/// Log lines kept with one finished run.
const HISTORY_LOG: usize = 200;

/// The file a store's finished runs are kept in: JSON lines, oldest first,
/// beside `store.db` so it travels and is deleted with the store. An older
/// binary never opens it.
pub const HISTORY_FILE: &str = "runs.jsonl";

impl RunState {
    /// This run as the history keeps it: `{id, store, kind, started,
    /// finished, result, files, chunks, stages {stage: ms}, log}`.
    fn history_row(&self, store: &str) -> serde_json::Value {
        let result = match self.status {
            RunStatus::Stopped => "stopped",
            RunStatus::Failed => "failed",
            _ => "done",
        };
        // Flattened to one number per stage; each lane's wait is summed into
        // `embed`, which is what the page draws as one bar.
        let mut stages = serde_json::Map::new();
        if let Some(object) = self.stages.as_ref().and_then(serde_json::Value::as_object) {
            for (key, value) in object {
                if let (Some(stage), Some(ms)) = (key.strip_suffix("_ms"), value.as_u64())
                    && key != "wall_ms"
                {
                    stages.insert(stage.to_string(), serde_json::json!(ms));
                }
            }
            let embed: u64 = object
                .get("embed_wait_ms")
                .and_then(serde_json::Value::as_object)
                .map(|lanes| lanes.values().filter_map(serde_json::Value::as_u64).sum())
                .unwrap_or(0);
            stages.insert("embed".into(), serde_json::json!(embed));
        }
        let log: Vec<serde_json::Value> = self
            .log
            .iter()
            .rev()
            .take(HISTORY_LOG)
            .rev()
            .map(history_line)
            .collect();
        serde_json::json!({
            "id": self.id,
            "store": store,
            "kind": self.kind.label(self.paths.len()),
            "started": self.started.unwrap_or(self.submitted),
            "finished": self.finished.unwrap_or_else(now),
            "result": result,
            "files": self.indexed,
            "chunks": self.chunks,
            "stages": stages,
            "log": log,
        })
    }
}

/// One run event as a history log line: `{at, level, text}`.
fn history_line(event: &serde_json::Value) -> serde_json::Value {
    let text = |key: &str| {
        event
            .get(key)
            .and_then(serde_json::Value::as_str)
            .unwrap_or("")
            .to_string()
    };
    let name = text("event");
    let (level, line) = match name.as_str() {
        "file" => {
            let outcome = text("outcome");
            let level = match outcome.as_str() {
                "failed" => "error",
                "refused" | "skipped" => "warn",
                _ => "info",
            };
            let why = text("why");
            let path = text("path");
            (
                level,
                if why.is_empty() {
                    format!("{outcome} {path}")
                } else {
                    format!("{outcome} {path}: {why}")
                },
            )
        }
        "error" => ("error", format!("error: {}", text("error"))),
        "done" => (
            "info",
            format!(
                "done: {} indexed, {} chunks",
                event
                    .get("indexed")
                    .and_then(serde_json::Value::as_u64)
                    .unwrap_or(0),
                event
                    .get("chunks")
                    .and_then(serde_json::Value::as_u64)
                    .unwrap_or(0)
            ),
        ),
        _ if !text("text").is_empty() => ("info", text("text")),
        other => ("info", other.to_string()),
    };
    serde_json::json!({
        "at": event.get("at").cloned().unwrap_or(serde_json::Value::Null),
        "level": level,
        "text": line,
    })
}

/// Append one finished run to a store's history, trimming it to
/// [`HISTORY_KEEP`] when it has grown to twice that.
///
/// ponytail: a whole-file rewrite at 100 lines; a run is a few KB, so it is
/// a sub-megabyte write once every fifty runs.
fn remember_run(dir: &Path, row: &serde_json::Value) {
    use std::io::Write;
    let path = dir.join(HISTORY_FILE);
    let appended = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(&path)
        .and_then(|mut file| writeln!(file, "{row}"));
    if appended.is_err() {
        return;
    }
    crate::home::tighten_file(&path);
    if let Ok(text) = std::fs::read_to_string(&path) {
        let lines: Vec<&str> = text.lines().collect();
        if lines.len() > HISTORY_KEEP * 2 {
            let kept = lines[lines.len() - HISTORY_KEEP..].join("\n") + "\n";
            let _ = crate::home::write_private(&path, kept.as_bytes());
        }
    }
}

/// A store's finished runs, newest first, at most [`HISTORY_KEEP`].
pub fn run_history(dir: &Path) -> Vec<serde_json::Value> {
    let Ok(text) = std::fs::read_to_string(dir.join(HISTORY_FILE)) else {
        return Vec::new();
    };
    text.lines()
        .rev()
        .filter_map(|line| serde_json::from_str(line).ok())
        .take(HISTORY_KEEP)
        .collect()
}

/// Everything about one open store that a route can ask about.
pub struct Store {
    pub name: String,
    pub dir: PathBuf,
    /// Roots as the registry records them, including any that no longer exist.
    pub roots: Vec<PathBuf>,
    /// The subset of `roots` that is actually on disk and being watched.
    pub watched: Vec<PathBuf>,
    queue: Mutex<VecDeque<Queued>>,
    events: Mutex<VecDeque<Event>>,
    /// This store's runs, oldest first.
    ///
    /// Kept after a run ends so a page opened afterwards shows what it did
    /// rather than an empty card. A list rather than a slot because a store
    /// can be indexed twice: the second run used to replace the first one's
    /// record, so the card kept the old log under the new header and the only
    /// button a finished run offered was Stop — which then latched the store
    /// and killed whatever ran next.
    runs: Mutex<Vec<RunState>>,
    /// False once the watcher thread has returned, so the portal can say a
    /// store stopped being kept current rather than showing a stale count.
    pub watching: AtomicBool,
    /// Set to stop this store's watcher and release its lock, so the store can
    /// be deleted while the daemon keeps running. The daemon's own shutdown
    /// sets it on every open store alongside the global stop.
    pub stop: AtomicBool,
    /// A queued index run holds here between files while this is set.
    pub paused: AtomicBool,
    /// A queued index run gives up and undoes itself when this is set.
    pub cancelled: AtomicBool,
    /// Unix seconds of the last write this daemon made to the store.
    ///
    /// A cache in front of the store's own `MAX(indexed_at)`, not the answer:
    /// a store written before this daemon started has a last write and this
    /// counter does not know it. `/api/stores` reads the store.
    pub last_write: AtomicUsize,
    /// Rows dropped when this store was opened because they sat outside every
    /// root it is registered against. Shown for the session, so a store that
    /// was contaminated says so rather than quietly correcting itself.
    pub pruned: AtomicUsize,
    /// Unix seconds until which this store's watcher holds off its catch-up,
    /// because a run that is about to arrive is what should index it.
    ///
    /// A store the portal creates is opened and then indexed, in that order,
    /// and the watcher used to win the race: it embedded the corpus itself,
    /// and the run that followed correctly found every file unchanged and
    /// reported "1 indexed, 2 unchanged, 1 chunks" for a store that had just
    /// gained three files. A deadline rather than a flag, so a run that never
    /// arrives costs the watcher a minute rather than the session.
    pub expecting_run_until: AtomicUsize,
    /// Why the writer thread ended, when it has. `/api/stores` reports it
    /// beside `watching: false`, so a stopped store says what stopped it.
    pub stopped_because: Mutex<Option<String>>,
    /// The store's `watch` setting (0.35.0): while false, filesystem events
    /// are dropped rather than re-embedded. The writer thread itself keeps
    /// running, because it is also what drains the queue a portal index or a
    /// forwarded `semlith_index` lands on — stopping it would make the store
    /// unwritable, which is not what "stop watching" means.
    pub watch_events: AtomicBool,
    /// The store's `gitignore` setting, read by the writer between jobs.
    pub gitignore: AtomicBool,
}

impl Store {
    /// A store's record with nothing going on yet. One constructor, so a field
    /// added here is a field every caller gets.
    fn new(
        name: String,
        dir: PathBuf,
        roots: Vec<PathBuf>,
        watched: Vec<PathBuf>,
        watching: bool,
        expecting_run_until: usize,
    ) -> Self {
        Self {
            name,
            dir,
            roots,
            watched,
            queue: Mutex::new(VecDeque::new()),
            events: Mutex::new(VecDeque::new()),
            runs: Mutex::new(Vec::new()),
            watching: AtomicBool::new(watching),
            stop: AtomicBool::new(false),
            paused: AtomicBool::new(false),
            cancelled: AtomicBool::new(false),
            last_write: AtomicUsize::new(0),
            pruned: AtomicUsize::new(0),
            // Set before the watcher thread exists, which is the point: a flag
            // set afterwards is a flag the catch-up may already have run past.
            expecting_run_until: AtomicUsize::new(expecting_run_until),
            stopped_because: Mutex::new(None),
            watch_events: AtomicBool::new(true),
            gitignore: AtomicBool::new(true),
        }
    }

    /// Take the registry's settings for this store into the live flags.
    pub fn apply_settings(&self, settings: &crate::home::StoreSettings) {
        self.watch_events.store(settings.watch, Ordering::Relaxed);
        self.gitignore.store(settings.gitignore, Ordering::Relaxed);
    }

    fn note(&self, text: String) {
        let mut events = self.events.lock().unwrap_or_else(|e| e.into_inner());
        if events.len() == EVENT_HISTORY {
            events.pop_front();
        }
        events.push_back(Event { at: now(), text });
        changes::bump(changes::Domain::Events);
    }

    /// Start a run's record, before it is admitted to the writer.
    ///
    /// Appended: a second run against this store is a second card with its own
    /// header, its own log and its own buttons. The oldest finished runs are
    /// dropped once there are more than [`RUN_HISTORY`] of them, so a daemon
    /// left up for a week does not accumulate cards forever; a run that has
    /// not finished is never dropped.
    fn begin_run(&self, id: u64, paths: Vec<PathBuf>, kind: RunKind) {
        let mut runs = self.runs.lock().unwrap_or_else(|e| e.into_inner());
        runs.push(RunState::new(id, paths, kind));
        while runs.len() > RUN_HISTORY {
            match runs.iter().position(|run| run.status.finished()) {
                Some(at) => {
                    runs.remove(at);
                }
                None => break,
            }
        }
        drop(runs);
        self.runs_changed();
    }

    /// Put the scan phase's plan on a run, and hold it for review when asked.
    fn set_plan(&self, id: u64, plan: &crate::Plan, review: bool) {
        let mut runs = self.runs.lock().unwrap_or_else(|e| e.into_inner());
        // A store with no rate of its own yet, on the rate of the last run this
        // daemon watched embed: the model and the machine are most of it.
        let seen = runs
            .iter()
            .filter(|r| r.id != id)
            .filter_map(RunState::embed_rate)
            .last();
        if let Some(run) = runs.iter_mut().find(|r| r.id == id) {
            run.plan = serde_json::to_value(plan).ok();
            run.plan_eta_ms = plan
                .eta_ms
                .or_else(|| seen.map(|rate| (plan.embed_bytes as f64 / rate * 1000.0) as u64));
            if review {
                run.status = RunStatus::Review;
            }
        }
        drop(runs);
        self.runs_changed();
    }

    /// What the store held before a run that has not started.
    fn set_before(&self, id: u64, files: u64, chunks: u64) {
        let mut runs = self.runs.lock().unwrap_or_else(|e| e.into_inner());
        if let Some(run) = runs.iter_mut().find(|r| r.id == id) {
            run.files_before = Some(files);
            run.chunks_before = Some(chunks);
        }
        drop(runs);
        self.runs_changed();
    }

    /// A run held for review, moved to the queue or ended.
    fn leave_review(&self, id: u64, to: RunStatus) -> bool {
        let mut runs = self.runs.lock().unwrap_or_else(|e| e.into_inner());
        let Some(run) = runs
            .iter_mut()
            .find(|r| r.id == id && r.status == RunStatus::Review)
        else {
            return false;
        };
        run.status = to;
        drop(runs);
        self.runs_changed();
        true
    }

    /// The one place the runs counter moves.
    ///
    /// Four things change what `/api/index/runs` would answer — a run begins,
    /// an event lands, a card is removed, the finished cards are cleared — and
    /// a counter bumped from four places is four sources of truth. See
    /// `one_bump_site_per_domain`.
    fn runs_changed(&self) {
        changes::bump(changes::Domain::Runs);
    }

    /// Fold one of this run's events into the snapshot.
    ///
    /// Called from `perform`'s `say`, so the snapshot and the stream carry the
    /// same events in the same order and neither can be the newer of the two.
    /// Addressed by run id rather than by store: a store with a run going and
    /// another queued behind it has two records, and an event folded into the
    /// wrong one is a card describing work it did not do.
    fn record(&self, id: u64, event: &serde_json::Value) {
        let ended = self
            .runs
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .iter_mut()
            .find(|run| run.id == id)
            .and_then(|run| {
                let was = run.status.finished();
                run.absorb(event);
                // A burst of events that changed nothing is not a run anybody
                // will look for, as it leaves no card either.
                let quiet = run.kind == RunKind::Batch
                    && run.indexed == 0
                    && run
                        .summary
                        .as_ref()
                        .and_then(|s| s.get("removed"))
                        .and_then(serde_json::Value::as_u64)
                        .unwrap_or(0)
                        == 0;
                (!was && run.status.finished() && !quiet).then(|| run.history_row(&self.name))
            });
        // Written outside the lock: a page polling runs should not wait on
        // the disk.
        if let Some(row) = ended {
            remember_run(&self.dir, &row);
        }
        self.runs_changed();
    }

    /// Do something with the run that has not finished, if there is one.
    fn with_live<T>(&self, body: impl FnOnce(&mut RunState) -> T) -> Option<T> {
        self.runs
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .iter_mut()
            .find(|run| !run.status.finished())
            .map(body)
    }

    /// Say that a stop has been asked for.
    ///
    /// Told apart from `stopped` because undoing what a run embedded takes as
    /// long as the embedding did on a large corpus, and a card that jumped
    /// straight to "stopped" would be claiming the store was already back to
    /// what it was. A run that had already finished is left alone: a stop that
    /// arrived late did not stop anything.
    pub fn mark_stopping(&self) {
        self.with_live(|run| run.status = RunStatus::Stopping);
        self.runs_changed();
    }

    /// Say that a pause has been asked for, so the card answers on the click
    /// and moves to `paused` when the engine reaches its next batch.
    pub fn mark_pausing(&self) {
        self.with_live(|run| {
            if matches!(run.status, RunStatus::Running) {
                run.status = RunStatus::Pausing;
            }
        });
        self.runs_changed();
    }

    /// Resume, as the card shows it at once. A held run stays held: the
    /// queue, not the button, decides when it goes on.
    pub fn mark_resuming(&self) {
        self.with_live(|run| {
            if matches!(run.status, RunStatus::Paused | RunStatus::Pausing) {
                run.status = RunStatus::Running;
            }
        });
        self.runs_changed();
    }

    /// The files the store held before its live run started, if it has one
    /// that has started.
    pub fn live_files_before(&self) -> Option<u64> {
        self.with_live(|run| run.files_before).flatten()
    }

    /// Put an event on one run's record from outside the writer, such as
    /// the outcome of a delete that followed its stop.
    pub fn note_run(&self, id: u64, event: &serde_json::Value) {
        self.record(id, event);
    }

    /// What started a run.
    fn run_kind(&self, id: u64) -> Option<RunKind> {
        self.runs
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .iter()
            .find(|run| run.id == id)
            .map(|run| run.kind)
    }

    /// The paths a run was asked to index.
    fn run_paths(&self, id: u64) -> Vec<PathBuf> {
        self.runs
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .iter()
            .find(|run| run.id == id)
            .map(|run| run.paths.clone())
            .unwrap_or_default()
    }

    /// The id of the run a stop just ended, or is ending.
    pub fn last_run(&self) -> Option<u64> {
        self.runs
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .last()
            .map(|run| run.id)
    }

    /// How long a run has been going, in milliseconds.
    ///
    /// The one clock for a run, read by the events it emits as well as by the
    /// snapshot, so the two cannot disagree — and it spans every slice, which
    /// is the whole of what was wrong with measuring inside `perform`.
    fn run_elapsed_ms(&self, id: u64) -> u64 {
        self.runs
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .iter()
            .find(|run| run.id == id)
            .map(|run| run.elapsed().as_millis() as u64)
            .unwrap_or(0)
    }

    /// The id of this store's unfinished run, if it has one.
    fn current_run(&self) -> Option<u64> {
        self.runs
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .iter()
            .find(|run| !run.status.finished())
            .map(|run| run.id)
    }

    /// The paths and the committed count of a run that is still going.
    ///
    /// Read at shutdown, which is the only moment it means anything: a run
    /// that has not finished when the process ends is one that will not, and
    /// this is what the next start has to say about it.
    fn interrupted(&self) -> Option<(Vec<PathBuf>, u64)> {
        self.with_live(|run| (run.paths.clone(), run.indexed))
    }

    /// Whether this store has a run that has not finished.
    pub fn run_live(&self) -> bool {
        self.current_run().is_some()
    }

    /// Drop one finished run's card.
    ///
    /// Only a finished one: Remove is how a card that has nothing left to say
    /// is dismissed, and removing a live run's record would leave the run
    /// going with nothing watching it.
    pub fn remove_run(&self, id: u64) -> bool {
        let mut runs = self.runs.lock().unwrap_or_else(|e| e.into_inner());
        let Some(at) = runs
            .iter()
            .position(|run| run.id == id && run.status.finished())
        else {
            return false;
        };
        runs.remove(at);
        drop(runs);
        self.runs_changed();
        true
    }

    /// Drop every finished run's card, and say how many went.
    pub fn clear_finished_runs(&self) -> usize {
        let mut runs = self.runs.lock().unwrap_or_else(|e| e.into_inner());
        let before = runs.len();
        runs.retain(|run| !run.status.finished());
        let gone = before - runs.len();
        drop(runs);
        if gone > 0 {
            self.runs_changed();
        }
        gone
    }

    /// This store's runs as `/api/index/runs` reports them, oldest first, with
    /// the queue position a caller works out from the admission queue.
    ///
    /// The position belongs to whichever run is still waiting, so a finished
    /// card is never labelled with a place in a queue it has already left.
    pub fn run_snapshots(&self, position: Option<usize>) -> Vec<serde_json::Value> {
        let runs = self.runs.lock().unwrap_or_else(|e| e.into_inner());
        runs.iter()
            .map(|run| {
                let position = (!run.status.finished()).then_some(position).flatten();
                self.snapshot_of(run, position)
            })
            .collect()
    }

    fn snapshot_of(&self, run: &RunState, position: Option<usize>) -> serde_json::Value {
        let (rate, average) = run.rates();
        serde_json::json!({
            "id": run.id,
            "store": self.name,
            "kind": run.kind,
            "files_before": run.files_before,
            "chunks_before": run.chunks_before,
            // What the session was built with, not what was asked for.
            "threads": (run.threads > 0).then_some(run.threads),
            // Chunks/s over the last ten seconds of active time, and since the
            // first batch. Null only before the first batch.
            "rate": if run.status.finished() { None } else { rate },
            "rate_average": average,
            // e.g. `{"gpu": 48.1, "cpu": 25.0}`: which device is doing what.
            "lane_rates": if run.status.finished() { BTreeMap::new() } else { run.lane_rates() },
            "delete": run.delete,
            "paths": run.paths.iter()
                .map(|p| crate::plain(&p.display().to_string()))
                .collect::<Vec<_>>(),
            "status": run.status,
            "plan": run.plan,
            "position": position,
            "submitted": run.submitted,
            "started_at": run.started,
            "finished_at": run.finished,
            "queued_ms": run.queued_ms,
            "elapsed_ms": run.elapsed().as_millis() as u64,
            // Null until the rate has settled, and whenever the run is not
            // running: the card says `estimating…` rather than guessing.
            "eta_ms": run.eta_ms(),
            "bytes": run.bytes,
            "bytes_total": run.bytes_total,
            // Whether the clock should be ticking in the page between polls.
            // A finished or held run keeps its reading; nothing else should
            // make a page decide that for itself.
            "ticking": matches!(run.status, RunStatus::Running | RunStatus::Queued | RunStatus::Stopping | RunStatus::Pausing),
            "scanned": run.scanned,
            "total": run.total,
            "indexed": run.indexed,
            "chunks": run.chunks,
            "symbols": run.symbols,
            // What the run is doing when it is not reading files — or, while
            // no lane it may use is ready and one is on its way, what it is
            // waiting for, with that lane's progress and time left.
            "phase": (run.status == RunStatus::Running)
                .then(crate::accel::waiting_for)
                .flatten()
                .or_else(|| run.phase.clone()),
            "summary": run.summary,
            // Walk, read+hash, extract+scan, parse+chunk, tokenize, the wait
            // on each lane, and write, summing to the run's wall time.
            "stages": run.stages,
            // The share of the store the vector half does not cover yet: rows
            // this run wrote that it has not embedded, over what the store
            // holds with them. Null outside a running run.
            "pending_share": (run.status == RunStatus::Running && run.rows > 0).then(|| {
                let pending = run.rows.saturating_sub(run.chunks) as f64;
                let whole = (run.chunks_before.unwrap_or(0) + run.rows).max(1) as f64;
                (pending / whole * 1000.0).round() / 1000.0
            }),
            // The vector cache's share of the run: chunks asked for, and held.
            "cache_lookups": run.cache.0,
            "cache_hits": run.cache.1,
            "cache_hit_rate": (run.cache.0 > 0).then(|| {
                (run.cache.1 as f64 * 1000.0 / run.cache.0 as f64).round() / 10.0
            }),
            // What a page's log cursor should be if it has never read this
            // run: the oldest line still on the ring, minus one.
            "log_from": run.log.front()
                .and_then(|line| line.get("seq"))
                .and_then(serde_json::Value::as_u64)
                .map(|first| first.saturating_sub(1)),
            "log_to": run.next_seq.saturating_sub(1),
        })
    }

    /// Every log line one run emitted after `seq`.
    ///
    /// A cursor rather than an offset, so two clients reading the same run
    /// through their own cursors each see every line exactly once and neither
    /// is affected by what the other has read.
    ///
    /// The run is named, because a store can have two. With no id the live run
    /// answers, or the last one if none is live — which is what a caller that
    /// knows only a store name means.
    pub fn log_after(&self, id: Option<u64>, seq: Option<u64>) -> Vec<serde_json::Value> {
        let runs = self.runs.lock().unwrap_or_else(|e| e.into_inner());
        let run = match id {
            Some(id) => runs.iter().find(|run| run.id == id),
            None => runs
                .iter()
                .find(|run| !run.status.finished())
                .or_else(|| runs.last()),
        };
        run.map(|run| {
            run.log
                .iter()
                .filter(|line| match seq {
                    None => true,
                    Some(after) => line
                        .get("seq")
                        .and_then(serde_json::Value::as_u64)
                        .is_some_and(|n| n > after),
                })
                .cloned()
                .collect()
        })
        .unwrap_or_default()
    }

    /// The watcher's recent events, oldest first.
    pub fn events(&self) -> Vec<Event> {
        self.events
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .iter()
            .cloned()
            .collect()
    }

    /// Hand a job to the writer and read its progress back as it happens.
    ///
    /// The receiver is what a streaming route writes chunks from, so the
    /// browser sees an index run while it is running rather than when it ends.
    /// Queue a job and hand back the channel its progress arrives on.
    ///
    /// `notice` is sent before the job is queued, for a caller that streams
    /// and would otherwise hear nothing while the writer finishes what it is
    /// doing. A caller that takes the first message as its answer — forget —
    /// passes `None`, because a notice would be that answer.
    fn submit(
        &self,
        job: Job,
        notice: Option<serde_json::Value>,
    ) -> mpsc::Receiver<serde_json::Value> {
        let (report, progress) = mpsc::channel();
        let mut queue = self.queue.lock().unwrap_or_else(|e| e.into_inner());
        if let Some(mut notice) = notice {
            if let Some(object) = notice.as_object_mut() {
                object.insert("store".into(), serde_json::json!(self.name));
                object.insert("ahead".into(), serde_json::json!(queue.len()));
            }
            let _ = report.send(notice);
        }
        // Run ids start at 1, so 0 is "no run": a forget or a fetch is a job
        // the writer performs and not a run with a card, and its events belong
        // to no run's log.
        queue.push_back(Queued {
            run: NO_RUN,
            job,
            report,
        });
        progress
    }

    /// Queue a job that is a run with a card of its own, minted by the caller.
    fn submit_run(&self, run: u64, job: Job) -> mpsc::Receiver<serde_json::Value> {
        let (report, progress) = mpsc::channel();
        let mut queue = self.queue.lock().unwrap_or_else(|e| e.into_inner());
        queue.push_back(Queued { run, job, report });
        progress
    }

    /// [`Store::submit`], at the front of the queue.
    fn submit_front(&self, job: Job) -> mpsc::Receiver<serde_json::Value> {
        let (report, progress) = mpsc::channel();
        let mut queue = self.queue.lock().unwrap_or_else(|e| e.into_inner());
        queue.push_front(Queued {
            run: NO_RUN,
            job,
            report,
        });
        progress
    }

    /// Put an admitted index run on this store's writer queue.
    ///
    /// Apart from [`Store::submit`] because the channel belongs to the run
    /// rather than to this call: the run was minted when it was queued for
    /// admission, possibly minutes ago, and whoever asked for it may already
    /// have been answered.
    fn submit_index(
        &self,
        run: u64,
        paths: Vec<PathBuf>,
        kind: RunKind,
        report: mpsc::Sender<serde_json::Value>,
        mut notice: serde_json::Value,
    ) {
        let mut queue = self.queue.lock().unwrap_or_else(|e| e.into_inner());
        if let Some(object) = notice.as_object_mut() {
            object.insert("store".into(), serde_json::json!(self.name));
            object.insert("ahead".into(), serde_json::json!(queue.len()));
        }
        // The snapshot before the stream, so a page that polls between these
        // two lines sees the notice rather than nothing.
        self.record(run, &notice);
        let _ = report.send(notice);
        queue.push_back(Queued {
            run,
            job: Job::Index(Indexing {
                // A burst of events names files, some of them deleted, and a
                // walk would drop the deleted ones; it is indexed as the list
                // it is.
                work: match kind {
                    // A picked list is indexed as the list it is, like a burst.
                    RunKind::Batch | RunKind::Files => Work::Rest(paths),
                    // A compaction is never admitted, so never reaches here;
                    // named rather than folded into a wildcard so a new kind
                    // has to say where it goes.
                    RunKind::Run
                    | RunKind::CatchUp
                    | RunKind::Compact
                    | RunKind::Reindex
                    | RunKind::Rebuild => Work::Roots(paths),
                },
                first: true,
                already: Vec::new(),
                scanned: 0,
                total: 0,
                bytes: 0,
                bytes_total: 0,
                tally: Tally::default(),
            }),
            report,
        });
    }

    /// Whether the writer has anything waiting — the Index view's queue depth.
    pub fn queue_depth(&self) -> usize {
        self.queue.lock().unwrap_or_else(|e| e.into_inner()).len()
    }

    /// Put a slice's remainder back, behind whatever is waiting.
    fn requeue(&self, queued: Queued) {
        self.queue
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .push_back(queued);
    }

    /// Drop every index job that has not started, answering each as stopped.
    ///
    /// A stop asked for while the job is still waiting its turn used to sit
    /// until the writer reached it and then be cleared, so the page said
    /// "stopping…" for as long as the queue took. Nothing was embedded, so
    /// there is nothing to undo and the answer is immediate.
    /// Take this store's waiting index jobs off its queue, and say which runs
    /// they were.
    ///
    /// The ids are the point. A run cancelled here has already been admitted —
    /// it is on the store's queue precisely because the admission queue let it
    /// through — so its place among the runs that may be going at once has to
    /// be given back, or the daemon loses a slot every time someone stops a
    /// run before its writer reaches it. Three stops left `runs at once` at
    /// three with nothing running and everything queued behind them forever.
    pub fn cancel_queued(&self) -> Vec<u64> {
        let mut queue = self.queue.lock().unwrap_or_else(|e| e.into_inner());
        let mut dropped = Vec::new();
        queue.retain(|queued| {
            if !matches!(queued.job, Job::Index(..) | Job::Compact(..)) {
                return true;
            }
            let answer = serde_json::json!({
                "event": "done",
                "indexed": 0,
                "unchanged": 0,
                "skipped": 0,
                "removed": 0,
                "chunks": 0,
                "images": 0,
                "remaining": 0,
                "stopped": true,
            });
            let _ = queued.report.send(answer);
            dropped.push(queued.run);
            false
        });
        drop(queue);
        // Folded into each run's own record, so a page that polls the snapshot
        // rather than reading the stream sees the run end too.
        for run in &dropped {
            self.record(
                *run,
                &serde_json::json!({
                    "event": "done",
                    "indexed": 0,
                    "unchanged": 0,
                    "skipped": 0,
                    "removed": 0,
                    "chunks": 0,
                    "images": 0,
                    "remaining": 0,
                    "stopped": true,
                }),
            );
        }
        dropped
    }
}

/// A run waiting for its store's writer, and the channel its progress goes
/// back on once it is admitted.
struct Pending {
    run: u64,
    store: Arc<Store>,
    paths: Vec<PathBuf>,
    kind: RunKind,
    report: mpsc::Sender<serde_json::Value>,
}

/// The one queue in front of every store's writer.
///
/// Each store's writer takes the next job on its own queue with nothing above
/// it, so with N stores open that is N runs at once regardless of what the
/// machine can carry — which is exactly what indexing several folders at once
/// would have produced. A run is admitted to its writer only while fewer than
/// `limit` are running; past that it waits here, in one daemon-wide FIFO
/// ordered by submission.
///
/// The watcher's own re-embeds do not pass through here. They are small, they
/// already interleave with runs through slices, and holding a file save behind
/// eleven queued repositories would make the watcher useless exactly when the
/// machine is busy.
pub struct Admission {
    queue: Mutex<VecDeque<Pending>>,
    /// Run ids admitted and not yet finished. A count would not do: a finish
    /// arriving for a run that was already removed would decrement the next
    /// run's place in the world.
    running: Mutex<Vec<u64>>,
    /// Runs admitted once and held again because the limit was lowered below
    /// the number running. Ahead of everything still queued, oldest first:
    /// they have work in hand, and their written list rides with them.
    held: Mutex<Vec<u64>>,
    limit: AtomicUsize,
    next: AtomicU64,
}

impl Admission {
    pub fn new(limit: usize) -> Self {
        Self {
            queue: Mutex::new(VecDeque::new()),
            running: Mutex::new(Vec::new()),
            held: Mutex::new(Vec::new()),
            limit: AtomicUsize::new(limit.max(1)),
            next: AtomicU64::new(1),
        }
    }

    pub fn limit(&self) -> usize {
        self.limit.load(Ordering::Relaxed)
    }

    /// Change how many runs may be on at once.
    ///
    /// Raising it admits at once. Lowering it holds the newest runs over the
    /// limit at their next batch — before 0.28.0 it stopped nothing, so a
    /// machine asked to do less went on doing as much until runs finished.
    /// A held run keeps its place on its writer, its progress and its written
    /// list, and goes on in submission order as slots free.
    pub fn set_limit(&self, limit: usize) {
        let limit = limit.max(1);
        self.limit.store(limit, Ordering::Relaxed);
        {
            let mut running = self.running.lock().unwrap_or_else(|e| e.into_inner());
            let mut held = self.held.lock().unwrap_or_else(|e| e.into_inner());
            while running.len() > limit {
                // Newest first: the oldest runs are furthest along.
                let Some(newest) = running.iter().copied().max() else {
                    break;
                };
                running.retain(|id| *id != newest);
                held.push(newest);
            }
            held.sort_unstable();
        }
        self.pump();
    }

    /// Whether this run is held by a lowered limit.
    pub fn is_held(&self, run: u64) -> bool {
        self.held
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .contains(&run)
    }

    /// Runs held by a lowered limit, oldest first.
    pub fn held(&self) -> Vec<u64> {
        self.held.lock().unwrap_or_else(|e| e.into_inner()).clone()
    }

    pub fn running(&self) -> usize {
        self.running.lock().unwrap_or_else(|e| e.into_inner()).len()
    }

    /// A fresh run id, for a run that is scanned and held for review before
    /// it is queued.
    pub fn mint(&self) -> u64 {
        self.next.fetch_add(1, Ordering::Relaxed)
    }

    /// Queue a run whose card already exists — one a person has just
    /// reviewed.
    pub fn enqueue(&self, run: u64, store: &Arc<Store>, paths: Vec<PathBuf>, kind: RunKind) {
        let (report, _progress) = mpsc::channel();
        {
            let mut queue = self.queue.lock().unwrap_or_else(|e| e.into_inner());
            store.record(
                run,
                &serde_json::json!({ "event": "submitted", "run": run, "ahead": queue.len() }),
            );
            queue.push_back(Pending {
                run,
                store: Arc::clone(store),
                paths,
                kind,
                report,
            });
        }
        self.pump();
    }

    /// Queue a run and hand back its id and the channel it reports on.
    ///
    /// The id is minted here rather than by a caller, so two routes submitting
    /// at the same moment cannot agree on one.
    pub fn submit(
        &self,
        store: &Arc<Store>,
        paths: Vec<PathBuf>,
        kind: RunKind,
    ) -> (u64, mpsc::Receiver<serde_json::Value>) {
        let run = self.next.fetch_add(1, Ordering::Relaxed);
        let (report, progress) = mpsc::channel();
        store.begin_run(run, paths.clone(), kind);
        {
            let mut queue = self.queue.lock().unwrap_or_else(|e| e.into_inner());
            store.record(
                run,
                &serde_json::json!({
                    "event": "submitted",
                    "run": run,
                    "ahead": queue.len(),
                }),
            );
            queue.push_back(Pending {
                run,
                store: Arc::clone(store),
                paths,
                kind,
                report,
            });
        }
        self.pump();
        (run, progress)
    }

    /// Admit as many waiting runs as the limit allows.
    ///
    /// Called on every submission, every finish and every change of the limit,
    /// which is the whole of the queue's movement: there is no timer, so a
    /// queue that is not moving is one where nothing has finished.
    pub fn pump(&self) {
        loop {
            let mut running = self.running.lock().unwrap_or_else(|e| e.into_inner());
            if running.len() >= self.limit() {
                return;
            }
            // A held run goes on before anything new starts: it is already
            // on its writer, part-way through.
            {
                let mut held = self.held.lock().unwrap_or_else(|e| e.into_inner());
                if !held.is_empty() {
                    let oldest = held.remove(0);
                    running.push(oldest);
                    continue;
                }
            }
            let mut queue = self.queue.lock().unwrap_or_else(|e| e.into_inner());
            let Some(next) = queue.pop_front() else {
                return;
            };
            running.push(next.run);
            drop(queue);
            drop(running);
            // A store that was stopped keeps no memory of it into the next
            // run. `cancelled` is the store's flag rather than the run's, and
            // nothing used to clear it, so a Stop pressed on one run killed
            // the next one at 0% — with the file fetched, written to disk and
            // never indexed, and no error anywhere.
            next.store.cancelled.store(false, Ordering::Relaxed);
            next.store.paused.store(false, Ordering::Relaxed);
            next.store.submit_index(
                next.run,
                next.paths,
                next.kind,
                next.report,
                serde_json::json!({ "event": "queued" }),
            );
        }
    }

    /// Release a run's place and admit whatever was waiting behind it.
    pub fn finish(&self, run: u64) {
        self.running
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .retain(|id| *id != run);
        self.held
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .retain(|id| *id != run);
        self.pump();
    }

    /// Where each waiting run stands, oldest first.
    pub fn waiting(&self) -> Vec<(u64, String, Vec<PathBuf>)> {
        self.queue
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .iter()
            .map(|p| (p.run, p.store.name.clone(), p.paths.clone()))
            .collect()
    }

    /// This store's place in the queue, counted from 1, or `None` when it is
    /// not waiting.
    pub fn position_of(&self, store: &str) -> Option<usize> {
        // Held runs are at the front of the line, so everything queued is
        // behind them.
        let ahead = self.held.lock().unwrap_or_else(|e| e.into_inner()).len();
        self.queue
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .iter()
            .position(|p| p.store.name == store)
            .map(|at| ahead + at + 1)
    }

    /// Take a waiting run out of the queue.
    ///
    /// Answered at once and with nothing undone, because nothing of it was
    /// embedded — which is the whole difference between removing a queued
    /// folder and stopping a running one. The runs behind it move up by having
    /// been behind it; there is no numbering to rewrite.
    pub fn dequeue(&self, store: &str) -> usize {
        let mut queue = self.queue.lock().unwrap_or_else(|e| e.into_inner());
        let mut dropped = 0;
        queue.retain(|pending| {
            if pending.store.name != store {
                return true;
            }
            let answer = serde_json::json!({
                "event": "done",
                "indexed": 0,
                "unchanged": 0,
                "skipped": 0,
                "removed": 0,
                "chunks": 0,
                "images": 0,
                "remaining": 0,
                "stopped": true,
                "dequeued": true,
            });
            pending.store.record(pending.run, &answer);
            let _ = pending.report.send(answer);
            dropped += 1;
            false
        });
        dropped
    }
}

/// One monotonic counter per data domain the portal reads.
///
/// This is the whole of what makes the portal live. A page polls
/// `/api/changes` once a second, compares six integers with what it last saw,
/// and refetches only the domains that moved — so a quiet daemon with a tab
/// open costs one small request a second and nothing else, and there is one
/// clock in the page rather than one per feature.
///
/// Process-global rather than a field of [`State`]: two of the six are bumped
/// by code that has no `State` in hand — the store's own event and log writers,
/// and the ledger's one record call — and threading a handle into them purely
/// to count would be a worse cost than a static. There is one daemon per
/// process by construction: it binds a port and holds every store's lock.
///
/// **A counter is bumped where its domain is written, once, and nowhere else.**
/// A second bump site is a second source of truth, and the failure it produces
/// is the worst one available — one page stale while the others are live, so
/// the portal is trusted everywhere and wrong in one place.
pub mod changes {
    use std::sync::atomic::{AtomicU64, Ordering};

    /// The domains a page can watch. Adding one means adding its bump site in
    /// the one function that writes it, and a row in the per-domain test.
    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
    pub enum Domain {
        Stores,
        Runs,
        Clients,
        Ledger,
        Events,
        Privacy,
    }

    pub const DOMAINS: [Domain; 6] = [
        Domain::Stores,
        Domain::Runs,
        Domain::Clients,
        Domain::Ledger,
        Domain::Events,
        Domain::Privacy,
    ];

    impl Domain {
        pub fn as_str(self) -> &'static str {
            match self {
                Self::Stores => "stores",
                Self::Runs => "runs",
                Self::Clients => "clients",
                Self::Ledger => "ledger",
                Self::Events => "events",
                Self::Privacy => "privacy",
            }
        }

        fn cell(self) -> &'static AtomicU64 {
            static STORES: AtomicU64 = AtomicU64::new(0);
            static RUNS: AtomicU64 = AtomicU64::new(0);
            static CLIENTS: AtomicU64 = AtomicU64::new(0);
            static LEDGER: AtomicU64 = AtomicU64::new(0);
            static EVENTS: AtomicU64 = AtomicU64::new(0);
            static PRIVACY: AtomicU64 = AtomicU64::new(0);
            match self {
                Self::Stores => &STORES,
                Self::Runs => &RUNS,
                Self::Clients => &CLIENTS,
                Self::Ledger => &LEDGER,
                Self::Events => &EVENTS,
                Self::Privacy => &PRIVACY,
            }
        }
    }

    /// Say that this domain was written.
    pub fn bump(domain: Domain) {
        domain.cell().fetch_add(1, Ordering::Relaxed);
    }

    /// Notice a registry that another process wrote.
    ///
    /// `semlith index ~/work/new` in a terminal writes the registry and this
    /// daemon opens that store on the next read of `/api/stores`, which is
    /// what 0.19.0's reconciliation is for. But the page only reads that route
    /// when the stores counter moves, and nothing *in this process* moved it —
    /// so the counter waited for the open and the open waited for the counter,
    /// and the row appeared on the next navigation rather than by itself.
    /// That circle is the whole of "my new repository is not in the list".
    ///
    /// One `stat` of the registry against the last one seen, on the poll that
    /// is already happening. Cheaper than reconciling per poll, which would
    /// take the registry lock and try to open every unopened store once a
    /// second; this only says "something changed", and the existing route does
    /// the work exactly once in response.
    pub fn notice_registry() {
        use std::sync::Mutex;
        use std::time::SystemTime;
        static SEEN: Mutex<Option<SystemTime>> = Mutex::new(None);

        let Ok(path) = crate::home::registry_path() else {
            return;
        };
        let Ok(at) = std::fs::metadata(&path).and_then(|m| m.modified()) else {
            return;
        };
        let mut seen = SEEN.lock().unwrap_or_else(|e| e.into_inner());
        match *seen {
            // The first poll is a baseline, not a change: whatever the page
            // drew on load already covers the registry as it stands.
            None => *seen = Some(at),
            Some(was) if was != at => {
                *seen = Some(at);
                bump(Domain::Stores);
            }
            Some(_) => {}
        }
    }

    /// Notice a root folder that has gone, or come back.
    ///
    /// `/api/stores` says per root whether it is on disk, and the Stores page
    /// badges a store whose corpus has gone — but only when it reads that
    /// route, which it does when the stores counter moves. Deleting a folder
    /// writes nothing to any store, so an open page kept drawing the root as
    /// present until something unrelated moved the counter.
    ///
    /// One `stat` per root on the poll that is already happening, compared
    /// with the set missing at the last poll. A store has one or two roots, so
    /// this is a handful of `stat`s a second on a daemon a page is watching,
    /// and nothing on one nobody is.
    pub fn notice_roots<'a>(roots: impl Iterator<Item = &'a std::path::Path>) {
        use std::sync::Mutex;
        static SEEN: Mutex<Option<Vec<std::path::PathBuf>>> = Mutex::new(None);

        let mut missing: Vec<std::path::PathBuf> = roots
            .filter(|root| !root.exists())
            .map(|root| root.to_path_buf())
            .collect();
        missing.sort();
        let mut seen = SEEN.lock().unwrap_or_else(|e| e.into_inner());
        match seen.as_ref() {
            // A baseline, as the registry's first poll is.
            None => *seen = Some(missing),
            Some(was) if *was != missing => {
                *seen = Some(missing);
                bump(Domain::Stores);
            }
            Some(_) => {}
        }
    }

    /// What each counter stands at.
    pub fn read(domain: Domain) -> u64 {
        domain.cell().load(Ordering::Relaxed)
    }
}

/// How many runs may be on at once, when the environment is what says so.
///
/// An explicit variable is an instruction from whoever started the process, so
/// it wins over the saved setting and the page cannot raise past it — the same
/// standing [`crate::embed::THREADS_ENV`] and [`crate::index::INDEX_MEMORY_ENV`]
/// already have over their own values.
pub const PARALLEL_ENV: &str = "SEMLITH_INDEX_PARALLEL";

/// Where a value in force came from.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum Source {
    Derived,
    Saved,
    Environment,
}

impl Source {
    fn as_str(self) -> &'static str {
        match self {
            Self::Derived => "derived",
            Self::Saved => "saved",
            Self::Environment => "set by the environment",
        }
    }
}

/// One of the three settings: what is in force, where it came from, and what
/// this machine would have chosen, with the one-line reason it chose it.
///
/// The derivation travels with the value rather than being recomputed by
/// whoever draws it, so the page's warning about a field set above the derived
/// value is comparing against the number the daemon actually derived.
#[derive(Debug, Clone, Serialize)]
pub struct Limit {
    pub value: usize,
    pub source: Source,
    pub derived: usize,
    pub reason: String,
    /// The most this machine will accept for this setting.
    ///
    /// Distinct from `derived`, and the distinction is the point. `derived` is
    /// what this machine would choose left alone, and going above it is an
    /// ordinary thing to want — a laptop that is doing nothing else can index
    /// harder than the default. `ceiling` is where the answer stops being a
    /// choice and starts being a machine that swaps or thrashes, and a page
    /// that let somebody past it would be offering a setting that makes the
    /// product worse with no way to tell.
    ///
    /// Enforced in the route as well as drawn on the page: a cap only the page
    /// knows about is not a cap.
    pub ceiling: usize,
}

impl Limit {
    fn new(
        derived: crate::system::Derivation,
        saved: Option<usize>,
        variable: &str,
        ceiling: usize,
    ) -> Self {
        let from_env = std::env::var(variable)
            .ok()
            .and_then(|raw| raw.parse::<usize>().ok())
            .filter(|n| *n > 0);
        // Never below what this machine would choose on its own. The memory
        // ceiling is what is free now less the reserve, which on a busy
        // machine is a few hundred MiB one second and nothing the next, and a
        // saved value clamped to it read "311 MiB" on one poll and "0 MiB" on
        // the next — beside a field that offered the derived 512 as in range.
        let ceiling = ceiling.max(derived.value);
        let (value, source) = match (from_env, saved.filter(|n| *n > 0)) {
            // The environment is the machine's owner speaking directly and is
            // not clamped: somebody who exported a variable has said what they
            // mean more deliberately than somebody typing in a box.
            (Some(n), _) => (n, Source::Environment),
            (None, Some(n)) => (n.min(ceiling), Source::Saved),
            (None, None) => (derived.value, Source::Derived),
        };
        Self {
            value,
            source,
            derived: derived.value,
            reason: derived.reason,
            ceiling,
        }
    }
}

/// The three settings in force, and the machine they were derived from.
#[derive(Debug, Clone, Serialize)]
pub struct Limits {
    pub runs_at_once: Limit,
    pub embed_threads: Limit,
    pub index_memory_mb: Limit,
    /// The reading the three were derived from, as the page shows it.
    ///
    /// Carried as JSON rather than by deriving `Serialize` on the reading
    /// itself: `system` answers a question about this machine and owes nothing
    /// to a wire format, and the one place that puts it on a wire is here.
    pub machine: serde_json::Value,
}

impl Limits {
    /// Read the machine, derive the three defaults, and let the saved settings
    /// and then the environment override them.
    ///
    /// Re-read rather than cached, because one of the numbers behind the
    /// derivation is how much memory is free *now* — a figure taken at start
    /// and shown an hour later is a figure about a machine that has since done
    /// other things.
    pub fn in_force() -> Self {
        let machine = crate::system::read();
        let derived = crate::system::derive(&machine, crate::system::PER_RUN_PEAK_MB);
        let saved = home::Settings::load();
        // The ceilings. Every logical core may carry a run, and every logical
        // core may carry an embedding thread, but nothing may ask for more
        // parallelism than the machine has hardware for. Memory stops at what
        // is free now less the reserve, because a store budget larger than
        // that is a promise the machine cannot keep.
        let cores = machine.logical_cores.max(1);
        let memory_ceiling = machine
            .available_memory_mb
            .saturating_sub(crate::system::RESERVE_MB) as usize;
        let runs_at_once = Limit::new(
            derived.runs_at_once,
            saved.runs_at_once,
            PARALLEL_ENV,
            cores,
        );
        // Derived from the runs actually in force, not from the runs this
        // machine would have chosen. The two differ exactly when someone has
        // changed the setting, which is the moment the sentence under the
        // field is being read.
        let threads_derived = crate::system::threads_for(&machine, runs_at_once.value);
        Self {
            runs_at_once,
            embed_threads: Limit::new(
                threads_derived,
                saved.embed_threads,
                crate::embed::THREADS_ENV,
                cores,
            ),
            index_memory_mb: Limit::new(
                derived.index_memory_mb,
                saved.index_memory_mb,
                crate::index::INDEX_MEMORY_ENV,
                memory_ceiling,
            ),
            machine: serde_json::json!({
                "logical_cores": machine.logical_cores,
                "physical_cores": machine.physical_cores,
                "total_memory_mb": machine.total_memory_mb,
                "available_memory_mb": machine.available_memory_mb,
                "reserve_mb": crate::system::RESERVE_MB,
                "per_run_peak_mb": crate::system::PER_RUN_PEAK_MB,
            }),
        }
    }

    /// Hand the values in force to the engine: the thread count every writer
    /// builds its session with, and the budget every open index fits inside.
    /// Before 0.28.0 two of the three were saved, shown and never applied.
    pub fn apply(&self) {
        // A derived count is left to follow the runs actually going; see
        // `embed::threads_in_force`.
        crate::embed::set_threads_in_force(match self.embed_threads.source {
            Source::Derived => 0,
            _ => self.embed_threads.value,
        });
        crate::index::set_budget_mb(self.index_memory_mb.value);
    }

    /// These limits with the memory figure the engine is actually using.
    ///
    /// [`Self::in_force`] re-derives from the memory free now, which is right
    /// for deciding what to apply and wrong for saying what was applied: the
    /// budget is handed to the engine when the daemon starts and when a
    /// setting is saved, and nowhere else, so a page that polls is told that.
    pub fn as_applied(mut self) -> Self {
        self.index_memory_mb.value = crate::index::budget_mb();
        self
    }

    /// The line `semlith start` prints: all three values with their source.
    pub fn line(&self) -> String {
        format!(
            "indexing: {} run(s) at once ({}), {} embedder thread(s) each ({}), {} MiB of vectors per store ({})",
            self.runs_at_once.value,
            self.runs_at_once.source.as_str(),
            self.embed_threads.value,
            self.embed_threads.source.as_str(),
            self.index_memory_mb.value,
            self.index_memory_mb.source.as_str(),
        )
    }
}

/// A run waiting for review: its store and the paths it will index.
type Awaiting = (Arc<Store>, Vec<PathBuf>);

/// What every route is handed.
pub struct State {
    pub server: Arc<Server>,
    /// One reader over every open store.
    ///
    /// ponytail: one lock for every read route. A search is milliseconds and
    /// there is one browser, so contention is theoretical; if it ever is not,
    /// the upgrade is a reader per store rather than a shared `Fleet`.
    pub fleet: Mutex<Option<Fleet>>,
    /// The open stores.
    ///
    /// Behind a lock because the set is no longer fixed at startup: indexing a
    /// folder on a machine with no store yet has to be able to make one and
    /// serve it in the same breath, without the developer restarting the
    /// daemon they only just started. Readers take a snapshot rather than hold
    /// the guard, so a slow search never blocks a store being opened.
    stores: RwLock<Vec<Arc<Store>>>,
    /// The writer threads of stores opened after startup, joined at shutdown
    /// with the startup ones. They were detached until 0.33.0, so a stop while
    /// one was embedding let `main` return and tear ONNX Runtime down under a
    /// session still running, which printed ORT's own error after `stopped`
    /// (#163).
    late_writers: Mutex<Vec<std::thread::JoinHandle<()>>>,
    /// The one queue in front of every store's writer, and the only thing that
    /// decides how many runs are on at once.
    pub admission: Arc<Admission>,
    pub airgap: bool,
    pub started: SystemTime,
    /// How long a watcher waits for a file to stop changing. Kept so a store
    /// opened later gets the same debounce as the ones opened at startup.
    debounce: Duration,
    /// The daemon's log line, so a store opened at runtime reports itself the
    /// way the startup ones do.
    report: Arc<dyn Fn(&str) + Send + Sync>,
    /// Refusals by class, for the line the daemon logs on shutdown.
    pub refusals: Mutex<BTreeMap<&'static str, u64>>,
    /// Runs scanned and held for a person's review, by run id (2.7).
    awaiting: Mutex<BTreeMap<u64, Awaiting>>,
    /// `semlith mcp` processes forwarding here: pid to the unix second it was
    /// last heard from. A proxy has no disconnect to observe — its client may
    /// simply stop asking — so recency is the only honest answer to "how many
    /// are connected".
    pub proxies: Mutex<BTreeMap<u32, u64>>,
    /// What each connected MCP client said it was, keyed by session.
    ///
    /// Held here and lost when the daemon exits, which is the honest lifetime:
    /// there is no disconnect to observe over HTTP either, so a client is
    /// "connected" for as long as it has been heard from recently.
    pub clients: Mutex<BTreeMap<String, Client>>,
    /// A second reader, for forwarded MCP calls, opened on first use.
    ///
    /// Separate from `fleet` on purpose: a forwarded `semlith_index` blocks its
    /// caller until the writer has run it, and sharing one lock would mean an
    /// agent's index freezing the portal for as long as the slice lasts. A
    /// reader that is never used costs a SQLite handle and no vectors.
    pub mcp_fleet: Mutex<Option<Fleet>>,
    /// Whether retrievals are recorded into each store's `retrievals` table.
    ///
    /// On unless `--no-ledger` or `SEMLITH_LEDGER=0` says otherwise, which
    /// reverses what 0.12.0 to 0.14.0 did. The principle those releases stated
    /// was that a local tool which starts logging without being told to is not
    /// different from one that phones home; the principle that actually holds
    /// is narrower and checkable:
    ///
    /// - the rows never leave the store they were written into, which is still
    ///   provable with a packet capture;
    /// - the daemon says on every start that it is recording, and names the
    ///   flag that stops it;
    /// - erasing every row is one `DELETE`.
    ///
    /// What the old default cost was the thing the record is for. A ledger
    /// nobody switched on measured nobody, so the savings figure had no
    /// denominator and the audit trail had no rows — and the one client whose
    /// retrievals it did record was the portal's own search box, which is not
    /// who the product is for.
    pub ledger: bool,
    /// The schedules timer's handle.
    ///
    /// Here rather than in a global so a route can reach it: a surface that has
    /// changed `schedules.json` calls `state.schedules.wake()` and the timer
    /// re-reads it at once instead of waiting out its sleep.
    pub schedules: Arc<crate::schedule::Runner>,
    /// The final cards of runs whose store was deleted by their own stop, so
    /// the page can say the store went rather than the card vanishing.
    pub gone: Mutex<Vec<serde_json::Value>>,
}

/// How recently a proxy must have called to count as connected.
const PROXY_FRESH: u64 = 120;

/// Each open store's directory beside the name the daemon knows it by.
///
/// The fleet derives a label from the store directory's basename, which is the
/// registered name only for a store living in the store home. Everywhere else
/// the two differ, and the portal — which draws its chips from `/api/stores` —
/// ended up naming stores the search route said were not open.
fn named(stores: &[Arc<Store>]) -> Vec<(PathBuf, String)> {
    stores
        .iter()
        .map(|s| (s.dir.clone(), s.name.clone()))
        .collect()
}

/// One MCP client, as the Agents page shows it.
#[derive(Debug, Clone, serde::Serialize)]
pub struct Client {
    /// Which client this is: one of the documented names where `clientid`
    /// can tell (from what the client called itself in `initialize` and the
    /// app that started its proxy), else what it called itself, else
    /// "unnamed client". Invented names are worse than an honest one.
    pub name: String,
    /// `clientInfo.version`, empty when the client gave none.
    pub version: String,
    pub transport: String,
    /// The protocol revision that was negotiated, as the client asked for it.
    pub revision: String,
    /// How many tool calls it has made through this daemon.
    pub queries: u64,
    pub seen: u64,
}

/// A registered store the daemon could not open, and why.
///
/// Reported on the Stores page rather than swallowed: a store that is in the
/// registry and not in the list is a store the user will look for and not
/// find, and "another process is writing it" and "it is not there any more"
/// are two different things to do about it.
#[derive(Debug, Clone)]
pub struct Unopened {
    pub name: String,
    pub dir: PathBuf,
    pub why: String,
    /// Whether nothing is at that path at all, as against a store that is
    /// there and could not be opened.
    ///
    /// The portal draws the two differently and must: a store another process
    /// is writing comes back by itself on the next read, and a registry entry
    /// for a directory that does not exist never will. Offering "add to
    /// alpha" for the second is the same mechanism that put 262 of one store's
    /// files into another.
    pub missing: bool,
}

impl State {
    /// Every open store, as a snapshot.
    ///
    /// Cloned out rather than handed back under the guard: an `Arc` clone is a
    /// counter bump, and holding a read lock across a search would stop a new
    /// store being opened for as long as the search took.
    pub fn stores(&self) -> Vec<Arc<Store>> {
        self.stores.read().expect("the stores lock").clone()
    }

    /// A store by name, opening it from the registry if this daemon has not
    /// got it open yet.
    ///
    /// The miss is what triggers the reconciliation, not a timer: `semlith
    /// index ~/work/new-project` from a second process writes a store this
    /// daemon has never heard of, and until 0.19.0 every agent asking for it
    /// by name got "no store called new-project is open" until the daemon was
    /// restarted.
    pub fn store(self: &Arc<Self>, name: &str) -> Option<Arc<Store>> {
        if let Some(open) = self.stores().into_iter().find(|s| s.name == name) {
            return Some(open);
        }
        self.reconcile();
        self.stores().into_iter().find(|s| s.name == name)
    }

    /// Open every registered store this daemon does not have open, and report
    /// the ones it could not.
    ///
    /// Called on a by-name miss and on every `/api/stores` read, which is the
    /// cheap place to notice: both are already the slow path, and neither runs
    /// in the indexing loop.
    ///
    /// A directory another process holds the lock on is left alone and
    /// reported as being written, never forced: `open_store` takes the lock
    /// first, so the daemon cannot become a second writer of one store. The
    /// next read tries again, which is what makes a `semlith index` that is
    /// still running appear by itself when it finishes.
    pub fn reconcile(self: &Arc<Self>) -> Vec<Unopened> {
        let Ok(registry) = Registry::load() else {
            return Vec::new();
        };
        let open: Vec<PathBuf> = self.stores().iter().map(|s| s.dir.clone()).collect();
        let mut out = Vec::new();
        for name in registry.stores.keys() {
            let Ok(dir) = Registry::dir_of(name) else {
                continue;
            };
            let dir = crate::canonical(&dir);
            if open.contains(&dir) {
                continue;
            }
            if !dir.exists() {
                out.push(Unopened {
                    name: name.clone(),
                    dir,
                    why: "registered, but there is nothing at that path".to_string(),
                    missing: true,
                });
                continue;
            }
            if let Err(e) = self.open_store(&dir, false) {
                out.push(Unopened {
                    name: name.clone(),
                    dir,
                    why: format!("{e:#}"),
                    missing: false,
                });
            }
        }
        out
    }

    /// The one store a write means, or an error naming the alternatives.
    pub fn writable(self: &Arc<Self>, name: Option<&str>) -> Result<Arc<Store>> {
        // A write naming a store this daemon has not opened is the same miss
        // as a read naming one, and gets the same reconciliation.
        if let Some(name) = name
            && let Some(open) = self.store(name)
        {
            return Ok(open);
        }
        if name.is_none() && self.stores().is_empty() {
            self.reconcile();
        }
        let stores = self.stores();
        match (name, stores.as_slice()) {
            (Some(name), _) => self
                .store(name)
                .with_context(|| format!("no store called {name} is open")),
            (None, [one]) => Ok(Arc::clone(one)),
            (None, []) => bail!("no store is open"),
            (None, many) => bail!(
                "this daemon has {} stores open, so a write has to name one: {}",
                many.len(),
                many.iter()
                    .map(|s| s.name.clone())
                    .collect::<Vec<_>>()
                    .join(", ")
            ),
        }
    }

    /// Queue an index run, and hand back its id and the channel it reports on.
    ///
    /// The run goes to the daemon-wide admission queue rather than straight to
    /// the store's writer, so how many are on at once is a property of the
    /// machine rather than of how many stores happen to be open. A caller that
    /// wants the events reads the receiver; a caller that only wanted the run
    /// started drops it, and the run carries on — its progress is in the
    /// store's snapshot either way.
    pub fn index(
        &self,
        store: &Arc<Store>,
        paths: Vec<PathBuf>,
    ) -> Result<(u64, mpsc::Receiver<serde_json::Value>)> {
        Self::writer_alive(store)?;
        Ok(self.admission.submit(store, paths, RunKind::Run))
    }

    /// Queue a re-index run of the given kind (0.35.0's Re-index and
    /// Re-index files), admitted like any other run.
    pub fn reindex(&self, store: &Arc<Store>, paths: Vec<PathBuf>, kind: RunKind) -> Result<u64> {
        Self::writer_alive(store)?;
        Ok(self.admission.submit(store, paths, kind).0)
    }

    /// The scan phase for `paths` into `store`, without the writer or the
    /// model: a read-only open of the store beside the writer's.
    pub fn plan(&self, store: &Arc<Store>, paths: &[PathBuf]) -> Result<crate::Plan> {
        let mut reader = crate::Semlith::open_existing(&store.dir)?;
        reader.gitignore = store.gitignore.load(Ordering::Relaxed);
        reader.plan(paths)
    }

    /// Start a run with its scan phase (2.7): the plan goes on the card, and
    /// when `review` is asked for and the scan found something that is a
    /// person's to decide, the run holds at review rather than queue. `hold`
    /// holds it after any scan, clean or not — the portal's Scan button, whose
    /// second press is Start indexing. Nothing else waits: other runs and the
    /// watcher go on.
    pub fn index_planned(
        &self,
        store: &Arc<Store>,
        paths: Vec<PathBuf>,
        review: bool,
        hold: bool,
    ) -> Result<u64> {
        Self::writer_alive(store)?;
        // A run nobody reviews starts at once. Its plan — the card's summary
        // and first estimate — is read beside it, from a snapshot of the store
        // taken before the run is queued, so it describes the store the run
        // started from. Before 0.32.0 every run waited for a scan of every
        // file just to put the plan on its card, which on the benchmark corpus
        // was twenty seconds before the first row.
        if !review && !hold {
            let (ready, pinned) = mpsc::channel::<()>();
            let (give_run, run_id) = mpsc::channel::<u64>();
            // Weak: the store's lock goes with its last strong reference, and
            // a plan still reading must not keep a deleted store's lock held
            // against the same folder indexed again at once.
            let for_plan = Arc::downgrade(store);
            let dir = store.dir.clone();
            let plan_paths = paths.clone();
            let gitignore = store.gitignore.load(Ordering::Relaxed);
            std::thread::Builder::new()
                .name("semlith-plan".to_string())
                .spawn(move || {
                    let reader = crate::Semlith::open_existing(&dir).and_then(|mut reader| {
                        reader.gitignore = gitignore;
                        reader.pin_snapshot()?;
                        Ok(reader)
                    });
                    let _ = ready.send(());
                    let Ok(mut reader) = reader else { return };
                    let Ok(run) = run_id.recv() else { return };
                    if let Ok(plan) = reader.plan(&plan_paths)
                        && let Some(store) = for_plan.upgrade()
                    {
                        store.set_plan(run, &plan, false);
                    }
                    reader.release_snapshot();
                })
                .context("starting the plan beside the run")?;
            let _ = pinned.recv_timeout(Duration::from_secs(10));
            let (run, _progress) = self.admission.submit(store, paths, RunKind::Run);
            let _ = give_run.send(run);
            return Ok(run);
        }
        let plan = self.plan(store, &paths).ok();
        if (review || hold)
            && let Some(plan) = plan.as_ref().filter(|p| hold || !p.review.is_empty())
        {
            let run = self.admission.mint();
            store.begin_run(run, paths.clone(), RunKind::Run);
            store.set_plan(run, plan, true);
            // What the store held before, as a started run reports it: a
            // held run never starts, and Discard scan decides from this
            // whether the scan made the store and should take it away.
            if let Ok((files, chunks, _)) =
                crate::Semlith::open_existing(&store.dir).and_then(|reader| reader.stats())
            {
                store.set_before(run, files.max(0) as u64, chunks.max(0) as u64);
            }
            self.awaiting
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .insert(run, (Arc::clone(store), paths));
            return Ok(run);
        }
        let (run, _progress) = self.admission.submit(store, paths, RunKind::Run);
        if let Some(plan) = &plan {
            store.set_plan(run, plan, false);
        }
        Ok(run)
    }

    /// Queue a run a person has reviewed. Whatever they left undecided stays
    /// refused and on the Not indexed list.
    pub fn start_reviewed(&self, run: u64) -> Result<()> {
        let Some((store, paths)) = self
            .awaiting
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .remove(&run)
        else {
            anyhow::bail!("run {run} is not waiting for review");
        };
        store.leave_review(run, RunStatus::Queued);
        self.admission.enqueue(run, &store, paths, RunKind::Run);
        Ok(())
    }

    /// Drop a run that was waiting for review, as a stop of it.
    pub fn drop_reviewed(&self, run: u64) -> bool {
        let taken = self
            .awaiting
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .remove(&run);
        match taken {
            Some((store, _)) => store.leave_review(run, RunStatus::Stopped),
            None => false,
        }
    }

    pub fn forget(
        &self,
        store: &Arc<Store>,
        path: PathBuf,
    ) -> Result<mpsc::Receiver<serde_json::Value>> {
        Self::writer_alive(store)?;
        // No notice: this caller takes the first message as the answer.
        Ok(store.submit(Job::Forget(path), None))
    }

    /// Queue a compaction of `store` as a run with its own card, behind
    /// whatever its writer is already doing.
    pub fn compact(
        &self,
        store: &Arc<Store>,
        options: crate::compact::CompactOptions,
    ) -> Result<(u64, mpsc::Receiver<serde_json::Value>)> {
        Self::writer_alive(store)?;
        let run = self.admission.next.fetch_add(1, Ordering::Relaxed);
        store.begin_run(run, vec![store.dir.clone()], RunKind::Compact);
        Ok((run, store.submit_run(run, Job::Compact(options))))
    }

    /// Accept one refused file, at the front of the writer's queue: a person
    /// waiting on a confirm should not wait behind a long run's slices.
    pub fn accept(
        &self,
        store: &Arc<Store>,
        path: PathBuf,
        mode: &str,
        source: &str,
    ) -> Result<mpsc::Receiver<serde_json::Value>> {
        Self::writer_alive(store)?;
        Ok(store.submit_front(Job::Accept {
            path,
            mode: mode.to_string(),
            source: source.to_string(),
        }))
    }

    /// Undo an acceptance: the file leaves the store and returns to the list.
    pub fn revoke(
        &self,
        store: &Arc<Store>,
        path: PathBuf,
    ) -> Result<mpsc::Receiver<serde_json::Value>> {
        Self::writer_alive(store)?;
        Ok(store.submit_front(Job::Revoke(path)))
    }

    /// Close a store and delete everything it holds.
    ///
    /// The order is what makes this safe while the daemon runs: the watcher is
    /// told to stop and its lock is released when the thread returns, the
    /// store leaves the served set so no request can reach it, the readers are
    /// dropped so no SQLite handle is still open on the files, and only then
    /// is the directory removed and the registry entry dropped.
    ///
    /// The indexed files themselves are not touched — this deletes what
    /// semlith derived from them.
    pub fn delete_store(self: &Arc<Self>, name: &str) -> Result<PathBuf> {
        let Some(store) = self.store(name) else {
            anyhow::bail!("this daemon is not serving a store called {name}");
        };

        store.stop.store(true, Ordering::SeqCst);
        // The watcher checks the flag between filesystem events, so this is a
        // wait of one debounce, not of one filesystem event.
        let deadline = std::time::Instant::now() + Duration::from_secs(10);
        while store.watching.load(Ordering::Relaxed) {
            if std::time::Instant::now() > deadline {
                anyhow::bail!(
                    "{name}'s watcher did not stop, so its lock is still held; \
                     nothing was deleted"
                );
            }
            std::thread::sleep(Duration::from_millis(50));
        }

        self.stores
            .write()
            .expect("the stores lock")
            .retain(|s| s.name != name);
        // Before the files go: a reader holding this store open would keep its
        // SQLite handles alive, and on Windows an open handle refuses the
        // delete outright.
        self.reopen_readers();
        Discovery::remove(&store.dir);

        // The registry entry goes before the bump: a client that re-reads
        // /api/stores on the bump must not still see the store, and no further
        // bump would come to correct it.
        let removed = crate::home::delete_store(name);
        changes::bump(changes::Domain::Stores);
        removed
    }

    /// Make an empty named store and serve it at once (0.35.0).
    pub fn create_store(self: &Arc<Self>, name: &str, kind: &str) -> Result<Arc<Store>> {
        let dir = crate::home::create_store(name, kind)?;
        self.open_store(&dir, false)
    }

    /// Take a store's changed settings into the running daemon: the live
    /// flags, the ledger's list, and readers rebuilt so the next search
    /// reads the new `lean`. A `watch` turned back on queues a catch-up,
    /// because what changed while it was off was dropped, not deferred.
    pub fn settings_changed(self: &Arc<Self>, store: &Arc<Store>) {
        let settings = crate::home::store_settings(&store.dir);
        let was_watching = store.watch_events.load(Ordering::Relaxed);
        store.apply_settings(&settings);
        if settings.watch && !was_watching && !store.watched.is_empty() {
            let _ = self
                .admission
                .submit(store, store.watched.clone(), RunKind::CatchUp);
        }
        refresh_unrecorded();
        self.reopen_readers();
    }

    /// Rename a store while the daemon runs: close it as a delete would,
    /// move its directory and registry key together, and open it again
    /// under the new name. An agent's next `store: "<new>"` resolves; the
    /// old name stops resolving the moment the store leaves the served set.
    ///
    /// Refused while it has a run going or queued: the run's card, its queue
    /// entry and its undo all hold the store by its old record.
    pub fn rename_store(self: &Arc<Self>, old: &str, new: &str) -> Result<Arc<Store>> {
        let Some(store) = self.store(old) else {
            bail!("this daemon is not serving a store called {old}");
        };
        crate::home::check_store_name(new)?;
        if Registry::load()?.stores.contains_key(new) {
            bail!("“{new}” is already a store on this machine. Pick another name.");
        }
        if store.run_live() || self.admission.position_of(old).is_some() || store.queue_depth() > 0
        {
            bail!("{old} is indexing; stop or wait for its run, then rename it");
        }
        store.stop.store(true, Ordering::SeqCst);
        let deadline = std::time::Instant::now() + Duration::from_secs(10);
        while store.watching.load(Ordering::Relaxed) {
            if std::time::Instant::now() > deadline {
                bail!("{old}'s writer did not stop, so nothing was renamed");
            }
            std::thread::sleep(Duration::from_millis(50));
        }
        self.stores
            .write()
            .expect("the stores lock")
            .retain(|s| s.name != old);
        self.reopen_readers();
        Discovery::remove(&store.dir);
        let dir = match crate::home::rename_store(old, new) {
            Ok(dir) => dir,
            Err(e) => {
                // Serve it again under the name it still has.
                let _ = self.reopen_after_stop(&store.dir);
                return Err(e);
            }
        };
        drop(store);
        // `open_store` bumps the stores counter; the registry write the
        // settings made is noticed by `changes::notice_registry`.
        self.reopen_after_stop(&dir)
    }

    /// Open a store whose writer was just told to stop. The thread drops the
    /// store's lock a moment after it clears `watching`, so the first try can
    /// still find it held.
    fn reopen_after_stop(self: &Arc<Self>, dir: &Path) -> Result<Arc<Store>> {
        let mut last = None;
        for _ in 0..40 {
            match self.open_store(dir, false) {
                Ok(store) => return Ok(store),
                Err(e) => last = Some(e),
            }
            std::thread::sleep(Duration::from_millis(50));
        }
        Err(last.unwrap_or_else(|| anyhow::anyhow!("could not reopen {}", dir.display())))
    }

    /// Stop a store's run and, once its undo has finished, delete the store.
    ///
    /// On its own thread, because the undo runs on the store's writer and the
    /// delete has to wait for that writer to stop. The card keeps its record
    /// in `gone`, so it says the store was deleted rather than vanishing.
    pub fn delete_after_stop(self: &Arc<Self>, store: Arc<Store>) {
        let state = Arc::clone(self);
        std::thread::spawn(move || {
            let deadline = std::time::Instant::now() + Duration::from_secs(600);
            while store.run_live() || state.admission.position_of(&store.name).is_some() {
                if std::time::Instant::now() > deadline {
                    store.note(
                        "the stop did not finish in ten minutes; the store was not deleted".into(),
                    );
                    return;
                }
                std::thread::sleep(Duration::from_millis(100));
            }
            let run = store.last_run();
            let outcome = state.delete_store(&store.name);
            let (event, text) = match &outcome {
                Ok(dir) => (
                    "deleted",
                    format!(
                        "the store was deleted ({})",
                        crate::plain(&dir.display().to_string())
                    ),
                ),
                Err(e) => ("delete-failed", format!("the store was not deleted: {e:#}")),
            };
            (state.report)(&format!("{}: {text}", store.name));
            if let Some(run) = run {
                store.note_run(run, &serde_json::json!({ "event": event, "text": text }));
                let snapshots = store.run_snapshots(None);
                if outcome.is_ok() {
                    state.gone_changed(|gone| {
                        gone.extend(snapshots.into_iter().filter(|s| {
                            s.get("id").and_then(serde_json::Value::as_u64) == Some(run)
                        }))
                    });
                }
            }
        });
    }

    /// Change the cards of runs whose store is gone, and say the runs domain
    /// moved. The one place those cards are written.
    pub fn gone_changed<T>(&self, change: impl FnOnce(&mut Vec<serde_json::Value>) -> T) -> T {
        let out = change(&mut self.gone.lock().unwrap_or_else(|e| e.into_inner()));
        changes::bump(changes::Domain::Runs);
        out
    }

    /// Refuse to queue work for a store whose writer is gone.
    ///
    /// The queue is drained by the watcher thread and by nothing else, so a
    /// job submitted after that thread has returned would never be performed
    /// and never be answered — its caller would hold an HTTP worker open until
    /// the daemon stopped. A watcher only returns on a backend failure it has
    /// already reported, so this is a real condition, not a theoretical one.
    fn writer_alive(store: &Arc<Store>) -> Result<()> {
        if store.watching.load(Ordering::Relaxed) {
            return Ok(());
        }
        bail!(
            "the writer for {} is not running, so it cannot be written to; \
             the daemon's stderr says why, and restarting it is the fix",
            store.name
        )
    }

    /// Rotate the token and tell every discovery file about it, so a
    /// forwarding `semlith mcp` is not cut off by the Privacy page's button.
    pub fn rotate(&self) -> String {
        let fresh = self.server.rotate();
        for store in self.stores() {
            let _ = discovery(self.server.port(), &fresh).write(&store.dir);
        }
        fresh
    }

    /// Record what a client said about itself, and count its tool calls.
    ///
    /// `session` is how one client is told from another: over HTTP it is the
    /// `Mcp-Session-Id` this daemon hands out at `initialize`, and for a
    /// forwarding `semlith mcp` it is the proxy's pid. A client that echoes
    /// neither is counted as one unnamed client per transport rather than as a
    /// new one on every request.
    /// The name a session gave itself when it initialized.
    ///
    /// The MCP handshake carries `clientInfo` once, and every later request on
    /// that session carries none — so the name the ledger records has to come
    /// from what was noted at the handshake rather than from the request in
    /// hand.
    /// The name and version a session's rows are recorded under.
    pub fn client_name(&self, session: &str, transport: &str) -> Option<(String, String)> {
        let clients = self.clients.lock().unwrap_or_else(|e| e.into_inner());
        clients
            .get(&format!("{transport}:{session}"))
            .map(|client| (client.name.clone(), client.version.clone()))
            .filter(|(name, _)| name != "unnamed client")
    }

    #[allow(clippy::too_many_arguments)]
    pub fn note_client(
        &self,
        session: &str,
        transport: &str,
        name: Option<&str>,
        version: Option<&str>,
        host: Option<&str>,
        revision: Option<&str>,
        query: bool,
    ) {
        let now = now();
        let mut clients = self.clients.lock().unwrap_or_else(|e| e.into_inner());
        let entry = clients
            .entry(format!("{transport}:{session}"))
            .or_insert_with(|| Client {
                name: "unnamed client".to_string(),
                version: String::new(),
                transport: transport.to_string(),
                revision: revision.unwrap_or("—").to_string(),
                queries: 0,
                seen: now,
            });
        // The handshake names the client once; the host arrives on every
        // proxied request, so a session first heard mid-conversation (a
        // daemon restarted under it) is still told apart.
        if name.is_some() || (host.is_some() && entry.name == "unnamed client") {
            entry.name = crate::clientid::label(name, host);
        }
        if let Some(version) = version.filter(|_| !name.is_some_and(crate::clientid::generic)) {
            entry.version = version.chars().take(64).collect();
        }
        if let Some(revision) = revision {
            entry.revision = revision.to_string();
        }
        if query {
            entry.queries += 1;
        }
        entry.seen = now;
        clients.retain(|_, client| now.saturating_sub(client.seen) <= PROXY_FRESH);
        drop(clients);
        Self::clients_changed();
    }

    /// The one place the clients domain is bumped, for both of its writers.
    fn clients_changed() {
        changes::bump(changes::Domain::Clients);
    }

    /// Every client heard from recently.
    pub fn clients(&self) -> Vec<Client> {
        let now = now();
        self.clients
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .values()
            .filter(|client| now.saturating_sub(client.seen) <= PROXY_FRESH)
            .cloned()
            .collect()
    }

    /// Note that a forwarding `semlith mcp` is alive.
    pub fn saw_proxy(&self, pid: u32) {
        let mut proxies = self.proxies.lock().unwrap_or_else(|e| e.into_inner());
        let now = now();
        proxies.insert(pid, now);
        proxies.retain(|_, seen| now.saturating_sub(*seen) <= PROXY_FRESH);
    }

    /// A session that has ended: its row and, for a proxy, its pid go now.
    pub fn forget_client(&self, session: &str, transport: &str, proxy: Option<u32>) {
        self.clients
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .remove(&format!("{transport}:{session}"));
        if let Some(pid) = proxy {
            self.proxies
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .remove(&pid);
        }
        Self::clients_changed();
    }

    /// How many `semlith mcp` processes are currently forwarding here.
    pub fn proxy_count(&self) -> usize {
        let now = now();
        self.proxies
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .values()
            .filter(|seen| now.saturating_sub(**seen) <= PROXY_FRESH)
            .count()
    }

    /// The reader forwarded MCP calls answer from, opened on first use.
    pub fn open_mcp_fleet(&self) -> Result<()> {
        let mut fleet = self.mcp_fleet.lock().unwrap_or_else(|e| e.into_inner());
        if fleet.is_some() {
            return Ok(());
        }
        let open = self.stores();
        let dirs: Vec<PathBuf> = open.iter().map(|s| s.dir.clone()).collect();
        if dirs.is_empty() {
            bail!("this daemon has no store open");
        }
        let mut opened = Fleet::open(&dirs)?;
        // An agent scoping a tool call with `store:` names it the way the
        // portal and `/api/stores` do. See `open_fleet`.
        opened.name_from(&named(&open));
        opened.quiet = true;
        *fleet = Some(opened);
        Ok(())
    }

    /// Open a store that was not open when the daemon started, and serve it
    /// immediately.
    ///
    /// This is what makes the portal's first run work. A machine with nothing
    /// indexed starts the daemon with no stores, and the Index page then has
    /// nothing to write to — so indexing a folder there has to be able to
    /// create the store, take its lock, start watching it and put it in front
    /// of the reader without the developer restarting anything.
    ///
    /// The order is the same one `run` uses and matters for the same reason:
    /// the lock is taken before the store is announced, so nothing is offered
    /// that another process might already own.
    pub fn open_store(self: &Arc<Self>, dir: &Path, expect_run: bool) -> Result<Arc<Store>> {
        let dir = crate::canonical(dir);
        if let Some(open) = self.stores().into_iter().find(|s| s.dir == dir) {
            return Ok(open);
        }

        let lock = StoreLock::acquire(&dir)
            .with_context(|| format!("{} cannot be opened by the daemon", dir.display()))?;

        let registry = Registry::load()?;
        let (name, roots) = roots_for(&dir, &registry);
        let watched: Vec<PathBuf> = roots.iter().filter(|r| r.exists()).cloned().collect();

        let expecting = if expect_run {
            now() as usize + EXPECTED_RUN_GRACE_SECS
        } else {
            0
        };
        let store = Arc::new(Store::new(
            name.clone(),
            dir.clone(),
            roots,
            watched,
            true,
            expecting,
        ));
        // Before the writer thread exists, so its first catch-up already
        // knows whether this store is watched.
        store.apply_settings(&crate::home::store_settings(&dir));

        discovery(self.server.port(), &self.server.token()).write(&store.dir)?;
        self.stores
            .write()
            .expect("the stores lock")
            .push(Arc::clone(&store));

        // Same shape as the startup watchers: the lock moves into the thread so
        // its life is the thread's life, which is what makes "the daemon is the
        // writer" true rather than intended.
        let watching = Arc::clone(&store);
        let debounce = self.debounce;
        let report = Arc::clone(&self.report);
        let admission = Arc::clone(&self.admission);
        let writer = std::thread::spawn(move || {
            let _lock = lock;
            keep_writing(&watching, debounce, &*report, &admission);
        });
        let mut late = self.late_writers.lock().unwrap_or_else(|e| e.into_inner());
        late.retain(|handle| !handle.is_finished());
        late.push(writer);

        (self.report)(&format!(
            "opened {name} at {} — now serving it",
            dir.display()
        ));
        self.reopen_readers();
        changes::bump(changes::Domain::Stores);
        Ok(store)
    }

    /// Throw away the readers so the next request builds one that knows about
    /// every store, including any opened since.
    ///
    /// Dropped rather than rebuilt here: rebuilding loads the embedding model,
    /// and doing that while holding the lock would stall whichever request
    /// happened to be next. The reader is opened on demand anyway.
    fn reopen_readers(&self) {
        *self.fleet.lock().unwrap_or_else(|e| e.into_inner()) = None;
        *self.mcp_fleet.lock().unwrap_or_else(|e| e.into_inner()) = None;
    }

    /// The reader every read route answers from, opened on first use and
    /// reopened after a store joins.
    pub fn open_fleet(&self) -> Result<()> {
        let mut fleet = self.fleet.lock().unwrap_or_else(|e| e.into_inner());
        if fleet.is_some() {
            return Ok(());
        }
        let open = self.stores();
        // A store whose directory has gone since the daemon opened it — it was
        // deleted by another process, or the volume it sat on was unmounted —
        // is left out rather than taken as a reason to answer nothing. It used
        // to fail the whole call, so one missing directory made `/api/stores`
        // answer 500 for every store on the machine and the portal went blank.
        let open: Vec<Arc<Store>> = open
            .into_iter()
            .filter(|s| s.dir.join("store.db").exists())
            .collect();
        let dirs: Vec<PathBuf> = open.iter().map(|s| s.dir.clone()).collect();
        if dirs.is_empty() {
            return Ok(());
        }
        let mut opened = Fleet::open(&dirs)?;
        // The daemon already knows what each store is called — it is what
        // `/api/stores` reports and what the portal's chips are made of. Handing
        // those names to the fleet is what stops the two from disagreeing, and
        // a chip from naming a store the search route says is not open.
        opened.name_from(&named(&open));
        opened.quiet = true;
        *fleet = Some(opened);
        Ok(())
    }

    /// Put a line on the daemon's own log.
    ///
    /// The `report` closure is private because nothing outside this module
    /// should be choosing what a daemon says; the schedules timer is a thread
    /// this daemon owns and has failures only its log can carry, so it gets a
    /// way to say them rather than a way to hold the closure.
    pub fn say(&self, text: &str) {
        (self.report)(text);
    }

    pub fn refuse(&self, class: Refusal) {
        *self
            .refusals
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .entry(class.as_str())
            .or_insert(0) += 1;
    }
}

/// Tell the ledger which registered stores say `record: false`.
pub fn refresh_unrecorded() {
    let registry = Registry::load().unwrap_or_default();
    crate::ledger::set_unrecorded(
        registry
            .stores
            .iter()
            .filter(|(_, entry)| !entry.settings.record)
            .filter_map(|(name, _)| Registry::dir_of(name).ok())
            .collect(),
    );
}

fn discovery(port: u16, token: &str) -> Discovery {
    Discovery {
        pid: std::process::id(),
        port,
        token: token.to_string(),
        version: env!("CARGO_PKG_VERSION").to_string(),
        started: started_at(),
    }
}

/// The unix second this process first wrote a discovery file, which is its
/// start: the first write happens as `run` opens its stores, and every later
/// one (a store added, a key rotated) must not move it.
fn started_at() -> u64 {
    static STARTED: std::sync::OnceLock<u64> = std::sync::OnceLock::new();
    *STARTED.get_or_init(|| {
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_secs())
            .unwrap_or(0)
    })
}

/// The stores `semlith start` should open.
///
/// With no arguments: every registered store. With paths: whatever each path
/// resolves to, so `semlith start ~/work/api` opens the api store whether it
/// lives in the home or beside the corpus.
pub fn stores_to_open(flags: &[PathBuf], paths: &[PathBuf], cwd: &Path) -> Result<Vec<PathBuf>> {
    if !flags.is_empty() {
        return Ok(flags.to_vec());
    }
    if paths.is_empty() {
        return home::all_dirs(&[], cwd);
    }
    let mut out = Vec::new();
    for path in paths {
        let dir = home::resolve(&[], path, None)?.one()?;
        if !dir.join("store.db").exists() {
            bail!(
                "{} has no store yet — run `semlith index {}` first",
                path.display(),
                path.display()
            );
        }
        if !out.contains(&dir) {
            out.push(dir);
        }
    }
    Ok(out)
}

/// Split `dirs` into the ones a live daemon already holds and the ones free to
/// open, pairing each held store with the daemon that holds it.
///
/// A store counts as held only when both halves agree. The write lock has to
/// refuse — so something really is the writer — *and* the discovery file beside
/// it has to name a daemon that will answer. Either alone is a guess: a lock
/// with no discovery is an `index` or `watch` run, and a discovery file with no
/// lock is what a SIGKILLed daemon left behind.
///
/// The live half is `Discovery::read` rather than `lock::daemon_holds`, which
/// answers from the file existing. The criterion here is a daemon that is
/// *answering*, and `read` is the only predicate that checks all of it: the
/// registry trusts the store, the file is this user's and 0600, the token is a
/// token, and the pid is alive. A stale file passes `daemon_holds` and would
/// turn a genuine conflict into a cheerful "already running" pointing at a URL
/// nothing listens on.
///
/// The probe takes each free store's lock and drops it again. The window
/// between that and `run`'s own acquire is harmless both ways: a daemon that
/// exits inside it hands `run` the lock, and one that starts inside it hands
/// `run` the refusal a genuine conflict deserves.
pub fn held_by_daemon(dirs: &[PathBuf]) -> (Vec<(PathBuf, Discovery)>, Vec<PathBuf>) {
    let mut held = Vec::new();
    let mut free = Vec::new();
    for dir in dirs {
        match StoreLock::acquire(dir) {
            Ok(_) => free.push(dir.clone()),
            Err(_) => match Discovery::read(dir) {
                Some(found) => held.push((dir.clone(), found)),
                // Refused by something that is not a daemon we can reach. Left
                // in `free` on purpose: `run` acquires it again and fails with
                // the error a real conflict has always produced, rather than
                // this function inventing a second wording for it.
                None => free.push(dir.clone()),
            },
        }
    }
    (held, free)
}

/// What a store watches, and what it should be watching but cannot find.
fn roots_for(dir: &Path, registry: &Registry) -> (String, Vec<PathBuf>) {
    if let Some(name) = registry.name_of(dir) {
        let roots = registry.stores[name].roots.clone();
        return (name.to_string(), roots);
    }
    // Not in the registry: a `.semlith` beside its corpus, or a `--store`
    // path. The directory holding it is the corpus, which is the same rule
    // `adopt` uses.
    let name = dir
        .parent()
        .and_then(Path::file_name)
        .or_else(|| dir.file_name())
        .map(|n| n.to_string_lossy().into_owned())
        .unwrap_or_else(|| "store".to_string());
    let root = dir
        .parent()
        .map(crate::canonical)
        .unwrap_or_else(|| dir.to_path_buf());
    (name, vec![root])
}

/// Run until a signal stops it.
///
/// `report` is the daemon's stderr: the bound port, each store opened, each
/// re-embed, and the refusal counts on the way out. Never the token — that
/// appears in the URL and in the discovery file, and nowhere else.
#[allow(clippy::too_many_arguments)]
pub fn run(
    dirs: &[PathBuf],
    port: u16,
    debounce: Duration,
    airgap: bool,
    ledger: bool,
    mcp_http: bool,
    report: impl Fn(&str) + Send + Sync + 'static,
) -> Result<Arc<State>> {
    let registry = Registry::load()?;

    // Named, not silently passed over. A registry entry whose directory is not
    // on disk was skipped without a word, so a machine with three of them
    // started clean and offered all three everywhere a real store is offered.
    for name in registry.stores.keys() {
        let Ok(dir) = Registry::dir_of(name) else {
            continue;
        };
        if dir.join("store.db").exists() {
            continue;
        }
        report(&format!(
            "skipped {name}: registered, but there is no store at {}",
            dir.display()
        ));
    }

    // Locks first, and all of them, before anything is watched or served: a
    // daemon that took three of four locks and then failed would leave three
    // stores unusable to the `semlith index` that is about to be tried.
    //
    // One exception: a store a terminal `semlith index` is writing. Refusing
    // to start over it made a login service fail until that run ended, and a
    // run ends by itself, so the store is opened when it does.
    let mut locks = Vec::new();
    let mut opening = Vec::new();
    let mut waiting = Vec::new();
    for dir in dirs {
        let Some(lock) = StoreLock::try_acquire(dir)
            .with_context(|| format!("{} cannot be opened by the daemon", dir.display()))?
        else {
            let (name, _) = roots_for(dir, &registry);
            report(&format!(
                "{name}: being indexed by another process; opened when that run ends"
            ));
            waiting.push(dir.clone());
            continue;
        };
        let (name, roots) = roots_for(dir, &registry);
        // Canonical, as `open_store` records it, because `reconcile` compares
        // what it computes from the registry against what is already open. A
        // home reached through a symlink — `/tmp` on macOS is one — gave those
        // two different spellings of one directory, so every store opened at
        // startup was reported as registered-but-not-open as well, and the
        // Stores page listed each of them twice.
        opening.push((name, crate::canonical(dir), roots));
        locks.push(lock);
    }

    // The port is the other thing worth failing on before any work: a daemon
    // that indexed for a minute and then could not listen has wasted the
    // minute.
    let server = Arc::new(Server::bind(port)?);
    report(&format!("listening on 127.0.0.1:{}", server.port()));

    // Said on every start, in both states, before anything is recorded.
    //
    // This is the whole of what makes default-on recording honest rather than
    // a surprise: the person who started the daemon is told, in the same
    // breath as the port, that it keeps a record and how to stop it. A default
    // nobody is told about is the thing 0.12.0 was right to refuse.
    crate::ledger::set_session_off(!ledger);
    let settings = crate::home::Settings::load();
    // The Privacy page's switch, as it was left. Before anything below can
    // reach for a model or a pack.
    crate::add::set_runtime_airgap(settings.airgap.unwrap_or(false));
    crate::add::start_counting();
    if settings.airgap == Some(true) && !airgap {
        report("airgap: on from the Privacy page; nothing will be fetched");
    }
    let paused = settings.ledger_paused.unwrap_or(false);
    crate::ledger::set_paused(paused);
    report(if !ledger {
        "ledger: off for this session"
    } else if paused {
        "ledger: paused from the portal (Resume recording on the Ledger page)"
    } else {
        "ledger: recording (local only; --no-ledger to stop)"
    });

    // The agent key is read, or written if this machine has none. It survives
    // restarts and upgrades on purpose: a client's configuration is written
    // once and has to keep working, which the per-run session token can never
    // do.
    let key = crate::home::agent_key()?;
    server.set_agent_key(&key);
    server.set_mcp_open(mcp_http);
    if mcp_http {
        report(&format!(
            "MCP over HTTP at http://127.0.0.1:{}{}",
            server.port(),
            crate::http::MCP_PATH
        ));
    } else {
        report("MCP over HTTP is closed (--no-mcp-http); `semlith mcp` over stdio is unaffected");
    }

    let mut stores = Vec::new();
    for (name, dir, roots) in opening {
        let watched: Vec<PathBuf> = roots.iter().filter(|r| r.exists()).cloned().collect();
        for missing in roots.iter().filter(|r| !r.exists()) {
            // Never fatal. A corpus on an unmounted disk is a Monday morning,
            // not a corruption, and the store's vectors are still searchable.
            report(&format!(
                "{name}: root {} is not there; not watching it",
                missing.display()
            ));
        }
        report(&format!(
            "opened {name} at {} — watching {} root(s)",
            dir.display(),
            watched.len()
        ));
        let store = Store::new(name, dir, roots, watched, false, 0);
        store.apply_settings(&crate::home::store_settings(&store.dir));
        stores.push(Arc::new(store));
    }
    refresh_unrecorded();

    let fleet = if stores.is_empty() {
        None
    } else {
        // Canonical, as every store in `stores` is recorded: `/api/stores`
        // pairs each store with its fleet member by directory, and a home
        // reached through a symlink (`/var` on macOS) gave the two different
        // spellings, so every row read zero files until the fleet was reopened.
        let canonical: Vec<PathBuf> = stores.iter().map(|s| s.dir.clone()).collect();
        let mut fleet = Fleet::open(&canonical)?;
        // The fleet every route answers from, so this is the one that has to
        // agree with `/api/stores` about what each store is called.
        fleet.name_from(&named(&stores));
        // stderr is this process's log, and a model download progress bar in
        // the middle of it is noise; the Stores view reports readiness.
        fleet.quiet = true;
        Some(fleet)
    };

    let token = server.token();
    for store in &stores {
        discovery(server.port(), &token).write(&store.dir)?;
    }

    let report_line: Arc<dyn Fn(&str) + Send + Sync> = Arc::new(report);

    // Read before the first run can be submitted, and printed below with where
    // each value came from: a machine that derived three when the user set one
    // should say so rather than leave them to infer it from how fast the queue
    // moves.
    let limits = Limits::in_force();
    report_line(&limits.line());
    limits.apply();

    // Index runs hand batches to the GPU lanes as well as the CPU. Only the
    // daemon does; a terminal `semlith index` stays on the CPU.
    crate::accel::manage();
    // The Neural Engine's first start compiles its models for this Mac, which
    // takes minutes once. Begun now, in the background, so the first run does
    // not wait for it; Machine limits shows how far it has got.
    if crate::accel::enabled().ane
        && crate::accel::unavailable_here("ane").is_none()
        && crate::model_cache_dir()
            .ok()
            .is_some_and(|cache| crate::packs::installed(&cache, &crate::packs::coreml()).is_some())
        && let Some(lane) = crate::accel::lane("ane")
    {
        lane.wake();
    }

    // Background while idle, normal while embedding. Before the watchers
    // start, so the catch-up they run is the first thing that lifts it.
    {
        let log = Arc::clone(&report_line);
        crate::priority::manage(move |line| log(line));
    }

    let state = Arc::new(State {
        server: Arc::clone(&server),
        fleet: Mutex::new(fleet),
        stores: RwLock::new(stores),
        late_writers: Mutex::new(Vec::new()),
        admission: Arc::new(Admission::new(limits.runs_at_once.value)),
        airgap,
        started: SystemTime::now(),
        debounce,
        report: Arc::clone(&report_line),
        refusals: Mutex::new(BTreeMap::new()),
        awaiting: Mutex::new(BTreeMap::new()),
        proxies: Mutex::new(BTreeMap::new()),
        clients: Mutex::new(BTreeMap::new()),
        mcp_fleet: Mutex::new(None),
        ledger,
        schedules: crate::schedule::Runner::new(),
        gone: Mutex::new(Vec::new()),
    });

    // The timer, over whatever `~/.semlith/schedules.json` holds. A home with
    // no such file is the normal state: nothing is created, nothing is said,
    // and the thread sleeps until somebody adds one and wakes it.
    match crate::schedule::Schedules::load() {
        Ok(file) if !file.schedules.is_empty() => {
            report_line(&format!("schedules: {} registered", file.schedules.len()))
        }
        Ok(_) => {}
        // Never fatal. A schedules file this binary cannot read is a report
        // that will not be written, not a reason to refuse to serve a corpus.
        Err(e) => report_line(&format!("schedules: {e:#}")),
    }
    crate::schedule::Runner::spawn(Arc::clone(&state));

    // Installed before the first thread starts: the signal is how this process
    // ends, so the ordinary exit has to be the safe one.
    watch::stop_on_signal();

    let report = Arc::clone(&report_line);
    let mut watchers = Vec::new();
    for (store, lock) in state.stores().into_iter().zip(locks) {
        let report = Arc::clone(&report);
        // Set here rather than inside the thread: the flag is what the write
        // queue checks before accepting a job, and opening the store takes
        // long enough that a request arriving in that window would otherwise
        // be told the writer was gone when it was merely starting.
        store.watching.store(true, Ordering::Relaxed);
        let admission = Arc::clone(&state.admission);
        watchers.push(std::thread::spawn(move || {
            // Moved in so the lock's life is the thread's life, which is what
            // makes "the daemon is the writer" true rather than intended.
            let _lock = lock;
            keep_writing(&store, debounce, &*report, &admission);
        }));
    }

    // After the watchers, so a note lands on a store whose feed is already
    // being kept, and before the URL, so it is on the page from the first read.
    report_dropped_queue(&state.stores(), &*report_line);

    // The stores another process was writing at startup, each opened as soon
    // as its lock comes free: `open_store` is the portal's path for a store
    // that appears while the daemon runs, watcher and catch-up included.
    if !waiting.is_empty() {
        let opener = Arc::clone(&state);
        std::thread::spawn(move || {
            while !waiting.is_empty() && !watch::STOP.load(Ordering::Relaxed) {
                std::thread::sleep(Duration::from_secs(2));
                waiting.retain(|dir| match StoreLock::try_acquire(dir) {
                    Ok(Some(free)) => {
                        // Let go first: `open_store` takes the lock itself.
                        drop(free);
                        opener.open_store(dir, false).is_err()
                    }
                    Ok(None) => true,
                    Err(e) => {
                        (opener.report)(&format!("{}: {e:#}", dir.display()));
                        false
                    }
                });
            }
        });
    }

    // Behind the URL, not in front of it: loading the model costs a second or
    // two, and a developer staring at a blank terminal waiting for a link is
    // paying that cost twice. The first search would otherwise pay it instead,
    // which is worse — it looks like the search is slow.
    {
        let warming = Arc::clone(&state);
        let report = report_line.clone();
        std::thread::spawn(move || {
            let mut fleet = warming.fleet.lock().unwrap_or_else(|e| e.into_inner());
            if let Some(fleet) = fleet.as_mut()
                && let Err(e) = fleet.warm()
            {
                report(&format!("could not load the embedding model: {e}"));
            }
        });
    }

    // stdout, not stderr: the token is in this URL, and stderr is the log the
    // operational criterion says never carries it. One line, once, so a
    // terminal's scrollback holds exactly one copy.
    println!("{}", server.url());
    report("open the URL on stdout to reach the portal; Ctrl-C stops the daemon");

    Ok(state.clone()).and_then(|state| {
        let handler = crate::routes::handler(Arc::clone(&state));
        let for_refusals = Arc::clone(&state);
        let result = server.serve(handler, &watch::STOP, move |class| {
            for_refusals.refuse(class);
        });

        // Shutdown, in the order the contract states: the watcher stops, the
        // index is checkpointed by the loop leaving cleanly, the lock is
        // released when its thread joins, and the discovery file goes last so
        // nothing is pointed at a daemon that is no longer answering.
        watch::STOP.store(true, Ordering::SeqCst);
        // Before the watchers are told to stop, while a run is still marked as
        // going: what this records is which folders the next start owes the
        // user an explanation for.
        remember_dropped_queue(&state.stores(), &state.admission.waiting());
        // Nothing waits on the timer — it holds no lock and owns no store —
        // but a thread asleep for an hour should not be what keeps a process
        // alive, so it is told to go.
        state.schedules.stop();
        // Each watcher waits on its own store's flag now, so that a single
        // store can be closed and deleted without stopping the daemon. A
        // shutdown is every store at once.
        for store in state.stores() {
            store.stop.store(true, Ordering::SeqCst);
        }
        for watcher in watchers {
            let _ = watcher.join();
        }
        let late =
            std::mem::take(&mut *state.late_writers.lock().unwrap_or_else(|e| e.into_inner()));
        for writer in late {
            let _ = writer.join();
        }
        for store in state.stores() {
            Discovery::remove(&store.dir);
        }

        let refusals = state
            .refusals
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .clone();
        if !refusals.is_empty() {
            let counts: Vec<String> = refusals
                .iter()
                .map(|(class, n)| format!("{class} {n}"))
                .collect();
            report(&format!("refused requests: {}", counts.join(", ")));
        }
        report("stopped");
        result.map(|()| state)
    })
}

/// Run a store's writer until it returns, and say why it did.
///
/// A panic is caught here as well as an error. Either way the thread is gone,
/// so `watching` goes false and the reason is on the store's feed; before
/// 0.28.0 a panic skipped both, and `/api/stores` went on reporting a store
/// nobody was watching as watched.
fn keep_writing(
    store: &Arc<Store>,
    debounce: Duration,
    report: &(dyn Fn(&str) + Send + Sync),
    admission: &Arc<Admission>,
) {
    let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        tend(store, debounce, &store.stop, report, admission)
    }));
    let why = match outcome {
        Ok(Ok(())) => None,
        Ok(Err(e)) => Some(format!("{e:#}")),
        Err(panic) => Some(format!(
            "the writer panicked: {}",
            panic
                .downcast_ref::<&str>()
                .map(|s| s.to_string())
                .or_else(|| panic.downcast_ref::<String>().cloned())
                .unwrap_or_else(|| "no message".to_string())
        )),
    };
    store.watching.store(false, Ordering::Relaxed);
    if let Some(why) = why {
        report(&format!("{}: watcher stopped: {why}", store.name));
        store.note(format!("watcher stopped: {why}"));
        *store
            .stopped_because
            .lock()
            .unwrap_or_else(|e| e.into_inner()) = Some(why);
    }
    changes::bump(changes::Domain::Stores);
}

/// How long a writer keeps its embedding session with nothing to embed.
const SESSION_IDLE: Duration = Duration::from_secs(60);

/// Override for [`SESSION_IDLE`], in seconds, so a test can see a session go
/// without waiting a minute. Not part of the documented environment.
const SESSION_IDLE_ENV: &str = "SEMLITH_SESSION_IDLE_SECS";

fn session_idle() -> Duration {
    std::env::var(SESSION_IDLE_ENV)
        .ok()
        .and_then(|v| v.parse().ok())
        .map(Duration::from_secs)
        .unwrap_or(SESSION_IDLE)
}

/// One store's watcher thread: the same loop `semlith watch` runs, plus the
/// queue the HTTP routes and the MCP proxy put work on.
fn tend(
    store: &Arc<Store>,
    debounce: Duration,
    stop: &AtomicBool,
    report: &(dyn Fn(&str) + Send + Sync),
    admission: &Arc<Admission>,
) -> Result<()> {
    let mut writer = Semlith::open(&store.dir, None)?;
    writer.quiet = true;
    writer.gitignore = store.gitignore.load(Ordering::Relaxed);
    let mut next_compact_check = std::time::Instant::now() + COMPACT_FIRST_CHECK;

    // Before the watcher reads a single event: a store that holds files
    // outside every root it is registered against is a store whose search
    // results carry the wrong label, and whose rows compete with the store
    // that really owns those files. The old boundary rule let the whole home
    // directory into every store, so this is the one-time cost of tightening
    // it. The files on disk are untouched.
    match writer.prune_out_of_root(&store.roots) {
        Ok(0) => {}
        Ok(dropped) => {
            store.pruned.store(dropped, Ordering::Relaxed);
            let text =
                format!("reconciled — {dropped} file(s) dropped, held outside this store's roots");
            report(&format!("{}: {text}", store.name));
            store.note(text);
        }
        Err(e) => {
            report(&format!("{}: could not reconcile: {e}", store.name));
        }
    }

    // A store with no root on disk still gets a thread, because it still has a
    // queue: the portal can index a new path into it even though there is
    // nothing to watch yet.
    let roots = store.watched.clone();

    // The catch-up is a run, admitted like any other. It used to be the
    // watcher's own walk, which the admission queue could not see: three
    // stores catching up at startup were three sessions embedding at once
    // whatever `runs at once` said, and stopping a run let the watcher take
    // the same walk up again on its own. A store the portal has just made
    // skips it, because the run on its way covers the same roots.
    let expecting = (now() as usize) < store.expecting_run_until.load(Ordering::Relaxed);
    // A store whose `watch` setting is off is not caught up either: what
    // changed while nothing watched is exactly what it asked not to follow.
    let deferred = !roots.is_empty() && !expecting && store.watch_events.load(Ordering::Relaxed);
    if deferred {
        let _ = admission.submit(store, roots.clone(), RunKind::CatchUp);
    }

    watch::run_held(
        &mut writer,
        &roots,
        debounce,
        stop,
        watch::Held {
            catch_up: false,
            waiting: &|| false,
            // A burst larger than a save is admitted like a run, so a `git
            // checkout` of thousands of files waits its turn and shows as a
            // card rather than embedding beside every admitted run.
            defer: &|paths| {
                // Watching off: the batch is taken and dropped. Turning it
                // back on queues a catch-up, which is what picks these up.
                if !store.watch_events.load(Ordering::Relaxed) {
                    return true;
                }
                if paths.len() <= BATCH_ADMIT_OVER {
                    return false;
                }
                let _ = admission.submit(store, paths.to_vec(), RunKind::Batch);
                true
            },
        },
        |progress| {
            use watch::Progress;
            match progress {
                Progress::Ready {
                    catch_up,
                    files,
                    chunks,
                } => {
                    // The catch-up is a queued run here, not the watcher's
                    // own pass, so "0 indexed at startup" would be a claim
                    // about work that has not happened yet (1.17).
                    let _ = &catch_up;
                    let text = if deferred {
                        format!(
                            "watching — {files} files, {chunks} chunks \
                             (catch-up deferred: {files} files queued)"
                        )
                    } else {
                        format!("watching — {files} files, {chunks} chunks")
                    };
                    report(&format!("{}: {text}", store.name));
                    store.note(text);
                }
                Progress::Batch(batch, elapsed) => {
                    let text = format!(
                        "{} re-embedded, {} removed, {} chunks in {:.1}s",
                        batch.indexed,
                        batch.removed,
                        batch.chunks,
                        elapsed.as_secs_f32()
                    );
                    report(&format!("{}: {text}", store.name));
                    store.note(text);
                    store.last_write.store(now() as usize, Ordering::Relaxed);
                }
                Progress::Error(e) => {
                    report(&format!("{}: watch error: {e}", store.name));
                    store.note(format!("error: {e}"));
                }
                Progress::Failed { path, why } => {
                    // A note on the store's feed, not a watch error: the
                    // thread is still running and `watching` is still true.
                    // One corrupt file used to take the whole watcher with it.
                    let text = format!(
                        "could not index {}: {why}",
                        crate::plain(&path.display().to_string())
                    );
                    report(&format!("{}: {text}", store.name));
                    store.note(text);
                }
                Progress::File(_) => {}
            }
        },
        |writer| {
            // Between batches, on the writer's own thread: fit the indexes to
            // the budget in force, and let go of a session nothing has used
            // for a minute.
            writer.follow_budget();
            writer.release_if_idle(session_idle());
            // The setting may have moved on the page since the last batch.
            writer.gitignore = store.gitignore.load(Ordering::Relaxed);
            if std::time::Instant::now() >= next_compact_check {
                next_compact_check = std::time::Instant::now() + COMPACT_CHECK;
                auto_compact(store, writer, admission);
            }
            // The writer is this thread, so a queued job runs here or nowhere.
            // The roots of any run that was stopped go back to the watcher,
            // which drops the events it queued for them meanwhile.
            let mut stopped = Vec::new();
            loop {
                // A shutdown leaves what is queued queued: a run that yielded
                // its slice requeues the rest, and without this the writer
                // took it straight back and a shutdown joining it never ended.
                if store.stop.load(Ordering::Relaxed) {
                    return Ok(stopped);
                }
                let Some(next) = store
                    .queue
                    .lock()
                    .unwrap_or_else(|e| e.into_inner())
                    .pop_front()
                else {
                    return Ok(stopped);
                };
                if let Some(roots) = perform(store, writer, next, admission) {
                    stopped.extend(roots);
                }
            }
        },
    )
}

/// How often the writer asks whether its idle store has passed the
/// auto-compaction threshold. A footprint is file sizes and three small
/// queries, so this is about not deciding every 250 ms, not about cost.
const COMPACT_CHECK: std::time::Duration = std::time::Duration::from_secs(600);

/// The first check, a minute after the writer starts: long enough that a
/// daemon's opening catch-ups are queued first and the store is not idle.
const COMPACT_FIRST_CHECK: std::time::Duration = std::time::Duration::from_secs(60);

/// Whether a store should be compacted on the daemon's own initiative.
///
/// Only an idle store -- no run on its card, nothing in its writer's queue or
/// the admission queue, not paused, no run about to arrive -- and only past
/// both the threshold and the floor. A threshold of 0 is off.
fn compaction_due(threshold: u64, idle: bool, disk: &crate::compact::Footprint) -> bool {
    threshold > 0
        && idle
        && disk.dead_percent() > threshold
        && disk.reclaimable >= crate::compact::AUTO_FLOOR_BYTES
}

/// Queue a compaction of `store` when [`compaction_due`] says so, as a run
/// with its own card, so it is seen and can be stopped like any other.
fn auto_compact(store: &Arc<Store>, writer: &Semlith, admission: &Arc<Admission>) {
    let idle = !store.run_live()
        && store.queue_depth() == 0
        && !store.paused.load(Ordering::Relaxed)
        && admission.position_of(&store.name).is_none()
        && (now() as usize) >= store.expecting_run_until.load(Ordering::Relaxed);
    let threshold = crate::compact::threshold_in_force();
    let retention = crate::compact::retention_in_force();
    let Ok(disk) = writer.footprint(retention) else {
        return;
    };
    if !compaction_due(threshold, idle, &disk) {
        return;
    }
    let run = admission.next.fetch_add(1, Ordering::Relaxed);
    store.begin_run(run, vec![store.dir.clone()], RunKind::Compact);
    store.note(format!(
        "{} % of this store is reclaimable, past the {threshold} % threshold: compacting",
        disk.dead_percent()
    ));
    let _ = store.submit_run(
        run,
        Job::Compact(crate::compact::CompactOptions {
            retention_days: retention,
            dry_run: false,
        }),
    );
}

/// Run one queued job. For a run that was stopped, the roots it covered.
fn perform(
    store: &Arc<Store>,
    writer: &mut Semlith,
    queued: Queued,
    admission: &Arc<Admission>,
) -> Option<Vec<PathBuf>> {
    let Queued { run, job, report } = queued;
    // The run the watcher was holding off for has arrived, so the hold ends
    // here rather than when its grace period runs out.
    store.expecting_run_until.store(0, Ordering::Relaxed);
    let back = report.clone();
    let say = |value: serde_json::Value| {
        // The snapshot is written by the same closure that emits the event, so
        // there is one path and a page polling the snapshot cannot be shown
        // something the stream never carried, or miss something it did. This
        // is the whole of why a run can outlive the request that started it.
        store.record(run, &value);
        // A closed receiver means the browser navigated away mid-run. The work
        // still finishes — it is the store's, not the request's.
        let _ = report.send(value);
    };

    match job {
        Job::Index(Indexing {
            work,
            first,
            already,
            scanned: scanned_before,
            total: total_before,
            bytes: bytes_before,
            bytes_total: bytes_total_before,
            mut tally,
        }) => {
            // Only the first slice announces itself; the rest are the same run
            // continuing, and a second "started" would read as a second run.
            // It carries what the store held before, which is what decides
            // whether Stop's "Also delete the store" starts ticked.
            if first {
                let names: Vec<String> = match &work {
                    Work::Roots(roots) => roots.iter().map(|p| p.display().to_string()).collect(),
                    Work::Rest(files) => files.iter().map(|p| p.display().to_string()).collect(),
                };
                let (files, chunks, _) = writer.stats().unwrap_or((0, 0, 0));
                say(serde_json::json!({
                    "event": "started",
                    "paths": names,
                    "files_before": files.max(0),
                    "chunks_before": chunks.max(0),
                }));
            }
            writer.follow_budget();
            // Every queued job arrived through the portal or through a
            // forwarded `semlith_index`, so both are held to the boundary: this
            // store's registered roots and the home directory, and never a
            // credential by name. The watcher's own re-embeds are inside those
            // roots by construction.
            writer.boundary = crate::Boundary::within(crate::home::index_roots(&store.dir));
            // A pause belongs to the run that was on when it was asked for.
            // A stop does not need clearing here: a job that was queued when
            // one arrived has already been dropped from the queue, so reaching
            // this line means the flag is for this run.
            store.paused.store(false, Ordering::Relaxed);
            let control = {
                let store = Arc::clone(store);
                let admission = Arc::clone(admission);
                let told = std::sync::atomic::AtomicBool::new(false);
                let held = std::sync::atomic::AtomicBool::new(false);
                move || {
                    if store.cancelled.load(Ordering::Relaxed) {
                        return crate::Flow::Stop;
                    }
                    // The daemon is shutting down: keep what is done and
                    // step aside now, as the slice budget would in 45 s,
                    // and before a pause or a hold, which would otherwise
                    // wait for ever under a shutdown joining this writer.
                    // Without it a stop mid-run returned from `main` under
                    // an embedding pass still going, and ONNX Runtime was
                    // torn down beneath it (#163).
                    if store.stop.load(Ordering::Relaxed) {
                        return crate::Flow::Yield;
                    }
                    // A lowered limit holds the newest runs at their next
                    // batch. Asked before the pause, because a held run that
                    // is also paused is still waiting for a slot first.
                    if admission.is_held(run) {
                        if !held.swap(true, Ordering::Relaxed) {
                            say(serde_json::json!({ "event": "held" }));
                        }
                        return crate::Flow::Pause;
                    }
                    if held.swap(false, Ordering::Relaxed) {
                        say(serde_json::json!({ "event": "resumed" }));
                    }
                    if store.paused.load(Ordering::Relaxed) {
                        // Once per pause, not once per tick.
                        if !told.swap(true, Ordering::Relaxed) {
                            say(serde_json::json!({ "event": "paused" }));
                        }
                        return crate::Flow::Pause;
                    }
                    if told.swap(false, Ordering::Relaxed) {
                        say(serde_json::json!({ "event": "resumed" }));
                    }
                    crate::Flow::Run
                }
            };
            // The run's own progress, not the slice's. A continuation's
            // counters start at one because it is indexing a shorter list; the
            // page is drawing one run, so it is told where the run is.
            // What the run had committed before this slice. A slice's own
            // report starts at zero, so these are what turn its counters back
            // into the run's — the same correction `scanned` has had since
            // 0.20.0, applied to the three counters that were left behind.
            // Without it the chunk counter climbed to a flush boundary and
            // fell back to single figures, which reads as a run losing work.
            let indexed_before = tally.indexed;
            let rows_before = tally.rows;
            let chunks_before = tally.chunks;
            let symbols_before = tally.symbols;
            let on_file = |path: &Path, progress: crate::IndexProgress| {
                let scanned = scanned_before + progress.scanned as u64;
                say(serde_json::json!({
                    // A batch inside a file is progress, not a verdict about
                    // the file: it moves the counters and stays off the log.
                    "event": if progress.outcome == crate::FileOutcome::Progress { "progress" } else { "file" },
                    // Plain, like every other path semlith hands out. The
                    // store keeps the verbatim form; a `\\?\C:\` prefix in
                    // an event is a path no editor opens and no shell
                    // completes.
                    "path": crate::plain(&path.display().to_string()),
                    // What is happening to this file, so a page can say
                    // "unchanged" rather than showing nothing at all.
                    "outcome": progress.outcome.as_str(),
                    // Why, for the outcomes that owe an explanation. A
                    // page showing "skipped" against two thousand files
                    // and nothing else is a page nobody can act on.
                    "why": progress.why,
                    "scanned": scanned,
                    // The walk's total, settled on the first slice. A later
                    // slice only knows how many it was handed.
                    "total": if total_before > 0 { total_before } else { progress.total as u64 },
                    "bytes": bytes_before + progress.bytes,
                    "bytes_total": if bytes_total_before > 0 { bytes_total_before } else { progress.bytes_total },
                    "indexed": indexed_before + progress.indexed as u64,
                    "chunks": chunks_before + progress.chunks as u64,
                    "rows": rows_before + progress.rows as u64,
                    "symbols": symbols_before + progress.symbols as u64,
                    "threads": progress.threads,
                    "lanes": progress.lanes,
                    "elapsed_ms": store.run_elapsed_ms(run),
                }));
            };
            // Every slice of a forced run is forced: a slice only carries the
            // files the run has not reached yet.
            writer.force = store.run_kind(run).is_some_and(RunKind::forced);
            let outcome = match work {
                Work::Roots(ref roots) => {
                    writer.index_within_held_under(roots, SLICE, &control, on_file)
                }
                Work::Rest(rest) => writer.index_rest_held_under(
                    rest,
                    SLICE,
                    (bytes_total_before > 0).then_some(bytes_total_before),
                    &control,
                    on_file,
                ),
            };
            writer.force = false;
            match outcome {
                Ok(mut done) => {
                    store.last_write.store(now() as usize, Ordering::Relaxed);

                    // Everything this run has embedded, across every slice of
                    // it, so a stop undoes the run rather than the slice that
                    // happened to be going.
                    let mut written = already;
                    written.append(&mut done.written);

                    if done.stopped {
                        // The same eviction a forget performs, for exactly the
                        // files this run wrote. Done here rather than inside
                        // the index pass because only this loop knows how many
                        // slices the run has had.
                        // One eviction pass and one save for the whole run.
                        // Per file, each forget rewrote the index.
                        let undone = writer.undo_held(&written).unwrap_or_else(|e| {
                            store.note(format!("the undo failed: {e:#}"));
                            0
                        });
                        store.note(format!(
                            "an index run was stopped; {undone} file(s) it had embedded were undone"
                        ));
                        say(serde_json::json!({
                            "event": "done",
                            "indexed": 0,
                            "unchanged": 0,
                            "skipped": 0,
                            "removed": undone,
                            "chunks": 0,
                            "images": 0,
                            "remaining": 0,
                            "stopped": true,
                        }));
                        store.paused.store(false, Ordering::Relaxed);
                        store.cancelled.store(false, Ordering::Relaxed);
                        release(run, admission);
                        return Some(store.run_paths(run));
                    }

                    // More to do: the rest goes back on the queue with the same
                    // channel, so the watcher gets a turn between slices and
                    // the reader keeps one stream rather than being asked to
                    // press the button again.
                    tally.add(&done);
                    let (slice_indexed, slice_chunks) = (tally.indexed, tally.chunks);
                    let slice_stages = tally.stages.clone();
                    let slice_cache = (tally.cache_lookups, tally.cache_hits);
                    if done.remaining > 0 {
                        // The remainder of the walk, not the roots. Handing the
                        // roots back meant the next slice walked the tree from
                        // the top and re-opened and re-hashed every file the
                        // run had already done, to be told each time that it
                        // was unchanged — so the progress bar restarted at one
                        // every forty-five seconds, and a run of N files over S
                        // slices read N×S files instead of N.
                        let total = if total_before > 0 {
                            total_before
                        } else {
                            // Settled once, on the slice that did the walk.
                            (scanned_before + done.scanned as u64) + done.remaining as u64
                        };
                        store.requeue(Queued {
                            run,
                            job: Job::Index(Indexing {
                                work: Work::Rest(std::mem::take(&mut done.pending)),
                                first: false,
                                already: written,
                                scanned: scanned_before + done.scanned as u64,
                                total,
                                bytes: bytes_before + done.bytes,
                                bytes_total: if bytes_total_before > 0 {
                                    bytes_total_before
                                } else {
                                    done.bytes_total
                                },
                                tally,
                            }),
                            report: back,
                        });
                        say(serde_json::json!({
                            "event": "slice",
                            "remaining": done.remaining,
                            "indexed": slice_indexed,
                            "chunks": slice_chunks,
                            "stages": slice_stages,
                            "cache_lookups": slice_cache.0,
                            "cache_hits": slice_cache.1,
                        }));
                        return None;
                    }

                    // A burst of watcher events that turned out to change
                    // nothing leaves no card: it was the daemon checking, not
                    // work anybody asked for or needs to dismiss.
                    let quiet_batch = store.run_kind(run) == Some(RunKind::Batch)
                        && tally.indexed == 0
                        && tally.removed == 0;
                    if store.run_kind(run) == Some(RunKind::CatchUp) {
                        // The other half of the startup line: what the
                        // deferred catch-up found, now that it has run.
                        let text = format!(
                            "catch-up done: {} indexed, {} unchanged, {} removed",
                            tally.indexed, tally.unchanged, tally.removed
                        );
                        eprintln!("semlith: {}: {text}", store.name);
                        store.note(text);
                    } else {
                        store.note(format!("{} indexed from the portal", tally.indexed));
                    }
                    // The daemon's log carries where the run's time went, so
                    // the next bottleneck is read rather than guessed.
                    if tally.indexed > 0 {
                        eprintln!("semlith: {}: {}", store.name, tally.stages.line());
                    }
                    say(serde_json::json!({
                        "event": "done",
                        "stages": tally.stages,
                        "cache_lookups": tally.cache_lookups,
                        "cache_hits": tally.cache_hits,
                        // The run's, not this slice's. A run of 35 files over
                        // three slices used to end by announcing the four the
                        // last slice reached.
                        "indexed": tally.indexed,
                        "unchanged": tally.unchanged,
                        "skipped": tally.skipped,
                        "removed": tally.removed,
                        "chunks": tally.chunks,
                        "images": tally.images,
                        // Above zero means the slice ran out of time, not that
                        // anything failed: asking again continues where it
                        // stopped and redoes nothing.
                        "remaining": done.remaining,
                        // A stopped run undid itself: nothing it embedded is
                        // in the store, so the next attempt starts from zero.
                        "stopped": done.stopped,
                        // Named, with the rule that refused each. A count would
                        // be a number somebody has to go and investigate.
                        "refused": tally.refused.iter().map(|(path, why)| {
                            serde_json::json!({ "path": crate::plain(path), "why": why })
                        }).collect::<Vec<_>>(),
                        // Named for the same reason as the refused, and the
                        // reason this release exists: a run that ends with
                        // "one file failed" and no name is a run whose one
                        // failure nobody can find.
                        "failed": tally.failed.iter().map(|(path, why)| {
                            serde_json::json!({ "path": crate::plain(path), "why": why })
                        }).collect::<Vec<_>>(),
                        // How the skipped divide up. The total alone is the
                        // number that made an Angular tree look like a run
                        // that had lost two thousand files.
                        "skipped_reasons": tally.skipped_reasons,
                        // What waits for a person, so a forwarded
                        // `semlith_index` says it as the in-process one does.
                        "review": crate::store::refusals(writer.db())
                            .map(|rows| rows.iter().filter(|r| r.reviewable && r.accepted.is_none()).count())
                            .unwrap_or(0),
                        // The daemon's own elapsed, so the page's clock is
                        // corrected to the run rather than to the tab.
                        "elapsed_ms": store.run_elapsed_ms(run),
                    }));
                    if quiet_batch {
                        store.remove_run(run);
                    }
                }
                Err(e) => say(serde_json::json!({ "event": "error", "error": e.to_string() })),
            }
            // The flags belong to a run, and this one is over.
            store.paused.store(false, Ordering::Relaxed);
            store.cancelled.store(false, Ordering::Relaxed);
            release(run, admission);
            None
        }
        Job::Accept { path, mode, source } => {
            let key = path.to_string_lossy().into_owned();
            match writer.accept(&key, &mode, &source) {
                Ok(accepted) => {
                    // Indexed at once, in the same job: the person who
                    // accepted it is looking at the page for the result.
                    let indexed = if mode == "refused" {
                        Ok(0)
                    } else {
                        writer
                            .index_changed(vec![PathBuf::from(&accepted.path)], |_, _| {})
                            .map(|r| r.indexed)
                    };
                    store.last_write.store(now() as usize, Ordering::Relaxed);
                    store.note(format!(
                        "{} {} ({})",
                        if mode == "refused" {
                            "refused"
                        } else {
                            "accepted"
                        },
                        path.display(),
                        mode
                    ));
                    match indexed {
                        Ok(n) => say(serde_json::json!({
                            "event": "done",
                            "accepted": accepted,
                            "indexed": n,
                        })),
                        Err(e) => {
                            say(serde_json::json!({ "event": "error", "error": format!("{e:#}") }))
                        }
                    }
                }
                Err(e) => say(serde_json::json!({ "event": "error", "error": format!("{e:#}") })),
            }
            None
        }
        Job::Revoke(path) => {
            let key = path.to_string_lossy().into_owned();
            match writer.revoke(&key) {
                Ok(revoked) => {
                    // Scanned again at once, so the file is back on the list
                    // with today's reasons rather than missing from both.
                    if revoked {
                        let _ = writer.index_changed(vec![path.clone()], |_, _| {});
                    }
                    store.note(format!("revoked {}", path.display()));
                    say(serde_json::json!({ "event": "done", "revoked": revoked }))
                }
                Err(e) => say(serde_json::json!({ "event": "error", "error": format!("{e:#}") })),
            }
            None
        }
        Job::Compact(options) => {
            store.cancelled.store(false, Ordering::Relaxed);
            say(
                serde_json::json!({ "event": "started", "paths": [store.dir.display().to_string()] }),
            );
            let stop = || store.cancelled.load(Ordering::Relaxed);
            match writer.compact_held(&options, &stop) {
                Ok(report) => {
                    store.cancelled.store(false, Ordering::Relaxed);
                    let text = if report.stopped {
                        "a compaction was stopped before its swap; nothing changed".to_string()
                    } else {
                        store.last_write.store(now() as usize, Ordering::Relaxed);
                        format!(
                            "compacted: {} to {} on disk ({} vectors and {} retired definitions dropped)",
                            crate::human_bytes(report.before.total() as i64),
                            crate::human_bytes(report.after.total() as i64),
                            report.vectors_dropped,
                            report.history_dropped,
                        )
                    };
                    // The daemon log's line, with every part's bytes, beside
                    // the event the card shows.
                    eprintln!(
                        "semlith: {}: {text} [database {} -> {}, exact {} -> {}, vectors {} -> {}]",
                        store.name,
                        report.before.database,
                        report.after.database,
                        report.before.exact,
                        report.after.exact,
                        report.before.vectors,
                        report.after.vectors,
                    );
                    store.note(text);
                    say(serde_json::json!({
                        "event": "done",
                        "stopped": report.stopped,
                        "compact": report,
                    }));
                }
                Err(e) => {
                    store.cancelled.store(false, Ordering::Relaxed);
                    say(serde_json::json!({ "event": "error", "error": format!("{e:#}") }))
                }
            }
            None
        }
        Job::Forget(path) => {
            match writer.forget_held(&path) {
                // Counted apart, because an image has no chunks: a single number
                // would report forgetting a picture as having done nothing.
                Ok((chunks, images)) => {
                    store.last_write.store(now() as usize, Ordering::Relaxed);
                    store.note(format!("forgot {}", path.display()));
                    say(serde_json::json!({
                        "event": "done",
                        "forgot": chunks,
                        "images": images,
                        "path": path.display().to_string(),
                    }));
                }
                Err(e) => say(serde_json::json!({ "event": "error", "error": e.to_string() })),
            }
            None
        }
    }
}

/// Where a queue that never started is left for the next daemon to report.
fn dropped_queue_path() -> Option<PathBuf> {
    home::home_or_error().ok().map(|h| h.join("queued.json"))
}

/// Remember the folders that were still waiting when this daemon stopped.
///
/// A queued run is not resumed on the next start: the run's state was this
/// daemon's, and the process ending is one of the two ways a run ends. What is
/// owed to the user is the list, because a folder that never started left no
/// trace anywhere — not in a store, not in a log — and without this they would
/// find out by noticing the search results are thin.
/// A run that was on when the process ended is remembered too, with what it
/// had committed. The checkpoint invariant means those files are in the store
/// and a re-run reports them as `unchanged`, but nothing on the page would
/// have said where it got to.
fn remember_dropped_queue(stores: &[Arc<Store>], waiting: &[(u64, String, Vec<PathBuf>)]) {
    let Some(path) = dropped_queue_path() else {
        return;
    };
    let mut rows: Vec<serde_json::Value> = waiting
        .iter()
        .map(|(_, store, paths)| {
            serde_json::json!({
                "store": store,
                "paths": paths.iter().map(|p| p.display().to_string()).collect::<Vec<_>>(),
                "started": false,
            })
        })
        .collect();
    for store in stores {
        if let Some((paths, kept)) = store.interrupted() {
            rows.push(serde_json::json!({
                "store": store.name,
                "paths": paths.iter().map(|p| p.display().to_string()).collect::<Vec<_>>(),
                "started": true,
                "kept": kept,
            }));
        }
    }
    if rows.is_empty() {
        let _ = std::fs::remove_file(&path);
        return;
    }
    if let Ok(body) = serde_json::to_string_pretty(&rows) {
        let _ = home::write_private(&path, (body + "\n").as_bytes());
    }
}

/// Say on each store's feed what the previous daemon never got to.
///
/// Read once and the file removed, so the line appears on the next start and
/// not on every start after it.
fn report_dropped_queue(stores: &[Arc<Store>], report: &(dyn Fn(&str) + Send + Sync)) {
    let Some(path) = dropped_queue_path() else {
        return;
    };
    let Ok(text) = std::fs::read_to_string(&path) else {
        return;
    };
    let _ = std::fs::remove_file(&path);
    let Ok(rows) = serde_json::from_str::<Vec<serde_json::Value>>(&text) else {
        return;
    };
    for row in rows {
        let Some(name) = row.get("store").and_then(serde_json::Value::as_str) else {
            continue;
        };
        let paths: Vec<String> = row
            .get("paths")
            .and_then(serde_json::Value::as_array)
            .map(|list| {
                list.iter()
                    .filter_map(serde_json::Value::as_str)
                    .map(crate::plain)
                    .collect()
            })
            .unwrap_or_default();
        let text = if row.get("started").and_then(serde_json::Value::as_bool) == Some(true) {
            let kept = row
                .get("kept")
                .and_then(serde_json::Value::as_u64)
                .unwrap_or(0);
            format!(
                "stopped during a run of {}; {kept} file(s) were kept, \
                 and indexing it again reports those as unchanged",
                paths.join(", ")
            )
        } else {
            format!(
                "queued when the daemon stopped and never started: {}",
                paths.join(", ")
            )
        };
        report(&format!("{name}: {text}"));
        if let Some(store) = stores.iter().find(|s| s.name == name) {
            store.note(text);
        }
    }
}

/// Give a finished run's place back, so whatever is waiting starts.
///
/// Called at the two points a run ends and nowhere else. A slice that hands
/// the writer back to the watcher is not one of them: the run is still the
/// store's, and releasing it there would admit a second run onto a machine
/// that is already carrying this one.
///
/// The run is named rather than looked up. It used to ask the store for "the
/// run", which worked while a store had exactly one; asking for its unfinished
/// run at the moment a run finishes returns nothing, and the place is never
/// given back — after `runs at once` runs the daemon admits nothing at all.
fn release(run: u64, admission: &Arc<Admission>) {
    admission.finish(run);
}

/// The daemon standing in as the writer for a forwarded `semlith_index` or
/// `semlith_forget`.
///
/// The agent's call blocks here until the watcher thread — the only thread
/// allowed to write — has run it, which is exactly the guarantee the agent
/// wanted and could not have in 0.8.0: the call either happens or reports why,
/// and never fails because somebody else holds the lock.
pub struct Writer(pub Arc<State>);

impl crate::mcp::Writer for Writer {
    fn index(&self, store: Option<&str>, paths: &[PathBuf]) -> Result<String, String> {
        let store = self.0.writable(store).map_err(|e| e.to_string())?.clone();
        // The receiver, not the run id: an agent's `semlith_index` blocks until
        // the writer has run it and answers with the summary, which is the
        // guarantee it has always had. The run is in the store's snapshot too,
        // so the portal shows an agent's index beside the page's own.
        let (_run, progress) = self
            .0
            .index(&store, paths.to_vec())
            .map_err(|e| e.to_string())?;
        let mut last = None;
        for event in progress {
            match event["event"].as_str() {
                Some("error") => {
                    return Err(event["error"]
                        .as_str()
                        .unwrap_or("indexing failed")
                        .to_string());
                }
                Some("done") => last = Some(event),
                // "file" events are the portal's progress bar; an agent gets
                // the summary it has always got.
                _ => {}
            }
        }
        let done = last.ok_or_else(|| "the writer stopped before answering".to_string())?;
        let count = |key: &str| done[key].as_u64().unwrap_or(0);
        // How the skipped divide up, in the same parenthesis and the same
        // order the in-process server writes (#121). Without it a forwarded
        // call answered "1 847 skipped" and nothing about why, while the same
        // call performed in process named every reason — the two paths are
        // one call in two places and may not say different things about it.
        let by_reason = {
            let mut parts: Vec<(u64, String)> = done["skipped_reasons"]
                .as_object()
                .into_iter()
                .flatten()
                .map(|(kind, n)| (n.as_u64().unwrap_or(0), kind.clone()))
                .collect();
            parts.sort_by(|a, b| b.0.cmp(&a.0).then(a.1.cmp(&b.1)));
            if parts.is_empty() {
                String::new()
            } else {
                format!(
                    " ({})",
                    parts
                        .iter()
                        .map(|(n, kind)| format!("{n} {kind}"))
                        .collect::<Vec<_>>()
                        .join(", ")
                )
            }
        };
        let mut text = format!(
            "{} indexed, {} unchanged, {} skipped{by_reason}, {} removed ({} chunks)",
            count("indexed"),
            count("unchanged"),
            count("skipped"),
            count("removed"),
            count("chunks"),
        );
        if count("remaining") > 0 {
            text.push_str(&format!(
                "\nStopped at the time limit with {} paths remaining. \
                 Call semlith_index again with the same arguments to continue; \
                 nothing already indexed is redone.",
                count("remaining")
            ));
        }
        // Named, one per line, with the rule that refused or failed each —
        // the same lines `mcp::answer` writes when it performs the run
        // itself. Without them a forwarded call answered a refused path with
        // "0 indexed, 0 unchanged, 0 skipped, 0 removed (0 chunks)" and
        // nothing else, so an agent held to the boundary was never told there
        // was a boundary, and retried the same path forever. The two paths are
        // the same call performed in two places; they may not say different
        // things about it.
        for (kind, lines) in [("refused", &done["refused"]), ("failed", &done["failed"])] {
            for entry in lines.as_array().into_iter().flatten() {
                let (Some(path), Some(why)) = (entry["path"].as_str(), entry["why"].as_str())
                else {
                    continue;
                };
                text.push_str(&format!("\n{kind}: {path} — {why}"));
            }
        }
        let waiting = count("review");
        if waiting > 0 {
            text.push_str(&format!(
                "\n{waiting} file{} await{} the owner's review (semlith refused, or the portal's Files ▸ Not indexed).",
                if waiting == 1 { "" } else { "s" },
                if waiting == 1 { "s" } else { "" }
            ));
        }
        Ok(text)
    }

    fn add(&self, store: Option<&str>, url: &str) -> Result<String, String> {
        let store = self.0.writable(store).map_err(|e| e.to_string())?.clone();

        // The fetch happens here rather than in the writer thread: it is
        // network work, and holding the store's write queue open for the
        // length of a download would stall every other write behind it. Only
        // the indexing of what landed goes through the queue, which is the
        // part the one-writer rule is actually about.
        let fetched = crate::add::fetch(url, &store.dir).map_err(|e| format!("{e:#}"))?;

        let indexed = self.index(Some(&store.name), std::slice::from_ref(&fetched.path))?;
        Ok(format!(
            "Fetched {} into {}. {indexed}",
            fetched.url,
            fetched.path.display()
        ))
    }

    fn forget(&self, store: Option<&str>, path: &str) -> Result<String, String> {
        let store = self.0.writable(store).map_err(|e| e.to_string())?.clone();
        let progress = self
            .0
            .forget(&store, PathBuf::from(path))
            .map_err(|e| e.to_string())?;
        let done = progress
            .recv()
            .map_err(|_| "the writer stopped before answering".to_string())?;
        if let Some(error) = done["error"].as_str() {
            return Err(error.to_string());
        }
        let chunks = done["forgot"].as_u64().unwrap_or(0);
        let images = done["images"].as_u64().unwrap_or(0);
        Ok(match (chunks, images) {
            (0, 0) => format!("{path} was not indexed; nothing removed."),
            (0, images) => format!("Removed {images} image vector(s) for {path}."),
            (chunks, 0) => format!("Removed {chunks} chunks for {path}."),
            (chunks, images) => {
                format!("Removed {chunks} chunks and {images} image vector(s) for {path}.")
            }
        })
    }
}

/// The URL the daemon prints, and the only place the token is shown.
pub fn url(state: &State) -> String {
    state.server.url()
}

/// Where the portal's About view gets its numbers.
pub fn uptime(state: &State) -> u64 {
    state
        .started
        .elapsed()
        .map(|d| d.as_secs())
        .unwrap_or_default()
}

pub fn now() -> u64 {
    SystemTime::now()
        .duration_since(SystemTime::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}

/// The port `semlith start` should use: the flag, then the environment, then
/// the default.
pub fn port_of(flag: Option<u16>) -> u16 {
    flag.or_else(|| {
        std::env::var(http::PORT_ENV)
            .ok()
            .and_then(|v| v.parse().ok())
    })
    .unwrap_or(http::DEFAULT_PORT)
}

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

    /// Six false privacy warnings on every `semlith mcp` start, one per store,
    /// naming six paths that do not exist.
    #[test]
    fn a_discovery_file_that_is_not_there_is_not_a_privacy_rejection() {
        let dir = tempfile::tempdir().expect("tempdir");
        let missing = dir.path().join(DISCOVERY_FILE);
        assert!(
            !missing.exists(),
            "the fixture wrote a file it should not have"
        );

        // The old predicate still answers `false` — that is correct and is
        // what made it the wrong thing to warn on.
        assert!(!owner_only(&missing));
        assert!(
            !worth_warning_about(&missing),
            "a missing {DISCOVERY_FILE} was reported as a privacy rejection",
        );

        // A file that is there and fails the check still gets its line.
        let present = dir.path().join("present.json");
        std::fs::write(&present, "{}").expect("write");
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            std::fs::set_permissions(&present, std::fs::Permissions::from_mode(0o644))
                .expect("chmod");
            assert!(
                worth_warning_about(&present),
                "a world-readable discovery file stopped being worth a line",
            );
        }
    }

    /// The change counters are process-global — deliberately, because two of
    /// the six are bumped by code that has no `State` in hand. That makes them
    /// shared between every test in this binary running in parallel threads.
    ///
    /// **Every test here that causes a bump takes this, not only the one that
    /// reads the counters.** Guarding the reader alone passed on macOS and
    /// failed on Windows, where the scheduling differs: a clock test's
    /// `record` landed inside the isolation test's window and moved `runs`
    /// while it was asserting that writing `stores` had not.
    fn counters() -> std::sync::MutexGuard<'static, ()> {
        static LOCK: Mutex<()> = Mutex::new(());
        LOCK.lock().unwrap_or_else(|e| e.into_inner())
    }

    /// A finished run is written to the store's history with its name, its
    /// stages flattened, and its log; a burst that changed nothing is not;
    /// the history reads back newest first and is trimmed.
    #[test]
    fn a_finished_run_outlives_the_daemon_in_its_store_history() {
        let _held = counters();
        let dir = tempfile::tempdir().unwrap();
        let store = Arc::new(Store::new(
            "notes".into(),
            dir.path().to_path_buf(),
            Vec::new(),
            Vec::new(),
            true,
            0,
        ));
        store.begin_run(
            1,
            vec![PathBuf::from("/a"), PathBuf::from("/b")],
            RunKind::Files,
        );
        store.record(1, &serde_json::json!({ "event": "started" }));
        store.record(
            1,
            &serde_json::json!({ "event": "file", "outcome": "failed", "path": "/a", "why": "unreadable" }),
        );
        store.record(
            1,
            &serde_json::json!({ "event": "done", "indexed": 1, "chunks": 4,
                "stages": { "wall_ms": 9, "read_ms": 2, "write_ms": 3,
                            "embed_wait_ms": { "cpu": 4 }, "prepare_cpu_ms": {} } }),
        );
        // A late event on a finished run does not write it twice.
        store.record(1, &serde_json::json!({ "event": "deleted", "text": "x" }));

        store.begin_run(2, Vec::new(), RunKind::Batch);
        store.record(
            2,
            &serde_json::json!({ "event": "done", "indexed": 0, "removed": 0 }),
        );

        let history = run_history(dir.path());
        assert_eq!(history.len(), 1, "{history:?}");
        let row = &history[0];
        assert_eq!(row["kind"], "Re-index 2 files");
        assert_eq!(row["result"], "done");
        assert_eq!(row["files"], 1);
        assert_eq!(row["chunks"], 4);
        assert_eq!(
            row["stages"],
            serde_json::json!({ "read": 2, "write": 3, "embed": 4 })
        );
        let log = row["log"].as_array().unwrap();
        assert!(log.iter().any(|l| l["level"] == "error"), "{log:?}");
        assert!(log.iter().all(|l| l["at"].is_u64()));

        for id in 10..10 + (HISTORY_KEEP as u64 * 2 + 5) {
            store.begin_run(id, Vec::new(), RunKind::Run);
            store.record(id, &serde_json::json!({ "event": "error", "error": "no" }));
        }
        let history = run_history(dir.path());
        assert_eq!(history.len(), HISTORY_KEEP);
        assert_eq!(history[0]["result"], "failed");
        let lines = std::fs::read_to_string(dir.path().join(HISTORY_FILE)).unwrap();
        assert!(
            lines.lines().count() <= HISTORY_KEEP * 2,
            "the file was never trimmed"
        );
    }

    /// A store with nothing behind it, for the parts of a run that are
    /// bookkeeping rather than embedding.
    fn bare_store(name: &str) -> Arc<Store> {
        Arc::new(Store::new(
            name.to_string(),
            PathBuf::from("/nowhere").join(name),
            Vec::new(),
            Vec::new(),
            true,
            0,
        ))
    }

    /// The clock restarted from zero every forty-five seconds, because it was
    /// measured inside `perform` and a slice is a fresh `Job::Index`. A page
    /// faithfully redrew that as a run that had just begun.
    #[test]
    fn a_runs_clock_spans_its_slices_and_never_goes_backwards() {
        // Recording bumps the runs counter, which another test in this
        // module asserts the isolation of. Same guard, or they race.
        let _held = counters();
        let store = bare_store("api");
        store.begin_run(1, vec![PathBuf::from("/work/api")], RunKind::Run);
        store.record(
            1,
            &serde_json::json!({ "event": "started", "paths": ["/work/api"] }),
        );
        std::thread::sleep(Duration::from_millis(30));
        let first = store.run_elapsed_ms(1);
        // A slice hands the writer back and the run returns as a new job. The
        // clock is the run's, so it carries on.
        store.record(1, &serde_json::json!({ "event": "slice", "remaining": 12 }));
        std::thread::sleep(Duration::from_millis(30));
        let second = store.run_elapsed_ms(1);
        assert!(second >= first, "{second} went backwards from {first}");
        assert!(second >= 50, "the clock did not span the slice: {second}ms");
    }

    /// Held time is not time anything is happening, and a run that is over
    /// keeps its total rather than counting on.
    #[test]
    fn a_held_run_stops_counting_and_a_finished_one_freezes() {
        // Recording bumps the runs counter, which another test in this
        // module asserts the isolation of. Same guard, or they race.
        let _held = counters();
        let store = bare_store("api");
        store.begin_run(1, Vec::new(), RunKind::Run);
        store.record(1, &serde_json::json!({ "event": "started", "paths": [] }));
        store.record(1, &serde_json::json!({ "event": "paused" }));
        let held = store.run_elapsed_ms(1);
        std::thread::sleep(Duration::from_millis(40));
        assert_eq!(
            store.run_elapsed_ms(1),
            held,
            "the clock counted while the run was held"
        );

        store.record(1, &serde_json::json!({ "event": "resumed" }));
        std::thread::sleep(Duration::from_millis(20));
        store.record(
            1,
            &serde_json::json!({ "event": "done", "indexed": 3, "stopped": false }),
        );
        let total = store.run_elapsed_ms(1);
        std::thread::sleep(Duration::from_millis(30));
        assert_eq!(
            store.run_elapsed_ms(1),
            total,
            "a finished run's clock kept going"
        );
    }

    /// Two clients read one run through their own cursors, and neither is
    /// affected by what the other has read.
    #[test]
    fn two_cursors_over_one_log_each_see_every_line_once() {
        // Recording bumps the runs counter, which another test in this
        // module asserts the isolation of. Same guard, or they race.
        let _held = counters();
        let store = bare_store("api");
        store.begin_run(1, Vec::new(), RunKind::Run);
        for scanned in 0..5 {
            store.record(
                1,
                &serde_json::json!({ "event": "file", "scanned": scanned }),
            );
        }

        let seqs = |lines: &[serde_json::Value]| {
            lines
                .iter()
                .map(|l| l["seq"].as_u64().unwrap())
                .collect::<Vec<_>>()
        };

        let first = store.log_after(None, None);
        assert_eq!(seqs(&first), vec![0, 1, 2, 3, 4]);

        // One client reads on from where it was; the other has read nothing
        // and still sees the whole run.
        store.record(1, &serde_json::json!({ "event": "file", "scanned": 5 }));
        assert_eq!(seqs(&store.log_after(None, Some(4))), vec![5]);
        assert_eq!(seqs(&store.log_after(None, None)), vec![0, 1, 2, 3, 4, 5]);
        // A cursor past the end is caught up, not an error.
        assert!(store.log_after(None, Some(5)).is_empty());
    }

    /// The queue is first in, first out by submission, and nothing reorders it.
    #[test]
    fn the_queue_admits_in_submission_order_and_only_up_to_the_limit() {
        // Recording bumps the runs counter, which another test in this
        // module asserts the isolation of. Same guard, or they race.
        let _held = counters();
        let admission = Admission::new(2);
        let stores: Vec<Arc<Store>> = ["a", "b", "c", "d"].into_iter().map(bare_store).collect();
        for store in &stores {
            admission.submit(
                store,
                vec![PathBuf::from("/work").join(&store.name)],
                RunKind::Run,
            );
        }

        assert_eq!(admission.running(), 2, "the limit admitted more than two");
        let waiting: Vec<String> = admission.waiting().into_iter().map(|(_, s, _)| s).collect();
        assert_eq!(waiting, vec!["c", "d"]);
        assert_eq!(admission.position_of("c"), Some(1));
        assert_eq!(admission.position_of("d"), Some(2));
        assert_eq!(
            admission.position_of("a"),
            None,
            "a running run is not waiting"
        );

        // A run finishing admits exactly the head, and the one behind it moves
        // up by having been behind it.
        let first = stores[0].current_run().expect("a is running");
        admission.finish(first);
        assert_eq!(admission.position_of("d"), Some(1));
        assert_eq!(admission.running(), 2);

        // Raising the limit admits immediately; nothing waits for a timer.
        admission.set_limit(4);
        assert!(admission.waiting().is_empty());
        assert_eq!(admission.running(), 3);
    }

    /// A run that ends gives its place back, however it ended.
    ///
    /// The place is released by run id. It used to be released by asking the
    /// store for "the run", which held while a store had exactly one — and
    /// once a store held several, asking for its *unfinished* run at the
    /// moment a run finished returned nothing, so the place was never given
    /// back. After `runs at once` runs the daemon admitted nothing ever again,
    /// and every later run sat at `queued` forever with nothing running.
    #[test]
    fn every_run_that_ends_gives_its_place_back() {
        let _held = counters();
        let admission = Arc::new(Admission::new(2));
        let stores: Vec<Arc<Store>> = ["a", "b", "c"].into_iter().map(bare_store).collect();
        let mut ids = Vec::new();
        for store in &stores {
            let (id, _) = admission.submit(store, Vec::new(), RunKind::Run);
            ids.push(id);
        }
        assert_eq!(admission.running(), 2);

        // Each run ends the way `perform` ends one: its record is marked done
        // and then its place is released by id.
        for (store, id) in stores.iter().zip(&ids) {
            store.record(
                *id,
                &serde_json::json!({ "event": "done", "indexed": 0, "stopped": false }),
            );
            // The reason `release` takes the id: by now the store has no
            // unfinished run to be asked for, so a lookup would find nothing
            // and release nothing.
            assert_eq!(store.current_run(), None);
            release(*id, &admission);
        }

        assert_eq!(
            admission.running(),
            0,
            "a finished run kept its place, so the daemon can admit nothing more"
        );
        assert!(admission.waiting().is_empty(), "the queue never drained");
    }

    /// Taking a folder out of the queue costs nothing, because nothing of it
    /// was embedded — which is the whole difference from stopping a run.    /// Taking a folder out of the queue costs nothing, because nothing of it
    /// was embedded — which is the whole difference from stopping a run.
    #[test]
    fn a_dequeued_run_is_answered_at_once_and_undoes_nothing() {
        // Recording bumps the runs counter, which another test in this
        // module asserts the isolation of. Same guard, or they race.
        let _held = counters();
        let admission = Admission::new(1);
        let (a, b) = (bare_store("a"), bare_store("b"));
        admission.submit(&a, Vec::new(), RunKind::Run);
        let (_, waiting) = admission.submit(&b, Vec::new(), RunKind::Run);

        assert_eq!(admission.position_of("b"), Some(1));
        assert_eq!(admission.dequeue("b"), 1);
        assert_eq!(admission.position_of("b"), None);
        assert_eq!(admission.running(), 1, "dequeuing b disturbed a");

        let answer = waiting.recv().expect("the dequeued run was answered");
        assert_eq!(answer["event"], "done");
        assert_eq!(answer["stopped"], true);
        assert_eq!(answer["dequeued"], true);
        assert_eq!(answer["removed"], 0, "a dequeued run undid something");
        // And nothing is left to dequeue.
        assert_eq!(admission.dequeue("b"), 0);
    }

    /// Each counter moves when, and only when, its own domain is written.
    ///
    /// The bump sites themselves are held to "once, and nowhere else" by
    /// `one_bump_site_per_domain` below; this is the other half — that the six
    /// are genuinely separate and one write does not move two.
    #[test]
    fn a_counter_moves_for_its_own_domain_and_no_other() {
        let _held = counters();
        for domain in changes::DOMAINS {
            let before: Vec<u64> = changes::DOMAINS.iter().map(|d| changes::read(*d)).collect();
            changes::bump(domain);
            for (other, was) in changes::DOMAINS.iter().zip(&before) {
                let now = changes::read(*other);
                if *other == domain {
                    assert_eq!(now, was + 1, "{} did not move", domain.as_str());
                } else {
                    assert_eq!(
                        now,
                        *was,
                        "writing {} moved {}",
                        domain.as_str(),
                        other.as_str()
                    );
                }
            }
        }
    }

    /// How many places may write each domain.
    ///
    /// One, except two. `stores`: a store joining, a store being deleted and
    /// a store's writer ending are three writes of one domain, and there is no
    /// single function they go through — `open_store` takes a lock and spawns
    /// a watcher, `delete_store` stops one and removes the directory, and
    /// `keep_writing` is the thread itself ending. `runs`: a store's own runs,
    /// and the cards of runs whose stop deleted their store, which no store is
    /// left to hold. Every other domain has exactly one writer and a second
    /// site would be a second source of truth.
    const BUMP_SITES: &[(&str, usize)] = &[
        ("stores", 3),
        ("runs", 2),
        ("clients", 1),
        ("ledger", 1),
        ("events", 1),
        ("privacy", 1),
    ];

    /// A bump site that is not in the one function that writes its domain is
    /// how one page goes stale while the others are live — the worst outcome
    /// available, because a portal that is wrong in one place is trusted in all
    /// of them.
    #[test]
    fn one_bump_site_per_domain() {
        // Every source file, read as text. `include_str!` rather than walking
        // the directory, so a file added without a line here is a compile
        // error rather than a silently unchecked file.
        const SOURCES: &[(&str, &str)] = &[
            ("daemon.rs", include_str!("daemon.rs")),
            ("routes.rs", include_str!("routes.rs")),
            ("store.rs", include_str!("store.rs")),
            ("doctor.rs", include_str!("doctor.rs")),
            ("ledger.rs", include_str!("ledger.rs")),
            ("lib.rs", include_str!("lib.rs")),
            ("watch.rs", include_str!("watch.rs")),
            ("mcp.rs", include_str!("mcp.rs")),
            ("http.rs", include_str!("http.rs")),
            ("home.rs", include_str!("home.rs")),
        ];
        for domain in changes::DOMAINS {
            let needle = format!("Domain::{}", title_case(domain.as_str()));
            let sites: usize = SOURCES
                .iter()
                .map(|(_, text)| {
                    text.lines()
                        .filter(|line| {
                            line.contains("changes::bump") && line.contains(needle.as_str())
                        })
                        .count()
                })
                .sum();
            let allowed = BUMP_SITES
                .iter()
                .find(|(name, _)| *name == domain.as_str())
                .map(|(_, n)| *n)
                .expect("every domain declares how many places may write it");
            assert_eq!(
                sites,
                allowed,
                "{} is bumped at {sites} site(s) and {allowed} is what it may have; a counter \
                 is bumped where its domain is written, and a site beyond that list is a \
                 second source of truth",
                domain.as_str()
            );
        }
    }

    fn title_case(word: &str) -> String {
        let mut chars = word.chars();
        match chars.next() {
            Some(first) => first.to_uppercase().collect::<String>() + chars.as_str(),
            None => String::new(),
        }
    }

    /// A `.semlith` beside its corpus is not in the registry, and the daemon
    /// still has to know what to call it and what to watch. The rule is the
    /// one `adopt` uses: the directory holding the store is the corpus.
    #[test]
    fn an_unregistered_store_is_named_after_the_directory_holding_it() {
        let registry = Registry::default();
        let (name, roots) = roots_for(Path::new("/work/api/.semlith"), &registry);
        assert_eq!(name, "api");
        assert_eq!(roots.len(), 1);
        assert!(roots[0].ends_with("api") || roots[0] == Path::new("/work/api"));
    }

    /// Lowering the limit holds the newest runs, which keep their place ahead
    /// of anything queued and go on oldest first as slots free.
    #[test]
    fn a_lowered_limit_holds_the_newest_runs_and_releases_them_in_order() {
        let _held = counters();
        let admission = Admission::new(3);
        let stores: Vec<Arc<Store>> = ["a", "b", "c", "d"].into_iter().map(bare_store).collect();
        let ids: Vec<u64> = stores
            .iter()
            .map(|store| admission.submit(store, Vec::new(), RunKind::Run).0)
            .collect();
        assert_eq!(admission.running(), 3);

        admission.set_limit(1);
        assert_eq!(admission.running(), 1, "one left running");
        assert_eq!(admission.held(), vec![ids[1], ids[2]], "held oldest first");
        assert!(!admission.is_held(ids[0]), "the oldest keeps going");
        assert_eq!(
            admission.position_of("d"),
            Some(3),
            "a queued run waits behind the held ones"
        );

        admission.finish(ids[0]);
        assert_eq!(admission.held(), vec![ids[2]], "the older held run went on");
        assert!(!admission.is_held(ids[1]));
        admission.finish(ids[1]);
        assert!(admission.held().is_empty());
        admission.finish(ids[2]);
        assert_eq!(admission.running(), 1, "then the queue moves");
    }

    /// The card's rate is the chunks of the last ten active seconds over
    /// their time, blank before the first batch, and the average leaves out
    /// the time before embedding began.
    #[test]
    fn the_rate_is_rolling_and_blank_before_the_first_batch() {
        let mut run = RunState::new(1, Vec::new(), RunKind::Run);
        assert_eq!(run.rates(), (None, None), "nothing embedded yet");
        run.first_sample = Some((0, 0, 0));
        run.samples = VecDeque::from(vec![(0, 0, 0), (20_000, 400, 0), (25_000, 500, 0)]);
        run.chunks = 600;
        // Frozen at 30 s of active time.
        run.ended = Some(Duration::from_secs(30));
        let (rate, average) = run.rates();
        // Base is the newest sample at or before 20 s: 200 chunks in 10 s.
        assert_eq!(rate, Some(20.0));
        assert_eq!(average, Some(20.0));
    }

    /// The time left is the bytes still to go over the bytes per second of
    /// the last ten active seconds, and there is none until the rate settles
    /// or while the run is not running.
    #[test]
    fn the_estimate_is_bytes_left_over_the_rolling_byte_rate() {
        let mut run = RunState::new(1, Vec::new(), RunKind::Run);
        run.status = RunStatus::Running;
        run.bytes_total = 10_000_000;
        run.first_sample = Some((0, 0, 0));
        run.samples = VecDeque::from(vec![(0, 0, 0), (2_000, 40, 200_000)]);
        run.bytes = 300_000;
        run.ended = Some(Duration::from_secs(3));
        assert_eq!(
            run.eta_ms(),
            None,
            "three seconds in, the rate has not settled"
        );

        // 20 s in; the newest sample at or before 10 s is the base: 3 MB in
        // 10 s is 300 KB/s, and 6 MB left is 20 s.
        run.samples = VecDeque::from(vec![
            (0, 0, 0),
            (10_000, 200, 1_000_000),
            (15_000, 300, 2_500_000),
        ]);
        run.bytes = 4_000_000;
        run.ended = Some(Duration::from_secs(20));
        assert_eq!(run.eta_ms(), Some(20_000));

        run.status = RunStatus::Paused;
        assert_eq!(
            run.eta_ms(),
            None,
            "a paused run has no time left to count down"
        );
    }

    /// The plan's estimate is on the card from the first batch, before the
    /// measured rate settles, and a finished run's rate is what a store with
    /// no rate of its own is planned on (2.7, measured failing at the 0.30.0
    /// walk stop: the first estimate came five seconds into embedding).
    #[test]
    fn the_first_estimate_comes_with_the_first_batch() {
        let mut done = RunState::new(1, Vec::new(), RunKind::Run);
        done.first_sample = Some((1_000, 0, 0));
        done.samples = VecDeque::from(vec![(1_000, 0, 0), (5_000, 90, 2_000_000)]);
        assert_eq!(done.embed_rate(), Some(500_000.0));

        let mut run = RunState::new(2, Vec::new(), RunKind::Run);
        run.status = RunStatus::Running;
        run.bytes_total = 3_000_000;
        run.plan_eta_ms = Some(6_000);
        run.first_sample = Some((500, 0, 0));
        run.samples = VecDeque::from(vec![(500, 0, 0)]);
        run.bytes = 50_000;
        run.ended = Some(Duration::from_millis(600));
        assert_eq!(
            run.eta_ms(),
            Some(5_400),
            "no estimate with the first batch"
        );
    }

    /// Issue #143: a run that stalls for eight seconds in a phase — writing its
    /// index — shows no estimate above twice the one before it.
    #[test]
    fn a_stalled_phase_holds_the_estimate_and_it_never_more_than_doubles() {
        let mut run = RunState::new(1, Vec::new(), RunKind::Run);
        run.status = RunStatus::Running;
        run.bytes_total = 10_000_000;
        run.first_sample = Some((0, 0, 0));
        run.samples = VecDeque::from(vec![
            (0, 0, 0),
            (10_000, 200, 1_000_000),
            (15_000, 300, 2_500_000),
        ]);
        run.bytes = 4_000_000;
        run.ended = Some(Duration::from_secs(20));
        let before = run.eta_ms().expect("settled");
        // Eight seconds writing the index: no bytes move at all.
        run.phase = Some("writing the index".to_string());
        for t in 21..=28 {
            run.samples.push_back((t * 1_000, 300, 4_000_000));
            run.ended = Some(Duration::from_secs(t));
            let eta = run.eta_ms().expect("held, not dropped");
            assert!(eta <= before * 2, "{eta} ms after {before} ms at {t} s");
        }
        // Out of the phase, the rate over the window is far lower than
        // before; the estimate may rise, but by at most 2x a poll.
        run.phase = None;
        run.samples.push_back((29_000, 310, 4_200_000));
        run.bytes = 4_200_000;
        run.ended = Some(Duration::from_secs(30));
        let after = run.eta_ms().expect("moving again");
        assert!(after <= before * 2, "{after} ms after {before} ms");
    }

    /// 0.30.1: a long run's estimate follows its rate. Before, a window that
    /// moved under 1 % of the run's bytes counted as stalled, and a run that
    /// never moves 1 % in ten seconds kept its first estimate for ever.
    #[test]
    fn a_long_runs_estimate_follows_its_rate() {
        let mut run = RunState::new(1, Vec::new(), RunKind::Run);
        run.status = RunStatus::Running;
        // A gigabyte at 100 KB/s, then at 200 KB/s: 0.1 % and 0.2 % a window.
        run.bytes_total = 1_000_000_000;
        run.first_sample = Some((0, 0, 0));
        run.samples = VecDeque::from(vec![(0, 0, 0)]);
        let mut shown = Vec::new();
        for t in 1..=40u64 {
            run.bytes = if t <= 20 {
                t * 100_000
            } else {
                2_000_000 + (t - 20) * 200_000
            };
            run.samples.push_back((t * 1_000, t, run.bytes));
            run.ended = Some(Duration::from_secs(t));
            if let Some(eta) = run.eta_ms() {
                shown.push((t, eta));
            }
        }
        let at = |t: u64| {
            shown
                .iter()
                .find(|(at, _)| *at == t)
                .expect("an estimate")
                .1
        };
        // 998 MB left at 100 KB/s is 9,980 s.
        assert_eq!(at(20) / 1_000, 9_980);
        // Ten seconds into the faster rate the window is all 200 KB/s:
        // 994 MB left is 4,970 s.
        assert_eq!(at(40) / 1_000, 4_970);
    }

    /// 0.30.1: a saved memory figure is never clamped below the derived one,
    /// however little is free this second, and so reads the same poll to poll.
    #[test]
    fn a_saved_figure_does_not_follow_the_free_memory_below_the_derived_one() {
        let derived = |value| crate::system::Derivation {
            value,
            reason: String::new(),
        };
        let unset = "SEMLITH_TEST_UNSET_LIMIT";
        for free_less_reserve in [0, 311, 400] {
            let limit = Limit::new(derived(512), Some(512), unset, free_less_reserve);
            assert_eq!(limit.value, 512, "at {free_less_reserve} MiB of room");
            assert_eq!(limit.ceiling, 512);
        }
        // Above the derived figure the room still caps it.
        assert_eq!(
            Limit::new(derived(512), Some(4096), unset, 1024).value,
            1024
        );
    }

    /// A write with several stores open has no "the" store, and guessing one
    /// writes into somebody's other repository.
    #[test]
    fn an_unnamed_write_among_several_stores_is_refused() {
        let state = State {
            server: Arc::new(Server::bind(0).expect("an ephemeral port")),
            fleet: Mutex::new(None),
            stores: RwLock::new(
                ["api", "cli"]
                    .into_iter()
                    .map(|name| {
                        Arc::new(Store::new(
                            name.to_string(),
                            PathBuf::from("/nowhere").join(name),
                            Vec::new(),
                            Vec::new(),
                            false,
                            0,
                        ))
                    })
                    .collect(),
            ),
            admission: Arc::new(Admission::new(1)),
            airgap: false,
            started: SystemTime::now(),
            debounce: Duration::from_millis(500),
            report: Arc::new(|_| {}),
            refusals: Mutex::new(BTreeMap::new()),
            late_writers: Mutex::new(Vec::new()),
            awaiting: Mutex::new(BTreeMap::new()),
            proxies: Mutex::new(BTreeMap::new()),
            clients: Mutex::new(BTreeMap::new()),
            mcp_fleet: Mutex::new(None),
            ledger: false,
            schedules: crate::schedule::Runner::new(),
            gone: Mutex::new(Vec::new()),
        };

        // An `Arc` because the reconciliation a miss triggers opens stores,
        // and opening one hands `Arc<State>` to the watcher thread it spawns.
        let state = Arc::new(state);
        let err = match state.writable(None) {
            Err(e) => e.to_string(),
            Ok(store) => panic!("an unnamed write picked {}", store.name),
        };
        assert!(err.contains("api, cli"), "{err}");
        assert_eq!(
            state.writable(Some("cli")).map(|s| s.name.clone()).ok(),
            Some("cli".to_string())
        );
        assert!(state.writable(Some("nope")).is_err());
    }

    #[test]
    fn a_store_is_compacted_only_when_idle_and_past_the_threshold_and_floor() {
        use crate::compact::{AUTO_FLOOR_BYTES, Footprint};
        let mb = 1024 * 1024;
        let dead = |total: u64, reclaimable: u64| Footprint {
            database: total,
            exact: 0,
            vectors: 0,
            reclaimable,
        };
        // 40 % dead, 40 MB: due at 25 %, idle.
        assert!(compaction_due(25, true, &dead(100 * mb, 40 * mb)));
        // A run, a queued job, a pause or a run about to arrive: not idle.
        assert!(!compaction_due(25, false, &dead(100 * mb, 40 * mb)));
        // Threshold 0 is off.
        assert!(!compaction_due(0, true, &dead(100 * mb, 40 * mb)));
        // Under the threshold.
        assert!(!compaction_due(25, true, &dead(100 * mb, 20 * mb)));
        // Past the share and under the floor: a tiny store is left alone.
        assert!(!compaction_due(
            25,
            true,
            &dead(2 * mb, AUTO_FLOOR_BYTES - 1)
        ));
    }
}