onepipeline 0.38.1

//! Release adoption: when a node launches relative to its dependencies'
//! *releases* rather than only their branches.
//!
//! A plan node declares an [`Adoption`] mode, and the mode decides one thing:
//! whether a dependency's finished **work** is enough to launch the node, or
//! whether the node waits for the **release** that carries that work.
//!
//! Under `fast` the node launches on branch readiness alone — today's readiness,
//! unchanged — and its dispatch is handed the git references of every dependency
//! that lands outside its own repository, so the worker pins against git rather
//! than against a version that does not exist yet. When those releases arrive it
//! is sent a note naming the versions, into the live turn where the dispatch has
//! a controllable one and onto its next dispatch where it does not.
//!
//! Under `published` the node is not scheduled at all until every one of those
//! dependencies answers released. That wait **blocks indefinitely and never
//! fails a node**: there is no timeout, no deadline, no retry budget, and no
//! automatic degrade to fast adoption. Only an answer of *released* starts it,
//! and "not answered" is never evidence that a release has not happened.
//!
//! `onevcs` records a release **style** per target and this module consumes it.
//! The scheduler's behaviour is identical for both styles — one hold, indefinite,
//! never failing — and what differs is only where the readiness answer comes from
//! and what is reported: an automated target's answer is its probe, which is a
//! subprocess and is therefore paced on its own interval and asked off the
//! reconcile loop's thread, and a human-step target's answer is the
//! acknowledgement record `onevcs release acknowledge` writes, for which this
//! crate runs no probe because there is none to run. An automated landing that
//! captured no baseline is the one place the two meet: no probe answer can
//! release it, and the acknowledgement record does. Nothing here performs a
//! human release step, prompts for one, or acknowledges one on somebody's behalf.

// llmlint: ignore-file[invalid_states_unrepresentable] a node id, a dependency
// reference, and a repository identity are the plain strings the plan schema spells and
// the journal payload carries, for the reason `src/plan.rs` records; and a
// [`Dependency`]'s cells are each `Option` because the whole point of the rendering is
// that a cell the run cannot name is *empty* rather than the row being dropped.

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

use onevcs::releases::{ReleaseStyle, RepositoryReleases, TargetName};
use onevcs::{Adoption, InstructionTemplate, ReleaseStatus};
use serde_json::{json, Value};

use crate::channel::Surface;
use crate::error::Result;
use crate::graph::NodeStatus;
use crate::journal::{self, Journal};
use crate::ledger::RunPaths;
use crate::plan::{CrossRepoReference, Node};
use crate::projection::RunState;

/// The environment variable bounding how often an **automated** target's probe
/// is run.
pub const POLL_ENV: &str = "ONEPIPELINE_RELEASE_POLL_SECONDS";

/// How often an automated target's probe is run when nothing overrides it.
///
/// A probe is a subprocess, so this bounds what a held run costs the host rather
/// than promising latency: a release arriving between two asks is noticed at the
/// second one.
///
/// **A minute, and never longer** — the bound every other answer this loop owes
/// on a clock is held to. A host wanting to spend less lengthens [`POLL_ENV`];
/// the shipped value stays inside the promise.
pub const DEFAULT_POLL_SECONDS: u64 = 60;

/// The environment variable bounding how often a held node's wait is surfaced.
pub const SURFACE_ENV: &str = "ONEPIPELINE_RELEASE_SURFACE_SECONDS";

/// The environment variable holding how long the asker goes on asking a question
/// the loop has **withdrawn**. Zero, and unset, is every build in the field.
///
/// What a value here models is the question that was already in flight when a
/// release arrived: its probe had been started, the loop took up an answer
/// carrying the version and withdrew the question, and the answer of the run
/// already begun landed *after* that. An answer with no version following one
/// that carried it, for the same key, is the whole of what
/// [`Watch::take_up`](Watch::take_up)'s latch is for — and it is the one ordering
/// a journey cannot otherwise schedule from outside, because the asker takes up
/// the loop's newest question set at the top of every iteration, so a key the
/// loop has answered is out of the set before the probe's interval comes round
/// again. `tests/e2e/adoption.rs` sets it to drive that ordering through the
/// compiled binary; nothing else in this crate reads it, and unset it changes
/// neither what is asked nor when.
pub const WITHDRAWN_ASK_ENV: &str = "ONEPIPELINE_RELEASE_WITHDRAWN_ASK_SECONDS";

/// How often a held node's wait is surfaced to the planner when nothing
/// overrides it.
///
/// Much longer than [`DEFAULT_POLL_SECONDS`], and deliberately: asking whether a
/// release has happened is cheap and repeating the question to a person is not.
/// The wait is repeated rather than stated once so it cannot go silent, and a
/// person reading it decides whether to keep waiting, flip the node to fast
/// adoption, or stop the run.
pub const DEFAULT_SURFACE_SECONDS: u64 = 900;

/// The kind a held node's wait is surfaced under.
pub const WAIT_SURFACE_KIND: &str = "release-wait";

/// What one awaited release is tracked by: the node waiting, and the dependency
/// it is waiting on.
type Key = (String, String);

/// One answer, on its way back from the asker to the reconcile loop, and every
/// wait [`questions_of`] put the question on behalf of.
///
/// One message for all of them, so the loop applies them in one go and two nodes
/// awaiting one release are never caught disagreeing about it.
type Answered = (Vec<Key>, Answer);

/// Which rung of the adoption chain one node resolves to.
///
/// Exactly four rungs, in this order and with no fifth:
///
/// 1. the node's own [`adoption`](Node::adoption);
/// 2. the repository rung, and
/// 3. the global rung — both of which are `onevcs`'s and are answered together
///    by [`onevcs::adoption_for`], which falls from the first to the second
///    itself;
/// 4. [`Adoption::Fast`].
///
/// There is deliberately no plan-level tier and no run-only override: the
/// operator specified four rungs, and a fifth changes what they asked for.
///
/// A node with no `repo` has no repository rung. The global rung is not
/// reachable without naming a repository — see `docs/contract-divergences.md`
/// entry 40 — so such a node falls to the floor, which is what the global rung
/// answers on any host that has not configured otherwise.
pub(crate) fn adoption_of(node: &Node) -> Adoption {
    if let Some(declared) = node.adoption {
        return declared;
    }
    if let Some(repo) = node.repo.as_deref() {
        if let Ok(resolved) = onevcs::adoption_for(repo) {
            return resolved;
        }
    }
    Adoption::Fast
}

/// The last thing `onevcs` said about one awaited release.
///
/// Five answers, and **none of them is folded into another**. "Awaiting a human
/// step" is not "not released", which means a probe answered and the version has
/// not moved, and it is not "not answered", which means a probe failed — folding
/// it into the last would report a perfectly healthy wait on a person as a broken
/// probe. And "not answered" is never recorded as "not released" anywhere: not in
/// the scheduler, not in an event payload, and not in a rendering.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum Answer {
    /// A release carrying the dependency's work is out, at this version. The one
    /// answer that releases a hold.
    Released {
        /// The version that carries it.
        version: String,
    },
    /// A probe answered, and the baseline has not been passed.
    NotReleased,
    /// It landed, and nobody has recorded the human step yet.
    AwaitingHumanStep,
    /// The question was not answered. **Never "not released".**
    NotAnswered,
    /// The work has not reached its base, so there is no release to ask about.
    NotLanded,
}

/// How a wait names an awaited release **nothing has answered yet**.
///
/// Deliberately not [`Answer::NotAnswered`], which is a probe this host *ran*
/// and got no usable answer out of and which sends its reader to go and look at
/// that probe. This one is a question still out, which is the state every wait
/// is in before its first probe completes.
///
/// A wait with no question to put at all — no reference the sibling resolves work
/// by, or no target that answers — is neither: nothing will ever answer it, and
/// [`Answer::NotAnswered`] is what that is.
const NO_ANSWER_YET: &str = "no-answer-yet";

impl Answer {
    /// The word an event payload and a rendering name this answer with.
    pub(crate) fn as_str(&self) -> &'static str {
        match self {
            Self::Released { .. } => "released",
            Self::NotReleased => "not-released",
            Self::AwaitingHumanStep => "awaiting-human-step",
            Self::NotAnswered => "not-answered",
            Self::NotLanded => "not-landed",
        }
    }

    /// The version this answer carries, when it is the one that releases a hold.
    fn version(&self) -> Option<&str> {
        match self {
            Self::Released { version } => Some(version),
            _ => None,
        }
    }

    /// What the sibling answered, read as this crate's own vocabulary.
    ///
    /// A refusal — the repository declares no targets, no default target, or the
    /// reference names nothing — is [`NotAnswered`](Self::NotAnswered) and never
    /// anything else: a question that could not be put is not an answer that the
    /// release has not happened.
    fn of(status: &onevcs::Result<ReleaseStatus>) -> Self {
        match status {
            // A version this crate cannot **render** is one it cannot report: it
            // is written into a worker's own note, into a surface, and into an
            // event payload, so one carrying whitespace or a control character
            // forges a line in each. Answered unusably is the sibling's own
            // reading of a probe whose output is not one usable line, and it is
            // this one's of a version that is not one usable word — which holds
            // the node, exactly as every answer but a release does.
            Ok(ReleaseStatus::Released { version, .. }) => match renderable(version) {
                Some(version) => Self::Released { version },
                None => Self::NotAnswered,
            },
            Ok(ReleaseStatus::NotReleased { .. }) => Self::NotReleased,
            Ok(ReleaseStatus::AwaitingHumanStep { .. }) => Self::AwaitingHumanStep,
            Ok(ReleaseStatus::NotAnswered { .. }) | Err(_) => Self::NotAnswered,
            Ok(ReleaseStatus::NotLanded) => Self::NotLanded,
        }
    }
}

/// One dependency of a node whose work lands **outside** that node's repository,
/// in a repository that releases something.
///
/// A dependency inside the same repository is not one of these: the lifecycle
/// already prepares the stacked or merged-stacked branch for it, and nothing
/// here changes that. Neither is one whose repository declares no release
/// targets — there is no release to wait for, which is every repository on a host
/// that has configured none.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct Dependency {
    /// The dependency as the plan names it: a node id, or a cross-DAG
    /// `run:<id>#<node>` reference.
    pub dep: String,
    /// The repository identity its work lands in.
    pub identity: String,
    /// The branch the work is on, where the run recorded one.
    pub branch: Option<String>,
    /// The commit that work reached its base at, where the run observed one.
    pub commit: Option<String>,
    /// Where the work landed, where an operator settling the node from evidence
    /// said it did: the commit it reached its base at, or the change request a
    /// person reads it in.
    ///
    /// A run's own settlement writes a branch and a landing; a `settle` writes
    /// what the run could not see, and this is the half of it that says *where*.
    /// `None` for every node this run watched settle itself, which is most of
    /// them — and for one an operator settled without naming where the work
    /// went, which is exactly the record this run already had.
    pub landing: Option<String>,
    /// The release target this node consumes that repository at.
    pub target: Option<TargetName>,
    /// How that target is released.
    pub style: Option<ReleaseStyle>,
    /// What a person has to do, for a human-step target.
    pub action: Option<String>,
    /// What that repository states about adopting this target, as `onevcs`
    /// resolved it through the three layers.
    ///
    /// Producer knowledge, carried rather than composed here: `None` is a
    /// producer that declares none, whose consumers get this engine's own
    /// default.
    pub instructions: Option<InstructionTemplate>,
}

impl Dependency {
    /// What `onevcs` is asked about.
    ///
    /// The **branch**, because that is the spelling the sibling resolves work by:
    /// a reference is a change request's URL, a session token, a branch a
    /// registered checkout or run clone holds, or a commit one of *those branches*
    /// carries — and a landing commit sitting on the base alone is none of them.
    /// The sibling resolves the branch to the landing itself, which is what a
    /// release is measured against; the commit is what the reference block shows a
    /// worker, and the fallback for work whose branch this run did not record.
    ///
    /// Ahead of both, a [`landing`](Self::landing) an operator stated: a `settle`
    /// is a correction of this run's record, and the branch is the part of that
    /// record it corrects. See `docs/contract-divergences.md` entry 40.
    fn reference(&self) -> Option<&str> {
        self.landing
            .as_deref()
            .or(self.branch.as_deref())
            .or(self.commit.as_deref())
    }

    /// Whether there is a question about this dependency to put at all.
    ///
    /// Both halves are needed to ask one: the reference the sibling resolves the
    /// work by, and the style that says where the answer comes from. A dependency
    /// missing either is one no probe is ever run for, so its wait is not one
    /// still expecting a first answer — it is [`Answer::NotAnswered`], which is
    /// what a question that could not be put is.
    fn askable(&self) -> bool {
        self.reference().is_some() && self.style.is_some()
    }

    /// The row this dependency renders as in the node's task, at whatever version
    /// the run can name — empty where no release has answered yet.
    fn row(&self, version: Option<&str>) -> CrossRepoReference {
        CrossRepoReference {
            dependency: self.dep.clone(),
            repository: self.identity.clone(),
            branch: self.branch.clone().unwrap_or_default(),
            commit: self.commit.clone().unwrap_or_default(),
            release_target: self
                .target
                .as_ref()
                .map(TargetName::to_string)
                .unwrap_or_default(),
            version: version.unwrap_or_default().to_owned(),
            adoption_instructions: self.instructions.clone(),
        }
    }

    /// How this dependency's release is named where one is reported.
    fn named(&self) -> String {
        match &self.target {
            Some(target) => format!("{} {target}", self.identity),
            None => self.identity.clone(),
        }
    }
}

/// What each repository releases, read once per driver.
///
/// Cached because asking is not free — `onevcs` reads the release-targets
/// document and inspects the repository's publication checkout to decide where a
/// script probe could run — and a held node asks the same question on every
/// reconcile pass. Configuration is a fact about the host rather than about the
/// run, so one read per driver is the right number.
///
/// **An answer is cached; a failure is not**, or not for good. What the sibling
/// could not say — a release-targets document half-written, a registry another
/// process was rewriting as it was read — is a fact about that moment, and a
/// driver that pinned it would hold every node depending on that repository for
/// its whole life over a read that would have gone through a second later. So a
/// failed read is kept for one poll interval and then asked again, which is the
/// pace every other question here is put at.
#[derive(Debug)]
struct Repositories {
    known: BTreeMap<String, std::result::Result<RepositoryReleases, UnreadRepository>>,
    retry_every: Duration,
}

#[derive(Debug, Clone)]
struct UnreadRepository {
    reason: String,
    at: Instant,
}

impl Default for Repositories {
    fn default() -> Self {
        Self::retrying_every(Duration::from_secs(poll_seconds()))
    }
}

impl Repositories {
    fn retrying_every(retry_every: Duration) -> Self {
        Self {
            known: BTreeMap::new(),
            retry_every,
        }
    }

    fn of(&mut self, repo: &str) -> std::result::Result<&RepositoryReleases, String> {
        let stale = self.known.get(repo).is_some_and(|known| match known {
            Ok(_) => false,
            Err(unread) => unread.at.elapsed() >= self.retry_every,
        });
        if stale {
            self.known.remove(repo);
        }
        self.known
            .entry(repo.to_owned())
            .or_insert_with(|| {
                onevcs::release_targets(repo).map_err(|failure| UnreadRepository {
                    reason: failure.to_string(),
                    at: Instant::now(),
                })
            })
            .as_ref()
            .map_err(|unread| unread.reason.clone())
    }
}

/// One question the asker puts to `onevcs`, and every wait it answers — see
/// [`questions_of`] for why those are not the same count.
#[derive(Debug, Clone, PartialEq, Eq)]
struct Question {
    /// Every node-and-dependency pair this one answer belongs to.
    keys: Vec<Key>,
    /// What the landed work is named by.
    reference: String,
    /// The target, or `None` for the repository's own default.
    target: Option<TargetName>,
    /// Which of the two styles this is, which decides only how it is paced.
    style: ReleaseStyle,
}

/// The thread that asks `onevcs` whether a release has happened.
///
/// Off the reconcile loop's own thread, and that is the whole reason it exists: an
/// automated target's answer is a **subprocess**, and a slow or hanging probe asked
/// inline would stall the loop every other node in the run depends on. The loop
/// hands over the current question set and reads whatever answers have arrived; it
/// never waits for one.
struct Asker {
    /// The current question set. Dropping this ends the thread.
    questions: Sender<Vec<Question>>,
    /// What has been answered since the loop last looked.
    answers: Receiver<Answered>,
}

/// How long the asker waits for a new question set before re-asking the one it
/// has.
///
/// Short, because a **human-step** answer is a record read rather than a probe
/// and is therefore asked as fast as it completes rather than on the probe's
/// interval; long enough that asking is not a spin. An automated question is
/// skipped on a tick the poll interval has not come due on.
const TICK: Duration = Duration::from_millis(250);

impl Asker {
    /// Start asking, pacing automated questions at `poll`.
    fn start(poll: Duration) -> Self {
        let (questions, asked): (Sender<Vec<Question>>, Receiver<Vec<Question>>) = mpsc::channel();
        let (answered, answers): (Sender<Answered>, Receiver<Answered>) = mpsc::channel();
        // Detached: it holds no run state, writes nothing, and ends when the
        // reconcile loop drops its end of `questions`.
        // llmlint: ignore-block[changed_behavior_has_e2e] no invocation a user can type
        // reaches this arm: it is a host that will not start a thread at all, which no
        // plan, flag, or environment of this crate's decides. What it does when it is
        // reached is the safe direction of every answer staying unarrived, which is the
        // hold a published node is under before its first probe answers and is driven end
        // to end by `tests/e2e/adoption.rs`.
        std::thread::Builder::new()
            .name("release-asker".to_owned())
            .spawn(move || ask_until_dropped(&asked, &answered, poll))
            .map(drop)
            .unwrap_or_else(|error| {
                // A thread this host would not start is reported and nothing
                // more: every answer then stays unarrived, which holds a
                // published node exactly as an unanswered probe does and leaves
                // a fast node with the git pin it launched under. Neither is a
                // node failed for a reason that is not about the node.
                eprintln!("onepipeline: cannot start the release watch: {error}");
            }); // llmlint: ignore-end[changed_behavior_has_e2e]
        Self { questions, answers }
    }

    /// Hand over the questions to ask from now on.
    fn ask(&self, questions: Vec<Question>) {
        // A dead asker is not an error here for the reason its refusal to start
        // is not: it leaves every answer unarrived, which is the safe direction.
        let _ = self.questions.send(questions);
    }

    /// Everything answered since the last look. Never blocks.
    fn answered(&self) -> Vec<Answered> {
        self.answers.try_iter().collect()
    }
}

/// The asker's own loop.
fn ask_until_dropped(asked: &Receiver<Vec<Question>>, answered: &Sender<Answered>, poll: Duration) {
    // Zero everywhere but a journey that needs the one ordering it cannot
    // otherwise schedule; see [`WITHDRAWN_ASK_ENV`]. Read once, because the
    // environment a thread was started under is what it runs under.
    let linger = withdrawn_ask();
    let mut questions: Vec<Question> = Vec::new();
    // Questions the loop has withdrawn and that are still being asked, with when
    // each was withdrawn. Always empty unless `linger` is set.
    let mut withdrawn: Vec<(Question, Instant)> = Vec::new();
    let mut probed: Option<Instant> = None;
    loop {
        match asked.recv_timeout(TICK) {
            Ok(fresh) => questions = retire(&questions, fresh, linger, &mut withdrawn),
            Err(RecvTimeoutError::Timeout) => {}
            // The reconcile loop has gone, so there is nobody to answer.
            Err(RecvTimeoutError::Disconnected) => return,
        }
        // Every question set the loop queued while this one was being asked, so
        // a slow probe leaves the asker working from the newest set rather than
        // from a backlog of stale ones.
        while let Ok(fresh) = asked.try_recv() {
            questions = retire(&questions, fresh, linger, &mut withdrawn);
        }
        withdrawn.retain(|(_, since)| since.elapsed() < linger);
        let due = probed.is_none_or(|last| last.elapsed() >= poll);
        let mut ran_a_probe = false;
        for question in questions
            .iter()
            .chain(withdrawn.iter().map(|(held, _)| held))
        {
            if question.style == ReleaseStyle::Automated {
                if !due {
                    continue;
                }
                ran_a_probe = true;
            }
            // The one call, for both styles. An automated target is answered by
            // running its probe under the probe's own timeout; a human-step
            // target executes nothing at all and is answered from the
            // acknowledgement record. There is no spelling of this that could
            // start a subprocess for a human-step target, because the probe
            // lives on the other variant.
            crate::loopstats::release_asked();
            let answer = Answer::of(&onevcs::release_status(
                &question.reference,
                question.target.as_ref(),
            ));
            if answered.send((question.keys.clone(), answer)).is_err() {
                return;
            }
        }
        if ran_a_probe {
            probed = Some(Instant::now());
        }
    }
}

/// The releases this run is waiting on, and what it has already said about them.
pub(crate) struct Watch {
    repositories: Repositories,
    /// The out-of-repository dependencies of each node, once the run could name
    /// them.
    ///
    /// Frozen per node the first time every one of its dependencies resolves,
    /// because by then each has settled `done` and a settled node's repository,
    /// branch, and landing commit do not move: a `retry` replaces the node
    /// under a new id, and a `requeue` continues the branch this already names.
    dependencies: BTreeMap<String, Vec<Dependency>>,
    /// The last answer about each awaited release.
    answers: BTreeMap<Key, Answer>,
    /// When each wait was first observed, in epoch milliseconds.
    since: BTreeMap<Key, u64>,
    /// The nodes an arrival note has already reached, seeded from the journal so
    /// a fresh driver does not deliver one twice.
    adopted: BTreeSet<String>,
    /// The awaited releases already reported as arrived, seeded the same way.
    arrived: BTreeSet<Key>,
    /// When each held node's wait was last surfaced.
    surfaced: BTreeMap<String, Instant>,
    /// How often a held node's wait is surfaced.
    surface_every: Duration,
    /// How often the identity's release records are read for what they say
    /// about this run's own landed work.
    ///
    /// The **probe's** interval, because it is the same question asked from the
    /// other end: a release nobody has asked about yet has nothing recorded
    /// about it either, so reading more often than the run asks would read the
    /// same file again for an answer nothing has written.
    relay_every: Duration,
    /// When the release records were last read. `None` before the first read,
    /// which is due immediately.
    relayed: Option<Instant>,
    /// Where each node's work landed, as an operator settling it from evidence
    /// stated it, keyed by the run whose journal it was read from.
    ///
    /// Read from a **journal** rather than taken off the folded state, because
    /// this run's fold is not the only one it is asked about: a cross-DAG
    /// dependency's landing was stated on the upstream run's journal, and both
    /// are read here through one function.
    stated: BTreeMap<String, BTreeMap<String, String>>,
    /// When a node's landings were last re-read. `None` before the first read,
    /// which is due immediately.
    read_landings: Option<Instant>,
    /// The nodes the last refresh watched whose dependencies it could not all
    /// describe, and which of them it could not.
    ///
    /// **The marker a hold reads.** A node is in here or in `dependencies` and
    /// never both: its set is frozen the first pass every dependency resolves,
    /// and until then this says which are still unreadable and why. An entry
    /// here holds a `published` node exactly as an unreleased dependency does,
    /// because the two are told apart only by what a later pass can read — and a
    /// node read as having *no* out-of-repository dependency because one could
    /// not be described is a node launched against work no release carries.
    unresolved: BTreeMap<String, Vec<Unresolved>>,
    asker: Asker,
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct Unresolved {
    dep: String,
    reason: String,
    /// When it was first found unreadable, in epoch milliseconds.
    since: u64,
}

/// The word a wait names a dependency the run could not describe by.
///
/// Beside the five answers and [`NO_ANSWER_YET`], and not one of them: no
/// question was put, because there was nothing yet to put one about.
const UNRESOLVED: &str = "unresolved";

impl Watch {
    /// Start watching one run, taking up whatever a previous driver said.
    ///
    /// The journal is external input, and what it seeds here **suppresses** a
    /// delivery — so a record that does not name both halves of what it claims
    /// was said is skipped rather than read as something. A record naming a node
    /// or a dependency this graph does not have suppresses nothing, because
    /// nothing is ever looked up under it; what a lax reading would cost is a
    /// record with an *empty* node or dep, which is a key a real one could
    /// collide with.
    pub(crate) fn of_run(paths: &RunPaths) -> Self {
        let mut adopted = BTreeSet::new();
        let mut arrived = BTreeSet::new();
        for event in journal::read(&paths.journal()) {
            let Some(node) = event.labels.node.clone().and_then(|node| renderable(&node)) else {
                continue;
            };
            match journal::PipelineKind::from_wire(&event.kind) {
                Some(journal::PipelineKind::ReleaseAdopted) => {
                    // Usable only if it names a release this build can read: a
                    // record whose versions are gone is one nothing can say what
                    // the node was told, so suppressing on it would leave a node
                    // never told anything and never told again.
                    if !Released::of_payload(event.payload.get("versions").unwrap_or(&Value::Null))
                        .is_empty()
                    {
                        adopted.insert(node);
                    }
                }
                Some(journal::PipelineKind::ReleaseArrived) => {
                    if let Some(dep) = event
                        .payload
                        .get("dep")
                        .and_then(Value::as_str)
                        .and_then(renderable)
                    {
                        arrived.insert((node, dep));
                    }
                }
                _ => {}
            }
        }
        Self {
            repositories: Repositories::default(),
            dependencies: BTreeMap::new(),
            answers: BTreeMap::new(),
            since: BTreeMap::new(),
            adopted,
            arrived,
            surfaced: BTreeMap::new(),
            unresolved: BTreeMap::new(),
            surface_every: Duration::from_secs(surface_every_seconds()),
            relay_every: Duration::from_secs(poll_seconds()),
            relayed: None,
            stated: BTreeMap::new(),
            read_landings: None,
            asker: Asker::start(Duration::from_secs(poll_seconds())),
        }
    }

    /// The rows a node's dispatch is handed, in the order its `deps` name them.
    ///
    /// **Both adoption modes**, because the block serves both. A fast-adoption
    /// node meets it as the git references it pins against, with the version cell
    /// empty because that is its whole condition; a `published` node — which was
    /// not started until every one of these answered released — meets it as the
    /// versions it is building against, and this is the **only** place that node
    /// ever sees one, since nothing sends it an arrival note it never needed.
    /// Which of the two the block says of itself is decided by the rows.
    pub(crate) fn references(&self, node: &Node) -> Vec<CrossRepoReference> {
        self.dependencies
            .get(&node.id)
            .map(|dependencies| {
                dependencies
                    .iter()
                    .map(|dependency| {
                        dependency.row(
                            self.answers
                                .get(&(node.id.clone(), dependency.dep.clone()))
                                .and_then(Answer::version),
                        )
                    })
                    .collect()
            })
            .unwrap_or_default()
    }

    /// How long the loop may go without taking this watch up.
    ///
    /// An arriving *answer* does not wait for it —
    /// [`take_up_answers`](Self::take_up_answers) is what the loop waits on, and
    /// it wakes within a fifth of a second of the asker answering. What this
    /// paces is the work that is due on a clock rather than on an answer:
    /// re-surfacing a wait that has not ended, and re-reading what the sibling
    /// recorded about this run's own landed work. So it is the shorter of those
    /// two intervals, held to a minute — which is the bound this loop promises
    /// for an arriving release, and the ceiling a host that configures neither
    /// falls back to.
    pub(crate) fn take_up_every(&self) -> Duration {
        self.surface_every
            .min(self.relay_every)
            .min(Duration::from_secs(60))
    }

    /// Take up whatever the asker has answered, and say whether it answered
    /// anything.
    ///
    /// Named for the taking up rather than for the question it answers: the
    /// asker's queue is drained here and the answers are stored, so a caller
    /// reading this as a look at the run's state would be the caller that lost
    /// them.
    ///
    /// What the reconcile loop waits on. The asker runs on its own thread at its
    /// own pace, so an answer arrives without anything about this run changing —
    /// and this is how it wakes the loop, rather than the loop going back and
    /// asking on a timer of its own.
    pub(crate) fn take_up_answers(&mut self) -> bool {
        let mut arrived = false;
        for (keys, answer) in self.asker.answered() {
            arrived = true;
            self.take_up(&keys, &answer);
        }
        arrived
    }

    /// Record what one ask answered, about every wait it was put on behalf of.
    ///
    /// **A release does not un-happen**, so the version latches: every other
    /// answer is a statement about *now* — a probe that failed, a target awaiting
    /// a person, a version that has not moved — and any of them written over one
    /// that carried a version un-releases a hold the run has already acted on.
    ///
    /// The **first** version, not the newest: what the run reported and what the
    /// node was told is the release that ended its wait.
    fn take_up(&mut self, keys: &[Key], answer: &Answer) {
        for key in keys {
            if self.answers.get(key).and_then(Answer::version).is_some() {
                continue;
            }
            self.answers.insert(key.clone(), answer.clone());
        }
    }

    /// Whether any node in this run names a dependency outside its own
    /// repository, which is the cheap half of asking whether one is waiting.
    ///
    /// What it decides is whether the loop wakes for a release at all, so it is
    /// deliberately the wider question: a run whose nodes name no such dependency
    /// has nothing to ask about and nothing to take up, and never pays the
    /// interval above — which is most runs. A run that names one goes on waking
    /// for it after the answer has arrived, because the answer this holds is a
    /// cache of a thing another process writes.
    pub(crate) fn names_a_release_dependency(&self) -> bool {
        !self.unresolved.is_empty() || self.dependencies.values().any(|of| !of.is_empty())
    }

    /// Whether anything this run landed could have a release recorded against it.
    ///
    /// The same question [`relay_releases`](Self::relay_releases) asks per node,
    /// asked of the run: a repository declaring no release target has nothing
    /// recorded about it, so there is nothing to go and read. Answered off the
    /// per-driver cache, so asking again costs nothing.
    pub(crate) fn relays_anything(&mut self, state: &RunState) -> bool {
        let repositories: Vec<String> = state
            .sessions
            .keys()
            .filter_map(|node| state.graph.get(node).and_then(|node| node.repo.clone()))
            .collect();
        repositories.into_iter().any(|repo| {
            self.repositories
                .of(&repo)
                .is_ok_and(|releases| !releases.targets.is_empty())
        })
    }

    /// The dependencies one node is still waiting on the release of, by id.
    ///
    /// The ids alone. What each of those waits *is* — its identity, its target,
    /// its style, how long it has been on — is
    /// [`ReleaseWait`](journal::PipelineKind::ReleaseWait)'s account of it and is
    /// not copied anywhere else.
    ///
    /// A dependency the run could not describe is among them: what holds the
    /// node is the release it may be waiting on, and that it could not yet be
    /// asked about is the wait's account of itself rather than a reason to name
    /// nothing.
    pub(crate) fn awaited_deps(&self, node: &str) -> Vec<String> {
        self.dependencies
            .get(node)
            .map(Vec::as_slice)
            .unwrap_or_default()
            .iter()
            .filter(|dependency| {
                self.answers
                    .get(&(node.to_owned(), dependency.dep.clone()))
                    .and_then(Answer::version)
                    .is_none()
            })
            .map(|dependency| dependency.dep.clone())
            .chain(self.unresolved_of(node).iter().map(|it| it.dep.clone()))
            .collect()
    }

    fn unresolved_of(&self, node: &str) -> &[Unresolved] {
        self.unresolved
            .get(node)
            .map(Vec::as_slice)
            .unwrap_or_default()
    }

    /// Take up the answers that have arrived, and ask about what is awaited now.
    ///
    /// `watching` is the nodes whose releases matter this pass: every node that
    /// is ready to start, and every fast-adoption node still running. Neither
    /// blocks on anything.
    pub(crate) fn refresh(&mut self, paths: &RunPaths, state: &RunState, watching: &[Node]) {
        self.take_up_answers();
        let now = crate::sys::now_millis();
        let re_read = self.landings_are_due();
        let mut waits: Vec<(Key, Dependency)> = Vec::new();
        for node in watching {
            for dependency in self.resolve(paths, state, node, re_read) {
                let key = (node.id.clone(), dependency.dep.clone());
                // A release that has arrived is not waited on any more: no
                // question is put about it again, and the clock the wait was
                // measured on is dropped rather than left running. Both are the
                // same fact — this wait is over — and leaving either behind is
                // what let a satisfied hold be reported as an hour-long one.
                if self.answers.get(&key).and_then(Answer::version).is_some() {
                    self.since.remove(&key);
                    continue;
                }
                self.since.entry(key.clone()).or_insert(now);
                waits.push((key, dependency));
            }
        }
        // A node whose dependencies this run could not describe yet is answered
        // again rather than left: what it is waiting on may be another run's
        // ledger or this host's own release-targets document, both of which move
        // without anything here changing, so the loop has to keep coming back for
        // it. `resolve` marked every watched node this pass could not describe;
        // one no longer watched is not waiting on anything here.
        self.unresolved
            .retain(|node, _| watching.iter().any(|watched| watched.id == *node));
        self.asker.ask(questions_of(&waits));
    }

    /// Whether every release one node awaits has arrived.
    ///
    /// `false` where it awaits none, so a node with no out-of-repository
    /// dependency is never reported as having adopted anything.
    fn all_released(&self, node: &str) -> bool {
        let dependencies = self.dependencies.get(node).filter(|of| !of.is_empty());
        dependencies.is_some_and(|dependencies| {
            dependencies.iter().all(|dependency| {
                self.answers
                    .get(&(node.to_owned(), dependency.dep.clone()))
                    .and_then(Answer::version)
                    .is_some()
            })
        })
    }

    /// The nodes a release hold will not let start yet.
    ///
    /// A `published` node whose out-of-repository dependencies have not all
    /// answered released — **or could not all be described**, which is held on
    /// the same footing: "not resolved this pass" says nothing about whether a
    /// release carries the work, and a node this cannot yet be asked for is not
    /// a node with nothing to wait on. Nothing else holds anything: a `fast` node
    /// launches on branch readiness alone, and a node whose every dependency
    /// resolved to nothing outside its repository is never held.
    pub(crate) fn held(&self, watching: &[Node]) -> BTreeSet<String> {
        watching
            .iter()
            .filter(|node| adoption_of(node) == Adoption::Published)
            .filter(|node| {
                self.unresolved.contains_key(&node.id) || self.awaits_a_release(&node.id)
            })
            .map(|node| node.id.clone())
            .collect()
    }

    fn awaits_a_release(&self, node: &str) -> bool {
        self.dependencies
            .get(node)
            .is_some_and(|dependencies| !dependencies.is_empty())
            && !self.all_released(node)
    }

    /// Report every release that has arrived, and every wait that is still on.
    ///
    /// The wait is surfaced when it begins and again on its own interval, so it
    /// cannot go silent; the arrival of one release is reported once.
    pub(crate) fn report(
        &mut self,
        paths: &RunPaths,
        journal: &mut Journal,
        held: &BTreeSet<String>,
        watching: &[Node],
    ) -> Result<()> {
        for node in watching {
            let dependencies = self.dependencies.get(&node.id).cloned().unwrap_or_default();
            for dependency in &dependencies {
                let key = (node.id.clone(), dependency.dep.clone());
                let Some(version) = self.answers.get(&key).and_then(Answer::version) else {
                    continue;
                };
                if !self.arrived.insert(key) {
                    continue;
                }
                journal.emit(
                    journal::PipelineKind::ReleaseArrived,
                    journal::labels(&paths.run, Some(&node.id)),
                    journal::payload(&[
                        ("node", json!(node.id)),
                        ("dep", json!(dependency.dep)),
                        ("identity", json!(dependency.identity)),
                        (
                            "target",
                            json!(dependency.target.as_ref().map(ToString::to_string)),
                        ),
                        ("style", json!(dependency.style.map(|style| style.as_str()))),
                        ("version", json!(version)),
                    ]),
                )?;
            }
        }
        for node in held {
            let due = self
                .surfaced
                .get(node)
                .is_none_or(|last| last.elapsed() >= self.surface_every);
            if !due {
                continue;
            }
            self.surfaced.insert(node.clone(), Instant::now());
            // The surface first and the record second. They are two appends
            // saying one thing, so a reader holding the record and reading the
            // surface beside it gets whichever surface was there when it looked:
            // raised second, that is the previous one, carrying the previous
            // answer about a probe that has since stopped answering. This way
            // round the surface is never older than the record beside it.
            crate::engine::raise(paths, journal, self.wait_surface(node))?;
            let awaiting = self.awaiting(node);
            journal.emit(
                journal::PipelineKind::ReleaseWait,
                journal::labels(&paths.run, Some(node)),
                journal::payload(&[("node", json!(node)), ("awaiting", json!(awaiting))]),
            )?;
        }
        // A node that is no longer held says nothing more: the arrival is
        // reported by `release-arrived`, and repeating the wait after it ended
        // would report a run as waiting on something it has.
        self.surfaced.retain(|node, _| held.contains(node));
        Ok(())
    }

    /// Relay what the sibling recorded about the releases carrying this run's
    /// own landed work.
    ///
    /// A release happens long after the dispatch that produced the work has
    /// ended, outside every session — so `onevcs` records it on the repository's
    /// own release record and joins it back to the session whose **landing
    /// commit** it names. That join is the sibling's, and the address of that
    /// record is the sibling's too: this asks only for the session it already
    /// knows, through the same phase-aware reader every followed session is read
    /// through, and takes back whatever that reader says belongs to it.
    ///
    /// Only the [`Release`](crate::event::Phase::Release) phase, and only past
    /// the mark each stream already stands at in this run's own store. The
    /// session's own records were relayed by the follow that watched them, and
    /// `release-probed` — which the publication wrote on the session's stream
    /// while that follow was reading it — is among them: reporting one ask as
    /// two is the same defect as losing it.
    ///
    /// The marks are the store's own, so what stops a record arriving twice is
    /// the record.
    // llmlint: ignore-block[changed_behavior_has_e2e] the two failures this reaches are
    // not decided here: an unreadable stream is [`crate::vcs::events`]'s own answer and
    // an unwritable journal is [`Journal::relay`]'s, each held where it lives.
    pub(crate) fn relay_releases(
        &mut self,
        paths: &RunPaths,
        journal: &mut Journal,
        state: &RunState,
        statuses: &BTreeMap<String, NodeStatus>,
        filter: Option<&crate::filter::EventFilter>,
    ) -> Result<()> {
        if !self
            .relayed
            .is_none_or(|last| last.elapsed() >= self.relay_every)
        {
            return Ok(());
        }
        self.relayed = Some(Instant::now());
        let mut relayed = crate::vcs::Watermarks::of_relayed(&journal::read(&paths.journal()));
        for (node, session) in &state.sessions {
            // A dispatch still in flight has a follow of its own reading that
            // session as it is written, and two readers of one stream deciding
            // separately what the store already holds is how a record arrives
            // twice. Its releases are read once it has settled, which is the
            // earliest its work can have landed anyway.
            if statuses.get(node) == Some(&NodeStatus::Running) {
                continue;
            }
            let releases_nothing = state
                .graph
                .get(node)
                .and_then(|node| node.repo.clone())
                .and_then(|repo| self.repositories.of(&repo).ok())
                .is_none_or(|releases| releases.targets.is_empty());
            // A repository that declares no release targets releases nothing, so
            // there is nothing recorded about it to read — which is every host
            // that has configured none, and is what keeps such a run's store
            // exactly the store it held before there was a release record at all.
            if releases_nothing {
                continue;
            }
            let known = crate::engine::dispatch_labels(
                &paths.run,
                node,
                None,
                state
                    .graph
                    .get(node)
                    .and_then(|node| node.persona.as_deref()),
            );
            for mut envelope in crate::vcs::events(session.token(), filter) {
                if envelope.dimensions.phase != Some(crate::event::Phase::Release) {
                    continue;
                }
                if !relayed.beyond(&envelope) {
                    continue;
                }
                // An enricher, never a rewriter: the producer could not know
                // which node's work this release carries, and everything it did
                // stamp stands.
                crate::lifecycle::stamp(&mut envelope.labels, &known);
                journal.relay(&envelope)?;
                relayed.reached(&envelope);
            }
        }
        Ok(())
    } // llmlint: ignore-end[changed_behavior_has_e2e]

    /// The word a payload and a surface name one awaited release's last answer
    /// by.
    ///
    /// Three states and not two, because a reader acts differently on each: the
    /// answer this run has, a question still out ([`NO_ANSWER_YET`]), and a
    /// question that could never be put — no reference to ask about or no target
    /// that answers — which is [`Answer::NotAnswered`] and never
    /// [`Answer::NotReleased`].
    fn last_answer(&self, key: &Key, dependency: &Dependency) -> &'static str {
        match self.answers.get(key) {
            Some(answer) => answer.as_str(),
            None if dependency.askable() => NO_ANSWER_YET,
            None => Answer::NotAnswered.as_str(),
        }
    }

    /// The `awaiting` list one held node's wait carries.
    fn awaiting(&self, node: &str) -> Vec<Value> {
        let now = crate::sys::now_millis();
        self.dependencies
            .get(node)
            .map(Vec::as_slice)
            .unwrap_or_default()
            .iter()
            .filter(|dependency| {
                self.answers
                    .get(&(node.to_owned(), dependency.dep.clone()))
                    .and_then(Answer::version)
                    .is_none()
            })
            .map(|dependency| {
                let key = (node.to_owned(), dependency.dep.clone());
                let since = self.since.get(&key).copied().unwrap_or(now);
                let mut entry = journal::payload(&[
                    ("dep", json!(dependency.dep)),
                    ("identity", json!(dependency.identity)),
                    (
                        "target",
                        json!(dependency.target.as_ref().map(ToString::to_string)),
                    ),
                    ("style", json!(dependency.style.map(|style| style.as_str()))),
                ]);
                // Only a human-step wait carries the action: it is the text a
                // person needs, and an automated wait has nobody to hand it to.
                if dependency.style == Some(ReleaseStyle::HumanStep) {
                    entry.insert("action".to_owned(), json!(dependency.action));
                }
                entry.insert(
                    "since".to_owned(),
                    json!(crate::sys::rfc3339_from_millis(since)),
                );
                entry.insert(
                    "waited_seconds".to_owned(),
                    json!(now.saturating_sub(since) / 1_000),
                );
                entry.insert(
                    "last_answer".to_owned(),
                    json!(self.last_answer(&key, dependency)),
                );
                Value::Object(entry)
            })
            .chain(self.unresolved_of(node).iter().map(|unresolved| {
                // No identity, target or style: those are what could not be
                // read. The dependency, how long, and why are what there is.
                let entry = journal::payload(&[
                    ("dep", json!(unresolved.dep)),
                    ("identity", Value::Null),
                    ("target", Value::Null),
                    ("style", Value::Null),
                    (
                        "since",
                        json!(crate::sys::rfc3339_from_millis(unresolved.since)),
                    ),
                    (
                        "waited_seconds",
                        json!(now.saturating_sub(unresolved.since) / 1_000),
                    ),
                    ("last_answer", json!(UNRESOLVED)),
                    ("reason", json!(unresolved.reason)),
                ]);
                Value::Object(entry)
            }))
            .collect()
    }

    /// The surface a held node's wait raises.
    ///
    /// Non-blocking: the hold is the scheduler's, and a blocking surface would
    /// hold the same subtree twice while reading, in every planner view, as a
    /// decision somebody has to answer before the run can move. This one is a
    /// report — the decision it informs is whether to go on waiting at all.
    fn wait_surface(&self, node: &str) -> Surface {
        let now = crate::sys::now_millis();
        let mut lines: Vec<String> = Vec::new();
        for dependency in self
            .dependencies
            .get(node)
            .map(Vec::as_slice)
            .unwrap_or(&[])
        {
            let key = (node.to_owned(), dependency.dep.clone());
            if self.answers.get(&key).and_then(Answer::version).is_some() {
                continue;
            }
            let waited = crate::telemetry::duration(
                now.saturating_sub(self.since.get(&key).copied().unwrap_or(now)),
            );
            let answered = self.last_answer(&key, dependency);
            // The style is named in the sentence itself, so an automated wait
            // and a wait on a person are tellable apart from this text alone —
            // without the reader opening the release-targets file to find out
            // which kind of wait they are looking at.
            let style = match (dependency.style, dependency.action.as_deref()) {
                (Some(ReleaseStyle::HumanStep), Some(action)) => {
                    format!("human-step release — a person has to: {action}")
                }
                (Some(ReleaseStyle::HumanStep), None) => "human-step release".to_owned(),
                (Some(ReleaseStyle::Automated), _) => "automated release".to_owned(),
                (None, _) => "no release target this host can name".to_owned(),
            };
            lines.push(format!(
                "- {named} — {style}, waited {waited}, last answer: {answered}",
                named = dependency.named(),
            ));
        }
        // A dependency the run could not describe is named by the id the plan
        // gives it, which is all the run has for it, and the line says why: the
        // hold is on the reading, and the reader is who can fix the reading.
        for unresolved in self.unresolved_of(node) {
            let waited = crate::telemetry::duration(now.saturating_sub(unresolved.since));
            lines.push(format!(
                "- {dep} — not yet resolved ({reason}), waited {waited}, last answer: {UNRESOLVED}",
                dep = unresolved.dep,
                reason = unresolved.reason,
            ));
        }
        Surface {
            id: 0,
            kind: WAIT_SURFACE_KIND.to_owned(),
            message: format!(
                "node '{node}' is held under published adoption, waiting on {count} \
                 release(s):\n{lines}\nNothing times this out and nothing will fail the node. \
                 Keep waiting, flip this node to `adoption: fast` by live edit, or stop the run.",
                count = lines.len(),
                lines = lines.join("\n"),
            ),
            source: crate::channel::source::PROPOSAL.to_owned(),
            blocking: false,
            queued_at: now,
            abandoned: false,
            asker: None,
            workstream: Some(node.to_owned()),
            correlation: None,
        }
    }

    /// The nodes whose awaited releases have all arrived and which have not been
    /// told yet, with the versions to tell them.
    ///
    /// **Fast adoption only**, for the reason [`references`](Self::references)
    /// gives: a `published` node launched against those versions in the first
    /// place, so a note telling it to move off a git pin it never held is noise
    /// aimed at a worker who cannot act on it.
    ///
    /// `told` is every node an arrival is owed to: the ones still running, whose
    /// live turn it reaches, **and** the ones whose change is held as a draft,
    /// which is where it does more than inform — the note is what puts a worker
    /// back on that branch to move the pin, and lifting the draft is what that
    /// worker's own publication then does.
    pub(crate) fn ready_to_adopt(&self, told: &[Node]) -> Vec<(String, Vec<Released>)> {
        told.iter()
            .filter(|node| adoption_of(node) == Adoption::Fast)
            .filter(|node| !self.adopted.contains(&node.id))
            .filter(|node| self.all_released(&node.id))
            .map(|node| (node.id.clone(), self.released(&node.id)))
            .collect()
    }

    /// The versions one node's awaited releases arrived at.
    fn released(&self, node: &str) -> Vec<Released> {
        self.dependencies
            .get(node)
            .map(Vec::as_slice)
            .unwrap_or_default()
            .iter()
            .filter_map(|dependency| {
                Some(Released {
                    dep: dependency.dep.clone(),
                    identity: dependency.identity.clone(),
                    branch: dependency.branch.clone().unwrap_or_default(),
                    commit: dependency.commit.clone().unwrap_or_default(),
                    target: dependency
                        .target
                        .as_ref()
                        .map(ToString::to_string)
                        .unwrap_or_default(),
                    version: self
                        .answers
                        .get(&(node.to_owned(), dependency.dep.clone()))
                        .and_then(Answer::version)?
                        .to_owned(),
                    instructions: dependency.instructions.clone(),
                })
            })
            .collect()
    }

    /// Record that one node has been told, so it is told exactly once.
    pub(crate) fn adopted(&mut self, node: &str) {
        self.adopted.insert(node.to_owned());
    }

    /// Whether a landing re-read is due, and take the tick if it is.
    ///
    /// One tick for the whole pass rather than one per node, so a run holding
    /// several nodes re-reads its journal once and they all see the same answer.
    /// Paced by the probe's own interval, for the reason
    /// [`relay_releases`](Self::relay_releases) is: a landing is stated by a
    /// person over the channel, and asking oftener than the run puts its release
    /// question re-reads a file for an answer that could not have changed what is
    /// asked.
    fn landings_are_due(&mut self) -> bool {
        if !self
            .read_landings
            .is_none_or(|last| last.elapsed() >= self.relay_every)
        {
            return false;
        }
        self.read_landings = Some(Instant::now());
        self.stated.clear();
        true
    }

    /// Where one node's work landed, as an operator stated it on that run's own
    /// journal.
    ///
    /// Read once per run per re-read tick and kept for the rest of it, because a
    /// run's journal answers this for every node in it.
    fn stated_landing(&mut self, paths: &RunPaths, node: &str) -> Option<String> {
        if !self.stated.contains_key(&paths.run) {
            self.stated.insert(
                paths.run.clone(),
                stated_landings(&journal::read(&paths.journal())),
            );
        }
        self.stated.get(&paths.run)?.get(node).cloned()
    }

    /// One node's out-of-repository dependencies, resolved once — and their
    /// landings re-read on every re-read tick for as long as the node is watched.
    ///
    /// **The set is frozen and the landing is not.** Which dependencies land
    /// outside the node's repository, and what those repositories release, is
    /// settled by the time it is asked. *Where the work went* is not: a landing
    /// this run observed is relayed before the node settles, but one nobody
    /// observed — a change request merged after its node settled, stated by an
    /// operator from evidence — arrives after the fact by definition, and frozen
    /// at settlement it answers `not-landed` for ever. Nor is a stated landing
    /// final once there is one: an operator who stated a squash-merge commit
    /// `onevcs` cannot resolve corrects it by stating the change request, and a
    /// landing frozen at the first statement would ask the unanswerable one for
    /// ever.
    fn resolve(
        &mut self,
        paths: &RunPaths,
        state: &RunState,
        node: &Node,
        re_read: bool,
    ) -> Vec<Dependency> {
        if let Some(known) = self.dependencies.get(&node.id) {
            let known = known.clone();
            if !re_read {
                return known;
            }
            let re_read: Vec<Dependency> = known
                .into_iter()
                .map(|mut dependency| {
                    dependency.landing = self.landing_of(paths, &dependency.dep);
                    dependency
                })
                .collect();
            self.dependencies.insert(node.id.clone(), re_read.clone());
            return re_read;
        }
        let mine = identity_of(&mut self.repositories, node);
        let now = crate::sys::now_millis();
        let mut resolved: Vec<Dependency> = Vec::new();
        let mut unresolved: Vec<Unresolved> = Vec::new();
        for dep in &node.deps {
            match self.dependency(paths, state, node, dep, mine.as_deref()) {
                // A dependency this run cannot describe leaves the whole set
                // unfrozen: it is answered again next pass rather than the node
                // being launched against a set with a row missing from it. Every
                // one of them is read rather than the first, so the hold names
                // all it is waiting on; and the clock on each is the first pass
                // that found it unreadable, not this one.
                Resolution::Unreadable(reason) => {
                    let since = self
                        .unresolved_of(&node.id)
                        .iter()
                        .find(|it| it.dep == *dep)
                        .map_or(now, |it| it.since);
                    unresolved.push(Unresolved {
                        dep: dep.clone(),
                        reason,
                        since,
                    });
                }
                Resolution::NothingToAwait => {}
                Resolution::Outside(dependency) => resolved.push(dependency),
            }
        }
        if !unresolved.is_empty() {
            self.unresolved.insert(node.id.clone(), unresolved);
            return Vec::new();
        }
        self.unresolved.remove(&node.id);
        self.dependencies.insert(node.id.clone(), resolved.clone());
        resolved
    }

    /// What one `deps` entry is, to the node that names it.
    fn dependency(
        &mut self,
        paths: &RunPaths,
        state: &RunState,
        node: &Node,
        dep: &str,
        mine: Option<&str>,
    ) -> Resolution {
        let target = node.consumes.get(dep).cloned();
        // A cross-DAG dependency is out-of-repository whatever repository it
        // lands in: the branch is another run's, so the stacked-branch machinery
        // this crate has cannot reach it and a git pin is the only thing a
        // worker can hold.
        if let Some(reference) = crate::crossdag::parse(dep) {
            let Some(upstream) = upstream_of(paths, &reference) else {
                return Resolution::Unreadable(format!(
                    "the run '{}' it names has no ledger under this run root",
                    reference.run
                ));
            };
            let Some(repo) = upstream
                .graph
                .get(&reference.node)
                .and_then(|node| node.repo.clone())
            else {
                // The upstream node lands in no repository, so it releases
                // nothing and there is nothing to pin against.
                return Resolution::NothingToAwait;
            };
            let stated = upstream_paths(paths, &reference)
                .and_then(|upstream| self.stated_landing(&upstream, &reference.node));
            return self.outside(
                dep,
                &repo,
                upstream.branches.get(&reference.node).cloned(),
                upstream.landing_commits.get(&reference.node).cloned(),
                stated,
                target,
            );
        }
        let repo = match across_repositories(&mut self.repositories, state, dep, mine) {
            Ok(repo) => repo,
            Err(ended) => return ended.into(),
        };
        let stated = self.stated_landing(paths, dep);
        self.outside(
            dep,
            &repo,
            state.branches.get(dep).cloned(),
            state.landing_commits.get(dep).cloned(),
            stated,
            target,
        )
    }

    /// Where one dependency's work landed, as an operator stated it — whichever
    /// run's journal that statement is on.
    fn landing_of(&mut self, paths: &RunPaths, dep: &str) -> Option<String> {
        match crate::crossdag::parse(dep) {
            Some(reference) => {
                let upstream = upstream_paths(paths, &reference)?;
                self.stated_landing(&upstream, &reference.node)
            }
            None => self.stated_landing(paths, dep),
        }
    }

    /// One out-of-repository dependency, with what its repository releases.
    fn outside(
        &mut self,
        dep: &str,
        repo: &str,
        branch: Option<String>,
        commit: Option<String>,
        landing: Option<String>,
        named: Option<TargetName>,
    ) -> Resolution {
        outside(
            &mut self.repositories,
            dep,
            repo,
            (branch, commit, landing),
            named,
        )
    }
}

fn identity_of(repositories: &mut Repositories, node: &Node) -> Option<String> {
    node.repo
        .as_deref()
        .and_then(|repo| repositories.of(repo).ok())
        .map(|releases| releases.identity.clone())
}

/// Whether an in-run dependency lands **outside** its consumer's repository, and
/// in which repository: `Ok` with that repository, or the resolution that ends the
/// question here. `mine` is the consumer's own identity.
fn across_repositories(
    repositories: &mut Repositories,
    state: &RunState,
    dep: &str,
    mine: Option<&str>,
) -> std::result::Result<String, Ended> {
    let Some(upstream) = state.graph.get(dep) else {
        return Err(Ended::Unreadable(format!(
            "the graph has no node '{dep}' to say where its work landed"
        )));
    };
    let Some(repo) = upstream.repo.clone() else {
        // A dependency that lands in no repository releases nothing.
        return Err(Ended::NothingToAwait);
    };
    let identity = match repositories.of(&repo) {
        Ok(releases) => releases.identity.clone(),
        Err(why) => return Err(Ended::Unreadable(unread(&repo, &why))),
    };
    if Some(identity.as_str()) == mine {
        // The lifecycle already prepares the stacked or merged-stacked
        // branch for this one, exactly as it does today.
        return Err(Ended::NothingToAwait);
    }
    Ok(repo)
}

enum Ended {
    NothingToAwait,
    Unreadable(String),
}

impl From<Ended> for Resolution {
    fn from(ended: Ended) -> Self {
        match ended {
            Ended::NothingToAwait => Resolution::NothingToAwait,
            Ended::Unreadable(why) => Resolution::Unreadable(why),
        }
    }
}

/// Why a dependency could not be described: what its repository releases could
/// not be read, in the sibling's own words.
fn unread(repo: &str, why: &str) -> String {
    format!("what the repository {repo} releases could not be read: {why}")
}

/// One out-of-repository dependency, with what its repository releases: its
/// branch, commit and stated landing as the run recorded them, and the target its
/// consumer names, if it names one.
fn outside(
    repositories: &mut Repositories,
    dep: &str,
    repo: &str,
    (branch, commit, landing): (Option<String>, Option<String>, Option<String>),
    named: Option<TargetName>,
) -> Resolution {
    let releases = match repositories.of(repo) {
        Ok(releases) => releases,
        Err(why) => return Resolution::Unreadable(unread(repo, &why)),
    };
    // A repository that declares **no release targets releases nothing**, so
    // there is no release to wait for and nothing to pin against instead of
    // one. That is every repository on a host that has configured none —
    // which is every host there was before `onevcs` had a release-targets
    // document at all — and it is what keeps a plan naming neither new field
    // producing exactly the run it produced then: no row, no hold, and a
    // rendered task byte-identical to the one it rendered before.
    if releases.targets.is_empty() {
        return Resolution::NothingToAwait;
    }
    let identity = releases.identity.clone();
    // A repository declaring no target that answers to this name leaves the
    // cell empty rather than the row absent: a worker still needs to see the
    // dependency, and a `published` node still waits — an unanswerable
    // question is not an answer that the release has happened.
    let selected = releases.select(named.as_ref()).ok();
    let (target, style, action, instructions) = match selected {
        Some(target) => (
            Some(target.name.clone()),
            Some(target.style()),
            target.action().map(str::to_owned),
            target.adoption_instructions.clone(),
        ),
        None => (named, None, None, None),
    };
    Resolution::Outside(Dependency {
        dep: dep.to_owned(),
        identity,
        branch,
        commit,
        landing,
        target,
        style,
        action,
        instructions,
    })
}

/// What one `deps` entry turned out to be.
///
/// [`Unreadable`](Resolution::Unreadable) **is** reachable from the release watch
/// of a run, and a `published` node it is reached for is held on it. The graph
/// settles a dependency `done` before its consumer is ever asked about, but what
/// that dependency's repository releases is read from the host — the
/// release-targets document, the registry — on the pass the consumer becomes
/// ready, and a read that does not go through on that pass is not a dependency
/// the node does not have. Read as one, the node launched at once against work no
/// release carried, with no probe and no hold on the record. So the arm leaves the
/// set unfrozen, marks the node unresolved, and the hold names what could not be
/// read; `tests/e2e/adoption.rs` drives it, and `awaited_targets` says the same
/// thing in its own line where a settle reaches it from a reply.
enum Resolution {
    /// There is no release to wait for: it lands in this node's own repository,
    /// in none at all, or in one that declares no release targets.
    NothingToAwait,
    /// It lands elsewhere, and this is what the run can say about it.
    Outside(Dependency),
    /// The run cannot say yet, and this is why.
    Unreadable(String),
}

/// One release a node was waiting on, as the note and the event name it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct Released {
    /// The dependency, as the plan names it.
    pub dep: String,
    /// The repository identity that released it.
    pub identity: String,
    /// The branch the work was on, where the run recorded one.
    pub branch: String,
    /// The commit that work reached its base at, where the run observed one.
    pub commit: String,
    /// The target that carries the work.
    pub target: String,
    /// The version it arrived at.
    pub version: String,
    /// What that repository states about adopting the target.
    pub instructions: Option<InstructionTemplate>,
}

impl Released {
    /// The payload entry this release is recorded as.
    ///
    /// The three the note has always named are always written. The four beside
    /// them are what a producer's own instruction is rendered against, and each is
    /// **omitted when empty** — so a run with nothing to say about a branch, a
    /// commit, or an instruction writes the record it always wrote, and a build
    /// that predates them reads one that carries them exactly as it read one that
    /// did not.
    pub(crate) fn payload(&self) -> Value {
        let mut entry = json!({
            "identity": self.identity,
            "target": self.target,
            "version": self.version,
        });
        let mut put = |key: &str, value: &str| {
            if !value.is_empty() {
                entry[key] = json!(value);
            }
        };
        put("dep", &self.dep);
        put("branch", &self.branch);
        put("commit", &self.commit);
        if let Some(instructions) = &self.instructions {
            entry["instructions"] = json!(instructions);
        }
        entry
    }

    pub(crate) fn row(&self) -> CrossRepoReference {
        CrossRepoReference {
            dependency: self.dep.clone(),
            repository: self.identity.clone(),
            branch: self.branch.clone(),
            commit: self.commit.clone(),
            release_target: self.target.clone(),
            version: self.version.clone(),
            adoption_instructions: self.instructions.clone(),
        }
    }

    /// The releases an event payload recorded, read back.
    ///
    /// A journal is external input like any other — a record this build reads may
    /// have been written by a different one, and it is a file on disk either way —
    /// so every field is checked where it is read and an entry that fails is
    /// **skipped**, exactly as every other reader of a journal skips a record it
    /// cannot read. What is at stake is the rendering: all three land in a
    /// worker's own task, and one carrying a control character forges a line
    /// there.
    pub(crate) fn of_payload(payload: &Value) -> Vec<Self> {
        payload
            .as_array()
            .map(Vec::as_slice)
            .unwrap_or_default()
            .iter()
            .filter_map(|entry| {
                let field = |key: &str| renderable(entry.get(key)?.as_str()?);
                Some(Self {
                    identity: field("identity")?,
                    // The sibling's own conversion, which is the one thing that
                    // decides what may spell a release target: a replayed name
                    // this host would refuse to configure is refused here too.
                    target: entry
                        .get("target")?
                        .as_str()?
                        .parse::<TargetName>()
                        .ok()?
                        .to_string(),
                    version: field("version")?,
                    // The four a record may not carry: a run that could not name
                    // one wrote nothing, and a build older than they are wrote
                    // none of them. Absent is empty rather than a record skipped,
                    // because what a note has to name is the release.
                    dep: field("dep").unwrap_or_default(),
                    branch: field("branch").unwrap_or_default(),
                    commit: field("commit").unwrap_or_default(),
                    // The sibling's own conversion again, for the same reason: a
                    // template is external input here, and what may spell one —
                    // non-empty, bounded, and parsing as a template — is decided
                    // by the type the producer declared it as. One that does not
                    // is dropped, which renders this engine's own default.
                    //
                    // llmlint: ignore-block[changed_behavior_has_e2e] a record no
                    // run of this build writes is not reachable from a journey:
                    // every field of this payload is checked the same way for the
                    // same reason, and `docs/contract-divergences.md` entry 40
                    // records why what a driver takes up out of this record is
                    // held by a fold rather than by a journey. Both deliveries
                    // either side of it — the producer's own instruction reaching
                    // a live turn, and reaching the next dispatch — are driven end
                    // to end by `tests/e2e/adoption.rs`.
                    instructions: entry
                        .get("instructions")
                        .and_then(Value::as_str)
                        .and_then(|declared| declared.parse::<InstructionTemplate>().ok()),
                    // llmlint: ignore-end[changed_behavior_has_e2e]
                })
            })
            .collect()
    }
}

/// One replayed field, held to what it will be rendered as.
///
/// The same rule `src/vcs.rs` holds a session token and a branch name to, and for
/// the same reason: these are printed into a task, a surface, and an event
/// payload, so a value carrying whitespace or a control character renders as
/// something other than what it is wherever it lands. `None` for one that would.
fn renderable(value: &str) -> Option<String> {
    if value.is_empty() || value.len() >= crate::event::MAX_PAYLOAD_TEXT_BYTES {
        return None;
    }
    if value.chars().any(|c| c.is_whitespace() || c.is_control()) {
        return None;
    }
    Some(value.to_owned())
}

/// Record, with `onevcs release acknowledge`'s own semantics, every release an
/// envelope's `settle`s state beside a landing.
///
/// Asked by the reply **before anything of the envelope is queued or applied**,
/// once, and never by a reconciler or a replay: the acknowledgement is a record in
/// `onevcs`'s store rather than in this run's, so it is made by the process a
/// person typed the settle into, and a refusal turns the whole envelope away
/// naming what `onevcs` said — a landing it cannot resolve, one whose target has an
/// established baseline its probe answers instead, a target the repository does
/// not declare, a version that is not a semantic version, or a different version
/// already recorded. Recording the same version again changes nothing, so an
/// envelope sent a second time after a refusal elsewhere in it is harmless.
///
/// **An acknowledgement made here stands if the envelope is then refused** — by a
/// later settle's release `onevcs` will not record, or by the run's reconciler
/// judging a command against what only it knows, such as a dispatch still in
/// flight. It is not undone, because it is not this run's to undo: it is the
/// operator's own statement of which release carries that landing, true whether or
/// not the settle beside it applied, and exactly what `onevcs release acknowledge`
/// would have recorded had they typed it. Sending the corrected envelope again
/// records nothing twice. `tests/e2e/adoption.rs` drives both refusals.
///
/// Nothing is inferred here. The version is the operator's, verified against the
/// release itself; no baseline is derived from when the work was committed or
/// from what a probe answers now, because neither proves what was public when the
/// work landed.
pub(crate) fn acknowledge_stated_releases(commands: &[crate::channel::Command]) -> Result<()> {
    for command in commands {
        let crate::channel::Command::Settle {
            id,
            landing: Some(landing),
            release: Some(release),
            ..
        } = command
        else {
            continue;
        };
        onevcs::acknowledge_release(landing, &release.target, &release.version, false).map_err(
            |refused| {
                crate::Error::Refused(format!(
                    "settle: node '{id}' states that {target} {version} carries the landing \
                     {landing}, and `onevcs` would not record it: {refused}; nothing was queued",
                    target = release.target,
                    version = release.version,
                ))
            },
        )?;
    }
    Ok(())
}

/// The words `onevcs` states an automated target's landing has no release
/// baseline in.
///
/// Read off its own answer because that library exposes the fact only there: a
/// landing with no baseline is `ReleaseStatus::NotAnswered`, and so is a probe that
/// failed just now, and only the reason tells a hold nothing will ever lift apart
/// from one the next probe may. `tests/e2e/adoption.rs` drives the linked `onevcs`
/// into exactly that state and holds this crate's message to it, so a sibling
/// release that rewords it fails there rather than falling silent here.
const NO_BASELINE: &str = "no baseline was captured";

/// What a person who settled a node at a landing, and stated no release carrying
/// it, has to be told **now** about releasing that work.
///
/// One line per release target a node waiting on this one would wait on — as
/// [`awaited_targets`] resolves them through the release watch's own resolution —
/// for which `onevcs` answers one of the two holds nothing in the run will ever
/// lift:
///
/// - the landing has **no release baseline**: nothing recorded what the target had
///   published when the work landed, so no probe answer can say a release carries
///   it, and the line carries the exact `onevcs release acknowledge` command with
///   the reference and the target filled in, for the version a person verifies;
/// - `onevcs` cannot resolve the stated reference to landed work at all, so no
///   release will ever be correlated through it.
///
/// And one line where what the node's repository releases could not be read, so
/// the question was not asked at all. Every other answer — released, not released
/// against a baseline, not landed yet, a human step awaiting its person, a probe
/// that did not answer this time — says nothing, because none of them is a hold the
/// next answer cannot lift. Nothing here infers a baseline; `onevcs` is asked exactly
/// what the release watch asks it.
pub(crate) fn hold_warnings_for_stated_landings(
    state: &RunState,
    commands: &[crate::channel::Command],
) -> Vec<String> {
    let mut said = Vec::new();
    for command in commands {
        let crate::channel::Command::Settle {
            id,
            landing: Some(landing),
            release: None,
            ..
        } = command
        else {
            continue;
        };
        let Some(repo) = state.graph.get(id).and_then(|node| node.repo.clone()) else {
            continue;
        };
        let mut repositories = Repositories::default();
        let Some(awaited) = awaited_targets(&mut repositories, state, id, &repo) else {
            said.push(format!(
                "onepipeline: settle: node '{id}' was settled at {landing}, and what its \
                 repository {repo} releases could not be read, so whether that landing has a \
                 release baseline was not asked; `onevcs release targets {repo}` says why"
            ));
            continue;
        };
        let identity = repositories
            .of(&repo)
            .map(|releases| releases.identity.clone())
            .unwrap_or_else(|_| repo.clone());
        for target in awaited {
            let named = target.to_string();
            let reference = shell_word(landing);
            match onevcs::release_status(landing, Some(&target)) {
                Ok(ReleaseStatus::NotAnswered { reason }) if reason.contains(NO_BASELINE) => {
                    said.push(format!(
                        "onepipeline: settle: node '{id}' was settled at the landing {landing}, \
                         and that landing has no release baseline for the release target \
                         '{named}' of {identity}: nothing recorded what that target had \
                         published when the work landed, so no probe answer can show a release \
                         carries it, and a node waiting on that release holds until one is \
                         recorded. Once you have verified the version that first carries it, \
                         record it:\n  onevcs release acknowledge {reference} --target {named} \
                         --version <VERSION>",
                    ));
                }
                Err(refused) => said.push(format!(
                    "onepipeline: settle: node '{id}' was settled at {landing}, which `onevcs` \
                     cannot resolve to landed work, so no release of the release target \
                     '{named}' of {identity} will be attributed through it and a node waiting \
                     on that release holds: {refused}. State a landing `onevcs` can resolve \
                     by settling the node again at the change request that carried the work, \
                     with the `release` that carries it once you have verified it; or settle it \
                     there without one and record the release:\n  onevcs release acknowledge \
                     '<CHANGE-REQUEST-URL>' --target {named} --version <VERSION>",
                )),
                Ok(_) => {}
            }
        }
    }
    said
}

/// The release targets a wait on one node's release is asked about, each once — or
/// `None` where what a repository releases could not be read.
///
/// **The release watch's own resolution, not a restatement of it**: each dependent
/// in this run is resolved through [`across_repositories`] and [`outside`], exactly
/// as [`Watch`] resolves the dependency it waits on, so a dependent in the node's
/// own repository awaits nothing and one naming no target awaits the repository's
/// default. Where no dependent awaits a release — nothing depends on the node yet,
/// or nothing outside its repository does — the node's own repository is resolved
/// the same way with no target named, which is its default: what a consumer added
/// later, or one in another run, waits on unless it says otherwise.
fn awaited_targets(
    repositories: &mut Repositories,
    state: &RunState,
    node: &str,
    repo: &str,
) -> Option<BTreeSet<TargetName>> {
    let mut resolutions: Vec<Resolution> = Vec::new();
    for dependent in state
        .graph
        .iter()
        .filter(|dependent| dependent.deps.iter().any(|dep| dep == node))
    {
        let mine = identity_of(repositories, dependent);
        resolutions.push(
            match across_repositories(repositories, state, node, mine.as_deref()) {
                Ok(repo) => outside(
                    repositories,
                    node,
                    &repo,
                    (None, None, None),
                    dependent.consumes.get(node).cloned(),
                ),
                Err(ended) => ended.into(),
            },
        );
    }
    let mut awaited = asked_about(resolutions)?;
    if awaited.is_empty() {
        awaited = asked_about(vec![outside(
            repositories,
            node,
            repo,
            (None, None, None),
            None,
        )])?;
    }
    Some(awaited)
}

/// The targets a set of resolutions puts a question about: every dependency the
/// watch would ask a target of, and `None` where one could not be described.
fn asked_about(resolutions: Vec<Resolution>) -> Option<BTreeSet<TargetName>> {
    let mut asked = BTreeSet::new();
    for resolution in resolutions {
        match resolution {
            Resolution::Unreadable(_) => return None,
            Resolution::NothingToAwait => {}
            Resolution::Outside(dependency) => {
                // A target the repository declares, which is what gives it a style: a
                // name it does not declare is a question `onevcs` would only refuse.
                asked.extend(dependency.style.and(dependency.target));
            }
        }
    }
    Some(asked)
}

/// One word of a command line a person pastes into a POSIX shell, single-quoted:
/// a URL may carry `&`, `?` or `#`, and a word quoted every time is one reading of
/// it rather than two.
fn shell_word(word: &str) -> String {
    format!("'{}'", word.replace('\'', "'\\''"))
}

/// Why one fast-adoption node's change request is opened as a **draft**, asked of
/// `onevcs` at the moment the publication is composed.
///
/// The rows are the node's own reference block, so what is asked about is exactly
/// what the worker was told to pin against. `None` is a node with nothing to wait
/// for.
///
/// **Asked here rather than off the reconcile loop's [`Watch`]** because a
/// dispatch is long and a release can arrive in the middle of one: the question is
/// whether the pin is still temporary *now*.
///
/// A row this run cannot put a question about is **not** drafted against — the
/// judgement [`Dependency::askable`] already makes for the hold — because a draft
/// nothing can ever lift is a change request nobody can finish.
pub(crate) fn draft_reason(references: &[CrossRepoReference]) -> Option<onevcs::DraftReason> {
    // llmlint: ignore-block[changed_behavior_has_e2e] no plan reaches the dropped
    // half of `askable`: a row is only composed for a dependency that settled `done`
    // in a repository declaring release targets, so it has the branch or the landing
    // commit the run recorded, and its target cell is either one that repository
    // declares or the one the plan's own `consumes` named — both of which parse.
    // What the arm does is leave such a row undrafted, which is this crate behaving
    // exactly as it did before drafting existed.
    let unreleased: Vec<(&CrossRepoReference, &str, TargetName)> = references
        .iter()
        // A row the run already has the version of is a pin that is not temporary:
        // that is every row a `published` node carries, since it was not started
        // until each of them released. Filtered on what the row says rather than on
        // the mode the plan declared, because asking a probe about an arrived
        // release again would let one transient "not answered" open a change
        // request as a draft against a pin that node never held.
        .filter(|row| row.version.is_empty())
        .filter_map(|row| {
            let (reference, target) = askable(row)?;
            Some((row, reference, target))
        }) // llmlint: ignore-end[changed_behavior_has_e2e]
        .filter(|(_, reference, target)| !released_already(reference, target))
        .collect();
    let (row, reference, target) = unreleased.first()?;
    // One reason names one dependency, because [`onevcs::DraftReason`] carries one
    // target — so how many are being waited on is said in the sentence a person
    // reads rather than left for them to count.
    Some(onevcs::DraftReason::AwaitingRelease {
        because: format!(
            "this node adopted {identity} early and is pinned to {reference} rather than to a \
             released version; it is one of {count} release(s) this node adopted early, and \
             landing now would make that pin permanent",
            identity = row.repository,
            count = unreleased.len(),
        ),
        awaiting: row.repository.clone(),
        target: target.clone(),
        reference: (*reference).to_owned(),
    })
}

/// What a draft reason names a settlement by, on the one line a settlement holds.
///
/// For a release that is awaited, three of the reason's four fields — the target,
/// the repository, and the reference — read off the value the publication was
/// made with, so the settlement cannot come to name a different release from the
/// one the host is holding the change for. `because` is not among them: it is the
/// sentence composed for a person reading the change request's own record, and a
/// settlement's detail is read on one line beside an outcome word that has
/// already said the node is a draft. A draft the **plan** asked for has no
/// release to name and says only that it is one, and whose to lift. Rendered here
/// rather than in `src/lifecycle.rs` so there is one spelling of each.
pub(crate) fn drafted_detail(reason: &onevcs::DraftReason) -> String {
    match reason {
        onevcs::DraftReason::AwaitingRelease {
            target,
            awaiting,
            reference,
            ..
        } => format!(
            "complete, and held as a draft: awaiting the {target} release of {awaiting}, \
             pinned to {reference} until it arrives"
        ),
        onevcs::DraftReason::Held { .. } => HELD_DETAIL.to_owned(),
    }
}

/// The one line a node that was asked to leave its change request as a draft
/// settles with.
///
/// Said once, because the settlement composes it beside what the worker did to
/// the change request and the two have to agree about the words: a person reads
/// this to learn why a `done` node's change did not land, and the answer is that
/// the plan said so and that lifting it is theirs.
pub(crate) const HELD_DETAIL: &str =
    "complete, and left as a draft as the plan asked, for a person to mark ready for review";

/// Why a change request the plan asked to leave as a draft is one, in the words
/// `onevcs` records on the change request's own stream.
///
/// The sentence a person reads off `change-drafted`, so it says whose decision
/// the draft is: the plan's, and not a release's. Composed here beside the other
/// reason this crate hands that library so both are spelled in one module.
pub(crate) fn held_reason(node: &str) -> onevcs::DraftReason {
    onevcs::DraftReason::Held {
        because: format!(
            "node '{node}' is declared `draft: true`, so its change request is left as a draft \
             for a person to mark ready for review"
        ),
    }
}

/// The pair a reference row can be asked about: what the sibling resolves the
/// work by, and which of that repository's targets carries it.
///
/// Both cells are empty where the run could not name one — the block renders an
/// empty cell rather than dropping the row — so this is where a row that cannot
/// be a question is told from one that can.
fn askable(row: &CrossRepoReference) -> Option<(&str, TargetName)> {
    let reference = [row.branch.as_str(), row.commit.as_str()]
        .into_iter()
        .find(|value| renderable(value).is_some())?;
    // The sibling's own conversion, which is the one thing that decides what may
    // spell a release target — the same rule [`Released::of_payload`] reads a
    // replayed one back through.
    Some((reference, row.release_target.parse::<TargetName>().ok()?))
}

/// Whether the release one reference row awaits has already happened.
///
/// [`Answer::of`] and no second reading of the sibling: *released* is the one
/// answer that means the pin can go, and every other — not released, awaiting a
/// person, a probe that could not answer — leaves the change a draft. "Not
/// answered" is never "not released" here either; what it is is a change that
/// must not land yet, which is the safe direction and the only one.
fn released_already(reference: &str, target: &TargetName) -> bool {
    Answer::of(&onevcs::release_status(reference, Some(target)))
        .version()
        .is_some()
}

/// The note a fast-adoption node is sent when the releases it was waiting on
/// arrive.
///
/// It **adds no bar**: it reports observed state and says what to do with it, in
/// the same frame a carried planner note is rendered in, so no worker can read it
/// as a new acceptance criterion. What it says to do with it is the **producer's
/// own** instruction, rendered by [`crate::plan::arrival_note`] against the
/// versions that arrived — the one rendering the reference block uses too, so an
/// instruction cannot read one way in a note and another in a task.
///
/// One function, called both where the note is delivered and where a journalled
/// delivery is folded back, so a note replayed from the record is the note that
/// was sent — which is why the record carries what the template is rendered
/// against and not only the rendering.
pub(crate) fn arrival_note(released: &[Released]) -> String {
    crate::plan::arrival_note(&released.iter().map(Released::row).collect::<Vec<_>>())
}

/// The distinct questions one pass has to put, and every wait each one answers.
///
/// [`onevcs::release_status`] is answered by the reference and the target alone,
/// so two waits naming the same pair are asking the **identical** question: it is
/// put once and its answer belongs to both. Put once per waiting node instead,
/// each copy runs the probe subprocess again on every poll and the last node in
/// the list waits out every probe before it — which is how it comes to read as a
/// release nothing has answered while a node beside it reads as answered.
///
/// A wait with nothing to ask — no reference the sibling resolves work by, or no
/// target that answers — puts no question and joins none. The wait stands: an
/// unanswerable question is not an answer that the release has happened.
fn questions_of(waits: &[(Key, Dependency)]) -> Vec<Question> {
    let mut questions: Vec<Question> = Vec::new();
    // Where the question about one release already stands, so the next wait
    // naming it joins that one. Keyed by everything the answer depends on — the
    // reference and the target — with the style beside them, so a pairing this
    // crate has not foreseen joins nothing rather than taking an answer obtained
    // another way.
    let mut asked: BTreeMap<(&str, Option<&TargetName>, &'static str), usize> = BTreeMap::new();
    for (key, dependency) in waits {
        let Some(reference) = dependency.reference() else {
            continue;
        };
        let Some(style) = dependency.style else {
            continue;
        };
        let about = (reference, dependency.target.as_ref(), style.as_str());
        match asked.get(&about) {
            Some(&already) => questions[already].keys.push(key.clone()),
            None => {
                asked.insert(about, questions.len());
                questions.push(Question {
                    keys: vec![key.clone()],
                    reference: reference.to_owned(),
                    target: dependency.target.clone(),
                    style,
                });
            }
        }
    }
    questions
}

/// The nodes whose releases matter on this pass.
///
/// Every node that is ready to start — which is where a hold applies and where a
/// dispatch's reference block is composed — every fast-adoption node still
/// running, which is where an arrival note is delivered, and every node whose
/// change is **held as a draft**, which is where the release that lifts it is
/// still being watched for. The third is the one that matters after a node has
/// stopped running: a draft nobody goes on asking about is a change request that
/// never becomes ready.
pub(crate) fn watching(
    state: &RunState,
    statuses: &BTreeMap<String, NodeStatus>,
    running: &BTreeSet<String>,
) -> Vec<Node> {
    state
        .graph
        .iter()
        .filter(|node| {
            matches!(
                statuses.get(&node.id),
                Some(&NodeStatus::Ready | &NodeStatus::CompleteDraft)
            ) || running.contains(&node.id)
        })
        .cloned()
        .collect()
}

/// Another run's folded state, for a cross-DAG dependency.
fn upstream_of(paths: &RunPaths, reference: &crate::crossdag::Reference) -> Option<RunState> {
    let upstream = upstream_paths(paths, reference)?;
    Some(crate::projection::fold(&journal::read(&upstream.journal())))
}

/// Where the run a cross-DAG dependency names keeps its own record.
fn upstream_paths(paths: &RunPaths, reference: &crate::crossdag::Reference) -> Option<RunPaths> {
    let upstream = RunPaths::under(paths.dir.parent()?, &reference.run);
    upstream.exists().then_some(upstream)
}

/// Where each node of one run's journal was stated to have landed.
///
/// **The projection's fold, and not a reading of its own**: a stated landing has
/// one interpretation — [`StatedLanding`](crate::edits::StatedLanding) — and the
/// views, `results` and the status write-back take it from
/// [`RunState::stated_landings`], so this does too, and what a release is
/// correlated through cannot come to differ from what those surfaces say landed.
/// That fold already holds a journal to what a `settle` may state, passes over an
/// operation list this build cannot read whole, lets the newest statement about a
/// node win, and lets a later settlement the run recorded itself supersede one.
///
/// A node id that would forge a line where a wait is rendered is passed over here
/// as well, because this is where it is rendered from.
fn stated_landings(events: &[crate::event::Envelope]) -> BTreeMap<String, String> {
    crate::projection::fold(events)
        .stated_landings
        .into_iter()
        .filter_map(|(node, stated)| Some((renderable(&node)?, String::from(stated))))
        .collect()
}

/// How often an automated target's probe is run.
///
/// An unusable value falls back to the default rather than to zero, which would
/// spend the host on probes, or to no bound at all, which would ask once and
/// never again.
fn poll_seconds() -> u64 {
    std::env::var(POLL_ENV)
        .ok()
        .and_then(|value| value.parse().ok())
        .filter(|seconds| *seconds > 0)
        .unwrap_or(DEFAULT_POLL_SECONDS)
}

/// The longest a withdrawn question may go on being asked.
///
/// What the seam models is one probe run that was already in flight, so the
/// bound is what one such run can cost: `onevcs` gives a probe its target's own
/// timeout, and this crate's own fixtures give it thirty seconds. Five minutes is
/// ten of those, and past it a question the loop has withdrawn is being asked for
/// a reason no in-flight probe explains.
const MAX_WITHDRAWN_ASK_SECONDS: u64 = 300;

/// How long the asker goes on asking a question the loop has withdrawn.
///
/// Zero for an unset value, one this build cannot read, and one past
/// [`MAX_WITHDRAWN_ASK_SECONDS`] — all three of which are the behaviour every
/// build in the field has, a withdrawn question asked no more. Falling **off**
/// rather than to the bound is the safe direction here for the reason
/// [`poll_seconds`] falls back rather than to zero: a value nobody can have meant
/// should leave the shipped behaviour alone rather than pick a number of its own.
fn withdrawn_ask() -> Duration {
    Duration::from_secs(
        std::env::var(WITHDRAWN_ASK_ENV)
            .ok()
            .and_then(|value| value.parse().ok())
            .filter(|seconds| *seconds <= MAX_WITHDRAWN_ASK_SECONDS)
            .unwrap_or(0),
    )
}

/// Take up a fresh question set, keeping for `linger` the questions it no longer
/// names.
///
/// With `linger` zero — every build in the field — the fresh set simply replaces
/// the one before it and nothing is kept, which is what this did before there was
/// anything to keep.
fn retire(
    current: &[Question],
    fresh: Vec<Question>,
    linger: Duration,
    withdrawn: &mut Vec<(Question, Instant)>,
) -> Vec<Question> {
    if !linger.is_zero() {
        let now = Instant::now();
        // A question is the one it is by what it asks about, so a set that names
        // the same release with a different wait behind it has withdrawn nothing.
        let names = |question: &Question, other: &Question| {
            question.reference == other.reference && question.target == other.target
        };
        for question in current {
            if !fresh.iter().any(|other| names(question, other))
                && !withdrawn.iter().any(|(held, _)| names(question, held))
            {
                withdrawn.push((question.clone(), now));
            }
        }
    }
    fresh
}

/// How often a held node's wait is surfaced.
fn surface_every_seconds() -> u64 {
    std::env::var(SURFACE_ENV)
        .ok()
        .and_then(|value| value.parse().ok())
        .filter(|seconds| *seconds > 0)
        .unwrap_or(DEFAULT_SURFACE_SECONDS)
}

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

    fn dependency(target: Option<&str>, style: Option<ReleaseStyle>) -> Dependency {
        Dependency {
            dep: "engine".to_owned(),
            identity: "github.com/owner/engine".to_owned(),
            branch: Some("onevcs/s-1".to_owned()),
            commit: Some("9f3c1ab".to_owned()),
            landing: None,
            target: target.map(|name| name.parse().expect("a target name")),
            style,
            action: style
                .filter(|style| *style == ReleaseStyle::HumanStep)
                .map(|_| "cut a release on PyPI".to_owned()),
            instructions: None,
        }
    }

    /// Every answer `onevcs` can give, and the refusal it can give instead, read
    /// as this crate's own vocabulary.
    ///
    /// Arm by arm, because the whole design rests on three of them staying
    /// apart: "awaiting a human step" is neither of its neighbours, and neither a
    /// probe that failed nor a question that could not be *put* is ever "not
    /// released".
    #[test]
    fn no_answer_the_sibling_gives_is_folded_into_another() {
        let cases: Vec<(onevcs::Result<ReleaseStatus>, &str)> = vec![
            (
                Ok(ReleaseStatus::Released {
                    target: "crate".parse().expect("a target name"),
                    style: ReleaseStyle::Automated,
                    version: "0.2.0".to_owned(),
                    source: onevcs::ReleaseSource::Probed,
                }),
                "released",
            ),
            // An automated target whose landing has no baseline, released by the
            // acknowledgement a person recorded: the same answer, whatever
            // produced it.
            (
                Ok(ReleaseStatus::Released {
                    target: "crate".parse().expect("a target name"),
                    style: ReleaseStyle::Automated,
                    version: "0.2.0".to_owned(),
                    source: onevcs::ReleaseSource::Acknowledged,
                }),
                "released",
            ),
            (
                Ok(ReleaseStatus::NotReleased {
                    at_landing: Baseline::At {
                        version: "0.1.0".to_owned(),
                    },
                    now: "0.1.0".to_owned(),
                }),
                "not-released",
            ),
            (
                Ok(ReleaseStatus::AwaitingHumanStep {
                    target: "wheel".parse().expect("a target name"),
                    action: "cut a release on PyPI".to_owned(),
                    since: "2026-08-24T00:00:00.000Z".to_owned(),
                }),
                "awaiting-human-step",
            ),
            (
                Ok(ReleaseStatus::NotAnswered {
                    reason: "the probe timed out".to_owned(),
                }),
                "not-answered",
            ),
            (Ok(ReleaseStatus::NotLanded), "not-landed"),
            // A question that could not be *put* at all — the repository
            // declares no target answering to the name, or names no default.
            // Not an answer that the release has not happened.
            (
                Err(onevcs::Error::Invalid {
                    reason: "the repository declares no release targets".to_owned(),
                }),
                "not-answered",
            ),
        ];
        for (status, expected) in cases {
            assert_eq!(
                Answer::of(&status).as_str(),
                expected,
                "{status:?} was read as another answer"
            );
        }
        // Exactly one of them releases a hold, and it is the one that names a
        // version.
        assert_eq!(
            Answer::Released {
                version: "0.2.0".to_owned()
            }
            .version(),
            Some("0.2.0")
        );
        for answer in [
            Answer::NotReleased,
            Answer::AwaitingHumanStep,
            Answer::NotAnswered,
            Answer::NotLanded,
        ] {
            assert_eq!(answer.version(), None, "{answer:?} released a hold");
        }
    }

    /// The two rungs a plan and this crate own: the node's own field, and the
    /// floor beneath every rung.
    ///
    /// The two in between are `onevcs`'s, and a node held against them is
    /// `tests/e2e/adoption.rs`'s
    /// `the_adoption_mode_resolves_through_exactly_four_rungs`, which drives all
    /// four against a real host — a rung is not a value anything reports, so what
    /// it decides is whether the node is scheduled.
    #[test]
    fn the_node_rung_wins_outright_and_a_node_with_no_repository_falls_to_the_floor() {
        let stated = Node {
            id: "stated".to_owned(),
            adoption: Some(Adoption::Published),
            ..Node::default()
        };
        assert_eq!(adoption_of(&stated), Adoption::Published);
        // No repository, so no repository rung — and no way to reach the global
        // one without naming one, which is the floor.
        assert_eq!(adoption_of(&Node::default()), Adoption::Fast);
        // A repository the sibling cannot answer for is a question that was not
        // put, which is the floor too rather than a node held for ever.
        let unknown = Node {
            id: "unknown".to_owned(),
            repo: Some("no-such-repository-on-this-host".to_owned()),
            ..Node::default()
        };
        assert_eq!(adoption_of(&unknown), Adoption::Fast);
    }

    /// A cell the run cannot name is **empty**, and the row is still there.
    #[test]
    fn a_dependency_the_run_cannot_fully_name_is_rendered_with_the_cell_empty() {
        let named = dependency(Some("crate"), Some(ReleaseStyle::Automated)).row(None);
        assert_eq!(named.repository, "github.com/owner/engine");
        assert_eq!(named.branch, "onevcs/s-1");
        assert_eq!(named.commit, "9f3c1ab");
        assert_eq!(named.release_target, "crate");

        // A repository declaring targets but no default, asked for none: the
        // sibling names no target, so the cell is empty and the row stands.
        let mut unnamed = dependency(None, None);
        unnamed.branch = None;
        unnamed.commit = None;
        let row = unnamed.row(None);
        assert_eq!(row.dependency, "engine");
        assert_eq!(row.repository, "github.com/owner/engine");
        assert!(row.branch.is_empty() && row.commit.is_empty() && row.release_target.is_empty());
    }

    /// The branch is what the sibling is asked about, and the commit is the
    /// fallback — see [`Dependency::reference`].
    #[test]
    fn the_reference_the_sibling_is_asked_about_is_the_branch() {
        assert_eq!(
            dependency(Some("crate"), None).reference(),
            Some("onevcs/s-1")
        );
        let mut branchless = dependency(Some("crate"), None);
        branchless.branch = None;
        assert_eq!(branchless.reference(), Some("9f3c1ab"));
        branchless.commit = None;
        assert_eq!(branchless.reference(), None);
    }

    /// A landing an operator stated is asked about ahead of both, in either
    /// spelling of one — and is what makes a dependency the run could not
    /// otherwise name askable at all.
    #[test]
    fn a_landing_an_operator_stated_is_what_the_sibling_is_asked_about() {
        for landing in [
            "https://github.com/owner/engine/pull/12",
            "3f9a1c2e5b7d9081f2a3b4c5d6e7f8091a2b3c4d",
        ] {
            let mut stated = dependency(Some("crate"), Some(ReleaseStyle::Automated));
            stated.landing = Some(landing.to_owned());
            assert_eq!(
                stated.reference(),
                Some(landing),
                "the branch this settle corrects is still what the release is measured against"
            );
            // And it is what makes a dependency the run could not otherwise name
            // askable at all: a node settled from evidence that never published
            // has no branch and no commit of its own.
            stated.branch = None;
            stated.commit = None;
            assert!(stated.askable(), "a stated landing is no question at all");
            assert_eq!(stated.reference(), Some(landing));
        }
    }

    /// Where a node's work was stated to have landed is read off the journal, and
    /// a record this build cannot read whole says nothing.
    ///
    /// The journal is external input — another build wrote it and a person can
    /// edit it — and what a mis-read costs here is a release question put about
    /// the wrong work, which is answered with somebody else's release.
    #[test]
    fn a_stated_landing_is_read_off_the_journal_and_the_last_one_wins() {
        let edit = |operations: Value| {
            serde_json::from_value::<crate::event::Envelope>(json!({
                "v": 1,
                "ts": "2026-09-07T00:00:00.000Z",
                "stream": "pipeline",
                "seq": 0,
                "source": "pipeline",
                "kind": journal::PipelineKind::EditCommitted.as_str(),
                "labels": {"run_id": "settled"},
                "payload": {"operations": operations},
            }))
            .expect("an envelope")
        };
        let stated = |node: &str, landing: &str| json!([{"kind": "landing-from-evidence", "node": node, "landing": landing}]);
        assert_eq!(
            stated_landings(&[
                edit(stated("publish", "3f9a1c2ab")),
                edit(stated("other", "https://example.invalid/pull/1")),
                // A record corrected twice: the newest correction is the one.
                edit(stated("publish", "9d8c7b6ef")),
            ]),
            [
                (
                    "other".to_owned(),
                    "https://example.invalid/pull/1".to_owned()
                ),
                ("publish".to_owned(), "9d8c7b6ef".to_owned()),
            ]
            .into_iter()
            .collect::<BTreeMap<String, String>>()
        );
        // Nothing this build cannot read whole is read as a landing: an
        // operation list it cannot parse, an operation of another kind, and a
        // value that would forge a line wherever it is rendered.
        for unreadable in [
            json!("not a list of operations"),
            json!([{"kind": "settled-from-evidence", "node": "publish", "outcome": "done",
                    "evidence": "it merged"}]),
            stated("publish", ""),
            stated("publish", "3f9a1c2 and the one before it"),
            stated("", "3f9a1c2"),
            // Neither spelling a `settle` may state, so neither is a question
            // this run puts: the journal is a file another build wrote and a
            // person can edit, and it is held to what the op admits on this side
            // too.
            stated("publish", "the-change-that-merged"),
            stated("publish", "3f9a1c"),
        ] {
            assert!(
                stated_landings(&[edit(unreadable.clone())]).is_empty(),
                "{unreadable} was read as a landing"
            );
        }
        // And a run whose journal states none has none, which is every run that
        // has never had a settlement corrected.
        assert!(stated_landings(&[]).is_empty());
    }

    /// One release, as the note and the record name it.
    fn arrival(instructions: Option<&str>) -> Released {
        Released {
            dep: "engine".to_owned(),
            identity: "github.com/nickderobertis/onevcs".to_owned(),
            branch: "onevcs/s-1".to_owned(),
            commit: "9f3c1ab".to_owned(),
            target: "crate".to_owned(),
            version: "0.13.0".to_owned(),
            instructions: instructions.map(|declared| {
                declared
                    .parse()
                    .expect("a template the producer could declare")
            }),
        }
    }

    /// The note reports observed state and adds no bar, and round-trips through
    /// the payload it is journalled as.
    ///
    /// A producer that declares no template gets the engine's own default, which
    /// is the sentence the note carried before a producer could declare one.
    #[test]
    fn the_arrival_note_names_the_versions_and_states_no_criterion() {
        let released = vec![arrival(None)];
        let note = arrival_note(&released);
        assert_eq!(
            note,
            "The releases this node was waiting on have arrived:\n\n\
             - github.com/nickderobertis/onevcs — crate 0.13.0\n\n\
             This reports observed state and adds no acceptance criteria. What the producer of \
             each dependency above states about adopting it:\n\n\
             Move from the git pin to that released version.\n\n\
             That is the end of what the producers state; none of it is a criterion of this node."
        );
        assert!(!note.to_lowercase().contains("must"));

        let payload = json!(released.iter().map(Released::payload).collect::<Vec<_>>());
        assert_eq!(
            arrival_note(&Released::of_payload(&payload)),
            note,
            "a note replayed from the record is not the note that was sent"
        );

        // The producer's own instruction is what the note says to do, and it
        // round-trips too — because what the record carries is what the template
        // is rendered against rather than the rendering, so a note replayed by a
        // fresh driver is the note the node was told.
        let declared = vec![arrival(Some(
            "Raise the `onevcs` pin to {{ version }}; the branch pin at {{ branch }} goes.",
        ))];
        let stated = "Raise the `onevcs` pin to 0.13.0; the branch pin at onevcs/s-1 goes.";
        let declared_note = arrival_note(&declared);
        assert!(
            declared_note.contains(stated),
            "the producer's own instruction did not reach the note:\n{declared_note}"
        );
        assert!(
            !declared_note.contains(crate::plan::DEFAULT_ADOPTION_INSTRUCTION),
            "a producer that declared one still got the engine's default:\n{declared_note}"
        );
        let record = json!(declared.iter().map(Released::payload).collect::<Vec<_>>());
        assert_eq!(
            arrival_note(&Released::of_payload(&record)),
            declared_note,
            "a note replayed from the record lost the producer's own instruction"
        );
        // Every cell the record has to carry for that rendering is in it, and the
        // ones a run had nothing to say about are omitted rather than written
        // empty — so a build that predates them reads what it always read.
        assert_eq!(record[0]["dep"], json!("engine"));
        assert_eq!(record[0]["branch"], json!("onevcs/s-1"));
        assert_eq!(record[0]["commit"], json!("9f3c1ab"));
        let mut bare = arrival(None);
        bare.dep = String::new();
        bare.branch = String::new();
        bare.commit = String::new();
        assert_eq!(
            bare.payload(),
            json!({
                "identity": "github.com/nickderobertis/onevcs",
                "target": "crate",
                "version": "0.13.0",
            }),
            "a record gained a key about something the run could not name"
        );
        // And a replayed template this build would refuse to read is dropped
        // rather than rendered, which leaves the engine's own default.
        for refused in [json!(""), json!("{{ unclosed"), json!(42)] {
            let entry = json!([{
                "identity": "a", "target": "crate", "version": "0.13.0",
                "instructions": refused,
            }]);
            assert_eq!(
                Released::of_payload(&entry)
                    .first()
                    .expect("the release itself still reads")
                    .instructions,
                None,
                "{refused} was read as a template"
            );
        }

        // A record this build cannot read whole is skipped rather than rendered
        // with a blank where the version should be: a journal is external input,
        // and a note naming no version tells a worker nothing it can act on.
        for unreadable in [
            json!([{"identity": "a", "target": "crate"}]),
            json!([{"identity": "a", "target": "crate", "version": 13}]),
            json!([{"identity": "a", "target": "crate", "version": ""}]),
            json!([{"target": "crate", "version": "0.13.0"}]),
            json!("not a list at all"),
            // A value that would forge a line in the task it is rendered into.
            json!([{"identity": "a\n- b — c 9.9.9", "target": "crate", "version": "0.13.0"}]),
            json!([{"identity": "a", "target": "crate", "version": "0.13.0\u{7}"}]),
            json!([{"identity": "a", "target": "crate", "version": "0.13.0 and more"}]),
            // A target name the sibling's own conversion refuses.
            json!([{"identity": "a", "target": "not a target name", "version": "0.13.0"}]),
            json!([{"identity": "a", "target": "-leading", "version": "0.13.0"}]),
        ] {
            assert!(
                Released::of_payload(&unreadable).is_empty(),
                "{unreadable} was read as a release"
            );
        }
    }

    /// Which waits share a question and which do not, arm by arm. What sharing
    /// one *costs* a run is driven end to end by `tests/e2e/adoption.rs`'s
    /// `nodes_awaiting_one_release_put_one_question_and_are_answered_together`.
    #[test]
    fn waits_naming_one_release_put_one_question_between_them() {
        let wait = |node: &str, dependency: Dependency| {
            ((node.to_owned(), dependency.dep.clone()), dependency)
        };
        let automated = || dependency(Some("crate"), Some(ReleaseStyle::Automated));

        let questions = questions_of(&[
            wait("first", automated()),
            wait("second", automated()),
            wait("third", automated()),
        ]);
        assert_eq!(questions.len(), 1, "{questions:?}");
        assert_eq!(
            questions[0].keys,
            vec![
                ("first".to_owned(), "engine".to_owned()),
                ("second".to_owned(), "engine".to_owned()),
                ("third".to_owned(), "engine".to_owned()),
            ],
        );
        assert_eq!(questions[0].reference, "onevcs/s-1");
        assert_eq!(questions[0].style, ReleaseStyle::Automated);

        // One target's answer says nothing about another's.
        let mut wheel = automated();
        wheel.target = Some("wheel".parse().expect("a target name"));
        wheel.style = Some(ReleaseStyle::HumanStep);
        let questions = questions_of(&[wait("first", automated()), wait("second", wheel.clone())]);
        assert_eq!(questions.len(), 2, "{questions:?}");
        assert_eq!(questions[1].style, ReleaseStyle::HumanStep);

        // And one landing's says nothing about another's.
        let mut other_branch = automated();
        other_branch.branch = Some("onevcs/s-2".to_owned());
        let questions = questions_of(&[
            wait("first", automated()),
            wait("second", other_branch.clone()),
        ]);
        assert_eq!(questions.len(), 2, "{questions:?}");

        let mut styleless = automated();
        styleless.style = None;
        let mut referenceless = automated();
        referenceless.branch = None;
        referenceless.commit = None;
        let questions = questions_of(&[
            wait("unanswerable", styleless),
            wait("unnameable", referenceless),
            wait("asked", automated()),
        ]);
        assert_eq!(questions.len(), 1, "{questions:?}");
        assert_eq!(
            questions[0].keys,
            vec![("asked".to_owned(), "engine".to_owned())]
        );
        assert!(questions_of(&[]).is_empty());
    }

    /// The three states of a wait with no version yet, read off both places the
    /// distinction has to exist: the payload and the surface a person reads.
    #[test]
    fn a_wait_still_expecting_its_first_answer_is_not_a_probe_that_could_not_answer() {
        let mut watch = Watch::of_run(&RunPaths::under(std::path::Path::new("/nowhere"), "demo"));
        watch.dependencies.insert(
            "asked".to_owned(),
            vec![dependency(Some("crate"), Some(ReleaseStyle::Automated))],
        );
        // A target this host declares nothing for: no question can be put, so
        // nothing will ever answer it.
        watch
            .dependencies
            .insert("unanswerable".to_owned(), vec![dependency(None, None)]);

        assert_eq!(
            watch.awaiting("asked")[0]["last_answer"],
            json!("no-answer-yet"),
            "a probe that has not come back yet was reported as one that failed"
        );
        assert!(watch
            .wait_surface("asked")
            .message
            .contains("last answer: no-answer-yet"));
        assert_eq!(
            watch.awaiting("unanswerable")[0]["last_answer"],
            json!("not-answered"),
            "a question that could not be put was reported as one still in flight"
        );

        watch.answers.insert(
            ("asked".to_owned(), "engine".to_owned()),
            Answer::NotAnswered,
        );
        assert_eq!(
            watch.awaiting("asked")[0]["last_answer"],
            json!("not-answered")
        );
        watch.answers.insert(
            ("asked".to_owned(), "engine".to_owned()),
            Answer::NotReleased,
        );
        assert_eq!(
            watch.awaiting("asked")[0]["last_answer"],
            json!("not-released")
        );
    }

    /// The surface a held node raises names the **style** of each release it
    /// awaits, and a human-step wait carries the action a person needs.
    ///
    /// Read off the surface's own text, because that is the promise: a reader
    /// tells the two waits apart without opening a configuration file.
    #[test]
    fn a_wait_on_a_machine_and_a_wait_on_a_person_read_differently() {
        let mut watch = Watch::of_run(&RunPaths::under(std::path::Path::new("/nowhere"), "demo"));
        watch.dependencies.insert(
            "auto".to_owned(),
            vec![dependency(Some("crate"), Some(ReleaseStyle::Automated))],
        );
        watch.dependencies.insert(
            "person".to_owned(),
            vec![dependency(Some("wheel"), Some(ReleaseStyle::HumanStep))],
        );
        watch.answers.insert(
            ("auto".to_owned(), "engine".to_owned()),
            Answer::NotReleased,
        );
        watch.answers.insert(
            ("person".to_owned(), "engine".to_owned()),
            Answer::AwaitingHumanStep,
        );

        let automated = watch.wait_surface("auto").message;
        assert!(
            automated.contains("automated release") && !automated.contains("human-step"),
            "{automated}"
        );
        assert!(
            automated.contains("last answer: not-released"),
            "{automated}"
        );
        let person = watch.wait_surface("person").message;
        assert!(
            person.contains("human-step release — a person has to: cut a release on PyPI"),
            "{person}"
        );
        assert!(
            person.contains("last answer: awaiting-human-step"),
            "a wait on a person read as a probe that failed: {person}"
        );
        // Neither is a decision point: the hold is the scheduler's.
        for surface in [watch.wait_surface("auto"), watch.wait_surface("person")] {
            assert!(!surface.blocking, "a release wait held a subtree twice");
            assert_eq!(surface.kind, WAIT_SURFACE_KIND);
        }

        // The payload carries the same distinction, so the surface is not the
        // only place it exists.
        let entries = watch.awaiting("person");
        assert_eq!(entries[0]["style"], json!("human-step"));
        assert_eq!(entries[0]["last_answer"], json!("awaiting-human-step"));
        assert_eq!(entries[0]["action"], json!("cut a release on PyPI"));
        assert!(watch.awaiting("auto")[0].get("action").is_none());
    }

    /// A release that has arrived stays arrived, and the wait it ended is over:
    /// no question is put about it again, and its clock is dropped rather than
    /// left running under a later hold.
    #[test]
    fn a_release_that_arrived_is_never_awaited_again() {
        let published = Node {
            id: "held".to_owned(),
            adoption: Some(Adoption::Published),
            ..Node::default()
        };
        let paths = RunPaths::under(std::path::Path::new("/nowhere"), "demo");
        let mut watch = Watch::of_run(&paths);
        let key = ("held".to_owned(), "engine".to_owned());
        watch.dependencies.insert(
            "held".to_owned(),
            vec![dependency(Some("crate"), Some(ReleaseStyle::Automated))],
        );
        let watching = vec![published];

        // The hold, and the clock it is measured on, from an hour ago.
        let long_ago = crate::sys::now_millis().saturating_sub(4_394_000);
        watch.since.insert(key.clone(), long_ago);
        assert!(watch.held(&watching).contains("held"));

        // The release arrives, so the hold is over.
        watch.take_up(
            std::slice::from_ref(&key),
            &Answer::Released {
                version: "0.2.0".to_owned(),
            },
        );
        watch.refresh(&paths, &RunState::default(), &watching);
        assert!(watch.held(&watching).is_empty(), "a released hold held");
        assert!(
            watch.awaiting("held").is_empty(),
            "a released dependency is still on the awaited list"
        );
        assert!(
            !watch.since.contains_key(&key),
            "the clock of a wait that ended is still running"
        );

        // Then the probe stops answering, which is a statement about the probe
        // and not about the release.
        for gone in [Answer::NotAnswered, Answer::NotReleased, Answer::NotLanded] {
            watch.take_up(std::slice::from_ref(&key), &gone);
            assert_eq!(
                watch.answers.get(&key).and_then(Answer::version),
                Some("0.2.0"),
                "{gone:?} un-released a release that had happened"
            );
        }
        watch.refresh(&paths, &RunState::default(), &watching);
        assert!(
            watch.held(&watching).is_empty() && watch.awaiting("held").is_empty(),
            "a satisfied hold was resurrected by a probe that stopped answering"
        );
        // And what the wait would have reported is what made this urgent: an hour
        // and a quarter waited, for a node that had been dispatched.
        assert!(
            !watch.since.contains_key(&key),
            "the resurrected wait would have counted from the hold that ended"
        );

        // A hold this run opens **after** that one is timed from itself. The
        // clock a satisfied hold left behind is the one that reported 4394
        // seconds, so what proves it was dropped is a second wait on the same key
        // starting at zero.
        watch.answers.remove(&key);
        watch.refresh(&paths, &RunState::default(), &watching);
        let waited = watch.awaiting("held")[0]["waited_seconds"]
            .as_u64()
            .expect("a wait says how long it has been");
        assert!(
            waited < 60,
            "a second hold on the same node counted {waited}s from a hold that had ended"
        );
    }

    /// Only an answer of released lets a `published` node start, and a node the
    /// run cannot name a dependency for is not held at all.
    #[test]
    fn nothing_but_released_releases_a_hold() {
        let published = Node {
            id: "held".to_owned(),
            adoption: Some(Adoption::Published),
            ..Node::default()
        };
        let mut watch = Watch::of_run(&RunPaths::under(std::path::Path::new("/nowhere"), "demo"));
        watch.dependencies.insert(
            "held".to_owned(),
            vec![dependency(Some("crate"), Some(ReleaseStyle::Automated))],
        );
        let watching = vec![published.clone()];
        // No answer at all holds it, and so does every answer but one.
        assert!(watch.held(&watching).contains("held"));
        for answer in [
            Answer::NotReleased,
            Answer::AwaitingHumanStep,
            Answer::NotAnswered,
            Answer::NotLanded,
        ] {
            watch
                .answers
                .insert(("held".to_owned(), "engine".to_owned()), answer.clone());
            assert!(
                watch.held(&watching).contains("held"),
                "{answer:?} released the hold"
            );
        }
        watch.answers.insert(
            ("held".to_owned(), "engine".to_owned()),
            Answer::Released {
                version: "0.2.0".to_owned(),
            },
        );
        assert!(watch.held(&watching).is_empty());

        // A node with nothing outside its repository is never held, whatever it
        // declares — there is no release for it to be waiting on.
        watch.dependencies.insert("held".to_owned(), Vec::new());
        watch.answers.clear();
        assert!(watch.held(&watching).is_empty());
        // Nor is a `fast` node, ever.
        let fast = Node {
            adoption: Some(Adoption::Fast),
            ..published
        };
        watch.dependencies.insert(
            "held".to_owned(),
            vec![dependency(Some("crate"), Some(ReleaseStyle::Automated))],
        );
        assert!(watch.held(&[fast]).is_empty());
    }

    /// A dependency the run could not describe holds a `published` node exactly
    /// as an unreleased one does, and the hold names it — on the record, in the
    /// wait, and on the surface — with why it could not be read.
    ///
    /// The distinction this holds is between *no* out-of-repository dependency
    /// and one that could not be resolved this pass: the first is a node with
    /// nothing to wait on, the second was read as one and launched against work
    /// no release carried. Driven end to end in `tests/e2e/adoption.rs`; this is
    /// the seam, held against every reader of the marker.
    #[test]
    fn a_dependency_the_run_cannot_describe_holds_the_node_and_is_named() {
        let published = Node {
            id: "held".to_owned(),
            adoption: Some(Adoption::Published),
            ..Node::default()
        };
        let mut watch = Watch::of_run(&RunPaths::under(std::path::Path::new("/nowhere"), "demo"));
        let since = crate::sys::now_millis().saturating_sub(5_000);
        watch.unresolved.insert(
            "held".to_owned(),
            vec![Unresolved {
                dep: "engine".to_owned(),
                reason: "what the repository engine releases could not be read: malformed"
                    .to_owned(),
                since,
            }],
        );
        let watching = vec![published.clone()];
        assert!(watch.held(&watching).contains("held"));
        assert!(watch.names_a_release_dependency());
        assert_eq!(watch.awaited_deps("held"), vec!["engine".to_owned()]);

        let awaiting = watch.awaiting("held");
        assert_eq!(awaiting.len(), 1, "{awaiting:?}");
        assert_eq!(awaiting[0]["dep"], json!("engine"));
        assert_eq!(awaiting[0]["last_answer"], json!(UNRESOLVED));
        assert_eq!(
            awaiting[0]["reason"],
            json!("what the repository engine releases could not be read: malformed")
        );
        assert!(awaiting[0]["identity"].is_null() && awaiting[0]["target"].is_null());
        assert!(
            awaiting[0]["waited_seconds"]
                .as_u64()
                .is_some_and(|waited| waited >= 5),
            "the clock does not run from the first pass that found it unreadable: {awaiting:?}"
        );
        let surface = watch.wait_surface("held").message;
        assert!(
            surface.contains(
                "- engine — not yet resolved (what the repository engine releases \
                              could not be read: malformed), waited"
            ) && surface.contains("last answer: unresolved"),
            "{surface}"
        );

        // The marker holds nothing but a `published` node, and holds it no longer
        // once the dependency resolves.
        let fast = Node {
            adoption: Some(Adoption::Fast),
            ..published.clone()
        };
        assert!(watch.held(&[fast]).is_empty());
        watch.unresolved.remove("held");
        assert!(watch.held(&watching).is_empty());
        assert!(watch.awaited_deps("held").is_empty());
    }

    /// A fresh driver takes up what its predecessor already said, so a node is
    /// told **once** across a driver that died holding it.
    ///
    /// Read out of the journal, which is the only thing that outlives a driver.
    /// Without it the node a fresh driver finds still running is told a second
    /// time — a correction the worker has already acted on, arriving again with
    /// nothing to tell it from a new one.
    ///
    /// Held here rather than by a journey, and the reason is written down in
    /// `docs/contract-divergences.md` entry 40: a journey for it has to kill a
    /// driver mid-dispatch, adopt the run, and get a *second* node told before it
    /// can assert about the first. One was written and it is green on its own —
    /// and it timed out against the 120-second deadline on three of four runs of
    /// the instrumented suite, while holding e2e slots the rest of it needs. A
    /// test that reports the tree's health by how loaded the host was is worse
    /// than none.
    ///
    /// What the seeding *is*, is a fold of a durable record, and that is what this
    /// drives — both directions, so it holds that the seeding **narrowed** rather
    /// than silenced. The deliveries either side of it, into a live turn and onto
    /// the next dispatch, are driven end to end in `tests/e2e/adoption.rs`.
    #[test]
    fn a_fresh_driver_takes_up_what_its_predecessor_already_said() {
        let root = std::env::temp_dir().join(format!("op-release-seed-{}", std::process::id()));
        let paths = RunPaths::under(&root, "restarted");
        std::fs::create_dir_all(&paths.dir).expect("a scratch run directory");
        let record = |kind: journal::PipelineKind, node: &str, payload: Value| {
            serde_json::json!({
                "v": 1,
                "ts": "2026-08-24T00:00:00.000Z",
                "stream": "predecessor",
                "seq": 0,
                "source": "pipeline",
                "kind": kind.as_str(),
                "labels": {"run_id": "restarted", "node": node},
                "payload": payload,
            })
            .to_string()
        };
        std::fs::write(
            paths.journal(),
            format!(
                "{}\n{}\n",
                record(
                    journal::PipelineKind::ReleaseAdopted,
                    "told",
                    json!({
                        "node": "told",
                        "delivery": "live",
                        "versions": [{
                            "identity": "github.com/owner/engine",
                            "target": "crate",
                            "version": "0.2.0"
                        }]
                    }),
                ),
                record(
                    journal::PipelineKind::ReleaseArrived,
                    "told",
                    json!({"node": "told", "dep": "engine"}),
                ),
            ),
        )
        .expect("the predecessor's journal is written");

        let mut watch = Watch::of_run(&paths);
        watch.dependencies.insert(
            "told".to_owned(),
            vec![dependency(Some("crate"), Some(ReleaseStyle::Automated))],
        );
        watch.answers.insert(
            ("told".to_owned(), "engine".to_owned()),
            Answer::Released {
                version: "0.2.0".to_owned(),
            },
        );
        let running = vec![Node {
            id: "told".to_owned(),
            adoption: Some(Adoption::Fast),
            ..Node::default()
        }];
        assert!(
            watch.ready_to_adopt(&running).is_empty(),
            "a fresh driver told a node its releases had arrived a second time"
        );
        assert!(
            watch
                .arrived
                .contains(&("told".to_owned(), "engine".to_owned())),
            "a fresh driver did not take up the arrival its predecessor reported"
        );

        // A record naming no release this build can read says nothing about what
        // its node was told, so it suppresses nothing.
        std::fs::write(
            paths.journal(),
            format!(
                "{}\n",
                record(
                    journal::PipelineKind::ReleaseAdopted,
                    "told",
                    json!({"node": "told", "delivery": "live", "versions": [{"identity": ""}]}),
                ),
            ),
        )
        .expect("the predecessor's journal is written");
        assert!(
            Watch::of_run(&paths).adopted.is_empty(),
            "an unreadable record suppressed a delivery nothing can say happened"
        );

        // A node its predecessor never told is told, which is what says the
        // seeding narrowed rather than silenced.
        let fresh = vec![Node {
            id: "fresh".to_owned(),
            adoption: Some(Adoption::Fast),
            ..Node::default()
        }];
        watch.dependencies.insert(
            "fresh".to_owned(),
            vec![dependency(Some("crate"), Some(ReleaseStyle::Automated))],
        );
        watch.answers.insert(
            ("fresh".to_owned(), "engine".to_owned()),
            Answer::Released {
                version: "0.2.0".to_owned(),
            },
        );
        assert_eq!(watch.ready_to_adopt(&fresh).len(), 1);
        std::fs::remove_dir_all(&root).ok();
    }

    /// The two bounds fall back rather than to zero or to no bound at all.
    #[test]
    fn an_unusable_bound_falls_back_to_the_shipped_one() {
        for (key, read) in [
            (POLL_ENV, poll_seconds as fn() -> u64),
            (SURFACE_ENV, surface_every_seconds as fn() -> u64),
        ] {
            let shipped = read();
            for unusable in ["0", "", "soon", "-1"] {
                std::env::set_var(key, unusable);
                assert_eq!(read(), shipped, "{key}={unusable:?}");
            }
            std::env::set_var(key, "7");
            assert_eq!(read(), 7);
            std::env::remove_var(key);
        }
        assert_eq!(poll_seconds(), DEFAULT_POLL_SECONDS);
        assert_eq!(surface_every_seconds(), DEFAULT_SURFACE_SECONDS);
    }

    /// The shipped probe interval is inside the bound this loop promises for
    /// every other answer it owes on a clock.
    ///
    /// Read through the same function a driver reads it through rather than off
    /// the constant, so a build that shipped a longer default fails here whether
    /// it moved the constant or the fallback around it. What holds the *rate* end
    /// to end — that a held run really is asked about no oftener than this,
    /// however fast its loop runs — is
    /// `adoption::a_held_release_is_asked_about_on_its_own_interval_however_fast_the_loop_runs`.
    #[test]
    fn the_shipped_probe_interval_is_no_longer_than_the_loop_promises() {
        std::env::remove_var(POLL_ENV);
        let shipped = Duration::from_secs(poll_seconds());
        assert!(
            shipped <= Duration::from_secs(60),
            "the shipped probe interval is {shipped:?}, which is longer than the minute a held \
             node is promised for every other answer this loop owes on a clock"
        );
    }

    /// The e2e suite carries its own copy of [`DEFAULT_POLL_SECONDS`], because
    /// this module is private and the suite drives the compiled binary from
    /// outside the crate. This is the gate that keeps that copy exact: it reads
    /// the literal out of the suite's own source and fails on drift in either
    /// direction, which neither the suite's ceiling assertion nor the bound
    /// above can see — both are `<=`, so a shipped value that *shrank* would
    /// pass both while the suite went on asserting a bound this build no longer
    /// ships.
    #[test]
    fn the_suites_copy_of_the_shipped_probe_interval_is_this_one() {
        let suite = include_str!("../tests/e2e/adoption.rs");
        let declaration = "const SHIPPED_POLL_SECONDS: u64 = ";
        let start = suite
            .find(declaration)
            .expect("tests/e2e/adoption.rs declares SHIPPED_POLL_SECONDS")
            + declaration.len();
        let copied: u64 = suite[start..]
            .split(';')
            .next()
            .expect("the declaration ends in a semicolon")
            .trim()
            .parse()
            .expect("SHIPPED_POLL_SECONDS is a plain integer literal");
        assert_eq!(
            copied, DEFAULT_POLL_SECONDS,
            "tests/e2e/adoption.rs holds a real build to a probe interval of {copied}s, but this \
             build ships {DEFAULT_POLL_SECONDS}s"
        );
    }

    /// The divergence record states this interval to the planner who owns the contract,
    /// in prose, in both of the places it describes a paced read.
    ///
    /// Prose is where a copy rots unnoticed, so it is reconciled here rather than read:
    /// every "N seconds by default" that record states is this constant. Stated as *every*
    /// occurrence rather than as a count, so a third place that describes the interval is
    /// covered the day it is written instead of failing this as a miscount.
    #[test]
    fn the_divergence_records_copies_of_the_shipped_probe_interval_are_this_one() {
        let record = include_str!("../docs/contract-divergences.md");
        let phrase = " seconds by default";
        let stated: Vec<u64> = record
            .match_indices(phrase)
            .map(|(at, _)| {
                record[..at]
                    .rsplit(|c: char| !c.is_ascii_digit())
                    .next()
                    .filter(|digits| !digits.is_empty())
                    .unwrap_or_else(|| {
                        panic!("docs/contract-divergences.md states \"{phrase}\" after no number")
                    })
                    .parse()
                    .expect("the interval the divergence record states is a plain integer")
            })
            .collect();
        assert!(
            !stated.is_empty(),
            "docs/contract-divergences.md no longer states the probe interval at all, so this \
             gate is reconciling nothing"
        );
        for interval in stated {
            assert_eq!(
                interval, DEFAULT_POLL_SECONDS,
                "docs/contract-divergences.md tells the contract's owner this build polls \
                 every {interval}s, but it ships {DEFAULT_POLL_SECONDS}s"
            );
        }
    }
}