feat(#2448): emit one multi-agent DAG for hivectl restart / restart-all

A hive-wide restart was N separate single-agent DAGs (one submit::restart
per agent). Now that agent is per-node (#2445), make it ONE DAG with a
per-agent restart subgraph each.

- templates::restart takes an agent list: each agent gets an independent
  subgraph (a head SetWanted(Up) root, then its restart chain), so the N
  subgraphs run concurrently on their own leases. One agent = the ordinary
  single-agent restart; unifies the old restart + graceful_restart fns.
- submit::restart / graceful_restart stay as single-agent wrappers over
  the new submit::restart_many(agents, graceful).
- server.rs handle_restart_all + handle_restart_scoped submit one
  restart_many call instead of looping per agent. Infra containers
  unchanged (no lease/DAG, synchronous).

Scope: restart + restart-all only. Broad stop+start is the same pattern
(stop/start templates take agent lists) — a follow-up increment.

hive-c0re/src/job_queue/submit.rs

@@ -27,12 +27,27 @@ pub fn rebuild(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: St
     submit_and_emit(coord, templates::rebuild(agent, source, reason, None, true))
 }
 
-/// Restart: mechanical stop + converge to `wanted = Up`. The intent
+/// Restart a single agent: mechanical stop + converge to `wanted = Up`.
-/// write is the template's head `SetWanted(Up)` node — it matters when
+/// The intent write is the template's head `SetWanted(Up)` node — it
-/// `wanted` drifted `Offline` under a running agent (an operator asking
+/// matters when `wanted` drifted `Offline` under a running agent (an
-/// for a restart plainly wants it running, not a stop).
+/// operator asking for a restart plainly wants it running, not a stop).
+/// Thin wrapper over [`restart_many`] with a one-agent slice.
 pub fn restart(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: String) -> u64 {
-    submit_and_emit(coord, templates::restart(agent, source, reason))
+    restart_many(coord, &[agent.to_owned()], false, source, reason)
+}
+
+/// Restart `agents` (one or many) in a **single** DAG — one per-agent
+/// subgraph each, running concurrently on their own leases. `graceful`
+/// prepends the signal→drain quiesce per agent. The whole hive-wide
+/// `hivectl restart` / `restart-all` is now one DAG instead of N.
+pub fn restart_many(
+    coord: &Arc<Coordinator>,
+    agents: &[String],
+    graceful: bool,
+    source: Source,
+    reason: String,
+) -> u64 {
+    submit_and_emit(coord, templates::restart(agents, graceful, source, reason))
 }
 
 /// Start: `SetWanted(Up)` (a DAG node now) then reconcile. A stale rev
@@ -67,17 +82,18 @@ pub fn graceful_stop(coord: &Arc<Coordinator>, agent: &str, source: Source, reas
     submit_and_emit(coord, templates::graceful_stop(agent, source, reason))
 }
 
-/// Graceful restart: signal → drain → mechanical stop → reconcile (starts
+/// Graceful restart of a single agent: signal → drain → mechanical stop →
-/// it back up) — one atomic DAG, no client-side "await the stop DAG then
+/// reconcile (starts it back up) — one atomic DAG, no client-side "await
-/// submit a start DAG" split. The head `SetWanted(Up)` node writes the
+/// the stop DAG then submit a start DAG" split. The head `SetWanted(Up)`
-/// intent as part of the DAG.
+/// node writes the intent as part of the DAG. Thin wrapper over
+/// [`restart_many`] with `graceful = true`.
 pub fn graceful_restart(
     coord: &Arc<Coordinator>,
     agent: &str,
     source: Source,
     reason: String,
 ) -> u64 {
-    submit_and_emit(coord, templates::graceful_restart(agent, source, reason))
+    restart_many(coord, &[agent.to_owned()], true, source, reason)
 }
 
 /// Perm change: commit the JSON file(s) then rebuild.

hive-c0re/src/job_queue/templates.rs

@@ -4,9 +4,11 @@
 //! `Vec<Node>` + `deps`).
 //!
 //! Every node carries its own `agent` (there is no DAG-level agent) — the
-//! `node` helper stamps the template's agent onto each. Today's templates
+//! `node` helper stamps the template's agent onto each. `restart` takes an
-//! are single-agent (every node shares one agent); a future multi-agent
+//! agent *list* and stamps each agent onto its own subgraph, so a hive-wide
-//! template would stamp different agents per subgraph.
+//! `hivectl restart` is ONE DAG with N independent per-agent subgraphs
+//! (each a root chain, run concurrently on its own lease) rather than N
+//! separate DAGs. The other templates are still single-agent.
 //!
 //! The power ops write the durable `wanted` intent via a head
 //! `SetWanted(w)` node (not a pre-submit side effect); it holds the agent
@@ -146,37 +148,36 @@ pub fn graceful_stop(agent: &str, source: Source, reason: String) -> DagSpec {
     }
 }
 
-/// Restart: write `wanted = Up` (head `SetWanted` node), mechanical stop,
+/// Restart one or more agents in a **single** DAG. Each agent gets an
-/// then converge — a stop + start like the old `lifecycle::restart`
+/// independent subgraph — a head `SetWanted(Up)` (a root: no cross-agent
-/// regardless of prior intent drift.
+/// dep) then its restart chain — so all agents' restarts run concurrently,
-pub fn restart(agent: &str, source: Source, reason: String) -> DagSpec {
+/// each taking its own agent lease. `graceful` inserts `Signal → Drain`
-    DagSpec {
+/// before the mechanical `StopForUpdate` (each agent's subgraph is still one
-        template: Template::Restart,
+/// atomic restart). One agent = the ordinary single-agent restart; many =
-        source,
+/// a hive-wide `hivectl restart` as one DAG instead of N separate ones.
-        reason,
+pub fn restart(agents: &[String], graceful: bool, source: Source, reason: String) -> DagSpec {
-        parent_id: None,
+    let mut nodes = Vec::new();
-        approval_id: None,
+    for agent in agents {
-        inputs: Vec::new(),
+        let base = u32::try_from(nodes.len()).unwrap_or(u32::MAX);
-        perm_payload: None,
+        // Head of this agent's subgraph — a root (empty deps), so the N
-        transient: Some(TransientKind::Restarting),
+        // per-agent subgraphs are independent and run concurrently.
-        nodes: vec![
+        nodes.push(node(agent, NodeKind::SetWanted { up: true }, Vec::new()));
-            node(agent, NodeKind::SetWanted { up: true }, Vec::new()),
+        if graceful {
-            node(agent, NodeKind::StopForUpdate, after_ok(0)),
+            nodes.push(node(agent, NodeKind::Signal, after_ok(base)));
-            node(agent, NodeKind::Reconcile, after_ok(1)),
+            nodes.push(node(agent, NodeKind::Drain, after_ok(base + 1)));
-        ],
+            nodes.push(node(agent, NodeKind::StopForUpdate, after_ok(base + 2)));
+            nodes.push(node(agent, NodeKind::Reconcile, after_ok(base + 3)));
+        } else {
+            nodes.push(node(agent, NodeKind::StopForUpdate, after_ok(base)));
+            nodes.push(node(agent, NodeKind::Reconcile, after_ok(base + 1)));
+        }
     }
-}
-
-/// Graceful restart: write `wanted = Up` (head `SetWanted`), signal →
-/// drain → mechanical stop → converge. One atomic DAG start to finish
-/// (no client- or server-side "submit one DAG, await it, submit the next"
-/// composition): the `Drain` node is the same bounded harness-checkpoint
-/// wait `graceful_stop` uses, then `StopForUpdate` (mechanical, ignores
-/// `wanted`) and the tail `Reconcile` (converges to `wanted = Up`, i.e.
-/// starts it back up) chain exactly like `restart`'s tail.
-pub fn graceful_restart(agent: &str, source: Source, reason: String) -> DagSpec {
     DagSpec {
-        template: Template::GracefulRestart,
+        template: if graceful {
+            Template::GracefulRestart
+        } else {
+            Template::Restart
+        },
         source,
         reason,
         parent_id: None,
@@ -184,13 +185,7 @@ pub fn graceful_restart(agent: &str, source: Source, reason: String) -> DagSpec
         inputs: Vec::new(),
         perm_payload: None,
         transient: Some(TransientKind::Restarting),
-        nodes: vec![
+        nodes,
-            node(agent, NodeKind::SetWanted { up: true }, Vec::new()),
-            node(agent, NodeKind::Signal, after_ok(0)),
-            node(agent, NodeKind::Drain, after_ok(1)),
-            node(agent, NodeKind::StopForUpdate, after_ok(2)),
-            node(agent, NodeKind::Reconcile, after_ok(3)),
-        ],
     }
 }
 

hive-c0re/src/server.rs

@@ -552,28 +552,29 @@ fn submit_single(coord: &Arc<Coordinator>, name: &str, verb: Verb) -> HostRespon
     HostResponse::queued(vec![id])
 }
 
-/// Restart every container by submitting one restart DAG per agent —
+/// Restart every container in **one** DAG — a per-agent restart subgraph
-/// each serializes on its own lease, so unrelated agents' restarts
+/// each, running concurrently on their own leases (so unrelated agents'
-/// overlap while nothing races an in-flight rebuild. Returns once all
+/// restarts overlap while nothing races an in-flight rebuild). Returns
-/// are queued; per-agent results surface on the queue.
+/// once queued; per-node progress surfaces on the single DAG.
 async fn handle_restart_all(coord: &Arc<Coordinator>) -> Result<HostResponse> {
     tracing::info!("restart-all");
-    let agents = lifecycle::list().await?;
+    let containers = lifecycle::list().await?;
-    let mut ok_agents: Vec<String> = Vec::new();
+    let agents: Vec<String> = containers
-    let mut queued: Vec<u64> = Vec::new();
+        .iter()
-    for agent in &agents {
+        .filter_map(|a| a.strip_prefix(lifecycle::AGENT_PREFIX).map(str::to_owned))
-        let Some(logical) = agent.strip_prefix(lifecycle::AGENT_PREFIX) else {
+        .collect();
-            continue;
+    let queued = if agents.is_empty() {
-        };
+        Vec::new()
-        queued.push(crate::job_queue::submit::restart(
+    } else {
+        vec![crate::job_queue::submit::restart_many(
             coord,
-            logical,
+            &agents,
+            false,
             crate::job_queue::Source::Manual,
             "manual restart via hivectl restart-all".to_owned(),
-        ));
+        )]
-        ok_agents.push(logical.to_owned());
+    };
-    }
+    let mut resp = HostResponse::list(agents);
-    let mut resp = HostResponse::list(ok_agents);
     resp.queued_dags = Some(queued);
     Ok(resp)
 }
@@ -718,16 +719,14 @@ async fn handle_start(
 }
 
 /// Restart containers hive-wide (`hivectl restart`) — the DAG-based
-/// sibling of [`handle_stop`]/[`handle_start`], replacing the old
+/// sibling of [`handle_stop`]/[`handle_start`]. All targeted agents ride
-/// client-side stop-then-start composition (issue tracker "dagify hivectl
+/// **one** DAG (a per-agent restart subgraph each: `SetWanted → [Signal →
-/// commands"). Each targeted agent gets exactly one atomic DAG submitted
+/// Drain →] StopForUpdate → Reconcile`, independent roots that run
-/// up front: the `Restart` template (mechanical stop + reconcile) in the
+/// concurrently on their own leases), not N separate DAGs — a hive-wide
-/// common case, or `GracefulRestart` (signal → drain → mechanical stop →
+/// restart is one job. `graceful` prepends signal→drain per agent. No
-/// reconcile) with `graceful` set — no "submit a DAG, wait for it, submit
+/// client-side stop-then-start composition, so a dropped `hivectl`
-/// another" composition on either path, so a dropped `hivectl` connection
+/// connection never strands an agent. Infra containers have no lease/DAG
-/// never strands an agent, the same way `handle_restart_all` already
+/// and restart synchronously (stop then start).
-/// avoids it. Infra containers have no lease/DAG and restart
-/// synchronously (stop then start).
 async fn handle_restart_scoped(
     coord: &Arc<Coordinator>,
     scope: &LifecycleScope,
@@ -740,30 +739,25 @@ async fn handle_restart_scoped(
     let mut errors: Vec<String> = Vec::new();
     let mut queued: Vec<u64> = Vec::new();
 
-    // One atomic DAG per agent, submitted up front — `Restart`
+    // One DAG for all targeted agents — a per-agent restart subgraph each
-    // (mechanical stop + reconcile) or, with `--graceful`,
+    // (`SetWanted → [Signal → Drain →] StopForUpdate → Reconcile`),
-    // `GracefulRestart` (signal → drain → mechanical stop → reconcile).
+    // independent roots that run concurrently on their own leases. A
-    // No client- or server-side "submit a stop DAG, await it, then
+    // hive-wide `hivectl restart` is now a single DAG, not N. No
-    // submit a start DAG" composition: that window is exactly the
+    // client-side stop-then-start composition — the whole restart survives
-    // dropped-connection gap this DAG-based path exists to close.
+    // a dropped connection because the DAG owns it.
-    for agent in &agents {
+    if !agents.is_empty() {
-        let id = if graceful {
+        queued.push(crate::job_queue::submit::restart_many(
-            crate::job_queue::submit::graceful_restart(
+            coord,
-                coord,
+            &agents,
-                agent,
+            graceful,
-                crate::job_queue::Source::Manual,
+            crate::job_queue::Source::Manual,
-                "manual via hivectl restart --graceful".to_owned(),
+            if graceful {
-            )
+                "manual via hivectl restart --graceful".to_owned()
-        } else {
+            } else {
-            crate::job_queue::submit::restart(
+                "manual restart via hivectl restart".to_owned()
-                coord,
+            },
-                agent,
+        ));
-                crate::job_queue::Source::Manual,
+        ok_items.extend(agents.iter().cloned());
-                "manual restart via hivectl restart".to_owned(),
-            )
-        };
-        queued.push(id);
-        ok_items.push(agent.clone());
     }
 
     for &container in &infra {