onepipeline 0.38.1

//! The per-run **summary document**: what a listing reads instead of a journal.
//!
//! One `summary.json` beside each run's `plan.json` and `result.json`, holding
//! the row a listing renders and the launch record's own account of the run. It
//! exists because the only constructor for a run — [`views::RunView::open`] —
//! folds the run's *entire* merged event store into memory, and a listing builds
//! one per run root: so asking a host what is running read every byte every run
//! had ever recorded, and asking it for one row cost more than asking it for
//! fifty.
//!
//! # Who writes it, and why that one
//!
//! **The journal writer**, folding each appended record into the summary as it
//! appends it — [`Maintainer`], held by [`journal::Journal`]. Two properties
//! come from that and from nothing else: the document is current for a run that
//! is *still recording*, which is the run a listing most needs to be right
//! about; and it costs O(1) per record rather than a pass over the store.
//!
//! # What a reader does when it is not there
//!
//! Folds, once, and caches what it folded. A run recorded by a build that
//! predates this document lists exactly as it does today and only more slowly —
//! which is what makes this landing non-breaking — and a run whose summary is
//! **stale** against the journal's own length or modification time is refolded
//! rather than served. Neither answer differs from the other: the same
//! derivation runs over both paths, so the row a listing serves and the row a
//! full fold produces are one row.
//!
//! # What is deliberately not in it
//!
//! **Liveness.** How a run is being driven is read from the host at the moment
//! of the question — a stored answer is stale the instant it is written, and a
//! stored `ACTIVE` is exactly the reading that sends nobody to a run whose
//! driver died. What this document carries is what
//! [`views::liveness`](crate::views::liveness) takes as *input*: the launch
//! record's pid, host, and start token, and the run's last recorded write. The
//! answer stays computed.
//!
//! **The observer.** Not even its inputs, and for a sharper reason: a driver
//! *rewrites* them under a live run without a record following. One that finds
//! its observer gone starts another and records the new graph run, and one that
//! stops starting another records why — both in the launch record, and neither
//! is a journal append, so this document would go on naming a graph run that
//! ended and saying nothing about a run nobody will watch again. The launch
//! record is where that question is asked, at the moment it is asked.
//!
//! [`views::RunView::open`]: crate::views::RunView::open

// llmlint: ignore-file[invalid_states_unrepresentable] every identifier and timestamp on
// `RunSummary` is a `String` for the reason `src/ledger.rs`'s own file-level suppression
// states, and this document is that file's records read back: a run id, a project id, a
// launching session and an instant are *serialized* fields a consumer parses and an older
// build wrote, so every reader has to accept what is there rather than what this build
// would mint. `docs/contract.md` names no `RunId` and no timestamp type, so a newtype here
// would be a public vocabulary the contract did not ask for — and the contract that does
// exist is enforced where it can be: `schema_version` is refused by the deserializer, and
// the one value that could be a nonsense pid is `NonZeroU32` rather than a checked `u32`.
use std::collections::{BTreeMap, BTreeSet};
use std::num::NonZeroU32;
use std::path::{Path, PathBuf};

use serde::{Deserialize, Serialize};

use crate::event::Envelope;
use crate::graph;
use crate::journal;
use crate::ledger::{self, LaunchRecord, RunPaths, Skipped};
use crate::projection::{self, RunState};
use crate::telemetry::{self, RunTelemetry};

/// The schema version of the summary document.
///
/// The whole compatibility statement: a reader that met a document it does not
/// understand and served it anyway would report a run's state out of fields that
/// mean something else. A version this build does not write is **refused**, and
/// a refused document is not an error — it is a run that folds, which is the
/// answer every run had before this document existed.
///
/// The version is why this document is read **closed** —
/// `deny_unknown_fields` — while the launch record beside it is read
/// permissively. That record has no version and no fallback: refusing it takes
/// the whole run away from every view, which is the incident the ledger's own
/// module documentation records. This one has both, so a key this build does not
/// know costs a fold and nothing else, and the alternative — reading a document
/// half of whose meaning is a build's this one is not — is exactly what the
/// version exists to refuse.
///
/// **3** since a row reads whether the driver its record names let go of the run
/// to fire a run-end hook: `let_go_by` is a field version 2 never had, and a
/// version-2 document carries no answer to that question rather than "no".
///
/// **4** since the fold reads a settle's stated landing as the node's landing: a
/// [`NodeLanding`] keeps its shape, but a version-3 document written over a
/// journal holding one says what a build that did not read it made of that node —
/// work nobody landed — and serving it would contradict every view beside it.
///
/// **5** since `driver-adopted` clears the run's recorded stop in the fold
/// (`RunState::stop`): `stop_recorded` keeps its name, but a version-4 document
/// written over a journal holding a stop and then an adoption says the run is
/// stopped, and serving it would report the adopting driver dead and the run
/// settled — the reading the fold change exists to end.
pub const SUMMARY_SCHEMA_VERSION: u32 = 5;

/// Read the version, refusing a document this build cannot honestly read.
fn this_version<'de, D: serde::Deserializer<'de>>(reader: D) -> Result<u32, D::Error> {
    let found = u32::deserialize(reader)?;
    if found != SUMMARY_SCHEMA_VERSION {
        return Err(serde::de::Error::custom(format!(
            "summary schema_version {found}, and this build reads {SUMMARY_SCHEMA_VERSION}"
        )));
    }
    Ok(found)
}

/// The version a document *declares*, where it declares one this build has
/// **moved past**.
///
/// Asked only where the strict read above has already refused, so a document at
/// this build's own version is still parsed exactly once. What it answers is a
/// **decision** rather than the absence of one: the file is there, it is
/// well-formed, and it says outright that its fields mean what an earlier build
/// meant by them. That settles the one thing `unwatched` ever asks a document —
/// whether it proves the run stopped — in the negative, because nothing in it may
/// be read as this build's `stop_recorded` or `graph_complete`.
///
/// **Only backwards.** A version *ahead* of this build was written by a build
/// that knows things this one does not, and this one has no reading of it at all,
/// not even that negative one. It stays what every other reader here makes of it:
/// refused, and reported as a document that could not be read.
pub(crate) fn version_this_build_moved_past(text: &str) -> Option<u32> {
    /// The version field on its own, read **open** — every other key ignored,
    /// because the whole point is to read a document whose other keys are another
    /// build's to mean.
    #[derive(Deserialize)]
    struct Declared {
        schema_version: u32,
    }

    serde_json::from_str::<Declared>(text)
        .ok()
        .map(|declared| declared.schema_version)
        .filter(|declared| *declared < SUMMARY_SCHEMA_VERSION)
}

/// One node's change, as the **inputs** a listing decides its landing from.
///
/// Everything here is a record of what the run observed; the decision itself is
/// taken when a view renders. See [`RunSummary::landings`], which states why the
/// answer is deliberately not stored.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct NodeLanding {
    /// What the node's own settlement observed: `landed` or `unlanded`, in the
    /// run's own words.
    ///
    /// A word rather than a closed set, on the terms this file's own suppression
    /// states: it is a serialized field an older build wrote and a newer one may
    /// spell differently, and `docs/contract.md` names no type for it.
    pub landing: String,
    /// The branch the dispatch reported, which is the only spelling of this work
    /// a repository can resolve. Absent for a node that published none.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub branch: Option<String>,
    /// The repository the question is narrowed to. Absent for a node whose plan
    /// names none — and a node with no repository publishes nothing, so asking
    /// after its branch would search every identity this host knows.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub repo: Option<String>,
    /// Whether the run is holding this change back as a **draft**.
    ///
    /// Counted under neither of a listing's two headings: a draft is a change
    /// this run is deliberately holding and will lift itself, not one nobody
    /// merged, and it has a line of its own saying what it waits on.
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
    pub drafted: bool,
}

/// One run, as a listing reads it: a bounded read that does not grow with the
/// run's journal.
///
/// Everything here is a **record of what the store said**, never a reading of
/// the host: see the module's note on liveness. Every field the launch record
/// contributes carries that record's own absence policy — a value the record
/// does not state is absent here rather than invented, because a listing that
/// fabricated a launch instant, a host, or a pid would be the more expensive
/// mistake by far.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RunSummary {
    /// The document's own version, so a reader can refuse one it does not
    /// understand. See [`SUMMARY_SCHEMA_VERSION`].
    #[serde(deserialize_with = "this_version")]
    pub schema_version: u32,
    /// The run.
    pub run_id: String,
    /// The run's last recorded write, in milliseconds since the epoch.
    ///
    /// **The ordering key.** A listing orders by it, and it is stored rather
    /// than derived for exactly that reason: an order taken from the journal
    /// would drag the whole fold back in for every row on the list. Absent for a
    /// run whose store carries no record this build can date, which is not the
    /// same fact as a run last written at the epoch.
    pub last_write_at: Option<u64>,
    /// The wire string of the last record in the run's merged store, or absent
    /// for a run that has recorded none.
    ///
    /// A wire string rather than a kind of this crate's own: the merged store
    /// interleaves three producers and two of the three spell their kinds in
    /// their own vocabulary.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub last_event_kind: Option<String>,
    /// How many records the run's merged store holds.
    pub event_count: u64,
    /// Each recorded status word to the number of nodes carrying it, in the
    /// run's own words.
    ///
    /// Derived by the **same precedence [`views`](crate::views) uses** — the
    /// projection's own `statuses`, which recomputes the derived gates against
    /// the graph as it stands — so a row and the graph it opens cannot describe
    /// different graphs. A word no node carries is absent rather than present
    /// and zero.
    pub node_counts: BTreeMap<String, u64>,
    /// Whether a stop has been recorded at all, however it went — and no
    /// adoption has driven the run since, which is what the fold's
    /// `RunState::stop` answers.
    pub stop_recorded: bool,
    /// Whether every node of the graph reached a state the loop is finished
    /// with, so no further pass is coming.
    ///
    /// Every node settled, over the same statuses
    /// [`node_counts`](Self::node_counts) is counted from, and `false` for a run
    /// whose graph has no nodes at all: a run that has recorded no plan has not
    /// converged, it has not started. With [`stop_recorded`](Self::stop_recorded)
    /// this is what a settled-run filter and a timing-quality reading are decided
    /// from — telemetry over a run still moving is a partial measurement.
    pub graph_complete: bool,
    /// How many decision points are reported as holding dependents back and not
    /// yet reported as released.
    pub decisions_pending: u64,
    /// How many surfaces the run has sent.
    pub surfaces_queued: u64,
    /// How many a planner has consumed.
    pub surfaces_read: u64,
    /// Whether a ready human action is outstanding: one nobody has attested.
    ///
    /// The **graph's** half of a decision point. The other half is a blocking
    /// surface, which lives in the channel rather than the store and is not a
    /// fact this document records — see
    /// [`views::decision_outstanding`](crate::views::decision_outstanding),
    /// which is the whole question.
    pub awaiting_human_action: bool,
    /// The qualified onetaskgraph project id the run was launched with. Empty on
    /// a record written before the store was where a plan came from.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub project: String,
    /// The launcher, as the launch record states it.
    pub launcher: String,
    /// The launching session, or **empty** for an unattributed launch — which is
    /// what a record carrying no session, and one carrying a blank one, both
    /// mean, and what a view labels `[unknown]`.
    pub session: String,
    /// When the run was launched, when the record says.
    ///
    /// Absent, and never an instant standing in for one: a launch instant nobody
    /// recorded is a different fact from one recorded at the epoch, and only the
    /// second is a measurement.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub started_at: Option<String>,
    /// The driver process, when the record names one a reader may act on.
    ///
    /// Absent for the `0` a record naming none defaults to. A pid is one third
    /// of a claim — which process, on which host, and the
    /// [`started`](Self::started) stamp saying it is still that process — and a
    /// pid nobody wrote has no stamp beside it by construction, so nothing that
    /// acts on a pid may act on this one.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub pid: Option<NonZeroU32>,
    /// The host that pid is meaningful on, when the record names one.
    ///
    /// Absent rather than claimed: a pid means nothing across machines, so a
    /// host the record does not name resolves toward *not this one*.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub host: Option<String>,
    /// The driver's own process start token, when the record carries one.
    ///
    /// The proof that the pid beside it is still the process it was written for.
    /// Absent on a record that predates the stamp — and an absent stamp never
    /// matches, so a reader of this row acts on the pid above exactly as far as
    /// this field lets it.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub started: Option<String>,
    /// The driver that let go of this run to fire its run-end hook, while no
    /// adoption has driven it since.
    ///
    /// What the listing reads its liveness through beside the claim above, for
    /// the reason the fold carries it: that driver is alive while it awaits the
    /// hook and holds nothing. Absent for every run that fired no hook.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub let_go_by: Option<crate::projection::DriverClaim>,
    /// The run's aggregate wall clock and usage.
    ///
    /// The **whole of [`RunTelemetry`](crate::views::RunTelemetry)**, referenced
    /// rather than restated: its fields are declared once, on the type this
    /// crate already aggregates, so there are not two accounts of one run's
    /// clock to drift apart. It is here so that listing a host's runs no longer
    /// costs a process per row to get it.
    pub timing: RunTelemetry,
    /// The nodes a planner's `cancel` idled, which no later pass dispatches
    /// until a `requeue`.
    ///
    /// The **names**, rather than a count, because the one thing a view says
    /// about them is what to requeue. Recorded whatever the run's convergence —
    /// a reader decides for itself whether a park is holding a settled run back,
    /// which is a question about the run rather than about the node.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub parked: Vec<String>,
    /// The nodes that failed on a judge's own verdict, which nothing dispatches
    /// as they stand.
    ///
    /// **Three records at once**, and the reason this is stored rather than
    /// derived: a failed status and a task-failed outcome are on this document
    /// already, and the verdict itself is in the run's merged store — so a
    /// listing that had to open the store to tell a rejected node from one whose
    /// task simply failed would be the fold this document exists to remove.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub judge_rejected: Vec<String>,
    /// What each node that recorded a landing published, as the **inputs** to
    /// the landing question rather than as its answer.
    ///
    /// Deliberately not a stored verdict. Whether a change has reached its base
    /// is decided when a view renders — a change merged after its node settled is
    /// not work nobody landed, and a count that said it was is what sent a
    /// supervisor to re-dispatch work that was already on the base. So this
    /// carries what the run *observed* and where to ask again, and every reader
    /// re-asks; a node whose settlement already recorded `landed` is the one that
    /// is never asked again, because a base does not stop carrying what it
    /// carries.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub landings: BTreeMap<String, NodeLanding>,
    /// The journal's length in bytes when this document was written.
    ///
    /// Half of the stamp a stale summary is detected by. The journal is
    /// append-only, so a length that has moved is a record this document does
    /// not know about.
    pub journal_len: u64,
    /// The journal's modification time when this document was written, in
    /// milliseconds since the epoch.
    ///
    /// The other half, for the change a length cannot see: a store rewritten to
    /// the same size — healed of a torn tail, or edited by hand — is a store
    /// this document no longer describes.
    pub journal_mtime_ms: u64,
}

/// The journal's length and modification time, as it stands.
///
/// A journal that is not there stamps as `(0, 0)`: a run with no store yet is a
/// real state — a directory and a launch record, written before the first record
/// — and it is one a summary describes exactly as well as any other.
fn journal_stamp(paths: &RunPaths) -> (u64, u64) {
    let Ok(about) = std::fs::metadata(paths.journal()) else {
        return (0, 0);
    };
    let modified = about
        .modified()
        .ok()
        .and_then(|at| at.duration_since(std::time::UNIX_EPOCH).ok())
        .map_or(0, |since| {
            u64::try_from(since.as_millis()).unwrap_or(u64::MAX)
        });
    (about.len(), modified)
}

/// What a summary stamps: the bytes of the journal it **accounted for**, and
/// when the file was last written.
///
/// The length is the writer's own count rather than whatever the file holds at
/// the moment of the stat, and the difference is the whole safety of the stamp.
/// A run's journal has several appenders — the launcher relaying its driver's
/// stream, and the engine loop's own writer — so one landing between our append
/// and our stat would have us declare a document fresh for a record it does not
/// carry, and a reader would serve a row one record behind with nothing saying
/// so. Counting what was folded can only ever fall *short* of the file, and a
/// stamp that falls short reads as stale, which costs a fold and never an
/// answer.
type Stamp = (u64, u64);

/// What a run's **merged store** contributed to its summary.
///
/// One value rather than five parameters side by side, because they are read
/// together and only ever together: both producers of this document — the
/// journal writer folding a record at a time, and the reader folding a whole
/// store — hand over exactly these, and a caller that took four of the five
/// would be describing a store nobody recorded.
struct Store<'a> {
    /// The run's journal, folded into the plan of record.
    state: &'a RunState,
    /// How many records that store holds.
    event_count: u64,
    /// The wire string of the record the merge order ends with.
    last_event_kind: Option<String>,
    /// The run's aggregate wall clock and usage.
    timing: &'a RunTelemetry,
    /// The nodes some record carries a failing judge verdict for.
    judged: &'a BTreeSet<String>,
}

impl RunSummary {
    /// One run's summary: the stored document where it is current, and a fold
    /// where it is not.
    ///
    /// The **only** public way to this document, so no caller decides for itself
    /// whether a stored one may be served. A run with no summary, or one whose
    /// summary is stale against the journal's recorded length or modification
    /// time, is folded exactly as [`views::RunView::open`](crate::views::RunView::open)
    /// folds it and the fold is cached — so the next reader of that run pays a
    /// bounded read, and a run recorded by an older build answers identically
    /// and only more slowly.
    ///
    /// The cache is written best-effort and its failure is never reported: a
    /// read-only runs root, or a directory this reader may not write, costs the
    /// next reader a fold and costs this one nothing. A cache written beside a
    /// live driver may be overtaken by that driver's own next append, which is
    /// the same staleness this document is read through in the first place.
    pub fn of(paths: &RunPaths) -> crate::Result<Self> {
        if !paths.exists() {
            return Err(crate::Error::NoSuchRun {
                run: paths.run.clone(),
                root: paths.dir.parent().unwrap_or(Path::new(".")).to_path_buf(),
            });
        }
        let stamp = journal_stamp(paths);
        if let Some(stored) = ledger::read_json_opt::<Self>(&paths.summary()) {
            if (stored.journal_len, stored.journal_mtime_ms) == stamp && stored.run_id == paths.run
            {
                return Ok(stored.with_the_report_the_run_left(paths));
            }
        }
        // Stamped with what the store held **before** the fold, for the reason
        // [`Stamp`] states: a record appended while this read was folding is one
        // this row may not claim to carry.
        let folded = Self::folded(paths, stamp)?;
        let _ = ledger::write_json(&paths.summary(), &folded);
        Ok(folded)
    }

    /// Take the landings the run's **own settled report** re-read, where they say
    /// more than its journal did.
    ///
    /// The reader's half of a document that is complete about the journal and one
    /// step behind the report beside it. A driver re-reads every unlanded change
    /// as it closes out — `engine`'s `landings_after_asking_again` — and writes
    /// what it found to `result.json`, **after** the last record it appends: so
    /// the writer maintaining this document has already written its last version
    /// by the time that report exists. A fold takes the report
    /// ([`views::landings_the_run_re_read`](crate::views::landings_the_run_re_read)),
    /// so a served row that did not would be the two accounts of one run this
    /// document exists to keep as one — and it is the account that costs, because
    /// the listing would then go and ask a repository about a change the run had
    /// already watched land.
    ///
    /// **Only ever `unlanded` → `landed`, and only where there is an `unlanded`
    /// to move.** A base does not stop carrying what it carries, so the later
    /// answer in that direction alone cannot un-land anything; and a run with
    /// nothing unlanded has the same row either way, which keeps this read off the
    /// listing path for every run with nothing to gain from it.
    fn with_the_report_the_run_left(mut self, paths: &RunPaths) -> Self {
        let unlanded = graph::Landing::Unlanded.as_str();
        if !self.landings.values().any(|node| node.landing == unlanded) {
            return self;
        }
        // Read leniently: a report this build cannot parse — one a newer build
        // wrote, at a version this one refuses — leaves the row exactly as the
        // journal left it, which is the answer it always had.
        let Some(result) = ledger::read_json_opt::<crate::engine::RunResult>(&paths.result())
        else {
            return self;
        };
        for node in result.nodes {
            if node.landing != Some(graph::Landing::Landed) {
                continue;
            }
            if let Some(held) = self.landings.get_mut(&node.id) {
                if held.landing == unlanded {
                    held.landing = graph::Landing::Landed.as_str().to_string();
                }
            }
        }
        self
    }

    /// The same summary, always by folding the whole store.
    ///
    /// What the fallback runs, and what the writer's own account is held equal
    /// to: one derivation, so the two accounts of a run cannot drift apart.
    fn folded(paths: &RunPaths, stamp: Stamp) -> crate::Result<Self> {
        let view = crate::views::RunView::open(paths)?;
        let judged: BTreeSet<String> = view
            .events
            .iter()
            .filter_map(crate::report::a_judge_failed)
            .map(str::to_string)
            .collect();
        Ok(Self::derive(
            &paths.run,
            &view.launch,
            &Store {
                state: &view.state,
                event_count: view.events.len() as u64,
                last_event_kind: view.events.last().map(|event| event.kind.0.clone()),
                timing: &telemetry::of_run(paths, &view.events),
                judged: &judged,
            },
            stamp,
        ))
    }

    /// Compose the document out of a run's launch record and its folded state.
    ///
    /// The one place either producer builds a row, which is what makes the two
    /// the same row.
    fn derive(
        run: &str,
        launch: &LaunchRecord,
        store: &Store<'_>,
        (journal_len, journal_mtime_ms): Stamp,
    ) -> Self {
        let Store {
            state,
            event_count,
            last_event_kind,
            timing,
            judged,
        } = store;
        let (event_count, last_event_kind, timing) =
            (*event_count, last_event_kind.clone(), *timing);
        let statuses = state.statuses();
        let mut node_counts: BTreeMap<String, u64> = BTreeMap::new();
        for status in statuses.values() {
            *node_counts.entry(status.as_str().to_string()).or_insert(0) += 1;
        }
        let with_status = |wanted: graph::NodeStatus| -> Vec<String> {
            statuses
                .iter()
                .filter(|(_, status)| **status == wanted)
                .map(|(id, _)| id.clone())
                .collect()
        };
        Self {
            schema_version: SUMMARY_SCHEMA_VERSION,
            run_id: run.to_string(),
            last_write_at: state.last_write_at,
            last_event_kind,
            event_count,
            node_counts,
            stop_recorded: state.stop_recorded(),
            graph_complete: !statuses.is_empty() && graph::is_terminal(&statuses),
            decisions_pending: state.decisions_pending.len() as u64,
            surfaces_queued: state.surfaces_queued,
            surfaces_read: state.surfaces_read,
            awaiting_human_action: state.awaiting_human_action(),
            project: launch.project.clone(),
            launcher: launch.launcher.clone(),
            session: launch.session.clone(),
            started_at: launch.launched_at().map(str::to_string),
            pid: launch.driver_pid(),
            host: launch.recorded_host().map(str::to_string),
            started: launch.driver_stamp().map(str::to_string),
            let_go_by: state.let_go_by.clone(),
            timing: timing.clone(),
            parked: with_status(graph::NodeStatus::Parked),
            // The same three records `views::rejected_by_a_judge` reads, taken
            // where all three are in hand: the derived status, the settlement's
            // own outcome, and the verdict the store carries. Either record alone
            // names the wrong nodes — a node that simply failed its task is not a
            // node a judge turned down.
            judge_rejected: with_status(graph::NodeStatus::Failed)
                .into_iter()
                .filter(|id| {
                    matches!(
                        state.outcomes.get(id).map(String::as_str),
                        Some(crate::engine::TASK_FAILED | crate::engine::TASK_FAILED_CHANGE_OPEN)
                    )
                })
                .filter(|id| judged.contains(id))
                .collect(),
            landings: state
                .landings
                .iter()
                .map(|(node, landing)| {
                    (
                        node.clone(),
                        NodeLanding {
                            landing: landing.as_str().to_string(),
                            branch: state.branches.get(node).cloned(),
                            repo: state.graph.get(node).and_then(|node| node.repo.clone()),
                            drafted: statuses.get(node) == Some(&graph::NodeStatus::CompleteDraft),
                        },
                    )
                })
                .collect(),
            journal_len,
            journal_mtime_ms,
        }
    }
}

/// What a bounded listing over a whole runs root read, and what it refused.
///
/// The second half is why this type has one: a listing that reported only what
/// it could read would reintroduce, at the cheap surface, the silent omission
/// [`Survey`](crate::views::Survey) exists to remove — a host with thirty run
/// roots on it rendering as nothing at all. A refused root is a fact about the
/// root, and it arrives on the same terms
/// [`Survey::skipped`](crate::views::Survey::skipped) already states.
#[derive(Debug, Clone, PartialEq)]
pub struct Listing {
    /// The runs root this listing read. Named on the output, because it is the
    /// scope of every claim made from it.
    pub root: PathBuf,
    /// The runs that read, **most recently written first** — the order
    /// [`RunSummary::last_write_at`] is stored to make answerable without a
    /// fold. A run whose store carries nothing datable sorts last, then by id,
    /// so the order is total and stable.
    pub summaries: Vec<RunSummary>,
    /// The run roots that did not, each with the reason it was refused.
    pub skipped: Vec<Skipped>,
}

impl Listing {
    /// Read every run under a root, keeping what could not be read.
    ///
    /// The bounded counterpart of [`Survey::of`](crate::views::Survey::of), and
    /// the same account of a refusal: a root the ledger refused and a run this
    /// build could neither read nor fold are the same fact to a reader — one
    /// directory that claimed to be a run and is not being reported as one — so
    /// they arrive on one list.
    pub fn of(root: &Path) -> Self {
        let index = ledger::all_runs(root);
        let mut summaries = Vec::new();
        let mut skipped = index.skipped;
        for paths in index.runs {
            match RunSummary::of(&paths) {
                Ok(summary) => summaries.push(summary),
                // The refusal as the folding reader already words it. Nothing is
                // added to it here — a second wording of one refusal is a second
                // thing to keep true.
                Err(error) => skipped.push(Skipped {
                    path: paths.dir,
                    reason: error.to_string(),
                }),
            }
        }
        summaries.sort_by(|a, b| {
            b.last_write_at
                .cmp(&a.last_write_at)
                .then_with(|| a.run_id.cmp(&b.run_id))
        });
        skipped.sort_by(|a, b| a.path.cmp(&b.path));
        Self {
            root: root.to_path_buf(),
            summaries,
            skipped,
        }
    }
}

/// A run's store folded: everything the summary is derived from except the
/// launch record.
///
/// One value rather than four fields side by side, because it is carried
/// forward as a unit — [`Maintainer`] holds the fold of everything settled and
/// re-folds the records still arriving at one instant onto a copy of it, and a
/// copy that took three of the four would be a state describing a store nobody
/// recorded.
#[derive(Debug, Default, Clone)]
struct Folded {
    state: RunState,
    aggregate: telemetry::Aggregate,
    events: u64,
    /// The kind of the record the **merge order** ends with, which for a store
    /// folded in that order is whatever was taken last.
    last_event_kind: Option<String>,
    /// The nodes some record in this store carries a **failing judge verdict**
    /// for.
    ///
    /// Folded a record at a time because that is all it takes: a verdict is on
    /// the settlement that carried it and nothing later withdraws one, so the
    /// set only grows. It is the one input to `judge_rejected` that lives in the
    /// store rather than in the state beside it.
    judged: BTreeSet<String>,
}

impl Folded {
    /// The state of an empty store, as [`projection::fold`] starts from.
    fn new() -> Self {
        Self {
            state: RunState {
                strict: true,
                ..RunState::default()
            },
            ..Self::default()
        }
    }

    /// Take one record that belongs at the end of the merge order.
    fn take(&mut self, paths: &RunPaths, event: &Envelope) {
        projection::fold_one(&mut self.state, event);
        self.aggregate.fold(paths, event);
        self.events += 1;
        self.last_event_kind = Some(event.kind.0.clone());
        if let Some(node) = crate::report::a_judge_failed(event) {
            self.judged.insert(node.to_string());
        }
    }
}

/// One run's summary, kept current a record at a time by the process appending
/// to its journal.
///
/// Held by [`journal::Journal`], which is the only writer of a run's merged
/// store, so the document is written by whatever wrote the record it describes
/// and is current for a live run rather than as of some later pass.
///
/// # Why an appended record can be folded at all
///
/// Because the derivations are over the store in **merge order**, and while
/// every record arrives at or past the last instant already recorded, that order
/// is: records by timestamp, ties between streams broken by stream id, each
/// stream's own `seq` preserved. So this holds the fold of everything stamped
/// *before* the newest instant, and the handful of records stamped **at** it —
/// two producers relay their own timestamps into this store and a millisecond
/// holds several records, so a record arriving beside one already folded has to
/// be able to sort in front of it. Serving the summary re-folds that handful
/// onto a copy, which is bounded by one instant's records rather than by the
/// run's length.
///
/// Two arrivals do not belong anywhere this can place them, and the answer to
/// both is to read the store again. One is a record stamped *behind* the newest
/// instant **any** stream has reached — a producer's clock is not this one's, and
/// the merge holds a stream stamped ahead of the others back until they drain, so
/// the order can end behind an instant the store already carries. The other is a
/// record whose
/// stream has already had a **higher** `seq` folded and settled: a stream is
/// merged in its own `seq` order whatever its stamps say, so a producer that
/// publishes `seq` 10 stamped before its `seq` 5 puts the later record first in
/// the merge and there is no instant this can hold it at. Both are real — the
/// second is what an `oneharness-session` record does, published out of band and
/// stamped when its session opened — and both are rare, which is what makes
/// reading again affordable.
#[derive(Debug)]
pub(crate) struct Maintainer {
    paths: RunPaths,
    /// The fold of every record stamped **before** [`open_ts`](Self::open_ts).
    settled: Folded,
    /// The records stamped **at** it, unfolded, so one arriving beside them can
    /// still sort in front of them.
    open: Vec<Envelope>,
    /// The instant the **merge order ends at**. Empty for a store with nothing in
    /// it, which every stamp is past.
    open_ts: String,
    /// The newest instant **any** record in this store carries, which is not
    /// always the one above.
    ///
    /// The merge is a k-way one — each stream in its own `seq`, streams
    /// interleaved by `ts` — so a stream whose head is stamped ahead of every
    /// other stream's remaining records is held back until those are drained, and
    /// the order then ends on a record stamped *behind* one already placed. A
    /// producer whose clock runs ahead of this one's is ordinary between hosts, so
    /// this is not a rare shape. An arrival behind this instant has records to
    /// sort in front of wherever the order happens to end, and is not one this
    /// state can place.
    newest_ts: String,
    /// How many bytes of the journal this state has accounted for. See
    /// [`Stamp`].
    accounted: u64,
    /// The highest `seq` each stream has **settled**, which is what an arriving
    /// record of that stream has to be past: the merge orders a stream by its own
    /// `seq` and by nothing else, so a lower one arriving now belongs in front of
    /// a record this state has already frozen.
    settled_seq: BTreeMap<String, u64>,
}

impl Maintainer {
    /// Build the state of a run's store as it stands, by reading it once.
    ///
    /// The one unbounded read in this file, and it happens where an unbounded
    /// read already did: a journal writer opens by reading the store to find the
    /// sequence number it may claim.
    pub(crate) fn of(paths: &RunPaths) -> Self {
        // What is accounted for is exactly what was read: the finished records the
        // bus reader handed back, up to the boundary after the last of them. A final
        // line whose writer had not finished it is left out of both, so the count is
        // always a position that reader resumes from — counting its bytes would put
        // the next tail read inside that record once its writer finished it, which
        // the reader refuses as no boundary at all. An appender that landed across
        // the read is past the count, which reads as stale rather than as an answer.
        let read = journal::finished_records_after(&paths.journal(), 0);
        let accounted: u64 = read.iter().map(|(_, bytes)| *bytes).sum();
        let mut events: Vec<Envelope> = read.into_iter().filter_map(|(record, _)| record).collect();
        journal::merge_order(&mut events);

        // The last instant's records are held open rather than folded, because
        // the next record to arrive may be stamped at that same instant and
        // belong in front of one of them.
        //
        // The **trailing run** of them, counted from the end, and not every
        // record that happens to carry that stamp: the merge orders by each
        // stream's own `seq` first, so a store some producer stamped out of
        // order can carry that instant earlier as well — and taking those with
        // it would re-sort records the merge had already placed, which is a
        // different store from the one on disk.
        let open_ts = events
            .last()
            .map(|event| event.ts.clone())
            .unwrap_or_default();
        let opened = events
            .iter()
            .rposition(|event| event.ts != open_ts)
            .map_or(0, |before| before + 1);
        let mut settled = Folded::new();
        let mut settled_seq = BTreeMap::new();
        for event in &events[..opened] {
            settled.take(paths, event);
            seq_reached(&mut settled_seq, event);
        }
        Self {
            paths: paths.clone(),
            settled,
            open: events[opened..].to_vec(),
            newest_ts: events
                .iter()
                .map(|event| &event.ts)
                .max()
                .cloned()
                .unwrap_or_default(),
            open_ts,
            accounted,
            settled_seq,
        }
    }

    /// The whole store folded: everything settled, plus the newest instant's
    /// records in the order the merge puts them.
    ///
    /// Within one instant the merge orders by stream and then by each stream's
    /// own `seq`, which is what this sorts by — the same order
    /// [`journal::merge_order`] would put them in, decided in one place so the
    /// two cannot disagree.
    fn current(&self) -> Folded {
        let mut folded = self.settled.clone();
        let mut open: Vec<&Envelope> = self.open.iter().collect();
        open.sort_by(|a, b| a.stream.cmp(&b.stream).then(a.seq.cmp(&b.seq)));
        for event in open {
            folded.take(&self.paths, event);
        }
        folded
    }

    /// Take one appended record, and write the run's summary.
    ///
    /// Called after the record has reached the file, so a read of the store —
    /// the answer to a record this state cannot place — reads a store that
    /// already holds it.
    ///
    /// `bytes` is the record's own size, terminator included. This state's count is
    /// always a record boundary the bus reader handed back — it never counts a line
    /// whose writer had not finished it — and that is what makes the arithmetic
    /// below sound. The append that wrote this record may first have healed a
    /// fragment a dead writer left ([`ledger::append_line_healed`] reports it), but a
    /// heal cuts only what follows the file's last boundary, which is never in front
    /// of this count, so the count needs nothing from it. And a line that was
    /// half-written when this state last read, and has been finished since, is read
    /// whole from that boundary rather than resumed from inside. A store shorter than
    /// the count plus this record is not the store this state was holding, and is
    /// read again rather than folded from an offset nothing can place.
    pub(crate) fn appended(&mut self, event: &Envelope, bytes: u64) {
        let len = journal_stamp(&self.paths).0;
        if len == self.accounted + bytes {
            // Ours alone: the file grew by exactly this record, so what is folded
            // here and what the file holds are the same store.
            self.fold(event, bytes);
        } else if len > self.accounted + bytes {
            // Somebody else appended beside us, which is the ordinary shape of a
            // run being driven: the relay thread writes the observer's envelopes
            // while the engine thread writes the graph's. What the store grew by
            // is read **from where this state left off** rather than from the
            // beginning — reading it whole per append would make recording a run
            // quadratic in its own length, which is the cost this document exists
            // to remove, reintroduced at the writer.
            self.catch_up();
        } else {
            // The file is not the file this state was holding: it is shorter than
            // what was accounted for and the record just written to it — replaced,
            // or cut back past a boundary this state counted. Nothing here can be
            // placed against it.
            *self = Self::of(&self.paths);
        }
        self.write();
    }

    /// Fold one record, or read the store again where it does not belong at or
    /// past the newest instant — and say which it did.
    ///
    /// A read has taken the whole store, so a caller walking a tail of it has
    /// nothing left to fold and must stop rather than fold the same records
    /// twice.
    fn fold(&mut self, event: &Envelope, bytes: u64) -> Rebuilt {
        let settled_past_it = self
            .settled_seq
            .get(&event.stream)
            .is_some_and(|reached| event.seq <= *reached);
        // A record stamped later than the instant still open closes that instant,
        // so one whose own stream is *already* further on inside it cannot be
        // placed either: the merge would put this record in front of one about to
        // be frozen behind it.
        let closing_over_it = event.ts > self.open_ts
            && self
                .open
                .iter()
                .any(|held| held.stream == event.stream && held.seq > event.seq);
        if settled_past_it || closing_over_it || event.ts < self.newest_ts {
            *self = Self::of(&self.paths);
            return Rebuilt::Yes;
        }
        if event.ts > self.open_ts {
            self.settled = self.current();
            for settled in &self.open {
                seq_reached(&mut self.settled_seq, settled);
            }
            self.open_ts = event.ts.clone();
            self.open.clear();
        }
        self.open.push(event.clone());
        if event.ts > self.newest_ts {
            self.newest_ts = event.ts.clone();
        }
        self.accounted += bytes;
        Rebuilt::No
    }

    /// Fold every finished record the store has grown by since this state last
    /// accounted for it.
    ///
    /// Bounded by what arrived rather than by what the run has ever recorded: the
    /// tail is read from the byte this state stopped at. A record in it stamped
    /// behind the newest instant takes the whole store again, which is the same
    /// answer [`fold`](Self::fold) gives for one appended here — and one this
    /// build cannot read still advances the count, because it is still a line the
    /// file holds. A final line whose writer has not finished it does not: the
    /// count stays at the boundary in front of it, which is where the next read
    /// resumes once its newline lands.
    fn catch_up(&mut self) {
        for (record, bytes) in
            journal::finished_records_after(&self.paths.journal(), self.accounted)
        {
            match record {
                // A read answered the whole tail; there is nothing left of it
                // this state has not already taken.
                Some(event) if self.fold(&event, bytes) == Rebuilt::Yes => return,
                Some(_) => {}
                None => self.accounted += bytes,
            }
        }
    }

    /// Write what the run stands at now.
    ///
    /// Best effort, and its failure is never reported: a summary that could not
    /// be written costs the next reader a fold, and a journal append that
    /// refused because the document beside it could not be written would be this
    /// cache taking a run's own record down with it.
    fn write(&mut self) {
        let mut folded = self.current();
        // Cross-DAG edges are resolved the way a view resolves them, so a row
        // and the graph it opens cannot describe different graphs. A graph
        // naming no other run pays a walk of its own nodes and nothing else; one
        // that does names pays that edge's read, which is what any reader of it
        // already pays.
        folded.state.cross_dag = crate::crossdag::resolve_quietly(
            &self
                .paths
                .dir
                .parent()
                .map_or_else(ledger::runs_root, Path::to_path_buf),
            &folded.state.graph,
        );
        // The landings a closing driver re-read and wrote to its result, taken
        // exactly where a folding reader takes them. Without it the two accounts
        // of one run would disagree on the one field that outlives the journal:
        // a change the run watched reach its base after the record that said it
        // had not.
        //
        // Asked only where there is something for it to move. That re-read only
        // ever turns `unlanded` into `landed`, so a run with no unlanded change
        // has the same state either way — and this runs on the **append path**,
        // where a file opened per record is a cost the whole run pays to answer
        // a question almost every record has no stake in.
        if folded
            .state
            .landings
            .values()
            .any(|landing| *landing == graph::Landing::Unlanded)
        {
            crate::views::landings_the_run_re_read(&mut folded.state, &self.paths);
        }
        let summary = RunSummary::derive(
            &self.paths.run,
            &self.launch(),
            &Store {
                state: &folded.state,
                event_count: folded.events,
                last_event_kind: folded.last_event_kind.clone(),
                timing: &folded.aggregate.finish(&self.paths.run, &folded.state),
                judged: &folded.judged,
            },
            (self.accounted, journal_stamp(&self.paths).1),
        );
        let _ = ledger::write_json(&self.paths.summary(), &summary);
    }

    /// The run's launch record, as the row's attribution needs it.
    ///
    /// Read from the file on each write rather than held: the record is rewritten
    /// under a live run — an adoption claims the run for a fresh driver, and the
    /// observer's graph run is recorded after the launch that made it — and a
    /// copy taken once would go on naming the driver that died. A record this
    /// build cannot read leaves the row with the launch's defaults, which say
    /// exactly what they say everywhere else: the record does not say.
    fn launch(&self) -> LaunchRecord {
        ledger::read_json_opt::<LaunchRecord>(&self.paths.launch()).unwrap_or(LaunchRecord {
            run_id: self.paths.run.clone(),
            project: String::new(),
            dir: PathBuf::new(),
            graph: String::new(),
            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: crate::sys::UNKNOWN_LAUNCHER.to_string(),
            session: String::new(),
            pid: 0,
            host: String::new(),
            started: String::new(),
            started_at: String::new(),
            heartbeat_interval: 0,
            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: crate::filter::Filters::default(),
            bus_config: Default::default(),
            envelope_reviewer_bar: Default::default(),
        })
    }
}

/// Fold a run's store as it stands and write its summary document, stamped for
/// the journal as it stands now.
///
/// The driver's closeout calls this once **every appender it holds has made its
/// last append** — the engine loop, the observer relay, the lifecycle relay,
/// observer teardown and an adopted driver included — so a run this engine drove
/// to settlement leaves a document that accounts for its journal by construction,
/// without any reader having to refresh it. That is clause 2 of the verb's rule
/// (entry 68 of `docs/contract-divergences.md`): `onepipeline start` handing back
/// at settlement never leaves a settled run that `unwatched` reports.
///
/// A full read rather than one more incremental append, and that is the point:
/// the two `Maintainer`s a driven run keeps over one journal each stamp only what
/// **they** folded, so a document left by whichever of them wrote last can be
/// stamped short of the file the other grew. This reads the whole store once,
/// with nothing left appending to it, so the length it stamps is the file's own.
/// Best effort like every write on this path — a document that could not be
/// written costs the next reader a fold, exactly as [`Maintainer::write`] says.
// llmlint: ignore[changed_behavior_has_e2e] a closeout that leaves a *current* document is
// read off the files by the three `unwatched::*_leaves_a_current_document` journeys, and what
// this write does to a document left behind its journal — the state no offline closeout
// reaches — is held by `the_seal_leaves_a_document_current_for_the_journal_as_it_stands`
// below. The only untested branch is the write failing, which is not new: `seal` inherits
// [`Maintainer::write`]'s discard-on-failure, whose recovery a reader re-folds and which
// inducing needs host sabotage rather than a user journey.
pub(crate) fn seal(paths: &RunPaths) {
    Maintainer::of(paths).write();
}

/// Record how far a stream has been folded, keeping the highest `seq` seen.
///
/// The highest rather than the last, because a producer may publish its own
/// records out of `seq` order and the question this answers is what the merge
/// has already placed.
fn seq_reached(reached: &mut BTreeMap<String, u64>, event: &Envelope) {
    let held = reached.entry(event.stream.clone()).or_insert(event.seq);
    *held = (*held).max(event.seq);
}

/// Whether folding a record took the whole store again — which is what a caller
/// walking a tail has to stop on, or it folds the same records twice.
#[derive(Debug, PartialEq, Eq)]
enum Rebuilt {
    No,
    Yes,
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::event::{EventKind, Labels, Source, ENVELOPE_VERSION};
    use crate::journal::{Journal, PipelineKind};
    use crate::plan::{Node, Plan, PLAN_SCHEMA_VERSION};
    use crate::sys;
    use serde_json::json;

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

    fn plan(nodes: &[&str]) -> Plan {
        Plan {
            schema_version: PLAN_SCHEMA_VERSION,
            goal: Some(crate::plan::Goal {
                text: "list a host without folding it".into(),
            }),
            name: Some("demo".into()),
            concurrency: 4,
            tasks: nodes
                .iter()
                .map(|id| Node {
                    id: (*id).to_string(),
                    persona: Some("engineer".into()),
                    task: Some("## What\ndo it".into()),
                    ..Node::default()
                })
                .collect(),
        }
    }

    /// A run root with a launch record, as `start` leaves one.
    fn a_run(root: &Path, run: &str) -> RunPaths {
        let paths = RunPaths::under(root, run);
        paths.create().expect("the run directory");
        let mut record = LaunchRecord {
            run_id: run.to_string(),
            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: String::new(),
            pr_author_graph: String::new(),
            node_validator: String::new(),
            envelope_reviewer: String::new(),
            launcher: "e2e".into(),
            session: "a-session".into(),
            pid: 0,
            host: String::new(),
            started: String::new(),
            started_at: sys::now_rfc3339(),
            heartbeat_interval: 1_800,
            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: crate::filter::Filters::default(),
            bus_config: Default::default(),
            envelope_reviewer_bar: Default::default(),
        };
        record.driven_by_this_process();
        ledger::write_json(&paths.launch(), &record).expect("a launch record");
        paths
    }

    /// One of this crate's own records, as its writer emits it.
    fn emit(journal: &mut Journal, kind: PipelineKind, node: Option<&str>, run: &str) {
        journal
            .emit(
                kind,
                crate::journal::labels(run, node),
                crate::journal::payload(&[("status", json!("done"))]),
            )
            .expect("appended");
    }

    /// A run whose store holds `records` of this crate's own, written through
    /// the real journal writer — which is what keeps the summary current.
    fn recorded(root: &Path, run: &str, records: usize) -> RunPaths {
        let paths = a_run(root, run);
        let mut journal = Journal::open(&paths);
        journal
            .emit(
                PipelineKind::RunStarted,
                crate::journal::labels(run, None),
                crate::journal::payload(&[("plan", json!(plan(&["build", "ship"])))]),
            )
            .expect("appended");
        for nth in 0..records {
            emit(
                &mut journal,
                PipelineKind::NodeReady,
                Some(if nth % 2 == 0 { "build" } else { "ship" }),
                run,
            );
        }
        paths
    }

    /// What one read of a run's summary cost, in bytes off the ledger.
    fn cost_of(paths: &RunPaths) -> (RunSummary, u64) {
        let before = ledger::bytes_read();
        let summary = RunSummary::of(paths).expect("the run reads");
        (summary, ledger::bytes_read() - before)
    }

    /// The whole point of the document: a listing's cost does not grow with the
    /// journals it is listing.
    ///
    /// Two runs three orders of magnitude apart, measured on **bytes read off
    /// the ledger** rather than on a clock — a wall-clock ratio says nothing
    /// about why it was fast, and this says exactly what was opened. The fold
    /// beside it is the control: it reads the whole store, and it is the reading
    /// every listing did before this document existed.
    #[test]
    fn a_summary_read_is_bounded_and_the_fold_it_replaces_is_not() {
        let root = scratch("bounded");
        let small = recorded(&root, "small", 10);
        let large = recorded(&root, "large", 10_000);
        assert!(
            std::fs::metadata(large.journal()).expect("a store").len()
                > 100 * std::fs::metadata(small.journal()).expect("a store").len(),
            "the two stores are not orders of magnitude apart"
        );

        let (small_row, small_cost) = cost_of(&small);
        let (large_row, large_cost) = cost_of(&large);
        assert_eq!(small_row.event_count, 11);
        assert_eq!(large_row.event_count, 10_001);
        // Both read one document each and stat one journal, so the cost is the
        // document's — which is the graph's size and not the store's.
        assert!(
            large_cost < 2 * small_cost,
            "reading the larger run's summary cost {large_cost} bytes against \
             {small_cost} for a store a thousandth the size"
        );

        // And the reading it replaces, over the same two stores: proportional,
        // which is what makes the measurement above mean something.
        let folded = |paths: &RunPaths| {
            std::fs::remove_file(paths.summary()).expect("the document");
            cost_of(paths).1
        };
        let small_fold = folded(&small);
        let large_fold = folded(&large);
        assert!(
            large_fold > 100 * small_fold,
            "the fold this replaces cost {large_fold} against {small_fold}, so the \
             measurement above is not measuring what a listing reads"
        );
        std::fs::remove_dir_all(&root).ok();
    }

    /// A run with no summary answers identically, folding once, and the next
    /// reader of it pays a bounded read.
    #[test]
    fn a_run_with_no_summary_folds_once_and_caches_what_it_folded() {
        let root = scratch("fallback");
        let paths = recorded(&root, "demo", 40);
        let maintained = RunSummary::of(&paths).expect("the run reads");

        std::fs::remove_file(paths.summary()).expect("the document");
        let (folded, fold_cost) = cost_of(&paths);
        assert_eq!(
            folded, maintained,
            "the row a fold produces differs from the row the writer maintained"
        );
        assert!(
            paths.summary().is_file(),
            "the fold was not cached, so every later reader folds again"
        );

        let (again, cached_cost) = cost_of(&paths);
        assert_eq!(again, folded);
        assert!(
            cached_cost < fold_cost,
            "the cached read cost {cached_cost} against a fold's {fold_cost}"
        );
        std::fs::remove_dir_all(&root).ok();
    }

    /// A summary that no longer describes the store beside it is refolded.
    ///
    /// Both halves of the stamp, because they catch different things: a record
    /// appended behind the writer's back moves the journal's **length**, and a
    /// store rewritten to the same size — healed of a torn tail, or edited by
    /// hand — moves only its **modification time**.
    #[test]
    fn a_stale_summary_is_refolded_rather_than_served() {
        let root = scratch("stale");
        let paths = recorded(&root, "demo", 6);
        let served = RunSummary::of(&paths).expect("the run reads");
        assert_eq!(served.event_count, 7);

        // A record appended by something that is not this writer: the length
        // moves, and the document no longer describes the store.
        let mut appended = event(PipelineKind::NodeReady, "demo", "other-stream", 0);
        appended.ts = sys::now_rfc3339();
        ledger::append_line(
            &paths.journal(),
            &serde_json::to_string(&appended).expect("a record"),
        )
        .expect("appended");
        let refolded = RunSummary::of(&paths).expect("the run reads");
        assert_eq!(
            refolded.event_count, 8,
            "a summary that predates a record in the store was served anyway"
        );

        // And the same size, rewritten: only the modification time says so.
        let store = std::fs::read(paths.journal()).expect("the store");
        let len = store.len();
        // Far enough ahead that no filesystem's timestamp granularity hides it.
        std::thread::sleep(std::time::Duration::from_millis(1_100));
        std::fs::write(paths.journal(), &store[..len - 1]).expect("a store rewritten in place");
        std::fs::write(paths.journal(), [&store[..len - 1], b"\n"].concat())
            .expect("a store rewritten to its own length");
        assert_eq!(
            std::fs::metadata(paths.journal()).expect("the store").len() as usize,
            len,
            "the rewrite changed the length, so this is not the case under test"
        );
        let served = RunSummary::of(&paths).expect("the run reads");
        assert_eq!(
            (served.journal_len, served.journal_mtime_ms),
            journal_stamp(&paths),
            "a summary written against a store that has since been rewritten was served"
        );
        std::fs::remove_dir_all(&root).ok();
    }

    /// A run that is **still recording** is listed at bounded cost, because the
    /// document is kept current as the store grows rather than written once.
    ///
    /// Several rounds, and after each one both halves: the served row says what
    /// the run now is, and serving it cost what it cost the round before.
    #[test]
    fn a_run_still_recording_stays_current_and_stays_bounded() {
        let root = scratch("growing");
        let paths = a_run(&root, "demo");
        let mut journal = Journal::open(&paths);
        journal
            .emit(
                PipelineKind::RunStarted,
                crate::journal::labels("demo", None),
                crate::journal::payload(&[("plan", json!(plan(&["build"])))]),
            )
            .expect("appended");

        let mut costs = Vec::new();
        let mut written = 1;
        for round in 1..=4 {
            for _ in 0..(round * 500) {
                emit(&mut journal, PipelineKind::NodeReady, Some("build"), "demo");
                written += 1;
            }
            let (row, cost) = cost_of(&paths);
            assert_eq!(
                row.event_count, written,
                "the served summary is behind the store it describes"
            );
            assert_eq!(
                row.last_event_kind.as_deref(),
                Some(PipelineKind::NodeReady.as_str())
            );
            costs.push(cost);
        }
        let (first, last) = (costs[0], costs[costs.len() - 1]);
        assert!(
            last < 2 * first,
            "serving the summary grew with the journal: {costs:?}"
        );
        std::fs::remove_dir_all(&root).ok();
    }

    /// A record that arrives **behind** the newest instant recorded is not folded
    /// onto the end of it.
    ///
    /// Two producers relay their own timestamps into this store, so a record
    /// landing out of order is a case rather than a hypothetical — and the whole
    /// of the timing account is a walk of the timeline in order. The answer is
    /// to rebuild, and what proves it is the fold beside it: the two rows are
    /// one row.
    #[test]
    fn a_record_stamped_behind_the_newest_instant_is_reread_rather_than_folded_onto_the_end() {
        let root = scratch("out-of-order");
        let paths = recorded(&root, "demo", 4);

        // A sibling's record, stamped before everything already in the store.
        let mut behind = event(PipelineKind::NodeReady, "demo", "a-sibling", 7);
        behind.source = Source::Agentgraph;
        behind.kind = EventKind("turn-completed".into());
        behind.ts = "2020-01-01T00:00:00.000Z".into();
        behind
            .payload
            .insert("usage".into(), json!({"input_tokens": 11}));
        let mut journal = Journal::open(&paths);
        journal.relay(&behind).expect("relayed");

        let served = RunSummary::of(&paths).expect("the run reads");
        std::fs::remove_file(paths.summary()).expect("the document");
        let folded = RunSummary::of(&paths).expect("the run reads");
        assert_eq!(
            served, folded,
            "a record stamped behind the newest instant left the two accounts apart"
        );
        // And it is measured where it belongs: the run's clock now starts at the
        // record that is stamped first.
        assert!(
            folded.timing.wall_ms > 0,
            "a record stamped years before the store left no wall clock"
        );
        std::fs::remove_dir_all(&root).ok();
    }

    /// A producer that publishes its records out of its own `seq` order.
    ///
    /// Not a hypothetical: `oneagentgraph` relays an `oneharness-session` record
    /// out of band, stamped when the session opened and carrying a `seq` far past
    /// the turn records around it — so one stream arrives `2, 3, 10, 4, 5` with
    /// `10` stamped *before* `5`. The merge orders a stream by its own `seq` and
    /// by nothing else, so `10` belongs after `5` however it is stamped, and
    /// there is no instant a state folding forward can hold it at. What the
    /// writer does is read the store again — and what this holds is that it does,
    /// by holding its row equal to the fold's.
    ///
    /// **Both ways the record can be reached**, because they are caught by
    /// different readings and only one of them was: the out-of-order `seq` can
    /// arrive while the instant it belongs behind is still open, and it can
    /// arrive after something else has already closed that instant.
    #[test]
    fn a_producer_publishing_out_of_its_own_seq_order_leaves_one_account_not_two() {
        let root = scratch("out-of-seq");
        // The stream, exactly as it arrives, and then the same stream with a
        // record of *another* producer closing the instant in between.
        let closed_by_another = |also: bool| -> Vec<(&'static str, u64, &'static str, bool)> {
            let mut relayed = vec![
                ("a-sibling", 2, "turn-activity", false),
                ("a-sibling", 3, "turn-message", false),
                // The record out of `seq` order, and one that means something: it
                // opens the judge side's turn, so where it is folded decides
                // whose the run's last stretch of clock is.
                ("a-sibling", 10, "member-started", true),
                ("a-sibling", 4, "turn-completed", false),
            ];
            if also {
                relayed.push(("b-sibling", 0, "turn-activity", false));
            }
            relayed.push(("a-sibling", 5, crate::report::MEMBER_SETTLED, false));
            relayed
        };

        for (run, relayed) in [
            ("still-open", closed_by_another(false)),
            ("already-closed", closed_by_another(true)),
        ] {
            let paths = a_run(&root, run);
            let mut engine = Journal::open(&paths);
            engine
                .emit(
                    PipelineKind::RunStarted,
                    crate::journal::labels(run, None),
                    crate::journal::payload(&[("plan", json!(plan(&["build"])))]),
                )
                .expect("appended");
            emit(
                &mut engine,
                PipelineKind::NodeDispatched,
                Some("build"),
                run,
            );

            let mut relay = Journal::open(&paths);
            let base = sys::now_millis();
            let instant = sys::rfc3339_from_millis(base);
            let later = sys::rfc3339_from_millis(base + 5);
            for (stream, seq, kind, judge) in relayed {
                let mut record = event(PipelineKind::NodeReady, run, stream, seq);
                record.source = Source::Agentgraph;
                record.kind = EventKind(kind.into());
                // The two records past the burst are stamped at the later
                // instant; everything in the burst shares the first one.
                record.ts = if seq == 5 || stream == "b-sibling" {
                    later.clone()
                } else {
                    instant.clone()
                };
                if judge {
                    record.payload.insert("role".into(), json!("judge"));
                }
                relay.relay(&record).expect("relayed");
            }
            // Far enough after that the span the run's clock ends on is a real
            // one: the record out of `seq` order is stamped *before* the one
            // merged in front of it, so where the two orders leave the clock is
            // where they differ — and a span of nought would hide it.
            std::thread::sleep(std::time::Duration::from_millis(30));
            emit(&mut engine, PipelineKind::NodeSettled, Some("build"), run);

            let served = RunSummary::of(&paths).expect("the run reads");
            std::fs::remove_file(paths.summary()).expect("the document");
            assert_eq!(
                served,
                RunSummary::of(&paths).expect("the run folds"),
                "on '{run}' a producer's out-of-order `seq` left the maintained row \
                 and the folded row apart"
            );
        }
        std::fs::remove_dir_all(&root).ok();
    }

    /// Two writers on one run's journal, which is the ordinary shape of a run
    /// being driven: the relay thread appends the observer's envelopes while the
    /// engine thread appends the graph's.
    ///
    /// Both keep the document current, and neither re-reads the store to do it.
    /// Reading the store again per append is the obvious answer to "somebody
    /// else appended", and it makes recording a run quadratic in its own length
    /// — which is the cost this document exists to remove, reintroduced at the
    /// writer.
    #[test]
    fn two_writers_on_one_journal_keep_the_summary_current_without_rereading_the_store() {
        let root = scratch("two-writers");
        let paths = recorded(&root, "demo", 400);
        let store = std::fs::metadata(paths.journal()).expect("a store").len();

        let mut engine = Journal::open(&paths);
        let mut relay = Journal::open(&paths);
        let before = ledger::bytes_read();
        const ROUNDS: usize = 40;
        for nth in 0..ROUNDS {
            emit(&mut engine, PipelineKind::NodeReady, Some("build"), "demo");
            let mut relayed = event(PipelineKind::NodeReady, "demo", "a-sibling", nth as u64);
            relayed.source = Source::Agentgraph;
            relayed.kind = EventKind("turn-completed".into());
            relay.relay(&relayed).expect("relayed");
        }
        let maintaining = ledger::bytes_read() - before;
        // Reading the store again for each of these would cost the store's whole
        // length every time. What keeping the document current may cost is the
        // tails that arrived, plus the launch record each row's attribution is
        // read from — both real reads, and neither proportional to the run's
        // length. The bound rules out the one that is proportional to both.
        let rereading = store * (ROUNDS as u64) * 2;
        assert!(
            maintaining * 4 < rereading,
            "keeping the document current across {} interleaved appends read {maintaining} \
             bytes, against {rereading} for reading a store of {store} again each time: \
             the writer is re-reading what it already holds",
            ROUNDS * 2
        );

        // And what it kept is what the store says: the two accounts of a run
        // written by two processes at once are still one account.
        let served = RunSummary::of(&paths).expect("the run reads");
        assert_eq!(served.event_count as usize, 401 + ROUNDS * 2);
        std::fs::remove_file(paths.summary()).expect("the document");
        assert_eq!(
            served,
            RunSummary::of(&paths).expect("the run folds"),
            "two writers left the maintained row and the folded row apart"
        );
        std::fs::remove_dir_all(&root).ok();
    }

    /// A bounded listing names the roots it could not read, on the same terms
    /// the folding survey does.
    #[test]
    fn a_listing_reports_the_roots_it_could_not_read() {
        let root = scratch("listing");
        recorded(&root, "readable", 3);
        recorded(&root, "also-readable", 3);
        std::fs::create_dir_all(root.join("half-written")).expect("a run root with no launch");

        let listing = Listing::of(&root);
        let named: Vec<&str> = listing
            .summaries
            .iter()
            .map(|row| row.run_id.as_str())
            .collect();
        assert_eq!(named.len(), 2, "{named:?}");
        assert!(named.contains(&"readable"));
        let survey = crate::views::Survey::of(&root);
        assert_eq!(
            listing
                .skipped
                .iter()
                .map(|root| (root.path.clone(), root.reason.clone()))
                .collect::<Vec<_>>(),
            survey
                .skipped
                .iter()
                .map(|root| (root.path.clone(), root.reason.clone()))
                .collect::<Vec<_>>(),
            "the bounded listing and the survey report different refusals"
        );
        std::fs::remove_dir_all(&root).ok();
    }

    /// The checked-in shape of a document at [`SUMMARY_SCHEMA_VERSION`].
    ///
    /// Read rather than restated: this is the wire a consumer parses, and the
    /// only thing that stops a field being renamed, an absence becoming a zero,
    /// or the version moving without anyone deciding to move it.
    const GOLDEN: &str = include_str!("../tests/golden/run-summary-v5.json");

    /// The documents earlier builds wrote, kept exactly as those builds wrote them.
    ///
    /// What proves each growth is a version and not a quiet widening: a real
    /// schema 1 document, which the bounded listing's build grew five fields past,
    /// and a real schema 2 document, which carries no answer to whether a driver
    /// let go of its run — a real schema 3 document, written by a build that did
    /// not read a stated landing as the node's landing — and a real schema 4
    /// document, written by a build whose `stop_recorded` outlived the adoption
    /// that answered it. The reader below has to refuse all four rather than read
    /// any as one of its own.
    const GOLDEN_EARLIER: [(u32, &str); 4] = [
        (1, include_str!("../tests/golden/run-summary-v1.json")),
        (2, include_str!("../tests/golden/run-summary-v2.json")),
        (3, include_str!("../tests/golden/run-summary-v3.json")),
        (4, include_str!("../tests/golden/run-summary-v4.json")),
    ];

    /// The document the golden pins, built through the types.
    ///
    /// Every absence policy the launch record establishes is on it at least
    /// once, because the wire is where an absence either survives or becomes a
    /// zero: this run's record named no host, no pid, no stamp, and no launch
    /// instant, and it is the record 141 roots on one host actually hold.
    fn golden() -> RunSummary {
        RunSummary {
            schema_version: SUMMARY_SCHEMA_VERSION,
            run_id: "golden".into(),
            last_write_at: Some(1_786_000_000_000),
            last_event_kind: Some("node-settled".into()),
            event_count: 42,
            node_counts: BTreeMap::from([("done".to_string(), 2), ("failed".to_string(), 1)]),
            stop_recorded: false,
            graph_complete: true,
            decisions_pending: 0,
            surfaces_queued: 2,
            surfaces_read: 1,
            awaiting_human_action: false,
            project: "plans:golden".into(),
            launcher: "claude-code".into(),
            // Unattributed: the launch record named no session, which is the
            // `[unknown]` owner every view has always printed for one.
            session: String::new(),
            started_at: None,
            pid: None,
            host: None,
            started: None,
            let_go_by: Some(
                crate::projection::DriverClaim::of_stream("golden-host-4242")
                    .expect("the golden stream names a driver"),
            ),
            timing: serde_json::from_str(include_str!("../tests/golden/telemetry-v2.json"))
                .expect("the telemetry golden reads back into the types"),
            // Nothing parked, which is an absent key rather than an empty list on
            // the wire.
            parked: Vec::new(),
            judge_rejected: vec!["publish".into()],
            landings: BTreeMap::from([
                // Observed landed, and with nothing to ask again about: a base
                // does not stop carrying what it carries.
                (
                    "build".to_string(),
                    NodeLanding {
                        landing: "landed".into(),
                        branch: None,
                        repo: None,
                        drafted: false,
                    },
                ),
                // The inputs to the question, not its answer.
                (
                    "publish".to_string(),
                    NodeLanding {
                        landing: "unlanded".into(),
                        branch: Some("onepipeline/golden".into()),
                        repo: Some("nickderobertis/onepipeline".into()),
                        drafted: false,
                    },
                ),
            ]),
            journal_len: 8_192,
            journal_mtime_ms: 1_786_000_000_100,
        }
    }

    #[test]
    fn a_schema_5_document_is_the_shape_the_golden_pins() {
        let rendered = serde_json::to_string_pretty(&golden()).expect("it serialises");
        assert_eq!(
            rendered.trim(),
            GOLDEN.trim(),
            "the summary document changed shape. If that was deliberate, bump \
             SUMMARY_SCHEMA_VERSION and update tests/golden/run-summary-v5.json together"
        );
    }

    /// The document a build before the bounded listing wrote is **refused**.
    ///
    /// The whole compatibility statement, exercised on a real document rather
    /// than on a number: schema 1 carries none of the five fields a listing
    /// answers a row out of, so reading it as one of this build's would report a
    /// run with no observer, nothing parked, nothing a judge turned down, and
    /// nothing outstanding — every one of them an absence standing in for a fact
    /// nobody recorded. A refusal costs a fold and nothing else.
    #[test]
    fn the_documents_earlier_builds_wrote_are_refused_rather_than_read() {
        for (version, earlier) in GOLDEN_EARLIER {
            let refused = serde_json::from_str::<RunSummary>(earlier).expect_err("it is refused");
            assert!(
                refused
                    .to_string()
                    .contains(&format!("schema_version {version}")),
                "the refusal does not name the version it met: {refused}"
            );
        }
    }

    #[test]
    fn a_schema_5_document_round_trips_and_a_version_this_build_does_not_read_is_refused() {
        let read: RunSummary =
            serde_json::from_str(GOLDEN).expect("the golden reads back into the types");
        assert_eq!(read, golden());
        // Every absence survives the round trip as an absence, which is the one
        // thing the wire can quietly turn into a measurement.
        assert_eq!(read.started_at, None);
        assert_eq!(read.pid, None);
        assert_eq!(read.host, None);
        assert_eq!(read.started, None);
        assert!(read.session.is_empty());
        assert!(read.parked.is_empty());
        // And the one field a listing must never read as a stored verdict is on
        // the wire as the inputs it is decided from.
        assert_eq!(read.landings["build"].branch, None);
        assert_eq!(
            read.landings["publish"].branch.as_deref(),
            Some("onepipeline/golden")
        );

        // And a document from a schema this build does not read is refused
        // rather than read as one it does — which is a run that folds, not a run
        // that vanishes.
        let mut later: serde_json::Value = serde_json::from_str(GOLDEN).expect("it parses");
        later["schema_version"] = json!(SUMMARY_SCHEMA_VERSION + 1);
        let refused = serde_json::from_value::<RunSummary>(later).expect_err("it is refused");
        assert!(
            refused.to_string().contains("schema_version"),
            "the refusal does not name the version: {refused}"
        );
    }

    /// A run whose summary is a version this build does not read folds, and the
    /// fold replaces the document it could not read.
    #[test]
    fn a_summary_from_a_schema_this_build_does_not_read_folds_rather_than_vanishes() {
        let root = scratch("later-schema");
        let paths = recorded(&root, "demo", 5);
        let mut document: serde_json::Value =
            serde_json::from_str(&std::fs::read_to_string(paths.summary()).expect("the document"))
                .expect("a summary");
        document["schema_version"] = json!(SUMMARY_SCHEMA_VERSION + 1);
        std::fs::write(paths.summary(), document.to_string()).expect("a later build's summary");

        let served = RunSummary::of(&paths).expect("the run reads");
        assert_eq!(served.schema_version, SUMMARY_SCHEMA_VERSION);
        assert_eq!(served.event_count, 6);
        std::fs::remove_dir_all(&root).ok();
    }

    /// The closeout seal writes a document that accounts for the journal as it
    /// stands, whatever the incremental writers left.
    ///
    /// The state staged is the one the seal exists for and no offline journey
    /// can reach: a record on the file that no document followed — what one of
    /// two maintainers over one journal leaves when its write lands after the
    /// other's append. Before the seal the document is behind the journal; after
    /// it the document is current and records the settlement, and the listing
    /// serves it without a fold.
    #[test]
    fn the_seal_leaves_a_document_current_for_the_journal_as_it_stands() {
        let root = scratch("sealed");
        let paths = recorded(&root, "demo", 2);
        let mut journal = Journal::open(&paths);
        emit(
            &mut journal,
            PipelineKind::NodeSettled,
            Some("build"),
            "demo",
        );
        emit(
            &mut journal,
            PipelineKind::NodeSettled,
            Some("ship"),
            "demo",
        );
        let stray = event(PipelineKind::PlannerSurfaced, "demo", "another-writer", 0);
        let line = serde_json::to_string(&stray).expect("a record");
        ledger::append_line_healed(&paths.journal(), &line).expect("appended");
        let behind: RunSummary =
            serde_json::from_str(&std::fs::read_to_string(paths.summary()).expect("the document"))
                .expect("a summary");
        assert_ne!(
            (behind.journal_len, behind.journal_mtime_ms),
            journal_stamp(&paths),
            "the staged document is not behind its journal, so the seal has nothing to prove"
        );

        seal(&paths);

        let sealed: RunSummary =
            serde_json::from_str(&std::fs::read_to_string(paths.summary()).expect("the document"))
                .expect("a summary");
        assert_eq!(
            (sealed.journal_len, sealed.journal_mtime_ms),
            journal_stamp(&paths)
        );
        assert!(sealed.graph_complete, "{sealed:?}");
        assert_eq!(sealed.event_count, 6);
        let (served, bytes) = cost_of(&paths);
        assert_eq!(served, sealed);
        assert!(
            bytes < sealed.journal_len,
            "the sealed document was folded rather than served: {bytes} bytes read"
        );
        std::fs::remove_dir_all(&root).ok();
    }

    /// One record of a run's own, built by hand where the writer's own stream
    /// would not do — a relayed record, or one stamped out of order.
    fn event(kind: PipelineKind, run: &str, stream: &str, seq: u64) -> Envelope {
        Envelope {
            v: ENVELOPE_VERSION,
            ts: sys::now_rfc3339(),
            stream: stream.to_string(),
            seq,
            source: Source::Pipeline,
            kind: EventKind(kind.as_str().into()),
            dimensions: Default::default(),
            labels: Labels {
                run_id: Some(run.to_string()),
                node: Some("build".into()),
                ..Labels::default()
            },
            payload: crate::journal::payload(&[("status", json!("done"))]),
            artifacts: Vec::new(),
        }
    }
}