wsb-daemon 0.258.6441

//! Lightweight background-operation registry.
//!
//! Every background task the server runs (git status scan, sync, git mutations)
//! registers itself here, and so does every command line that announces itself
//! (`adopt`). The registry is in memory; a command line's output is in the log
//! file that command line writes, which is read back after a restart.
//!
//! # Design
//!
//! - `OpsRegistry` is a cheap-to-clone handle (Arc inside).
//! - `start_op(kind, label)` creates a `Running` op and returns an `OpHandle`.
//! - `OpHandle` is an RAII guard: call `.complete()` or `.fail(msg)` when the
//!   operation finishes; if the handle is dropped without being resolved the op
//!   is automatically marked `Failed` with a "dropped without completion" error.
//! - A `broadcast::Sender<OpsEvent>` fans out every state change to all SSE
//!   subscribers in real-time.
//! - Completed / failed ops are kept in a ring-buffer (capacity 100) so new
//!   SSE clients get recent history on connect.
use wsb::prelude::*;

use axum::{
    extract::{Path as AxumPath, State},
    http::StatusCode,
    response::sse::{Event, KeepAlive, Sse},
    Json,
};
use chrono::{DateTime, Utc};
use futures::stream::Stream;
use std::collections::{HashMap, VecDeque};
use std::convert::Infallible;
use tokio::sync::{broadcast, RwLock};
use uuid::Uuid;

use super::state::WebState;

// ---------------------------------------------------------------------------
// Public types
// ---------------------------------------------------------------------------

/// The kind of background operation.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum OpKind {
    /// Background git status scan of all repos in the workspace.
    GitScan,
    /// A user-initiated git mutation (stage/commit/push/pull/checkout/etc.).
    GitMutation,
    /// A file system operation (create, delete, rename, copy, move).
    FileOperation,
    /// Reading every repository in the workspace into the test catalogue.
    CatalogueScan,
    /// A command line this registry did not start.
    ///
    /// A run typed in a terminal announces itself and is shown here like any
    /// other: `wsb update`, `wsb sync`, a verb that has no button.
    Command,
}

impl OpKind {
    /// Whether this is upkeep the daemon does for itself, over and over.
    ///
    /// Such an operation is ONE row in the history, each run replacing the
    /// last under the same id and counting itself in `runs`. A row per run
    /// filled the whole history with them — 83 "Sync history" and 17
    /// "Scan 38 repos" were all of it — and every build and test that had
    /// finished in those minutes was pushed out.
    ///
    /// Exhaustive on purpose: a new kind has to be put on one side or the
    /// other by whoever adds it.
    pub fn recurs(self) -> bool {
        match self {
            // Re-reading what is there; each run supersedes the one before.
            OpKind::GitScan | OpKind::CatalogueScan => true,
            // Something somebody asked for, each its own event.
            OpKind::GitMutation | OpKind::FileOperation | OpKind::Command => false,
        }
    }
}

/// Lifecycle state of an operation. An op always starts `Running`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum OpStatus {
    Running,
    Completed,
    Failed,
}

/// A single operation record, snapshotted for serialization.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Op {
    pub id: Uuid,
    pub kind: OpKind,
    pub label: String,
    pub status: OpStatus,
    /// 0.0 → 1.0; meaningful while Running.
    pub progress: f32,
    /// Non-null only when status is Failed.
    pub error: Option<String>,
    pub started_at: DateTime<Utc>,
    pub ended_at: Option<DateTime<Utc>>,
    /// Whether this operation's output is streamable from `/ops/:id/output`.
    /// True exactly for command lines — announced, or read back from their
    /// logs; the daemon's in-process background tasks have no output.
    pub has_output: bool,
    /// Who started it.
    ///
    /// Shown because the two are not interchangeable to a person looking at
    /// the panel: one of these rows is something this daemon is doing, and the
    /// other is a window they have open somewhere with a build running in it.
    pub started_from: StartedFrom,
    /// The absolute path of the file this operation's output is written to.
    ///
    /// Shown by the page, always, because "where is that build's log" is the
    /// question a workbench cannot answer by pointing at itself: the answer
    /// has to work in a terminal, in an issue, and on the machine that ran it.
    /// None for the in-process operations, which produce no output, and for a
    /// command line that announced itself without one.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub log_path: Option<String>,
    /// The key this operation recurs under, for the background tasks that run
    /// again and again — history syncs, repository scans.
    ///
    /// Such a task is ONE row: each run replaces the last, under the same id,
    /// and `runs` counts them. Each run used to be a row of its own, and a sync
    /// every few seconds filled the whole history — 83 "Sync history" and 17
    /// "Scan 38 repos" were the entire panel, and every build and test that had
    /// finished in those minutes had been pushed out of it.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub recurring: Option<String>,
    /// How many times a recurring task has run; 1 for everything else.
    #[serde(default = "one_run")]
    pub runs: u64,
    /// How many of those runs failed, this one included once it has.
    ///
    /// Carried from run to run, because a row that reported only its latest
    /// outcome would hide a failure behind the next success.
    #[serde(default)]
    pub failed_runs: u64,
}

fn one_run() -> u64 {
    1
}

/// Where an operation was started.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum StartedFrom {
    /// This daemon is doing it, in process: a scan, a sync, a git mutation
    /// the page asked for.
    Workbench,
    /// It is running in a terminal and told this daemon about itself.
    ///
    /// The daemon does not own such a process: it did not start it, it holds
    /// no handle on it, and it deliberately does not accept one. A pid taken
    /// on trust from an unauthenticated request, used later to signal a
    /// process group, would turn the ops API into a way to kill anything on
    /// the machine. It is stopped in its terminal, where Ctrl-C is.
    CommandLine,
}

/// Events broadcast to SSE subscribers on every state change.
#[derive(Debug, Clone, Serialize)]
#[serde(tag = "event", rename_all = "snake_case")]
pub enum OpsEvent {
    Started(Op),
    Progress {
        id: Uuid,
        progress: f32,
        label: Option<String>,
    },
    Completed(Op),
    Failed(Op),
    /// Emitted once on initial SSE connect to replay the current snapshot.
    /// Contains all ops (running + recent completed/failed).
    Snapshot {
        ops: Vec<Op>,
    },
}

// ---------------------------------------------------------------------------
// Command output
// ---------------------------------------------------------------------------

/// One captured line of a command line's output.
///
/// A record, not a string. A command line reports `--jsonl` records, so what
/// arrives from it is already one of a closed set of document types — a test
/// finishing, a child process starting, a line of prose. Flattening that back
/// into text here would throw away exactly what the page needs to show a run
/// as something other than a wall of output.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct OutputLine {
    /// Monotonic per-operation sequence number. A client replays the snapshot,
    /// remembers the highest seq it saw, and drops live lines at or below it —
    /// so the subscribe-then-snapshot overlap is resolved exactly instead of
    /// by string comparison or luck.
    pub seq: u64,
    /// When the child said it, RFC 3339, as the child stamped it.
    ///
    /// Kept rather than re-stamped on arrival: the log file is read outside
    /// this daemon — tailed in a terminal, pasted into an issue — and a time
    /// that means "when the server got round to it" is the wrong answer to
    /// every question somebody asks a log. Absent for the lines that did not
    /// come through the packaged stream.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub ts: Option<String>,
    #[serde(flatten)]
    pub record: wsb::output::Record,
}

/// One line as an adopted run sends it.
///
/// The run's OWN sequence numbers are deliberately not carried. Those are the
/// run's, counted from the first thing it said including everything it said
/// before it was announced; this buffer numbers what IT holds, and a page
/// resolving the snapshot/live overlap has to compare like with like. What the
/// file records under the run's numbering is unaffected — that file is the
/// run's, and this is a view of it.
#[derive(Debug, Clone, Deserialize)]
pub struct AdoptedLine {
    #[serde(default)]
    pub ts: Option<String>,
    #[serde(flatten)]
    pub record: wsb::output::Record,
}

/// What an output subscriber receives after the snapshot.
#[derive(Debug, Clone)]
pub enum OutputEvent {
    Line(OutputLine),
    /// No further lines will arrive; the command line has finished.
    Closed,
}

/// Lines kept per operation. Beyond this the oldest lines are dropped and
/// `dropped` counts them, so a client can say "N earlier lines not shown"
/// instead of presenting a truncated log as the whole log.
const OUTPUT_LINE_CAPACITY: usize = 4000;

struct OutputBuffer {
    lines: VecDeque<OutputLine>,
    next_seq: u64,
    /// Lines evicted from the front to stay under capacity.
    dropped: u64,
    closed: bool,
    tx: broadcast::Sender<OutputEvent>,
}

impl OutputBuffer {
    fn new() -> Self {
        let (tx, _) = broadcast::channel(1024);
        Self {
            lines: VecDeque::new(),
            next_seq: 0,
            dropped: 0,
            closed: false,
            tx,
        }
    }

    fn append(&mut self, record: wsb::output::Record, ts: Option<String>) -> OutputLine {
        let line = OutputLine {
            seq: self.next_seq,
            ts,
            record,
        };
        self.next_seq += 1;
        self.lines.push_back(line.clone());
        while self.lines.len() > OUTPUT_LINE_CAPACITY {
            self.lines.pop_front();
            self.dropped += 1;
        }
        let _ = self.tx.send(OutputEvent::Line(line.clone()));
        line
    }

    fn close(&mut self) {
        if self.closed {
            return;
        }
        self.closed = true;
        let _ = self.tx.send(OutputEvent::Closed);
    }
}

// ---------------------------------------------------------------------------
// Internal state
// ---------------------------------------------------------------------------

struct Inner {
    /// All currently-running ops plus the last `HISTORY_CAPACITY` completed/failed.
    ops: VecDeque<Op>,
    /// Captured output, for exactly the command lines that announced
    /// themselves. Evicted together with its op.
    outputs: HashMap<Uuid, OutputBuffer>,
}

const HISTORY_CAPACITY: usize = 100;

impl Inner {
    fn new() -> Self {
        Self {
            ops: VecDeque::new(),
            outputs: HashMap::new(),
        }
    }

    fn insert(&mut self, op: Op) {
        self.ops.push_back(op);
    }

    /// Drop everything held IN MEMORY for an op that left the history ring.
    ///
    /// Not its log: the file is the record, and an operation that has aged out
    /// of the ring is exactly the one somebody is about to go looking for.
    fn forget(&mut self, id: Uuid) {
        self.outputs.remove(&id);
    }

    fn update_progress(&mut self, id: Uuid, progress: f32, label: Option<String>) {
        if let Some(op) = self.ops.iter_mut().find(|o| o.id == id) {
            op.progress = progress.clamp(0.0, 1.0);
            if let Some(ref l) = label {
                op.label = l.clone();
            }
        }
    }

    fn resolve(&mut self, id: Uuid, status: OpStatus, error: Option<String>) {
        if let Some(op) = self.ops.iter_mut().find(|o| o.id == id) {
            if status == OpStatus::Failed && op.status != OpStatus::Failed {
                op.failed_runs += 1;
            }
            op.status = status;
            op.error = error;
            op.ended_at = Some(Utc::now());
            if status != OpStatus::Running {
                op.progress = if status == OpStatus::Completed {
                    1.0
                } else {
                    op.progress
                };
            }
        }
        // Its output can gain no further lines, and a subscriber is told so.
        if let Some(buffer) = self.outputs.get_mut(&id) {
            buffer.close();
        }
        // Evict oldest non-running entries beyond history cap.
        while self
            .ops
            .iter()
            .filter(|o| o.status != OpStatus::Running)
            .count()
            > HISTORY_CAPACITY
        {
            // Find index of the oldest finished op.
            if let Some(idx) = self.ops.iter().position(|o| o.status != OpStatus::Running) {
                let evicted = self.ops.remove(idx);
                if let Some(op) = evicted {
                    self.forget(op.id);
                }
            } else {
                break;
            }
        }
    }

    fn snapshot(&self) -> Vec<Op> {
        self.ops.iter().cloned().collect()
    }
}

// ---------------------------------------------------------------------------
// OpsRegistry — the shared handle
// ---------------------------------------------------------------------------

/// Thread-safe, cheaply-cloneable handle to the ops registry.
#[derive(Clone)]
pub struct OpsRegistry {
    inner: Arc<RwLock<Inner>>,
    tx: broadcast::Sender<OpsEvent>,
    /// Where operation logs are written. None in tests and in any context with
    /// no workspace, where operations still run and simply are not recorded.
    logs_root: Option<Arc<PathBuf>>,
}

impl OpsRegistry {
    /// A registry that keeps output in memory only.
    pub fn new() -> Self {
        let (tx, _) = broadcast::channel(256);
        Self {
            inner: Arc::new(RwLock::new(Inner::new())),
            tx,
            logs_root: None,
        }
    }

    /// A registry that reads command lines' operation logs under `logs_root`,
    /// which is what makes an operation's output outlive the ring buffer's
    /// capacity and this process.
    pub fn with_logs(logs_root: PathBuf) -> Self {
        Self {
            logs_root: Some(Arc::new(logs_root)),
            ..Self::new()
        }
    }

    /// Where this registry reads operation logs from, when it reads them.
    pub fn logs_root(&self) -> Option<&Path> {
        self.logs_root.as_deref().map(PathBuf::as_path)
    }

    /// The log files a sweep must not remove: everything being written now.
    ///
    /// A command line opens its own log and announces itself with `adopt`, so
    /// its row carries a `log_path` — and that file is being appended to for
    /// as long as the run is running. Leaving it out once let a Mac filled to
    /// 1.3 GiB trip the free-space floor, and the janitor swept the log of an
    /// `wsb test` that was still running: the run carried on writing to a path
    /// that no longer existed.
    ///
    /// Only RUNNING operations: a finished log is an ordinary candidate, which
    /// is the whole point of bounding the directory.
    pub async fn protected_logs(&self) -> Vec<PathBuf> {
        let inner = self.inner.read().await;
        let mut held: Vec<PathBuf> = inner
            .ops
            .iter()
            .filter(|op| op.status == OpStatus::Running)
            .filter_map(|op| op.log_path.as_deref())
            .map(PathBuf::from)
            .collect();
        held.sort();
        held.dedup();
        held
    }

    /// Start a new in-process operation. Returns an `OpHandle` that must be
    /// resolved.
    pub async fn start_op(&self, kind: OpKind, label: impl Into<String>) -> OpHandle {
        let label = label.into();
        let started_at = Utc::now();
        // Upkeep the daemon does for itself recurs under its kind: one row,
        // each run replacing the last.
        let recurring = kind.recurs().then(|| {
            // Its serialised name, so the key is the one the page reads.
            serde_json::to_value(kind)
                .ok()
                .and_then(|value| value.as_str().map(str::to_string))
                .expect("an OpKind serialises to its name")
        });
        // A recurring task's last finished run gives this one its id and its
        // count, and leaves the history.
        let (id, runs, failed_runs) = match &recurring {
            Some(key) => {
                let mut inner = self.inner.write().await;
                let previous = inner.ops.iter().position(|o| {
                    o.recurring.as_deref() == Some(key.as_str()) && o.status != OpStatus::Running
                });
                match previous.and_then(|index| inner.ops.remove(index)) {
                    Some(last) => (last.id, last.runs + 1, last.failed_runs),
                    None => (Uuid::new_v4(), 1, 0),
                }
            }
            None => (Uuid::new_v4(), 1, 0),
        };

        let op = Op {
            id,
            kind,
            label,
            status: OpStatus::Running,
            progress: 0.0,
            error: None,
            started_at,
            ended_at: None,
            has_output: false,
            started_from: StartedFrom::Workbench,
            log_path: None,
            recurring,
            runs,
            failed_runs,
        };
        let event = OpsEvent::Started(op.clone());
        self.inner.write().await.insert(op);
        let _ = self.tx.send(event);
        OpHandle {
            id,
            registry: self.clone(),
            resolved: false,
        }
    }

    /// Take in an operation this registry did not start.
    ///
    /// A command line running in a terminal, which has told this daemon about
    /// itself. It is an ordinary op from here on — one row in the panel, one
    /// output stream, one log path:
    ///
    /// * its ID is the RUN's, minted where the run's log was opened, so the
    ///   row and the file are one thing rather than two records of it;
    /// * this daemon writes nothing to that file, because the run is writing
    ///   it, and a second writer on one path would interleave two copies of
    ///   every line;
    /// * this daemon holds nothing on the process. Taking a pid on trust from
    ///   the request and signalling its process group later would make the
    ///   ops API a way to kill anything on the machine.
    ///
    /// Announcing the same run twice replaces nothing and reports it: a second
    /// announcement means either a retry after a lost response or a bug, and
    /// silently starting a second row for one run is the wrong answer to both.
    #[allow(clippy::too_many_arguments)]
    pub async fn adopt(
        &self,
        id: Uuid,
        kind: OpKind,
        label: String,
        started_at: DateTime<Utc>,
        log_path: Option<String>,
        lines: Vec<AdoptedLine>,
        dropped: u64,
    ) -> Result<(), String> {
        let op = Op {
            id,
            kind,
            label,
            status: OpStatus::Running,
            progress: 0.0,
            error: None,
            started_at,
            ended_at: None,
            has_output: true,
            started_from: StartedFrom::CommandLine,
            log_path,
            recurring: None,
            runs: 1,
            failed_runs: 0,
        };
        {
            let mut inner = self.inner.write().await;
            if inner.ops.iter().any(|held| held.id == id) {
                return Err(format!("operation {id} is already known to this daemon"));
            }
            inner.insert(op.clone());
            let mut buffer = OutputBuffer::new();
            buffer.dropped = dropped;
            for line in lines {
                buffer.append(line.record, line.ts);
            }
            inner.outputs.insert(id, buffer);
        }
        let _ = self.tx.send(OpsEvent::Started(op));
        Ok(())
    }

    /// Lines an adopted run has said since the last time it said any.
    ///
    /// Refused when the id is not an operation this daemon is holding — a run
    /// whose announcement was lost, or one that has aged out of history. The
    /// caller stops reporting on that; it does not retry into a registry that
    /// would have to invent an operation to accept them.
    ///
    /// Held in memory only: the run is writing its own log.
    pub async fn adopted_lines(&self, id: Uuid, lines: Vec<AdoptedLine>) -> Result<(), String> {
        let mut inner = self.inner.write().await;
        let buffer = inner
            .outputs
            .get_mut(&id)
            .ok_or_else(|| format!("no operation {id} is being reported to this daemon"))?;
        for line in lines {
            buffer.append(line.record, line.ts);
        }
        Ok(())
    }

    /// How an adopted run ended.
    pub async fn adopted_finished(&self, id: Uuid, exit_code: i32) -> Result<(), String> {
        {
            let inner = self.inner.read().await;
            if !inner.ops.iter().any(|held| held.id == id) {
                return Err(format!("no operation {id} is being reported to this daemon"));
            }
        }
        let (status, error) = if exit_code == 0 {
            (OpStatus::Completed, None)
        } else {
            (
                OpStatus::Failed,
                Some(format!("exited with status {exit_code}")),
            )
        };
        self.resolve_op(id, status, error).await;
        self.close_output(id).await;
        Ok(())
    }

    /// Mark an op's output finished; subscribers receive `Closed`.
    pub async fn close_output(&self, id: Uuid) {
        let mut inner = self.inner.write().await;
        if let Some(buffer) = inner.outputs.get_mut(&id) {
            buffer.close();
        }
    }

    /// The captured lines so far, how many earlier lines were dropped, and
    /// whether the output is finished. None when the op has no output.
    ///
    /// Memory only. `output_snapshot` is what serves a page, and it prefers
    /// the file — see there.
    async fn buffered_snapshot(&self, id: Uuid) -> Option<(Vec<OutputLine>, u64, bool)> {
        let inner = self.inner.read().await;
        inner.outputs.get(&id).map(|buffer| {
            (
                buffer.lines.iter().cloned().collect(),
                buffer.dropped,
                buffer.closed,
            )
        })
    }

    /// Everything this operation has said, for a page that just opened it.
    ///
    /// The FILE is preferred whenever the memory buffer has lost lines, and it
    /// is the only source for an operation this process never ran — one from
    /// before a restart, or one that has aged out of the history ring. That is
    /// the whole point of writing it: what a page shows must not depend on how
    /// long ago the run happened or how much it said.
    ///
    /// The memory buffer still answers for a live operation that has not
    /// dropped anything, which is the common case and does no I/O.
    pub async fn output_snapshot(&self, id: Uuid) -> Option<(Vec<OutputLine>, u64, bool)> {
        let buffered = self.buffered_snapshot(id).await;
        match &buffered {
            // Whole in memory: nothing on disk can be more complete.
            Some((_, 0, _)) => return buffered,
            _ => {}
        }

        // No file — a registry with no logs root, or an operation whose log
        // could not be opened. What memory holds is then the whole of what
        // there is, and the dropped count is what says so.
        let Some(path) = self.log_path_of(id).await else {
            return buffered;
        };
        match tokio::task::spawn_blocking(move || super::op_log::read(&path)).await {
            Ok(Ok(log)) => {
                // Closed when the file says the operation ended. A running
                // operation's file has no footer, and the page keeps its live
                // subscription.
                let closed = log.footer.is_some();
                let lines = log
                    .lines
                    .into_iter()
                    .map(super::op_log::output_line_of)
                    .collect();
                Some((lines, 0, closed))
            }
            // A file that cannot be read is not a reason to show nothing: what
            // memory has is still what the operation said, and the dropped
            // count still says what is missing from it.
            Ok(Err(error)) => {
                tracing::warn!(%error, "falling back to the in-memory output buffer");
                buffered
            }
            Err(error) => {
                tracing::warn!(%error, "reading an operation log panicked");
                buffered
            }
        }
    }

    /// The file backing one operation: from the op record while it is in
    /// history, from the ops directory once it is not.
    async fn log_path_of(&self, id: Uuid) -> Option<PathBuf> {
        {
            let inner = self.inner.read().await;
            if let Some(op) = inner.ops.iter().find(|o| o.id == id) {
                if let Some(path) = op.log_path.as_ref() {
                    return Some(PathBuf::from(path));
                }
            }
        }
        // Memory does not know it. That is the ORDINARY case for an operation
        // this daemon did not run — one from before a restart — and it is
        // exactly when the file matters most, because the live stream it would
        // otherwise be served from no longer exists. The log is on disk under
        // a name derived from the id; find it there.
        let root = self.logs_root.clone()?;
        let found =
            tokio::task::spawn_blocking(move || wsb::logs::find_op_log(&root, &id)).await;
        found.ok().flatten()
    }

    /// Subscribe to an op's live output. None when the op has no output.
    pub async fn subscribe_output(&self, id: Uuid) -> Option<broadcast::Receiver<OutputEvent>> {
        let inner = self.inner.read().await;
        inner.outputs.get(&id).map(|buffer| buffer.tx.subscribe())
    }

    /// Update an op's progress (0.0–1.0) and optionally change its label.
    /// Silently ignored if the op id is not found (e.g. already completed).
    pub async fn report_progress(&self, id: Uuid, progress: f32, label: Option<String>) {
        {
            let mut inner = self.inner.write().await;
            inner.update_progress(id, progress, label.clone());
        }
        let _ = self.tx.send(OpsEvent::Progress {
            id,
            progress,
            label,
        });
    }

    /// Mark an op as completed.
    async fn complete(&self, id: Uuid) {
        self.resolve_op(id, OpStatus::Completed, None).await;
    }

    /// Mark an op as failed with an error message.
    async fn fail(&self, id: Uuid, error: String) {
        self.resolve_op(id, OpStatus::Failed, Some(error)).await;
    }

    /// Resolve an op and tell subscribers how it ended.
    async fn resolve_op(&self, id: Uuid, status: OpStatus, error: Option<String>) {
        let op = {
            let mut inner = self.inner.write().await;
            inner.resolve(id, status, error);
            inner.ops.iter().find(|o| o.id == id).cloned()
        };
        if let Some(op) = op {
            let event = match status {
                OpStatus::Completed => OpsEvent::Completed(op),
                _ => OpsEvent::Failed(op),
            };
            let _ = self.tx.send(event);
        }
    }

    /// Resolve adopted operations whose process is gone, from their own logs.
    ///
    /// An in-process operation cannot be lost: `OpHandle` is an RAII guard and
    /// its `Drop` fails anything unresolved. An ADOPTED one is a
    /// process in somebody's terminal, and it resolves only when that process
    /// reports its exit — so a run that was killed, or whose machine went
    /// down, leaves a row saying "running" for ever. Four sat in the panel at
    /// once, the oldest an hour old, all of them long finished.
    ///
    /// The pid is deliberately not the answer. This daemon holds no handle on
    /// an adopted process and refuses to accept one, because a pid taken on
    /// trust from an unauthenticated request and later signalled would make
    /// the ops API a way to kill anything on the machine — see
    /// [`StartedFrom::CommandLine`].
    ///
    /// Its LOG is the answer. Every command line writes a footer saying how
    /// it ended, and that footer is what rebuilds history when this daemon
    /// restarts. It is the running
    /// process's own account of its exit: it needs no pid and trusts nothing
    /// from a request.
    ///
    /// A log with NO footer is left alone. That is a process still running, or
    /// one killed so hard it never wrote one, and the two are indistinguishable
    /// from here — resolving on absence would report a live build as finished,
    /// which is worse than the staleness, because somebody acts on it.
    pub async fn resolve_finished_adopted(&self) {
        let candidates: Vec<(Uuid, String)> = {
            let inner = self.inner.read().await;
            inner
                .ops
                .iter()
                .filter(|op| {
                    op.status == OpStatus::Running
                        && op.started_from == StartedFrom::CommandLine
                })
                .filter_map(|op| op.log_path.clone().map(|path| (op.id, path)))
                .collect()
        };
        for (id, path) in candidates {
            // Off the async executor: one file read per unresolved adopted op,
            // and there are single digits of those.
            let envelope = tokio::task::spawn_blocking(move || {
                super::op_log::read_envelope(std::path::Path::new(&path))
            })
            .await;
            let Ok(Ok((_, Some(footer)))) = envelope else {
                continue;
            };
            let (status, error) = match footer.status {
                super::op_log::Status::Completed => (OpStatus::Completed, None),
                super::op_log::Status::Failed => (OpStatus::Failed, footer.error.clone()),
            };
            self.resolve_op(id, status, error).await;
        }
    }

    /// Return a snapshot of all current ops (running + recent history).
    pub async fn snapshot(&self) -> Vec<Op> {
        self.inner.read().await.snapshot()
    }

    /// Fill history from the operation logs on disk, newest last.
    ///
    /// This is what makes a restarted daemon continue to know about the build
    /// somebody ran before it restarted. Without it the workbench's history
    /// starts empty every time the daemon is bounced, and the log files sit
    /// there unreachable from the page — visible only to somebody who thought
    /// to look in the directory.
    ///
    /// Only the ENVELOPE of each file is read: what it was, and how it ended.
    /// The lines are read when somebody opens one.
    ///
    /// An operation with no footer did not finish while the last daemon was
    /// watching. It is adopted as failed and says so: showing it as still
    /// running would claim a process nobody is reporting on.
    pub async fn adopt_recent_logs(&self, limit: usize) -> usize {
        let Some(root) = self.logs_root.as_deref().cloned() else {
            return 0;
        };
        let found = tokio::task::spawn_blocking(move || {
            super::op_log::recent_logs(&root, limit)
                .into_iter()
                .filter_map(|path| {
                    super::op_log::read_envelope(&path)
                        .ok()
                        .map(|(header, footer)| (path, header, footer))
                })
                .collect::<Vec<_>>()
        })
        .await
        .unwrap_or_default();

        let mut adopted = 0;
        let mut inner = self.inner.write().await;
        // Oldest first: history is a queue, and adopting newest-first would
        // put yesterday's run after today's in every list drawn from it.
        for (path, header, footer) in found.into_iter().rev() {
            if inner.ops.iter().any(|op| op.id == header.id) {
                continue;
            }
            let (status, ended_at, error) = match &footer {
                Some(footer) => (
                    match footer.status {
                        super::op_log::Status::Completed => OpStatus::Completed,
                        super::op_log::Status::Failed => OpStatus::Failed,
                    },
                    Some(footer.ended_at),
                    footer.error.clone(),
                ),
                None => (
                    OpStatus::Failed,
                    None,
                    Some("the daemon stopped while this was running".to_string()),
                ),
            };
            // Every log on disk is a command line's: this daemon's own work is
            // in process and writes none.
            inner.insert(Op {
                id: header.id,
                kind: OpKind::Command,
                label: header.label,
                status,
                progress: if status == OpStatus::Completed { 1.0 } else { 0.0 },
                error,
                started_at: header.started_at,
                ended_at,
                has_output: true,
                started_from: StartedFrom::CommandLine,
                log_path: Some(path.display().to_string()),
                recurring: None,
                runs: 1,
                failed_runs: 0,
            });
            adopted += 1;
        }
        adopted
    }

    /// Subscribe to the broadcast stream of events.
    pub fn subscribe(&self) -> broadcast::Receiver<OpsEvent> {
        self.tx.subscribe()
    }
}

// ---------------------------------------------------------------------------
// OpHandle — RAII guard
// ---------------------------------------------------------------------------

/// A handle to a running operation. Must be explicitly resolved via
/// `.complete()` or `.fail(msg)`. If dropped without resolution, the op is
/// automatically failed with "dropped without completion".
pub struct OpHandle {
    id: Uuid,
    registry: OpsRegistry,
    resolved: bool,
}

impl OpHandle {
    pub fn id(&self) -> Uuid {
        self.id
    }

    /// Report progress without resolving. Safe to call many times.
    pub async fn progress(&self, progress: f32) {
        self.registry.report_progress(self.id, progress, None).await;
    }

    /// Mark the op as successfully completed. Consumes the handle.
    pub async fn complete(mut self) {
        self.resolved = true;
        self.registry.complete(self.id).await;
    }

    /// Mark the op as failed. Consumes the handle.
    pub async fn fail(mut self, error: impl Into<String>) {
        self.resolved = true;
        self.registry.fail(self.id, error.into()).await;
    }
}

impl Drop for OpHandle {
    fn drop(&mut self) {
        if self.resolved {
            return;
        }
        // Synchronously schedule the fail — we can't await inside Drop.
        // Use try_write so we don't block the async runtime; if the lock
        // is contended we skip the eviction step (benign: the entry stays
        // Running until garbage-collected by the next successful write).
        let id = self.id;
        let registry = self.registry.clone();
        tokio::spawn(async move {
            registry
                .fail(id, "dropped without completion".to_string())
                .await;
        });
    }
}

// ---------------------------------------------------------------------------
// HTTP handlers
// ---------------------------------------------------------------------------

/// GET /api/v1/ops — every operation this registry holds, running and recent.
pub async fn ops_snapshot(State(state): State<WebState>) -> Json<Vec<Op>> {
    Json(state.ops.snapshot().await)
}

/// GET /api/v1/ops/live — the ids whose output can still be shown.
///
/// Asked once when the workbench restores its layout, so a tab pointing at an
/// operation whose log is gone is dropped rather than opened. Three separate
/// mechanisms delete underneath such a tab — history eviction, the log
/// janitor, and a daemon restart — and none of them knows the tab exists.
///
/// The answer is the SAME predicate `/ops/:id/output` serves on: an id is here
/// exactly when that endpoint would not 404. Anything else would be a second
/// definition of "has output", and a tab restored against one while the stream
/// answered the other is precisely the spinner this removes.
///
/// One call rather than one per tab: a restore with a dozen tabs should not be
/// a dozen round trips before anything is on screen.
pub async fn op_live_ids(State(state): State<WebState>) -> Json<Vec<Uuid>> {
    let known = state.ops.snapshot().await;
    let mut live = Vec::with_capacity(known.len());
    for op in known {
        if state.ops.output_snapshot(op.id).await.is_some() {
            live.push(op.id);
        }
    }
    Json(live)
}

/// GET /api/v1/ops/:id/output — SSE stream of one command line's output.
///
/// The subscription is taken BEFORE the snapshot, and every line carries a
/// sequence number, so the overlap between the two is dropped exactly: a line
/// is shown once whether it arrived in the snapshot or live. 404 when the op
/// does not exist, is an in-process task, or is gone from history and disk.
pub async fn op_output_stream(
    State(state): State<WebState>,
    AxumPath(id): AxumPath<Uuid>,
) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, (StatusCode, Json<serde_json::Value>)>
{
    let registry = state.ops.clone();
    // No live subscription is not the same as no output. An operation that
    // finished before this daemon started has a file and nothing else, and it
    // is served from the file as a stream that is already closed — which is
    // what it is.
    let rx = registry.subscribe_output(id).await;
    let Some((lines, dropped, closed)) = registry.output_snapshot(id).await else {
        return Err((
            StatusCode::NOT_FOUND,
            Json(serde_json::json!({
                "error": format!("operation {id} has no captured output"),
            })),
        ));
    };
    // Nothing to subscribe to and nothing left to say.
    let closed = closed || rx.is_none();

    let stream = async_stream::stream! {
        let mut last_seq: i64 = lines.last().map(|l| l.seq as i64).unwrap_or(-1);
        let snapshot = serde_json::json!({
            "event": "snapshot",
            "lines": lines,
            "dropped": dropped,
            "closed": closed,
        });
        yield Ok(Event::default().event("output").data(snapshot.to_string()));
        if closed {
            return;
        }

        let Some(mut rx) = rx else {
            return;
        };
        loop {
            match rx.recv().await {
                Ok(OutputEvent::Line(line)) => {
                    if (line.seq as i64) <= last_seq {
                        continue;
                    }
                    last_seq = line.seq as i64;
                    let payload = serde_json::json!({
                        "event": "line",
                        "line": line,
                    });
                    yield Ok(Event::default().event("output").data(payload.to_string()));
                }
                Ok(OutputEvent::Closed) => {
                    let payload = serde_json::json!({ "event": "closed" });
                    yield Ok(Event::default().event("output").data(payload.to_string()));
                    break;
                }
                Err(broadcast::error::RecvError::Lagged(n)) => {
                    // Fell behind; re-send the snapshot so the client is whole
                    // again, and say how many lines the buffer dropped.
                    tracing::warn!("op output SSE client lagged by {} lines", n);
                    let Some((lines, dropped, closed)) = registry.output_snapshot(id).await else {
                        break;
                    };
                    last_seq = lines.last().map(|l| l.seq as i64).unwrap_or(-1);
                    let payload = serde_json::json!({
                        "event": "snapshot",
                        "lines": lines,
                        "dropped": dropped,
                        "closed": closed,
                    });
                    yield Ok(Event::default().event("output").data(payload.to_string()));
                    if closed {
                        break;
                    }
                }
                Err(broadcast::error::RecvError::Closed) => break,
            }
        }
    };

    Ok(Sse::new(stream).keep_alive(KeepAlive::default()))
}

/// What a command line says about itself when it announces.
#[derive(Debug, Deserialize)]
pub struct AdoptRequest {
    /// The run's own id — the one its log file was opened under.
    pub id: Uuid,
    pub label: String,
    pub started_at: DateTime<Utc>,
    #[serde(default)]
    pub log_path: Option<String>,
    /// Everything it said before it was worth announcing, so the panel opens
    /// on the whole run rather than on its middle.
    #[serde(default)]
    pub lines: Vec<AdoptedLine>,
    /// How many earlier lines the run had to drop to hold the rest.
    #[serde(default)]
    pub dropped: u64,
}

/// POST /api/v1/ops/adopt — a command line reports itself as an operation.
///
/// A run typed in a terminal appears in the same registry, with the same shape,
/// next to the daemon's own work — as a command, under the label it gave.
pub async fn op_adopt(
    State(state): State<WebState>,
    Json(request): Json<AdoptRequest>,
) -> Result<Json<serde_json::Value>, (StatusCode, Json<serde_json::Value>)> {
    let kind = OpKind::Command;
    let label = request.label;

    state
        .ops
        .adopt(
            request.id,
            kind,
            label,
            request.started_at,
            request.log_path,
            request.lines,
            request.dropped,
        )
        .await
        .map_err(|error| {
            (
                StatusCode::CONFLICT,
                Json(serde_json::json!({ "error": error })),
            )
        })?;
    Ok(Json(serde_json::json!({ "adopted": request.id })))
}

#[derive(Debug, Deserialize)]
pub struct AdoptedLinesRequest {
    pub lines: Vec<AdoptedLine>,
}

/// POST /api/v1/ops/:id/lines — more of what an adopted run is saying.
pub async fn op_adopted_lines(
    State(state): State<WebState>,
    AxumPath(id): AxumPath<Uuid>,
    Json(request): Json<AdoptedLinesRequest>,
) -> Result<Json<serde_json::Value>, (StatusCode, Json<serde_json::Value>)> {
    let taken = request.lines.len();
    state
        .ops
        .adopted_lines(id, request.lines)
        .await
        .map_err(|error| {
            (
                StatusCode::NOT_FOUND,
                Json(serde_json::json!({ "error": error })),
            )
        })?;
    Ok(Json(serde_json::json!({ "lines": taken })))
}

#[derive(Debug, Deserialize)]
pub struct AdoptedFinishedRequest {
    pub exit_code: i32,
}

/// POST /api/v1/ops/:id/finished — how an adopted run ended.
pub async fn op_adopted_finished(
    State(state): State<WebState>,
    AxumPath(id): AxumPath<Uuid>,
    Json(request): Json<AdoptedFinishedRequest>,
) -> Result<Json<serde_json::Value>, (StatusCode, Json<serde_json::Value>)> {
    state
        .ops
        .adopted_finished(id, request.exit_code)
        .await
        .map_err(|error| {
            (
                StatusCode::NOT_FOUND,
                Json(serde_json::json!({ "error": error })),
            )
        })?;
    Ok(Json(serde_json::json!({ "finished": id })))
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

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

    /// Verifies that a Running op transitions to Completed and broadcasts the
    /// right events. Also checks that the snapshot reflects the final state.
    #[tokio::test]
    async fn op_lifecycle_running_to_completed() {
        let registry = OpsRegistry::new();
        let mut rx = registry.subscribe();

        let handle = registry.start_op(OpKind::CatalogueScan, "sync test").await;
        let id = handle.id();

        // Should have broadcast a Started event.
        let started_event = rx.try_recv().expect("Started event should be immediate");
        match started_event {
            OpsEvent::Started(ref op) => {
                assert_eq!(op.id, id);
                assert_eq!(op.status, OpStatus::Running);
                assert_eq!(op.label, "sync test");
            }
            other => panic!("expected Started, got {:?}", other),
        }

        // Update progress.
        handle.progress(0.5).await;
        let progress_event = rx.try_recv().expect("Progress event should be broadcast");
        match progress_event {
            OpsEvent::Progress {
                id: eid, progress, ..
            } => {
                assert_eq!(eid, id);
                assert!((progress - 0.5).abs() < f32::EPSILON);
            }
            other => panic!("expected Progress, got {:?}", other),
        }

        // Complete.
        handle.complete().await;
        let completed_event = rx.try_recv().expect("Completed event should be broadcast");
        match completed_event {
            OpsEvent::Completed(ref op) => {
                assert_eq!(op.id, id);
                assert_eq!(op.status, OpStatus::Completed);
                assert!((op.progress - 1.0).abs() < f32::EPSILON);
                assert!(op.ended_at.is_some());
            }
            other => panic!("expected Completed, got {:?}", other),
        }

        // Snapshot reflects final state.
        let snap = registry.snapshot().await;
        let found = snap
            .iter()
            .find(|o| o.id == id)
            .expect("op must be in snapshot");
        assert_eq!(found.status, OpStatus::Completed);
    }

    /// Verifies that a Running op transitions to Failed when fail() is called.
    #[tokio::test]
    async fn op_lifecycle_running_to_failed() {
        let registry = OpsRegistry::new();
        let mut rx = registry.subscribe();

        let handle = registry.start_op(OpKind::GitMutation, "push origin").await;
        let id = handle.id();
        let _ = rx.try_recv(); // consume Started

        handle.fail("authentication failure").await;

        let event = rx.try_recv().expect("Failed event should be broadcast");
        match event {
            OpsEvent::Failed(ref op) => {
                assert_eq!(op.id, id);
                assert_eq!(op.status, OpStatus::Failed);
                assert_eq!(op.error.as_deref(), Some("authentication failure"));
            }
            other => panic!("expected Failed, got {:?}", other),
        }
    }

    /// Verifies that dropping an unresolved OpHandle marks the op Failed.
    #[tokio::test]
    async fn unresolved_drop_fails_the_op() {
        let registry = OpsRegistry::new();
        let mut rx = registry.subscribe();

        let id = {
            let handle = registry.start_op(OpKind::GitScan, "scan").await;
            let id = handle.id();
            let _ = rx.try_recv(); // consume Started
            id
            // handle dropped here without complete/fail
        };

        // Give the spawned task time to run.
        tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;

        let snap = registry.snapshot().await;
        let found = snap
            .iter()
            .find(|o| o.id == id)
            .expect("op must still be in snapshot");
        assert_eq!(
            found.status,
            OpStatus::Failed,
            "dropped op must be Failed, not {:?}",
            found.status
        );
        assert_eq!(found.error.as_deref(), Some("dropped without completion"));
    }

    /// Verifies that concurrent ops each receive independent broadcast events
    /// and that their IDs never cross-contaminate.
    #[tokio::test]
    async fn concurrent_ops_are_independent() {
        let registry = OpsRegistry::new();
        // Spawn 10 concurrent ops, each progresses to 1.0 and completes.
        let mut handles: Vec<OpHandle> = Vec::new();
        let mut ids: Vec<Uuid> = Vec::new();
        for i in 0..10 {
            let h = registry
                .start_op(OpKind::GitScan, format!("scan {}", i))
                .await;
            ids.push(h.id());
            handles.push(h);
        }

        // Complete all.
        for h in handles {
            h.progress(1.0).await;
            h.complete().await;
        }

        let snap = registry.snapshot().await;
        for id in &ids {
            let op = snap
                .iter()
                .find(|o| o.id == *id)
                .expect("every op must be in snapshot");
            assert_eq!(
                op.status,
                OpStatus::Completed,
                "op {} must be Completed",
                id
            );
        }
    }

    /// Verifies that the ring-buffer evicts old completed ops when it exceeds
    /// HISTORY_CAPACITY, while active (Running) ops are never evicted.
    #[tokio::test]
    async fn history_ring_buffer_evicts_old_completed() {
        let registry = OpsRegistry::new();

        // Fill beyond capacity.
        // A kind that is a row per run: a recurring one could not fill it.
        for i in 0..(HISTORY_CAPACITY + 20) {
            let h = registry.start_op(OpKind::GitMutation, format!("commit {}", i)).await;
            h.complete().await;
        }

        let snap = registry.snapshot().await;
        // Only HISTORY_CAPACITY completed ops retained.
        let completed: Vec<_> = snap
            .iter()
            .filter(|o| o.status == OpStatus::Completed)
            .collect();
        assert!(
            completed.len() <= HISTORY_CAPACITY,
            "completed ops {} must not exceed capacity {}",
            completed.len(),
            HISTORY_CAPACITY
        );
    }

    /// TEST12151: a recurring background task is one row, and cannot push real
    /// work out of the history.
    ///
    /// A history sync every few seconds used to add a row per run: the panel
    /// held 83 "Sync history" and 17 "Scan 38 repos" and nothing else, and every
    /// test and build that had finished in those minutes had been evicted.
    #[tokio::test]
    async fn test12151_a_recurring_task_is_one_row_and_evicts_nothing() {
        let registry = OpsRegistry::new();

        let build = registry.start_op(OpKind::GitMutation, "a real commit").await;
        let build_id = build.id();
        build.complete().await;

        let mut ids = std::collections::HashSet::new();
        for _ in 0..(HISTORY_CAPACITY + 50) {
            let h = registry
                .start_op(OpKind::CatalogueScan, "Sync history")
                .await;
            ids.insert(h.id());
            h.complete().await;
        }

        let snap = registry.snapshot().await;
        let syncs: Vec<_> = snap.iter().filter(|o| o.label == "Sync history").collect();
        assert_eq!(syncs.len(), 1, "every run of one recurring task is one row: {}", syncs.len());
        assert_eq!(syncs[0].runs, (HISTORY_CAPACITY + 50) as u64, "the row counts its runs");
        assert_eq!(ids.len(), 1, "the row keeps one id, so a page updates it in place");
        assert!(
            snap.iter().any(|o| o.id == build_id),
            "a finished build was evicted by a recurring task's runs"
        );

        // Two runs at once are both real, so both are rows.
        let first = registry.start_op(OpKind::GitScan, "Scan").await;
        let second = registry.start_op(OpKind::GitScan, "Scan").await;
        assert_ne!(first.id(), second.id(), "a run that overlaps another is a row of its own");
        first.complete().await;
        second.complete().await;
    }

    /// Verifies that the snapshot the event stream emits on connect (the
    /// Snapshot event) contains all currently-running ops so a new client sees
    /// live work.
    #[tokio::test]
    async fn snapshot_includes_running_ops() {
        let registry = OpsRegistry::new();

        let h1 = registry.start_op(OpKind::GitScan, "scan").await;
        let h2 = registry.start_op(OpKind::CatalogueScan, "sync").await;
        let id1 = h1.id();
        let id2 = h2.id();

        let snap = registry.snapshot().await;
        let ids_in_snap: Vec<Uuid> = snap.iter().map(|o| o.id).collect();
        assert!(
            ids_in_snap.contains(&id1),
            "snapshot must include running op1"
        );
        assert!(
            ids_in_snap.contains(&id2),
            "snapshot must include running op2"
        );

        h1.complete().await;
        h2.fail("err").await;
    }

    fn adopted(text: &str) -> AdoptedLine {
        AdoptedLine {
            ts: Some("2026-08-29T00:00:00Z".to_string()),
            record: wsb::output::Record::Stdout {
                text: text.to_string(),
            },
        }
    }

    /// Announce a command line with nothing said yet and no log of its own.
    async fn announce(registry: &OpsRegistry, id: Uuid, log_path: Option<String>) {
        registry
            .adopt(
                id,
                OpKind::Command,
                "build".to_string(),
                Utc::now(),
                log_path,
                Vec::new(),
                0,
            )
            .await
            .expect("a run this registry has not seen is adoptable");
    }

    /// Write a command line's operation log the way a command line writes it:
    /// through the library's writer, under the logs root, one line per text,
    /// with a footer only when the run finished.
    fn command_line_log(
        logs_root: &std::path::Path,
        id: Uuid,
        label: &str,
        texts: &[String],
        finished: bool,
    ) -> String {
        let writer = wsb::logs::op_log::OpLogWriter::create(
            logs_root,
            id,
            "test",
            label,
            &["wsb".to_string(), "test".to_string()],
            Utc::now(),
        )
        .expect("a log is created");
        for (seq, text) in texts.iter().enumerate() {
            writer.write_line(&wsb::output::Line {
                seq: seq as u64,
                ts: "2026-08-29T00:00:00Z".to_string(),
                record: wsb::output::Record::Stdout { text: text.clone() },
            });
        }
        if finished {
            writer.finish(wsb::logs::op_log::Status::Completed, None);
        }
        writer.path().display().to_string()
    }

    /// TEST8946: output lines are sequenced, capped, and the cap is counted.
    ///
    /// A client reconciles snapshot and live stream by seq; if seqs repeated or
    /// skipped under eviction, lines would double or vanish silently. And a
    /// buffer that dropped lines without saying so would present a truncated
    /// log as the whole log.
    #[tokio::test]
    async fn test8946_output_is_sequenced_and_eviction_is_counted() {
        let registry = OpsRegistry::new();
        let id = Uuid::new_v4();
        announce(&registry, id, None).await;

        let lines = (0..(OUTPUT_LINE_CAPACITY + 25))
            .map(|i| adopted(&format!("line {i}")))
            .collect();
        registry
            .adopted_lines(id, lines)
            .await
            .expect("an announced run may say more");

        let (lines, dropped, closed) = registry
            .output_snapshot(id)
            .await
            .expect("an announced run has an output buffer");
        assert!(!closed);
        assert_eq!(lines.len(), OUTPUT_LINE_CAPACITY);
        assert_eq!(dropped, 25, "eviction must be counted, not silent");
        // Seqs are contiguous and continue across eviction.
        assert_eq!(lines.first().unwrap().seq, 25);
        assert_eq!(
            lines.last().unwrap().seq as usize,
            OUTPUT_LINE_CAPACITY + 24
        );
    }

    /// TEST8986: the file is what a page gets once the ring has lost lines.
    ///
    /// This is the whole reason the file exists. A long build says more than
    /// the in-memory ring can hold; served from memory, opening that build's
    /// tab showed the tail and a note that some number of earlier lines were
    /// gone — and the beginning of a build is where the answer usually is.
    #[tokio::test]
    async fn test8986_a_long_run_is_served_whole_from_its_file() {
        let dir = tempfile::tempdir().unwrap();
        let id = Uuid::new_v4();
        let total = OUTPUT_LINE_CAPACITY + 25;
        let texts: Vec<String> = (0..total).map(|i| format!("line {i}")).collect();
        let path = command_line_log(dir.path(), id, "build", &texts, true);

        let registry = OpsRegistry::with_logs(dir.path().to_path_buf());
        announce(&registry, id, Some(path)).await;
        registry
            .adopted_lines(id, texts.iter().map(|text| adopted(text)).collect())
            .await
            .expect("an announced run may say more");
        registry
            .adopted_finished(id, 0)
            .await
            .expect("an announced run may end");

        let (lines, dropped, closed) = registry
            .output_snapshot(id)
            .await
            .expect("the operation has output");
        assert_eq!(lines.len(), total, "every line, not the ring's tail");
        assert_eq!(dropped, 0, "nothing is missing, so nothing is reported as missing");
        assert!(closed);
        match &lines[0].record {
            wsb::output::Record::Stdout { text } => assert_eq!(text, "line 0"),
            other => panic!("the first line is not the first line: {other:?}"),
        }
    }

    /// TEST8987: an operation this process never saw is still readable.
    ///
    /// The daemon restarts — a rebuild, a crash, a machine reboot — and the
    /// build somebody ran ten minutes ago must still be in the list and still
    /// open. Without reading the log files, restarting the daemon emptied the
    /// history and every tab pointing into it answered 404.
    #[tokio::test]
    async fn test8987_a_log_outlives_the_process_that_wrote_it() {
        let dir = tempfile::tempdir().unwrap();
        let id = Uuid::new_v4();
        command_line_log(dir.path(), id, "the declared suite", &["it ran".to_string()], true);

        // A daemon that has never heard of it.
        let registry = OpsRegistry::with_logs(dir.path().to_path_buf());

        // The OUTPUT is reachable immediately, before any adoption: the file
        // is on disk under a name derived from the id, and a page asking for
        // an operation's output is asking about one it can already name. This
        // is what a person refreshing the page on a finished run relies on —
        // adoption is capped, and a run older than the cap still has its log.
        let (lines, _, closed) = registry
            .output_snapshot(id)
            .await
            .expect("a log on disk is reachable from its id alone");
        assert!(closed, "a log with a footer is a finished operation");
        assert_eq!(lines.len(), 1, "the line the command line wrote");

        // The LISTING is what adoption fills: being able to fetch one
        // operation you can name is not the same as offering a history.
        assert!(
            registry.snapshot().await.is_empty(),
            "history is empty until the directory is read"
        );
        assert_eq!(registry.adopt_recent_logs(50).await, 1);

        let listed = registry.snapshot().await;
        assert_eq!(listed.len(), 1);
        assert_eq!(listed[0].id, id);
        assert_eq!(listed[0].status, OpStatus::Completed);
        assert_eq!(listed[0].label, "the declared suite");
        assert_eq!(listed[0].kind, OpKind::Command);
        assert_eq!(listed[0].started_from, StartedFrom::CommandLine);
        assert!(
            listed[0].log_path.is_some(),
            "an adopted operation must say where its log is"
        );

        let (lines, _, closed) = registry
            .output_snapshot(id)
            .await
            .expect("the log is readable from disk");
        assert!(closed, "a run from before this process is not still running");
        match &lines[0].record {
            wsb::output::Record::Stdout { text } => assert_eq!(text, "it ran"),
            other => panic!("{other:?}"),
        }
    }

    /// TEST8988: a run whose log never got a footer does not read as finished.
    #[tokio::test]
    async fn test8988_an_interrupted_run_is_adopted_as_failed() {
        let dir = tempfile::tempdir().unwrap();
        let id = Uuid::new_v4();
        // Started, spoke, and never wrote how it ended.
        command_line_log(dir.path(), id, "the app", &["starting".to_string()], false);

        let registry = OpsRegistry::with_logs(dir.path().to_path_buf());
        registry.adopt_recent_logs(50).await;
        let listed = registry.snapshot().await;
        let op = listed.iter().find(|o| o.id == id).expect("it is listed");
        assert_eq!(
            op.status,
            OpStatus::Failed,
            "a run with no footer must not present itself as one that finished"
        );
        assert!(op.error.as_deref().unwrap_or_default().contains("stopped"));
    }

    /// TEST8947: an ended run's output is closed for subscribers.
    ///
    /// A viewer of a finished op must learn the log is complete; without
    /// Closed it would show a spinner over a log that will never grow.
    #[tokio::test]
    async fn test8947_resolving_closes_output() {
        let registry = OpsRegistry::new();
        let id = Uuid::new_v4();
        announce(&registry, id, None).await;
        let mut rx = registry
            .subscribe_output(id)
            .await
            .expect("an announced run is subscribable");

        registry
            .adopted_lines(id, vec![adopted("one")])
            .await
            .expect("an announced run may say more");
        registry
            .adopted_finished(id, 1)
            .await
            .expect("an announced run may end");

        // First event is the line, second is Closed.
        match rx.try_recv().expect("line event") {
            OutputEvent::Line(line) => assert_eq!(
                line.record,
                wsb::output::Record::Stdout {
                    text: "one".to_string()
                }
            ),
            OutputEvent::Closed => panic!("Closed arrived before the line"),
        }
        match rx.try_recv().expect("closed event") {
            OutputEvent::Closed => {}
            OutputEvent::Line(line) => panic!("expected Closed, got line {:?}", line.record),
        }
        let (_, _, closed) = registry.output_snapshot(id).await.expect("still in history");
        assert!(closed, "a resolved op's output must be closed");
    }

    /// TEST8949: eviction from the history ring drops the op's output buffer
    /// — otherwise every finished build would hold its full log in memory
    /// forever.
    #[tokio::test]
    async fn test8949_eviction_forgets_output() {
        let registry = OpsRegistry::new();
        let first_id = Uuid::new_v4();
        announce(&registry, first_id, None).await;
        registry
            .adopted_lines(first_id, vec![adopted("kept?")])
            .await
            .expect("an announced run may say more");
        registry
            .adopted_finished(first_id, 0)
            .await
            .expect("an announced run may end");
        assert!(registry.output_snapshot(first_id).await.is_some());

        // Filled with something that is its own row each time. A recurring
        // kind — a sync — is one row however often it runs, so it cannot
        // push anything out.
        for i in 0..(HISTORY_CAPACITY + 1) {
            let h = registry.start_op(OpKind::GitMutation, format!("commit {i}")).await;
            h.complete().await;
        }

        assert!(
            registry.output_snapshot(first_id).await.is_none(),
            "an evicted op's output must be forgotten"
        );
        assert!(
            registry.subscribe_output(first_id).await.is_none(),
            "an evicted op must not be subscribable"
        );
    }

    /// TEST11224: a run in a terminal is an operation here, under its own id.
    ///
    /// The whole point of adoption. A command line opened its log file under
    /// an id before it knew whether anything was watching; the registry takes
    /// THAT id rather than minting one, so the row in the panel and the file
    /// on disk are one run. A registry that minted its own would give an
    /// operator two identifiers for one thing and no way to tell they matched.
    #[tokio::test]
    async fn test11224_an_adopted_run_keeps_the_id_its_log_was_opened_under() {
        let registry = OpsRegistry::new();
        let id = Uuid::new_v4();
        let started = Utc::now();

        registry
            .adopt(
                id,
                OpKind::Command,
                "wsb update --recursive".to_string(),
                started,
                Some("/logs/update.jsonl".to_string()),
                vec![adopted("reading the workspace")],
                0,
            )
            .await
            .expect("a run this registry has not seen is adoptable");

        let snapshot = registry.snapshot().await;
        let op = snapshot.iter().find(|op| op.id == id).expect("listed");
        assert_eq!(op.status, OpStatus::Running);
        assert_eq!(op.started_from, StartedFrom::CommandLine);
        assert_eq!(op.log_path.as_deref(), Some("/logs/update.jsonl"));
        assert_eq!(
            op.started_at, started,
            "the run's own start time was replaced with the moment it was announced"
        );
        assert!(op.has_output, "an adopted run's output must be readable");

        let (lines, dropped, closed) = registry.output_snapshot(id).await.expect("output");
        assert_eq!(lines.len(), 1, "the announcement's lines were not kept");
        assert_eq!(dropped, 0);
        assert!(!closed);
    }

    /// TEST11405: a command-line run's log is protected while it is running.
    ///
    /// `protected_logs` is what a sweep is told it may never remove, and it
    /// read `inner.logs` — the writers THIS daemon opened. A run started from
    /// a terminal opens its own log and announces itself with `adopt`, so it
    /// lands in `inner.ops` carrying a `log_path` and in `inner.logs` not at
    /// all. Its log was therefore unprotected for its whole life.
    ///
    /// That is not theoretical. A Mac filled to 1.3 GiB tripped the janitor's
    /// free-space floor, and the sweep took the operation logs of an `wsb
    /// test` that was still running — including the file being appended to at
    /// that moment. The run carried on writing to a path that no longer
    /// existed, and the only record of a forty-minute CUDA run vanished while
    /// somebody was reading it.
    ///
    /// A running operation's log is in use whoever opened it.
    #[tokio::test]
    async fn test11405_a_command_line_runs_log_is_protected_while_it_runs() {
        let registry = OpsRegistry::new();
        let id = Uuid::new_v4();

        registry
            .adopt(
                id,
                OpKind::Command,
                "wsb test --on pve-windows-11-24h2-x86-64-cuda-pve".to_string(),
                Utc::now(),
                Some("/logs/ops/2026-09-18/125417-test.jsonl".to_string()),
                vec![adopted("mg-test: starting")],
                0,
            )
            .await
            .expect("a run started from a terminal announces itself");

        let protected = registry.protected_logs().await;
        let shown: Vec<String> = protected
            .iter()
            .map(|path| path.display().to_string())
            .collect();
        assert!(
            shown.iter().any(|p| p.ends_with("125417-test.jsonl")),
            "the log of a RUNNING command-line operation is not protected from \
             the sweep: {shown:?}"
        );
    }

    /// TEST11406: a FINISHED run's log is sweepable again.
    ///
    /// The other half of TEST11405. Protection is about work in progress, not
    /// about who opened the file — so it has to end when the work does, or the
    /// directory this sweep exists to bound can never be reclaimed and the
    /// limit becomes a number nothing enforces.
    #[tokio::test]
    async fn test11406_a_finished_runs_log_is_no_longer_protected() {
        let registry = OpsRegistry::new();
        let id = Uuid::new_v4();

        registry
            .adopt(
                id,
                OpKind::Command,
                "wsb test --on pve".to_string(),
                Utc::now(),
                Some("/logs/ops/2026-09-18/done.jsonl".to_string()),
                vec![adopted("mg-test: starting")],
                0,
            )
            .await
            .expect("announced");

        assert_eq!(
            registry.protected_logs().await.len(),
            1,
            "a running run's log is protected"
        );

        registry
            .adopted_finished(id, 0)
            .await
            .expect("a run that ended says so");

        let protected = registry.protected_logs().await;
        assert!(
            protected.is_empty(),
            "a finished run's log must be sweepable again, or the limit is a \
             number nothing enforces: {protected:?}"
        );
    }

    /// TEST11225: what a terminal run says arrives, and how it ended is what
    /// the panel shows.
    #[tokio::test]
    async fn test11225_an_adopted_runs_lines_and_outcome_reach_the_registry() {
        let registry = OpsRegistry::new();
        let id = Uuid::new_v4();
        registry
            .adopt(
                id,
                OpKind::Command,
                "wsb ship".to_string(),
                Utc::now(),
                None,
                vec![adopted("first")],
                0,
            )
            .await
            .unwrap();

        registry
            .adopted_lines(id, vec![adopted("second"), adopted("third")])
            .await
            .expect("a run this registry is holding may say more");
        registry
            .adopted_finished(id, 2)
            .await
            .expect("a run this registry is holding may end");

        let (lines, _, closed) = registry.output_snapshot(id).await.expect("output");
        assert_eq!(lines.len(), 3, "lines sent after the announcement were lost");
        assert!(closed, "a finished run's output must be closed to subscribers");
        // Sequence numbers are this buffer's, contiguous from zero — a page
        // resolving the snapshot/live overlap compares them against each
        // other, and numbering carried from the run would have holes in it
        // wherever the run said something before it was announced.
        assert_eq!(
            lines.iter().map(|line| line.seq).collect::<Vec<_>>(),
            vec![0, 1, 2]
        );

        let snapshot = registry.snapshot().await;
        let op = snapshot.iter().find(|op| op.id == id).expect("listed");
        assert_eq!(
            op.status,
            OpStatus::Failed,
            "a run that exited non-zero was shown as having succeeded"
        );
        assert!(op.error.is_some(), "a failed run must say something about why");
    }

    /// TEST11226: a run nothing announced cannot be spoken for.
    ///
    /// Lines and outcomes are refused for an id the registry is not holding —
    /// an announcement that was lost, a run that has aged out of history, or a
    /// caller inventing one. Accepting them would mean creating an operation
    /// out of a line of output, with no start time, no label and no log.
    #[tokio::test]
    async fn test11226_lines_for_an_unknown_run_are_refused() {
        let registry = OpsRegistry::new();
        let unknown = Uuid::new_v4();

        assert!(
            registry.adopted_lines(unknown, vec![adopted("x")]).await.is_err(),
            "lines were accepted for a run this registry never heard of"
        );
        assert!(
            registry.adopted_finished(unknown, 0).await.is_err(),
            "an outcome was accepted for a run this registry never heard of"
        );
        assert!(registry.snapshot().await.is_empty());
    }

    /// TEST11227: announcing twice is refused rather than duplicated.
    ///
    /// A second announcement is either a retry after a lost response or a bug.
    /// Two rows for one run is the wrong answer to both — and the second would
    /// take over the output buffer, so the first row would stop moving while
    /// the run carried on.
    #[tokio::test]
    async fn test11227_a_run_cannot_be_announced_twice() {
        let registry = OpsRegistry::new();
        let id = Uuid::new_v4();
        let announce = || {
            registry.adopt(
                id,
                OpKind::Command,
                "wsb update".to_string(),
                Utc::now(),
                None,
                Vec::new(),
                0,
            )
        };
        announce().await.expect("the first announcement is taken");
        assert!(
            announce().await.is_err(),
            "the same run was announced twice and got two rows"
        );
        assert_eq!(registry.snapshot().await.len(), 1);
    }

    /// A log file as one is actually written: a header line, then a footer.
    ///
    /// Written rather than mocked, because what is under test is the daemon
    /// reading a REAL log — a fixture returning a parsed footer would assert
    /// about a test helper.
    fn log_with_footer(dir: &std::path::Path, id: Uuid, status: &str) -> String {
        let path = dir.join(format!("{id}.jsonl"));
        let header = serde_json::json!({
            "log": "header", "format": "wsb-op-log/1", "id": id,
            "kind": "checkpoint", "label": "checkpointing",
            "argv": ["wsb", "checkpoint"], "started_at": Utc::now(),
        });
        let footer = serde_json::json!({
            "log": "footer", "status": status, "ended_at": Utc::now(), "lines": 1,
        });
        std::fs::write(&path, format!("{header}\n{footer}\n")).expect("write log");
        path.display().to_string()
    }

    /// TEST11228: an adopted run whose process is gone resolves from its log.
    ///
    /// The daemon holds no handle on an adopted process and refuses to accept
    /// one, so a run killed in its terminal never reports an exit and its row
    /// said "running" until the daemon restarted. Four sat in the panel at
    /// once, the oldest an hour old and all long finished.
    ///
    /// The footer is the process's own account of how it ended, and the daemon
    /// already reads footers to rebuild history when it starts. This is that
    /// same fact, applied while it is running.
    #[tokio::test]
    async fn test11228_a_finished_adopted_run_resolves_from_its_log() {
        let dir = std::env::temp_dir().join(format!("wsb-ops-{}", Uuid::new_v4()));
        std::fs::create_dir_all(&dir).expect("scratch");
        let registry = OpsRegistry::new();

        let finished = Uuid::new_v4();
        let path = log_with_footer(&dir, finished, "failed");
        registry
            .adopt(
                finished,
                OpKind::Command,
                "wsb checkpoint".to_string(),
                Utc::now(),
                Some(path),
                vec![],
                0,
            )
            .await
            .expect("adoptable");

        // Still running as far as the registry knows: nothing reported an exit.
        let before = registry.snapshot().await;
        assert_eq!(
            before.iter().find(|o| o.id == finished).expect("listed").status,
            OpStatus::Running,
        );

        registry.resolve_finished_adopted().await;

        let after = registry.snapshot().await;
        assert_eq!(
            after.iter().find(|o| o.id == finished).expect("listed").status,
            OpStatus::Failed,
            "a row stayed running for a process that had already written how it ended"
        );
        std::fs::remove_dir_all(&dir).ok();
    }

    /// TEST11229: a run that is STILL GOING is left alone.
    ///
    /// A log with no footer is a process still writing, or one killed so hard
    /// it never wrote one, and the two are indistinguishable from here.
    /// Resolving on absence would report a live build as finished — worse than
    /// the staleness it fixes, because somebody acts on it.
    #[tokio::test]
    async fn test11229_a_running_adopted_run_is_not_resolved() {
        let dir = std::env::temp_dir().join(format!("wsb-ops-{}", Uuid::new_v4()));
        std::fs::create_dir_all(&dir).expect("scratch");
        let registry = OpsRegistry::new();

        // A header and nothing else: what a run that is still going looks like.
        let live = Uuid::new_v4();
        let path = dir.join(format!("{live}.jsonl"));
        let header = serde_json::json!({
            "log": "header", "format": "wsb-op-log/1", "id": live,
            "kind": "test_run", "label": "running the declared suite",
            "argv": ["wsb", "test"], "started_at": Utc::now(),
        });
        std::fs::write(&path, format!("{header}\n")).expect("write log");
        registry
            .adopt(
                live,
                OpKind::Command,
                "wsb test".to_string(),
                Utc::now(),
                Some(path.display().to_string()),
                vec![],
                0,
            )
            .await
            .expect("adoptable");

        registry.resolve_finished_adopted().await;

        let after = registry.snapshot().await;
        assert_eq!(
            after.iter().find(|o| o.id == live).expect("listed").status,
            OpStatus::Running,
            "a run with no footer was reported as finished; it may still be going"
        );
        std::fs::remove_dir_all(&dir).ok();
    }
}