diff --git a/docs/coordinator.md b/docs/coordinator.md
index 3cf09fdb..b02ed1ed 100644
--- a/docs/coordinator.md
+++ b/docs/coordinator.md
@@ -52,16 +52,32 @@ Cheap — no build slot:
 
 There is deliberately **no `GitCommit` node**: `meta.rs` fuses each mutation
 with its commit under its internal `META_LOCK` mutex, so a standalone commit
-node would open a dirty-working-tree window between nodes. That same
-`META_LOCK` is also why the scheduler needs no meta-repo resource class — any
-executor touching the meta repo serializes inside `meta.rs`.
+node would open a dirty-working-tree window between nodes.
+
+Two further layers protect the meta repo across *windows* that span multiple
+`META_LOCK` acquisitions — above all the approval deploy's prepare→finalize
+span, which keeps a bumped `flake.lock` **staged uncommitted** for the whole
+container build:
+
+- **The deploy-window gate** (`meta::exclusive()`): every executor that
+  mutates the meta repo (`Prebuild`'s sync+relock, `MetaLock`,
+  `WritePermFile`, `Create`'s agent registration, and `ApprovalDeploy` for
+  its whole span) holds this async mutex for its mutation span, so no commit
+  can land inside another node's staged window. `Prebuild` drops it before
+  the long toplevel build (store reads only), preserving `buildSlots > 1`
+  concurrency.
+- **Path-limited commits**: the targeted meta committers (perm files,
+  topology, lock bumps, finalize) commit `-- <their paths>` with path-scoped
+  dirty checks, so even a non-queue caller (boot migration, destroy's
+  `sync_agents`) can never sweep someone else's staged content into its
+  commit.
 
 ### Every operation as a DAG
 
 ```text
 rebuild(a):        Prebuild(a) → StopForUpdate(a) → Swap(a) →(after-any) Reconcile(a)
 graceful-stop(a):  [wanted=Offline]  Signal(a) → Drain(a) → Reconcile(a)
-restart(a):        StopForUpdate(a) → Reconcile(a)          (wanted unchanged)
+restart(a):        [wanted=Up]       StopForUpdate(a) → Reconcile(a)
 start(a):          [wanted=Up]       Reconcile(a)     (stale rev ⇒ upgraded to rebuild)
 stop(a):           [wanted=Offline]  Reconcile(a)
 spawn(a):          [wanted=Up]  Create(a) → WriteDropin(a) → Reconcile(a)
diff --git a/frontend/packages/dashboard/src/swarm.js b/frontend/packages/dashboard/src/swarm.js
index ce2014d0..27c2a4a9 100644
--- a/frontend/packages/dashboard/src/swarm.js
+++ b/frontend/packages/dashboard/src/swarm.js
@@ -811,6 +811,7 @@ export function renderContainers(s) {
                 : op.kind === 'start'       ? 'starting'
                 : op.kind === 'stop'        ? 'stopping'
                 : op.kind === 'graceful_stop' ? 'stopping'
+                : op.kind === 'reconcile'   ? 'reconciling'
                 : 'rebuilding')
             : (op.kind === 'meta_update'    ? 'meta-update queued'
                 : op.kind === 'destroy'     ? 'destroy queued'
@@ -818,6 +819,7 @@ export function renderContainers(s) {
                 : op.kind === 'start'       ? 'start queued'
                 : op.kind === 'stop'        ? 'stop queued'
                 : op.kind === 'graceful_stop' ? 'stop queued'
+                : op.kind === 'reconcile'   ? 'reconcile queued'
                 : 'rebuild queued')));
     const opRunning = transientKind != null
       || (op != null && op.state === 'running');
diff --git a/hive-c0re/src/audit_log.rs b/hive-c0re/src/audit_log.rs
index c145dde9..d6a26153 100644
--- a/hive-c0re/src/audit_log.rs
+++ b/hive-c0re/src/audit_log.rs
@@ -119,11 +119,8 @@ impl AuditLog {
     /// Returns an error if the directory can't be created, the sqlite
     /// file can't be opened, or applying the schema fails.
     pub fn open(db_dir: &Path) -> Result<Self> {
-        std::fs::create_dir_all(db_dir)
-            .with_context(|| format!("create audit_log db parent {}", db_dir.display()))?;
         let path = db_dir.join("audit_log.sqlite");
-        let conn = Connection::open(&path)
-            .with_context(|| format!("open audit_log db {}", path.display()))?;
+        let conn = crate::db::open(&path, "audit_log")?;
         conn.execute_batch(SCHEMA)
             .context("apply audit_log schema")?;
         Ok(Self {
diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs
index 7c316243..93cb974f 100644
--- a/hive-c0re/src/job_queue/exec.rs
+++ b/hive-c0re/src/job_queue/exec.rs
@@ -106,11 +106,17 @@ async fn run_prebuild(
     // Idempotent meta sync so a manual rebuild can also recover from a
     // divergent meta repo; then bump just this agent's input. `relock =
     // false` only for meta-update cascade children, where re-locking
-    // would revert the bump the cascade just committed.
-    let agents = crate::lifecycle::agents_for_meta_listing().await?;
-    crate::meta::sync_agents(&hive, &agents).await?;
-    if relock {
-        crate::meta::lock_update_for_rebuild(name).await?;
+    // would revert the bump the cascade just committed. Both run under
+    // the deploy-window gate so they can never land inside another
+    // node's staged prepare→finalize window; the gate drops before the
+    // (long) toplevel build, which only reads the store.
+    {
+        let _window = crate::meta::exclusive().await;
+        let agents = crate::lifecycle::agents_for_meta_listing().await?;
+        crate::meta::sync_agents(&hive, &agents).await?;
+        if relock {
+            crate::meta::lock_update_for_rebuild(name).await?;
+        }
     }
     ctx.step("nix build");
     let flake_ref = format!("{}#{name}", crate::meta::meta_dir().display());
@@ -140,13 +146,10 @@ async fn run_swap(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Res
             {
                 tracing::warn!(%name, error = ?e, "write rev marker failed");
             }
-            coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt {
-                agent: name.clone(),
-                ok: true,
-                note: None,
-                sha: None,
-                tag: None,
-            });
+            // The `Rebuilt` manager event fires exactly once per DAG
+            // from the terminal hook — emitting ok here and letting a
+            // failed tail `Reconcile` add a contradictory !ok would
+            // double-report the same rebuild.
             ctx.step("forge sync");
             // Full forge + matrix sync on every successful rebuild so
             // the rebuild path is equivalent to the startup sweep:
@@ -179,6 +182,11 @@ async fn run_create(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> R
     let hive = coord.hive_env();
     let paths = Coordinator::agent_paths(name, agent_dir);
     ctx.step("nixos-container create");
+    // create_container registers the new agent in the meta flake
+    // (sync_agents commit) before `nixos-container create` — hold the
+    // deploy-window gate so that commit can't land inside another
+    // node's staged deploy window.
+    let _window = crate::meta::exclusive().await;
     crate::lifecycle::create_container(name, &hive, &paths).await?;
     Ok(NodeOutput::default())
 }
@@ -196,6 +204,7 @@ async fn run_meta_lock(
 ) -> Result<NodeOutput> {
     if sweep {
         ctx.step("nix flake update hyperhive");
+        let _window = crate::meta::exclusive().await;
         if let Err(e) = crate::meta::lock_update_hyperhive().await {
             tracing::warn!(error = ?e, "startup sweep: meta lock_update_hyperhive failed");
         }
@@ -205,7 +214,10 @@ async fn run_meta_lock(
     }
     let _progress = coord.meta_update_guard();
     ctx.step("nix flake update");
-    crate::meta::lock_update(&claim.inputs).await?;
+    {
+        let _window = crate::meta::exclusive().await;
+        crate::meta::lock_update(&claim.inputs).await?;
+    }
     // Lock file changed — meta-inputs panel re-renders.
     crate::dashboard::emit_meta_inputs_snapshot(coord);
     let cascade = match fanout {
@@ -258,8 +270,8 @@ async fn run_reconcile(
     Ok(NodeOutput::default())
 }
 
-/// Mechanical stop for the profile swap. Never touches `wanted`; noop
-/// when already stopped.
+/// Mechanical stop for the profile swap. Never *changes* `wanted`;
+/// noop when already stopped.
 async fn run_stop_for_update(
     coord: &Arc<Coordinator>,
     claim: &Claim,
@@ -267,6 +279,13 @@ async fn run_stop_for_update(
 ) -> Result<NodeOutput> {
     let name = &claim.agent;
     if crate::lifecycle::is_running(name).await {
+        // Seed a missing agent_power row from the PRE-stop observation
+        // — the DAG's tail `Reconcile` observes only the mechanically
+        // stopped state and would otherwise seed a running-but-unknown
+        // agent as `Offline`, stranding it down after its own rebuild.
+        if let Err(e) = coord.power.get_or_seed(name, true) {
+            tracing::warn!(%name, error = ?e, "agent_power: pre-stop seed failed");
+        }
         ctx.step("nixos-container stop");
         crate::lifecycle::kill(name).await?;
         coord.rescan_containers_and_emit().await;
@@ -322,6 +341,11 @@ async fn run_write_perm_file(
     use super::model::PermPayload;
     let name = &claim.agent;
     ctx.step("writing + committing perm file");
+    // Deploy-window gate: a perm commit landing inside another node's
+    // staged prepare→finalize window would sweep the staged deploy
+    // lock into its commit (the commits are also path-limited in
+    // meta.rs — belt and braces).
+    let _window = crate::meta::exclusive().await;
     match &claim.perm_payload {
         Some(PermPayload::ToolGroups { groups }) => {
             crate::meta::commit_tool_groups(name, groups)
@@ -363,6 +387,11 @@ async fn run_approval_deploy(coord: &Arc<Coordinator>, claim: &Claim) -> Result<
     let approval_id = claim
         .approval_id
         .with_context(|| format!("approval_deploy dag {} has no approval_id", claim.dag_id))?;
+    // Hold the deploy-window gate for the whole prepare→finalize span:
+    // `prepare_deploy` stages `flake.lock` uncommitted for the entire
+    // container build, and no other meta mutation may land inside that
+    // window (it would sweep the staged lock and neuter `abort_deploy`).
+    let _window = crate::meta::exclusive().await;
     let kind = coord
         .approvals
         .get(approval_id)
@@ -377,26 +406,55 @@ async fn run_approval_deploy(coord: &Arc<Coordinator>, claim: &Claim) -> Result<
     result.map(|()| NodeOutput::default())
 }
 
-/// Terminal-roll-up hook, fired exactly once per DAG. Approval DAGs
-/// resolve their approval row (except the opaque deploy pipeline,
-/// which resolves inside its node); non-approval rebuild-shaped DAGs
-/// surface the `Rebuilt { ok: false }` manager event on failure —
-/// success fires from the `Swap` tail, matching today's timing.
+/// Terminal-roll-up hook, fired exactly once per DAG (node completion
+/// and cancel paths alike — the queue buffers roll-ups and the
+/// scheduler drains them). Three concerns:
+/// - approval DAGs resolve their approval row (except the opaque
+///   deploy pipeline, which resolves inside its node — unless it was
+///   cancelled while still queued and the node never ran);
+/// - non-approval rebuild-shaped DAGs emit exactly one `Rebuilt`
+///   manager event: ok on `Done`, !ok on `Failed`, none on cancel;
+/// - a cancelled power-op DAG reverts the `wanted` intent its submit
+///   wrote: the operator's cancel means "don't do it", so intent
+///   snaps back to the observed state instead of the flip executing
+///   as a surprise side effect of some later reconcile.
 pub(super) async fn on_dag_terminal(coord: &Arc<Coordinator>, terminal: &TerminalDag) {
+    if terminal.state == State::Cancelled
+        && matches!(
+            terminal.template,
+            Template::Start | Template::Stop | Template::GracefulStop | Template::Restart
+        )
+    {
+        let running = crate::lifecycle::is_running(&terminal.agent).await;
+        if let Err(e) = coord
+            .power
+            .set(&terminal.agent, crate::power::Wanted::from_running(running))
+        {
+            tracing::warn!(agent = %terminal.agent, error = ?e, "agent_power: cancel revert failed");
+        }
+    }
     if terminal.approval_id.is_some() {
         crate::actions::resolve_approval_dag(coord, terminal).await;
         return;
     }
-    if matches!(terminal.template, Template::Rebuild | Template::PermChange)
-        && terminal.state == State::Failed
-    {
-        coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt {
-            agent: terminal.agent.clone(),
-            ok: false,
-            note: terminal.error.clone(),
-            sha: None,
-            tag: None,
-        });
+    if matches!(terminal.template, Template::Rebuild | Template::PermChange) {
+        match terminal.state {
+            State::Done => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt {
+                agent: terminal.agent.clone(),
+                ok: true,
+                note: None,
+                sha: None,
+                tag: None,
+            }),
+            State::Failed => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt {
+                agent: terminal.agent.clone(),
+                ok: false,
+                note: terminal.error.clone(),
+                sha: None,
+                tag: None,
+            }),
+            _ => {}
+        }
     }
 }
 
diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs
index 9aa91d07..1db5d53d 100644
--- a/hive-c0re/src/job_queue/mod.rs
+++ b/hive-c0re/src/job_queue/mod.rs
@@ -76,15 +76,6 @@ pub struct TerminalDag {
     pub error: Option<String>,
 }
 
-/// Report from [`JobQueue::complete_node`].
-#[derive(Debug, Default)]
-pub struct CompletionReport {
-    /// DAGs that became terminal as a result of this completion
-    /// (the completed node's own DAG, plus none others — but kept as a
-    /// Vec so cancel paths can reuse the same settle plumbing).
-    pub terminal: Vec<TerminalDag>,
-}
-
 #[derive(Debug, Default)]
 struct Inner {
     dags: VecDeque<Dag>,
@@ -93,6 +84,12 @@ struct Inner {
     slots_used: usize,
     /// agent → dag id currently holding that agent's lifecycle lease.
     leases: HashMap<String, u64>,
+    /// Terminal roll-ups not yet consumed by the scheduler
+    /// ([`JobQueue::drain_terminal`]). Fed by every path that settles
+    /// state — node completion AND the cancel surfaces — so the
+    /// terminal hooks (approval resolution, intent revert, transient
+    /// release) fire exactly once per DAG no matter how it ended.
+    pending_terminal: Vec<TerminalDag>,
 }
 
 /// The queue. Lives on `Coordinator` (one per hive-c0re process); a
@@ -324,13 +321,9 @@ impl JobQueue {
 
     /// Mark a claimed node terminal, release its build slot, cascade
     /// cancellations, and settle terminal DAGs (lease release + history
-    /// trim). `error` is stored (truncated) when `result` is `Err`.
-    pub fn complete_node(
-        &self,
-        dag_id: u64,
-        node_id: NodeId,
-        result: Result<(), String>,
-    ) -> CompletionReport {
+    /// trim; the terminal roll-up lands in the [`Self::drain_terminal`]
+    /// buffer). `error` is stored (truncated) when `result` is `Err`.
+    pub fn complete_node(&self, dag_id: u64, node_id: NodeId, result: Result<(), String>) {
         let mut inner = self.inner.lock().expect("job_queue mutex poisoned");
         if let Some(dag) = inner.dags.iter_mut().find(|d| d.id == dag_id)
             && let Some(node) = dag.node_mut(node_id)
@@ -360,21 +353,27 @@ impl JobQueue {
                 inner.slots_used = inner.slots_used.saturating_sub(1);
             }
         }
-        let report = Self::settle(&mut inner);
+        Self::settle(&mut inner);
         drop(inner);
         self.notify.notify_one();
-        report
+    }
+
+    /// Take the terminal roll-ups accumulated since the last drain.
+    /// The scheduler calls this after every wakeup and runs the
+    /// terminal hooks on each entry.
+    pub fn drain_terminal(&self) -> Vec<TerminalDag> {
+        let mut inner = self.inner.lock().expect("job_queue mutex poisoned");
+        std::mem::take(&mut inner.pending_terminal)
     }
 
     /// Propagate cancellations, release the leases of newly-terminal
-    /// DAGs, and trim history. Each terminal DAG is reported exactly
-    /// once (the `terminal_reported` flag) so the scheduler's hooks —
-    /// approval resolution, transient-guard release — fire once per
-    /// DAG.
-    fn settle(inner: &mut Inner) -> CompletionReport {
+    /// DAGs, buffer each terminal roll-up exactly once (the
+    /// `terminal_reported` flag) for [`Self::drain_terminal`], and trim
+    /// history.
+    fn settle(inner: &mut Inner) {
         Self::propagate_cancellations(inner);
-        let mut report = CompletionReport::default();
         let mut freed: Vec<String> = Vec::new();
+        let mut reports: Vec<TerminalDag> = Vec::new();
         for dag in &mut inner.dags {
             if !dag.is_terminal() || dag.terminal_reported {
                 continue;
@@ -383,7 +382,7 @@ impl JobQueue {
             if inner.leases.get(dag.agent.as_str()) == Some(&dag.id) {
                 freed.push(dag.agent.clone());
             }
-            report.terminal.push(TerminalDag {
+            reports.push(TerminalDag {
                 dag_id: dag.id,
                 template: dag.template,
                 agent: dag.agent.clone(),
@@ -392,11 +391,11 @@ impl JobQueue {
                 error: dag.first_error().map(str::to_owned),
             });
         }
+        inner.pending_terminal.append(&mut reports);
         for agent in freed {
             inner.leases.remove(&agent);
         }
         Self::trim_history(inner);
-        report
     }
 
     /// Cancel a DAG that hasn't started yet (roll-up `Queued`): every
@@ -416,7 +415,10 @@ impl JobQueue {
             n.state = State::Cancelled;
             n.finished_at = Some(now);
         }
-        let _ = Self::settle(&mut inner);
+        // Settle buffers the terminal roll-up; the notify wakes the
+        // scheduler, which drains it and fires the terminal hooks
+        // (approval resolution, power-intent revert).
+        Self::settle(&mut inner);
         drop(inner);
         self.notify.notify_one();
         true
@@ -438,7 +440,7 @@ impl JobQueue {
             }
         }
         if count > 0 {
-            let _ = Self::settle(&mut inner);
+            Self::settle(&mut inner);
             drop(inner);
             self.notify.notify_one();
         }
@@ -534,15 +536,24 @@ impl JobQueue {
     }
 
     /// Keep only the newest `MAX_HISTORY_PER_TEMPLATE` terminal DAGs
-    /// per template; live DAGs are never evicted.
+    /// per template. Live DAGs are never evicted — and neither is a
+    /// terminal parent that still has live children (a fan-out parent
+    /// is terminal the moment its `MetaLock` completes; evicting it
+    /// while cascade rebuilds run would orphan their dashboard group).
     fn trim_history(inner: &mut Inner) {
+        let live_parents: std::collections::HashSet<u64> = inner
+            .dags
+            .iter()
+            .filter(|d| !d.is_terminal())
+            .filter_map(|d| d.parent_id)
+            .collect();
         let mut counts: HashMap<Template, usize> = HashMap::new();
         let kept: Vec<Dag> = inner
             .dags
             .iter()
             .rev()
             .filter(|d| {
-                if !d.is_terminal() {
+                if !d.is_terminal() || live_parents.contains(&d.id) {
                     return true;
                 }
                 let n = counts.entry(d.template).or_insert(0);
diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs
index 8c4ebcf6..39933504 100644
--- a/hive-c0re/src/job_queue/model.rs
+++ b/hive-c0re/src/job_queue/model.rs
@@ -33,8 +33,9 @@ pub enum Template {
     /// non-fatal) node; stale agents' `Rebuild` DAGs fan out on
     /// completion.
     StartupSweep,
-    /// `StopForUpdate → Reconcile` — stop + converge back to `wanted`
-    /// (unchanged), i.e. a restart for a wanted-up agent.
+    /// `StopForUpdate → Reconcile` with `wanted` set to `Up` at submit
+    /// time — a mechanical stop + start, like the old
+    /// `lifecycle::restart`, regardless of prior intent drift.
     Restart,
     /// `WritePermFile → Prebuild → StopForUpdate → Swap → Reconcile` —
     /// perm-file commit followed by the rebuild subgraph.
diff --git a/hive-c0re/src/job_queue/scheduler.rs b/hive-c0re/src/job_queue/scheduler.rs
index 3962cbc2..fb7997a4 100644
--- a/hive-c0re/src/job_queue/scheduler.rs
+++ b/hive-c0re/src/job_queue/scheduler.rs
@@ -38,6 +38,11 @@ pub async fn run_worker(coord: Arc<Coordinator>) {
     // DAG id → transient guard held for the lease window.
     let mut transients: HashMap<u64, crate::coordinator::TransientGuard> = HashMap::new();
     loop {
+        // Terminal roll-ups can appear without a node completion —
+        // the cancel surfaces settle DAGs directly and wake this loop
+        // via notify — so drain on every iteration, not just inside
+        // handle_completion.
+        process_terminals(&coord, &mut transients).await;
         let claims = coord.job_queue.claim_ready();
         if !claims.is_empty() {
             for claim in claims {
@@ -109,20 +114,28 @@ async fn handle_completion(
             (Err(msg), Vec::new())
         }
     };
-    let report = coord
+    coord
         .job_queue
         .complete_node(claim.dag_id, claim.node_id, queue_result);
     if !fanout.is_empty() {
         let specs = fanout_specs(&claim, fanout);
         coord.job_queue.append_children(specs);
     }
-    for terminal in report.terminal {
-        // Drop the lease-window transient guard, then let the hook
-        // fire approval resolution / failure events.
+    process_terminals(coord, transients).await;
+    coord.emit_rebuild_queue_snapshot();
+}
+
+/// Drain buffered terminal roll-ups: drop each DAG's lease-window
+/// transient guard, then run the terminal hook (approval resolution,
+/// `Rebuilt` events, cancelled-power-op intent revert).
+async fn process_terminals(
+    coord: &Arc<Coordinator>,
+    transients: &mut HashMap<u64, crate::coordinator::TransientGuard>,
+) {
+    for terminal in coord.job_queue.drain_terminal() {
         transients.remove(&terminal.dag_id);
         exec::on_dag_terminal(coord, &terminal).await;
     }
-    coord.emit_rebuild_queue_snapshot();
 }
 
 /// Child `Rebuild` specs for a completed `MetaLock` fan-out, grouped
diff --git a/hive-c0re/src/job_queue/submit.rs b/hive-c0re/src/job_queue/submit.rs
index 324a1c0c..fa889fe7 100644
--- a/hive-c0re/src/job_queue/submit.rs
+++ b/hive-c0re/src/job_queue/submit.rs
@@ -34,8 +34,12 @@ 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 back to `wanted` (unchanged).
+/// Restart: mechanical stop + converge to `wanted = Up`. The intent
+/// write matters when `wanted` drifted `Offline` under a running
+/// agent — the old `kill + start` always ended up, and an operator
+/// asking for a restart plainly wants it running, not a stop.
 pub fn restart(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: String) -> u64 {
+    set_wanted(coord, agent, Wanted::Up);
     submit_and_emit(coord, templates::restart(agent, source, reason))
 }
 
diff --git a/hive-c0re/src/job_queue/templates.rs b/hive-c0re/src/job_queue/templates.rs
index 536a7e50..46f817c2 100644
--- a/hive-c0re/src/job_queue/templates.rs
+++ b/hive-c0re/src/job_queue/templates.rs
@@ -6,7 +6,7 @@
 //! ```text
 //! rebuild(a):        Prebuild(a) → StopForUpdate(a) → Swap(a) →(any) Reconcile(a)
 //! graceful-stop(a):  [wanted=Offline]  Signal(a) → Drain(a) → Reconcile(a)
-//! restart(a):        StopForUpdate(a) → Reconcile(a)      (wanted unchanged)
+//! restart(a):        [wanted=Up]       StopForUpdate(a) → Reconcile(a)
 //! start(a):          [wanted=Up]       Reconcile(a)
 //! stop(a):           [wanted=Offline]  Reconcile(a)
 //! spawn(a):          [wanted=Up]  Create(a) → WriteDropin(a) → Reconcile(a)
@@ -141,8 +141,9 @@ pub fn graceful_stop(agent: &str, source: Source, reason: String) -> DagSpec {
     }
 }
 
-/// Restart: mechanical stop, then converge back to `wanted`
-/// (unchanged) — a stop + start for a wanted-up agent.
+/// Restart: mechanical stop, then converge to `wanted` — the submit
+/// layer writes `wanted = Up` first, so this is a stop + start like
+/// the old `lifecycle::restart` regardless of prior intent drift.
 pub fn restart(agent: &str, source: Source, reason: String) -> DagSpec {
     DagSpec {
         template: Template::Restart,
diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs
index 128a352e..f7ec112e 100644
--- a/hive-c0re/src/job_queue/tests.rs
+++ b/hive-c0re/src/job_queue/tests.rs
@@ -626,13 +626,15 @@ fn terminal_dag_reported_exactly_once_and_lease_released() {
         templates::restart("agent-a", Source::Manual, "r".to_owned()),
     );
     let stop = claim_one(&q);
-    let r1 = q.complete_node(id, stop.node_id, Ok(()));
-    assert!(r1.terminal.is_empty(), "dag not terminal yet");
+    q.complete_node(id, stop.node_id, Ok(()));
+    assert!(q.drain_terminal().is_empty(), "dag not terminal yet");
     let rec = claim_one(&q);
-    let r2 = q.complete_node(id, rec.node_id, Ok(()));
-    assert_eq!(r2.terminal.len(), 1);
-    assert_eq!(r2.terminal[0].dag_id, id);
-    assert_eq!(r2.terminal[0].state, State::Done);
+    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,
@@ -649,20 +651,105 @@ fn terminal_dag_reported_exactly_once_and_lease_released() {
     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() {
+fn cancelled_dag_reports_terminal_once() {
     let q = JobQueue::new(1);
-    let id = submit(&q, rebuild("agent-a", "r"));
+    let id = submit(
+        &q,
+        templates::approval_deploy("agent-a", 7, "approval #7".to_owned()),
+    );
     assert!(q.cancel(id));
-    // The cancel path settles internally; a subsequent completion
-    // report must not re-report it. Verify via a second dag's cycle.
+    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);
-    let report = q.complete_node(other, c.node_id, Err("boom".to_owned()));
-    // agent-b's dag isn't terminal (reconcile still pending) and
-    // agent-a's was already reported by cancel → nothing here.
-    assert!(report.terminal.iter().all(|t| t.dag_id != id));
+    q.complete_node(other, c.node_id, Err("boom".to_owned()));
+    assert!(q.drain_terminal().iter().all(|t| t.dag_id != id));
+}
+
+#[test]
+fn cancel_children_reports_terminals() {
+    let q = JobQueue::new(1);
+    let meta = submit(
+        &q,
+        templates::meta_update(vec![], Source::Manual, "bump".to_owned(), None),
+    );
+    let _lock = claim_one(&q);
+    let child = submit(
+        &q,
+        templates::rebuild(
+            "agent-a",
+            Source::MetaUpdate,
+            "cascade".to_owned(),
+            Some(meta),
+            false,
+        ),
+    );
+    assert_eq!(q.cancel_children(meta), 1);
+    let reports = q.drain_terminal();
+    assert_eq!(reports.len(), 1);
+    assert_eq!(reports[0].dag_id, child);
+    assert_eq!(reports[0].state, State::Cancelled);
+}
+
+/// History trim must not evict a terminal fan-out parent while its
+/// children are still live — the dashboard groups children under it.
+#[test]
+fn trim_keeps_terminal_parent_with_live_children() {
+    let q = JobQueue::new(1);
+    // Pin agent-x's lease with a running stop DAG so the child below
+    // stays fully queued while we churn history.
+    let pin = submit(
+        &q,
+        templates::reconcile_only(
+            Template::Stop,
+            "agent-x",
+            Source::Manual,
+            "lease pin".to_owned(),
+            None,
+        ),
+    );
+    let pin_claim = claim_one(&q);
+    assert_eq!(pin_claim.dag_id, pin);
+    // Terminal fan-out parent + a lease-blocked child under it.
+    let meta = submit(
+        &q,
+        templates::meta_update(vec![], Source::Manual, "bump".to_owned(), None),
+    );
+    let lock = claim_one(&q);
+    q.complete_node(meta, lock.node_id, Ok(()));
+    let mut child_spec = templates::restart("agent-x", Source::MetaUpdate, "cascade".to_owned());
+    child_spec.parent_id = Some(meta);
+    let child = submit(&q, child_spec);
+    // Churn > MAX_HISTORY_PER_TEMPLATE terminal meta_update DAGs.
+    for i in 0..7 {
+        let id = submit(
+            &q,
+            templates::meta_update(
+                vec![format!("input-{i}")],
+                Source::Manual,
+                "churn".to_owned(),
+                None,
+            ),
+        );
+        let c = claim_one(&q);
+        assert_eq!(c.dag_id, id, "child is lease-blocked; churn claims freely");
+        q.complete_node(id, c.node_id, Ok(()));
+    }
+    let snap = q.snapshot();
+    assert!(
+        snap.iter().any(|d| d.id == meta),
+        "terminal parent with live child must survive trim"
+    );
+    assert!(snap.iter().any(|d| d.id == child));
 }
 
 // ---- steps, build logs, history ----
diff --git a/hive-c0re/src/meta.rs b/hive-c0re/src/meta.rs
index c5764faa..ad9034bf 100644
--- a/hive-c0re/src/meta.rs
+++ b/hive-c0re/src/meta.rs
@@ -29,6 +29,25 @@ const GIT_EMAIL: &str = "c0re@hyperhive.local";
 /// take turns instead of colliding.
 static META_LOCK: Mutex<()> = Mutex::const_new(());
 
+/// Coarse exclusivity for meta-repo *windows* that span multiple
+/// `META_LOCK` acquisitions — above all the two-phase deploy
+/// (`prepare_deploy` stages `flake.lock` uncommitted for the whole
+/// container build; `finalize_deploy` / `abort_deploy` resolve it).
+/// `META_LOCK` serializes individual git ops but cannot keep another
+/// op out of that staged window: a perm-file or lock-bump commit
+/// landing mid-window would sweep the staged deploy lock into its own
+/// commit and neuter `abort_deploy`. Job-queue executors that mutate
+/// the meta repo hold this gate for their mutation span; the opaque
+/// approval-deploy node holds it across its whole prepare→finalize
+/// span. Never acquired inside this module's functions (they run
+/// *under* a caller's window — nesting would deadlock).
+static DEPLOY_GATE: Mutex<()> = Mutex::const_new(());
+
+/// Acquire the deploy/meta-mutation window gate. See [`DEPLOY_GATE`].
+pub async fn exclusive() -> tokio::sync::MutexGuard<'static, ()> {
+    DEPLOY_GATE.lock().await
+}
+
 /// Where the manager sees this directory inside its container (RO bind).
 pub const CONTAINER_MANAGER_META_MOUNT: &str = "/meta";
 
@@ -239,11 +258,16 @@ pub async fn prepare_deploy(name: &str) -> Result<()> {
 pub async fn finalize_deploy(name: &str, sha: &str, tag: &str) -> Result<()> {
     let _guard = META_LOCK.lock().await;
     let dir = meta_dir();
-    if !has_staged_changes(&dir).await? {
+    if !paths_dirty(&dir, &["flake.lock"]).await? {
         return Ok(());
     }
     let short = &sha[..sha.len().min(12)];
-    git_commit(&dir, &format!("deploy {name} {tag} {short}")).await
+    git_commit_paths(
+        &dir,
+        &format!("deploy {name} {tag} {short}"),
+        &["flake.lock"],
+    )
+    .await
 }
 
 /// Phase 2-failure. Unstage + restore the lock so meta returns to
@@ -256,21 +280,6 @@ pub async fn abort_deploy() -> Result<()> {
     git(&dir, &["restore", "flake.lock"]).await
 }
 
-async fn has_staged_changes(dir: &Path) -> Result<bool> {
-    let st = lifecycle::git_command()
-        .current_dir(dir)
-        .args(["diff", "--cached", "--quiet"])
-        .status()
-        .await
-        .with_context(|| format!("git diff --cached in {}", dir.display()))?;
-    // exit 1 = differences present, 0 = no diff, other = error
-    match st.code() {
-        Some(0) => Ok(false),
-        Some(1) => Ok(true),
-        _ => bail!("git diff --cached exited unexpectedly"),
-    }
-}
-
 /// One-shot used by the manual-rebuild path: relock just one
 /// agent's input and commit the lock change if any. Single-phase
 /// (no separate finalize) because rebuild has no failure-revert
@@ -280,11 +289,16 @@ pub async fn lock_update_for_rebuild(name: &str) -> Result<()> {
     let dir = meta_dir();
     let input = format!("agent-{name}");
     nix(&dir, &["flake", "update", &input]).await?;
-    if git_is_clean(&dir).await? {
+    if !paths_dirty(&dir, &["flake.lock"]).await? {
         return Ok(());
     }
     git(&dir, &["add", "flake.lock"]).await?;
-    git_commit(&dir, &format!("rebuild {name}: lock update")).await
+    git_commit_paths(
+        &dir,
+        &format!("rebuild {name}: lock update"),
+        &["flake.lock"],
+    )
+    .await
 }
 
 /// Build the `--override-input` value pinning an agent's config repo to
@@ -349,7 +363,7 @@ pub async fn lock_update(inputs: &[String]) -> Result<()> {
         args.push(i.as_str());
     }
     nix(&dir, &args).await?;
-    if git_is_clean(&dir).await? {
+    if !paths_dirty(&dir, &["flake.lock"]).await? {
         return Ok(());
     }
     git(&dir, &["add", "flake.lock"]).await?;
@@ -360,7 +374,7 @@ pub async fn lock_update(inputs: &[String]) -> Result<()> {
     } else {
         format!("lock update: {}", inputs.join(", "))
     };
-    git_commit(&dir, &msg).await
+    git_commit_paths(&dir, &msg, &["flake.lock"]).await
 }
 
 /// One-shot used by the auto-update path: pin the latest hyperhive
@@ -370,11 +384,11 @@ pub async fn lock_update_hyperhive() -> Result<()> {
     let _guard = META_LOCK.lock().await;
     let dir = meta_dir();
     nix(&dir, &["flake", "update", "hyperhive"]).await?;
-    if git_is_clean(&dir).await? {
+    if !paths_dirty(&dir, &["flake.lock"]).await? {
         return Ok(());
     }
     git(&dir, &["add", "flake.lock"]).await?;
-    git_commit(&dir, "bump hyperhive").await
+    git_commit_paths(&dir, "bump hyperhive", &["flake.lock"]).await
 }
 
 /// Write the tool-groups file for `agent` and commit it atomically
@@ -388,8 +402,13 @@ pub async fn commit_tool_groups(agent: &str, groups: &[String]) -> Result<()> {
     if crate::tool_groups::tool_groups_path().exists() {
         git(&dir, &["add", "tool-groups.json"]).await?;
     }
-    if has_staged_changes(&dir).await? {
-        git_commit(&dir, &format!("set tool-groups for {agent}")).await?;
+    if paths_dirty(&dir, &["tool-groups.json"]).await? {
+        git_commit_paths(
+            &dir,
+            &format!("set tool-groups for {agent}"),
+            &["tool-groups.json"],
+        )
+        .await?;
     }
     Ok(())
 }
@@ -404,8 +423,13 @@ pub async fn commit_capabilities(agent: &str, caps: &[String]) -> Result<()> {
     if crate::capabilities::capabilities_path().exists() {
         git(&dir, &["add", "capabilities.json"]).await?;
     }
-    if has_staged_changes(&dir).await? {
-        git_commit(&dir, &format!("set capabilities for {agent}")).await?;
+    if paths_dirty(&dir, &["capabilities.json"]).await? {
+        git_commit_paths(
+            &dir,
+            &format!("set capabilities for {agent}"),
+            &["capabilities.json"],
+        )
+        .await?;
     }
     Ok(())
 }
@@ -445,8 +469,14 @@ pub async fn commit_perms(
         }
         parts.push("capabilities");
     }
-    if has_staged_changes(&dir).await? {
-        git_commit(&dir, &format!("set {} for {agent}", parts.join(" + "))).await?;
+    let paths = ["tool-groups.json", "capabilities.json"];
+    if paths_dirty(&dir, &paths).await? {
+        git_commit_paths(
+            &dir,
+            &format!("set {} for {agent}", parts.join(" + ")),
+            &paths,
+        )
+        .await?;
     }
     Ok(())
 }
@@ -467,10 +497,11 @@ pub async fn commit_topology(
     let dir = meta_dir();
     let stage = async {
         git(&dir, &["add", "topology.json"]).await?;
-        if has_staged_changes(&dir).await? {
-            git_commit(
+        if paths_dirty(&dir, &["topology.json"]).await? {
+            git_commit_paths(
                 &dir,
                 &format!("topology: {} → {}", child, new_parent.unwrap_or("<root>")),
+                &["topology.json"],
             )
             .await?;
         }
@@ -541,8 +572,8 @@ pub async fn bulk_commit_topology(
     };
     let stage = async {
         git(&dir, &["add", "topology.json"]).await?;
-        if has_staged_changes(&dir).await? {
-            git_commit(&dir, &commit_msg).await?;
+        if paths_dirty(&dir, &["topology.json"]).await? {
+            git_commit_paths(&dir, &commit_msg, &["topology.json"]).await?;
         }
         Ok::<_, anyhow::Error>(())
     };
@@ -1158,16 +1189,6 @@ where
     out
 }
 
-async fn git_is_clean(dir: &Path) -> Result<bool> {
-    let out = lifecycle::git_command()
-        .current_dir(dir)
-        .args(["status", "--porcelain"])
-        .output()
-        .await
-        .with_context(|| format!("git status in {}", dir.display()))?;
-    Ok(out.stdout.iter().all(u8::is_ascii_whitespace))
-}
-
 /// Return the list of file names that are currently staged (index differs
 /// from HEAD). On the initial commit (`HEAD` doesn't exist yet) falls back
 /// to `git diff --cached --name-only HEAD` failing gracefully by using
@@ -1248,6 +1269,43 @@ async fn git_commit(dir: &Path, message: &str) -> Result<()> {
     Ok(())
 }
 
+/// Path-limited commit: commits ONLY the given paths, so unrelated
+/// staged content — above all a `prepare_deploy`-staged `flake.lock`
+/// — can never be swept into someone else's commit. Every targeted
+/// meta commit (perm files, topology, lock bumps) goes through this;
+/// only `sync_agents` uses the bare [`git_commit`], because its
+/// staged set *is* its intentional commit set.
+async fn git_commit_paths(dir: &Path, message: &str, paths: &[&str]) -> Result<()> {
+    let name = format!("user.name={GIT_NAME}");
+    let email = format!("user.email={GIT_EMAIL}");
+    let mut args = vec!["-c", &name, "-c", &email, "commit", "-m", message, "--"];
+    args.extend_from_slice(paths);
+    git(dir, &args).await?;
+    if let Err(e) = crate::forge::push_meta(dir).await {
+        tracing::warn!(error = ?e, "forge: meta push after commit failed (non-fatal)");
+    }
+    Ok(())
+}
+
+/// True when any of `paths` differs between HEAD and the index or
+/// working tree — the path-scoped replacement for whole-tree
+/// `git_is_clean` / `has_staged_changes` guards, which a concurrently
+/// staged deploy lock would otherwise trip.
+async fn paths_dirty(dir: &Path, paths: &[&str]) -> Result<bool> {
+    let mut args = vec!["diff", "--quiet", "HEAD", "--"];
+    args.extend_from_slice(paths);
+    let out = lifecycle::git_command()
+        .current_dir(dir)
+        .args(&args)
+        .output()
+        .await
+        .with_context(|| format!("git diff --quiet in {}", dir.display()))?;
+    // Exit 0 = no differences; 1 = differences; anything else (e.g.
+    // no HEAD yet on a fresh repo) → treat as dirty so the commit
+    // path runs and surfaces real errors loudly.
+    Ok(!out.status.success())
+}
+
 async fn nix(dir: &Path, args: &[&str]) -> Result<()> {
     // `--extra-experimental-features` belt-and-suspenders for hosts
     // that haven't set this in nix.conf. The hyperhive module's
@@ -1276,6 +1334,43 @@ async fn nix(dir: &Path, args: &[&str]) -> Result<()> {
 mod tests {
     use super::*;
 
+    /// The regression the deploy-window bug review surfaced: a
+    /// path-limited commit must leave an unrelated staged file (the
+    /// prepare_deploy-staged `flake.lock`) untouched, so a later
+    /// `abort_deploy` still has something to restore.
+    #[tokio::test]
+    async fn path_limited_commit_leaves_unrelated_staged_file_alone() {
+        let tmp = tempfile::tempdir().expect("tempdir");
+        let dir = tmp.path();
+        git(dir, &["init", "--initial-branch=main"])
+            .await
+            .expect("git init");
+        std::fs::write(dir.join("tool-groups.json"), "{}").expect("write");
+        std::fs::write(dir.join("flake.lock"), "v1").expect("write");
+        git(dir, &["add", "-A"]).await.expect("add");
+        git_commit(dir, "seed").await.expect("seed commit");
+        // A deploy stages a new lock (uncommitted)…
+        std::fs::write(dir.join("flake.lock"), "v2-staged-by-deploy").expect("write");
+        git(dir, &["add", "flake.lock"]).await.expect("stage lock");
+        // …and a perm change commits, path-limited.
+        std::fs::write(dir.join("tool-groups.json"), r#"{"alice":[]}"#).expect("write");
+        git(dir, &["add", "tool-groups.json"]).await.expect("add");
+        git_commit_paths(dir, "set tool-groups for alice", &["tool-groups.json"])
+            .await
+            .expect("path-limited commit");
+        // The perm file is committed; the deploy's staged lock is not.
+        assert!(
+            !paths_dirty(dir, &["tool-groups.json"])
+                .await
+                .expect("check"),
+            "perm file must be committed"
+        );
+        assert!(
+            paths_dirty(dir, &["flake.lock"]).await.expect("check"),
+            "staged deploy lock must survive the perm commit"
+        );
+    }
+
     fn sample_spec(name: &str, is_manager: bool, port: u16) -> AgentSpec {
         AgentSpec {
             name: name.to_owned(),
diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs
index 0572306c..c64be3e1 100644
--- a/hive-c0re/src/server.rs
+++ b/hive-c0re/src/server.rs
@@ -93,10 +93,19 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
             HostRequest::Kill { name } => handle_kill(&coord, name).await?,
             HostRequest::Restart { name } => {
                 tracing::info!(%name, "restart");
-                lifecycle::restart(name).await?;
+                // Through the queue: serializes against in-flight
+                // rebuilds via the agent lease, writes `wanted = Up`,
+                // and gets the transient/crash-watch suppression the
+                // direct kill+start lacked. Returns once queued.
+                crate::job_queue::submit::restart(
+                    &coord,
+                    name,
+                    crate::job_queue::Source::Manual,
+                    "manual restart via hivectl".to_owned(),
+                );
                 HostResponse::success()
             }
-            HostRequest::RestartAll => handle_restart_all().await?,
+            HostRequest::RestartAll => handle_restart_all(&coord).await?,
             HostRequest::Stop { scope, graceful } => {
                 // Resolve the scope to explicit container names at the entry
                 // point, then operate on names — never pass the bare "all
@@ -241,33 +250,27 @@ async fn handle_kill(coord: &Arc<Coordinator>, name: &str) -> Result<HostRespons
     Ok(HostResponse::success())
 }
 
-/// Restart every container, aggregating per-agent failures into one
-/// response rather than aborting on the first error.
-async fn handle_restart_all() -> Result<HostResponse> {
+/// Restart every container by submitting one restart DAG per agent —
+/// each serializes on its own lease, so unrelated agents' restarts
+/// overlap while nothing races an in-flight rebuild. Returns once all
+/// are queued; per-agent results surface on the queue.
+async fn handle_restart_all(coord: &Arc<Coordinator>) -> Result<HostResponse> {
     tracing::info!("restart-all");
     let agents = lifecycle::list().await?;
     let mut ok_agents: Vec<String> = Vec::new();
-    let mut errors: Vec<String> = Vec::new();
     for agent in &agents {
-        if let Err(e) = lifecycle::restart(agent).await {
-            tracing::warn!(%agent, error = ?e, "restart-all: failed to restart agent");
-            errors.push(format!("{agent}: {e:#}"));
-        } else {
-            ok_agents.push(agent.clone());
-        }
-    }
-    if errors.is_empty() {
-        Ok(HostResponse::list(ok_agents))
-    } else {
-        Ok(HostResponse {
-            ok: false,
-            error: Some(errors.join("; ")),
-            agents: Some(ok_agents),
-            approvals: None,
-            urls: None,
-            agent_statuses: None,
-        })
+        let Some(logical) = agent.strip_prefix(lifecycle::AGENT_PREFIX) else {
+            continue;
+        };
+        crate::job_queue::submit::restart(
+            coord,
+            logical,
+            crate::job_queue::Source::Manual,
+            "manual restart via hivectl restart-all".to_owned(),
+        );
+        ok_agents.push(logical.to_owned());
     }
+    Ok(HostResponse::list(ok_agents))
 }
 
 /// Stop the given `agents` (resolved logical names) then `infra` containers