swarm-controller: create every agent's subagent stream

swarm-controller now creates `term-sub-<agent>` for every agent a hive is
declared to run, at start and every minute after, with the config
`swarm_queue_client::subagent_term::open_or_create` spells (subjects
`$SWARM.term.<agent>.sub.>`, max_age 24h). An existing stream is opened as
it is, as the controller does for its other streams and buckets, under the
`$JS.API.STREAM.CREATE.*` grant it already holds.

The agent no longer creates the stream: its token is granted publish on
`$SWARM.term.<agent>.sub.>` and no `$JS.API.STREAM.CREATE|INFO` subject,
and the subagent daemon only publishes. A `CREATE` carries the stream's
config in its payload, which no subject grant narrows, so the agent could
otherwise pick the stream's subjects and limits.

swarm-controller/Cargo.toml

@@ -90,7 +90,9 @@ swarm-authelia-bridge-sock.workspace = true
 # creation config are shared with the hive that writes it, so this end does
 # not get to declare them privately.
 #
-swarm-queue-client = { workspace = true, features = ["kv"] }
+# `subagent-term`: this daemon creates every agent's subagent stream, with the
+# config the crate spells.
+swarm-queue-client = { workspace = true, features = ["kv", "subagent-term"] }
 # `matrix_account.rs` writes the credential this daemon's route accepts. Same
 # crate the hive reads it back with, which is the point: the path, the field
 # name and the object's shape are agreements between the two ends, and a

swarm-controller/src/main.rs

@@ -2392,6 +2392,16 @@ fn spawn_agent_renewal(
     });
 }
 
+/// Start creating every live agent's subagent stream when a swarm queue is
+/// wired up. Lifted out of `main` for `clippy::too_many_lines`.
+fn spawn_subagent_streams(status: Option<&Arc<status::StatusReader>>, hives: &[HiveEntry]) {
+    let (Some(status), Some(wanted)) = (status, wanted_writer(status)) else {
+        return;
+    };
+    let hives = hives.iter().map(|h| h.name.clone()).collect();
+    subagent_term::spawn(status.queue_client(), wanted, hives);
+}
+
 /// Check an agent's forge token now, and mint one if it is missing or stale.
 ///
 /// The periodic pass (`forge::agent_token::spawn`) does the same every five
@@ -2920,6 +2930,7 @@ async fn main() -> Result<()> {
     let state_forge = keep_forge_for_state(forge_client, webhook_secret.clone());
 
     spawn_agent_renewal(&jobq, wanted_writer(status.as_ref()), &hives);
+    spawn_subagent_streams(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 —
     // either way it cannot collect what this daemon writes for it. A store

swarm-controller/src/subagent_term.rs

@@ -1,14 +1,13 @@
-//! An agent's subagents: which ones the swarm has rows for, and a live SSE
-//! relay of one subagent's terminal.
+//! An agent's subagents: the stream that keeps their rows, which ones the
+//! swarm has rows for, and a live SSE relay of one subagent's terminal.
 //!
 //! The agent's subagent daemon publishes each subagent's classified rows on
 //! `$SWARM.term.{agent}.sub.{subagent}` under the agent's own queue credential,
-//! into the stream `term-sub-{agent}` it creates itself
-//! (`swarm_queue_client::subagent_term`). Subagents are spawned on demand
+//! which may publish there and nothing else of these streams. [`spawn`]
+//! creates the stream `term-sub-{agent}` for every live agent, with the config
+//! `swarm_queue_client::subagent_term` spells. Subagents are spawned on demand
 //! under caller-chosen names, so the list is the set of subjects that stream
-//! holds rows for, read with `STREAM.INFO`. This daemon never creates the
-//! stream: a missing one is an agent with no subagent output yet, and lists
-//! nothing.
+//! holds rows for, read with `STREAM.INFO`. A missing stream lists nothing.
 //!
 //! **No hive lookup.** Unlike [`crate::term_stream`], there is no
 //! hive-scoped subject to follow: only the agent's own credential is granted
@@ -19,7 +18,10 @@
 //! while it is attached. Nothing is sent toward a subagent: subagents take no
 //! input from the swarm.
 
+use std::collections::BTreeSet;
 use std::convert::Infallible;
+use std::sync::Arc;
+use std::time::Duration;
 
 use axum::Json;
 use axum::extract::{Path, State};
@@ -29,6 +31,58 @@ use futures_util::{Stream, StreamExt as _, TryStreamExt as _};
 use swarm_queue_client::subagent_term;
 
 use crate::AppState;
+use crate::wanted::WantedWriter;
+
+/// How often [`spawn`] checks every live agent's stream. Also how long a newly
+/// declared agent's subagents can go unlisted.
+const ENSURE_INTERVAL: Duration = Duration::from_mins(1);
+
+/// Create the subagent stream of every live agent that has none, now and every
+/// [`ENSURE_INTERVAL`] after. An existing stream is left as it is.
+///
+/// The live agents are the ones some hive's wanted-state declaration names in
+/// a state other than `Destroyed` (`agent_renewal::live_agents`). A pass skips
+/// a hive whose declaration cannot be read and an agent whose stream cannot be
+/// created, logging each; a pass while the queue is down does nothing. It
+/// never stops the daemon.
+pub(crate) fn spawn(client: async_nats::Client, wanted: Arc<WantedWriter>, hives: Vec<String>) {
+    tokio::spawn(async move {
+        let mut ticker = tokio::time::interval(ENSURE_INTERVAL);
+        loop {
+            ticker.tick().await;
+            if swarm_queue_client::ensure_connected(&client).is_err() {
+                tracing::debug!("subagent streams: queue not connected; next pass");
+                continue;
+            }
+            for agent in live_agents(&wanted, &hives).await {
+                if let Err(e) = subagent_term::open_or_create(&client, &agent).await {
+                    tracing::warn!(
+                        %agent,
+                        error = %swarm_queue_client::chain(&e),
+                        "subagent streams: creating the agent's stream failed; next pass"
+                    );
+                }
+            }
+        }
+    });
+}
+
+/// Every agent the readable declarations of `hives` keep alive.
+async fn live_agents(wanted: &WantedWriter, hives: &[String]) -> BTreeSet<String> {
+    let mut declarations = Vec::with_capacity(hives.len());
+    for hive in hives {
+        match wanted.view(hive).await {
+            Ok(Some(d)) => declarations.push(d),
+            Ok(None) => {}
+            Err(e) => tracing::warn!(
+                %hive,
+                error = %format!("{e:#}"),
+                "subagent streams: reading the hive's wanted-state declaration failed"
+            ),
+        }
+    }
+    crate::agent_renewal::live_agents(&declarations)
+}
 
 #[utoipa::path(
     get,