feat(#2450): collapse the startup sweep into one inline DAG

The boot is now ONE DAG, assembled inline in submit_boot_tree — no boot_root
Noop anchor, no per-agent child DAGs, no display-only parent_id grouping: a
sweep MetaLock root (only when something is stale) that grows one rebuild
subgraph per stale agent into the same DAG (via append_subgraph, previous
commit), plus one Reconcile root per drifted agent (independent — a boot
reconcile needs no lock bump).

- submit_boot_tree builds the DagSpec inline; removed the single-use
  templates::{boot_root, startup_sweep} builders (inlined per the operator's
  "don't force single-use shapes into templates.rs" steer).
- fanout_specs simplified to the meta-update cascade path only — the startup
  sweep no longer fans out child DAGs, so its branch was dead.
- test: append_subgraph_roots_on_emitter_and_rebases_local_deps.

Vestigial after this (deliberately left as follow-ups, flagged in the PR):
NodeKind::Noop is now unconstructed (contained to hive-c0re, removable);
Template::StartupSweep is unconstructed but a hive-sh4re wire type
(frontend-coordinated removal).

hive-c0re/src/job_queue/exec.rs

@@ -10,7 +10,7 @@ use std::sync::Arc;
 
 use anyhow::{Context as _, Result};
 
-use super::model::{NodeKind, State, Template};
+use super::model::{NodeKind, NodeSpec, State, Template};
 use super::{Claim, TerminalDag};
 use crate::coordinator::Coordinator;
 use crate::power::{ReconcileAction, reconcile_action};
@@ -38,6 +38,16 @@ pub struct NodeOutput {
     /// these *before* the emitting node's completion so the DAG never
     /// rolls terminal with the appended work still pending.
     pub append_nodes: Vec<NodeKind>,
+    /// Whole per-agent *subgraphs* to append into *this same* DAG at
+    /// runtime — the multi-node generalisation of `append_nodes`. Each
+    /// inner `Vec<NodeSpec>` is one independent subgraph whose `deps` are
+    /// local (0-based within that subgraph); the scheduler appends each via
+    /// [`JobQueue::append_subgraph`], which rebases the deps onto the DAG's
+    /// node-id space and roots the subgraph on the emitting node. The
+    /// startup sweep's `MetaLock` uses this to grow one stale-agent rebuild
+    /// subgraph per agent into the same boot DAG instead of fanning out
+    /// child DAGs. Same before-completion ordering as `append_nodes`.
+    pub append_subgraph: Vec<Vec<NodeSpec>>,
 }
 
 /// Step-label + build-log sink for one claimed node.
@@ -276,8 +286,17 @@ async fn run_meta_lock(
         if let Err(e) = crate::meta::lock_update_hyperhive().await {
             tracing::warn!(error = ?e, "startup sweep: meta lock_update_hyperhive failed");
         }
+        // Grow one rebuild subgraph per stale agent into *this* boot DAG
+        // (rooted on this `MetaLock`, so they build against the post-bump
+        // lock), rather than fanning out child DAGs. `relock = true` — a
+        // boot sweep relocks per-agent like a manual rebuild.
+        let append_subgraph = fanout
+            .unwrap_or_default()
+            .iter()
+            .map(|agent| super::templates::rebuild_nodes(agent, true, 0))
+            .collect();
         return Ok(NodeOutput {
-            fanout: fanout.unwrap_or_default(),
+            append_subgraph,
             ..Default::default()
         });
     }

hive-c0re/src/job_queue/mod.rs

@@ -34,7 +34,8 @@ use hive_sh4re::wire_time::now_unix;
 use tokio::sync::Notify;
 
 pub use model::{
-    Dag, DagSpec, DagView, DepWhen, Node, NodeId, NodeKind, PermPayload, Source, State, Template,
+    Dag, DagSpec, DagView, DepWhen, Node, NodeId, NodeKind, NodeSpec, PermPayload, Source, State,
+    Template,
 };
 
 /// How many terminal DAGs (`Done` / `Failed` / `Cancelled`) to retain
@@ -206,6 +207,71 @@ impl JobQueue {
         Some(new_id)
     }
 
+    /// Append a whole *subgraph* into a live (non-terminal) DAG at runtime
+    /// — the multi-node, multi-agent generalisation of [`Self::append_node`].
+    /// Each [`NodeSpec`] carries its own `agent` and subgraph-relative `deps`
+    /// (indices into `nodes`); this rebases those onto the DAG's node-id
+    /// space (`id == index`, an invariant `append_node` also maintains) and
+    /// attaches every subgraph *root* — a node with no internal deps — to
+    /// `dep_on` with an `AfterOk` edge. Used by the startup sweep's
+    /// `MetaLock` to grow per-agent rebuild subgraphs into the same boot DAG
+    /// instead of fanning out child DAGs. Same call-*before*-`complete_node`
+    /// contract as `append_node` (so the DAG can't roll terminal with the
+    /// appended work still pending). Returns the new node ids; empty if the
+    /// DAG is gone or `nodes` is empty.
+    pub fn append_subgraph(
+        &self,
+        dag_id: u64,
+        nodes: Vec<NodeSpec>,
+        dep_on: NodeId,
+    ) -> Vec<NodeId> {
+        if nodes.is_empty() {
+            return Vec::new();
+        }
+        let mut inner = self.inner.lock().expect("job_queue mutex poisoned");
+        let Some(dag) = inner.dags.iter_mut().find(|d| d.id == dag_id) else {
+            return Vec::new();
+        };
+        let base: NodeId = u32::try_from(dag.nodes.len()).unwrap_or(u32::MAX);
+        let mut new_ids = Vec::with_capacity(nodes.len());
+        for (i, spec) in nodes.into_iter().enumerate() {
+            let new_id: NodeId = base + u32::try_from(i).unwrap_or(u32::MAX);
+            // Subgraph roots (no internal deps) hang off the emitting node;
+            // internal deps rebase from subgraph-relative onto the DAG id
+            // space (both start at `base`).
+            let deps = if spec.deps.is_empty() {
+                vec![model::Dep {
+                    on: dep_on,
+                    when: DepWhen::AfterOk,
+                }]
+            } else {
+                spec.deps
+                    .into_iter()
+                    .map(|d| model::Dep {
+                        on: base + d.on,
+                        when: d.when,
+                    })
+                    .collect()
+            };
+            dag.nodes.push(Node {
+                id: new_id,
+                agent: spec.agent,
+                kind: spec.kind,
+                deps,
+                state: State::Queued,
+                step: None,
+                build_log_id: None,
+                started_at: None,
+                finished_at: None,
+                error: None,
+            });
+            new_ids.push(new_id);
+        }
+        drop(inner);
+        self.notify.notify_one();
+        new_ids
+    }
+
     fn push_dag(inner: &mut Inner, spec: DagSpec) -> u64 {
         inner.next_id += 1;
         let id = inner.next_id;

hive-c0re/src/job_queue/scheduler.rs

@@ -17,7 +17,7 @@ use std::collections::HashMap;
 use std::sync::Arc;
 
 use super::exec::{self, NodeOutput};
-use super::{Claim, Source, Template, templates};
+use super::{Claim, Source, templates};
 use crate::coordinator::Coordinator;
 
 struct NodeDone {
@@ -116,6 +116,15 @@ async fn handle_completion(
                     .job_queue
                     .append_node(claim.dag_id, kind, claim.node_id);
             }
+            // Same before-completion ordering as `append_nodes`, but for
+            // whole per-agent subgraphs (the startup sweep's rebuild
+            // subgraphs growing into the boot DAG) — each an independent
+            // subgraph rooted on this node.
+            for subgraph in output.append_subgraph {
+                coord
+                    .job_queue
+                    .append_subgraph(claim.dag_id, subgraph, claim.node_id);
+            }
             coord
                 .job_queue
                 .complete_node(claim.dag_id, claim.node_id, Ok(()));
@@ -157,26 +166,29 @@ async fn process_terminals(
     }
 }
 
-/// Child `Rebuild` specs for a completed `MetaLock` fan-out, grouped
-/// under the parent via `parent_id`. Meta-update children skip the
-/// per-agent relock (it would revert the bump the parent just
-/// committed); sweep children relock like a manual rebuild.
+/// Child `Rebuild` specs for a completed meta-update `MetaLock` fan-out,
+/// grouped under the parent via `parent_id`. Meta-update children skip the
+/// per-agent relock (`relock = false`) — it would revert the bump the parent
+/// just committed. This is now the meta-update cascade path only: the startup
+/// sweep no longer fans out child DAGs — it grows one rebuild subgraph per
+/// stale agent into its own DAG via `append_subgraph` (see
+/// `exec::run_meta_lock`).
 fn fanout_specs(claim: &Claim, agents: Vec<String>) -> Vec<super::DagSpec> {
-    let sweep = claim.template == Template::StartupSweep;
-    let (source, relock) = if sweep {
-        (Source::StartupSweep, true)
-    } else {
-        (Source::MetaUpdate, false)
-    };
-    let reason = if sweep {
-        "startup sweep".to_owned()
-    } else if let Some(approval_id) = claim.approval_id {
+    let reason = if let Some(approval_id) = claim.approval_id {
         format!("approval #{approval_id} meta input cascade")
     } else {
         "meta-update cascade".to_owned()
     };
     agents
         .into_iter()
-        .map(|agent| templates::rebuild(&agent, source, reason.clone(), Some(claim.dag_id), relock))
+        .map(|agent| {
+            templates::rebuild(
+                &agent,
+                Source::MetaUpdate,
+                reason.clone(),
+                Some(claim.dag_id),
+                false,
+            )
+        })
         .collect()
 }

hive-c0re/src/job_queue/templates.rs

@@ -216,49 +216,11 @@ pub fn meta_update(
     }
 }
 
-/// Boot-time root anchor DAG: a single [`NodeKind::Noop`] node that groups
-/// this boot's `StartupSweep` + per-agent `Reconcile` child DAGs (linked via
-/// `parent_id`) into one tree so the dashboard renders the boot as one entry.
-/// Holds no lease and does no work — the children it anchors still run
-/// concurrently. `auto_update::run` submits it first (when there's any boot
-/// work), then parents the sweep + reconciles onto its id.
-pub fn boot_root(reason: String) -> DagSpec {
-    DagSpec {
-        template: Template::Boot,
-        source: Source::AutoUpdate,
-        reason,
-        parent_id: None,
-        approval_id: None,
-        inputs: Vec::new(),
-        perm_payload: None,
-        transient: None,
-        nodes: vec![node("hyperhive", NodeKind::Noop, Vec::new())],
-    }
-}
-
-/// Boot-time sweep parent: bump meta's hyperhive input (non-fatal),
-/// then fan out `Rebuild` children for the precomputed stale agent
-/// list (topology-sorted by the caller).
-pub fn startup_sweep(reason: String, stale_agents: Vec<String>) -> DagSpec {
-    DagSpec {
-        template: Template::StartupSweep,
-        source: Source::AutoUpdate,
-        reason,
-        parent_id: None,
-        approval_id: None,
-        inputs: Vec::new(),
-        perm_payload: None,
-        transient: None,
-        nodes: vec![node(
-            "hyperhive",
-            NodeKind::MetaLock {
-                sweep: true,
-                fanout: Some(stale_agents),
-            },
-            Vec::new(),
-        )],
-    }
-}
+// The boot is now assembled inline in `workers/auto_update.rs::submit_boot_tree`
+// as ONE DAG (a sweep `MetaLock` root that grows rebuild subgraphs in-DAG, plus
+// a `Reconcile` root per drifted agent) — no `boot_root` Noop anchor, no
+// `startup_sweep` parent template, no per-agent child DAGs. The old single-use
+// `boot_root` / `startup_sweep` builders were inlined there and removed.
 
 /// Validate a spec before it enters the queue: node ids are dense
 /// (index = id), deps reference existing nodes, and the dep graph is

hive-c0re/src/job_queue/tests.rs

@@ -450,6 +450,58 @@ fn offline_agents_skip_mechanical_nodes_but_keep_reconcile() {
     );
 }
 
+#[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(),
+        parent_id: None,
+        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"
+    );
+}
+
 // ---- failure: cancel-downstream + AfterAny ----
 
 #[test]

hive-c0re/src/workers/auto_update.rs

@@ -288,12 +288,14 @@ pub async fn run(coord: Arc<Coordinator>) -> Result<()> {
     Ok(())
 }
 
-/// Submit this boot's DAGs under one `Boot` root: a `Noop` anchor with the
-/// startup sweep + per-agent reconciles hung off it via `parent_id`, so the
-/// dashboard renders the boot as a single tree instead of N+1 rows. No-op when
-/// there's nothing to do. The `parent_id` link is a display grouping, not a
-/// dependency edge — the children run concurrently, so the reconciles never
-/// wait behind the lock bump.
+/// Submit this boot's work as **one DAG** (no `boot_root` Noop anchor, no
+/// per-agent child DAGs). Node 0 is the sweep `MetaLock` (only when
+/// something is stale) — its executor bumps the hyperhive lock, then grows
+/// one rebuild subgraph per stale agent into *this same* DAG (rooted on the
+/// `MetaLock`, so they build against the post-bump lock; see
+/// `exec::run_meta_lock`). Every drifted agent gets a boot `Reconcile` as an
+/// independent root — a boot reconcile needs no lock bump, so it converges
+/// concurrently with the sweep. No-op when there's nothing to do.
 fn submit_boot_tree(
     coord: &Arc<Coordinator>,
     any_stale: bool,
@@ -302,59 +304,61 @@ fn submit_boot_tree(
     n_deferred: usize,
     n_skipped: usize,
 ) {
-    // Only emit a boot root when there's actually boot work — a fully-quiet
-    // boot (nothing stale, nothing drifted) submits nothing, exactly as before.
-    let boot_root_id = if any_stale || !drifted.is_empty() {
-        let reason = format!(
-            "boot: {} rebuild(s), {} reconcile(s), {} deferred (offline), {} up-to-date",
-            fanout.len(),
-            drifted.len(),
-            n_deferred,
-            n_skipped,
-        );
-        match coord
-            .job_queue
-            .submit(crate::job_queue::templates::boot_root(reason))
-        {
-            Ok(id) => Some(id),
-            Err(e) => {
-                tracing::warn!(error = ?e, "boot reconcile: boot-root submit failed");
-                None
-            }
-        }
-    } else {
-        None
-    };
+    use crate::job_queue::{DagSpec, NodeKind, NodeSpec, Source, Template};
 
+    // Fully-quiet boot (nothing stale, nothing drifted) submits nothing.
+    if !any_stale && drifted.is_empty() {
+        return;
+    }
+    let reason = format!(
+        "boot: {} rebuild(s), {} reconcile(s), {} deferred (offline), {} up-to-date",
+        fanout.len(),
+        drifted.len(),
+        n_deferred,
+        n_skipped,
+    );
+
+    let mut nodes: Vec<NodeSpec> = Vec::new();
     // Sweep whenever ANY marker is stale — even when every stale agent is
     // wanted-offline: the hyperhive lock bump must land now so their later
-    // start-upgrade rebuilds build against it. No stale agents ⇒ no sweep ⇒ no
-    // meta commit on a no-change boot.
+    // start-upgrade rebuilds build against it. No stale agents ⇒ no MetaLock
+    // ⇒ no meta commit on a no-change boot. The `fanout` list rides the
+    // MetaLock into `run_meta_lock`, which appends the rebuild subgraphs.
     if any_stale {
-        let reason = format!(
-            "startup sweep: {} rebuild(s), {} deferred (offline), {} up-to-date",
-            fanout.len(),
-            n_deferred,
-            n_skipped,
-        );
-        let mut spec = crate::job_queue::templates::startup_sweep(reason, fanout);
-        spec.parent_id = boot_root_id;
-        if let Err(e) = coord.job_queue.submit(spec) {
-            tracing::warn!(error = ?e, "boot reconcile: sweep submit failed");
-        }
+        nodes.push(NodeSpec {
+            agent: "hyperhive".to_owned(),
+            kind: NodeKind::MetaLock {
+                sweep: true,
+                fanout: Some(fanout),
+            },
+            deps: Vec::new(),
+        });
     }
+    // One boot Reconcile per drifted agent — independent roots.
     for name in drifted {
-        let mut spec = crate::job_queue::templates::reconcile_only(
-            crate::job_queue::Template::Reconcile,
-            &name,
-            crate::job_queue::Source::AutoUpdate,
-            "boot reconcile".to_owned(),
-            None,
-        );
-        spec.parent_id = boot_root_id;
-        if let Err(e) = coord.job_queue.submit(spec) {
-            tracing::warn!(%name, error = ?e, "boot reconcile: submit failed");
-        }
+        nodes.push(NodeSpec {
+            agent: name,
+            kind: NodeKind::Reconcile,
+            deps: Vec::new(),
+        });
+    }
+
+    let spec = DagSpec {
+        template: Template::Boot,
+        source: Source::AutoUpdate,
+        reason,
+        parent_id: None,
+        approval_id: None,
+        inputs: Vec::new(),
+        perm_payload: None,
+        // Rebuilding when the sweep will grow rebuild subgraphs (per-agent
+        // crash-watch suppression during their Swap, applied at claim time);
+        // a reconcile-only boot needs no transient.
+        transient: any_stale.then_some(crate::coordinator::TransientKind::Rebuilding),
+        nodes,
+    };
+    if let Err(e) = coord.job_queue.submit(spec) {
+        tracing::warn!(error = ?e, "boot: sweep DAG submit failed");
     }
     coord.emit_rebuild_queue_snapshot();
 }