fix(#2398): graceful restart as one atomic DAG, not compose-and-await

mara's review on #2436: no submit-await-submit composition, even
server-side. Adds Template::GracefulRestart (Signal -> Drain ->
StopForUpdate -> Reconcile, wanted=Up) mirroring how Restart already
does StopForUpdate -> Reconcile, plus submit::graceful_restart and
templates::graceful_restart. handle_restart_scoped now submits exactly
one DAG per agent up front for both the graceful and non-graceful
case -- no await_dags in the loop anymore.

hive-c0re/src/bin/hivectl.rs

@@ -1643,10 +1643,12 @@ async fn start(socket: &Path, scope: hive_host_sock::LifecycleScope, no_wait: bo
 }
 
 /// Restart — one `RestartScoped` daemon call, server-side DAG-based (see
-/// issue tracker "dagify hivectl commands"). Each targeted agent rides one
-/// atomic `Restart` DAG (or, with `--graceful`, a server-awaited
-/// graceful-stop DAG followed by a start DAG); infra containers restart
-/// synchronously. Unlike the old client-side stop-then-start compose, a
+/// issue tracker "dagify hivectl commands"). Each targeted agent rides
+/// exactly one atomic DAG queued up front — `Restart` (mechanical stop +
+/// reconcile), or `GracefulRestart` with `--graceful` (signal → drain →
+/// mechanical stop → reconcile); infra containers restart synchronously.
+/// No "submit one DAG, wait for it, submit another" composition on either
+/// side of the wire: unlike the old client-side stop-then-start compose, a
 /// dropped `hivectl` connection mid-restart no longer leaves an agent
 /// stopped with no automatic follow-up — the daemon owns the whole
 /// sequence once this call is made.

hive-c0re/src/job_queue/exec.rs

@@ -507,7 +507,11 @@ pub(super) async fn on_dag_terminal(coord: &Arc<Coordinator>, terminal: &Termina
     if terminal.state == State::Cancelled
         && matches!(
             terminal.template,
-            Template::Start | Template::Stop | Template::GracefulStop | Template::Restart
+            Template::Start
+                | Template::Stop
+                | Template::GracefulStop
+                | Template::Restart
+                | Template::GracefulRestart
         )
     {
         let running = crate::lifecycle::is_running(&terminal.agent).await;

hive-c0re/src/job_queue/submit.rs

@@ -100,6 +100,19 @@ pub fn graceful_stop(coord: &Arc<Coordinator>, agent: &str, source: Source, reas
     submit_and_emit(coord, templates::graceful_stop(agent, source, reason))
 }
 
+/// Graceful restart: persist `wanted = Up`, then signal → drain →
+/// mechanical stop → reconcile (starts it back up) — one atomic DAG,
+/// no client-side "await the stop DAG then submit a start DAG" split.
+pub fn graceful_restart(
+    coord: &Arc<Coordinator>,
+    agent: &str,
+    source: Source,
+    reason: String,
+) -> u64 {
+    set_wanted(coord, agent, Wanted::Up);
+    submit_and_emit(coord, templates::graceful_restart(agent, source, reason))
+}
+
 /// Perm change: commit the JSON file(s) then rebuild.
 pub fn perm_change(
     coord: &Arc<Coordinator>,

hive-c0re/src/job_queue/templates.rs

@@ -7,6 +7,7 @@
 //! rebuild(a):        Prebuild(a) → StopForUpdate(a) → Swap(a) →(any) Reconcile(a)
 //! graceful-stop(a):  [wanted=Offline]  Signal(a) → Drain(a) → Reconcile(a)
 //! restart(a):        [wanted=Up]       StopForUpdate(a) → Reconcile(a)
+//! graceful-restart(a): [wanted=Up]     Signal(a) → Drain(a) → StopForUpdate(a) → Reconcile(a)
 //! start(a):          [wanted=Up]       Reconcile(a)
 //! stop(a):           [wanted=Offline]  Reconcile(a)
 //! spawn(a):          [wanted=Up]  Provision(a) → Create(a) → WriteDropin(a) → Reconcile(a)
@@ -168,6 +169,46 @@ pub fn restart(agent: &str, source: Source, reason: String) -> DagSpec {
     }
 }
 
+/// Graceful restart: signal → drain → mechanical stop → converge to
+/// `wanted` — the caller writes `wanted = Up` first, same as `restart`.
+/// 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,
+        agent: agent.to_owned(),
+        source,
+        reason,
+        parent_id: None,
+        approval_id: None,
+        inputs: Vec::new(),
+        perm_payload: None,
+        transient: Some(TransientKind::Restarting),
+        nodes: vec![
+            NodeSpec {
+                kind: NodeKind::Signal,
+                deps: Vec::new(),
+            },
+            NodeSpec {
+                kind: NodeKind::Drain,
+                deps: after_ok(0),
+            },
+            NodeSpec {
+                kind: NodeKind::StopForUpdate,
+                deps: after_ok(1),
+            },
+            NodeSpec {
+                kind: NodeKind::Reconcile,
+                deps: after_ok(2),
+            },
+        ],
+    }
+}
+
 /// Single-`Reconcile` DAG: `Start` / `Stop` (caller writes `wanted`
 /// first) and the boot-time `Reconcile` converge (wanted untouched).
 pub fn reconcile_only(

hive-c0re/src/server.rs

@@ -657,39 +657,30 @@ async fn handle_restart_scoped(
     let mut errors: Vec<String> = Vec::new();
     let mut queued: Vec<u64> = Vec::new();
 
-    if graceful {
-        let mut stop_ids = Vec::new();
-        for agent in &agents {
-            stop_ids.push(crate::job_queue::submit::graceful_stop(
+    // One atomic DAG per agent, submitted up front — `Restart`
+    // (mechanical stop + reconcile) or, with `--graceful`,
+    // `GracefulRestart` (signal → drain → mechanical stop → reconcile).
+    // No client- or server-side "submit a stop DAG, await it, then
+    // submit a start DAG" composition: that window is exactly the
+    // dropped-connection gap this DAG-based path exists to close.
+    for agent in &agents {
+        let id = if graceful {
+            crate::job_queue::submit::graceful_restart(
                 coord,
                 agent,
                 crate::job_queue::Source::Manual,
                 "manual via hivectl restart --graceful".to_owned(),
-            ));
-        }
-        // Graceful drains can take a while; bound the wait generously —
-        // same shape as `handle_stop`'s hard-stop-before-infra wait, just
-        // a longer ceiling since drains are minutes, not seconds.
-        await_dags(coord, &stop_ids, std::time::Duration::from_mins(10)).await;
-        for agent in &agents {
-            queued.push(crate::job_queue::submit::start(
-                coord,
-                agent,
-                crate::job_queue::Source::Manual,
-                "manual via hivectl restart --graceful".to_owned(),
-            ));
-            ok_items.push(agent.clone());
-        }
-    } else {
-        for agent in &agents {
-            queued.push(crate::job_queue::submit::restart(
+            )
+        } else {
+            crate::job_queue::submit::restart(
                 coord,
                 agent,
                 crate::job_queue::Source::Manual,
                 "manual restart via hivectl restart".to_owned(),
-            ));
-            ok_items.push(agent.clone());
-        }
+            )
+        };
+        queued.push(id);
+        ok_items.push(agent.clone());
     }
 
     for &container in &infra {