onepipeline 0.44.2

//! The committed contract drives the public types.
//!
//! Every fixture here is read out of `docs/contract.md` at compile time rather
//! than copied beside it, so the document and this crate's wire shapes cannot
//! drift: edit one without the other and this suite fails.
//!
//! The contract's Rust block is an interface sketch, not compilable source, so
//! it is used the way the document uses it — as the naming authority. Each test
//! below builds the real value with the compiler proving the fields exist, then
//! asserts the document still names them.

use std::collections::BTreeSet;
use std::num::{NonZeroU32, NonZeroU64};
use std::path::{Path, PathBuf};

use clap::{CommandFactory, Parser};
use oneagentgraph::config::{ConfigRef, GraphConfig, JudgeSide, Member};
use oneagentgraph::persona::{merge, Persona};
use onepipeline::channel::{
    allows, Author, Command as Edit, Dependents, Reply, Surface, SurfaceKind,
};
use onepipeline::cli::{
    Cli, Command, DAG_GRAPH_OFF, DEFAULT_DISPATCH_ENV_HOOK_TIMEOUT_SECONDS,
    DEFAULT_HEARTBEAT_INTERVAL_SECONDS, DEFAULT_HOOK_TIMEOUT_SECONDS,
    DEFAULT_WRITEBACK_ITEM_BUDGET_SECONDS, WRITEBACK_CLASSIFIED_COMMANDS,
    WRITEBACK_COMMAND_FLOOR_SECONDS, WRITEBACK_FAILURE_CLASS_MEMBER, WRITEBACK_FAILURE_EXIT,
    WRITEBACK_ITEM_BUDGET_ENV, WRITEBACK_MEMBERS_FROM, WRITEBACK_MEMBER_READ,
    WRITEBACK_PARTIAL_CLASS_MEMBER, WRITEBACK_PARTIAL_EXIT, WRITEBACK_PROJECTIONS_FILE,
    WRITEBACK_REFUSED_CLASS, WRITEBACK_STORE_FILE,
};
use onepipeline::controls::NodeControls;
use onepipeline::error::{
    EXIT_NODE_SETTLED, EXIT_NOTHING_DRIVING, EXIT_QUEUED, EXIT_REFUSED, EXIT_SUCCESS,
    EXIT_SURFACE_WAITING, EXIT_WATCH_ELAPSED,
};
use onepipeline::event::{
    ArtifactId, ArtifactRef, Envelope, EventKind, Labels, Phase, PipelineKind, Source,
    ENVELOPE_VERSION, ENVELOPE_VERSIONS_READ, PIPELINE_KINDS,
};
use onepipeline::executor::{
    CancelMode, CancellationToken, Capabilities, CapacityReport, DispatchRequest, Executor,
    LocalExecutor, WorkspaceSpec,
};
use onepipeline::filter::{
    EventFilter, Filters, LaunchConfig, Matcher, LAUNCH_CONFIG_SCHEMA_VERSION,
    LAUNCH_CONFIG_SCHEMA_VERSIONS_READ,
};
use onepipeline::note::{Addressee, Delivered, Note, Party, Reached};
use onepipeline::plan::{
    adoption_instructions, arrival_note, CrossRepoReference, Node, NodeKind, Plan, RepoType,
    Resume, Step, Workflow, ADOPTION_INSTRUCTION_VARIABLES, AMENDMENT_HEADING,
    AMENDMENT_PRECEDENCE, CROSS_REPO_REFERENCES_HEADING, DEFAULT_ADOPTION_INSTRUCTION,
    OBSERVED_STATE, PLANNER_CONTEXT_HEADING, PLAN_SCHEMA_VERSION, PLAN_SCHEMA_VERSIONS_READ,
};
use onepipeline::report::{
    retain, ACCEPTED_REPORT_FILE, MAX_REPORT_BYTES, MEMBER_SETTLED, REPORT_PATH,
};
use onepipeline::rules::{ExecutorKind, ExecutorRules, Predicate};
use onepipeline::verbs;
use onepipeline::views::{
    FailureClass, Listing, NodeLanding, ProjectGroup, ProjectionActions, ProjectionEnded,
    ProjectionFailure, ProjectionRecord, ProjectionScope, Projects, RunPaths, RunSummary,
    RunTelemetry, WholeBecause, GROUP_HEADER, NO_PROJECT, SUMMARY_SCHEMA_VERSION,
};
use onevcs::{Adoption, MergePolicy, SessionRequest};
use serde_json::{json, Value};

/// The approved contract itself.
const CONTRACT: &str = include_str!("../docs/contract.md");

/// The repository root, so a test can open the shipped content the contract
/// names. `CARGO_MANIFEST_DIR` is the crate root, which here is the repo root.
fn repo_root() -> PathBuf {
    PathBuf::from(env!("CARGO_MANIFEST_DIR"))
}

/// The fenced blocks in the contract carrying the given info string.
///
/// The scanner tracks whether it is *inside* a block rather than matching the
/// opening fence, so a closing fence never reads as an unlabelled opening one.
fn fenced_blocks(language: &str) -> Vec<String> {
    let mut blocks = Vec::new();
    let mut open: Option<String> = None;
    let mut body = String::new();
    for line in CONTRACT.lines() {
        match &open {
            Some(info) => {
                if line.trim_end() == "```" {
                    if info == language {
                        blocks.push(std::mem::take(&mut body));
                    }
                    body.clear();
                    open = None;
                } else {
                    body.push_str(line);
                    body.push('\n');
                }
            }
            None => {
                if let Some(info) = line.trim_end().strip_prefix("```") {
                    open = Some(info.trim().to_string());
                    body.clear();
                }
            }
        }
    }
    assert!(open.is_none(), "unterminated ``` block in docs/contract.md");
    blocks
}

/// The one fenced block of that language whose body names `needle`.
///
/// The contract carries more than one example in a given language, so a fixture
/// says which of them it is about by naming a key only that one has — rather
/// than by an index, which a block inserted above it would silently shift onto
/// the wrong example.
fn fenced_block_naming(language: &str, needle: &str) -> String {
    let mut matching: Vec<String> = fenced_blocks(language)
        .into_iter()
        .filter(|body| body.contains(needle))
        .collect();
    assert_eq!(
        matching.len(),
        1,
        "expected exactly one ```{language} block naming {needle:?} in docs/contract.md, found {}",
        matching.len()
    );
    matching.pop().expect("one block")
}

/// Every `` `backticked` `` token in the contract.
fn backticked() -> BTreeSet<String> {
    let mut out = BTreeSet::new();
    let mut rest = CONTRACT;
    while let Some(open) = rest.find('`') {
        rest = &rest[open + 1..];
        let Some(close) = rest.find('`') else { break };
        out.insert(rest[..close].to_string());
        rest = &rest[close + 1..];
    }
    out
}

/// Assert the contract names each of these, so a document edit that drops one
/// fails here rather than leaving the surface unproven.
fn assert_contract_names(what: &str, names: &[&str]) {
    for name in names {
        assert!(
            CONTRACT.contains(name),
            "docs/contract.md no longer names the {what} `{name}`"
        );
    }
}

#[test]
fn the_contracts_rules_example_parses_and_round_trips() {
    let yaml = fenced_block_naming("yaml", "executors:");
    let rules: ExecutorRules = serde_norway::from_str(&yaml).expect("the rules example parses");

    assert_eq!(
        rules.executors.len(),
        1,
        "the example declares one executor"
    );
    let local = &rules.executors[0];
    assert_eq!(local.name, "local");
    assert_eq!(local.kind, ExecutorKind::Local);
    assert_eq!(local.max_load1, Some(8.0));
    assert_eq!(
        local.min_free_mem.as_deref(),
        Some("2GiB"),
        "the size is carried as the contract writes it"
    );

    assert_eq!(
        rules.rules.len(),
        2,
        "the example declares two ordered rules"
    );
    assert_eq!(
        rules.rules[0].when,
        Some(Predicate {
            executor_has_capacity: Some("local".into()),
            ..Predicate::default()
        }),
        "the first rule tests capacity"
    );
    assert_eq!(rules.rules[0].use_executor, "local");
    assert_eq!(
        rules.rules[1].when, None,
        "the last rule is the unconditional fallback"
    );
    assert_eq!(rules.rules[1].use_executor, "local");

    let round_tripped: ExecutorRules =
        serde_norway::from_str(&serde_norway::to_string(&rules).expect("serializes"))
            .expect("re-parses");
    assert_eq!(round_tripped, rules);
}

#[test]
fn the_contract_states_both_predicate_families_and_what_each_matches_on() {
    let prose = CONTRACT.split_whitespace().collect::<Vec<_>>().join(" ");
    assert!(
        prose.contains("`executor_has_capacity: NAME` matches on **capacity**"),
        "the contract no longer says what the capacity family matches on"
    );
    assert!(
        prose.contains("`node_label: {KEY: VALUE, ...}` matches on the **node's labels**"),
        "the contract no longer says what the label family matches on"
    );
    assert!(
        prose.contains("Several conditions in one `when` conjoin"),
        "the contract no longer says how two conditions in one `when` combine"
    );

    // The keys the contract lists are the keys the grammar accepts. Both halves
    // of that sentence are gated: a key the code accepts and the contract does
    // not name is undocumented surface, and the other way round is a promise
    // nothing keeps.
    for key in onepipeline::rules::SELECTABLE_LABELS {
        assert!(
            prose.contains(&format!("`{key}`")),
            "the contract does not name the selectable label `{key}`"
        );
    }
    let rules: ExecutorRules = serde_norway::from_str(
        "executors: [{name: local, type: local}]\n\
         rules: [{when: {node_label: {step: implement}}, use: local}]\n",
    )
    .expect("it parses");
    let err = rules
        .validate()
        .expect_err("`step` is not a key the contract lists");
    assert!(err.to_string().contains("step"), "{err}");
}

#[test]
fn an_unknown_rules_key_is_refused_at_the_boundary() {
    let bad = "executors:\n  - {name: local, type: local, mx_load1: 8.0}\nrules:\n  - use: local\n";
    let err = serde_norway::from_str::<ExecutorRules>(bad)
        .expect_err("a mistyped key is rejected, not silently dropped");
    assert!(
        err.to_string().contains("mx_load1"),
        "the error names the offending key: {err}"
    );
}

#[test]
fn the_shipped_rules_example_is_the_contracts_own() {
    let shipped = std::fs::read_to_string(repo_root().join("examples/executors.yaml"))
        .expect("examples/executors.yaml ships");
    let shipped: ExecutorRules = serde_norway::from_str(&shipped).expect("it parses");
    let documented: ExecutorRules =
        serde_norway::from_str(&fenced_block_naming("yaml", "executors:"))
            .expect("the contract's example parses");
    assert_eq!(
        shipped, documented,
        "the shipped executor-rules example must be the contract's own"
    );
}

#[test]
fn the_dispatch_request_carries_every_field_the_contract_declares() {
    let request = DispatchRequest {
        graph: ConfigRef("./graphs/node-scope.yaml".into()),
        task: "## What\nDo the thing.".into(),
        labels: Labels {
            run_id: Some("run-1".into()),
            round: Some(2),
            node: Some("service".into()),
            step: Some("implement".into()),
            persona: Some("engineer".into()),
            ..Labels::default()
        },
        controls: NodeControls {
            max_turns: NonZeroU32::new(24),
        },
        workspace: WorkspaceSpec::VcsSession(SessionRequest {
            repo: "nickderobertis/some-service".into(),
            branch: None,
            base: None,
            execution_checkout: None,
            pool: None,
            overflow: None,
            labels: Default::default(),
        }),
        cancel: CancellationToken::new(),
        attempt: NonZeroU32::new(2).expect("two is an attempt"),
    };

    assert_contract_names(
        "DispatchRequest field",
        &[
            "graph",
            "task",
            "labels",
            "controls",
            "workspace",
            "cancel",
            "attempt",
        ],
    );
    assert_eq!(
        request.attempt.get(),
        2,
        "the request carries which attempt of the node it is"
    );
    assert_eq!(
        request.controls.max_turns,
        NonZeroU32::new(24),
        "the request carries the node's own controls, not only its labels"
    );
    assert_contract_names(
        "reserved label",
        &["run_id", "round", "node", "step", "persona"],
    );
    assert_contract_names(
        "WorkspaceSpec variant",
        &["Path(PathBuf)", "VcsSession(SessionRequest"],
    );

    // The contract's `WorkspaceSpec::VcsSession` means the machine running the
    // dispatch opens the session, so the request carries the *ask*, never an
    // already-opened session.
    match &request.workspace {
        WorkspaceSpec::VcsSession(session) => {
            assert_eq!(session.repo, "nickderobertis/some-service");
        }
        WorkspaceSpec::Path(path) => panic!("built a VcsSession, got a path: {}", path.display()),
    }

    let local = WorkspaceSpec::Path(Path::new("/tmp/work").to_path_buf());
    assert_ne!(local, request.workspace);
}

#[test]
fn the_local_executor_is_the_one_v1_ships_and_takes_both_workspaces() {
    let local = LocalExecutor;
    assert_eq!(local.name(), "local");
    assert_eq!(
        local.capabilities(),
        Capabilities { vcs_sessions: true },
        "the contract says LocalExecutor supports both WorkspaceSpec variants"
    );
    assert!(CONTRACT.contains("v1 ships `LocalExecutor` only (supports both variants)"));
}

#[test]
fn the_local_executors_capacity_reports_the_three_numbers_the_contract_names() {
    // A rules file selects on these, so each has to be a number a predicate can
    // compare. Every unreadable input resolves toward "has capacity" rather than
    // toward a zero that would stall a healthy host.
    let report = LocalExecutor.capacity();
    assert!(
        report.load1.is_finite() && report.load1 >= 0.0,
        "{report:?}"
    );
    assert!(report.mem_free_bytes > 0, "{report:?}");
    assert_ne!(report, CapacityReport::default(), "nothing was probed");
    assert_contract_names(
        "CapacityReport field",
        &["slots_free", "load1", "mem_free_bytes"],
    );
}

/// Serialises every test below that sets one of the process-global variables the
/// `oneagentgraph` seam reads — `ONEPIPELINE_ONEAGENTGRAPH_BIN`, which names the
/// executable, and `ONEAGENTGRAPH_STATE_DIR`, which names where its runs are
/// recorded.
///
/// nextest — the runner this repository uses — gives each test its own process,
/// so under it neither variable reaches anything else. Plain `cargo test` runs a
/// binary's tests as *threads of one process*, and under that runner one test's
/// value decides another's assertion:
/// `dispatching_goes_through_the_oneagentgraph_seam_and_says_so_when_it_cannot`
/// points the seam at an executable that does not exist, and
/// `a_dispatch_built_outside_a_run_still_carries_its_controls_into_the_launch`
/// then launches *that* instead of the real sibling and is refused for the wrong
/// reason. Observed, one run in three. The lock costs nothing under nextest and
/// makes both runners say the same thing.
static SEAM_ENV: std::sync::Mutex<()> = std::sync::Mutex::new(());

/// Held for the length of a test that sets one of those variables. A poisoned
/// lock is recovered rather than propagated: the test that panicked holding it
/// has already failed, and refusing to run the next one would report a second
/// failure belonging to nobody.
fn seam_env_lock() -> std::sync::MutexGuard<'static, ()> {
    SEAM_ENV
        .lock()
        .unwrap_or_else(|poisoned| poisoned.into_inner())
}

#[test]
fn dispatching_goes_through_the_oneagentgraph_seam_and_says_so_when_it_cannot() {
    // The seam is a subprocess boundary: this crate composes `oneagentgraph`
    // rather than reimplementing it. Pointed at an executable that does not
    // exist, the failure names that sibling instead of reading as a node the
    // agent failed.
    //
    // The seam is *named* rather than left to `PATH`: `oneagentgraph` is a
    // published CLI, so a host that has it installed would otherwise make this
    // assertion depend on whose machine it ran on. The variable is this whole
    // process's, so it is set under [`seam_env_lock`] and removed again — a
    // sibling test launches the real executable through it.
    let _seam = seam_env_lock();
    std::env::set_var(
        "ONEPIPELINE_ONEAGENTGRAPH_BIN",
        "oneagentgraph-that-is-not-installed",
    );
    // `Box<dyn DispatchHandle>` is not `Debug`, so the success arm is destructured
    // rather than unwrapped.
    let Err(err) = LocalExecutor.dispatch(DispatchRequest {
        graph: ConfigRef("./graphs/node-scope.yaml".into()),
        task: "anything".into(),
        labels: Labels::default(),
        controls: NodeControls::default(),
        workspace: WorkspaceSpec::Path(PathBuf::from(".")),
        cancel: CancellationToken::new(),
        attempt: NonZeroU32::MIN,
    }) else {
        panic!("no `oneagentgraph` is installed here, so the dispatch cannot start");
    };
    let message = err.to_string();
    assert!(
        message.contains("oneagentgraph"),
        "the seam is unnamed: {message}"
    );
    // Put back what this test pointed at an executable that is not there, so it
    // decides nothing but its own assertion.
    std::env::remove_var("ONEPIPELINE_ONEAGENTGRAPH_BIN");
}

/// The `filters:` block in the contract is a block this crate's own types read.
///
/// Driven out of the document, like every other fixture here: the grammar is
/// `onemessagebus`'s, re-exported, and this document keeps a marked copy of it, so
/// a bus release whose grammar stopped matching that text fails this gate.
#[test]
fn the_contracts_launch_config_example_parses_and_round_trips() {
    let yaml = fenced_block_naming("yaml", "schema_version: 2");
    let config: LaunchConfig = serde_norway::from_str(&yaml).expect("the launch config parses");
    // A version this build **reads**, which is what the contract's committed
    // example is: the document is approved verbatim and is never edited to
    // follow the schema, so a later version having been written since does not
    // stop the example being a config an operator still has on disk.
    assert!(
        LAUNCH_CONFIG_SCHEMA_VERSIONS_READ.contains(&config.schema_version),
        "the contract's example declares a version this build does not read: {}",
        config.schema_version
    );
    assert_eq!(
        config.pr_author_graph.as_deref(),
        Some("./graphs/pr-author.yaml"),
        "the contract's example declares the launch's other decision and this build \
         does not read it"
    );
    let filters = config.filters;

    let agentgraph = filters
        .agentgraph
        .as_ref()
        .expect("it names a source filter");
    assert_eq!(agentgraph.include, Vec::new(), "an absent include is empty");
    assert_eq!(agentgraph.exclude.len(), 1);
    assert_eq!(agentgraph.exclude[0].kind.as_deref(), Some("turn-activity"));
    let vcs = filters.vcs.as_ref().expect("it names a vcs filter");
    assert_eq!(vcs.include.len(), 2);
    assert_eq!(vcs.include[0].kind.as_deref(), Some("gate-*"));

    // The two shipped profiles, exactly as the contract states them: an
    // override that changed either would be a run whose default view is not the
    // documented one.
    assert_eq!(
        filters.profiles["planner"],
        EventFilter {
            include: vec![Matcher {
                source: Some(Source::Pipeline),
                ..Matcher::default()
            }],
            exclude: Vec::new(),
        }
    );
    assert_eq!(filters.profiles["detailed"], EventFilter::default());

    let round_tripped: Filters =
        serde_json::from_str(&serde_json::to_string(&filters).expect("serializes"))
            .expect("re-parses");
    assert_eq!(round_tripped, filters);

    // The checked-in golden **is** the contract's own example in every
    // respect but one name. Two documents that both claim to pin the launch
    // config's shape and could disagree would be two sources; this is the one
    // place they are held to being one. The one name: the golden is a recording,
    // written when the unfiltered profile shipped as `monitor`, and a recording
    // is never renamed — so it declares its unfiltered override under that
    // retired name, where the contract's example declares it under `detailed`.
    // Same filter, same shape, the operator's own name for it in each.
    let golden: LaunchConfig = serde_json::from_str(
        &std::fs::read_to_string(repo_root().join("tests/golden/launch-config-v2.json"))
            .expect("the golden ships"),
    )
    .expect("the golden parses");
    let mut recorded = golden.filters.clone();
    let retired = recorded
        .profiles
        .remove("monitor")
        .expect("the golden records an override under the name the profile shipped as then");
    assert_eq!(retired, EventFilter::default());
    let mut stated = filters.clone();
    let detailed = stated
        .profiles
        .remove("detailed")
        .expect("the contract's example overrides the shipped `detailed` profile");
    assert_eq!(detailed, EventFilter::default());
    assert_eq!(
        (golden.schema_version, recorded, golden.pr_author_graph),
        (config.schema_version, stated, config.pr_author_graph),
        "tests/golden/launch-config-v2.json and the contract's own example are \
         different documents beyond the recorded override's name"
    );

    // The version before it is still a document this build reads, and it ships
    // as its own golden: the bump is additive, and that promise is to every
    // config already written beside a plan.
    let earlier: LaunchConfig = serde_json::from_str(
        &std::fs::read_to_string(repo_root().join("tests/golden/launch-config-v1.json"))
            .expect("the earlier golden ships"),
    )
    .expect("the earlier golden parses");
    assert_eq!(earlier.schema_version, 1);
    assert_eq!(
        earlier.pr_author_graph, None,
        "the earlier golden carries a key that version never had"
    );
    assert!(
        CONTRACT.contains("a version-1 config is a complete document this build still reads"),
        "the contract no longer says an earlier launch config still reads"
    );
}

/// A launch config declaring only its version is a launch that says nothing.
///
/// The contract's promise to a document written before the block existed, and to
/// one that never wanted it: the block is optional, an empty one is omitted from
/// what this crate writes, and neither is an error.
#[test]
fn the_contracts_launch_config_omits_an_empty_block_and_still_reads() {
    for version in LAUNCH_CONFIG_SCHEMA_VERSIONS_READ {
        let bare: LaunchConfig = serde_norway::from_str(&format!("schema_version: {version}\n"))
            .expect("a config may declare only a version");
        assert!(bare.filters.is_empty());
        assert_eq!(bare.pr_author_graph, None);
        assert_eq!(bare.node_validator, None);
    }
    // And what this crate *writes* for one is the version alone: every optional
    // key is omitted, so a launch that declared none is a document an earlier
    // reader accepts.
    assert_eq!(
        serde_json::to_string(&LaunchConfig::default()).expect("serializes"),
        format!(r#"{{"schema_version":{LAUNCH_CONFIG_SCHEMA_VERSION}}}"#),
        "an empty filters block, an absent drafting graph, or an absent node \
         validator was written out"
    );
    assert_contract_names(
        "launch config surface",
        &["--launch-config FILE", "schema_version: 2"],
    );
}

/// The shipped defaults are the contract's, and both are overridable by name.
#[test]
fn the_shipped_profiles_are_the_contracts_own_and_are_overridable() {
    let empty = Filters::default();
    assert_eq!(
        empty.profile("planner").expect("planner ships"),
        EventFilter {
            include: vec![Matcher {
                source: Some(Source::Pipeline),
                ..Matcher::default()
            }],
            exclude: Vec::new(),
        }
    );
    assert_eq!(
        empty.profile("detailed").expect("detailed ships"),
        EventFilter::default(),
        "the shipped detailed profile is unfiltered"
    );
    // The name the profile shipped under before it was `detailed` is no alias
    // of it: a launch that declares no profile of that name has none.
    let retired = empty
        .profile("monitor")
        .expect_err("the retired profile name is not an alias")
        .to_string();
    assert!(
        retired.contains("'monitor' is not a filter profile") && retired.contains("detailed"),
        "{retired}"
    );

    let mine = EventFilter::parse(r#"{"include": [{"kind": "node-*"}]}"#).expect("a filter");
    let overridden = Filters {
        profiles: [
            ("planner".to_string(), mine.clone()),
            ("detailed".to_string(), mine.clone()),
        ]
        .into_iter()
        .collect(),
        ..Filters::default()
    };
    assert_eq!(overridden.profile("planner").expect("overridden"), mine);
    assert_eq!(overridden.profile("detailed").expect("overridden"), mine);

    let unknown = empty
        .profile("planer")
        .expect_err("a profile this run does not have is refused");
    let said = unknown.to_string();
    assert!(said.contains("planer"), "{said}");
    assert!(
        said.contains("planner") && said.contains("detailed"),
        "{said}"
    );
}

/// The grammar's refusal semantics, at the boundary a spec crosses.
#[test]
fn a_filter_spec_is_refused_by_the_shared_grammars_own_rules() {
    let unknown_field = EventFilter::parse(r#"{"include": [{"role": "agent"}]}"#)
        .expect_err("a matcher field the grammar does not have is refused");
    let said = unknown_field.to_string();
    assert!(said.contains("role"), "the refusal names the field: {said}");
    assert!(
        said.contains("include") && said.contains('1'),
        "the refusal says which list and where in it: {said}"
    );

    let stray = EventFilter::parse(r#"{"includes": []}"#)
        .expect_err("a filter names include and exclude and nothing else");
    assert!(stray.to_string().contains("includes"), "{stray}");

    // `round` is a reserved label the approved matcher list does not name, so it
    // is refused here like any other non-field rather than quietly accepted.
    let deprecated = EventFilter::parse(r#"{"include": [{"round": "1"}]}"#)
        .expect_err("`round` is not in the grammar");
    assert!(deprecated.to_string().contains("round"), "{deprecated}");

    // Both refusals are at the one boundary: a spec reaches this crate from a
    // command line, from a file, and from the launch record every later read
    // opens, and a filter checked only where an operator typed it would be a
    // record that could be edited into a matcher this build says it will not
    // honour — and then honoured.
    let empty_matcher = EventFilter::parse(r#"{"exclude": [{}]}"#)
        .expect_err("a matcher naming no field matches everything");
    assert!(
        empty_matcher.to_string().contains("exclude"),
        "{empty_matcher}"
    );

    let empty_field = EventFilter::parse(r#"{"include": [{"kind": ""}]}"#)
        .expect_err("nothing on the stream carries an empty kind");
    assert!(empty_field.to_string().contains("kind"), "{empty_field}");

    // The launch record is that boundary too, and it is the one an operator
    // never typed at: a block edited into a matcher naming nothing is refused
    // where the record is read.
    let record = serde_json::from_str::<Filters>(r#"{"vcs": {"exclude": [{}]}}"#)
        .expect_err("a launch record carrying an unusable filter is refused");
    assert!(record.to_string().contains("exclude"), "{record}");
}

/// Every property a JSON Schema document declares for its object, following the
/// `$ref`s and `allOf`s a flattened field is composed through.
fn property_names(node: &Value, root: &Value) -> BTreeSet<String> {
    let mut names = BTreeSet::new();
    if let Some(properties) = node.get("properties").and_then(Value::as_object) {
        names.extend(properties.keys().cloned());
    }
    if let Some(target) = node
        .get("$ref")
        .and_then(Value::as_str)
        .and_then(|reference| reference.strip_prefix('#'))
        .and_then(|pointer| root.pointer(pointer))
    {
        names.extend(property_names(target, root));
    }
    if let Some(parts) = node.get("allOf").and_then(Value::as_array) {
        for part in parts {
            names.extend(property_names(part, root));
        }
    }
    names
}

/// Every `` `backticked` `` token in one passage of the contract.
fn backticked_in(passage: &str) -> BTreeSet<String> {
    passage
        .split('`')
        .skip(1)
        .step_by(2)
        .map(str::to_owned)
        .collect()
}

/// The contract's envelope and grammar paragraphs are a **marked copy** of
/// `onemessagebus`'s contract, so they are reconciled against the released bus
/// itself, not only against fixtures written here: a bus release that adds a
/// matcher field or a source, or a lock that moves to a release the copy was not
/// decided against, fails this gate instead of leaving the copy stale in silence.
#[test]
fn the_marked_copy_of_the_bus_contract_is_reconciled_against_the_released_bus() {
    // The release the copy was decided against is the one this build links.
    let decided = CONTRACT
        .split_once("decided against `onemessagebus` ")
        .and_then(|(_, rest)| {
            rest.split(|c: char| !(c.is_ascii_digit() || c == '.'))
                .next()
        })
        .map(|version| version.trim_end_matches('.'))
        .filter(|version| !version.is_empty())
        .expect("the contract names the bus release its marked copy was decided against");
    let lock = std::fs::read_to_string(repo_root().join("Cargo.lock")).expect("the lock ships");
    for bus in ["onemessagebus", "onemessagebus-agent"] {
        let resolved: Vec<&str> = lock
            .split("[[package]]")
            .filter(|package| {
                package
                    .lines()
                    .any(|line| line == format!("name = \"{bus}\""))
            })
            .filter_map(|package| {
                package
                    .lines()
                    .find_map(|line| line.strip_prefix("version = \""))
                    .map(|version| version.trim_end_matches('"'))
            })
            .collect();
        assert_eq!(
            resolved,
            vec![decided],
            "docs/contract.md's copy was decided against `onemessagebus` {decided}, and the lock \
             resolves {bus} at {resolved:?}: re-decide the copy against the linked release"
        );
    }

    // The matcher fields the grammar paragraph names are exactly the fields the
    // bus's own matcher document declares.
    let grammar = CONTRACT
        .lines()
        .find(|line| line.starts_with("**Filtering is owned by the stream's source.**"))
        .expect("the contract states the filter grammar");
    let fields = grammar
        .split_once("A matcher's fields are all optional")
        .and_then(|(_, rest)| rest.split_once("An absent or empty"))
        .map(|(clause, _)| clause)
        .expect("the grammar paragraph lists the matcher's fields");
    let named: BTreeSet<String> = backticked_in(fields)
        .into_iter()
        .filter(|token| token.chars().all(|c| c.is_ascii_lowercase() || c == '_'))
        // The clause names `onevcs` as the producer that stamps `phase`; it is
        // the one lowercase token in it that is not a field.
        .filter(|token| token != "onevcs")
        .collect();
    let document = serde_json::to_value(schemars::schema_for!(Matcher))
        .expect("the bus's matcher document serializes");
    assert_eq!(
        named,
        property_names(&document, &document),
        "the grammar paragraph's matcher fields are not the ones the linked bus declares"
    );

    // And the sources the merged-stream paragraph names are the bus's own set.
    let sources = CONTRACT
        .split_once("interleaving the three sources ")
        .and_then(|(_, rest)| rest.split_once('.'))
        .map(|(list, _)| backticked_in(list))
        .expect("the contract names the merged stream's sources");
    assert_eq!(
        sources,
        Source::every()
            .iter()
            .map(|source| source.as_str().to_owned())
            .collect::<BTreeSet<_>>(),
        "the contract's sources are not the ones the linked bus declares"
    );
}

/// The contract's statement of a chain-stopping death names the linked
/// producer's own words for the rule and both causes, so a variant renamed
/// there fails here rather than leaving the paragraph describing a word nothing
/// publishes. `tests/e2e/boundary.rs` drives both causes through the binary.
#[test]
fn the_contract_names_the_producers_words_for_a_chain_that_stopped() {
    use oneagentgraph::event::Cause;
    let passage = CONTRACT
        .split_once("**A chain that stopped raises a finding.**")
        .and_then(|(_, rest)| rest.split_once("\n\n"))
        .map(|(passage, _)| passage)
        .expect("the contract states which death is a chain that stopped");
    let named = backticked_in(passage);
    for word in [
        oneagentgraph::member::Rule::ProviderFailure.as_str(),
        Cause::Unclassified.as_str(),
        Cause::FallbackChainExhausted.as_str(),
        "finding",
        "proposal",
    ] {
        assert!(
            named.contains(word),
            "the chain-stopping paragraph does not name `{word}`: {passage}"
        );
    }
}

/// The committed planner-channel document is the one this build generates
/// from its compiled-in layout, through the bus's own document types: a change
/// to the layout that is not regenerated here fails, and so does a hand edit.
///
/// `ONEPIPELINE_WRITE_LAYOUT_DOCUMENT=1` writes the generated document in place
/// of the committed one instead — how it is regenerated, and never set by a
/// check.
#[test]
fn the_committed_planner_channel_document_is_the_compiled_in_layout() {
    use onepipeline::channel::layout::{bundle_json, DOCUMENT_PATH};
    let path = repo_root().join(DOCUMENT_PATH);
    let generated = bundle_json();
    if std::env::var_os("ONEPIPELINE_WRITE_LAYOUT_DOCUMENT").is_some() {
        std::fs::write(&path, &generated).expect("the document is written");
    }
    let committed = std::fs::read_to_string(&path)
        .unwrap_or_else(|error| panic!("{DOCUMENT_PATH} is committed: {error}"));
    assert!(
        committed == generated,
        "{DOCUMENT_PATH} is not what the compiled-in layout generates. If the layout \
         changed on purpose, raise `DOCUMENT_VERSION` and regenerate it with \
         `ONEPIPELINE_WRITE_LAYOUT_DOCUMENT=1 cargo test --test contract \
         the_committed_planner_channel_document_is_the_compiled_in_layout`; never edit it \
         by hand.\n--- committed\n{committed}\n--- generated\n{generated}"
    );
}

/// The contract states where the document is, the version it declares and
/// that the compiled-in layout is its source — and each of those is true of the
/// committed document, read back through the bus's own bundle type.
#[test]
fn the_contract_states_the_planner_channel_documents_path_version_and_source() {
    use onepipeline::channel::layout::{document, DOCUMENT_PATH, DOCUMENT_VERSION};
    let passage = CONTRACT
        .split_once("**The planner-channel layout is published as a document.**")
        .and_then(|(_, rest)| rest.split_once("\n\n"))
        .map(|(passage, _)| passage)
        .expect("the contract states the published layout document");
    let named = backticked_in(passage);
    for word in [
        DOCUMENT_PATH,
        DOCUMENT_VERSION,
        "LayoutDocument",
        "SchemaBundle",
    ] {
        assert!(
            named.contains(word),
            "the layout-document paragraph does not name `{word}`: {passage}"
        );
    }
    assert!(
        passage.contains("compiled-in layout is its one source"),
        "the paragraph does not say where the document comes from: {passage}"
    );
    let committed = std::fs::read_to_string(repo_root().join(DOCUMENT_PATH))
        .expect("the document is committed");
    let bundle =
        onemessagebus::SchemaBundle::from_json(&committed).expect("the bus reads the document");
    assert_eq!(bundle.version().to_string(), DOCUMENT_VERSION);
    assert_eq!(bundle.layouts(), [document()]);
}

/// `exclude` wins, an absent `include` admits everything, and a glob is `*`.
#[test]
fn the_grammar_matches_the_way_the_contract_says_it_does() {
    let envelope = |source: Source, kind: &str, labels: Labels| Envelope {
        v: ENVELOPE_VERSION,
        ts: "2026-08-15T00:00:00.000Z".into(),
        stream: "s".into(),
        seq: 0,
        source,
        kind: EventKind(kind.into()),
        dimensions: Default::default(),
        labels,
        payload: Default::default(),
        artifacts: Vec::new(),
    };
    let plain = envelope(Source::Agentgraph, "turn-activity", Labels::default());

    assert!(
        EventFilter::default().matches(&plain),
        "an absent include admits everything"
    );
    let excluded = EventFilter::parse(r#"{"exclude": [{"kind": "turn-*"}]}"#).expect("a filter");
    assert!(!excluded.matches(&plain), "a glob matches the wire string");
    let both = EventFilter::parse(
        r#"{"include": [{"source": "agentgraph"}], "exclude": [{"kind": "turn-activity"}]}"#,
    )
    .expect("a filter");
    assert!(!both.matches(&plain), "exclude wins over include");

    // A label the envelope never stamped is not a wildcard.
    let asks_node = EventFilter::parse(r#"{"include": [{"node": "build"}]}"#).expect("a filter");
    assert!(
        !asks_node.matches(&plain),
        "an unstamped label never matches"
    );
    assert!(asks_node.matches(&envelope(
        Source::Pipeline,
        "node-settled",
        Labels {
            node: Some("build".into()),
            ..Labels::default()
        }
    )));

    // `member` has no typed slot on this crate's labels, so it is read out of
    // the extras a relayed envelope stamps it in.
    let asks_member =
        EventFilter::parse(r#"{"include": [{"member": "worker"}]}"#).expect("a filter");
    let mut relayed = plain.clone();
    relayed
        .labels
        .extra
        .insert("member".into(), json!("worker"));
    assert!(asks_member.matches(&relayed));
    assert!(!asks_member.matches(&plain));
}

#[test]
fn a_dispatch_is_cancelled_the_two_ways_the_contract_names() {
    assert_ne!(CancelMode::Cooperative, CancelMode::Kill);
    assert_contract_names("CancelMode variant", &["Cooperative | Kill"]);
}

#[test]
fn the_contract_declares_the_seams_traits_and_methods() {
    let sketch = fenced_block_naming("rust", "pub trait Executor");
    for item in [
        "pub trait Executor",
        "fn name(",
        "fn capabilities(",
        "fn capacity(",
        "fn dispatch(",
        "pub struct DispatchRequest",
        "pub trait DispatchHandle",
        "fn events(",
        "fn wait(",
        "fn cancel(",
    ] {
        assert!(
            sketch.contains(item),
            "the contract's Rust block no longer declares `{item}`"
        );
    }
}

/// The host-shutdown seam the contract spells, held to the types the crate
/// publishes.
///
/// The block is a **sketch** — the contract writes the seam as a consumer reads
/// it — so what is held here is that every item it names exists under that name,
/// and the signatures are held by the coercions in
/// `the_contract_names_every_post_launch_verb_the_sdk_publishes_and_no_other`,
/// which is the drift gate this file uses everywhere. A type added to the block
/// and not to `verbs` fails to compile here before the document can promise it.
#[test]
fn the_contract_declares_the_host_shutdown_seam_this_crate_publishes() {
    use onepipeline::verbs::{
        BranchPreserved, DispatchEnding, DispatchStopped, Preserved, RunShutdown, Shutdown,
        ShutdownRequest, ShutdownScope, StopTeardown,
    };
    let sketch = fenced_block_naming("rust", "pub enum ShutdownScope");
    for item in [
        "pub enum ShutdownScope { Run(String), Mine, Host }",
        "pub struct ShutdownRequest",
        "pub enum DispatchEnding { Graceful, Killed, StillRunning }",
        "pub enum Preserved { Pushed, AlreadyOnOrigin, NoRemote, Refused }",
        "pub struct DispatchStopped",
        "pub struct BranchPreserved",
        "pub struct RunShutdown",
        "pub struct Shutdown",
        "pub fn shutdown(root: &Path, request: ShutdownRequest) -> Result<Shutdown>;",
        "pub fn render_shutdown(shutdown: &Shutdown) -> String;",
        "impl Shutdown { pub fn exit_code(&self) -> i32; }",
        "pub not_pushed: Vec<(String, String)>",
        "pub not_pushed_unread: Option<String>",
    ] {
        assert!(
            sketch.contains(item),
            "the contract's shutdown block no longer declares `{item}`"
        );
    }

    // Every field the block spells, with its type, is exactly the set the
    // exhaustive literals below name — so a field the document adds, drops,
    // renames or retypes fails here, and one the crate adds or drops fails
    // those literals to compile.
    for (name, fields) in [
        (
            "ShutdownRequest",
            &[
                ("scope", "ShutdownScope"),
                ("session", "String"),
                ("grace", "Duration"),
                ("force", "bool"),
            ][..],
        ),
        (
            "DispatchStopped",
            &[
                ("node", "String"),
                ("pid", "u32"),
                ("interrupt", "String"),
                ("detail", "String"),
                ("ended", "DispatchEnding"),
                ("waited", "Duration"),
            ][..],
        ),
        (
            "BranchPreserved",
            &[
                ("identity", "String"),
                ("branch", "String"),
                ("outcome", "Preserved"),
                ("remote", "Option<String>"),
                ("commit", "Option<String>"),
                ("detail", "String"),
            ][..],
        ),
        (
            "RunShutdown",
            &[
                ("run", "String"),
                ("owner", "String"),
                ("forced_over_owner", "bool"),
                ("dispatches", "Vec<DispatchStopped>"),
                ("teardown", "journal::StopTeardown"),
                ("branches", "Vec<BranchPreserved>"),
            ][..],
        ),
        (
            "Shutdown",
            &[
                ("root", "PathBuf"),
                ("scope", "ShutdownScope"),
                ("grace", "Duration"),
                ("forced", "bool"),
                ("runs", "Vec<RunShutdown>"),
                ("not_pushed", "Vec<(String, String)>"),
                ("not_pushed_unread", "Option<String>"),
            ][..],
        ),
    ] {
        let expected: Vec<(String, String)> = fields
            .iter()
            .map(|(field, ty)| ((*field).to_string(), (*ty).to_string()))
            .collect();
        assert_eq!(
            sketch_struct_fields(&sketch, name),
            expected,
            "the contract's `{name}` fields have drifted from the type this crate publishes"
        );
    }

    // The values the block names, built here so a variant renamed or dropped
    // fails to compile rather than leaving the document promising it.
    let stopped = DispatchStopped {
        node: "build".into(),
        pid: 4_242,
        interrupt: "delivered".into(),
        detail: "the running turn took the redirection".into(),
        ended: DispatchEnding::Graceful,
        waited: std::time::Duration::from_secs(3),
    };
    let preserved = BranchPreserved {
        identity: "github.com/owner/service".into(),
        branch: "feat/thing".into(),
        outcome: Preserved::Pushed,
        remote: Some("https://github.com/owner/service.git".into()),
        commit: Some("abc1234".into()),
        detail: String::new(),
    };
    let run = RunShutdown {
        run: "run-1".into(),
        owner: "[mine]".into(),
        forced_over_owner: false,
        dispatches: vec![stopped],
        teardown: StopTeardown::Signalled,
        branches: vec![preserved],
    };
    let shutdown = Shutdown {
        root: PathBuf::from("/runs"),
        scope: ShutdownScope::Host,
        grace: std::time::Duration::from_secs(600),
        forced: false,
        runs: vec![run],
        not_pushed: vec![("github.com/owner/other".into(), "feat/left".into())],
        not_pushed_unread: None,
    };
    // Everything went as it was asked to, so this is exit 0 — and the report
    // still names the branch nobody pushed, which is a fact rather than a
    // failure.
    assert_eq!(shutdown.exit_code(), 0);
    let report = onepipeline::verbs::render_shutdown(&shutdown);
    assert!(report.contains("/runs"), "{report}");
    assert!(
        report.contains("github.com/owner/other@feat/left"),
        "{report}"
    );
    assert!(report.contains("on its origin unproven"), "{report}");

    // And a dispatch the deadline reaped is the refusal the contract assigns.
    let killed = Shutdown {
        runs: vec![RunShutdown {
            dispatches: vec![DispatchStopped {
                ended: DispatchEnding::Killed,
                ..shutdown.runs[0].dispatches[0].clone()
            }],
            ..shutdown.runs[0].clone()
        }],
        ..shutdown
    };
    assert_eq!(killed.exit_code(), EXIT_REFUSED);

    // The three scopes and the default grace are the document's own.
    assert_ne!(ShutdownScope::Mine, ShutdownScope::Host);
    let _: ShutdownRequest = ShutdownRequest {
        scope: ShutdownScope::Run("run-1".into()),
        session: "s".into(),
        grace: std::time::Duration::from_secs(onepipeline::cli::DEFAULT_SHUTDOWN_GRACE_SECONDS),
        force: false,
    };
    assert_eq!(onepipeline::cli::DEFAULT_SHUTDOWN_GRACE_SECONDS, 600);
    assert_contract_names(
        "shutdown paragraph's",
        &[
            "`onepipeline shutdown [RUN] [--mine] [--host] [--grace SECONDS] [--force]`",
            "**default 600**",
            "A shutdown journals no `run-stopped` and fires no run-end hook",
            "on its origin *unproven*",
        ],
    );
}

/// The `pub name: Type` fields a sketched `pub struct` spells, in order, with
/// each type's whitespace collapsed and the block's doc comments left out.
fn sketch_struct_fields(sketch: &str, name: &str) -> Vec<(String, String)> {
    let code: String = sketch
        .lines()
        .filter(|line| !line.trim_start().starts_with("///"))
        .collect::<Vec<_>>()
        .join("\n");
    let open = format!("pub struct {name} {{");
    let start = code
        .find(&open)
        .unwrap_or_else(|| panic!("the contract's shutdown block declares no `{open}`"))
        + open.len();
    // The body ends at the brace that closes it; a field type such as
    // `Vec<(String, String)>` nests parentheses and angles but never braces.
    let body = &code[start..start + code[start..].find('}').expect("the struct closes")];
    let mut fields = Vec::new();
    let mut depth = 0_i32;
    let mut field = String::new();
    for ch in body.chars().chain(std::iter::once(',')) {
        match ch {
            '<' | '(' => depth += 1,
            '>' | ')' => depth -= 1,
            _ => {}
        }
        if ch == ',' && depth == 0 {
            let spelled = field.split_whitespace().collect::<Vec<_>>().join(" ");
            if let Some(rest) = spelled.strip_prefix("pub ") {
                let (field_name, ty) = rest.split_once(':').expect("a field has a type");
                fields.push((field_name.trim().to_string(), ty.trim().to_string()));
            }
            field.clear();
        } else {
            field.push(ch);
        }
    }
    fields
}

/// A plan exercising every node shape the contract names.
fn every_node_shape() -> Value {
    json!({
        "schema_version": PLAN_SCHEMA_VERSION,
        "name": "every-shape",
        "concurrency": 3,
        "goal": {"text": "prove the schema"},
        "tasks": [
            {
                "id": "direct",
                "persona": "engineer",
                "task": "## What\nx\n\n## Why\ny\n\n## Acceptance criteria\n- z",
                "max_turns": 24,
                "expects_no_diff": true,
                "context": "the earlier round already landed the schema",
                "executor": "local",
                "agent_graph": "./graphs/node-scope.yaml",
                "deps": ["run:other-run#upstream"]
            },
            {
                "id": "approval",
                "kind": "human",
                "task": "Approve the design.",
                "deps": ["direct"]
            },
            {
                "id": "lifecycle",
                "repo": "nickderobertis/some-service",
                "repo_type": "team",
                "workflow": "remote",
                "merge_policy": "change-auto",
                "base_branch": "main",
                "branch": "feat/thing",
                "title": "feat: thing",
                "execution_checkout": "isolated",
                "parked": true,
                "resume": {
                    "branch": "feat/thing",
                    "checkpoint": "abc1234",
                    "completed_steps": ["implement"]
                },
                "deps": ["approval"],
                "steps": [
                    {
                        "id": "implement",
                        "persona": "engineer",
                        "task": "## What\nx",
                        "max_turns": 32,
                        "expects_no_diff": false,
                        "executor": "local",
                        "agent_graph": "./graphs/node-scope.yaml"
                    },
                    {
                        "id": "sign-off",
                        "kind": "human",
                        "task": "Exercise staging and approve.",
                        "deps": ["implement"]
                    }
                ]
            }
        ]
    })
}

#[test]
fn the_plan_schema_carries_every_node_shape_the_contract_names() {
    let plan: Plan = serde_json::from_value(every_node_shape()).expect("the plan parses");

    assert_eq!(plan.schema_version, PLAN_SCHEMA_VERSION);
    assert_eq!(plan.concurrency, 3);
    assert_eq!(plan.goal.as_ref().expect("a goal").text, "prove the schema");
    assert_eq!(plan.tasks.len(), 3);

    let direct = &plan.tasks[0];
    assert_eq!(direct.kind, NodeKind::Agent, "`agent` is the default kind");
    assert!(direct.expects_no_diff);
    assert_eq!(
        direct.max_turns,
        Some(24),
        "a turn budget is a node-level control the schema keeps"
    );
    assert_eq!(direct.executor.as_deref(), Some("local"));
    assert_eq!(
        direct.agent_graph,
        Some(ConfigRef("./graphs/node-scope.yaml".into())),
        "`agent_graph` is an oneagentgraph config reference"
    );
    assert_eq!(
        direct.context.as_deref(),
        Some("the earlier round already landed the schema")
    );
    assert_eq!(
        direct.deps,
        vec!["run:other-run#upstream"],
        "a cross-DAG reference is a dependency like any other"
    );

    assert_eq!(plan.tasks[1].kind, NodeKind::Human);

    let lifecycle = &plan.tasks[2];
    assert_eq!(
        lifecycle.repo.as_deref(),
        Some("nickderobertis/some-service")
    );
    assert_eq!(lifecycle.repo_type, Some(RepoType::Team));
    assert_eq!(lifecycle.workflow, Some(Workflow::Remote));
    assert_eq!(lifecycle.merge_policy, Some(MergePolicy::ChangeAuto));
    assert!(lifecycle.parked);
    assert_eq!(
        lifecycle.resume,
        Some(Resume {
            branch: "feat/thing".into(),
            checkpoint: Some("abc1234".into()),
            completed_steps: vec!["implement".into()],
        })
    );
    let steps = lifecycle
        .steps
        .as_ref()
        .expect("nested steps on one branch");
    assert_eq!(steps.len(), 2);
    assert_eq!(steps[0].kind, NodeKind::Agent);
    assert_eq!(steps[1].kind, NodeKind::Human);
    assert_eq!(
        steps[0].max_turns,
        Some(32),
        "a step carries its own turn budget"
    );

    assert_contract_names(
        "node shape",
        &[
            "`agent` direct",
            "lifecycle with `repo`",
            "`kind: human`",
            "nested `steps` on one branch",
            "`expects_no_diff`",
            "`context`",
            "cross-DAG `run:<id>#<node>` refs",
            "per-node `max_turns`",
            "`executor: NAME`",
            "`agent_graph: REF`",
        ],
    );
}

/// The contract's schema version and this crate's are the same number, and every
/// version the document says this build reads, it reads.
///
/// The plan schema is a serialized contract: a document says which version it
/// was written at, and a reader decides by that. So the number the document
/// states and the number the code writes are gated against each other here — and
/// so is the set below it, because "an earlier plan still runs" is a promise to
/// every plan already written on a host and there is nothing else holding it.
#[test]
fn the_contracts_plan_schema_version_is_the_one_this_crate_writes() {
    assert!(
        CONTRACT.contains(&format!("Plan schema v{PLAN_SCHEMA_VERSION} =")),
        "the contract states a different plan schema version than this crate writes \
         ({PLAN_SCHEMA_VERSION})"
    );
    assert!(
        CONTRACT.contains("this build reads **3, 2, and 1**"),
        "the contract no longer names the versions this build reads"
    );
    assert_eq!(
        PLAN_SCHEMA_VERSIONS_READ,
        [3, 2, 1],
        "this crate reads a different set of versions than the contract states"
    );

    // Every version the contract names, as a document an operator wrote: each
    // one reads, and each keeps the version it declares — a reader decides by
    // that number, so a schema that normalized it would answer for a document
    // nobody wrote. That a *project* declaring each one executes is driven
    // through the binary and a real store, in `tests/e2e/plan.rs`, and all the
    // way to a publication in `tests/e2e/lifecycle.rs`, because that is where a
    // planner meets either answer.
    for version in PLAN_SCHEMA_VERSIONS_READ {
        let plan: Plan = serde_json::from_value(json!({
            "schema_version": version,
            "tasks": [{"id": "a", "persona": "engineer", "task": "Do it."}],
        }))
        .unwrap_or_else(|why| panic!("a version {version} plan is a readable document: {why}"));
        assert_eq!(plan.schema_version, version);
    }

    // What this crate *writes* carries the current number, whatever it read.
    let earlier: Plan = serde_json::from_value(json!({
        "schema_version": 1,
        "tasks": [{"id": "a", "persona": "engineer", "task": "Do it."}],
    }))
    .expect("it still reads");
    let current = Plan {
        schema_version: PLAN_SCHEMA_VERSION,
        ..earlier
    };
    let written = serde_json::to_value(&current).expect("it serialises");
    assert_eq!(written["schema_version"], PLAN_SCHEMA_VERSION);
}

/// Every field the contract says a task carries under `onepipeline.<field>` is
/// a field of this schema.
///
/// The mapping itself is behind the surface — a plan is read out of the store by
/// the private engine, and `tests/e2e/plan.rs` drives it through the real binary
/// against a real project. What this holds is the half a document can drift on
/// alone: the contract names each reserved key in writing, and a key naming no
/// field of `Plan` or `Node` would be a promise nothing could keep.
#[test]
fn every_reserved_metadata_key_the_contract_names_is_a_field_of_this_schema() {
    let plan = serde_json::to_value(Plan {
        schema_version: PLAN_SCHEMA_VERSION,
        goal: Some(onepipeline::plan::Goal { text: "why".into() }),
        name: Some("named".into()),
        concurrency: 2,
        tasks: Vec::new(),
    })
    .expect("a plan serialises");
    // Every optional field set, so the serialized shape names all of them: an
    // omitted one would read here as a field the schema does not have.
    let node = serde_json::to_value(Node {
        id: "n".into(),
        kind: NodeKind::Human,
        task: Some("t".into()),
        persona: Some("engineer".into()),
        deps: vec!["other".into()],
        max_turns: Some(1),
        expects_no_diff: true,
        context: Some("note".into()),
        parked: true,
        executor: Some("local".into()),
        agent_graph: Some(ConfigRef("g".into())),
        repo: Some("github.com/owner/name".into()),
        repo_type: Some(RepoType::Team),
        workflow: Some(Workflow::Remote),
        merge_policy: Some(MergePolicy::ChangeAuto),
        base_branch: Some("main".into()),
        branch: Some("topic".into()),
        title: Some("feat: x".into()),
        body: Some("why".into()),
        draft: true,
        execution_checkout: Some("checkout".into()),
        pool: Some(1),
        overflow: Some(onevcs::Bound::Bounded(2)),
        steps: Some(Vec::new()),
        resume: Some(Resume {
            branch: "topic".into(),
            checkpoint: None,
            completed_steps: Vec::new(),
        }),
        adoption: Some(Adoption::Published),
        amendment: Some("changed requirements".into()),
        consumes: std::collections::BTreeMap::new(),
        delivers: vec!["tickets:t-1".into()],
    })
    .expect("a node serialises");
    let fields: BTreeSet<String> = plan
        .as_object()
        .expect("a mapping")
        .keys()
        .chain(node.as_object().expect("a mapping").keys())
        .cloned()
        .collect();

    // The run-history paragraph names a second `onepipeline.`-prefixed
    // namespace — the labels the engine stamps on a oneharness session — which
    // is the `oneharness_history` block's and not plan metadata at all, so what
    // that block names is left out of this inventory.
    let history: Value = serde_json::from_str(&fenced_block_naming("json", "oneharness_history"))
        .expect("the run-history block is JSON");
    let history = &history["oneharness_history"];
    let stamped: BTreeSet<String> = history["labels"]
        .as_object()
        .expect("the block names its labels")
        .values()
        .filter_map(Value::as_str)
        .map(ToOwned::to_owned)
        .chain(history["label_prefix"].as_str().map(ToOwned::to_owned))
        .collect();
    let named: BTreeSet<String> = backticked()
        .iter()
        .filter(|token| !stamped.contains(*token))
        .filter_map(|token| token.strip_prefix("onepipeline.").map(ToOwned::to_owned))
        .filter(|field| field != "<field>")
        .collect();
    assert!(
        !named.is_empty(),
        "the contract names no reserved metadata key at all"
    );
    for field in &named {
        assert!(
            fields.contains(field),
            "the contract reserves `onepipeline.{field}`, which is not a field of the plan schema"
        );
    }
    // The four the contract calls the plan's own sit on the **project** rather
    // than on a task, and it names them in one sentence rather than one key at a
    // time — so that sentence is what is held, and each of the four is a field
    // of the plan.
    assert!(
        CONTRACT.contains(
            "The plan-level fields — `schema_version`, `goal`, `name`, `concurrency` — are \
             reserved metadata keys `onepipeline.<field>` on the **project**"
        ),
        "the contract no longer states which fields the project carries"
    );
    for field in ["schema_version", "goal", "name", "concurrency"] {
        assert!(
            plan.as_object().expect("a mapping").contains_key(field),
            "the contract reserves `{field}` on the project, which is not a plan field"
        );
    }
}

/// The retired field, refused **by name** at every boundary a plan crosses.
///
/// `deny_unknown_fields` would answer a plan still carrying it with a bare
/// `unknown field`, which tells a planner that a field does not exist and not
/// where the review bar they wrote belongs. Every plan written before this schema
/// change carries one, so the refusal has to say where the bar goes instead.
///
/// Two halves. Here: the field is not in the schema at any version this build
/// reads, which is what makes the named refusal the only answer there can be.
/// In `tests/e2e/plan.rs`: a real project carrying `onepipeline.done_when` is
/// refused **by that name**, through the binary, with the bar's new home in the
/// message — which is where a planner meets it.
#[test]
fn a_plan_still_carrying_done_when_is_refused_by_name_and_told_where_the_bar_goes() {
    assert!(
        CONTRACT.contains("A plan still carrying `done_when` is refused **by name**"),
        "the contract no longer states the refusal"
    );
    assert!(
        CONTRACT.contains(
            "A retired field is refused by its own name here exactly as it is in a \
                           plan document"
        ),
        "the contract no longer carries the refusal into the store"
    );
    for version in PLAN_SCHEMA_VERSIONS_READ {
        for field in ["done_when", "verify_via_ci"] {
            let refused = serde_json::from_value::<Plan>(json!({
                "schema_version": version,
                "tasks": [{"id": "contract", "persona": "engineer", "task": "t", field: true}],
            }))
            .expect_err("a retired field is not a field of this schema");
            assert!(
                refused.to_string().contains(field),
                "the schema's own refusal does not name the field: {refused}"
            );
        }
    }
    // A step carries the same fields and is held to the same schema.
    for field in ["done_when", "verify_via_ci"] {
        serde_json::from_value::<Step>(json!({"id": "implement", field: true}))
            .expect_err("a retired field is not a field of a step either");
    }
    // And a plan that carries none reads, with what it did declare intact.
    let plan: Plan = serde_json::from_value(json!({
        "schema_version": PLAN_SCHEMA_VERSION,
        "tasks": [{"id": "contract", "persona": "engineer", "task": "t", "max_turns": 45}],
    }))
    .expect("a plan without the retired field reads");
    assert_eq!(plan.tasks[0].max_turns, Some(45));
}

/// A dispatch an external caller builds carries its node's controls into the
/// launch, and a control the graph has no field for refuses that launch.
///
/// Through the public seam, with no `run_id`: nothing here reads a launch record,
/// so what reaches `oneagentgraph` is the request's own `controls` or nothing at
/// all. The sibling checks an override against the same schema it reads the graph
/// with, so a `max_turns` addressed to a single-sided member — which has no such
/// field — is refused by name. That refusal is only reachable if the control was
/// transmitted; a dispatch that dropped it would launch the graph happily.
#[test]
fn a_dispatch_built_outside_a_run_still_carries_its_controls_into_the_launch() {
    // Both of these are the whole process's: the root is keyed on this test
    // rather than on the process it runs in, and the variable naming it is set
    // under the lock a sibling test setting the seam's own variables also holds.
    let _seam = seam_env_lock();
    let root = scratch("seam");
    std::env::set_var("ONEAGENTGRAPH_STATE_DIR", root.join("state"));
    let graph = root.join("single-sided.yaml");
    std::fs::write(
        &graph,
        "version: 1\nname: single-sided\nmembers:\n  worker:\n    kind: oneharness\n    \
         oneharness_config: ./nothing.toml\n",
    )
    .expect("the graph is written");

    let request = |controls| DispatchRequest {
        graph: ConfigRef(graph.display().to_string()),
        task: "## What\nDo the thing.".into(),
        labels: Labels::default(),
        controls,
        workspace: WorkspaceSpec::Path(root.clone()),
        cancel: CancellationToken::new(),
        attempt: NonZeroU32::MIN,
    };

    let Err(refused) = LocalExecutor.dispatch(request(NodeControls {
        max_turns: NonZeroU32::new(45),
    })) else {
        panic!("a single-sided member has no `max_turns`, so the launch cannot start");
    };
    let refused = refused.to_string();
    assert!(
        refused.contains("max_turns"),
        "the control never reached the launch: {refused}"
    );

    // Without one, nothing addresses that field at all, and the launch fails
    // further in — on the config this graph names and this test never wrote.
    let Err(other) = LocalExecutor.dispatch(request(NodeControls::default())) else {
        panic!("the graph names a config that does not exist");
    };
    assert!(
        !other.to_string().contains("max_turns"),
        "a control nobody declared was sent anyway: {other}"
    );
    std::env::remove_var("ONEAGENTGRAPH_STATE_DIR");
    std::fs::remove_dir_all(&root).ok();
}

/// A node's turn budget reaches the graph that runs it, read off the effective
/// configuration rather than off the code path that composed it.
///
/// The overrides this crate renders are applied to the **shipped** node-scope
/// graph by `oneagentgraph`'s own applier — the same call its `run` makes before
/// it builds a member — and the worker is read out of the result. A budget the
/// overrides never carried cannot survive that, and neither can one addressed to
/// a member or a field the sibling does not have.
#[test]
fn a_declared_turn_budget_reaches_the_effective_configuration_of_the_worker() {
    assert!(
        CONTRACT.contains("`max_turns` is the worker member's own turn ceiling"),
        "the contract no longer says where a turn budget lands"
    );
    let text = std::fs::read_to_string(repo_root().join("graphs/node-scope.yaml"))
        .expect("the node-scope graph ships");

    let effective = |controls: NodeControls| -> GraphConfig {
        let overrides: Vec<_> = controls
            .overrides()
            .expect("a declared budget is appliable")
            .iter()
            .map(|set| oneagentgraph::run::parse_set(set).expect("the sibling parses the override"))
            .collect();
        let mut document: Value = serde_norway::from_str(&text).expect("the graph parses");
        oneagentgraph::run::apply_overrides(&mut document, &overrides)
            .expect("the sibling applies the override");
        serde_norway::from_value(serde_norway::to_value(&document).expect("a value"))
            .expect("the overridden graph is still a valid graph config")
    };

    let turns_of = |graph: &GraphConfig| match graph.members.get("worker") {
        Some(Member::Onejudge(worker)) => worker.max_turns,
        other => panic!("the node-scope worker is a two-party member: {other:?}"),
    };

    assert_eq!(
        turns_of(&effective(NodeControls::default())),
        None,
        "the shipped graph must state no budget, or this proves nothing"
    );
    assert_eq!(
        turns_of(&effective(NodeControls {
            max_turns: NonZeroU32::new(45)
        })),
        Some(45),
        "the node's turn budget did not reach the member that runs its work"
    );
}

#[test]
fn resume_carries_what_the_contract_says_a_preserved_branch_needs() {
    let prose = CONTRACT.split_whitespace().collect::<Vec<_>>().join(" ");
    assert!(
        prose.contains("`{branch, checkpoint?, completed_steps?}`"),
        "the contract no longer states the `resume` shape"
    );
    assert!(
        prose.contains("`completed_steps` names the steps that branch already carries"),
        "the contract no longer says what `completed_steps` means"
    );
    assert!(
        prose.contains("`checkpoint` must be a commit reachable on the remote"),
        "the contract no longer says what a checkpoint is"
    );

    // The shape the contract states is the shape the schema reads, including a
    // continuation that names no steps at all.
    let full: Resume = serde_json::from_value(json!({
        "branch": "feat/thing",
        "checkpoint": "abc1234",
        "completed_steps": ["implement", "review"]
    }))
    .expect("the stated shape parses");
    assert_eq!(full.completed_steps, ["implement", "review"]);

    let minimal: Resume =
        serde_json::from_value(json!({"branch": "feat/thing"})).expect("branch alone is a resume");
    assert!(
        minimal.completed_steps.is_empty(),
        "an absent list re-runs the whole workstream"
    );
    // And an empty list is omitted again, so an old consumer sees no new field.
    assert_eq!(
        serde_json::to_value(&minimal).expect("serializes"),
        json!({"branch": "feat/thing"})
    );
}

#[test]
fn a_plan_round_trips_without_losing_a_field() {
    let plan: Plan = serde_json::from_value(every_node_shape()).expect("parses");
    let again: Plan = serde_json::from_value(serde_json::to_value(&plan).expect("serializes"))
        .expect("re-parses");
    assert_eq!(again, plan);
}

#[test]
fn a_mistyped_node_key_is_refused_at_the_boundary() {
    let err = serde_json::from_value::<Plan>(json!({
        "schema_version": PLAN_SCHEMA_VERSION,
        "tasks": [{"id": "x", "persna": "engineer"}]
    }))
    .expect_err("a mistyped key is rejected, not silently dropped");
    assert!(
        err.to_string().contains("persna"),
        "the error names it: {err}"
    );
}

#[test]
fn a_node_and_a_step_default_to_the_shapes_the_contract_states() {
    let node = Node {
        id: "x".into(),
        ..Node::default()
    };
    assert_eq!(node.kind, NodeKind::Agent);
    assert!(!node.expects_no_diff);
    assert!(!node.parked);
    assert!(node.deps.is_empty());

    let step = Step {
        id: "s".into(),
        ..Step::default()
    };
    assert_eq!(step.kind, NodeKind::Agent);
    assert!(!step.expects_no_diff);

    // An unset optional is omitted, so an old consumer never sees a null it did
    // not have before.
    let rendered = serde_json::to_value(&node).expect("serializes");
    assert_eq!(
        rendered,
        json!({"id": "x"}),
        "the default kind is omitted, so an old consumer sees no field it did not have"
    );
}

/// The shipped example store holds one project per example plan, and each
/// declares the version this crate writes.
///
/// The documents themselves rather than the graph they read as: what they read
/// as is `tests/e2e/shipped.rs`'s, which drives them through the real
/// `onetaskgraph` binary and executes one. What is held here is that the store
/// the contract names is on disk, in the shape the mapping needs — a project per
/// plan, a task per node, and the reserved key that carries the schema version.
#[test]
fn the_shipped_example_store_holds_a_project_per_example_plan() {
    let store = repo_root().join("examples").join("plan-store");
    assert!(
        store.join("onetaskgraph.yaml").is_file(),
        "the example store configures no source, so nothing can read it"
    );
    for name in ["single-node", "tracked-release"] {
        let project = store.join("projects").join(format!("{name}.md"));
        let text = std::fs::read_to_string(&project)
            .unwrap_or_else(|e| panic!("the {name} project ships: {e}"));
        assert!(
            text.contains(&format!(
                "\"onepipeline.schema_version\": {PLAN_SCHEMA_VERSION}"
            )),
            "the {name} project does not declare the schema version this crate writes:\n{text}"
        );
        let tasks = store.join("tasks").join(name);
        assert!(
            std::fs::read_dir(&tasks)
                .unwrap_or_else(|e| panic!("the {name} project has tasks: {e}"))
                .count()
                > 0,
            "the {name} project holds no task, so its plan has no node"
        );
    }
}

/// The `op` a [`Edit`] serializes as.
///
/// The match has no wildcard, so a tenth variant stops this suite compiling
/// until the contract, [`OPS`], and the round-trip below all name it. That is
/// the half of "exactly the ops this crate accepts" a hand-written list cannot
/// prove.
fn op_of(command: &Edit) -> &'static str {
    match command {
        Edit::Add { .. } => "add",
        Edit::Drop { .. } => "drop",
        Edit::Reparent { .. } => "reparent",
        Edit::Retry { .. } => "retry",
        Edit::Cancel { .. } => "cancel",
        Edit::Settle { .. } => "settle",
        Edit::Requeue { .. } => "requeue",
        Edit::Attest { .. } => "attest",
        Edit::Complete { .. } => "complete",
        Edit::Amend { .. } => "amend",
        Edit::Note { .. } => "note",
        Edit::Finding { .. } => "finding",
    }
}

/// The ops the contract lists, in the order it lists them.
const OPS: &[&str] = &[
    "add", "drop", "reparent", "retry", "cancel", "requeue", "attest", "complete", "context",
];

/// The ops the contract lists that this build **no longer accepts**, per
/// divergence entry 60.
///
/// The only *subtraction* from the contract's own list, and it is a subtraction
/// rather than a divergence beyond it: `context` was collapsed into `note`, so an
/// envelope naming it is refused at the wire. Held here so that every place [`OPS`]
/// is used says which of the two it means, and so that a second removal cannot
/// arrive without a line in the record.
const REMOVED_OPS: &[&str] = &["context"];

/// The contract's op list, less what entry 60 removed: exactly what this build
/// accepts of it.
fn contract_ops_this_build_accepts() -> Vec<&'static str> {
    OPS.iter()
        .copied()
        .filter(|op| !REMOVED_OPS.contains(op))
        .collect()
}

/// The contract's op list, less the one entry 60 removed, is exactly what this
/// crate accepts of it.
///
/// The contract is committed as approved and still names `context`, so the
/// subtraction lives in the divergence record and is asserted from there rather
/// than by editing the document to match the code.
#[test]
fn the_contract_lists_exactly_the_ops_this_crate_accepts() {
    let listed = "`add | drop | reparent | retry | cancel | requeue | attest | complete | context`";
    assert!(
        CONTRACT.contains(listed),
        "the contract's op list moved; update OPS with it"
    );
    assert_eq!(OPS.len(), 9);

    let removed: Vec<String> = serde_json::from_value(divergence_block("60.")["removed"].clone())
        .expect("entry 60 names what it removes");
    assert_eq!(
        removed,
        REMOVED_OPS
            .iter()
            .map(ToString::to_string)
            .collect::<Vec<_>>(),
        "the record and this suite disagree about which of the contract's ops are gone"
    );
    for op in REMOVED_OPS {
        let err = serde_json::from_value::<Edit>(json!({"op": op, "id": "x", "note": "hello"}))
            .expect_err("an op the contract lists and this build removed is refused");
        assert!(
            err.to_string().contains(op),
            "the refusal does not name `{op}`: {err}"
        );
    }

    assert_eq!(
        op_of(&Edit::Cancel {
            id: "x".into(),
            reason: None
        }),
        "cancel",
        "the exhaustive match above is what proves the variant set, and it runs"
    );
}

/// One divergence entry's fenced JSON block, by the number it opens with.
fn divergence_block(number: &str) -> Value {
    let record = std::fs::read_to_string(repo_root().join("docs/contract-divergences.md"))
        .expect("the divergence record reads");
    let entry = record
        .split("\n## ")
        .find(|entry| entry.starts_with(number))
        .unwrap_or_else(|| panic!("the divergence record still carries entry {number}"));
    let block = entry
        .split("```json")
        .nth(1)
        .and_then(|rest| rest.split("```").next())
        .unwrap_or_else(|| panic!("entry {number} carries the json block this test drives"));
    serde_json::from_str(block).unwrap_or_else(|e| panic!("entry {number}'s block is JSON: {e}"))
}

/// The release-adoption surface this build carries **beyond** the contract is
/// exactly what the divergence record proposes.
///
/// The contract is committed as approved and names none of it, so entry 40 is the
/// only place it is written down — and a divergence nothing gates quietly stops
/// being true. The entry's own block is the source: what parses here is what a
/// planner would write in a plan.
#[test]
fn the_release_adoption_surface_is_what_the_divergence_record_names() {
    let block = divergence_block("40.");

    // The node shape, exactly as the entry writes it.
    let written = block["node"].clone();
    let node: Node = serde_json::from_value(written.clone()).expect("entry 40's node parses");
    assert_eq!(
        serde_json::to_value(&node).expect("serializes"),
        written,
        "entry 40's node does not round-trip as written"
    );
    assert_eq!(
        node.adoption,
        Some(Adoption::Published),
        "`adoption` is the node rung of the chain"
    );
    assert_eq!(
        node.consumes
            .get("engine")
            .map(std::string::ToString::to_string),
        Some("crate".to_string()),
        "`consumes` is keyed by dependency node id"
    );
    // At schema 3, and optional: a plan naming neither field is the plan it
    // always was, and round-trips without either appearing.
    let plain = json!({"id": "solo", "persona": "engineer", "task": "## What\nx"});
    let bare: Node = serde_json::from_value(plain.clone()).expect("a node naming neither parses");
    assert_eq!(bare.adoption, None);
    assert!(bare.consumes.is_empty());
    assert_eq!(
        serde_json::to_value(&bare).expect("serializes"),
        plain,
        "a node naming neither field gained one on the way out"
    );
    // The event kinds, held against the enum itself and against the contract.
    let kinds: Vec<String> =
        serde_json::from_value(block["event_kinds"].clone()).expect("entry 40 names its kinds");
    assert!(!kinds.is_empty());
    for kind in &kinds {
        assert!(
            PipelineKind::from_wire(&EventKind(kind.clone())).is_some(),
            "`{kind}` is not a kind this crate emits"
        );
    }

    // The heading, which is a published constant so a reader finds the block by
    // name rather than by matching prose.
    assert_eq!(
        block["heading"].as_str(),
        Some(CROSS_REPO_REFERENCES_HEADING),
        "entry 40 names a different heading than this crate publishes"
    );
    assert_ne!(CROSS_REPO_REFERENCES_HEADING, PLANNER_CONTEXT_HEADING);

    // The wait surface and the rule that withholds a stale one: the kind is one
    // the entry's own kinds carry, what fixes its epoch is a field every surface
    // carries, and each record named as ending a hold is a kind this crate
    // emits. The rule itself is driven through the binary by the adoption
    // journey the entry names; what is held here is that the entry's words and
    // the types cannot drift apart.
    let wait = &block["wait_surface"];
    assert_eq!(wait["kind"].as_str(), Some("release-wait"));
    assert!(kinds.contains(&"release-wait".to_string()));
    assert_eq!(wait["epoch"].as_str(), Some("queued_at"));
    let queued = serde_json::to_value(Surface {
        id: 1,
        kind: "release-wait".into(),
        message: "held".into(),
        source: "proposal".into(),
        blocking: false,
        queued_at: 7,
        abandoned: false,
        asker: None,
        workstream: Some("consumer".into()),
        correlation: None,
    })
    .expect("a surface serializes");
    assert_eq!(
        queued["queued_at"],
        json!(7),
        "the field the entry says fixes a wait's epoch is not one a surface carries"
    );
    // Which kinds they are is reconciled against the set `wait_outlived` reads
    // by `src/release.rs`'s own test of this block, since that set is private.
    let ending: Vec<String> = serde_json::from_value(wait["withheld_after"].clone())
        .expect("entry 40 names the records that end a hold");
    assert!(
        !ending.is_empty(),
        "entry 40 names no record that ends a hold"
    );
    for kind in &ending {
        assert!(
            PipelineKind::from_wire(&EventKind(kind.clone())).is_some(),
            "`{kind}` is not a kind this crate emits"
        );
    }
    let record = std::fs::read_to_string(repo_root().join("docs/contract-divergences.md"))
        .expect("the divergence record ships");
    let prose = record.split_whitespace().collect::<Vec<_>>().join(" ");
    for said in [
        "fixed by the instant it was queued",
        "at or after that instant",
        "is **withheld**, never handed out",
        "recorded neither as `planner-surfaced`",
    ] {
        assert!(
            prose.contains(said),
            "entry 40 no longer states the rule that withholds a stale wait: {said:?}"
        );
    }
}

/// The placement overrides, the hold, the surface and the requeue this build
/// carries are exactly what entry 82 records and the amended contract names.
///
/// The node half is driven through the public type: the entry's node parses,
/// round-trips as written, carries the sibling's own `Bound`, and a node naming
/// neither field is the node it always was. The kind is held against the enum.
/// The hold entry's shape is private vocabulary and is held by `pool::tests`
/// against the same block.
#[test]
fn the_workspace_placement_surface_is_what_the_divergence_record_names() {
    let block = divergence_block("82.");

    let written = block["node"].clone();
    let node: Node = serde_json::from_value(written.clone()).expect("entry 82's node parses");
    assert_eq!(
        serde_json::to_value(&node).expect("serializes"),
        written,
        "entry 82's node does not round-trip as written"
    );
    assert_eq!(node.pool, Some(1));
    assert_eq!(node.overflow, Some(onevcs::Bound::Bounded(0)));
    // The word, as the sibling spells it, both ways.
    let unlimited: Node = serde_json::from_value(json!({
        "id": "n", "persona": "engineer", "task": "## What\nx", "overflow": "unlimited"
    }))
    .expect("the sibling's word parses");
    assert_eq!(unlimited.overflow, Some(onevcs::Bound::Unlimited));
    assert_eq!(
        serde_json::to_value(&unlimited).expect("serializes")["overflow"],
        json!("unlimited")
    );
    // At schema 3, and optional: a plan naming neither field round-trips
    // without either appearing.
    let plain = json!({"id": "solo", "persona": "engineer", "task": "## What\nx"});
    let bare: Node = serde_json::from_value(plain.clone()).expect("a node naming neither parses");
    assert_eq!(bare.pool, None);
    assert_eq!(bare.overflow, None);
    assert_eq!(
        serde_json::to_value(&bare).expect("serializes"),
        plain,
        "a node naming neither field gained one on the way out"
    );
    assert_eq!(
        block["refused_below_schema"].as_u64(),
        Some(u64::from(PLAN_SCHEMA_VERSION))
    );

    // The requeue is a kind this crate emits and the contract lists, and the
    // contract names the fields a pinned re-dispatch's carries beside the two
    // every requeue does — on any attempt, settling nothing.
    let kind = block["requeue"]["kind"].as_str().expect("a kind");
    assert_eq!(
        PipelineKind::from_wire(&EventKind(kind.to_owned())),
        Some(PipelineKind::NodeRequeued)
    );
    assert_eq!(block["requeue"]["fields"], json!(["reason", "detail"]));
    assert_eq!(
        block["requeue"]["pinned_fields"],
        json!(["branch", "attempt"])
    );
    for names in [
        "`pool: u32` and `overflow: N|unlimited`",
        "**held under `workspace`**",
        "`workspace-wait` surface",
        "`reason: workspace-exhausted`",
        "neither settled nor `awaiting-planner`",
        "on **any attempt of any node** — no pool-exhausted refusal writes a settlement or \
         spends a boundary attempt",
        "**keeps that pin through the queue**: its `node-requeued` also carries the `branch`",
        "A `retry` is never what continues a node the host was merely too busy for",
    ] {
        assert!(
            CONTRACT.contains(names),
            "the contract no longer states {names}"
        );
    }
}

/// The pool-maintenance schedule this build takes is exactly what entry 83 records
/// and the amended contract names.
///
/// The block is the source: the flag is asked of the parser, the key of the launch
/// config at the version the block states and at none before it, the schedule of
/// the public type — parsed through the sibling's own span and match, and
/// round-tripping as written — the kind of the enum, and every environment name of
/// the contract's own tokens. `tests/e2e/maintenance.rs` drives the schedule through
/// the real binary and the real sibling.
#[test]
fn the_pool_maintenance_schedule_is_what_the_divergence_record_names() {
    use onepipeline::maintenance::{
        MaintenanceConfig, DEFAULT_PACE_SECONDS, FLAG, KEY, PACE_ENV, SCHEDULE_VERSION,
    };
    let block = divergence_block("83.");
    assert_eq!(block["flag"].as_str(), Some(FLAG));
    assert_eq!(block["config_key"].as_str(), Some(KEY));
    assert!(
        CONTRACT.contains("[--maintenance-config FILE]"),
        "the driver invocation no longer names the maintenance flag"
    );

    // The flag, and `adopt` not taking it.
    let Command::Start(started) = Cli::try_parse_from([
        "onepipeline",
        "start",
        "plans:demo",
        FLAG,
        "./maintenance.yml",
    ])
    .expect("the flag the block names is one `start` takes")
    .command
    else {
        panic!("that is not a start")
    };
    assert_eq!(
        started.maintenance_config.as_deref(),
        Some("./maintenance.yml")
    );
    let Command::Start(unset) = Cli::try_parse_from(["onepipeline", "start", "plans:demo"])
        .expect("it parses")
        .command
    else {
        panic!("that is not a start")
    };
    assert_eq!(unset.maintenance_config, None);
    let refused = Cli::try_parse_from(["onepipeline", "adopt", "demo", FLAG, "x"])
        .expect_err("adopt takes no maintenance flag");
    assert!(refused.to_string().contains(FLAG), "{refused}");

    // The key, at the version the block states and refused by name before it.
    let at = u32::try_from(
        block["config_schema_version"]
            .as_u64()
            .expect("the block states the version the key arrived at"),
    )
    .expect("a version fits");
    assert_eq!(at, LAUNCH_CONFIG_SCHEMA_VERSION);
    let named: LaunchConfig = serde_json::from_value(json!({
        "schema_version": at,
        KEY: "./maintenance.yml",
    }))
    .expect("a config naming the schedule parses");
    assert_eq!(
        named.maintenance_config.as_deref(),
        Some("./maintenance.yml")
    );
    let example = fenced_block_naming("yaml", "maintenance_config:");
    let example: LaunchConfig =
        serde_norway::from_str(&example).expect("the contract's launch example parses");
    assert_eq!(example.schema_version, at);
    assert_eq!(
        example.maintenance_config.as_deref(),
        Some("./maintenance.yml")
    );
    let dir = std::env::temp_dir().join(format!(
        "onepipeline-contract-maintenance-{}",
        std::process::id()
    ));
    std::fs::create_dir_all(&dir).expect("a scratch directory");
    for version in LAUNCH_CONFIG_SCHEMA_VERSIONS_READ
        .into_iter()
        .filter(|version| *version < at)
    {
        let path = dir.join(format!("{version}.yaml"));
        std::fs::write(
            &path,
            format!("schema_version: {version}\n{KEY}: ./maintenance.yml\n"),
        )
        .expect("the config is written");
        let refused = LaunchConfig::load(&path)
            .expect_err("a version that never had the key refuses it")
            .to_string();
        assert!(
            refused.contains(&format!("`{KEY}`")) && refused.contains(&format!("schema {at} key")),
            "schema {version} did not refuse `{KEY}` by its name: {refused}"
        );
    }

    // The schedule, through the public type and the sibling's own grammar.
    let written = block["schedule"].clone();
    let schedule: MaintenanceConfig =
        serde_json::from_value(written.clone()).expect("entry 83's schedule parses");
    assert_eq!(schedule.version, SCHEDULE_VERSION);
    assert_eq!(
        block["schedule_version"].as_u64(),
        Some(u64::from(SCHEDULE_VERSION))
    );
    assert_eq!(schedule.default.every.to_string(), "7d");
    assert_eq!(schedule.rules.len(), 1);
    assert_eq!(schedule.rules[0].every.to_string(), "3d");
    assert_eq!(schedule.rules[0].r#match.name.as_deref(), Some("onevcs"));
    assert_eq!(
        serde_json::to_value(&schedule).expect("serializes"),
        written,
        "entry 83's schedule does not round-trip as written"
    );
    // The contract's own schedule block is the same document.
    let contract_schedule: MaintenanceConfig =
        serde_norway::from_str(&fenced_block_naming("yaml", "every: 7d"))
            .expect("the contract's schedule example parses");
    assert_eq!(contract_schedule, schedule);
    // Loaded off disk, each refusal the block lists names its key.
    let schedule_path = dir.join("schedule.yml");
    for (document, names) in [
        (
            "version: 1\ndefault:\n  every: 7d\nat: \"03:00\"\n",
            "unknown field `at`",
        ),
        ("version: 2\ndefault:\n  every: 7d\n", "`version` is 2"),
        (
            "version: 1\ndefault:\n  every: 7d\nrules:\n  - match: {}\n    every: 1d\n",
            "`match` names no field",
        ),
        ("version: 1\ndefault:\n  every: 7x\n", "`every`"),
    ] {
        std::fs::write(&schedule_path, document).expect("the schedule is written");
        let refused = MaintenanceConfig::load(FLAG, &schedule_path)
            .expect_err("the document is refused")
            .to_string();
        assert!(
            refused.contains(names),
            "{document:?} was refused as: {refused}"
        );
    }
    assert_eq!(
        block["refused"].as_array().map(Vec::len),
        Some(4),
        "the block lists a different number of refusals than the test drives"
    );
    let _ = std::fs::remove_dir_all(&dir);

    // The kind, the pace and the load: the enum's, the constants', and the
    // contract's own tokens.
    let kinds: Vec<String> =
        serde_json::from_value(block["event_kinds"].clone()).expect("entry 83 names its kind");
    assert_eq!(kinds, ["pool-maintenance"]);
    assert_eq!(
        PipelineKind::from_wire(&EventKind(kinds[0].clone())),
        Some(PipelineKind::PoolMaintenance)
    );
    assert_eq!(block["pace_env"].as_str(), Some(PACE_ENV));
    assert_eq!(
        block["default_pace_seconds"].as_u64(),
        Some(DEFAULT_PACE_SECONDS)
    );
    assert_eq!(
        block["load_env"].as_str(),
        Some(onepipeline::executor::LOAD1_ENV)
    );
    let tokens = backticked();
    for token in [KEY, PACE_ENV, "pool-maintenance", "schema_version: 9"] {
        assert!(
            tokens.contains(token),
            "the contract no longer names `{token}`"
        );
    }
    assert!(
        tokens.contains(&format!("{FLAG} FILE")),
        "the contract no longer names `{FLAG} FILE`"
    );
    for names in [
        "the flag beats the config, and a blank value, flag or key, is this launch saying it has none",
        "`every` is the **only** key a rule or the default carries",
        "a slot is due when `last_maintained < now − every`, and the sibling decides it",
        "nothing is written when every identity answered `no-maintain-command`, `no-slots` or `not-due`",
        "A launch naming no schedule runs no maintenance, spawns no thread, and behaves exactly as before",
    ] {
        assert!(CONTRACT.contains(names), "the contract no longer states {names}");
    }
}

/// Every field a summary document carries, taken off a value built through the
/// type with every absence filled in.
///
/// Built rather than listed: a field added to `RunSummary` does not compile here
/// until it is given a value, and an optional one left absent would not reach the
/// wire — so what this answers is the whole inventory a consumer can read.
fn summary_fields() -> BTreeSet<String> {
    let document = RunSummary {
        schema_version: SUMMARY_SCHEMA_VERSION,
        run_id: "gated".into(),
        last_write_at: Some(1_786_000_000_000),
        last_event_kind: Some("node-settled".into()),
        event_count: 11,
        node_counts: [("done".to_string(), 1)].into_iter().collect(),
        stop_recorded: true,
        graph_complete: true,
        decisions_pending: 1,
        surfaces_queued: 1,
        surfaces_read: 1,
        awaiting_human_action: true,
        project: "plans:gated".into(),
        name: Some("Gated".into()),
        launcher: "claude-code".into(),
        session: "a-session".into(),
        started_at: Some("2026-01-01T00:00:00.000Z".into()),
        pid: NonZeroU32::new(4_242),
        host: Some("a-host".into()),
        started: Some("linux-proc-stat:1".into()),
        // Read rather than built: the claim's type is the engine's, and what this
        // inventory counts is the key it is written under.
        let_go_by: serde_json::from_value(json!({"host": "a-host", "pid": 4242}))
            .expect("a driver claim reads"),
        timing: serde_json::from_str::<RunTelemetry>(include_str!("golden/telemetry-v2.json"))
            .expect("the telemetry golden reads back into the types"),
        parked: vec!["idle".into()],
        judge_rejected: vec!["publish".into()],
        landings: [(
            "publish".to_string(),
            NodeLanding {
                landing: "unlanded".into(),
                branch: Some("onepipeline/gated".into()),
                repo: Some("nickderobertis/onepipeline".into()),
                drafted: false,
            },
        )]
        .into_iter()
        .collect(),
        oneharness_sessions: Some("/runs/gated/oneharness-sessions.jsonl".into()),
        journal_len: 8_192,
        journal_mtime_ms: 1_786_000_000_100,
    };
    serde_json::to_value(&document)
        .expect("a summary is an object")
        .as_object()
        .expect("a summary is an object")
        .keys()
        .cloned()
        .collect()
}

/// Every field one node's landing inputs carry, on the same terms.
///
/// A second inventory because the document grew a **nested** type, and a field
/// added inside one is exactly as unreadable to a consumer nobody told as a field
/// added beside it.
fn landing_fields() -> BTreeSet<String> {
    let landing = NodeLanding {
        landing: "unlanded".into(),
        branch: Some("onepipeline/gated".into()),
        repo: Some("nickderobertis/onepipeline".into()),
        drafted: true,
    };
    serde_json::to_value(&landing)
        .expect("a landing is an object")
        .as_object()
        .expect("a landing is an object")
        .keys()
        .cloned()
        .collect()
}

/// The summary document's schema version and field inventory are the type's own.
///
/// The contract names neither, so entry 56 is the only place the document a
/// consumer parses is written down — and a divergence nothing gates quietly stops
/// being true. Both directions, as entries 39 to 41 already hold their own: a
/// field the type grows and the entry does not name fails here, and so does a name
/// the entry keeps after the type dropped it.
#[test]
fn the_summary_document_is_what_the_divergence_record_names() {
    let block = divergence_block("56.");
    assert_eq!(
        block["schema_version"].as_u64(),
        Some(u64::from(SUMMARY_SCHEMA_VERSION)),
        "entry 56 states a schema version this build does not write"
    );
    let named: BTreeSet<String> = serde_json::from_value(block["fields"].clone())
        .expect("entry 56 names the document's fields");
    assert_eq!(
        named,
        summary_fields(),
        "entry 56's inventory is not the document this build writes"
    );
    let landings: BTreeSet<String> = serde_json::from_value(block["landing_fields"].clone())
        .expect("entry 56 names the landing inputs' fields");
    assert_eq!(
        landings,
        landing_fields(),
        "entry 56's landing inventory is not what this build writes"
    );
}

/// The amendment lever and the node-validator hook this build carries **beyond**
/// the contract are exactly what the divergence record proposes.
///
/// The contract is committed as approved and names neither, so entry 41 is the
/// only place they are written down — and a divergence nothing gates quietly
/// stops being true. The entry's own block is the source: what parses here is
/// what a planner would type and what an operator would write in a config.
#[test]
fn the_amendment_and_validator_surface_is_what_the_divergence_record_names() {
    let block = divergence_block("41.");

    // The op, as the wire carries it.
    let fixtures: Vec<Value> =
        serde_json::from_value(block["ops"].clone()).expect("entry 41 names the op it adds");
    let monitor_may: BTreeSet<String> = serde_json::from_value(block["monitor_may_issue"].clone())
        .expect("entry 41 says which of them the monitor may issue");
    assert!(!fixtures.is_empty(), "{block}");
    for fixture in &fixtures {
        let op = fixture["op"].as_str().expect("the fixture names its op");
        assert!(
            !OPS.contains(&op),
            "`{op}` is on the contract's own list, so it is no divergence"
        );
        let edit: Edit = serde_json::from_value(fixture.clone())
            .unwrap_or_else(|e| panic!("`{op}` deserializes: {e}"));
        assert_eq!(op_of(&edit), op, "`{op}` deserialized into another variant");
        assert_eq!(
            &serde_json::to_value(&edit).expect("serializes"),
            fixture,
            "`{op}` round-trips unchanged"
        );
        allows(Author::planner(), &edit)
            .unwrap_or_else(|e| panic!("the planner was refused `{op}`: {e}"));
        let verdict = allows(Author::from("monitor"), &edit);
        if monitor_may.contains(op) {
            verdict.unwrap_or_else(|e| panic!("the monitor was refused `{op}`: {e}"));
            continue;
        }
        // Refused, and the refusal names the op — an observer told only "no"
        // has nothing to act on.
        let refusal = verdict
            .expect_err(&format!("the monitor was allowed `{op}`"))
            .to_string();
        assert!(
            refusal.contains(op) && refusal.contains("Surface it to the planner"),
            "the refusal does not name `{op}` and what to do instead: {refusal}"
        );
    }

    // The node field, at schema 3 and optional, exactly as the entry writes it.
    let written = block["node"].clone();
    let node: Node = serde_json::from_value(written.clone()).expect("entry 41's node parses");
    assert_eq!(
        serde_json::to_value(&node).expect("serializes"),
        written,
        "entry 41's node does not round-trip as written"
    );
    let text = node.amendment.clone().expect("the node carries one");
    let plain = json!({"id": "solo", "persona": "engineer", "task": "## What\nx"});
    let bare: Node = serde_json::from_value(plain.clone()).expect("a node naming none parses");
    assert_eq!(bare.amendment, None);
    assert_eq!(
        serde_json::to_value(&bare).expect("serializes"),
        plain,
        "a node naming no amendment gained one on the way out"
    );

    // The heading and the precedence sentence, which are published constants,
    // and the rendering they open: the amendment's clauses are criteria of the
    // one section the entry names, above the notes, opening with the sentence
    // that states their precedence over the whole task — and never a section of
    // their own beside the criteria they override.
    assert_eq!(
        block["heading"].as_str(),
        Some(AMENDMENT_HEADING),
        "entry 41 names a different heading than this crate publishes"
    );
    assert_eq!(
        block["precedence"].as_str(),
        Some(AMENDMENT_PRECEDENCE),
        "entry 41 states a different precedence sentence than this crate publishes"
    );
    assert_ne!(AMENDMENT_HEADING, PLANNER_CONTEXT_HEADING);
    let section = block["rendered_into"]
        .as_str()
        .expect("entry 41 names the section the amendment is rendered into");
    assert!(
        section.starts_with("## ") && AMENDMENT_HEADING.starts_with("### "),
        "the amendment's heading is not a sub-heading of the section it is rendered into"
    );
    for task in [
        // A task stating the section: the amendment joins it, above the notes.
        "## What\nship it\n\n## Acceptance criteria\n\n- it ships\n\n## Additional info\n\nrun the gate.\n",
        // A task stating none: one is opened for it, above the notes.
        "## What\nship it\n\n## Additional info\n\nrun the gate.\n",
    ] {
        let rendered = Node {
            task: Some(task.into()),
            ..node.clone()
        }
        .rendered_task();
        let at = |needle: &str| {
            rendered
                .find(needle)
                .unwrap_or_else(|| panic!("{needle} is not rendered: {rendered}"))
        };
        assert!(
            at(section) < at(AMENDMENT_HEADING) && at(AMENDMENT_HEADING) < at("## Additional info"),
            "the amendment is not inside the criteria and above the notes: {rendered}"
        );
        assert_eq!(
            rendered.matches(section).count(),
            1,
            "the amendment made a second acceptance section: {rendered}"
        );
        assert!(
            !rendered.contains("\n## Amendment"),
            "the amendment is a section of its own beside the criteria: {rendered}"
        );
        assert!(
            rendered.contains(&format!("{AMENDMENT_HEADING}\n{AMENDMENT_PRECEDENCE}\n\n- {text}\n")),
            "the amendment's clause is not a criterion under the sentence stating its \
             precedence: {rendered}"
        );
        assert!(
            AMENDMENT_PRECEDENCE.contains("takes precedence over the whole task"),
            "the amendment does not state its precedence over the whole task"
        );
    }

    // The validator, named three ways, with the config key at the version the
    // entry states.
    let validator = &block["validator"];
    assert_eq!(validator["config_key"].as_str(), Some("node_validator"));
    let at = validator["config_schema_version"]
        .as_u64()
        .expect("entry 41 states the version the key arrived at");
    // The version the key **arrived** at, which stays where it is when a later
    // key moves the schema on: each key is refused by its own arrival version
    // rather than by the schema's current number, so what has to hold here is
    // that a document declaring it is still one this build reads.
    let arrived = u32::try_from(at).expect("a version fits");
    assert!(
        LAUNCH_CONFIG_SCHEMA_VERSIONS_READ.contains(&arrived)
            && arrived <= LAUNCH_CONFIG_SCHEMA_VERSION,
        "entry 41 states the validator key arrived at schema {arrived}, which is not a          version this build reads"
    );
    let named: LaunchConfig = serde_json::from_value(json!({
        "schema_version": at,
        "node_validator": "./scripts/check-node.sh",
    }))
    .expect("a launch config naming a validator parses");
    assert_eq!(
        named.node_validator.as_deref(),
        Some("./scripts/check-node.sh")
    );
    // The flag is the one `start` actually takes, asked of the parser rather
    // than of a list beside it.
    let flag = validator["flag"].as_str().expect("entry 41 names the flag");
    let parsed = Cli::try_parse_from(["onepipeline", "start", "plan.json", flag, "check-node"])
        .expect("the flag entry 41 names is one `start` takes");
    let Command::Start(started) = parsed.command else {
        panic!("that is not a start")
    };
    assert_eq!(started.node_validator.as_deref(), Some("check-node"));
    // And the ops it is offered, which the journeys drive.
    let offered: BTreeSet<String> = serde_json::from_value(validator["ops_offered"].clone())
        .expect("entry 41 names the ops a validator is offered");
    assert_eq!(
        offered,
        ["add", "amend", "requeue", "retry"]
            .into_iter()
            .map(str::to_string)
            .collect::<BTreeSet<String>>()
    );

    // The README is a **second copy** of all of this, in the prose an operator
    // meets it in, and nothing compiles that. So the entry is held against it
    // here: every name and every op above has to appear there, and the two
    // headings the guidance turns on have to be the constants this crate
    // publishes rather than prose that drifted away from them.
    let readme = std::fs::read_to_string(repo_root().join("README.md")).expect("the README ships");
    let prose = readme.split_whitespace().collect::<Vec<_>>().join(" ");
    let precedence: Vec<String> = serde_json::from_value(validator["precedence"].clone())
        .expect("entry 41 states the order it proposes");
    let mut at: Vec<usize> = Vec::new();
    for named in &precedence {
        let spelling = validator[named.as_str()]
            .as_str()
            .expect("entry 41 names it");
        let found = prose.find(spelling).unwrap_or_else(|| {
            panic!("the README does not name the validator's {named}, `{spelling}`")
        });
        at.push(found);
    }
    // And in the order the entry proposes: a README listing them the other way
    // round would read as a different rule while naming the same three things.
    assert!(
        at.windows(2).all(|pair| pair[0] < pair[1]),
        "the README names the three spellings in an order entry 41 does not propose: \
         {precedence:?} at {at:?}"
    );
    for op in offered.iter().chain(std::iter::once(&"note".to_string())) {
        assert!(
            prose.contains(&format!("`{op}`")),
            "the README's live-edit guidance does not name `{op}`"
        );
    }
    for heading in [AMENDMENT_HEADING, PLANNER_CONTEXT_HEADING] {
        assert!(
            prose.contains(heading),
            "the README does not name `{heading}`, which is where this crate renders one of \
             the two levers"
        );
    }
    // And the distinction itself, which is the whole reason both exist: the
    // README has to say which one moves the bar and which one only steers.
    assert!(
        prose.contains("adds no acceptance criteria")
            && prose.contains("the worker and the judge reviewing it read the same ruling"),
        "the README no longer states which lever changes what a node is judged against \
         and which one only steers its worker"
    );
    // And *when* it binds, which is where the two levers differ in practice and
    // where a paraphrase drifts: the journeys establish that a turn already in
    // flight is not reached, so both copies have to say so.
    let record = std::fs::read_to_string(repo_root().join("docs/contract-divergences.md"))
        .expect("the divergence record ships");
    for (which, copy) in [
        ("the README", prose.clone()),
        (
            "divergence entry 41",
            record.split_whitespace().collect::<Vec<_>>().join(" "),
        ),
    ] {
        assert!(
            copy.contains("on the dispatch that follows")
                && copy.contains("A turn already in flight is not reached"),
            "{which} no longer says which dispatch an amendment binds; the journeys \
             establish that a turn already running is not reached"
        );
    }
    // The refusal it promises for a rejected edit, and the default it promises
    // for a launch that names none.
    assert!(
        prose.contains("exit `1` refuses it with the command's own stderr as the reason")
            && prose.contains("naming none is the default and runs no validator at all"),
        "the README no longer states what a validator's answers mean"
    );
}

/// The envelope reviewer this build carries **beyond** the contract.
///
/// The contract is committed as approved and names none of it, so entry 45 is
/// the only place it is written down — and a divergence nothing gates quietly
/// stops being true. The entry's own block is the source: what parses here is
/// what an operator would write in a config and what a host would read on the
/// hook's stdin.
#[test]
fn the_envelope_reviewer_surface_is_what_the_divergence_record_names() {
    let block = divergence_block("45.");
    let reviewer = &block["reviewer"];

    // Named three ways, with the config key at the version the entry states.
    assert_eq!(reviewer["config_key"].as_str(), Some("envelope_reviewer"));
    let at = reviewer["config_schema_version"]
        .as_u64()
        .expect("entry 45 states the version the key arrived at");
    let arrived = u32::try_from(at).expect("a version fits");
    assert!(
        LAUNCH_CONFIG_SCHEMA_VERSIONS_READ.contains(&arrived)
            && arrived <= LAUNCH_CONFIG_SCHEMA_VERSION,
        "entry 45 states the reviewer key arrived at schema {arrived}, which is not a          version this build reads"
    );
    let named: LaunchConfig = serde_json::from_value(json!({
        "schema_version": at,
        "envelope_reviewer": "./scripts/review-envelope.sh",
    }))
    .expect("a launch config naming a reviewer parses");
    assert_eq!(
        named.envelope_reviewer.as_deref(),
        Some("./scripts/review-envelope.sh")
    );
    // And every version below it is still one this build reads, so a config
    // written before this hook existed still loads.
    for version in LAUNCH_CONFIG_SCHEMA_VERSIONS_READ {
        let earlier: LaunchConfig =
            serde_json::from_value(json!({"schema_version": version})).expect("it parses");
        assert_eq!(earlier.envelope_reviewer, None);
    }

    // The flag is the one `start` actually takes, asked of the parser rather
    // than of a list beside it.
    let flag = reviewer["flag"].as_str().expect("entry 45 names the flag");
    let parsed = Cli::try_parse_from(["onepipeline", "start", "plan.json", flag, "review-edit"])
        .expect("the flag entry 45 names is one `start` takes");
    let Command::Start(started) = parsed.command else {
        panic!("that is not a start")
    };
    assert_eq!(started.envelope_reviewer.as_deref(), Some("review-edit"));

    // The document, built out of this crate's own published shapes: a host
    // reading it reads a plan `Node` and a `Plan`, not a shape of this hook's
    // own.
    let document = &reviewer["document"];
    let listed: BTreeSet<String> =
        serde_json::from_value(reviewer["ops_listed_as_changes"].clone())
            .expect("entry 45 names the ops it lists as changes");
    let known: BTreeSet<String> = OPS
        .iter()
        .map(|op| (*op).to_string())
        .chain(std::iter::once("amend".to_string()))
        .collect();
    assert!(
        listed.is_subset(&known),
        "entry 45 lists an op this build has no such thing as: {listed:?}"
    );
    let changes = document["changes"]
        .as_array()
        .expect("the document carries a changes list");
    assert!(!changes.is_empty(), "{document}");
    for change in changes {
        let op = change["op"].as_str().expect("each change names its op");
        assert!(
            listed.contains(op),
            "entry 45's document carries a change under `{op}`, which it does not list"
        );
        let written = change["node"].clone();
        let node: Node = serde_json::from_value(written.clone())
            .unwrap_or_else(|e| panic!("a changed node is a plan node: {e}"));
        assert_eq!(
            serde_json::to_value(&node).expect("serializes"),
            written,
            "a changed node does not round-trip as the entry writes it"
        );
    }
    let written = document["plan"].clone();
    let plan: Plan =
        serde_json::from_value(written.clone()).expect("the plan under review is a plan");
    assert_eq!(plan.schema_version, PLAN_SCHEMA_VERSION);
    assert_eq!(
        serde_json::to_value(&plan).expect("serializes"),
        written,
        "the plan under review does not round-trip as the entry writes it"
    );
    // The goal is hoisted out of the plan it is also part of, and says the same
    // thing in both places — a document whose two copies disagreed would leave a
    // reviewer judging the edit against a goal the run is not being run for.
    assert_eq!(
        document["goal"].as_str(),
        plan.goal.as_ref().map(|goal| goal.text.as_str()),
        "{document}"
    );

    // Offered once per accepted envelope, which is the property this hook does
    // **not** inherit from the per-node validator's two offers.
    assert_eq!(
        reviewer["offers_per_accepted_envelope"].as_u64(),
        Some(1),
        "entry 45 no longer states how many times an accepted envelope is offered"
    );

    // The README is a **second copy** of all of this, in the prose an operator
    // meets it in, and nothing compiles that. So the entry is held against it
    // here: every spelling, in the order the entry proposes, and the four
    // promises a host is relying on.
    let readme = std::fs::read_to_string(repo_root().join("README.md")).expect("the README ships");
    let prose = readme.split_whitespace().collect::<Vec<_>>().join(" ");
    let precedence: Vec<String> = serde_json::from_value(reviewer["precedence"].clone())
        .expect("entry 45 states the order it proposes");
    let mut at: Vec<usize> = Vec::new();
    for spelling in &precedence {
        let named = reviewer[spelling.as_str()]
            .as_str()
            .expect("entry 45 names it");
        at.push(prose.find(named).unwrap_or_else(|| {
            panic!("the README does not name the reviewer's {spelling}, `{named}`")
        }));
    }
    assert!(
        at.windows(2).all(|pair| pair[0] < pair[1]),
        "the README names the three spellings in an order entry 45 does not propose: \
         {precedence:?} at {at:?}"
    );
    for promise in [
        "refuses the whole envelope",
        "once per envelope",
        "reviewed by nothing",
        "reported as having declared none",
        "naming none is the default and runs no reviewer at all",
    ] {
        assert!(
            prose.contains(promise),
            "the README no longer states that the envelope reviewer {promise}"
        );
    }
    // And the line a reviewer declares the node it objected to on, which a host
    // writing one reads out of the README rather than out of this record.
    let prefix = reviewer["objection_prefix"]
        .as_str()
        .expect("entry 45 states the line a reviewer declares its objection on");
    assert!(
        prose.contains(prefix),
        "the README does not state the `{prefix}` line a reviewer declares on"
    );
}

/// The write-back item budget this build carries **beyond** the contract.
///
/// The contract is committed as approved and names none of it, so entry 71 is
/// the only place it is written down — and a divergence nothing gates quietly
/// stops being true. The entry's own block is the source: what parses here is
/// what an operator would type, export, or write in a config, and the two
/// numbers beneath the three spellings are the constants the code carries.
#[test]
fn the_writeback_budget_surface_is_what_the_divergence_record_names() {
    let block = divergence_block("71.");
    let budget = &block["budget"];

    assert_eq!(budget["config_key"].as_str(), Some("writeback_item_budget"));
    let at = budget["config_schema_version"]
        .as_u64()
        .expect("entry 71 states the version the key arrived at");
    let arrived = u32::try_from(at).expect("a version fits");
    assert!(
        LAUNCH_CONFIG_SCHEMA_VERSIONS_READ.contains(&arrived)
            && arrived <= LAUNCH_CONFIG_SCHEMA_VERSION,
        "entry 71 states the budget key arrived at schema {arrived}, which is not a \
         version this build reads"
    );
    let named: LaunchConfig = serde_json::from_value(json!({
        "schema_version": at,
        "writeback_item_budget": 12,
    }))
    .expect("a launch config naming a budget parses");
    assert_eq!(named.writeback_item_budget, NonZeroU64::new(12));
    // At **no** earlier version this build reads: the key is refused by its own
    // name there, and a config written before the setting existed still loads
    // and says nothing about it.
    for version in LAUNCH_CONFIG_SCHEMA_VERSIONS_READ {
        let earlier: LaunchConfig =
            serde_json::from_value(json!({"schema_version": version})).expect("it parses");
        assert_eq!(earlier.writeback_item_budget, None);
        if version < arrived {
            let dir = std::env::temp_dir().join(format!(
                "onepipeline-contract-budget-{}-{version}",
                std::process::id()
            ));
            std::fs::create_dir_all(&dir).expect("a scratch directory");
            let path = dir.join("launch.yaml");
            std::fs::write(
                &path,
                format!("schema_version: {version}\nwriteback_item_budget: 12\n"),
            )
            .expect("the config is written");
            let refused = LaunchConfig::load(&path)
                .expect_err("a version that never had the key refuses it")
                .to_string();
            assert!(
                refused.contains("`writeback_item_budget`")
                    && refused.contains(&format!("schema {arrived} key")),
                "schema {version} did not refuse the key by its name: {refused}"
            );
            let _ = std::fs::remove_dir_all(&dir);
        }
    }

    // The flag is the one `start` actually takes, asked of the parser rather
    // than of a list beside it.
    let flag = budget["flag"].as_str().expect("entry 71 names the flag");
    let parsed = Cli::try_parse_from(["onepipeline", "start", "plan.json", flag, "12"])
        .expect("the flag entry 71 names is one `start` takes");
    let Command::Start(started) = parsed.command else {
        panic!("that is not a start")
    };
    assert_eq!(started.writeback_item_budget, Some(12));
    // And naming none leaves the flag unset, so the rungs beneath it are consulted.
    let parsed = Cli::try_parse_from(["onepipeline", "start", "plan.json"]).expect("it parses");
    let Command::Start(unset) = parsed.command else {
        panic!("that is not a start")
    };
    assert_eq!(unset.writeback_item_budget, None);

    // The environment name is the constant the engine reads, and the precedence
    // is the order the driver resolves in — the same order the two hooks take.
    assert_eq!(
        budget["environment"].as_str(),
        Some(WRITEBACK_ITEM_BUDGET_ENV),
        "entry 71 names a different variable than the engine reads"
    );
    let precedence: Vec<String> = serde_json::from_value(budget["precedence"].clone())
        .expect("entry 71 states the order it proposes");
    assert_eq!(precedence, ["flag", "environment", "config_key"]);

    // The two numbers beneath the spellings are the constants the code carries.
    assert_eq!(
        budget["default_seconds"].as_u64(),
        Some(DEFAULT_WRITEBACK_ITEM_BUDGET_SECONDS.get()),
        "entry 71 states a different shipped default than the code carries"
    );
    assert_eq!(
        budget["floor_seconds"].as_u64(),
        Some(WRITEBACK_COMMAND_FLOOR_SECONDS),
        "entry 71 states a different floor than the code carries"
    );
    // A shipped default of zero would be no budget at all, so it is held against
    // the block's own number rather than asserted constant.
    assert!(
        budget["default_seconds"]
            .as_u64()
            .is_some_and(|seconds| seconds > 0),
        "entry 71 states a shipped default of zero"
    );
    // The formula the entry states is the one each example below is computed by
    // — the floor or the product, whichever is larger — so a block whose formula
    // and examples disagreed would fail here rather than read as two truths.
    assert_eq!(
        budget["deadline"].as_str(),
        Some("max(floor_seconds, budget × items)"),
        "entry 71 states a deadline other than the arithmetic its examples are held to"
    );
    // And each worked example is that arithmetic, computed from those two
    // constants, with the refusal naming the one that governed.
    let examples = budget["examples"]
        .as_array()
        .expect("entry 71 works an example");
    assert!(!examples.is_empty(), "{budget}");
    for example in examples {
        let items = example["items"].as_u64().expect("an item count");
        let per_item = example["budget_seconds"].as_u64().expect("a budget");
        let deadline = example["deadline_seconds"].as_u64().expect("a deadline");
        assert_eq!(
            deadline,
            (per_item * items).max(WRITEBACK_COMMAND_FLOOR_SECONDS),
            "entry 71's example is not the arithmetic it states: {example}"
        );
        let refusal = example["refusal"].as_str().expect("the refusal");
        assert!(
            refusal.starts_with(&format!("project-copy exceeded {deadline} seconds ("))
                && refusal.contains(&format!("{items} items × {per_item} seconds per item")),
            "entry 71's refusal does not name what it computed: {refusal}"
        );
        assert_eq!(
            refusal.contains("floor"),
            per_item * items < WRITEBACK_COMMAND_FLOOR_SECONDS,
            "entry 71's refusal says the floor governed where it did not, or the reverse: \
             {refusal}"
        );
    }

    // The README is a **second copy** of all of this, in the prose an operator
    // meets it in, and nothing compiles that. So the entry is held against it
    // here: every spelling, in the order the entry proposes, and the default.
    let readme = std::fs::read_to_string(repo_root().join("README.md")).expect("the README ships");
    let prose = readme.split_whitespace().collect::<Vec<_>>().join(" ");
    let mut found: Vec<usize> = Vec::new();
    for spelling in &precedence {
        let named = budget[spelling.as_str()]
            .as_str()
            .expect("entry 71 names it");
        found.push(prose.find(named).unwrap_or_else(|| {
            panic!("the README does not name the budget's {spelling}, `{named}`")
        }));
    }
    assert!(
        found.windows(2).all(|pair| pair[0] < pair[1]),
        "the README names the three spellings in an order entry 71 does not propose: \
         {precedence:?} at {found:?}"
    );
    for promise in [
        "multiplied by the number of items",
        "never below the sixty-second floor",
        "zero is refused",
        "naming none takes ten seconds per item",
    ] {
        assert!(
            prose.contains(promise),
            "the README no longer states that the write-back budget is {promise}"
        );
    }
}

/// The run-end hooks, as the contract's own block names them, are what this
/// build takes, reads and emits.
///
/// The block is the source: the flags are asked of the parser, the keys of the
/// launch config at the version the block states and at none before it, the
/// default of the constant a launch resolves to, and the kinds of the enum that
/// emits them. What a hook is handed and what `results` renders are driven
/// against the real binary by `tests/e2e/run_end_hooks.rs`, out of this same
/// block.
#[test]
fn the_run_end_hooks_surface_is_what_the_contract_names() {
    let block: Value = serde_json::from_str(&fenced_block_naming("json", "run_end_hooks"))
        .expect("the run-end hooks block is JSON");
    let hooks = &block["run_end_hooks"];
    let spelled = |group: &str, which: &str| -> String {
        hooks[group][which]
            .as_str()
            .unwrap_or_else(|| panic!("the block names no {group}.{which}"))
            .to_string()
    };

    assert!(
        CONTRACT
            .contains("[--success-hook COMMAND] [--failure-hook COMMAND] [--hook-timeout SECONDS]"),
        "the driver invocation no longer names the run-end hook flags"
    );
    let parsed = Cli::try_parse_from([
        "onepipeline".to_string(),
        "start".to_string(),
        "plans:demo".to_string(),
        spelled("flags", "success"),
        "./follow-up.sh".to_string(),
        spelled("flags", "failure"),
        "./report-failure.sh".to_string(),
        spelled("flags", "timeout"),
        "30".to_string(),
    ])
    .expect("the flags the block names are ones `start` takes");
    let Command::Start(started) = parsed.command else {
        panic!("that is not a start")
    };
    assert_eq!(started.success_hook.as_deref(), Some("./follow-up.sh"));
    assert_eq!(started.failure_hook.as_deref(), Some("./report-failure.sh"));
    assert_eq!(started.hook_timeout, NonZeroU64::new(30));
    let Command::Start(unset) = Cli::try_parse_from(["onepipeline", "start", "plans:demo"])
        .expect("it parses")
        .command
    else {
        panic!("that is not a start")
    };
    assert_eq!(
        (unset.success_hook, unset.failure_hook, unset.hook_timeout),
        (None, None, None),
        "a launch naming no hook named one"
    );

    // The keys, at the version the block states and refused by name before it.
    let at = hooks["config_schema_version"]
        .as_u64()
        .expect("the block states the version the keys arrived at");
    let arrived = u32::try_from(at).expect("a version fits");
    assert!(
        LAUNCH_CONFIG_SCHEMA_VERSIONS_READ.contains(&arrived)
            && arrived <= LAUNCH_CONFIG_SCHEMA_VERSION,
        "the block states the hook keys arrived at schema {arrived}, which this build does not read"
    );
    let keys = [
        (spelled("config_keys", "success"), json!("./follow-up.sh")),
        (
            spelled("config_keys", "failure"),
            json!("./report-failure.sh"),
        ),
        (spelled("config_keys", "timeout"), json!(45)),
    ];
    let mut document = serde_json::Map::new();
    document.insert("schema_version".into(), json!(at));
    for (key, value) in &keys {
        document.insert(key.clone(), value.clone());
    }
    let named: LaunchConfig =
        serde_json::from_value(Value::Object(document)).expect("a config naming both hooks parses");
    assert_eq!(named.success_hook.as_deref(), Some("./follow-up.sh"));
    assert_eq!(named.failure_hook.as_deref(), Some("./report-failure.sh"));
    assert_eq!(named.hook_timeout, NonZeroU64::new(45));
    let dir =
        std::env::temp_dir().join(format!("onepipeline-contract-hooks-{}", std::process::id()));
    std::fs::create_dir_all(&dir).expect("a scratch directory");
    for version in LAUNCH_CONFIG_SCHEMA_VERSIONS_READ
        .into_iter()
        .filter(|version| *version < arrived)
    {
        for (key, value) in &keys {
            let path = dir.join(format!("{version}-{key}.yaml"));
            std::fs::write(
                &path,
                format!("schema_version: {version}\n{key}: {value}\n"),
            )
            .expect("the config is written");
            let refused = LaunchConfig::load(&path)
                .expect_err("a version that never had the key refuses it")
                .to_string();
            assert!(
                refused.contains(&format!("`{key}`"))
                    && refused.contains(&format!("schema {arrived} key")),
                "schema {version} did not refuse `{key}` by its name: {refused}"
            );
        }
    }
    let _ = std::fs::remove_dir_all(&dir);
    let mut zero = serde_json::Map::new();
    zero.insert("schema_version".into(), json!(at));
    zero.insert(spelled("config_keys", "timeout"), json!(0));
    let refused = serde_json::from_value::<LaunchConfig>(Value::Object(zero))
        .expect_err("a timeout of zero is refused")
        .to_string();
    assert!(refused.contains("zero"), "{refused}");
    assert_eq!(
        hooks["default_timeout_seconds"].as_u64(),
        Some(DEFAULT_HOOK_TIMEOUT_SECONDS.get()),
        "the block states a different shipped timeout than the code carries"
    );

    // The kinds are the enum's own and the contract's list names each.
    let payloads = hooks["payloads"]
        .as_object()
        .expect("the block names the payloads");
    let tokens = backticked();
    assert_eq!(payloads.len(), 3, "{payloads:?}");
    for kind in payloads.keys() {
        assert!(
            PipelineKind::from_wire(&EventKind(kind.clone())).is_some(),
            "`{kind}` is not a kind this build emits"
        );
        assert!(
            tokens.contains(kind),
            "the contract's kind list does not name `{kind}`"
        );
    }

    // And the block agrees with itself: the reason a hook is handed is the one its
    // firing records, in one of the kinds it states, over nodes of one shape.
    let endings: Vec<String> =
        serde_json::from_value(hooks["endings"].clone()).expect("the block names the endings");
    let reason_kinds: Vec<String> = serde_json::from_value(hooks["reason_kinds"].clone())
        .expect("the block names the reason kinds");
    assert!(endings.contains(
        &payloads["run-hook-finished"]["ending"]
            .as_str()
            .expect("an ending")
            .to_string()
    ));
    assert_eq!(hooks["stdin"]["success"]["reason"], Value::Null);
    let failure = &hooks["stdin"]["failure"]["reason"];
    assert!(reason_kinds.contains(&failure["kind"].as_str().expect("a kind").to_string()));
    assert_eq!(failure, &payloads["run-hook-fired"]["reason"]);
    for node in failure["nodes"].as_array().expect("the reason lists nodes") {
        let fields: BTreeSet<&str> = node
            .as_object()
            .expect("a node is an object")
            .keys()
            .map(String::as_str)
            .collect();
        assert_eq!(fields, BTreeSet::from(["id", "status", "outcome"]));
    }

    // The README is a **second copy** of the operator-facing half, in the prose an
    // operator meets it in, and nothing compiles that. So it is held to the block
    // here: every spelling the block names, where a hook's output is kept, the
    // shipped timeout, and the promises that make the two hooks safe to wire.
    let readme = std::fs::read_to_string(repo_root().join("README.md")).expect("the README ships");
    let prose = readme.split_whitespace().collect::<Vec<_>>().join(" ");
    for group in ["flags", "config_keys"] {
        for which in ["success", "failure", "timeout"] {
            let named = spelled(group, which);
            assert!(
                prose.contains(&named),
                "the README does not name the run-end hooks' {group} `{named}`"
            );
        }
    }
    let environment: Vec<String> = serde_json::from_value(hooks["environment"].clone())
        .expect("the block names a hook's environment");
    for variable in &environment {
        assert!(
            prose.contains(variable.as_str()),
            "the README does not name `{variable}`, which a hook is given"
        );
    }
    let log = hooks["log"].as_str().expect("the block names the log");
    assert!(
        prose.contains(log),
        "the README does not say a hook's output is kept in `{log}`"
    );
    assert_eq!(
        DEFAULT_HOOK_TIMEOUT_SECONDS.get(),
        600,
        "the README's shipped timeout is written in words; move them with the constant"
    );
    for promise in [
        "six hundred seconds when unnamed, zero refused",
        "has not ended and fires neither",
        "a hook never changes how the run settled",
    ] {
        assert!(
            prose.contains(promise),
            "the README no longer states that {promise}"
        );
    }

    // The **epoch** half of that copy is held to the contract as well as stated,
    // because it is the half a reader acts on — which endings fire a hook — and a
    // README that drifted from it would tell an operator to expect a hook the
    // engine never fires. Each phrase is one sentence both documents carry word for
    // word, so rewording the rule fails here until both move together. The three
    // promises above are older paraphrases the two documents never shared; widening
    // this to them would be a rewrite of the contract's prose rather than a gate.
    let contract = CONTRACT.split_whitespace().collect::<Vec<_>>().join(" ");
    for promise in [
        "at most once per ending",
        "judged by what the edit left behind, not by which operation it carried",
        "a run recovered from a failure still fires the hook it then reaches",
        "a settle that carries a failed ending to a complete one",
    ] {
        assert!(
            contract.contains(promise),
            "the contract no longer states that {promise}"
        );
        assert!(
            prose.contains(promise),
            "the README states the epoch rule differently from the contract: {promise}"
        );
    }
}

/// The dispatch-env hook, as the contract's own block names it, is what this
/// build takes and reads.
///
/// The block is the source, on the run-end hooks' terms: the flags are asked of
/// the parser, the keys of the launch config at the version the block states and
/// at none before it, the default of the constant a launch resolves to, the
/// document of the reader that reads a hook's stdout — driven through the real
/// binary by `tests/e2e/dispatch_env_hook.rs`, where every environment name, the
/// log, each ending and the refusal outcome are read out of this same block —
/// and the README's copy of the operator-facing half to every spelling here.
/// The run-history surface: the environment the engine overlays, the key names,
/// the scope words, the pointer file's name and the merge rule are the block's,
/// and the crate's constants are held to it — both directions, so a word the
/// crate renames fails here and so does one the block keeps after the crate
/// dropped it. The CLI takes the two spellings the paragraph states and
/// refuses the one it does not.
#[test]
fn the_run_history_surface_is_what_the_contract_names() {
    use onepipeline::agents;

    let block: Value = serde_json::from_str(&fenced_block_naming("json", "oneharness_history"))
        .expect("the run-history block is JSON");
    let history = &block["oneharness_history"];
    let spelled = |path: &[&str]| -> String {
        path.iter()
            .fold(history, |value, key| &value[*key])
            .as_str()
            .unwrap_or_else(|| panic!("the block names no {}", path.join(".")))
            .to_string()
    };

    assert_eq!(spelled(&["environment", "history"]), agents::HISTORY_ENV);
    assert_eq!(
        spelled(&["environment", "pointer_file"]),
        agents::POINTER_FILE_ENV
    );
    assert_eq!(spelled(&["environment", "labels"]), agents::LABELS_ENV);
    assert_eq!(spelled(&["never_set"]), agents::HISTORY_DIR_ENV);
    assert_eq!(spelled(&["pointer_file"]), agents::SESSIONS_FILE);
    assert_eq!(
        RunPaths::under(Path::new("/runs"), "demo")
            .oneharness_sessions()
            .file_name()
            .and_then(|name| name.to_str()),
        Some(agents::SESSIONS_FILE)
    );
    assert_eq!(spelled(&["label_prefix"]), agents::LABEL_PREFIX);
    let keys = [
        ("run_id", agents::RUN_ID_LABEL),
        ("project", agents::PROJECT_LABEL),
        ("scope", agents::SCOPE_LABEL),
        ("node", agents::NODE_LABEL),
        ("step", agents::STEP_LABEL),
        ("attempt", agents::ATTEMPT_LABEL),
    ];
    for (name, constant) in keys {
        assert_eq!(spelled(&["labels", name]), constant, "labels.{name}");
        assert!(
            constant.starts_with(agents::LABEL_PREFIX),
            "{constant} is not under the engine's prefix"
        );
    }
    let named_keys: BTreeSet<String> = history["labels"]
        .as_object()
        .expect("labels is an object")
        .keys()
        .cloned()
        .collect();
    assert_eq!(
        named_keys,
        keys.iter().map(|(name, _)| (*name).to_string()).collect(),
        "the block names a key the crate does not, or the crate one the block does not"
    );
    let scopes: std::collections::BTreeMap<String, String> =
        serde_json::from_value(history["scopes"].clone()).expect("the scopes are words");
    assert_eq!(
        scopes.values().cloned().collect::<BTreeSet<_>>(),
        agents::Scope::ALL
            .iter()
            .map(|scope| scope.as_str().to_string())
            .collect(),
        "the scope words are not the crate's"
    );

    // The merge rule, on the block's own example: the inherited repository key
    // survives, the stale engine key is replaced, and the wire is key-ordered.
    let stamp = &history["stamp"];
    let attempt =
        u32::try_from(stamp["attempt"].as_u64().expect("an attempt")).expect("an attempt fits");
    let attempt = NonZeroU32::new(attempt).expect("an attempt counts from one");
    let node = stamp["node"].as_str().expect("the example names a node");
    let launched = match spelled(&["stamp", "scope"]).as_str() {
        "node" => agents::Launched::Node {
            node,
            step: stamp["step"].as_str(),
            attempt,
        },
        "pr-author" => agents::Launched::PrAuthor { node, attempt },
        "observer" => agents::Launched::Observer,
        other => panic!("the example's scope `{other}` is not one of the words"),
    };
    assert_eq!(
        launched.scope().as_str(),
        spelled(&["stamp", "scope"]),
        "the launch's scope word is not the example's"
    );
    let composed = agents::compose_labels(
        Some(&spelled(&["inherited"])),
        &agents::Stamp {
            run: &spelled(&["stamp", "run_id"]),
            project: stamp["project"].as_str(),
            launched,
        },
    )
    .expect("the block's example composes");
    assert_eq!(composed, spelled(&["composed"]));

    let at = history["summary_schema_version"]
        .as_u64()
        .expect("the block states the summary version the field arrived at");
    assert_eq!(
        u64::from(SUMMARY_SCHEMA_VERSION),
        at,
        "the summary schema moved past the version the block states"
    );
    assert!(
        CONTRACT.contains(&spelled(&["opt_out"])),
        "the paragraph no longer states the opt-out"
    );

    // The CLI: a run with or without a node, a project, and never both.
    for args in [
        vec!["onepipeline", "agents", "demo"],
        vec!["onepipeline", "agents", "demo", "service"],
        vec!["onepipeline", "agents", "--project", "plans:demo"],
    ] {
        let parsed = Cli::try_parse_from(args.clone())
            .unwrap_or_else(|error| panic!("{args:?} is not a spelling `agents` takes: {error}"));
        assert!(
            matches!(parsed.command, Command::Agents(_)),
            "{args:?} parsed as something else"
        );
    }
    for args in [
        vec!["onepipeline", "agents"],
        vec!["onepipeline", "agents", "demo", "--project", "plans:demo"],
    ] {
        Cli::try_parse_from(args.clone())
            .err()
            .unwrap_or_else(|| panic!("{args:?} is a spelling `agents` refuses"));
    }
    assert_contract_names(
        "agents verb",
        &[
            "onepipeline agents RUN [NODE]",
            "onepipeline agents --project PROJECT",
            "verbs::agents(&RunPaths, AgentScope::Run | AgentScope::Node(NODE)) -> Agents",
            "verbs::project_agents(root, PROJECT) -> Agents",
            "RunPaths::oneharness_sessions()",
            "RunSummary.oneharness_sessions",
            "DispatchRequest.attempt",
        ],
    );
}

#[test]
fn the_dispatch_env_hook_surface_is_what_the_contract_names() {
    let block: Value = serde_json::from_str(&fenced_block_naming("json", "dispatch_env_hook"))
        .expect("the dispatch-env hook block is JSON");
    let hook = &block["dispatch_env_hook"];
    let spelled = |group: &str, which: &str| -> String {
        hook[group][which]
            .as_str()
            .unwrap_or_else(|| panic!("the block names no {group}.{which}"))
            .to_string()
    };

    assert!(
        CONTRACT.contains("[--dispatch-env-hook COMMAND] [--dispatch-env-hook-timeout SECONDS]"),
        "the driver invocation no longer names the dispatch-env hook flags"
    );
    let parsed = Cli::try_parse_from([
        "onepipeline".to_string(),
        "start".to_string(),
        "plans:demo".to_string(),
        spelled("flags", "command"),
        "./dispatch-env.sh".to_string(),
        spelled("flags", "timeout"),
        "30".to_string(),
    ])
    .expect("the flags the block names are ones `start` takes");
    let Command::Start(started) = parsed.command else {
        panic!("that is not a start")
    };
    assert_eq!(
        started.dispatch_env_hook.as_deref(),
        Some("./dispatch-env.sh")
    );
    assert_eq!(started.dispatch_env_hook_timeout, NonZeroU64::new(30));
    let Command::Start(unset) = Cli::try_parse_from(["onepipeline", "start", "plans:demo"])
        .expect("it parses")
        .command
    else {
        panic!("that is not a start")
    };
    assert_eq!(
        (unset.dispatch_env_hook, unset.dispatch_env_hook_timeout),
        (None, None),
        "a launch naming no hook named one"
    );
    // `adopt` takes neither flag.
    for flag in ["command", "timeout"] {
        let refused = Cli::try_parse_from([
            "onepipeline".to_string(),
            "adopt".to_string(),
            "demo".to_string(),
            spelled("flags", flag),
            "x".to_string(),
        ])
        .expect_err("adopt takes no dispatch-env hook flag");
        assert!(
            refused.to_string().contains(&spelled("flags", flag)),
            "{refused}"
        );
    }

    // The keys, at the version the block states and refused by name before it.
    let at = hook["config_schema_version"]
        .as_u64()
        .expect("the block states the version the keys arrived at");
    let arrived = u32::try_from(at).expect("a version fits");
    assert!(
        LAUNCH_CONFIG_SCHEMA_VERSIONS_READ.contains(&arrived)
            && arrived <= LAUNCH_CONFIG_SCHEMA_VERSION,
        "the block states the hook keys arrived at schema {arrived}, which this build does not read"
    );
    let keys = [
        (
            spelled("config_keys", "command"),
            json!("./dispatch-env.sh"),
        ),
        (spelled("config_keys", "timeout"), json!(45)),
    ];
    let mut document = serde_json::Map::new();
    document.insert("schema_version".into(), json!(at));
    for (key, value) in &keys {
        document.insert(key.clone(), value.clone());
    }
    let named: LaunchConfig =
        serde_json::from_value(Value::Object(document)).expect("a config naming the hook parses");
    assert_eq!(
        named.dispatch_env_hook.as_deref(),
        Some("./dispatch-env.sh")
    );
    assert_eq!(named.dispatch_env_hook_timeout, NonZeroU64::new(45));
    let dir = std::env::temp_dir().join(format!(
        "onepipeline-contract-dispatch-env-{}",
        std::process::id()
    ));
    std::fs::create_dir_all(&dir).expect("a scratch directory");
    for version in LAUNCH_CONFIG_SCHEMA_VERSIONS_READ
        .into_iter()
        .filter(|version| *version < arrived)
    {
        for (key, value) in &keys {
            let path = dir.join(format!("{version}-{key}.yaml"));
            std::fs::write(
                &path,
                format!("schema_version: {version}\n{key}: {value}\n"),
            )
            .expect("the config is written");
            let refused = LaunchConfig::load(&path)
                .expect_err("a version that never had the key refuses it")
                .to_string();
            assert!(
                refused.contains(&format!("`{key}`"))
                    && refused.contains(&format!("schema {arrived} key")),
                "schema {version} did not refuse `{key}` by its name: {refused}"
            );
        }
    }
    let _ = std::fs::remove_dir_all(&dir);
    let mut zero = serde_json::Map::new();
    zero.insert("schema_version".into(), json!(at));
    zero.insert(spelled("config_keys", "timeout"), json!(0));
    let refused = serde_json::from_value::<LaunchConfig>(Value::Object(zero))
        .expect_err("a timeout of zero is refused")
        .to_string();
    assert!(refused.contains("zero"), "{refused}");
    assert_eq!(
        hook["default_timeout_seconds"].as_u64(),
        Some(DEFAULT_DISPATCH_ENV_HOOK_TIMEOUT_SECONDS.get()),
        "the block states a different shipped timeout than the code carries"
    );

    // The block agrees with itself and with the prose: the document it shows is
    // the one the prose states, the hook's name is what `ONEPIPELINE_HOOK` says
    // and the stem of its log, and the outcome a refusal settles under is one
    // the contract already names.
    let stdout = &hook["stdout"];
    assert_eq!(stdout["version"], json!(1));
    let fields: BTreeSet<&str> = stdout
        .as_object()
        .expect("the document is an object")
        .keys()
        .map(String::as_str)
        .collect();
    assert_eq!(fields, BTreeSet::from(["version", "env"]));
    for (name, value) in stdout["env"].as_object().expect("`env` is an object") {
        assert!(
            value.is_string(),
            "`env.{name}` is not a string in the block"
        );
    }
    let name = hook["hook"].as_str().expect("the block names the hook");
    assert_eq!(
        hook["log"].as_str(),
        Some(format!("hooks/{name}.log").as_str())
    );
    assert!(
        CONTRACT.contains(&format!("`ONEPIPELINE_HOOK={name}`")),
        "the prose does not say what ONEPIPELINE_HOOK says"
    );
    let environment: Vec<String> = serde_json::from_value(hook["environment"].clone())
        .expect("the block names the hook's environment");
    let tokens = backticked();
    for variable in &environment {
        assert!(
            tokens.contains(variable),
            "the contract's prose does not name `{variable}`, which the block says a hook is given"
        );
    }
    let endings: Vec<String> =
        serde_json::from_value(hook["endings"].clone()).expect("the block names the endings");
    assert_eq!(endings, ["exit", "timeout", "could-not-start", "malformed"]);
    // The outcome word is the engine's own and private, so it is held to the
    // contract's own closed list of them rather than imported: the prose that
    // names every outcome a node settles under names this one already.
    let outcome = hook["refusal_outcome"]
        .as_str()
        .expect("the block names the outcome a refused launch settles under");
    assert_eq!(outcome, "infrastructure-failure");
    assert!(
        tokens.contains(outcome),
        "a refused launch settles under a word the contract does not already name"
    );

    // The README is a **second copy** of the operator-facing half, held to the
    // block as the run-end hooks' is.
    let readme = std::fs::read_to_string(repo_root().join("README.md")).expect("the README ships");
    let prose = readme.split_whitespace().collect::<Vec<_>>().join(" ");
    for group in ["flags", "config_keys"] {
        for which in ["command", "timeout"] {
            let named = spelled(group, which);
            assert!(
                prose.contains(&named),
                "the README does not name the dispatch-env hook's {group} `{named}`"
            );
        }
    }
    for variable in &environment {
        assert!(
            prose.contains(variable.as_str()),
            "the README does not name `{variable}`, which the hook is given"
        );
    }
    let log = hook["log"].as_str().expect("the block names the log");
    assert!(
        prose.contains(log),
        "the README does not say the hook's stderr is kept in `{log}`"
    );
    assert_eq!(
        DEFAULT_DISPATCH_ENV_HOOK_TIMEOUT_SECONDS.get(),
        60,
        "the README's shipped timeout is written in words; move them with the constant"
    );
    for promise in [
        "sixty seconds when unnamed, zero refused",
        "for that one child launch only",
        "no value it prints is written anywhere the run keeps",
        "Naming no hook dispatches exactly as before",
    ] {
        assert!(
            prose.contains(promise),
            "the README no longer states that {promise}"
        );
    }
}

/// The write-back's refusal rule this build carries **beyond** the contract.
///
/// The contract says write-back is "retried off the reconcile loop" and nothing
/// narrower, so entry 72 is the only place the narrowing is written down — and a
/// divergence nothing gates quietly stops being true. The entry's own block is the
/// source: the member the worker branches on, the value that stops its timer, the
/// commands whose failures it reads and the partial-answer rule are each the
/// constant the code carries. The contract's own sentence is held too, because it
/// is approved as written and this entry is a proposal against it rather than an
/// edit of it.
#[test]
fn the_writeback_refusal_rule_is_what_the_divergence_record_names() {
    let block = divergence_block("72.");
    let rule = &block["failure"];
    assert_eq!(
        rule["member"].as_str(),
        Some(WRITEBACK_FAILURE_CLASS_MEMBER),
        "entry 72 names a different member than the worker branches on"
    );
    assert_eq!(
        rule["stops_the_timer"].as_str(),
        Some(WRITEBACK_REFUSED_CLASS),
        "entry 72 names a different class than the one that stops the retry timer"
    );
    assert_eq!(
        rule["failure_document_exit"].as_i64(),
        Some(i64::from(WRITEBACK_FAILURE_EXIT)),
        "entry 72 reads the failure document under a different exit than the worker does"
    );
    assert_eq!(
        rule["commands"],
        json!(WRITEBACK_CLASSIFIED_COMMANDS),
        "entry 72 names different commands than the worker reads a class off"
    );

    let partial = &rule["partial_answer"];
    assert_eq!(
        partial["exit"].as_i64(),
        Some(i64::from(WRITEBACK_PARTIAL_EXIT)),
        "entry 72 reads a partial answer under a different exit than the worker does"
    );
    assert_eq!(
        partial["member"].as_str(),
        Some(WRITEBACK_PARTIAL_CLASS_MEMBER),
        "entry 72 names a different partial-answer member than the worker reads"
    );
    assert_eq!(
        partial["refused_when"].as_str(),
        Some("every"),
        "entry 72 states a partial answer is refused on something other than every entry"
    );

    let contract =
        std::fs::read_to_string(repo_root().join("docs/contract.md")).expect("the contract reads");
    assert!(
        contract.contains("Write-back is best effort and retried off the reconcile loop"),
        "the sentence entry 72 proposes to narrow is no longer the contract's"
    );
}

/// The JSON type one written value is, in the words entry 73's `fields` use.
fn json_type(value: &Value) -> &'static str {
    match value {
        Value::Null => "null",
        Value::Bool(_) => "boolean",
        Value::Number(number) if number.is_u64() || number.is_i64() => "integer",
        Value::Number(_) => "number",
        Value::String(_) => "string",
        Value::Array(_) => "array",
        Value::Object(_) => "object",
    }
}

/// The write-back's projection record and member copy this build carries **beyond** the
/// contract.
///
/// The contract says every accepted edit updates the project's tasks and names no record of
/// it, so entry 73 is the only place either is written down — and a divergence nothing gates
/// quietly stops being true. The entry's own block is the source: the two paths and the release
/// are the constants the worker writes and compares by, the member projection's reads are the
/// commands it runs, and every field, type and admitted value of the record is held against the
/// record type on each shape of line it writes, with the entry's example read and written back
/// unchanged.
#[test]
fn the_writeback_projection_record_is_what_the_divergence_record_names() {
    let block = divergence_block("73.");
    let projection = &block["projection"];
    let detection = &block["detection"];
    let member = &block["member_projection"];

    assert_eq!(
        projection["record"].as_str(),
        Some(format!("<run dir>/{WRITEBACK_PROJECTIONS_FILE}").as_str()),
        "entry 73 names a different record than the worker appends to"
    );
    assert_eq!(
        detection["record"].as_str(),
        Some(format!("<run dir>/{WRITEBACK_STORE_FILE}").as_str()),
        "entry 73 keeps the store's answer somewhere other than the worker does"
    );
    assert_eq!(
        detection["members_from"].as_str(),
        Some(WRITEBACK_MEMBERS_FROM),
        "entry 73 names a different first release offering a member copy"
    );
    assert_eq!(
        detection["command"].as_str(),
        Some("onetaskgraph --version")
    );
    assert_eq!(
        member["reads"],
        json!([WRITEBACK_CLASSIFIED_COMMANDS[0], WRITEBACK_MEMBER_READ]),
        "entry 73 names other reads for a member projection than the worker runs"
    );
    assert_eq!(
        member["never_reads"],
        json!([WRITEBACK_CLASSIFIED_COMMANDS[1]]),
        "entry 73 no longer says a member projection runs no page of tasks"
    );

    // The example is one line of the record, and written back it is the same line.
    let example = &projection["example"];
    let record: ProjectionRecord = serde_json::from_value(example.clone())
        .unwrap_or_else(|error| panic!("entry 73's example is not a record: {error}"));
    assert_eq!(
        serde_json::to_value(&record).expect("a record serializes"),
        *example,
        "entry 73's example does not write back as itself"
    );

    // Every shape of line the worker writes: a member copy that landed, a whole one that failed
    // with the store's class, and one that landed with no report read.
    let lines = [
        record.clone(),
        ProjectionRecord {
            scope: ProjectionScope::Whole(WholeBecause::AfterFailure),
            ended: ProjectionEnded::Failed {
                classified: Some(ProjectionFailure {
                    class: FailureClass::Refused,
                    kind: "stale-origin".to_owned(),
                }),
                reason: "copy exited 1: the store's own words".to_owned(),
            },
            ..record.clone()
        },
        ProjectionRecord {
            scope: ProjectionScope::Whole(WholeBecause::First),
            ended: ProjectionEnded::Projected {
                actions: None,
                spent: None,
            },
            ..record.clone()
        },
        ProjectionRecord {
            scope: ProjectionScope::Whole(WholeBecause::StoreLacksMembers),
            ended: ProjectionEnded::Failed {
                classified: None,
                reason: "project-copy exceeded 60 seconds".to_owned(),
            },
            ..record.clone()
        },
        // A partial copy: the deliverer's own write landed and a ticket it delivers did not.
        ProjectionRecord {
            scope: ProjectionScope::Whole(WholeBecause::First),
            ended: ProjectionEnded::Failed {
                classified: Some(ProjectionFailure {
                    class: FailureClass::Transient,
                    kind: "unavailable".to_owned(),
                }),
                reason: "the copy landed, but the store could not keep every delivered ticket \
                         in step"
                    .to_owned(),
            },
            delivered: vec![json!({
                "ticket": "tickets:t/one", "deliverer": "plans:p/a", "outcome": "failed",
                "from": "queued",
                "failure": {"class": "transient", "kind": "unavailable", "source": "tickets",
                            "message": "cannot write", "retry_after_seconds": null}
            })
            .as_object()
            .cloned()
            .expect("a delivered entry is an object")],
            ..record.clone()
        },
    ];
    let fields = projection["fields"]
        .as_object()
        .expect("entry 73 names the record's fields");
    // A field the entry says a line leaves off under a condition is the one kind of key a line
    // may lack; every other key is on every line.
    let always: BTreeSet<&String> = fields
        .iter()
        .filter(|(_, field)| field["omitted_when"].is_null())
        .map(|(name, _)| name)
        .collect();
    let mut written_keys: BTreeSet<String> = BTreeSet::new();
    let mut written_words: std::collections::BTreeMap<String, BTreeSet<String>> =
        std::collections::BTreeMap::new();
    for line in &lines {
        let written = serde_json::to_value(line).expect("a record serializes");
        let written = written.as_object().expect("a record is one object");
        let keys: BTreeSet<&String> = written.keys().collect();
        assert!(
            always.is_subset(&keys) && keys.iter().all(|key| fields.contains_key(*key)),
            "the record writes other keys than entry 73 names: {keys:?}"
        );
        written_keys.extend(written.keys().cloned());
        if line.delivered.is_empty() {
            assert!(
                !written.contains_key("delivered"),
                "a line that reached no ticket wrote `delivered`: {written:?}"
            );
        }
        for (name, field) in fields {
            let Some(value) = written.get(name) else {
                continue;
            };
            let types: Vec<&str> = match &field["type"] {
                Value::String(one) => vec![one.as_str()],
                Value::Array(several) => several.iter().filter_map(Value::as_str).collect(),
                other => panic!("entry 73 types `{name}` as {other}"),
            };
            assert!(
                types.contains(&json_type(value)),
                "`{name}` is written as {} where entry 73 types it {types:?}: {value}",
                json_type(value)
            );
            if let (Some(admitted), Some(word)) = (field["values"].as_array(), value.as_str()) {
                assert!(
                    admitted.contains(&json!(word)),
                    "`{name}` is written as `{word}`, which entry 73 does not admit"
                );
                written_words
                    .entry(name.clone())
                    .or_default()
                    .insert(word.to_owned());
            }
            if let (Some(members), Some(object)) = (field["members"].as_array(), value.as_object())
            {
                assert_eq!(
                    object.keys().map(String::as_str).collect::<BTreeSet<_>>(),
                    members
                        .iter()
                        .filter_map(Value::as_str)
                        .collect::<BTreeSet<_>>(),
                    "`{name}` is written with other members than entry 73 names"
                );
            }
        }
    }
    assert_eq!(
        written_keys.iter().collect::<BTreeSet<_>>(),
        fields.keys().collect::<BTreeSet<_>>(),
        "a key entry 73 names is written on no shape of line"
    );
    // And every word entry 73 admits for the scope, the outcome and the reasons is one the
    // record writes — asked of the serializer rather than restated.
    for name in ["scope", "outcome"] {
        let admitted: BTreeSet<String> = serde_json::from_value(fields[name]["values"].clone())
            .expect("entry 73 lists the words");
        assert_eq!(
            written_words.get(name),
            Some(&admitted),
            "entry 73 admits other `{name}` words than the record writes"
        );
    }
    let because: BTreeSet<String> = [
        WholeBecause::First,
        WholeBecause::AfterFailure,
        WholeBecause::StoreLacksMembers,
    ]
    .into_iter()
    .map(|reason| {
        serde_json::to_value(reason)
            .ok()
            .and_then(|word| word.as_str().map(str::to_owned))
            .expect("a reason is a word")
    })
    .collect();
    assert_eq!(
        serde_json::from_value::<BTreeSet<String>>(fields["whole_because"]["values"].clone())
            .expect("entry 73 lists the reasons"),
        because
    );
    let classes: BTreeSet<String> = [FailureClass::Refused, FailureClass::Transient]
        .into_iter()
        .map(|class| {
            serde_json::to_value(class)
                .ok()
                .and_then(|word| word.as_str().map(str::to_owned))
                .expect("a class is a word")
        })
        .collect();
    assert_eq!(
        serde_json::from_value::<BTreeSet<String>>(fields["class"]["values"].clone())
            .expect("entry 73 lists the classes"),
        classes
    );
    assert_eq!(
        fields["actions"]["members"],
        json!(serde_json::to_value(ProjectionActions::default())
            .expect("counts serialize")
            .as_object()
            .expect("an object")
            .keys()
            .collect::<Vec<_>>())
    );

    // A line saying a contradiction, or naming a key the entry does not, is refused.
    for (contradiction, patch) in [
        (
            "a member copy with a reason to be whole",
            json!({"whole_because": "first"}),
        ),
        ("a whole copy with no reason", json!({"scope": "whole"})),
        (
            "a landed attempt with a failure's reason",
            json!({"reason": "it failed"}),
        ),
        (
            "a failed attempt with a copy report",
            json!({"outcome": "failed", "reason": "it failed"}),
        ),
        (
            "a failed attempt with no reason",
            json!({"outcome": "failed", "actions": null, "spent": null}),
        ),
        (
            "a class without its kind",
            json!({"outcome": "failed", "reason": "it failed", "actions": null, "spent": null,
                   "class": "refused"}),
        ),
        ("a key the entry does not name", json!({"cost": 94})),
        ("a start that is not a time", json!({"at": "yesterday"})),
        (
            "a start that is not UTC",
            json!({"at": "2026-09-13T12:00:00+02:00"}),
        ),
        (
            "a project that is not a qualified id",
            json!({"project": "writeback-quota-plan"}),
        ),
        ("an item with no node id", json!({"items": [""]})),
    ] {
        let mut line = example.clone();
        for (key, value) in patch.as_object().expect("a patch") {
            line[key] = value.clone();
        }
        assert!(
            serde_json::from_value::<ProjectionRecord>(line.clone()).is_err(),
            "{contradiction} was read as a record: {line}"
        );
    }

    // The record's schema version: the constant the worker writes is the one the entry names,
    // and both golden lines are written at it.
    let schema = &projection["schema"];
    let current = onepipeline::cli::WRITEBACK_PROJECTIONS_SCHEMA_VERSION;
    assert_eq!(
        schema["current"].as_u64(),
        Some(u64::from(current)),
        "entry 73 names a different current schema version than the worker writes"
    );
    assert_eq!(schema["read"], json!([1, 2, current]));
    assert_eq!(schema["absent_means"], json!(1));
    assert_eq!(schema["added_at_2"], json!(["delivered"]));
    assert_eq!(schema["added_at_3"], json!(["actions.reopened"]));
    let delivered_example = &projection["example_delivered"];
    for golden in [example, delivered_example] {
        assert_eq!(
            golden["schema_version"].as_u64(),
            Some(u64::from(current)),
            "a golden line of entry 73 is not at the current schema version: {golden}"
        );
    }

    // `delivered` round-trips verbatim — every member the store wrote, one this build never
    // names included — and a line that reached no ticket writes no `delivered` key at all.
    assert!(
        delivered_example["delivered"][0].get("pruned").is_some(),
        "the golden entry carries no member this build never names, so verbatim proves nothing"
    );
    let with_tickets: ProjectionRecord = serde_json::from_value(delivered_example.clone())
        .unwrap_or_else(|error| panic!("entry 73's delivered example is not a record: {error}"));
    assert_eq!(
        json!(with_tickets.delivered),
        delivered_example["delivered"],
        "the delivered entries were not kept verbatim"
    );
    assert_eq!(
        serde_json::to_value(&with_tickets).expect("a record serializes"),
        *delivered_example,
        "entry 73's delivered example does not write back as itself"
    );
    assert!(
        record.delivered.is_empty() && example.get("delivered").is_none(),
        "the plain golden line was meant to name no ticket"
    );
    assert!(
        serde_json::to_value(&record)
            .expect("a record serializes")
            .get("delivered")
            .is_none(),
        "a line that reached no ticket wrote a `delivered` key"
    );

    // The golden that reopened an item says so, or `reopened` on a golden proves nothing.
    assert_eq!(
        delivered_example["actions"]["reopened"],
        json!(1),
        "entry 73's delivered example counts no reopen, so the key's round trip is untested"
    );
    assert_eq!(example["actions"]["reopened"], json!(0));

    // A line an earlier build wrote names no version, no `delivered` and no
    // `actions.reopened`: it still reads, as version 1, and is written back at the current
    // version with `reopened` read as zero. A version 2 line — `delivered` allowed, still no
    // `reopened` — reads the same way.
    let mut older_actions = example["actions"].clone();
    older_actions
        .as_object_mut()
        .expect("actions is an object")
        .remove("reopened");
    let mut unversioned = example.clone();
    unversioned
        .as_object_mut()
        .expect("a line is an object")
        .remove("schema_version");
    unversioned["actions"] = older_actions.clone();
    let older: ProjectionRecord = serde_json::from_value(unversioned)
        .unwrap_or_else(|error| panic!("a version 1 line did not read: {error}"));
    assert_eq!(
        serde_json::to_value(&older).expect("a record serializes"),
        *example,
        "a version 1 line was not written back at the current version"
    );
    let mut second = delivered_example.clone();
    second["schema_version"] = json!(2);
    second["actions"] = older_actions.clone();
    let second: ProjectionRecord = serde_json::from_value(second)
        .unwrap_or_else(|error| panic!("a version 2 line did not read: {error}"));
    let mut expected = delivered_example.clone();
    expected["actions"]["reopened"] = json!(0);
    assert_eq!(
        serde_json::to_value(&second).expect("a record serializes"),
        expected,
        "a version 2 line was not written back at the current version with `reopened` zero"
    );
    for (refused, patch) in [
        (
            "a version 1 line naming `delivered`",
            json!({"schema_version": 1, "actions": older_actions,
                   "delivered": delivered_example["delivered"]}),
        ),
        // Refused by the key's own name rather than by what it holds: a line that names it
        // empty is still a version 1 line naming a key version 2 added, and reading it as
        // one that left the key off would let that shape through the boundary unremarked.
        (
            "a version 1 line naming an empty `delivered`",
            json!({"schema_version": 1, "actions": older_actions, "delivered": []}),
        ),
        // The same, for the key version 3 added: named as zero is still named.
        (
            "a version 1 line naming `actions.reopened`",
            json!({"schema_version": 1}),
        ),
        (
            "a version 2 line naming `actions.reopened`",
            json!({"schema_version": 2}),
        ),
        (
            "a version 3 line naming `actions` without `reopened`",
            json!({"actions": older_actions}),
        ),
        (
            "a version this build has never written",
            json!({"schema_version": current + 1}),
        ),
        ("version 0", json!({"schema_version": 0})),
    ] {
        let mut line = example.clone();
        for (key, value) in patch.as_object().expect("a patch") {
            line[key] = value.clone();
        }
        assert!(
            serde_json::from_value::<ProjectionRecord>(line.clone()).is_err(),
            "{refused} was read as a record: {line}"
        );
    }
}

/// The criterion check this build carries **beyond** the contract is exactly what
/// the divergence record proposes.
///
/// The contract is committed as approved and names none of it, so entry 47 is the
/// only place it is written down — and a divergence nothing gates quietly stops
/// being true. The kind is held against the enum that emits it and against the
/// contract's own list, and the surface a mismatch is raised under against the
/// kinds this build parses: a check that started reporting under a kind no
/// reader recognises would be a finding nobody is handed.
#[test]
fn the_criterion_check_is_what_the_divergence_record_names() {
    let block = divergence_block("47.");

    let kinds: Vec<String> =
        serde_json::from_value(block["event_kinds"].clone()).expect("entry 47 names its kinds");
    assert!(!kinds.is_empty());
    for kind in &kinds {
        assert!(
            PipelineKind::from_wire(&EventKind(kind.clone())).is_some(),
            "`{kind}` is not a kind this crate emits"
        );
        assert!(
            !CONTRACT.contains(&format!("`{kind}`")),
            "the contract names `{kind}`, so it is no divergence"
        );
    }

    // A mismatch is a finding and not a kind of its own: the planner already has
    // one word for "something a watcher saw and decided you should know", and a
    // second would be a queue a reader has to learn to read.
    let raised = block["surface_kind"]
        .as_str()
        .expect("entry 47 names the surface a mismatch is raised under");
    let parsed: SurfaceKind = serde_json::from_value(json!(raised))
        .unwrap_or_else(|e| panic!("`{raised}` is a kind this build parses: {e}"));
    assert_eq!(parsed, SurfaceKind::finding());

    // Three answers, kept apart. The words themselves are private vocabulary and
    // are held against the enum that spells them by the module's own test; what
    // is checkable from out here is that the entry proposes three distinct ones
    // rather than folding "could not read it" into "it disagreed".
    let answers: BTreeSet<String> =
        serde_json::from_value(block["answers"].clone()).expect("entry 47 names its answers");
    assert_eq!(
        answers,
        ["match", "mismatch", "unread"]
            .into_iter()
            .map(str::to_string)
            .collect::<BTreeSet<String>>(),
        "entry 47 no longer keeps a file it could not read apart from one that disagreed"
    );
}

#[test]
fn every_op_deserializes_with_the_fields_the_protocol_requires() {
    let envelopes: Vec<(&str, Value)> = vec![
        ("add", json!({"op": "add", "node": {"id": "new"}})),
        (
            "drop",
            json!({"op": "drop", "id": "slow", "dependents": "detach"}),
        ),
        (
            "reparent",
            json!({"op": "reparent", "id": "pending", "deps": ["slow"]}),
        ),
        (
            "retry",
            json!({"op": "retry", "id": "failed", "node": {"id": "retry"}}),
        ),
        ("cancel", json!({"op": "cancel", "id": "sweep"})),
        (
            "requeue",
            json!({"op": "requeue", "id": "sweep", "amend": {"max_turns": 32}}),
        ),
        ("attest", json!({"op": "attest", "ref": "approve"})),
        (
            "complete",
            json!({"op": "complete", "reason": "closeout verified"}),
        ),
    ];
    let seen: Vec<&str> = envelopes.iter().map(|(op, _)| *op).collect();
    assert_eq!(
        seen,
        contract_ops_this_build_accepts(),
        "every op the contract lists and this build still accepts is exercised, in order"
    );

    for (op, value) in &envelopes {
        let edit: Edit = serde_json::from_value(value.clone())
            .unwrap_or_else(|e| panic!("`{op}` deserializes: {e}"));
        assert_eq!(
            &op_of(&edit),
            op,
            "`{op}` deserialized into another variant"
        );
        let again = serde_json::to_value(&edit).expect("serializes");
        assert_eq!(&again, value, "`{op}` round-trips unchanged");
    }
}

/// The two axes the survivor carries, the defaults it omits, and the third
/// delivery value that is gone — as divergence entry 60 declares them.
///
/// The contract is committed as approved and still describes `context` with
/// `deliver: auto|live|next`; entry 60 is the proposal that replaces that
/// paragraph, so what this build does is asserted from the entry rather than by
/// editing the document to match the code. Both directions: a build that grows a
/// field or a value the entry does not name fails here as loudly as one that
/// drops one it does.
#[test]
fn a_note_carries_two_axes_and_the_defaults_entry_60_declares() {
    let prose = CONTRACT.split_whitespace().collect::<Vec<_>>().join(" ");
    assert!(
        prose.contains("`deliver: auto|live|next`, defaulting to `auto`"),
        "the contract's `context` paragraph moved; entry 60 names the sentence it replaces"
    );

    let block = divergence_block("60.");
    let deliver = &block["deliver"];
    let values: Vec<String> =
        serde_json::from_value(deliver["values"].clone()).expect("entry 60 names them");
    let default = deliver["default"]
        .as_str()
        .expect("entry 60 names the default");
    let gone = deliver["gone"]
        .as_str()
        .expect("entry 60 names what is gone");
    let persists = block["persist"]["default"]
        .as_bool()
        .expect("entry 60 names `persist`'s default");

    let note = |extra: Value| {
        let mut value =
            json!({"op": "note", "id": "slow", "addressee": "worker", "text": "the fix landed"});
        for (key, field) in extra.as_object().expect("an object") {
            value[key] = field.clone();
        }
        serde_json::from_value::<Edit>(value).expect("the note parses")
    };

    // Every value the entry names is one the wire accepts, and each is a
    // different command from the others.
    let spelled: Vec<Edit> = values
        .iter()
        .map(|value| note(json!({"deliver": value})))
        .collect();
    assert_eq!(spelled.len(), 2, "entry 60 names two delivery values");
    assert_ne!(spelled[0], spelled[1]);

    // The defaults a caller gets by omitting both fields are the ones the entry
    // names, and they are omitted again when the note is re-serialized.
    let bare = note(json!({}));
    assert_eq!(
        bare,
        note(json!({"deliver": default, "persist": persists})),
        "a note that says nothing about either axis is not the declared default"
    );
    assert_eq!(
        serde_json::to_value(&bare).expect("serializes"),
        json!({"op": "note", "id": "slow", "addressee": "worker", "text": "the fix landed"}),
        "a note carrying the defaults writes them out"
    );
    // And the non-defaults are written, so a re-serialized note says what it is.
    let other = values
        .iter()
        .find(|value| value.as_str() != default)
        .expect("entry 60 names a second value");
    assert_eq!(
        serde_json::to_value(note(json!({"deliver": other, "persist": !persists})))
            .expect("serializes"),
        json!({
            "op": "note", "id": "slow", "addressee": "worker", "text": "the fix landed",
            "deliver": other, "persist": !persists
        })
    );

    // The value that named a combination of both axes is gone, refused by the
    // same rule any other unknown value is — and its meaning has an exact
    // spelling in the two fields.
    let err = serde_json::from_value::<Edit>(
        json!({"op": "note", "id": "slow", "addressee": "worker", "text": "n", "deliver": gone}),
    )
    .expect_err("the removed delivery value is refused");
    assert!(err.to_string().contains(gone), "the error names it: {err}");
    let was_auto = block["combinations"]
        .as_array()
        .expect("entry 60 names the four combinations")
        .iter()
        .find(|one| one["default"] == json!(true))
        .expect("one of them is the default");
    assert_eq!(
        note(json!({"deliver": was_auto["deliver"], "persist": was_auto["persist"]})),
        bare,
        "the combination entry 60 calls the default is not what a bare note carries"
    );
    assert_eq!(block["combinations"].as_array().expect("an array").len(), 4);
}

#[test]
fn drop_must_state_the_dependents_fate() {
    let err = serde_json::from_value::<Edit>(json!({"op": "drop", "id": "slow"}))
        .expect_err("`dependents` is required");
    assert!(
        err.to_string().contains("dependents"),
        "the error names it: {err}"
    );

    assert_ne!(Dependents::Drop, Dependents::Detach);
    assert!(CONTRACT.contains("drop"));
}

#[test]
fn an_unknown_op_is_refused_rather_than_ignored() {
    let err = serde_json::from_value::<Edit>(json!({"op": "rewrite", "id": "x"}))
        .expect_err("an op outside the protocol is refused");
    assert!(
        err.to_string().contains("rewrite"),
        "the error names it: {err}"
    );
}

#[test]
fn a_command_only_envelope_and_a_verdict_envelope_are_both_replies() {
    let commands_only: Reply = serde_json::from_value(json!({
        "version": 1,
        "commands": [{"op": "attest", "ref": "approve"}]
    }))
    .expect("a command-only envelope parses");
    assert_eq!(commands_only.version, Some(1));
    assert_eq!(commands_only.commands.len(), 1);
    assert_eq!(commands_only.completion, None);

    let both: Reply = serde_json::from_value(json!({
        "completion": false,
        "message": "apply the replacement and continue",
        "reason": "the failed node is retryable",
        "version": 1,
        "commands": [{"op": "retry", "id": "failed", "node": {"id": "retry", "expects_no_diff": true}}]
    }))
    .expect("commands may accompany a legacy verdict");
    assert_eq!(both.completion, Some(false));
    assert_eq!(both.commands.len(), 1);

    let legacy: Reply = serde_json::from_value(json!({"completion": true, "reason": "done"}))
        .expect("a legacy verdict alone parses");
    assert!(legacy.commands.is_empty());

    assert!(CONTRACT.contains(r#"{"version": 1, "commands": [...]}"#));
}

/// The halves the contract routes a reply by, as the wire shape declares them.
///
/// A pending surface is answered by a **verdict half** and nothing else, so the
/// three fields it is spelled with are what tells the two readers' envelopes
/// apart — and a commands-only envelope declares none of them. No addressing
/// field was minted for this: the discrimination is already in the wire shape,
/// which is what this asserts, alongside the routing the document states.
#[test]
fn a_reply_declares_the_halves_the_contract_routes_it_by() {
    let read = |value: Value| serde_json::from_value::<Reply>(value).expect("the envelope parses");

    let commands_only = read(json!({
        "version": 1,
        "commands": [
            {"op": "note", "id": "plan", "addressee": "worker", "text": "the scope changed"}
        ]
    }));
    assert_eq!(commands_only.completion, None);
    assert_eq!(commands_only.message, None);
    assert_eq!(commands_only.reason, None);
    assert_eq!(commands_only.commands.len(), 1);

    for (half, value) in [
        ("completion", json!({"completion": false})),
        ("message", json!({"message": "keep going"})),
        ("reason", json!({"reason": "keep going"})),
    ] {
        let alone = read(value);
        assert!(
            alone.completion.is_some() || alone.message.is_some() || alone.reason.is_some(),
            "`{half}` alone declares no verdict half"
        );
        assert!(
            alone.commands.is_empty(),
            "`{half}` alone declares a commands half"
        );
    }

    let both = read(json!({
        "completion": false,
        "reason": "retry it",
        "version": 1,
        "commands": [{"op": "cancel", "id": "slow"}]
    }));
    assert_eq!(both.reason.as_deref(), Some("retry it"));
    assert_eq!(both.commands.len(), 1);

    assert!(
        CONTRACT.contains(
            "**A reply is routed by the halves it carries, never by which reader reaches the \
             queue first.**"
        ),
        "the contract no longer states that a reply is routed by its halves"
    );
    assert!(
        CONTRACT.contains(
            "It answers a pending surface only when it carries a **verdict half** — \
             `completion`, `message`, or `reason`"
        ),
        "the contract no longer says which half answers a pending surface"
    );
    assert!(
        CONTRACT.contains(
            "belongs to the command path alone: it leaves the pending surface, and any reader \
             waiting there for a verdict, untouched"
        ),
        "the contract no longer says where a commands-only envelope goes"
    );
    assert!(
        CONTRACT.contains("One carrying both is delivered to both"),
        "the contract no longer says what an envelope carrying both halves does"
    );
    assert!(
        CONTRACT.contains("Neither reader advances the other's cursor"),
        "the contract no longer promises the two cursors stay apart"
    );
}

#[test]
fn the_contract_declares_an_open_surface_kind_vocabulary() {
    let check_in: SurfaceKind = serde_json::from_value(json!("check-in")).expect("parses");
    assert_eq!(check_in, SurfaceKind::check_in());
    let host_kind: SurfaceKind = serde_json::from_value(json!("host-defined")).expect("parses");
    assert_eq!(host_kind.as_str(), "host-defined");
    assert!(CONTRACT.contains("--kind KIND"));
    assert!(CONTRACT.contains("^[a-z][a-z0-9-]{0,63}$"));
    assert!(
        CONTRACT.contains(
            "restarts the check-in clock of **every member the run's recorded observer graph \
             declares `resettable` in its `schedule`**"
        ),
        "consuming a surface restarts the resettable clocks"
    );
    assert!(
        !CONTRACT.contains("reset-timer RUN check-in"),
        "the contract names a member of one host's observer graph"
    );
    // The one kind the engine raises of its own, stated with its source and its
    // message.
    assert!(CONTRACT.contains("`edit-applied`, queued non-blocking"));
    assert!(CONTRACT.contains("`<author> applied an edit: <command>`"));
}

/// The refusals the contract lists for an author granted nothing are the
/// linked bus's own words, op by op and for the legacy verdict.
///
/// The block is the contract's; each entry is driven through the same `allows`
/// and `allows_completion` the channel refuses with, under the profile's own
/// allowlist — which grants an author the launch configuration never declared
/// nothing, exactly as a declared author granted nothing is refused. The author
/// is a word nothing built in knows.
#[test]
fn the_refusals_the_contract_lists_are_the_channels_own_words() {
    let examples: Vec<Value> = serde_json::from_str(&fenced_block_naming("json", "\"refused\""))
        .expect("the block is JSON");
    assert!(examples.len() >= 8, "{examples:?}");
    for example in &examples {
        let author = Author::from(example["author"].as_str().expect("an author word"));
        assert_ne!(author.as_str(), "planner", "{example}");
        let refused = example["refused"].as_str().expect("a refusal");
        let answered = if example.get("verdict").is_some() {
            onepipeline::channel::allows_completion(author, Some(true))
        } else {
            let edit: Edit = serde_json::from_value(example["command"].clone())
                .unwrap_or_else(|e| panic!("the example's command parses: {e}: {example}"));
            allows(author, &edit)
        }
        .expect_err("an author granted nothing is refused")
        .to_string();
        assert_eq!(
            answered,
            format!("refused: {refused}"),
            "the channel refuses in different words than the contract states"
        );
    }
}

#[test]
fn the_contracts_open_author_grammar_is_the_channels_complete_boundary() {
    assert!(CONTRACT.contains("Channel authors are open words"));
    assert!(CONTRACT.contains("^[a-z][a-z0-9-]{0,63}$"));

    for word in ["planner", "monitor", "sentinel", "a", &"a".repeat(64)] {
        let author: Author = serde_json::from_value(json!(word))
            .unwrap_or_else(|error| panic!("the documented author `{word}` parses: {error}"));
        assert_eq!(author.as_str(), word);
        assert_eq!(
            serde_json::to_value(author).expect("serializes"),
            json!(word)
        );
    }
    for word in ["", "Monitor", "two_words", "-leading", &"a".repeat(65)] {
        let error = serde_json::from_value::<Author>(json!(word))
            .expect_err(&format!("the undocumented author `{word}` was accepted"));
        assert!(
            error.to_string().contains("^[a-z][a-z0-9-]{0,63}$"),
            "the refusal does not name the contract grammar: {error}"
        );
    }
}

#[test]
fn the_reply_exit_codes_are_the_ones_the_contract_assigns() {
    // The contract's own sentence, verbatim, and the constants it assigns — both
    // unchanged. What `reply` answers for an *accepted and still queued*
    // envelope is `0` rather than the `1` this sentence gives it, which is
    // divergence 67 and is stated there: `1` is still this build's constant and
    // still what an unfinished run and a refused plan check exit at.
    assert!(CONTRACT.contains(
        "reply exit 0 = applied, 1 = accepted-not-yet-reconciled, 2 = refused/malformed"
    ));
    assert_eq!(EXIT_SUCCESS, 0);
    assert_eq!(EXIT_QUEUED, 1);
    assert_eq!(EXIT_REFUSED, 2);

    assert!(CONTRACT.contains("exit 3 = nothing is driving the run"));
    assert_eq!(EXIT_NOTHING_DRIVING, 3);

    // Each code means one thing: a caller that reads the status must not have
    // to guess which of two verdicts it got.
    let spent = [
        EXIT_SUCCESS,
        EXIT_QUEUED,
        EXIT_REFUSED,
        EXIT_NOTHING_DRIVING,
    ];
    let mut unique = spent.to_vec();
    unique.sort_unstable();
    unique.dedup();
    assert_eq!(unique.len(), spent.len(), "two verdicts share an exit code");
}

/// The two version numbers entry 65 declares are the constants this build
/// publishes.
///
/// A consumer in another repository decides what a record may contain from that
/// entry, and the merged stream's version is the whole of what tells a v1
/// `edit-committed` from a v2 one. An entry naming a version this build does not
/// write, or a read set it does not honour, is a document describing a build that
/// does not exist.
#[test]
fn the_divergence_record_names_the_envelope_version_this_build_writes_and_reads() {
    let block = divergence_block("65.");

    assert_eq!(
        block["journal_envelope_version"],
        json!(ENVELOPE_VERSION),
        "entry 65 names an envelope version this build does not write"
    );
    assert_eq!(
        serde_json::from_value::<Vec<u32>>(block["journal_envelope_versions_read"].clone())
            .expect("entry 65 names the versions this build reads"),
        ENVELOPE_VERSIONS_READ,
        "entry 65 names a different read set from the one this build honours"
    );
    // The move is additive, which is what makes reading the older one whole a
    // promise rather than a hope: the version this build writes is ahead of every
    // other one it reads, and the one before it is still in the set.
    assert!(
        ENVELOPE_VERSIONS_READ.contains(&ENVELOPE_VERSION)
            && ENVELOPE_VERSIONS_READ.contains(&1)
            && ENVELOPE_VERSIONS_READ
                .iter()
                .all(|version| *version <= ENVELOPE_VERSION),
        "the read set is not the versions up to the one this build writes: \
         {ENVELOPE_VERSIONS_READ:?}"
    );

    // And it is not the *reply* envelope's version line, which entries 57 and 60
    // move on their own cadence. Two documents, two numbers, and a consumer that
    // took one for the other would refuse every edit envelope it was sent.
    assert!(
        block.get("envelope_version").is_none(),
        "entry 65 spells its version under the key entries 57 and 60 use for the reply \
         envelope's, which a consumer reads as that one"
    );
}

/// What entry 66 says a delivered surface's record carries is what this build
/// writes on it.
///
/// The contract names the kind and says nothing about its payload, so that entry
/// is the only place the shape a consumer reads is written down.
#[test]
fn the_delivered_surfaces_instant_is_what_the_divergence_record_names() {
    let block = divergence_block("66.");

    let kind = block["delivery_kind"]
        .as_str()
        .expect("entry 66 names the kind that delivers a surface");
    assert!(
        PIPELINE_KINDS
            .iter()
            .any(|carried| carried.as_str() == kind),
        "entry 66 names `{kind}`, which is not a kind this build emits"
    );
    assert_eq!(
        block["queued_at_units"], "epoch-milliseconds",
        "entry 66 spells the instant in units this crate does not record in"
    );
    assert_eq!(
        serde_json::from_value::<Vec<String>>(block["carried_beside"].clone())
            .expect("entry 66 names the fields the instant joins"),
        ["kind", "message", "source", "blocking"]
    );
    let field = block["queued_at_field"]
        .as_str()
        .expect("entry 66 names the field");
    let verbs = std::fs::read_to_string(repo_root().join("src/verbs.rs")).expect("the verbs ship");
    assert!(
        verbs.contains(&format!("(\"{field}\", json!(surface.{field}))")),
        "entry 66 names a field the hand-out does not write: {field}"
    );
    let journeys = std::fs::read_to_string(repo_root().join("tests/e2e/channel.rs"))
        .expect("the channel journeys ship");
    assert!(
        journeys.contains("fn a_delivered_surface_is_recorded_with_the_instant_it_was_queued("),
        "entry 66 names a journey the channel suite does not run"
    );
}

/// The statuses entry 67 declares a reply exits at are this build's constants.
///
/// The contract fixes three codes for this verb and the code no longer writes
/// one of them, so that entry is where the mapping a caller branches on lives.
#[test]
fn the_replys_exit_statuses_are_what_the_divergence_record_names() {
    let block = divergence_block("67.");

    assert_eq!(
        block["reply_applied_exit"],
        json!(EXIT_SUCCESS),
        "entry 67 names a status this build does not exit at for an applied edit"
    );
    assert_eq!(
        block["reply_queued_exit"],
        json!(EXIT_SUCCESS),
        "entry 67 names a status this build does not exit at for a queued edit"
    );
    assert_eq!(
        block["reply_refused_exit"],
        json!(EXIT_REFUSED),
        "entry 67 names a status this build does not exit at for a refused edit"
    );
    assert_eq!(
        block["reply_queued_exit"], block["reply_applied_exit"],
        "entry 67 has a queued envelope answering with a status of its own again"
    );
    assert_ne!(
        block["reply_queued_exit"],
        json!(EXIT_QUEUED),
        "entry 67 has a queued reply answering with the unfinished-run status again"
    );
    let verbs = std::fs::read_to_string(repo_root().join("src/verbs.rs")).expect("the verbs ship");
    assert!(
        verbs.contains("durable command queue")
            && verbs.contains("has to drive the run for them to")
            && verbs.contains("They are not to be sent"),
        "the reply no longer says what a queued envelope is waiting for"
    );
}

#[test]
fn an_envelope_round_trips_through_the_merged_streams_shape() {
    let wire = json!({
        "v": ENVELOPE_VERSION,
        "ts": "2026-08-07T12:00:00.000Z",
        "stream": "onepipeline-7f3a",
        "seq": 42,
        "source": "pipeline",
        "kind": "node-settled",
        "labels": {"run_id": "run-1", "round": 2, "node": "service", "attempt": 1},
        "payload": {"status": "done"},
        "artifacts": [{"id": "gate-log", "kind": "log", "bytes": 8192}]
    });

    let envelope: Envelope = serde_json::from_value(wire.clone()).expect("the envelope parses");
    assert_eq!(envelope.source, Source::Pipeline);
    assert_eq!(envelope.kind, EventKind("node-settled".into()));
    assert_eq!(envelope.labels.run_id.as_deref(), Some("run-1"));
    assert_eq!(envelope.labels.round, Some(2));
    assert_eq!(
        envelope.labels.extra.get("attempt"),
        Some(&json!(1)),
        "a label outside the reserved keys rides in `extra`"
    );
    assert_eq!(
        envelope.artifacts,
        vec![ArtifactRef {
            id: "gate-log".into(),
            kind: "log".into(),
            bytes: 8192
        }]
    );

    assert_eq!(serde_json::to_value(&envelope).expect("serializes"), wire);
}

/// Every item `onepipeline::event` and `onepipeline::filter` published before the
/// wire moved onto the bus still resolves at the same path, and the envelope, its
/// labels, source, phase and artifact, the kind, the filter and the matcher are
/// the bus's own types rather than copies of them.
///
/// A type named twice is proven one type by handing both names to a function
/// that takes one: a copy, however faithful, does not compile here.
#[test]
fn the_wire_types_resolve_where_they_did_and_are_the_buss_own() {
    fn one_type<T>(_: std::marker::PhantomData<T>, _: std::marker::PhantomData<T>) {}
    use std::marker::PhantomData as Named;
    one_type(Named::<Envelope>, Named::<onemessagebus_agent::Envelope>);
    one_type(Named::<Labels>, Named::<onemessagebus_agent::Labels>);
    one_type(Named::<Source>, Named::<onemessagebus_agent::Source>);
    one_type(Named::<Phase>, Named::<onemessagebus_agent::Phase>);
    one_type(Named::<ArtifactRef>, Named::<onemessagebus::ArtifactRef>);
    one_type(Named::<EventKind>, Named::<onemessagebus::Kind>);
    one_type(
        Named::<EventFilter>,
        Named::<onemessagebus_agent::EventFilter>,
    );
    one_type(Named::<Matcher>, Named::<onemessagebus_agent::Matcher>);
    // And the sibling that relays into this crate holds the same ones.
    one_type(Named::<Envelope>, Named::<onevcs::Envelope>);
    one_type(Named::<Envelope>, Named::<oneagentgraph::event::Envelope>);

    // The rest of what the two modules published, at the same paths.
    assert_eq!(ENVELOPE_VERSIONS_READ.first(), Some(&ENVELOPE_VERSION));
    assert_eq!(onepipeline::event::MAX_PAYLOAD_TEXT_BYTES, 4096);
    assert_eq!(
        PipelineKind::from_wire(&EventKind::from("run-started")),
        PIPELINE_KINDS.first().copied()
    );
    assert_eq!(ArtifactId("gate-log".into()).0, "gate-log");
    assert_eq!(onepipeline::filter::DEFAULT_PROFILE, "planner");
    assert_eq!(onepipeline::filter::DETAILED_PROFILE, "detailed");
    assert_eq!(
        LAUNCH_CONFIG_SCHEMA_VERSIONS_READ.first(),
        Some(&LAUNCH_CONFIG_SCHEMA_VERSION)
    );
    assert_eq!(LaunchConfig::default().filters, Filters::default());
}

#[test]
fn the_three_merged_streams_are_the_three_libraries_the_contract_composes() {
    assert!(CONTRACT.contains("merges the three event streams"));
    for (library, source) in [
        ("oneagentgraph", Source::Agentgraph),
        ("onevcs", Source::Vcs),
        ("onepipeline", Source::Pipeline),
    ] {
        assert!(
            CONTRACT.contains(library),
            "the contract no longer names `{library}` as a composed library"
        );
        let _ = source;
    }
    assert_eq!(
        serde_json::to_value(Source::Agentgraph).expect("serializes"),
        json!("agentgraph")
    );
    assert_eq!(
        serde_json::to_value(Source::Vcs).expect("serializes"),
        json!("vcs")
    );
    assert_eq!(
        serde_json::to_value(Source::Pipeline).expect("serializes"),
        json!("pipeline")
    );

    // A fourth source is not a stream this crate merges.
    serde_json::from_value::<Source>(json!("harness")).expect_err("an unknown source is refused");
}

#[test]
fn the_contract_enumerates_exactly_this_librarys_own_event_kinds() {
    // Both directions. A kind the crate emits and the contract does not list is
    // undocumented wire; a kind the contract lists and the enum does not carry is
    // a promise nothing keeps. `PIPELINE_KINDS` is what `Journal::emit` accepts,
    // so this is the emitted set and not a second copy of it.
    assert_eq!(PIPELINE_KINDS.len(), 35, "the closed set changed size");
    let listed: BTreeSet<String> = backticked()
        .into_iter()
        .filter(|token| {
            token.chars().all(|c| c.is_ascii_lowercase() || c == '-') && token.contains('-')
        })
        .collect();
    // The kinds the contract does not list are exactly the ones the divergence
    // record proposes, and no others: a kind neither document names fails here.
    let proposed: BTreeSet<String> = ["40.", "47.", "55.", "65.", "70."]
        .into_iter()
        .flat_map(|entry| {
            serde_json::from_value::<Vec<String>>(divergence_block(entry)["event_kinds"].clone())
                .unwrap_or_else(|e| panic!("entry {entry} names the kinds it adds: {e}"))
        })
        .collect();
    let undocumented: BTreeSet<String> = PIPELINE_KINDS
        .iter()
        .map(|kind| kind.as_str().to_string())
        .filter(|kind| !listed.contains(kind))
        .collect();
    assert_eq!(
        undocumented, proposed,
        "the kinds this crate emits that docs/contract.md does not list are not entry 40's"
    );

    // The wire spelling is the enum's, not a string beside it.
    assert_eq!(PipelineKind::RunStarted.as_str(), "run-started");
    assert_eq!(
        PipelineKind::from_wire(&EventKind("node-settled".into())),
        Some(PipelineKind::NodeSettled)
    );
    // A sibling's kind stays a wire string: the enum declines it rather than
    // rejecting the envelope.
    assert_eq!(
        PipelineKind::from_wire(&EventKind("gate-finished".into())),
        None
    );
}

/// The envelope's `phase` is the **sibling's** vocabulary, spelled the same way
/// on both sides of the relay, and every one of it.
///
/// The one field on the merged envelope that neither this crate nor
/// `docs/contract.md` owns: `onevcs` classifies what an event of its own belongs
/// to and this crate carries the classification rather than making one. So the
/// gate is exhaustive in both directions and over that library's own list — a
/// phase it adds fails here rather than arriving as a value every reader
/// silently declines, and one this copy grew alone fails here too.
///
/// `docs/contract-divergences.md` entry 40 is where the field is proposed to the
/// planner who owns the contract; this is what holds it to the sibling's while
/// it is a proposal.
#[test]
fn the_envelopes_phase_is_the_siblings_own_vocabulary_and_all_of_it() {
    // Spelled by a match rather than by a list, so a variant added to this
    // copy has to be spelled here as well as there.
    let spelled = |phase: Phase| match phase {
        Phase::Development => "development",
        Phase::Integrate => "integrate",
        Phase::Review => "review",
        Phase::Release => "release",
    };
    let theirs = onevcs::Phase::every();
    assert_eq!(
        theirs.len(),
        4,
        "the sibling's phase vocabulary changed size"
    );
    for phase in theirs {
        let wire = serde_json::to_value(phase).expect("the sibling's phase serializes");
        let mine: Phase = serde_json::from_value(wire.clone())
            .unwrap_or_else(|e| panic!("this copy does not read the sibling's {wire}: {e}"));
        assert_eq!(json!(spelled(mine)), wire, "the two copies spell it apart");
        // And back, so neither side carries a phase the other cannot read.
        let round: onevcs::Phase = serde_json::from_value(json!(spelled(mine)))
            .expect("the sibling reads what this copy writes");
        assert_eq!(round, phase);
    }

    // On the wire it is optional and omitted when absent: a store written before
    // there was a phase round-trips as its writer wrote it, and one written with
    // a phase keeps it.
    let without = json!({
        "v": ENVELOPE_VERSION,
        "ts": "2026-08-07T12:00:01.500Z",
        "stream": "onevcs-1a2b",
        "seq": 3,
        "source": "vcs",
        "kind": "session-opened",
        "labels": {},
        "payload": {},
        "artifacts": []
    });
    let envelope: Envelope = serde_json::from_value(without.clone()).expect("parses");
    assert_eq!(envelope.dimensions.phase, None);
    assert_eq!(
        serde_json::to_value(&envelope).expect("serializes"),
        without
    );

    let mut with = without.clone();
    with["phase"] = json!("release");
    let envelope: Envelope = serde_json::from_value(with.clone()).expect("parses");
    assert_eq!(envelope.dimensions.phase, Some(Phase::Release));
    assert_eq!(serde_json::to_value(&envelope).expect("serializes"), with);
}

#[test]
fn a_relayed_envelope_keeps_its_producers_own_kind() {
    // onepipeline merges rather than rewrites, so an envelope a sibling produced
    // survives the trip through this crate's shape unchanged.
    let wire = json!({
        "v": ENVELOPE_VERSION,
        "ts": "2026-08-07T12:00:01.500Z",
        "stream": "onevcs-1a2b",
        "seq": 3,
        "source": "vcs",
        "kind": "gate-finished",
        "labels": {},
        "payload": {},
        "artifacts": []
    });
    let envelope: Envelope = serde_json::from_value(wire.clone()).expect("parses");
    assert_eq!(envelope.source, Source::Vcs);
    assert_eq!(envelope.kind, EventKind("gate-finished".into()));
    assert_eq!(serde_json::to_value(&envelope).expect("serializes"), wire);
}

/// A scratch root for one test, removed when it ends.
///
/// The retention path writes to a real filesystem, so these journeys need
/// somewhere to hold a consumer's own runs root beside a producing process's own
/// scratch — which is the whole distinction the path is about.
fn scratch(name: &str) -> PathBuf {
    let dir = std::env::temp_dir().join(format!(
        "onepipeline-contract-{name}-{}-{:?}",
        std::process::id(),
        std::thread::current().id()
    ));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).expect("a scratch root");
    dir
}

/// The `member-settled` a producing process wrote to its own stdout, as the wire
/// line it arrives as.
///
/// Built out of the published constants rather than a second spelling of them:
/// what a consumer has to be able to compose is exactly this envelope, and the
/// crate's own words are what it composes it from.
fn member_settled(stream: &str, seq: u64, named: &Path) -> Envelope {
    let mut wire = json!({
        "v": ENVELOPE_VERSION,
        "ts": "2026-08-18T09:00:00.000Z",
        "stream": stream,
        "seq": seq,
        "source": "agentgraph",
        "kind": MEMBER_SETTLED,
        "labels": {"node": "build", "member": "worker"},
        "payload": {},
        // The artifact names the stream and not the seq, which is why the copy
        // is derivable from this envelope and never from the id alone.
        "artifacts": [{"id": format!("report-{stream}"), "kind": "report", "bytes": 0}]
    });
    wire["payload"][REPORT_PATH] = json!(named.display().to_string());
    serde_json::from_value(wire).expect("the settlement parses")
}

/// The whole published path, driven the way a consumer downstream of this crate
/// drives it: mint the run's paths, ingest the settlement a producing process
/// wrote, and read the report back from the path `report_for` derives.
///
/// The stream carries characters the sanitiser rewrites, so what is proved is
/// the *derived* name rather than a passthrough one — and the consumer obtains
/// it by calling, never by restating the rule behind it.
#[test]
fn a_consumer_retains_a_report_and_reads_it_back_from_the_path_it_derives() {
    let root = scratch("retention-journey");
    // The producing library's own scratch: a directory this crate neither
    // chooses nor can attest.
    let produced = root.join("producer");
    std::fs::create_dir_all(&produced).expect("a producer scratch");
    let named = produced.join(ACCEPTED_REPORT_FILE);
    let body = r#"{"results":[{"harness":"claude-code","text":"Ran the gate."}]}"#;
    std::fs::write(&named, body).expect("the report the producer wrote");

    let runs = root.join("runs");
    let paths = RunPaths::under(&runs, "demo");
    assert_eq!(paths.run, "demo");
    assert_eq!(paths.dir, runs.join("demo"));
    assert!(paths.reports_dir().starts_with(&paths.dir));

    let stream = "node-scope/1786925518098 3163646";
    retain(&paths, &member_settled(stream, 7, &named));

    let kept = paths.report_for(stream, 7);
    assert_eq!(
        std::fs::read_to_string(&kept).expect("this run's own copy of the report"),
        body,
        "the copy at the derived path is not the report the producer wrote"
    );

    // One segment under the run's own storage, whatever the producer's stream id
    // said — the sanitiser is reached only through this call.
    let leaf = kept
        .strip_prefix(paths.reports_dir())
        .expect("the copy is under the run's own reports directory");
    assert_eq!(
        leaf.components().count(),
        1,
        "the derived name is not a single segment: {leaf:?}"
    );

    // Copied, not referenced: the producer's own file going away costs the run
    // nothing, which is what makes the derived path the one a reader opens.
    std::fs::remove_file(&named).expect("the producer's own copy goes away");
    assert!(std::fs::read_to_string(&kept).is_ok());

    std::fs::remove_dir_all(&root).ok();
}

/// A producer's stream id is a string; the path it derives is a name.
///
/// The seq is part of that name because the settlement is what identifies the
/// report — the artifact id names the stream alone — so two settlements of one
/// stream never resolve to one file.
#[test]
fn a_derived_report_path_is_one_segment_under_the_runs_own_storage() {
    let paths = RunPaths::under(Path::new("/nowhere"), "demo");
    for stream in [
        "oneagentgraph-1",
        "../../elsewhere",
        "..",
        "",
        "node scope:1786925518098/3163646",
    ] {
        let kept = paths.report_for(stream, 3);
        let leaf = kept
            .strip_prefix(paths.reports_dir())
            .unwrap_or_else(|_| panic!("'{stream}' derived a path outside the run: {kept:?}"));
        assert_eq!(
            leaf.components().count(),
            1,
            "'{stream}' derived more than one segment: {leaf:?}"
        );
        assert_ne!(
            kept,
            paths.report_for(stream, 4),
            "'{stream}' resolves two settlements to one file"
        );
    }
}

/// Every refusal the writer makes, met the way a consumer meets it: nothing is
/// at the path `report_for` derives, so no reader of this run's store finds one.
#[test]
fn the_published_writer_refuses_anything_that_is_not_the_producers_own_plain_file() {
    let root = scratch("retention-refusals");
    let paths = RunPaths::under(&root.join("runs"), "demo");
    let secret = root.join("secret.json");
    std::fs::write(&secret, r#"{"transcript":{"messages":[]}}"#)
        .expect("a file the producing library never wrote");

    let planted = root.join("planted");
    std::fs::create_dir_all(&planted).expect("somewhere to plant a link");
    let link = planted.join(ACCEPTED_REPORT_FILE);
    #[cfg(unix)]
    std::os::unix::fs::symlink(&secret, &link).expect("a symlink");
    #[cfg(windows)]
    std::os::windows::fs::symlink_file(&secret, &link).expect("a symlink");

    let directory = root.join("as-a-directory").join(ACCEPTED_REPORT_FILE);
    std::fs::create_dir_all(&directory).expect("a directory wearing the accepted name");

    for (seq, (case, named)) in [
        ("a base name the producing library never writes", secret),
        ("a symlink wearing the accepted name", link),
        ("a directory wearing the accepted name", directory),
        (
            "nothing at all",
            root.join("gone").join(ACCEPTED_REPORT_FILE),
        ),
    ]
    .into_iter()
    .enumerate()
    {
        let seq = seq as u64;
        retain(&paths, &member_settled("oneagentgraph-1", seq, &named));
        let kept = paths.report_for("oneagentgraph-1", seq);
        assert!(
            std::fs::symlink_metadata(&kept).is_err(),
            "{case}: '{}' reached the run's own storage at {kept:?}",
            named.display()
        );
    }
    std::fs::remove_dir_all(&root).ok();
}

/// The bound is the writer's, and it is published so a caller knows what will be
/// refused rather than discovering it.
#[test]
fn the_published_writer_refuses_a_report_past_the_bound_it_publishes() {
    let root = scratch("retention-oversize");
    let paths = RunPaths::under(&root.join("runs"), "demo");
    let produced = root.join("producer");
    std::fs::create_dir_all(&produced).expect("a producer scratch");
    let named = produced.join(ACCEPTED_REPORT_FILE);
    // Claimed rather than written: the bound is on the size the filesystem
    // reports, and a real 32MiB fixture would be a slow way to say so.
    let file = std::fs::File::create(&named).expect("the stored report");
    file.set_len(MAX_REPORT_BYTES + 1)
        .expect("a report past the bound");
    drop(file);

    retain(&paths, &member_settled("oneagentgraph-1", 4, &named));
    let kept = paths.report_for("oneagentgraph-1", 4);
    assert!(
        std::fs::symlink_metadata(&kept).is_err(),
        "a report past {MAX_REPORT_BYTES} bytes reached the run's own storage"
    );
    std::fs::remove_dir_all(&root).ok();
}

/// The writer ingests a settlement of the producing library's, and nothing else
/// of the same shape.
#[test]
fn the_published_writer_ingests_only_an_oneagentgraph_settlement() {
    let root = scratch("retention-kinds");
    let paths = RunPaths::under(&root.join("runs"), "demo");
    let produced = root.join("producer");
    std::fs::create_dir_all(&produced).expect("a producer scratch");
    let named = produced.join(ACCEPTED_REPORT_FILE);
    std::fs::write(&named, "{}").expect("the report the producer wrote");

    // This library's own event, of the very same shape.
    let mut ours = member_settled("onepipeline-1", 1, &named);
    ours.source = Source::Pipeline;
    retain(&paths, &ours);

    // And a kind of the producing library's that settles nothing.
    let mut other = member_settled("oneagentgraph-1", 2, &named);
    other.kind = EventKind("turn-completed".into());
    retain(&paths, &other);

    assert!(
        !paths.reports_dir().exists(),
        "an envelope that is not an agentgraph {MEMBER_SETTLED} was ingested as a report"
    );
    std::fs::remove_dir_all(&root).ok();
}

/// The document names every item the retention path is promised through, and the
/// release a consumer links it as.
#[test]
fn the_contract_names_the_retention_path_and_the_release_it_ships_in() {
    // The package's own version, read from the manifest rather than from a
    // second spelling of it: the whole point of publishing it is that a host
    // pinning this engine and a reader of its run store can prove they are one
    // release.
    let manifest =
        std::fs::read_to_string(repo_root().join("Cargo.toml")).expect("the manifest ships");
    let declared = manifest
        .split_once("\n[package]")
        .expect("the manifest declares this package")
        .1
        .lines()
        .find_map(|line| line.trim().strip_prefix("version = \"")?.strip_suffix('"'))
        .expect("the package declares a version");
    assert_eq!(
        onepipeline::VERSION,
        declared,
        "`VERSION` is not this crate's own package version"
    );

    assert_contract_names(
        "published retention path",
        &[
            "views::RunPaths",
            "the run id `run` and the run's own directory `dir`",
            "RunPaths::new",
            "RunPaths::under",
            "reports_dir()",
            "report_for(STREAM, SEQ)",
            "reports/<sanitised stream>-<seq>.json",
            "report::retain(&RunPaths, &Envelope)",
            "report::MEMBER_SETTLED",
            "report::REPORT_PATH",
            "report::ACCEPTED_REPORT_FILE",
            "report::MAX_REPORT_BYTES",
            "onepipeline::VERSION",
        ],
    );
    // The precondition rides with the function: publishing it did not widen when
    // it is safe to call, and a document that dropped this leaves a caller free
    // to hand it a path it has no authority for.
    assert!(
        CONTRACT.contains("the caller holds the producing process's authority for the path"),
        "docs/contract.md no longer states `retain`'s precondition"
    );
    assert!(
        CONTRACT.contains("the sanitiser is not public"),
        "docs/contract.md no longer says the sanitiser is unreachable"
    );
}

#[test]
fn the_driver_contracts_invocation_parses_exactly_as_written() {
    let documented = "onepipeline start PROJECT [--attach|--detach] \
                      [--dag-graph off|REF] [--pr-author-graph REF] \
                      [--heartbeat-interval 1800] \
                      [--set PATH=VALUE]... [--node-set PATH=VALUE]... \
                      [--acknowledge-concurrent]";
    assert!(CONTRACT.contains(documented), "the driver invocation moved");

    let cli = Cli::try_parse_from([
        "onepipeline",
        "start",
        "otg:plan-store",
        "--detach",
        "--dag-graph",
        "graphs/dag-scope.yaml",
        "--pr-author-graph",
        "graphs/pr-author.yaml",
        "--heartbeat-interval",
        "1800",
        "--set",
        "members.monitor.agent.model=dag one",
        "--set=members.check-in.model=dag=two",
        "--node-set",
        "members.worker.agent.model=node one",
        "--node-set=members.worker.judge.model=node=two",
        "--acknowledge-concurrent",
    ])
    .expect("the documented invocation parses");
    let Command::Start(args) = cli.command else {
        panic!("expected `start`");
    };
    assert_eq!(args.project, "otg:plan-store");
    assert!(args.detach);
    assert!(!args.attach);
    assert_eq!(args.dag_graph, "graphs/dag-scope.yaml");
    assert_eq!(
        args.pr_author_graph.as_deref(),
        Some("graphs/pr-author.yaml")
    );
    assert_eq!(args.heartbeat_interval, 1_800);
    assert!(args.acknowledge_concurrent);
    assert_eq!(
        args.dag_sets,
        [
            "members.monitor.agent.model=dag one",
            "members.check-in.model=dag=two"
        ]
    );
    assert_eq!(
        args.node_sets,
        [
            "members.worker.agent.model=node one",
            "members.worker.judge.model=node=two"
        ]
    );

    // The document's numbers and its default are this crate's.
    assert_eq!(DEFAULT_HEARTBEAT_INTERVAL_SECONDS, 1_800);
    assert!(
        CONTRACT.contains("`--dag-graph` defaults to `off`"),
        "the contract no longer states the shipped default"
    );
    let defaulted = Cli::try_parse_from(["onepipeline", "start", "plans:demo"]).expect("parses");
    let Command::Start(args) = defaulted.command else {
        panic!("expected `start`");
    };
    assert_eq!(
        args.dag_graph, DAG_GRAPH_OFF,
        "a plan runs with no agent graph unless one is asked for"
    );
    assert_eq!(
        args.pr_author_graph, None,
        "a change request is drafted by no graph unless one is asked for"
    );
    assert_eq!(args.heartbeat_interval, DEFAULT_HEARTBEAT_INTERVAL_SECONDS);
}

/// The verbs an agent used to drive the engine with, which no longer exist.
///
/// Refused rather than merely absent: a caller that still spells one is told, by
/// clap, that there is no such command — and this is what stops them being
/// reintroduced by habit.
#[test]
fn the_round_verbs_are_gone_from_the_command_surface() {
    for retired in [
        vec!["round", "run", "run-1"],
        vec!["round", "next", "run-1"],
    ] {
        Cli::try_parse_from(std::iter::once("onepipeline").chain(retired.iter().copied()))
            .expect_err("a round verb still parses");
    }
    Cli::try_parse_from(["onepipeline", "start", "p.json", "--round-budget", "10"])
        .expect_err("--round-budget still parses");
    assert!(
        !CONTRACT.contains("round run") && !CONTRACT.contains("--round-budget"),
        "the contract still names a retired verb or flag"
    );
}

#[test]
fn attach_and_detach_are_the_alternatives_the_contract_writes_them_as() {
    Cli::try_parse_from(["onepipeline", "start", "p.json", "--attach"]).expect("attach parses");
    Cli::try_parse_from(["onepipeline", "start", "p.json", "--attach", "--detach"])
        .expect_err("`--attach|--detach` are alternatives, not a pair");
}

#[test]
fn every_command_the_contract_names_parses() {
    let invocations: &[(&str, &[&str])] = &[
        ("start", &["start", "plans:demo"]),
        ("adopt", &["adopt", "run-1"]),
        ("next", &["next", "run-1"]),
        ("reply", &["reply", "run-1"]),
        ("reply FILE", &["reply", "run-1", "edits.json"]),
        (
            "surface",
            &[
                "surface",
                "run-1",
                "--kind",
                "check-in",
                "--message",
                "all clear",
            ],
        ),
        ("attest", &["attest", "run-1", "approve"]),
        ("stop", &["stop", "run-1"]),
        ("stop --force", &["stop", "run-1", "--force"]),
        ("shutdown RUN", &["shutdown", "run-1"]),
        ("shutdown --mine", &["shutdown", "--mine"]),
        ("shutdown --host", &["shutdown", "--host"]),
        (
            "shutdown --host --grace",
            &["shutdown", "--host", "--grace", "60"],
        ),
        (
            "shutdown --mine --force",
            &["shutdown", "--mine", "--force"],
        ),
        ("runs", &["runs"]),
        ("runs --mine", &["runs", "--mine"]),
        ("status", &["status"]),
        ("host", &["host"]),
        ("monitor", &["monitor", "run-1"]),
        ("results", &["results", "run-1"]),
        ("goals", &["goals"]),
        ("transcript", &["transcript", "run-1"]),
        ("transcript NODE", &["transcript", "run-1", "build"]),
        ("telemetry", &["telemetry"]),
        ("telemetry --breakdown", &["telemetry", "--breakdown"]),
    ];

    for (name, args) in invocations {
        let argv: Vec<&str> = std::iter::once("onepipeline")
            .chain(args.iter().copied())
            .collect();
        Cli::try_parse_from(&argv).unwrap_or_else(|e| panic!("`{name}` does not parse: {e}"));
    }
}

#[test]
fn the_contract_names_every_command_and_view_this_crate_offers() {
    assert_contract_names(
        "channel command",
        &[
            "`onepipeline next RUN [--filter NAME|SPEC] [--all]`",
            "reply RUN [FILE]",
            "surface RUN --kind KIND --message TEXT",
            "attest RUN REF",
            "stop RUN",
        ],
    );
    assert_contract_names("driver verb", &["onepipeline adopt RUN"]);

    // The views, as the contract lists them.
    let tokens = backticked();
    for view in ["runs", "status", "host", "results", "goals"] {
        assert!(
            tokens.contains(view),
            "the contract no longer lists the `{view}` view"
        );
    }
    assert!(tokens.contains("monitor RUN [--filter NAME|SPEC] [--all]"));
    assert!(tokens.contains("telemetry [--breakdown]"));
    assert!(tokens.contains("transcript RUN [NODE]"));
    assert!(tokens.contains("runs --mine"));
}

/// The words the telemetry document writes, gated against the contract that
/// names them.
///
/// Read out of the source rather than through the types: `telemetry` is behind
/// the contract's surface — the document reaches a consumer through the CLI —
/// so this suite cannot build a `BucketName` to ask it what it spells.
#[test]
fn the_contract_names_every_bucket_and_every_usage_party_the_document_writes() {
    let source = std::fs::read_to_string(repo_root().join("src/telemetry.rs"))
        .expect("the telemetry view ships");
    let tokens = backticked();

    for (what, list) in [
        ("bucket", "pub const ALL: [Self; 8]"),
        ("party", "pub const ALL: [Self; 4]"),
    ] {
        let declared: Vec<String> = source
            .split_once(list)
            .unwrap_or_else(|| panic!("telemetry declares its {what} list"))
            .1
            .split_once("];")
            .expect("the list is closed")
            .0
            .split(',')
            .filter_map(|entry| entry.trim().strip_prefix("Self::"))
            .map(wire_word)
            .collect();
        assert!(!declared.is_empty(), "the {what} list is empty");
        for name in declared {
            assert!(
                tokens.contains(&name),
                "the contract does not name the `{name}` {what}"
            );
        }
    }

    // And the fields a party's usage carries, each one a number an operator
    // budgets against.
    for field in ["input", "output", "cache_read", "cache_write", "cost_usd"] {
        assert!(
            tokens.contains(field),
            "the contract does not name the `{field}` usage field"
        );
    }
}

/// One `CamelCase` variant as the wire spells it: `snake_case`.
fn wire_word(variant: &str) -> String {
    let mut out = String::new();
    for (at, letter) in variant.char_indices() {
        if letter.is_uppercase() && at > 0 {
            out.push('_');
        }
        out.extend(letter.to_lowercase());
    }
    out
}

#[test]
fn a_command_outside_the_surface_is_refused() {
    Cli::try_parse_from(["onepipeline", "publish", "run-1"])
        .expect_err("the surface is exactly what the contract names");
}

#[test]
fn the_dag_scope_graph_is_a_monitor_plus_a_resettable_check_in() {
    assert!(CONTRACT.contains(
        "the shipped example, `graphs/dag-scope.yaml`, is an observer member beside a \
         resettable-cron `check-in` member; a host copies it or brings its own, and the \
         engine names no member of it"
    ));

    let text = std::fs::read_to_string(repo_root().join("graphs/dag-scope.yaml"))
        .expect("the dag-scope graph ships");
    // It parses as oneagentgraph's own schema — the library that launches it.
    let graph: GraphConfig = serde_norway::from_str(&text).expect("it is a valid graph config");
    assert_eq!(graph.name, "dag-scope");

    let monitor = graph.members.get("monitor").expect("a monitor member");
    let Member::Onejudge(monitor) = monitor else {
        panic!("the monitor is a two-party member");
    };
    match &monitor.judge[..] {
        [JudgeSide::Command(judge)] => assert_eq!(
            judge.command,
            [
                "onemessagebus",
                "serve",
                "--codec",
                "onejudge",
                "--transport-dir",
                "${ONEPIPELINE_RUNS_DIR}/${ONEPIPELINE_RUN_ID}/channel",
                "surfaces",
            ],
            "the monitor's judge side is exactly the adopted bus server command"
        ),
        other => panic!("the contract makes the judge side one command provider, not {other:?}"),
    }

    let check_in = graph.members.get("check-in").expect("a check-in member");
    let Member::Oneharness(check_in) = check_in else {
        panic!("the pacemaker is a single-sided member");
    };
    let schedule = check_in.schedule.expect("it is a cron member");
    assert!(
        schedule.resettable,
        "the contract makes the check-in resettable"
    );
    assert_eq!(
        schedule.every, DEFAULT_HEARTBEAT_INTERVAL_SECONDS,
        "its period is the driver's default heartbeat interval"
    );
}

/// The engine names no member of the observer graph a host runs.
///
/// The words that named one host's members — `monitor`, the observer, and
/// `pacemaker`, its clock — are gone from `src/` and `docs/`: identifiers,
/// comments, help text, messages and documentation describe the observer graph,
/// a scheduled member, an author other than the planner, and the check-in clock
/// instead. What the shipped example graph and personas under `graphs/` and
/// `personas/` call their members is theirs, as examples a host may copy, and the
/// engine reads none of them at runtime.
///
/// **The one allowance, by name, is the `onepipeline monitor` view verb**: it
/// means "watch a run" and names no member. An occurrence is that verb when it
/// is spelled as its invocation (`onepipeline monitor`); as its bare name in
/// backticks in prose (`` `monitor` ``) — provided the next word is not a
/// member, an author, a persona, a profile or a graph, which are the things a
/// member word would name — or with its arguments (`` `monitor RUN``,
/// `` `monitor --all``); as a string on a line that also spells its invocation
/// (the key beside `onepipeline monitor {run}`) or reads clap's name for it
/// (`get_name()`, `find_subcommand(`); or as the verb's own identifiers in code
/// (`Verb::Monitor(..)`, `fn monitor(..)` and calls of it). Every other
/// occurrence fails, named by file and line — a `"monitor"` author or member in
/// a fixture among them. `MonitorArgs` and `monitor_of`-style identifiers are
/// not whole words and are not asked about.
///
/// `CHANGELOG.md` and `docs/contract-divergences.md` are records of what was and
/// are exempt; so are recordings and fixtures under `tests/`, which this walk
/// never enters.
#[test]
fn no_member_word_names_the_observer_graph_in_the_engine_or_its_docs() {
    const VERB: &str = "monitor";
    const NAMED_THINGS: [&str; 7] = [
        "member", "author", "persona", "profile", "graph", "op", "role",
    ];
    fn walk(dir: &Path, into: &mut Vec<PathBuf>) {
        for entry in std::fs::read_dir(dir).expect("a directory to walk") {
            let path = entry.expect("an entry").path();
            if path.is_dir() {
                walk(&path, into);
            } else if path
                .extension()
                .is_some_and(|ext| ext == "rs" || ext == "md")
            {
                into.push(path);
            }
        }
    }
    let is_word = |c: char| c.is_alphanumeric() || c == '_';
    // Whether the whole word at `at` in `line` (lowercased) is the view verb.
    let names_the_verb = |line: &str, lower: &str, at: usize| -> bool {
        let before = &line[..at];
        let after = &line[at + VERB.len()..];
        let spelled = &line[at..at + VERB.len()];
        // `onepipeline monitor`: the verb invoked.
        if before.ends_with("onepipeline ") {
            return true;
        }
        // `Verb::Monitor(..)`, `fn monitor(..)`, `views::monitor(..)`.
        if after.starts_with('(') && (spelled == "monitor" || spelled == "Monitor") {
            return true;
        }
        if spelled != VERB {
            return false;
        }
        // The bare name in backticks, or the name with its arguments.
        if before.ends_with('`') && after.starts_with('`') {
            let rest = after[1..].trim_start_matches(|c: char| !c.is_alphanumeric());
            let next_word: String = rest.chars().take_while(|c| c.is_alphanumeric()).collect();
            return !NAMED_THINGS.contains(&next_word.as_str());
        }
        if before.ends_with('`')
            && (after.starts_with(" RUN")
                || after.starts_with(" <RUN>")
                || after.starts_with(" --"))
        {
            return true;
        }
        // The name as a string: only beside its invocation, or as the name
        // clap gives the verb.
        if before.ends_with('"') && after.starts_with('"') {
            return lower.contains("onepipeline monitor")
                || line.contains("get_name()")
                || line.contains("find_subcommand(");
        }
        false
    };

    let mut files = Vec::new();
    walk(&repo_root().join("src"), &mut files);
    walk(&repo_root().join("docs"), &mut files);
    files.sort();
    let mut offences = Vec::new();
    for path in files {
        if path.ends_with("docs/contract-divergences.md") {
            continue;
        }
        let text = std::fs::read_to_string(&path).expect("a source or document reads");
        for (number, line) in text.lines().enumerate() {
            let lower = line.to_lowercase();
            for word in [VERB, "pacemaker"] {
                let mut from = 0;
                while let Some(found) = lower[from..].find(word) {
                    let at = from + found;
                    from = at + word.len();
                    let whole = !line[..at].chars().next_back().is_some_and(is_word)
                        && !line[at + word.len()..].chars().next().is_some_and(is_word);
                    if !whole || (word == VERB && names_the_verb(line, &lower, at)) {
                        continue;
                    }
                    offences.push(format!(
                        "{}:{}: `{}` names a member of one host's observer graph: {}",
                        path.strip_prefix(repo_root()).unwrap_or(&path).display(),
                        number + 1,
                        &line[at..at + word.len()],
                        line.trim()
                    ));
                }
            }
        }
    }
    assert!(
        offences.is_empty(),
        "the engine names a member of the observer graph; the only allowance is the \
         `onepipeline monitor` view verb:\n{}",
        offences.join("\n")
    );
}

#[test]
fn the_default_node_scope_graph_is_a_worker_and_a_judge() {
    assert!(CONTRACT.contains("a default node-scope config (worker+judge)"));
    let text = std::fs::read_to_string(repo_root().join("graphs/node-scope.yaml"))
        .expect("the node-scope graph ships");
    let graph: GraphConfig = serde_norway::from_str(&text).expect("it is a valid graph config");
    assert_eq!(graph.name, "node-scope");

    let worker = graph.members.get("worker").expect("a worker member");
    let Member::Onejudge(worker) = worker else {
        panic!("worker+judge is a two-party member");
    };
    assert!(
        matches!(worker.judge[..], [JudgeSide::Harness(_)]),
        "the node-scope judge is one harness-backed side, not a panel: {:?}",
        worker.judge
    );
    assert_eq!(
        graph.members.len(),
        1,
        "worker+judge is one onejudge member"
    );
}

/// The shipped persona files, and the role each one carries.
///
/// The monitor's file keeps the `orchestrator.yaml` name it shipped under: the
/// orchestrator persona was **rewritten** into the observer, not replaced by a
/// file beside it, so the path a consumer already names keeps resolving. The
/// role is what changed, and the role is what the contract, the graph member,
/// and the channel's author allowlist all spell `monitor`.
const SHIPPED_PERSONAS: [(&str, &str); 3] = [
    ("orchestrator", "monitor"),
    ("check-in", "check-in"),
    ("pr-author", "pr-author"),
];

#[test]
fn every_persona_the_contract_ships_is_present_and_has_both_sides() {
    assert!(CONTRACT.contains(
        "example personas — an observer (at `personas/orchestrator.yaml`, the shipped file \
         the orchestrator persona was rewritten into), `check-in`, `pr-author`"
    ));
    assert!(
        CONTRACT
            .contains("The files under `graphs/` and `personas/` are **examples a host may copy**"),
        "the contract no longer says the shipped graph and personas are examples"
    );
    for (file, role) in SHIPPED_PERSONAS {
        let path = repo_root().join("personas").join(format!("{file}.yaml"));
        let text =
            std::fs::read_to_string(&path).unwrap_or_else(|e| panic!("{file} persona ships: {e}"));
        // Through the sibling's own reader, which is the trust boundary a member
        // resolving this file crosses: a persona in a shape it refuses ships
        // broken, and asserting on the YAML alone would not notice.
        let persona = Persona::parse(&text, &format!("personas/{file}.yaml"))
            .unwrap_or_else(|e| panic!("{file} is not a persona oneagentgraph loads: {e}"));
        assert_eq!(
            persona.label(),
            Some(role),
            "personas/{file}.yaml carries the {role} role"
        );
        // What it layers onto a base, read off the merge rather than off the
        // file: that is the config the member is actually launched with.
        let effective = merge("{}\n", "an empty base config", &persona)
            .unwrap_or_else(|e| panic!("{file} does not layer onto a base config: {e}"));
        assert!(
            effective.pointer("/system_prompt").is_some(),
            "{file} states the agent's role"
        );
        assert!(
            effective.pointer("/user/persona").is_some(),
            "{file} states the supervisor's review bar"
        );
    }
}

/// The monitor persona's own account of its allowlist is the allowlist.
///
/// The persona is the only thing the member ever reads, so a persona naming an
/// op the channel refuses sends it to be refused every run, and one silent about
/// an op the channel allows leaves that op unreachable however carefully it was
/// wired. Both directions, off the same `allows` the channel enforces with.

#[test]
fn the_pr_author_never_blocks_publication() {
    assert!(
        CONTRACT.contains("Drafting is never on the publication path."),
        "the contract no longer keeps the drafting dispatch off the publication path"
    );
    assert!(
        CONTRACT.contains(
            "the change request opens with no body and the node settles on its \
                           publication as before"
        ),
        "the contract no longer says what a drafting dispatch that ended badly costs"
    );
    let text = std::fs::read_to_string(repo_root().join("personas/pr-author.yaml"))
        .expect("the pr-author persona ships");
    // The persona is wrapped prose, so match on its words rather than its line
    // breaks.
    let flattened = text.split_whitespace().collect::<Vec<_>>().join(" ");
    assert!(
        flattened.contains("not on the publication path"),
        "the persona itself says the dispatch is not on the publication path"
    );
}

/// Every divergence the record raises, and the name its ruling adopted.
///
/// A divergence is closed by the contract *saying* the thing, so each entry
/// gates both halves: the record marks the item resolved, and the contract names
/// what was ruled. An entry that quietly loses its ruling, or a contract that
/// stops naming it, fails here.
const RULINGS: &[(&str, &str)] = &[
    ("1.", "ConfigRef"),
    ("2.", "SessionRequest"),
    ("3.", "DispatchOutcome"),
    ("4.", "node_label"),
    ("5.", "min_free_mem"),
    ("6.", "PipelineKind"),
    ("7.", "completed_steps"),
    ("8.", "cross-dag-satisfied"),
    ("9.", "publication_wait"),
    ("23.", "drive GRAPH"),
    ("24.", "NodeControls"),
    ("25.", "drive-run RUN"),
    ("26.", "nothing else able to move"),
    ("27.", "ending that parked driver politely"),
    ("28.", "`attempt`, `attempts`"),
    ("29.", "inherits both"),
    ("30.", "--launch-config FILE"),
    ("31.", "shaped event view beside the surface"),
    ("32.", "any run of characters including none"),
    ("34.", "body-not-drafted"),
    ("44.", "the minimum this build requires"),
    ("63.", "--correlation C"),
    ("74.", "holds any node that is not `done`"),
    (
        "75.",
        "`onemessagebus`'s own `docs/contract.md` is the one source of their shape",
    ),
    ("77.", "The planner channel runs on `onemessagebus`"),
    ("78.", "Surface kinds are an open vocabulary"),
    ("79.", "edit-applied"),
    ("81.", "the CLI is argument parsing over them"),
    ("82.", "held under `workspace`"),
    ("83.", "pool-maintenance"),
];

#[test]
fn every_recorded_divergence_is_ruled_on_or_states_the_proposal_it_waits_on() {
    let divergences = std::fs::read_to_string(repo_root().join("docs/contract-divergences.md"))
        .expect("the divergence record ships");

    let sections: Vec<&str> = divergences.split("\n## ").skip(1).collect();
    assert!(sections.len() >= RULINGS.len(), "{sections:?}");
    let section_of = |number: &str| {
        sections
            .iter()
            .find(|section| section.starts_with(number))
            .unwrap_or_else(|| panic!("the record has no divergence {number}"))
    };

    // Every entry a ruling has not closed is a proposal the planner who owns the
    // contract has not answered. It is recorded and marked, never resolved from
    // this repository — which is the whole point of the file. Matched by the
    // entry's own number rather than by its position, so a later ruling does not
    // have to be renumbered into the leading block to be recognised.
    for section in &sections {
        let heading = section.lines().next().expect("a heading");
        if RULINGS
            .iter()
            .any(|(number, _)| heading.starts_with(number))
        {
            continue;
        }
        // Or superseded by a later entry, which is the third state an unruled
        // divergence can be in: the proposal was answered by a *better* proposal
        // from this repository, not by a ruling. It stays for its history and
        // says where its shape went, and the entry it names has to be one the
        // record still carries and still open — a pointer at a resolved or
        // missing entry is a shape nobody is holding.
        if let Some((_, superseder)) = heading.split_once("— SUPERSEDED BY ") {
            let superseder = format!("{}.", superseder.trim());
            let later = sections
                .iter()
                .find(|section| section.starts_with(&superseder))
                .unwrap_or_else(|| {
                    panic!("`{heading}` names entry {superseder}, which the record does not have")
                });
            assert!(
                later.lines().next().expect("a heading").ends_with("— OPEN"),
                "`{heading}` is superseded by an entry that is not itself open"
            );
            assert!(
                section.contains("Superseded by entry"),
                "divergence `{heading}` is marked superseded and its prose does not say so"
            );
            continue;
        }
        assert!(
            heading.ends_with("— OPEN"),
            "an unruled divergence is not marked open: {heading}"
        );
        assert!(
            section.contains("**Proposal"),
            "divergence `{heading}` is open and states no proposal"
        );
    }

    for (number, named) in RULINGS {
        let section = section_of(number);
        let heading = section.lines().next().expect("a heading");
        assert!(
            heading.ends_with("— RESOLVED"),
            "divergence {number} is not marked resolved: {heading}"
        );
        assert!(
            section.contains("**Ruling:"),
            "divergence {number} is marked resolved but records no ruling"
        );
        // `executor_has_capacity` is the one name the record and the contract
        // shared before any ruling; every other is what a ruling adopted.
        assert!(
            CONTRACT.contains(named),
            "the contract does not name `{named}`, which divergence {number} was ruled onto it"
        );
    }
    assert!(CONTRACT.contains("executor_has_capacity"));
}

/// The `draft` field and the two environment names entry 69 proposes are what
/// this build carries, at the surfaces this file can reach.
///
/// The field: named in the contract's own reserved-key list, and a field of the
/// plan schema that round-trips as `body` does — omitted when false, carried when
/// true. The names: what the README documents a dispatch's environment as, since
/// the constants themselves are engine vocabulary `src/lifecycle.rs` holds to the
/// same block. Both directions, so a field or a name this build grows without a
/// line in that entry fails here as loudly as one the entry names that this build
/// dropped.
#[test]
fn the_draft_closeout_is_what_the_divergence_record_names() {
    let block = divergence_block("69.");
    let fields: Vec<String> =
        serde_json::from_value(block["node_fields"].clone()).expect("entry 69 names its fields");
    assert_eq!(fields, vec!["draft".to_string()]);
    for field in &fields {
        assert!(
            CONTRACT.contains(&format!("`{field}`")),
            "the contract's reserved-key list does not name `{field}`"
        );
    }
    let written = serde_json::to_value(Node {
        id: "held".into(),
        repo: Some("github.com/owner/name".into()),
        title: Some("feat: hold it".into()),
        draft: true,
        ..Node::default()
    })
    .expect("a node serialises");
    assert_eq!(written["draft"], json!(true));
    let read: Node = serde_json::from_value(written).expect("it re-reads");
    assert!(read.draft);
    let omitted = serde_json::to_value(Node::default()).expect("a node serialises");
    assert!(
        omitted.get("draft").is_none(),
        "`draft` is written when false, so a plan no longer round-trips as the file wrote it"
    );

    let environment: Vec<String> =
        serde_json::from_value(block["environment"].clone()).expect("entry 69 names the names");
    assert_eq!(
        environment,
        vec![
            "ONEVCS_SESSION".to_string(),
            "ONEPIPELINE_RUNS_DIR".to_string()
        ]
    );
    let readme = std::fs::read_to_string(repo_root().join("README.md")).expect("the README ships");
    for name in &environment {
        assert!(
            readme.contains(name),
            "the README does not document `{name}`, which entry 69 says every dispatch carries"
        );
    }
    // The opening sentence is the one the approved contract's drafter has always
    // been given, and the entry says so byte for byte.
    assert_eq!(
        block["drafting_task"]["opening"].as_str(),
        Some(
            "Read this branch's diff and write the change request's body, following the \
             repository's own template. The task this branch delivered:"
        )
    );
}

/// The sentence entry 54 proposes to amend is one the contract still carries.
///
/// The entry is open, so the contract says nothing about what a live edit does to
/// a node's `consumes` and there is no fenced block to drive. What can drift is
/// the entry's *citation*: an amendment that reworded the `retry` paragraph would
/// leave the register naming a sentence nobody can find, and an open proposal
/// nothing gates quietly stops being true.
#[test]
fn the_consumes_divergence_quotes_a_sentence_the_contract_still_carries() {
    let record = std::fs::read_to_string(repo_root().join("docs/contract-divergences.md"))
        .expect("the divergence record ships");
    let entry = record
        .split("\n## ")
        .find(|entry| entry.starts_with("54."))
        .expect("the divergence record carries entry 54");
    let quoted = entry
        .lines()
        .skip_while(|line| !line.starts_with("> "))
        .take_while(|line| line.starts_with("> "))
        .map(|line| line.trim_start_matches("> "))
        .collect::<Vec<_>>()
        .join(" ");
    assert!(
        !quoted.is_empty(),
        "entry 54 quotes no sentence of the contract"
    );
    let contract = CONTRACT.split_whitespace().collect::<Vec<_>>().join(" ");
    assert!(
        contract.contains(&quoted),
        "entry 54 names a sentence the contract no longer carries: {quoted}"
    );
    assert!(
        !contract.contains("consumes` with it"),
        "the contract now states what an edit does to `consumes`; entry 54 has been ruled on"
    );
}

/// The templated adoption instruction this build renders is exactly what the
/// divergence record proposes, at both sites and inside the frame.
///
/// The contract is committed as approved and names none of it, so entry 53 is the
/// only place it is written down — and a divergence nothing gates quietly stops
/// being true. The entry's own block is the source: the template written there is
/// one a producer could declare, and what it renders to is what a worker reads.
#[test]
fn the_templated_adoption_instruction_is_what_the_divergence_record_names() {
    let block = divergence_block("53.");

    // The three sentences the entry names are this crate's own constants, so a
    // build that reworded one fails here rather than leaving the record wrong.
    assert_eq!(
        block["default_instruction"].as_str(),
        Some(DEFAULT_ADOPTION_INSTRUCTION),
        "entry 53 names a different default instruction than this crate publishes"
    );
    assert_eq!(
        block["observed_state"].as_str(),
        Some(OBSERVED_STATE),
        "entry 53 names a different observed-state frame than this crate publishes"
    );
    assert_eq!(
        block["heading"].as_str(),
        Some(CROSS_REPO_REFERENCES_HEADING),
    );
    let variables: Vec<String> =
        serde_json::from_value(block["variables"].clone()).expect("entry 53 names its variables");
    assert_eq!(
        variables, ADOPTION_INSTRUCTION_VARIABLES,
        "entry 53 names a different variable set than this crate publishes"
    );

    // The row, and the producer's own template on it. Both are the entry's.
    let row = &block["row"];
    let field = |key: &str| {
        row[key]
            .as_str()
            .unwrap_or_else(|| panic!("entry 53's row states `{key}`"))
            .to_owned()
    };
    let reference = CrossRepoReference {
        dependency: field("dependency"),
        repository: field("repository"),
        branch: field("branch"),
        commit: field("commit"),
        release_target: field("release_target"),
        version: field("version"),
        adoption_instructions: Some(
            block["adoption_instructions"]
                .as_str()
                .expect("entry 53 states the template")
                .parse()
                .expect("the template entry 53 states is one `onevcs` accepts"),
        ),
    };
    let rendered = block["rendered"]
        .as_str()
        .expect("entry 53 states what its template renders to");
    assert_eq!(
        reference.instruction(),
        rendered,
        "the template entry 53 states does not render what it says it does"
    );

    // Both sites, through this crate's own API rather than through a dispatch:
    // the block a node's task carries, and the note its arrival is delivered as.
    let node = Node {
        id: "consumer".into(),
        persona: Some("engineer".into()),
        task: Some("## What\nbuild against the released engine".into()),
        ..Node::default()
    };
    let references = [reference];
    let calls: Vec<String> =
        serde_json::from_value(block["api"]["calls"].clone()).expect("entry 53 names its calls");
    assert_eq!(calls, ["adoption_instructions", "arrival_note"]);
    assert_eq!(block["api"]["module"].as_str(), Some("onepipeline::plan"));

    for (site, text) in [
        ("the reference block", node.rendered_task_with(&references)),
        ("the arrival note", arrival_note(&references)),
    ] {
        assert!(
            text.contains(rendered),
            "{site} does not carry the producer's own instruction:\n{text}"
        );
        // And it is enclosed by the frame, so a worker cannot read it as a bar.
        let opened = text
            .find(OBSERVED_STATE)
            .unwrap_or_else(|| panic!("{site} states no observed-state frame:\n{text}"));
        assert!(
            opened < text.find(rendered).expect("the instruction is there"),
            "{site} renders the instruction ahead of the frame that holds it:\n{text}"
        );
    }
    // The one rendering both sites share, reachable on its own.
    assert!(adoption_instructions(&references).contains(rendered));
    assert!(adoption_instructions(&[]).is_empty());

    // A producer that declares none is unaffected: the same call, the engine's
    // own default, and no template anywhere in it.
    let undeclared = CrossRepoReference {
        adoption_instructions: None,
        ..references[0].clone()
    };
    assert!(adoption_instructions(&[undeclared]).contains(DEFAULT_ADOPTION_INSTRUCTION));
}

/// The adoption flags are the one CLI extension the contract has not yet
/// approved, so entry 42 and the parser gate each other until the planner rules.
#[test]
fn the_adoption_flags_are_exactly_the_open_divergence_the_record_names() {
    let divergences = std::fs::read_to_string(repo_root().join("docs/contract-divergences.md"))
        .expect("the divergence record ships");
    let entry = divergences
        .split("\n## ")
        .find(|section| section.starts_with("42."))
        .expect("the divergence record carries entry 42");
    assert!(entry
        .lines()
        .next()
        .is_some_and(|line| line.ends_with("— OPEN")));
    assert!(entry.contains("`onepipeline adopt RUN [--attach|--detach]`"));
    assert!(CONTRACT.contains("`onepipeline adopt RUN` attaches"));
    assert!(!CONTRACT.contains("adopt RUN [--attach|--detach]"));
    let readme = std::fs::read_to_string(repo_root().join("README.md")).expect("the README ships");
    assert!(readme.contains("the same `--attach`/`--detach` pair `start` does"));

    Cli::try_parse_from(["onepipeline", "adopt", "run-1", "--attach"])
        .expect("the proposed attached form parses");
    Cli::try_parse_from(["onepipeline", "adopt", "run-1", "--detach"])
        .expect("the proposed detached form parses");
    Cli::try_parse_from(["onepipeline", "adopt", "run-1", "--attach", "--detach"])
        .expect_err("the proposed alternatives refuse each other");
}

/// The verb a consumer checks a plan with, and the flags the contract spells.
///
/// A test about the *surface* rather than about the loader: the journeys in
/// `tests/e2e/plan_check.rs` drive what it does. This is the gate that stops the
/// document and the argument parser drifting — a flag renamed here without the
/// contract saying so is interface drift, and a consumer pins to the spelling.
#[test]
fn the_plan_check_verb_is_the_one_the_contract_spells() {
    let contract = CONTRACT.split_whitespace().collect::<Vec<_>>().join(" ");
    assert!(
        contract.contains("`onepipeline plan check <SOURCE:PROJECT> [--check <PATH>]... [--json]`"),
        "the contract no longer states the `plan check` command line"
    );
    for claim in [
        "first the engine's own plan loader",
        "`ONEPIPELINE_PLAN_CHECK_SCHEMA=1` in its environment",
        "the registered checks do **not** run",
        "carry `\"source\": \"engine\"`",
    ] {
        assert!(
            contract.contains(claim),
            "the contract no longer states '{claim}'"
        );
    }

    let plan = Cli::command()
        .get_subcommands()
        .find(|sub| sub.get_name() == "plan")
        .expect("the binary offers `plan`")
        .clone();
    let check = plan
        .get_subcommands()
        .find(|sub| sub.get_name() == "check")
        .expect("`plan` offers `check`")
        .clone();
    let flags: BTreeSet<String> = check
        .get_arguments()
        .filter_map(|arg| arg.get_long().map(str::to_string))
        .collect();
    assert!(
        flags.contains("check") && flags.contains("json"),
        "{flags:?}"
    );

    // And it parses the way the contract writes it: repeatable `--check`, in the
    // order the flags were given.
    let parsed = Cli::parse_from([
        "onepipeline",
        "plan",
        "check",
        "otg:plan-store",
        "--check",
        "./first",
        "--check",
        "./second",
        "--json",
    ]);
    let Command::Plan(onepipeline::cli::PlanCommand::Check(args)) = parsed.command else {
        panic!("`plan check` did not parse as itself");
    };
    assert_eq!(args.project, "otg:plan-store");
    assert!(args.json);
    assert_eq!(
        args.checks,
        vec![PathBuf::from("./first"), PathBuf::from("./second")]
    );
}

/// The README's own copy of the `plan check` interface, reconciled clause by
/// clause with the contract it copies.
///
/// The README states the wire and the exit codes in an operator's prose, which
/// is a second copy of the contract — and a consumer writes a check against
/// whichever of the two it read. A gate that sampled a few phrases out of that
/// passage would let the copy drift in everything it did not sample, so this
/// reconciles the **whole** passage: every backticked token it writes has to be
/// one the contract's own plan-check paragraph writes, every claim it makes in
/// prose is paired with the contract clause that licenses it, the exit codes are
/// asserted against the constants the binary exits with **and** against the
/// contract, the flags against the argument parser that offers them, and the
/// example it shows against the parser that has to accept it.
///
/// The contract side is that one paragraph rather than the whole document: a
/// token this passage uses is only licensed by the passage it is a copy of.
#[test]
fn the_readmes_plan_check_passage_is_a_gated_copy_of_the_contract() {
    let raw = std::fs::read_to_string(repo_root().join("README.md")).expect("the README ships");
    // Wrapped prose on both sides, so match on words rather than line breaks.
    let readme = raw.split_whitespace().collect::<Vec<_>>().join(" ");
    let contract = CONTRACT
        .lines()
        .find(|line| line.contains("**A plan is checkable without launching it"))
        .expect("the contract has a plan-check paragraph")
        .split_whitespace()
        .collect::<Vec<_>>()
        .join(" ");
    let passage = readme
        .split_once("A plan is checkable before it is launched")
        .expect("the README has a plan-check passage")
        .1
        .split_once("The run's own record does not move")
        .expect("that passage ends where the README says it does")
        .0
        .to_string();

    // The invocation the passage shows is one the binary accepts, taken out of
    // the prose so a fence marker is not read as one of its tokens. A README
    // that showed a command the parser refuses is the drift a reader meets
    // first.
    let (before, rest) = passage
        .split_once("```bash ")
        .expect("the passage shows an example invocation");
    let (example, after) = rest
        .split_once(" ```")
        .expect("that example's fence closes");
    let parsed = Cli::try_parse_from(example.split_whitespace()).unwrap_or_else(|error| {
        panic!("the README shows `{example}`, which does not parse: {error}")
    });
    assert!(
        matches!(
            parsed.command,
            Command::Plan(onepipeline::cli::PlanCommand::Check(_))
        ),
        "the README's example is not a `plan check`: {example}"
    );
    let prose = format!("{before}{after}");

    // The wire, mechanically: every backticked run in the passage is one the
    // contract's own paragraph writes. No hand-picked list, so a field, a key, a
    // flag, or a literal added to the README without the contract stating it
    // fails here. A run the README elides with `...` is reconciled by the parts
    // it did write, because an elision is a shortening of the contract's shape
    // rather than a claim of its own.
    for token in backticked_runs(&prose) {
        for part in token
            .split("...")
            .map(str::trim)
            .filter(|part| !part.is_empty())
        {
            assert!(
                contract.contains(part),
                "the README's plan-check passage writes `{token}`, and the contract's \
                 plan-check paragraph does not state '{part}'"
            );
        }
    }

    // And the prose, clause by clause: each claim the passage makes, paired with
    // the contract clause it is a copy of. The two are worded for different
    // readers, so each side is asserted against its own document — and either
    // one rewritten without the other fails this.
    for (stated, licensed) in [
        (
            "every refusal `start` makes before it dispatches anything, and no other rule",
            "every refusal `onepipeline start` would make before dispatching anything, and no other rule",
        ),
        (
            "Each repeatable `--check <PATH>` names an executable, resolved against the directory the verb ran in",
            "a repeatable `--check <PATH>` flag naming an executable, resolved against the working directory `plan check` was run from",
        ),
        (
            "handed the **loaded** plan as one JSON document on its stdin",
            "spawned with the loaded plan as a single JSON document on its **stdin**",
        ),
        ("every default resolved", "with every default already resolved"),
        (
            "each node carrying its task's own metadata map verbatim",
            "the store's own metadata map for that task, verbatim",
        ),
        (
            "`ONEPIPELINE_PLAN_CHECK_SCHEMA=1` in its environment",
            "`ONEPIPELINE_PLAN_CHECK_SCHEMA=1` in its environment",
        ),
        (
            "answers on stdout with `{\"refusals\": [...]}`",
            "On **stdout** a check answers with one JSON object, `{\"refusals\":",
        ),
        ("and exit 0", "A check that ran answers with exit status **0**"),
        (
            "`node` and `field` present on each and null where it is about neither",
            "`node` and `field` are always present and may be null",
        ),
        (
            "Engine refusals come first and carry `\"source\": \"engine\"`",
            "Engine refusals come first and carry `\"source\": \"engine\"`",
        ),
        (
            "each check's follow in the order its flags were given, under the path as it was given",
            "each check's refusals follow in flag order and carry `\"source\": \"<the path as given>\"`",
        ),
        (
            "`--json` prints them as one object carrying `project`, `accepted`, `refusals` and `unrunnable`, always all four",
            "`{\"project\": <string>, \"accepted\": <bool>, \"refusals\": [{\"source\",\"node\",\"field\",\"reason\"}, ...], \"unrunnable\":",
        ),
        (
            "reported separately from a refusal",
            "which is reported separately from a refusal",
        ),
        ("never read as an accept", "is never read as an accept"),
        (
            "A loader refusal short-circuits: there is no loaded plan to hand a check, so each is reported as not run",
            "there is no loaded plan to hand it: the registered checks do **not** run, and each is reported as not run",
        ),
    ] {
        assert!(
            prose.contains(stated),
            "the README's plan-check passage no longer states '{stated}'"
        );
        assert!(
            contract.contains(licensed),
            "the README states '{stated}', and the contract's plan-check paragraph \
             no longer states '{licensed}'"
        );
    }

    // The exit codes, which are the binary's: the README names the numbers this
    // build actually exits with, and the contract names the same three.
    for (code, meaning, stated) in [
        (
            EXIT_SUCCESS,
            "the loader and every check accepting",
            "**0** — the loader and every check accepted",
        ),
        (
            EXIT_QUEUED,
            "at least one refusal from either source",
            "**1** — at least one refusal, from either source",
        ),
        (
            EXIT_REFUSED,
            "a project that could not be read",
            "**2** — the project could not be read at all, or a registered check could not be run",
        ),
    ] {
        assert!(
            prose.contains(&format!("`{code}` is {meaning}")),
            "the README's plan-check passage no longer maps exit {code} to {meaning}"
        );
        assert!(
            contract.contains(stated),
            "the contract's plan-check paragraph no longer states '{stated}'"
        );
    }

    // And the flags, which are the parser's.
    let check = Cli::command()
        .get_subcommands()
        .find(|sub| sub.get_name() == "plan")
        .expect("the binary offers `plan`")
        .clone()
        .get_subcommands()
        .find(|sub| sub.get_name() == "check")
        .expect("`plan` offers `check`")
        .clone();
    for flag in check
        .get_arguments()
        .filter_map(|arg| arg.get_long().map(str::to_string))
    {
        // Either alone or with the value it takes, which is how the passage
        // writes the one that has a value.
        assert!(
            prose.contains(&format!("`--{flag}`")) || prose.contains(&format!("`--{flag} ")),
            "the README's plan-check passage does not name `--{flag}`, which the verb takes"
        );
    }
}

/// Every backticked run in one piece of prose, in the order it wrote them.
///
/// An unterminated backtick ends the scan: what follows it is not a run, and
/// reading to the end of the text as though it were would let any prose past
/// this gate.
fn backticked_runs(prose: &str) -> Vec<String> {
    let mut runs = Vec::new();
    let mut rest = prose;
    while let Some((_, after)) = rest.split_once('`') {
        let Some((run, tail)) = after.split_once('`') else {
            break;
        };
        if !run.trim().is_empty() {
            runs.push(run.to_string());
        }
        rest = tail;
    }
    runs
}

#[test]
fn the_smoke_scripts_command_list_is_the_binarys_whole_surface() {
    // The published-artifact smoke checks `--help` against a hand-written word
    // list. Without this gate a command added to the contract could be missing
    // from every published binary and the smoke would still pass.
    let script = std::fs::read_to_string(repo_root().join("scripts/smoke-published.sh"))
        .expect("the smoke script ships");
    let listed = script
        .lines()
        .find_map(|line| {
            line.trim()
                .strip_prefix("for command in ")?
                .strip_suffix("; do")
        })
        .expect("the smoke script iterates a `for command in ...; do` list")
        .split_whitespace()
        .map(str::to_string)
        .collect::<BTreeSet<String>>();

    let (documented, hidden): (BTreeSet<String>, BTreeSet<String>) = Cli::command()
        .get_subcommands()
        .map(|sub| (sub.get_name().to_string(), sub.is_hide_set()))
        .fold(
            Default::default(),
            |(mut shown, mut hidden), (name, hide)| {
                if hide {
                    hidden.insert(name);
                } else {
                    shown.insert(name);
                }
                (shown, hidden)
            },
        );

    assert_eq!(
        listed, documented,
        "scripts/smoke-published.sh checks a different command set than the CLI offers"
    );

    // A hidden verb is not on `--help`, so the loop above has nothing to find it
    // in — and it is exactly the kind a published artifact could lack without
    // anything noticing, because no user types it. `drive` is what `start
    // --detach` spawns of *itself*, so a build without it cannot launch a
    // detached run at all. The script reaches each one directly instead.
    for command in &hidden {
        assert!(
            script.contains(&format!("onepipeline {command} ")),
            "scripts/smoke-published.sh never runs the hidden `{command}` command, which \
             `--help` does not list for it to check"
        );
    }
}

#[test]
fn the_readmes_interface_claims_match_the_code_they_describe() {
    // The README restates numbers that live in the code. Each copy is gated
    // here, so a change to the code fails the suite instead of leaving the
    // README quietly wrong.
    let raw = std::fs::read_to_string(repo_root().join("README.md")).expect("the README ships");
    // Wrapped prose, so match on its words rather than its line breaks.
    let readme = raw.split_whitespace().collect::<Vec<_>>().join(" ");

    assert!(
        readme.contains(&format!(
            "exit `{EXIT_NOTHING_DRIVING}` means nothing is driving"
        )),
        "the README states a different code for an undriven run than the crate uses"
    );
    assert!(
        readme.contains(&format!(
            "exits `{EXIT_SUCCESS}` when the reconciler applied it, `{EXIT_SUCCESS}` when it is \
             accepted and still queued"
        )) && readme.contains(&format!("and `{EXIT_REFUSED}` when it was refused")),
        "the README's reply exit-code mapping no longer matches the crate's constants"
    );
    assert!(
        !readme.contains(&format!("`{EXIT_QUEUED}` when it is queued")),
        "the README still maps a queued reply to a non-zero status"
    );

    // Every view the README lists is a command the binary actually offers.
    let surface = Cli::command()
        .get_subcommands()
        .map(|sub| sub.get_name().to_string())
        .collect::<BTreeSet<String>>();
    let views = readme
        .split_once("Read-only views")
        .expect("the README has a read-only views paragraph")
        .1
        .split_once("without touching a run")
        .expect("that paragraph ends where the README says it does")
        .0
        .to_string();
    for view in [
        "runs",
        "status",
        "host",
        "monitor",
        "results",
        "goals",
        "transcript",
        "agents",
        "telemetry",
    ] {
        assert!(
            views.contains(&format!("`{view}`")),
            "the README's view list omits `{view}`"
        );
        assert!(
            surface.contains(view),
            "`{view}` is not a command the binary offers"
        );
    }

    // The run-history passage restates, in prose, what `onepipeline::agents`
    // holds as constants and the contract's `oneharness_history` block spells:
    // the pointer file's name, the one variable the engine never sets, the
    // label prefix, the scope words, the opt-out, and the two shapes of the
    // verb. Each spelling is read out of the code or the contract here, so a
    // renamed file or scope word fails the suite rather than leaving a reader
    // looking for a file the run no longer writes.
    {
        use onepipeline::agents;
        let passage = readme
            .split_once("**Every agent a run launches is visible, automatically")
            .expect("the README documents the run's agent visibility")
            .1
            .split_once("**A listing groups its runs by project.**")
            .expect("that passage ends where the README's next one begins")
            .0
            .to_string();
        assert!(
            passage.contains(&format!("`<run root>/{}`", agents::SESSIONS_FILE)),
            "the README's pointer file is not the one the run writes"
        );
        assert!(
            passage.contains(&format!("never sets `{}`", agents::HISTORY_DIR_ENV)),
            "the README does not name the store variable the engine leaves alone"
        );
        assert!(
            passage.contains(&format!("under the `{}` prefix", agents::LABEL_PREFIX)),
            "the README's label prefix is not the engine's"
        );
        let scopes = agents::Scope::ALL
            .iter()
            .map(|scope| format!("`{}`", scope.as_str()))
            .collect::<Vec<_>>();
        assert!(
            passage.contains(&format!("({}, {} or {})", scopes[0], scopes[1], scopes[2])),
            "the README's scope words are not the crate's"
        );
        let block: Value = serde_json::from_str(&fenced_block_naming("json", "oneharness_history"))
            .expect("the run-history block is JSON");
        let opt_out = block["oneharness_history"]["opt_out"]
            .as_str()
            .expect("the contract spells the opt-out");
        assert!(
            passage.contains(&format!("with `{opt_out}` writes no line")),
            "the README's opt-out is not the contract's"
        );
        for usage in [
            "`onepipeline agents RUN [NODE]`",
            "`onepipeline agents --project PROJECT`",
        ] {
            assert!(
                passage.contains(usage),
                "the README's run-history passage does not show {usage}"
            );
        }
        let agents_verb = Cli::command()
            .get_subcommands()
            .find(|sub| sub.get_name() == "agents")
            .expect("the binary offers `agents`")
            .clone();
        assert!(
            agents_verb
                .get_arguments()
                .any(|arg| arg.get_long() == Some("project")),
            "`agents` takes no `--project`, which the README shows"
        );
    }

    // `watch` is documented in a passage of its own, and the two things it
    // restates that a caller writes code against are read out of the code here:
    // the flags are clap's, and the four terminal statuses are the exit-code
    // constants. A status moved in the crate and left in the README fails the
    // suite rather than sending a supervisor to branch on one the binary no
    // longer returns. The same passage's event set and NDJSON record schema are
    // reconciled against `MEANINGFUL` and `Record`, which are private to the
    // module, by that module's own `the_readme_passage_...` unit test.
    assert!(
        surface.contains("watch"),
        "`watch` is not a command the binary offers"
    );
    let passage = readme
        .split_once("`onepipeline watch RUN` is the bounded wait")
        .expect("the README documents the watch verb")
        .1
        .split_once("## Where a dispatch runs")
        .expect("that passage ends where the README's next heading begins")
        .0
        .to_string();
    let watch = Cli::command()
        .get_subcommands()
        .find(|sub| sub.get_name() == "watch")
        .expect("the binary offers `watch`")
        .clone();
    for flag in watch
        .get_arguments()
        .filter_map(|arg| arg.get_long().map(str::to_string))
    {
        assert!(
            passage.contains(&format!("`--{flag}`")) || passage.contains(&format!("`--{flag} ")),
            "the README's watch passage does not name `--{flag}`, which the verb takes"
        );
    }
    // The `--until` vocabulary, both ways, out of the constant a caller's
    // condition is parsed against. A condition this build accepts and the README
    // never mentions is one nobody was told they could ask for — and the wait it
    // ends is the one that may now have no bound at all — while a condition the
    // README keeps past the code sends a supervisor to spell something this
    // binary refuses.
    for condition in onepipeline::cli::watch_conditions() {
        assert!(
            passage.contains(&format!("`--until {condition}`")),
            "the README's watch passage does not name `--until {condition}`, which the verb \
             accepts"
        );
    }
    for named in passage
        .split('`')
        .skip(1)
        .step_by(2)
        .filter_map(|span| span.strip_prefix("--until "))
    {
        assert!(
            onepipeline::cli::watch_conditions().contains(&named),
            "the README's watch passage names `--until {named}`, which the verb does not accept"
        );
    }

    for (what, stated) in [
        (
            "the run settling",
            format!("exit `{EXIT_SUCCESS}` when the run settled complete"),
        ),
        (
            "nothing driving the run",
            format!("`{EXIT_NOTHING_DRIVING}` when nothing is driving it"),
        ),
        (
            "a blocking surface waiting",
            format!("`{EXIT_SURFACE_WAITING}` when a blocking surface is waiting"),
        ),
        (
            "the wait elapsing",
            format!("`{EXIT_WATCH_ELAPSED}` when the"),
        ),
        (
            "a node the wait named settling",
            format!("`{EXIT_NODE_SETTLED}` when a node the wait was told to return on settled"),
        ),
    ] {
        assert!(
            passage.contains(&stated),
            "the README's watch passage states a different status than the crate does for \
             {what}: it should say '{stated}'"
        );
    }

    // And the other way, which the loops above cannot see. Everything so far
    // fails on a claim the README is *missing*; a claim it has kept past the code
    // would pass all of it, and a stale one is the worse failure — a supervisor
    // reading it writes a script against a flag or a status this binary does not
    // have, which is the prose-matching the verb exists to end.
    let offered: BTreeSet<String> = watch
        .get_arguments()
        .filter_map(|arg| arg.get_long().map(str::to_string))
        .collect();
    for mentioned in passage
        .split('`')
        .skip(1)
        .step_by(2)
        .filter_map(|span| span.split_whitespace().next())
        .filter_map(|word| word.strip_prefix("--"))
    {
        // The one flag the passage names that this verb must not take: it is
        // `start`'s pacemaker cadence, and the sentence saying the two are
        // different clocks is the point of naming it.
        if mentioned == "heartbeat-interval" {
            assert!(
                !offered.contains(mentioned),
                "`watch` took `start`'s pacemaker flag, which the README says it refuses"
            );
            continue;
        }
        assert!(
            offered.contains(mentioned),
            "the README's watch passage names `--{mentioned}`, which the verb does not take"
        );
    }
    let returned: BTreeSet<String> = [
        EXIT_SUCCESS,
        EXIT_NOTHING_DRIVING,
        EXIT_SURFACE_WAITING,
        EXIT_WATCH_ELAPSED,
        EXIT_NODE_SETTLED,
    ]
    .iter()
    .map(i32::to_string)
    .collect();
    for mentioned in passage
        .split('`')
        .skip(1)
        .step_by(2)
        .filter(|span| span.parse::<i32>().is_ok())
    {
        assert!(
            returned.contains(mentioned),
            "the README's watch passage states exit status `{mentioned}`, which this verb \
             never returns"
        );
    }
}

/// The one manager-note op this build carries where the contract names two is
/// exactly what the divergence record proposes.
///
/// The contract is committed as approved: it lists `context` and names no `note`,
/// so entry 60 is the only place the survivor's shape is written down — and a
/// divergence nothing gates quietly stops being true. The entry's own block is the
/// source: what parses here is what a manager would type, and the refusals are what
/// a manager would be told.
#[test]
fn the_note_delivery_surface_is_what_the_divergence_record_names() {
    let block = divergence_block("60.");

    // The envelope this op travels in declares the version the entry names, and
    // the contract's own sentence still names the one it was cut from — which is
    // what makes this a recorded divergence rather than a document edited to
    // match the code.
    let version = block["envelope_version"]
        .as_u64()
        .expect("entry 60 names the envelope version") as u32;
    assert!(
        onepipeline::channel::REPLY_ENVELOPE_VERSIONS_READ.contains(&version),
        "the build no longer reads the envelope version entry 60 declares"
    );
    assert!(
        onepipeline::channel::REPLY_ENVELOPE_VERSION >= version,
        "the build writes an envelope version older than the one entry 60 moved it to"
    );
    assert!(
        CONTRACT.contains(r#"{"version": 1, "commands": [...]}"#),
        "the contract's envelope sentence moved; entry 60 names the version it replaces"
    );

    // The op, as the wire carries it, and the authors it is for.
    let fixtures: Vec<Value> =
        serde_json::from_value(block["ops"].clone()).expect("entry 60 names the op it adds");
    let monitor_may: BTreeSet<String> = serde_json::from_value(block["monitor_may_issue"].clone())
        .expect("entry 60 says which of them the monitor may issue");
    assert!(!fixtures.is_empty(), "{block}");
    for fixture in &fixtures {
        let op = fixture["op"].as_str().expect("the fixture names its op");
        assert!(
            !OPS.contains(&op),
            "`{op}` is on the contract's own list, so it is no divergence"
        );
        let edit: Edit = serde_json::from_value(fixture.clone())
            .unwrap_or_else(|e| panic!("`{op}` deserializes: {e}"));
        assert_eq!(op_of(&edit), op, "`{op}` deserialized into another variant");
        assert_eq!(
            &serde_json::to_value(&edit).expect("serializes"),
            fixture,
            "`{op}` round-trips unchanged"
        );
        allows(Author::planner(), &edit)
            .unwrap_or_else(|e| panic!("the planner was refused `{op}`: {e}"));
        let verdict = allows(Author::from("monitor"), &edit);
        if monitor_may.contains(op) {
            verdict.unwrap_or_else(|e| panic!("the monitor was refused `{op}`: {e}"));
            continue;
        }
        // Refused, and the refusal names the op and what to do instead — an
        // observer told only "no" has nothing to act on.
        let refusal = verdict
            .expect_err(&format!("the monitor was allowed `{op}`"))
            .to_string();
        assert!(
            refusal.contains(op) && refusal.contains("Surface it to the planner"),
            "the refusal does not name `{op}` and what to do instead: {refusal}"
        );
    }

    // The whole envelope a manager sends, parsed as one: a `note` is an edit like
    // any other and travels the same way.
    let envelope: Reply =
        serde_json::from_value(json!({"version": version, "commands": block["ops"].clone()}))
            .expect("entry 60's ops travel in a reply envelope");
    assert_eq!(envelope.commands.len(), fixtures.len());

    // The whole field set, and nothing outside it. Every field the entry names is
    // carried by one of its fixtures and every key a fixture carries is one it
    // names, so the set is proven from both ends — a field added to the op
    // without a line in the record fails here. No single fixture can carry all
    // six, because the two defaults are omitted from what a note serializes and
    // the one combination that writes both out is the one the run refuses.
    let fields: BTreeSet<String> =
        serde_json::from_value(block["fields"].clone()).expect("entry 60 names the field set");
    let carried_keys: BTreeSet<String> = fixtures
        .iter()
        .flat_map(|fixture| fixture.as_object().expect("an object").keys().cloned())
        .filter(|key| key != "op")
        .collect();
    assert_eq!(
        carried_keys, fields,
        "entry 60's fixtures and its own field set disagree about what a note carries"
    );

    // The fields an envelope refuses to do without, each dropped on its own so
    // the refusal is about that field rather than about the note being empty.
    let required: Vec<String> =
        serde_json::from_value(block["required"].clone()).expect("entry 60 names them");
    for field in &required {
        for fixture in &fixtures {
            let mut without = fixture.clone();
            without
                .as_object_mut()
                .expect("an object")
                .remove(field.as_str());
            let err = serde_json::from_value::<Edit>(without)
                .expect_err(&format!("a note without `{field}` is refused"));
            assert!(
                err.to_string().contains(field),
                "the refusal does not name `{field}`: {err}"
            );
        }
    }

    // Every addressee the seam has, spelled as the entry writes them — and the
    // spelling is the seam's own rather than a second one this crate keeps.
    let addressees: Vec<String> =
        serde_json::from_value(block["addressees"].clone()).expect("entry 60 names the addressees");
    for named in &addressees {
        let parsed: Addressee = serde_json::from_value(json!(named))
            .unwrap_or_else(|e| panic!("`{named}` is an addressee: {e}"));
        assert_eq!(parsed.as_str(), named, "`{named}` round-trips");
    }
    assert!(addressees.contains(&"worker".to_string()));

    // And every disposition, likewise: what a delivery can answer is the set the
    // record carries, so a party added upstream — or a disposition this crate
    // composes itself — cannot land here unnamed.
    let reached: Vec<String> =
        serde_json::from_value(block["reached"].clone()).expect("entry 60 names the dispositions");
    let carried = [
        Reached::Queued,
        Reached::Worker,
        Reached::Supervisor,
        Reached::JudgedWith {
            completion_reason: "the work is done".into(),
        },
        Reached::Carried,
    ];
    assert_eq!(
        carried
            .iter()
            .map(|one| one.as_str().to_string())
            .collect::<Vec<_>>(),
        reached,
        "entry 60 names dispositions this build does not carry, or the other way round"
    );
    for one in &carried {
        let written = serde_json::to_value(one).expect("a disposition serializes");
        let read: Reached = serde_json::from_value(written.clone()).expect("and reads back");
        assert_eq!(&read, one, "{written} did not round-trip");
    }

    // And what each disposition confirms at the acknowledgement and what it
    // only routes onward, as entry 70 tabulates them — the two facts a reader
    // of the record no longer has to tell apart by inference. Keyed by the
    // disposition's own word, and exhaustive both ways: a disposition a table
    // does not name, or a row naming no disposition, fails here.
    let tabulated_by = divergence_block("70.");
    let words = |party: &Party| -> String {
        serde_json::to_value(party)
            .expect("a party serializes")
            .as_str()
            .expect("a party is a word")
            .to_string()
    };
    for (table, answer) in [
        (
            "shown_at_delivery",
            Reached::shown_at_delivery as fn(&Reached) -> &'static [Party],
        ),
        (
            "routed_to",
            Reached::routed_to as fn(&Reached) -> &'static [Party],
        ),
    ] {
        let tabulated: std::collections::BTreeMap<String, Vec<String>> =
            serde_json::from_value(tabulated_by[table].clone())
                .unwrap_or_else(|e| panic!("entry 70 tabulates `{table}` per disposition: {e}"));
        assert_eq!(
            tabulated.keys().cloned().collect::<Vec<_>>(),
            {
                let mut named = reached.clone();
                named.sort();
                named
            },
            "entry 70's `{table}` table and entry 60's dispositions are not one set"
        );
        for one in &carried {
            assert_eq!(
                &answer(one).iter().map(words).collect::<Vec<_>>(),
                &tabulated[one.as_str()],
                "`{}`'s `{table}` is not what entry 70 says",
                one.as_str()
            );
        }
    }

    // The boundary: an envelope this seam cannot act on is refused where it
    // arrives, by serde and by the seam's own newtypes, rather than somewhere
    // later. The removed op is in this list, so its refusal is proven by the
    // same rule that proves a malformed note's.
    let refused: Vec<Value> =
        serde_json::from_value(block["refused"].clone()).expect("entry 60 names what is refused");
    let removed: Vec<String> =
        serde_json::from_value(block["removed"].clone()).expect("entry 60 names what it removes");
    for fixture in &refused {
        let read = serde_json::from_value::<Edit>(fixture.clone());
        let err = read
            .err()
            .unwrap_or_else(|| {
                panic!("the envelope's boundary accepted a note it cannot deliver: {fixture}")
            })
            .to_string();
        let op = fixture["op"].as_str().expect("the fixture names its op");
        if removed.contains(&op.to_string()) {
            assert!(
                err.contains(op),
                "the removed op was refused without being named: {err}"
            );
        }
    }
    assert!(
        refused.iter().any(
            |fixture| removed.contains(&fixture["op"].as_str().unwrap_or_default().to_string())
        ),
        "entry 60 removes an op and drives no envelope carrying it"
    );

    // The same delivery on this crate's own surface, named as the entry names it.
    // A note carrying an unusable criterion is unrepresentable here too, so the
    // two spellings refuse the same things.
    let api = &block["api"];
    assert_eq!(api["module"].as_str(), Some("onepipeline::note"));
    let note = Note::to(Addressee::Worker, "stop editing src/old.rs");
    assert!(!note.binds());
    assert_eq!(note.addressee, Addressee::Worker);
    let bound = note
        .clone()
        .binding("`version.txt` holds `v: 2`")
        .expect("a criterion the seam accepts");
    assert!(bound.binds());
    assert!(
        Note::new(Addressee::Worker, "  ").is_err(),
        "a blank note is not a note"
    );
    // The two answers, by the names the entry proposes.
    let answers: Vec<String> =
        serde_json::from_value(api["answers"].clone()).expect("entry 60 names what it answers");
    let spelled = [
        format!("{:?}", Delivered::To(Reached::Worker)),
        format!("{:?}", Delivered::Queued),
    ];
    for (answer, spelled) in answers.iter().zip(spelled.iter()) {
        assert!(
            spelled.starts_with(answer),
            "entry 60 names `{answer}`, which this build spells `{spelled}`"
        );
    }
    assert_eq!(answers.len(), spelled.len());
    // The calls themselves are the ones the entry names, asked of the compiler
    // rather than of a list beside it: the one that takes the defaults, and the
    // one that names both axes.
    assert_eq!(api["call"].as_str(), Some("deliver"));
    assert_eq!(api["with"].as_str(), Some("deliver_with"));
    let _: fn(&RunPaths, &str, &Note) -> onepipeline::Result<Delivered> =
        onepipeline::note::deliver;
    let _: fn(
        &RunPaths,
        &str,
        &Note,
        onepipeline::channel::Deliver,
        bool,
    ) -> onepipeline::Result<Delivered> = onepipeline::note::deliver_with;
}

/// The paragraph of the contract that states the post-launch verbs.
fn verbs_paragraph() -> &'static str {
    CONTRACT
        .split("\n\n")
        .find(|paragraph| {
            paragraph.starts_with(
                "**The post-launch verbs are the SDK, and the CLI is argument parsing over them.**",
            )
        })
        .expect("docs/contract.md states the post-launch verbs")
}

/// Every function `src/verbs.rs` publishes.
fn verb_functions_in_source() -> BTreeSet<String> {
    std::fs::read_to_string(repo_root().join("src/verbs.rs"))
        .expect("the verbs ship")
        .lines()
        .filter_map(|line| line.strip_prefix("pub fn "))
        .map(|rest| {
            rest.chars()
                .take_while(|c| c.is_ascii_alphanumeric() || *c == '_')
                .collect::<String>()
        })
        .collect()
}

/// Every function the contract's verbs paragraph names.
///
/// A verb is written as its signature — `` `name(args) -> Type` `` — and a
/// renderer as its bare name or its own signature: what is read here is every
/// backticked span that opens with an identifier and either carries `->` or is a
/// `render_` name. A method a result type answers (`pending()`, `refusal()`)
/// carries neither, and belongs to its type rather than to this list.
fn verb_functions_in_contract() -> BTreeSet<String> {
    backticked_in(verbs_paragraph())
        .into_iter()
        .filter_map(|span| {
            let name: String = span
                .chars()
                .take_while(|c| c.is_ascii_alphanumeric() || *c == '_')
                .collect();
            let rest = &span[name.len()..];
            let signature = rest.starts_with('(') && span.contains("->");
            let renderer =
                name.starts_with("render_") && (rest.is_empty() || rest.starts_with('('));
            (!name.is_empty() && (signature || renderer)).then_some(name)
        })
        .collect()
}

/// The contract's list of verb functions and `src/verbs.rs` agree, in both
/// directions: a function published there and not named here is surface the
/// contract does not promise, and a name here the source dropped is a promise
/// to nobody.
///
/// The names are held by reading the two; the **signatures** are held by the
/// compiler, which is the drift gate this file uses everywhere: each verb and
/// renderer below is coerced to a function pointer of the signature the contract
/// spells, so a parameter or a result type that moves fails to compile here
/// before the document can go on stating the old one.
#[test]
fn the_contract_names_every_post_launch_verb_the_sdk_publishes_and_no_other() {
    use onemessagebus::Correlation;
    use onepipeline::channel::SurfaceKind;
    use onepipeline::filter::EventFilter;
    use onepipeline::verbs::{
        Adopt, Adopted, ChannelQueue, Goals, Grouping, Host, Monitored, Next, Receipt, Results,
        Retained, Shutdown, ShutdownRequest, Status, StopRequest, Stopped, Surfaced, Transcript,
        Unwatched, WatchFrame, WatchLines, WatchOutcome, WatchRequest,
    };
    use onepipeline::views::{DriverLiveness, Projects};
    use onepipeline::Result;

    let _: fn(&Path, &str, bool) -> Projects = verbs::runs;
    let _: fn(&Projects, Grouping, &str) -> String = verbs::render_runs;
    let _: fn(&Path, Option<&str>) -> Result<Status> = verbs::status;
    let _: fn(&Status) -> String = verbs::render_status;
    let _: fn(&Path) -> Host = verbs::host;
    let _: fn(&Host) -> String = verbs::render_host;
    let _: fn(&Path, Option<&str>) -> Result<Goals> = verbs::goals;
    let _: fn(&Goals) -> String = verbs::render_goals;
    let _: fn(&RunPaths) -> Result<Results> = verbs::results;
    let _: fn(&Results) -> String = verbs::render_results;
    let _: fn(&RunPaths, Option<&str>) -> Result<Transcript> = verbs::transcript;
    let _: fn(&Transcript) -> String = verbs::render_transcript;
    let _: fn(&Path, Option<&str>) -> Result<Vec<RunTelemetry>> = verbs::telemetry;
    let _: fn(&[RunTelemetry], bool) -> Result<String> = verbs::render_telemetry;
    let _: fn(&RunTelemetry) -> String = verbs::render_telemetry_breakdown;
    let _: fn(&RunPaths, &[Envelope]) -> RunTelemetry = onepipeline::telemetry::of_run;
    let _: fn(&RunPaths, &EventFilter, Option<&str>) -> Result<Monitored> = verbs::monitor;
    let _: fn(&Monitored) -> String = verbs::render_monitor;
    let _: fn(&RunPaths, &EventFilter) -> Result<Next> = verbs::next;
    let _: fn(&Next) -> String = verbs::render_next;
    let _: fn(&RunPaths) -> Result<ChannelQueue> = verbs::channel;
    let _: fn(&ChannelQueue) -> Result<String> = verbs::render_channel;
    /// The sink a watch hands its frames to, as the contract spells it.
    type WatchSink<'a> = &'a mut dyn FnMut(WatchFrame<'_>) -> Result<()>;
    let _: fn(&RunPaths, &WatchRequest, WatchSink<'_>) -> Result<WatchOutcome> = verbs::watch;
    let _: fn(&WatchFrame<'_>) -> Result<WatchLines> = verbs::render_watch_frame;
    let _: fn(&Path, &str) -> Result<Unwatched> = verbs::unwatched;
    let _: fn(&Unwatched) -> String = verbs::render_unwatched;
    let _: fn(&RunPaths, Option<&Correlation>, &str) -> Result<Receipt> = verbs::reply;
    let _: fn(&Receipt) -> Result<String> = verbs::render_receipt;
    let _: fn(&RunPaths, &str) -> Result<Receipt> = verbs::attest;
    let _: fn(&RunPaths, SurfaceKind, String) -> Result<Surfaced> = verbs::surface;
    let _: fn(&Surfaced) -> String = verbs::render_surfaced;
    let _: fn(&RunPaths, StopRequest<'_>) -> Result<Stopped> = verbs::stop;
    let _: fn(&Stopped) -> String = verbs::render_stopped;
    let _: fn(&Path, ShutdownRequest) -> Result<Shutdown> = verbs::shutdown;
    let _: fn(&Shutdown) -> String = verbs::render_shutdown;
    let _: fn(&RunPaths, Adopt) -> Result<Adopted> = verbs::adopt;
    let _: fn(&Adopted) -> String = verbs::render_adopted;
    let _: fn(&RunPaths, Retained) -> Result<i32> = verbs::drive_run;
    let _: fn(&RunSummary) -> DriverLiveness = onepipeline::views::liveness_of;
    let _: fn(&RunPaths) -> Result<Plan> = onepipeline::views::plan_of;

    let in_source = verb_functions_in_source();
    let in_contract = verb_functions_in_contract();
    assert_eq!(
        in_contract, in_source,
        "the verbs paragraph and src/verbs.rs disagree\n  named and not published: {:?}\n  published and not named: {:?}",
        in_contract.difference(&in_source).collect::<Vec<_>>(),
        in_source.difference(&in_contract).collect::<Vec<_>>()
    );
    // Each of the verbs the plan fixed, by name, so the list cannot quietly lose
    // one and still agree with a source that lost it too.
    for verb in [
        "next",
        "reply",
        "surface",
        "attest",
        "stop",
        "shutdown",
        "render_shutdown",
        "adopt",
        "drive_run",
        "watch",
        "unwatched",
        "runs",
        "status",
        "host",
        "monitor",
        "channel",
        "results",
        "goals",
        "transcript",
        "telemetry",
    ] {
        assert!(
            in_source.contains(verb),
            "src/verbs.rs publishes no `{verb}`"
        );
    }
    assert_contract_names(
        "verbs paragraph's",
        &[
            "`onepipeline::verbs`",
            "`tests/parity.rs`",
            "`telemetry::of_run`",
            "`views::liveness_of(&RunSummary) -> DriverLiveness`",
            "`views::plan_of(&RunPaths) -> Plan`",
            "`channel::{Surface, QueuedReply, QueuedCommands, CommandOutcome, CommandResult, CommandVerdict}`",
            "`channel queue RUN`",
        ],
    );
}

/// The grouped listing is what the contract states, and the types it names are
/// the ones this build carries: groups keyed by project, ordered newest activity
/// first and then by project id, the no-project group an ordinary group, and the
/// header marker one no run line can start with.
#[test]
fn the_grouped_listing_is_what_the_contract_states() {
    assert_contract_names(
        "grouped listing's",
        &[
            "`views::Projects { root, groups: Vec<ProjectGroup>, skipped }`",
            "`ProjectGroup { project: Option<String>, name: Option<String>, last_write_at: Option<u64>, runs: Vec<RunSummary> }`",
            "`Projects::of(&Listing)`",
            "`Projects::flat()`",
            "`views::GROUP_HEADER`",
            "`runs --flat`",
            "(no project)",
        ],
    );
    assert_eq!(NO_PROJECT, "(no project)");
    assert!(
        GROUP_HEADER
            .starts_with(|c: char| !c.is_alphanumeric() && c != '.' && c != '_' && c != '-'),
        "a group header must open with something no run id can: {GROUP_HEADER:?}"
    );

    // Rows over the checked-in golden document, so a row here is a document this
    // build reads rather than one assembled by hand.
    let golden: RunSummary =
        serde_json::from_str(include_str!("golden/run-summary-v7.json")).expect("the golden reads");
    let row = |run: &str, project: &str, name: Option<&str>, at: Option<u64>| RunSummary {
        run_id: run.into(),
        project: project.into(),
        name: name.map(str::to_owned),
        last_write_at: at,
        ..golden.clone()
    };
    let listing = Listing {
        root: PathBuf::from("/runs"),
        summaries: vec![
            row("newer-b", "plans:b", Some("B"), Some(300)),
            row("orphan", "", None, Some(250)),
            row("older-b", "plans:b", None, Some(200)),
            row("only-a", "plans:a", Some("A"), Some(100)),
            row("undated", "plans:c", None, None),
        ],
        skipped: Vec::new(),
    };
    let projects = Projects::of(&listing);
    assert_eq!(projects.root, listing.root);
    /// One group as this test reads it: its project, its name, its recency, and
    /// its runs in order.
    type Shape<'a> = (Option<&'a str>, Option<&'a str>, Option<u64>, Vec<&'a str>);
    let shape: Vec<Shape<'_>> = projects
        .groups
        .iter()
        .map(|group| {
            (
                group.project.as_deref(),
                group.name.as_deref(),
                group.last_write_at,
                group.runs.iter().map(|run| run.run_id.as_str()).collect(),
            )
        })
        .collect();
    assert_eq!(
        shape,
        vec![
            (
                Some("plans:b"),
                Some("B"),
                Some(300),
                vec!["newer-b", "older-b"]
            ),
            (None, None, Some(250), vec!["orphan"]),
            (Some("plans:a"), Some("A"), Some(100), vec!["only-a"]),
            (Some("plans:c"), None, None, vec!["undated"]),
        ]
    );
    assert_eq!(
        projects
            .flat()
            .iter()
            .map(|run| run.run_id.as_str())
            .collect::<Vec<_>>(),
        vec!["newer-b", "orphan", "older-b", "only-a", "undated"],
        "the flat list is the listing's own order"
    );
    let orphans: &ProjectGroup = &projects.groups[1];
    assert_eq!(orphans.header(), format!("{GROUP_HEADER}{NO_PROJECT}\n"));
    assert_eq!(
        projects.groups[0].header(),
        format!("{GROUP_HEADER}plans:b — B\n")
    );
}

/// The paragraph that says this crate owns the `planner-channel` layout names
/// what the layout at `onepipeline::channel::layout` declares — its queues, the
/// asker variable, the ids it registers and the schema documents behind the
/// reply envelope — and the three rulings it states are what the layout does.
#[test]
fn the_planner_channel_layout_is_this_crates_and_is_what_the_contract_states() {
    use onemessagebus::{Layout as _, LocalTransport, Transport};
    use onepipeline::channel::layout::{
        self, source, Channel, PlannerChannel, Surface, ASKER_ENV, COMMANDS, PLANNER_CHANNEL,
        REPLIES, REPLY_ENVELOPE_FAMILY, REPLY_ENVELOPE_VERSION, REPLY_ENVELOPE_VERSIONS_READ,
    };

    let passage = CONTRACT
        .lines()
        .find(|line| line.starts_with("**The planner channel runs on `onemessagebus`"))
        .expect("the contract states who owns the planner-channel layout");
    let named = backticked_in(passage);
    assert!(named.contains("onepipeline::channel::layout"));
    assert_eq!(PlannerChannel.name(), PLANNER_CHANNEL);
    assert!(named.contains(PLANNER_CHANNEL) && named.contains(ASKER_ENV));
    for file in layout::FILES {
        assert!(
            named.contains(file),
            "the contract does not name the layout's file `{file}`"
        );
    }
    for spec in layout::queues() {
        assert!(
            named.contains(&spec.name.to_string()),
            "the contract does not name the layout's queue `{}`",
            spec.name
        );
    }

    // The ids it registers, and the reply envelope's documents.
    let registry = layout::registry();
    for id in [
        layout::SURFACE_SCHEMA,
        layout::QUEUED_REPLY_SCHEMA,
        layout::QUEUED_COMMANDS_SCHEMA,
        layout::COMMAND_OUTCOME_SCHEMA,
    ] {
        assert!(registry.schema(&id).is_some(), "{id} is not registered");
        assert!(
            named.contains(&id.to_string()),
            "the contract does not name {id}"
        );
    }
    assert!(named.contains(REPLY_ENVELOPE_FAMILY));
    for version in REPLY_ENVELOPE_VERSIONS_READ {
        let id = onemessagebus::SchemaId::literal("agent", "reply-envelope", *version);
        assert!(registry.schema(&id).is_some(), "{id} is not registered");
        let document = format!("schemas/reply-envelope-v{version}.schema.json");
        assert!(
            named.contains(&document),
            "the contract does not name {document}"
        );
        assert!(
            repo_root().join(&document).is_file(),
            "{document} does not ship"
        );
    }

    // An edit envelope requires version 3, refused in the contract's words.
    let replies: onemessagebus::QueueName = REPLIES.parse().expect("a queue name");
    let grants = PlannerChannel.allowlist();
    let refused = PlannerChannel
        .prepare(
            &replies,
            json!({"version": 1, "commands": [{"op": "finding", "message": "m"}]}),
            &grants,
        )
        .expect_err("an edit envelope at version 1");
    assert!(
        named.contains(&refused),
        "the contract does not quote `{refused}`"
    );
    assert_eq!(REPLY_ENVELOPE_VERSION, 3);

    // A bare reply is routed by its halves.
    let routed = |envelope: Value| -> Vec<String> {
        PlannerChannel
            .prepare(&replies, envelope, &grants)
            .expect("the planner's reply is routed")
            .into_iter()
            .map(|(queue, _)| queue.to_string())
            .collect()
    };
    let edit = json!([{"op": "finding", "message": "m"}]);
    assert_eq!(
        routed(json!({"version": 3, "message": "go on", "completion": false, "commands": edit})),
        vec![COMMANDS, REPLIES]
    );
    assert_eq!(
        routed(json!({"version": 3, "commands": edit})),
        vec![COMMANDS]
    );
    assert_eq!(routed(json!({"message": "carry on"})), vec![REPLIES]);

    // A waiting surface is superseded on `source == check-in`, not on `kind`.
    assert!(named.contains("source == check-in"));
    let dir = std::env::temp_dir().join(format!(
        "onepipeline-contract-layout-{}",
        std::process::id()
    ));
    let _ = std::fs::remove_dir_all(&dir);
    let transport: std::sync::Arc<dyn Transport> =
        std::sync::Arc::new(LocalTransport::open(&dir).expect("the transport opens"));
    let channel = Channel::open(&transport).expect("the channel opens");
    let surface = |message: &str, from: &str| Surface {
        id: 0,
        kind: "check-in".to_owned(),
        message: message.to_owned(),
        source: from.to_owned(),
        blocking: false,
        queued_at: 1,
        workstream: None,
        abandoned: false,
        asker: None,
        correlation: None,
    };
    for (message, from) in [
        ("an observer's first", source::PROPOSAL),
        ("an observer's second", source::PROPOSAL),
        ("a pacemaker's first", source::CHECK_IN),
        ("a pacemaker's second", source::CHECK_IN),
    ] {
        channel.push(&surface(message, from)).expect("queued");
    }
    let mut handed = Vec::new();
    while let Some(next) = channel.claim().expect("a claim") {
        handed.push(next.message);
    }
    let _ = std::fs::remove_dir_all(&dir);
    assert_eq!(
        handed,
        vec![
            "an observer's first",
            "an observer's second",
            "a pacemaker's second"
        ]
    );
}

/// The contract's best-effort reading of an older record's bus configuration
/// names exactly the codec fields the linked bus requires, and quotes the
/// refusals that bus's `serve` gives a codec whose fields were read empty.
#[test]
fn an_older_records_bus_config_is_read_best_effort_as_the_contract_states() {
    let passage = CONTRACT
        .split("**A launch record's bus configuration is read best-effort.**")
        .nth(1)
        .and_then(|rest| rest.split("**").next())
        .expect("the contract states how an older record's bus configuration is read");
    let named = backticked_in(passage);

    // The fields it names are the ones the linked bus's codec requires.
    let schema = schemars::schema_for!(onemessagebus::CodecConfig).to_value();
    let required: BTreeSet<String> = schema["required"]
        .as_array()
        .expect("the codec schema requires fields")
        .iter()
        .map(|field| field.as_str().expect("a field name").to_owned())
        .collect();
    assert_eq!(
        required,
        BTreeSet::from(["select".to_owned(), "frames".to_owned()])
    );
    for field in &required {
        assert!(
            named.contains(field),
            "the contract does not name `{field}`"
        );
    }

    // A real older record's codec, with those fields read empty, is refused by
    // the bus's own serve resolution in the contract's words, one field at a time.
    let older: Value = serde_json::from_str(
        &std::fs::read_to_string(
            repo_root().join("tests/recorded/launch/otg-closed-state-writes-status.json"),
        )
        .expect("the older record ships"),
    )
    .expect("the older record is JSON");
    let (name, codec) = older["bus_config"]["codecs"]
        .as_object()
        .and_then(|codecs| codecs.iter().next())
        .expect("the older record names a codec");
    let mut emptied = codec.clone();
    emptied["select"] = json!("");
    emptied["frames"] = json!({});
    let refusal = |codec: Value| {
        onemessagebus::ConfiguredCodec::new(
            name.parse().expect("a codec name"),
            serde_json::from_value(codec).expect("a codec configuration"),
        )
        .expect_err("a codec with an empty field")
        .replace(&format!("codecs.{name}."), "codecs.<name>.")
    };
    assert!(named.contains(&refusal(emptied.clone())));
    emptied["select"] = json!("op");
    assert!(named.contains(&refusal(emptied)));
}