swarm-controller: mint each hive's matrix sender token

A hive whose homeserver runs on another host has no local
matrix-appservice-token, so hive-c0re's matrix sweep returned before
reaching the store read in ensure_hive_user: no @hive-<name>: token, no
Space, no chat room, no invites, and a sweep-health banner.

swarm-controller now mints @hive-<name>: with the swarm appservice
token for every hive in its directory, as a MintHiveSenderToken job
node queued by a five-minute pass, and stores it at
swarm/hives/<name>/matrix/sender-token, the same matrix::Credential
swarm-matrix-ctl writes there. It is keep-if-live, reusing agent_token's
classify/plan: a stored token whoami confirms as @hive-<name>: is left
alone, so only an absent or dead one is minted. agent_token's probe and
mint steps are lifted into probe_at/mint_at so both passes share them.

swarm-matrix-ctl mint still writes the path for its own hive when it is
empty. If both mint an empty path at once, one token is invalidated
(same pinned device); the next pass classifies it Revoked and re-mints.

hive-c0re's ensure_all no longer returns when there is no local
as_token. ensure_hive_user reads the store first on every sweep and
overwrites its token file when the store's token differs, keeps the
file when the store has none, mints with the local as_token only when
neither holds one, and fails with one error when there is nothing at
all. The decision is sender_source, unit-tested.

The controller's bao policy gains create/read/update on
swarm/hives/+/matrix/sender-token (`+`, since `*` is a glob only at the
end of a path), pinned in module-eval.

Refs #4427

swarm-controller/src/main.rs

@@ -123,6 +123,10 @@ enum SwarmNodeKind {
     /// Carries no hive: the path it writes has no hive segment, so an agent
     /// that moves between hives keeps one matrix identity.
     MintAgentMatrixAccount { agent: String },
+    /// Make sure `hive`'s sender account (`@hive-<hive>:`) holds a live token
+    /// in the swarm store, where that hive's matrix sweep reads it. See
+    /// `matrix_account::hive_sender`.
+    MintHiveSenderToken { hive: String },
     /// Re-mint `agent`'s queue secret if it is still stored and old enough.
     /// See `agent_renewal`.
     ///
@@ -154,6 +158,7 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind {
             SwarmNodeKind::MintAgentIdentity { .. } => "mint_agent_identity".to_owned(),
             SwarmNodeKind::MintAgentForgeToken { .. } => "mint_agent_forge_token".to_owned(),
             SwarmNodeKind::MintAgentMatrixAccount { .. } => "mint_agent_matrix_account".to_owned(),
+            SwarmNodeKind::MintHiveSenderToken { .. } => "mint_hive_sender_token".to_owned(),
             SwarmNodeKind::RenewAgentQueueCredential { .. } => {
                 "renew_agent_queue_credential".to_owned()
             }
@@ -181,6 +186,7 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind {
             | SwarmNodeKind::RenewAgentQueueCredential { agent } => {
                 serde_json::json!({ "agent": agent })
             }
+            SwarmNodeKind::MintHiveSenderToken { hive } => serde_json::json!({ "hive": hive }),
             SwarmNodeKind::TriggerDeploy { hive, agent }
             | SwarmNodeKind::SetAgentWanted { hive, agent } => {
                 serde_json::json!({ "agent": agent, "hive": hive })
@@ -329,6 +335,9 @@ async fn run_swarm_node(
         SwarmNodeKind::MintAgentMatrixAccount { agent } => {
             mint_matrix_account(deps.matrix_homeserver.as_deref(), &agent).await
         }
+        SwarmNodeKind::MintHiveSenderToken { hive } => {
+            mint_hive_sender(deps.matrix_homeserver.as_deref(), &hive).await
+        }
         SwarmNodeKind::RenewAgentQueueCredential { agent } => {
             match agent_renewal::renew(&agent).await {
                 Ok(()) => Outcome::Done,
@@ -408,6 +417,23 @@ async fn mint_matrix_account(
     }
 }
 
+/// The `MintHiveSenderToken` arm, lifted out for the same reason.
+async fn mint_hive_sender(homeserver: Option<&str>, hive: &str) -> hive_jobq::scheduler::Outcome {
+    use hive_jobq::scheduler::Outcome;
+
+    let Some(base) = homeserver else {
+        return Outcome::Failed(format!(
+            "no matrix homeserver configured on this host ({} unset), so no hive \
+             sender token can be minted",
+            matrix_account::DEFAULT_HOMESERVER_ENV
+        ));
+    };
+    match matrix_account::hive_sender::ensure_hive_sender_token(base, hive).await {
+        Ok(()) => Outcome::Done,
+        Err(e) => Outcome::Failed(format!("{e:#}")),
+    }
+}
+
 /// Declare a brand-new agent at [`NEW_AGENT_WANTED_STATE`] — unless it turns
 /// out not to be new: an agent that already has a declaration (other than
 /// `Destroyed`, which this treats as reusable) is left alone, so a retried
@@ -2132,6 +2158,27 @@ fn queue_matrix_account_mints(
     Ok(ids)
 }
 
+/// Insert one `MintHiveSenderToken` job per hive and return the nodes' ids.
+/// `matrix_account::hive_sender::spawn`'s periodic pass is the one caller.
+fn queue_hive_sender_mints(
+    sched: &Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>,
+    hives: Vec<String>,
+) -> Result<Vec<hive_jobq::NodeId>> {
+    let mut sched = sched
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
+    let mut ids = Vec::with_capacity(hives.len());
+    for hive in hives {
+        let queued = sched
+            .insert_job(None, |b| {
+                vec![b.node(SwarmNodeKind::MintHiveSenderToken { hive }).guid()]
+            })
+            .map_err(|e| anyhow::anyhow!("{e}"))?;
+        ids.extend(queued);
+    }
+    Ok(ids)
+}
+
 /// The homeserver agents' own matrix accounts are minted on. Unset is this
 /// controller creating agents with no matrix account, which the mint node
 /// reports by name rather than `main` refusing to start.
@@ -2141,21 +2188,30 @@ fn configured_matrix_homeserver() -> Option<Arc<str>> {
         .map(Arc::from)
 }
 
-/// Start the matrix-account backfill when a homeserver is configured. Lifted
-/// out of `main` for `clippy::too_many_lines`.
+/// Start the matrix-account backfill and the hive sender-token pass when a
+/// homeserver is configured. Lifted out of `main` for
+/// `clippy::too_many_lines`.
 fn spawn_matrix_account_backfill(
     jobq: &Arc<Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>>,
     homeserver: Option<Arc<str>>,
+    hives: &[HiveEntry],
 ) {
     let Some(base) = homeserver else {
         return;
     };
     let sched = Arc::clone(jobq);
-    matrix_account::agent_token::spawn(base, move |agents| {
+    matrix_account::agent_token::spawn(Arc::clone(&base), move |agents| {
         if let Err(e) = queue_matrix_account_mints(&sched, agents) {
             tracing::warn!(error = %format!("{e:#}"), "agent matrix accounts: queueing failed");
         }
     });
+    let sched = Arc::clone(jobq);
+    let hive_names = hives.iter().map(|h| h.name.clone()).collect();
+    matrix_account::hive_sender::spawn(base, hive_names, move |hives| {
+        if let Err(e) = queue_hive_sender_mints(&sched, hives) {
+            tracing::warn!(error = %format!("{e:#}"), "hive sender tokens: queueing failed");
+        }
+    });
 }
 
 /// Start the forge's periodic passes — the swarm-wide objects, and agents'
@@ -2714,7 +2770,8 @@ async fn main() -> Result<()> {
         hive_jobq::Graph::new(),
         hive_jobq::resources::ResourceTable::new(),
     )));
-    spawn_matrix_account_backfill(&jobq, deps.matrix_homeserver.clone());
+    let hives = load_hives();
+    spawn_matrix_account_backfill(&jobq, deps.matrix_homeserver.clone(), &hives);
     spawn_jobq_worker(Arc::clone(&jobq), deps);
     // Bound to a named variable, not `_` — dropping the provider stops its
     // `PeriodicReader`, so it must live as long as `main` does (which it
@@ -2771,7 +2828,6 @@ async fn main() -> Result<()> {
     let config_prs = forge_client.clone().map(config_pr::spawn);
     let state_forge = keep_forge_for_state(forge_client, webhook_secret.clone());
 
-    let hives = load_hives();
     spawn_agent_renewal(&jobq, wanted_writer(status.as_ref()), &hives);
     // Before serving, because a hive whose role does not exist cannot log in,
     // and one whose policy does not exist logs in able to read nothing —
@@ -4370,6 +4426,37 @@ mod tests {
         assert_eq!(agents, ["a", "b"]);
     }
 
+    /// The hive sender-token pass's queueing: one mint node per hive, and
+    /// the node names the hive rather than an agent.
+    #[test]
+    fn a_queued_hive_sender_mint_is_one_node_per_hive() {
+        use hive_jobq_wire::WireNode as _;
+
+        let sched = std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new(
+            hive_jobq::Graph::new(),
+            hive_jobq::resources::ResourceTable::new(),
+        ));
+        let ids =
+            super::queue_hive_sender_mints(&sched, vec!["alpha".to_owned(), "beta".to_owned()])
+                .expect("two jobs insert");
+        assert_eq!(ids.len(), 2);
+        let guard = sched
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        let mut hives: Vec<String> = guard
+            .graph()
+            .nodes()
+            .map(|n| {
+                assert_eq!(n.payload.label(), "mint_hive_sender_token");
+                let data = n.payload.data(n.id.get());
+                assert!(data.get("agent").is_none(), "{data}");
+                data["hive"].as_str().expect("hive is a string").to_owned()
+            })
+            .collect();
+        hives.sort();
+        assert_eq!(hives, ["alpha", "beta"]);
+    }
+
     /// The manual route and the periodic pass both go through
     /// `queue_forge_token_mints`, so this is the assertion that a backfilled
     /// mint is the same pair of nodes agent creation inserts: the forge user,

swarm-controller/src/matrix_account.rs

@@ -4,6 +4,7 @@
 //! the swarm's homeserver, minted with the swarm's appservice token and stored
 //! where the agent's daemon reads it. See docs/swarm/credentials.md for why
 //! `main` is the one name [`put_matrix_account`] refuses to write.
+//! [`hive_sender`] mints each **hive's** sender account the same way, sharing this module's helpers with [`agent_token`].
 //!
 //! **The external one** — [`put_matrix_account`] and everything under it — is
 //! an account somewhere else that an operator hands us a credential for: put it
@@ -39,6 +40,7 @@ use utoipa::ToSchema;
 use super::{AppState, error_problem, swarm_hive};
 
 pub mod agent_token;
+pub mod hive_sender;
 
 fn default_mode() -> String {
     "token".to_owned()

swarm-controller/src/matrix_account/agent_token.rs

@@ -107,7 +107,7 @@ pub fn plan(observed: &[(String, Observed)]) -> Vec<String> {
 }
 
 /// The swarm appservice token, or an error naming why there is none.
-async fn appservice_token(store: &SecretStore) -> Result<String> {
+pub(super) async fn appservice_token(store: &SecretStore) -> Result<String> {
     let path = matrix::swarm_appservice_token_path()?;
     let stored: Option<matrix::Credential> = store
         .read_optional(&path)
@@ -129,9 +129,18 @@ async fn probe(
     base: &str,
     agent: &str,
 ) -> Result<Probe> {
-    let path = matrix::account_path(agent, ACCOUNT)?;
+    probe_at(store, http, base, &matrix::account_path(agent, ACCOUNT)?).await
+}
+
+/// What the store and the homeserver say about the token stored at `path`.
+pub(super) async fn probe_at(
+    store: &SecretStore,
+    http: &reqwest::Client,
+    base: &str,
+    path: &str,
+) -> Result<Probe> {
     let stored: Option<matrix::Credential> = store
-        .read_optional(&path)
+        .read_optional(path)
         .await
         .with_context(|| format!("reading {path}"))?;
     let Some(stored) = stored else {
@@ -170,35 +179,52 @@ pub async fn ensure_agent_matrix_account(base: &str, agent: &str) -> Result<()>
         Decision::Mint(reason) => reason,
     };
 
-    let token = match homeserver::register(&http, base, agent, &as_token).await? {
+    let path = matrix::account_path(agent, ACCOUNT)?;
+    mint_at(&store, &http, base, agent, &as_token, &path).await?;
+    tracing::info!(agent, ?reason, %path, "agent matrix account token minted and stored");
+    Ok(())
+}
+
+/// Create the account `account` (a localpart), or log in to it as the appservice when it
+/// exists, store the token at `path`, and read it back with `whoami`.
+///
+/// # Errors
+/// When the store or the homeserver refuses a step, or the new token
+/// authenticates as someone else.
+pub(super) async fn mint_at(
+    store: &SecretStore,
+    http: &reqwest::Client,
+    base: &str,
+    account: &str,
+    as_token: &str,
+    path: &str,
+) -> Result<()> {
+    let token = match homeserver::register(http, base, account, as_token).await? {
         homeserver::Registered::Token(token) => token,
-        // An agent minted before, or one a hive created back when hives did
-        // this: log in as the appservice on the same device instead.
+        // An account minted before, or one a hive created itself: log in as
+        // the appservice on the same device instead.
         homeserver::Registered::AlreadyExists => {
-            homeserver::appservice_login(&http, base, agent, &as_token).await?
+            homeserver::appservice_login(http, base, account, as_token).await?
         }
     };
-    let path = matrix::account_path(agent, ACCOUNT)?;
     store
         .write(
-            &path,
+            path,
             &matrix::Credential {
                 value: token.clone(),
                 homeserver: Some(base.to_owned()),
             },
         )
         .await
-        .with_context(|| format!("storing {agent}'s matrix token at {path}"))?;
+        .with_context(|| format!("storing @{account}'s matrix token at {path}"))?;
 
-    match homeserver::whoami(&http, base, &token)
+    match homeserver::whoami(http, base, &token)
         .await
-        .with_context(|| format!("using {agent}'s new matrix token"))?
+        .with_context(|| format!("using @{account}'s new matrix token"))?
     {
-        Whoami::User(user) if localpart(&user) == Some(agent) => {}
-        _ => bail!("{agent}'s new matrix token does not authenticate as {agent}"),
+        Whoami::User(user) if localpart(&user) == Some(account) => Ok(()),
+        _ => bail!("@{account}'s new matrix token does not authenticate as @{account}"),
     }
-    tracing::info!(agent, ?reason, %path, "agent matrix account token minted and stored");
-    Ok(())
 }
 
 /// One pass: every agent holding a store identity, observed.

swarm-controller/src/matrix_account/hive_sender.rs

@@ -0,0 +1,201 @@
+//! Each hive's sender account, `@hive-<hive>:`, on the swarm's homeserver:
+//! created here with the **swarm's** appservice token and stored at
+//! `swarm/hives/<hive>/matrix/sender-token`, where hive-c0re's matrix sweep
+//! reads it under the hive's own store identity. A hive whose homeserver runs
+//! elsewhere holds no appservice token, so this is its only sender token.
+//!
+//! Runs for every hive in the directory, local or remote, and decides the
+//! same way [`super::agent_token`] does: a stored token that `whoami`
+//! confirms as `@hive-<hive>:` is kept, so this writes only when the path is
+//! empty or its token is dead.
+//!
+//! ⚠️ `swarm-matrix-ctl mint` also writes this path for the hive whose host
+//! runs the homeserver, and skips when it is non-empty. Both log in on the
+//! same pinned device, so if both find it empty at once, one of the two
+//! tokens is dead on arrival. Whichever of them lands in the store, the next
+//! [`RECONCILE_INTERVAL`] pass either keeps it (live) or re-mints it
+//! (`Revoked`), and matrix-ctl never writes a non-empty path — so the store
+//! converges on one live token, and the hive's sweep takes whatever it holds.
+
+use std::sync::Arc;
+
+use anyhow::{Context, Result};
+use swarm_matrix_client as homeserver;
+use swarm_secret_client::matrix;
+
+use super::agent_token::{Decision, Observed, appservice_token, classify, mint_at, plan, probe_at};
+
+/// How often [`spawn`] re-checks every hive's sender token.
+const RECONCILE_INTERVAL: std::time::Duration = std::time::Duration::from_mins(5);
+
+/// Decide what to do about `hive`'s sender token from what the stored one is.
+fn classify_hive(hive: &str, probe: &super::agent_token::Probe) -> Decision {
+    classify(&matrix::hive_localpart(hive), probe)
+}
+
+/// Make sure `hive`'s sender account holds a live token in the store,
+/// creating the account or logging in to it when it does not. The whole job
+/// of the `MintHiveSenderToken` node.
+///
+/// # Errors
+/// When the store or the homeserver refuses a step, the swarm appservice
+/// token has not been published, or the new token authenticates as someone
+/// else.
+pub async fn ensure_hive_sender_token(base: &str, hive: &str) -> Result<()> {
+    let store = crate::store::connect()
+        .await
+        .context("logging in to the swarm secret store")?;
+    let as_token = appservice_token(&store).await?;
+    let http = homeserver::client()?;
+    let path = matrix::sender_token_path(hive)?;
+    let reason = match classify_hive(hive, &probe_at(&store, &http, base, &path).await?) {
+        Decision::Keep => {
+            tracing::debug!(hive, "hive sender token is current; left as it is");
+            return Ok(());
+        }
+        Decision::Mint(reason) => reason,
+    };
+    let localpart = matrix::hive_localpart(hive);
+    mint_at(&store, &http, base, &localpart, &as_token, &path).await?;
+    tracing::info!(hive, ?reason, %path, "hive sender token minted and stored");
+    Ok(())
+}
+
+/// One pass: every hive in `hives`, observed.
+///
+/// Fails as a whole when the swarm appservice token is not published, for
+/// the reason `agent_token`'s pass does.
+async fn observe_all(base: &str, hives: &[String]) -> Result<Vec<(String, Observed)>> {
+    let store = crate::store::connect()
+        .await
+        .context("logging in to the swarm secret store")?;
+    appservice_token(&store).await?;
+    let http = homeserver::client()?;
+    let mut observed = Vec::with_capacity(hives.len());
+    for hive in hives {
+        let o = match matrix::sender_token_path(hive) {
+            Ok(path) => match probe_at(&store, &http, base, &path).await {
+                Ok(p) => Observed::Decided(classify_hive(hive, &p)),
+                Err(e) => {
+                    tracing::warn!(hive, error = %format!("{e:#}"), "hive sender token: read failed");
+                    Observed::Unknown
+                }
+            },
+            Err(e) => {
+                tracing::warn!(hive, error = %e, "hive sender token: the hive name forms no store path");
+                Observed::Unknown
+            }
+        };
+        observed.push((hive.clone(), o));
+    }
+    Ok(observed)
+}
+
+/// Check every hive's sender token now and every [`RECONCILE_INTERVAL`]
+/// after, and hand the hives that need one to `enqueue`, which inserts a
+/// `MintHiveSenderToken` node for each.
+///
+/// The roster is the controller's hive directory. A pass that fails is
+/// logged and retried on the next tick; it never stops the daemon.
+pub fn spawn(base: Arc<str>, hives: Vec<String>, enqueue: impl Fn(Vec<String>) + Send + 'static) {
+    tokio::spawn(async move {
+        let mut ticker = tokio::time::interval(RECONCILE_INTERVAL);
+        loop {
+            ticker.tick().await;
+            match observe_all(&base, &hives).await {
+                Ok(observed) => {
+                    let minting = plan(&observed);
+                    if minting.is_empty() {
+                        tracing::debug!(
+                            checked = observed.len(),
+                            "hive sender tokens: all current"
+                        );
+                    } else {
+                        tracing::info!(
+                            checked = observed.len(),
+                            minting = minting.len(),
+                            "hive sender tokens: queueing mints"
+                        );
+                        enqueue(minting);
+                    }
+                }
+                Err(e) => tracing::warn!(
+                    error = %format!("{e:#}"),
+                    retry_in_s = RECONCILE_INTERVAL.as_secs(),
+                    "hive sender tokens: pass failed; retrying next tick"
+                ),
+            }
+        }
+    });
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::agent_token::{MintReason, Probe};
+    use super::*;
+    use swarm_matrix_client::Whoami;
+
+    fn user(id: &str) -> Probe {
+        Probe::Whoami(Whoami::User(id.to_owned()))
+    }
+
+    #[test]
+    fn a_live_token_for_the_hive_account_is_kept() {
+        assert_eq!(
+            classify_hive("pr1ma", &user("@hive-pr1ma:t.local")),
+            Decision::Keep
+        );
+    }
+
+    #[test]
+    fn nothing_stored_mints() {
+        // A hive whose homeserver is remote: nothing has ever written its path.
+        assert_eq!(
+            classify_hive("pr1ma", &Probe::NotStored),
+            Decision::Mint(MintReason::NotStored)
+        );
+    }
+
+    #[test]
+    fn a_dead_token_mints() {
+        // What the loser of a simultaneous mint with matrix-ctl leaves behind.
+        assert_eq!(
+            classify_hive("pr1ma", &Probe::Whoami(Whoami::UnknownToken)),
+            Decision::Mint(MintReason::Revoked)
+        );
+    }
+
+    #[test]
+    fn a_token_for_another_account_mints() {
+        // The bare hive name is an agent's localpart, not the hive's account,
+        // and another hive's account is not this one's.
+        for other in ["@pr1ma:t.local", "@hive-beta:t.local", "@hive:t.local"] {
+            assert_eq!(
+                classify_hive("pr1ma", &user(other)),
+                Decision::Mint(MintReason::OtherUser),
+                "{other}"
+            );
+        }
+    }
+
+    #[test]
+    fn an_outage_plans_nothing() {
+        // A mint replaces the token the hive is running on, so a pass that
+        // could not read must not mint for every hive.
+        let observed = [
+            ("alpha".to_owned(), Observed::Unknown),
+            ("beta".to_owned(), Observed::Unknown),
+        ];
+        assert!(plan(&observed).is_empty());
+    }
+
+    #[test]
+    fn the_published_path_is_the_one_the_hive_reads() {
+        // hive-c0re's `stored_sender_token` and `swarm-matrix-ctl mint` both
+        // resolve this path; the literal is what the bao grant names.
+        assert_eq!(
+            matrix::sender_token_path("pr1ma").expect("a plain name is legal"),
+            "swarm/hives/pr1ma/matrix/sender-token"
+        );
+    }
+}