archetype flows on the board list the agents the router sends to, and the docs say a switch-off key outlives its flow

docs/2-handbook/swarm/archetypes.md

@@ -163,7 +163,9 @@ within about 10 seconds, and the card has no `edit` button.
   flow can also request a review, which an archetype's flow can't. Switch the
   archetype's flow off to replace it with your copy.
 - A key for a flow or agent that no longer exists stays in the list and does nothing, so
-  editing an archetype never breaks the stored flows.
+  editing an archetype never breaks the stored flows. The key stays
+  after its flow goes: a later flow with the same archetype and id comes up
+  already switched off; switch it on from its card.
 
 ## How the list stays current
 

docs/4-implementation/swarm/notify-router.md

@@ -303,7 +303,7 @@ it refuses any id that's no built-in.
 `[{archetype, key, flow, members}]`, one per enabled archetype flow that passes
 the archetype check and has an agent, with the flow as the archetype declares it
 (a `todo` to `"members"`). The controller reads the archetype records and the
-`agent-archetype` keys itself; a read that fails leaves the list empty and the
+`agent-archetype` keys itself, and lists as members the agents the router sends to: those with a key that are in the agent-status roster. A read that fails leaves the list empty and the
 document still comes back. The page draws a card for each, with the `on` switch (the
 key), a checkbox per member (the key with `@<agent>`) and `duplicate`, which
 opens the editor on `?from=<key>`. The copy replaces `"members"` with the one

swarm-controller/src/notify_flows.rs

@@ -192,7 +192,12 @@ pub(crate) async fn get_flows(
 /// logged and leaves the list empty: the document is still served.
 async fn archetype_flows(client: &async_nats::Client) -> Vec<ArchetypeFlowView> {
     let found = async {
-        let refs = crate::agent_archetype::all(client).await?;
+        // The router's members: the archetype keys of the agents in the swarm.
+        let roster = swarm_queue_client::agent_status::roster(client)
+            .await
+            .map_err(|e| anyhow::anyhow!("listing the agents: {e}"))?;
+        let mut refs = crate::agent_archetype::all(client).await?;
+        refs.retain(|agent, _| roster.contains(agent));
         let mut records = BTreeMap::new();
         if let Some(store) = archetypes::open(client).await? {
             let sources: BTreeSet<&str> = refs.values().map(|r| r.source.as_str()).collect();
@@ -592,6 +597,8 @@ mod tests {
             ("alice", "triage"),
             ("bob", "triage"),
             ("carol", "reviewer"),
+            // In the bucket but not in the swarm: the router sends it nothing.
+            ("dave", "triage"),
         ] {
             let choice = ArchetypeRef {
                 source: BUILTIN_SOURCE.to_owned(),
@@ -603,6 +610,16 @@ mod tests {
                 .expect("put the choice");
         }
 
+        let status = swarm_queue_client::agent_status::open_or_create(&client)
+            .await
+            .expect("status");
+        for agent in ["alice", "bob", "carol"] {
+            status
+                .put(format!("hive.{agent}"), "{}".into())
+                .await
+                .expect("put the status");
+        }
+
         let Json(view) = get_flows(State(state.clone())).await.expect("get");
         assert_eq!(view.archetype_flows.len(), 1, "{:?}", view.archetype_flows);
         let listed = &view.archetype_flows[0];

swarm-notify-router/src/api.rs

@@ -95,11 +95,12 @@ async fn deliveries_answer(js: &jetstream::Context, payload: &[u8]) -> Vec<u8> {
 /// cannot be read leaves them out of the counts.
 async fn archetype_flows(
     client: &async_nats::Client,
-    js: &jetstream::Context,
     document: &swarm_wire::notify_flows::Document,
 ) -> Vec<crate::flow::ArchetypeFlow> {
     let found = async {
-        let roster = crate::roster_of(js).await.map_err(|e| format!("{e:#}"))?;
+        let roster = crate::roster_of(client)
+            .await
+            .map_err(|e| format!("{e:#}"))?;
         crate::archetype_flows::load(client, &roster).await
     }
     .await;
@@ -125,7 +126,7 @@ async fn dry_run_answer(
                 .check()
                 .map_err(|why| anyhow::anyhow!("the document fails its check: {why}"))?;
             let (flows, _) = request.document.active();
-            let archetype = archetype_flows(client, js, &request.document).await;
+            let archetype = archetype_flows(client, &request.document).await;
             let mut events: Vec<(String, Vec<u8>)> =
                 log::newest(js, forge_events::STREAM, DRY_RUN_EVENTS)
                     .await?

swarm-notify-router/src/main.rs

@@ -35,7 +35,7 @@ use std::time::{Duration, Instant};
 
 use anyhow::{Context as _, Result, bail};
 use async_nats::jetstream::{self, consumer::pull, kv, message::PublishMessage};
-use futures_util::{StreamExt as _, TryStreamExt as _};
+use futures_util::StreamExt as _;
 use metrics::Handled;
 use swarm_queue_client::{ClientSecret, QueueConfig};
 use swarm_secret_client::stored_secret::StoredSecret;
@@ -366,35 +366,18 @@ impl Router {
         {
             return Ok(agents.clone());
         }
-        let agents = roster_of(&self.js).await?;
+        let agents = roster_of(&self.nats).await?;
         self.roster = Some((Instant::now(), agents.clone()));
         Ok(agents)
     }
 }
 
 /// The agents of this swarm, read from the agent-status bucket.
-pub(crate) async fn roster_of(js: &jetstream::Context) -> Result<BTreeSet<String>> {
+pub(crate) async fn roster_of(client: &async_nats::Client) -> Result<BTreeSet<String>> {
-    let bucket = swarm_wire::agent_status::BUCKET;
+    swarm_queue_client::agent_status::roster(client)
-    let stream = js
-        .get_stream(format!("KV_{bucket}"))
         .await
-        .context("opening the agent-status bucket")?;
+        .map_err(|e| anyhow::anyhow!("{e}"))
-    let subjects: Vec<(String, usize)> = stream
+        .context("listing the agents in the agent-status bucket")
-        .info_with_subjects(format!("$KV.{bucket}.>"))
-        .await
-        .context("listing agent-status keys")?
-        .try_collect()
-        .await
-        .context("listing agent-status keys")?;
-    let prefix = format!("$KV.{bucket}.");
-    let agents: BTreeSet<String> = subjects
-        .into_iter()
-        .filter_map(|(subject, _)| {
-            let key = subject.strip_prefix(&prefix)?;
-            swarm_wire::agent_status::split_key(key).map(|(_, agent)| agent.to_owned())
-        })
-        .collect();
-    Ok(agents)
 }
 
 /// Why an event was not handled.

swarm-queue-client/Cargo.toml

@@ -16,7 +16,7 @@ edition.workspace = true
 # client; the exception exists because an agreement between two crates has
 # to live in one of them, and neither end of that bucket is senior to the
 # other.
-kv = ["async-nats/kv"]
+kv = ["async-nats/kv", "dep:futures-util"]
 # 🩸 `jetstream` is NOT in async-nats's default feature set here — the
 # workspace-level dependency turns default features off entirely (see
 # root `Cargo.toml`: `server_2_14`/`nkeys`/`ring` only). `kv` above works
@@ -83,6 +83,8 @@ router-todo = ["async-nats/jetstream"]
 # above - the connect itself needs none of them.
 async-nats.workspace = true
 hive-types = { workspace = true, optional = true }
+# `agent_status::roster` collects the stream's subject listing.
+futures-util = { workspace = true, optional = true }
 # `blocking` on top of the workspace default (`form`/`json`/`rustls`) —
 # `mint_token_for_blocking` needs `reqwest::blocking::Client` for a caller
 # with no tokio reactor to `.await` an async request on (an OTLP exporter's

swarm-queue-client/src/agent_status.rs

@@ -20,3 +20,34 @@ pub async fn open_or_create(
     )
     .await
 }
+
+/// The agents of the swarm: the agent names in the bucket's keys
+/// (`<hive>.<agent>`), read from the stream's subject list. This is the
+/// roster the notification router sends to.
+///
+/// # Errors
+///
+/// The bucket's stream cannot be opened or its subjects listed.
+#[cfg(feature = "kv")]
+pub async fn roster(
+    client: &async_nats::Client,
+) -> Result<std::collections::BTreeSet<String>, Box<dyn std::error::Error + Send + Sync>> {
+    use futures_util::TryStreamExt as _;
+
+    let bucket = swarm_wire::agent_status::BUCKET;
+    let js = async_nats::jetstream::new(client.clone());
+    let stream = js.get_stream(format!("KV_{bucket}")).await?;
+    let subjects: Vec<(String, usize)> = stream
+        .info_with_subjects(format!("$KV.{bucket}.>"))
+        .await?
+        .try_collect()
+        .await?;
+    let prefix = format!("$KV.{bucket}.");
+    Ok(subjects
+        .into_iter()
+        .filter_map(|(subject, _)| {
+            let key = subject.strip_prefix(&prefix)?;
+            swarm_wire::agent_status::split_key(key).map(|(_, agent)| agent.to_owned())
+        })
+        .collect())
+}