onepipeline 0.28.3

//! Lifecycle nodes: composing a `onevcs` session with the dispatches that work
//! in it.
//!
//! A lifecycle node names a `repo`, so its work happens on an isolated branch
//! and is published through that repository's registered policy. This module is
//! the composition and nothing more — the branch, the worktree, and the
//! publication are all `onevcs`'s, and the dispatch inside them is
//! `oneagentgraph`'s. Nothing here verifies the change: that is the repository's
//! own merge path, at the publishing push.
//!
//! Several `steps` share one branch and run **serially in topological order**,
//! because concurrent writers cannot safely share a worktree.

use std::collections::{BTreeMap, BTreeSet};
use std::sync::mpsc::Sender;
use std::time::{Duration, Instant};

use onevcs::SessionRequest;

use crate::controls::NodeControls;
use crate::engine::{self, Message, Settlement};
use crate::event::{Envelope, Labels};
use crate::executor::{DispatchRequest, Executor, WorkspaceSpec};
use crate::filter::EventFilter;
use crate::graph::NodeStatus;
use crate::ledger::RunPaths;
use crate::plan::{Node, NodeKind, Step};

/// The persona that drafts a change request's body.
pub const PR_AUTHOR_PERSONA: &str = "pr-author";

// llmlint: ignore-block[invalid_states_unrepresentable] the graph references below are the
// same validated, launch-recorded strings the engine carries — the launch record's own
// fields, read off it strictly and passed straight back into oneagentgraph's transparent
// ConfigRef. Another newtype would duplicate that sibling type and widen this
// path-resolution change across unrelated composition.
/// What this run's launch decides about a lifecycle node's dispatches.
///
/// One value rather than three parameters, and it is the launch record's own
/// three: every one of them is read off the record the loop read **strictly** at
/// the start of the pass, so a dispatch cannot pick up a config this build could
/// not honour by re-reading `launch.json` leniently where nothing can refuse it.
#[derive(Debug, Clone, Default)]
pub struct Launch {
    /// The default node-scope agent-graph config every dispatch launches, unless
    /// the node or the step names one of its own.
    pub node_graph: String,
    /// The agent graph a change request's body is drafted by, when the launch
    /// named one. `None` is the shipped default: this crate ships the flag, not
    /// the document, and a launch that names no graph drafts nothing.
    pub pr_author_graph: Option<String>,
    /// What every followed `onevcs` session's stream is read through.
    pub vcs_filter: Option<EventFilter>,
}

/// Run one lifecycle node to settlement, re-dispatching it while its
/// publication fails in a way that leaves the work behind.
///
/// A publication that ends `checks-failed`, `checks-unsettled`, `push-rejected`,
/// `pushed-unverified`, or `sync-conflict` did not reject the node — it rejected
/// the **tree** the node produced, and that tree is still on the branch the
/// session handed back (`pushed-unverified` more than the rest: that one is
/// already on the origin, and what a further attempt re-reads is the merge path
/// rather than the push). There
/// is nothing left in the run that would ever look at it again: the node settles
/// `failed`, its dependents never start, and an operator hand-builds a
/// replacement node out of the settlement's detail. So the node is asked again
/// instead, on that same branch, with the diagnosis in its hands.
///
/// Bounded, because a check that will never pass would otherwise be answered by
/// an unbounded series of dispatches, each of them producing the same tree and
/// paying for the same refusal. The node that spends the budget settles `failed`
/// saying how many attempts were made and what each one ended with, which is what
/// tells a reader the difference between a failure and a loop.
#[allow(
    clippy::too_many_arguments,
    reason = "one node's whole execution: the executor, the run, the launch, the node, \
              its cross-repository references, its cancellation, and where to report"
)]
pub fn execute(
    executor: &dyn Executor,
    paths: &RunPaths,
    launch: &Launch,
    node: &Node,
    references: &[crate::plan::CrossRepoReference],
    cancel: &crate::executor::CancellationToken,
    tx: &Sender<Message>,
) -> Settlement {
    let attempts = engine::publication_attempts();
    // What each attempt's publication ended with, in order, for the settlement
    // that stops the loop. One word per attempt: the last attempt's own reason
    // leads the detail as it always has, and a detail carrying three sibling
    // diagnostics in full would carry none of them — every payload text this
    // crate writes is bounded.
    let mut endings: Vec<crate::vcs::Preserving> = Vec::new();
    let launched = node;
    let mut node = std::borrow::Cow::Borrowed(node);
    let mut attempt = std::num::NonZeroU32::MIN;
    // The commit the branch stood at when the previous attempt published it.
    let mut published: Option<String> = None;
    // The manager's notes each attempt is composed with. The first attempt is
    // composed with none: what it was owed rode in as the node's own `context`,
    // and a note delivered into a conversation it has not opened yet cannot
    // exist. Every attempt after it is composed with what the attempt before it
    // was — read off the run's record, which is the one place both the delivery
    // and this thread can see.
    let mut notes: Vec<crate::note::RecordedNote> = Vec::new();
    loop {
        let preserved = match attempt_once(
            executor, paths, launch, &node, references, &notes, cancel, tx,
        ) {
            Attempt::Settled(settlement) => return *settlement,
            Attempt::Preserving(preserved) => preserved,
        };
        endings.push(preserved.outcome);
        if let Some(same) = republished(published.as_deref(), &preserved.tip) {
            return republished_the_same_commit(&node.id, &preserved, &endings, &same, attempt);
        }
        published = match &preserved.tip {
            crate::vcs::SessionTip::At(commit) => Some(commit.as_str().to_owned()),
            // The branch stands where the attempt before it left it, which is
            // what the next attempt is compared against.
            crate::vcs::SessionTip::Unmoved => published,
            // Nothing is known about where it stands, so there is nothing for
            // the next attempt to be compared against.
            crate::vcs::SessionTip::Unknown => None,
        };
        // Two reasons to stop, and one settlement for both: the budget is spent,
        // or the run is being stopped. A cancelled run must not be given another
        // dispatch — the teardown is on its way to reap it, and the node would
        // then settle as the cancellation rather than as the publication failure
        // that is the useful half of what happened.
        if attempt >= attempts || cancel.is_cancelled() {
            return stopped_retrying(&node.id, &preserved, &endings);
        }
        attempt = attempt.saturating_add(1);
        // What the conversation that just ended was shown, and what reached the
        // node while nothing of it was live: both are composed into the attempt
        // that replaces it, and the record of that dispatch names them, so a
        // ruling issued during one attempt is in the hands of the judge that
        // rules on the next. Read here, once, and handed to both the record and
        // the composition, so the two cannot name different notes.
        notes = match crate::note::standing_for(paths, &node.id) {
            Ok(standing) => standing.notes(),
            // A record this build cannot read is refused rather than read past:
            // continuing without the notes would compose the very dispatch this
            // fold exists to prevent, and saying so is the one honest answer.
            //
            // llmlint: ignore-block[changed_behavior_has_e2e] no invocation a user can
            // type reaches this arm: the records the fold refuses are this crate's own,
            // and its writers check what they write, so reaching it means a journal
            // edited by hand — which would prove the fixture rather than the code. The
            // refusal itself is held by `note::tests`, which drives the fold over such a
            // record; what this arm adds is the settlement, which is `stopped_retrying`'s
            // shape with the fold's own sentence as its detail.
            Err(why) => {
                return Settlement {
                    detail: Some(format!(
                        "the node was not dispatched again: {why}. The branch {} still \
                         carries the work; re-issue any note the earlier attempt was given \
                         and `retry` the node",
                        preserved.branch
                    )),
                    branch: Some(preserved.branch.clone()),
                    ..Settlement::plain(
                        &node.id,
                        NodeStatus::Failed,
                        Some(engine::INFRASTRUCTURE_FAILURE),
                    )
                };
            } // llmlint: ignore-end[changed_behavior_has_e2e]
        };
        // Another `node-dispatched` rather than a kind of its own, so a reader
        // counting dispatches sees the retry without a second word to learn.
        let _ = tx.send(Message::Redispatched(Box::new(engine::Redispatch {
            node: node.id.clone(),
            attempt,
            attempts,
            reason: format!("{}: {}", preserved.outcome.outcome(), preserved.reason),
            carried: notes.clone(),
        })));
        node =
            std::borrow::Cow::Owned(continued(launched, &preserved, attempt, attempts, &endings));
    }
}

#[allow(
    clippy::too_many_arguments,
    reason = "one attempt's whole context, which is `execute`'s own — see the reason there"
)]
fn attempt_once(
    executor: &dyn Executor,
    paths: &RunPaths,
    launch: &Launch,
    node: &Node,
    references: &[crate::plan::CrossRepoReference],
    notes: &[crate::note::RecordedNote],
    cancel: &crate::executor::CancellationToken,
    tx: &Sender<Message>,
) -> Attempt {
    let run = paths.run.as_str();
    let vcs_filter = launch.vcs_filter.as_ref();
    let Some(request) = crate::vcs::request_for(node) else {
        return Attempt::settled(Settlement {
            detail: Some("a lifecycle node needs a repo".into()),
            ..Settlement::plain(&node.id, NodeStatus::Failed, Some(engine::INVALID_NODE))
        });
    };

    // A node that declared no steps has one dispatch and no step, so nothing
    // stamps a `step` label the plan never wrote.
    let declared_steps = node.steps.is_some();
    // Every step's controls are narrowed here, before a session exists: a
    // workstream that cannot dispatch one of its steps must not first cut a
    // branch and run the steps before it, because that leaves work on a branch
    // for a node that was never going to finish.
    // llmlint: ignore-block[changed_behavior_has_e2e] what this arm newly carries — a
    // step whose declaration no dispatch can run under — is refused by `graph::validate`
    // at `start` and at every live edit, so only a graph
    // folded from a journal an *earlier build* wrote reaches it. Reaching that end to
    // end means writing that journal by hand, which proves the fixture rather than the
    // code, and deleting the arm would reinstate the silent default this control exists
    // to remove. Held instead by the unit test below, which drives the real
    // `LocalExecutor`; the step cycle this arm already reported keeps its own journey.
    let steps = match dispatchable_steps(node) {
        Ok(steps) => steps,
        Err(reason) => {
            return Attempt::settled(Settlement {
                detail: Some(reason),
                ..Settlement::plain(&node.id, NodeStatus::Failed, Some(engine::INVALID_NODE))
            })
        }
    }; // llmlint: ignore-end[changed_behavior_has_e2e]

    // When this dispatch began, which is what tells its commits from ones an
    // earlier dispatch left in a worktree the session took up again. The second
    // it fell in is waited out before the first session opens, so that nothing
    // this dispatch writes shares a stamp with what was there before it.
    let began = std::time::SystemTime::now();
    let mut session: Option<onevcs::SessionToken> = None;
    // The session's own stream, followed from the moment there is a token to
    // follow, so the publication that comes after the steps is visible while it
    // runs rather than only once it is over.
    let mut stream: Option<crate::vcs::Follower> = None;
    // The one worktree this node's dispatches work in, once its session has
    // opened one. See where it is read: every dispatch after the first runs
    // *there* rather than opening a session beside it.
    let mut worktree: Option<std::path::PathBuf> = None;
    // And what a publication of that session measures its branch against, taken
    // from the same read: asking again at publication would ask the sibling a
    // question it has already answered, on a path where the answer cannot have
    // changed and a failure could not happen — a publication that succeeded is a
    // record that was readable.
    let mut base: Option<String> = None;
    // Where in the run the session's own envelopes belong. The node, not a
    // step: a session outlives every step that wrote in it, and the publication
    // that follows them belongs to none.
    let whose = engine::dispatch_labels(run, &node.id, None, node.persona.as_deref());
    let mut branch: Option<String> = node.branch.clone();
    // The steps the preserved branch already carries, plus the ones this attempt
    // adds. Carried forward whole, because the branch a later attempt preserves
    // is the same branch: a step skipped on one attempt is still on it.
    let mut completed: Vec<String> = node
        .resume
        .as_ref()
        .map(|resume| resume.completed_steps.clone())
        .unwrap_or_default();

    for (step, controls) in &steps {
        if declared_steps && completed.iter().any(|id| id == &step.id) {
            // Already on the preserved branch. Re-running it would redo work the
            // branch carries, which for a step that opened a change is not
            // idempotent.
            continue;
        }
        if step.kind == NodeKind::Human {
            // A ready human step needs a person, and the workstream holds its
            // branch until one acts. The harness never infers that it happened.
            // The session stays open for them and the follow does not: dropping
            // it here ends a process that would otherwise read a stream nobody
            // is waiting for, for as long as the driver lives.
            //
            // Read against its criteria here, because this is where the node
            // settles and the person about to act is the reader a finding is
            // for: a branch that contradicts the bar is worth knowing *before*
            // an approval, not after. And it is the only place: the attempt that
            // follows an attestation dispatches nothing where the human step was
            // last, so it opens no session and has no branch in hand to read.
            check_criteria(node, worktree.as_deref(), tx);
            return Attempt::settled(Settlement {
                branch,
                completed_steps: completed,
                ..Settlement::plain(&node.id, NodeStatus::Waiting, None)
            });
        }
        if step.expects_no_diff {
            continue;
        }
        // Every step after the first names the branch the first opened, which
        // is what makes them one workstream rather than several beside it.
        let request = SessionRequest {
            branch: branch.clone().or_else(|| request.branch.clone()),
            ..request.clone()
        };
        // And works in the worktree the first step's session opened, rather than
        // asking for a session of its own. `onevcs` cuts every session its own
        // clone from the execution checkout, so a second session on the same
        // branch starts from the base with none of the earlier steps' work — and
        // opening it reclaims the first session's workspace, uncommitted work
        // and all. Steps are serial by construction, so one worktree is what
        // "several steps share one branch" has always meant.
        let workspace = match &worktree {
            Some(dir) => WorkspaceSpec::Path(dir.clone()),
            None => WorkspaceSpec::VcsSession(request.clone()),
        };
        let graph = engine::node_graph(
            step.agent_graph.as_ref().or(node.agent_graph.as_ref()),
            &launch.node_graph,
        );
        let build = || DispatchRequest {
            graph: graph.clone(),
            task: step.rendered_task_carrying(node, references, notes),
            labels: engine::dispatch_labels(
                run,
                &node.id,
                declared_steps.then_some(step.id.as_str()),
                step.persona.as_deref(),
            ),
            controls: *controls,
            workspace: workspace.clone(),
            cancel: cancel.clone(),
        };
        if worktree.is_none() {
            crate::vcs::wait_out_the_second(began);
        }
        let drained = engine::attempt(executor, node, cancel, tx, &build);
        // The session the dispatch opened is what publication needs, whether or
        // not the step succeeded: a cancelled step's commits are preserved on
        // the branch it left behind.
        session = drained.session.or(session);
        branch = drained.branch.or(branch);
        if stream.is_none() {
            if let Some(token) = &session {
                let opened = crate::vcs::working_session(token);
                worktree = opened.as_ref().map(|open| open.worktree.clone());
                base = opened.map(|open| open.base);
                stream = crate::vcs::follow(token, vcs_filter, relay_into(tx, whose.clone()));
            }
        }
        if drained.settlement.status != NodeStatus::Done {
            // The node is settling on this, so the branch it settles on is read
            // against its own criteria before the session that holds it goes.
            check_criteria(node, worktree.as_deref(), tx);
            // Where a human reads the change request the worker opened, on a node
            // that settles without publishing — a task that failed, a dispatch
            // that was cancelled or died — so `results` shows where the work is.
            // The engine names it where the dispatch handed a session back; a
            // later step's dispatch runs in a worktree and hands back none, so
            // the session this node holds is asked here.
            let change_url = drained
                .settlement
                .change_url
                .clone()
                .or_else(|| session.as_ref().and_then(crate::vcs::change_opened_in));
            end_session(stream, tx, session.as_ref(), &whose, vcs_filter);
            return Attempt::settled(Settlement {
                branch,
                completed_steps: completed,
                change_url,
                ..drained.settlement
            });
        }
        if declared_steps {
            completed.push(step.id.clone());
        }
    }

    let Some(token) = session else {
        // Every step declared no diff, so there is nothing to publish and the
        // node settles on the existing no-changes outcome.
        return Attempt::settled(Settlement {
            branch,
            ..Settlement::plain(&node.id, NodeStatus::Done, Some(engine::NO_CHANGES))
        });
    };

    let attempted = publish(
        executor,
        paths,
        launch,
        node,
        references,
        worktree.as_deref(),
        base.as_deref(),
        began,
        cancel,
        tx,
        &token,
        branch,
    );
    // Only where this attempt is the node's answer. A publication that failed
    // leaving the work on its branch is asked again, and reporting a criterion
    // against every attempt of a node that is still being re-dispatched would
    // put three findings on the queue for one branch nobody has settled on yet.
    if matches!(attempted, Attempt::Settled(_)) {
        check_criteria(node, worktree.as_deref(), tx);
    }
    end_session(stream, tx, Some(&token), &whose, vcs_filter);
    attempted
}

/// Hand [`crate::criteria`]'s reading of this node's branch to the loop.
///
/// **When**, which is all this call site decides. Every caller is a settlement
/// `attempt_once` has already made, so nothing here can change one; and it is
/// once per settlement rather than once per node, which differ only for a
/// workstream held at a human step and dispatched again afterwards — two
/// settlements, and not the same branch between them.
///
/// No worktree is no branch to read: every step declared no diff, the session
/// never opened, or the remaining steps were all done on a previous attempt.
/// Silence rather than an unread answer — nothing was named that could not be
/// read.
fn check_criteria(node: &Node, worktree: Option<&std::path::Path>, tx: &Sender<Message>) {
    // Both halves of "there is something to read, on behalf of somebody": a
    // branch in hand, and a node the graph carries to report it against.
    let (Some(worktree), Some(whose)) = (worktree, crate::graph::NodeRef::of(node)) else {
        return;
    };
    for check in crate::criteria::checkable_of(node) {
        let answer = crate::criteria::answer(worktree, &check);
        let _ = tx.send(Message::CriterionChecked(Box::new(
            engine::CriterionChecked {
                node: whose.clone(),
                check,
                answer,
            },
        )));
    }
}

/// Draft the change request's body, write it onto the change request the
/// session already holds, and publish through `onevcs`.
///
/// The order is divergence entry 69's, and the steps below run in it.
#[allow(
    clippy::too_many_arguments,
    reason = "publication needs the dispatch context (executor, the run's paths, what its \
              launch decided, the node, cancellation, and the event stream) as well as what \
              the steps left behind (the session token, its branch, the worktree and base \
              its record named, and when the dispatch began); the first six are the node's \
              own dispatch identity and bundling them would only move the same list one \
              indirection away"
)]
fn publish(
    executor: &dyn Executor,
    paths: &RunPaths,
    launch: &Launch,
    node: &Node,
    references: &[crate::plan::CrossRepoReference],
    worktree: Option<&std::path::Path>,
    base: Option<&str>,
    began: std::time::SystemTime,
    cancel: &crate::executor::CancellationToken,
    tx: &Sender<Message>,
    token: &onevcs::SessionToken,
    branch: Option<String>,
) -> Attempt {
    // A branch level with its base has nothing to draft a body for and nothing
    // to push, so neither is spent on it. Asked of the session's own record —
    // the worktree it opened and the base it measures against — and only where
    // both were read: a session whose record could not be read publishes as it
    // always did, and the sibling answers `NothingToPublish` after the drafting
    // dispatch has run, which is what this read exists to get ahead of.
    if let Some(level) = worktree
        .zip(base)
        .and_then(|(worktree, base)| crate::vcs::level_with_base(worktree, base, began))
    {
        return level_branch_settlement(node, &level, branch);
    }
    // A level branch is answered above before the host is asked anything, because
    // there is nothing to publish whatever the session holds: a change request the
    // worker opened from a branch that never diverged carries no diff for this
    // closeout to describe, and it is left exactly as the worker left it.
    // Asked next, because the answer decides which closeout this is. The rule
    // is about the **branch**: any change request open from the session's branch
    // into its base is finished here, whether the worker opened it as a draft or
    // an earlier attempt of this node opened it and this attempt is continuing.
    //
    // A host that could not be asked has not said the session holds nothing, so
    // that is said out loud rather than read as `None`: the publication below
    // asks the same host again and settles on its own words.
    let held = match crate::vcs::session_change(token) {
        Ok(held) => held,
        Err(error) => {
            eprintln!(
                "onepipeline: node '{}': onevcs could not say whether session {} holds a change \
                 request, so the closeout publishes as it always has: {error}",
                node.id, token.0
            );
            None
        }
    };
    // Whether it was this session's worker that drafted it, read before this
    // closeout writes anything of its own onto the same stream.
    let worker_drafted = held.is_some() && crate::vcs::change_drafted_in(token);
    // The plan's own body wins outright and spends no dispatch: a planner who
    // wrote the change request has already done the drafting.
    let (body, undrafted) = match node.body.clone() {
        Some(body) => (Some(body), None),
        // `None` is a launch that named no drafting graph, which drafts nothing
        // and is not a failure: this crate ships the flag, not the document.
        None => match drafted(
            executor,
            paths,
            launch,
            node,
            worktree,
            held.as_ref(),
            cancel,
            tx,
        ) {
            None => (None, None),
            Some(Drafted::Body(body)) => (Some(body), None),
            Some(Drafted::Undrafted(ending)) => (None, Some(ending)),
        },
    };
    // Said twice, and only where a drafting dispatch was configured and
    // attempted: in the run's own record at the moment it happened, and on the
    // node's settlement, where a planner reading `results` is shown it without
    // opening the store. Without either, a bodyless change request cannot say
    // whether the drafter ran and failed or was never wired at all — and those
    // need different fixes.
    //
    // On the settlement whatever the publication went on to do, not only where
    // it succeeded: what the drafter did is true either way, and a reader
    // looking for it must not have to know which failure came first. The
    // publication's own reason leads there, because that is what settled the
    // node.
    let undrafted = undrafted.map(|ending| {
        let why = ending.why();
        let _ = tx.send(Message::BodyNotDrafted(Box::new(engine::UndraftedBody {
            node: node.id.clone(),
            ending,
        })));
        why
    });
    // Written **onto** the change request the session holds, under the node's
    // own title, before the publication adopts it: a body handed to `publish` is
    // written only where the publication opens the change request, which here
    // it never does. `true` is a description the host now carries. A write the
    // host refused leaves the description as the worker left it, and the
    // settlement says so beside the publication's own words, exactly as a
    // drafting failure is said — the body exists, and the reader is owed the
    // reason it is not on the change request.
    let (described, undescribed) = match (&held, &body) {
        (Some(change), Some(body)) => {
            match crate::vcs::describe_change(token, node.title.as_deref(), body) {
                Ok(_) => (true, None),
                Err(error) => {
                    let why = format!(
                        "the drafted description was not written onto {}: {error}",
                        change.url
                    );
                    eprintln!("onepipeline: node '{}': {why}", node.id);
                    (false, Some(why))
                }
            }
        }
        _ => (false, None),
    };
    // One aside beside a publication's own words, and **not** only an undrafted
    // body: a body that was not drafted, or one that was drafted and could not
    // be written onto the change request. Never both — the second needs a body
    // the first says there is none of.
    let body_aside = undrafted.or(undescribed);
    // Through the one composition, so every place a publication's own words and
    // this aside are put together agrees about the order and the punctuation —
    // including the failure paths below, which compose the same two values from
    // a different function.
    let with_aside = |detail: String| compose(&detail, body_aside.as_deref());
    // The residual: a publication this crate can say nothing more about than
    // that it failed. Every failure `onevcs` names a kind for goes through
    // `failed_publication` below instead, which is where the word and the routing
    // are decided together.
    let publication_failed = |detail: String| {
        Attempt::settled(Settlement {
            branch: branch.clone(),
            detail: Some(with_aside(detail)),
            ..Settlement::plain(
                &node.id,
                NodeStatus::Failed,
                Some(crate::vcs::Failure::RESIDUAL),
            )
        })
    };

    // Whether this node's temporary git pin is still temporary, asked at the one
    // moment it decides anything. A node with no reference block — every node that
    // is not fast-adoption, and every fast one whose dependencies land where it
    // does — has nothing to ask about and gets `None`, which is the publication
    // this crate has always made. The release reason wins over the plan's own
    // `draft`: a draft a release will lift holds the run, and one the plan asked
    // for holds nothing, so where both apply the node settles on the first.
    let draft = crate::release::draft_reason(references)
        .or_else(|| node.draft.then(|| crate::release::held_reason(&node.id)));
    // No body where the session already holds the change request: the
    // description was written above, and a publication writes a body only where
    // it opens the change request.
    let opening_body = if held.is_some() {
        None
    } else {
        body.as_deref()
    };
    let publication =
        publish_rereading_the_merge_path(node, token, opening_body, draft.as_ref(), cancel);
    match publication.answered {
        Ok(published) => {
            // A publication that did not land is an ending of the publication,
            // not a refused request: `onevcs` draws that line itself, in
            // `PublishOutcome::Failed`, and this crate reads its line rather
            // than a second one. The reason is the sibling's own — what turned
            // the publication down, and what it said — and it is what the node
            // settles with, or what a re-dispatch carries back to the worker.
            if let onevcs::PublishOutcome::Failed {
                kind,
                reason,
                retained,
            } = &published.outcome
            {
                // The push reached the remote and the reads behind it are spent.
                // Settled here rather than routed with the four the tree was
                // rejected by: there is nothing about the tree to fix, and the
                // node says what it is — where the work is, what commit it is at,
                // and what stopped the read — instead of being asked for again.
                if crate::vcs::failure_of(*kind) == crate::vcs::Failure::Unread {
                    return unread_merge_path(
                        &node.id,
                        token,
                        branch.or_else(|| Some(published.branch.clone())),
                        reason,
                        publication.reads,
                        body_aside.clone(),
                    );
                }
                return failed_publication(
                    &node.id,
                    token,
                    branch.or_else(|| Some(published.branch.clone())),
                    *kind,
                    reason,
                    retained.as_ref(),
                    body_aside.clone(),
                );
            }
            let labels =
                engine::dispatch_labels(&paths.run, &node.id, None, node.persona.as_deref());
            let _ = tx.send(Message::Event(Box::new(crate::vcs::published_event(
                &published, &labels,
            ))));
            // What a `no-changes` compared against. It is the one outcome whose
            // word says nothing a reader can act on — a worker that wrote
            // nothing, a worker whose change was already on the base, and a base
            // that has since taken the work another way all settle identically,
            // and only the ref separates them. The base is the session's own,
            // because a node naming none took the identity's default and this
            // crate never saw it; a session whose record could not be read has
            // none, and the settlement then says what it always did rather than
            // naming a ref nothing established.
            let compared = match published.outcome {
                // Composed out of two of the sibling's own strings and stripped
                // of controls on the way, by the same rule every view applies to
                // a relayed value it renders — see `views::one_line`. A
                // settlement's detail is read back on one line, and this is where
                // this crate makes one out of somebody else's.
                onevcs::PublishOutcome::NothingToPublish => base.map(|base| {
                    crate::views::one_line(&format!(
                        "compared against {base}: {} carries nothing it does not",
                        published.branch
                    ))
                }),
                _ => None,
            };
            // A change the host is holding as a draft says why, on the line a
            // settlement is read back on. It leads whatever else the settlement
            // had to say for the reason a publication's own reason leads a
            // drafting failure: this is what the node settled as.
            let left_as_draft = matches!(published.outcome, onevcs::PublishOutcome::ChangeDraft(_));
            let drafted = left_as_draft
                .then(|| draft.as_ref().map(crate::release::drafted_detail))
                .flatten();
            // And a change request the session already held says what the
            // closeout did to it — the one sentence that tells such a node from
            // one that published afresh, since the outcome word is deliberately
            // the same.
            let finished = held
                .as_ref()
                .map(|change| finished_detail(change, worker_drafted, described, left_as_draft));
            let detail: Vec<String> = [drafted, finished, compared]
                .into_iter()
                .flatten()
                .collect();
            Attempt::settled(Settlement {
                // What the node settles on is its publication, exactly as
                // before; a drafting failure only ever adds words to it.
                detail: (!detail.is_empty())
                    .then(|| with_aside(detail.join(". ")))
                    .or_else(|| body_aside.clone()),
                // The branch the publication says carried the change, where a
                // dispatch reported none: they are the same branch, and the
                // sibling is the one that knows it.
                branch: branch.or_else(|| Some(published.branch.clone())),
                change_url: crate::vcs::change_url(&published.outcome),
                // Every ending has its own name. A branch whose base already
                // carries it settles `no-changes` rather than a bare
                // "published", which is what let a node whose worker wrote
                // nothing report as one that landed work.
                outcome: Some(crate::vcs::outcome_of(&published.outcome).to_owned()),
                // The node is done either way — publishing is the whole of what
                // the plan asked of it — so whether the change *landed* is
                // carried beside the status rather than folded into it. Taken
                // from what the publication answered and never from the policy
                // it ran under: an identity that asks the host to merge
                // immediately still has to be observed doing it.
                landing: crate::vcs::landing_of(&published.outcome),
                // **Complete, and not `done`.** Every step ran and the branch is
                // published, so nothing here failed and nothing is left for this
                // run to ask a worker for; what is left is a release outside it.
                // Settled `done`, the node's dependents would start on work that
                // cannot land and the run would report finished with the git pin
                // it launched against sitting in an open change — which is the
                // one ending fast adoption must not have.
                ..Settlement::plain(
                    &node.id,
                    drafted_status(&published.outcome, draft.as_ref()),
                    None,
                )
            })
        }
        // llmlint: ignore[changed_behavior_has_e2e] this arm is `onevcs` refusing the
        // call outright, which no double can produce: the fake host answers a publish
        // request, and the one refusal this crate can provoke — a title the sibling
        // will not commit under — is caught before any dispatch runs, by
        // `a_title_the_sibling_will_not_commit_under_is_refused_before_any_dispatch`.
        // What the arm does with a drafting failure is not its own composition either:
        // it is the same `publication_failed` the outcome arm above takes, which
        // `a_publication_its_merge_path_refuses_settles_the_node_failed_by_name` drives
        // end to end beside an undrafted body.
        Err(error) => publication_failed(error.to_string()),
    }
}

/// The settlement of a node whose branch is level with its base, split on
/// whether this dispatch wrote a commit rather than on the ahead-count alone.
///
/// Three situations are zero commits ahead, and the ahead-count reads all three
/// as one. The node **declared** it expects no diff: the success `engine::NO_CHANGES`
/// is documented for, and the word it settles under. This dispatch **wrote a
/// commit** the branch carries and the base already has it: the second reading
/// of that same word — a publication whose base already carried the branch —
/// and a legitimate success, which is why the split is on the commit.
/// Split on the count alone, it would be reported as the third: the declaration
/// absent and the dispatch having committed nothing, which is a worker asked for
/// a change that produced none. That one settles `failed` under
/// [`engine::EMPTY_BRANCH`], naming the branch and what it was measured against,
/// so a manager decides — a retry carrying the declaration accepts it as the
/// deterministic success that declaration is, and an amended task sends it back
/// to produce a diff. Settled `done` under the no-change word, as it used to be,
/// the modelling error was invisible in every view.
///
/// The detail says what was compared, exactly as the `NothingToPublish` arm of
/// [`publish`] does for the same word: `no-changes` alone covers a worker that
/// wrote nothing and a branch measured against itself, and only the ref tells
/// them apart.
fn level_branch_settlement(
    node: &Node,
    level: &crate::vcs::LevelBranch,
    branch: Option<String>,
) -> Attempt {
    let branch = branch.unwrap_or_else(|| level.branch.clone());
    let compared = format!(
        "compared against {}: {branch} carries nothing it does not",
        level.base
    );
    let carried = level.wrote == crate::vcs::Wrote::ACommitTheBaseCarries;
    if node.expects_no_diff || carried {
        let detail = if carried {
            format!(
                "{compared}; the base already carries what this dispatch committed to it, so \
                 there was nothing to draft or publish"
            )
        } else {
            compared
        };
        return Attempt::settled(Settlement {
            branch: Some(branch),
            detail: Some(crate::views::one_line(&detail)),
            ..Settlement::plain(&node.id, NodeStatus::Done, Some(engine::NO_CHANGES))
        });
    }
    Attempt::settled(Settlement {
        branch: Some(branch.clone()),
        detail: Some(crate::views::one_line(&format!(
            "{compared}. This dispatch committed nothing to it and the node does not declare \
             `expects_no_diff`, so nothing was drafted or published. If no diff was ever \
             expected, retry the node with `expects_no_diff: true`, which settles it done \
             without a dispatch; otherwise amend its task and retry it to produce one"
        ))),
        ..Settlement::plain(&node.id, NodeStatus::Failed, Some(engine::EMPTY_BRANCH))
    })
}

/// The status a publication settles its node at.
///
/// Read off what the publication **answered** rather than off the reason that was
/// asked for, because the two can differ: a host that would not draft the change
/// is a change that can land, and reporting it as held back would leave a landable
/// pin reported as safe. The reason decides only *which* draft a drafted change
/// is: one a release will arrive to lift holds the run, and one the plan asked
/// to leave for a person holds nothing — nothing in the run will ever lift it,
/// so the node is `done` and its dependents proceed.
fn drafted_status(
    outcome: &onevcs::PublishOutcome,
    reason: Option<&onevcs::DraftReason>,
) -> NodeStatus {
    match (outcome, reason) {
        (
            onevcs::PublishOutcome::ChangeDraft(_),
            Some(onevcs::DraftReason::AwaitingRelease { .. }),
        ) => NodeStatus::CompleteDraft,
        _ => NodeStatus::Done,
    }
}

/// What the closeout did to a change request the session already held, in the
/// one sentence a settlement carries about it.
///
/// Three facts and no more, each read off what happened rather than off what was
/// asked: who opened it — this session's worker, as a draft; an earlier
/// publication of this same branch that left it as one; or one that left it
/// open — whether the drafted description reached it, and whether it was marked
/// ready for review or left as the draft the plan asked for. Written once so
/// `results` and `status` read the same words.
fn finished_detail(
    change: &onevcs::SessionChange,
    worker_drafted: bool,
    described: bool,
    left_as_draft: bool,
) -> String {
    let opened = match (change.draft, worker_drafted) {
        (true, true) => "the worker opened the change request as a draft",
        (true, false) => "an earlier publication of this branch left the change request as a draft",
        (false, _) => {
            "the change request was already open from an earlier publication of this branch"
        }
    };
    let description = if described {
        "the closeout wrote the drafted description onto it"
    } else {
        "the closeout left the description as the worker left it"
    };
    let lifted = match (change.draft, left_as_draft) {
        (true, false) => " and marked it ready for review",
        (true, true) => " and left it as a draft",
        (false, _) => "",
    };
    format!("{opened}; {description}{lifted}")
}

/// One publication, and how many reads of the merge path it took to answer.
struct Published {
    /// What the last read answered, which is the publication the node settles on.
    answered: crate::error::Result<onevcs::Publication>,
    /// How many reads were made, for the settlement a spent budget writes. One
    /// on every path but the re-read, which is every publication that answered
    /// the first time it was asked.
    reads: std::num::NonZeroU32,
}

/// Publish, and where the push reached the remote with the merge path unread,
/// **re-read that path** rather than sending the agent back to the tree.
///
/// The one publication failure whose fix is not more work: the push landed, and
/// re-dispatching the agent buys a fresh clone and a fresh gate to re-push what
/// the remote already carries.
///
/// The re-read is [`crate::vcs::publish`] on the **same session**, which for a
/// branch already on the remote is `onevcs`'s own guidance for this state and is a
/// read of the host rather than a second push. Nothing here asks a host anything
/// itself: a second route to a change request's state would be host knowledge
/// regrown in the composition layer.
///
/// Bounded by [`engine::merge_path_reads`]. Whatever a read answers is what the
/// node settles on, and the routing that follows is the one that word has always
/// had.
fn publish_rereading_the_merge_path(
    node: &Node,
    token: &onevcs::SessionToken,
    body: Option<&str>,
    draft: Option<&onevcs::DraftReason>,
    cancel: &crate::executor::CancellationToken,
) -> Published {
    let publish =
        || crate::vcs::publish(token, node.merge_policy, node.title.as_deref(), body, draft);
    let budget = engine::merge_path_reads();
    let mut backoff = engine::merge_path_backoff();
    let mut reads = std::num::NonZeroU32::MIN;
    let mut answered = publish();
    while reads < budget && still_unread(&answered) && waited(backoff, cancel) {
        backoff = engine::doubled(backoff);
        reads = reads.saturating_add(1);
        answered = publish();
    }
    Published { answered, reads }
}

/// Wait out one backoff, answering `false` if the run was stopped instead.
///
/// In steps rather than one sleep because the backoff doubles to two minutes, and
/// a teardown held open that long by a poll of somebody else's API is the thing
/// the cancellation is for.
fn waited(backoff: std::time::Duration, cancel: &crate::executor::CancellationToken) -> bool {
    /// How often the wait looks up. Short enough that a stop is not felt as a
    /// pause, long enough that waiting out a two-minute backoff is not a spin.
    const STEP: std::time::Duration = std::time::Duration::from_millis(50);

    let until = std::time::Instant::now() + backoff;
    loop {
        if cancel.is_cancelled() {
            return false;
        }
        let left = until.saturating_duration_since(std::time::Instant::now());
        if left.is_zero() {
            return true;
        }
        std::thread::sleep(left.min(STEP));
    }
}

/// Whether a publication ended with its push on the remote and the path behind it
/// unread — the one ending another read of the host could answer.
///
/// A publication `onevcs` **refused** is not one: the sibling draws the line
/// between a request it would not take and a publication that ran and did not
/// land, and a refusal is the first of those. Nothing about it says a push reached
/// anything, so re-reading it would ask the host about work that may never have
/// left this machine. It settles as the residual, exactly as it always did.
fn still_unread(answered: &crate::error::Result<onevcs::Publication>) -> bool {
    let Ok(published) = answered else {
        return false;
    };
    matches!(
        &published.outcome,
        onevcs::PublishOutcome::Failed { kind, .. }
            if crate::vcs::failure_of(*kind) == crate::vcs::Failure::Unread
    )
}

/// The settlement of a node whose work reached the remote and whose merge path
/// the reads never answered.
///
/// **Not re-dispatched**: nothing rejected the tree, so asking a worker again
/// would republish what the origin already carries. `reason` is `onevcs`'s own
/// sentence, which names the branch, the commit, what stopped the read, and the
/// `publish-branch` that lands it once the host is back.
fn unread_merge_path(
    node: &str,
    token: &onevcs::SessionToken,
    branch: Option<String>,
    reason: &str,
    reads: std::num::NonZeroU32,
    body_aside: Option<String>,
) -> Attempt {
    let how_many = format!(
        "the merge path was read {reads} time{} and never answered",
        if reads.get() == 1 { "" } else { "s" }
    );
    Attempt::settled(Settlement {
        branch,
        head: crate::vcs::branch_head_in(token),
        detail: Some(compose(
            &format!("onevcs: {reason}. {how_many}"),
            body_aside.as_deref(),
        )),
        ..Settlement::plain(node, NodeStatus::Failed, Some(crate::vcs::Failure::UNREAD))
    })
}

/// How one attempt at a lifecycle node ended.
///
/// Two cases, and the split is the whole of the routing: a settlement is the
/// node's answer and the loop stops on it, and a publication that failed leaving
/// the work on its branch is an attempt rather than an answer. Cases rather than
/// a settlement a caller inspects afterwards, because everything a continuation
/// needs exists only on the second and would otherwise be `Option`s on every
/// settlement this crate makes.
enum Attempt {
    Settled(Box<Settlement>),
    Preserving(Box<Preserved>),
}

impl Attempt {
    /// One settled attempt, boxed as the other case is: a settlement is much the
    /// larger of the two, and an enum carrying it inline would move that whole
    /// value through every return between here and the loop.
    fn settled(settlement: Settlement) -> Self {
        Self::Settled(Box::new(settlement))
    }
}

/// A publication that failed and handed its branch back.
///
/// What the *next* attempt is dispatched with, and what the settlement says if
/// there is no next attempt — the same four values serve both, because a budget
/// that runs out has to report exactly the failure it stopped re-dispatching.
struct Preserved {
    branch: String,
    outcome: crate::vcs::Preserving,
    /// Already bounded and folded onto one line, because it reaches a reader
    /// through an envelope payload and a settlement detail, both of which are
    /// read back a line at a time.
    reason: String,
    evidence: Vec<crate::vcs::Evidence>,
    /// What this attempt has to say about the change request's body beside the
    /// publication's own words — that none was drafted, or that one was and the
    /// host refused to write it — carried so that the settlement a spent budget
    /// writes says it exactly as one that settled straight away does.
    body_aside: Option<String>,
    /// Where the session left this branch, which is the tip that was published.
    ///
    /// Read off `onevcs`'s own record of the session — the library that made the
    /// commit is the one that knows what it is at — so comparing two attempts is
    /// comparing two of its answers rather than a second account of a branch
    /// grown here. Three cases rather than an absent commit, because
    /// [`SessionTip::Unmoved`] is evidence the branch did not move and
    /// [`SessionTip::Unknown`] is the absence of evidence either way.
    ///
    /// [`SessionTip::Unmoved`]: crate::vcs::SessionTip::Unmoved
    /// [`SessionTip::Unknown`]: crate::vcs::SessionTip::Unknown
    tip: crate::vcs::SessionTip,
}

/// Settle or continue one failed publication.
///
/// Preserving is **two** conditions and not one. The failure has to be a kind a
/// further attempt could answer — [`crate::vcs::failure_of`] decides that — and
/// the branch has to still exist outside the session, which is `onevcs`'s
/// [`Retention`](onevcs::Retention) answer: a session's clone is disposable, so a
/// branch the execution checkout refused went with it and there is nothing left
/// to continue. Sending a node back to work on a branch nobody kept would cut a
/// fresh one from the base and republish an empty tree, reporting a recovery that
/// recovered nothing.
#[allow(
    clippy::too_many_arguments,
    reason = "the failure's own five values — which kind, what it said, what became of the \
              branch, which branch, and which node — plus the session the evidence is read \
              off and the drafting ending the settlement carries either way. Bundling them \
              would name a struct whose only constructor is this call site"
)]
fn failed_publication(
    node: &str,
    token: &onevcs::SessionToken,
    branch: Option<String>,
    kind: onevcs::FailureKind,
    reason: &str,
    retained: Option<&onevcs::Retention>,
    body_aside: Option<String>,
) -> Attempt {
    let failure = crate::vcs::failure_of(kind);
    let handed_back = matches!(retained, Some(onevcs::Retention::HandedBack(_)));
    let settled = || {
        Attempt::settled(Settlement {
            branch: branch.clone(),
            detail: Some(compose(&format!("onevcs: {reason}"), body_aside.as_deref())),
            ..Settlement::plain(node, NodeStatus::Failed, Some(failure.outcome()))
        })
    };
    // llmlint: ignore-block[changed_behavior_has_e2e] the second arm covers two cases and
    // only one of them is new. The **terminal** one — a failure no further attempt can
    // answer — is driven end to end by
    // `a_publication_onevcs_refuses_outright_settles_the_residual_and_is_not_retried`,
    // which asserts the residual word and that the node was dispatched exactly once. It
    // reaches the arm through a hosted identity this build has no `RemoteHost` for,
    // because a repository's own verification no longer produces a terminal kind at all:
    // `onevcs` 0.11.0 runs no gate, and the merge path refusing a push is `push-rejected`,
    // which is preserving. The other case is a preserving failure whose branch the
    // execution checkout refused: `onevcs` hands a branch back on every failure it can and
    // reports `Refused` only when that copy itself failed — a checkout that could not be
    // written to — which no double here injects and which the hook script deliberately
    // keeps out of the repository. Reaching it would mean breaking the checkout
    // mid-publication, which proves the fixture rather than this arm, and what it does is
    // exactly what the terminal case does.
    match (failure, handed_back, branch.clone()) {
        (crate::vcs::Failure::Preserving(outcome), true, Some(branch)) => {
            Attempt::Preserving(Box::new(Preserved {
                branch,
                outcome,
                reason: engine::bounded(&crate::views::one_line(reason)),
                evidence: crate::vcs::evidence_in(token),
                body_aside,
                tip: crate::vcs::session_tip(token),
            }))
        }
        _ => settled(),
    } // llmlint: ignore-end[changed_behavior_has_e2e]
}

/// A publication's own words and what is to be said about the body, in that
/// order.
///
/// Written once because three settlements compose the pair, and three spellings
/// of it would come to disagree about the order or the punctuation.
fn compose(detail: &str, body_aside: Option<&str>) -> String {
    match body_aside {
        Some(why) => format!("{detail}. {why}"),
        None => detail.to_owned(),
    }
}

/// The commit an attempt republished, where the attempt before it published one
/// and this one left the branch at the same place.
///
/// Named for the **tip** and not for the tree, because that is what is compared:
/// two commits carrying identical trees are two commits, and this answers `None`
/// for them, which is the safe direction. `None` too where nothing is known — a
/// record that could not be read is not evidence that a branch did not move, and
/// the attempt is spent exactly as it was before any of this existed.
fn republished(published: Option<&str>, tip: &crate::vcs::SessionTip) -> Option<String> {
    let published = published?;
    match tip {
        // The same commit, named twice.
        crate::vcs::SessionTip::At(commit) => {
            (commit.as_str() == published).then(|| published.to_owned())
        }
        // A session that committed nothing left the branch at what the attempt
        // before it published.
        crate::vcs::SessionTip::Unmoved => Some(published.to_owned()),
        crate::vcs::SessionTip::Unknown => None,
    }
}

/// The settlement of a node whose attempt republished the commit its previous
/// attempt published — the branch did not move, so its tree did not either.
///
/// **The residual and not the failure's own word.** Each of the four preserving
/// words says something about the branch, and the branch is not what refused
/// this; [`Failure::Terminal`] is the ending whose meaning covers it. The attempt
/// is handed back rather than spent, and the settlement says how much is left.
///
/// [`Failure::Terminal`]: crate::vcs::Failure::Terminal
fn republished_the_same_commit(
    node: &str,
    preserved: &Preserved,
    endings: &[crate::vcs::Preserving],
    head: &str,
    attempt: std::num::NonZeroU32,
) -> Settlement {
    let attempts = engine::publication_attempts();
    // The attempt this settles on is given back, so what it cost the budget is
    // every attempt *before* it.
    let unspent = attempts
        .get()
        .saturating_sub(attempt.get().saturating_sub(1));
    let roll_up = format!(
        "{count} publication attempt{plural} on {branch} published the same commit, {head}, \
         and ended {endings}: the refusal is not about the branch, so no further attempt on \
         it could answer differently. {unspent} of {attempts} publication attempts are \
         unspent",
        count = endings.len(),
        plural = if endings.len() == 1 { "" } else { "s" },
        branch = preserved.branch,
        endings = endings
            .iter()
            .map(|ending| ending.outcome())
            .collect::<Vec<_>>()
            .join(", "),
    );
    Settlement {
        branch: Some(preserved.branch.clone()),
        head: Some(head.to_owned()),
        detail: Some(compose(
            &format!("onevcs: {}. {roll_up}", preserved.reason),
            preserved.body_aside.as_deref(),
        )),
        ..Settlement::plain(
            node,
            NodeStatus::Failed,
            Some(crate::vcs::Failure::RESIDUAL),
        )
    }
}

/// The settlement of a node that will not be dispatched again — because its
/// publication budget is spent, or because the run is being cancelled before it
/// could be. Both endings settle as the publication failure, so both are written
/// here.
///
/// The **last** failure's word, because that is the one standing in the way, over
/// a roll-up of every attempt: without it a reader sees one failure and cannot
/// tell it from a node that failed once — which is the difference between "fix
/// this check" and "this check is never going to pass".
fn stopped_retrying(
    node: &str,
    preserved: &Preserved,
    endings: &[crate::vcs::Preserving],
) -> Settlement {
    let each: Vec<String> = endings
        .iter()
        .enumerate()
        .map(|(index, ending)| format!("{} {}", index + 1, ending.outcome()))
        .collect();
    let roll_up = format!(
        "{} publication attempt{} on {}: {}",
        endings.len(),
        if endings.len() == 1 { "" } else { "s" },
        preserved.branch,
        each.join(", ")
    );
    Settlement {
        branch: Some(preserved.branch.clone()),
        detail: Some(compose(
            &format!("onevcs: {}. {roll_up}", preserved.reason),
            preserved.body_aside.as_deref(),
        )),
        // **No step** is recorded as completed, and that is the same rule the
        // re-dispatch was made under seen from the other end. The branch carries
        // a tree the merge path rejected, so a `retry` that skipped the steps it
        // already holds would publish that tree again unaltered and meet the same
        // refusal — the ending this whole loop exists to avoid, reached by hand
        // instead of automatically.
        ..Settlement::plain(node, NodeStatus::Failed, Some(preserved.outcome.outcome()))
    }
}

/// The node the next attempt is dispatched as.
///
/// Three changes and no others, made to the node **as it was launched** rather
/// than to the attempt before. It is **pinned to the preserved branch**, so the
/// session `onevcs` opens continues that branch from its own tip rather than
/// cutting a second one beside committed work. It records **no step as
/// completed**, so every step runs again — against the tree that was rejected,
/// which is the tree that has to change; a continuation that skipped the steps
/// the branch already carries would republish it unaltered and meet the same
/// refusal, which is the one failure this must not have. And it carries the
/// **diagnosis** as its node context, so the worker meets the failure rather than
/// having to go and find it.
///
/// The planner's own note **survives** it, above the diagnosis: this attempt is
/// the engine's continuation of the dispatch that note was composed into, not a
/// dispatch anybody asked for, so a ruling the first attempt was given is not
/// spent by a retry nobody issued. It is the launched node's context that is
/// kept, because the attempt before carried the diagnosis of the one before it,
/// and each diagnosis already names every ending so far.
fn continued(
    launched: &Node,
    preserved: &Preserved,
    attempt: std::num::NonZeroU32,
    attempts: std::num::NonZeroU32,
    endings: &[crate::vcs::Preserving],
) -> Node {
    let diagnosis = diagnosis(preserved, attempt, attempts, endings);
    let context = match launched.context.as_deref().map(str::trim) {
        Some(note) if !note.is_empty() => format!("{note}\n\n{diagnosis}"),
        _ => diagnosis,
    };
    Node {
        branch: Some(preserved.branch.clone()),
        resume: None,
        context: Some(context),
        ..launched.clone()
    }
}

/// What the next attempt is told about the one before it.
///
/// The failure's own reason and a pointer to every artifact its publication
/// recorded — the check's log, the push's output, the conflict's hunks — by id,
/// because the artifact is somebody else's megabytes and what a worker needs is
/// the fetch that gets it. Named as *observed state* by the section it is
/// rendered into, so a worker cannot read a failure report as a new bar to clear.
fn diagnosis(
    preserved: &Preserved,
    attempt: std::num::NonZeroU32,
    attempts: std::num::NonZeroU32,
    endings: &[crate::vcs::Preserving],
) -> String {
    let mut note = format!(
        "The previous attempt's publication failed and its branch was preserved. This is \
         attempt {attempt} of {attempts}, and it continues that same branch — {branch} — so \
         the tree that was rejected is the tree this dispatch starts from. Change what the \
         failure below is about; republishing it unaltered meets the same refusal.\n\n\
         The publication ended `{ending}`, and `onevcs` said:\n\n{reason}\n",
        branch = preserved.branch,
        ending = preserved.outcome.outcome(),
        reason = preserved.reason,
    );
    if endings.len() > 1 {
        let each: Vec<&str> = endings.iter().map(|ending| ending.outcome()).collect();
        note.push_str(&format!(
            "\nEvery attempt so far ended: {}.\n",
            each.join(", ")
        ));
    }
    if !preserved.evidence.is_empty() {
        note.push_str(
            "\nThe publication recorded this evidence, each fetched with \
             `onevcs artifact cat ID`:\n",
        );
        for evidence in &preserved.evidence {
            note.push_str(&format!("- {} — {}\n", evidence.kind.0, evidence.id.0));
        }
    }
    note
}

/// The task a drafting dispatch is given, ahead of the node's own.
///
/// Byte for byte what it has always been: the consumer host reads it out of this
/// binary to hold its own out-of-band drafter to the same prompt, and so does it
/// read each of the four literals below, which is why every one of them is a
/// constant on a line of its own rather than a piece of a format string.
const DRAFTING_TASK: &str = "Read this branch's diff and write the change request's body, \
     following the repository's own template. The task this branch delivered:";

/// The heading under which a drafting dispatch is shown the change request its
/// session already holds. Only there: a session holding none gets no section.
const CHANGE_REQUEST_HEADING: &str = "## Change request";

/// The line under that heading saying whether the worker is holding the change
/// request as a draft, followed by `yes` or `no`.
const HELD_AS_DRAFT_LINE: &str = "Held as a draft by the worker:";

/// The heading under which the change request's description is shown to the
/// drafter as the worker left it, verbatim — or [`NO_DESCRIPTION`].
const DESCRIPTION_HEADING: &str = "### Description as the worker left it";

const NO_DESCRIPTION: &str = "(the worker left no description)";

/// The heading every drafting dispatch is told how to read the worker's
/// transcript under: the verb, with this run's and this node's ids substituted,
/// and the two environment variables that verb reads.
const WORKER_TRANSCRIPT_HEADING: &str = "## Worker transcript";

/// The whole task one drafting dispatch is given.
///
/// One composition, in a fixed order, every heading a `##` so it sits beside the
/// headings the rendered task already carries: the opening sentence, the node's
/// task as every drafting dispatch already gets it, the change request the
/// session holds — only where it holds one — and how to read the worker's
/// transcript. The last is always there, because what evidence the worker
/// produced, and when, is the worker's to decide and the drafter's to find: a
/// demonstration change request it opened, a comment a bot posted, a check it
/// triggered.
fn drafting_task(node: &Node, run: &str, held: Option<&onevcs::SessionChange>) -> String {
    let mut task = format!("{DRAFTING_TASK}\n\n{}", node.rendered_task());
    if let Some(change) = held {
        let description = if change.body.trim().is_empty() {
            NO_DESCRIPTION
        } else {
            change.body.as_str()
        };
        task.push_str(&format!(
            "\n\n{CHANGE_REQUEST_HEADING}\n{url}\n{HELD_AS_DRAFT_LINE} {held}\n\n\
             {DESCRIPTION_HEADING}\n{description}",
            url = change.url,
            held = if change.draft { "yes" } else { "no" },
        ));
    }
    task.push_str(&format!(
        "\n\n{WORKER_TRANSCRIPT_HEADING}\n\
         `onepipeline transcript {run} {node}` renders every tool call the worker made on this \
         branch and what each returned; `{run_env}` and `{runs_dir_env}` in this dispatch's \
         environment are what that command reads. Read it to find evidence the worker \
         produced that belongs in the description — a demonstration change request it opened, \
         a comment a bot posted, a check it triggered.",
        node = node.id,
        run_env = crate::agentgraph::RUN_ID_ENV,
        runs_dir_env = crate::agentgraph::RUNS_DIR_ENV,
    ));
    task
}

/// What one drafting dispatch ended as.
///
/// A body or an ending that is not one, because **every** ending here leaves the
/// publication to proceed with no body: the two are what a change request opens
/// with, not whether it opens.
enum Drafted {
    /// It drafted the change request's body.
    Body(String),
    /// It ended with none, and which of the three endings it was.
    Undrafted(Undrafted),
}

/// A drafting dispatch that produced no body, and how.
///
/// Three endings kept apart rather than one "it did not work", because they need
/// three different fixes: a graph that will not start or will not finish, one
/// whose answers the schema refuses, and one that answers inside the schema with
/// nothing in it. A run that had just wired a drafter could tell none of them
/// from a launch that had wired no drafter at all.
pub(crate) enum Undrafted {
    /// It could not be run, or it ran and did not succeed — in its own words.
    ///
    /// One ending rather than three: a dispatch that never started, one that
    /// failed, and one that was cancelled differ in the reason they carry and in
    /// nothing else a publication carrying no body either way can act on.
    Dispatch(String),
    /// It succeeded, and the schema it was validated against refused every
    /// answer it made.
    SchemaRefused,
    /// It succeeded and there was no body in what it answered with.
    ///
    /// The widest of the three on purpose. It is where a dispatch lands that
    /// succeeded and had nothing refused: one that answered inside its schema
    /// and put nothing in it, one no schema was asked of, and one whose reports
    /// this run holds no readable copy of. They differ in nothing a reader acts
    /// on differently — a drafter that succeeded and produced no prose is the
    /// same fix in each — and none of them is a schema to correct, which is
    /// what keeps them out of [`SchemaRefused`](Self::SchemaRefused).
    Bodyless,
}

impl Undrafted {
    /// The ending, as the event names it.
    pub(crate) fn ending(&self) -> &'static str {
        match self {
            Self::Dispatch(_) => "dispatch-failed",
            Self::SchemaRefused => "schema-refused",
            Self::Bodyless => "no-body",
        }
    }

    /// Why the change request opened with no body, in the words a planner reads
    /// off `results`.
    pub(crate) fn why(&self) -> String {
        match self {
            Self::Dispatch(reason) => {
                format!("the change request's body was not drafted: {reason}")
            }
            Self::SchemaRefused => "the change request's body was not drafted: the drafting \
                 dispatch answered nothing the schema it was validated against accepted"
                .to_owned(),
            Self::Bodyless => "the change request's body was not drafted: the drafting \
                 dispatch succeeded and there was no body in what it answered with"
                .to_owned(),
        }
    }
}

/// One post-verification dispatch drafting the change request's body, when the
/// launch named a graph to draft it with and the node carries none of its own.
///
/// It runs **after** the branch has been verified and is not on the publication
/// path: every way it can end badly leaves the change request to open with no
/// body, and the node settles on its publication as before. That is the whole
/// point of running it here rather than making it a step. What each of those
/// ways *was* is [`Undrafted`], reported beside the publication rather than
/// folded into it.
///
/// It runs in the node's **own** worktree, which is the only place the diff it is
/// asked to read exists: a session of its own would be a fresh clone cut from the
/// base, carrying nothing this node wrote — and opening one reclaims the session
/// still holding the work. So a node with no worktree to run it in drafts
/// nothing, out loud, rather than dispatching an agent to read an empty diff.
///
/// `None` is the one ending that is not a failure and is not reported: a launch
/// that named no drafting graph. This crate ships the flag, not the document, so
/// naming none is the shipped default and there is nothing to say about it. A
/// node carrying its own `body` never reaches here at all.
#[allow(
    clippy::too_many_arguments,
    reason = "the draft is a dispatch inside one lifecycle execution and needs that \
              execution's executor, the run's own paths, what its launch decided, the node, \
              the workspace, the change request the session holds, cancellation, and the \
              event stream"
)]
fn drafted(
    executor: &dyn Executor,
    paths: &RunPaths,
    launch: &Launch,
    node: &Node,
    worktree: Option<&std::path::Path>,
    held: Option<&onevcs::SessionChange>,
    cancel: &crate::executor::CancellationToken,
    tx: &Sender<Message>,
) -> Option<Drafted> {
    let graph = launch.pr_author_graph.as_deref()?;
    let Some(worktree) = worktree else {
        // A dispatch that was configured and could not be run at all, which is
        // the same ending as one the executor refused: it is said out loud, as
        // it always was, and now recorded as well.
        let why = "there was no worktree to read this branch's diff in";
        eprintln!(
            "onepipeline: node '{}': no worktree to draft its change request in, \
             so it publishes with no body",
            node.id
        );
        return Some(Drafted::Undrafted(Undrafted::Dispatch(why.to_owned())));
    };
    let dispatch = executor.dispatch(DispatchRequest {
        graph: oneagentgraph::config::ConfigRef(graph.to_owned()),
        task: drafting_task(node, &paths.run, held),
        labels: engine::dispatch_labels(&paths.run, &node.id, None, Some(PR_AUTHOR_PERSONA)),
        // None of the node's own: the drafting dispatch is not the node's work,
        // and a turn budget written for that work would be spent twice — once on
        // it and once here — if this dispatch inherited it.
        controls: NodeControls::default(),
        workspace: WorkspaceSpec::Path(worktree.to_path_buf()),
        cancel: cancel.clone(),
    });
    let mut handle = match dispatch {
        Ok(handle) => handle,
        Err(error) => {
            return Some(undrafted(
                &node.id,
                format!("the drafting dispatch could not start: {error}"),
            ))
        }
    };
    let mut retained = Vec::new();
    for envelope in handle.events() {
        // A line off this stream that will not parse costs that line and nothing
        // more: what the loop is looking for is a `member-settled` naming a
        // retained report, and a drafting run that never produces one already
        // publishes with no body two lines below. Skipping is therefore the same
        // ending an unreadable stream would reach by any other route, reported
        // the same way.
        // llmlint: ignore[changed_behavior_has_e2e] no double can produce this: the
        // envelopes come off the real `oneagentgraph`'s own stdout, which is well-formed
        // by construction, and there is no fault-injection seam here to drive one through.
        // The ending it falls back to — a dispatch that yields no body, and a publication
        // that proceeds without one — is driven end to end by
        // `a_drafting_graph_the_runner_refuses_still_publishes_the_change_request`.
        let Ok(envelope) = envelope else { continue };
        // **Ingest**, and the same ingest the engine performs on the envelopes
        // this relays to it: the line is arriving on the stdout of a process
        // this crate started, which is the one moment the path it names carries
        // the producer's authority rather than the journal's. What is read
        // below is that copy, at a path derived from the settlement — this
        // crate never opens the path a producer named.
        crate::report::retain(paths, &envelope);
        if envelope.source == crate::event::Source::Agentgraph
            && envelope.kind.0 == crate::report::MEMBER_SETTLED
        {
            retained.push(paths.report_for(&envelope.stream, envelope.seq));
        }
        let _ = tx.send(Message::Event(Box::new(envelope)));
    }
    // llmlint: ignore-block[changed_behavior_has_e2e] the last arm below is reached by a
    // dispatch that failed and by one that was cancelled; the first has a journey of its
    // own in `tests/e2e/lifecycle.rs` and the second has none because it is not separately
    // reachable. Nothing cancels a drafting dispatch except the node's own token being
    // flipped, which happens when the run is being stopped — and a run whose driver is
    // being torn down has no publication left to protect, so a journey claiming "it
    // published anyway" would be asserting the opposite of what a stop means. Deleting the
    // arm is not the alternative either: it is the same `_` a failed settlement takes.
    match handle.wait() {
        Ok(outcome) if outcome.succeeded => {
            // Every report the dispatch retained, read as **one** answer: a
            // fallback chain records a candidate per identity it tried, and
            // which report an entry landed in is not the reader's business.
            let kept: Vec<serde_json::Value> = retained
                .iter()
                .filter_map(|kept| crate::report::read(kept))
                .collect();
            Some(match crate::report::drafted(&kept) {
                crate::report::Drafted::Body(body) => Drafted::Body(body),
                crate::report::Drafted::SchemaRefused => {
                    Drafted::Undrafted(Undrafted::SchemaRefused)
                }
                crate::report::Drafted::Bodyless => Drafted::Undrafted(Undrafted::Bodyless),
            })
        }
        Ok(outcome) => Some(undrafted(
            &node.id,
            format!(
                "the drafting dispatch settled without succeeding: {}",
                first_line(&outcome.detail)
            ),
        )),
        Err(error) => Some(undrafted(
            &node.id,
            format!("the drafting dispatch could not be waited on: {error}"),
        )),
    } // llmlint: ignore-end[changed_behavior_has_e2e]
}

/// A drafting dispatch that produced no body because the **dispatch** failed,
/// said out loud and then recorded.
///
/// Out loud for every one of those endings rather than one of them, because the
/// reason does not distinguish between them: a launch that named a drafting graph
/// and silently drafted nothing is indistinguishable from one that named none,
/// and the change request it opens carries no sign of it either way. Which
/// ending a given failure takes depends on where the dispatch runs — a graph the
/// runner refuses is a refusal to the caller that launched it in its own process
/// and a settlement to the one that gave it a process of its own — and an
/// operator reading stderr must not be told only on the arms one deployment
/// happens to reach.
fn undrafted(node: &str, why: String) -> Drafted {
    eprintln!("onepipeline: node '{node}': {why}, so it publishes with no body");
    Drafted::Undrafted(Undrafted::Dispatch(why))
}

/// A dispatch's own words, as one bounded line of a settlement detail.
///
/// The reason a dispatch gives is its stderr, which is many lines of a sibling's
/// diagnostics; what belongs beside a publication is the first of them, held to
/// the same bound every other payload text this crate writes is held to.
fn first_line(detail: &str) -> String {
    match detail.lines().find(|line| !line.trim().is_empty()) {
        Some(line) => engine::bounded(line.trim()),
        None => "it reported nothing".to_owned(),
    }
}
// llmlint: ignore-end[invalid_states_unrepresentable]

/// Put every envelope a followed session writes into the merged stream.
fn relay_into(tx: &Sender<Message>, node: Labels) -> Box<dyn Fn(Envelope) + Send> {
    let tx = tx.clone();
    Box::new(move |mut envelope| {
        stamp(&mut envelope.labels, &node);
        let _ = tx.send(Message::Event(Box::new(envelope)));
    })
}

/// Say which node a session's envelope belongs to, where its producer could not.
///
/// `onevcs` stamps what it knows, and a session does not know it is a graph
/// node: the crate that opened it does. Without this a whole publication —
/// push, change request, merge — lands in the merged store belonging to no node,
/// so every per-node view reads it as work that happened to nobody.
///
/// An enricher, so it never rewrites: a key the producer stamped stands.
pub(crate) fn stamp(labels: &mut Labels, known: &Labels) {
    labels.run_id = labels.run_id.take().or_else(|| known.run_id.clone());
    labels.node = labels.node.take().or_else(|| known.node.clone());
    labels.persona = labels.persona.take().or_else(|| known.persona.clone());
}

/// How long ending a session waits for the terminator its stream ends with.
///
/// Nothing in the ordinary case: a session whose terminator is written answers on
/// the first read. The grace bounds the case where none is.
const TERMINATOR_GRACE: Duration = Duration::from_secs(5);

/// How often that wait re-reads the stream, and re-asks a close that refused.
const TERMINATOR_POLL: Duration = Duration::from_millis(250);

/// Close the session, and collect what its stream said.
///
/// **The follow ends first, and the close comes after it.** `onevcs` refuses to
/// release a session while any live process is working inside its run root, and
/// the follow's own poll asks the library whether the session closed — a call
/// that shells out to `git` in that session's clone. Closed first, as this once
/// was, the close raced the follow's own children and lost intermittently: the
/// session stayed open and the refusal was the only account of why.
///
/// Closing is still what writes the session's last record, so a follow can end
/// having relayed everything but the tail, and the stream is read again
/// afterwards from the point it reached. **Again, rather than once more**: every
/// way a read can go badly hands back an empty batch, so one read made the last
/// record conditional on that read having gone well, and a node settled in
/// silence with a hole in its record.
fn end_session(
    stream: Option<crate::vcs::Follower>,
    tx: &Sender<Message>,
    token: Option<&onevcs::SessionToken>,
    node: &Labels,
    filter: Option<&EventFilter>,
) {
    // Empty from either side is the whole stream still to read: no follow was
    // started, or one was and relayed nothing.
    let followed_through = stream.map(crate::vcs::Follower::finish).unwrap_or_default();
    let refused = close(token);
    relay_session_events(tx, token, node, followed_through, filter, refused);
}

/// Fold the part of the session's stream nothing has relayed into the merged one.
///
/// `onevcs` records the commits and the publication against the session, the
/// merge path's own verdict on the `push` among them; without this the merged
/// store would carry a lifecycle node's settlement with none of the evidence
/// behind it. `followed_through` is the highest `seq` the follow already relayed
/// **per stream**, so a record arrives **once**: each stream is numbered
/// monotonically and resumes its series across the processes that write to it,
/// which makes those marks the whole of the bookkeeping. Per stream and not one
/// mark over all of them, because reading one session hands back the identity's
/// release records as well as the session's own, in a series of their own.
fn relay_session_events(
    tx: &Sender<Message>,
    token: Option<&onevcs::SessionToken>,
    node: &Labels,
    followed_through: crate::vcs::Watermarks,
    filter: Option<&EventFilter>,
    refused: Option<String>,
) {
    let Some(token) = token else { return };
    let relay = relay_into(tx, node.clone());
    // Carried across the reads rather than re-derived from each: what stops a
    // record arriving twice is the mark the relay left on it, and a second read of
    // a stream hands back everything the first one did.
    let mut relayed = followed_through;
    let mut refused = refused;
    let deadline = Instant::now() + TERMINATOR_GRACE;
    // llmlint: ignore-block[changed_behavior_has_e2e] every lifecycle journey drives the
    // first pass. A second one needs a stream that gains its last record late or never,
    // which no plan can ask for — `a_session_whose_terminator_arrives_late_still_relays_it_and_one_with_none_says_so`
    // holds both, against a real stream through the real reader.
    loop {
        // The same filter the follow was opened with: this read covers the tail of
        // the *same* stream, so a run that filtered what it followed and not what
        // it caught up on would relay events it said it did not want, for no
        // reason but which side of a settlement they landed on.
        let read = crate::vcs::events(token, filter);
        // Asked of the whole read rather than of what this pass relays: the follow
        // may have relayed the terminator already, and what the wait is for is a
        // session that has *ended*, however its last record reached the store.
        let ended = read.iter().any(crate::vcs::is_terminator);
        for envelope in beyond(read, &relayed) {
            relayed.reached(&envelope);
            relay(envelope);
        }
        if ended {
            return;
        }
        if Instant::now() >= deadline {
            // The one thing a reader of the merged store cannot work out for
            // itself: the records are all there and the last one is missing, which
            // is indistinguishable from a publication still running.
            eprintln!(
                "onepipeline: session {} ended with no `session-closed` record{}",
                token.0,
                refused
                    .map(|why| format!("; its close refused: {why}"))
                    .unwrap_or_default()
            );
            return;
        }
        // Asked again only where the ask before it refused, and only after the read
        // above found no terminator — so a close that emitted one and then failed is
        // never asked to emit a second. `onevcs` refuses a close over a live process
        // working inside the run root, which is exactly what the dispatch that has
        // just finished is until the host reaps it: a refusal that answers
        // differently a moment later, and the only one that leaves a session with no
        // last record at all.
        if refused.is_some() {
            refused = close(Some(token));
        }
        std::thread::sleep(TERMINATOR_POLL);
    }
    // llmlint: ignore-end[changed_behavior_has_e2e]
}

/// The part of what a read handed back that a follow did not already relay.
///
/// Empty marks are the whole of it — no follow was started, or one was and
/// relayed nothing — which is a stream still to read rather than a stream that
/// held nothing. Otherwise everything numbered past the mark **its own stream**
/// stands at, and nothing at or below it: a record relayed twice is the same
/// defect as one lost, seen from the other side.
fn beyond(envelopes: Vec<Envelope>, followed_through: &crate::vcs::Watermarks) -> Vec<Envelope> {
    envelopes
        .into_iter()
        .filter(|envelope| followed_through.beyond(envelope))
        .collect()
}

/// Close the session, and answer with what refused it.
///
/// Still best effort where it mattered — the refusal settles no node, so a node
/// that already failed is not reported as a different failure because its cleanup
/// also failed. It is kept because it is the one thing that explains a stream with
/// no terminator on it.
fn close(token: Option<&onevcs::SessionToken>) -> Option<String> {
    crate::vcs::session_close(token?)
        .err()
        .map(|refusal| refusal.to_string())
}

/// A node's steps in dispatch order, each with the controls its dispatch runs
/// under, or why the node has none it can run.
///
/// One function for both refusals a workstream can carry before it starts — a
/// dependency cycle among its steps, and a step whose declaration no dispatch
/// can honour — because they cost the same thing if they are found late: a
/// branch cut for a node that was never going to finish.
fn dispatchable_steps(node: &Node) -> std::result::Result<Vec<(Step, NodeControls)>, String> {
    ordered_steps(node)?
        .into_iter()
        .map(|step| {
            NodeControls::of_step(&step)
                .map(|controls| (step.clone(), controls))
                .map_err(|why| format!("node '{}': step '{}': {why}", node.id, step.id))
        })
        .collect()
}

/// A node's steps in topological order, or why they have none.
///
/// Steps share one branch and run serially, so the order is a total one: ties
/// are broken by the order the plan wrote them, which keeps a workstream
/// reproducible.
pub fn ordered_steps(node: &Node) -> std::result::Result<Vec<Step>, String> {
    let Some(steps) = &node.steps else {
        // A lifecycle node with no steps is one implicit step: its own persona
        // and task, on its own branch.
        return Ok(vec![Step {
            id: node.id.clone(),
            kind: node.kind,
            task: node.task.clone(),
            persona: node.persona.clone(),
            deps: Vec::new(),
            max_turns: node.max_turns,
            expects_no_diff: node.expects_no_diff,
            executor: node.executor.clone(),
            agent_graph: node.agent_graph.clone(),
        }]);
    };

    let by_id: BTreeMap<&str, &Step> = steps.iter().map(|s| (s.id.as_str(), s)).collect();
    let mut settled: BTreeSet<&str> = BTreeSet::new();
    let mut order: Vec<Step> = Vec::new();
    while order.len() < steps.len() {
        let mut progressed = false;
        for step in steps {
            if settled.contains(step.id.as_str()) {
                continue;
            }
            if step
                .deps
                .iter()
                .all(|dep| settled.contains(dep.as_str()) || !by_id.contains_key(dep.as_str()))
            {
                settled.insert(step.id.as_str());
                order.push(step.clone());
                progressed = true;
            }
        }
        if !progressed {
            return Err(format!(
                "node '{}': its steps have a dependency cycle",
                node.id
            ));
        }
    }
    Ok(order)
}

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

    /// What an attempt left the branch at decides whether it republished, and the
    /// three answers are three different things.
    ///
    /// The two an e2e drives — a session that committed nothing over a tree its
    /// worker did not change, and one that committed something new — are in
    /// `tests/e2e/lifecycle.rs`. The third is not reachable through the binary:
    /// the hook that makes a session's stream unreadable runs on every
    /// publishing push, so a run cannot be driven to a *later* attempt that
    /// cannot say where the branch is while an earlier one could. It is the one
    /// answer that must not be read as either of the others, so it is stated
    /// here.
    #[test]
    fn a_tip_nothing_could_read_is_not_evidence_that_the_branch_did_not_move() {
        let commit = |sha: &str| crate::vcs::Commit::of(sha).expect("a commit this crate carries");
        let published = Some("c0ffee");
        assert_eq!(
            republished(published, &crate::vcs::SessionTip::At(commit("c0ffee"))),
            Some("c0ffee".to_string()),
            "the same commit, published twice, was read as two"
        );
        assert_eq!(
            republished(published, &crate::vcs::SessionTip::At(commit("decaf"))),
            None,
            "a branch that moved was read as one that stood still"
        );
        assert_eq!(
            republished(published, &crate::vcs::SessionTip::Unmoved),
            Some("c0ffee".to_string()),
            "a session that committed nothing moved the branch it did not touch"
        );
        assert_eq!(
            republished(published, &crate::vcs::SessionTip::Unknown),
            None,
            "a record nothing could read was taken as proof the branch stood still"
        );
        // And nothing to compare against is nothing to conclude, whatever this
        // attempt says: the first attempt of every node arrives here.
        assert_eq!(republished(None, &crate::vcs::SessionTip::Unmoved), None);
    }

    /// The attempt that continues a preserved branch keeps the note the node was
    /// launched with, above the diagnosis — and keeps the *launched* node's, so a
    /// third attempt carries one diagnosis rather than a stack of them.
    ///
    /// The half an e2e cannot hold still: `tests/note/` drives the continuation
    /// against a real conversation and reads the task it was handed, but a
    /// third attempt on a host whose check flips is scheduling rather than a
    /// claim, so which diagnosis a later attempt carries is stated here.
    #[test]
    fn a_continuation_keeps_the_launched_note_and_carries_one_diagnosis() {
        let launched = Node {
            id: "service".into(),
            context: Some("the reviewer asked for a smaller diff".into()),
            ..Node::default()
        };
        let preserved = |reason: &str| Preserved {
            branch: "feat/service".into(),
            outcome: crate::vcs::Preserving::ChecksFailed,
            reason: reason.into(),
            evidence: Vec::new(),
            body_aside: None,
            tip: crate::vcs::SessionTip::Unknown,
        };
        let two = std::num::NonZeroU32::new(2).expect("two");
        let three = std::num::NonZeroU32::new(3).expect("three");
        let endings = [
            crate::vcs::Preserving::ChecksFailed,
            crate::vcs::Preserving::ChecksFailed,
        ];

        let second = continued(
            &launched,
            &preserved("llmlint red"),
            two,
            three,
            &endings[..1],
        );
        let context = second
            .context
            .as_deref()
            .expect("the continuation carries context");
        assert!(
            context.starts_with("the reviewer asked for a smaller diff\n\n"),
            "the launched note does not lead the continuation's context:\n{context}"
        );
        assert!(context.contains("attempt 2 of 3") && context.contains("llmlint red"));
        assert_eq!(second.branch.as_deref(), Some("feat/service"));
        assert!(second.resume.is_none());

        // The third is composed from the launched node too, not from the second:
        // one note, one diagnosis, and the endings so far named inside it.
        let third = continued(
            &launched,
            &preserved("llmlint still red"),
            three,
            three,
            &endings,
        );
        let context = third
            .context
            .as_deref()
            .expect("the continuation carries context");
        assert_eq!(
            context
                .matches("the reviewer asked for a smaller diff")
                .count(),
            1
        );
        assert_eq!(
            context
                .matches("The previous attempt's publication failed")
                .count(),
            1
        );
        assert!(!context.contains("llmlint red\n"), "{context}");
        assert!(context.contains("attempt 3 of 3") && context.contains("llmlint still red"));
        assert!(
            context.contains("Every attempt so far ended: checks-failed, checks-failed."),
            "{context}"
        );

        // And a node launched with no note carries the diagnosis alone.
        let bare = continued(
            &Node::default(),
            &preserved("red"),
            two,
            three,
            &endings[..1],
        );
        assert!(bare
            .context
            .as_deref()
            .is_some_and(|context| context.starts_with("The previous attempt's publication")));
    }

    /// The endings this module emits and the endings the contract names are one
    /// set.
    ///
    /// The wire spellings are stated twice — in `docs/contract.md`'s pr-author
    /// paragraph and in [`Undrafted::ending`] — and only one of them is
    /// compiled, so the document needs a gate the way the closed set of kinds
    /// has one in `tests/contract.rs`. It cannot live there: the type is private
    /// to this module, and a public one would widen the surface past what the
    /// contract names.
    #[test]
    fn every_ending_this_module_emits_is_one_the_contract_names() {
        let contract = std::fs::read_to_string(
            std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("docs/contract.md"),
        )
        .expect("the contract ships");
        let endings = [
            Undrafted::Dispatch(String::new()),
            Undrafted::SchemaRefused,
            Undrafted::Bodyless,
        ];
        for ending in &endings {
            assert!(
                contract.contains(&format!("`{}`", ending.ending())),
                "docs/contract.md does not name the `{}` ending this module emits",
                ending.ending()
            );
        }

        // And the other direction: a spelling the document carries and nothing
        // emits is a promise nobody keeps. The contract lists them in one
        // clause, so the clause is read and its backticked tokens compared with
        // the set above rather than the whole document searched.
        let clause = contract
            .split_once("carrying `ending` —")
            .expect("the contract lists the endings `body-not-drafted` carries")
            .1
            .split_once("— and `detail`")
            .expect("the clause ends where the detail begins")
            .0;
        let listed: Vec<&str> = clause.split('`').skip(1).step_by(2).collect();
        assert_eq!(
            listed,
            endings
                .iter()
                .map(Undrafted::ending)
                .collect::<Vec<&'static str>>(),
            "the contract's endings are not the ones this module emits"
        );

        // The sentences a reader is given are each the ending's own, so two
        // endings cannot arrive under one set of words.
        let why: std::collections::BTreeSet<String> = endings.iter().map(Undrafted::why).collect();
        assert_eq!(why.len(), endings.len(), "two endings say the same thing");
    }

    /// The README summarises the same set, so it is gated the same way.
    ///
    /// It is a third copy of the endings — the enum, the contract, and the
    /// user-facing prose — and the first two already hold each other. Left
    /// ungated the README is the one that goes quietly stale: nothing compiles
    /// it, and a reader meeting an ending it does not list has no way to know
    /// which of the two is behind.
    #[test]
    fn the_readmes_ending_summary_is_the_set_this_module_emits() {
        let raw = std::fs::read_to_string(
            std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("README.md"),
        )
        .expect("the README ships");
        // Wrapped prose, so match on its words rather than its line breaks.
        let readme = raw.split_whitespace().collect::<Vec<_>>().join(" ");
        let clause = readme
            .split_once("under one of three endings —")
            .expect("the README summarises the endings a drafting dispatch can reach")
            .1
            .split_once("— and the node's own settlement")
            .expect("the clause ends where the settlement's own half begins")
            .0;
        let listed: Vec<&str> = clause.split('`').skip(1).step_by(2).collect();
        assert_eq!(
            listed,
            [
                Undrafted::Dispatch(String::new()),
                Undrafted::SchemaRefused,
                Undrafted::Bodyless,
            ]
            .iter()
            .map(Undrafted::ending)
            .collect::<Vec<&'static str>>(),
            "the README's endings are not the ones this module emits"
        );
    }

    /// The literals the drafting task is composed from, and the environment
    /// names every dispatch carries, are exactly what divergence entry 69 names.
    ///
    /// Private vocabulary, so `tests/contract.rs` cannot reach it and the entry
    /// is the only place it is written down beside the code — and the consumer
    /// host reads each literal out of this binary to hold its own drafter to the
    /// same prompt, so one reworded here without the record moving is a prompt
    /// two hosts no longer share. Held both ways: the entry's words are the
    /// constants, and the composed task carries each of them where the entry
    /// says it does.
    #[test]
    fn the_drafting_task_and_the_dispatch_environment_are_what_the_divergence_record_names() {
        let record = std::fs::read_to_string(
            std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("docs/contract-divergences.md"),
        )
        .expect("the divergence record ships");
        let entry = record
            .split("\n## ")
            .find(|entry| entry.starts_with("69."))
            .expect("the record still carries entry 69");
        let block: serde_json::Value = entry
            .split("```json")
            .nth(1)
            .and_then(|rest| rest.split("```").next())
            .and_then(|block| serde_json::from_str(block).ok())
            .expect("entry 69 carries the json block this test drives");

        let task = &block["drafting_task"];
        for (key, constant) in [
            ("opening", DRAFTING_TASK),
            ("change_request_heading", CHANGE_REQUEST_HEADING),
            ("held_as_draft_line", HELD_AS_DRAFT_LINE),
            ("description_heading", DESCRIPTION_HEADING),
            ("no_description", NO_DESCRIPTION),
            ("worker_transcript_heading", WORKER_TRANSCRIPT_HEADING),
        ] {
            assert_eq!(
                task[key].as_str(),
                Some(constant),
                "entry 69's `{key}` is not the literal this build composes with"
            );
        }
        let environment: Vec<String> =
            serde_json::from_value(block["environment"].clone()).expect("entry 69 names the names");
        assert_eq!(
            environment,
            vec![
                crate::agentgraph::SESSION_ENV.to_string(),
                crate::agentgraph::RUNS_DIR_ENV.to_string()
            ],
            "entry 69 names environment variables this build does not compose"
        );

        // And the composition itself, in the order the entry states: a session
        // holding a change request gets every section, and one holding none
        // gets the opening, the task, and the transcript alone.
        let node = lifecycle(None);
        let held = onevcs::SessionChange {
            url: onevcs::Url::parse("https://github.com/owner/service/pull/7").expect("a url"),
            id: onevcs::ChangeId("7".to_owned()),
            base: "main".to_owned(),
            draft: true,
            title: "wip: the worker's".to_owned(),
            body: String::new(),
        };
        let with = drafting_task(&node, "run-1", Some(&held));
        let without = drafting_task(&node, "run-1", None);
        assert!(with.starts_with(DRAFTING_TASK) && without.starts_with(DRAFTING_TASK));
        let positions: Vec<usize> = [
            DRAFTING_TASK,
            CHANGE_REQUEST_HEADING,
            HELD_AS_DRAFT_LINE,
            DESCRIPTION_HEADING,
            NO_DESCRIPTION,
            WORKER_TRANSCRIPT_HEADING,
        ]
        .iter()
        .map(|literal| {
            with.find(literal)
                .unwrap_or_else(|| panic!("the composed task lacks {literal:?}:\n{with}"))
        })
        .collect();
        assert!(
            positions.windows(2).all(|pair| pair[0] < pair[1]),
            "the sections are out of order:\n{with}"
        );
        assert!(with.contains(&format!(
            "{CHANGE_REQUEST_HEADING}\n{}\n{HELD_AS_DRAFT_LINE} yes",
            held.url
        )));
        assert!(with.contains("`onepipeline transcript run-1 service`"));
        assert!(
            !without.contains(CHANGE_REQUEST_HEADING)
                && without.contains(WORKER_TRANSCRIPT_HEADING),
            "a session holding no change request was shown one:\n{without}"
        );
        // The described body goes in verbatim, and a worker that left one is not
        // told it left none.
        let described = drafting_task(
            &node,
            "run-1",
            Some(&onevcs::SessionChange {
                body: "## What\nHalf written.".to_owned(),
                draft: false,
                ..held
            }),
        );
        assert!(described.contains(&format!("{DESCRIPTION_HEADING}\n## What\nHalf written.")));
        assert!(described.contains(&format!("{HELD_AS_DRAFT_LINE} no")));
        assert!(!described.contains(NO_DESCRIPTION));
    }

    /// A branch level with its base settles on whether this dispatch wrote a
    /// commit, and on the declaration, rather than on the ahead-count alone.
    ///
    /// The three situations `level_branch_settlement` tells apart, driven through
    /// the real function on a session's own reading. The two the engine reaches
    /// end to end — an empty branch nothing declared, and a branch the base took
    /// — are `tests/e2e/lifecycle.rs`'s; the declared one is held here, because
    /// the engine settles a declared node before any dispatch and only a direct
    /// `execute` can put one in front of this arm.
    #[test]
    fn a_level_branch_settles_on_the_commit_and_the_declaration_rather_than_the_count() {
        let level = |wrote: crate::vcs::Wrote| crate::vcs::LevelBranch {
            branch: "work/service".into(),
            base: "origin/main".into(),
            wrote,
        };
        let (nothing, a_commit) = (
            crate::vcs::Wrote::Nothing,
            crate::vcs::Wrote::ACommitTheBaseCarries,
        );
        let settled = |node: &Node, level: &crate::vcs::LevelBranch| match level_branch_settlement(
            node, level, None,
        ) {
            Attempt::Settled(settlement) => *settlement,
            Attempt::Preserving(_) => panic!("a level branch is an answer, not an attempt"),
        };
        let node = lifecycle(None);

        // Nothing declared and nothing committed: the modelling error, named.
        let empty = settled(&node, &level(nothing));
        assert_eq!(empty.status, NodeStatus::Failed);
        assert_eq!(empty.outcome.as_deref(), Some(engine::EMPTY_BRANCH));
        assert_eq!(empty.branch.as_deref(), Some("work/service"));
        let detail = empty.detail.expect("the settlement says why");
        for claim in [
            "compared against origin/main: work/service carries nothing it does not",
            "committed nothing",
            "does not declare `expects_no_diff`",
            "nothing was drafted or published",
            "retry the node with `expects_no_diff: true`",
        ] {
            assert!(detail.contains(claim), "{detail}");
        }

        // The base already carries a commit the dispatch wrote: the second
        // reading of `no-changes`, and a success whatever the node declares.
        for declared in [false, true] {
            let node = Node {
                expects_no_diff: declared,
                ..node.clone()
            };
            let carried = settled(&node, &level(a_commit));
            assert_eq!(carried.status, NodeStatus::Done, "declared: {declared}");
            assert_eq!(carried.outcome.as_deref(), Some(engine::NO_CHANGES));
            let detail = carried
                .detail
                .expect("the settlement says what it compared");
            assert!(detail.contains("compared against origin/main"), "{detail}");
            assert!(
                detail.contains("already carries what this dispatch committed"),
                "{detail}"
            );
        }

        // Declared, and nothing committed: the first reading, as documented.
        let declared = settled(
            &Node {
                expects_no_diff: true,
                ..node
            },
            &level(nothing),
        );
        assert_eq!(declared.status, NodeStatus::Done);
        assert_eq!(declared.outcome.as_deref(), Some(engine::NO_CHANGES));
        assert_eq!(
            declared.detail.as_deref(),
            Some("compared against origin/main: work/service carries nothing it does not")
        );
        // The branch the caller knew wins over the one the read named.
        let named =
            match level_branch_settlement(&lifecycle(None), &level(nothing), Some("kept".into())) {
                Attempt::Settled(settlement) => settlement.branch,
                Attempt::Preserving(_) => unreachable!(),
            };
        assert_eq!(named.as_deref(), Some("kept"));
    }

    /// A workstream refuses before it cuts a branch.
    ///
    /// The step's declaration is one no dispatch can run under, and it is found
    /// while the node is still a document: nothing here opens a session, so the
    /// refusal cannot leave commits on a branch for a node that was never going
    /// to finish. `execute` reports it through the arm a step cycle already
    /// takes.
    #[test]
    fn a_step_whose_budget_no_dispatch_can_run_under_stops_the_workstream() {
        let node = Node {
            id: "service".into(),
            repo: Some("owner/service".into()),
            steps: Some(vec![
                Step {
                    max_turns: Some(45),
                    ..step("implement", &[])
                },
                Step {
                    max_turns: Some(0),
                    ..step("review", &["implement"])
                },
            ]),
            ..Node::default()
        };
        let why = dispatchable_steps(&node)
            .expect_err("a step that can take no turn is not dispatchable");
        assert!(why.contains("node 'service': step 'review':"), "{why}");
        assert!(why.contains("no turn at all"), "{why}");

        // And the workstream itself stops there. The repository is one nothing
        // has registered, so if this refusal came any later the failure would be
        // `onevcs`'s — which is the same as saying a branch would already exist
        // for a node that was never going to finish. The executor is the real
        // one for the same reason: a regression here would go looking for
        // `oneagentgraph` rather than quietly running the step.
        let (tx, rx) = std::sync::mpsc::channel();
        let settlement = execute(
            &crate::executor::LocalExecutor,
            &RunPaths::under(std::path::Path::new("/nowhere"), "demo"),
            &Launch {
                node_graph: "graphs/node-scope.yaml".into(),
                ..Launch::default()
            },
            &node,
            &[],
            &crate::executor::CancellationToken::new(),
            &tx,
        );
        assert_eq!(settlement.status, NodeStatus::Failed);
        assert_eq!(settlement.outcome.as_deref(), Some("invalid-node"));
        let detail = settlement.detail.expect("the settlement says why");
        assert!(detail.contains("step 'review'"), "{detail}");
        assert!(detail.contains("no turn at all"), "{detail}");
        assert_eq!(
            rx.try_iter().count(),
            0,
            "a workstream that could not dispatch a step opened a session anyway"
        );

        // The step that *can* run keeps the budget it declared, narrowed.
        let node = Node {
            steps: Some(vec![Step {
                max_turns: Some(45),
                ..step("implement", &[])
            }]),
            ..node
        };
        let dispatchable = dispatchable_steps(&node).expect("45 is a budget a step can run under");
        assert_eq!(
            dispatchable[0].1.max_turns,
            std::num::NonZeroU32::new(45),
            "the step's own budget did not survive the conversion"
        );
    }

    fn step(id: &str, deps: &[&str]) -> Step {
        Step {
            id: id.into(),
            persona: Some("engineer".into()),
            task: Some("## What\ndo it".into()),
            deps: deps.iter().map(|d| (*d).to_string()).collect(),
            ..Step::default()
        }
    }

    fn lifecycle(steps: Option<Vec<Step>>) -> Node {
        Node {
            id: "service".into(),
            repo: Some("owner/repo".into()),
            persona: steps.is_none().then(|| "engineer".into()),
            task: steps.is_none().then(|| "## What\nship".into()),
            steps,
            ..Node::default()
        }
    }

    #[test]
    fn steps_run_serially_in_topological_order() {
        let node = lifecycle(Some(vec![
            step("publish", &["review"]),
            step("implement", &[]),
            step("review", &["implement"]),
        ]));
        let order: Vec<String> = ordered_steps(&node)
            .expect("the steps order")
            .into_iter()
            .map(|s| s.id)
            .collect();
        assert_eq!(order, vec!["implement", "review", "publish"]);
    }

    #[test]
    fn steps_with_a_cycle_are_reported_rather_than_run_in_some_order() {
        let node = lifecycle(Some(vec![step("a", &["b"]), step("b", &["a"])]));
        let message = ordered_steps(&node).unwrap_err();
        assert!(message.contains("dependency cycle"), "{message}");
    }

    #[test]
    fn a_lifecycle_node_with_no_steps_is_one_implicit_step() {
        let node = lifecycle(None);
        let steps = ordered_steps(&node).expect("one implicit step");
        assert_eq!(steps.len(), 1);
        assert_eq!(steps[0].id, "service");
        assert_eq!(steps[0].persona.as_deref(), Some("engineer"));
    }

    /// The record a follow ended one read short of, relayed exactly once — and
    /// counted against **its own** stream.
    ///
    /// The window this covers is inside a library call now — closing a session
    /// flips its record and only then writes `session-closed`, and the follow
    /// reads and only then asks whether the session closed — so it cannot be
    /// forced from an e2e the way a delayed subprocess once could. This is the
    /// arithmetic that makes "once" true either way, held on its own.
    ///
    /// Two streams rather than one, because reading a session hands back two:
    /// its own records, and the identity's release records that name its
    /// landing, numbered in a series of their own over every session in that
    /// repository. Under one mark over both, a release numbered higher than the
    /// session had reached would hide the session's next record — a relayed
    /// record lost, silently.
    #[test]
    fn relays_only_what_the_follow_did_not_counting_each_stream_on_its_own() {
        let wrote = |stream: &str, seq: u64| Envelope {
            v: crate::event::ENVELOPE_VERSION,
            ts: "2026-01-01T00:00:00.000Z".into(),
            stream: stream.to_owned(),
            seq,
            source: crate::event::Source::Vcs,
            kind: crate::event::EventKind("session-closed".into()),
            phase: None,
            labels: Labels::default(),
            payload: serde_json::Map::new(),
            artifacts: Vec::new(),
        };
        let session: Vec<Envelope> = (1..=4).map(|seq| wrote("s-1", seq)).collect();

        // A follow that reached the third record leaves the tail and nothing
        // else: re-reading the whole stream would put the first three in twice.
        let mut reached = crate::vcs::Watermarks::default();
        for envelope in &session[..3] {
            reached.reached(envelope);
        }
        let tail = beyond(session.clone(), &reached);
        assert_eq!(tail.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![4]);

        // A follow that ended having relayed everything leaves nothing.
        reached.reached(&session[3]);
        assert!(beyond(session.clone(), &reached).is_empty());

        // A release the same read handed back is another stream's record, so a
        // mark four records into the session's says nothing about it.
        let released = wrote("releases-0a1b2c3d4e5f", 2);
        assert_eq!(
            beyond(vec![released.clone()], &reached)
                .iter()
                .map(|e| e.stream.clone())
                .collect::<Vec<_>>(),
            vec!["releases-0a1b2c3d4e5f".to_owned()],
            "a release was hidden by how far the session's own stream had got"
        );
        // And once it has been relayed, the session's tail is still relayable
        // from the mark it stands at rather than from that release's.
        reached.reached(&released);
        assert!(beyond(vec![released], &reached).is_empty());
        assert_eq!(
            beyond(vec![wrote("s-1", 5)], &reached)
                .iter()
                .map(|e| e.seq)
                .collect::<Vec<_>>(),
            vec![5]
        );

        // And a follow that relayed nothing at all leaves the whole stream,
        // which is a stream still to read rather than a stream that held
        // nothing.
        assert_eq!(
            beyond(session, &crate::vcs::Watermarks::default())
                .iter()
                .map(|e| e.seq)
                .collect::<Vec<_>>(),
            vec![1, 2, 3, 4]
        );
    }

    /// A terminator written after the follow ended still reaches the merged
    /// store, and a session that never writes one is **said** rather than left as
    /// a hole in it.
    ///
    /// Driven against the stream **file**, because that is what decides: no
    /// journey can make a real close write its terminator late or not at all. The
    /// sibling is not doubled — the reader under test is its own.
    ///
    /// One test for all three cases: `ONEVCS_HOME` is process-global.
    #[test]
    fn a_session_whose_terminator_arrives_late_still_relays_it_and_one_with_none_says_so() {
        let _home = crate::vcs::scratch_home_held();
        let root =
            std::env::temp_dir().join(format!("onepipeline-terminator-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&root);
        std::fs::create_dir_all(root.join("streams")).expect("a scratch state root");
        std::env::set_var("ONEVCS_HOME", &root);

        let record = |token: &str, seq: u64, kind: &str| {
            serde_json::json!({
                "v": crate::event::ENVELOPE_VERSION,
                "ts": "2026-01-01T00:00:00.000Z",
                "stream": token,
                "seq": seq,
                "source": "vcs",
                "kind": kind,
                "labels": {},
                "payload": {},
                "artifacts": [],
            })
            .to_string()
        };
        let path = |token: &str| root.join("streams").join(format!("{token}.ndjson"));
        let write = |token: &str, body: &str| {
            std::fs::write(path(token), body).expect("the stream is written");
        };
        // What the loop put in the merged store, in the order it put it there —
        // which is where a record relayed twice shows up as well as one lost.
        let relayed = |rx: &std::sync::mpsc::Receiver<Message>| {
            let mut kinds = Vec::new();
            while let Ok(message) = rx.try_recv() {
                if let Message::Event(envelope) = message {
                    kinds.push(envelope.kind.0.clone());
                }
            }
            kinds
        };
        let end = |token: &str, tx: &Sender<Message>| {
            end_session(
                None,
                tx,
                Some(&onevcs::SessionToken(token.to_owned())),
                &Labels::default(),
                None,
            );
        };

        // A stream that already carries its terminator is read once and answered
        // on: the wait below must cost an ordinary settlement nothing.
        let closed = "s-terminated";
        write(
            closed,
            &format!(
                "{}\n{}\n",
                record(closed, 1, "push"),
                record(closed, 2, "session-closed")
            ),
        );
        let (tx, rx) = std::sync::mpsc::channel();
        let began = Instant::now();
        end(closed, &tx);
        assert_eq!(relayed(&rx), vec!["push", "session-closed"]);
        assert!(
            began.elapsed() < TERMINATOR_GRACE,
            "a session that had already ended was waited on anyway"
        );

        // A terminator the first read did not find. Before this it was lost: the
        // one read had been made, and nothing afterwards reads a session's stream.
        let late = "s-latelyclosed";
        let pushed = record(late, 1, "push");
        let ended = record(late, 2, "session-closed");
        write(late, &format!("{pushed}\n"));
        let writing = {
            let path = path(late);
            std::thread::spawn(move || {
                std::thread::sleep(TERMINATOR_POLL * 2);
                std::fs::write(&path, format!("{pushed}\n{ended}\n"))
                    .expect("the terminator is written");
            })
        };
        let (tx, rx) = std::sync::mpsc::channel();
        end(late, &tx);
        writing.join().expect("the writer finishes");
        assert_eq!(
            relayed(&rx),
            vec!["push", "session-closed"],
            "a terminator written after the first read never reached the merged store"
        );

        // And one that never arrives: bounded, so a node cannot hang on its own
        // cleanup, and the records before it relayed exactly once however many
        // reads the wait made.
        let never = "s-neverclosed";
        write(never, &format!("{}\n", record(never, 1, "push")));
        let (tx, rx) = std::sync::mpsc::channel();
        let began = Instant::now();
        end(never, &tx);
        assert!(
            began.elapsed() >= TERMINATOR_GRACE,
            "the wait for a terminator gave up early"
        );
        assert_eq!(
            relayed(&rx),
            vec!["push"],
            "a re-read handed the same record back a second time"
        );

        let _ = std::fs::remove_dir_all(&root);
    }

    #[test]
    fn a_step_ordering_ignores_a_dependency_on_something_outside_the_node() {
        let node = lifecycle(Some(vec![step("only", &["elsewhere"])]));
        let steps = ordered_steps(&node).expect("an outside reference does not deadlock");
        assert_eq!(steps.len(), 1);
    }
}