hive-c0re/hivectl/hive-agent: pause as a job-queue DAG node (closes #3056)

hive-c0re/src/coordinator.rs

@@ -162,6 +162,14 @@ pub struct Coordinator {
     /// agent is in this set — the inbound fence. Cleared when the agent
     /// reports `GracefulStopComplete` or the container is stopped.
     graceful_stop_pending: Mutex<HashSet<String>>,
+    /// Agents with a pause acknowledgement in progress — set by the pause
+    /// DAG's signal node, cleared when the agent reports
+    /// `PauseAcknowledged` (or the drain node's timeout fallback fires).
+    /// Same shape as `graceful_stop_pending`, one set per concern rather
+    /// than reusing it: a pause and a graceful stop are independent
+    /// orchestrations that can be in flight for different agents
+    /// simultaneously.
+    pause_pending: Mutex<HashSet<String>>,
     /// Logical agent names that were running at the last broad-scope
     /// `hivectl stop`. A subsequent broad-scope `hivectl start` restores
     /// only this set (intersected with the requested scope) rather than
@@ -582,6 +590,7 @@ impl Coordinator {
             recent_transient: Mutex::new(HashMap::new()),
             recent_crashes: Mutex::new(HashMap::new()),
             graceful_stop_pending: Mutex::new(HashSet::new()),
+            pause_pending: Mutex::new(HashSet::new()),
             last_stopped_running: Mutex::new(None),
             dashboard_events,
             event_seq: AtomicU64::new(0),
@@ -1188,6 +1197,26 @@ impl Coordinator {
         self.graceful_stop_pending.lock().unwrap().remove(name);
     }
 
+    /// Mark `name` as having a pause acknowledgement in progress. Set by
+    /// the pause DAG's signal node; the drain node polls
+    /// `is_pause_pending` until the agent reports back or the timeout
+    /// fallback fires.
+    pub fn mark_pause_pending(&self, name: &str) {
+        self.pause_pending.lock().unwrap().insert(name.to_owned());
+    }
+
+    /// Whether a pause acknowledgement is pending for `name`.
+    #[must_use]
+    pub fn is_pause_pending(&self, name: &str) -> bool {
+        self.pause_pending.lock().unwrap().contains(name)
+    }
+
+    /// Clear the pause-pending flag for `name` (agent reported
+    /// `PauseAcknowledged`, or the drain node's wait finished / timed out).
+    pub fn clear_pause_pending(&self, name: &str) {
+        self.pause_pending.lock().unwrap().remove(name);
+    }
+
     /// Record the set of agents that were running at a broad-scope
     /// `hivectl stop`, so the next broad-scope `start` restores exactly
     /// this set. See the `last_stopped_running` field doc. Persists a

hive-c0re/src/job_queue/exec.rs

@@ -24,6 +24,15 @@ use crate::power::{ReconcileAction, reconcile_action};
 /// N × this timeout.
 pub const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(3);
 
+/// Max time `PauseDrain` waits for the harness to report
+/// `PauseAcknowledged` before giving up and resolving anyway (the
+/// marker itself — not this node — is what actually gates the turn
+/// loop, so "giving up" costs nothing but a slightly-late dashboard
+/// badge). Same ceiling as `GRACEFUL_STOP_TIMEOUT` — no reason for the
+/// two to diverge yet, but aliased under its own name so a future
+/// change to one doesn't silently retune the other.
+const PAUSE_ACK_TIMEOUT: std::time::Duration = GRACEFUL_STOP_TIMEOUT;
+
 /// Run one claimed node to completion. Called from a task the
 /// scheduler spawns per claim; the `Result` (stringified) becomes the
 /// node's terminal state.
@@ -97,6 +106,8 @@ pub(super) async fn run_node(
             Ok(())
         }
         NodeKind::Drain { .. } => run_drain(coord, agent).await,
+        NodeKind::PauseSignal { .. } => run_pause_signal(coord, agent).await,
+        NodeKind::PauseDrain { .. } => run_pause_drain(coord, agent).await,
         NodeKind::WriteDropin { .. } => run_write_dropin(coord, agent).await,
         // The payload rides the node and is destructured here, so the executor
         // takes it directly instead of re-matching the kind behind a `bail!`
@@ -490,6 +501,54 @@ async fn run_drain(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
     Ok(())
 }
 
+/// Write the pause marker + mark `pause_pending`, no kick (unlike
+/// `run_signal`) — the harness's own between-turns poll (`PAUSE_POLL`,
+/// 1s default) is already responsive enough, and `run_signal`'s kick
+/// message ("you were just (re)started") would be actively misleading
+/// here.
+///
+/// Skips marking `pause_pending` (the marker write still happens,
+/// harmlessly idempotent either way) if the agent is already paused:
+/// the harness reports `PauseAcknowledged` only on the marker's
+/// `false → true` edge, so re-pausing an already-paused agent produces
+/// no edge for `run_pause_drain` to observe — marking pending here
+/// would just burn its timeout every time an operator re-confirms a
+/// pause that already took.
+async fn run_pause_signal(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
+    let agent = hive_types::Ident::parse(name)
+        .map_err(|e| anyhow::anyhow!("invalid agent name for pause {name:?}: {e}"))?;
+    let already_paused = Coordinator::is_paused(&agent);
+    Coordinator::set_paused(&agent, true).await?;
+    if !already_paused {
+        coord.mark_pause_pending(name);
+    }
+    // Same pattern `run_start`/`run_stop` use: refresh the dashboard's
+    // view right after the state change so the paused badge flips
+    // immediately instead of waiting on the next periodic rescan.
+    coord.rescan_containers_and_emit().await;
+    Ok(())
+}
+
+/// Await the harness reporting `PauseAcknowledged`, bounded by
+/// `PAUSE_ACK_TIMEOUT`. Resolves ok either way, mirroring `run_drain` —
+/// pausing is best-effort from the queue's perspective; the marker
+/// (not this node) is what actually gates the harness's turn loop, so
+/// a timed-out wait doesn't leave the agent un-paused, just leaves the
+/// dashboard's "pausing…" badge running a little longer than it needed
+/// to.
+async fn run_pause_drain(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
+    let deadline = std::time::Instant::now() + PAUSE_ACK_TIMEOUT;
+    while coord.is_pause_pending(name) {
+        if std::time::Instant::now() >= deadline {
+            tracing::warn!(agent = %name, "pause: ack wait timed out — marker is set regardless");
+            break;
+        }
+        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
+    }
+    coord.clear_pause_pending(name);
+    Ok(())
+}
+
 /// `set_nspawn_flags` + `set_resource_limits` + daemon-reload.
 async fn run_write_dropin(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
     // write_dropins only needs the path value to build AgentPaths; the

hive-c0re/src/job_queue/model.rs

@@ -119,6 +119,16 @@ pub enum NodeKind {
     /// `GRACEFUL_STOP_TIMEOUT`. Resolves ok either way — the
     /// downstream `Reconcile` performs the actual stop.
     Drain { agent: String },
+    /// Write the pause marker (`Coordinator::set_paused`) + mark
+    /// `pause_pending`. No kick, unlike `Signal` — the harness's own
+    /// between-turns poll (`PAUSE_POLL`, 1s default) is already
+    /// responsive enough, and `Signal`'s kick-message body ("you were
+    /// just (re)started") would be actively misleading here.
+    PauseSignal { agent: String },
+    /// Await the harness reporting `PauseAcknowledged`, bounded by a
+    /// timeout. Resolves ok either way — pausing is best-effort from
+    /// the queue's perspective, same as `Drain`.
+    PauseDrain { agent: String },
     /// `set_nspawn_flags` + `set_resource_limits` + daemon-reload.
     WriteDropin { agent: String },
     /// Commit `tool-groups.json` / `capabilities.json` per its `payload`
@@ -338,6 +348,8 @@ impl NodeKind {
             NodeKind::StopForUpdate { .. } => "stop_for_update",
             NodeKind::Signal { .. } => "signal",
             NodeKind::Drain { .. } => "drain",
+            NodeKind::PauseSignal { .. } => "pause_signal",
+            NodeKind::PauseDrain { .. } => "pause_drain",
             NodeKind::WriteDropin { .. } => "write_dropin",
             NodeKind::WritePermFile { .. } => "write_perm_file",
             NodeKind::Reparent { .. } => "reparent",
@@ -372,6 +384,8 @@ impl NodeKind {
             | NodeKind::StopForUpdate { agent }
             | NodeKind::Signal { agent }
             | NodeKind::Drain { agent }
+            | NodeKind::PauseSignal { agent }
+            | NodeKind::PauseDrain { agent }
             | NodeKind::WriteDropin { agent }
             | NodeKind::WritePermFile { agent, .. }
             | NodeKind::DeployWindow { agent, .. }

hive-c0re/src/job_queue/power.rs

@@ -327,3 +327,31 @@ pub async fn stop_many(
     coord.emit_rebuild_queue_snapshot();
     Ok(ids)
 }
+
+/// Declare the pause DAG for `agents` — one `PauseSignal → PauseDrain`
+/// pair (see [`super::templates::pause_quiesce`]) per agent, independent
+/// roots on their own agent lease so a whole-hive pause overlaps rather
+/// than serialising.
+pub(crate) fn pause_nodes(builder: &JobBuilder, agents: &[String]) -> Vec<hive_jobq::NodeGuid> {
+    agents
+        .iter()
+        .map(|agent| super::templates::pause_quiesce(builder, agent).guid())
+        .collect()
+}
+
+/// Pause `agents` in a **single** DAG. Unlike `stop`/`start`/`restart`,
+/// this needs no live-state read first — `Coordinator::set_paused`
+/// works on a stopped container too (the marker is sticky), so there's
+/// no `running`/`stale` branch to resolve async before building.
+///
+/// # Errors
+/// Propagates a graph-insert error.
+pub async fn pause_many(
+    coord: &Arc<Coordinator>,
+    agents: &[String],
+) -> anyhow::Result<Vec<NodeId>> {
+    let targets: Vec<String> = agents.to_vec();
+    let ids = coord.job_queue.insert_job(|b| pause_nodes(b, &targets))?;
+    coord.emit_rebuild_queue_snapshot();
+    Ok(ids)
+}

hive-c0re/src/job_queue/templates.rs

@@ -137,6 +137,46 @@ pub(crate) fn quiesce<'a>(builder: &'a JobBuilder, agent: &str, brace: Handle<'a
         .after_ok(signal)
 }
 
+/// The pause quiesce pair — `PauseSignal` then `PauseDrain`: write the
+/// pause marker, then wait for the harness to acknowledge it. Returns
+/// the **group root** — the brace, not the tail. Unlike [`quiesce`],
+/// nothing chains onto `PauseDrain` (there's no downstream
+/// `Reconcile`-shaped node the way a stop has one), so the handle a
+/// caller actually needs is the brace whose roll-up covers the whole
+/// pair, same as [`super::power`]'s `stop_chain` returning `wanted`'s
+/// guid rather than `quiesce`'s own returned `Drain` handle.
+///
+/// Unlike [`quiesce`], there's no natural resource-holding head to
+/// borrow a `brace` from — pausing isn't a `wanted`-state transition,
+/// so there's no `SetWanted`-shaped parent the way `stop_chain` has
+/// one. This is exactly the case [`NodeKind::AgentWindow`] exists for
+/// (see the module header's _brace_ paragraph): a pure-resource-holder
+/// root with no work of its own, so `PauseSignal`/`PauseDrain` can be
+/// plain siblings under it, both borrowing its lease via `part_of`.
+///
+/// ⚠️ `PauseSignal` can *not* hold the lease itself with `PauseDrain`
+/// nested under it (`.part_of(signal)`) — that was the first shape
+/// tried here, and `hive_jobq` rejects it at insert: `PauseDrain`
+/// depending on `PauseSignal` via `after_ok` while also being its
+/// *child* reaches outside `PauseDrain`'s own group (its parent, not a
+/// sibling) — "an edge must stay within the depender's own group".
+/// `AgentWindow` as a separate, actual brace is what makes them
+/// siblings instead.
+pub(crate) fn pause_quiesce<'a>(builder: &'a JobBuilder, agent: &str) -> Handle<'a> {
+    let a = || agent.to_owned();
+    let brace = builder
+        .node(NodeKind::AgentWindow { agent: a() })
+        .needs(Resource::Agent(a()));
+    let signal = builder
+        .node(NodeKind::PauseSignal { agent: a() })
+        .part_of(brace);
+    let _drain = builder
+        .node(NodeKind::PauseDrain { agent: a() })
+        .part_of(brace)
+        .after_ok(signal);
+    brace
+}
+
 /// The group-roots a [`rebuild_nodes`] subgraph exposes to its caller: what a
 /// tail node edges onto, and what a follow-up node waits for.
 ///

hive-c0re/src/job_queue/tests.rs

@@ -1587,6 +1587,47 @@ fn graceful_stop_shape_signal_drain_reconcile() {
     );
 }
 
+#[test]
+fn pause_shape_signal_drain() {
+    let q = JobQueue::new(1);
+    insert(&q, |builder| {
+        power::pause_nodes(builder, &["agent-a".to_owned()]);
+    });
+    assert_eq!(
+        declared_shape(&q),
+        vec![
+            // Unlike the stop quiesce pair (which hangs under an existing
+            // `set_wanted` head), pausing has no natural parent to reuse, so
+            // this shape declares its own `AgentWindow` brace — the pair are
+            // plain siblings under it, not a signal-holds-the-lease-itself
+            // shape (that was tried first and rejected at insert: a child
+            // can't `after_ok` its own parent — see `pause_quiesce`'s doc
+            // comment for the exact error).
+            row("agent_window", None, &[]),
+            row("pause_signal", Some("agent_window"), &[]),
+            row(
+                "pause_drain",
+                Some("agent_window"),
+                &[("pause_signal", "done")]
+            ),
+        ]
+    );
+    assert_eq!(
+        declared_resources(&q, node_of(&q, "agent_window")),
+        vec![Resource::Agent("agent-a".to_owned())],
+        "the brace holds the lease for the whole pair"
+    );
+    assert_eq!(
+        [
+            declared_resources(&q, node_of(&q, "pause_signal")),
+            declared_resources(&q, node_of(&q, "pause_drain")),
+        ],
+        [vec![], vec![]],
+        "the pause pair borrows the brace's lease and declares nothing itself \
+         — same shape the stop quiesce pair uses"
+    );
+}
+
 #[test]
 fn spawn_shape_provision_create_dropin_reconcile() {
     let q = JobQueue::new(1);

hive-c0re/src/server.rs

@@ -329,24 +329,43 @@ async fn handle_spawn(coord: &Arc<Coordinator>, name: &str) -> Result<HostRespon
 /// `hivectl pause|resume` / the dashboard toggle: write or remove the
 /// agent's pause marker.
 ///
-/// Deliberately not a lifecycle DAG. There's no container operation to
-/// sequence — it's one marker file, and the harness picks it up on its
-/// next poll — so queueing it would only add latency and a lease. That
-/// also means it works on a stopped agent: the marker is sticky, so the
-/// agent comes up paused.
+/// **Resume** stays synchronous, direct marker removal — there's
+/// nothing to acknowledge (a resumed agent just starts driving turns
+/// again on its own next poll, no handshake needed), and nobody has
+/// asked resume to wait.
+///
+/// **Pause** rides the job queue (`PauseSignal → PauseDrain`, see
+/// `job_queue::power::pause_many`) instead of writing the marker
+/// synchronously here: pausing wants the same "confirmed, not just
+/// requested" signal a graceful stop gets from its `Signal → Drain`
+/// pair, and the DAG's agent lease is what stops a pause from racing
+/// an in-flight rebuild/stop of the same agent — a synchronous write
+/// had no such protection. Still works on a stopped agent (the queued
+/// `PauseSignal` writes the same sticky marker either way, container
+/// running or not).
 async fn handle_set_paused(
     coord: &std::sync::Arc<Coordinator>,
     name: &hive_types::Ident,
     paused: bool,
 ) -> HostResponse {
-    if let Err(e) = Coordinator::set_paused(name, paused).await {
-        return HostResponse::error(format!("set paused={paused} for {name}: {e}"));
+    if !paused {
+        if let Err(e) = Coordinator::set_paused(name, false).await {
+            return HostResponse::error(format!("set paused=false for {name}: {e}"));
+        }
+        tracing::info!(%name, "agent pause marker cleared");
+        // Refresh the dashboard's view so the paused badge flips without
+        // waiting for the next periodic rescan.
+        coord.rescan_containers_and_emit().await;
+        return HostResponse::success();
+    }
+    match crate::job_queue::power::pause_many(coord, std::slice::from_ref(&name.to_string())).await
+    {
+        Ok(ids) => {
+            tracing::info!(%name, "agent pause queued");
+            HostResponse::queued(ids.into_iter().map(hive_jobq::NodeId::get).collect())
+        }
+        Err(e) => HostResponse::error(format!("queue pause for {name}: {e}")),
     }
-    tracing::info!(%name, paused, "agent pause marker updated");
-    // Refresh the dashboard's view so the paused badge flips without
-    // waiting for the next periodic rescan.
-    coord.rescan_containers_and_emit().await;
-    HostResponse::success()
 }
 
 /// Collect per-agent status rows for `hivectl status` and the dashboard.

hive-c0re/src/socket_server/mod.rs

@@ -254,6 +254,13 @@ pub(crate) async fn dispatch_shared(
             coord.clear_graceful_stop(agent);
             hive_core_agent_sock::Response::Ok
         }
+        hive_core_agent_sock::Request::PauseAcknowledged => {
+            // Harness saw its own pause marker flip between turns: clear
+            // the fence so the pause DAG's drain node (which polls this
+            // flag) resolves without waiting out its timeout.
+            coord.clear_pause_pending(agent);
+            hive_core_agent_sock::Response::Ok
+        }
         hive_core_agent_sock::Request::GetHostJournal {
             unit,
             container,