an archetype carries its own notification-router flows for its agents

ArchetypeMeta gets flows, checked when the controller stores the record: every todo must go to the new members recipient, which the router expands to the agents of the archetype. The operator's flows document refuses members. The router reads the archetype records and each roster agent's archetype key, and its NATS grant gains read on both buckets.

swarm-notify-router/Cargo.toml

@@ -28,6 +28,9 @@ serde_json.workspace = true
 swarm-queue-client = { workspace = true, features = [
   "forge-events",
   "router-todo",
+  # `archetypes::open` and `agent_archetype::read`, for the flows archetypes carry.
+  "kv",
+  "agent-archetype",
 ] }
 # Its queue client secret, read with its own store leaf.
 swarm-secret-client = { workspace = true, features = ["otel"] }

swarm-notify-router/src/archetype_flows.rs

@@ -0,0 +1,364 @@
+//! The flows the archetypes carry, and who they run for.
+//!
+//! An archetype's record (`swarm_queue_client::archetypes`) lists flows in its
+//! `meta.flows`. Each runs, next to the built-ins and the operator's document
+//! ([`crate::stored`]), with a `members` recipient meaning the agents whose
+//! `agent-archetype` key (`swarm_queue_client::agent_archetype`) names that
+//! archetype.
+//!
+//! The router reads both buckets again every [`REFRESH`], for the agents of the
+//! roster only, so an agent that is not in the swarm is no member. Each read it
+//! cannot make or decode leaves the flows it already has; before any read
+//! succeeds there are none.
+
+use std::collections::{BTreeMap, BTreeSet};
+use std::time::{Duration, Instant};
+
+use async_nats::jetstream::kv;
+use swarm_queue_client::agent_archetype::{self, Read};
+use swarm_queue_client::archetypes::{self, SourceRecord};
+use swarm_wire::agent_archetype::ArchetypeRef;
+
+use crate::flow::ArchetypeFlow;
+
+/// How long a read of the buckets is reused.
+const REFRESH: Duration = Duration::from_secs(10);
+
+/// The archetype flows in force.
+#[derive(Default)]
+pub struct ArchetypeFlows {
+    pub active: Vec<ArchetypeFlow>,
+    /// The last problem logged, so a standing one is logged once.
+    problem: Option<String>,
+    checked: Option<Instant>,
+}
+
+impl ArchetypeFlows {
+    /// Read the buckets if the last read is older than [`REFRESH`]. `roster`
+    /// is the agents of the swarm.
+    pub async fn refresh(&mut self, client: &async_nats::Client, roster: &BTreeSet<String>) {
+        if self.checked.is_some_and(|at| at.elapsed() < REFRESH) {
+            return;
+        }
+        self.checked = Some(Instant::now());
+        match read(client, roster).await {
+            Ok((refs, records)) => {
+                self.problem = None;
+                self.set(build(&refs, &records));
+            }
+            Err(problem) => {
+                if self.problem.as_ref() != Some(&problem) {
+                    tracing::error!(%problem, "notify router: keeping the archetype flows it has");
+                    self.problem = Some(problem);
+                }
+            }
+        }
+    }
+
+    fn set(&mut self, (active, skipped): (Vec<ArchetypeFlow>, Vec<String>)) {
+        for why in &skipped {
+            tracing::error!(%why, "notify router: skipping an archetype flow");
+        }
+        if active != self.active {
+            tracing::info!(
+                flows = active.len(),
+                "notify router: running the archetype flows"
+            );
+            self.active = active;
+        }
+    }
+}
+
+/// Every roster agent's archetype, and the record of every source they name.
+async fn read(
+    client: &async_nats::Client,
+    roster: &BTreeSet<String>,
+) -> Result<
+    (
+        BTreeMap<String, ArchetypeRef>,
+        BTreeMap<String, SourceRecord>,
+    ),
+    String,
+> {
+    let none = Ok((BTreeMap::new(), BTreeMap::new()));
+    let Some(store) = archetypes::open(client)
+        .await
+        .map_err(|e| format!("opening the archetypes bucket failed: {e}"))?
+    else {
+        return none;
+    };
+    let mut refs = BTreeMap::new();
+    for agent in roster {
+        match agent_archetype::read(client, agent)
+            .await
+            .map_err(|e| format!("reading the archetype of {agent} failed: {e}"))?
+        {
+            Read::Current(Some(r)) => {
+                refs.insert(agent.clone(), r);
+            }
+            Read::Current(None) => {}
+            Read::NoBucket => return none,
+        }
+    }
+    let mut records = BTreeMap::new();
+    for source in refs
+        .values()
+        .map(|r| r.source.as_str())
+        .collect::<BTreeSet<_>>()
+    {
+        let entry = store
+            .entry(source)
+            .await
+            .map_err(|e| format!("reading the archetypes of {source} failed: {e}"))?;
+        if let Some(entry) = entry.filter(|e| e.operation == kv::Operation::Put) {
+            let record = serde_json::from_slice(&entry.value)
+                .map_err(|e| format!("the archetypes of {source} cannot be read: {e}"))?;
+            records.insert(source.to_owned(), record);
+        }
+    }
+    Ok((refs, records))
+}
+
+/// The flows to run: each flow of each archetype that has an agent, with those
+/// agents, and why any flow was left out. A flow's id is prefixed with its
+/// archetype's name, so it cannot be a built-in's or the document's, and no two
+/// archetypes' collide.
+fn build(
+    refs: &BTreeMap<String, ArchetypeRef>,
+    records: &BTreeMap<String, SourceRecord>,
+) -> (Vec<ArchetypeFlow>, Vec<String>) {
+    let mut agents: BTreeMap<(&str, &str), Vec<String>> = BTreeMap::new();
+    for (agent, r) in refs {
+        agents
+            .entry((r.source.as_str(), r.name.as_str()))
+            .or_default()
+            .push(agent.clone());
+    }
+    let mut active = Vec::new();
+    let mut skipped = Vec::new();
+    for ((source, name), members) in agents {
+        let Some(meta) = records
+            .get(source)
+            .and_then(|record| record.archetypes.get(name))
+        else {
+            continue;
+        };
+        let mut ids = BTreeSet::new();
+        for flow in meta.flows.iter().filter(|f| f.enabled) {
+            if !ids.insert(flow.id.as_str()) {
+                skipped.push(format!("{name}: flow {}: its id is used twice", flow.id));
+            } else if let Err(why) = flow.validate_archetype() {
+                skipped.push(format!("{name}: flow {}: {why}", flow.id));
+            } else {
+                let mut flow = flow.clone();
+                flow.id = format!("{name}:{}", flow.id);
+                active.push(ArchetypeFlow {
+                    flow,
+                    members: members.clone(),
+                });
+            }
+        }
+    }
+    (active, skipped)
+}
+
+#[cfg(test)]
+mod tests {
+    use swarm_queue_client::archetypes::ArchetypeMeta;
+    use swarm_wire::notify_flows::{Edge, Flow, Node, NodeKind, Recipients, TodoSink};
+    use swarm_wire::router_todo::Op;
+
+    use super::*;
+    use crate::test_nats::TestNats;
+
+    /// An archetype flow `id` on new issues, sending to `to`.
+    pub(crate) fn flow(id: &str, to: Recipients) -> Flow {
+        Flow {
+            id: id.to_owned(),
+            name: id.to_owned(),
+            enabled: true,
+            nodes: vec![
+                Node {
+                    id: "src".to_owned(),
+                    kind: NodeKind::Source {
+                        events: vec!["issues".to_owned()],
+                    },
+                },
+                Node {
+                    id: "tell".to_owned(),
+                    kind: NodeKind::Todo(TodoSink {
+                        to,
+                        op: Op::Upsert,
+                        key: "forge:{{ repo }}#{{ number }}:new".to_owned(),
+                        prefix: false,
+                        text: "new issue {{ repo }}#{{ number }}".to_owned(),
+                    }),
+                },
+            ],
+            edges: vec![Edge {
+                from: "src".to_owned(),
+                to: "tell".to_owned(),
+            }],
+        }
+    }
+
+    fn reference(name: &str) -> ArchetypeRef {
+        ArchetypeRef {
+            source: "hyperhive".to_owned(),
+            name: name.to_owned(),
+        }
+    }
+
+    fn record(archetypes: &[(&str, Vec<Flow>)]) -> BTreeMap<String, SourceRecord> {
+        let archetypes = archetypes
+            .iter()
+            .map(|(name, flows)| {
+                (
+                    (*name).to_owned(),
+                    ArchetypeMeta {
+                        description: "d".to_owned(),
+                        flows: flows.clone(),
+                        ..ArchetypeMeta::default()
+                    },
+                )
+            })
+            .collect();
+        BTreeMap::from([(
+            "hyperhive".to_owned(),
+            SourceRecord {
+                source: "hyperhive".to_owned(),
+                archetypes,
+                refused: BTreeMap::new(),
+            },
+        )])
+    }
+
+    fn refs(agents: &[(&str, &str)]) -> BTreeMap<String, ArchetypeRef> {
+        agents
+            .iter()
+            .map(|(agent, archetype)| ((*agent).to_owned(), reference(archetype)))
+            .collect()
+    }
+
+    fn members(built: &(Vec<ArchetypeFlow>, Vec<String>)) -> Vec<(&str, &[String])> {
+        built
+            .0
+            .iter()
+            .map(|a| (a.flow.id.as_str(), a.members.as_slice()))
+            .collect()
+    }
+
+    /// An archetype's flow runs for the agents of that archetype and no
+    /// others, under an id prefixed with the archetype's name.
+    #[test]
+    fn a_flow_runs_for_the_agents_of_its_archetype_only() {
+        let records = record(&[
+            ("triage", vec![flow("new-issues", Recipients::Members)]),
+            ("reviewer", vec![flow("new-issues", Recipients::Members)]),
+            ("unused", vec![flow("never", Recipients::Members)]),
+        ]);
+        let built = build(
+            &refs(&[
+                ("alice", "triage"),
+                ("bob", "triage"),
+                ("carol", "reviewer"),
+            ]),
+            &records,
+        );
+        assert!(built.1.is_empty(), "{:?}", built.1);
+        let alice_bob = ["alice".to_owned(), "bob".to_owned()];
+        let carol = ["carol".to_owned()];
+        assert_eq!(
+            members(&built),
+            [
+                ("reviewer:new-issues", &carol[..]),
+                ("triage:new-issues", &alice_bob[..])
+            ]
+        );
+    }
+
+    /// A flow that is not valid for an archetype, a disabled one and a second
+    /// one under the same id are left out and reported; the rest run.
+    #[test]
+    fn a_flow_that_is_not_valid_is_left_out_and_the_rest_run() {
+        let mut off = flow("off", Recipients::Members);
+        off.enabled = false;
+        let records = record(&[(
+            "triage",
+            vec![
+                flow("good", Recipients::Members),
+                flow("named", Recipients::Agent("atlas".to_owned())),
+                flow("good", Recipients::Members),
+                off,
+            ],
+        )]);
+        let built = build(&refs(&[("alice", "triage")]), &records);
+        assert_eq!(
+            built
+                .0
+                .iter()
+                .map(|a| a.flow.id.as_str())
+                .collect::<Vec<_>>(),
+            ["triage:good"]
+        );
+        assert_eq!(built.1.len(), 2, "{:?}", built.1);
+        assert!(built.1[0].contains("named"), "{:?}", built.1);
+        assert!(built.1[1].contains("used twice"), "{:?}", built.1);
+    }
+
+    /// Against a real server: the roster's agents are read from
+    /// `agent-archetype`, an agent outside the roster is no member, an agent
+    /// with no key is none either, and each source's record is read once.
+    #[tokio::test]
+    async fn the_buckets_are_read_for_the_roster_agents() {
+        let nats = TestNats::start();
+        let client = nats.client().await;
+        let roster: BTreeSet<String> = ["alice", "bob", "carol", "dave"].map(str::to_owned).into();
+
+        // Before anything made the buckets there is nothing to run.
+        let (r, rec) = read(&client, &roster).await.expect("no bucket is no flows");
+        assert!(r.is_empty() && rec.is_empty());
+
+        let records = archetypes::open_or_create(&client).await.expect("records");
+        let choices = agent_archetype::open_or_create(&client)
+            .await
+            .expect("choices");
+        let source = record(&[("triage", vec![flow("new-issues", Recipients::Members)])]);
+        records
+            .put(
+                "hyperhive",
+                serde_json::to_vec(&source["hyperhive"]).unwrap().into(),
+            )
+            .await
+            .expect("put the record");
+        for (agent, archetype) in [("alice", "triage"), ("bob", "triage"), ("eve", "triage")] {
+            choices
+                .put(
+                    agent,
+                    serde_json::to_vec(&reference(archetype)).unwrap().into(),
+                )
+                .await
+                .expect("put the choice");
+        }
+        // Carol's key was deleted: she is a custom agent again.
+        choices
+            .put(
+                "carol",
+                serde_json::to_vec(&reference("triage")).unwrap().into(),
+            )
+            .await
+            .expect("put");
+        choices.delete("carol").await.expect("delete");
+
+        let (r, rec) = read(&client, &roster).await.expect("a read");
+        let (active, skipped) = build(&r, &rec);
+        assert!(skipped.is_empty(), "{skipped:?}");
+        assert_eq!(
+            members(&(active, skipped)),
+            [(
+                "triage:new-issues",
+                &["alice".to_owned(), "bob".to_owned()][..]
+            )]
+        );
+    }
+}

swarm-notify-router/src/dry_run.rs

@@ -38,7 +38,7 @@ pub fn run(flows: &[Flow], events: &[(String, Vec<u8>)]) -> DryRun {
                 }
             }
         }
-        let routed = route::route(event_type, &delivery, &before, flows, |_| true);
+        let routed = route::route(event_type, &delivery, &before, flows, &[], |_| true);
         snapshots.insert(key, routed.next);
     }
     DryRun {

swarm-notify-router/src/flow.rs

@@ -82,13 +82,33 @@ pub fn reached<'f>(flow: &'f Flow, event_type: &str, fields: &Fields) -> Vec<&'f
     ids
 }
 
+/// A flow an archetype carries, with the agents of that archetype: who a
+/// `members` recipient expands to.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct ArchetypeFlow {
+    pub flow: Flow,
+    pub members: Vec<String>,
+}
+
 /// Walk `flow` for one event of type `event_type` and return the todo ops it
-/// emits. Assumes [`Flow::validate`] passed.
+/// emits. Assumes [`Flow::validate`] passed, so it has no `members` recipient;
+/// [`evaluate_for`] runs an archetype's flow.
 pub fn evaluate(flow: &Flow, event_type: &str, fields: &Fields) -> Vec<Action> {
+    evaluate_for(flow, event_type, fields, &[])
+}
+
+/// [`evaluate`] for a flow that passed [`Flow::validate_archetype`]: a
+/// `members` recipient expands to `members`.
+pub fn evaluate_for(
+    flow: &Flow,
+    event_type: &str,
+    fields: &Fields,
+    members: &[String],
+) -> Vec<Action> {
     let mut actions = Vec::new();
     walk(flow, event_type, fields, |node| {
         if let NodeKind::Todo(sink) = &node.kind {
-            actions.extend(emit(flow, node, sink, fields));
+            actions.extend(emit(flow, node, sink, fields, members));
         }
     });
     actions
@@ -126,9 +146,16 @@ fn request(flow: &Flow, node: &Node, reviewer: &str, fields: &Fields) -> Option<
     })
 }
 
-fn emit(flow: &Flow, node: &Node, sink: &TodoSink, fields: &Fields) -> Vec<Action> {
+fn emit(
+    flow: &Flow,
+    node: &Node,
+    sink: &TodoSink,
+    fields: &Fields,
+    members: &[String],
+) -> Vec<Action> {
     let recipients = match &sink.to {
         Recipients::Agent(agent) => vec![agent.clone()],
+        Recipients::Members => members.to_vec(),
         Recipients::Field(name) => match fields.get(name.as_str()) {
             Some(Value::List(items)) => items.clone(),
             Some(Value::Text(one)) => vec![one.clone()],

swarm-notify-router/src/main.rs

@@ -6,6 +6,7 @@
 //! recipient agent's subject of `ROUTER_TODOS` (`swarm_wire::router_todo`).
 //! Each agent's harness applies them to its own todo list. The flows are the
 //! built-in ones, changed by the `notify-flows` document ([`stored`]).
+//! The archetypes add flows of their own for their agents ([`archetype_flows`]).
 //!
 //! A `request_reviewer` node is a request to swarm-controller, which holds the
 //! forge token (`swarm_wire::router_sink`). It is sent once, after the event's
@@ -43,6 +44,7 @@ use swarm_wire::router_sink::{self, Outcome, RequestReviewer};
 use swarm_wire::{forge_events, notify_router, router_todo};
 
 mod api;
+mod archetype_flows;
 mod dry_run;
 mod event;
 mod flow;
@@ -129,6 +131,7 @@ struct Router {
     events: jetstream::consumer::Consumer<pull::Config>,
     state: kv::Store,
     flows: stored::Flows,
+    archetype_flows: archetype_flows::ArchetypeFlows,
     roster: Option<(Instant, BTreeSet<String>)>,
     metrics: metrics::Metrics,
 }
@@ -175,6 +178,7 @@ impl Router {
             events,
             state,
             flows: stored::Flows::builtins(),
+            archetype_flows: archetype_flows::ArchetypeFlows::default(),
             roster: None,
             metrics: metrics::Metrics::from_env(),
         })
@@ -234,9 +238,15 @@ impl Router {
         };
         self.flows.refresh(&self.js).await;
         let roster = self.roster().await.map_err(Skip::Retry)?;
-        let routed = route::route(&event_type, &delivery, &snapshot, &self.flows.active, |a| {
-            roster.contains(a)
-        });
+        self.archetype_flows.refresh(&self.nats, &roster).await;
+        let routed = route::route(
+            &event_type,
+            &delivery,
+            &snapshot,
+            &self.flows.active,
+            &self.archetype_flows.active,
+            |a| roster.contains(a),
+        );
         let flows: BTreeSet<&str> = routed
             .actions
             .iter()

swarm-notify-router/src/route.rs

@@ -8,7 +8,7 @@ use swarm_wire::router_todo::Op;
 use crate::event::{Delivery, Snapshot, conflicting_after, fields, is_assign};
 use swarm_wire::notify_flows::Flow;
 
-use crate::flow::{Action, ReviewRequest, evaluate, requests};
+use crate::flow::{Action, ArchetypeFlow, ReviewRequest, evaluate, evaluate_for, requests};
 
 /// What one event comes to.
 #[derive(Debug, PartialEq, Eq)]
@@ -23,13 +23,15 @@ pub struct Routed {
     pub next: Snapshot,
 }
 
-/// Run `flows` over one delivery of `event_type` on an issue whose snapshot is
-/// `snapshot`. `is_agent` says whether a recipient is an agent of this swarm.
+/// Run `flows`, and then the flows of `archetype`, over one delivery of
+/// `event_type` on an issue whose snapshot is `snapshot`. `is_agent` says
+/// whether a recipient is an agent of this swarm.
 pub fn route(
     event_type: &str,
     delivery: &Delivery,
     snapshot: &Snapshot,
     flows: &[Flow],
+    archetype: &[ArchetypeFlow],
     is_agent: impl Fn(&str) -> bool,
 ) -> Routed {
     let is_run = event_type.starts_with("workflow_run_");
@@ -54,7 +56,15 @@ pub fn route(
     for flow in flows {
         review_requests.extend(requests(flow, event_type, &fields));
     }
-    for action in flows.iter().flat_map(|f| evaluate(f, event_type, &fields)) {
+    let walked = flows
+        .iter()
+        .flat_map(|f| evaluate(f, event_type, &fields))
+        .chain(
+            archetype
+                .iter()
+                .flat_map(|a| evaluate_for(&a.flow, event_type, &fields, &a.members)),
+        );
+    for action in walked {
         if is_agent(&action.agent) {
             actions.push(action);
         } else if !skipped.contains(&action.agent) {
@@ -105,6 +115,17 @@ mod tests {
     use super::*;
     use swarm_wire::notify_flows::{Condition, Edge, Node, NodeKind, builtins};
 
+    /// `super::route` with no archetype flows.
+    fn route(
+        event_type: &str,
+        delivery: &Delivery,
+        snapshot: &Snapshot,
+        flows: &[Flow],
+        is_agent: impl Fn(&str) -> bool,
+    ) -> Routed {
+        super::route(event_type, delivery, snapshot, flows, &[], is_agent)
+    }
+
     fn delivery(action: &str, assignees: &[&str]) -> Delivery {
         Delivery {
             action: action.to_owned(),
@@ -631,4 +652,90 @@ mod tests {
         assert!(routed("pull_request_review_approved", "mara").is_empty());
         assert!(routed("pull_request_review_rejected", "review-classifier").is_empty());
     }
+
+    /// An archetype's flow on newly assigned issues, to its members.
+    fn members_flow() -> Flow {
+        let mut flow = builtins().remove(0);
+        flow.nodes[2].kind = NodeKind::Todo(swarm_wire::notify_flows::TodoSink {
+            to: swarm_wire::notify_flows::Recipients::Members,
+            op: Op::Upsert,
+            key: "forge:{{ repo }}#{{ number }}:triage".to_owned(),
+            prefix: false,
+            text: "triage {{ repo }}#{{ number }}".to_owned(),
+        });
+        assert_eq!(flow.validate_archetype(), Ok(()));
+        flow.id = "triage:new".to_owned();
+        flow
+    }
+
+    /// A `members` recipient is every agent of the archetype the flow came
+    /// from, in the roster; the agents of another archetype get nothing.
+    #[test]
+    fn an_archetype_flow_reaches_its_members_and_no_other_agent() {
+        let archetype = [
+            ArchetypeFlow {
+                flow: members_flow(),
+                members: vec!["alice".to_owned(), "bob".to_owned()],
+            },
+            ArchetypeFlow {
+                flow: {
+                    let mut other = members_flow();
+                    other.id = "reviewer:new".to_owned();
+                    other
+                },
+                members: vec!["carol".to_owned()],
+            },
+        ];
+        let d = delivery("assigned", &[]);
+        // `carol` is a member of another archetype's flow of the same shape;
+        // `bob` has left the roster.
+        let routed = super::route(
+            "issue_assign",
+            &d,
+            &Snapshot::default(),
+            &[],
+            &archetype[..1],
+            |a| a != "bob",
+        );
+        assert_eq!(agents(&routed), [("alice", Op::Upsert)]);
+        assert_eq!(routed.skipped, ["bob"]);
+        assert_eq!(routed.actions[0].op.key, "forge:o/r#9:triage");
+        assert_eq!(routed.actions[0].op.source, "router:triage:new");
+        assert!(routed.next.holders.contains("alice"));
+
+        // Control: both archetypes' flows, every agent in the roster.
+        let routed = super::route(
+            "issue_assign",
+            &d,
+            &Snapshot::default(),
+            &[],
+            &archetype,
+            |_| true,
+        );
+        assert_eq!(
+            agents(&routed),
+            [
+                ("alice", Op::Upsert),
+                ("bob", Op::Upsert),
+                ("carol", Op::Upsert)
+            ]
+        );
+        // The document's own flow for the event is not displaced.
+        let both = super::route(
+            "issue_assign",
+            &delivery("assigned", &["dave"]),
+            &Snapshot::default(),
+            &builtins(),
+            &archetype[..1],
+            |_| true,
+        );
+        assert_eq!(
+            agents(&both),
+            [
+                ("dave", Op::Upsert),
+                ("alice", Op::Upsert),
+                ("bob", Op::Upsert)
+            ]
+        );
+    }
 }