hyperhive/hive-c0re/src/job_queue/tests.rs

//! 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::{Dep, DepWhen, NodeKind, NodeSpec};
use super::*;

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

fn rebuild(agent: &str, reason: &str) -> DagSpec {
    templates::rebuild(agent, Source::Manual, reason.to_owned(), true)
}

/// Restart DAG spec with every agent treated as **running** — the online
/// shape (`SetWanted → [Signal→Drain→] StopForUpdate → Reconcile`) 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 {
    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 {
    let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect();
    submit::stop_spec(&targets, graceful, Source::Manual, reason.to_owned())
}

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

fn state_of(q: &JobQueue, dag_id: u64) -> State {
    q.snapshot()
        .iter()
        .find(|d| d.id == dag_id)
        .expect("dag present")
        .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() {
    let q = JobQueue::new(1);
    let a = submit(&q, rebuild("agent-a", "first"));
    let claim = claim_one(&q); // Prebuild running
    assert_eq!(claim.dag_id, a);
    // While the original runs, re-submit is legitimate new work.
    let again = submit(&q, rebuild("agent-a", "config bumped during build"));
    assert_ne!(a, again);
    assert_eq!(q.snapshot().len(), 2);
}

// ---- cycle rejection ----

#[test]
fn cyclic_dag_is_rejected_at_submit() {
    let q = JobQueue::new(1);
    let mut spec = rebuild("agent-a", "cyclic");
    // 0 → 1 → 0 cycle.
    spec.nodes = vec![
        NodeSpec {
            agent: "agent-a".to_owned(),
            kind: NodeKind::StopForUpdate,
            deps: vec![Dep {
                on: 1,
                when: DepWhen::AfterOk,
            }],
        },
        NodeSpec {
            agent: "agent-a".to_owned(),
            kind: NodeKind::Reconcile,
            deps: vec![Dep {
                on: 0,
                when: DepWhen::AfterOk,
            }],
        },
    ];
    assert!(q.submit(spec).is_err(), "cyclic spec must be refused");
    assert!(q.snapshot().is_empty());
}

#[test]
fn unknown_dep_is_rejected_at_submit() {
    let q = JobQueue::new(1);
    let mut spec = rebuild("agent-a", "bad dep");
    spec.nodes = vec![NodeSpec {
        agent: "agent-a".to_owned(),
        kind: NodeKind::Reconcile,
        deps: vec![Dep {
            on: 9,
            when: DepWhen::AfterOk,
        }],
    }];
    assert!(q.submit(spec).is_err());
}

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

#[test]
fn rebuild_chain_claims_in_dep_order() {
    let q = JobQueue::new(1);
    let id = submit(&q, rebuild("agent-a", "r"));
    for expected in [
        "prebuild",
        "stop_for_update",
        "swap",
        "post_swap",
        "reconcile",
    ] {
        let c = claim_one(&q);
        assert_eq!(c.dag_id, id);
        assert_eq!(c.kind.as_str(), expected);
        assert!(
            q.claim_ready().is_empty(),
            "chain must serialize: nothing ready while {expected} runs"
        );
        q.complete_node(id, c.node_id, Ok(()));
    }
    assert_eq!(state_of(&q, id), State::Done);
}

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

#[test]
fn build_slot_serializes_nix_heavy_nodes() {
    let q = JobQueue::new(1);
    let a = submit(&q, rebuild("agent-a", "r"));
    let b = submit(&q, rebuild("agent-b", "r"));
    let first = claim_one(&q); // a's Prebuild takes the only slot
    assert_eq!(first.dag_id, a);
    assert_eq!(first.kind.as_str(), "prebuild");
    q.complete_node(a, first.node_id, Ok(()));
    // With the slot free again, FIFO gives... a's StopForUpdate is
    // slot-free (lease) and b's Prebuild takes the slot — both run.
    let claims = q.claim_ready();
    let kinds: Vec<(u64, &str)> = claims.iter().map(|c| (c.dag_id, c.kind.as_str())).collect();
    assert!(kinds.contains(&(a, "stop_for_update")));
    assert!(kinds.contains(&(b, "prebuild")));
    assert_eq!(claims.len(), 2);
}

#[test]
fn two_build_slots_run_two_prebuilds() {
    let q = JobQueue::new(2);
    submit(&q, rebuild("agent-a", "r"));
    submit(&q, rebuild("agent-b", "r"));
    let claims = q.claim_ready();
    assert_eq!(claims.len(), 2, "two slots → two concurrent prebuilds");
    assert!(claims.iter().all(|c| c.kind.as_str() == "prebuild"));
}

#[test]
fn fifo_fairness_for_the_slot() {
    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, rebuild("agent-c", "r"));
    let first = claim_one(&q);
    assert_eq!(first.dag_id, a, "submit order wins the slot");
    q.complete_node(a, first.node_id, Ok(()));
    let next: Vec<u64> = q.claim_ready().iter().map(|cl| cl.dag_id).collect();
    assert!(next.contains(&b), "b's prebuild before c's");
    assert!(!next.contains(&c));
}

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

#[test]
fn lease_serializes_two_lifecycle_dags_for_same_agent() {
    let q = JobQueue::new(4);
    let restart = submit(&q, restart_online(&["agent-a"], false, "restart"));
    let stop = submit(
        &q,
        templates::reconcile_only(
            Template::Stop,
            "agent-a",
            Source::Manual,
            "stop".to_owned(),
            None,
        ),
    );
    // Restart's head SetWanted takes the lease; stop's Reconcile must
    // wait even though slots are free.
    let first = claim_one(&q);
    assert_eq!(first.dag_id, restart);
    assert_eq!(first.kind.as_str(), "set_wanted");
    assert!(first.lease_acquired);
    q.complete_node(restart, first.node_id, Ok(()));
    // Same DAG keeps the lease through StopForUpdate then Reconcile.
    let second = claim_one(&q);
    assert_eq!(second.dag_id, restart);
    assert_eq!(second.kind.as_str(), "stop_for_update");
    assert!(!second.lease_acquired, "lease already held by this DAG");
    q.complete_node(restart, second.node_id, Ok(()));
    let third = claim_one(&q);
    assert_eq!(third.dag_id, restart);
    assert_eq!(third.kind.as_str(), "reconcile");
    q.complete_node(restart, third.node_id, Ok(()));
    // Restart terminal → lease released → stop's Reconcile runs.
    let fourth = claim_one(&q);
    assert_eq!(fourth.dag_id, stop);
    q.complete_node(stop, fourth.node_id, Ok(()));
    assert_eq!(state_of(&q, restart), State::Done);
    assert_eq!(state_of(&q, stop), State::Done);
}

#[test]
fn lease_exempt_prebuild_overlaps_other_dag_on_same_agent() {
    let q = JobQueue::new(2);
    submit(&q, rebuild("agent-a", "rebuild"));
    let stop = submit(
        &q,
        templates::reconcile_only(
            Template::Stop,
            "agent-a",
            Source::Manual,
            "stop".to_owned(),
            None,
        ),
    );
    // Prebuild is lease-exempt: the stop's Reconcile takes the lease
    // and runs concurrently with the rebuild's out-of-band nix build.
    let claims = q.claim_ready();
    let kinds: Vec<&str> = claims.iter().map(|c| c.kind.as_str()).collect();
    assert!(kinds.contains(&"prebuild"));
    assert!(kinds.contains(&"reconcile"));
    // But the rebuild's StopForUpdate must then wait for the stop DAG
    // to finish (lease).
    let prebuild = claims
        .iter()
        .find(|c| c.kind.as_str() == "prebuild")
        .expect("prebuild claim")
        .clone();
    q.complete_node(prebuild.dag_id, prebuild.node_id, Ok(()));
    assert!(
        q.claim_ready().is_empty(),
        "StopForUpdate blocked while stop DAG holds the lease"
    );
    let reconcile = claims
        .iter()
        .find(|c| c.kind.as_str() == "reconcile")
        .expect("reconcile claim")
        .clone();
    q.complete_node(stop, reconcile.node_id, Ok(()));
    let next = claim_one(&q);
    assert_eq!(next.kind.as_str(), "stop_for_update");
}

#[test]
fn agents_do_not_contend_on_each_others_leases() {
    let q = JobQueue::new(4);
    submit(&q, restart_online(&["agent-a"], false, "r"));
    submit(&q, restart_online(&["agent-b"], false, "r"));
    let claims = q.claim_ready();
    assert_eq!(claims.len(), 2, "different agents run concurrently");
}

#[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 (SetWanted) is a root, so both are
    // claimable at once — each takes its OWN agent's lease (no contention
    // across distinct agents), all inside the single DAG.
    let claims = q.claim_ready();
    assert!(claims.iter().all(|c| c.dag_id == id));
    let mut heads: Vec<(&str, &str, bool)> = claims
        .iter()
        .map(|c| (c.agent.as_str(), c.kind.as_str(), c.lease_acquired))
        .collect();
    heads.sort_unstable();
    assert_eq!(
        heads,
        vec![
            ("agent-a", "set_wanted", true),
            ("agent-b", "set_wanted", true),
        ],
        "both per-agent subgraphs start concurrently, each acquiring its own lease"
    );
}

/// A multi-agent DAG frees an agent's lease the moment THAT agent's
/// subgraph is terminal — not when the whole DAG finishes. So a
/// concurrent DAG wanting the finished agent can proceed while the rest
/// of the first DAG runs on.
#[test]
fn multi_agent_lease_frees_per_subgraph_not_whole_dag() {
    let q = JobQueue::new(4);
    let id = submit(&q, restart_online(&["agent-a", "agent-b"], false, "r"));

    // Drive agent-a's ENTIRE subgraph to Done while leaving agent-b's
    // head running (so agent-b keeps holding its lease).
    let mut b_in_flight = false;
    loop {
        let mut progressed = false;
        for c in q.claim_ready() {
            if c.agent == "agent-a" {
                q.complete_node(id, c.node_id, Ok(()));
                progressed = true;
            } else {
                b_in_flight = true; // leave agent-b's node running
            }
        }
        if !progressed {
            break;
        }
    }
    assert!(b_in_flight, "agent-b subgraph should still be in flight");
    // The DAG as a whole is NOT terminal — agent-b runs on.
    assert_eq!(state_of(&q, id), State::Running);

    // agent-a's lease is freed early → a concurrent agent-a DAG runs;
    // an agent-b DAG still blocks on the lease agent-b's subgraph holds.
    submit(&q, restart_online(&["agent-a"], false, "concurrent-a"));
    submit(&q, restart_online(&["agent-b"], false, "concurrent-b"));
    let claims = q.claim_ready();
    let agents: Vec<&str> = claims.iter().map(|c| c.agent.as_str()).collect();
    assert!(
        agents.contains(&"agent-a"),
        "agent-a lease freed the moment its subgraph settled"
    );
    assert!(
        !agents.contains(&"agent-b"),
        "agent-b lease still held — its subgraph is still in flight"
    );
}

#[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, bool)> = claims
        .iter()
        .map(|c| (c.agent.as_str(), c.kind.as_str(), c.lease_acquired))
        .collect();
    heads.sort_unstable();
    assert_eq!(
        heads,
        vec![
            ("agent-a", "set_wanted", true),
            ("agent-b", "set_wanted", true),
        ],
        "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 (prebuild first).
    for c in &heads {
        q.complete_node(id, 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", "prebuild")],
        "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 → SetWanted(Up) → Reconcile (no StopForUpdate): a
    // restart of a down agent is really a start.
    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!["set_wanted".to_owned(), "reconcile".to_owned()],
        "offline restart skips StopForUpdate, keeps Reconcile (it's a start)"
    );
}

#[test]
fn append_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 {
        template: Template::Boot,
        source: Source::AutoUpdate,
        reason: "sweep".to_owned(),
        approval_id: None,
        inputs: Vec::new(),
        perm_payload: None,
        transient: None,
        nodes: vec![NodeSpec {
            agent: "hyperhive".to_owned(),
            kind: NodeKind::MetaLock {
                sweep: true,
                fanout: None,
            },
            deps: Vec::new(),
        }],
    };
    let id = 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 (`rebuild_nodes(_, true, 0)`: root Prebuild →
    // StopForUpdate → Swap → Reconcile, local 0-based deps), so this test
    // tracks any drift in that builder's root-first (`base = 0`) shape.
    let subgraph = |agent: &str| templates::rebuild_nodes(agent, true, 0);
    // Must append BEFORE completing the emitter (the documented contract).
    q.append_subgraph(id, subgraph("a"), emitter.node_id);
    q.append_subgraph(id, subgraph("b"), emitter.node_id);
    q.complete_node(id, emitter.node_id, Ok(()));
    // Still ONE DAG; both subgraph roots become ready once the emitter is
    // Done (rooted on it), each on its own agent lease.
    assert_eq!(q.snapshot().len(), 1);
    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![("a", "prebuild"), ("b", "prebuild")],
        "both rebuild subgraphs root on the emitter and run concurrently in one DAG"
    );
}

#[test]
fn meta_update_carries_rebuilding_transient_and_grows_cascade_in_dag() {
    // The meta-update `MetaLock` grows one rebuild subgraph per affected
    // agent into its OWN DAG (via append_subgraph), not child DAGs.
    // The DAG carries `Rebuilding` so the folded rebuilds keep crash-watch
    // suppression (the property the old child Rebuild DAGs had via their own
    // transient).
    let spec = templates::meta_update(
        vec!["nixpkgs".to_owned()],
        Source::Manual,
        "bump".to_owned(),
        None,
    );
    assert!(
        matches!(
            spec.transient,
            Some(crate::coordinator::TransientKind::Rebuilding)
        ),
        "meta-update DAG must carry Rebuilding so cascade rebuilds get suppression"
    );
    let q = JobQueue::new(4);
    let id = 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).
    for agent in ["alice", "bob"] {
        q.append_subgraph(
            id,
            templates::rebuild_nodes(agent, false, 0),
            meta_lock.node_id,
        );
    }
    q.complete_node(id, meta_lock.node_id, Ok(()));
    // Still ONE DAG — no child DAGs — and both cascade rebuild subgraphs root
    // on the MetaLock, each on its own agent lease.
    assert_eq!(q.snapshot().len(), 1);
    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![("alice", "prebuild"), ("bob", "prebuild")],
        "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 prebuild = claim_one(&q);
    q.complete_node(id, 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(id, reconcile.node_id, Ok(()));
    let snap = q.snapshot();
    let dag = snap.iter().find(|d| d.id == id).expect("dag");
    assert_eq!(dag.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);
    assert_eq!(by_kind("stop_for_update"), State::Cancelled);
    assert_eq!(by_kind("swap"), State::Cancelled);
    assert_eq!(by_kind("post_swap"), State::Cancelled);
    assert_eq!(by_kind("reconcile"), State::Done);
    assert_eq!(
        dag.nodes
            .iter()
            .find(|n| n.kind == "prebuild")
            .and_then(|n| n.error.as_deref()),
        Some("nix build exploded")
    );
}

/// The swap-failure recovery: `Swap` fails → the `AfterOk` `PostSwap` is
/// cancel-cascaded → its terminal state still satisfies `Reconcile`'s
/// `AfterAny(PostSwap)` edge, so recovery-start runs and brings a wanted-up
/// agent back on its old config.
#[test]
fn swap_failure_still_runs_reconcile() {
    let q = JobQueue::new(1);
    let id = submit(&q, rebuild("agent-a", "r"));
    for _ in 0..2 {
        let c = claim_one(&q);
        q.complete_node(id, c.node_id, Ok(()));
    }
    let swap = claim_one(&q);
    assert_eq!(swap.kind.as_str(), "swap");
    q.complete_node(id, swap.node_id, Err("update failed".to_owned()));
    // PostSwap (AfterOk on the failed Swap) is cancel-cascaded; Reconcile is
    // next-claimable via its AfterAny(PostSwap) edge.
    let reconcile = claim_one(&q);
    assert_eq!(reconcile.kind.as_str(), "reconcile");
    q.complete_node(id, reconcile.node_id, Ok(()));
    let all_dags = q.snapshot();
    let dag = all_dags.iter().find(|d| d.id == id).expect("dag");
    assert_eq!(
        dag.nodes
            .iter()
            .find(|n| n.kind == "post_swap")
            .expect("post_swap node")
            .state,
        State::Cancelled,
        "PostSwap must cancel-cascade when Swap fails"
    );
    assert_eq!(state_of(&q, id), State::Failed);
}

/// 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 swap_ok_runs_post_swap_before_reconcile() {
    let q = JobQueue::new(1);
    let id = submit(&q, rebuild("agent-a", "r"));
    // prebuild + stop_for_update
    for _ in 0..2 {
        let c = claim_one(&q);
        q.complete_node(id, c.node_id, Ok(()));
    }
    let swap = claim_one(&q);
    assert_eq!(swap.kind.as_str(), "swap");
    q.complete_node(id, swap.node_id, Ok(()));
    // PostSwap runs next, and nothing else is claimable while it does — the
    // tail serializes ahead of Reconcile.
    let post_swap = claim_one(&q);
    assert_eq!(post_swap.kind.as_str(), "post_swap");
    assert!(
        q.claim_ready().is_empty(),
        "Reconcile must wait for PostSwap, not race it"
    );
    q.complete_node(id, post_swap.node_id, Ok(()));
    let reconcile = claim_one(&q);
    assert_eq!(reconcile.kind.as_str(), "reconcile");
    q.complete_node(id, reconcile.node_id, Ok(()));
    assert_eq!(state_of(&q, id), State::Done);
}

#[test]
fn failed_reconcile_marks_dag_failed() {
    let q = JobQueue::new(1);
    let id = submit(
        &q,
        templates::reconcile_only(
            Template::Start,
            "agent-a",
            Source::Manual,
            "start".to_owned(),
            None,
        ),
    );
    let c = claim_one(&q);
    q.complete_node(id, c.node_id, Err("start failed".to_owned()));
    assert_eq!(state_of(&q, id), State::Failed);
}

// ---- cancel ----

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

#[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);
}

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

#[test]
fn terminal_dag_reported_exactly_once_and_lease_released() {
    let q = JobQueue::new(1);
    let id = submit(&q, restart_online(&["agent-a"], false, "r"));
    // restart = SetWanted → StopForUpdate → Reconcile; not terminal until
    // the last node completes.
    let set_wanted = claim_one(&q);
    q.complete_node(id, set_wanted.node_id, Ok(()));
    assert!(q.drain_terminal().is_empty(), "dag not terminal yet");
    let stop = claim_one(&q);
    q.complete_node(id, stop.node_id, Ok(()));
    assert!(q.drain_terminal().is_empty(), "dag not terminal yet");
    let rec = claim_one(&q);
    q.complete_node(id, rec.node_id, Ok(()));
    let reports = q.drain_terminal();
    assert_eq!(reports.len(), 1);
    assert_eq!(reports[0].dag_id, id);
    assert_eq!(reports[0].state, State::Done);
    assert!(q.drain_terminal().is_empty(), "reported exactly once");
    // Lease released: a new DAG for the agent can claim immediately.
    let next = submit(
        &q,
        templates::reconcile_only(
            Template::Stop,
            "agent-a",
            Source::Manual,
            "stop".to_owned(),
            None,
        ),
    );
    let c = claim_one(&q);
    assert_eq!(c.dag_id, next);
    assert!(c.lease_acquired);
}

/// A DAG cancelled while fully queued must still surface a terminal
/// roll-up for the scheduler's hooks — otherwise a queued approval
/// DAG cancelled by the operator would dangle its approval forever.
#[test]
fn cancelled_dag_reports_terminal_once() {
    let q = JobQueue::new(1);
    let id = submit(
        &q,
        templates::approval_deploy("agent-a", 7, "approval #7".to_owned()),
    );
    assert!(q.cancel(id));
    let reports = q.drain_terminal();
    assert_eq!(reports.len(), 1);
    assert_eq!(reports[0].dag_id, id);
    assert_eq!(reports[0].state, State::Cancelled);
    assert_eq!(reports[0].approval_id, Some(7));
    // Never re-reported by later activity.
    let other = submit(&q, rebuild("agent-b", "r"));
    let c = claim_one(&q);
    assert_eq!(c.dag_id, other);
    q.complete_node(other, c.node_id, Err("boom".to_owned()));
    assert!(q.drain_terminal().iter().all(|t| t.dag_id != id));
}

// ---- steps, build logs, history ----

#[test]
fn set_step_only_on_running_and_signals_change() {
    let q = JobQueue::new(1);
    let id = submit(&q, rebuild("agent-a", "r"));
    assert!(!q.set_step(id, 0, "too early"), "queued node refuses step");
    let c = claim_one(&q);
    assert!(q.set_step(id, c.node_id, "nix build"));
    assert!(
        !q.set_step(id, c.node_id, "nix build"),
        "same label → false"
    );
    assert!(q.set_step(id, c.node_id, "next phase"));
    assert!(q.set_step_running(id, "via running lookup"));
    q.complete_node(id, c.node_id, Ok(()));
    let snap = q.snapshot();
    let node = &snap.iter().find(|d| d.id == id).expect("dag").nodes[0];
    assert_eq!(node.step, None, "step cleared on completion");
}

#[test]
fn set_build_log_id_links_running_node() {
    let q = JobQueue::new(1);
    let id = submit(&q, rebuild("agent-a", "r"));
    assert!(!q.set_build_log_id(id, 0, 41), "queued node refuses log id");
    let c = claim_one(&q);
    assert!(q.set_build_log_id(id, c.node_id, 42));
    assert!(q.set_build_log_id_running(id, 43));
    q.complete_node(id, c.node_id, Ok(()));
    let snap = q.snapshot();
    let node = &snap.iter().find(|d| d.id == id).expect("dag").nodes[0];
    assert_eq!(node.build_log_id, Some(43), "log id survives completion");
}

#[test]
fn history_evicts_old_terminals_per_template() {
    let q = JobQueue::new(1);
    for i in 0..8 {
        let id = submit(
            &q,
            templates::reconcile_only(
                Template::Start,
                &format!("agent-{i}"),
                Source::Manual,
                "start".to_owned(),
                None,
            ),
        );
        let c = claim_one(&q);
        q.complete_node(id, c.node_id, Ok(()));
    }
    // Fresh terminals are inside the grace window: nothing evicts yet,
    // so a ~1s QueueDag poller can still observe every terminal state
    // (a broad stop/start settles many same-template DAGs at once).
    assert_eq!(
        q.snapshot().len(),
        8,
        "grace window protects fresh terminals"
    );
    // Past the grace window the per-template cap applies.
    q.trim_ignoring_grace();
    assert_eq!(q.snapshot().len(), 5, "per-template history cap");
    assert_eq!(q.live_count(), 0);
}

#[test]
fn error_is_truncated() {
    let q = JobQueue::new(1);
    let id = submit(&q, rebuild("agent-a", "r"));
    let c = claim_one(&q);
    q.complete_node(id, c.node_id, Err("x".repeat(5000)));
    let snap = q.snapshot();
    let err = snap.iter().find(|d| d.id == id).expect("dag").nodes[0]
        .error
        .clone()
        .expect("error stored");
    assert!(err.chars().count() <= 2001, "truncated + ellipsis");
    assert!(err.ends_with('…'));
}

// ---- 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"));
    for expected in ["set_wanted", "signal", "drain", "reconcile"] {
        let c = claim_one(&q);
        assert_eq!(c.kind.as_str(), expected);
        q.complete_node(id, c.node_id, Ok(()));
    }
    assert_eq!(state_of(&q, id), State::Done);
}

#[test]
fn graceful_signal_and_drain_hold_no_build_slot() {
    // A whole-hive graceful stop overlaps every drain even at
    // buildSlots = 1 while a rebuild hogs the slot.
    let q = JobQueue::new(1);
    submit(&q, rebuild("builder", "slot hog"));
    submit(&q, stop_online(&["agent-a"], true, "g"));
    submit(&q, stop_online(&["agent-b"], true, "g"));
    let claims = q.claim_ready();
    let kinds: Vec<&str> = claims.iter().map(|c| c.kind.as_str()).collect();
    assert_eq!(
        kinds,
        vec!["prebuild", "set_wanted", "set_wanted"],
        "both agents' graceful-stop heads (SetWanted, build-slot-exempt) run \
         while the slot is held; their signals follow"
    );
}

#[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()),
    );
    for expected in ["provision", "create", "write_dropin", "reconcile"] {
        let c = claim_one(&q);
        assert_eq!(c.kind.as_str(), expected);
        assert_eq!(c.approval_id, Some(7));
        q.complete_node(id, c.node_id, Ok(()));
    }
    let report_terminal = state_of(&q, id);
    assert_eq!(report_terminal, State::Done);
}

#[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,
            },
        ),
    );
    for expected in [
        "write_perm_file",
        "prebuild",
        "stop_for_update",
        "swap",
        "post_swap",
        "reconcile",
    ] {
        let c = claim_one(&q);
        assert_eq!(c.kind.as_str(), expected);
        q.complete_node(id, c.node_id, Ok(()));
    }
    assert_eq!(state_of(&q, id), State::Done);
}