onepipeline 0.38.0

//! The run ledger: where a run's durable state lives, and who may write it.
//!
//! One directory per run under the runs root. The launch record says who owns
//! the run and how to relaunch it; the ownership lock is what makes the driving
//! process a single writer; the journal beside them is the merged event store.
//!
//! Writes that a reader could catch half-finished are atomic — written to a
//! temporary beside the target and renamed over it — because every view here
//! reads a live run's directory while its driver is writing to it. An append
//! cannot be written that way, since the file is everything already recorded, so
//! it holds the file's exclusive lock instead: it heals a fragment a dead writer
//! left before it writes, reports what that cost, and takes its own bytes back
//! off the file when the write fails.
//!
//! # How a record here is read, and what that costs
//!
//! **A field this build does not know is ignored, not refused.** These records
//! are this crate's own — written by some build of it and read back by another,
//! never by a person — so the reader that meets an unfamiliar key is meeting a
//! build that is not this one, which is a fact about versions and not a typo to
//! catch. Refusing it takes away the *whole* record over one key, and a run root
//! whose launch record cannot be read is a run that vanishes from the view an
//! operator opens to see what is running on their host. Read permissively,
//! adding a field can never break a reader again — no per-version migration, no
//! compatibility shim.
//!
//! This is the opposite rule from the one that governs a **plan**, an
//! **executor-rules file**, a **launch config**, or a **reply envelope**. Those
//! are external input — a document somebody wrote — where an unknown key is a
//! typo and dropping it silently is how an operator finds out their setting did
//! nothing. Those types keep `deny_unknown_fields`, and the [`Filters`] block
//! nested in [`LaunchRecord`] is one of them: it is the same type an operator
//! writes in a launch config, so it stays closed, and a key added inside that
//! block is not covered by the paragraph above.
//!
//! **A record this build genuinely cannot read is still reported.** Permissive
//! is not silent: a document that is not JSON, one torn mid-write, or one
//! missing a field this build requires still comes back as an error naming the
//! file and the reason, and [`Survey::of`](crate::views::Survey::of) puts it on
//! the reader's `run root(s) skipped` list. The distinction is the journal's
//! own: unreadable is reported, unfamiliar is ignored, and neither is ever
//! dropped on the floor.
//!
//! **So retire a field or a variant, never delete one.** Permissive parsing
//! fixes the additive half only: a record naming something whose *meaning* this
//! build removed is one it cannot act on however leniently it parses, and that
//! loss is accepted rather than engineered around. What stops it is keeping a
//! retired field or variant recognised and **inert**: leave it declared, stop
//! writing it, `#[serde(default, skip_serializing_if = …)]` so it is omitted
//! from anything written now, and say in its doc comment that it is retired.
//! A build that reads it then goes on reading every record already on disk.

// llmlint: ignore-file[invalid_states_unrepresentable] a run id, a host name, and a
// timestamp are `String`s in these records because each one is a *serialized* field: the
// launch record and the lock are JSON a human reads and another process parses, and every
// reader — including one written against an older build — has to accept what is there
// rather than what this build would mint. `docs/contract.md` names no `RunId`, so a
// newtype would also be a public vocabulary the contract did not ask for. What is
// enforced instead is the thing that matters: `owned_by` is the one place ownership is
// decided, and `unknown` is never anybody's.

use serde::{Deserialize, Serialize};
use std::ffi::OsStr;
use std::fs;
use std::io;
use std::num::{NonZeroU32, NonZeroU64};
use std::path::{Path, PathBuf};

use crate::error::{Error, Result};
use crate::filter::Filters;
use crate::sys;

// Every byte of a run's ledger this process reads is counted through the two
// functions below, and the counter they reach is `crate::loopstats`'s rather than
// a second one kept here.
//
// Accounting rather than a promise. What it is *for* is holding a bounded read
// bounded: a run's summary served from its own document reads that document and
// stats the journal, while one that has to be folded reads the entire store — and
// the difference between those two is the whole reason the document exists. That
// is a property to be **measured** rather than inferred from a clock, and this is
// what measures it.
//
// One account rather than two, because a driver launched to report what its loop
// cost writes that same number out as `store_bytes`. Two counters here would not
// fail by disagreeing — nobody compares them — but by a reader being added to one
// and not the other, leaving the number the checks hold and the number the host is
// told describing different sets of reads.

/// What this thread's share of that counter stands at.
///
/// Read by the tests that hold the bounded read bounded, and by nothing else —
/// a number this crate acted on would be a second account of a run's cost, and
/// there is no second account here to keep true. Per thread rather than per
/// process for the reason [`crate::loopstats::store_bytes`] gives: a check
/// measuring a delta wants the reads it performed, not the ones a test beside it
/// performed at the same time.
#[cfg(test)]
pub(crate) fn bytes_read() -> u64 {
    crate::loopstats::store_bytes()
}

/// Count what a read just cost.
fn counted<T>(bytes: usize, read: T) -> T {
    crate::loopstats::store_read(bytes as u64);
    // And, where a view is deciding one node's landing while this happens, the
    // same read again under that node's name: a store read that is per-node
    // work is exactly what the render bound forbids, and only the scope it
    // happened inside can tell the two apart.
    crate::rendercost::store_read(bytes as u64);
    read
}

/// The environment variable that moves the runs root.
pub const RUNS_DIR_ENV: &str = "ONEPIPELINE_RUNS_DIR";

/// The runs root when the environment names none.
pub const DEFAULT_RUNS_DIR: &str = "runs";

/// The runs root this process reads and writes.
pub fn runs_root() -> PathBuf {
    std::env::var_os(RUNS_DIR_ENV)
        .map(PathBuf::from)
        .filter(|p| !p.as_os_str().is_empty())
        .unwrap_or_else(|| PathBuf::from(DEFAULT_RUNS_DIR))
}

/// Where one run's durable state lives.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RunPaths {
    /// The run id, as every command and view names it.
    pub run: String,
    /// The run's own directory.
    pub dir: PathBuf,
}

/// Whether a run id names one directory under the runs root and nothing else.
///
/// A run id is external input on every verb that takes one, and — since
/// cross-DAG references carry one — inside plan files too. It is joined onto the
/// runs root, so a separator, a `..`, or an absolute path would read and write
/// *outside* the ledger this process was pointed at: `onepipeline status
/// ../../elsewhere` would render another root's run, and a plan naming
/// `run:../../elsewhere#node` would resolve its schedule against one.
///
/// One segment, and nothing that navigates. `mint_run_id` already produces only
/// this alphabet; this is the boundary for the ids that arrive from outside.
pub fn is_valid_run_id(run: &str) -> bool {
    !run.is_empty()
        && run != "."
        && run != ".."
        && !run.contains('/')
        && !run.contains('\\')
        && !Path::new(run).is_absolute()
        && Path::new(run).components().count() == 1
}

/// One producer-supplied name, as a single path segment.
///
/// Everything outside `[A-Za-z0-9._-]` becomes a `-`, and a name that is empty
/// or navigates gets one of its own: a segment built from a stranger's string
/// has to be a *name*, never a path, and `..` is the shortest path there is.
fn path_segment(name: &str) -> String {
    let mapped: String = name
        .chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() || c == '-' || c == '_' || c == '.' {
                c
            } else {
                '-'
            }
        })
        .collect();
    if mapped.is_empty() || mapped.chars().all(|c| c == '.') {
        return "unnamed".to_string();
    }
    mapped
}

impl RunPaths {
    /// The paths for `run` under the process's runs root.
    pub fn new(run: &str) -> Self {
        Self::under(&runs_root(), run)
    }

    /// The paths for `run` under an explicit root.
    pub fn under(root: &Path, run: &str) -> Self {
        Self {
            run: run.to_string(),
            dir: root.join(run),
        }
    }

    /// Whether this run has a directory at all.
    pub fn exists(&self) -> bool {
        self.dir.is_dir()
    }

    /// Create the run's directory and the two subdirectories a run always has.
    ///
    /// The dispatch registry among them, and empty is the answer it is created to
    /// be able to give: a reader that meets no registry at all cannot tell a run
    /// with nothing running from one whose record of what it is running has gone,
    /// and refuses. So a run has one from the moment it exists, and its absence
    /// afterwards means something took it away.
    pub fn create(&self) -> Result<()> {
        for dir in [self.channel_dir(), self.dispatches()] {
            fs::create_dir_all(&dir).map_err(|e| Error::Ledger {
                path: dir,
                source: e,
            })?;
        }
        Ok(())
    }

    /// The merged three-stream event store.
    pub fn journal(&self) -> PathBuf {
        self.dir.join("events.jsonl")
    }

    /// The launch record: who owns the run, and what to relaunch it with.
    pub fn launch(&self) -> PathBuf {
        self.dir.join("launch.json")
    }

    /// The plan the run was launched with.
    pub fn plan(&self) -> PathBuf {
        self.dir.join("plan.json")
    }

    /// The run's summary document: the row a bounded listing reads instead of
    /// this run's whole journal.
    ///
    /// Beside [`plan`](Self::plan) and the run's result, and for the same
    /// reason: it is a derived record of the run, kept where the run's own state
    /// is kept, so a run thrown away takes it with it. See
    /// [`summary::RunSummary`](crate::summary::RunSummary).
    pub fn summary(&self) -> PathBuf {
        self.dir.join("summary.json")
    }

    /// The run's fold checkpoint: what a reader resumes a fold from instead of
    /// replaying this run's whole journal.
    ///
    /// Beside [`summary`](Self::summary) and for the same reason: it is a derived
    /// record of the run, kept where the run's own state is kept, so a run thrown
    /// away takes it with it. See [`checkpoint`](crate::checkpoint).
    ///
    /// Crate-visible, unlike its neighbours here: `docs/contract.md` names six
    /// members of this type and says nothing else on it is contracted, and this
    /// document is read and written by one private module — so publishing the
    /// path would be a promise nobody asked for about a cache.
    pub(crate) fn checkpoint(&self) -> PathBuf {
        self.dir.join("checkpoint.json")
    }

    /// The directory holding one record per live **watch** of this run.
    ///
    /// A directory of small records rather than one document, for the reason
    /// [`dispatches`](Self::dispatches) is one: any number of watches may sit on
    /// one run at once and they start and end independently, so a single file
    /// would be read, edited, and rewritten by several of them and a lost update
    /// there is a watch nothing can find.
    ///
    /// Crate-visible, like [`checkpoint`](Self::checkpoint) and unlike its
    /// neighbours here: `docs/contract.md` names six members of this type, and a
    /// reader of these records reaches them through
    /// [`views::Watchers::of`](crate::views::Watchers::of) rather than by
    /// composing a path — so publishing the directory would be a second promise
    /// about the same documents.
    pub(crate) fn watchers(&self) -> PathBuf {
        self.dir.join("watchers")
    }

    /// One watch's record, named by the process holding it and a value unique to
    /// that watch.
    ///
    /// Both halves are load-bearing. The pid alone is not a name: several watches
    /// of one run may run in one process, and two of them would write one entry
    /// with the first to end taking the survivor's record away. And a pid the
    /// kernel has handed round again is not its predecessor, so a name that were
    /// only a pid would let a new watch land on an old record — which is why
    /// staleness is decided by a record's *content* and never by its name.
    pub(crate) fn watcher(&self, pid: u32, nonce: &str) -> PathBuf {
        self.watchers().join(format!("{pid}-{nonce}.json"))
    }

    /// The single-writer ownership lock the engine verbs hold.
    pub fn lock(&self) -> PathBuf {
        self.dir.join("owner.lock")
    }

    /// Where a **detached** driver's own output goes.
    ///
    /// A file rather than a pipe, because the process that would read the pipe
    /// is the launcher, and `--detach` means the launcher is about to exit. A
    /// driver holding the write end of a pipe nobody holds the read end of dies
    /// on its first line of output — and it dies mid-run, leaving a run whose
    /// graph never settles and whose driver is gone.
    pub fn driver_log(&self) -> PathBuf {
        self.dir.join("driver.log")
    }

    /// The channel's transport state.
    pub fn channel_dir(&self) -> PathBuf {
        self.dir.join("channel")
    }

    /// Where this run keeps its own copy of the evidence its dispatches left.
    ///
    /// Run-**owned**: a sibling's report lives in that library's scratch, which
    /// is a directory this crate neither chooses nor can attest, and a reader
    /// that opened whatever a journal line pointed at would be an
    /// arbitrary-file reader driven by whatever wrote to the journal. So the
    /// evidence is copied here as it is ingested, and every reader afterwards
    /// opens only what is under this directory.
    pub fn reports_dir(&self) -> PathBuf {
        self.dir.join("reports")
    }

    /// This run's copy of one relayed settlement's report.
    ///
    /// Named from the producing stream and its sequence number, which identify
    /// the settlement and nothing else — so a reader derives the name rather
    /// than following a path, and both sides agree without either trusting one.
    /// The stream is written as a single sanitised segment: it is a producer's
    /// string, and joining one raw is how a name becomes a path.
    pub fn report_for(&self, stream: &str, seq: u64) -> PathBuf {
        self.reports_dir()
            .join(format!("{}-{seq}.json", path_segment(stream)))
    }

    /// A file within the channel's transport state.
    pub fn channel(&self, name: &str) -> PathBuf {
        self.channel_dir().join(name)
    }

    /// The run's recorded result, rewritten whenever a driver closes out.
    ///
    /// One document, at the run's own root: the frontier is continuous, so what
    /// the ledger records is where the whole graph has got to.
    pub fn result(&self) -> PathBuf {
        self.dir.join("result.json")
    }

    /// The dispatch ownership registry: one record per process this run has
    /// work running in.
    ///
    /// A directory of small records rather than one document, because its
    /// writers are the run's dispatch threads and they start and finish
    /// independently: a single file would be read, edited, and rewritten by
    /// several of them at once, and a lost update there is a live dispatch no
    /// later stop can find.
    pub fn dispatches(&self) -> PathBuf {
        self.dir.join("dispatches")
    }

    /// One dispatch's record, named by the process it runs in and the claim that
    /// wrote it.
    ///
    /// Named from a pid because a pid is always a safe file name and a node id is
    /// not: an id is plan text, required to be non-empty and unique and nothing
    /// else, so joining one raw is how a name becomes a path — and sanitising it
    /// would map two distinct nodes onto one record.
    ///
    /// A pid alone is **not** an identity, which is the other half. A run's
    /// dispatches can share one process: that is what the library backend is —
    /// several nodes running concurrently inside the driver — so two live
    /// dispatches would write one entry, the second would overwrite the first,
    /// and the first to end would take the survivor's registration with it,
    /// leaving a live dispatch nothing could find. `claim` is what tells them
    /// apart, and it is unique for the life of the process that mints it.
    pub fn dispatch(&self, pid: u32, claim: u64) -> PathBuf {
        self.dispatches().join(format!("{pid}-{claim}.json"))
    }
}

/// A run root this build refused, and the reason it gave.
///
/// A rejection, never an absence. A reader who is told nothing is there acts on
/// "nothing is running"; a reader who is told which directory was refused and
/// why can fix it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Skipped {
    /// The directory that was refused.
    pub path: PathBuf,
    /// Why, in the words the reader needs to act on it.
    pub reason: String,
}

/// Every run a root records, and every run root under it that records none.
///
/// The two halves are returned together because dropping the second is what
/// made an unreadable root indistinguishable from an empty one: a host with
/// thirty run roots on it rendered as a host with nothing running.
#[derive(Debug, Default, PartialEq, Eq)]
pub struct RunIndex {
    /// The runs the root records, in id order.
    pub runs: Vec<RunPaths>,
    /// The run roots it refused, in path order.
    pub skipped: Vec<Skipped>,
}

/// Every run the root records, in id order, and every run root it refused.
///
/// A directory under the runs root is a *claim* to be a run, and one this build
/// cannot read is a claim it is rejecting — so it comes back named. A plain file
/// beside the runs is not such a claim and is passed over silently: nothing ever
/// said it was a run.
pub fn all_runs(root: &Path) -> RunIndex {
    let mut index = RunIndex::default();
    // llmlint: ignore-block[changed_behavior_has_e2e] a runs root that exists and will not
    // open, and an entry the filesystem lists and then refuses to describe, are host
    // conditions no portable journey can set. The arm a user reaches — a run root with no
    // launch record — is driven in `tests/e2e/views.rs`.
    let entries = match fs::read_dir(root) {
        Ok(entries) => entries,
        // A root nobody has written to yet holds nothing to reject, which is the
        // one reading of an empty view that is honest. Anything else is a root
        // this process was pointed at and could not read, and reporting that as
        // "no runs recorded" is the lie this whole index exists to stop.
        Err(error) if error.kind() == io::ErrorKind::NotFound => return index,
        Err(error) => {
            index.skipped.push(Skipped {
                path: root.to_path_buf(),
                reason: format!("the runs root cannot be read: {error}"),
            });
            return index;
        }
    };
    for entry in entries {
        let entry = match entry {
            Ok(entry) => entry,
            Err(error) => {
                index.skipped.push(Skipped {
                    path: root.to_path_buf(),
                    reason: format!("an entry under the runs root cannot be read: {error}"),
                });
                continue;
            }
        };
        let path = entry.path();
        // Asked for, rather than tested with `is_dir`: that helper answers
        // `false` both for "not a directory" and for "this host would not say",
        // and reading the second as the first is the collapse this whole index
        // exists to undo — the entry would be dropped as though it had never
        // claimed to be a run.
        let about = match fs::metadata(&path) {
            Ok(about) => about,
            // Gone between the listing and the look. A run swept while this scan
            // was running is not a root to make any claim about.
            Err(error) if error.kind() == io::ErrorKind::NotFound => continue,
            Err(error) => {
                index.skipped.push(Skipped {
                    path,
                    reason: format!("this host will not describe it: {error}"),
                });
                continue;
            }
        };
        // llmlint: ignore-end[changed_behavior_has_e2e]
        if !about.is_dir() {
            continue;
        }
        // llmlint: ignore-block[changed_behavior_has_e2e] a directory name that is not text
        // is not portably creatable — Windows refuses one outright — and no run id this
        // crate mints is one. What lands here is a directory an operator left beside the
        // runs, which this arm names rather than drops.
        let Ok(name) = entry.file_name().into_string() else {
            index.skipped.push(Skipped {
                path,
                reason: "its name is not text this host can read, so no run id names it".into(),
            });
            continue;
        };
        // llmlint: ignore-end[changed_behavior_has_e2e]
        let paths = RunPaths::under(root, &name);
        let launch = paths.launch();
        let named = launch
            .file_name()
            .map_or_else(|| "launch record".into(), |name| name.to_string_lossy());
        // Every answer kept apart, for the same reason as above: absent, present
        // as something that is not a record, and unreadable are three different
        // things to tell a reader, and `is_file` says `false` to all three.
        let refused = match fs::metadata(&launch) {
            Ok(about) if about.is_file() => None,
            Ok(_) => Some(format!(
                "its {named} is not a file, so it records no launch"
            )),
            Err(error) if error.kind() == io::ErrorKind::NotFound => Some(format!(
                "no {named}: a run root records the launch that owns it"
            )),
            // llmlint: ignore-block[changed_behavior_has_e2e] a launch record this host
            // lists and will not describe is a host condition no portable journey can set.
            // The two answers a user reaches are both driven in `tests/e2e/views.rs`.
            Err(error) => Some(format!("its {named} cannot be read: {error}")),
            // llmlint: ignore-end[changed_behavior_has_e2e]
        };
        if let Some(reason) = refused {
            index.skipped.push(Skipped {
                path: paths.dir,
                reason,
            });
            continue;
        }
        index.runs.push(paths);
    }
    index.runs.sort_by(|a, b| a.run.cmp(&b.run));
    index.skipped.sort_by(|a, b| a.path.cmp(&b.path));
    index
}

/// Whether a path field carries nothing, for the records that omit it then.
fn is_unset(path: &Path) -> bool {
    path.as_os_str().is_empty()
}

/// What a launch record naming no launcher launched under.
///
/// The same word [`sys::launcher`] answers when this host's environment says
/// nothing, so a record that predates the key and one written where nothing
/// could attribute the launch read alike — which is what they are.
fn unattributed_launcher() -> String {
    sys::UNKNOWN_LAUNCHER.to_string()
}

/// What `start` recorded about a run, and what `adopt` replays it from.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct LaunchRecord {
    /// The run id.
    pub run_id: String,
    /// The qualified onetaskgraph project id the run was launched with.
    ///
    /// The plan's **definition** lives in that store; what this run executes is
    /// the graph projected from its own journal, so this names where the plan
    /// came from and is never re-read to decide what the run is doing.
    ///
    /// Empty on a record written before the store was where a plan came from —
    /// one that named a plan file instead — and omitted when empty, like every
    /// other field added to this record after it shipped, so a run an earlier
    /// build launched is still a run this one can read, adopt, and report on.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub project: String,
    /// The directory every member of this run works in.
    ///
    /// The launch directory, made absolute once — at `start`, by the process
    /// the operator ran — and replayed verbatim by `adopt`. It reaches
    /// `oneagentgraph` as the run's `dir`, which is the `--cwd` each harness is
    /// given, so a relative value would resolve against whichever process
    /// happened to spawn the graph rather than against the directory the
    /// operator launched from, and `start` and `adopt` would name two different
    /// places for one run.
    ///
    /// Empty only on a record written before this field existed; the launcher
    /// then falls back to its own working directory, which is what it did then.
    /// Omitted when empty, like every other field added to this record after
    /// it shipped, so a build that predates it still reads what it wrote.
    #[serde(default, skip_serializing_if = "is_unset")]
    pub dir: PathBuf,
    /// The dag-scope agent-graph config launched as this run's observer.
    ///
    /// Absent when the launch named none, which is the shipped default: no
    /// agent is required to execute a plan. Read it through
    /// [`observer_graph`](Self::observer_graph) rather than testing this field —
    /// the serialized shape omits an absent value, and the one place that turns
    /// "omitted" back into "there is none" is that accessor.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub graph: String,
    /// The `oneagentgraph` run this run's observer graph is, as that library
    /// minted it — **not** this run's id, which names something else entirely.
    ///
    /// Written after the launch that produced it, and rewritten by every
    /// `adopt`, because an adoption starts a fresh graph run with an id of its
    /// own. It is how a later `onepipeline next` — a different process, with no
    /// handle on the observer — addresses the run's check-in clocks. Empty when no
    /// observer graph was launched.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub graph_run: String,
    /// Every `oneagentgraph` run this run's observer graph has been, oldest
    /// first, ending with the one [`graph_run`](Self::graph_run) names.
    ///
    /// A different question from that field, and it becomes one the moment an
    /// observer is replaced: `graph_run` says what to address **now** — the
    /// clocks a later `next` restarts — while this says what has watched the
    /// run at all. A graph run that ended is still the producer of every
    /// envelope it wrote, and once `graph_run` has moved past it a reader
    /// meeting one of its records in the merged store has nothing else to tell
    /// this run's observer from one of its node dispatches.
    ///
    /// Written by [`watched_by`](Self::watched_by), which is the one writer of
    /// both fields, so the two cannot come apart. Omitted when empty, like every
    /// other field added to this record after it shipped, so a build that
    /// predates it still reads what it wrote.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub observer_runs: Vec<String>,
    /// Why nothing is starting this run's observer graph again, or empty while
    /// something still would.
    ///
    /// The driver's own statement, which is stronger evidence than any probe of
    /// the graph run beside it: a driver that has stopped restarting knows the
    /// run is not going to be watched again, where
    /// [`agentgraph::graph_run_ended`] can only report what this host can prove.
    /// So it is what a view reads first — see
    /// [`views`](crate::views) — and it is cleared by
    /// [`watched_by`](Self::watched_by), because whatever ended the last
    /// observer is not true of the one now watching.
    ///
    /// Empty on every record written before this field existed, which reads as
    /// the run those builds always had: one whose observer nothing was going to
    /// restart, and which said so by leaving this unset.
    ///
    /// [`agentgraph::graph_run_ended`]: crate::agentgraph::graph_run_ended
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub observer_ending: String,
    /// The default node-scope agent-graph config every dispatch launches.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub node_graph: String,
    /// The agent-graph config a lifecycle node's change request body is drafted
    /// by, when the launch named one.
    ///
    /// Absent when it named none, which is the shipped default: this crate ships
    /// the flag and not the document, so a launch that says nothing drafts
    /// nothing. Read it through [`pr_author_graph`](Self::pr_author_graph)
    /// rather than testing this field, for the reason [`graph`](Self::graph) is
    /// read through its own accessor. Omitted when empty, like every other field
    /// added to this record after it shipped, so a build that predates it still
    /// reads what it wrote.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub pr_author_graph: String,
    /// The command every op that introduces or changes a node's task is offered
    /// to before it is applied, when the launch named one.
    ///
    /// **Resolved once, at the launch**, out of the flag, the environment, and
    /// the launch config in that order — so an `adopt` replays the validator its
    /// launch resolved rather than re-reading an environment that has since
    /// moved. Absent when the launch named none, which is the shipped default
    /// and is exactly the behaviour every run had before this field existed.
    /// Read it through [`node_validator`](Self::node_validator) rather than
    /// testing this field, for the reason [`graph`](Self::graph) is read through
    /// its own accessor. Omitted when empty, like every other field added to
    /// this record after it shipped, so a build that predates it still reads
    /// what it wrote.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub node_validator: String,
    /// The command a whole reply envelope is reviewed by before any of its
    /// edits is committed, when the launch named one.
    ///
    /// **Resolved once, at the launch**, out of the flag, the environment, and
    /// the launch config in that order — so an `adopt`, and a `reply` typed in
    /// another shell, use the reviewer this run was launched under rather than
    /// whatever the reading process's own environment says. Absent when the
    /// launch named none, which is the shipped default and is exactly the
    /// behaviour every run had before this field existed. Read it through
    /// [`envelope_reviewer`](Self::envelope_reviewer) rather than testing this
    /// field, for the reason [`graph`](Self::graph) is read through its own
    /// accessor. Omitted when empty, like every other field added to this record
    /// after it shipped, so a build that predates it still reads what it wrote.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub envelope_reviewer: String,
    /// The command whose output fingerprints the bar this run's envelope
    /// reviewer judges against, when the launch named one.
    ///
    /// **Resolved once, at the launch**, out of the flag, the environment, and
    /// the launch config in that order. What it prints is read each time a pass
    /// is looked for rather than here, so a bar that moves between two envelopes
    /// of one run runs the reviewer again. Read it through
    /// [`envelope_reviewer_bar`](Self::envelope_reviewer_bar). Omitted when empty,
    /// like every other field added to this record after it shipped.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub envelope_reviewer_bar: String,
    /// The launcher, as the environment reported it.
    ///
    /// Defaulted to [`sys::UNKNOWN_LAUNCHER`] on a record that carries no such
    /// key, which is the answer a record carrying an *empty* one already gives:
    /// an unattributed launch. **Always written**, unlike every optional field
    /// above — this build knows its own launcher — so the default is a
    /// concession to what somebody else wrote rather than a shape this crate
    /// produces. Refusing it cost a whole view: a third of the run roots on one
    /// host had been written before this key existed, and every one of them read
    /// as `missing field \`launcher\`` on the view an operator opens to see what
    /// is running — 141 refusals burying the live verdict under them.
    #[serde(default = "unattributed_launcher")]
    pub launcher: String,
    /// The launching session. A view labels a foreign one by
    /// [`sys::session_digest`], never by this value.
    ///
    /// Defaulted to **empty** on a record that carries no such key, and empty is
    /// already the unattributed launch every view renders `[unknown]` — see
    /// [`owner_label`](Self::owner_label). So a record written before this key
    /// existed and one written where nothing could attribute the launch read
    /// alike, which is what they are: neither says who owns the run, and an
    /// unattributed run is nobody's. The default states nothing rather than
    /// naming a session, because a reader that took the *reading* process's
    /// session for the record's would hand a stranger's run every command
    /// ownership guards.
    #[serde(default)]
    pub session: String,
    /// The driver process, or `0` on a record that names none.
    ///
    /// Read through [`driver_pid`](Self::driver_pid) rather than tested here.
    /// A pid is one third of a claim — see [`started`](Self::started) — and a
    /// defaulted one has no stamp beside it by construction, so nothing that
    /// acts on a pid may act on this one. `0` is never a driver on any platform
    /// this builds for, which is what makes it the honest default: it is the
    /// value `start` itself writes in the moment before
    /// [`driven_by_this_process`](Self::driven_by_this_process) claims the run.
    #[serde(default)]
    pub pid: u32,
    /// The host that pid is meaningful on, or empty on a record that names none.
    ///
    /// Read through [`recorded_host`](Self::recorded_host), which is where empty
    /// becomes "this record does not say" again. A reader must not claim a host
    /// the record does not name: a pid means nothing across machines, so a
    /// nameless host resolves toward *not this one* and the pid beside it is
    /// left alone.
    #[serde(default)]
    pub host: String,
    /// That driver's own process start token, as [`sys::process_start_token`]
    /// read it when it claimed the run.
    ///
    /// The same proof, and for the same reason, as the ownership lock's and the
    /// registry's: the pid says *which* process and this says it is still that
    /// one. This record outlives every driver it names — a driver that died
    /// leaves its pid sitting here until something adopts the run — so by the
    /// time a `stop` reads it the host may have handed that pid to a stranger,
    /// and a teardown aimed at it would end work this run never started.
    ///
    /// Written only by [`driven_by_this_process`](Self::driven_by_this_process),
    /// which writes all three fields together: a pid recorded without the stamp
    /// beside it is a pid no later reader may act on.
    ///
    /// Empty when this host would not say, and on a record written before the
    /// field existed. Omitted when empty, like every other field added to this
    /// record after it shipped, so a build that predates it still reads what it
    /// wrote. Empty is **not** a match — see [`sys::StartToken::matches`].
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub started: String,
    /// When the run was launched, or empty on a record that does not say.
    ///
    /// Read through [`launched_at`](Self::launched_at), which serves the absence
    /// **absent**. Never an instant standing in for one: a launch nobody
    /// recorded is a different fact from a launch recorded at the epoch, and
    /// only the second is a measurement. A reader handed epoch zero here would
    /// report every record that predates this key as a run launched in 1970 and
    /// age it accordingly.
    #[serde(default)]
    pub started_at: String,
    /// The check-in interval, in seconds, or `0` on a record that names none.
    ///
    /// Read through [`pacemaker_interval`](Self::pacemaker_interval), which is
    /// where `0` becomes "the record does not say" again: the shipped default,
    /// [`cli::DEFAULT_HEARTBEAT_INTERVAL_SECONDS`], or no check-in at all —
    /// and never a zero-second one. A check-in that fires continuously is worse
    /// than a run with none: it buries every real surface under its own.
    ///
    /// [`cli::DEFAULT_HEARTBEAT_INTERVAL_SECONDS`]: crate::cli::DEFAULT_HEARTBEAT_INTERVAL_SECONDS
    #[serde(default)]
    pub heartbeat_interval: u64,
    /// The write-back's per-item budget, in seconds, or `0` on a record that
    /// names none.
    ///
    /// **Resolved once, at the launch**, out of the flag, the environment, and
    /// the launch config in that order — so an `adopt` bounds the copies it
    /// projects as its launch chose rather than by an environment that has
    /// since moved. Read through [`item_budget`](Self::item_budget), which is
    /// where `0` becomes "the record does not say" again: the shipped default,
    /// [`cli::DEFAULT_WRITEBACK_ITEM_BUDGET_SECONDS`] — and never a budget of
    /// zero, which is no budget at all. Defaulted so a record written before
    /// this field existed still reads, and resolves to that default.
    ///
    /// [`cli::DEFAULT_WRITEBACK_ITEM_BUDGET_SECONDS`]: crate::cli::DEFAULT_WRITEBACK_ITEM_BUDGET_SECONDS
    #[serde(default)]
    pub writeback_item_budget: u64,
    /// The command this run fires once when it ends with every node `done`, when
    /// the launch named one.
    ///
    /// **Resolved once, at the launch**, out of the flag and the launch config in
    /// that order, and replayed by every driver that adopts the run — `adopt`
    /// takes none of its own. Read it through
    /// [`success_hook`](Self::success_hook). Omitted when empty, like every other
    /// field added to this record after it shipped, so a record written by a
    /// build that predates it reads as naming no hook, and fires none.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub success_hook: String,
    /// The command this run fires once when it ends any other way, when the
    /// launch named one. Resolved, replayed and omitted exactly as
    /// [`success_hook`](Self::success_hook) is.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub failure_hook: String,
    /// How long a run-end hook is awaited, in seconds, or `0` on a record that
    /// names no hook.
    ///
    /// Retained only beside a hook, because it bounds nothing else: a launch
    /// naming no hook writes the record it always wrote. Read through
    /// [`hook_timeout`](Self::hook_timeout), which is where `0` becomes the
    /// shipped default and never a timeout that ends a hook before it begins.
    #[serde(default, skip_serializing_if = "is_zero")]
    pub hook_timeout: u64,
    /// The command this run runs immediately before every node-scope dispatch,
    /// whose stdout adds environment to that one child launch, when the launch
    /// named one.
    ///
    /// **Resolved once, at the launch**, out of the flag and the launch config in
    /// that order, and replayed by every driver that adopts the run — `adopt`
    /// takes none of its own. Read it through
    /// [`dispatch_env_hook`](Self::dispatch_env_hook). Omitted when empty, like
    /// every other field added to this record after it shipped, so a record
    /// written by a build that predates it reads as naming no hook, and runs none.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub dispatch_env_hook: String,
    /// How long the dispatch-env hook is awaited, in seconds, or `0` on a record
    /// that names no such hook.
    ///
    /// Retained only beside the hook it bounds, as
    /// [`hook_timeout`](Self::hook_timeout) is. Read through
    /// [`dispatch_env_hook_timeout`](Self::dispatch_env_hook_timeout), which is
    /// where `0` becomes the shipped default and never a timeout that ends the
    /// hook before it begins.
    #[serde(default, skip_serializing_if = "is_zero")]
    pub dispatch_env_hook_timeout: u64,
    /// Opaque overrides replayed on the dag-scope graph launch.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub dag_sets: Vec<String>,
    /// Opaque overrides replayed on every node-scope graph launch.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub node_sets: Vec<String>,
    /// How many times a fresh driver has been attached by `adopt`.
    #[serde(default)]
    pub adoptions: u32,
    /// What this launch said about its run's events: the two source filters it
    /// passes through, and the read-time profiles it defines.
    ///
    /// Retained here rather than derived per command because both halves outlive
    /// the launching process: `adopt` replays the source filters onto the graphs
    /// it restarts, and every later `next` and `monitor` — different processes,
    /// with no handle on the launch — reads through the profiles this run was
    /// given. Omitted when empty, like every other field added to this record
    /// after it shipped, so a build that predates it still reads what it wrote.
    #[serde(default, skip_serializing_if = "Filters::is_empty")]
    pub filters: Filters,
    /// The `onemessagebus` configuration this run's channel is kept under, when
    /// the launch named one.
    ///
    /// The document as the launch read and checked it — its transport the local
    /// one with no directory of its own, its profile `planner-channel` — retained
    /// whole rather than as the path it was read from, so a `reply` typed in
    /// another shell, a later the host bus server, and every driver that adopts the
    /// run enforce the configuration the run was launched under rather than
    /// whatever that file says now. Omitted when absent, so a record written
    /// before this field existed reads as a run under the profile as declared.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub bus_config: Option<onemessagebus::Config>,
}

impl LaunchRecord {
    /// Record that **this** process is now the run's driver.
    ///
    /// The one writer of the three fields that name a driver, because they are
    /// one fact and a record carrying two of them is a pid nothing can act on:
    /// a `stop` reading a pid with no stamp beside it cannot tell the driver it
    /// was written for from whatever the host has since given that pid to. Every
    /// path that claims a run — the launch, the driver a detached launch
    /// retains, and each adoption — goes through here.
    pub fn driven_by_this_process(&mut self) {
        self.pid = sys::pid();
        self.host = sys::hostname();
        self.started = sys::process_start_token(self.pid)
            .map(|token| token.recorded().to_string())
            .unwrap_or_default();
    }

    /// Record the observer graph run that is watching this run now.
    ///
    /// The one writer of [`graph_run`](Self::graph_run),
    /// [`observer_runs`](Self::observer_runs) and
    /// [`observer_ending`](Self::observer_ending), because they are one fact
    /// about one graph: what to address, what has watched, and whether anything
    /// still will. Every path that starts an observer — the launch, the driver a
    /// detached launch retains, each adoption, and each restart — goes through
    /// here, so a record naming a graph run that is not in its own history is a
    /// shape this crate cannot write.
    ///
    /// An empty id is a graph that started without announcing itself, which is
    /// the one case that leaves nothing to address: it moves the field, because
    /// the old address is not this observer's, and adds nothing to the history,
    /// because there is nothing to add.
    pub fn watched_by(&mut self, graph_run: String) {
        self.graph_run = graph_run;
        if !self.graph_run.is_empty() {
            self.observer_runs.push(self.graph_run.clone());
        }
        // Whatever stopped the observer before this one is not true of this one.
        self.observer_ending.clear();
    }

    /// The observer graph this run was launched with, when it was launched with
    /// one.
    ///
    /// The record is a serialized schema and an absent string field is written
    /// as no field at all, so the absence arrives back as an empty one. This is
    /// where that becomes an [`Option`] again, so no caller decides for itself
    /// what an empty graph reference means.
    pub fn observer_graph(&self) -> Option<&str> {
        (!self.graph.is_empty()).then_some(self.graph.as_str())
    }

    /// The graph this run drafts change request bodies with, when it was
    /// launched with one.
    ///
    /// The same reading [`observer_graph`](Self::observer_graph) has, for the
    /// same reason: an absent string field is written as no field at all, and
    /// this is the one place that absence becomes "there is none" again.
    pub fn pr_author_graph(&self) -> Option<&str> {
        (!self.pr_author_graph.is_empty()).then_some(self.pr_author_graph.as_str())
    }

    /// The node validator this run was launched with, when it was launched with
    /// one.
    ///
    /// The same reading [`observer_graph`](Self::observer_graph) has, and for
    /// the same reason: an absent string field is written as no field at all,
    /// and this is the one place that absence becomes "there is none" again.
    pub fn node_validator(&self) -> Option<&str> {
        (!self.node_validator.is_empty()).then_some(self.node_validator.as_str())
    }

    /// The envelope reviewer this run was launched with, when it was launched
    /// with one.
    ///
    /// The same reading [`observer_graph`](Self::observer_graph) has, and for
    /// the same reason: an absent string field is written as no field at all,
    /// and this is the one place that absence becomes "there is none" again.
    pub fn envelope_reviewer(&self) -> Option<&str> {
        (!self.envelope_reviewer.is_empty()).then_some(self.envelope_reviewer.as_str())
    }

    /// The bar this run's envelope reviewer judges against, when the launch
    /// named one — read as [`envelope_reviewer`](Self::envelope_reviewer) is.
    pub fn envelope_reviewer_bar(&self) -> Option<&str> {
        (!self.envelope_reviewer_bar.is_empty()).then_some(self.envelope_reviewer_bar.as_str())
    }

    /// Whether `session` is the session that launched this run.
    ///
    /// An unattributed launch is nobody's, including the reader's — a
    /// provenance-less run never displays as the caller's, and never accepts a
    /// command that ownership guards.
    pub fn owned_by(&self, session: &str) -> bool {
        owned_by(&self.session, session)
    }

    /// How a view names this run's owner.
    pub fn owner_label(&self, session: &str) -> String {
        owner_label(&self.launcher, &self.session, session)
    }

    /// The driver pid this record names, when it names one a reader may act on.
    ///
    /// `None` for the `0` a record carrying no pid defaults to, and a
    /// [`NonZeroU32`] so that no later reader can put it back: `0` is never a
    /// driver on any platform this builds for, so the one value that would be a
    /// nonsense pid is not a value this answer has. A pid is one third of a
    /// claim — which process, on which host, and the stamp saying it
    /// is still that process, all written together by
    /// [`driven_by_this_process`](Self::driven_by_this_process) — so a defaulted
    /// pid has no stamp beside it by construction and no reader may act on it:
    /// not the liveness readings, which would report a live run's driver dead on
    /// the strength of a pid nobody wrote, and not the stop path, which refuses
    /// an unstamped claim outright and would otherwise be aiming a teardown at
    /// whatever this host has given pid `0` a meaning of.
    pub fn driver_pid(&self) -> Option<NonZeroU32> {
        NonZeroU32::new(self.pid)
    }

    /// The stamp that proves the pid this record names, when it carries one.
    ///
    /// The third of the claim [`driver_pid`](Self::driver_pid) is one third of,
    /// served the way the other two are: an absent or unwritten stamp arrives
    /// back as an empty string, and this is the one place that becomes "there
    /// is none" again. What a reader does with `None` is its own affair — a
    /// stop leaves the pid alone, a liveness reading resolves toward live —
    /// but neither may read an empty stamp as agreeing with anything.
    pub fn driver_stamp(&self) -> Option<&str> {
        (!self.started.is_empty()).then_some(self.started.as_str())
    }

    /// The host this record names, when it names one.
    ///
    /// The same reading [`observer_graph`](Self::observer_graph) has: an absent
    /// string field arrives back as an empty one, and this is the one place that
    /// becomes "there is none" again. A reader must not claim a host the record
    /// does not name — the pid beside it would then be read as this machine's.
    pub fn recorded_host(&self) -> Option<&str> {
        (!self.host.is_empty()).then_some(self.host.as_str())
    }

    /// When the run was launched, when the record says.
    ///
    /// Reported **absent** rather than as an instant: a launch instant nobody
    /// recorded is a different fact from one recorded at the epoch, and only the
    /// second is a measurement. This is the one place that decision is made, so
    /// no caller invents a date for a record that carries none.
    pub fn launched_at(&self) -> Option<&str> {
        (!self.started_at.is_empty()).then_some(self.started_at.as_str())
    }

    /// The check-in interval this launch recorded, when it recorded one.
    ///
    /// `None` for the `0` a record carrying no interval defaults to, and a
    /// caller that must pace something takes the shipped default —
    /// [`cli::DEFAULT_HEARTBEAT_INTERVAL_SECONDS`] — or paces nothing. **Never
    /// `Some(0)`**: a zero-second check-in fires continuously, and a run whose
    /// every surface is its own check-in's is worse off than a run with no
    /// check-in at all.
    ///
    /// [`cli::DEFAULT_HEARTBEAT_INTERVAL_SECONDS`]: crate::cli::DEFAULT_HEARTBEAT_INTERVAL_SECONDS
    pub fn pacemaker_interval(&self) -> Option<u64> {
        (self.heartbeat_interval > 0).then_some(self.heartbeat_interval)
    }

    /// The write-back's per-item budget this launch recorded, in seconds, when
    /// it recorded one.
    ///
    /// `None` for the `0` a record carrying no budget defaults to, and the
    /// caller takes the shipped default —
    /// [`cli::DEFAULT_WRITEBACK_ITEM_BUDGET_SECONDS`]. A [`NonZeroU64`] so that
    /// no later reader can put zero back: a budget of zero seconds per item is
    /// no budget at all, leaving the floor as the whole deadline for every plan
    /// — the outgrown minute the setting exists to end.
    ///
    /// [`cli::DEFAULT_WRITEBACK_ITEM_BUDGET_SECONDS`]: crate::cli::DEFAULT_WRITEBACK_ITEM_BUDGET_SECONDS
    pub fn item_budget(&self) -> Option<NonZeroU64> {
        NonZeroU64::new(self.writeback_item_budget)
    }

    /// The success hook this run was launched with, when it was launched with
    /// one.
    ///
    /// The same reading [`observer_graph`](Self::observer_graph) has, and for
    /// the same reason: an absent string field is written as no field at all,
    /// and this is the one place that absence becomes "there is none" again.
    pub fn success_hook(&self) -> Option<&str> {
        (!self.success_hook.is_empty()).then_some(self.success_hook.as_str())
    }

    /// The failure hook this run was launched with, when it was launched with
    /// one. Read as [`success_hook`](Self::success_hook) is.
    pub fn failure_hook(&self) -> Option<&str> {
        (!self.failure_hook.is_empty()).then_some(self.failure_hook.as_str())
    }

    /// How long this run's hooks are awaited.
    ///
    /// The shipped default,
    /// [`cli::DEFAULT_HOOK_TIMEOUT_SECONDS`], for the `0` a record naming no
    /// timeout defaults to — and a [`NonZeroU64`], so no later reader can put a
    /// timeout of zero back.
    ///
    /// [`cli::DEFAULT_HOOK_TIMEOUT_SECONDS`]: crate::cli::DEFAULT_HOOK_TIMEOUT_SECONDS
    pub fn hook_timeout(&self) -> NonZeroU64 {
        NonZeroU64::new(self.hook_timeout).unwrap_or(crate::cli::DEFAULT_HOOK_TIMEOUT_SECONDS)
    }

    /// The dispatch-env hook this run was launched with, when it was launched
    /// with one. Read as [`success_hook`](Self::success_hook) is.
    pub fn dispatch_env_hook(&self) -> Option<&str> {
        (!self.dispatch_env_hook.is_empty()).then_some(self.dispatch_env_hook.as_str())
    }

    /// How long this run's dispatch-env hook is awaited.
    ///
    /// The shipped default, [`cli::DEFAULT_DISPATCH_ENV_HOOK_TIMEOUT_SECONDS`],
    /// for the `0` a record naming no timeout defaults to — and a
    /// [`NonZeroU64`], so no later reader can put a timeout of zero back.
    ///
    /// [`cli::DEFAULT_DISPATCH_ENV_HOOK_TIMEOUT_SECONDS`]: crate::cli::DEFAULT_DISPATCH_ENV_HOOK_TIMEOUT_SECONDS
    pub fn dispatch_env_hook_timeout(&self) -> NonZeroU64 {
        NonZeroU64::new(self.dispatch_env_hook_timeout)
            .unwrap_or(crate::cli::DEFAULT_DISPATCH_ENV_HOOK_TIMEOUT_SECONDS)
    }
}

/// Whether a count this record carries is the `0` that stands for "the record
/// does not say", so the field is omitted rather than written.
fn is_zero(value: &u64) -> bool {
    *value == 0
}

/// Whether a recorded session attributes a launch to anybody at all.
///
/// Two spellings of the same nothing: `unknown`, which is what
/// [`sys::launching_session`] answers where the environment identifies nobody,
/// and **empty**, which is what a record carrying no `session` key defaults to
/// and what a launcher that wrote a blank one recorded. Reading them apart would
/// make a run written before the key existed *ownable* by whichever reader also
/// had no session.
fn attributed(recorded: &str) -> bool {
    !recorded.is_empty() && recorded != sys::UNKNOWN_LAUNCHER
}

/// Whether `reader` is the session a launch was recorded under.
///
/// A free function rather than a method, because two documents carry that
/// recorded session — the launch record, and the summary document a listing
/// reads instead of it — and one reading over both is what keeps a run from
/// being this session's on one view and nobody's on the other.
pub(crate) fn owned_by(recorded: &str, reader: &str) -> bool {
    attributed(recorded) && recorded == reader
}

/// How a view names the owner of a launch recorded under `recorded`.
///
/// The counterpart of [`owned_by`], shared by both documents for the same
/// reason.
pub(crate) fn owner_label(launcher: &str, recorded: &str, reader: &str) -> String {
    if !attributed(recorded) {
        "[unknown]".to_string()
    } else if recorded == reader {
        "[mine]".to_string()
    } else {
        format!("[{launcher}:{}]", sys::session_digest(recorded))
    }
}

/// Read a JSON document, refusing anything the type does not accept.
pub fn read_json<T: serde::de::DeserializeOwned>(path: &Path) -> Result<T> {
    let text = fs::read_to_string(path).map_err(|e| Error::Ledger {
        path: path.to_path_buf(),
        source: e,
    })?;
    counted(text.len(), ());
    serde_json::from_str(&text).map_err(|e| Error::Invalid(format!("{}: {e}", path.display())))
}

/// Read a JSON document, or `None` when it is absent or unreadable.
///
/// Used only where the contract says an unreadable input withholds a verdict
/// rather than ending the read: a view must still render the rest of a run.
pub fn read_json_opt<T: serde::de::DeserializeOwned>(path: &Path) -> Option<T> {
    fs::read_to_string(path)
        .ok()
        .map(|text| counted(text.len(), text))
        .and_then(|text| serde_json::from_str(&text).ok())
}

/// Write a JSON document so no reader can observe it half-written.
pub fn write_json<T: Serialize>(path: &Path, value: &T) -> Result<()> {
    let body = serde_json::to_string_pretty(value)
        .map_err(|e| Error::Invalid(format!("{}: {e}", path.display())))?;
    write_atomic(path, body.as_bytes())
}

/// Write bytes so no reader can observe them half-written.
pub fn write_atomic(path: &Path, bytes: &[u8]) -> Result<()> {
    let ledger = |e: io::Error| Error::Ledger {
        path: path.to_path_buf(),
        source: e,
    };
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent).map_err(ledger)?;
    }
    // The temporary carries this process's pid so two writers racing the same
    // target cannot truncate each other's partial file before the rename.
    let temp = path.with_extension(format!("tmp.{}", sys::pid()));
    fs::write(&temp, bytes).map_err(ledger)?;
    fs::rename(&temp, path).map_err(ledger)
}

/// A record fragment an append found at the end of a file and discarded.
///
/// A writer that dies mid-record — the disk it was writing to ran out, the
/// process it was in was killed — leaves bytes that are not a whole line. The
/// next append heals the file back to its last record boundary, and this is the
/// account of what that cost: the loss is *reported* rather than quietly
/// repaired, because a store that silently patches itself is a store whose own
/// record of a run is wrong with nothing saying so.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct TornTail {
    /// When the append that met the fragment healed the file.
    pub at: String,
    /// Where the fragment began: the offset just past the last terminated
    /// record.
    pub offset: u64,
    /// How many bytes were discarded.
    pub bytes: u64,
    /// The process that healed it.
    pub healed_by: u32,
}

/// Where the fragments healed out of one append-only file are recorded.
///
/// Beside the file itself rather than inside it: the journal's own kinds are a
/// closed set the contract names, so a loss cannot be written there as an event,
/// and a reader that had to interpret a non-record line in the store would be
/// reading the very ambiguity this is about.
pub fn torn_tail_log(path: &Path) -> PathBuf {
    let mut name = path.file_name().map_or_else(
        || std::ffi::OsString::from("torn"),
        std::ffi::OsStr::to_os_string,
    );
    name.push(".torn");
    path.with_file_name(name)
}

/// Every fragment an append has healed out of one file, oldest first.
///
// llmlint: ignore[boundary_inputs_validated] the loss log is not external input: it is
// written by `report_torn_tail` in this build and by nothing else, and `TornTail` is a
// record of this crate's own, read by the module rule above — a key this build does not
// know is ignored, and a line that is not a record at all is dropped rather than taking
// the file with it. What a lenient read costs is one loss going unmentioned; what
// refusing would cost is the whole run's record, taken away over the file that exists to
// report a loss — which is the rule `read_json_opt` above already states for every ledger
// record this crate wrote.
pub fn torn_tails(path: &Path) -> Vec<TornTail> {
    // llmlint: ignore-block[no_panics_on_recoverable_errors] a line of this log the build cannot read costs that one loss going unmentioned; refusing the read would take away the whole account of what a run lost, over the file that exists to report a loss.
    read_lines(&torn_tail_log(path))
        .iter()
        .filter_map(|line| serde_json::from_str(line).ok())
        .collect()
    // llmlint: ignore-end[no_panics_on_recoverable_errors]
}

/// One line of an append-only file, and where in the file it is.
///
/// The terminator is carried because it is the only signal a store holds that a
/// writer finished: a final line with no `\n` reads exactly like a terminated
/// one through [`str::lines`], which is how a torn record used to reach a reader
/// as an ordinary one.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Record {
    /// The 1-based line number, counting every line including the blank ones.
    pub line: usize,
    /// Where the line begins in the file.
    pub offset: u64,
    /// The line, without its terminator.
    ///
    /// Decoded leniently: a record torn mid-character is not UTF-8, and what it
    /// decodes to is a line no reader can parse — which is what it is.
    pub text: String,
    /// How many bytes of the file the line is, without its terminator.
    ///
    /// Not `text.len()`: a lossy decoding is not the same length as what it was
    /// decoded from, and a view that reported a loss's size from the decoding
    /// would be describing its own reading rather than the file.
    pub bytes: u64,
    /// Whether the line ended with a newline. A final line without one is a
    /// record whose writer did not finish it.
    pub terminated: bool,
}

/// Append one line to a durable append-only file.
///
/// Three things happen under one exclusive lock, and each of them is about a
/// writer that failed rather than about the ordinary case:
///
/// - **Heal.** A file that does not end in `\n` ends in a fragment a dead writer
///   left. It is truncated back to just past the last terminator *before*
///   anything is appended, so a whole record is never glued onto half of one —
///   the measured shape of this loss was a line holding a fragment and, after
///   it, the complete record that reported the death of the process that left
///   the fragment. The discarded bytes are reported, never silently repaired.
/// - **One `write_all` of the record and its terminator**, never `writeln!`:
///   that macro writes the text and the newline as two calls, and a run's
///   journal is appended by several processes at once. A second appender landing
///   between the two tears the record in half.
/// - **Roll back.** `write_all` loops on short writes, and a full disk answers
///   the first `write(2)` with a partial count: those bytes are in the file
///   before the retry returns the error. The length is captured before the write
///   and restored after a failure, so an append that fails leaves the file on
///   the record boundary it started on.
///
/// The lock is what makes the first and third safe: a truncation by a writer
/// that excluded nobody destroys a record another writer appended in between,
/// which is the loss this exists to stop.
// llmlint: ignore[names_match_behavior] the healing is not a side effect beside the
// append, it is what appending to a store a writer died in the middle of *means* here —
// and there is deliberately no variant that skips it, because one would glue a whole
// record onto half of one again. The loss log is that heal's report rather than a second
// output a caller chooses. A name carrying both would be describing the two failures this
// function exists to survive rather than the operation every caller asks for.
#[cfg(test)]
pub fn append_line(path: &Path, line: &str) -> Result<()> {
    append_line_healed(path, line).map(|_| ())
}

/// [`append_line`], answering **how many bytes the heal in front of it
/// discarded**.
///
/// Zero where there was no fragment, which is every append but the one after a
/// writer died mid-record.
///
/// The journal writer is the one caller that has to know. It keeps the run's
/// summary current by *counting*: the document is stamped with the bytes of the
/// store it has accounted for, and every append moves that count up by the
/// record it just wrote. A heal moves it **down**, by bytes the record does not
/// replace — and a writer that did not hear about it goes on counting from an
/// offset the file no longer has a record boundary at, which is
/// the one thing a reader folding from that offset needs. Every other caller appends to a
/// file nothing accounts for and takes the shorter form above.
///
/// An append that failed still reports nothing: the error is what happened, and
/// the heal in front of it is recorded in the loss log either way. What the
/// writer does then is leave its count alone, which reads as stale — a fold,
/// never a wrong answer.
pub fn append_line_healed(path: &Path, line: &str) -> Result<u64> {
    let (healed, appended) = append_line_locked(path, line);
    if let Some(torn) = &healed {
        report_torn_tail(path, torn);
    }
    appended.map(|()| healed.map_or(0, |torn| torn.bytes))
}

/// The append itself: what it healed, and whether it wrote.
///
/// Two answers rather than one, because a heal and a failed write happen on the
/// same call and the loss must be reported either way — an append that healed a
/// fragment and then failed on a disk that is still full has still discarded the
/// fragment.
fn append_line_locked(path: &Path, line: &str) -> (Option<TornTail>, Result<()>) {
    let (torn, opened) = open_healed(path);
    let mut file = match opened {
        Ok(file) => file,
        Err(e) => return (torn, Err(e)),
    };
    let appended = write_record(path, &mut file, line);
    (torn, appended)
}

/// Open an append-only file as its only appender, healed of any fragment a dead
/// writer left: what was healed, and the handle.
///
/// Two answers rather than one, because a heal and a failed open happen on the
/// same call and the loss must be reported either way — an open that healed a
/// fragment and then could not read the file's own tail has still discarded the
/// fragment.
fn open_healed(path: &Path) -> (Option<TornTail>, Result<fs::File>) {
    let ledger = |e: io::Error| Error::Ledger {
        path: path.to_path_buf(),
        source: e,
    };
    if let Some(parent) = path.parent() {
        if let Err(e) = fs::create_dir_all(parent) {
            return (None, Err(ledger(e)));
        }
    }
    let mut file = match sys::open_locked_append(path) {
        Ok(file) => file,
        Err(e) => return (None, Err(ledger(e))),
    };
    match heal_tail(&mut file) {
        Ok(torn) => (torn, Ok(file)),
        Err(e) => (None, Err(ledger(e))),
    }
}

/// Write one record at the end of a healed, locked handle, or leave the file on
/// the boundary the write started on.
fn write_record(path: &Path, file: &mut fs::File, line: &str) -> Result<()> {
    use std::io::{Seek, SeekFrom, Write};

    let ledger = |e: io::Error| Error::Ledger {
        path: path.to_path_buf(),
        source: e,
    };
    // To the end explicitly rather than on the strength of the open mode: one of
    // the two platforms hands back a plain write handle, because an append-only
    // one cannot be truncated and truncating is half of what happens above. The
    // answer is the boundary this append starts on, which is what a failure
    // restores.
    let boundary = file.seek(SeekFrom::End(0)).map_err(ledger)?;
    let written = file
        .write_all(format!("{line}\n").as_bytes())
        .and_then(|()| file.flush());
    match written {
        Ok(()) => Ok(()),
        Err(e) => {
            // Whatever of the record reached the file goes back off it. A
            // failure to undo leaves the original failure as the one reported:
            // it is the one that says what went wrong with the host, and a
            // second error naming the same disk would displace it.
            //
            // llmlint: ignore-block[no_panics_on_recoverable_errors] a second error naming the disk the write already named would displace the one that says what happened; the undo is best-effort by design.
            // llmlint: ignore-block[changed_behavior_has_e2e] no journey reaches this: a host that refuses to truncate a descriptor this function owns refused the write before it too, and that is the error handed back.
            let _ = file.set_len(boundary);
            // llmlint: ignore-end[changed_behavior_has_e2e]
            // llmlint: ignore-end[no_panics_on_recoverable_errors]
            Err(ledger(e))
        }
    }
}

/// Truncate a fragment a dead writer left, and say what was discarded.
///
/// Called with the file's exclusive lock already held, so nothing is appending
/// while the tail is read and cut.
fn heal_tail(file: &mut fs::File) -> io::Result<Option<TornTail>> {
    use std::io::{Read, Seek, SeekFrom};

    let length = file.metadata()?.len();
    if length == 0 {
        return Ok(None);
    }
    let mut last = [0u8; 1];
    file.seek(SeekFrom::Start(length - 1))?;
    file.read_exact(&mut last)?;
    if last[0] == b'\n' {
        return Ok(None);
    }
    // Backwards a chunk at a time: a run's journal reaches megabytes, and the
    // fragment is at the end of it.
    const CHUNK: u64 = 64 * 1024;
    let mut buffer = vec![0u8; CHUNK as usize];
    let mut end = length;
    let mut boundary = 0;
    while end > 0 {
        let start = end.saturating_sub(CHUNK);
        let size = (end - start) as usize;
        file.seek(SeekFrom::Start(start))?;
        file.read_exact(&mut buffer[..size])?;
        if let Some(at) = buffer[..size].iter().rposition(|byte| *byte == b'\n') {
            boundary = start + at as u64 + 1;
            break;
        }
        end = start;
    }
    file.set_len(boundary)?;
    Ok(Some(TornTail {
        at: sys::now_rfc3339(),
        offset: boundary,
        bytes: length - boundary,
        healed_by: sys::pid(),
    }))
}

/// Report a healed fragment: durably beside the file, and on stderr.
///
/// Durably, because stderr on a detached run is a log nobody opens, and the
/// whole point is that a reader of the run can see what the run lost — see
/// [`torn_tail_log`]. On stderr as well, because the process that heals is often
/// the one a person is watching.
// llmlint: ignore-block[no_panics_on_recoverable_errors] every failure below is
// deliberately dropped, and the reason is the same one each time: the loss is already on
// stderr by the line above, and the append that healed the file **succeeded**. Failing it
// because the *report* of a heal could not be written would turn a store this process put
// back together into a run that could not record anything — the loss the whole of this
// exists to stop, arrived at through its own diagnostics.
fn report_torn_tail(path: &Path, torn: &TornTail) {
    eprintln!(
        "onepipeline: {}: discarded a {}-byte record fragment at byte {}, left by a writer \
         that did not finish it; the record it was is lost",
        path.display(),
        torn.bytes,
        torn.offset
    );
    let Ok(line) = serde_json::to_string(torn) else {
        return;
    };
    // The log of tears is appended through the same locked path, and its own
    // heal is reported on stderr alone: recording it would need a log of its
    // own, and that recursion has no end. A failed write is not propagated —
    // the loss is already on stderr, and the append that healed the file
    // succeeded.
    let (healed, _) = append_line_locked(&torn_tail_log(path), &line);
    if let Some(torn) = healed {
        eprintln!(
            "onepipeline: {}: discarded a {}-byte fragment of the loss log itself",
            torn_tail_log(path).display(),
            torn.bytes
        );
    }
}
// llmlint: ignore-end[no_panics_on_recoverable_errors]

/// Every line of an append-only file, with where it is and whether its writer
/// finished it, or nothing when the file does not exist yet.
///
/// Read as **bytes** and decoded a line at a time, which is not a detail: a
/// record torn mid-character leaves a byte sequence that is not UTF-8, and
/// decoding the file whole would fail on it and hand back an empty store — every
/// view of that run rendering as a run that recorded nothing, over one bad byte
/// at the end. Decoded per line, the tear is one unreadable line and every whole
/// record before it survives; the replacement characters it decodes to are what
/// makes it unreadable, which is what it is. The offsets stay the file's own,
/// because they are counted off the bytes rather than off the decoding.
///
/// A file this process cannot open reads as one that is not there. That is the
/// rule every ledger reader here follows — [`read_json_opt`] states it — and it
/// is what lets a view render a live run rather than refuse it over one input.
// llmlint: ignore[changed_behavior_has_e2e] the per-line decoding is three lines of one function, and `tests/e2e/journal.rs` drives them through the compiled binary against the store where a tear actually happens. Its other callers — the channel queue, the replies, the commands — differ only in the path they hand it, so a journey each would be the same three lines tested four times; the unit test below holds the decoding itself.
pub fn read_records(path: &Path) -> Vec<Record> {
    // llmlint: ignore-block[no_panics_on_recoverable_errors] the leniency is the documented rule above and predates this reader; changing it changes every ledger reader in the crate — the channel queue, the replies, the commands — rather than this one.
    let Ok(bytes) = fs::read(path) else {
        return Vec::new();
    };
    // llmlint: ignore-end[no_panics_on_recoverable_errors]
    counted(bytes.len(), ());
    records_of(&bytes, 0)
}

/// Split read bytes into records, numbering and placing them from `base`.
///
/// One splitter for the whole file and for a tail of it, so a record read either
/// way is the same record: the decoding, the terminator, and the byte count are
/// decided here and in no second place.
fn records_of(bytes: &[u8], base: u64) -> Vec<Record> {
    let mut records = Vec::new();
    let mut offset = base;
    for (index, line) in bytes.split_inclusive(|byte| *byte == b'\n').enumerate() {
        let terminated = line.ends_with(b"\n");
        let text = String::from_utf8_lossy(line)
            .trim_end_matches('\n')
            .trim_end_matches('\r')
            .to_string();
        let bytes = line.len() as u64 - u64::from(terminated);
        records.push(Record {
            line: index + 1,
            offset,
            text,
            bytes,
            terminated,
        });
        offset += line.len() as u64;
    }
    records
}

/// One line of a run's journal as the bus [`Reader`](onemessagebus::Reader) read
/// it, and where in the file it is.
///
/// The journal's envelope lines are decoded by that reader and by nothing here:
/// a line it read whole carries the envelope, and any other line carries only its
/// place, for a caller deciding what a line that is not an envelope is.
pub(crate) struct EnvelopeLine {
    /// The 1-based line number, counted from where the read began.
    pub(crate) line: usize,
    /// Where the line begins in the file.
    pub(crate) offset: u64,
    /// How many bytes of the file the line is, without its terminator.
    pub(crate) bytes: u64,
    /// Whether a newline ended it: the reader's own torn report is the one line
    /// that is `false`.
    pub(crate) terminated: bool,
    /// The envelope, where the reader read the line as a whole one.
    pub(crate) envelope: Option<crate::event::Envelope>,
}

impl EnvelopeLine {
    /// The line's own bytes, decoded leniently and without a trailing `\r`, or
    /// `None` where they could not be read again.
    ///
    /// Asked only of a line the reader did not read as an envelope — a torn tail
    /// or a refused line — which is rare, so it is read again from the file
    /// rather than held for every line of every read. A reread that fails is not
    /// an empty line: the reader refused bytes that were there.
    pub(crate) fn text(&self, path: &Path) -> Option<String> {
        read_range(path, self.offset, self.bytes).map(|bytes| {
            String::from_utf8_lossy(&bytes)
                .trim_end_matches('\r')
                .to_string()
        })
    }
}

/// Every line of a run's journal from its first `from` bytes, as the bus reader
/// read each one, or nothing where the file cannot be read from there.
///
/// The one decoder of journal envelope lines. `from` is a record boundary, as
/// [`read_records_from`] requires; a file that does not exist, cannot be opened,
/// or is shorter than `from` reads as holding nothing past it, which is the rule
/// every ledger reader here follows.
pub(crate) fn read_envelope_lines(path: &Path, from: u64) -> Vec<EnvelopeLine> {
    // llmlint: ignore-block[no_panics_on_recoverable_errors] the same leniency every
    // ledger reader here follows, stated on `read_records` above: a file this process
    // cannot open reads as one that is not there.
    let Ok(reader) = onemessagebus_agent::Reader::open_at(path, from) else {
        return Vec::new();
    };
    // llmlint: ignore-end[no_panics_on_recoverable_errors]
    let mut lines = Vec::new();
    let mut start = from;
    for (index, reading) in reader.enumerate() {
        let line = index + 1;
        match reading {
            onemessagebus::Reading::Record(record) => {
                lines.push(EnvelopeLine {
                    line,
                    offset: start,
                    bytes: record.position - start - 1,
                    terminated: true,
                    envelope: Some(record.envelope),
                });
                start = record.position;
            }
            onemessagebus::Reading::Refused(refused) => {
                lines.push(EnvelopeLine {
                    line,
                    offset: refused.at,
                    bytes: refused.position - refused.at - 1,
                    terminated: true,
                    envelope: None,
                });
                start = refused.position;
            }
            onemessagebus::Reading::Torn(torn) => {
                lines.push(EnvelopeLine {
                    line,
                    offset: torn.at,
                    bytes: torn.bytes,
                    terminated: false,
                    envelope: None,
                });
                start = torn.at + torn.bytes;
            }
        }
    }
    counted(usize::try_from(start - from).unwrap_or(usize::MAX), lines)
}

/// The **raw bytes** of a byte range of a file, or `None` where the file does not
/// hold all of them.
///
/// The one reader here that hands back bytes rather than records, and it exists
/// because a checkpoint's marker is corroborated against the journal's own bytes:
/// a record decoded into a [`Record`] has been through `from_utf8_lossy` and had
/// its terminator trimmed, so a digest taken over that is a digest of the
/// decoding rather than of the file. `None` for a file that cannot be opened or
/// that is shorter than the range asked for, which is the same answer a marker
/// the journal cannot corroborate gets.
///
/// Counted like every other read here — see the note at the head of this file.
pub(crate) fn read_range(path: &Path, from: u64, len: u64) -> Option<Vec<u8>> {
    use std::io::{Read, Seek, SeekFrom};
    let mut file = fs::File::open(path).ok()?;
    file.seek(SeekFrom::Start(from)).ok()?;
    let want = usize::try_from(len).ok()?;
    let mut bytes = vec![0u8; want];
    file.read_exact(&mut bytes).ok()?;
    Some(counted(bytes.len(), bytes))
}

/// Every line of an append-only file, or nothing when it does not exist yet.
pub fn read_lines(path: &Path) -> Vec<String> {
    read_records(path)
        .into_iter()
        .filter(|record| !record.text.trim().is_empty())
        .map(|record| record.text)
        .collect()
}

/// Who holds a run's single-writer lock.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct LockRecord {
    /// The holding process.
    pub pid: u32,
    /// The host that pid is meaningful on.
    pub host: String,
    /// When it was taken.
    pub acquired_at: String,
    /// What the holder is doing, for the refusal message.
    pub verb: String,
    /// The holder's own process start token, as
    /// [`sys::process_start_token`] read it when the lock was taken.
    ///
    /// The pid beside it says *which* process; this says it is still that
    /// process. A pid is reused, so a lock left behind by a driver that died two
    /// days ago names a pid the host may since have handed to something else —
    /// and a view reading the pid alone renders that as a live dispatch. Compared
    /// for equality against a fresh reading and never parsed.
    ///
    /// Empty when this host would not say, and on a record written before the
    /// field existed. Omitted when empty, like every other field added to a
    /// record after it shipped. Empty is **not** a match: it leaves a reader
    /// unable to prove the holder either way, which is the answer it has.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub started: String,
}

/// Take `path` as this process's claim, or report who holds it.
///
/// [`OwnershipLock`]'s own acquisition: a run has one writer, and this is the
/// file that says which process it is.
///
/// Taking a claim nobody holds is exclusive, because creating a file exclusively
/// is what the filesystem decides — and the file is created **with** its record
/// already in it, so no reader ever finds the lock without a holder to name. See
/// [`create_exclusively_filled`] for the operation that gives both on each
/// platform. Reclaiming the claim of a holder this host can prove is gone — which is what `adopt` recovers a dead driver's run by, and
/// what a `reply` takes a dead driver's queue over by — is exclusive for the same
/// reason: the reclaim is contended for by creating a file exclusively, and only
/// the process that created it may put its own record where the dead one was.
/// See [`reclaim`] for the shape, and why two processes that both proved the
/// same holder dead at the same instant cannot both end up holding the run.
fn claim_or_report_the_holder(path: &Path, run: &str, verb: &str) -> Result<()> {
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent).map_err(|e| Error::Ledger {
            path: parent.to_path_buf(),
            source: e,
        })?;
    }
    let record = LockRecord {
        pid: sys::pid(),
        host: sys::hostname(),
        acquired_at: sys::now_rfc3339(),
        verb: verb.to_string(),
        started: sys::process_start_token(sys::pid())
            .map(|token| token.recorded().to_string())
            .unwrap_or_default(),
    };
    let body = serde_json::to_string(&record)
        .map_err(|e| Error::Invalid(format!("{}: {e}", path.display())))?;

    loop {
        if create_exclusively_filled(path, &body)? {
            return Ok(());
        }
        let held = match read_lock_file(path)? {
            LockFile::Record(held) => held,
            // Let go between the create that found it and this read, so there
            // is nobody to report: it is taken the way a lock nobody holds is.
            LockFile::Absent => continue,
            // An unreadable lock is still a claim. Refusing is the safe
            // reading: the alternative is a second writer on a run
            // whose first writer cannot be identified.
            LockFile::Unreadable => return Err(unreadable_lock(path, run)),
        };
        // A holder on this host that this host can prove is gone leaves a lock
        // nothing will release. Reclaim it — or report whoever did.
        if !(held.host == sys::hostname() && !sys::process_may_be_live(held.pid)) {
            return Err(locked_by(run, &held));
        }
        match reclaim(path, &held, &body)? {
            Reclaimed::Won => return Ok(()),
            Reclaimed::HeldBy(holder) => return Err(locked_by(run, &holder)),
            Reclaimed::Unreadable(at) => return Err(unreadable_lock(&at, run)),
            // The lock was let go while this process was contending for it,
            // so there is nobody to reclaim it from: it is taken the way a
            // lock nobody holds is taken, exclusively.
            Reclaimed::Released => {}
        }
    }
}

/// How a reclaim of a dead holder's lock ended for the process that attempted it.
enum Reclaimed {
    /// This process's record is now the lock.
    Won,
    /// Another process holds the run, or is taking it over at this instant.
    HeldBy(LockRecord),
    /// A claim on the run exists that this build cannot read, at this path.
    Unreadable(PathBuf),
    /// The lock was released while this process was contending for it.
    Released,
}

/// What the lock path holds, told apart three ways because a reclaim answers
/// each differently.
enum LockFile {
    Record(LockRecord),
    Absent,
    /// Bytes that are not a record this build reads.
    Unreadable,
}

/// How long a read of a lock the filesystem refuses is tried again before the
/// refusal is reported as the filesystem said it.
///
/// A refusal is not a record nobody can read. Windows refuses to open a name
/// whose deletion is pending — answering access denied rather than not found —
/// for as long as another process still has it open, and the reclaim takes its
/// entries away while its losers are reading them; what the name holds a moment
/// later is the answer. A refusal that outlasts this is not that.
const LOCK_READ_PATIENCE: std::time::Duration = std::time::Duration::from_secs(1);

fn read_lock_file(path: &Path) -> Result<LockFile> {
    let deadline = std::time::Instant::now() + LOCK_READ_PATIENCE;
    loop {
        match fs::read_to_string(path) {
            Ok(text) => {
                return Ok(match serde_json::from_str(&counted(text.len(), text)) {
                    Ok(record) => LockFile::Record(record),
                    Err(_) => LockFile::Unreadable,
                })
            }
            Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(LockFile::Absent),
            Err(_) if std::time::Instant::now() < deadline => {
                std::thread::sleep(std::time::Duration::from_millis(1));
            }
            Err(e) => {
                return Err(Error::Ledger {
                    path: path.to_path_buf(),
                    source: e,
                })
            }
        }
    }
}

/// Put this process's record where a dead holder's is, or say who got there
/// first.
///
/// Two processes that both read `dead` and both proved it gone would, each
/// writing its own record and reading it back, each be able to read its own —
/// the moment the driver of a run dies with two replies queued behind it is
/// exactly the moment two processes do that, and the run's queue was then
/// reconciled twice. So the write is never what decides. The dead record is
/// replaced by the process that **created its reclaim entry**, exclusively — a
/// file named after the dead record, so every process that proved *that* record
/// dead contends under one name and the filesystem picks one — and that process
/// re-reads the lock before it writes, so a record already put there by an
/// earlier winner is reported rather than overwritten. A loser reads the winner's
/// record, or the winner's entry while its record is on the way, and reports it
/// as the holder: it never writes.
///
/// The entries are numbered, and a number is stepped over only when the process
/// that created it is one this host can prove is gone with the dead record still
/// in place — the shape [`Handover`] describes, keyed to one dead record. A
/// reclaimer that dies between creating its entry and writing the lock therefore
/// costs the next one a name, not the run. The winner takes the entries away
/// once its record is the lock, and never before: with the dead record gone
/// from the path, a late contender that creates a fresh entry re-reads the lock,
/// finds the winner, and reports it.
///
/// The dead record is replaced by [`write_atomic`]'s rename rather than by a
/// link, because the name is already there to replace: `rename(2)` on Linux and
/// macOS, and the replacing `MoveFileExW` or `SetFileInformationByHandle` that
/// `std::fs::rename` is on Windows, swap which complete file the name refers to,
/// so a reader finds the dead record or the winner's and never part of either.
/// A rename is not exclusive on any of the three, and needs not be here: only
/// the process that created the entry makes it.
fn reclaim(path: &Path, dead: &LockRecord, body: &str) -> Result<Reclaimed> {
    let key = reclaim_key(dead);
    let mut number = 1u64;
    loop {
        let entry = reclaim_entry(path, &key, number);
        if create_exclusively_filled(&entry, body)? {
            // This process alone may replace `dead` — provided it is still
            // what the path holds, which is what an earlier winner changes.
            let outcome = match read_lock_file(path) {
                Ok(LockFile::Record(now)) if now == *dead => {
                    write_atomic(path, body.as_bytes()).map(|()| Reclaimed::Won)
                }
                Ok(LockFile::Record(now)) => Ok(Reclaimed::HeldBy(now)),
                Ok(LockFile::Absent) => Ok(Reclaimed::Released),
                Ok(LockFile::Unreadable) => Ok(Reclaimed::Unreadable(path.to_path_buf())),
                Err(refused) => Err(refused),
            };
            // Either way this process is done contending, so the entries go: a
            // live creator's entry left behind would name it the holder of a run
            // it never took.
            for done in 1..=number {
                let _ = fs::remove_file(reclaim_entry(path, &key, done));
            }
            return outcome;
        }
        // Another process created this number. If the lock has moved on, it is
        // that process's — or its successor's — and is reported as it stands.
        // llmlint: ignore-block[changed_behavior_has_e2e] the states below are a
        // reclaimer dying, or the run being let go, between one filesystem operation
        // and the next, and no journey can place a subprocess there; the unit tests
        // beside this file put the entries in place by hand, which needs the private
        // names this file owns. What a user reaches — two replies taking one dead
        // driver's run over — is driven in `tests/e2e/driver.rs`.
        match read_lock_file(path)? {
            LockFile::Record(now) if now == *dead => {}
            LockFile::Record(now) => return Ok(Reclaimed::HeldBy(now)),
            LockFile::Absent => return Ok(Reclaimed::Released),
            LockFile::Unreadable => return Ok(Reclaimed::Unreadable(path.to_path_buf())),
        }
        // The dead record was still in place, so whoever created this number
        // is between that and writing the lock — or died there.
        match read_lock_file(&entry)? {
            LockFile::Record(reclaimer)
                if reclaimer.host == sys::hostname()
                    && !sys::process_may_be_live(reclaimer.pid) =>
            {
                number += 1;
            }
            LockFile::Record(reclaimer) => return Ok(Reclaimed::HeldBy(reclaimer)),
            // Gone since the create was refused: its creator has written the
            // lock and taken the entries away. The next look at the lock
            // finds that record.
            LockFile::Absent => {}
            LockFile::Unreadable => return Ok(Reclaimed::Unreadable(entry)),
        }
        // llmlint: ignore-end[changed_behavior_has_e2e]
    }
}

/// Create `path` exclusively, **with** `body` already in it.
///
/// One operation rather than a create followed by a write, because what a
/// contender learns from a lock or an entry is who created it, and it reads the
/// file at exactly the moment its creator would be between the two — which a
/// reader on Linux finds thousands of times in a few thousand claims, and one on
/// Windows found in the race it lost. So the body is written to a name only this
/// process uses and *linked* to `path`: the link is refused where `path` exists,
/// and where it is not refused `path` was never observable empty. `Ok(true)`
/// when this process created it, `Ok(false)` when something already had.
///
/// The link is **exclusive and atomic in one call** on every platform the merge
/// path runs: `link(2)` on Linux and macOS, which fails with `EEXIST` where the
/// name exists, and `CreateHardLinkW` on Windows, which fails with
/// `ERROR_ALREADY_EXISTS` — and on all three the name, once it appears, is a
/// second name for a file that was already complete. Neither nearer call gives
/// both: `create_new` followed by a write is exclusive and exposes the name empty
/// in between, and a rename exposes no partial file and replaces whatever holds
/// the name.
fn create_exclusively_filled(path: &Path, body: &str) -> Result<bool> {
    static NONCE: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
    let ledger = |e: io::Error| Error::Ledger {
        path: path.to_path_buf(),
        source: e,
    };
    let mut name = path
        .file_name()
        .map(OsStr::to_os_string)
        .unwrap_or_default();
    name.push(format!(
        ".tmp.{}.{}",
        sys::pid(),
        NONCE.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
    ));
    let temp = path.with_file_name(name);
    if let Err(e) = fs::write(&temp, body) {
        let _ = fs::remove_file(&temp);
        return Err(ledger(e));
    }
    let linked = match fs::hard_link(&temp, path) {
        Ok(()) => Ok(true),
        Err(e) if e.kind() == io::ErrorKind::AlreadyExists => Ok(false),
        Err(e) => Err(ledger(e)),
    };
    let _ = fs::remove_file(&temp);
    linked
}

/// The name every reclaim of one dead record contends under.
///
/// Spelled from the record itself — the pid, the instant it was taken, and the
/// start token beside it — so two processes that read the same record derive the
/// same name without agreeing on anything else, and two records never share one.
fn reclaim_key(dead: &LockRecord) -> String {
    let plain =
        |text: &str| -> String { text.chars().filter(char::is_ascii_alphanumeric).collect() };
    format!(
        "{}-{}-{}",
        dead.pid,
        plain(&dead.acquired_at),
        plain(&dead.started)
    )
}

/// One numbered entry of one dead record's reclaim, beside the lock it is for.
fn reclaim_entry(path: &Path, key: &str, number: u64) -> PathBuf {
    let mut name = path
        .file_name()
        .map(OsStr::to_os_string)
        .unwrap_or_default();
    name.push(format!(".reclaim.{key}.{number}"));
    path.with_file_name(name)
}

fn locked_by(run: &str, holder: &LockRecord) -> Error {
    Error::Locked {
        run: run.to_string(),
        pid: holder.pid,
        host: holder.host.clone(),
        verb: holder.verb.clone(),
    }
}

/// What a claim refused over a lock nobody can be named as holding says, as the
/// source of the [`Error::Ledger`] that reports it.
///
/// Not an [`Error::Locked`]: that names a holding process, and the one thing
/// known here is that no process can be named. Reporting one as pid 0 sent
/// whoever read the refusal looking for a process that does not exist.
#[derive(Debug)]
struct UnreadableLock {
    run: String,
}

impl std::fmt::Display for UnreadableLock {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "run '{}' is claimed by an unreadable lock: what it holds is not a record this \
             build can read, so no process can be named as its holder",
            self.run
        )
    }
}

impl std::error::Error for UnreadableLock {}

fn unreadable_lock(path: &Path, run: &str) -> Error {
    Error::Ledger {
        path: path.to_path_buf(),
        source: io::Error::new(
            io::ErrorKind::InvalidData,
            UnreadableLock {
                run: run.to_string(),
            },
        ),
    }
}

/// Whether `error` refused a claim over a lock nobody can be named as holding —
/// which is still a claim on the run, and answered as one.
pub(crate) fn is_unreadable_lock(error: &Error) -> bool {
    matches!(
        error,
        Error::Ledger { source, .. }
            if source.get_ref().is_some_and(|inner| inner.is::<UnreadableLock>())
    )
}

/// The gate that makes **accepting a command** and **letting the run go** two
/// things that cannot interleave.
///
/// A departing owner reads the command queue and, finding it empty, releases the
/// run; a submitter asks whether anything is driving the run and, finding one,
/// queues behind it. Interleaved, those two produce the outcome neither party
/// would accept: an envelope accepted onto the queue of a run whose owner has
/// just left. **No absence of I/O closes that** — the window is a preemption, not
/// a duration — so both parties hold this across their pair, and one of the two
/// orders happens instead.
///
/// What it deliberately does *not* guard is applying an edit: what it serializes
/// is a file read and a file remove, so a holder is never inside a subprocess, a
/// conversation or a graph fold. A party that cannot take it does neither of its
/// two things and says so.
///
/// # What arbitrates it
///
/// The gate is the **highest-numbered entry that exists**, and its holder is
/// whoever created that entry. Taking it means creating the number above the
/// highest one seen, exclusively — so two parties that looked at the same moment
/// and picked the same number meet **on one path**, where the filesystem decides:
/// exactly one create succeeds, and the other looks again and finds a live holder
/// to wait for.
///
/// That is the whole of it, and the two things it does not do are why it holds:
///
/// * **Nothing is ever numbered above a holder that is alive.** A number above
///   the top is taken only when the top's holder is one this host can prove is
///   gone, and every uncertainty — another host's entry, a body that cannot be
///   read, a listing that cannot be read — counts as alive.
/// * **No party ever removes an entry it did not create.** A dead holder's entry
///   is left exactly where it is and stepped *over*, so there is no moment at
///   which one process is deleting a file another has just taken. Stale entries
///   are the residue of a process dying inside a two-file-operation section; they
///   cost a name each and block nobody.
///
/// A name carrying its maker's own clock could do neither. Clocks disagree
/// between machines, and one process can read its own before another and write it
/// after, so a party arriving later could sort below the holder and both would
/// read themselves in. A name carrying the maker's *identity* beside the number
/// is worse still: two parties picking one number then create two different
/// paths, and the filesystem is never asked to arbitrate at all.
#[derive(Debug)]
pub(crate) struct Handover {
    /// The entry this process created, which nothing else ever removes.
    entry: PathBuf,
}

/// Where the entries live: one directory per run, beside its channel.
fn handover_entries(paths: &RunPaths) -> PathBuf {
    paths.channel("handover")
}

/// The verb a `reply` writes into a run's ownership lock while it applies what
/// the run's driver did not.
///
/// Read as well as written: it is how one reply taking a run over tells itself
/// from a driver driving it, and the two answer differently.
pub(crate) const REPLY_VERB: &str = "reply";

/// The verb a driver writes into a run's ownership lock while it drives it.
pub(crate) const DRIVE_VERB: &str = "drive";

/// How long a party waits for a **live holder** before refusing.
///
/// A bound on patience and never a licence: what waits it out is a peer inside a
/// section of two file operations, so reaching it means something is wrong with
/// that peer rather than that the wait is over. Time establishes nothing about
/// whether a holder is still inside — a live process can be stopped for as long
/// as anyone likes — so the answer here is that the gate was **not taken**, and
/// every caller's job is to do nothing that needed it.
const HANDOVER_PATIENCE: std::time::Duration = std::time::Duration::from_secs(30);

/// How many times one attempt will lose the race for a number before it reports
/// the gate untaken.
const NUMBERS_TRIED: usize = 1_000;

// llmlint: ignore-block[changed_behavior_has_e2e] what a journey can drive of this is
// driven: `driver::an_edit_that_cannot_be_gated_is_refused_and_nothing_reaches_the_queue`
// and `driver::a_driver_that_cannot_be_gated_on_its_way_out_leaves_the_run_claimed` are
// real runs meeting a gate this host cannot take. What no journey can drive is which party
// is inside the section when the other arrives, or which of two writes lands first: both
// are decided in microseconds inside two processes, and no CLI input places anything
// there. Those are driven instead by the cases beside this code in `tests` — two parties
// that looked at one gate, an arrival while another is inside, a holder this host knows is
// gone, and three it cannot account for — and from both sides at once by
// `engine::tests::a_submission_either_reaches_the_departing_owners_queue_or_finds_the_run_free`.
impl Handover {
    /// Hold the gate for this run, or report that this process is not inside it.
    pub(crate) fn hold(paths: &RunPaths) -> Result<Self> {
        Self::hold_within(paths, HANDOVER_PATIENCE)
    }

    /// The same, with the patience stated — which is how a test drives a
    /// contention that outlasts it without waiting out the shipped bound.
    pub(crate) fn hold_within(paths: &RunPaths, patience: std::time::Duration) -> Result<Self> {
        let dir = handover_entries(paths);
        fs::create_dir_all(&dir).map_err(|e| not_taken(&paths.run, &e.to_string()))?;
        let host = sys::hostname();
        let deadline = std::time::Instant::now() + patience;
        for _ in 0..NUMBERS_TRIED {
            let top = highest_entry_in(&dir).map_err(|e| not_taken(&paths.run, &e.to_string()))?;
            // Somebody holds the gate, so far as this process can establish —
            // and everything it cannot establish counts as somebody.
            if top.is_some_and(|top| !holder_is_gone(&dir, top, &host)) {
                if std::time::Instant::now() >= deadline {
                    return Err(not_taken(
                        &paths.run,
                        &format!(
                            "it has been held for {}s by a party this process cannot show \
                             has gone",
                            patience.as_secs()
                        ),
                    ));
                }
                std::thread::sleep(std::time::Duration::from_millis(5));
                continue;
            }
            // A gate numbered to the ceiling is one nothing can be taken above,
            // which is the refusal every other way of not getting in gives rather
            // than a number that wraps. It is also what an entry this build
            // cannot place reads as, so this is the arm that answers those.
            let Some(below) = top.unwrap_or(0).checked_add(1).map(|_| top.unwrap_or(0)) else {
                return Err(not_taken(
                    &paths.run,
                    "its order is at the highest number this build can write",
                ));
            };
            match Self::take_above(paths, &dir, below, sys::pid(), &host)? {
                Some(held) => return Ok(held),
                // The number went to somebody else between the look and the
                // write, which is the race this shape exists to arbitrate: look
                // again, and find the party that won it.
                None => continue,
            }
        }
        Err(not_taken(
            &paths.run,
            &format!("{NUMBERS_TRIED} numbers in its order were taken while this process looked"),
        ))
    }

    /// One attempt to take the number above `observed`, as `pid` of `host`.
    ///
    /// `None` where another party took that number first — which is the whole
    /// arbitration, and is why the number alone names the file: two parties that
    /// looked at the same moment write to **one path**, and the filesystem says
    /// which of them holds it. The identity goes in the body, where it tells a
    /// later waiter whether this holder is still there; in the name it would give
    /// each party a path of its own and arbitrate nothing.
    ///
    /// Takes the identity rather than reading it, so that a test can be two
    /// parties that looked at one gate.
    fn take_above(
        paths: &RunPaths,
        dir: &Path,
        observed: u64,
        pid: u32,
        host: &str,
    ) -> Result<Option<Self>> {
        let entry = dir.join(entry_named(observed + 1));
        let mut file = match fs::OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(&entry)
        {
            Ok(file) => file,
            Err(e) if e.kind() == io::ErrorKind::AlreadyExists => return Ok(None),
            Err(e) => return Err(not_taken(&paths.run, &e.to_string())),
        };
        use std::io::Write;
        if let Err(e) = file.write_all(body_of_entry(pid, host).as_bytes()) {
            // An entry whose body never landed is one no waiter can read an
            // identity out of, and so one they all wait on for ever. It goes with
            // the attempt that failed to write it — this process created it, so
            // this process is the one that may.
            drop(file);
            let _ = fs::remove_file(&entry);
            return Err(not_taken(&paths.run, &e.to_string()));
        }
        Ok(Some(Self { entry }))
    }
}

// llmlint: ignore-end[changed_behavior_has_e2e]

/// The highest-numbered entry in the gate, or `None` where nobody is in it.
///
/// A name outside the shape [`entry_named`] writes is **not** passed over: it is
/// a claim this build cannot read, and the answer to a claim it cannot read is
/// the same as to one it can — somebody is in there. So it reports the highest
/// number it did read, and [`holder_is_gone`] answers that such a holder is not
/// one this host can show has gone.
fn highest_entry_in(dir: &Path) -> Result<Option<u64>> {
    let listing = fs::read_dir(dir).map_err(|e| Error::Ledger {
        path: dir.to_path_buf(),
        source: e,
    })?;
    let mut highest: Option<u64> = None;
    let mut a_name_this_build_did_not_write = false;
    for entry in listing {
        let name = entry
            .map_err(|e| Error::Ledger {
                path: dir.to_path_buf(),
                source: e,
            })?
            .file_name();
        match number_of_entry(Path::new(&name)) {
            Some(number) => highest = Some(highest.map_or(number, |high: u64| high.max(number))),
            None => a_name_this_build_did_not_write = true,
        }
    }
    match (highest, a_name_this_build_did_not_write) {
        // Something is in the gate that this build cannot place. It is not
        // stepped over: `u64::MAX` is a holder no number can be taken above and
        // no host can show has gone, so every party waits and then refuses.
        (_, true) => Ok(Some(u64::MAX)),
        (highest, false) => Ok(highest),
    }
}

/// Whether the holder of entry `number` is one **this host** can prove is gone.
///
/// Every uncertainty answers `false`, because the only thing this decides is
/// whether a party may number itself above that holder: a body it cannot read, an
/// identity it cannot parse, and an entry another host wrote are each a holder it
/// cannot rule out. A pid is only the host that issued it to judge.
fn holder_is_gone(dir: &Path, number: u64, this_host: &str) -> bool {
    let Ok(body) = fs::read_to_string(dir.join(entry_named(number))) else {
        return false;
    };
    let Some((pid, host)) = identity_of_body(&body) else {
        return false;
    };
    host == this_host && !sys::process_may_be_live(pid)
}

/// One entry's name: the number it holds in the gate's order, and nothing else.
///
/// **Nothing else on purpose.** The name is the path two parties picking the same
/// number contend on, so anything in it that differs between them — an identity,
/// a clock reading — would give each a path of its own and leave the race
/// unarbitrated.
fn entry_named(number: u64) -> String {
    format!("{number:020}")
}

/// The number an entry's name holds, or `None` for a name outside that shape.
fn number_of_entry(entry: &Path) -> Option<u64> {
    let name = entry.file_name()?.to_str()?;
    (name.len() == 20 && name.bytes().all(|byte| byte.is_ascii_digit()))
        .then(|| name.parse().ok())
        .flatten()
}

/// What an entry carries: who holds it, for a waiter deciding whether they are
/// still there, and for an operator reading one left behind.
fn body_of_entry(pid: u32, host: &str) -> String {
    format!("{pid} {host}")
}

/// The identity an entry's body carries, or `None` for a body this build cannot
/// read — which is a holder it cannot show has gone.
fn identity_of_body(body: &str) -> Option<(u32, &str)> {
    let (pid, host) = body.trim().split_once(' ')?;
    Some((pid.parse().ok()?, host))
}

/// The one answer a party that is not inside the gate gets, whatever kept it
/// out: what it must not do, and why.
fn not_taken(run: &str, because: &str) -> Error {
    Error::Refused(format!(
        "the handover gate of run '{run}' could not be taken, so nothing was accepted onto \
         its command queue and nothing was released: {because}. This process is not inside \
         the gate, and going on without it is what would let an edit be accepted by a run \
         whose owner has already left"
    ))
}

impl Drop for Handover {
    /// Takes away the entry this process created, and never another's.
    fn drop(&mut self) {
        let _ = fs::remove_file(&self.entry);
    }
}

/// The run's ownership lock, released when this value is dropped.
///
/// The process driving a run is the only writer of its graph and its journal's
/// graph records, and this is what makes that true across processes: the lock
/// file is created exclusively, so a second writer loses the race rather than
/// interleaving with the first.
#[derive(Debug)]
pub struct OwnershipLock {
    path: PathBuf,
    /// Whether this value still holds the lock. A lock released by hand does
    /// not release again on drop.
    held: bool,
}

impl OwnershipLock {
    /// Take the run's lock, or report who holds it.
    ///
    /// A lock whose holder this host can prove is gone is reclaimed: a driver
    /// that died mid-run must not leave its run unwritable forever, which is
    /// the state `adopt` exists to recover from.
    pub fn acquire(paths: &RunPaths, verb: &str) -> Result<Self> {
        let path = paths.lock();
        claim_or_report_the_holder(&path, &paths.run, verb)?;
        Ok(Self { path, held: true })
    }

    /// Release the lock now rather than at the end of the scope.
    pub fn release(mut self) {
        self.remove();
    }

    /// Stop holding the lock **without** releasing the run.
    ///
    /// What a writer does when it cannot hand the run over safely: releasing it
    /// there is the one thing that would let an edit be accepted by a run whose
    /// owner has gone, and this leaves the claim standing instead. The record
    /// then names a process that is about to end, which the next writer reclaims
    /// on the spot — so the run is recovered by the same path that recovers one
    /// whose driver died, rather than by nothing.
    pub(crate) fn abandon(mut self) {
        self.held = false;
    }

    fn remove(&mut self) {
        if self.held {
            let _ = fs::remove_file(&self.path);
            self.held = false;
        }
    }
}

impl Drop for OwnershipLock {
    fn drop(&mut self) {
        self.remove();
    }
}

/// One live dispatch's claim on the process it is running in.
///
/// The registry answers a question neither the launch record nor the ownership
/// lock can: *what is this run actually running, and where*. Both of those name
/// a **driver**, and a driver is not the work — it starts the work, and when it
/// dies the work it started is reparented away and outlives it, findable by
/// nothing that descends from a pid either record holds. That is a live dispatch
/// a stop cannot reach and an operator is told is over.
///
/// Written by the machine running the dispatch, which is the one that knows
/// which process the work is in, and removed when that dispatch ends. Every
/// field is required, the stamp included: a record that cannot prove its own pid
/// is not a weaker entry but an unusable one, and the type is what stops one
/// being written or read.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DispatchRecord {
    /// The node this dispatch is running.
    pub node: String,
    /// The process the work is running in.
    pub pid: u32,
    /// The host that pid is meaningful on.
    pub host: String,
    /// When the dispatch was recorded.
    pub dispatched_at: String,
    /// That process's own start token, as [`sys::process_start_token`] read it
    /// when the dispatch started.
    ///
    /// The same proof, and for the same reason, as the ownership lock's: the pid
    /// says *which* process and this says it is still that one. A record outlives
    /// the driver that wrote it — that is the case it exists for — so by the time
    /// anything reads it the host may have handed the pid to a stranger, and a
    /// teardown aimed at that would end work this run never started.
    pub started: String,
}

impl DispatchRecord {
    /// Whether this entry is one a reader may act on.
    ///
    /// An empty stamp parses and proves nothing, which is the one state the
    /// field's type cannot rule out. A registry holding one cannot say whether
    /// the pid beside it is still this run's work, and *cannot say* is the answer
    /// this registry exists to stop being read as *nothing is running*.
    fn is_usable(&self) -> bool {
        !self.started.trim().is_empty()
    }
}

/// A dispatch's entry in the registry, removed when this value is dropped.
///
/// RAII for the same reason [`OwnershipLock`] is: a dispatch ends in more ways
/// than it settles, and every one of them drops this. The single ending that
/// leaves the entry behind is the process itself dying, which is exactly when a
/// stop needs it.
#[derive(Debug)]
pub struct DispatchClaim {
    path: PathBuf,
    /// What this claim recorded, so it removes its **own** entry and never a
    /// later dispatch's: the host reissues pids, and a record keyed by one is
    /// only this dispatch's while the process behind it is.
    started: String,
}

impl Drop for DispatchClaim {
    fn drop(&mut self) {
        let ours = read_json_opt::<DispatchRecord>(&self.path)
            .is_some_and(|held| held.started == self.started);
        if ours {
            let _ = fs::remove_file(&self.path);
        }
    }
}

/// Record that this run is running `node` in `pid`, on this host, or refuse.
///
/// A **trust boundary**, not bookkeeping. The registry is the only record of
/// where a run's work actually is, so a dispatch this run cannot register is a
/// process nothing will ever find: not the operator reading a view, and not the
/// `stop` they run when they need the work to end. Continuing anyway would buy
/// one dispatch at the price of the guarantee every later stop rests on — so the
/// caller is given the failure and ends the dispatch with it.
///
/// Two ways to fail, and both are refusals rather than empty entries. A host that
/// will not say when `pid` started leaves nothing that could prove the pid is
/// still this process, and an entry a reader cannot act on is one that would make
/// a later stop refuse instead. A write that did not land — or landed as
/// something other than what was written — is the same absence with a file in the
/// way, so what was written is read back before the claim is handed over.
pub fn claim_dispatch(paths: &RunPaths, node: &str, pid: u32) -> Result<DispatchClaim> {
    // Unique for the life of this process, which is what separates two dispatches
    // running inside it. Across processes the pid separates them, and across a
    // pid this host has reissued the stamp does.
    static CLAIMED: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
    let claim = CLAIMED.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    let Some(started) = sys::process_start_token(pid) else {
        return Err(Error::Refused(format!(
            "run '{}': node '{node}': this host will not say when pid {pid} started, so its \
             dispatch cannot be recorded as running there and nothing could prove that pid is \
             still this run's work",
            paths.run
        )));
    };
    let record = DispatchRecord {
        node: node.to_string(),
        pid,
        host: sys::hostname(),
        dispatched_at: sys::now_rfc3339(),
        started: started.recorded().to_string(),
    };
    let path = paths.dispatch(pid, claim);
    write_dispatch(paths, &path, claim, &record)?;
    // Read back through the same reader a stop uses, because what this promises
    // its caller is not that a write returned but that the registry now holds an
    // entry that reader will act on.
    //
    // llmlint: ignore-block[changed_behavior_has_e2e] a write that lands as something other
    // than what was written is a filesystem lying to a process, not anything a user can type
    // or a suite can arrange: every fault a journey *can* set — no directory, a file where
    // one has to be, an entry rewritten afterwards — fails earlier, in branches
    // `a_dispatch_this_run_cannot_record_is_refused_and_does_not_run` and
    // `stopping_a_run_whose_registry_cannot_be_read_refuses_and_leaves_the_run_retryable`
    // drive end to end. What this arm adds is that the promise is checked rather than
    // assumed, and `a_dispatch_the_registry_cannot_record_is_refused` holds the refusal it
    // produces against the real filesystem.
    match read_json::<DispatchRecord>(&path) {
        Ok(held) if held == record => Ok(DispatchClaim {
            path,
            started: record.started,
        }),
        Ok(_) | Err(_) => Err(Error::Refused(format!(
            "run '{}': node '{node}': its dispatch in pid {pid} was written to {} and did not \
             read back as itself, so the run cannot say where that work is",
            paths.run,
            path.display()
        ))),
    } // llmlint: ignore-end[changed_behavior_has_e2e]
}

/// Write one registry entry so no reader can catch it half-written.
///
/// Renamed into the registry from a temporary **outside** it, which is the whole
/// difference from [`write_atomic`]: every file in that directory is an entry a
/// reader acts on, and a reader is now entitled to fail on one it cannot read. A
/// temporary written beside its target would be a half-written entry in the set,
/// and a `stop` racing a dispatch would refuse over this crate's own scratch.
fn write_dispatch(
    paths: &RunPaths,
    path: &Path,
    claim: u64,
    record: &DispatchRecord,
) -> Result<()> {
    let body = serde_json::to_string_pretty(record)
        .map_err(|e| Error::Invalid(format!("{}: {e}", path.display())))?;
    let ledger = |at: &Path| {
        let at = at.to_path_buf();
        move |source: io::Error| Error::Ledger { path: at, source }
    };
    fs::create_dir_all(paths.dispatches()).map_err(ledger(&paths.dispatches()))?;
    // Named from the claim as well as the process, for the reason
    // [`RunPaths::dispatch`] gives: two dispatches inside one process would
    // otherwise write one another's temporary, and a reader is entitled to fail
    // on an entry it cannot parse.
    let temp = paths
        .dir
        .join(format!("dispatch-{}-{claim}.tmp", record.pid));
    fs::write(&temp, body.as_bytes()).map_err(ledger(&temp))?;
    fs::rename(&temp, path).map_err(ledger(path))
}

/// Every dispatch this run has recorded, in pid order — or why this build cannot
/// say.
///
/// Errors are **preserved**, never flattened into an empty registry, and that is
/// this reader's whole job. "Nothing is registered" and "what is registered
/// cannot be read" are opposite answers for the caller that acts on them: the
/// first says a run has no work running, and the second says nobody knows — and a
/// stop that read the second as the first would report a run ended over work it
/// never looked for. So a registry that is not there, a directory this host will
/// not enumerate, an entry that cannot be read, one carrying a field this build
/// does not know, and one whose stamp proves nothing are all failures with the
/// path that caused them.
///
/// Ordered because a caller acts on them — a teardown signals what they name —
/// and a directory listing comes in whatever order the host gives.
pub fn dispatches_of(paths: &RunPaths) -> Result<Vec<DispatchRecord>> {
    let registry = paths.dispatches();
    let listed = fs::read_dir(&registry).map_err(|source| Error::Ledger {
        path: registry.clone(),
        source,
    })?;
    let mut found = Vec::new();
    for entry in listed {
        // llmlint: ignore-block[changed_behavior_has_e2e] an enumeration that fails *part way* is
        // the host withdrawing a directory it has already begun to list — a condition no
        // portable journey can set, and one this reader answers exactly as it answers the
        // directory it could not open at all, which
        // `stopping_a_run_whose_registry_cannot_be_read_refuses_and_leaves_the_run_retryable`
        // drives end to end for both the missing registry and the entry it cannot read.
        let entry = entry.map_err(|source| Error::Ledger {
            path: registry.clone(),
            source,
        })?; // llmlint: ignore-end[changed_behavior_has_e2e]
        let held: DispatchRecord = read_json(&entry.path())?;
        if !held.is_usable() {
            return Err(Error::Invalid(format!(
                "{}: the dispatch it records carries no start token, so nothing says pid {} is \
                 still this run's work",
                entry.path().display(),
                held.pid
            )));
        }
        found.push(held);
    }
    found.sort_by_key(|held| held.pid);
    Ok(found)
}

#[cfg(test)]
mod tests {
    use super::{body_of_entry, entry_named, identity_of_body, number_of_entry, Handover};
    use std::time::Duration;

    /// A run directory of this test's own, emptied first.
    fn gate_scratch(name: &str) -> RunPaths {
        let dir = std::env::temp_dir().join(format!("onepipeline-gate-{name}-{}", sys::pid()));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(dir.join("channel")).expect("a run directory");
        RunPaths {
            run: name.to_owned(),
            dir,
        }
    }

    /// Every entry the gate is holding, by number.
    fn entries_in(paths: &RunPaths) -> Vec<u64> {
        let mut numbers: Vec<u64> = std::fs::read_dir(paths.channel("handover"))
            .expect("the gate's entries")
            .filter_map(|entry| number_of_entry(std::path::Path::new(&entry.ok()?.file_name())))
            .collect();
        numbers.sort_unstable();
        numbers
    }

    /// **Two parties that looked at the same gate do not both enter it.**
    ///
    /// This is the interleaving no ordering by name survives and no sequential
    /// test reaches: both parties look while the gate is empty, and only then
    /// does either write. A name carrying its maker's identity beside the number
    /// gives each of them a path of its own — so both creates succeed, both are
    /// numbered 1, and whichever sorts lower reads itself in while the other is
    /// already inside. The number alone is the name here, so the two meet on one
    /// path and the filesystem says which of them holds it.
    ///
    /// The identities are the test's rather than this process's, because one
    /// process cannot be two — and it is exactly two *identities* picking one
    /// number that the broken shape let through.
    #[test]
    fn two_parties_that_looked_at_the_same_gate_do_not_both_enter_it() {
        let paths = gate_scratch("looked-together");
        let dir = paths.channel("handover");
        std::fs::create_dir_all(&dir).expect("the gate's entries");

        // Both look, and both see a gate nobody is in. Neither has written yet:
        // this is the moment the second party is paused in.
        let the_first_looked = super::highest_entry_in(&dir).expect("the gate reads");
        let the_second_looked = super::highest_entry_in(&dir).expect("the gate reads");
        assert_eq!((the_first_looked, the_second_looked), (None, None));

        let first = Handover::take_above(&paths, &dir, 0, 200, "this-host")
            .expect("the first party's write is answered")
            .expect("the first party takes the gate");
        let second = Handover::take_above(&paths, &dir, 0, 100, "this-host")
            .expect("the second party's write is answered");

        assert!(
            second.is_none(),
            "two parties took the same number, so both are inside the gate: {:?}",
            entries_in(&paths)
        );
        assert_eq!(
            entries_in(&paths),
            vec![1],
            "the gate holds an entry no party is accountable for"
        );
        // And the party that lost the race finds a holder rather than a way in.
        Handover::hold_within(&paths, Duration::from_millis(50))
            .expect_err("the gate another party is inside is not taken");

        drop(first);
        std::fs::remove_dir_all(&paths.dir).ok();
    }

    /// **A party arriving while another is inside is numbered above it**, and a
    /// holder nothing can show has gone is never stepped over.
    #[test]
    fn a_party_arriving_while_another_is_inside_does_not_get_in() {
        let paths = gate_scratch("late-arrival");
        let inside = Handover::hold(&paths).expect("this thread is inside the gate");
        assert_eq!(entries_in(&paths), vec![1]);

        Handover::hold_within(&paths, Duration::from_millis(50))
            .expect_err("a gate another party is inside is not taken");
        assert_eq!(
            entries_in(&paths),
            vec![1],
            "an arrival numbered itself over a holder that is alive"
        );

        // And once the holder lets go, the next party is let in — above the
        // entry it left, never in place of it.
        drop(inside);
        let next = Handover::hold(&paths).expect("the gate is free once its holder lets go");
        drop(next);
        std::fs::remove_dir_all(&paths.dir).ok();
    }

    /// A holder this host can prove is gone is **stepped over**, not removed.
    ///
    /// Removing it is what would let one party delete a file another has just
    /// taken; the number above it is free, so nothing needs to be removed. The
    /// entry a process that died inside the section leaves behind costs a name
    /// and blocks nobody.
    #[test]
    fn a_holder_this_host_knows_is_gone_is_stepped_over_rather_than_removed() {
        let paths = gate_scratch("holder-gone");
        let dir = paths.channel("handover");
        std::fs::create_dir_all(&dir).expect("the gate's entries");
        std::fs::write(dir.join(entry_named(1)), body_of_entry(0, &sys::hostname()))
            .expect("the entry a holder that died left behind");

        let held = Handover::hold_within(&paths, Duration::from_millis(50))
            .expect("a gate whose holder is gone is taken");
        assert_eq!(
            entries_in(&paths),
            vec![1, 2],
            "the entry of the holder that is gone was removed rather than stepped over"
        );
        // And it is a gate like any other once taken.
        Handover::hold_within(&paths, Duration::from_millis(50))
            .expect_err("the gate is held, so a second party is refused");
        drop(held);
        std::fs::remove_dir_all(&paths.dir).ok();
    }

    /// A holder this host **cannot** show has gone is waited on, whatever it is.
    ///
    /// Another host's pid, a body this build cannot read, and a name outside the
    /// shape it writes are each a claim it cannot rule out — and the answer to a
    /// claim it cannot rule out is the one it gives a live holder.
    #[test]
    fn a_holder_this_host_cannot_account_for_is_waited_on_rather_than_stepped_over() {
        for (what, name, body) in [
            (
                "another host's",
                entry_named(1),
                body_of_entry(0, "somewhere-else"),
            ),
            (
                "an unreadable body",
                entry_named(1),
                "not an identity".to_owned(),
            ),
            (
                "a name from outside this build",
                "handover.lock".to_owned(),
                String::new(),
            ),
        ] {
            let paths = gate_scratch("holder-unknown");
            let dir = paths.channel("handover");
            std::fs::create_dir_all(&dir).expect("the gate's entries");
            std::fs::write(dir.join(&name), &body).expect("the entry is written");

            Handover::hold_within(&paths, Duration::from_millis(50))
                .err()
                .unwrap_or_else(|| {
                    panic!("a gate held by {what} was taken, so two parties are inside it")
                });
            std::fs::remove_dir_all(&paths.dir).ok();
        }
    }

    /// The entry's name and body are written by one pair of functions and read by
    /// another, so the two are held to each other here rather than by a reader
    /// noticing.
    #[test]
    fn a_gate_entrys_name_carries_its_number_and_its_body_the_holder() {
        assert_eq!(
            number_of_entry(std::path::Path::new(&entry_named(1))),
            Some(1)
        );
        assert_eq!(
            identity_of_body(&body_of_entry(4242, "a-host")),
            Some((4242, "a-host"))
        );
        // Numbers compare as text in the order they are handed out, which is what
        // makes the highest entry the holder.
        assert!(entry_named(2) > entry_named(1));
        assert!(entry_named(10) > entry_named(2));
        assert!(entry_named(u64::MAX) > entry_named(1_000_000));

        // A name or a body this build did not write is read as neither.
        for stranger in [
            "handover.lock",
            "1",
            "0000000000000000000x",
            "-0000000000000000001",
        ] {
            assert_eq!(
                number_of_entry(std::path::Path::new(stranger)),
                None,
                "'{stranger}' was read as a name this build wrote"
            );
        }
        for stranger in ["", "notanumber a-host", "4242"] {
            assert_eq!(
                identity_of_body(stranger),
                None,
                "'{stranger}' was read as an identity this build wrote"
            );
        }
    }

    use super::*;

    fn scratch(name: &str) -> PathBuf {
        let dir = std::env::temp_dir().join(format!("onepipeline-ledger-{name}-{}", sys::pid()));
        let _ = fs::remove_dir_all(&dir);
        fs::create_dir_all(&dir).expect("a scratch root");
        dir
    }

    /// The record a launch that declared no filters writes is the record every
    /// build before the block existed wrote.
    ///
    /// Checked at the wire rather than through the types: `Filters::default()`
    /// and an explicit `"filters": {}` are the same value in Rust whatever the
    /// serializer does, but writing the empty block out would make a record this
    /// build wrote unreadable to a build that predates the field — which is the
    /// one thing an added field must not do — and would have every reader
    /// branching on a key that is always present and usually meaningless.
    #[test]
    fn a_launch_that_declared_no_filters_writes_the_record_it_always_wrote() {
        let record = LaunchRecord {
            filters: Filters::default(),
            ..a_record()
        };
        let text = serde_json::to_string(&record).expect("it serialises");
        assert!(
            !text.contains("filters"),
            "an empty filters block reached the record: {text}"
        );
        assert_eq!(
            serde_json::from_str::<LaunchRecord>(&text).expect("it re-parses"),
            record
        );

        // Which is the same thing as saying a record written *before* the field
        // existed still reads, as the launch it was: nothing filtered, and the
        // shipped profiles to read through. Spelled as its own document rather
        // than inferred from the omission above, because that is the file on
        // disk this build has to keep opening.
        let older = serde_json::json!({
            "run_id": "demo",
            "plan": "plan.json",
            "node_graph": "graphs/node-scope.yaml",
            "launcher": "claude-code",
            "session": "a-session",
            "pid": 1,
            "host": "h",
            "started_at": "2026-08-15T00:00:00.000Z",
            "heartbeat_interval": 1800,
        });
        let read: LaunchRecord =
            serde_json::from_value(older).expect("a record predating the field still reads");
        assert!(read.filters.is_empty());
    }

    /// A record that *did* declare filters carries every one of them back.
    #[test]
    fn a_launchs_filters_survive_the_record_they_are_retained_in() {
        let declared = Filters {
            agentgraph: Some(
                crate::filter::EventFilter::parse(r#"{"exclude": [{"kind": "turn-*"}]}"#)
                    .expect("a filter"),
            ),
            vcs: Some(
                crate::filter::EventFilter::parse(r#"{"include": [{"kind": "gate-*"}]}"#)
                    .expect("a filter"),
            ),
            profiles: [(
                "planner".to_string(),
                crate::filter::EventFilter::parse(r#"{"include": [{"source": "pipeline"}]}"#)
                    .expect("a filter"),
            )]
            .into_iter()
            .collect(),
        };
        let record = LaunchRecord {
            filters: declared.clone(),
            ..a_record()
        };
        let text = serde_json::to_string(&record).expect("it serialises");
        let read: LaunchRecord = serde_json::from_str(&text).expect("it re-parses");
        assert_eq!(read.filters, declared);
        assert_eq!(read, record);
    }

    /// A launch record with nothing said about its events.
    fn a_record() -> LaunchRecord {
        LaunchRecord {
            run_id: "demo".into(),
            project: "plans:demo".into(),
            dir: PathBuf::from("/tmp/launch"),
            graph: String::new(),
            graph_run: String::new(),
            observer_runs: Vec::new(),
            observer_ending: String::new(),
            node_graph: "graphs/node-scope.yaml".into(),
            pr_author_graph: String::new(),
            node_validator: String::new(),
            envelope_reviewer: String::new(),
            launcher: "claude-code".into(),
            session: "a-session".into(),
            pid: 1,
            host: "h".into(),
            started: "Fri Aug 15 00:00:00 2026".into(),
            started_at: sys::now_rfc3339(),
            heartbeat_interval: 1_800,
            writeback_item_budget: 10,
            success_hook: "./scripts/follow-up.sh".into(),
            failure_hook: "./scripts/report-failure.sh".into(),
            hook_timeout: 45,
            dispatch_env_hook: "./scripts/dispatch-env.sh".into(),
            dispatch_env_hook_timeout: 20,
            dag_sets: Vec::new(),
            node_sets: Vec::new(),
            adoptions: 0,
            filters: Filters::default(),
            bus_config: Default::default(),
            envelope_reviewer_bar: Default::default(),
        }
    }

    /// A driver is claimed as three facts at once, and a record that predates
    /// the stamp is read as proving nothing rather than as proving its pid.
    ///
    /// The pid and the stamp are one claim: a `stop` reading a pid with nothing
    /// beside it cannot tell the driver the record was written for from whatever
    /// the host has since handed that pid to, so the writer that records one
    /// records both. The compatibility half is the same promise every field
    /// added to this record makes — an older document still reads — and the
    /// answer it must give is the *empty* stamp, which never matches.
    #[test]
    fn claiming_a_run_records_the_stamp_that_proves_its_pid_and_an_older_record_carries_none() {
        let mut record = a_record();
        record.driven_by_this_process();
        assert_eq!(record.pid, sys::pid());
        assert_eq!(record.host, sys::hostname());
        assert!(
            sys::process_start_token(sys::pid())
                .expect("this host says when a process started")
                .matches(&record.started),
            "a run claimed by this process recorded a stamp that does not prove it"
        );
        let text = serde_json::to_string(&record).expect("it serialises");
        assert_eq!(
            serde_json::from_str::<LaunchRecord>(&text).expect("it re-parses"),
            record
        );

        // A record written before the field existed, which is the file on disk
        // this build has to keep opening.
        let older = serde_json::json!({
            "run_id": "demo",
            "plan": "plan.json",
            "node_graph": "graphs/node-scope.yaml",
            "launcher": "claude-code",
            "session": "a-session",
            "pid": sys::pid(),
            "host": sys::hostname(),
            "started_at": "2026-08-15T00:00:00.000Z",
            "heartbeat_interval": 1800,
        });
        let read: LaunchRecord =
            serde_json::from_value(older).expect("a record predating the stamp still reads");
        assert!(read.started.is_empty());
        assert!(
            !sys::process_start_token(sys::pid())
                .expect("this host says when a process started")
                .matches(&read.started),
            "a record carrying no stamp proved a live pid"
        );

        // And a record that carries none writes none, so a build that predates
        // the field still reads what this one wrote.
        let text = serde_json::to_string(&LaunchRecord {
            started: String::new(),
            ..a_record()
        })
        .expect("it serialises");
        assert!(
            !text.contains("started\""),
            "an empty stamp reached the record: {text}"
        );
    }

    /// The five keys a record on this host was written before, each absent on its
    /// own and then all five at once — read off disk, as the reader meets them.
    ///
    /// The same incident [`launcher`](LaunchRecord::launcher) records, and the
    /// rest of it: 141 of 443 run roots on one host hold a record written before
    /// one or more of these keys existed, and every one of them was refused by
    /// name — so a third of that host's history was invisible to the view an
    /// operator opens to see what is running. One field of six was repaired
    /// then; these are the other five.
    ///
    /// What each absence **means** is asserted beside the read, because a lenient
    /// reader that fabricates a value is the failure this must not become: the
    /// default stands for *the record does not say*, and every reading of it says
    /// so. Each assertion here fails if the value were invented — a session that
    /// named somebody, a pid a reader would probe, a host a reader would claim,
    /// an instant nobody measured, a check-in that fires every zero seconds, a
    /// write-back budget of zero seconds per item, a run-end or dispatch-env
    /// hook nobody named, or a hook timeout of zero.
    #[test]
    fn a_launch_record_written_before_any_of_these_eleven_keys_still_reads() {
        /// Every key added to this record after it shipped whose absence this
        /// reader has to answer for.
        const HISTORICAL: [&str; 11] = [
            "session",
            "pid",
            "host",
            "started_at",
            "heartbeat_interval",
            "writeback_item_budget",
            "success_hook",
            "failure_hook",
            "hook_timeout",
            "dispatch_env_hook",
            "dispatch_env_hook_timeout",
        ];
        let root = scratch("historical-launch");
        let whole = serde_json::to_value(a_record()).expect("a record this build writes");

        // Each shape written where a run root keeps one, and read back the way
        // every view reads it: off the file, through the ledger's own reader.
        let read_one = |name: &str, without: &[&str]| -> LaunchRecord {
            let mut document = whole.clone();
            let fields = document.as_object_mut().expect("a launch record");
            for key in without {
                assert!(
                    fields.remove(*key).is_some(),
                    "the record this build writes carries no `{key}` to take away"
                );
            }
            let dir = root.join(name);
            fs::create_dir_all(&dir).expect("a run root");
            let launch = dir.join("launch.json");
            fs::write(&launch, document.to_string()).expect("a record an older build wrote");
            read_json::<LaunchRecord>(&launch).unwrap_or_else(|error| {
                panic!("a record written before `{without:?}` existed was refused: {error}")
            })
        };

        for key in HISTORICAL {
            let read = read_one(key, &[key]);
            // Everything the record still says, it still says: the default
            // answers for the missing key and for nothing beside it.
            assert_eq!(read.run_id, "demo");
            match key {
                "session" => {
                    assert!(read.session.is_empty(), "a session was invented: {read:?}");
                    assert!(
                        !read.owned_by(""),
                        "a record naming no session was owned by a reader that names none either"
                    );
                    assert_eq!(read.owner_label("a-session"), "[unknown]");
                }
                "pid" => {
                    assert_eq!(read.pid, 0);
                    assert_eq!(
                        read.driver_pid(),
                        None,
                        "a pid nobody wrote was served as one a reader may act on"
                    );
                }
                "host" => {
                    assert!(read.host.is_empty());
                    assert_eq!(
                        read.recorded_host(),
                        None,
                        "a host nobody named was served as one"
                    );
                }
                "started_at" => {
                    assert_eq!(
                        read.launched_at(),
                        None,
                        "a launch instant nobody recorded was served as an instant"
                    );
                }
                "heartbeat_interval" => {
                    assert_eq!(read.heartbeat_interval, 0);
                    assert_eq!(
                        read.pacemaker_interval(),
                        None,
                        "a record naming no interval produced a check-in interval"
                    );
                    assert_ne!(
                        read.pacemaker_interval(),
                        Some(0),
                        "a zero-second check-in was served as an interval"
                    );
                }
                "writeback_item_budget" => {
                    assert_eq!(read.writeback_item_budget, 0);
                    assert_eq!(
                        read.item_budget(),
                        None,
                        "a record naming no budget produced a per-item budget"
                    );
                    assert_eq!(
                        read.item_budget().map(NonZeroU64::get),
                        None,
                        "a zero-second budget was served as one"
                    );
                }
                "success_hook" => {
                    assert_eq!(read.success_hook(), None, "a success hook was invented");
                    assert_eq!(read.failure_hook(), Some("./scripts/report-failure.sh"));
                }
                "failure_hook" => {
                    assert_eq!(read.failure_hook(), None, "a failure hook was invented");
                    assert_eq!(read.success_hook(), Some("./scripts/follow-up.sh"));
                }
                "hook_timeout" => {
                    assert_eq!(read.hook_timeout, 0);
                    assert_eq!(
                        read.hook_timeout(),
                        crate::cli::DEFAULT_HOOK_TIMEOUT_SECONDS,
                        "a record naming no timeout was not given the shipped one"
                    );
                }
                "dispatch_env_hook" => {
                    assert_eq!(
                        read.dispatch_env_hook(),
                        None,
                        "a dispatch-env hook was invented"
                    );
                    assert_eq!(read.success_hook(), Some("./scripts/follow-up.sh"));
                }
                "dispatch_env_hook_timeout" => {
                    assert_eq!(read.dispatch_env_hook_timeout, 0);
                    assert_eq!(
                        read.dispatch_env_hook_timeout(),
                        crate::cli::DEFAULT_DISPATCH_ENV_HOOK_TIMEOUT_SECONDS,
                        "a record naming no dispatch-env timeout was not given the shipped one"
                    );
                }
                other => unreachable!("{other} is not one of the eleven"),
            }
        }

        // And the record 141 roots on that host actually hold: none of the
        // eleven.
        let oldest = read_one("all-eleven", &HISTORICAL);
        assert_eq!(oldest.success_hook(), None);
        assert_eq!(oldest.failure_hook(), None);
        assert_eq!(
            oldest.hook_timeout(),
            crate::cli::DEFAULT_HOOK_TIMEOUT_SECONDS
        );
        assert_eq!(oldest.dispatch_env_hook(), None);
        assert_eq!(
            oldest.dispatch_env_hook_timeout(),
            crate::cli::DEFAULT_DISPATCH_ENV_HOOK_TIMEOUT_SECONDS
        );
        assert!(oldest.session.is_empty());
        assert_eq!(oldest.owner_label("a-session"), "[unknown]");
        assert!(!oldest.owned_by(sys::UNKNOWN_LAUNCHER));
        assert_eq!(oldest.driver_pid(), None);
        assert_eq!(oldest.recorded_host(), None);
        assert_eq!(oldest.launched_at(), None);
        assert_eq!(oldest.pacemaker_interval(), None);
        assert_eq!(oldest.item_budget(), None);
        // What it does say, it still says — the run it names most of all, since
        // a row a reader cannot key is a row that reaches nobody.
        assert_eq!(oldest.run_id, "demo");
        assert_eq!(oldest.launcher, "claude-code");
        assert_eq!(oldest.node_graph, "graphs/node-scope.yaml");

        // A record that carries the eleven reads them, so none of the defaults
        // above is standing in front of a value somebody wrote.
        let whole = read_one("whole", &[]);
        assert_eq!(whole.success_hook(), Some("./scripts/follow-up.sh"));
        assert_eq!(whole.failure_hook(), Some("./scripts/report-failure.sh"));
        assert_eq!(whole.hook_timeout(), NonZeroU64::new(45).expect("nonzero"));
        assert_eq!(whole.dispatch_env_hook(), Some("./scripts/dispatch-env.sh"));
        assert_eq!(
            whole.dispatch_env_hook_timeout(),
            NonZeroU64::new(20).expect("nonzero")
        );
        assert_eq!(whole.session, "a-session");
        assert_eq!(whole.driver_pid(), NonZeroU32::new(1));
        assert_eq!(whole.recorded_host(), Some("h"));
        assert!(whole.launched_at().is_some());
        assert_eq!(whole.pacemaker_interval(), Some(1_800));
        assert_eq!(whole.item_budget(), NonZeroU64::new(10));
    }

    /// A run's journal has several appenders at once — the launcher relaying its
    /// driver's stream, and the engine loop's own writer — so a record has to reach the
    /// file whole. Written as concurrent appenders because that is the only way
    /// the tearing shows: each opens its own descriptor, exactly as the separate
    /// processes do.
    #[test]
    fn concurrent_appenders_each_land_a_whole_line() {
        let root = scratch("append");
        let path = root.join("events.jsonl");
        const WRITERS: usize = 8;
        const EACH: usize = 60;

        std::thread::scope(|scope| {
            for writer in 0..WRITERS {
                let path = path.clone();
                scope.spawn(move || {
                    for n in 0..EACH {
                        let line = serde_json::json!({
                            "writer": writer,
                            "seq": n,
                            // Long enough that a torn write is not hidden by the
                            // kernel's small-write behaviour.
                            "payload": "x".repeat(512),
                        })
                        .to_string();
                        append_line(&path, &line).expect("the line is appended");
                    }
                });
            }
        });

        let lines = read_lines(&path);
        assert_eq!(lines.len(), WRITERS * EACH, "a record was torn or lost");
        for line in &lines {
            serde_json::from_str::<serde_json::Value>(line)
                .unwrap_or_else(|e| panic!("a torn record reached the file: {e}: {line}"));
        }
        fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn a_lock_refuses_a_second_writer_and_names_the_first() {
        let root = scratch("lock");
        let paths = RunPaths::under(&root, "demo");
        paths.create().expect("the run directory");

        let first = OwnershipLock::acquire(&paths, "start").expect("the first writer wins");
        let second = OwnershipLock::acquire(&paths, "adopt");
        match second {
            Err(Error::Locked { run, pid, verb, .. }) => {
                assert_eq!(run, "demo");
                assert_eq!(pid, sys::pid());
                assert_eq!(verb, "start");
            }
            other => panic!("a second writer was not refused: {other:?}"),
        }

        first.release();
        OwnershipLock::acquire(&paths, "adopt").expect("the lock was released");
        fs::remove_dir_all(&root).ok();
    }

    /// Every verb takes a run id, and a plan's cross-DAG reference carries one.
    /// It is joined onto the runs root, so anything that navigates reaches a
    /// ledger this process was never pointed at.
    #[test]
    fn a_run_id_names_one_directory_and_never_a_path() {
        for good in ["demo", "run-2", "a_b", "tracked-release", "R1"] {
            assert!(is_valid_run_id(good), "{good} was refused");
        }
        for bad in [
            "",
            ".",
            "..",
            "../elsewhere",
            "../../elsewhere",
            "a/b",
            "a\\b",
            "/absolute",
            "./here",
        ] {
            assert!(!is_valid_run_id(bad), "{bad:?} was accepted");
        }
    }

    #[test]
    fn a_lock_whose_holder_is_proved_gone_is_reclaimed() {
        let root = scratch("stale");
        let paths = RunPaths::under(&root, "demo");
        paths.create().expect("the run directory");

        let dead = sys::reaped_pid();

        write_json(
            &paths.lock(),
            &LockRecord {
                pid: dead,
                host: sys::hostname(),
                acquired_at: sys::now_rfc3339(),
                verb: "start".to_string(),
                started: String::new(),
            },
        )
        .expect("a stale lock");

        OwnershipLock::acquire(&paths, "start").expect("a dead holder's lock is reclaimed");
        fs::remove_dir_all(&root).ok();
    }

    fn a_dead_holders_lock(paths: &RunPaths) -> LockRecord {
        let dead = LockRecord {
            pid: sys::reaped_pid(),
            host: sys::hostname(),
            acquired_at: sys::now_rfc3339(),
            verb: "drive".to_string(),
            started: String::new(),
        };
        write_json(&paths.lock(), &dead).expect("a dead holder's lock");
        dead
    }

    fn reclaim_entries_beside(paths: &RunPaths) -> Vec<String> {
        let mut names: Vec<String> = fs::read_dir(&paths.dir)
            .expect("the run directory lists")
            .map(|entry| {
                entry
                    .expect("an entry")
                    .file_name()
                    .to_string_lossy()
                    .into_owned()
            })
            .filter(|name| name.contains(".reclaim."))
            .collect();
        names.sort();
        names
    }

    /// Several processes proving one holder dead at the same instant is the
    /// state a run's driver dying with replies queued behind it puts them in, and
    /// what came of it once was both replies taking the run over and one edit
    /// committed twice. Real contenders — threads doing the real filesystem
    /// operations, released together — against one dead lock, over enough rounds
    /// that a reclaim which merely narrows the window is caught: exactly one of
    /// them holds the lock afterwards and every other one is told who does.
    #[test]
    fn racing_reclaimers_of_one_dead_lock_leave_exactly_one_holder() {
        const CONTENDERS: usize = 16;
        const ROUNDS: usize = 25;
        let root = scratch("reclaim-race");
        let paths = RunPaths::under(&root, "demo");
        paths.create().expect("the run directory");

        for round in 0..ROUNDS {
            a_dead_holders_lock(&paths);
            let gate = std::sync::Arc::new(std::sync::Barrier::new(CONTENDERS));
            let outcomes: Vec<Result<OwnershipLock>> = (0..CONTENDERS)
                .map(|_| {
                    let gate = gate.clone();
                    let paths = paths.clone();
                    std::thread::spawn(move || {
                        gate.wait();
                        OwnershipLock::acquire(&paths, "reply")
                    })
                })
                .collect::<Vec<_>>()
                .into_iter()
                .map(|contender| contender.join().expect("a contender finishes"))
                .collect();

            let winners = outcomes.iter().filter(|outcome| outcome.is_ok()).count();
            assert_eq!(
                winners, 1,
                "round {round}: {winners} contenders each believe they hold the run"
            );
            for lost in outcomes.iter().filter_map(|outcome| outcome.as_ref().err()) {
                match lost {
                    Error::Locked {
                        pid, host, verb, ..
                    } => {
                        assert_eq!(*pid, sys::pid(), "round {round}: {lost}");
                        assert_eq!(*host, sys::hostname(), "round {round}: {lost}");
                        assert_eq!(verb, "reply", "round {round}: {lost}");
                    }
                    other => {
                        panic!("round {round}: a loser was not told who holds the run: {other}")
                    }
                }
            }
            let held: LockRecord = read_json(&paths.lock()).expect("the lock reads back");
            assert_eq!(
                held.pid,
                sys::pid(),
                "round {round}: the lock does not name the winner"
            );
            assert_eq!(
                reclaim_entries_beside(&paths),
                Vec::<String>::new(),
                "round {round}: the reclaim left its entries behind"
            );
            // Drop the winner's lock, which releases it, so the next round
            // starts from a dead holder's lock and not a live one.
            drop(outcomes);
        }
        fs::remove_dir_all(&root).ok();
    }

    /// A lock's name is never visible before the record it carries is whole.
    ///
    /// A reader that finds the name and cannot read a record in it has nobody to
    /// report, and a loser of the lock that reports nobody names no process to look
    /// at. So a reader watching the name while it is taken and let go, over and over,
    /// reads either no lock or a complete record — never an empty or a partial one.
    #[test]
    fn a_lock_is_never_observable_before_its_record_is_complete() {
        const CLAIMS: usize = 500;
        let root = scratch("lock-observed");
        let paths = RunPaths::under(&root, "demo");
        paths.create().expect("the run directory");
        let taking = std::sync::atomic::AtomicBool::new(true);

        let incomplete = std::thread::scope(|scope| {
            let observer = scope.spawn(|| {
                let mut incomplete = Vec::new();
                while taking.load(std::sync::atomic::Ordering::Relaxed) {
                    if let Ok(text) = fs::read_to_string(paths.lock()) {
                        if serde_json::from_str::<LockRecord>(&text).is_err() {
                            incomplete.push(text);
                        }
                    }
                }
                incomplete
            });
            for _ in 0..CLAIMS {
                OwnershipLock::acquire(&paths, "drive")
                    .expect("a lock nobody holds is taken")
                    .release();
            }
            taking.store(false, std::sync::atomic::Ordering::Relaxed);
            observer.join().expect("the observer finishes")
        });

        assert!(
            incomplete.is_empty(),
            "a reader found the lock's name before its record was complete, {} times: {:?}",
            incomplete.len(),
            incomplete.iter().take(3).collect::<Vec<_>>()
        );
        fs::remove_dir_all(&root).ok();
    }

    /// A reclaimer that died between creating its entry and writing the lock
    /// costs the next reclaimer a name, not the run.
    #[test]
    fn a_dead_reclaimers_entry_is_stepped_over_and_taken_away() {
        let root = scratch("reclaim-stale");
        let paths = RunPaths::under(&root, "demo");
        paths.create().expect("the run directory");
        let dead = a_dead_holders_lock(&paths);

        let key = reclaim_key(&dead);
        let abandoned = reclaim_entry(&paths.lock(), &key, 1);
        write_json(
            &abandoned,
            &LockRecord {
                pid: sys::reaped_pid(),
                host: sys::hostname(),
                acquired_at: sys::now_rfc3339(),
                verb: "reply".to_string(),
                started: String::new(),
            },
        )
        .expect("an entry a reclaimer died holding");

        let held = OwnershipLock::acquire(&paths, "adopt")
            .expect("a dead reclaimer's entry does not keep the run from being reclaimed");
        let record: LockRecord = read_json(&paths.lock()).expect("the lock reads back");
        assert_eq!(record.pid, sys::pid());
        assert_eq!(record.verb, "adopt");
        assert!(
            !abandoned.exists(),
            "the dead reclaimer's entry was left behind"
        );
        assert_eq!(reclaim_entries_beside(&paths), Vec::<String>::new());
        held.release();
        fs::remove_dir_all(&root).ok();
    }

    /// A reclaimer that is between creating its entry and writing the lock is
    /// the run's holder to anyone arriving then, named with what it is doing —
    /// which is how a second reply learns to wait for the first's answer rather
    /// than reporting the edit queued behind a driver that is gone.
    #[test]
    fn a_live_reclaimers_entry_names_it_as_the_holder() {
        let root = scratch("reclaim-live");
        let paths = RunPaths::under(&root, "demo");
        paths.create().expect("the run directory");
        let dead = a_dead_holders_lock(&paths);

        let key = reclaim_key(&dead);
        let taking_over = reclaim_entry(&paths.lock(), &key, 1);
        write_json(
            &taking_over,
            &LockRecord {
                pid: sys::pid(),
                host: sys::hostname(),
                acquired_at: sys::now_rfc3339(),
                verb: "reply".to_string(),
                started: String::new(),
            },
        )
        .expect("an entry a live reclaimer holds");

        match OwnershipLock::acquire(&paths, "adopt") {
            Err(Error::Locked { run, pid, verb, .. }) => {
                assert_eq!(run, "demo");
                assert_eq!(pid, sys::pid());
                assert_eq!(verb, "reply");
            }
            other => panic!("a run being taken over was not reported as held: {other:?}"),
        }
        let untouched: LockRecord = read_json(&paths.lock()).expect("the lock reads back");
        assert_eq!(untouched, dead, "the loser wrote the lock");
        assert!(
            taking_over.exists(),
            "the loser took away an entry it did not create"
        );
        fs::remove_dir_all(&root).ok();
    }

    /// An entry this build cannot read is a claim on the run, for the reason an
    /// unreadable lock is.
    #[test]
    fn an_unreadable_reclaim_entry_is_still_a_claim() {
        let root = scratch("reclaim-unreadable");
        let paths = RunPaths::under(&root, "demo");
        paths.create().expect("the run directory");
        let dead = a_dead_holders_lock(&paths);
        let entry = reclaim_entry(&paths.lock(), &reclaim_key(&dead), 1);
        fs::write(&entry, "not json at all").expect("a corrupt entry");

        match OwnershipLock::acquire(&paths, "adopt") {
            Err(unreadable) if is_unreadable_lock(&unreadable) => {
                assert!(
                    matches!(&unreadable, Error::Ledger { path, .. } if *path == entry),
                    "the refusal did not name the entry nobody can read: {unreadable}"
                );
            }
            other => panic!("an unreadable entry was not reported as one: {other:?}"),
        }
        let untouched: LockRecord = read_json(&paths.lock()).expect("the lock reads back");
        assert_eq!(untouched, dead);
        fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn an_unreadable_lock_is_still_a_claim() {
        let root = scratch("unreadable");
        let paths = RunPaths::under(&root, "demo");
        paths.create().expect("the run directory");
        fs::write(paths.lock(), "not json at all").expect("a corrupt lock");

        match OwnershipLock::acquire(&paths, "start") {
            Err(unreadable) if is_unreadable_lock(&unreadable) => {
                let said = unreadable.to_string();
                assert!(said.contains("an unreadable lock"), "{said}");
                assert!(said.contains("'demo'"), "{said}");
                assert!(
                    !said.contains("pid 0"),
                    "a holder nobody wrote was named: {said}"
                );
                assert!(
                    matches!(&unreadable, Error::Ledger { path, .. } if *path == paths.lock()),
                    "the refusal did not name the lock nobody can read: {said}"
                );
            }
            other => panic!("an unreadable lock was not reported as one: {other:?}"),
        }
        assert_eq!(
            fs::read_to_string(paths.lock()).expect("the lock is still there"),
            "not json at all",
            "a claim nobody can read was overwritten"
        );
        fs::remove_dir_all(&root).ok();
    }

    /// A lock the filesystem refuses to open for a moment is not a lock nobody
    /// can read: once it opens, the holder it names is the holder reported.
    ///
    /// The moment is a mode taken away and given back, which is what this host can
    /// arrange; what it stands for is a name another process is part way through
    /// replacing or taking away, which Windows refuses to open while it does.
    #[cfg(unix)]
    #[test]
    fn a_lock_the_filesystem_refuses_for_a_moment_names_its_holder_once_it_opens() {
        use std::os::unix::fs::PermissionsExt;
        let root = scratch("lock-refused-briefly");
        let paths = RunPaths::under(&root, "demo");
        paths.create().expect("the run directory");
        let holder = LockRecord {
            pid: sys::pid(),
            host: sys::hostname(),
            acquired_at: sys::now_rfc3339(),
            verb: "drive".to_string(),
            started: String::new(),
        };
        write_json(&paths.lock(), &holder).expect("a live holder's lock");
        fs::set_permissions(paths.lock(), fs::Permissions::from_mode(0o000))
            .expect("the lock's mode is taken away");
        assert!(
            fs::read(paths.lock()).is_err(),
            "this process reads a file whose mode refuses it, so the refusal this is about \
             cannot be arranged here"
        );

        let refused = std::thread::scope(|scope| {
            scope.spawn(|| {
                std::thread::sleep(std::time::Duration::from_millis(100));
                fs::set_permissions(paths.lock(), fs::Permissions::from_mode(0o644))
                    .expect("the lock's mode is given back");
            });
            OwnershipLock::acquire(&paths, "reply")
        });

        match refused {
            Err(Error::Locked {
                pid, host, verb, ..
            }) => {
                assert_eq!(pid, holder.pid);
                assert_eq!(host, holder.host);
                assert_eq!(verb, "drive");
            }
            other => {
                panic!("a lock refused for a moment was not reported by its holder: {other:?}")
            }
        }
        fs::remove_dir_all(&root).ok();
    }

    /// A lock the filesystem goes on refusing is reported as that refusal: not
    /// as a holder, and not as a record nobody can read, because nothing was read.
    #[test]
    fn a_lock_the_filesystem_goes_on_refusing_is_reported_as_the_refusal() {
        let root = scratch("lock-refused");
        let paths = RunPaths::under(&root, "demo");
        paths.create().expect("the run directory");
        fs::create_dir(paths.lock()).expect("something that is not a file holds the name");

        match OwnershipLock::acquire(&paths, "start") {
            Err(refused @ Error::Ledger { .. }) if !is_unreadable_lock(&refused) => {
                assert!(
                    matches!(&refused, Error::Ledger { path, .. } if *path == paths.lock()),
                    "the refusal did not name the lock: {refused}"
                );
            }
            other => panic!("a lock the filesystem refuses was not reported as that: {other:?}"),
        }
        assert!(paths.lock().is_dir(), "what held the name was replaced");
        fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn an_unknown_launch_is_nobodys_run() {
        let record = LaunchRecord {
            run_id: "demo".into(),
            project: "plans:demo".into(),
            dir: PathBuf::from("/tmp/launch"),
            graph: "graphs/dag-scope.yaml".into(),
            graph_run: String::new(),
            observer_runs: Vec::new(),
            observer_ending: String::new(),
            node_graph: String::new(),
            pr_author_graph: String::new(),
            node_validator: String::new(),
            envelope_reviewer: String::new(),
            launcher: sys::UNKNOWN_LAUNCHER.into(),
            session: sys::UNKNOWN_LAUNCHER.into(),
            pid: 1,
            host: "h".into(),
            started: String::new(),
            started_at: sys::now_rfc3339(),
            heartbeat_interval: 1,
            writeback_item_budget: 0,
            success_hook: String::new(),
            failure_hook: String::new(),
            hook_timeout: 0,
            dispatch_env_hook: String::new(),
            dispatch_env_hook_timeout: 0,
            dag_sets: Vec::new(),
            node_sets: Vec::new(),
            adoptions: 0,
            filters: Filters::default(),
            bus_config: Default::default(),
            envelope_reviewer_bar: Default::default(),
        };
        assert!(!record.owned_by(sys::UNKNOWN_LAUNCHER));
        assert_eq!(record.owner_label("anyone"), "[unknown]");
    }

    #[test]
    fn a_foreign_owner_is_labelled_without_naming_the_session() {
        let record = LaunchRecord {
            run_id: "demo".into(),
            project: "plans:demo".into(),
            dir: PathBuf::from("/tmp/launch"),
            graph: "graphs/dag-scope.yaml".into(),
            graph_run: String::new(),
            observer_runs: Vec::new(),
            observer_ending: String::new(),
            node_graph: String::new(),
            pr_author_graph: String::new(),
            node_validator: String::new(),
            envelope_reviewer: String::new(),
            launcher: "claude-code".into(),
            session: "secret-session-id".into(),
            pid: 1,
            host: "h".into(),
            started: String::new(),
            started_at: sys::now_rfc3339(),
            heartbeat_interval: 1,
            writeback_item_budget: 0,
            success_hook: String::new(),
            failure_hook: String::new(),
            hook_timeout: 0,
            dispatch_env_hook: String::new(),
            dispatch_env_hook_timeout: 0,
            dag_sets: Vec::new(),
            node_sets: Vec::new(),
            adoptions: 0,
            filters: Filters::default(),
            bus_config: Default::default(),
            envelope_reviewer_bar: Default::default(),
        };
        let label = record.owner_label("mine");
        assert!(!label.contains("secret-session-id"), "{label} leaks the id");
        assert!(label.starts_with("[claude-code:"));
        assert_eq!(record.owner_label("secret-session-id"), "[mine]");
        assert!(record.owned_by("secret-session-id"));
    }

    #[test]
    fn an_atomic_write_leaves_no_temporary_behind() {
        let root = scratch("atomic");
        let target = root.join("nested").join("record.json");
        write_json(&target, &serde_json::json!({"ok": true})).expect("written");
        let value: serde_json::Value = read_json(&target).expect("read back");
        assert_eq!(value["ok"], serde_json::json!(true));
        let leftovers: Vec<_> = fs::read_dir(root.join("nested"))
            .expect("the directory")
            .flatten()
            .filter(|e| e.file_name().to_string_lossy().contains("tmp"))
            .collect();
        assert!(leftovers.is_empty(), "a temporary survived the rename");
        fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn appended_lines_read_back_in_order_and_skip_blanks() {
        // A scratch name of its own: `scratch` derives the directory from the
        // process, so two tests naming one share it — and these two run at once,
        // which made the concurrent-appender count above fail on this test's
        // writes rather than on a torn record.
        let root = scratch("append-order");
        let path = root.join("queue.jsonl");
        assert!(read_lines(&path).is_empty());
        append_line(&path, "first").expect("appended");
        append_line(&path, "").expect("appended");
        append_line(&path, "second").expect("appended");
        assert_eq!(read_lines(&path), vec!["first", "second"]);
        fs::remove_dir_all(&root).ok();
    }

    /// A record's own position and terminator survive the read.
    ///
    /// `str::lines` discards the one signal a store holds that a writer
    /// finished: a final line with no `\n` reads exactly like a terminated one,
    /// which is how a torn record used to reach a reader as an ordinary one.
    #[test]
    fn a_record_carries_where_it_is_and_whether_its_writer_finished_it() {
        let root = scratch("records");
        let path = root.join("events.jsonl");
        fs::write(&path, "first\n\nsecond\nhalf").expect("the file is written");

        let records = read_records(&path);
        assert_eq!(records.len(), 4);
        assert_eq!(records[0].offset, 0);
        assert_eq!(records[2].text, "second");
        assert_eq!(records[2].offset, 7);
        assert!(records[2].terminated);
        assert_eq!(records[3].text, "half");
        assert_eq!(records[3].offset, 14);
        assert!(
            !records[3].terminated,
            "a fragment read as a record its writer had finished"
        );
        // The blank line is still skipped by every reader that only wants the
        // records, which is what `read_lines` has always answered.
        assert_eq!(read_lines(&path), vec!["first", "second", "half"]);
        fs::remove_dir_all(&root).ok();
    }

    /// A file torn mid-character is read for the records it holds.
    ///
    /// A record cut in the middle of a multi-byte character is not UTF-8, and
    /// decoding the file whole failed on it — so *every* reader here, the
    /// channel's queue and its replies as much as the journal, handed back an
    /// empty file over one bad byte at the end of it. Per line, the tear costs
    /// itself and nothing before it, and the offsets stay the file's own.
    #[test]
    fn a_file_torn_mid_character_still_reads_back_the_records_before_the_tear() {
        let root = scratch("lossy");
        let path = root.join("queue.jsonl");
        fs::create_dir_all(&root).ok();
        let mut bytes = b"{\"first\":1}\n{\"second\":2}\n{\"third\":".to_vec();
        // The first byte of a three-byte character, and nothing after it.
        bytes.push(0xE2);
        fs::write(&path, &bytes).expect("the file is written");

        let records = read_records(&path);
        assert_eq!(
            records.len(),
            3,
            "the tear took the file with it: {records:?}"
        );
        assert_eq!(records[0].text, "{\"first\":1}");
        assert_eq!(records[1].text, "{\"second\":2}");
        assert_eq!(records[2].offset, 25);
        assert_eq!(
            records[2].bytes, 10,
            "the loss was measured in the reading rather than in the file"
        );
        assert!(!records[2].terminated);
        assert_eq!(read_lines(&path).len(), 3);
        fs::remove_dir_all(&root).ok();
    }

    /// A fragment a dead writer left is cut back to the last boundary, and what
    /// it cost is recorded beside the store rather than quietly repaired.
    #[test]
    fn an_append_heals_a_fragment_and_records_what_it_discarded() {
        let root = scratch("heal");
        let path = root.join("events.jsonl");
        append_line(&path, "first").expect("appended");
        let whole = fs::read_to_string(&path).expect("the file reads");
        fs::OpenOptions::new()
            .append(true)
            .open(&path)
            .and_then(|mut file| std::io::Write::write_all(&mut file, b"{\"half\":"))
            .expect("the fragment is written");

        append_line(&path, "second").expect("appended");

        assert_eq!(
            fs::read_to_string(&path).expect("the file reads"),
            format!("{whole}second\n"),
            "the heal cut into a whole record, or left the fragment in"
        );
        let recorded = torn_tails(&path);
        assert_eq!(recorded.len(), 1, "{recorded:?}");
        assert_eq!(recorded[0].offset, whole.len() as u64);
        assert_eq!(recorded[0].bytes, 8);
        assert_eq!(recorded[0].healed_by, sys::pid());
        // The next append meets a store that ends on a boundary and records
        // nothing further: healing is not something a reader sees twice.
        append_line(&path, "third").expect("appended");
        assert_eq!(torn_tails(&path).len(), 1);
        fs::remove_dir_all(&root).ok();
    }

    /// A store whose *whole* content is one unterminated fragment.
    ///
    /// The measured shape of this loss: sixteen kilobytes with no newline in
    /// them at all. There is no boundary to cut back to but the start of the
    /// file, and the whole of what is discarded is reported.
    #[test]
    fn a_store_that_is_nothing_but_a_fragment_heals_to_empty_and_says_so() {
        let root = scratch("heal-whole");
        let path = root.join("events.jsonl");
        fs::write(&path, "x".repeat(70 * 1024)).expect("the fragment is written");

        append_line(&path, "first").expect("appended");

        assert_eq!(
            fs::read_to_string(&path).expect("the file reads"),
            "first\n"
        );
        let recorded = torn_tails(&path);
        assert_eq!(recorded.len(), 1, "{recorded:?}");
        assert_eq!(recorded[0].offset, 0);
        assert_eq!(recorded[0].bytes, 70 * 1024);
        fs::remove_dir_all(&root).ok();
    }

    /// The log of tears is named from the store it is about, and reads as
    /// nothing when there has never been one.
    #[test]
    fn a_store_that_lost_nothing_records_nothing() {
        let root = scratch("torn-absent");
        let path = root.join("events.jsonl");
        assert_eq!(
            torn_tail_log(&path),
            root.join("events.jsonl.torn"),
            "the loss log is not beside the store it is about"
        );
        assert!(torn_tails(&path).is_empty());
        append_line(&path, "first").expect("appended");
        assert!(
            torn_tails(&path).is_empty(),
            "an append that healed nothing reported a loss"
        );
        fs::remove_dir_all(&root).ok();
    }

    /// A directory that is not a run is a **rejection**, and it comes back named.
    ///
    /// The reading it replaces: the same root reported two runs and said nothing
    /// at all about the third directory, so a reader could not tell a root that
    /// holds nothing from one whose contents were refused.
    #[test]
    fn only_directories_with_a_launch_record_are_runs_and_the_rest_are_named() {
        let root = scratch("index");
        for name in ["b-run", "a-run"] {
            let paths = RunPaths::under(&root, name);
            paths.create().expect("a run directory");
            write_json(&paths.launch(), &serde_json::json!({})).expect("a launch record");
        }
        fs::create_dir_all(root.join("scratch")).expect("a directory that records no run");
        fs::write(root.join("notes.txt"), "not a run root").expect("a file beside the runs");
        // A launch record that is there and is not a record: absent and "present
        // as something else" are different things to tell a reader.
        fs::create_dir_all(RunPaths::under(&root, "impostor").launch())
            .expect("a launch record that is a directory");

        let index = all_runs(&root);
        let ids: Vec<String> = index.runs.iter().map(|r| r.run.clone()).collect();
        assert_eq!(ids, vec!["a-run".to_string(), "b-run".to_string()]);
        // Each directory is named with its own reason; the file never claimed to
        // be a run, so nothing is claimed about it either.
        let refused: Vec<(PathBuf, String)> = index
            .skipped
            .iter()
            .map(|root| (root.path.clone(), root.reason.clone()))
            .collect();
        assert_eq!(refused.len(), 2, "{refused:?}");
        assert_eq!(refused[0].0, root.join("impostor"));
        assert!(refused[0].1.contains("is not a file"), "{refused:?}");
        assert_eq!(refused[1].0, root.join("scratch"));
        assert!(refused[1].1.contains("no launch.json"), "{refused:?}");

        // A root nobody has written to holds nothing to reject.
        assert_eq!(all_runs(&root.join("missing")), RunIndex::default());
        fs::remove_dir_all(&root).ok();
    }

    /// The lock names the process *and* proves it is still that process.
    ///
    /// A pid alone is what a two-day-old lock has, and a pid the host has since
    /// reused answers a liveness probe as alive — which is how a dead run's
    /// dispatches were rendered as a live fleet.
    #[test]
    fn a_lock_records_the_holders_start_token_beside_its_pid() {
        let root = scratch("stamp");
        let paths = RunPaths::under(&root, "demo");
        paths.create().expect("the run directory");

        let held = OwnershipLock::acquire(&paths, "drive").expect("the lock is taken");
        let record: LockRecord = read_json(&paths.lock()).expect("the lock reads back");
        assert_eq!(record.pid, sys::pid());
        assert!(
            sys::process_start_token(sys::pid())
                .expect("this host says when a process started")
                .matches(&record.started),
            "the lock's stamp is not this process's own start"
        );
        assert!(!record.started.is_empty());
        held.release();
        fs::remove_dir_all(&root).ok();
    }

    /// A dispatch claims the process its work is in, and gives the claim up when
    /// the dispatch ends.
    ///
    /// The registry's whole contract in one place: while a dispatch is running
    /// the run says which process it is running in and proves that pid is still
    /// that process, and the moment the dispatch ends — however it ends, because
    /// the claim is given up by being dropped — the run stops saying so. A
    /// registry that kept the entry would send a later stop at whatever the host
    /// had handed the pid to next.
    #[test]
    fn a_dispatch_claims_the_process_it_runs_in_and_gives_it_up_when_it_ends() {
        let root = scratch("dispatches");
        let paths = RunPaths::under(&root, "demo");
        paths.create().expect("the run directory");
        assert!(
            dispatches_of(&paths)
                .expect("a run that has dispatched nothing has an empty registry")
                .is_empty(),
            "a run that has dispatched nothing claimed a process"
        );

        let claim = claim_dispatch(&paths, "build", sys::pid()).expect("the dispatch is recorded");
        let recorded = dispatches_of(&paths).expect("the registry reads");
        assert_eq!(recorded.len(), 1, "{recorded:?}");
        assert_eq!(recorded[0].node, "build");
        assert_eq!(recorded[0].pid, sys::pid());
        assert_eq!(recorded[0].host, sys::hostname());
        assert!(
            sys::process_start_token(sys::pid())
                .expect("this host says when a process started")
                .matches(&recorded[0].started),
            "the entry's stamp is not this process's own start: {recorded:?}"
        );

        drop(claim);
        assert!(
            dispatches_of(&paths)
                .expect("the registry reads")
                .is_empty(),
            "a dispatch that ended left the run claiming its process"
        );
        fs::remove_dir_all(&root).ok();
    }

    /// Two dispatches inside one process are two entries, and each ends alone.
    ///
    /// The library backend runs a node's dispatch **in the driver**, so a run at
    /// any concurrency above one has several live dispatches sharing a pid. Keyed
    /// by pid alone they were one entry: the second overwrote the first, and
    /// whichever ended first took the survivor's registration with it — leaving a
    /// live dispatch nothing could find, which is the failure this registry
    /// exists to make impossible.
    #[test]
    fn two_dispatches_in_one_process_are_two_entries_and_each_ends_alone() {
        let root = scratch("dispatches-shared-process");
        let paths = RunPaths::under(&root, "demo");
        paths.create().expect("the run directory");

        let first = claim_dispatch(&paths, "first", sys::pid()).expect("the first is recorded");
        let second = claim_dispatch(&paths, "second", sys::pid()).expect("the second is recorded");
        let nodes = |paths: &RunPaths| {
            let mut named: Vec<String> = dispatches_of(paths)
                .expect("the registry reads")
                .into_iter()
                .map(|held| held.node)
                .collect();
            named.sort();
            named
        };
        assert_eq!(
            nodes(&paths),
            vec!["first".to_string(), "second".to_string()],
            "two dispatches in one process did not record two entries"
        );

        drop(first);
        assert_eq!(
            nodes(&paths),
            vec!["second".to_string()],
            "a dispatch that ended took a live one's registration with it"
        );
        drop(second);
        assert!(dispatches_of(&paths)
            .expect("the registry reads")
            .is_empty());
        fs::remove_dir_all(&root).ok();
    }

    /// A claim removes its **own** entry and never the one that replaced it.
    ///
    /// The host reissues pids, so the entry under one is this dispatch's only
    /// while the process behind it is. A claim that removed the file by name
    /// would, on the ordering that matters — a dispatch ending just as a later
    /// one starts in a pid the host has recycled — delete a live dispatch's
    /// entry and leave that process findable by nothing.
    #[test]
    fn a_claim_that_ends_leaves_a_later_dispatchs_entry_alone() {
        let root = scratch("dispatches-reused");
        let paths = RunPaths::under(&root, "demo");
        paths.create().expect("the run directory");

        let first = claim_dispatch(&paths, "build", sys::pid()).expect("the dispatch is recorded");
        // The same pid, claimed again by what stands in here for a later process
        // wearing it: the entry is rewritten with a start this one does not have.
        write_json(
            &paths.dispatch(sys::pid(), 0),
            &DispatchRecord {
                started: "the process that took it, which is not the first one".into(),
                ..dispatches_of(&paths).expect("the registry reads")[0].clone()
            },
        )
        .expect("the entry is rewritten");

        drop(first);
        let recorded = dispatches_of(&paths).expect("the registry reads");
        assert_eq!(
            recorded.len(),
            1,
            "a dispatch that ended removed an entry it did not write: {recorded:?}"
        );
        fs::remove_dir_all(&root).ok();
    }

    /// Every entry, in pid order.
    ///
    /// Ordered because a teardown acts on them, and a directory listing comes in
    /// whatever order the host gives — so a run's stop would reach its processes
    /// in a different order each time it was asked.
    #[test]
    fn the_registry_reads_in_pid_order() {
        let root = scratch("dispatches-order");
        let paths = RunPaths::under(&root, "demo");
        paths.create().expect("the run directory");
        for (node, pid) in [("later", 900_u32), ("earlier", 90), ("middle", 300)] {
            write_json(
                &paths.dispatch(pid, 0),
                &DispatchRecord {
                    node: node.to_string(),
                    pid,
                    host: sys::hostname(),
                    dispatched_at: sys::now_rfc3339(),
                    started: "a start this host once reported".into(),
                },
            )
            .expect("an entry");
        }

        let read: Vec<u32> = dispatches_of(&paths)
            .expect("the registry reads")
            .iter()
            .map(|held| held.pid)
            .collect();
        assert_eq!(read, vec![90, 300, 900]);
        fs::remove_dir_all(&root).ok();
    }

    /// Every way a registry can fail to be read is a failure, and none of them
    /// is an empty registry.
    ///
    /// "Nothing is registered" and "what is registered cannot be read" are
    /// opposite answers for the caller that acts on them, and this reader's whole
    /// job is to keep them apart: the first says a run has no work running, the
    /// second says nobody knows. Each of these was once the same empty vector.
    ///
    /// A field this build does not know is **not** one of them, and that is the
    /// first thing asserted below: an entry from another build of this crate is a
    /// live process a stop has to reach, and refusing the whole registry over a
    /// key is how a run's work becomes unreachable.
    #[test]
    fn a_registry_this_build_cannot_read_is_reported_and_never_read_as_an_empty_one() {
        let root = scratch("dispatches-unreadable");
        let paths = RunPaths::under(&root, "demo");
        paths.create().expect("the run directory");
        let usable = DispatchRecord {
            node: "build".into(),
            pid: 4_242,
            host: sys::hostname(),
            dispatched_at: sys::now_rfc3339(),
            started: "a start this host once reported".into(),
        };

        fs::write(
            paths.dispatch(usable.pid, 0),
            serde_json::to_string(&serde_json::json!({
                "node": usable.node,
                "pid": usable.pid,
                "host": usable.host,
                "dispatched_at": usable.dispatched_at,
                "started": usable.started,
                "reaped_by": "a build that came later",
            }))
            .expect("an entry from a newer writer"),
        )
        .expect("an entry");
        assert_eq!(
            dispatches_of(&paths).expect("an entry from a newer writer reads"),
            vec![usable.clone()],
            "an entry carrying a field this build does not know took the whole registry with it"
        );

        for (what, entry) in [
            (
                "a record that is not JSON at all",
                "not an entry".to_string(),
            ),
            (
                "a record missing the stamp entirely",
                serde_json::to_string(&serde_json::json!({
                    "node": usable.node,
                    "pid": usable.pid,
                    "host": usable.host,
                    "dispatched_at": usable.dispatched_at,
                }))
                .expect("an entry from a writer that recorded no stamp"),
            ),
            (
                "a record whose stamp proves nothing",
                serde_json::to_string(&DispatchRecord {
                    started: String::new(),
                    ..usable.clone()
                })
                .expect("an unstamped entry"),
            ),
        ] {
            fs::write(paths.dispatch(usable.pid, 0), entry).expect("an entry");
            let refused =
                dispatches_of(&paths).expect_err(&format!("{what} was read as a registry"));
            assert!(
                refused.to_string().contains(&usable.pid.to_string()),
                "the refusal over {what} does not name what caused it: {refused}"
            );
        }

        // And the registry that is not there at all. Every run this build creates
        // has one, so its absence is something having taken it away — which is
        // not the same fact as a run that has dispatched nothing, and answering
        // it the same way is what this reader refuses to do.
        fs::remove_dir_all(paths.dispatches()).expect("the registry is taken away");
        let refused = dispatches_of(&paths)
            .expect_err("a registry that is not there was read as a run with nothing running");
        assert!(
            refused
                .to_string()
                .contains(&paths.dispatches().display().to_string()),
            "the refusal does not name the registry it could not read: {refused}"
        );
        fs::remove_dir_all(&root).ok();
    }

    /// A dispatch this run cannot record is a dispatch this run does not run.
    ///
    /// The claim is the trust boundary, not bookkeeping around one: an entry
    /// that was not written is a process no view will show and no stop will
    /// reach, on a run whose own records say it has nothing running. So the
    /// caller is handed the failure — both ways it can happen — and ends the
    /// dispatch with it rather than running work nothing can find.
    #[test]
    fn a_dispatch_the_registry_cannot_record_is_refused() {
        let root = scratch("dispatches-unwritable");
        let paths = RunPaths::under(&root, "demo");
        paths.create().expect("the run directory");

        // A host that will not say when the process started: nothing could prove
        // that pid is still this run's work, so there is no entry to write.
        let reaped = sys::reaped_pid();
        let refused = claim_dispatch(&paths, "build", reaped)
            .expect_err("a dispatch nothing can stamp was recorded anyway");
        assert!(
            refused.to_string().contains(&reaped.to_string()),
            "the refusal does not name the process it could not stamp: {refused}"
        );

        // And a registry that cannot be written at all, with a file where its
        // directory has to go — a host that is otherwise perfectly healthy.
        fs::remove_dir_all(paths.dispatches()).expect("the registry is taken away");
        fs::write(paths.dispatches(), "not a directory").expect("something in the way");
        let refused = claim_dispatch(&paths, "build", sys::pid())
            .expect_err("a claim that could not be written was reported as held");
        assert!(
            refused
                .to_string()
                .contains(&paths.dispatches().display().to_string()),
            "the refusal does not name what it could not write: {refused}"
        );
        fs::remove_dir_all(&root).ok();
    }

    /// A lock written before the stamp existed still reads, and still round-trips
    /// without gaining a field it never had.
    #[test]
    fn a_lock_without_a_start_token_reads_and_is_written_back_without_one() {
        let record: LockRecord = serde_json::from_str(
            r#"{"pid":1,"host":"h","acquired_at":"2026-01-01T00:00:00.000Z","verb":"drive"}"#,
        )
        .expect("a lock from a build that predates the stamp");
        assert!(record.started.is_empty());
        let written = serde_json::to_string(&record).expect("it serializes");
        assert!(!written.contains("started"), "{written}");
    }
}