//! Queue-core unit tests: submit / no-dedup, cycle rejection, resource
//! serialization (build slots / per-agent leases), lease-exempt
//! overlap, FIFO fairness, cancel semantics, `AfterAny` failure
//! routing, in-DAG subgraph growth, and history retention. All
//! synchronous — the
//! scheduler's async loop is a thin claim/complete pump over the same
//! methods exercised here.

use super::model::NodeKind;
use super::*;

fn submit<F: FnOnce(&Job)>(q: &JobQueue, spec: DagSpec<F>) -> u64 {
    q.submit(spec).expect("valid spec")
}

fn ident(s: &str) -> hive_types::Ident {
    hive_types::Ident::parse(s).expect("valid test ident")
}

fn rebuild(agent: &str, reason: &str) -> DagSpec<impl FnOnce(&Job) + use<>> {
    templates::rebuild(agent, Source::Manual, reason.to_owned(), true)
}

/// Restart DAG spec with every agent treated as **running** — the online
/// shape (`[Signal→Drain→] StopForUpdate → Reconcile`, no `SetWanted` head)
/// most queue-mechanics tests assume. Mirrors the pre-dynamic
/// `templates::restart` (which is now the state-aware `submit::restart_spec`).
fn restart_online(
    agents: &[&str],
    graceful: bool,
    reason: &str,
) -> DagSpec<impl FnOnce(&Job) + use<>> {
    let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect();
    submit::restart_spec(&targets, graceful, Source::Manual, reason.to_owned())
}

/// Stop DAG spec with every agent treated as **running** — the online shape
/// (`SetWanted → [Signal→Drain→](graceful) Reconcile`).
fn stop_online(
    agents: &[&str],
    graceful: bool,
    reason: &str,
) -> DagSpec<impl FnOnce(&Job) + use<>> {
    let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect();
    submit::stop_spec(&targets, graceful, Source::Manual, reason.to_owned())
}

/// What a test observes about a node the settle loop just started: its id, its
/// DAG, and its payload.
///
/// **Test-only, and deliberately not a production type.** `exec::run_node`
/// takes `(NodeId, &NodeKind)` and derives the DAG id on the two arms that
/// actually want it — nothing in production needs a claim snapshot to exist.
/// The assertions here are about *which* node the graph let run, which does
/// need the payload next to the id.
#[derive(Debug, Clone)]
struct Claimed {
    dag_id: u64,
    node_id: NodeId,
    kind: NodeKind,
    agent: String,
}

/// Drive one settle wave and report every node that started.
///
/// An **extension trait rather than a method on [`JobQueue`]**: production
/// claims one node at a time ([`hive_jobq::scheduler::Scheduler::claim_next`])
/// and has no use for a whole wave, so this must not be reachable from
/// non-test code. `settle()` is that same claim primitive in a loop, so a test
/// driving it here exercises the production path.
trait ClaimReady {
    fn claim_ready(&self) -> Vec<Claimed>;
}

impl ClaimReady for JobQueue {
    fn claim_ready(&self) -> Vec<Claimed> {
        let mut sched = self.sched().lock().expect("job_queue mutex poisoned");
        let started = sched.settle();
        started
            .into_iter()
            .filter_map(|node_id| {
                let kind = sched.graph().node(node_id)?.payload.clone();
                Some(Claimed {
                    dag_id: sched.graph().root_of(node_id)?.get(),
                    node_id,
                    agent: kind.agent().to_owned(),
                    kind,
                })
            })
            .collect()
    }
}

/// Drive a node terminal by hand.
///
/// Also an extension trait, for the same reason as [`ClaimReady`]: production
/// completes a node **inside** the future
/// [`hive_jobq::scheduler::Scheduler::claim_next`] hands back, so "run the node,
/// then remember to complete it" is not an expressible sequence there — which
/// was the whole point of the seam. These tests need to express it, because
/// they exercise the graph without running any executor.
trait CompleteNode {
    fn complete_node(&self, node_id: NodeId, result: Result<(), String>);
    fn complete_node_growing(&self, node_id: NodeId, result: Result<(), String>, grown: Job);
    fn new_job(&self) -> Job;
}

impl CompleteNode for JobQueue {
    /// Mint a builder to declare growth into.
    ///
    /// Test-only for the same reason as the rest of this trait: production
    /// never mints one, because `claim_next` hands each running node its
    /// builder and takes it back. That leaves `Scheduler::new_job` with no
    /// non-test caller either — see the note on that fn.
    fn new_job(&self) -> Job {
        self.sched()
            .lock()
            .expect("job_queue mutex poisoned")
            .new_job()
    }

    fn complete_node(&self, node_id: NodeId, result: Result<(), String>) {
        self.sched()
            .lock()
            .expect("job_queue mutex poisoned")
            .complete(node_id, outcome_of(result));
        self.notify.notify_one();
    }

    fn complete_node_growing(&self, node_id: NodeId, result: Result<(), String>, grown: Job) {
        // A rejected grown job is logged, not propagated: the node's own work
        // already ran, and refusing to complete it here would both misreport
        // that and wedge the DAG on a node stuck `Running`.
        if let Err(e) = self
            .sched()
            .lock()
            .expect("job_queue mutex poisoned")
            .complete_growing(node_id, outcome_of(result), grown)
        {
            tracing::error!(
                node = node_id.get(),
                error = %e,
                "job_queue: work grown by a completing node was rejected"
            );
        }
        self.notify.notify_one();
    }
}

/// One node's **declared** shape: what it is, what it hangs under, and what it
/// waits for — all by kind, since ids are not stable across runs.
#[derive(Debug, PartialEq, Eq)]
struct Declared {
    kind: &'static str,
    /// Parent kind, or `None` when the node hangs directly under the DAG
    /// container (i.e. it is a group root).
    parent: Option<&'static str>,
    /// Kinds this node declared a node-dep on, in declaration order, each with
    /// the outcome set that satisfies it.
    ///
    /// The outcome set is **not** decoration: a template emits its tails as a
    /// pair edged on the same upstream nodes, and the *only* thing telling the
    /// ok-tail from the fail-tail is which outcomes each accepts. Without it
    /// two structurally different nodes read as identical.
    after: Vec<(&'static str, String)>,
}

/// Render a dep's outcome set as the outcomes it actually accepts.
///
/// ⚠️ Spelled out rather than bucketed into `ok` / `any` / other. The first
/// version of this did bucket, and a template's three `ResolveApproval` tails —
/// which differ ONLY in their accepted outcomes — all rendered as `"other"`.
/// A helper that prints two structurally different nodes identically turns an
/// assertion into a tautology.
fn when_tag(when: hive_jobq::DepWhen) -> String {
    [
        (TerminalState::Done, "done"),
        (TerminalState::Failed, "failed"),
        (TerminalState::Cancelled, "cancelled"),
        (TerminalState::Skipped, "skipped"),
    ]
    .into_iter()
    .filter(|(outcome, _)| when.accepts(*outcome))
    .map(|(_, name)| name)
    .collect::<Vec<_>>()
    .join("|")
}

/// Every work node under `dag`, in insertion order, as its declared shape.
///
/// **This is what the template tests are actually about.** A template's output
/// is fully determined the moment `submit` returns: the kinds, the parent
/// nesting and the dep edges are all sitting in the graph. Reading them here
/// keeps the assertion on c0re's own product. Whether the scheduler then
/// *honours* those edges — runs a chain serially, holds a grant across a
/// subtree — is `hive_jobq`'s property and is tested in `hive_jobq`.
fn declared_shape(q: &JobQueue, dag: u64) -> Vec<Declared> {
    let sched = q.sched().lock().expect("job_queue mutex poisoned");
    let graph = sched.graph();
    let root = graph.resolve_id(dag).expect("dag id is a real node id");
    let kind_of = |id: NodeId| graph.node(id).map(|n| n.payload.as_str());
    graph
        .nodes()
        .filter(|n| n.id != root && graph.root_of(n.id) == Some(root))
        .map(|n| Declared {
            kind: n.payload.as_str(),
            parent: n.parent.filter(|p| *p != root).and_then(kind_of),
            after: n
                .deps
                .iter()
                .filter_map(|dep| match dep {
                    hive_jobq::Dep::Node { id, when } => kind_of(*id).map(|k| (k, when_tag(*when))),
                    hive_jobq::Dep::Resource { .. } => None,
                })
                .collect(),
        })
        .collect()
}

/// The id of the one node of `kind` under `dag`, for the resource assertions.
///
/// Panics unless there is exactly one — every caller is about a shape where the
/// kind is unique, so two would mean the assertion had quietly stopped being
/// about the node the test names.
fn node_of(q: &JobQueue, dag: u64, kind: &str) -> hive_jobq::NodeId {
    let sched = q.sched().lock().expect("job_queue mutex poisoned");
    let graph = sched.graph();
    let root = graph.resolve_id(dag).expect("dag id is a real node id");
    let mut found: Vec<_> = graph
        .nodes()
        .filter(|n| n.id != root && graph.root_of(n.id) == Some(root) && n.payload.as_str() == kind)
        .map(|n| n.id)
        .collect();
    assert_eq!(
        found.len(),
        1,
        "expected exactly one {kind} node in the dag"
    );
    found.pop().expect("checked above")
}

/// The payload of the one node of `kind` under `dag`, for assertions about what
/// a node *carries* rather than how it is wired.
fn payload_of(q: &JobQueue, dag: u64, kind: &str) -> NodeKind {
    let id = node_of(q, dag, kind);
    let sched = q.sched().lock().expect("job_queue mutex poisoned");
    sched.graph().node(id).expect("node exists").payload.clone()
}

/// Kinds of every node under `dag` still `Pending` — the nodes that could yet
/// run. Stronger than asking the scheduler what is *ready right now*: a node
/// blocked on a dep is not ready but is very much still alive.
fn pending_kinds(q: &JobQueue, dag: u64) -> Vec<&'static str> {
    let sched = q.sched().lock().expect("job_queue mutex poisoned");
    let graph = sched.graph();
    let root = graph.resolve_id(dag).expect("dag id is a real node id");
    graph
        .nodes()
        .filter(|n| n.id != root && graph.root_of(n.id) == Some(root) && n.state == State::Pending)
        .map(|n| n.payload.as_str())
        .collect()
}

/// Shorthand for one expected row, so the tables below read as a shape.
fn row(
    kind: &'static str,
    parent: Option<&'static str>,
    after: &[(&'static str, &str)],
) -> Declared {
    Declared {
        kind,
        parent,
        after: after.iter().map(|(k, w)| (*k, (*w).to_owned())).collect(),
    }
}

/// Claim helper asserting exactly one node comes back.
fn claim_one(q: &JobQueue) -> Claimed {
    let mut claims = q.claim_ready();
    assert_eq!(
        claims.len(),
        1,
        "expected exactly one claim, got {claims:?}"
    );
    claims.pop().expect("one claim")
}

/// Claim an approval DAG's `ResolveApproval` tail and complete it, asserting it
/// is the one built for `expect`.
///
/// A template emits one tail per outcome and the graph runs exactly one, so the
/// assertion is on *which node was claimed* — that alone says what the approval
/// row is about to be resolved as. Nothing computes it.
fn settle_approval_tail(q: &JobQueue, approval_id: i64, expect: TerminalState) {
    let tail = claim_one(q);
    assert!(
        matches!(
            tail.kind,
            NodeKind::ResolveApproval { approval_id: got, outcome }
                if got == approval_id && outcome == expect
        ),
        "expected the {expect:?} ResolveApproval tail for #{approval_id}, got {:?}",
        tail.kind
    );
    q.complete_node(tail.node_id, Ok(()));
}

/// The `EmitRebuilt` counterpart of [`settle_approval_tail`] — claim the tail the
/// graph let run and assert it's the `ok` one expected.
fn settle_rebuild_tail(q: &JobQueue, agent: &str, expect_ok: bool) {
    let tail = claim_one(q);
    assert!(
        matches!(&tail.kind, NodeKind::EmitRebuilt { agent: a, ok } if a == agent && *ok == expect_ok),
        "expected the ok={expect_ok} EmitRebuilt tail for {agent}, got {:?}",
        tail.kind
    );
    q.complete_node(tail.node_id, Ok(()));
}

/// The resources a node **declared**, read off its graph edges.
///
/// The declaration is the thing under test now that construction sites state
/// their own holdings: asking the `NodeKind` what it "should" need would just
/// re-run the derivation this module removed, and would pass even if the
/// construction site declared nothing.
fn declared_resources(q: &JobQueue, node_id: hive_jobq::NodeId) -> Vec<Resource> {
    let inner = q.lock();
    inner
        .graph()
        .node(node_id)
        .expect("node exists")
        .deps
        .iter()
        .filter_map(|dep| match dep {
            hive_jobq::Dep::Resource { name, .. } => Some(name.clone()),
            hive_jobq::Dep::Node { .. } => None,
        })
        .collect()
}

fn state_of(q: &JobQueue, dag_id: u64) -> State {
    // A DAG whose nodes have all settled `Done` or `Skipped` drops out of the
    // snapshot — absence is the completion signal, so map it to `Done`.
    // Otherwise derive the roll-up from the node set, exactly as every wire
    // consumer does.
    q.snapshot()
        .iter()
        .find(|d| d.id == dag_id)
        .map_or(State::Done, DagView::rollup_state)
}

// ---- submit (dedup removed — every submit is a fresh DAG) ----

#[test]
fn submit_assigns_distinct_ids() {
    let q = JobQueue::new(1);
    let a = submit(&q, rebuild("agent-a", "first"));
    let b = submit(&q, rebuild("agent-b", "second"));
    assert_ne!(a, b);
    assert_eq!(q.snapshot().len(), 2);
}

/// Submit-time dedup was removed with the agent-per-node refactor (a
/// multi-agent DAG has no single agent to key a dedup on), so an identical
/// resubmit — same template + agent, still queued — now enqueues a distinct
/// DAG instead of collapsing into the pending one. Whether any dedup needs
/// reintroducing is tracked as a follow-up.
#[test]
fn identical_resubmit_is_a_distinct_dag() {
    let q = JobQueue::new(1);
    let a = submit(&q, rebuild("agent-a", "first"));
    let b = submit(&q, rebuild("agent-a", "again"));
    assert_ne!(a, b, "no dedup: identical resubmit is a new DAG");
    assert_eq!(q.snapshot().len(), 2);
}

#[test]
fn distinct_submits_never_collapse() {
    let q = JobQueue::new(1);
    let a = submit(&q, rebuild("agent-a", "r"));
    let b = submit(&q, rebuild("agent-b", "r"));
    let c = submit(&q, restart_online(&["agent-a"], false, "r"));
    assert_ne!(a, b);
    assert_ne!(a, c);
    assert_eq!(q.snapshot().len(), 3);
}

#[test]
fn resubmit_while_running_is_new_dag() {
    // The "while running" is not load-bearing and used to be staged by claiming
    // a node first. `submit` appends a container and inserts the declared
    // group; it never consults the state of any existing node, so whether an
    // earlier DAG is running cannot change the outcome. What is actually being
    // asserted — no dedup, ever — is `identical_resubmit_is_a_distinct_dag`.
    //
    // Kept as the *named* case because "a config bump mid-build must not be
    // swallowed" is the scenario people worry about, and a reader looking for
    // it should find it.
    let q = JobQueue::new(1);
    let a = submit(&q, rebuild("agent-a", "first"));
    let again = submit(&q, rebuild("agent-a", "config bumped during build"));
    assert_ne!(a, again);
    assert_eq!(q.snapshot().len(), 2);
}

// ---- malformed specs: no longer expressible ----
//
// Three tests lived here — a dependency cycle, a dependency on a node that
// does not exist, and an out-of-range parent index — each asserting that
// `submit` refused the spec. All three built their spec by hand out of
// positional indices, which is exactly the representation that made those
// shapes possible: an index can name a node that isn't there, or one that
// comes later.
//
// A job is now declared against handles that only exist for nodes already
// declared, so there is no index to put out of range, and every edge points
// backwards — a cycle needs a forward edge. The guard those tests covered was
// deleted along with the failure mode. What remains — a handle used against a
// builder that never issued it — is `hive_jobq`'s to reject, and its builder
// tests cover it (`a_forward_edge_is_rejected_by_name`,
// `a_forward_parent_is_rejected_by_name`, `graph_rejection_surfaces_as_is`).

// ---- dependency order within a DAG ----

#[test]
fn rebuild_chain_is_declared_serial() {
    // Was `rebuild_chain_claims_in_dep_order`, which drove the whole DAG to
    // observe an order that is fully declared the moment `submit` returns.
    //
    // ⚠️ The old name was also wrong about the mechanism, and reading it rather
    // than the graph is how you'd stay wrong: **only half this chain is dep
    // edges.** `stop_for_update` and `swap` declare no deps at all — they are
    // ordered by *parent nesting* ("a node's sub-nodes run after its own
    // logic"). Both axes are asserted below because a template can break either
    // one independently.
    let q = JobQueue::new(1);
    let id = submit(&q, rebuild("agent-a", "r"));
    assert_eq!(
        declared_shape(&q, id),
        vec![
            row("meta_sync", None, &[]),
            row("prebuild", None, &[("meta_sync", "done")]),
            // No dep: ordered by hanging under `prebuild`.
            row("stop_for_update", Some("prebuild"), &[]),
            row("swap", Some("stop_for_update"), &[]),
            row("post_swap", Some("stop_for_update"), &[("swap", "done")]),
            row("reconcile", None, &[("prebuild", "done|failed|skipped")]),
            // The tail pair. The ok tail needs every root to succeed; the !ok
            // tail hangs off the ok tail's *elimination* (`skipped`), which is
            // what makes exactly one of them run.
            row(
                "emit_rebuilt",
                None,
                &[
                    ("meta_sync", "done"),
                    ("prebuild", "done"),
                    ("reconcile", "done"),
                ],
            ),
            row(
                "emit_rebuilt",
                None,
                &[
                    ("emit_rebuilt", "skipped"),
                    ("meta_sync", "done|failed|skipped"),
                    ("prebuild", "done|failed|skipped"),
                    ("reconcile", "done|failed|skipped"),
                ],
            ),
        ]
    );
}

/// The boot sweep's graceful shape: the agent gets `Signal` → `Drain` to
/// finish its turn before `StopForUpdate` takes the container down. `Signal`
/// *parents* the rest of the stop rather than sitting beside it, so the agent
/// lease is held continuously across the whole bounce — as siblings, each of
/// `Signal` / `Drain` / `StopForUpdate` would acquire the lease separately and
/// leave a window for another DAG to claim the agent mid-stop.
#[test]
fn graceful_rebuild_chain_drains_before_stopping() {
    let q = JobQueue::new(1);
    let id = submit(
        &q,
        DagSpec {
            source: Source::AutoUpdate,
            reason: "sweep".to_owned(),
            declare: Box::new(|b: &Job| {
                templates::rebuild_nodes(
                    b,
                    "agent-a",
                    templates::RebuildOpts {
                        relock: true,
                        graceful: true,
                    },
                    None,
                );
            }),
        },
    );
    assert_eq!(
        declared_shape(&q, id)
            .iter()
            .map(|d| d.kind)
            .collect::<Vec<_>>(),
        vec![
            "meta_sync",
            "prebuild",
            // The graceful window goes between the build and the stop: the
            // agent gets its turn to finish before the container goes down.
            "signal",
            "drain",
            "stop_for_update",
            "swap",
            "post_swap",
            "reconcile",
        ],
        "graceful inserts signal + drain ahead of the stop, and nothing else"
    );
}

/// The non-graceful shape is the default everywhere except the boot sweep:
/// a manual rebuild, a meta-update cascade child and a deploy must NOT spend a
/// drain window, so `StopForUpdate` still hangs straight off `Prebuild`.
#[test]
fn non_graceful_rebuild_has_no_signal_or_drain() {
    // Read the shape off the queue rather than out of a node list: a declared
    // job keeps its nodes to itself and inserts them, so what it built is
    // observable where it matters — in what the scheduler runs.
    let q = JobQueue::new(1);
    let id = submit(
        &q,
        DagSpec {
            source: Source::Manual,
            reason: "manual".to_owned(),
            declare: Box::new(|b: &Job| {
                templates::rebuild_nodes(
                    b,
                    "agent-a",
                    templates::RebuildOpts {
                        relock: true,
                        graceful: false,
                    },
                    None,
                );
            }),
        },
    );
    assert_eq!(
        declared_shape(&q, id)
            .iter()
            .map(|d| d.kind)
            .collect::<Vec<_>>(),
        vec![
            "meta_sync",
            "prebuild",
            "stop_for_update",
            "swap",
            "post_swap",
            "reconcile"
        ],
        "exactly six nodes, and neither of them is signal or drain"
    );
}

/// A cleanly-finished DAG leaves the snapshot even though its not-taken
/// failure branch is still in the graph as `Skipped`. Skipped nodes ride the
/// wire so the dashboard can mark them, which makes "the node list is empty"
/// and "nothing here is still worth showing" two different questions — only
/// the second one may drop the DAG. Conflating them pins every completed
/// deploy in the queue view forever.
#[test]
fn settled_dag_leaves_the_snapshot_despite_its_skipped_branch() {
    let q = JobQueue::new(1);
    let id = submit(&q, rebuild("agent-a", "r"));
    for _ in 0..6 {
        let c = claim_one(&q);
        q.complete_node(c.node_id, Ok(()));
    }
    settle_rebuild_tail(&q, "agent-a", true);
    assert!(
        q.snapshot().iter().all(|d| d.id != id),
        "a fully settled DAG drops out of the snapshot"
    );
}

// ---- build slots ----

#[test]
fn rebuild_chain_declares_the_slot_where_the_nix_work_is() {
    // Was `fifo_fairness_for_the_slot`, which submitted three rebuilds and
    // drove one to completion to watch the freed slot go to the earlier
    // waiter. **That fairness guarantee is hive_jobq's**, and it had no test
    // there at all — `claim_one` scans nodes in insertion order and takes the
    // first satisfiable one, and nothing pinned that. It does now:
    // `a_contended_resource_goes_to_the_oldest_waiter`.
    //
    // What is c0re's is *which* nodes contend for the slot in the first place,
    // and that is a declaration. "Uniform hold across the chain" then follows
    // from the parent nesting asserted in `rebuild_chain_is_declared_serial`:
    // a resource unit is held for the acquirer's whole subtree, so the slot
    // `Prebuild` takes covers `StopForUpdate` → `Swap` → `PostSwap` beneath it.
    let q = JobQueue::new(1);
    let id = submit(&q, rebuild("agent-a", "r"));
    let res = |kind: &str| declared_resources(&q, node_of(&q, id, kind));

    let agent = || Resource::Agent("agent-a".to_owned());
    assert_eq!(
        [
            res("meta_sync"),
            res("prebuild"),
            res("stop_for_update"),
            res("swap"),
            res("reconcile"),
        ],
        [
            // The meta preamble takes the global window and *nothing else* —
            // no slot (it does no nix work) and no lease.
            vec![Resource::MetaWindow],
            // The nix build is the slot-needer, and takes **no lease**. That is
            // what lets a prebuild overlap another DAG on the same agent: the
            // container is still up and untouched while it builds.
            vec![Resource::BuildSlot],
            // The lease starts here — the first node that touches the
            // container — and not one node earlier.
            vec![agent()],
            // Swap needs both. It re-enters the slot its `Prebuild` ancestor
            // holds rather than acquiring a second unit.
            vec![Resource::BuildSlot, agent()],
            vec![agent()],
        ],
        "the slot follows the nix work and the lease follows the container"
    );
}

// ---- per-agent lease ----

#[test]
fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() {
    let q = JobQueue::new(4);
    let id = submit(
        &q,
        restart_online(&["agent-a", "agent-b"], false, "hive-wide"),
    );
    // A hive-wide restart is ONE DAG, not one-per-agent.
    assert_eq!(q.snapshot().len(), 1);
    // Each agent's subgraph head (StopForUpdate, since both are running) is a
    // group root with no deps, so nothing orders them against each other; and
    // each declares only its OWN agent's lease, so nothing makes them contend.
    // Those two declared facts are what "they run concurrently" *means* here —
    // that a scheduler then does run independent, resource-disjoint roots at
    // once is hive_jobq's property, tested there.
    let heads: Vec<_> = declared_shape(&q, id)
        .into_iter()
        .filter(|d| d.kind == "stop_for_update")
        .collect();
    assert_eq!(
        heads,
        vec![
            row("stop_for_update", None, &[]),
            row("stop_for_update", None, &[]),
        ],
        "both per-agent heads are independent group roots"
    );
    let sched = q.sched().lock().expect("job_queue mutex poisoned");
    let graph = sched.graph();
    let root = graph.resolve_id(id).expect("dag id");
    let mut leases: Vec<String> = graph
        .nodes()
        .filter(|n| graph.root_of(n.id) == Some(root) && n.payload.as_str() == "stop_for_update")
        .flat_map(|n| {
            n.deps.iter().filter_map(|dep| match dep {
                hive_jobq::Dep::Resource { name, .. } => Some(format!("{name:?}")),
                hive_jobq::Dep::Node { .. } => None,
            })
        })
        .collect();
    leases.sort();
    assert_eq!(
        leases,
        vec![
            format!("{:?}", Resource::Agent("agent-a".to_owned())),
            format!("{:?}", Resource::Agent("agent-b".to_owned())),
        ],
        "each head declares only its own agent's lease — disjoint, so no contention"
    );
}

#[test]
fn multi_agent_stop_is_one_dag_with_concurrent_per_agent_subgraphs() {
    let q = JobQueue::new(4);
    let id = submit(
        &q,
        stop_online(&["agent-a", "agent-b"], false, "hive-wide stop"),
    );
    // A hive-wide stop is ONE DAG, not one-per-agent.
    assert_eq!(q.snapshot().len(), 1);
    let claims = q.claim_ready();
    assert!(claims.iter().all(|c| c.dag_id == id));
    let mut heads: Vec<(&str, &str)> = claims
        .iter()
        .map(|c| (c.agent.as_str(), c.kind.as_str()))
        .collect();
    heads.sort_unstable();
    assert_eq!(
        heads,
        vec![("agent-a", "set_wanted"), ("agent-b", "set_wanted")],
        "both per-agent stop subgraphs start concurrently, each on its own lease"
    );
}

#[test]
fn multi_agent_start_one_dag_folds_per_agent_stale_rebuild() {
    let q = JobQueue::new(4);
    let id = submit(
        &q,
        // fresh: offline + not stale → SetWanted → Reconcile.
        // stale: offline + stale → SetWanted → «rebuild subgraph».
        submit::start_spec(
            &[
                ("fresh".to_owned(), false, false),
                ("stale".to_owned(), false, true),
            ],
            Source::Manual,
            "hive-wide start".to_owned(),
        ),
    );
    // One DAG spanning both agents.
    assert_eq!(q.snapshot().len(), 1);
    // Both subgraph heads (SetWanted(Up)) are roots — claimable at once,
    // each acquiring its own agent lease.
    let heads = q.claim_ready();
    assert!(
        heads
            .iter()
            .all(|c| c.dag_id == id && c.kind.as_str() == "set_wanted")
    );
    let mut head_agents: Vec<&str> = heads.iter().map(|c| c.agent.as_str()).collect();
    head_agents.sort_unstable();
    assert_eq!(head_agents, vec!["fresh", "stale"]);
    // Complete both heads; the fresh agent then reconciles directly while
    // the stale agent's subgraph is the rebuild chain (meta_sync first).
    for c in &heads {
        q.complete_node(c.node_id, Ok(()));
    }
    let next = q.claim_ready();
    let mut kinds: Vec<(&str, &str)> = next
        .iter()
        .map(|c| (c.agent.as_str(), c.kind.as_str()))
        .collect();
    kinds.sort_unstable();
    assert_eq!(
        kinds,
        vec![("fresh", "reconcile"), ("stale", "meta_sync")],
        "fresh agent starts directly; stale agent rebuilds first, all in one DAG"
    );
}

#[test]
fn offline_agents_skip_mechanical_nodes_but_keep_reconcile() {
    // The dynamic build skips Signal/Drain/StopForUpdate for a down agent
    // (nothing to quiesce/stop) but ALWAYS keeps the Reconcile tail — the
    // convergence guarantee that catches a race-up between the is_running
    // read and node exec.
    let q = JobQueue::new(4);
    // Offline graceful stop → SetWanted(Off) → Reconcile (no Signal/Drain).
    let stop = submit(
        &q,
        submit::stop_spec(
            &[("down".to_owned(), false)],
            true,
            Source::Manual,
            "stop down".to_owned(),
        ),
    );
    // Offline restart → a lone Reconcile (no SetWanted, no StopForUpdate):
    // nothing to bounce, and restart never rewrites intent, so the tail
    // Reconcile converges the down agent to its existing `wanted`.
    let restart = submit(
        &q,
        submit::restart_spec(
            &[("down2".to_owned(), false)],
            true,
            Source::Manual,
            "restart down".to_owned(),
        ),
    );
    let shape = |id: u64| -> Vec<String> {
        q.snapshot()
            .iter()
            .find(|d| d.id == id)
            .expect("dag")
            .nodes
            .iter()
            .map(|n| n.kind.clone())
            .collect()
    };
    assert_eq!(
        shape(stop),
        vec!["set_wanted".to_owned(), "reconcile".to_owned()],
        "offline graceful stop skips the signal/drain quiesce, keeps Reconcile"
    );
    assert_eq!(
        shape(restart),
        vec!["reconcile".to_owned()],
        "offline restart is a lone Reconcile (no SetWanted head, nothing to stop)"
    );
}

#[test]
fn boot_sweep_nodes_declare_their_own_resources() {
    // Regression, and the reason it needs its own test: `workers::auto_update`
    // is the only place job nodes are constructed *outside* `job_queue/`, so
    // nothing in this module covered it. When resource derivation moved to the
    // construction sites, this path was missed and both kinds silently declared
    // nothing — dropping the agent lease a boot `Reconcile` needs to not race
    // another DAG's container ops, and letting the sweep `MetaLock` land its
    // meta commit inside another node's staged deploy window. Nothing failed to
    // compile; only an exhaustive caller list would have caught it.
    let q = JobQueue::new(4);
    let _id = submit(
        &q,
        DagSpec {
            source: Source::AutoUpdate,
            reason: "boot".to_owned(),
            declare: Box::new(|b: &Job| {
                crate::workers::auto_update::boot_nodes(
                    b,
                    true,
                    vec!["stale-agent".to_owned()],
                    vec!["drifted-agent".to_owned()],
                );
            }),
        },
    );

    // Both are independent roots on disjoint resources, so both start at once.
    let claims = q.claim_ready();
    let by_kind = |kind: &str| {
        claims
            .iter()
            .find(|c| c.kind.as_str() == kind)
            .unwrap_or_else(|| panic!("no {kind} claim in {claims:?}"))
    };

    let mut lock = declared_resources(&q, by_kind("meta_lock").node_id);
    lock.sort_by_key(|r| format!("{r:?}"));
    assert_eq!(
        lock,
        vec![Resource::BuildSlot, Resource::MetaWindow],
        "the sweep MetaLock runs a nix lock bump and commits to meta"
    );

    assert_eq!(
        declared_resources(&q, by_kind("reconcile").node_id),
        vec![Resource::Agent("drifted-agent".to_owned())],
        "a boot Reconcile touches the container, so it holds that agent's lease"
    );
}

#[test]
fn grown_subgraph_roots_on_emitter_and_rebases_local_deps() {
    // The startup-sweep mechanism: a `MetaLock` emitter grows one rebuild
    // subgraph per stale agent into its OWN DAG. Each subgraph is rooted on
    // the emitter and its LOCAL 0-based deps are rebased onto the DAG.
    let q = JobQueue::new(4);
    let spec = DagSpec {
        source: Source::AutoUpdate,
        reason: "sweep".to_owned(),
        declare: Box::new(|b: &Job| {
            let _lock = b.node(NodeKind::MetaLock {
                sweep: true,
                fanout: None,
                inputs: Vec::new(),
            });
        }),
    };
    submit(&q, spec);
    let emitter = claim_one(&q);
    assert_eq!(emitter.kind.as_str(), "meta_lock");
    // Two independent per-agent subgraphs — the REAL production shape the
    // sweep MetaLock grows: root MetaSync → root Prebuild → Signal → Drain →
    // StopForUpdate → Swap → Reconcile, local 0-based deps. `graceful` must
    // match the sweep arm of `run_meta_lock` or this stops tracking production.
    // Both subgraphs go into the emitter's own builder, exactly as
    // `run_meta_lock`'s sweep arm does. Insert-before-complete is no longer the
    // caller's job to remember: it is one call, and the ordering is inside it.
    let grown = q.new_job();
    for agent in ["a", "b"] {
        templates::rebuild_nodes(
            &grown,
            agent,
            templates::RebuildOpts {
                relock: true,
                graceful: true,
            },
            None,
        );
    }
    q.complete_node_growing(emitter.node_id, Ok(()), grown);
    // Still ONE DAG; both subgraph roots become ready once the emitter is
    // Done (rooted on it), each on its own agent lease. Their `MetaSync` heads
    // take turns on the cap-1 global meta window, so drain those first — what
    // must be concurrent is the builds.
    assert_eq!(q.snapshot().len(), 1);
    let mut kinds = drain_meta_syncs(&q);
    kinds.sort_unstable();
    assert_eq!(
        kinds,
        vec![
            ("a".to_owned(), "prebuild".to_owned()),
            ("b".to_owned(), "prebuild".to_owned())
        ],
        "both rebuild subgraphs root on the emitter and run concurrently in one DAG"
    );
}

/// Complete every `MetaSync` head the queue offers (they take turns on the
/// cap-1 global meta window) and return whatever else got claimed alongside
/// them, as `(agent, kind)` pairs left in flight.
fn drain_meta_syncs(q: &JobQueue) -> Vec<(String, String)> {
    let mut rest = Vec::new();
    for _ in 0..3 {
        for c in q.claim_ready() {
            if c.kind.as_str() == "meta_sync" {
                q.complete_node(c.node_id, Ok(()));
            } else {
                rest.push((c.agent.clone(), c.kind.as_str().to_owned()));
            }
        }
    }
    rest
}

/// Crash-watch suppression for a cascade rebuild, which the deleted half of
/// `meta_update_grows_cascade_in_dag` used to assert via `DagSpec::transient`.
///
/// The property is unchanged — a container going down under a rebuild must not
/// read as a crash — but it is no longer a DAG-level declaration: each node
/// answers for itself, so the assertion moves to the nodes a cascade actually
/// runs. Kept as its own test rather than dropped, because it is the *property*
/// that mattered, not the field that used to carry it.
#[test]
fn rebuild_chain_nodes_suppress_crash_watch() {
    for kind in [
        NodeKind::StopForUpdate {
            agent: "a".to_owned(),
        },
        NodeKind::Swap {
            agent: "a".to_owned(),
        },
        NodeKind::Drain {
            agent: "a".to_owned(),
        },
    ] {
        assert!(
            kind.takes_container_down(),
            "{} must suppress crash-watch — a rebuild takes the container down \
             on purpose",
            kind.as_str()
        );
    }
    // The counter-case, and the reason this can't be "any node in a rebuild":
    // the tail brings the container back up, so a container that dies there
    // really did crash.
    assert!(
        !NodeKind::Start {
            agent: "a".to_owned()
        }
        .takes_container_down()
    );
}

#[test]
fn meta_update_grows_cascade_in_dag() {
    // The meta-update `MetaLock` grows one rebuild subgraph per affected
    // agent into its OWN DAG (via the builder it is handed), not child DAGs.
    let spec = templates::meta_update(
        vec!["nixpkgs".to_owned()],
        Source::Manual,
        "bump".to_owned(),
        None,
    );
    let q = JobQueue::new(4);
    submit(&q, spec);
    let meta_lock = claim_one(&q);
    assert_eq!(meta_lock.kind.as_str(), "meta_lock");
    // Simulate the executor growing the cascade in-DAG (`relock = false` — a
    // cascade child must not re-lock and revert the parent's bump). Both
    // agents go into the one builder the node was handed, which is what
    // `run_meta_lock`'s fanout arm does.
    let grown = q.new_job();
    for agent in ["alice", "bob"] {
        templates::rebuild_nodes(
            &grown,
            agent,
            templates::RebuildOpts {
                relock: false,
                graceful: false,
            },
            None,
        );
    }
    q.complete_node_growing(meta_lock.node_id, Ok(()), grown);
    // Still ONE DAG — no child DAGs — and both cascade rebuild subgraphs root
    // on the MetaLock, each on its own agent lease. The per-agent `MetaSync`
    // heads serialize on the global meta window (they commit to the meta repo);
    // the builds behind them do not.
    assert_eq!(q.snapshot().len(), 1);
    let mut kinds = drain_meta_syncs(&q);
    kinds.sort_unstable();
    assert_eq!(
        kinds,
        vec![
            ("alice".to_owned(), "prebuild".to_owned()),
            ("bob".to_owned(), "prebuild".to_owned())
        ],
        "cascade rebuilds grow in the meta-update DAG, concurrent per agent"
    );
}

// ---- failure: cancel-downstream + AfterAny ----

#[test]
fn failed_node_cancels_downstream_but_afterany_reconcile_runs() {
    let q = JobQueue::new(1);
    let id = submit(&q, rebuild("agent-a", "r"));
    let meta_sync = claim_one(&q);
    assert_eq!(meta_sync.kind.as_str(), "meta_sync");
    q.complete_node(meta_sync.node_id, Ok(()));
    let prebuild = claim_one(&q);
    assert_eq!(prebuild.kind.as_str(), "prebuild");
    q.complete_node(prebuild.node_id, Err("nix build exploded".to_owned()));
    // StopForUpdate + Swap are cancelled (AfterOk on a failed chain);
    // the AfterAny Reconcile still runs once Swap is terminal.
    let reconcile = claim_one(&q);
    assert_eq!(reconcile.kind.as_str(), "reconcile");
    q.complete_node(reconcile.node_id, Ok(()));
    let snap = q.snapshot();
    let dag = snap.iter().find(|d| d.id == id).expect("dag");
    assert_eq!(dag.rollup_state(), State::Failed, "roll-up failed");
    let by_kind = |k: &str| {
        dag.nodes
            .iter()
            .find(|n| n.kind == k)
            .expect("node present")
            .state
    };
    assert_eq!(by_kind("prebuild"), State::Failed);
    // `StopForUpdate` / `Swap` / `PostSwap` were *ruled out* by the failed
    // `Prebuild`. They ride the wire as `Skipped` so an operator can see which
    // steps the run never reached, without them reading as failures of their
    // own — the roll-up ignores `Skipped` entirely.
    for ruled_out in ["stop_for_update", "swap", "post_swap"] {
        assert_eq!(
            by_kind(ruled_out),
            State::Skipped,
            "{ruled_out} was ruled out, so it is on the wire as skipped"
        );
    }
    // The AfterAny reconcile ran (claimed + completed Ok above) → it's `Done`,
    // and `Done` nodes are excluded from the wire, so it's absent here.
    assert!(
        dag.nodes.iter().all(|n| n.kind != "reconcile"),
        "the completed (Done) reconcile is filtered off the wire"
    );
    assert_eq!(
        dag.nodes
            .iter()
            .find(|n| n.kind == "prebuild")
            .and_then(|n| n.error.as_deref()),
        Some("nix build exploded")
    );
}

/// The swap-success path: `Swap` ok → the `AfterOk` `PostSwap` (bookkeeping
/// tail) runs, and only then does `Reconcile` fire — serialized behind
/// `PostSwap` (not racing it) because `Reconcile` deps `AfterAny(PostSwap)`.
#[test]
fn rebuild_reconcile_waits_for_the_whole_build_subtree() {
    // Replaces `swap_ok_runs_post_swap_before_reconcile` and
    // `swap_failure_still_runs_reconcile`, which walked the same DAG with the
    // swap succeeding in one and failing in the other.
    //
    // The interesting claim was "Reconcile must wait for PostSwap, not race
    // it" — and it does *not* come from an edge between them. `reconcile` deps
    // `AfterAny(prebuild)`, while `post_swap` sits inside prebuild's subtree
    // (post_swap → stop_for_update → prebuild). A parent is not terminal until
    // its subtree is, so prebuild cannot satisfy that edge while post_swap is
    // outstanding. **The ordering is the parent chain, not a dependency.**
    //
    // Both facts are asserted in `rebuild_chain_is_declared_serial`; this test
    // states the derived property explicitly because the indirection is the
    // easy thing to break — someone flattening the chain would keep every edge
    // and still lose the guarantee.
    let q = JobQueue::new(1);
    let id = submit(&q, rebuild("agent-a", "r"));
    let shape = declared_shape(&q, id);
    let parent_of = |kind: &str| {
        shape
            .iter()
            .find(|d| d.kind == kind)
            .unwrap_or_else(|| panic!("{kind} node"))
            .parent
    };
    assert_eq!(parent_of("post_swap"), Some("stop_for_update"));
    assert_eq!(parent_of("stop_for_update"), Some("prebuild"));
    assert_eq!(
        shape
            .iter()
            .find(|d| d.kind == "reconcile")
            .expect("reconcile node")
            .after,
        vec![("prebuild", "done|failed|skipped".to_owned())],
        "reconcile gates on prebuild's roll-up, which covers the whole build \
         subtree — including post_swap — and runs on failure too"
    );
}

// `swap_failure_still_runs_reconcile` lived here.
//
// It asserted that a failed swap leaves `post_swap` `Skipped` and `swap`
// `Failed`, and that reconcile still runs. All three are hive_jobq's cascade
// (`failed_after_ok_dep_cancels_dependents_but_after_any_still_runs`), and the
// "says so on the wire" half turned out to be nothing: `snapshot` fills
// `NodeView { state: node.state, .. }`, a straight copy of the same `State`
// type, so there is no c0re-side mapping to get wrong.
//
// `failed_reconcile_marks_dag_failed` lived here too — a one-node DAG whose
// node fails, asserting the DAG reads `Failed`. That is `failed_child_rolls_
// parent_up_to_failed` in hive_jobq, restated through a c0re template.

// `multi_agent_lease_frees_per_subgraph_not_whole_dag` and
// `dag_settles_terminal_and_releases_lease_after_work` lived here.
//
// Both drove a DAG to completion to watch an agent lease free up — one when a
// single agent's subgraph settled inside a still-running multi-agent DAG, the
// other when a whole power op finished. Releasing a grant once its owner's
// subtree is terminal is hive_jobq's (`owner_holds_grant_for_its_whole_subtree`,
// `child_borrows_ancestor_grant_released_when_subtree_done`,
// `leaf_owner_goes_done_directly_and_releases`).
//
// The c0re halves are declared and asserted elsewhere: that each agent's
// subgraph is an independent root holding only its own lease is in
// `multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs`, and
// that a power op emits no tail node is in
// `cancelled_power_op_runs_no_compensating_node`, which checks the DAG has no
// pending nodes left at all.

/// `Start` / `Stop` were lease-exempt *as kinds*, which was only safe because
/// every construction site fans them out from inside a lease-holding ancestor.
/// They declare the lease themselves now, and this pins that they do.
///
/// Was `a_fanned_out_start_declares_the_lease_and_re_enters_its_reconciles_
/// grant`, which submitted a `Reconcile`, claimed it, and then **re-declared
/// the fan-out inline** — *"same two calls the scheduler makes"*. That is a
/// copy of production in a test: had `exec.rs` stopped declaring the lease, it
/// would have kept passing. The declaration now lives in `templates::
/// fanned_out_mechanical`, so this calls the real thing.
///
/// The other half of the old test — that a descendant *re-enters* its
/// ancestor's grant rather than taking a second unit of a cap-1 lease — is
/// `hive_jobq`'s, and is tested there by
/// `child_borrows_ancestor_grant_released_when_subtree_done` and
/// `nested_borrowers_never_deadlock`.
#[test]
fn a_fanned_out_mechanical_node_declares_its_agent_lease() {
    let q = JobQueue::new(4);
    let id = submit(
        &q,
        DagSpec {
            source: Source::Manual,
            reason: "fan-out".to_owned(),
            declare: Box::new(|b: &Job| {
                templates::fanned_out_mechanical(
                    b,
                    NodeKind::Start {
                        agent: "agent-a".to_owned(),
                    },
                );
            }),
        },
    );
    assert_eq!(declared_shape(&q, id), vec![row("start", None, &[])]);
    assert_eq!(
        declared_resources(&q, node_of(&q, id, "start")),
        vec![Resource::Agent("agent-a".to_owned())],
        "the fanned-out node carries the lease itself, rather than relying on \
         whoever happened to fan it out"
    );
}

// ---- cancel ----

#[test]
fn cancel_clears_queued_dag() {
    let q = JobQueue::new(1);
    let id = submit(&q, rebuild("agent-a", "r"));
    assert!(q.cancel(id), "fully-queued dag cancels");
    // The operator sees `Cancelled` the moment the cancel returns — the spared
    // tail is still `Pending`, and a DAG must not read `Queued` back to the
    // operator who just cancelled it (the dashboard renders this roll-up from
    // the snapshot `post_rebuild_queue_cancel` emits synchronously).
    assert_eq!(state_of(&q, id), State::Cancelled, "no stale Queued gap");
    // Neither `EmitRebuilt` tail accepts a *dropped* dependency — the ok one is
    // `AFTER_OK`, the failure one keys on elimination — so both are cancelled
    // with the work and **nothing is emitted** for a rebuild that never ran.
    assert!(
        q.claim_ready().is_empty(),
        "a dropped rebuild reports nothing"
    );
    assert_eq!(state_of(&q, id), State::Cancelled);
}

/// `cancel` takes a **node** id, not a DAG id — so an interior node can be
/// dropped without touching the rest of the group.
///
/// This is the capability the DAG-scoped version couldn't express, and the
/// reason it reads naturally: a DAG id *is* its root node's id, so the
/// whole-group cancel every other test does is just this called on a root.
/// Here a hive-wide restart drops **one agent's** subgraph and the other agent
/// still runs.
#[test]
fn cancel_drops_one_agents_branch_leaving_the_rest() {
    let q = JobQueue::new(2);
    let id = submit(&q, restart_online(&["agent-a", "agent-b"], false, "r"));
    // Per-agent subgraphs are independent roots; find agent-a's.
    let snap = q.snapshot();
    let dag = snap.iter().find(|d| d.id == id).expect("dag in snapshot");
    let a_root = dag
        .nodes
        .iter()
        .find(|n| n.agent == "agent-a" && n.parent.is_none())
        .expect("agent-a has a group root");

    assert!(q.cancel(a_root.id), "an interior/group root cancels alone");

    // agent-b's work is untouched and still claimable; agent-a's is not.
    let claims = q.claim_ready();
    assert!(
        !claims.is_empty() && claims.iter().all(|c| c.agent == "agent-b"),
        "only agent-b remains runnable, got {:?}",
        claims.iter().map(|c| c.agent.as_str()).collect::<Vec<_>>()
    );
}

#[test]
fn cancel_refuses_running_dag() {
    let q = JobQueue::new(1);
    let id = submit(&q, rebuild("agent-a", "r"));
    let _ = claim_one(&q);
    assert!(!q.cancel(id));
    assert_eq!(state_of(&q, id), State::Running);
}

/// A cancelled power op must run **no** compensating node — not even one that
/// carries a `SetWanted` head.
///
/// Now structural rather than a property of a hook enum: a power op emits no
/// tail node at all, so once its work nodes cancel there is simply nothing left
/// to claim. `cancel` also refuses unless every work node is still `Pending`
/// (`cancel_refuses_running_dag`), so a `Cancelled` DAG provably never executed
/// a node: its `SetWanted` never ran and the agent's intent still reads whatever
/// the operator last set. A "revert" instead writes the agent's *observed*
/// state, which for a down-but-`wanted = Up` agent (crashed, or caught
/// mid-bounce) flips the intent to `Offline` and leaves it
/// deliberately-stopped as far as reconcile and crash-watch are concerned.
#[test]
fn cancelled_power_op_runs_no_compensating_node() {
    /// Submit-cancel-assert for one power op. Taking the already-submitted DAG
    /// id is what removes the need to put three differently-typed recipes in
    /// one array: each caller submits its own spec, so no closure type has to
    /// be erased to a boxed one.
    fn assert_cancels_clean(q: &JobQueue, id: u64, writes_intent: bool, case: &str) {
        // Read the intent head off the submitted DAG rather than out of the
        // spec: a declared job holds its own nodes and inserts them.
        let has_intent = declared_shape(q, id).iter().any(|d| d.kind == "set_wanted");
        assert_eq!(has_intent, writes_intent, "{case}: intent head");
        assert!(q.cancel(id), "{case}: cancelled while queued");
        assert_eq!(state_of(q, id), State::Cancelled);
        // Nothing is left that *could* run. Asserting on the pending set rather
        // than on "what is ready this instant" also covers a node that is alive
        // but blocked — which is exactly what a leftover compensating node
        // would look like.
        assert_eq!(
            pending_kinds(q, id),
            Vec::<&str>::new(),
            "{case}: a power op emits no tail node, so a cancelled one leaves nothing"
        );
    }

    for graceful in [false, true] {
        for running in [false, true] {
            let targets = vec![("agent-a".to_owned(), running)];
            let case = format!("graceful={graceful} running={running}");

            let q = JobQueue::new(1);
            let id = submit(
                &q,
                submit::restart_spec(&targets, graceful, Source::Manual, "bounce".to_owned()),
            );
            assert_cancels_clean(&q, id, false, &format!("restart {case}"));

            let q = JobQueue::new(1);
            let id = submit(
                &q,
                submit::stop_spec(&targets, graceful, Source::Manual, "stop".to_owned()),
            );
            assert_cancels_clean(&q, id, true, &format!("stop {case}"));

            let q = JobQueue::new(1);
            let id = submit(
                &q,
                submit::start_spec(
                    &[("agent-a".to_owned(), running, false)],
                    Source::Manual,
                    "start".to_owned(),
                ),
            );
            assert_cancels_clean(&q, id, true, &format!("start {case}"));
        }
    }
}

// ---- terminal reporting + lease release ----

/// A DAG cancelled while fully queued must still **run its tail**, or a queued
/// approval DAG cancelled by the operator would dangle its approval forever.
///
/// This is the load-bearing case for sparing tails in [`JobQueue::cancel`]: the
/// work nodes all cancel, but `ResolveApproval` is weak-edged, so a `Cancelled`
/// dep satisfies its edge and it becomes claimable instead of being cancelled
/// along with everything else. It reads `Cancelled` off its own deps and resolves
/// the approval as "cancelled before completion".
#[test]
fn cancelled_dag_still_runs_its_approval_tail() {
    let q = JobQueue::new(1);
    let id = submit(
        &q,
        templates::approval_deploy("agent-a", 7, "approval #7".to_owned()),
    );
    assert!(q.cancel(id), "fully-queued dag cancels");
    // The `Cancelled` tail is the only node whose edge accepts a dropped
    // dependency, so it is the only one `cancel` spares — and claiming it *is*
    // the assertion that the approval gets resolved as cancelled.
    settle_approval_tail(&q, 7, TerminalState::Cancelled);
    assert_eq!(state_of(&q, id), State::Cancelled);
    // Unrelated later activity doesn't disturb the settled DAG.
    let other = submit(&q, rebuild("agent-b", "r"));
    let c = claim_one(&q);
    assert_eq!(c.dag_id, other);
    q.complete_node(c.node_id, Err("boom".to_owned()));
    assert_eq!(state_of(&q, id), State::Cancelled);
}

// ---- approval deploy subtree ----

/// The config-PR deploy is a subtree, not one opaque node. The
/// resource-holding root completes immediately (its `Finishing` state is the
/// parent gate that releases the children), then the phases run strictly in
/// order — and the `AfterAny` tail still runs when the irreversible half fails,
/// because it's the node that compensates for it.
#[test]
fn deploy_dag_runs_phases_in_order_and_tails_a_failed_apply() {
    let q = JobQueue::new(1);
    let id = submit(
        &q,
        templates::approval_deploy("agent-a", 7, "approval #7".to_owned()),
    );

    assert_eq!(
        declared_shape(&q, id),
        vec![
            // The window is the group root and holds the meta window for the
            // whole subtree; the three phases are its sub-nodes.
            row("deploy_window", None, &[]),
            row("merge_verify", Some("deploy_window"), &[]),
            // Apply only on a clean verify — a failed verify cancel-cascades
            // it, which is what leaves the forge and the applied repo untouched.
            row(
                "deploy_apply",
                Some("deploy_window"),
                &[("merge_verify", "done")]
            ),
            // The compensation tail accepts every terminal outcome of apply,
            // *including `skipped`* — which is the state apply lands in when
            // verify failed and it never ran. That one edge is the entire
            // "still tails a failed apply / a failed verify" behaviour, and it
            // is why two separate DAG-driving tests collapsed into this table.
            row(
                "deploy_tail",
                Some("deploy_window"),
                &[("deploy_apply", "done|failed|skipped")]
            ),
            // One approval tail per outcome, gated on the window's roll-up.
            row("resolve_approval", None, &[("deploy_window", "done")]),
            row("resolve_approval", None, &[("deploy_window", "failed")]),
            row("resolve_approval", None, &[("deploy_window", "cancelled")]),
        ]
    );
}

/// The deploy's happy path: `DeployApply` does not build. It grows the ordinary
/// rebuild chain into the live DAG under itself, and `FinalizeDeploy` — gated on
/// that graft finishing — plants the deploy tag last.
///
/// The queue is built with **one** build slot on purpose. `DeployWindow` already
/// holds that slot (and the meta window) for the whole subtree, so the grafted
/// `Prebuild` can only ever claim by *re-entering* its ancestor's hold. If the
/// graft were rooted anywhere outside `DeployWindow`'s subtree it would block on
/// a resource its own DAG owns and deadlock — this test is what pins that down.
#[test]
fn deploy_apply_grows_rebuild_subgraph_and_finalizes_after_it() {
    // What `DeployApply` grows is `deploy_rebuild_nodes`' output, and that is a
    // pure declaration — so it is declared here directly rather than by running
    // a deploy far enough to graft it. **Reproducing the runtime path is not
    // needed to test what the runtime path declares.**
    //
    // The grafting mechanism itself is hive_jobq's and tested there: the work
    // lands under the emitter *before* it settles, and the emitter parks in
    // `Finishing` so a downstream `AfterAny` gate stays shut while the new
    // children run (`a_completing_node_grows_the_work_it_declared`,
    // `parent_parks_in_finishing_until_children_roll_up`).
    let q = JobQueue::new(1);
    let id = submit(
        &q,
        DagSpec {
            source: Source::Manual,
            reason: "deploy graft".to_owned(),
            declare: Box::new(|b: &Job| templates::deploy_rebuild_nodes(b, "agent-a", 11)),
        },
    );

    assert_eq!(
        declared_shape(&q, id),
        vec![
            row("meta_sync", None, &[]),
            row("prebuild", None, &[("meta_sync", "done")]),
            row("stop_for_update", Some("prebuild"), &[]),
            row("swap", Some("stop_for_update"), &[]),
            row("post_swap", Some("stop_for_update"), &[("swap", "done")]),
            row("reconcile", None, &[("prebuild", "done|failed|skipped")]),
            // The deploy tag is planted only after the rebuild came up clean:
            // `AfterOk` on **both** roots, so either one failing skips it. That
            // pair of edges is the whole "skips finalize on a failed graft"
            // behaviour — no run needed to see it.
            row(
                "finalize_deploy",
                None,
                &[("prebuild", "done"), ("reconcile", "done")]
            ),
        ]
    );
}

// `deploy_dag_skips_finalize_but_still_tails_a_failed_graft` lived here.
//
// A failure *inside* the grafted rebuild is the failure mode subgraph growth
// introduces: the deploy is already merged and the container half-swapped, so
// `FinalizeDeploy` must be cancel-cascaded (no `deployed/<id>` tag planted)
// while the tail still runs to compensate, and `Reconcile` is deliberately
// still reached — it boots the container back up.
//
// Every declared half of that is asserted by
// `deploy_apply_grows_rebuild_subgraph_and_finalizes_after_it`, which reads
// `deploy_rebuild_nodes`' shape directly:
//
//   - "a failed swap still reaches Reconcile" is the `reconcile` row's
//     `AfterAny` edge on `prebuild` (`done|failed|skipped`);
//   - "finalize is cancel-cascaded" is `finalize_deploy`'s `AfterOk` pair —
//     either root failing skips it;
//   - "the tail still runs" is `deploy_tail`'s own `done|failed|skipped` edge
//     on apply, asserted in the `approval_deploy` table above.
//
// The runtime halves are hive_jobq's: cascade on failure, roll-up, and
// `first_error` digging past a group root that rolled up `Failed` while
// carrying no error of its own (`first_error_skips_a_rolled_up_failure_
// carrying_no_error`). That last one is why the DAG reports "profile swap
// failed" rather than nothing — the mechanism, not this shape.

// `deploy_dag_skips_apply_but_still_runs_tail_when_verify_fails` lived here.
//
// A pre-merge rejection (drift gate, eval failure) cancel-cascades the
// irreversible half via its `AfterOk` edge, while the tail is still reached —
// it owns the forge mirror, not just compensation. That test and
// `deploy_dag_runs_phases_in_order_and_tails_a_failed_apply` differed only in
// *where* they injected the failure, and each drove the whole DAG to watch the
// tail run anyway.
//
// Both outcomes follow from one declared edge, which the surviving test now
// asserts directly: the tail accepts `done|failed|skipped` on apply, and
// `skipped` is exactly the state apply lands in when verify failed and it never
// ran. The runtime halves are hive_jobq's and tested there —
// `failed_after_ok_dep_cancels_dependents_but_after_any_still_runs` and
// `failed_child_rolls_parent_up_to_failed` (an Ok tail cannot launder a failed
// deploy into a success).

// ---- history ----
//
// The node → build-log link is no longer queue state: the log row carries
// `node_id` and the lookup lives in `stores::build_logs` (see
// `node_link_survives_completion_and_newest_wins` there). Nothing in the queue
// needs testing for it any more, which is the point of that move.

/// History retention is a **flat** newest-first cap over all terminal DAGs
/// (`MAX_HISTORY_DAGS`), not a per-template bucket behind a grace window.
/// The dashboard renders one recent-builds list, so one number bounds it —
/// and with no bucketing there's nothing for a burst of same-shaped DAGs to
/// evict early, which is what the grace window used to paper over.
#[test]
fn history_evicts_oldest_terminals_past_flat_cap() {
    const OVERFLOW: usize = 8;
    let q = JobQueue::new(1);
    let mut ids = Vec::new();
    for i in 0..(MAX_HISTORY_DAGS + OVERFLOW) {
        let id = submit(
            &q,
            templates::reconcile_only(&format!("agent-{i}"), Source::Manual, "start".to_owned()),
        );
        let c = claim_one(&q);
        // Fail the single work node so the DAG *lingers*: a fully-`Done` DAG
        // drops off the wire entirely, but a `Failed` one is retained (+
        // history-capped) so the operator can still triage it. Completing the
        // node rolls the container up terminal.
        q.complete_node(c.node_id, Err("boom".to_owned()));
        ids.push(id);
    }
    let kept: std::collections::HashSet<u64> = q.snapshot().iter().map(|d| d.id).collect();
    assert_eq!(kept.len(), MAX_HISTORY_DAGS, "flat history cap");
    // Newest-first: the oldest `OVERFLOW` fall off, everything after survives.
    // These DAGs settle within the same wall-clock second, so this also pins
    // the `NodeId`-descending tiebreak that orders them when `finished_at` ties.
    for old in &ids[..OVERFLOW] {
        assert!(!kept.contains(old), "oldest terminal {old} evicted");
    }
    for recent in &ids[OVERFLOW..] {
        assert!(kept.contains(recent), "recent terminal {recent} retained");
    }
    assert_eq!(q.live_count(), 0);
}

#[test]
fn error_truncation_cuts_on_a_char_boundary() {
    // `truncate_error` is a pure `&str -> String`. This used to submit a DAG,
    // claim its head, fail it with a long string and read the error back out of
    // a snapshot — four moving parts to observe one transform, and the DAG
    // round-trip is covered by its own tests either way.
    //
    // Testing it directly also reaches the case the round-trip never did: the
    // cap is a **byte** length, so a multibyte char straddling it would panic
    // the slice. That boundary scan is the only non-obvious line in the fn and
    // it had no coverage at all.
    assert_eq!(
        truncate_error("short"),
        "short",
        "under the cap is untouched"
    );

    let ascii = truncate_error(&"x".repeat(5000));
    assert!(ascii.ends_with('…'));
    assert!(ascii.len() <= MAX_ERROR_LEN + '…'.len_utf8());

    // 'é' is 2 bytes, so the 2000-byte cap lands mid-char.
    let multibyte = truncate_error(&"é".repeat(5000));
    assert!(multibyte.ends_with('…'));
    assert!(multibyte.len() <= MAX_ERROR_LEN + '…'.len_utf8());
}

// ---- template shapes ----

#[test]
fn graceful_stop_shape_signal_drain_reconcile() {
    let q = JobQueue::new(1);
    let id = submit(&q, stop_online(&["agent-a"], true, "graceful"));
    assert_eq!(
        declared_shape(&q, id),
        vec![
            // The whole stop hangs under `set_wanted`: the durable intent is
            // written first, and the mechanical steps are its sub-nodes.
            row("set_wanted", None, &[]),
            row("signal", Some("set_wanted"), &[]),
            row("drain", Some("set_wanted"), &[("signal", "done")]),
            row("reconcile", Some("set_wanted"), &[("drain", "done")]),
        ]
    );
    // Neither half of the graceful window takes a build slot. That is what lets
    // a whole-hive graceful stop overlap every agent's drain even at
    // `buildSlots = 1` while a rebuild hogs the slot — the cost ceiling is one
    // `GRACEFUL_STOP_TIMEOUT` in total, not one per agent.
    let agent = Resource::Agent("agent-a".to_owned());
    assert_eq!(
        [
            declared_resources(&q, node_of(&q, id, "signal")),
            declared_resources(&q, node_of(&q, id, "drain")),
        ],
        [vec![agent.clone()], vec![agent]],
        "signal and drain hold the lease but never a build slot"
    );
}

#[test]
fn spawn_shape_provision_create_dropin_reconcile() {
    let q = JobQueue::new(1);
    let id = submit(
        &q,
        templates::spawn("newbie", 7, "approval #7 spawn".to_owned()),
    );
    assert_eq!(
        declared_shape(&q, id),
        vec![
            row("provision", None, &[]),
            row("create", Some("provision"), &[]),
            row("write_dropin", Some("create"), &[]),
            row("reconcile", Some("create"), &[("write_dropin", "done")]),
            // One tail per outcome, each edged to accept only that one — so
            // *which* tail the graph lets run already is the answer, and
            // nothing branches at runtime. The three differ **only** in their
            // accepted outcome, which is why `declared_shape` spells the
            // outcome set out instead of bucketing it.
            row("resolve_approval", None, &[("provision", "done")]),
            row("resolve_approval", None, &[("provision", "failed")]),
            row("resolve_approval", None, &[("provision", "cancelled")]),
        ]
    );
}

#[test]
fn perm_change_shape_prefixes_rebuild_chain() {
    let q = JobQueue::new(1);
    let id = submit(
        &q,
        templates::perm_change(
            "agent-a",
            Source::Manual,
            "perm".to_owned(),
            PermPayload::Combined {
                groups: Some(vec![]),
                caps: None,
            },
        ),
    );
    assert_eq!(
        declared_shape(&q, id)
            .iter()
            .map(|d| d.kind)
            .collect::<Vec<_>>(),
        vec![
            "write_perm_file",
            "meta_sync",
            "prebuild",
            "stop_for_update",
            "swap",
            "post_swap",
            "reconcile",
            // the ok / !ok tail pair
            "emit_rebuilt",
            "emit_rebuilt",
        ],
        "the perm write prefixes an otherwise ordinary rebuild chain"
    );
}

#[test]
fn reparent_shape_is_a_lone_agentless_meta_window_node() {
    // Single-move `set-parent` shape: one node, no rebuild subgraph (no
    // container rebuild needed for a parent move), agentless like
    // `MetaLock`, and it must declare the meta window — a topology commit
    // must not land inside another node's staged deploy window.
    let q = JobQueue::new(1);
    let id = submit(
        &q,
        templates::reparent(
            vec![(ident("alice"), Some(ident("bob")))],
            Source::Manual,
            "set-parent".to_owned(),
        ),
    );
    assert_eq!(
        declared_shape(&q, id),
        vec![row("reparent", None, &[])],
        "one node, no rebuild subgraph"
    );
    let node = node_of(&q, id, "reparent");
    assert_eq!(
        declared_resources(&q, node),
        vec![Resource::MetaWindow],
        "a topology commit must declare the same MetaWindow as WritePermFile, \
         and nothing else — no lease (agentless), no build slot (no nix work)"
    );
}

#[test]
fn reparent_bulk_shape_carries_every_move_on_one_node() {
    // `set-parent-bulk`: still ONE node (one git commit, `moves.len() > 1`),
    // not one node per move — bulk atomicity across every move in the
    // request is the reason a single node was chosen in the first place.
    let moves = vec![(ident("alice"), Some(ident("bob"))), (ident("carol"), None)];
    let q = JobQueue::new(1);
    let id = submit(
        &q,
        templates::reparent(moves.clone(), Source::Manual, "set-parent-bulk".to_owned()),
    );
    assert_eq!(
        declared_shape(&q, id),
        vec![row("reparent", None, &[])],
        "one node for the whole request, not one per move"
    );
    let NodeKind::Reparent { moves: got } = payload_of(&q, id, "reparent") else {
        panic!("expected a Reparent node");
    };
    assert_eq!(got, moves, "every move rides the single node");
}