a hive-side pause holds against the swarm's declared state until an operator resumes

the harness writes the reason into the pause marker when it pauses itself and reports it as paused_by_hive in its turn-state header. swarm-controller declares such an agent paused (only if declared up) and shows the reason on the agents status row and the swarm ui card.

swarm-controller/src/agent_login.rs

@@ -7,6 +7,10 @@
 //! an agent has no entry for up to a minute after this daemon starts — the
 //! same gap [`crate::inference::Reported`] has on the same subject.
 //!
+//! A header may also carry `paused_by_hive`, the reason the agent's own harness
+//! paused its turn loop (`swarm_wire::agent_pause`); [`crate::pause_hold`]
+//! acts on it, and `GET /api/agents/status` shows it.
+//!
 //! The same header carries the harness's `capabilities`
 //! (`swarm_wire::hive_status`), which [`Reported::advertises`] answers from:
 //! the controller grants `logs` or `metrics` only to an agent whose running
@@ -18,14 +22,27 @@ use std::time::Duration;
 
 use futures_util::StreamExt as _;
 use swarm_wire::agent_login::Login;
+use swarm_wire::agent_pause::HivePause;
 use swarm_wire::hive_status::Capabilities;
 
+/// One agent's reported hive pause.
+#[derive(Debug)]
+struct Pause {
+    reason: String,
+    /// Whether [`crate::pause_hold`] has already dealt with this pause, so a
+    /// header re-sent after an operator resumed the agent does not pause it
+    /// again.
+    held: bool,
+}
+
 /// The latest login flag each agent reported, by agent.
 #[derive(Debug, Default)]
 pub(crate) struct Reported {
     latest: Mutex<HashMap<String, bool>>,
     /// The capabilities each agent's latest header listed.
     capabilities: Mutex<HashMap<String, Capabilities>>,
+    /// Each agent whose latest header carries a hive-pause reason.
+    pauses: Mutex<HashMap<String, Pause>>,
 }
 
 impl Reported {
@@ -75,6 +92,7 @@ impl Reported {
                 capabilities.remove(agent);
             }
         }
+        self.record_pause(agent, payload);
         let mut latest = self.latest.lock().unwrap_or_else(PoisonError::into_inner);
         match serde_json::from_slice::<Login>(payload) {
             Ok(Login {
@@ -92,6 +110,66 @@ impl Reported {
         }
     }
 
+    /// Keep one header's hive-pause reason. A header without one, or one that
+    /// does not decode, removes what was kept, which is how a resumed agent
+    /// is told from one that is still paused.
+    fn record_pause(&self, agent: &str, payload: &[u8]) {
+        let mut pauses = self.pauses.lock().unwrap_or_else(PoisonError::into_inner);
+        match serde_json::from_slice::<HivePause>(payload) {
+            Ok(HivePause {
+                paused_by_hive: Some(reason),
+            }) => {
+                pauses
+                    .entry(agent.to_owned())
+                    .and_modify(|p| p.reason.clone_from(&reason))
+                    .or_insert(Pause {
+                        reason,
+                        held: false,
+                    });
+            }
+            Ok(HivePause {
+                paused_by_hive: None,
+            })
+            | Err(_) => {
+                pauses.remove(agent);
+            }
+        }
+    }
+
+    /// `agent`'s hive-pause reason from its latest header; `None` when that
+    /// header carried none, or none arrived since this daemon started.
+    pub(crate) fn paused_by_hive(&self, agent: &str) -> Option<String> {
+        self.pauses
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+            .get(agent)
+            .map(|p| p.reason.clone())
+    }
+
+    /// The agents whose reported hive pause has not been held yet.
+    pub(crate) fn unheld_pauses(&self) -> Vec<String> {
+        self.pauses
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+            .iter()
+            .filter(|(_, p)| !p.held)
+            .map(|(agent, _)| agent.clone())
+            .collect()
+    }
+
+    /// Note that `agent`'s current hive pause has been dealt with. A no-op
+    /// once its header has stopped reporting one.
+    pub(crate) fn mark_pause_held(&self, agent: &str) {
+        if let Some(p) = self
+            .pauses
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+            .get_mut(agent)
+        {
+            p.held = true;
+        }
+    }
+
     /// Whether `agent`'s latest header lists `capability`. `false` for an
     /// agent that has sent none since this daemon started, or whose harness
     /// predates the field.
@@ -162,6 +240,44 @@ mod tests {
         assert_eq!(reported.get("iris"), None);
     }
 
+    fn pause_header(reason: Option<&str>) -> Vec<u8> {
+        let mut value: serde_json::Value = serde_json::from_slice(&header(None)).unwrap();
+        if let Some(reason) = reason {
+            value["paused_by_hive"] = reason.into();
+        }
+        serde_json::to_vec(&value).unwrap()
+    }
+
+    /// A header with the reason reports it, one without clears it, and a pause
+    /// held once stays held until a header without the reason has been seen.
+    #[test]
+    fn a_hive_pause_is_reported_until_a_header_clears_it() {
+        let reported = Reported::default();
+        let subject = swarm_queue_client::agent_state_subject("iris");
+        reported.record(&subject, &header(None));
+        assert_eq!(reported.paused_by_hive("iris"), None, "an old harness");
+        assert!(reported.unheld_pauses().is_empty());
+
+        reported.record(&subject, &pause_header(Some("skipped")));
+        assert_eq!(reported.paused_by_hive("iris").as_deref(), Some("skipped"));
+        assert_eq!(reported.unheld_pauses(), ["iris"]);
+        reported.mark_pause_held("iris");
+        reported.record(&subject, &pause_header(Some("skipped")));
+        assert!(reported.unheld_pauses().is_empty(), "held stays held");
+        assert_eq!(reported.paused_by_hive("iris").as_deref(), Some("skipped"));
+
+        reported.record(&subject, &pause_header(None));
+        assert_eq!(reported.paused_by_hive("iris"), None);
+        reported.record(&subject, &pause_header(Some("skipped")));
+        assert_eq!(
+            reported.unheld_pauses(),
+            ["iris"],
+            "a new pause is held anew"
+        );
+        reported.record(&subject, b"not json");
+        assert_eq!(reported.paused_by_hive("iris"), None);
+    }
+
     #[test]
     fn a_header_is_kept_under_the_agent_its_subject_names() {
         let reported = Reported::default();

swarm-controller/src/agent_status.rs

@@ -73,6 +73,11 @@ pub struct AgentStatusRow {
     /// `AppState::agent_login` — this reader sees the hive's bucket, and the
     /// flag comes from the agent itself.
     pub needs_login: Option<bool>,
+    /// Why the agent's own harness paused its turn loop, from its latest
+    /// turn-state header; `None` when that header named none, none arrived
+    /// since swarm-controller started, or the harness predates the field.
+    /// Filled in by the `GET /api/agents/status` handler like `needs_login`.
+    pub paused_by_hive: Option<String>,
     /// The archetype the agent is built from; `None` for a custom agent. Always
     /// `None` out of this reader, filled in by the `GET /api/agents/status`
     /// handler from the `agent-archetype` bucket.
@@ -202,6 +207,7 @@ fn render(
                 config_pr: None,
                 wanted: None,
                 needs_login: None,
+                paused_by_hive: None,
                 archetype: None,
             },
         })
@@ -248,6 +254,7 @@ fn row(
         config_pr: None,
         wanted: None,
         needs_login: None,
+        paused_by_hive: None,
         archetype: None,
     }
 }

swarm-controller/src/main.rs

@@ -70,6 +70,7 @@ mod notify_flows;
 mod notify_router_api;
 mod observability_grant;
 mod otel_http_client;
+mod pause_hold;
 mod pr_report;
 mod proxy_credential;
 mod pulls;
@@ -237,6 +238,7 @@ async fn main() -> Result<()> {
     read_policy::ensure_agent_policies().await;
 
     let state = AppState {
+        agent_login: startup::agent_login(status.as_ref(), &hives),
         hives: Arc::new(hives),
         links: Arc::new(load_links()),
         wanted: wanted_writer(status.as_ref()),
@@ -244,7 +246,6 @@ async fn main() -> Result<()> {
         agent_icons: agent_icon_reader(status.as_ref()),
         agents_md: agents_md_reader(status.as_ref()),
         reported_inference: reported_inference(status.as_ref()),
-        agent_login: startup::agent_login(status.as_ref()),
         runs: review_classifier::runs::Runs::Live(status.as_ref().map(|s| s.queue_client())),
         builds: builder::runs::Builds::Live(status.as_ref().map(|s| s.queue_client())),
         status,

swarm-controller/src/pause_hold.rs

@@ -0,0 +1,133 @@
+//! Holds an agent the hive paused, so the swarm's declaration stops undoing it.
+//!
+//! An agent whose harness paused its own turn loop says so in its turn-state
+//! header (`paused_by_hive`, see [`crate::agent_login::Reported`]). The hive
+//! resumes any paused agent that the swarm declares `Up`, so this declares it
+//! `Paused` instead, which the hive's pause marker already agrees with. Only an
+//! operator declaring the agent `Up` again resumes it.
+//!
+//! Each pause is held once. A header re-sent after the operator's resume and
+//! before the hive cleared the marker still reports it, and must not pause the
+//! agent a second time. The agent is only declared `Paused` while it is
+//! declared `Up`: an operator's `Offline` or `Destroyed` stays what it is.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use anyhow::Result;
+
+use crate::agent_login::Reported;
+use crate::wanted::WantedWriter;
+
+/// How often the reported pauses are looked at.
+const POLL: Duration = Duration::from_secs(5);
+
+/// Hold the pauses `login` learns of, for as long as this daemon runs.
+pub(crate) fn spawn(login: Arc<Reported>, wanted: Arc<WantedWriter>, hives: Vec<String>) {
+    tokio::spawn(async move {
+        loop {
+            tokio::time::sleep(POLL).await;
+            hold_pending(&login, &wanted, &hives).await;
+        }
+    });
+}
+
+/// Hold every reported pause not held yet. A pause that could not be held is
+/// tried again on the next call.
+pub(crate) async fn hold_pending(login: &Reported, wanted: &WantedWriter, hives: &[String]) {
+    for agent in login.unheld_pauses() {
+        match hold(wanted, hives, &agent).await {
+            Ok(()) => login.mark_pause_held(&agent),
+            Err(e) => tracing::warn!(
+                %agent,
+                error = %format!("{e:#}"),
+                "holding a hive-paused agent failed; will retry"
+            ),
+        }
+    }
+}
+
+/// Declare `agent` `Paused` on whichever hive declares it `Up`.
+async fn hold(wanted: &WantedWriter, hives: &[String], agent: &str) -> Result<()> {
+    for hive in hives {
+        if wanted.pause_if_up(hive, agent).await? {
+            tracing::info!(%hive, %agent, "the hive paused this agent; declared it paused");
+        }
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::hold_pending;
+    use crate::agent_login::Reported;
+    use crate::test_nats::TestNats;
+    use crate::wanted::WantedWriter;
+    use swarm_wire::wanted::AgentState;
+
+    fn header(reason: Option<&str>) -> Vec<u8> {
+        let mut header = serde_json::json!({ "agent_state": "paused", "needs_login": false });
+        if let Some(reason) = reason {
+            header["paused_by_hive"] = reason.into();
+        }
+        serde_json::to_vec(&header).unwrap()
+    }
+
+    async fn declared(wanted: &WantedWriter, agent: &str) -> Option<AgentState> {
+        let declaration = wanted.view("pr1ma").await.unwrap()?;
+        declaration.agents.get(agent).map(|a| a.state)
+    }
+
+    /// The whole fix: a hive-paused agent declared `Up` is declared `Paused`,
+    /// the operator's `Up` after that sticks, and the next pause is held too.
+    #[tokio::test]
+    async fn a_hive_pause_replaces_up_and_an_operator_up_sticks() {
+        let nats = TestNats::start();
+        let wanted = WantedWriter::new(nats.client().await);
+        let hives = vec!["pr1ma".to_owned()];
+        let login = Reported::default();
+        let subject = swarm_queue_client::agent_state_subject("atlas");
+        wanted.set("pr1ma", "atlas", AgentState::Up).await.unwrap();
+        wanted.set("pr1ma", "iris", AgentState::Up).await.unwrap();
+
+        login.record(&subject, &header(Some("skipped get_loose_ends")));
+        hold_pending(&login, &wanted, &hives).await;
+        assert_eq!(declared(&wanted, "atlas").await, Some(AgentState::Paused));
+        assert_eq!(declared(&wanted, "iris").await, Some(AgentState::Up));
+
+        // The operator resumes; the hive has not cleared its marker yet and
+        // re-sends the header.
+        wanted.set("pr1ma", "atlas", AgentState::Up).await.unwrap();
+        login.record(&subject, &header(Some("skipped get_loose_ends")));
+        hold_pending(&login, &wanted, &hives).await;
+        assert_eq!(declared(&wanted, "atlas").await, Some(AgentState::Up));
+
+        // The hive clears the marker, then pauses the agent again.
+        login.record(&subject, &header(None));
+        login.record(&subject, &header(Some("skipped get_loose_ends")));
+        hold_pending(&login, &wanted, &hives).await;
+        assert_eq!(declared(&wanted, "atlas").await, Some(AgentState::Paused));
+    }
+
+    /// An operator's `Offline` is not overruled by the agent's pause, and an
+    /// agent no hive declares leaves the pause to be tried again.
+    #[tokio::test]
+    async fn only_an_agent_declared_up_is_paused() {
+        let nats = TestNats::start();
+        let wanted = WantedWriter::new(nats.client().await);
+        let hives = vec!["pr1ma".to_owned()];
+        let login = Reported::default();
+        wanted
+            .set("pr1ma", "atlas", AgentState::Offline)
+            .await
+            .unwrap();
+
+        login.record(
+            &swarm_queue_client::agent_state_subject("atlas"),
+            &header(Some("skipped get_loose_ends")),
+        );
+        hold_pending(&login, &wanted, &hives).await;
+        assert_eq!(declared(&wanted, "atlas").await, Some(AgentState::Offline));
+        assert!(login.unheld_pauses().is_empty(), "nothing left to hold");
+    }
+}

swarm-controller/src/routes/agents.rs

@@ -471,6 +471,7 @@ pub(crate) async fn get_agents_status(
             if let Some(login) = state.agent_login.as_ref() {
                 for row in &mut rows {
                     row.needs_login = login.get(&row.name);
+                    row.paused_by_hive = login.paused_by_hive(&row.name);
                 }
             }
             // Absent on a failed read, like `wanted`: one unreadable bucket
@@ -1816,6 +1817,7 @@ mod tests {
             config_pr: None,
             wanted: None,
             needs_login: None,
+            paused_by_hive: None,
             archetype: None,
         }
     }

swarm-controller/src/startup.rs

@@ -31,9 +31,9 @@ use crate::{
     agent_permissions, agent_renewal, agent_state_stream, agent_status, agents_md, alerts,
     archetypes, auth, backend_env, builder, claude_login, config_drift, config_pr, egress_allow,
     egress_deny, forge, forge_account, github_account, gitlab_account, inference, issue_report,
-    linked_accounts, matrix_account, message_stream, notify_flows, notify_router_api, pr_report,
-    pulls, queue_identity, review_classifier, schedules, status, subagent_term, term_stream,
-    wanted, webhook,
+    linked_accounts, matrix_account, message_stream, notify_flows, notify_router_api, pause_hold,
+    pr_report, pulls, queue_identity, review_classifier, schedules, status, subagent_term,
+    term_stream, wanted, webhook,
 };
 
 /// Where the daemon binds, overridable via `SWARM_CONTROLLER_SOCKET`.
@@ -206,11 +206,17 @@ pub(crate) fn reported_inference(
 }
 
 /// Whether each agent last said it needs a login, read on the status
-/// reader's connection — same rationale as [`wanted_writer`].
+/// reader's connection — same rationale as [`wanted_writer`] — and, from the
+/// same headers, which agents their hive paused: those are declared `Paused`
+/// (see [`pause_hold`]).
 pub(crate) fn agent_login(
     status: Option<&Arc<status::StatusReader>>,
+    hives: &[HiveEntry],
 ) -> Option<Arc<agent_login::Reported>> {
-    Some(agent_login::Reported::spawn(status?.queue_client()))
+    let login = agent_login::Reported::spawn(status?.queue_client());
+    let hives = hives.iter().map(|h| h.name.clone()).collect();
+    pause_hold::spawn(Arc::clone(&login), wanted_writer(status)?, hives);
+    Some(login)
 }
 
 /// The homeserver agents' own matrix accounts are minted on. Unset is this

swarm-controller/src/wanted.rs

@@ -40,17 +40,23 @@ enum Wrote {
 /// and a caller that gets an error can ask again with fresh intent.
 const MAX_ATTEMPTS: usize = 5;
 
+/// A hive's current declaration; none yet reads as an empty one, an undecodable
+/// value as an error.
+fn decode(current: Option<&[u8]>) -> Result<HiveWanted> {
+    match current {
+        Some(raw) => serde_json::from_slice::<HiveWanted>(raw)
+            .context("the hive's current declaration is not decodable"),
+        None => Ok(HiveWanted::default()),
+    }
+}
+
 /// Apply one agent's declared state to a hive's current declaration.
 ///
 /// `current` is the hive's whole agent map — an undecodable value is an
 /// **error**, not treated as absent, so a bad read never discards every
 /// other agent's declaration under a fresh-start fallback.
 fn apply(current: Option<&[u8]>, agent: &str, state: AgentState) -> Result<(HiveWanted, Vec<u8>)> {
-    let mut declaration = match current {
-        Some(raw) => serde_json::from_slice::<HiveWanted>(raw)
-            .context("the hive's current declaration is not decodable")?,
-        None => HiveWanted::default(),
-    };
+    let mut declaration = decode(current)?;
     declaration
         .agents
         .insert(agent.to_owned(), AgentWanted { state });
@@ -98,11 +104,37 @@ impl WantedWriter {
 
     /// Declare `agent` on `hive` to be in `state`, unconditionally (see the
     /// module doc), and return the whole declaration as published.
+    pub async fn set(&self, hive: &str, agent: &str, state: AgentState) -> Result<HiveWanted> {
+        self.write(hive, agent, state, None)
+            .await
+            .map(|(declaration, _)| declaration)
+    }
+
+    /// Declare `agent` on `hive` `Paused` if, and only if, the declaration at
+    /// the moment of the write says `Up`: an operator who has since set the
+    /// agent offline, destroyed or already paused is not overruled. `true`
+    /// when the declaration now says `Paused` because of this call.
+    pub async fn pause_if_up(&self, hive: &str, agent: &str) -> Result<bool> {
+        self.write(hive, agent, AgentState::Paused, Some(AgentState::Up))
+            .await
+            .map(|(_, applied)| applied)
+    }
+
+    /// The write behind [`Self::set`] and [`Self::pause_if_up`]: the
+    /// declaration as published and whether `state` was written. With
+    /// `only_if`, the write happens only while the agent is declared in that
+    /// state.
     ///
     /// Read-modify-write against the entry's revision, not a plain `put`:
     /// the value is the hive's whole agent map, so a blind write would drop
     /// a concurrent change to a different agent.
-    pub async fn set(&self, hive: &str, agent: &str, state: AgentState) -> Result<HiveWanted> {
+    async fn write(
+        &self,
+        hive: &str,
+        agent: &str,
+        state: AgentState,
+        only_if: Option<AgentState>,
+    ) -> Result<(HiveWanted, bool)> {
         swarm_queue_client::ensure_connected(&self.client)?;
         let store = swarm_queue_client::wanted::open_or_create(&self.client, hive).await?;
 
@@ -112,8 +144,14 @@ impl WantedWriter {
                 .await
                 .with_context(|| format!("reading the declaration for {hive}"))?;
             let revision = entry.as_ref().map(|e| e.revision);
-            let (declaration, encoded) =
-                apply(entry.as_ref().map(|e| e.value.as_ref()), agent, state)?;
+            let current = entry.as_ref().map(|e| e.value.as_ref());
+            if let Some(from) = only_if {
+                let declaration = decode(current)?;
+                if declaration.agents.get(agent).map(|a| a.state) != Some(from) {
+                    return Ok((declaration, false));
+                }
+            }
+            let (declaration, encoded) = apply(current, agent, state)?;
 
             // A redeclare of the identical state re-encodes the declaration
             // unchanged (a caller that already sees this state and asks
@@ -124,7 +162,7 @@ impl WantedWriter {
             // between reads.
             if entry.as_ref().map(|e| e.value.as_ref()) == Some(encoded.as_slice()) {
                 tracing::debug!(hive, agent, "declaration already current, no write needed");
-                return Ok(declaration);
+                return Ok((declaration, true));
             }
 
             // `update` and `create` have separate error types, and only one
@@ -149,7 +187,7 @@ impl WantedWriter {
             match written {
                 Wrote::Ok => {
                     tracing::info!(hive, agent, ?state, "declared agent state");
-                    return Ok(declaration);
+                    return Ok((declaration, true));
                 }
                 // Re-read and re-apply onto the winner's value, not over it.
                 Wrote::LostRace => {