From 13ff33ead016fb255596253bc11f2375a85b6e92 Mon Sep 17 00:00:00 2001 From: atlas Date: Wed, 11 Nov 2026 20:45:07 +0100 Subject: [PATCH] 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 | 4 ++- docs/4-implementation/swarm/notify-router.md | 2 +- swarm-controller/src/notify_flows.rs | 19 +++++++++++- swarm-notify-router/src/api.rs | 7 +++-- swarm-notify-router/src/main.rs | 29 ++++-------------- swarm-queue-client/Cargo.toml | 4 ++- swarm-queue-client/src/agent_status.rs | 31 ++++++++++++++++++++ 7 files changed, 66 insertions(+), 30 deletions(-) diff --git a/docs/2-handbook/swarm/archetypes.md b/docs/2-handbook/swarm/archetypes.md index 5b99025f0..23bd5db3d 100644 --- a/docs/2-handbook/swarm/archetypes.md +++ b/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 diff --git a/docs/4-implementation/swarm/notify-router.md b/docs/4-implementation/swarm/notify-router.md index 6efdfc280..7e330f446 100644 --- a/docs/4-implementation/swarm/notify-router.md +++ b/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 `@`) and `duplicate`, which opens the editor on `?from=`. The copy replaces `"members"` with the one diff --git a/swarm-controller/src/notify_flows.rs b/swarm-controller/src/notify_flows.rs index 9c47b90fe..d37988c53 100644 --- a/swarm-controller/src/notify_flows.rs +++ b/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 { 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]; diff --git a/swarm-notify-router/src/api.rs b/swarm-notify-router/src/api.rs index c717c422f..078812c5d 100644 --- a/swarm-notify-router/src/api.rs +++ b/swarm-notify-router/src/api.rs @@ -95,11 +95,12 @@ async fn deliveries_answer(js: &jetstream::Context, payload: &[u8]) -> Vec { /// 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 { 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)> = log::newest(js, forge_events::STREAM, DRY_RUN_EVENTS) .await? diff --git a/swarm-notify-router/src/main.rs b/swarm-notify-router/src/main.rs index e3db39f03..fb3005b04 100644 --- a/swarm-notify-router/src/main.rs +++ b/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> { - let bucket = swarm_wire::agent_status::BUCKET; - let stream = js - .get_stream(format!("KV_{bucket}")) +pub(crate) async fn roster_of(client: &async_nats::Client) -> Result> { + swarm_queue_client::agent_status::roster(client) .await - .context("opening the agent-status bucket")?; - let subjects: Vec<(String, usize)> = stream - .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 = 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) + .map_err(|e| anyhow::anyhow!("{e}")) + .context("listing the agents in the agent-status bucket") } /// Why an event was not handled. diff --git a/swarm-queue-client/Cargo.toml b/swarm-queue-client/Cargo.toml index bd77e8984..99c744f6e 100644 --- a/swarm-queue-client/Cargo.toml +++ b/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 diff --git a/swarm-queue-client/src/agent_status.rs b/swarm-queue-client/src/agent_status.rs index 8a8696ec6..0023757c4 100644 --- a/swarm-queue-client/src/agent_status.rs +++ b/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 +/// (`.`), 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, Box> { + 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()) +}