//! What the read-only views report.
//!
//! The views are the CLI's `runs`, `status`, `host`, `monitor`, `results`,
//! `goals`, `transcript`, and `telemetry`. The one distinction that has cost
//! real supervision time is named here as a type rather than left to prose:
//! whether a run is being driven, and if not, how it stopped being driven.
//!
//! The second is not a type but a rule: **a view never reports an unmeasured
//! thing as a measured nothing.** A dispatch that has named no tool says so; a
//! report this host cannot read says so; a bucket nothing produces is absent.
//!
//! Everything here **reads**. A view opens a run's ledger and its merged event
//! store, probes the recorded driver, and renders — and writes nothing back, so
//! rendering a run never counts as supervising it. Consuming a surface is the
//! channel's job, not a view's.
// llmlint: ignore-block[names_match_behavior] `Parked` reads as a deliberate idle, and
// for a *node* it is one — but this is the *run* liveness verdict, and `PARKED` is the
// word `docs/contract.md` fixes for it ("DRIVER DEAD vs PARKED vs UNDRIVEN"). Renaming it
// would make this crate's views disagree with the contract and with the operators who
// already read that word off them; the collision is exactly what the doc comment below
// exists to disarm. Raise it with the planner who owns the contract, not here.
/// Whether a run is being driven, and if not, why not.
///
/// The three are deliberately distinct. A run whose *driver* died is not lost —
/// its ledger is intact and `onepipeline adopt` attaches a fresh driver to it —
/// while a parked node is a planner's own deliberate idle and nothing to
/// intervene in. Reading one as the other is what this distinction exists to
/// prevent.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum DriverLiveness {
/// A driver holds the run and this host has observed it working.
Driving,
/// This host has proved the recorded driver process is gone, or the run's own
/// journal records that driver letting go of it to fire its run-end hook.
/// Nothing is driving the run; `adopt` is the way back.
DriverDead,
/// The launch still holds its recorded pid, but nothing is happening — no
/// child process, no surface, no ledger write. Alive and not working, so
/// treat it as stopped and intervene.
Parked,
/// A *node* the ledger records as started that nothing is driving.
/// Deliberately not [`Parked`](Self::Parked), which is the state a planner's
/// own `cancel` produces.
Undriven,
}
// llmlint: ignore-end[names_match_behavior]
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use crate::event::{Envelope, Source};
use crate::filter::EventFilter;
use crate::graph::{self, Landing, NodeStatus};
use crate::journal::PipelineKind;
use crate::ledger::{self, LaunchRecord};
use crate::projection::{MemberLabel, Refusal, RunState, ServiceRecord};
use crate::rendercost::Rendered;
use crate::report::{ToolText, Truncation};
use crate::sys;
use crate::vcs::LandingRead;
/// A run root a view refused, and the reason it gave.
///
/// Re-exported where the views are, because a rejection is part of what a view
/// reports: a root that was skipped is named on the same output a run is.
pub use crate::ledger::Skipped;
/// Where one run's durable state lives.
///
/// Re-exported where the views are, because it is the type a consumer already
/// receives on [`RunView::paths`] and could not name — and because
/// [`report_for`](RunPaths::report_for) is how a reader of this run's store
/// resolves the copy [`report::retain`](crate::report::retain) wrote. What the
/// contract promises of it is `run`, `dir`, [`new`](RunPaths::new),
/// [`under`](RunPaths::under), [`reports_dir`](RunPaths::reports_dir), and
/// `report_for`; the segment sanitiser behind the last of those stays private,
/// so a report path is obtained by calling and never by restating.
// llmlint: ignore[invalid_states_unrepresentable] naming the type changes nothing about a run id: `run` is a `String` for the reason `src/ledger.rs`'s file-level suppression states, and `ledger::is_valid_run_id` remains the boundary every externally-supplied id crosses.
pub use crate::ledger::RunPaths;
/// One run, as a bounded listing reads it.
///
/// Re-exported where the views are, because it is a **view**: it answers the
/// questions `runs` and `status` answer, in the structure a consumer building
/// its own listing needs, and at a cost that does not grow with the run's
/// journal. `runs` and `status` given no run **are** this read; the entry points
/// that fold are the detail reads beside them.
///
/// [`NodeLanding`] comes with it because [`RunSummary::landings`] carries one per
/// node: a field a consumer cannot name is a field it cannot read.
pub use crate::summary::{Listing, NodeLanding, RunSummary, SUMMARY_SCHEMA_VERSION};
/// What says a run is being **watched**, and what a reader makes of it now.
///
/// Re-exported where the views are, because deciding whether anything is watching
/// a run is a read of the same kind as deciding whether anything is driving it —
/// and a consumer that cannot name the record cannot render a run's watchers.
/// [`Watchers::of`] is how the records are reached: the directory they sit in
/// stays private, so a reader asks rather than composing a path.
///
/// The asymmetry is **inverted** against [`DriverLiveness`], deliberately, and
/// `src/watchers.rs` states why: for a driver the worse error is reporting live
/// work as dead, and for a watch it is reporting a run as watched while nothing is
/// watching it.
pub use crate::watchers::{Watch, WatchStanding, WatcherRecord, Watchers, WATCHER_SCHEMA_VERSION};
/// One run's timing and usage, with a breakdown that sums exactly to its wall
/// clock.
///
/// Re-exported here because [`RunSummary::timing`] carries one: the summary
/// **references** this type rather than restating its fields, so there is one
/// declaration of what a run's clock is rather than two to drift apart — and a
/// field a consumer cannot name is a field it cannot read. The five types are
/// what one document is made of, and `telemetry [--breakdown]` renders exactly
/// this.
pub use crate::telemetry::{Bucket, BucketName, Party, RunTelemetry, Usage};
/// One settlement write-back projection attempt, as a run records it.
///
/// Re-exported where the views are, because it is read the way a view is: what a run's
/// projections carried, how each ended and what it spent, off the run's own directory rather
/// than off a store. The six types beside it are what one line is made of — a field a
/// consumer cannot name is a field it cannot read — and entry 73 of
/// `docs/contract-divergences.md` is the one statement of the line's shape.
pub use crate::writeback::{
FailureClass, ProjectionActions, ProjectionEnded, ProjectionFailure, ProjectionRecord,
ProjectionScope, WholeBecause,
};
/// How long a launch may hold its pid without doing anything before it is
/// reported [`Parked`](DriverLiveness::Parked).
///
/// The default planner-update interval: a run that has not written, surfaced,
/// or dispatched for a whole interval is not merely between turns.
pub const DEFAULT_PARKED_AFTER_SECONDS: u64 = 1_800;
/// The environment variable that moves that threshold.
pub const PARKED_AFTER_ENV: &str = "ONEPIPELINE_PARKED_AFTER_SECONDS";
/// How a node whose dispatch a `stop` ended is reported.
///
/// One phrasing, in the two views that report an in-flight node, because they
/// are read together and a run that says two things about one node is a run
/// nobody trusts. It says what happened to the *worker* rather than what the
/// node produced: a stop ends the run's whole dispatch tree, and the process
/// that would have settled the node was in that tree — so the last thing the
/// record holds for it is that it started, and a reader with nothing else to go
/// on takes that for a node that produced nothing.
const ENDED_BY_THE_STOP: &str = "worker ended when the run was stopped";
/// How the same node reads when nothing established what became of its worker.
///
/// It has to be a different sentence: the worker is very likely still running,
/// and "ended" there is the false completion `stop` itself refuses to report.
const OUTLIVED_THE_STOP: &str = "worker may still be running: the stop could not reach it";
/// Which of the two a stopped run's in-flight node gets.
fn became_of_the_worker(state: &crate::projection::RunState) -> &'static str {
match state.stop {
crate::projection::StopState::WorkersUndetermined => OUTLIVED_THE_STOP,
_ => ENDED_BY_THE_STOP,
}
}
/// Whether the run's **observer** graph is still watching, and if not, why not.
///
/// Beside [`DriverLiveness`] rather than inside it: the two are about different
/// processes and a run can be in any pairing of them, and a live driver
/// executing unwatched is the state this exists to report. Private, because
/// `docs/contract.md` names the driver tier and this is a rendering beside it.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
enum ObserverLiveness {
/// The launch named an observer graph and nothing says its run has ended.
Watching,
/// The launch named one and this host can prove that graph run is over. The
/// run is executing with nothing watching it — for now: a driver that
/// notices starts another, so this is what the window between them reads as.
ObserverDead,
/// The launch named one, its driver has stopped starting another, and the
/// record says why. Nothing is watching this run and nothing is going to,
/// which is the state an operator has to act on rather than wait out.
ObserverNotRestarted,
/// The launch named no observer graph at all — the shipped default, since no
/// agent is required to execute a plan. Nothing is watching this run either,
/// and nothing ever was: a different fact, and a different fix.
Unobserved,
}
impl ObserverLiveness {
/// The word a view prints beside the driver tier. Nothing for an observer
/// doing its job — which is the verdict [`observer_suffix`] renders as no
/// suffix at all rather than by printing this.
fn as_str(self) -> &'static str {
match self {
Self::Watching => "",
Self::ObserverDead => "OBSERVER DEAD",
Self::ObserverNotRestarted => "OBSERVER NOT RESTARTED",
Self::Unobserved => "NO OBSERVER",
}
}
}
/// Whether anything is watching this run.
///
/// The launch record answers the first half — whether there is an observer at
/// all — and names the graph run whose own liveness is the second, which
/// [`agentgraph::graph_run_ended`] decides and documents.
///
/// [`agentgraph::graph_run_ended`]: crate::agentgraph::graph_run_ended
fn observer_liveness(launch: &LaunchRecord) -> ObserverLiveness {
watching(
launch.observer_graph(),
&launch.graph_run,
&launch.observer_ending,
&launch.run_id,
)
}
/// The same verdict, over the three things a record says about the observer.
///
/// Taken apart from the record because two documents carry them — the launch
/// record, and the summary document a listing reads instead of it — and one
/// reading over both is what stops a run reading `NO OBSERVER` on one view and
/// `ACTIVE` on the other.
fn watching(graph: Option<&str>, graph_run: &str, ending: &str, run: &str) -> ObserverLiveness {
if graph.is_none() {
return ObserverLiveness::Unobserved;
}
// The driver's own account first, because it is the stronger evidence: a
// driver that has stopped starting another observer *knows* the run will not
// be watched again, where the probe below reports only what this host can
// prove and answers everything else with "still watching".
if !ending.is_empty() {
return ObserverLiveness::ObserverNotRestarted;
}
if crate::agentgraph::graph_run_ended(graph_run, run) {
return ObserverLiveness::ObserverDead;
}
ObserverLiveness::Watching
}
impl DriverLiveness {
/// The word a view prints for this verdict.
pub fn as_str(self) -> &'static str {
match self {
Self::Driving => "ACTIVE",
Self::DriverDead => "DRIVER DEAD",
Self::Parked => "PARKED",
Self::Undriven => "UNDRIVEN",
}
}
/// Whether this verdict means nothing is driving the run.
///
/// `adopt` is the way back from both of the two that do.
pub fn is_undriven(self) -> bool {
matches!(self, Self::DriverDead | Self::Parked)
}
}
/// How long a launch may be silent before it is reported parked.
pub fn parked_after_seconds() -> u64 {
std::env::var(PARKED_AFTER_ENV)
.ok()
.and_then(|value| value.parse().ok())
.filter(|seconds| *seconds > 0)
.unwrap_or(DEFAULT_PARKED_AFTER_SECONDS)
}
/// Whether a **decision point** is outstanding, in either of the two forms one
/// takes: a ready human action nobody has attested, or a blocking surface nobody
/// has answered.
///
/// The one question every verdict about a stalled run asks, so the settlement and
/// the liveness verdict cannot disagree about the same run — a run reported
/// `PARKED` invites an `adopt` that may end its driver, and doing that to a run
/// whose next move is already sitting in a planner's queue costs the work it
/// holds for nothing.
pub fn decision_outstanding(state: &RunState, paths: &RunPaths) -> bool {
state.awaiting_human_action() || blocking_surface(paths)
}
/// Whether a blocking surface is outstanding, read or not.
///
/// Unread counts. A question nobody has looked at is still a question the run is
/// waiting on, and a verdict that ignored it would report the run as abandoned —
/// sending an operator to intervene in a run whose next move is already sitting
/// in their own queue.
///
/// A surface nobody is waiting on does not count, and it is the one case where
/// the reasoning above runs out: the reader that asked the question has gone, so
/// there is no next move sitting in anybody's queue, and a run reported as
/// waiting for a planner would wait for ever.
pub(crate) fn blocking_surface(paths: &RunPaths) -> bool {
let queue = crate::channel::ChannelState::new(paths).queue();
queue
.waiting
.iter()
.chain(queue.pending.iter())
.any(|surface| surface.blocking && !surface.abandoned)
}
/// Whether a run is being driven, and if not, why not.
///
/// Every unreadable input resolves toward "still working", so a busy driver is
/// never misreported: one live process, one fresh surface, or one recent ledger
/// write is enough to keep it reported as running. A pid recorded on another
/// host is exactly such an unknown — a pid means nothing across machines — so a
/// run another driver is holding reads as the live work it is.
///
/// **A record that names no driver at all is that same unknown.** A record
/// written before those keys existed defaults to pid `0` on a nameless host, and
/// a reader that probed `0` would prove the absence it was handed and report
/// `DRIVER DEAD` over a run it knows nothing about — so the pid and the host are
/// each read through the accessor that turns the record's silence back into
/// silence rather than into an answer.
pub fn liveness(launch: &LaunchRecord, state: &RunState, paths: &RunPaths) -> DriverLiveness {
driver_liveness(
state.stop_recorded()
|| let_go(
launch.recorded_host(),
launch.driver_pid(),
state.let_go_by.as_ref(),
),
launch.recorded_host(),
launch.driver_pid(),
launch.driver_stamp(),
state.last_write_at,
state.awaiting_human_action(),
paths,
)
}
/// Whether the driver a record names is the one a run's journal records letting
/// go of the run to fire its run-end hook.
///
/// That driver is still a live process while it awaits the hook, and it holds
/// nothing: the ownership lock is released before the hook starts. So it is
/// matched by what the journal stamped — the host and pid of the stream that
/// wrote the firing — against the claim the record makes, and a driver that
/// adopted the run since is a different claim.
pub(crate) fn let_go(
recorded_host: Option<&str>,
pid: Option<std::num::NonZeroU32>,
let_go_by: Option<&crate::projection::DriverClaim>,
) -> bool {
let_go_by.is_some_and(|claim| claim.is(recorded_host, pid))
}
/// The same verdict, over the six things a record says and the run's own
/// channel.
///
/// Taken apart from the fold for the reason [`watching`] is: the bounded
/// listing reads these six off a summary document rather than off a whole
/// merged store, and a second implementation of this reading is a second run
/// liveness to keep true.
///
/// The pid is read with the stamp beside it, through the same
/// [`sys::claim_on`] a stop aims by; read alone it called a reissued pid a
/// live driver.
///
/// `undriven_by_record` is what the run's own record already says without asking
/// the host: a stop was recorded, or the driver the record names let go of the
/// run to fire its run-end hook — that driver is not driving, however alive it
/// is. See [`let_go`].
fn driver_liveness(
undriven_by_record: bool,
recorded_host: Option<&str>,
pid: Option<std::num::NonZeroU32>,
started: Option<&str>,
last_write_at: Option<u64>,
awaiting_human_action: bool,
paths: &RunPaths,
) -> DriverLiveness {
if undriven_by_record {
return DriverLiveness::DriverDead;
}
if driver_claim_is_over(recorded_host, pid, started) {
return DriverLiveness::DriverDead;
}
// A live pid is ownership, not progress.
//
// A run holding an outstanding decision point is *waiting*, not parked: the
// loop that would be writing is deliberately holding a subtree back until a
// person answers, and a driver reported dead there sends an operator to
// intervene in work that is doing exactly what it should.
if quiet_past_parked(last_write_at) && !(awaiting_human_action || blocking_surface(paths)) {
return DriverLiveness::Parked;
}
DriverLiveness::Driving
}
/// Whether the driver a record names is proved, on this host, to be over.
///
/// A pid recorded on another host means nothing here, so it is never over.
/// Apart from [`driver_liveness`] because a watch asks it between reads: a
/// driver ending moves no file, so what a wait watches for has to ask the host.
pub(crate) fn driver_claim_is_over(
recorded_host: Option<&str>,
pid: Option<std::num::NonZeroU32>,
started: Option<&str>,
) -> bool {
recorded_host == Some(sys::hostname().as_str())
&& pid.is_some_and(|pid| sys::claim_on(pid.get(), started.unwrap_or_default()).is_over())
}
/// Whether a run last written at `last_write_at` has been quiet past
/// [`parked_after_seconds`] — the other half of a liveness that moves with the
/// clock rather than with a file.
pub(crate) fn quiet_past_parked(last_write_at: Option<u64>) -> bool {
last_write_at
.is_some_and(|last| sys::now_millis().saturating_sub(last) / 1_000 > parked_after_seconds())
}
/// Take the landings the run's **own settled report** recorded, where they say
/// more than its journal does.
///
/// A driver re-reads every unlanded change as it closes out — `engine`'s
/// `landings_after_asking_again` — and writes what it found there; without this
/// the report a consumer parses said `landed` while the view an operator opens
/// said `NOT landed`.
///
/// **Only ever `unlanded` → `landed`.** A report written at an earlier close-out
/// is older than the journal beside it, and a run still going has one — but a
/// base does not stop carrying what it carries, so taking the later answer in
/// that direction alone cannot un-land anything.
pub(crate) fn landings_the_run_re_read(state: &mut RunState, paths: &RunPaths) {
// Read leniently: a report this build cannot parse — one a newer build
// wrote, at a version this one refuses — leaves the view 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;
};
for node in result.nodes {
if node.landing == Some(Landing::Landed)
&& state.landings.get(&node.id) == Some(&Landing::Unlanded)
{
state.landings.insert(node.id, Landing::Landed);
}
}
}
/// One render of one view, and the scope every landing read it takes is asked
/// once inside.
///
/// Two things at once, because they are the same scope: what a render costs is
/// counted against it — see [`crate::rendercost`] — and the memo below is
/// emptied when it opens, so **a node not already recorded landed is asked again
/// on every render**. That is the point: the two answers a later merge changes
/// are exactly `not landed` and `could not be decided`, and a view that reused
/// yesterday's would be the stale claim this exists to stop making.
///
/// Nested scopes are one render: `status` prints [`RunView::summary`] inside
/// itself, and a second scope there would empty the memo half-way through and
/// ask every node twice.
struct Rendering(crate::rendercost::Render);
impl Rendering {
/// Record that this render reported on one node's landing, which is the set
/// every read it takes is held against.
fn reported(&self, node: &str) {
self.0.reported(node);
}
}
thread_local! {
/// Every landing this render has already read, keyed by run and node.
///
/// A render asks the sibling **at most once for each node whose line it
/// prints**: `status` reports the same node on its own line and inside the
/// summary it prints above it, and two reads of one node would double what a
/// supervisory look costs to say one thing.
//
// llmlint: ignore[invalid_states_unrepresentable] a run id and a node id are the plain strings this whole crate spells them as — `RunPaths.run`, `RunState::landings`, `RunState::branches` — for the reason `src/ledger.rs`'s file-level suppression records, and `src/AGENTS.md` names a `RunId` newtype as interface drift. Newtypes at these two sites alone would disagree with every neighbour and convert at every boundary. Neither value is unchecked: the run id crossed `ledger::is_valid_run_id` when the run was launched, and the node id is one the run's own journal projected into its graph.
static READ_THIS_RENDER: std::cell::RefCell<BTreeMap<(String, String), std::rc::Rc<LandingRead>>> =
const { std::cell::RefCell::new(BTreeMap::new()) };
}
/// Begin rendering `view` over `run`.
fn rendering(view: Rendered, run: &str) -> Rendering {
let render = crate::rendercost::rendering(view, run);
if render.outermost() {
READ_THIS_RENDER.with_borrow_mut(BTreeMap::clear);
}
Rendering(render)
}
/// Where one node's change stands, as the render printing it can tell.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Stands {
/// On its base.
Landed,
/// There is work on its branch the base does not carry.
NotLanded,
/// Nothing this host can read decides it — which is neither of the other
/// two, and is reported as itself.
Undecided,
/// The node never had a change to land, so nothing is said about one.
NoChangeToLand,
}
/// What the settlement recorded, for the answers a later read can still move.
///
/// `landed` is deliberately not one of them: that answer is
/// [`Reported::TheRunObserved`] and is never read again, so a settlement value
/// carried alongside a read cannot be it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Settled {
/// The publication answered `unlanded`: there was a change, and it had not
/// reached the base when the node settled.
Unlanded,
/// The publication answered no landing at all — a node that failed, or one
/// whose base already carried its content.
Nothing,
}
/// What a render reports about one node's landing.
///
/// Four cases rather than two records beside each other, because the pairs that
/// would be contradictory are the ones a reader acts on: a landing the run
/// observed is never re-read, so it can carry no later read to disagree with and
/// no settlement value to contradict.
enum Reported {
/// The run itself observed the change reach its base. **Never re-read**: a
/// base does not stop carrying work it has taken, so the one read that could
/// move this could only agree with it, at a cost.
TheRunObserved,
/// An operator settling the node from evidence stated where its work landed.
/// **Never re-read** either, and above all not by reading the branch the
/// dispatch left behind: that branch is the record the settle corrected, so a
/// read of it could only contradict the statement it was superseded by.
Stated(crate::edits::StatedLanding),
/// A read taken while this view rendered, over what the settlement recorded.
ReadNow {
settled: Settled,
read: std::rc::Rc<LandingRead>,
},
/// The settlement said `unlanded` and there is no branch to ask about, so
/// its own claim is all there has ever been and it stands, dated.
UnlandedAtSettlement,
/// The node never had a change to land.
NoChangeToLand,
}
impl Reported {
fn stands(&self) -> Stands {
match self {
Reported::TheRunObserved | Reported::Stated(_) => Stands::Landed,
Reported::ReadNow { read, .. } => match read.as_ref() {
LandingRead::Answered(landed) => match landed {
// A landing the branch has since gone past is not this
// node's work finished: the sibling's own `is_landed` says
// so, and the commits above it are work still to publish.
onevcs::Landed::Yes { .. } => Stands::Landed,
onevcs::Landed::InPart { .. } | onevcs::Landed::No => Stands::NotLanded,
// Deliberately not a `no`: the base carries the change and
// nothing records why, which is equally what somebody else
// making the same change leaves behind.
onevcs::Landed::Unknown => Stands::Undecided,
},
LandingRead::Refused { .. } => Stands::Undecided,
},
Reported::UnlandedAtSettlement => Stands::NotLanded,
Reported::NoChangeToLand => Stands::NoChangeToLand,
}
}
}
/// What this render says about one node's landing, reading now wherever a read
/// could change the answer.
///
/// **Asked for every node whose line the render prints, and for no other** — the
/// render is told which those are as it prints them, so a read taken for a node
/// nobody is shown fails the bound `tests/e2e/landing.rs` holds.
fn reported_landing(view: &RunView, render: &Rendering, node: &str) -> Reported {
reported_landing_of(
&view.paths.run,
render,
node,
view.state.landings.get(node).copied(),
view.state.stated_landings.get(node),
view.state.branches.get(node).map(String::as_str),
view.state
.graph
.get(node)
.and_then(|node| node.repo.as_deref()),
)
}
/// The same report, over the four things a record says about one node's change.
///
/// Taken apart from the fold for the reason [`watching`] and [`driver_liveness`]
/// are: a bounded listing reads the settlement's own landing, the branch it
/// published, and the repository to narrow to off a summary document, and the
/// *decision* has to be the same decision either way — a second one would be a
/// second answer to "has this landed?" for a reader to meet.
fn reported_landing_of(
run: &str,
render: &Rendering,
node: &str,
settled: Option<Landing>,
stated: Option<&crate::edits::StatedLanding>,
branch: Option<&str>,
repo: Option<&str>,
) -> Reported {
render.reported(node);
// Ahead of everything else, because the statement is what the record now
// says: it is the correction of the very branch a read would be taken of.
if let Some(stated) = stated {
return Reported::Stated(stated.clone());
}
if settled == Some(Landing::Landed) {
return Reported::TheRunObserved;
}
// The branch the dispatch reported, which is the only spelling of this work
// the sibling can resolve, and the repository the question is narrowed to. A
// node that settled without a branch — every node that never published, and
// every node still in flight — has nothing to ask about; a node that names
// no repository publishes nothing at all, so it has no change to land and
// asking after one would search every identity this host knows for a branch
// no repository of theirs was ever asked to carry.
let asked = branch.zip(repo);
let settled = match settled {
Some(Landing::Unlanded) => Settled::Unlanded,
// `landed` is unreachable here: it returned above, which is what makes
// [`Settled`] the two answers a read can still move.
Some(Landing::Landed) | None => Settled::Nothing,
};
match (asked, settled) {
(Some((branch, repo)), settled) => Reported::ReadNow {
settled,
read: read_now(run, node, branch, repo),
},
(None, Settled::Unlanded) => Reported::UnlandedAtSettlement,
(None, Settled::Nothing) => Reported::NoChangeToLand,
}
}
/// One node's landing, read now — or handed back from the read this render
/// already took of it.
fn read_now(run: &str, node: &str, branch: &str, repo: &str) -> std::rc::Rc<LandingRead> {
let key = (run.to_owned(), node.to_owned());
if let Some(read) = READ_THIS_RENDER.with_borrow(|reads| reads.get(&key).cloned()) {
return read;
}
// Everything the decision costs is counted against this node, which is what
// makes "no other per-node work" a measurement rather than a promise.
let read = {
let _deciding = crate::rendercost::deciding(node);
std::rc::Rc::new(crate::vcs::landing_now(branch, Some(repo)))
};
READ_THIS_RENDER.with_borrow_mut(|reads| {
reads.insert(key, std::rc::Rc::clone(&read));
});
read
}
/// Everything a view needs about one run, read once.
#[derive(Debug)]
pub struct RunView {
/// Where the run's state lives.
pub paths: RunPaths,
/// Who launched it, and with what.
pub launch: LaunchRecord,
/// Its merged event store, in merge order.
pub events: Vec<Envelope>,
/// What the journal says about it.
pub state: RunState,
}
impl RunView {
/// Read one run, or report why it cannot be read.
pub fn open(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 launch: LaunchRecord = ledger::read_json(&paths.launch())?;
let mut events = crate::journal::read(&paths.journal());
crate::journal::merge_order(&mut events);
// Resumed from the run's checkpoint rather than folded from nothing: the
// events above are the merged store this view hands to its caller, and what
// no longer grows with them is the fold over them. See
// [`crate::checkpoint`].
let mut state = crate::checkpoint::fold_and_checkpoint(paths);
landings_the_run_re_read(&mut state, paths);
// A view resolves cross-DAG edges the same way the loop does, so a
// consumer this run is about to dispatch is not reported blocked to the
// person deciding whether to intervene. Reading only: rendering a run
// records nothing about it.
state.cross_dag = crate::crossdag::resolve_quietly(
&paths
.dir
.parent()
.map_or_else(ledger::runs_root, Path::to_path_buf),
&state.graph,
);
Ok(Self {
paths: paths.clone(),
launch,
events,
state,
})
}
/// How the run is being driven.
pub fn liveness(&self) -> DriverLiveness {
liveness(&self.launch, &self.state, &self.paths)
}
/// The surfaces nobody has read yet, and how stale the oldest is.
///
/// These are the state a planner who never attached is blind to: the row
/// above says only `ACTIVE`, and the delivery record they would look for is
/// written on consumption, which has not happened.
pub fn unread_surfaces(&self) -> (usize, Option<u64>) {
let unread = self.unread();
(unread.count, unread.oldest_seconds)
}
/// The same read, with what the queue is holding as well as how much.
///
/// Crate-visible and behind the pair above: the count and the staleness are
/// what the contract names, and which kinds are waiting is a rendering — see
/// [`Unread`] for why the line carries it.
pub(crate) fn unread(&self) -> Unread {
Unread::of(&crate::channel::ChannelState::new(&self.paths).queue())
}
/// A one-line summary of where the run has got to.
///
/// `n/n done` is the line a planner reads to decide a run is finished, and
/// on its own it is the false completion this crate exists to stop
/// reporting: every node can be done while every change is sitting in a pull
/// request nobody merged. So the count of what has not landed rides the same
/// line, and is absent — rather than a zero — when there is nothing to say.
///
/// Read **as this line renders**, rather than as each node settled: a change
/// that reached its base after the node that made it settled is not work
/// nobody landed, and a count that said it was is what sent a supervisor to
/// re-dispatch nodes whose work was already on the base. A landing this host
/// cannot decide is counted under its own word rather than folded into
/// either of the other two, because "not landed" is a claim and an undecided
/// read makes none.
// llmlint: ignore[contracts_have_one_source_or_a_drift_gate] the second source is `docs/contract-divergences.md` entry 33, and the gate reconciling it is `tests/contract.rs`: it refuses an entry that is neither marked resolved with the contract naming what its ruling adopted, nor marked open with the proposal it waits on. Entry 33 is open and states this exact proposal, because the contract is approved verbatim and amending it is the planner's, never this repository's.
pub fn summary(&self) -> String {
let render = rendering(Rendered::Summary, &self.paths.run);
let statuses = self.state.statuses();
let counted = |wanted: NodeStatus| statuses.values().filter(|s| **s == wanted).count();
tally(
counted(NodeStatus::Done),
statuses.len(),
counted(NodeStatus::Skipped),
&outstanding_landings(self, &render),
)
}
}
/// The `n/m done` line, from the counts behind it.
///
/// One phrasing, because two readers meet it: the folding views build the counts
/// out of a run's whole store and the bounded listing builds them out of that
/// run's summary document, and a line worded twice is a line that drifts.
fn tally(done: usize, total: usize, skipped: usize, outstanding: &Outstanding) -> String {
let unlanded = match outstanding.not_landed.len() {
0 => String::new(),
count => format!(", {count} not landed"),
};
let undecided = match outstanding.undecided.len() {
0 => String::new(),
count => format!(", {count} landing undecided"),
};
// What is *missing* from the count above splits two ways, and only one of
// them is work that was attempted: `n/n done` on its own left a reader
// unable to tell a node the run tried and lost from one it never asked at
// all. Absent rather than a zero, like the clause before it.
let skipped = match skipped {
0 => String::new(),
count => format!(", {count} never attempted"),
};
format!("{done}/{total} done{unlanded}{undecided}{skipped}")
}
/// What one run's unread surfaces are, as the one line reporting them needs them.
///
/// A blocking surface produces no other signal, and on a host holding thousands
/// of routine observer updates against a handful of questions a bare count read
/// the same either way.
#[derive(Debug, Default)]
pub(crate) struct Unread {
pub(crate) count: usize,
/// Absent when nothing is waiting, rather than a zero that reads as a queue
/// somebody has just emptied.
pub(crate) oldest_seconds: Option<u64>,
/// Blocking kinds first — that is what the run is held on — then rarest
/// first, since a rare kind behind a common one is the burial this repairs;
/// then by name, so the line is stable to read.
pub(crate) kinds: Vec<(String, usize)>,
/// How many surfaces are in the queue that nobody is waiting on: their
/// reader exited without an answer. Counted apart from [`count`](Self::count)
/// rather than inside it — the count above is the one line a supervisor may
/// not filter, and an entry nothing will ever answer degrades it — and
/// reported all the same, because their text is still there to read.
pub(crate) abandoned: usize,
}
/// How many kinds a line names before it summarises the rest.
///
/// The remainder is counted out loud rather than dropped: a silently truncated
/// list reads as the whole answer.
const MAX_NAMED_KINDS: usize = 4;
impl Unread {
pub(crate) fn of(queue: &crate::channel::Queue) -> Self {
let mut counts: BTreeMap<String, (bool, usize)> = BTreeMap::new();
let unread = || queue.waiting.iter().filter(|surface| !surface.abandoned);
for surface in unread() {
// The kind is a stranger's: an observer's frame names it in that
// persona's own vocabulary, so it goes through the same strip every
// other borrowed value on these views does.
let seen = counts.entry(one_line(&surface.kind)).or_insert((false, 0));
seen.0 |= surface.blocking;
seen.1 += 1;
}
let mut ordered: Vec<(bool, usize, String)> = counts
.into_iter()
.map(|(kind, (blocking, count))| (blocking, count, kind))
.collect();
ordered.sort_by(|a, b| {
b.0.cmp(&a.0)
.then(a.1.cmp(&b.1))
.then_with(|| a.2.cmp(&b.2))
});
Self {
count: unread().count(),
oldest_seconds: unread()
.map(|surface| sys::now_millis().saturating_sub(surface.queued_at) / 1_000)
.max(),
kinds: ordered
.into_iter()
.map(|(_, count, kind)| (kind, count))
.collect(),
abandoned: queue
.waiting
.iter()
.filter(|surface| surface.abandoned)
.count(),
}
}
/// The kinds, bounded: past [`MAX_NAMED_KINDS`] the line says how many it
/// left out rather than ending where a reader cannot tell it was cut.
pub(crate) fn phrase(&self) -> String {
let named: Vec<String> = self
.kinds
.iter()
.take(MAX_NAMED_KINDS)
.map(|(kind, count)| format!("{count} {kind}"))
.collect();
let rest = match self.kinds.len().saturating_sub(named.len()) {
0 => String::new(),
more => format!(", and {more} other kind(s)"),
};
format!("{}{rest}", named.join(", "))
}
}
/// What a view over a whole runs root read, and what it refused.
///
/// The second half is why this type exists. A view that opened every run it
/// could and silently dropped the rest reported a rejection as an *absence*: a
/// host with thirty run roots on it rendered as `no runs recorded`, which a
/// planner reads as "nothing is running". A refused root is a fact about the
/// root, and it is carried to the renderer rather than thrown away in the read.
#[derive(Debug)]
pub struct Survey {
/// The runs root this survey read. Named on the output, because it is the
/// scope of every claim made from it.
pub root: PathBuf,
/// The runs that read, oldest id first.
pub views: Vec<RunView>,
/// The run roots that did not, each with the reason it was refused.
pub skipped: Vec<Skipped>,
}
impl Survey {
/// Read every run under a root, keeping what could not be read.
///
/// A root the ledger refused and a run whose launch record this build cannot
/// accept 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 views = Vec::new();
let mut skipped = index.skipped;
for paths in index.runs {
match RunView::open(&paths) {
Ok(view) => views.push(view),
// The refusal as `results` already words it: the file, the
// offending field, and what was expected. 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(),
}),
}
}
skipped.sort_by(|a, b| a.path.cmp(&b.path));
Self {
root: root.to_path_buf(),
views,
skipped,
}
}
/// The one run a caller named, surveyed on its own.
///
/// It was opened by name, so a root it did not read is not this survey's to
/// report: a caller who named a run that could not be read was refused
/// outright rather than handed a view of it.
pub fn of_one(view: RunView) -> Self {
let root = view
.paths
.dir
.parent()
.map_or_else(ledger::runs_root, Path::to_path_buf);
Self {
root,
views: vec![view],
skipped: Vec::new(),
}
}
}
// llmlint: ignore-block[cli_output_contract] a refused run root is part of the answer, not
// a failure of the command: the empty case here *replaces* `no runs recorded`, so it cannot
// live on a stream other than the answer it replaces. The two driver sites that print these
// carry the exit code that goes with the same decision.
/// How many refused run roots a view names before it summarises the rest.
///
/// The remainder is counted out loud rather than dropped, exactly as
/// [`MAX_NAMED_KINDS`] counts the surface kinds it did not name: a silently
/// truncated list reads as the whole answer.
const MAX_NAMED_SKIPS: usize = 3;
/// What a view says about the run roots it refused, or nothing when it refused
/// none.
///
/// **One line: a count, and as many reasons as fit on it.** A count alone tells
/// a reader something is wrong and not which directory to look at, so a few
/// roots are named with the reason `results` prints for the same refusal — and a
/// line each is what buried the answer, because a third of the roots on a host
/// can be unreadable and the live verdict is what a reader opened the view
/// for.
fn skipped_lines(skipped: &[Skipped]) -> String {
if skipped.is_empty() {
return String::new();
}
// Every value on the line is a stranger's — a directory name on disk and a
// refusal built from it — so both go through the same strip.
let named: Vec<String> = skipped
.iter()
.take(MAX_NAMED_SKIPS)
.map(|root| {
format!(
"{}: {}",
one_line(&root.path.display().to_string()),
one_line(&root.reason)
)
})
.collect();
let rest = match skipped.len().saturating_sub(named.len()) {
0 => String::new(),
more => format!(", and {more} more"),
};
format!(
"{} run root(s) skipped: {}{rest}\n",
skipped.len(),
named.join("; ")
)
}
/// What a whole-root view says when it has no run to report.
///
/// Two different facts, which is the whole reason this is a function: a root
/// with nothing in it and a root whose every run was refused both rendered as
/// `no runs recorded`, and only one of them means there is nothing running.
fn nothing_to_report(survey: &Survey) -> String {
let mut out = if survey.views.is_empty() && !survey.skipped.is_empty() {
format!(
"no run under {} could be read\n",
one_line(&survey.root.display().to_string())
)
} else {
"no runs recorded\n".to_string()
};
out.push_str(&skipped_lines(&survey.skipped));
out
}
// llmlint: ignore-end[cli_output_contract]
/// Where a run stands: what is left for a driver to do, and whether anything is
/// doing it.
///
/// One reading behind both halves of what a view says about a run — the word on
/// its row and the advice beneath it — so no combination of graph state and
/// driver state can make the two disagree. Read apart, they did: a run whose
/// graph had completed printed `SETTLED` and, directly under it, an invitation
/// to attach a fresh driver to finished work.
struct Standing {
/// How the run is being driven.
liveness: DriverLiveness,
/// What remains for a driver or planner to do.
work: WorkStanding,
/// Whether the loop has anything left to converge.
///
/// Carried beside [`work`](Self::work) rather than read off it, because the
/// two answer different questions and one of the readings is not a function
/// of the other: a converged run holding a ready human action is
/// [`WorkStanding::Outstanding`] — an `attest` moves it — and it has still
/// settled, its driver having written its result and gone.
convergence: Convergence,
}
/// Whether every node of a run's graph reached a state the loop is finished
/// with, so no further pass is coming.
///
/// A named verdict rather than a bare flag, because it is what
/// [`has_settled`] answers and what the channel refuses a reply on: read as the
/// wrong polarity it queues a verdict where nothing drains it, or turns away the
/// one reply a live run was waiting for.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Convergence {
/// Nothing is left to converge: the run has settled.
Settled,
/// Something can still move, with or without a driver attached to move it.
Moving,
}
impl Convergence {
/// The verdict for a graph whose nodes have all settled, or not.
fn of(converged: bool) -> Self {
if converged {
Self::Settled
} else {
Self::Moving
}
}
}
/// One or more nodes an operator has to act on before a driver can move the run.
///
/// The invariant is the advice itself: a standing that says work is held names
/// the work it means, and a list that could be empty would put a prescription
/// naming nothing on an operator's screen.
struct HeldNodes(Vec<String>);
impl HeldNodes {
/// The nodes, or `None` where there are none — and so no standing to report.
fn of(nodes: Vec<String>) -> Option<Self> {
(!nodes.is_empty()).then_some(Self(nodes))
}
fn named(&self) -> String {
self.0.join(", ")
}
}
/// The readings of the graph behind a run's word and advice.
enum WorkStanding {
/// Every node completed successfully.
Complete,
/// The graph converged without completing and has no work an operator can
/// return to the frontier, as for a failed run.
Settled,
/// Something is ready, running, waiting, or blocked and a fresh driver can
/// pick it up now or when its gate opens.
Outstanding,
/// Work no driver moves on its own, which is what the run is told about.
Held(HeldWork),
}
/// The work a converged run holds that no driver moves on its own.
///
/// Three inhabited cases and no fourth. A run can hold both kinds at once and
/// they are different facts about it — parked work returns to the frontier on a
/// `requeue`, and a node its judge rejected does not return on anything a driver
/// does — so both are carried, and [`ranked`](Self::ranked) decides which is
/// said rather than the reading throwing one away to decide.
enum HeldWork {
/// Nodes a `requeue` returns to the frontier.
Parked(HeldNodes),
/// Nodes a judge rejected, which nothing dispatches as they stand.
Rejected(HeldNodes),
/// Both, on the one run.
Both {
/// The parked half, which is the half that has an answer a driver acts on.
parked: HeldNodes,
/// The judged half, still true of the run and still unanswered by that.
rejected: HeldNodes,
},
}
impl HeldWork {
/// What a converged run holds, or `None` where it holds neither kind — which
/// is also the run that has no held standing to report.
fn of(parked: Vec<String>, rejected: Vec<String>) -> Option<Self> {
match (HeldNodes::of(parked), HeldNodes::of(rejected)) {
(Some(parked), Some(rejected)) => Some(Self::Both { parked, rejected }),
(Some(parked), None) => Some(Self::Parked(parked)),
(None, Some(rejected)) => Some(Self::Rejected(rejected)),
(None, None) => None,
}
}
/// Every prescription this held work carries, most urgent first.
///
/// The order is the ranking and the head is the answer: a `requeue` is the
/// one of the two an operator can act on now — it returns work to the
/// frontier and a driver dispatches it — so a run holding both is told to
/// requeue, and meets the judgement on the read after that work has moved.
/// One run is given one answer, because [`Standing::intervention`] takes the
/// head; what the ranking does not do is decide by discarding the other
/// half.
fn ranked(&self) -> impl Iterator<Item = Intervention<'_>> {
let (parked, rejected) = match self {
Self::Parked(parked) => (Some(parked), None),
Self::Rejected(rejected) => (None, Some(rejected)),
Self::Both { parked, rejected } => (Some(parked), Some(rejected)),
};
parked
.map(Intervention::RequeueThenAdopt)
.into_iter()
.chain(rejected.map(Intervention::ReviewThenSupersede))
}
}
/// The nodes of a converged run whose work a judge rejected.
///
/// **Two records, because either alone names the wrong nodes.** The outcome says
/// the dispatch ended on the *task's* own verdict rather than on the machinery
/// or on a publication, which is the settlement a judgement produces; the failed
/// verdict is the judgement itself, and without one a node that simply failed
/// its task would be reported as judged by a judge that never scored it. Both
/// are read off records these views already hold — the run's folded outcomes and
/// the verdicts `oneagentgraph` copies onto the settlement — so nothing is
/// opened to answer the question.
fn rejected_by_a_judge(view: &RunView, statuses: &BTreeMap<String, NodeStatus>) -> Vec<String> {
statuses
.iter()
.filter(|(_, status)| **status == NodeStatus::Failed)
.filter(|(id, _)| {
matches!(
view.state.outcomes.get(*id).map(String::as_str),
Some(crate::engine::TASK_FAILED | crate::engine::TASK_FAILED_CHANGE_OPEN)
)
})
.filter(|(id, _)| !crate::report::failed_verdicts(&view.events, id).is_empty())
.map(|(id, _)| id.clone())
.collect()
}
/// Where a run's **graph** has got to, as one answer rather than as the several
/// questions it is read from.
///
/// A closed set, because the questions behind it are not independent and reading
/// them as though they were is how a contradiction gets a name: every node `done`
/// implies every node settled, so "complete but not converged" describes no graph
/// — except the empty one, which is not a finished run but a run that has not
/// started, and which is [`Unrecorded`](Self::Unrecorded) here rather than a
/// coincidence of two flags.
///
/// One vocabulary over two readers: the folding views read it off a status map
/// and the bounded listing off a summary document's status counts, and each
/// reader has its own constructor so no half-answer crosses between them.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum GraphStanding {
/// Nothing has recorded a graph for this run at all.
Unrecorded,
/// Something can still move, with or without a driver attached to move it.
Moving,
/// Every node settled, and every one of them is `done`.
Complete,
/// Every node settled, and something is waiting on a person or blocked
/// behind one.
ConvergedWaiting,
/// Every node settled, and nothing is waiting on anybody.
ConvergedStill,
}
impl GraphStanding {
/// Read off a run's folded status map.
fn of_statuses(statuses: &BTreeMap<String, NodeStatus>) -> Self {
// An empty status map is a run whose graph nothing has read, not a run
// with nothing left to do: `is_terminal` and `state_of` both answer an
// empty map with the settled reading, and taking it would report a run
// that has recorded nothing as finished.
if statuses.is_empty() {
return Self::Unrecorded;
}
if !graph::is_terminal(statuses) {
return Self::Moving;
}
if graph::state_of(statuses) == graph::GraphState::Complete {
return Self::Complete;
}
if statuses
.values()
.any(|status| matches!(status, NodeStatus::Waiting | NodeStatus::Blocked))
{
return Self::ConvergedWaiting;
}
Self::ConvergedStill
}
/// The same reading off a summary document: the same words, counted.
///
/// `converged` is the document's own [`RunSummary::graph_complete`], which is
/// that same "every node settled, and there is a node" — derived once, where
/// the graph was in hand, rather than guessed back out of a count.
fn of_counts(counts: &BTreeMap<String, u64>, converged: bool) -> Self {
if counts.is_empty() {
return Self::Unrecorded;
}
if !converged {
return Self::Moving;
}
if counts.keys().all(|word| word == NodeStatus::Done.as_str()) {
return Self::Complete;
}
if counts.contains_key(NodeStatus::Waiting.as_str())
|| counts.contains_key(NodeStatus::Blocked.as_str())
{
return Self::ConvergedWaiting;
}
Self::ConvergedStill
}
/// Whether the loop has anything left to converge.
fn converged(self) -> bool {
matches!(
self,
Self::Complete | Self::ConvergedWaiting | Self::ConvergedStill
)
}
/// The nodes a converged run holds, and nothing for one still moving.
///
/// A run with a node still ready, running, or pending has work a fresh driver
/// moves, whatever else is parked or a judge rejected, and `adopt` is the
/// whole of what it needs. Held work is the reading for the frontier that
/// cannot move.
///
/// The nodes are named by a closure rather than passed, because one of the
/// two callers reads them out of the run's whole event store: a moving run
/// must not pay for an answer this would discard.
fn holding(self, named: impl FnOnce() -> Vec<String>) -> Vec<String> {
if self.converged() {
named()
} else {
Vec::new()
}
}
}
impl WorkStanding {
/// What a graph leaves for somebody to do.
///
/// One derivation, because two readers reach it: the folding views read the
/// graph off a run's whole store and the bounded listing reads it off that
/// run's summary document, and the reading — not the reader — is what decides
/// the word on the row and the advice under it.
fn of(graph: GraphStanding, parked: Vec<String>, rejected: Vec<String>) -> Self {
if graph == GraphStanding::Complete {
return Self::Complete;
}
if let Some(held) = HeldWork::of(parked, rejected) {
return Self::Held(held);
}
match graph {
GraphStanding::ConvergedStill => Self::Settled,
_ => Self::Outstanding,
}
}
}
impl Standing {
/// Read one run's standing, once.
fn of(view: &RunView) -> Self {
let statuses = view.state.statuses();
let graph = GraphStanding::of_statuses(&statuses);
Self {
liveness: view.liveness(),
work: WorkStanding::of(
graph,
graph.holding(|| {
statuses
.iter()
.filter(|(_, status)| **status == NodeStatus::Parked)
.map(|(id, _)| id.clone())
.collect()
}),
graph.holding(|| rejected_by_a_judge(view, &statuses)),
),
convergence: Convergence::of(graph.converged()),
}
}
/// The same standing, off one row of a bounded listing.
///
/// The graph is read off the document's own status counts rather than off a
/// status map, in the same words and by the same reading. What a count cannot
/// answer, the document names outright: which nodes are parked, and which a
/// judge turned down.
fn of_row(row: &Row<'_>) -> Self {
let graph = GraphStanding::of_counts(&row.summary.node_counts, row.summary.graph_complete);
Self {
liveness: row.liveness(),
work: WorkStanding::of(
graph,
graph.holding(|| row.summary.parked.clone()),
graph.holding(|| row.summary.judge_rejected.clone()),
),
convergence: Convergence::of(graph.converged()),
}
}
/// The word a view prints for how the run is being driven.
fn word(&self) -> &'static str {
if matches!(&self.work, WorkStanding::Complete) {
"SETTLED"
} else {
self.liveness.as_str()
}
}
/// What this run needs before it can move again, where it needs anything.
///
/// `None` is the third answer and the one the advice used not to have: a run
/// whose work is over needs nothing, and a driver attached to it settles it
/// again in no time at all having dispatched nothing.
fn intervention(&self) -> Option<Intervention<'_>> {
if !self.liveness.is_undriven() {
return None;
}
match &self.work {
WorkStanding::Outstanding => Some(Intervention::Adopt),
// The head of the ranking, which held work always has: `HeldWork::of`
// answers `None` rather than build one holding nothing.
WorkStanding::Held(held) => held.ranked().next(),
WorkStanding::Complete | WorkStanding::Settled => None,
}
}
}
/// What a run nothing is driving needs before it can move again.
enum Intervention<'a> {
/// A fresh driver, and nothing else: work is waiting on the frontier for it.
Adopt,
/// The parked work returned to the frontier, and *then* a driver.
RequeueThenAdopt(&'a HeldNodes),
/// The judge's verdict read, and the node it rejected superseded. A driver
/// is not the first step here and on its own is not a step at all.
ReviewThenSupersede(&'a HeldNodes),
}
/// The prescription for a run nothing is driving whose unfinished work is
/// parked, phrased once for both views that give it.
///
/// The order is the whole content. A parked node is held out of every reconcile
/// pass until a `requeue`, so a driver adopted first derives an empty frontier
/// and returns at exit 0 having dispatched nothing — which is what an advice
/// line naming only `adopt` cost the operator who followed it, twice.
fn requeue_then_adopt(run: &str, parked: &HeldNodes) -> String {
format!(
"its unfinished work is parked, and no driver dispatches a parked node: return {} \
to the frontier with a `requeue` on: onepipeline reply {run} — and only then \
attach a fresh driver with: onepipeline adopt {run}",
parked.named()
)
}
/// The prescription for a run nothing is driving whose unfinished work is held
/// up by nodes a judge rejected, phrased once for both views that give it.
///
/// **The judgement is the content.** A rejection is deliberately outside the
/// publication attempt budget — asking again repeats the same work against the
/// same bar — so nothing dispatches the node as it stands, and a driver attached
/// here settles having moved nothing, which is what the `adopt` this replaces
/// cost an operator twice. What moves the run is the verdict, and then the
/// planner's own `amend` and `retry`.
fn review_then_supersede(run: &str, rejected: &HeldNodes) -> String {
format!(
"its unfinished work is held up by {}, whose work a judge rejected, and no driver \
dispatches a rejected node as it stands: read the verdict with: onepipeline \
results {run} — and decide from it, most likely amending the task and superseding \
the node with an `amend` and a `retry` on: onepipeline reply {run}",
rejected.named()
)
}
/// The word a view prints for how a run is being driven.
///
/// A run whose graph completed is **settled**, not abandoned: its driver is
/// gone because there was nothing left for it to do. Reporting `DRIVER DEAD`
/// there would send a planner to intervene in finished work.
pub fn liveness_word(view: &RunView) -> &'static str {
Standing::of(view).word()
}
/// The same word off one row of a **bounded listing**, for a verb that reports a
/// run without folding it.
///
/// The listing's own reading, reached rather than reimplemented: `runs` prints
/// this word for this run, and a second derivation of it would be a second
/// standing to keep true. Reading it costs what a row costs — the host's answer
/// about the recorded driver and the run's own channel — and no part of the run's
/// merged event store.
pub(crate) fn summary_standing_word(root: &Path, summary: &RunSummary) -> &'static str {
Standing::of_row(&Row::of(root, summary)).word()
}
/// Whether a run has **settled**: every node of its graph reached a state the
/// loop is finished with, so there is no pass left to read anything handed to it.
///
/// A fact about the run rather than about a process, and deliberately so.
/// [`DriverLiveness`] answers *who is driving*, which is a different question
/// with a different answer at the same instant: a driver that has written the
/// run's result and released its ownership lock is still in the process table
/// for as long as it takes to exit, and reads as `ACTIVE` throughout. Anything
/// deciding whether a run can still take work in — `reply` is the one that
/// costs, because a reply queued for a run nobody will drive again is delivered
/// to nothing — has to ask this instead, or its answer depends on the order in
/// which a driver exits.
///
/// A run whose graph nothing has recorded has **not** settled: an empty status
/// map is a run that has not started, and `Standing::of` states that reading
/// once for every reader of it.
///
/// Crate-visible: what the contract names here is what a view *renders*, and this
/// is a reading the channel's own refusal shares with them rather than a promise
/// to a consumer.
pub(crate) fn has_settled(view: &RunView) -> bool {
Standing::of(view).convergence == Convergence::Settled
}
/// Whether this run's driver tier leaves room to say anything about the graph
/// **watching** it.
///
/// Only a run that is actually executing does. A settled run needs no observer,
/// and a run nothing is driving has bigger news on the same line — reporting
/// either as unwatched would send an operator after a graph whose absence is not
/// the problem. The verdict itself is [`watching`]; this is the gate in front of
/// asking for one, which is also what keeps the launch record unread for every
/// row that could not print the answer.
fn may_report_an_observer(standing: &Standing) -> bool {
standing.word() == DriverLiveness::Driving.as_str()
}
/// That word, and — where the driver recorded one — why nothing is watching.
///
/// The reason travels on the same line rather than under it, because this is the
/// one observer verdict a reader has to act on and the act depends on which
/// reason it is: a bound that is spent is a run to take over, and a graph that
/// would not start is a graph to fix first.
fn observer_suffix(view: &RunView, standing: &Standing) -> String {
observer_phrase(
may_report_an_observer(standing).then(|| observer_liveness(&view.launch)),
&view.launch.observer_ending,
)
}
/// The suffix itself, from the verdict and the reason it may carry.
///
/// One phrasing over both readers, for the reason [`tally`] is one: the listing
/// reads the observer off a summary document and the folding views read it off a
/// launch record, and a suffix worded twice is a suffix that drifts.
fn observer_phrase(verdict: Option<ObserverLiveness>, ending: &str) -> String {
match verdict {
None | Some(ObserverLiveness::Watching) => String::new(),
Some(verdict @ ObserverLiveness::ObserverNotRestarted) => {
format!(" {}: {}", verdict.as_str(), ending)
}
Some(verdict) => format!(" {}", verdict.as_str()),
}
}
/// One run of a **bounded listing**, as the two views that render one read it.
///
/// The counterpart of [`RunView`] on the cheap path, and the distinction between
/// them is the whole of this type's reason to exist: a detail read may fold a
/// run's merged store, and a listing may never — a host holding five hundred runs
/// answers `runs` by reading five hundred bounded documents, not eleven gigabytes
/// of journal. What is *not* on the document stays a read of the host at the
/// moment of the question: whether the driver process is alive, what the run's
/// channel is holding, and whether a change has reached its base.
struct Row<'a> {
/// Where the run's durable state lives, for exactly those reads.
paths: RunPaths,
/// The run's own record of itself, read rather than folded.
summary: &'a RunSummary,
}
impl<'a> Row<'a> {
fn of(root: &Path, summary: &'a RunSummary) -> Self {
Self {
paths: RunPaths::under(root, &summary.run_id),
summary,
}
}
/// How the run is being driven, read from the host now.
fn liveness(&self) -> DriverLiveness {
driver_liveness(
self.summary.stop_recorded
|| let_go(
self.summary.host.as_deref(),
self.summary.pid,
self.summary.let_go_by.as_ref(),
),
self.summary.host.as_deref(),
self.summary.pid,
self.summary.started.as_deref(),
self.summary.last_write_at,
self.summary.awaiting_human_action,
&self.paths,
)
}
/// Whether anything is watching the run, where its driver tier leaves room
/// to say so.
///
/// Read off the **launch record**, and deliberately not off the summary
/// document beside it: a driver rewrites what this asks under a live run and
/// appends nothing — one that finds its observer gone starts another and
/// records the new graph run, one that stops starting another records why —
/// so a stored answer would name a graph run that had ended and say nothing
/// about a run nobody will watch again. The read is a small file, it is not
/// the run's merged store, and it happens only for a run this listing is
/// already reporting as being driven.
///
/// A record this build cannot read leaves the suffix off, which is where
/// every unreadable observer input resolves: a run reported unwatched invites
/// an operator to intervene, and doing that to a working observer is worse
/// than saying nothing.
fn observer_suffix(&self, standing: &Standing) -> String {
if !may_report_an_observer(standing) {
return String::new();
}
let Some(launch) = ledger::read_json_opt::<LaunchRecord>(&self.paths.launch()) else {
return String::new();
};
observer_phrase(Some(observer_liveness(&launch)), &launch.observer_ending)
}
/// The `n/m done` line, with every landing it counts decided **now** —
/// which is why the counts come off the document and the landings do not.
fn tally(&self) -> String {
let render = rendering(Rendered::Summary, &self.summary.run_id);
let counted = |word: NodeStatus| {
usize::try_from(*self.summary.node_counts.get(word.as_str()).unwrap_or(&0))
.unwrap_or(usize::MAX)
};
// Saturating rather than summed, because every one of these counts is
// *external input* — a document on disk that some other build wrote, and
// that a hand-edit can put any number in — and a plain sum of them
// overflows into a panic on a debug build rather than into a row that
// reads oddly.
let total = self
.summary
.node_counts
.values()
.copied()
.map(|count| usize::try_from(count).unwrap_or(usize::MAX))
.fold(0usize, usize::saturating_add);
tally(
counted(NodeStatus::Done),
total,
counted(NodeStatus::Skipped),
&self.outstanding(&render),
)
}
/// What of this run's work has not reached a base, asked of the document's
/// own landing inputs and answered by reading the repository now.
///
/// The one fact on this path that is deliberately *not* served out of the
/// document. See [`RunSummary::landings`].
fn outstanding(&self, render: &Rendering) -> Outstanding {
let mut outstanding = Outstanding::default();
for (node, landing) in &self.summary.landings {
if landing.drafted {
continue;
}
let reported = reported_landing_of(
&self.summary.run_id,
render,
node,
Landing::parse(&landing.landing),
// A stated landing reaches a summary as the `landed` word the
// fold wrote for it, which is all a count reads: it is never
// re-read here either.
None,
landing.branch.as_deref(),
landing.repo.as_deref(),
);
match reported.stands() {
Stands::NotLanded => outstanding.not_landed.push(node.clone()),
Stands::Undecided => outstanding.undecided.push(node.clone()),
Stands::Landed | Stands::NoChangeToLand => {}
}
}
outstanding
}
fn unread(&self) -> Unread {
Unread::of(&crate::channel::ChannelState::new(&self.paths).queue())
}
}
/// `onepipeline runs`.
///
/// A run root under this root that could not be read is named at the end rather
/// than dropped: an empty listing on a host that holds runs is the reading that
/// costs the most, because it is the one a planner acts on by starting more work.
pub fn runs(root: &Path, mine_only: bool, session: &str) -> String {
let listing = Listing::of(root);
let mut out = String::new();
for summary in listed(&listing) {
let owned = ledger::owned_by(&summary.session, session);
if mine_only && !owned {
continue;
}
let row = Row::of(root, summary);
let marker = if owned { '*' } else { ' ' };
let standing = Standing::of_row(&row);
out.push_str(&format!(
"{marker} {:<24} {:<24} {} {}{}\n",
summary.run_id,
ledger::owner_label(&summary.launcher, &summary.session, session),
row.tally(),
standing.word(),
row.observer_suffix(&standing)
));
// A run reported stopped keeps the line saying why it stopped rather
// than an invitation to read updates nothing will follow up on. A run
// that needs no intervention falls through to those updates instead:
// its work is over, and what is left to say about it is what nobody has
// read yet.
if let Some(intervention) = standing.intervention() {
out.push_str(&match intervention {
Intervention::Adopt => format!(
" {} — its ledger is intact; attach a fresh driver with: \
onepipeline adopt {}\n",
standing.word(),
summary.run_id
),
Intervention::RequeueThenAdopt(parked) => format!(
" {} — its ledger is intact; {}\n",
standing.word(),
requeue_then_adopt(&summary.run_id, parked)
),
Intervention::ReviewThenSupersede(rejected) => format!(
" {} — its ledger is intact; {}\n",
standing.word(),
review_then_supersede(&summary.run_id, rejected)
),
});
continue;
}
let unread = row.unread();
if let (count, Some(stale)) = (unread.count, unread.oldest_seconds) {
if count > 0 {
out.push_str(&format!(
" {count} planner update(s) waiting ({}), unread for {}; \
read them with: onepipeline next {}\n",
unread.phrase(),
crate::telemetry::duration(stale * 1_000),
summary.run_id
));
}
}
}
if out.is_empty() {
return nothing_listed(&listing);
}
out.push_str(&skipped_lines(&listing.skipped));
out
}
/// The rows of a listing in the order a view renders them: **by run id**.
///
/// [`Listing`] serves its rows most recently written first, which is the order
/// the ordering key on the document exists to make answerable without a fold. A
/// view renders them by id instead, because that is the order every reader of
/// `runs` and `status` has ever read them in and the row's own id is the thing
/// they scan for; the recency order is there for a consumer building its own
/// listing.
fn listed(listing: &Listing) -> Vec<&RunSummary> {
let mut rows: Vec<&RunSummary> = listing.summaries.iter().collect();
rows.sort_by(|a, b| a.run_id.cmp(&b.run_id));
rows
}
/// What a bounded listing says when it has no run to report.
///
/// The same two facts [`nothing_to_report`] tells apart, over the listing's own
/// account of what it refused: a root with nothing in it and a root whose every
/// run was refused both rendered as `no runs recorded`, and only one of them
/// means there is nothing running.
fn nothing_listed(listing: &Listing) -> String {
// llmlint: ignore-block[cli_output_contract] a refused run root is part of the answer,
// not a failure of the command, on the terms `skipped_lines` and its own block state:
// this empty case *replaces* `no runs recorded`, so it cannot live on a stream other
// than the answer it replaces.
let mut out = if listing.summaries.is_empty() && !listing.skipped.is_empty() {
format!(
"no run under {} could be read\n",
one_line(&listing.root.display().to_string())
)
} else {
"no runs recorded\n".to_string()
};
out.push_str(&skipped_lines(&listing.skipped));
out
// llmlint: ignore-end[cli_output_contract]
}
/// The **run-level** lines of `onepipeline status`: what it says about the run
/// itself rather than about any one of its nodes.
///
/// One phrasing over two readers, and the split is the point. `status` given one
/// run is a detail read and folds that run's whole store to say what each of its
/// nodes is doing; `status` given none is a **listing**, and answers these lines
/// out of each run's bounded summary document. Two wordings of the run's own
/// standing would be two things to keep true, and a supervisor reads them one
/// after the other.
fn status_run_lines(
paths: &RunPaths,
standing: &Standing,
observer: &str,
tally: &str,
unread: &Unread,
) -> String {
let run = &paths.run;
let mut out = format!("{run} {}{observer} {tally}\n", standing.word());
if let Some(intervention) = standing.intervention() {
out.push_str(&match intervention {
Intervention::Adopt => format!(
" {}: nothing is driving this run; adopt it or stop it\n",
standing.word()
),
Intervention::RequeueThenAdopt(parked) => format!(
" {}: nothing is driving this run and {}\n",
standing.word(),
requeue_then_adopt(run, parked)
),
Intervention::ReviewThenSupersede(rejected) => format!(
" {}: nothing is driving this run and {}\n",
standing.word(),
review_then_supersede(run, rejected)
),
});
}
// Whatever is in the slot, said either way — but not the same way. A surface
// nobody is waiting on any more is not a decision this run is held on, and
// reporting it as one is the defect this line used to have; saying nothing
// about it instead would lose the last place its text is shown, since it has
// been delivered and `next` does not hand it out twice.
if let Some(held) = crate::channel::ChannelState::new(paths).held() {
out.push_str(&format!(
" {}: {} — {}\n",
match (held.abandoned, held.blocking) {
(true, _) => "a planner update nobody is waiting on any more",
(false, true) => "waiting for planner decision",
(false, false) => "waiting for planner reply",
},
held.kind,
held.message
));
}
if unread.count > 0 {
out.push_str(&format!(
" {} planner update(s) waiting ({}), unread for {}\n",
unread.count,
unread.phrase(),
crate::telemetry::duration(unread.oldest_seconds.unwrap_or(0) * 1_000)
));
}
// Off the count above and said out loud anyway. Nothing is waiting for an
// answer to these, so counting them would inflate the one line a supervisor
// may not filter — but their text is still in the queue and still worth
// reading, and a surface that vanished without a word would be the same
// silence from the other direction.
if unread.abandoned > 0 {
out.push_str(&format!(
" {} planner update(s) nobody is waiting on: whatever raised them ended \
without an answer; read them with: onepipeline next {run}\n",
unread.abandoned
));
}
out
}
/// `onepipeline status` given **no run**: the bounded listing.
///
/// Every run under the root, at a cost that does not grow with any of their
/// journals — the run-level lines alone, read out of each run's summary
/// document. What each node of a run is doing is `status <RUN>`'s to say, and
/// that is a detail read which folds: a per-node block for every run on a host
/// is a fold of every store on it, which is the cost this path exists to remove.
pub(crate) fn status_listed(listing: &Listing) -> String {
let mut out = String::new();
for summary in listed(listing) {
let row = Row::of(&listing.root, summary);
// Opened before anything under it renders, so the tally this line
// carries and the landings it counts are one render and ask each node
// once between them.
let _render = rendering(Rendered::Status, &summary.run_id);
let standing = Standing::of_row(&row);
out.push_str(&status_run_lines(
&row.paths,
&standing,
&row.observer_suffix(&standing),
&row.tally(),
&row.unread(),
));
}
if out.is_empty() {
return nothing_listed(listing);
}
out.push_str(&skipped_lines(&listing.skipped));
out
}
/// `onepipeline status`.
// llmlint: ignore[contracts_have_one_source_or_a_drift_gate] the second source is `docs/contract-divergences.md` entry 33, and the gate reconciling it is `tests/contract.rs`: it refuses an entry that is neither marked resolved with the contract naming what its ruling adopted, nor marked open with the proposal it waits on. Entry 33 is open and states this exact proposal, because the contract is approved verbatim and amending it is the planner's, never this repository's.
pub fn status(survey: &Survey) -> String {
let mut out = String::new();
for view in &survey.views {
// Opened before anything under it renders, so the summary this line
// carries and the outstanding-landing lines below it are one render and
// ask each node once between them.
let render = rendering(Rendered::Status, &view.paths.run);
let standing = Standing::of(view);
out.push_str(&status_run_lines(
&view.paths,
&standing,
&observer_suffix(view, &standing),
&view.summary(),
&view.unread(),
));
let statuses = view.state.statuses();
for (id, node_status) in &statuses {
if *node_status != NodeStatus::Running {
continue;
}
let age = view
.state
.dispatched_at
.get(id)
.map(|at| sys::now_millis().saturating_sub(*at));
let age = crate::telemetry::duration(age.unwrap_or(0));
if view.state.stop_recorded() {
// What this node last did stays on the record and is
// deliberately not repeated here — see [`ENDED_BY_THE_STOP`].
let became = became_of_the_worker(&view.state);
out.push_str(&format!(" {id}: {became}, {age} in\n"));
continue;
}
out.push_str(&format!(" {id}: running for {age}"));
match view.state.activity.get(id) {
// A node the ledger records as running that no live dispatch is
// driving is `UNDRIVEN`. Deliberately not `parked`: that word
// means the opposite here — a node the planner idled with
// `cancel`.
None => out.push_str(&format!(" — {}", DriverLiveness::Undriven.as_str())),
Some(activity) => out.push_str(&format!(" — {}", working(activity))),
}
out.push('\n');
}
// A node whose cancellation is still out there. Under its own word,
// because neither of the two a reader has fits: `parked` is a planner's
// own idle with nothing running, and `ready` is a node about to start.
for (id, node_status) in &statuses {
if *node_status != NodeStatus::Parked {
continue;
}
let Some(pending) = cancelling_for(&view.state, id) else {
continue;
};
out.push_str(&format!(
" {id}: cancelling — asked to stop {pending} ago and its dispatch has not \
settled; it still holds the node's workspace, so wait for it rather than \
requeueing the node\n"
));
}
// A node that is ready and has not started. On its own that reads as
// "about to go", which is what a node waiting on an occupied workspace
// looks like for as long as it waits — one sat `ready` for forty minutes
// while a supervisor looked for a wedge that was not there. So the two
// are separate lines: what it is waiting for, or that it is waiting for
// nothing but a slot.
for (id, node_status) in &statuses {
if *node_status != NodeStatus::Ready {
continue;
}
out.push_str(&format!(
" {id}: ready — {}\n",
waiting_on(&view.state, id)
));
}
// What each amended node is currently judged against. Rendered whatever
// that node's status is, because the reader this line is for is a
// manager about to *replace* an amendment: `amend` replaces rather than
// appends, so a ruling that is not readable before the replacement lands
// is a ruling nobody can weigh against the one taking its place.
for node in view.state.graph.iter() {
if let Some(amendment) = &node.amendment {
out.push_str(&format!(
" {}: amended — {}\n",
node.id,
one_line(amendment)
));
}
}
// What refused, for the nodes that failed. A failed node otherwise reads
// the same whether its own gate failed or an identity chain ran out, and
// the two call for opposite actions from whoever is reading this.
//
// Only a chain that ran out is reported under the node's failure. One
// that fell through and was then served is reported as the recovery it
// was, under its own word.
for (id, node_status) in &statuses {
if *node_status != NodeStatus::Failed {
continue;
}
// A dispatch that died is not a node whose agent failed, and this
// view is where somebody decides whether to re-run one. Said here as
// well as in `results` for the reason the unlanded line below is:
// deciding there is nothing left to do is a decision made from this
// view, and a node reported as a plain failure over finished work is
// what that decision used to be made on.
let branch = view.state.branches.get(id).cloned().or_else(|| {
view.state
.graph
.get(id)
.and_then(|node| node.branch.clone())
});
if let Some(died) = death_phrase(&view.state, id, branch.as_deref()) {
out.push_str(&format!(" {id}: {died}\n"));
}
for record in chain_records(&view.state, id) {
out.push_str(&format!(
" {id}: {} — {}\n",
record.lead_in(),
chain_phrase(&record)
));
}
}
// Said on this view because it is the one somebody reads when a run has
// gone quiet and they are deciding whether it is stuck; nothing else here
// tells waiting-on-a-release from stalled.
let drafted = drafted_nodes(view);
if !drafted.is_empty() {
out.push_str(&format!(
" {} node(s) complete and held as a draft: the run is waiting on the \
release(s) each names, and is neither stalled nor finished\n",
drafted.len()
));
for (id, awaiting) in &drafted {
let says = awaiting
.as_deref()
.map(|detail| format!(" — {detail}"))
.unwrap_or_default();
out.push_str(&format!(" {id}: complete-but-draft{says}\n"));
}
}
// The settled nodes whose work has not reached anyone. This view
// otherwise reports only what is in flight, so a planner reading it saw
// a run go quiet and took that for a run whose work had landed. Named
// here rather than left to `results`, because deciding there is nothing
// left to do is a decision made from this view.
let outstanding = outstanding_landings(view, &render);
if !outstanding.not_landed.is_empty() {
out.push_str(&format!(
" {} node(s) have not landed: {} — read as this rendered; \
`results {}` names the change to open\n",
outstanding.not_landed.len(),
outstanding.not_landed.join(", "),
view.paths.run
));
}
// Kept apart from the count above, because it is not a count of work
// nobody landed: it is a count of nodes nothing here could decide, and
// reporting them as unlanded would put back the false claim the line
// above stopped making.
if !outstanding.undecided.is_empty() {
out.push_str(&format!(
" {} node(s) whose landing this host could not decide: {} — \
`results {}` says what refused, and names the change to open\n",
outstanding.undecided.len(),
outstanding.undecided.join(", "),
view.paths.run
));
}
out.push_str(&journal_loss_line(view));
if let Some(health) = crate::agentgraph::health() {
out.push_str(&format!(" providers: {health}\n"));
}
}
if out.is_empty() {
return nothing_to_report(survey);
}
out.push_str(&skipped_lines(&survey.skipped));
out
}
/// What a node that settled [`DISPATCH_DIED`] or [`PROVIDER_FAILED`] says, in one
/// sentence.
///
/// Named once because `results` and `status` both say it, and a manager reading
/// one after the other must not meet two accounts of the same node.
///
/// The two words open the sentence differently and end it identically, because
/// the difference between them is *what* killed the dispatch and the half a
/// reader acts on is the same either way. `provider-failed` says the provider
/// killed it outright: `task-failed` used to send that reader looking for what
/// the work got wrong, when nothing was wrong with the work at all.
///
/// It names the producer's own classification first, because that is what decides
/// whether anything is worth re-running: a rate limit twenty seconds after the
/// final report and a workspace deleted underneath a live turn call for opposite
/// next moves. Then — where the node left a branch, and where `onevcs` recorded
/// what that branch is at — it says the branch may carry finished work and names
/// the commit, which is the half that otherwise has to be dug out of the run's
/// journal by hand.
///
/// It says **may**, deliberately. This crate cannot tell a worker's prose report
/// from a gate verdict, and a sentence that was sometimes wrong about "the gate
/// passed" would be worse than no sentence at all. So it points at the branch and
/// at the node's own transcript and lets a person read them.
///
/// `None` for every node that settled any other way, which is what keeps this out
/// of the line of a node whose agent really did fail its task.
///
/// [`DISPATCH_DIED`]: crate::engine::DISPATCH_DIED
/// [`PROVIDER_FAILED`]: crate::engine::PROVIDER_FAILED
fn death_phrase(state: &RunState, id: &str, branch: Option<&str>) -> Option<String> {
let word = match state.outcomes.get(id).map(String::as_str) {
Some(word @ (crate::engine::DISPATCH_DIED | crate::engine::PROVIDER_FAILED)) => word,
_ => return None,
};
let classified = match state.causes.get(id) {
Some(cause) => format!(" ({cause})"),
// llmlint: ignore[changed_behavior_has_e2e] no invocation a user can type reaches
// this arm: `engine::failed_task` settles this word only where it lifted a
// classification, so a settlement carrying it without one is a journal an *older
// build* wrote. Reaching it end to end means writing that journal by hand, which
// would prove the fixture, and dropping the arm would put the run's own word behind
// an unwrap. Held by this module's unit test instead, which folds that record.
None => String::new(),
};
let where_the_work_is = match (branch, state.heads.get(id)) {
(Some(branch), Some(head)) => {
format!("{branch} may carry finished work, at {head}")
}
(Some(branch), None) => format!("{branch} may carry finished work"),
// The workspace went with the dispatch. Said outright, because the
// absence of a branch here is the difference between work to recover and
// work to redo.
(None, _) => "it left no branch, so nothing of it survived".to_owned(),
};
// The provider death says what it was rather than what it was not, because
// that is the reading `task-failed` used to deny it: nothing was wrong with
// the work, so "rather than failing its task" is not the contrast to draw.
let how = if word == crate::engine::PROVIDER_FAILED {
format!("the provider killed the dispatch{classified}, so nothing here is the work's fault")
} else {
format!("the dispatch died{classified} rather than failing its task")
};
Some(format!("{how}; {where_the_work_is}"))
}
/// How long a node's cancellation has been waiting on the dispatch it asked to
/// stop, when one is still out there — see [`Recorded::cancelling_since`].
///
/// [`Recorded::cancelling_since`]: crate::projection::Recorded::cancelling_since
fn cancelling_for(state: &RunState, id: &str) -> Option<String> {
let since = state.recorded.get(id)?.cancelling_since()?;
Some(crate::telemetry::duration(
sys::now_millis().saturating_sub(since),
))
}
/// What became of one candidate a node's identity chain stepped past.
///
/// Three answers rather than a bool, because the third is a different fact and
/// reading it as either of the others is a claim nothing supports: a chain this
/// run's records cannot follow is **not** a chain that ran out of candidates,
/// and a view that called it one would send every reader at a subscription that
/// was never the problem.
#[derive(Debug, Clone, PartialEq, Eq)]
enum Fallthrough {
/// Another identity went on to run that side's invocation on that turn.
PassedTo(String),
/// Nothing served it: the chain had no successful candidate.
Refused,
/// This run's records do not say. A single-sided member attributes nothing
/// per side or per turn, so nothing it publishes can be paired with.
Unrecorded,
}
/// One rendered line's worth of what a node's identity chains did.
///
/// The advance, what became of it, and how many records said the same thing —
/// the last collapsed **here** rather than in the fold, because two turns of one
/// chain can end differently and a record that had collapsed them could only be
/// rendered as one of the two.
struct ChainRecord<'a> {
/// The candidate the chain stepped past.
refusal: &'a Refusal,
/// What became of that side's turn afterwards.
became: Fallthrough,
/// How many records carried this same side, identity, reason, and ending.
///
/// Non-zero for the reason [`Refusal::records`] is: a line exists only by
/// having been recorded at least once, and a rendering that could hold a
/// zero would be one that could say a chain recorded nothing.
records: std::num::NonZeroU64,
}
impl ChainRecord<'_> {
/// The word this record is reported under.
///
/// A chain that ran out is why the node failed; one that recovered is
/// evidence beside it, and saying `failed` over it is exactly the confusion
/// this exists to end.
fn lead_in(&self) -> &'static str {
match self.became {
Fallthrough::Refused => "failed",
Fallthrough::PassedTo(_) | Fallthrough::Unrecorded => "fallback",
}
}
}
/// What one node's identity chains did, in arrival order and one entry per line
/// a view will render.
///
/// Records that agree on the side, the identity, the reason **and** the ending
/// are one fact recorded several times; records that differ in the ending are
/// two facts, and a run whose chain recovered on one turn and ran out on the
/// next says both.
fn chain_records<'a>(state: &'a RunState, node: &str) -> Vec<ChainRecord<'a>> {
let mut records: Vec<ChainRecord<'a>> = Vec::new();
for refusal in refusals_of(state, node) {
let became = became_of(state, node, refusal);
if let Some(same) = records.iter_mut().find(|seen| {
seen.refusal.advanced.identity == refusal.advanced.identity
&& seen.refusal.advanced.role == refusal.advanced.role
&& seen.refusal.advanced.reason == refusal.advanced.reason
&& seen.refusal.member == refusal.member
&& seen.became == became
}) {
same.records = same.records.saturating_add(refusal.records.get());
continue;
}
records.push(ChainRecord {
refusal,
became,
records: refusal.records,
});
}
records
}
/// What became of the turn one advance was recorded on.
///
/// Paired by **side and turn**, which is what the producer stamps on both
/// records: a two-party member runs one chain per side per turn, publishes an
/// advance per candidate that chain stepped past, and publishes the invocation
/// that actually ran beside them. An advance carrying neither — which is a
/// single-sided member's, the one kind that publishes no invocation at all — has
/// nothing to pair with, and is said to have nothing rather than assumed to have
/// run out.
fn became_of(state: &RunState, node: &str, refusal: &Refusal) -> Fallthrough {
let (Some(role), Some(turn)) = (refusal.advanced.role, refusal.advanced.turn) else {
return Fallthrough::Unrecorded;
};
match served_in(state, node, &refusal.member, role, turn) {
Some(served) => Fallthrough::PassedTo(served.session.identity.clone()),
None => Fallthrough::Refused,
}
}
/// The invocation that ran one member's side on one turn, if this run recorded
/// one.
///
/// The member is part of the key as well as the side and the turn: a dispatch
/// runs more than one member, each numbers its own turns, and pairing across two
/// of them would name an identity that served somebody else's chain.
fn served_in<'a>(
state: &'a RunState,
node: &str,
member: &MemberLabel,
role: oneagentgraph::event::Role,
turn: u64,
) -> Option<&'a ServiceRecord> {
state.served.get(node)?.iter().find(|served| {
served.member == *member && served.session.role == role && served.session.turn == turn
})
}
/// Every provider refusal one node's dispatches recorded, in arrival order.
fn refusals_of<'a>(state: &'a RunState, node: &str) -> &'a [Refusal] {
state.refusals.get(node).map_or(&[], Vec::as_slice)
}
/// How one candidate a chain stepped past reads on a rendered line.
///
/// The side first, because it is the half a reader most often gets wrong: the
/// two sides of a member prefer different identities, and an operator who
/// restored the wrong subscription spent a night watching the same failure.
///
/// A bare "refused" is reserved for a chain with **no** successful candidate.
/// Every other ending says the chain fell through and what happened next, so no
/// reader takes a recovery for the reason a node failed.
///
/// Every value on the line is a stranger's — an identity, a classification, a
/// role, and a member name, all read off a sibling's envelope, and for a
/// recovery a second identity read off a second one — so the whole phrase goes
/// through the same control strip the rest of this module uses.
fn chain_phrase(record: &ChainRecord) -> String {
let refusal = record.refusal;
// The role's own spelling, taken from the producing library's serialization
// rather than matched into words of this crate's: the sides are that
// library's vocabulary, and a second spelling of them here is a second thing
// to keep true.
let role = refusal
.advanced
.role
.and_then(|role| serde_json::to_value(role).ok());
let side = match (
role.as_ref().and_then(serde_json::Value::as_str),
&refusal.member,
) {
(Some(role), _) => format!("the {role} side"),
(None, MemberLabel::Named(member)) => format!("member '{member}'"),
// Neither was stamped. The identity is still the thing to act on, and
// naming a side the record does not carry would send the fix at a chain
// nobody named — which is the failure this line exists to end.
//
// llmlint: ignore-block[changed_behavior_has_e2e] `oneagentgraph` labels a
// member's envelopes with the member, so no producer reaches this arm; it is
// written for one that stamps neither. The arm a producer does reach is driven in
// `tests/e2e/views.rs`.
(None, MemberLabel::Unstamped) => "a side the record does not name".to_string(),
// llmlint: ignore-end[changed_behavior_has_e2e]
};
// llmlint: ignore-block[changed_behavior_has_e2e] `FallbackAdvanced::reason` is a
// required `String`, so no producer reaches the empty half; it is written for a newer
// sibling that relaxes the field. The half a producer does reach is driven end to end.
let reason = if refusal.advanced.reason.is_empty() {
"for a reason the record does not carry".to_string()
} else {
format!("({})", refusal.advanced.reason)
};
// llmlint: ignore-end[changed_behavior_has_e2e]
// What was counted, said as what it is: records carrying this same side,
// identity, reason, and ending. The producer stamps a turn on each advance
// and nothing here counts them, so "on N turns" would be a measurement this
// line never made.
let again = if record.records.get() > 1 {
format!(", recorded {} times", record.records)
} else {
String::new()
};
let identity = &refusal.advanced.identity;
one_line(&match &record.became {
Fallthrough::Refused => format!("{side}: identity '{identity}' refused {reason}{again}"),
Fallthrough::PassedTo(who) => {
format!("{side} fell through '{identity}' {reason} → served by '{who}'{again}")
}
Fallthrough::Unrecorded => format!(
"{side} fell through '{identity}' {reason}; nothing this run recorded names what \
served that turn{again}"
),
})
}
/// How one judge verdict that failed a node reads on a rendered line.
///
/// Both halves are a stranger's — the criterion a graph declared and the
/// sentence a judge wrote — so the phrase goes through the same control strip
/// every other relayed value on these views does.
fn verdict_phrase(verdict: &crate::report::FailedVerdict) -> String {
// A record that named neither is still worth a line: it says the node failed
// on its judge, which is the fact a provider line above it would otherwise
// be read as.
let criterion = match &verdict.criterion {
Some(criterion) => format!("'{criterion}'"),
None => "a criterion the record does not name".to_string(),
};
let reason = match &verdict.reason {
Some(reason) => reason.clone(),
None => "the record carries no reason".to_string(),
};
one_line(&format!("{criterion} failed — {reason}"))
}
/// How the dependencies that skipped a node read on that node's own line.
///
/// Each carries its own status: a `failed` cause is work attempted and lost and
/// a `skipped` one is another node never tried, so a reader following the chain
/// back knows whether the next hop is the end of it.
fn skipped_by_phrase(causes: &[(String, NodeStatus)]) -> String {
if causes.is_empty() {
// Unreachable by construction, and phrased anyway: rendering nothing
// would read as a fact the view lost.
return "a dependency this run can no longer name".to_string();
}
causes
.iter()
.map(|(dependency, status)| format!("{dependency} ({})", status.as_str()))
.collect::<Vec<_>>()
.join(", ")
}
/// Whether a person attested that a node this run **failed** had in fact landed.
///
/// Two records, because either alone says something else: an attestation is how
/// every human action completes, and the failure is what the status said before
/// anybody looked.
fn attested_after_failing(view: &RunView, node: &str) -> bool {
view.state.attestations.contains(node)
&& view.events.iter().any(|event| {
event.kind.0 == PipelineKind::NodeSettled.as_str()
&& event.labels.node.as_deref() == Some(node)
&& event
.payload
.get("status")
.and_then(serde_json::Value::as_str)
== Some(NodeStatus::Failed.as_str())
})
}
/// The nodes whose change is still outstanding, read as the view renders.
///
/// Two lists rather than one, because the two call for opposite things from
/// whoever is reading them: a change the base does not carry is work somebody
/// has to land, and a landing this host cannot decide is a question somebody has
/// to answer by opening the change.
#[derive(Debug, Default)]
struct Outstanding {
/// Read now, and the base does not carry it.
not_landed: Vec<String>,
/// Read now, and nothing this host can read decides it.
undecided: Vec<String>,
}
/// What of this run's work has not reached a base, in id order.
///
/// Asked of the nodes whose publication answered a landing at all, which is what
/// it has always been counted over: a node that published nothing has no change
/// outstanding, and one whose settlement recorded a landing already is never
/// asked again. So the read happens for exactly the two answers a later merge
/// changes — `unlanded`, and a landing an earlier read could not decide.
///
/// Not "what had not landed as of settlement". A settlement is an observation of
/// a moment and the moment passes: a run reported four of eleven nodes done
/// while nine of the eleven had landed, and an adoption node was dispatched three
/// times against work that was already on its base.
fn outstanding_landings(view: &RunView, render: &Rendering) -> Outstanding {
let statuses = view.state.statuses();
let mut outstanding = Outstanding::default();
for (node, _) in view
.state
.landings
.iter()
// A node whose change is held as a draft has not landed and is not one of
// these: this list is work an operator has to *decide* about — a change
// sitting open that nobody merged — and a draft is a change this run is
// deliberately holding and will lift itself. It has a line of its own,
// which says what it is waiting on.
.filter(|(node, _)| statuses.get(*node) != Some(&NodeStatus::CompleteDraft))
{
match reported_landing(view, render, node).stands() {
Stands::NotLanded => outstanding.not_landed.push(node.clone()),
Stands::Undecided => outstanding.undecided.push(node.clone()),
Stands::Landed | Stands::NoChangeToLand => {}
}
}
outstanding
}
/// The nodes whose work is complete and whose change is held back as a draft,
/// with what each is waiting on.
///
/// The reason is the settlement's own detail, which is the same sentence `results`
/// prints — so a reader meeting this line and a reader opening `results` are told
/// the same thing rather than two accounts of one node.
fn drafted_nodes(view: &RunView) -> Vec<(String, Option<String>)> {
view.state
.statuses()
.into_iter()
.filter(|(_, status)| *status == NodeStatus::CompleteDraft)
.map(|(node, _)| {
let awaiting = settled_detail(view, &node).map(|detail| one_line(&detail));
(node, awaiting)
})
.collect()
}
/// The `detail` the newest `node-settled` for one node carried.
///
/// One read for the two views that print it, so a settlement cannot be quoted two
/// ways.
fn settled_detail(view: &RunView, node: &str) -> Option<String> {
view.events
.iter()
.rev()
.find(|event| {
event.kind.0 == PipelineKind::NodeSettled.as_str()
&& event.labels.node.as_deref() == Some(node)
})
.and_then(|event| event.payload.get("detail"))
.and_then(|detail| detail.as_str())
.map(str::to_owned)
}
/// What a ready node is waiting on, as far as this host can tell.
///
/// A lifecycle node cannot start until it can open a `onevcs` session over its
/// repository, and a session is held under an occupancy lease — so a ready node
/// whose repository somebody is already in is waiting on that lease, and waits
/// silently. The commonest holder is the node's *own* previous dispatch, which
/// is the state right after a cancel: the node is back on the frontier and the
/// dispatch it cancelled has not let go. Reported so that node reads differently
/// from one waiting for nothing but a concurrency slot, because a supervisor
/// spent forty minutes looking for a wedge in the second when it was the first.
///
/// Three answers and not two. A workspace this host **could not ask about** is
/// neither held nor free, and saying "queued" there would report an unmeasured
/// thing as a measured nothing — the one rule every view here is written to. The
/// holders themselves are `onevcs`'s own verdict, liveness included, because a
/// pid alone cannot say whether a lease is real.
fn waiting_on(state: &RunState, id: &str) -> String {
const QUEUED: &str = "queued for dispatch";
// A node with no repository has no workspace to be held out of.
let Some(repo) = state.graph.get(id).and_then(|node| node.repo.as_deref()) else {
return QUEUED.to_string();
};
let holders = match crate::vcs::holders_of(repo) {
Ok(holders) => holders,
Err(why) => {
return format!(
"{QUEUED}, and this host cannot say whether the '{repo}' workspace is \
free: {}",
one_line(&why)
)
}
};
let held: Vec<String> = holders
.into_iter()
.filter(|holder| {
holder.state == onevcs::Lifecycle::Open && holder.liveness == onevcs::Liveness::Live
})
.map(|holder| {
format!(
"session '{}' (owner_pid {})",
holder.token.0, holder.owner_pid
)
})
.collect();
if held.is_empty() {
return QUEUED.to_string();
}
format!(
"waiting for the '{repo}' workspace, held by {}",
held.join(", ")
)
}
/// What one in-flight dispatch is doing now, on the line that reports it.
///
/// Four facts, because one alone misleads: what it last did, how much it has
/// done, how long ago, and — separately — whether it is still alive. A dispatch
/// that has recorded plenty and nothing recently is the wedged one; a first turn
/// that has run for twenty minutes and is still recording is healthy, and has
/// twice been reported dead for want of this line.
///
/// The age is of the last thing the dispatch **did**, and the heartbeat is
/// reported beside it rather than inside it: an age over every envelope can
/// never be older than one beat for anything that has not died, so a wedged
/// dispatch and a working one read the same.
fn working(activity: &crate::projection::NodeActivity) -> String {
let ago = |at: u64| crate::telemetry::duration(sys::now_millis().saturating_sub(at));
let alive = activity
.last_heartbeat_at
.filter(|beat| activity.progress.is_none_or(|done| *beat > done.last_at()))
.map(|beat| format!("; alive {} ago", ago(beat)))
.unwrap_or_default();
// A dispatch whose every envelope has been a heartbeat has *started* and
// produced nothing, which is neither a dispatch nothing is driving nor one
// that is doing something. Said in those words rather than as an age of
// nothing.
let Some(progress) = activity.progress else {
return format!("nothing recorded yet{alive}");
};
let counted = format!(
"{} event(s), {} ago{alive}",
progress.events(),
ago(progress.last_at())
);
match &activity.doing {
Some(doing) => format!("now {doing} ({counted})"),
// Absent rather than guessed: the dispatch has recorded something and
// has not named a tool, so the count and the age are the whole of what
// this line can claim.
None => counted,
}
}
/// Whether this host can prove a recorded dispatch is still running.
///
/// Three answers, because a row an operator acts on has to distinguish them.
/// `Stale` and `Held` are proofs in opposite directions; `Unproven` is the
/// answer this host does not have, and collapsing it into either is how a
/// registry row outlives the process it describes.
#[derive(Debug, Clone, PartialEq, Eq)]
enum Proof {
/// The dispatch's own registry entry names a live pid, and that pid started
/// when the entry says it did.
///
/// Deliberately not named for liveness: what this establishes is that the
/// evidence agrees, to the resolution the host reports a process start at —
/// one second, where that is `ps`. A pid reused *inside* that resolution by
/// a process that also started then is the one case this cannot separate,
/// and it is why the word is about the record rather than about the work.
Held,
/// This host proved the process behind the row is gone.
Stale(String),
/// This host cannot decide: the dispatch was recorded elsewhere, the
/// registry cannot be read, or it holds no entry for the node at all.
Unproven(String),
}
/// What a run's own dispatch registry says about the processes its work is in.
///
/// The registry, and not the ownership lock: the lock names the **driver**, which
/// starts the dispatch and does not outlive it, and judging a row by it answered
/// about the wrong process — a run stopped and then adopted carries its stop for
/// ever, so its fresh dispatches all read as ended. [`ledger::DispatchRecord`] is
/// what the run writes for each process its work is in, and what a `stop` aims
/// at, so a row here and the teardown an operator runs after reading it are
/// decided from one record.
// llmlint: ignore-block[invalid_states_unrepresentable] the key is the node id
// `DispatchRecord::node` carries, and a node id is the plain `String` every identifier in
// this crate is, for the reason `src/error.rs`'s file-level suppression states. The value
// is not this reader's to check either: it comes back from `ledger::dispatches_of`, which
// is the boundary that refuses an entry nothing may be acted on.
enum Registry {
/// What the registry holds, by the node each entry names.
Read(BTreeMap<String, Vec<ledger::DispatchRecord>>),
/// The registry could not be read, and this is why.
///
/// Not an absence: a run whose registry this build cannot enumerate is one
/// nobody can say is idle, which is the whole reading this view exists to
/// stop being made.
Unreadable(String),
}
// llmlint: ignore-end[invalid_states_unrepresentable]
impl Registry {
/// Read one run's registry, keeping the refusal where there is one.
fn of(paths: &RunPaths) -> Self {
match ledger::dispatches_of(paths) {
Ok(records) => {
let mut by_node: BTreeMap<String, Vec<ledger::DispatchRecord>> = BTreeMap::new();
for record in records {
by_node.entry(record.node.clone()).or_default().push(record);
}
Self::Read(by_node)
}
Err(error) => Self::Unreadable(error.to_string()),
}
}
/// What this registry proves about one node's dispatch.
///
/// **Most certain first, and a live entry wins.** One node can hold more than
/// one entry — a dispatch that was retried inside the run leaves the ended
/// attempt's record behind until the process it named is collected — so a
/// stale entry beside a live one says the older attempt ended, never that the
/// node has nothing running.
fn proves(&self, node: &str) -> Proof {
let by_node = match self {
Self::Unreadable(why) => {
return Proof::Unproven(format!(
"the run's dispatch registry cannot be read: {why}"
))
}
Self::Read(by_node) => by_node,
};
let records = by_node.get(node).map(Vec::as_slice).unwrap_or_default();
let mut unproven = None;
let mut stale = None;
for record in records {
match proof_of(record) {
Proof::Held => return Proof::Held,
Proof::Unproven(why) => unproven.get_or_insert(why),
Proof::Stale(why) => stale.get_or_insert(why),
};
}
match (unproven, stale) {
(Some(why), _) => Proof::Unproven(why),
(None, Some(why)) => Proof::Stale(why),
// No entry at all: the run says a dispatch is in flight and no
// process claims it. Deliberately not a proof of staleness — the
// entry is written by the executor as the dispatch starts, so a
// reader between those two writes would report live work as ended.
(None, None) => Proof::Unproven(
"the run's dispatch registry holds no entry for it, so nothing here says which \
process it is in"
.to_string(),
),
}
}
}
/// Whether this host can prove the process one registry entry names is still the
/// dispatch it was recorded for.
///
/// Its pid says which process, and its start token says the pid is still that
/// process — a pid alone is what a two-day-old entry has, and a reused one
/// answers a liveness probe as alive.
fn proof_of(record: &ledger::DispatchRecord) -> Proof {
if record.host != sys::hostname() {
return Proof::Unproven(format!(
"its dispatch runs on {}, and a pid means nothing across machines",
record.host
));
}
if !sys::process_may_be_live(record.pid) {
return Proof::Stale(format!("its dispatch (pid {}) is gone", record.pid));
}
match sys::process_start_token(record.pid) {
// llmlint: ignore-block[changed_behavior_has_e2e] the host declining to answer is a
// property of the machine rather than of anything a user types. The answers it does
// give, and the other unproven arms that resolve alike, are driven end to end.
None => Proof::Unproven(format!(
"this host will not say when pid {} started",
record.pid
)),
// llmlint: ignore-end[changed_behavior_has_e2e]
Some(token) if token.matches(&record.started) => Proof::Held,
Some(_) => Proof::Stale(format!(
"pid {} is a different process from the one its dispatch was recorded in",
record.pid
)),
}
}
/// `onepipeline host` — every live dispatch on this host, across every planner.
///
/// A row here is a claim that a dispatch exists **now**, and it is acted on: an
/// operator leaves it alone, or ends it. So a row is rendered as live only where
/// this host can prove the process the dispatch is in is still that process —
/// the pid its own registry entry names, and the start token beside it. The run's
/// ownership lock is deliberately *not* what decides this — it names the driver,
/// and a driver is not the work; the reasoning is on the private reader that
/// replaced it. A row
/// proved to have nothing behind it is dropped and counted, and one this host
/// cannot decide either way is rendered saying so. Never a bare row that reads
/// as live work.
pub fn host(survey: &Survey) -> String {
let mut out = format!("host {}\n", sys::hostname());
// The scope of the claim. This scan has an under-reporting direction it
// cannot see past — a run recorded under another runs root is a live
// dispatch this view will never list — and a reader who does not know which
// root was read cannot tell that absence from an idle host.
out.push_str(&format!(
" reading {}\n",
one_line(&survey.root.display().to_string())
));
let mut rendered = false;
let mut ignored: Vec<String> = Vec::new();
for view in &survey.views {
let registry = Registry::of(&view.paths);
for (id, status) in &view.state.statuses() {
if *status != NodeStatus::Running {
continue;
}
let proof = registry.proves(id);
if let Proof::Stale(why) = &proof {
ignored.push(one_line(&format!("{}/{id}: {why}", view.paths.run)));
continue;
}
let age = view
.state
.dispatched_at
.get(id)
.map(|at| sys::now_millis().saturating_sub(*at))
.unwrap_or(0);
rendered = true;
out.push_str(&format!(
" {:<24} {:<20} {:<16} {}",
view.paths.run,
id,
view.launch.launcher,
crate::telemetry::duration(age)
));
if let Proof::Unproven(why) = &proof {
out.push_str(&format!(" UNPROVEN: {}", one_line(why)));
}
out.push('\n');
}
}
if !rendered {
out.push_str(" no live dispatches\n");
}
if !ignored.is_empty() {
out.push_str(&format!(
" {} stale registry entr{} ignored: {}\n",
ignored.len(),
if ignored.len() == 1 { "y" } else { "ies" },
ignored.join("; ")
));
}
out.push_str(&skipped_lines(&survey.skipped));
out
}
/// One run's merged store as one reader is shown it.
///
/// **Read-time only.** The store is not touched and nothing is recorded: two
/// readers of the same run see it through different profiles and neither loses
/// an event the other keeps, which is the whole difference between this and the
/// source filters a launch passes through to `oneagentgraph` and `onevcs`.
///
/// Borrowed rather than cloned where nothing is dropped, because the common case
/// — `--all`, and the shipped `detailed` profile — is a filter that admits
/// everything.
pub fn shaped<'a>(view: &'a RunView, filter: &EventFilter) -> Vec<&'a Envelope> {
view.events
.iter()
.filter(|event| filter.matches(event))
.collect()
}
/// `onepipeline monitor` — one pass over the merged stream.
///
/// The first line is the contract, not a banner: every event line carries the
/// typed id a detail lookup resolves, and `monitor` never tries to *be* the
/// detail.
pub fn monitor(view: &RunView, filter: &EventFilter) -> String {
monitor_of(view, &view.events, filter)
}
/// The same pass over a slice of the store the caller read itself.
///
/// What the `monitor` verb renders through, because the resume line it ends on
/// is a byte of the journal and the events above that line have to come out of
/// the one read that byte was taken from — [`RunView::open`]'s own read is a
/// different moment, and a record appended between the two would be one the
/// cursor stepped past without it ever being rendered.
pub(crate) fn monitor_of(view: &RunView, events: &[Envelope], filter: &EventFilter) -> String {
let mut out = String::from(
"Concise graph events; ask the producing library for full detail by stream id.\n",
);
for event in events.iter().filter(|event| filter.matches(event)) {
out.push_str(&event_line(view, event));
out.push('\n');
}
// The run's own state has no node, so it has no graph id: it reaches the
// reader as a trailer rather than as an event line.
out.push_str(&format!(
"-- {} {} {} {}\n",
view.paths.run,
view.summary(),
liveness_word(view),
graph::state_of(&view.state.statuses()).as_str()
));
out
}
/// One event as a reader is shown it: the typed id a detail lookup resolves,
/// what the event says, and what this crate knows about it since.
///
/// Crate-visible and without its terminator, because two verbs render the same
/// line and only one of them is building a document: `monitor` joins these into
/// one pass over the store, and `watch` writes one as each record arrives. A
/// second spelling of the line would let a supervisor's live view and their
/// replay of the same run disagree about what happened.
pub(crate) fn event_line(view: &RunView, event: &Envelope) -> String {
let id = match event.source {
Source::Pipeline => format!("graph:{}", event.labels.node.as_deref().unwrap_or("-")),
Source::Agentgraph => format!("agent:{}", event.stream),
Source::Vcs => format!("vcs:{}", event.stream),
};
format!(
"{} {:<28} {}{}",
event.ts,
id,
summarize(event),
superseded_suffix(view, event)
)
}
/// What a line about a superseded node says beyond the event itself, or nothing
/// for every other line.
///
/// The stream keeps the `node-settled` that failed the node, and it is what the
/// run's own observer reads — see
/// [`crate::projection::RunState::superseded`] for what that observer did with an
/// unqualified one.
///
/// Only the records *about the node* carry it, which keeps it a qualification
/// rather than a banner: a relayed envelope of a sibling's is about the dispatch
/// that ran, and the supersession is this crate's own answer about the node.
fn superseded_suffix(view: &RunView, event: &Envelope) -> String {
if event.source != Source::Pipeline {
return String::new();
}
let Some(node) = event.labels.node.as_deref() else {
return String::new();
};
match view.state.superseded.get(node) {
Some(replacement) => format!(" — superseded, retried as {}", one_line(replacement)),
None => String::new(),
}
}
/// One control-stripped line derived from an event's recorded values.
fn summarize(event: &Envelope) -> String {
const CAP: usize = 96;
let mut detail = event.kind.0.clone();
let surface = matches!(
event.kind.0.as_str(),
"planner-surface-queued" | "planner-surfaced"
);
if surface {
for key in ["kind", "message", "source", "blocking"] {
if let Some(value) = event.payload.get(key) {
detail.push_str(&format!(" {key}={value}"));
}
}
}
// `landing` beside `status`: a `node-settled` and a `published` both carry
// it, and a `monitor` line that showed only `done` said the same thing about a
// merge and about an open change request.
for key in ["status", "landing", "outcome", "state", "message", "reason"] {
if surface && key == "message" {
continue;
}
if let Some(value) = event.payload.get(key).and_then(|v| v.as_str()) {
detail.push_str(&format!(" {value}"));
}
}
let stripped: String = detail
.chars()
.map(|c| if c.is_control() { ' ' } else { c })
.collect();
if surface {
return stripped;
}
if stripped.chars().count() <= CAP {
return stripped;
}
stripped.chars().take(CAP).collect()
}
/// How a landing reads on a rendered line.
///
/// A phrase rather than the bare word, because it has to say **what decided it**:
/// a node that settled `done (queued)` was still reporting the settlement's
/// answer hours after its change had merged, and a supervisor re-dispatched three
/// nodes against a base that already carried their work.
///
/// `None` where the node never had a change to land, which is what its absence
/// on every other node's line has always meant.
fn landed_phrase(reported: &Reported, settled_at: Option<u64>) -> Option<String> {
let ago = match settled_at {
Some(at) => format!(
" {} ago",
crate::telemetry::duration(sys::now_millis().saturating_sub(at))
),
// A settlement whose moment the ledger does not carry: the claim is
// still the settlement's, and saying *when* would be inventing one.
None => String::new(),
};
match reported {
// The run's own observation, and the reason no read was taken.
Reported::TheRunObserved => {
Some("landed on its base — the run observed the change reach it".to_string())
}
// The operator's statement, on the tier it rests on and at the reference
// it names, so a reader can tell it from an observation and go and read
// the same evidence.
Reported::Stated(stated) => Some(format!(
"landed on its base — {tier}: stated from evidence at {reference}",
tier = stated.tier(),
reference = stated.reference()
)),
Reported::ReadNow { settled, read } => Some(match read.as_ref() {
LandingRead::Answered(landed) => match landed {
onevcs::Landed::Yes { .. } => {
format!("landed on its base — read now: {}", evidence(landed))
}
onevcs::Landed::InPart { .. } | onevcs::Landed::No => format!(
"NOT landed: read now, {} — open the change for where it is",
evidence(landed)
),
onevcs::Landed::Unknown => undecided(&evidence(landed), settled, &ago),
},
// The one answer nothing gave, said in its own words rather than in
// either of the other two — with the settlement's own dated claim
// behind it, which is what it always was.
LandingRead::Refused { because } => undecided(because, settled, &ago),
}),
// No branch to ask about, so the settlement's own claim is all there has
// ever been: dated, and saying that nothing has moved it.
Reported::UnlandedAtSettlement => Some(format!(
"NOT landed: the change had not reached its base when this settled{ago}, and \
no later read has said otherwise — open the change for where it is now"
)),
Reported::NoChangeToLand => None,
}
}
/// How a landing nothing could decide reads, with the settlement's own dated
/// claim behind it where there was one.
///
/// One phrasing for the two ways a read fails to decide — the sibling answering
/// that it cannot tell, and the read refusing outright — because what a reader
/// does about either is the same: open the change and look.
fn undecided(because: &str, settled: &Settled, ago: &str) -> String {
match settled {
Settled::Unlanded => format!(
"landing UNDECIDED: read now, {because}; it had not reached its base when this \
settled{ago} — open the change for where it is now"
),
Settled::Nothing => {
format!("landing UNDECIDED: read now, {because} — open the change for where it is")
}
}
}
/// What decided one of the sibling's answers, in the words a line prints.
///
/// Composed here rather than in `crate::vcs`, because it is prose for a rendered
/// line: the tier is that library's own word for it, and the commit and the count
/// beside it are what a reader deciding what to do next needs.
fn evidence(landed: &onevcs::Landed) -> String {
match landed {
onevcs::Landed::Yes { evidence } => format!("{} at {}", landed.tier(), evidence.commit()),
onevcs::Landed::InPart { evidence, unlanded } => format!(
"{} at {}, with {unlanded} commit(s) above it the landing did not carry",
landed.tier(),
evidence.commit()
),
onevcs::Landed::No => format!(
"{}: the base does not carry what this branch changed",
landed.tier()
),
onevcs::Landed::Unknown => format!(
"{}: the base already carries what this branch changed, and nothing records why",
landed.tier()
),
}
}
/// What a view says about the records the run's own store does not hold whole,
/// or nothing when it holds them all.
///
/// Every line either view prints is folded from that store, so a loss inside it
/// is the one fact that makes the rest of them unprovable — a `node-settled`
/// nobody can read renders as a node that never settled, and an `edit-committed`
/// nobody can read renders as a node the run never had. It is said here because
/// the only place it used to be said was the driver's stderr, which a detached
/// run writes to a log file nobody opens.
fn journal_loss_line(view: &RunView) -> String {
let integrity = crate::journal::integrity(&view.paths.journal());
if integrity.is_whole() {
return String::new();
}
format!(
" journal: {} — this run's record of itself is incomplete\n",
integrity.phrase()
)
}
/// `onepipeline results` — per-node outcomes, with each node's own evidence.
// llmlint: ignore[contracts_have_one_source_or_a_drift_gate] the second source is `docs/contract-divergences.md` entry 33, and the gate reconciling it is `tests/contract.rs`: it refuses an entry that is neither marked resolved with the contract naming what its ruling adopted, nor marked open with the proposal it waits on. Entry 33 is open and states this exact proposal, because the contract is approved verbatim and amending it is the planner's, never this repository's.
pub fn results(view: &RunView) -> String {
let render = rendering(Rendered::Results, &view.paths.run);
// The run and how its graph stands — deliberately not the node tally the
// other views carry, because every line under this one is a node's own
// status and a header that also said `done` would read as one of them.
let mut out = format!(
"{} {}\n",
view.paths.run,
graph::state_of(&view.state.statuses()).as_str()
);
let statuses = view.state.statuses();
for node in view.state.graph.iter() {
let status = statuses
.get(&node.id)
.copied()
.unwrap_or(NodeStatus::Pending);
out.push_str(&format!(" {:<24} {}", node.id, status.as_str()));
if let Some(outcome) = view.state.outcomes.get(&node.id) {
out.push_str(&format!(" ({outcome})"));
}
// Beside the status word for the same reason `status` gives it a line:
// `parked` on its own says the planner idled the node, and says nothing
// about the dispatch still running for it.
if let Some(pending) = cancelling_for(&view.state, &node.id) {
out.push_str(&format!(" — cancelling, asked to stop {pending} ago"));
}
// Beside the status word, because it is the fact the status word does
// not carry: `done` is the same for a change that merged and one still
// sitting in a pull request. Rendered for both landings rather than only
// the unlanded one — a reader who sees the qualifier where a change was
// open and nothing where it merged is reading the absence, which is what
// every other node's absence already means.
let reported = reported_landing(view, &render, &node.id);
if let Some(phrase) = landed_phrase(&reported, view.state.settled_at.get(&node.id).copied())
{
out.push_str(&format!(" — {phrase}"));
}
// The attestation settles the node, so the status word alone would
// report a dispatch that failed as one that succeeded. Both records
// ride the line instead: what this run got, and what a person said
// afterwards — which is also what released everything under it.
if attested_after_failing(view, &node.id) {
out.push_str(" — settled failed, attested as landed");
}
if status == NodeStatus::Running && view.state.stop_recorded() {
out.push_str(&format!(" — {}", became_of_the_worker(&view.state)));
}
// What the dispatch reported, before what the plan asked for: an
// unpinned lifecycle node's branch is named by the sibling that cut it,
// so the plan does not know it and a reader looking for the work would
// find nothing.
let branch = view
.state
.branches
.get(&node.id)
.or(node.branch.as_ref())
.cloned();
if let (NodeStatus::Parked | NodeStatus::Failed | NodeStatus::Cancelled, Some(branch)) =
(status, &branch)
{
out.push_str(&format!(" — preserved on {branch}"));
}
// Where the dispatch an adoption cleared had got to. The node itself is
// dispatched again and settles under its own words, and none of them say
// that an *earlier* dispatch committed work somewhere: the driver that
// was running it exited without settling anything, so this line is the
// only place that branch and that session are ever named.
if let Some(session) = view.state.abandoned.get(&node.id) {
out.push_str(&format!(
" — a dispatch was abandoned when the run was adopted; its work is on {} \
(onevcs session {})",
session.branch(),
session.token().0
));
}
// The one piece of evidence a person actually opens.
if let Some(url) = view.state.change_urls.get(&node.id) {
out.push_str(&format!(" — {url}"));
}
out.push('\n');
// Before the detail, because it is what the detail is evidence *for*: the
// detail is the producer's own sentence about how the dispatch ended, and
// this says what that means for the node — which used to be a thing a
// manager derived by opening `events.jsonl` and counting commits.
if let Some(died) = death_phrase(&view.state, &node.id, branch.as_deref()) {
out.push_str(&format!(" died: {died}\n"));
}
if let Some(detail) = settled_detail(view, &node.id) {
out.push_str(&format!(" detail: {}\n", one_line(&detail)));
}
// What this node is judged against beyond its own task prose. Under the
// node rather than beside it, because it is prose rather than a word,
// and rendered before the settlement's own reasons because it is what
// those reasons were reached against.
if let Some(amendment) = &node.amendment {
out.push_str(&format!(" amendment: {}\n", one_line(amendment)));
}
// Why the judge failed it, then which chains ran out, then which
// recovered — in that order, because that is the order they matter in.
//
// The verdict is first because it is the thing that actually failed the
// node, and it used to be reachable only by opening the node's retained
// report by hand while three provider lines sat above it pointing
// somewhere else. A chain that ran out gets the `provider` line, which
// is a retry aimed at a subscription; a chain that recovered gets its
// own word, because a fix aimed at it changes nothing.
if status == NodeStatus::Failed {
for verdict in crate::report::failed_verdicts(&view.events, &node.id) {
out.push_str(&format!(" verdict: {}\n", verdict_phrase(&verdict)));
}
let chains = chain_records(&view.state, &node.id);
for record in chains
.iter()
.filter(|record| record.became == Fallthrough::Refused)
{
out.push_str(&format!(" provider: {}\n", chain_phrase(record)));
}
for record in chains
.iter()
.filter(|record| record.became != Fallthrough::Refused)
{
out.push_str(&format!(" fallback: {}\n", chain_phrase(record)));
}
}
// Why the run never asked this node to do anything. `skipped` on its own
// says a dependency of *some* kind went wrong and leaves a reader to
// rebuild the graph by hand to find which — which is how a node stayed
// skipped over work that had already merged. The dependency is known at
// the moment the skip is derived, so it is named where the skip is
// reported.
if status == NodeStatus::Skipped {
out.push_str(&format!(
" never attempted; skipped by: {}\n",
skipped_by_phrase(&graph::skipped_by(
&view.state.graph,
&statuses,
&view.state.settled(),
&node.id
))
));
}
if status == NodeStatus::Waiting {
if let Some(task) = &node.task {
out.push_str(&format!(" action: {task}\n"));
}
let unblocks = graph::unblocks(&view.state.graph, &node.id);
if !unblocks.is_empty() {
out.push_str(&format!(" unblocks: {}\n", unblocks.join(", ")));
}
}
}
out.push_str(&superseded_lines(view));
// Under the graph, because a hook is the run's rather than any node's.
out.push_str(&crate::hooks::results_lines(view));
out.push_str(&journal_loss_line(view));
out
}
/// What the run's results say about the nodes a `retry` replaced, or nothing
/// where it replaced none.
///
/// **Under the graph, because they are not in it** — every line above is a node
/// the run is still executing, and these vanished from the results altogether;
/// [`crate::projection::RunState::superseded`] is what that cost. The line says
/// what became of the node rather than what its last dispatch scored, and names
/// the replacement because that is where the work went: a supersession inherits
/// the branch, so the replacement's own line above is where the work is.
fn superseded_lines(view: &RunView) -> String {
let mut out = String::new();
for (node, replacement) in &view.state.superseded {
out.push_str(&format!(
" {:<24} superseded — retried as {}\n",
one_line(node),
one_line(replacement)
));
}
out
}
/// `onepipeline transcript` — one dispatch's turns, its tools, and its words.
///
/// Two sources, because they answer at different times and neither is the whole
/// answer. The merged store carries every `turn-activity` as it arrives, so a
/// turn's tools are readable *while it runs*; the onejudge report a
/// `member-settled` retained carries the conversation itself, which is what a
/// reader asking why a turn did what it did needs, and which exists only once
/// the member has settled.
///
/// A report this run did not keep a copy of is said to be unretained rather
/// than passed over: an absent transcript and an unread one are different facts.
/// The path the producer named is printed and **never opened** — the only file
/// this verb reads is the run's own copy, made when the settlement was ingested.
pub fn transcript(view: &RunView, only: Option<&str>) -> String {
let mut out = String::new();
// Derived once for the whole run rather than per node.
let settlements = crate::report::evidence(&view.paths, &view.events);
for node in nodes_with_agent_records(view, only) {
// Every value on a rendered line is a stranger's: a node label a
// producer stamped, a member it named, a path it chose, a role its
// report carried. One control character in any of them rewrites the
// line around it, so they all go through the same strip.
out.push_str(&format!("{} {}\n", view.paths.run, one_line(&node)));
for event in view
.events
.iter()
.filter(|event| event.source == Source::Agentgraph)
.filter(|event| event.labels.node.as_deref() == Some(node.as_str()))
{
let field = |key: &str| {
event
.payload
.get(key)
.and_then(|value| value.as_str())
.unwrap_or_default()
};
match event.kind.0.as_str() {
"turn-started" => out.push_str(&format!(
" turn {}\n",
event
.payload
.get("turn")
.map_or_else(|| "-".to_string(), ToString::to_string)
)),
"turn-activity" => out.push_str(&format!(
" {} {} {}\n",
one_line(field("kind")),
one_line(field("name")),
// Read out of the payload the same way a retained report's
// event is: a result's text is under `output` and it carries
// no `detail` at all, so a third column read out of `detail`
// was blank on every observation a turn made.
tool_text(&ToolText::of(field("kind"), |key| event.payload.get(key)))
)),
_ => {}
}
}
for settled in settlements
.iter()
.filter(|settled| settled.node.as_deref() == Some(node.as_str()))
{
// Named by the member that settled with it: a graph runs more than
// one, and a reader looking at two reports has to know whose is
// whose. The path is the producer's own, printed so a reader knows
// what the settlement claimed — and it is not what is opened.
out.push_str(&format!(
" report {} {}\n",
one_line(settled.member.as_deref().unwrap_or("-")),
one_line(&settled.named.display().to_string())
));
let Some(document) = crate::report::read(&settled.kept) else {
out.push_str(
" not retained by this run, so it is not read: only this run's own \
copy of a report is ever opened\n",
);
continue;
};
let turns = crate::report::turns(&document);
if turns.is_empty() {
out.push_str(" it carries no transcript\n");
}
for turn in turns {
out.push_str(&format!(" {}\n", one_line(&turn.role)));
for line in turn.text.lines() {
out.push_str(&format!(" {}\n", one_line(line)));
}
for tool in turn.tools {
out.push_str(&format!(
" {} {} {}\n",
one_line(&tool.kind),
one_line(&tool.name),
tool_text(&tool.text)
));
}
}
}
}
if out.is_empty() {
out.push_str("no dispatch has recorded a transcript\n");
}
out
}
/// The nodes this run's merged store carries an `oneagentgraph` record for, in
/// id order.
///
/// Any record, not only a settled turn: a node whose dispatch is still running
/// has a transcript worth reading, and that is most of what this verb is for.
///
/// Crate-visible: `docs/contract.md` names the views, not the parts one is
/// assembled from, and a public item the contract does not name is a promise
/// this crate did not make.
pub(crate) fn nodes_with_agent_records(view: &RunView, only: Option<&str>) -> Vec<String> {
let mut nodes: Vec<String> = view
.events
.iter()
.filter(|event| event.source == Source::Agentgraph)
.filter_map(|event| event.labels.node.clone())
.filter(|node| only.is_none_or(|wanted| wanted == node))
.collect();
nodes.sort_unstable();
nodes.dedup();
nodes
}
/// How much of a tool's own output one transcript line prints.
///
/// Explicit rather than incidental, because the two sources this verb reads are
/// bounded differently. A `turn-activity` summary is inside
/// [`MAX_PAYLOAD_TEXT_BYTES`](crate::event::MAX_PAYLOAD_TEXT_BYTES) by the time
/// it is in the store — this crate's own promise about its own envelope, cut and
/// marked at ingest — while a retained report's output is the harness's raw
/// bytes with nothing bounding them at all: reports on this host carry single
/// outputs past sixty kilobytes, and one of those splatted whole into a terminal
/// is the rendering this ceiling exists to prevent.
///
/// So it is that same bound, read as a **character** ceiling against a byte one:
/// a text inside 4096 bytes is inside 4096 characters, which is what makes this
/// unable to cut anything the run's own journal kept. What it does cut on the
/// report path it counts out loud.
const MAX_TOOL_OUTPUT_CHARS: usize = crate::event::MAX_PAYLOAD_TEXT_BYTES;
/// The third column of a tool's line: what a call acted on, or what a result
/// returned.
///
/// Whichever half the event is, because a `tool_result` carries its text under
/// `output` and carries no `detail` at all — reading only the latter rendered
/// every observation a turn made as a blank column, which is a run's own
/// evidence hidden by its reader. Either text is a stranger's, so either goes
/// through the same control-character strip every other borrowed value on these
/// views does.
///
/// **What is not printed is said rather than dropped.** An output the producer
/// had already cut short says so, and one this view cuts says how much of it a
/// reader is looking at — a truncated output rendered as though it were whole is
/// how a reader concludes a tool returned nothing further, which is the reading
/// this verb exists to correct. A result that returned nothing renders as the
/// empty column it is, and says nothing about what it left out, because it left
/// nothing out.
fn tool_text(text: &ToolText) -> String {
let (output, truncated) = match text {
ToolText::Acted(detail) => return one_line(detail),
ToolText::Returned { output, truncated } => (output, *truncated),
};
if output.is_empty() {
return String::new();
}
let stripped = one_line(output);
let whole = stripped.chars().count();
let mut text: String = stripped.chars().take(MAX_TOOL_OUTPUT_CHARS).collect();
let mut notes: Vec<String> = Vec::new();
if whole > MAX_TOOL_OUTPUT_CHARS {
notes.push(format!("{MAX_TOOL_OUTPUT_CHARS} of {whole} characters"));
}
match truncated {
Truncation::Whole => {}
Truncation::Cut => notes.push("already cut short by the producer".to_string()),
// Not silence, and not `false`: a flag this build cannot read leaves
// whether the output is whole unanswered, and a line that said nothing
// would be answering it.
Truncation::Unreadable => {
notes.push("the producer's truncation flag is unreadable".to_string());
}
}
if !notes.is_empty() {
text.push_str(&format!(" … [{}]", notes.join("; ")));
}
text
}
/// One control-stripped line, so a relayed value cannot rewrite the rendering
/// around it.
///
/// Shared with the settlements this crate *composes* out of a sibling's values —
/// `crate::lifecycle` names the ref a publication compared against — so the rule
/// is one rule rather than one per place a sibling's text reaches a line.
pub(crate) fn one_line(text: &str) -> String {
text.chars()
.map(|c| if c.is_control() { ' ' } else { c })
.collect()
}
/// `onepipeline goals` — what each run is for, and how far it has got.
pub fn goals(survey: &Survey) -> String {
let mut out = String::new();
for view in &survey.views {
let goal = view
.state
.plan
.as_ref()
.and_then(|plan| plan.goal.as_ref())
.map(|goal| goal.text.clone())
.unwrap_or_else(|| crate::plan::NO_GOAL.to_string());
out.push_str(&format!(
"{} {}\n {}\n {}\n",
view.paths.run,
liveness_word(view),
goal,
view.summary()
));
// The repository identities this run holds, so two planners can see
// whether they would share a checkout.
let mut repos: Vec<&str> = view
.state
.graph
.iter()
.filter_map(|node| node.repo.as_deref())
.collect();
repos.sort_unstable();
repos.dedup();
if !repos.is_empty() {
out.push_str(&format!(" identities: {}\n", repos.join(", ")));
}
}
if out.is_empty() {
return nothing_to_report(survey);
}
out.push_str(&skipped_lines(&survey.skipped));
out
}
#[cfg(test)]
mod tests {
use super::*;
use crate::event::{EventKind, Labels, ENVELOPE_VERSION};
use crate::filter::Filters;
use crate::plan::{Node, Plan, PLAN_SCHEMA_VERSION};
use serde_json::json;
use std::path::PathBuf;
fn scratch(name: &str) -> PathBuf {
let dir = std::env::temp_dir().join(format!("onepipeline-views-{name}-{}", sys::pid()));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).expect("a scratch root");
dir
}
/// Each of the four answers `onevcs` gives reads back as one of the three
/// things a line can say, with the evidence that decided it.
///
/// Two of the four are exactly the readings that must not be collapsed: a
/// landing the branch has gone past is **not** finished work — the sibling's
/// own `is_landed` says so, and the commits above it are work still to
/// publish — and a base that carries the change with nothing recording why is
/// **undecided** rather than landed. Driven through the answer rather than
/// through a repository, because what is decided here is which answer is
/// which; `tests/e2e/landing.rs` drives the repository.
#[test]
fn each_answer_the_sibling_gives_reads_back_as_one_of_the_three_with_its_evidence() {
use crate::vcs::LandingRead;
use onevcs::{Landed, LandingEvidence, Sha};
let read = |landed: Landed| Reported::ReadNow {
settled: Settled::Unlanded,
read: std::rc::Rc::new(LandingRead::Answered(landed)),
};
let recorded = LandingEvidence::RecordedLanding {
commit: Sha("abc1234".into()),
};
let yes = Landed::Yes {
evidence: recorded.clone(),
};
assert_eq!(read(yes.clone()).stands(), Stands::Landed);
assert_eq!(evidence(&yes), "a recorded landing at abc1234");
let part = Landed::InPart {
evidence: recorded,
unlanded: std::num::NonZeroUsize::new(2).expect("two"),
};
assert_eq!(read(part.clone()).stands(), Stands::NotLanded);
assert_eq!(
evidence(&part),
"a recorded landing at abc1234, with 2 commit(s) above it the landing did not carry"
);
assert_eq!(read(Landed::No).stands(), Stands::NotLanded);
assert!(evidence(&Landed::No).contains("the base does not carry"));
assert_eq!(
read(Landed::Unknown).stands(),
Stands::Undecided,
"a base that carries the change with nothing recording why was read as an answer"
);
assert!(evidence(&Landed::Unknown).contains("nothing records why"));
// And a read that got no answer at all is undecided too.
assert_eq!(
Reported::ReadNow {
settled: Settled::Nothing,
read: std::rc::Rc::new(LandingRead::Refused {
because: "this host could not decide it".into()
}),
}
.stands(),
Stands::Undecided
);
}
fn plan() -> Plan {
Plan {
schema_version: PLAN_SCHEMA_VERSION,
goal: Some(crate::plan::Goal {
text: "close the coverage gap".into(),
}),
name: Some("demo".into()),
concurrency: 4,
tasks: vec![Node {
id: "build".into(),
persona: Some("engineer".into()),
task: Some("## What\ndo it".into()),
..Node::default()
}],
}
}
fn launch(pid: u32) -> LaunchRecord {
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: "session-a".into(),
pid,
host: sys::hostname(),
started: sys::process_start_token(pid)
.map(|token| token.recorded().to_string())
.unwrap_or_default(),
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: Filters::default(),
bus_config: Default::default(),
envelope_reviewer_bar: Default::default(),
}
}
fn write_run(root: &Path, run: &str, pid: u32, events: &[Envelope]) -> RunPaths {
let paths = RunPaths::under(root, run);
paths.create().expect("the run directory");
let mut record = launch(pid);
record.run_id = run.to_string();
ledger::write_json(&paths.launch(), &record).expect("a launch record");
for event in events {
ledger::append_line(
&paths.journal(),
&serde_json::to_string(event).expect("an event"),
)
.expect("appended");
}
paths
}
/// One entry in a run's dispatch registry, as the executor writes it.
///
/// Written by hand rather than through `claim_dispatch` because what these
/// exercise is a reader meeting a record another process left behind — a pid
/// that has gone, one on another machine, one a reused pid now answers for —
/// and none of those is a state a claim this process takes can be in.
fn register(paths: &RunPaths, record: &ledger::DispatchRecord) {
std::fs::create_dir_all(paths.dispatches()).expect("the registry directory");
ledger::write_json(&paths.dispatch(record.pid, 0), record).expect("a registry entry");
}
/// The entry a live dispatch of this process leaves: the pid *and* the start
/// token that says the pid is still that process.
fn dispatched_here(node: &str) -> ledger::DispatchRecord {
ledger::DispatchRecord {
node: node.to_string(),
pid: sys::pid(),
host: sys::hostname(),
dispatched_at: sys::now_rfc3339(),
started: sys::process_start_token(sys::pid())
.map(|token| token.recorded().to_string())
.unwrap_or_default(),
}
}
/// A `monitor` renders from the same read of the journal its cursor is taken
/// from, not from the view's own earlier one.
///
/// A record is put between the two reads here, which no invocation of the
/// binary can be made to do on demand: had the events come from the view, the
/// cursor would step past that record and no later pass would render it.
#[test]
fn a_monitor_renders_a_record_appended_after_its_view_was_read_before_its_cursor_passes_it() {
let root = scratch("view-one-read");
let paths = write_run(
&root,
"demo",
sys::pid(),
&[event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan()))],
)],
);
let view = RunView::open(&paths).expect("the run reads");
let late = event(
crate::journal::PipelineKind::NodeReady,
Some("late-arrival"),
&[],
);
ledger::append_line(
&paths.journal(),
&serde_json::to_string(&late).expect("an event"),
)
.expect("appended");
assert!(
view.events
.iter()
.all(|seen| seen.labels.node.as_deref() != Some("late-arrival")),
"the view was read after the append, so this proves nothing"
);
let args = crate::cli::MonitorArgs {
read: crate::cli::ReadArgs {
run: "demo".to_string(),
filter: None,
all: true,
},
cursor: None,
};
let document =
crate::watch::monitor_document(&args, &paths, &view, &EventFilter::default())
.expect("the pass renders");
let held = std::fs::metadata(paths.journal())
.expect("the journal is there")
.len();
assert!(
document.ends_with(&format!("-- cursor 1:demo:{held}")),
"{document}"
);
assert!(
document.contains("graph:late-arrival"),
"the cursor passed a record this pass never rendered:\n{document}"
);
std::fs::remove_dir_all(&root).ok();
}
fn event(
kind: crate::journal::PipelineKind,
node: Option<&str>,
fields: &[(&str, serde_json::Value)],
) -> Envelope {
relayed(
EventKind(kind.as_str().into()),
Source::Pipeline,
node,
fields,
)
}
/// The same envelope, for a kind a *sibling* produced: those stay wire
/// strings, which is the half of the merged store this crate does not close.
fn relayed(
kind: EventKind,
source: Source,
node: Option<&str>,
fields: &[(&str, serde_json::Value)],
) -> Envelope {
Envelope {
v: ENVELOPE_VERSION,
ts: sys::now_rfc3339(),
stream: "s".into(),
seq: 0,
source,
kind,
dimensions: Default::default(),
labels: Labels {
run_id: Some("demo".into()),
round: Some(1),
node: node.map(str::to_string),
..Labels::default()
},
payload: crate::journal::payload(fields),
artifacts: Vec::new(),
}
}
fn dead_pid() -> u32 {
sys::reaped_pid()
}
/// The one line a supervisor may not filter out says *what* is waiting.
///
/// A blocking question behind a pile of routine updates is exactly the case
/// a bare count hid: this host's own history holds a handful of questions
/// against thousands of observer surfaces, and both rendered as a number.
/// So the kinds are named, the blocking one leads, and a queue of unrelated
/// kinds is summarised out loud rather than silently cut.
#[test]
fn the_unread_line_names_the_kinds_waiting_and_leads_with_a_blocking_one() {
let root = scratch("unread-kinds");
let paths = write_run(
&root,
"demo",
sys::pid(),
&[event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan()))],
)],
);
let channel = crate::channel::ChannelState::new(&paths);
let queue = |kind: &str, blocking: bool| {
channel
.push(crate::channel::Surface {
id: 0,
kind: kind.into(),
message: format!("something about {kind}"),
source: "proposal".into(),
blocking,
queued_at: sys::now_millis(),
abandoned: false,
asker: None,
workstream: None,
correlation: None,
})
.expect("the surface queues");
};
for _ in 0..6 {
queue("observation", false);
}
queue("planner-question", true);
for kind in ["edit-rejected", "quiet-worker", "check-in", "proposal"] {
queue(kind, false);
}
let view = RunView::open(&paths).expect("the run reads");
assert_eq!(view.unread_surfaces().0, 11);
let unread = view.unread();
// The blocking kind leads; then the rarest, so the common one that
// buries it never comes first; then the count that stands for the rest.
assert_eq!(
unread.phrase(),
"1 planner-question, 1 check-in, 1 edit-rejected, 1 proposal, and 2 other kind(s)"
);
let rendered = runs(&root, false, "session-a");
assert!(
rendered.contains("11 planner update(s) waiting (1 planner-question,"),
"{rendered}"
);
assert!(
status(&Survey::of(&root)).contains("1 planner-question,"),
"{}",
status(&Survey::of(&root))
);
std::fs::remove_dir_all(&root).ok();
}
/// A skip with no cause to name still says the node was never attempted.
///
/// The empty list is unreachable from a plan this crate executes, so the
/// phrase has no journey of its own — and it is held here rather than left
/// untested, because what it guards against is `results` printing a bare
/// `skipped by:` that a reader takes for a view that lost the fact.
#[test]
fn a_skip_with_no_cause_left_in_the_graph_is_still_phrased() {
assert_eq!(
skipped_by_phrase(&[]),
"a dependency this run can no longer name"
);
assert_eq!(
skipped_by_phrase(&[
("build".to_string(), NodeStatus::Failed),
("lint".to_string(), NodeStatus::Skipped),
]),
"build (failed), lint (skipped)"
);
}
#[test]
fn a_driver_this_host_can_prove_is_gone_reads_as_driver_dead() {
let root = scratch("dead");
write_run(
&root,
"demo",
dead_pid(),
&[event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan()))],
)],
);
let view = RunView::open(&RunPaths::under(&root, "demo")).expect("the run reads");
assert_eq!(view.liveness(), DriverLiveness::DriverDead);
assert!(view.liveness().is_undriven());
assert!(runs(&root, false, "session-a").contains("DRIVER DEAD"));
std::fs::remove_dir_all(&root).ok();
}
/// A live pid the host has handed on since the driver recorded it is a
/// driver that is gone, on both readings a run has.
///
/// The record names this test's own live process, stamped as a process it is
/// not — what a host that reissued a dead driver's pid leaves behind. Read on
/// the pid alone, that is a live driver, and it was: `adopt` refused a run
/// its own `start --attach` had settled as "still being driven" on Windows,
/// where a freed pid is handed on within moments. The stamp is what says
/// otherwise, and both the whole view and the bounded listing read it.
#[test]
fn a_live_pid_stamped_as_another_process_reads_as_driver_dead() {
let root = scratch("reissued");
let paths = write_run(
&root,
"demo",
sys::pid(),
&[event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan()))],
)],
);
let mut record: LaunchRecord = ledger::read_json(&paths.launch()).expect("the record");
record.started = "the driver it named, which is not this process".into();
ledger::write_json(&paths.launch(), &record).expect("the record is rewritten");
let view = RunView::open(&paths).expect("the run reads");
assert_eq!(view.liveness(), DriverLiveness::DriverDead);
assert!(view.liveness().is_undriven());
let listed = runs(&root, false, "session-a");
assert!(listed.contains("DRIVER DEAD"), "{listed}");
std::fs::remove_dir_all(&root).ok();
}
/// A run this quiet is parked — unless a decision is outstanding.
///
/// Both halves of the same setup, so the difference between them is only the
/// blocking surface: a live pid, and a last write old enough that silence
/// alone would park it.
fn quiet_run(root: &Path, run: &str) -> RunPaths {
let mut stale = event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan()))],
);
// Far older than `DEFAULT_PARKED_AFTER_SECONDS`, so the verdict does not
// depend on the threshold's environment override.
stale.ts = "2020-01-01T00:00:00Z".into();
write_run(root, run, sys::pid(), &[stale])
}
#[test]
fn a_live_driver_that_has_gone_quiet_with_nothing_outstanding_reads_as_parked() {
let root = scratch("quiet-parked");
let paths = quiet_run(&root, "demo");
let view = RunView::open(&paths).expect("the run reads");
assert_eq!(view.liveness(), DriverLiveness::Parked);
std::fs::remove_dir_all(&root).ok();
}
/// The same silence, with a blocking surface nobody has answered.
///
/// A decision point takes two forms, and `settlement_of` already reports this
/// run `awaiting-planner`. A liveness verdict that read only the graph's human
/// actions called the very same run `PARKED` — inviting an `adopt` that may
/// end its driver while the answer sits unread in a planner's queue.
#[test]
fn a_live_driver_quiet_behind_a_blocking_surface_reads_as_active() {
let root = scratch("quiet-blocking");
let paths = quiet_run(&root, "demo");
crate::channel::ChannelState::new(&paths)
.push(crate::channel::Surface {
id: 0,
kind: "blocker".into(),
message: "Node build needs a decision; proceed?".into(),
source: "watcher".into(),
blocking: true,
queued_at: sys::now_millis(),
abandoned: false,
asker: None,
workstream: Some("build".into()),
correlation: None,
})
.expect("the surface queues");
let view = RunView::open(&paths).expect("the run reads");
assert_eq!(view.liveness(), DriverLiveness::Driving);
assert!(!view.liveness().is_undriven());
std::fs::remove_dir_all(&root).ok();
}
#[test]
fn a_live_driver_that_is_writing_reads_as_active() {
let root = scratch("live");
write_run(
&root,
"demo",
sys::pid(),
&[event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan()))],
)],
);
let view = RunView::open(&RunPaths::under(&root, "demo")).expect("the run reads");
assert_eq!(view.liveness(), DriverLiveness::Driving);
assert!(!view.liveness().is_undriven());
std::fs::remove_dir_all(&root).ok();
}
/// A record that names **no driver** is not a driver this host can prove is
/// gone.
///
/// Two shapes, and the first is the one a guard on the host name alone would
/// miss: a record carrying this host and no pid at all. `0` is not a process
/// here, so a reader that probed it would prove the absence it was handed and
/// report `DRIVER DEAD` over a run it knows nothing about — and `DRIVER DEAD`
/// is what invites the `adopt` that displaces a driver. The second is the
/// record 141 roots on one host actually hold: none of the five keys, read
/// off disk exactly as a view meets it.
#[test]
fn a_launch_record_naming_no_driver_is_never_read_as_a_dead_one() {
let root = scratch("no-driver");
let live = event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan()))],
);
// The record as an older build left it: this host, and nothing that says
// which process on it.
let paths = write_run(&root, "unclaimed", dead_pid(), std::slice::from_ref(&live));
without(&paths, &["pid", "started"]);
let view = RunView::open(&paths).expect("a record naming no pid still reads");
assert_eq!(view.launch.driver_pid(), None);
assert_eq!(view.launch.recorded_host(), Some(sys::hostname().as_str()));
assert_eq!(
view.liveness(),
DriverLiveness::Driving,
"a pid nobody recorded was probed, and its absence read as a dead driver"
);
// And the whole historical shape: none of the five.
let older = write_run(&root, "oldest", dead_pid(), std::slice::from_ref(&live));
without(
&older,
&[
"session",
"pid",
"host",
"started",
"started_at",
"heartbeat_interval",
],
);
let view = RunView::open(&older).expect("a record predating all five still reads");
assert_eq!(view.launch.driver_pid(), None);
assert_eq!(view.launch.recorded_host(), None);
assert_eq!(view.launch.launched_at(), None);
assert_eq!(view.launch.pacemaker_interval(), None);
assert_ne!(view.liveness(), DriverLiveness::DriverDead);
// And both are on the view an operator opens, under the label an
// unattributed launch has always had.
let rendered = runs(&root, false, "session-a");
assert!(rendered.contains("unclaimed"), "{rendered}");
assert!(rendered.contains("oldest"), "{rendered}");
assert!(rendered.contains("[unknown]"), "{rendered}");
assert!(
!rendered.contains("run root(s) skipped"),
"a record an older build wrote was refused:\n{rendered}"
);
std::fs::remove_dir_all(&root).ok();
}
/// Take keys off a run's launch record, leaving the record an older build of
/// this crate wrote.
///
/// Edited on the file rather than built from the struct: what these read is a
/// *document* that never carried the key, and a record this build serialises
/// carries every one of them.
fn without(paths: &RunPaths, keys: &[&str]) {
let mut document: serde_json::Value =
serde_json::from_str(&std::fs::read_to_string(paths.launch()).expect("the record"))
.expect("a launch record");
let fields = document.as_object_mut().expect("a launch record");
for key in keys {
fields.remove(*key);
}
std::fs::write(paths.launch(), document.to_string()).expect("an older build's record");
}
#[test]
fn a_pid_recorded_on_another_host_never_reads_as_dead() {
let root = scratch("elsewhere");
let paths = RunPaths::under(&root, "demo");
paths.create().expect("the run directory");
let mut record = launch(dead_pid());
record.host = "some-other-host".into();
ledger::write_json(&paths.launch(), &record).expect("a launch record");
ledger::append_line(
&paths.journal(),
&serde_json::to_string(&event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan()))],
))
.expect("an event"),
)
.expect("appended");
let view = RunView::open(&paths).expect("the run reads");
assert_eq!(
view.liveness(),
DriverLiveness::Driving,
"a pid means nothing across machines"
);
std::fs::remove_dir_all(&root).ok();
}
#[test]
fn a_run_nobody_recorded_is_no_such_run() {
let root = scratch("missing");
let error = RunView::open(&RunPaths::under(&root, "nowhere")).unwrap_err();
assert!(matches!(error, crate::Error::NoSuchRun { .. }));
assert!(runs(&root, false, "session-a").contains("no runs recorded"));
std::fs::remove_dir_all(&root).ok();
}
#[test]
fn only_the_reader_sees_mine_and_a_foreign_run_is_labelled_by_digest() {
let root = scratch("owner");
write_run(
&root,
"demo",
sys::pid(),
&[event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan()))],
)],
);
let listing = runs(&root, false, "session-a");
assert!(listing.contains("[mine]"), "{listing}");
let foreign = runs(&root, false, "session-b");
assert!(!foreign.contains("[mine]"), "{foreign}");
assert!(
!foreign.contains("session-a"),
"{foreign} leaks the session id"
);
assert!(runs(&root, true, "session-b").contains("no runs recorded"));
std::fs::remove_dir_all(&root).ok();
}
/// A directory under the runs root that this build will not read is a
/// **rejection**, and every whole-root view names it beside the runs that
/// did read.
///
/// The reading it replaces: the same root listed one run and said nothing
/// about the other directory, so a planner could not tell a root holding one
/// run from a root holding one run and one it could not open.
///
/// **Unreadable** is what is named here, and it is not the same thing as
/// unfamiliar: a launch record carrying a key this build does not know is
/// read and listed, which is the run beside them below.
#[test]
fn a_run_root_this_build_refuses_is_named_rather_than_dropped() {
let root = scratch("skipped");
write_run(
&root,
"readable",
sys::pid(),
&[event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan()))],
)],
);
// A run whose launch record another build wrote, carrying a key this one
// has never had. It reads, and the run is on the view.
let newer = write_run(&root, "from-a-newer-build", sys::pid(), &[]);
let mut written: serde_json::Value = serde_json::from_str(
&std::fs::read_to_string(newer.launch()).expect("the launch record this build wrote"),
)
.expect("a launch record");
written["channel_id"] = json!("a field a later build removed");
std::fs::write(newer.launch(), written.to_string()).expect("a launch record");
std::fs::create_dir_all(root.join("no-launch")).expect("a directory with no launch");
// And one whose launch record is not a launch record: a document with
// none of what this build needs to say anything about the run.
let empty = RunPaths::under(&root, "not-a-record");
empty.create().expect("the run directory");
std::fs::write(empty.launch(), json!({"oops": true}).to_string())
.expect("a launch record this build cannot read");
let survey = Survey::of(&root);
assert_eq!(survey.views.len(), 2, "{:?}", survey.skipped);
assert_eq!(survey.skipped.len(), 2, "{:?}", survey.skipped);
for rendered in [
runs(&root, false, "session-a"),
status(&survey),
goals(&survey),
] {
assert!(rendered.contains("readable"), "{rendered}");
assert!(rendered.contains("from-a-newer-build"), "{rendered}");
}
// `host` lists dispatches rather than runs, so the runs it read are not
// on it — the roots it could not read still are.
for rendered in [
runs(&root, false, "session-a"),
status(&survey),
goals(&survey),
host(&survey),
] {
assert!(rendered.contains("2 run root(s) skipped"), "{rendered}");
assert!(rendered.contains("no-launch"), "{rendered}");
assert!(rendered.contains("launch.json"), "{rendered}");
// What the record was missing, as the schema named it: a refusal
// that only counted the roots would leave a reader nothing to act
// on.
assert!(rendered.contains("run_id"), "{rendered}");
}
std::fs::remove_dir_all(&root).ok();
}
/// A root whose every run was refused is not a root with nothing in it, and
/// the two must not read alike: one means nothing is running and the other
/// means this build cannot see what is.
#[test]
fn a_root_whose_every_run_is_refused_does_not_read_as_no_runs_recorded() {
let root = scratch("all-refused");
std::fs::create_dir_all(root.join("no-launch")).expect("a directory with no launch");
let survey = Survey::of(&root);
assert!(survey.views.is_empty());
for rendered in [
runs(&root, false, "session-a"),
status(&survey),
goals(&survey),
] {
assert!(
!rendered.contains("no runs recorded"),
"a rejected root reported as an absence: {rendered}"
);
assert!(rendered.contains("no run under"), "{rendered}");
assert!(rendered.contains("1 run root(s) skipped"), "{rendered}");
assert!(rendered.contains("no-launch"), "{rendered}");
}
// An empty root is still the other fact, and still says so.
let empty = scratch("all-refused-empty");
assert_eq!(runs(&empty, false, "session-a"), "no runs recorded\n");
std::fs::remove_dir_all(&root).ok();
std::fs::remove_dir_all(&empty).ok();
}
/// A dispatch row nothing is driving is not a live dispatch.
///
/// Measured on a real host: six rows aged 12h–52h rendered as a live fleet
/// while nothing matching them was running. The row is dropped and counted
/// rather than rendered, because an operator acts on this list — and the
/// action it invites for work that does not exist is the one that ends work
/// that does.
#[test]
fn a_host_row_whose_driver_is_gone_is_counted_rather_than_rendered_live() {
let root = scratch("host-stale");
let paths = write_run(
&root,
"ghosted",
sys::pid(),
&[
event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan()))],
),
event(
crate::journal::PipelineKind::NodeDispatched,
Some("build"),
&[],
),
],
);
// The registry entry a dispatch that died left behind: its pid is one
// this host can prove is gone.
register(
&paths,
&ledger::DispatchRecord {
pid: dead_pid(),
started: "a token from the process that died".into(),
..dispatched_here("build")
},
);
let rendered = host(&Survey::of(&root));
assert!(
!rendered.contains("ghosted "),
"a dispatch nothing is driving was rendered as a live row: {rendered}"
);
assert!(rendered.contains("no live dispatches"), "{rendered}");
assert!(
rendered.contains("1 stale registry entry ignored"),
"{rendered}"
);
assert!(rendered.contains("ghosted/build"), "{rendered}");
assert!(rendered.contains("is gone"), "{rendered}");
// And the scope of the claim is on the output, because the scan cannot
// see a run recorded under another root.
assert!(rendered.contains(&root.display().to_string()), "{rendered}");
std::fs::remove_dir_all(&root).ok();
}
/// The same run with a live process actually running the dispatch: the row
/// renders, and it renders as live.
#[test]
fn a_host_row_backed_by_a_live_registry_entry_renders_as_a_live_dispatch() {
let root = scratch("host-live");
let paths = write_run(
&root,
"driven",
sys::pid(),
&[
event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan()))],
),
event(
crate::journal::PipelineKind::NodeDispatched,
Some("build"),
&[],
),
],
);
register(&paths, &dispatched_here("build"));
let rendered = host(&Survey::of(&root));
assert!(rendered.contains("driven"), "{rendered}");
assert!(rendered.contains("build"), "{rendered}");
assert!(!rendered.contains("no live dispatches"), "{rendered}");
assert!(!rendered.contains("stale registry"), "{rendered}");
assert!(!rendered.contains("UNPROVEN"), "{rendered}");
std::fs::remove_dir_all(&root).ok();
}
/// A lock this host cannot check the pid against is neither proof. The row
/// stays visible — dropping a dispatch that may be running is the other
/// error — and it says outright that nothing backs it.
#[test]
fn a_host_row_this_host_cannot_prove_either_way_says_so_rather_than_reading_live() {
let root = scratch("host-unproven");
let paths = write_run(
&root,
"elsewhere",
sys::pid(),
&[
event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan()))],
),
event(
crate::journal::PipelineKind::NodeDispatched,
Some("build"),
&[],
),
],
);
register(
&paths,
&ledger::DispatchRecord {
host: "some-other-host".into(),
..dispatched_here("build")
},
);
let rendered = host(&Survey::of(&root));
assert!(rendered.contains("elsewhere"), "{rendered}");
assert!(rendered.contains("UNPROVEN"), "{rendered}");
assert!(rendered.contains("some-other-host"), "{rendered}");
assert!(!rendered.contains("stale registry"), "{rendered}");
// An entry from a build that predates the stamp is the same answer for a
// different reason: nothing says the pid is still the process the
// dispatch was recorded in, and the registry refuses the whole read
// rather than hand back a row nobody may act on.
register(
&paths,
&ledger::DispatchRecord {
started: String::new(),
..dispatched_here("build")
},
);
let rendered = host(&Survey::of(&root));
assert!(rendered.contains("UNPROVEN"), "{rendered}");
assert!(rendered.contains("no start token"), "{rendered}");
// And a live pid whose start token disagrees with the one recorded is a
// *different* process wearing a reused pid: proved stale.
register(
&paths,
&ledger::DispatchRecord {
started: "the process it was recorded in, which was not this one".into(),
..dispatched_here("build")
},
);
let rendered = host(&Survey::of(&root));
assert!(
rendered.contains("1 stale registry entry ignored"),
"{rendered}"
);
assert!(rendered.contains("different process"), "{rendered}");
// And with nothing in the registry at all, the run says a dispatch is in
// flight and no process claims it — which is neither proof.
std::fs::remove_dir_all(paths.dispatches()).expect("the registry is taken away");
std::fs::create_dir_all(paths.dispatches()).expect("an empty registry");
let rendered = host(&Survey::of(&root));
assert!(rendered.contains("UNPROVEN"), "{rendered}");
assert!(rendered.contains("holds no entry for it"), "{rendered}");
std::fs::remove_dir_all(&root).ok();
}
/// One advance a member's chain published, as the producer publishes it.
fn advanced(role: Option<&str>, turn: Option<u64>, identity: &str, reason: &str) -> Envelope {
advanced_for("worker", role, turn, identity, reason)
}
/// The same, for a named member: a dispatch runs more than one, and each
/// numbers its own turns.
fn advanced_for(
member: &str,
role: Option<&str>,
turn: Option<u64>,
identity: &str,
reason: &str,
) -> Envelope {
let mut fields = vec![("identity", json!(identity)), ("reason", json!(reason))];
if let Some(role) = role {
fields.push(("role", json!(role)));
}
if let Some(turn) = turn {
fields.push(("turn", json!(turn)));
}
let mut envelope = relayed(
EventKind("fallback-advanced".into()),
Source::Agentgraph,
Some("build"),
&fields,
);
envelope.stream = "oneagentgraph-1".into();
envelope.labels.member = Some(member.into());
envelope
}
/// One invocation that ran, built through the producing library's own
/// payload type so what is folded is what that library publishes.
fn invocation(role: oneagentgraph::event::Role, turn: u64, identity: &str) -> Envelope {
invocation_for("worker", role, turn, identity)
}
/// The same, for a named member.
fn invocation_for(
member: &str,
role: oneagentgraph::event::Role,
turn: u64,
identity: &str,
) -> Envelope {
let session = oneagentgraph::event::OneharnessSession {
role,
turn,
identity: identity.to_string(),
session_id: None,
history_id: "record-1".into(),
history_dir: "/store".into(),
history_project: "project".into(),
history_session: "record-1".into(),
truncated: false,
};
let mut envelope = relayed(
EventKind("oneharness-session".into()),
Source::Agentgraph,
Some("build"),
&[],
);
envelope.stream = "oneagentgraph-1".into();
envelope.payload = match serde_json::to_value(&session) {
Ok(serde_json::Value::Object(payload)) => payload,
other => panic!("a session is not an object: {other:?}"),
};
envelope.labels.member = Some(member.into());
envelope
}
/// What a dispatch that died says about where its work is, in each of the
/// three shapes a settlement of that kind comes in.
///
/// The e2e journeys drive the two ends of this — a dispatch that left a
/// branch and a commit, and one that left neither — because those are the two
/// a run can actually reach. The middle one is a branch `onevcs` recorded no
/// commit for, which is a settlement rather than a scenario: it is what a
/// dispatch that opened a session and committed nothing leaves, and the
/// sentence has to stay true of it. The fourth is a record an **older build**
/// wrote, carrying the word and no classification, which no run this build
/// drives can produce at all.
#[test]
fn a_dispatch_that_died_says_where_its_work_is_in_every_shape_a_settlement_has() {
let root = scratch("died");
let died = |node: &str, fields: &[(&str, serde_json::Value)]| {
let mut all = vec![
("status", json!("failed")),
("outcome", json!(crate::engine::DISPATCH_DIED)),
];
all.extend(fields.iter().cloned());
event(crate::journal::PipelineKind::NodeSettled, Some(node), &all)
};
let mut plan = plan();
for id in ["branchless", "uncommitted", "unclassified"] {
plan.tasks.push(Node {
id: id.into(),
task: Some("## What\ndo it".into()),
..Node::default()
});
}
write_run(
&root,
"died",
sys::pid(),
&[
event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan))],
),
died(
"build",
&[
("cause", json!("rate_limit")),
("branch", json!("b/one")),
("head", json!("abc123")),
],
),
died("branchless", &[("cause", json!("spawn-error"))]),
died(
"uncommitted",
&[("cause", json!("auth")), ("branch", json!("b/two"))],
),
died("unclassified", &[("branch", json!("b/three"))]),
],
);
let survey = Survey::of(&root);
let rendered = results(&survey.views[0]);
for said in [
"(rate_limit) rather than failing its task; b/one may carry finished work, at abc123",
"(spawn-error) rather than failing its task; it left no branch",
"(auth) rather than failing its task; b/two may carry finished work",
// No classification at all: the word still says what happened, and
// nothing here invents a reason the producer did not give.
"the dispatch died rather than failing its task; b/three may carry finished work",
] {
assert!(rendered.contains(said), "{said:?} is not in:{rendered}");
}
// And the same sentences reach the view a supervisor decides from.
let standing = status(&survey);
assert!(
standing.contains("the dispatch died (rate_limit) rather than failing its task"),
"{standing}"
);
}
/// A node that failed says which chain **ran out** and which merely fell
/// through and was served — and never the second under the first's word.
///
/// Both sides, because they are the point: a two-party member runs one chain
/// per side and they prefer different identities, so a fix aimed at the wrong
/// one changes nothing and the run fails the same way again. And a fix aimed
/// at a chain that recovered changes nothing at all.
#[test]
fn a_failed_node_tells_a_recovered_chain_from_one_that_ran_out() {
// The reason is a stranger's word on the line, but the stranger is
// oneharness: a producer writes that library's `FallThroughReason`
// spelling, so the record here carries one it could have written.
let rate_limited = oneharness_core::domain::fallback::FallThroughReason::RateLimit.as_str();
let root = scratch("refusal");
write_run(
&root,
"refused",
sys::pid(),
&[
event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan()))],
),
event(
crate::journal::PipelineKind::NodeDispatched,
Some("build"),
&[],
),
advanced(Some("agent"), Some(1), "claude-code", "quota"),
// The agent side's turn went on to run under the next candidate,
// so its chain recovered and nothing about it failed this node.
invocation(
oneagentgraph::event::Role::Agent,
1,
"claude-code:alternate",
),
advanced(Some("judge"), Some(1), "codex", rate_limited),
// The same side refusing the same way again is one fact
// recorded twice, not two facts.
advanced(Some("judge"), Some(1), "codex", rate_limited),
event(
crate::journal::PipelineKind::NodeSettled,
Some("build"),
&[
("status", json!("failed")),
("outcome", json!("task-failed")),
],
),
],
);
let survey = Survey::of(&root);
let rendered = results(&survey.views[0]);
assert!(
rendered.contains(
"fallback: the agent side fell through 'claude-code' (quota) → served by \
'claude-code:alternate'"
),
"{rendered}"
);
assert!(
rendered.contains(&format!(
"provider: the judge side: identity 'codex' refused ({rate_limited}), recorded 2 \
times"
)),
"{rendered}"
);
assert!(
!rendered.contains("provider: the agent side"),
"a recovered chain was reported as a refusal:\n{rendered}"
);
// The same attribution on the view a planner reads first.
let rendered = status(&survey);
assert!(
rendered.contains("build: failed — the judge side: identity 'codex' refused"),
"{rendered}"
);
assert!(
rendered.contains("build: fallback — the agent side fell through 'claude-code'"),
"{rendered}"
);
std::fs::remove_dir_all(&root).ok();
}
/// A record that names no side is rendered without one, and a chain that
/// named no identity is not rendered at all: an attribution nobody can act
/// on is what this whole line exists to replace.
#[test]
fn an_unattributed_refusal_is_never_given_a_side_it_did_not_carry() {
let advance = |reason: &str| oneagentgraph::event::FallbackAdvanced {
identity: "codex".into(),
reason: reason.into(),
role: None,
turn: None,
truncated: false,
};
let single = Refusal {
advanced: advance("auth"),
member: MemberLabel::Named("worker".into()),
records: std::num::NonZeroU64::MIN,
};
// A record carrying neither side nor turn has nothing to pair with, so
// it is neither a recovery nor a refusal — and saying "refused" over it
// would be naming a subscription that was never the problem.
assert_eq!(
chain_phrase(&ChainRecord {
refusal: &single,
became: Fallthrough::Unrecorded,
records: std::num::NonZeroU64::MIN,
}),
"member 'worker' fell through 'codex' (auth); nothing this run recorded names what \
served that turn"
);
let bare = Refusal {
advanced: advance(""),
member: MemberLabel::Unstamped,
records: std::num::NonZeroU64::MIN,
};
let phrase = chain_phrase(&ChainRecord {
refusal: &bare,
became: Fallthrough::Refused,
records: std::num::NonZeroU64::MIN,
});
assert!(
phrase.contains("a side the record does not name"),
"{phrase}"
);
assert!(
phrase.contains("for a reason the record does not carry"),
"{phrase}"
);
// An advance carrying no identity names nothing to act on. It is not an
// advance the producing library's own type accepts, so nothing here
// assembles an attribution out of what is left of it.
let mut nameless = relayed(
EventKind("fallback-advanced".into()),
Source::Agentgraph,
Some("build"),
&[("reason", json!("quota"))],
);
nameless.stream = "oneagentgraph-1".into();
assert!(crate::projection::fold(&[nameless]).refusals.is_empty());
}
/// One chain, two turns, two endings: the recovered turn and the one that
/// ran out are two facts, and each is rendered as itself.
///
/// The fold keeps them apart by turn for exactly this — a record that had
/// collapsed them could only ever be rendered as one of the two, and which
/// one it picked would decide where a reader went.
#[test]
fn one_chain_that_recovers_and_then_runs_out_says_both() {
let state = crate::projection::fold(&[
advanced(Some("agent"), Some(1), "claude-code", "quota"),
invocation(
oneagentgraph::event::Role::Agent,
1,
"claude-code:alternate",
),
// A second turn that ended the same way. Two records of one fact, so
// they are counted rather than repeated — the collapsing the fold
// cannot do, because it does not yet know how either ended.
advanced(Some("agent"), Some(2), "claude-code", "quota"),
invocation(
oneagentgraph::event::Role::Agent,
2,
"claude-code:alternate",
),
// And a third that ran out of candidates, which is a different fact
// about the same chain and says so on its own line.
advanced(Some("agent"), Some(3), "claude-code", "quota"),
]);
let records = chain_records(&state, "build");
let phrases = records.iter().map(chain_phrase).collect::<Vec<_>>();
assert_eq!(
phrases,
vec![
"the agent side fell through 'claude-code' (quota) → served by \
'claude-code:alternate', recorded 2 times"
.to_string(),
"the agent side: identity 'claude-code' refused (quota)".to_string(),
]
);
// An invocation of the *other* side, or of another member, never answers
// for this one: each numbers its own turns, so pairing across either
// would name an identity that served somebody else's chain. A dispatch
// runs more than one member, and the double every journey here drives
// labels every envelope with the one it runs — so the second member is
// stated at this level, where a record can carry the label the producer
// stamps on a member of its own.
for crossing in [
invocation(oneagentgraph::event::Role::Agent, 1, "claude-code"),
invocation_for("reviewer", oneagentgraph::event::Role::Judge, 1, "codex-2"),
] {
let crossed = crate::projection::fold(&[
advanced(Some("judge"), Some(1), "codex", "quota"),
crossing,
]);
assert_eq!(
chain_records(&crossed, "build")
.iter()
.map(chain_phrase)
.collect::<Vec<_>>(),
vec!["the judge side: identity 'codex' refused (quota)".to_string()]
);
}
// And the member's *own* invocation still answers for it, so the
// isolation above is a boundary rather than a chain nothing can pair.
let paired = crate::projection::fold(&[
advanced_for("reviewer", Some("judge"), Some(1), "codex", "quota"),
invocation_for("reviewer", oneagentgraph::event::Role::Judge, 1, "codex-2"),
]);
assert_eq!(
chain_records(&paired, "build")
.iter()
.map(chain_phrase)
.collect::<Vec<_>>(),
vec!["the judge side fell through 'codex' (quota) → served by 'codex-2'".to_string()]
);
}
/// A judge verdict that failed a node is rendered from the settlement's own
/// inline copy, so the reason reaches a reader without a file being opened.
#[test]
fn a_verdict_that_failed_a_node_names_its_criterion_and_its_reason() {
let root = scratch("verdict");
let settled = |verdicts: serde_json::Value| {
let mut envelope = relayed(
EventKind("member-settled".into()),
Source::Agentgraph,
Some("build"),
&[("completed", json!(false)), ("verdict", verdicts)],
);
envelope.stream = "oneagentgraph-1".into();
envelope
};
write_run(
&root,
"verdict",
sys::pid(),
&[
event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan()))],
),
event(
crate::journal::PipelineKind::NodeDispatched,
Some("build"),
&[],
),
settled(json!([
// Passed, so it failed nothing and is not the reason.
{"criterion": "the branch is pushed", "kind": "boolean",
"verdict": {"value": true, "reason": "it is"}},
{"criterion": "the change builds", "kind": "boolean",
"verdict": {"value": false, "reason": "cargo build fails in src/views.rs"}},
// Numeric, which onejudge reports and fails nothing over.
{"criterion": "how readable it is", "kind": "numeric",
"verdict": {"value": 2.0, "reason": "dense"}},
// A record that names neither half. It still says the node
// failed on its judge, which is the fact a provider line
// above it would otherwise be read as — and an empty string
// is a criterion nobody wrote, not one worth a bare pair of
// quotes on a line.
{"criterion": "", "kind": "boolean", "verdict": {"value": false}},
// Not one of the producing library's verdicts at all: it
// names no kind. Dropped whole rather than mined for the
// fields it does carry — a sentence lifted out of a record
// this build cannot read would be attributed to a criterion
// nobody scored.
{"criterion": "the tests pass",
"verdict": {"value": false, "reason": "the suite is red"}},
])),
event(
crate::journal::PipelineKind::NodeSettled,
Some("build"),
&[
("status", json!("failed")),
("outcome", json!("task-failed")),
],
),
],
);
let rendered = results(&Survey::of(&root).views[0]);
assert!(
rendered.contains(
"verdict: 'the change builds' failed — cargo build fails in src/views.rs"
),
"{rendered}"
);
assert!(
rendered.contains(
"verdict: a criterion the record does not name failed — the record carries no \
reason"
),
"{rendered}"
);
for absent in [
"the branch is pushed",
"how readable it is",
"the suite is red",
] {
assert!(
!rendered.contains(absent),
"a verdict that failed nothing, or that this build cannot read, was named as \
the failure:\n{rendered}"
);
}
std::fs::remove_dir_all(&root).ok();
}
/// The advice for a run nothing is driving names the nodes a judge
/// **rejected**, and no other node that failed.
///
/// Two records make a rejection, and this drives the run either one alone
/// would misname: a node that failed its own task with no judge verdict
/// against it is not a node a judge rejected, and reported as one it would
/// send a planner to read a verdict nobody wrote. The journey the operator
/// took is in `tests/e2e/views.rs`; what is held here is the discrimination.
#[test]
fn only_a_node_a_judge_rejected_is_named_as_one() {
let root = scratch("rejected-advice");
let failing = |node: &str, verdicts: Option<serde_json::Value>| {
let mut events = vec![event(
crate::journal::PipelineKind::NodeDispatched,
Some(node),
&[],
)];
if let Some(verdicts) = verdicts {
let mut settled = relayed(
EventKind("member-settled".into()),
Source::Agentgraph,
Some(node),
&[("completed", json!(false)), ("verdict", verdicts)],
);
settled.stream = "oneagentgraph-1".into();
events.push(settled);
}
events.push(event(
crate::journal::PipelineKind::NodeSettled,
Some(node),
&[
("status", json!("failed")),
("outcome", json!(crate::engine::TASK_FAILED)),
],
));
events
};
let rejection = json!([
{"criterion": "the change builds", "kind": "boolean",
"verdict": {"value": false, "reason": "cargo build fails"}},
]);
let held_up = Plan {
tasks: vec![
Node {
id: "build".into(),
..Node::default()
},
Node {
id: "later".into(),
deps: vec!["build".into()],
..Node::default()
},
],
..plan()
};
for (run, verdicts) in [("judged", Some(rejection)), ("brokeoff", None)] {
let mut events = vec![event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(held_up))],
)];
events.extend(failing("build", verdicts));
write_run(&root, run, dead_pid(), &events);
}
let listing = runs(&root, false, "session-a");
let status_of = |run: &str| {
let paths = RunPaths::under(&root, run);
status(&Survey {
root: root.clone(),
views: vec![RunView::open(&paths).expect("the run reads back")],
skipped: Vec::new(),
})
};
let judged = status_of("judged");
for rendered in [&listing, &judged] {
assert!(
rendered.contains("build, whose work a judge rejected"),
"the rejected node is not named as one a judge rejected:\n{rendered}"
);
assert!(
rendered.contains("onepipeline results judged"),
"the verdict a planner has to read is not named:\n{rendered}"
);
assert!(
rendered.contains("superseding the node"),
"the step that moves the run is not named:\n{rendered}"
);
}
assert!(
!judged.contains("adopt"),
"a driver is prescribed for a frontier it cannot move:\n{judged}"
);
let broke = status_of("brokeoff");
assert!(
!broke.contains("a judge rejected"),
"a node that failed its own task was reported as judged:\n{broke}"
);
assert!(
!listing.contains("onepipeline results brokeoff"),
"a run nothing judged was given the judgement's advice:\n{listing}"
);
std::fs::remove_dir_all(&root).ok();
}
/// Every value relayed from a sibling's record reaches a rendered line
/// through the same strip: an identity that served a turn is a stranger's
/// string exactly as the one that refused is.
#[test]
fn a_relayed_value_never_carries_a_control_character_onto_a_line() {
let refusal = Refusal {
advanced: oneagentgraph::event::FallbackAdvanced {
identity: "codex".into(),
reason: "quota".into(),
role: Some(oneagentgraph::event::Role::Agent),
turn: Some(1),
truncated: false,
},
member: MemberLabel::Named("worker".into()),
records: std::num::NonZeroU64::MIN,
};
let phrase = chain_phrase(&ChainRecord {
refusal: &refusal,
became: Fallthrough::PassedTo("codex\r\nprovider: forged".into()),
records: std::num::NonZeroU64::MIN,
});
assert!(!phrase.contains('\n') && !phrase.contains('\r'), "{phrase}");
let phrase = verdict_phrase(&crate::report::FailedVerdict {
criterion: Some("it builds".into()),
reason: Some("no\nit does not".into()),
});
assert!(!phrase.contains('\n'), "{phrase}");
}
#[test]
fn every_view_renders_from_the_merged_stream() {
let root = scratch("render");
let mut agent = relayed(
EventKind("turn-finished".into()),
Source::Agentgraph,
Some("build"),
&[("message", json!("ran the gate"))],
);
agent.stream = "oneagentgraph-1".into();
let mut vcs = relayed(
EventKind("session-opened".into()),
Source::Vcs,
Some("build"),
&[("branch", json!("feature"))],
);
vcs.stream = "onevcs-tok".into();
write_run(
&root,
"demo",
sys::pid(),
&[
event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan()))],
),
event(crate::journal::PipelineKind::NodeReady, Some("build"), &[]),
event(
crate::journal::PipelineKind::NodeDispatched,
Some("build"),
&[],
),
agent,
vcs,
],
);
let view = RunView::open(&RunPaths::under(&root, "demo")).expect("the run reads");
let stream = monitor(&view, &EventFilter::default());
assert!(stream.starts_with("Concise graph events;"), "{stream}");
assert!(stream.contains("agent:oneagentgraph-1"), "{stream}");
assert!(stream.contains("vcs:onevcs-tok"), "{stream}");
assert!(stream.contains("graph:build"), "{stream}");
// The run's own state has no node, so it has no typed id: it reaches the
// reader as a trailer, naming the run it belongs to.
assert!(stream.contains("-- demo 0/1 done"), "{stream}");
assert!(
!stream.contains("round"),
"a round reached a view: {stream}"
);
// The dispatch's own registry entry is what makes it a live one to the
// host view.
let paths = RunPaths::under(&root, "demo");
register(&paths, &dispatched_here("build"));
let survey = Survey::of(&root);
assert!(status(&survey).contains("build: running"));
assert!(host(&survey).contains("build"));
assert!(goals(&survey).contains("close the coverage gap"));
assert!(results(&survey.views[0]).contains("build"));
std::fs::remove_dir_all(&root).ok();
}
#[test]
fn a_node_the_ledger_calls_running_that_nothing_drives_is_undriven() {
let root = scratch("undriven");
write_run(
&root,
"demo",
sys::pid(),
&[
event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan()))],
),
event(
crate::journal::PipelineKind::NodeDispatched,
Some("build"),
&[],
),
],
);
let rendered = status(&Survey::of(&root));
assert!(rendered.contains("UNDRIVEN"), "{rendered}");
std::fs::remove_dir_all(&root).ok();
}
/// What the line says while a node is in flight, which is the whole of what
/// an operator has to decide between cancel, retry, and wait on.
#[test]
fn a_live_dispatch_reports_what_it_is_doing_now_with_a_count_and_an_age() {
let root = scratch("activity");
let mut turn = relayed(
EventKind("turn-activity".into()),
Source::Agentgraph,
Some("build"),
&[
("kind", json!("tool_call")),
("name", json!("Bash")),
("detail", json!("cargo llvm-cov --workspace")),
],
);
turn.stream = "oneagentgraph-1".into();
write_run(
&root,
"demo",
sys::pid(),
&[
event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan()))],
),
event(
crate::journal::PipelineKind::NodeDispatched,
Some("build"),
&[],
),
turn,
],
);
let rendered = status(&Survey::of(&root));
assert!(
rendered.contains("now Bash cargo llvm-cov --workspace"),
"{rendered}"
);
assert!(rendered.contains("1 event(s)"), "{rendered}");
assert!(rendered.contains("ago"), "{rendered}");
assert!(
!rendered.contains(DriverLiveness::Undriven.as_str()),
"a dispatch that is recording was reported as driving nothing: {rendered}"
);
std::fs::remove_dir_all(&root).ok();
}
/// One node's progress record, built through the transitions the fold builds
/// it through rather than assembled: `events` arrivals, the last of them at
/// `last_at`.
///
/// Started from nothing and advanced one arrival at a time, exactly as
/// `fold_activity` advances it — so no arrivals is no record, which is the
/// answer the fold gives too.
fn recorded(events: u64, last_at: u64) -> Option<crate::projection::Progress> {
(0..events).fold(None, |progress, _| match progress {
None => crate::projection::Progress::first(Some(last_at)),
Some(progress) => Some(progress.and(Some(last_at))),
})
}
/// A dispatch that has recorded something without naming a tool claims the
/// count and the age and nothing more.
#[test]
fn a_dispatch_that_has_named_no_tool_reports_its_count_rather_than_a_guess() {
let rendered = working(&crate::projection::NodeActivity {
doing: None,
progress: recorded(3, sys::now_millis()),
last_heartbeat_at: None,
});
assert_eq!(rendered, "3 event(s), 0s ago");
assert!(!rendered.contains("now"), "{rendered}");
}
/// A dispatch heartbeating over work it did long ago reports both, and the
/// age it reports is the work's.
#[test]
fn a_heartbeat_is_reported_beside_the_age_of_the_work_rather_than_as_work() {
let now = sys::now_millis();
let rendered = working(&crate::projection::NodeActivity {
doing: Some("Bash red-green.sh".into()),
progress: recorded(4, now - 600_000),
last_heartbeat_at: Some(now),
});
assert!(
rendered.contains("4 event(s), 10m00s ago"),
"the age of the work was taken from the heartbeat: {rendered}"
);
assert!(
rendered.contains("alive 0s ago"),
"a dispatch that is alive and doing nothing is not reported as alive: {rendered}"
);
}
/// A dispatch that has only ever heartbeated has produced nothing, and says
/// so — rather than claiming an age for work it has not done, and rather
/// than reading as a node nothing is driving.
#[test]
fn a_dispatch_that_has_only_heartbeated_reports_no_work_and_still_reads_as_alive() {
let rendered = working(&crate::projection::NodeActivity {
doing: None,
progress: None,
last_heartbeat_at: Some(sys::now_millis()),
});
assert_eq!(rendered, "nothing recorded yet; alive 0s ago");
}
/// Both halves of a transcript: the tools the store carries as the turn
/// runs, and the words out of the report the member settled with.
#[test]
fn a_transcript_renders_the_turns_tools_and_the_report_it_settled_with() {
let root = scratch("transcript");
let paths = RunPaths::under(&root, "demo");
// This run's own copy, at the name ingest gives it: the reader derives
// that name from the settlement rather than following the path on it.
let stored = paths.report_for("s", 0);
std::fs::create_dir_all(paths.reports_dir()).expect("the run's report storage");
std::fs::write(
&stored,
json!({
"schema_version": 7,
"transcript": {"messages": [
{"role": "assistant", "content": "Ran the gate.\nIt passed.", "events": [
{"kind": "tool_call", "name": "bash", "input": {"command": "just check"}},
]},
]},
})
.to_string(),
)
.expect("a stored report");
let mut started = relayed(
EventKind("turn-started".into()),
Source::Agentgraph,
Some("build"),
&[("turn", json!(1))],
);
started.stream = "oneagentgraph-1".into();
let mut activity = relayed(
EventKind("turn-activity".into()),
Source::Agentgraph,
Some("build"),
&[
("kind", json!("tool_call")),
("name", json!("bash")),
("detail", json!("just check")),
],
);
activity.stream = "oneagentgraph-1".into();
let settled = relayed(
EventKind(crate::report::MEMBER_SETTLED.into()),
Source::Agentgraph,
Some("build"),
&[(crate::report::REPORT_PATH, json!("/elsewhere/report.json"))],
);
write_run(
&root,
"demo",
sys::pid(),
&[
event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan()))],
),
started,
activity,
settled,
],
);
let view = RunView::open(&RunPaths::under(&root, "demo")).expect("the run reads");
let rendered = transcript(&view, None);
assert!(rendered.contains("demo build"), "{rendered}");
assert!(rendered.contains("turn 1"), "{rendered}");
assert!(
rendered.contains("tool_call bash just check"),
"{rendered}"
);
assert!(rendered.contains("assistant"), "{rendered}");
assert!(rendered.contains("Ran the gate."), "{rendered}");
assert!(rendered.contains("It passed."), "{rendered}");
// Scoped to a node that dispatched nothing, there is nothing to render.
assert!(transcript(&view, Some("elsewhere")).contains("no dispatch"));
std::fs::remove_dir_all(&root).ok();
}
/// Two records off a run this crate drove on 2026-08-22, verbatim: a
/// `tool_call` and the `tool_result` that answered it.
///
/// Recorded rather than composed here, because the payload under test is a
/// producer's shape and not this crate's. A fixture written beside the
/// renderer proves the renderer agrees with the fixture — which is exactly
/// what the defect below had, since a `tool_result` carries its text under
/// `output` and carries no `detail` at all.
const RECORDED_ACTIVITY: &str = include_str!("../tests/recorded/turn-activity.jsonl");
/// The `member-settled` a real dispatch relayed, and the report this run
/// kept its own copy of. Same run, same seq: the reader derives the copy's
/// name from the settlement.
const RECORDED_SETTLEMENT: &str = include_str!("../tests/recorded/member-settled.json");
const RECORDED_REPORT: &str = include_str!("../tests/recorded/settled-report.json");
/// The evidence a run recorded, read back off the verb an operator is sent
/// to when a settlement's evidence looks missing.
///
/// The third column used to be `detail` for every activity, and a
/// `tool_result`'s `detail` is the empty string — so every observation a
/// dispatch made rendered blank while its own journal carried the whole
/// text. A node was failed for want of a measurement its journal held.
#[test]
fn a_recorded_tool_results_own_output_is_what_the_transcript_renders() {
let root = scratch("transcript-recorded-journal");
let mut events = vec![event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan()))],
)];
events.extend(RECORDED_ACTIVITY.lines().map(|line| {
serde_json::from_str::<Envelope>(line).expect("a recorded envelope reads back")
}));
write_run(&root, "demo", sys::pid(), &events);
let view = RunView::open(&RunPaths::under(&root, "demo")).expect("the run reads");
let rendered = transcript(&view, None);
assert!(
rendered.contains("tool_call Bash gh issue view 28"),
"{rendered}"
);
assert!(
rendered.contains("**Accepted fix.** Four parts:"),
"the recorded output is not on the line that answered for it:\n{rendered}"
);
// Not a prefix of it: the end of what the producer recorded is there
// too, so a reader is not looking at a silently shortened output.
assert!(
rendered.contains("the gate then passed without it."),
"the recorded output was cut before its end:\n{rendered}"
);
// And the producer had already cut it short, which the line says rather
// than leaving a reader to conclude the tool returned nothing further.
assert!(
rendered.contains("… [already cut short by the producer]"),
"an output the producer marked truncated is rendered as a whole \
one:\n{rendered}"
);
assert!(
!rendered.lines().any(|line| line.trim() == "tool_result"),
"an observation still renders as an empty column:\n{rendered}"
);
std::fs::remove_dir_all(&root).ok();
}
/// The same evidence out of the other source this verb reads: the report a
/// settled member left behind, which is what a reader gets once the dispatch
/// is over.
///
/// A retained report's outputs are the harness's raw bytes and nothing
/// bounds them, so this is also where the view's own ceiling is proven: the
/// one output past it is cut, and the line counts out loud how much of it a
/// reader is looking at.
#[test]
fn a_recorded_reports_tool_output_is_rendered_and_bounded() {
let root = scratch("transcript-recorded-report");
let settled: Envelope =
serde_json::from_str(RECORDED_SETTLEMENT.trim()).expect("a recorded settlement");
let paths = RunPaths::under(&root, "demo");
paths.create().expect("the run directory");
std::fs::create_dir_all(paths.reports_dir()).expect("the run's report storage");
std::fs::write(
paths.report_for(&settled.stream, settled.seq),
RECORDED_REPORT,
)
.expect("this run's own copy of the report");
write_run(
&root,
"demo",
sys::pid(),
&[
event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan()))],
),
settled,
],
);
let view = RunView::open(&RunPaths::under(&root, "demo")).expect("the run reads");
let rendered = transcript(&view, None);
assert!(
rendered.contains("tool_result ///"),
"the report's first observation renders as an empty column:\n{rendered}"
);
assert!(
rendered.contains("// the second, because every family below it is the"),
"the report's longest observation is not rendered at all:\n{rendered}"
);
assert!(
rendered.contains("… [4096 of 4350 characters]"),
"an unbounded output was printed whole, or cut without saying \
so:\n{rendered}"
);
assert!(
!rendered.contains("Verdict::Reclaim(lease) => reclaim(&mut report"),
"the ceiling printed the tail of an output past it:\n{rendered}"
);
std::fs::remove_dir_all(&root).ok();
}
/// A `tool_result` a producer recorded with **no output at all**, which is a
/// shape the store really carries — two of the 525 results one run relayed.
///
/// It renders with an empty third column, and that is the honest answer: the
/// payload holds no text under any key, so there is nothing to show. Pinned
/// because the blank line it produces looks exactly like the defect this
/// renderer was fixed for, and the two have opposite fixes — one is the
/// reader looking under the wrong key, this one is a producer that recorded
/// nothing. A reader meeting it should not go looking for text that was
/// never there, and nobody should invent any.
const RECORDED_WITHOUT_OUTPUT: &str =
include_str!("../tests/recorded/turn-activity-no-output.json");
#[test]
fn a_recorded_result_carrying_no_output_renders_empty_because_it_is_empty() {
let root = scratch("transcript-recorded-outputless");
let recorded: Envelope =
serde_json::from_str(RECORDED_WITHOUT_OUTPUT.trim()).expect("a recorded envelope");
// The premise, checked against the record rather than assumed: nothing in
// this payload carries text, under the key the renderer reads or any
// other.
assert_eq!(recorded.payload.get("output"), None, "{recorded:?}");
assert_eq!(
recorded
.payload
.get("detail")
.and_then(serde_json::Value::as_str),
Some(""),
"{recorded:?}"
);
write_run(
&root,
"demo",
sys::pid(),
&[
event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan()))],
),
recorded,
],
);
let view = RunView::open(&RunPaths::under(&root, "demo")).expect("the run reads");
let rendered = transcript(&view, None);
assert!(
rendered.lines().any(|line| line.trim() == "tool_result"),
"{rendered}"
);
assert!(!rendered.contains('…'), "{rendered}");
std::fs::remove_dir_all(&root).ok();
}
/// A stranger's output cannot rewrite the line it is printed on, and the
/// note about what was left out cannot be forged into looking like the
/// view's own.
#[test]
fn a_control_character_in_an_output_is_stripped_like_every_other_value() {
let rendered = tool_text(&ToolText::Returned {
output: "first\r\nsecond\u{1b}[2K".to_string(),
truncated: Truncation::Whole,
});
assert_eq!(rendered, "first second [2K");
}
/// What a producer says about having cut an output short, in each of the
/// three things it can say — including the one that is neither answer.
///
/// A flag this build cannot read used to be read as `false`, which is a
/// claim that the output is whole made on behalf of a producer that claimed
/// nothing of the sort. It is the same claim this view was corrected for
/// making, so it is said rather than assumed.
#[test]
fn a_truncation_flag_this_build_cannot_read_is_said_rather_than_assumed() {
let text = |flag: Option<serde_json::Value>| {
tool_text(&ToolText::Returned {
output: "what it returned".to_string(),
truncated: Truncation::of(flag.as_ref()),
})
};
assert_eq!(text(None), "what it returned");
assert_eq!(text(Some(json!(false))), "what it returned");
assert_eq!(text(Some(json!(null))), "what it returned");
assert_eq!(
text(Some(json!(true))),
"what it returned … [already cut short by the producer]"
);
for unreadable in [json!("true"), json!(1), json!({"cut": true})] {
assert_eq!(
text(Some(unreadable.clone())),
"what it returned … [the producer's truncation flag is unreadable]",
"{unreadable}"
);
}
}
/// A settlement whose report this run kept no copy of is said to be
/// unretained, and the path it named is printed and not opened. An absent
/// transcript and an unread one are different facts.
#[test]
fn a_report_this_run_did_not_keep_is_named_as_unretained_and_never_opened() {
let root = scratch("transcript-unread");
let settled = relayed(
EventKind(crate::report::MEMBER_SETTLED.into()),
Source::Agentgraph,
Some("build"),
&[(
crate::report::REPORT_PATH,
json!("/nowhere/onepipeline/report.json"),
)],
);
write_run(
&root,
"demo",
sys::pid(),
&[
event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan()))],
),
settled,
],
);
let view = RunView::open(&RunPaths::under(&root, "demo")).expect("the run reads");
let rendered = transcript(&view, None);
assert!(rendered.contains("not retained by this run"), "{rendered}");
assert!(
rendered.contains("/nowhere/onepipeline/report.json"),
"the path that was not read is not named: {rendered}"
);
std::fs::remove_dir_all(&root).ok();
}
/// A report that carries no transcript says so, rather than reading as a
/// dispatch that did nothing.
#[test]
fn a_report_without_a_transcript_says_so() {
let root = scratch("transcript-none");
let paths = RunPaths::under(&root, "demo");
std::fs::create_dir_all(paths.reports_dir()).expect("the run's report storage");
std::fs::write(
paths.report_for("s", 0),
json!({"usage": {"input_tokens": 1}}).to_string(),
)
.expect("a stored report");
let settled = relayed(
EventKind(crate::report::MEMBER_SETTLED.into()),
Source::Agentgraph,
Some("build"),
&[(crate::report::REPORT_PATH, json!("/elsewhere/report.json"))],
);
write_run(
&root,
"demo",
sys::pid(),
&[
event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan()))],
),
settled,
],
);
let view = RunView::open(&RunPaths::under(&root, "demo")).expect("the run reads");
assert!(transcript(&view, None).contains("carries no transcript"));
std::fs::remove_dir_all(&root).ok();
}
#[test]
fn a_run_that_dispatched_nothing_has_no_transcript_to_render() {
let root = scratch("transcript-empty");
write_run(
&root,
"demo",
sys::pid(),
&[event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan()))],
)],
);
let view = RunView::open(&RunPaths::under(&root, "demo")).expect("the run reads");
assert_eq!(
transcript(&view, None),
"no dispatch has recorded a transcript\n"
);
std::fs::remove_dir_all(&root).ok();
}
#[test]
fn a_waiting_human_reports_its_action_and_what_it_unblocks() {
let root = scratch("waiting");
let mut waiting_plan = plan();
waiting_plan.tasks = vec![
Node {
id: "approve".into(),
kind: crate::plan::NodeKind::Human,
task: Some("approve the release".into()),
..Node::default()
},
Node {
id: "ship".into(),
persona: Some("engineer".into()),
task: Some("## What\nship".into()),
deps: vec!["approve".into()],
..Node::default()
},
];
write_run(
&root,
"demo",
sys::pid(),
&[
event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(waiting_plan))],
),
event(
crate::journal::PipelineKind::NodeSettled,
Some("approve"),
&[("status", json!("waiting"))],
),
],
);
let view = RunView::open(&RunPaths::under(&root, "demo")).expect("the run reads");
let rendered = results(&view);
assert!(rendered.contains("approve the release"), "{rendered}");
assert!(rendered.contains("unblocks: ship"), "{rendered}");
assert!(
rendered.contains("ship") && rendered.contains("blocked"),
"{rendered}"
);
std::fs::remove_dir_all(&root).ok();
}
/// A manager about to replace an amendment can read the one they are
/// replacing, from either view.
///
/// `amend` replaces rather than appends, so a ruling that is not readable
/// *before* the replacement lands is a ruling nobody can weigh against the
/// one taking its place. Both views, because deciding to replace one is made
/// from either.
#[test]
fn both_views_render_the_amendment_a_node_is_currently_judged_against() {
let root = scratch("amendment");
let mut amended = plan();
amended.tasks[0].amendment =
Some("The four comment lines are out of scope: leave them.".into());
write_run(
&root,
"demo",
sys::pid(),
&[event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(amended))],
)],
);
let paths = RunPaths::under(&root, "demo");
let view = RunView::open(&paths).expect("the run reads");
let survey = Survey::of(&root);
for (which, rendered) in [("status", status(&survey)), ("results", results(&view))] {
assert!(
rendered.contains("The four comment lines are out of scope: leave them."),
"`{which}` does not say what `build` is judged against:\n{rendered}"
);
assert!(
rendered.contains("amend"),
"`{which}` renders the text without naming it an amendment:\n{rendered}"
);
}
// A node carrying none says nothing, so the absence reads as the absence
// rather than as a blank line somebody has to interpret.
let plain = scratch("amendment-none");
write_run(
&plain,
"demo",
sys::pid(),
&[event(
crate::journal::PipelineKind::RunStarted,
None,
&[("plan", json!(plan()))],
)],
);
let view = RunView::open(&RunPaths::under(&plain, "demo")).expect("the run reads");
assert!(!results(&view).contains("amendment:"), "{}", results(&view));
std::fs::remove_dir_all(&root).ok();
std::fs::remove_dir_all(&plain).ok();
}
#[test]
fn a_summary_line_is_capped_and_control_stripped() {
let long = "x".repeat(500);
let stripped = summarize(&relayed(
EventKind("kind".into()),
Source::Agentgraph,
None,
&[("message", json!(format!("a\nb{long}")))],
));
assert!(!stripped.contains('\n'), "{stripped}");
assert_eq!(stripped.chars().count(), 96);
}
#[test]
fn the_parked_threshold_is_read_from_the_environment_or_defaults() {
assert!(parked_after_seconds() > 0);
}
#[test]
fn every_liveness_verdict_has_the_word_the_contract_fixes() {
assert_eq!(DriverLiveness::Driving.as_str(), "ACTIVE");
assert_eq!(DriverLiveness::DriverDead.as_str(), "DRIVER DEAD");
assert_eq!(DriverLiveness::Parked.as_str(), "PARKED");
assert_eq!(DriverLiveness::Undriven.as_str(), "UNDRIVEN");
}
}