diff --git a/docs/2-handbook/swarm/archetypes.md b/docs/2-handbook/swarm/archetypes.md index c3c4bc090..d3cee7d40 100644 --- a/docs/2-handbook/swarm/archetypes.md +++ b/docs/2-handbook/swarm/archetypes.md @@ -61,7 +61,9 @@ hyperhiveArchetypes.reviewer.meta.prompt = builtins.readFile ./prompts/reviewer. `prompt` is optional text that prefills `AGENTS.md`, [below](#the-prompt). `icon` is optional SVG text, at most 16 KiB, that the create form shows on the archetype's tile; `builtins.readFile ./icon.svg` gives it. The swarm refuses an - `icon` that isn't SVG text or is over the limit, like any bad `meta`, and ignores + `icon` that isn't SVG text or is over the limit, like any bad `meta`. + `flows` is an optional list of notification router flows, + [below](#notification-flows). The swarm ignores other keys in `meta`. - `module` is a NixOS module for the agent's container. The meta flake imports it into each agent created from the archetype. @@ -93,6 +95,54 @@ archetype without `prompt` leaves the box alone. Where the files live: [the implementation page](../../4-implementation/swarm/archetypes.md#the-built-in-prompts). +## Notification flows + +An archetype's `meta.flows` lists [notification router +flows](../../4-implementation/swarm/notify-router.md#flows) that run for every +agent of the archetype, next to the built-in flows and the ones in the +operator's document. A `todo` node in them sends to `members`: every agent +created from the archetype. + +```nix +hyperhiveArchetypes.triage.meta.flows = [ + { + id = "new-issues"; + name = "new issues go to triage"; + nodes = [ + { id = "src"; kind = "source"; events = [ "issues" ]; } + { id = "opened"; kind = "match"; when = [ { field = "action"; equals = "opened"; } ]; } + { + id = "tell"; + kind = "todo"; + to = "members"; + op = "upsert"; + key = "forge:{{ repo }}#{{ number }}:triage"; + text = "triage {{ repo }}#{{ number }}: {{ title }}"; + } + ]; + edges = [ + { from = "src"; to = "opened"; } + { from = "opened"; to = "tell"; } + ]; + } +]; +``` + +- A flow here has the shape of a flow in the operator's document, with two + differences. Every `todo` must send to `members`. A flow that names an agent + or an event field's agents, or asks for a review, belongs in the operator's + document, and the swarm refuses it as it refuses any bad `meta`: the + archetype leaves the list and the reason goes into `refused`. The operator's + document in turn refuses `members`, since it has no archetype. +- Flow ids are unique within the archetype. +- Use a todo key that starts with `forge:{{ repo }}#{{ number }}:`, as the example + does. The built-in `closed` flow clears todos by that prefix when the issue + or PR closes. +- A change to the archetype's flows, or to who belongs to it, takes effect within + about 10 seconds of the next forge event. An agent outside the swarm gets + nothing. +- Custom agents have no archetype and get no archetype flows. + ## How the list stays current The swarm controller asks the builder for the list when the queue connects and diff --git a/docs/4-implementation/swarm/archetypes.md b/docs/4-implementation/swarm/archetypes.md index 0e3676a4a..87a3256dd 100644 --- a/docs/4-implementation/swarm/archetypes.md +++ b/docs/4-implementation/swarm/archetypes.md @@ -25,6 +25,8 @@ it stores. The operator's view is under the key `hyperhive`, when it differs from the stored one. The list leaves out an archetype that fails the check, and its name and reason go into the record's `refused` map and the controller's journal. + An archetype's `flows` pass the router's [archetype check](notify-router.md#flows) here too; the + notification router reads them from this record. 4. `GET /api/archetypes` answers from that record, and the create form lists from it. diff --git a/docs/4-implementation/swarm/notify-router.md b/docs/4-implementation/swarm/notify-router.md index 4cf1d8dba..bf47a25fb 100644 --- a/docs/4-implementation/swarm/notify-router.md +++ b/docs/4-implementation/swarm/notify-router.md @@ -85,7 +85,7 @@ each one runs, so one event can reach more than one leaf. A flow must be a tree: one source, one parent for every other node, every node reachable, no children under a `todo` or a `request_reviewer`. The most is 64 nodes, 16 levels deep, and 100 custom flows. A `todo` node sends to the agents an event -field holds (`assignees_added`, `todo_holders`, …) or to one named agent. +field holds (`assignees_added`, `todo_holders`, …) or to one named agent. An archetype's flows also send to its members ([below](#archetype-flows)). The router ships the eight flows above, each a source → match on `action` → todo (`conflict` and `conflict-cleared` also match `mergeability_changed`), and `head-classifier`: source `pull_request` and `pull_request_sync` → @@ -274,6 +274,21 @@ Before the router has read any document, "before" means the built-ins, so a bad document never leaves it with no flows. The controller refuses to overwrite a document of a version it doesn't know. +**Archetype flows.** An archetype carries flows of its own in its record's +`meta.flows` ([operator page](../../2-handbook/swarm/archetypes.md#notification-flows)). +Each is an ordinary flow with one more recipient, `Recipients::Members` (`"members"` +on the wire): every agent whose `agent-archetype` key names the archetype the +flow came from. The controller checks each one with `Flow::validate_archetype` when it stores the record: the check +of any flow, except that every `todo` goes to `members` and no node requests a +review. The operator's document refuses `members` (`Flow::validate`), because it has +no archetype. The router reads the `agent-archetypes` record and the +`agent-archetype` key of each agent in its roster at most every 10 seconds, when an event arrives, and runs +each archetype's valid, enabled flows after the built-ins and the document's flows, for the +archetype's agents. The router logs a flow it can't use and leaves it out. A read that fails keeps the archetype flows it had. +Each runs under the id `:`, so its todos' source is +`router::`. A recipient not in the roster gets nothing, the same as a named agent that doesn't exist. +A todo key is the dedupe key as for any flow: the same key from the operator's document replaces the archetype's todo. + **Deploying in any order.** The controller, the router and the UI can be on different hosts. A router that predates `request_reviewer` reads a document holding such a node (a custom flow with one) as one it can't read and keeps the @@ -283,6 +298,7 @@ as before. A router that has `request_reviewer`, with a controller that predates it, gets no answer to the request and logs that. A router that predates the `conflict` flows ignores the `pull_request_mergeability` events a newer controller sends, and gives no conflict todos; a controller that predates them sends none, so the new flows never run. A controller that predates this leaves the key absent, and the router runs the built-ins as before. A router that predates this ignores the document, so the page and the API show a state the router isn't yet running. +A router that predates archetype flows ignores `meta.flows`, and a controller that predates them drops the field when it stores the record, so the agents of an archetype get no router todos from it until both are new. Deploy `swarm-nats-auth` before the router: without its read grant on the two archetype buckets, each read times out, the router logs it and runs without archetype flows, and the event that triggered the read waits for the timeout. ## Deliveries and dry runs @@ -355,11 +371,11 @@ route: the page, if newer, shows the failure. ## Credentials and grants -| Principal | Grant | -| --------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `swarm-notify-router`, in its container | store: `read` on `swarm/services//oidc/client`. Queue: its own consumer on `SWARM_EVENTS` (create, pull, ack) and `STREAM.MSG.GET` on it for the dry run, publish on `$SWARM.router.todo.*` and on `$SWARM.router.sink.request-reviewer`, create `ROUTER_TODOS`, create `NOTIFY_ROUTER_LOG`, publish its one subject and read it by sequence, its own `notify-router-state` bucket, read on the `notify-flows` bucket, `STREAM.INFO` on the agent-status bucket, which lists the agents, and subscribe to the two `$SWARM.router.api.*` request subjects and answer each once, for 10 seconds | -| swarm-controller | publish on `$SWARM.events.forge.*` and on the two `$SWARM.router.api.*` request subjects; subscribe to `$SWARM.router.sink.request-reviewer` and answer each request once, for 30 seconds; creates and writes the `notify-flows` bucket (every bucket, by its `$KV.*.>` grant) | -| an agent, under its own credential | create its own consumer `todo-` in the form whose filter is `$SWARM.router.todo.`, pull from it, and ack what it delivered | +| Principal | Grant | +| --------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | +| `swarm-notify-router`, in its container | store: `read` on `swarm/services//oidc/client`. Queue: its own consumer on `SWARM_EVENTS` (create, pull, ack) and `STREAM.MSG.GET` on it for the dry run, publish on `$SWARM.router.todo.*` and on `$SWARM.router.sink.request-reviewer`, create `ROUTER_TODOS`, create `NOTIFY_ROUTER_LOG`, publish its one subject and read it by sequence, its own `notify-router-state` bucket, read on the `notify-flows` bucket, read on the `agent-archetypes` bucket (with `STREAM.INFO`) and the `agent-archetype` bucket, `STREAM.INFO` on the agent-status bucket, which lists the agents, and subscribe to the two `$SWARM.router.api.*` request subjects and answer each once, for 10 seconds | +| swarm-controller | publish on `$SWARM.events.forge.*` and on the two `$SWARM.router.api.*` request subjects; subscribe to `$SWARM.router.sink.request-reviewer` and answer each request once, for 30 seconds; creates and writes the `notify-flows` bucket (every bucket, by its `$KV.*.>` grant) | +| an agent, under its own credential | create its own consumer `todo-` in the form whose filter is `$SWARM.router.todo.`, pull from it, and ack what it delivered | No other client may publish on `$SWARM.router.todo.*`, `$SWARM.router.log`, the two `$SWARM.router.api.*` subjects, `$SWARM.router.sink.request-reviewer` or diff --git a/swarm-controller/src/archetypes.rs b/swarm-controller/src/archetypes.rs index 5c49d4d95..7939a882b 100644 --- a/swarm-controller/src/archetypes.rs +++ b/swarm-controller/src/archetypes.rs @@ -24,6 +24,7 @@ use swarm_queue_client::archetypes::{ use swarm_wire::builder::{ ARCHETYPES_REPLY_DEADLINE, ARCHETYPES_SUBJECT, ArchetypesReply, EvalArchetypes, }; +use swarm_wire::notify_flows::{Flow, MAX_FLOWS}; use utoipa::ToSchema; use crate::routes::{AppState, error_problem}; @@ -61,6 +62,28 @@ struct DeclaredMeta { prompt: Option, #[serde(default)] icon: Option, + #[serde(default)] + flows: Vec, +} + +/// The flows an archetype carries: each passes [`Flow::validate_archetype`], +/// ids are unique and there are no more than the operator's document may hold. +fn check_flows(flows: &[Flow]) -> Result<(), String> { + if flows.len() > MAX_FLOWS { + return Err(format!( + "flows: {} flows, more than {MAX_FLOWS}", + flows.len() + )); + } + let mut ids = std::collections::BTreeSet::new(); + for flow in flows { + if !ids.insert(flow.id.as_str()) { + return Err(format!("flows: id {} is used twice", flow.id)); + } + flow.validate_archetype() + .map_err(|why| format!("flow {}: {why}", flow.id))?; + } + Ok(()) } /// `meta` as stored, or why a create would refuse it. @@ -84,6 +107,7 @@ fn check(name: &str, meta: serde_json::Value) -> Result { .map(|i| check_icon(&i).map(|()| i)) .transpose() .map_err(|e| format!("icon: {e}"))?; + check_flows(&declared.flows)?; let permissions = match declared.permissions { Some(perms) => { agent_permissions::unique(&perms)?; @@ -112,6 +136,7 @@ fn check(name: &str, meta: serde_json::Value) -> Result { .map_err(|e| format!("inference: {e}"))?, prompt: declared.prompt.filter(|p| !p.trim().is_empty()), icon, + flows: declared.flows, }) } @@ -548,6 +573,57 @@ mod tests { assert!(r.refused["twice"].contains("listed twice")); } + /// A `todo` node to `to`, below a source of `issues`, as an archetype's + /// `meta.flows` entry in the JSON a source flake declares. + fn flow_to(id: &str, to: &serde_json::Value) -> serde_json::Value { + json!({ + "id": id, + "name": "new issues", + "nodes": [ + { "id": "src", "kind": "source", "events": ["issues"] }, + { "id": "tell", "kind": "todo", "to": to, "op": "upsert", + "key": "forge:{{ repo }}#{{ number }}:new", "text": "new issue {{ title }}" } + ], + "edges": [{ "from": "src", "to": "tell" }] + }) + } + + #[test] + fn an_archetype_flow_to_members_is_stored_and_any_other_recipient_is_refused() { + let r = evaluated(json!({ + "ok": { "description": "x", "flows": [flow_to("new", &json!("members"))] }, + "agent": { "description": "x", "flows": [flow_to("new", &json!({ "agent": "triage" }))] }, + "field": { "description": "x", "flows": [flow_to("new", &json!({ "field": "author" }))] }, + "twice": { "description": "x", "flows": [ + flow_to("new", &json!("members")), flow_to("new", &json!("members")) + ] }, + "review": { "description": "x", "flows": [{ + "id": "rev", + "name": "review", + "nodes": [ + { "id": "src", "kind": "source", "events": ["pull_request"] }, + { "id": "ask", "kind": "request_reviewer", "reviewer": "argus" } + ], + "edges": [{ "from": "src", "to": "ask" }] + }] }, + "broken": { "description": "x", "flows": [{ "id": "x" }] } + })); + assert_eq!(r.archetypes.keys().collect::>(), ["ok"]); + assert_eq!(r.archetypes["ok"].flows.len(), 1); + assert_eq!(r.archetypes["ok"].flows[0].id, "new"); + for refused in ["agent", "field"] { + assert!( + r.refused[refused].contains("operator's flows document"), + "{refused}: {}", + r.refused[refused] + ); + } + assert!(r.refused["twice"].contains("used twice")); + assert!(r.refused["review"].contains("cannot request a review")); + assert!(r.refused.contains_key("broken")); + assert_eq!(r.refused.len(), 5); + } + /// Against a real server: no builder listening names the option; a /// listener's evaluation is checked and stored, and storing the same /// record again writes nothing. diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index 8f7bf3d98..39ac68180 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -292,9 +292,9 @@ impl Policy { /// subjects, which is what makes the router their sole writer. /// - **Its state bucket.** Create, look up, read and write /// `notify-router-state`, and no other bucket. - /// - **The flows document.** `STREAM.INFO` and `DIRECT.GET` on - /// `notify-flows`, which it reads and never creates or writes: the - /// controller creates the bucket and is its one writer. + /// - **The flows document and the archetypes' flows.** `STREAM.INFO` and + /// `DIRECT.GET` on `notify-flows` and `agent-archetypes`, `DIRECT.GET` on + /// `agent-archetype`; it never creates or writes them: the controller does. /// - **The roster.** `STREAM.INFO` on the agent-status bucket, which lists /// its keys (`.`) and writes none. /// - **Review requests.** Publish on the one sink subject @@ -313,6 +313,7 @@ impl Policy { let consumer = forge_events::ROUTER_CONSUMER; let state = format!("KV_{}", notify_router::STATE_BUCKET); let flows = format!("KV_{}", notify_flows::BUCKET); + let archetypes = format!("KV_{}", swarm_queue_client::archetypes::BUCKET); let mut publish = Self::jetstream_minimum().to_vec(); publish.extend([ format!("$JS.API.STREAM.INFO.{events}"), @@ -333,6 +334,12 @@ impl Policy { format!("$KV.{}.>", notify_router::STATE_BUCKET), format!("$JS.API.STREAM.INFO.{flows}"), format!("$JS.API.DIRECT.GET.{flows}.>"), + format!("$JS.API.STREAM.INFO.{archetypes}"), + format!("$JS.API.DIRECT.GET.{archetypes}.>"), + format!( + "$JS.API.DIRECT.GET.{}.>", + swarm_queue_client::agent_archetype::STREAM + ), format!("$JS.API.STREAM.INFO.{}", Self::agent_status_stream()), swarm_wire::router_sink::REQUEST_REVIEWER_SUBJECT.to_owned(), format!("$JS.API.STREAM.MSG.GET.{events}"), @@ -1985,6 +1992,10 @@ mod tests { "$JS.API.DIRECT.GET.KV_{flows}.$KV.{flows}.{}", notify_flows::KEY ), + // The archetypes' records and each agent's archetype: read them. + "$JS.API.STREAM.INFO.KV_agent-archetypes".to_owned(), + "$JS.API.DIRECT.GET.KV_agent-archetypes.$KV.agent-archetypes.hyperhive".to_owned(), + swarm_queue_client::agent_archetype::direct_get_subject("atlas"), ] { assert!(may(&allowed), "the router may not reach {allowed}"); } @@ -2022,6 +2033,11 @@ mod tests { format!("$KV.{flows}.{}", notify_flows::KEY), format!("$JS.API.STREAM.CREATE.KV_{flows}"), format!("$JS.API.STREAM.DELETE.KV_{flows}"), + // The archetype buckets: not write them, nor make them. + "$KV.agent-archetypes.hyperhive".to_owned(), + "$JS.API.STREAM.CREATE.KV_agent-archetypes".to_owned(), + swarm_queue_client::agent_archetype::key_subject("atlas"), + "$JS.API.STREAM.CREATE.KV_agent-archetype".to_owned(), swarm_queue_client::agent_input_subject("atlas"), "$JS.API.STREAM.CREATE.KV_agent-status".to_owned(), ] { diff --git a/swarm-notify-router/Cargo.toml b/swarm-notify-router/Cargo.toml index daa2672bd..eb988ff94 100644 --- a/swarm-notify-router/Cargo.toml +++ b/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"] } diff --git a/swarm-notify-router/src/archetype_flows.rs b/swarm-notify-router/src/archetype_flows.rs new file mode 100644 index 000000000..3f1ebaf90 --- /dev/null +++ b/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, + /// The last problem logged, so a standing one is logged once. + problem: Option, + checked: Option, +} + +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) { + 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, Vec)) { + 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, +) -> Result< + ( + BTreeMap, + BTreeMap, + ), + 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::>() + { + 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, + records: &BTreeMap, +) -> (Vec, Vec) { + let mut agents: BTreeMap<(&str, &str), Vec> = 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)]) -> BTreeMap { + 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 { + agents + .iter() + .map(|(agent, archetype)| ((*agent).to_owned(), reference(archetype))) + .collect() + } + + fn members(built: &(Vec, Vec)) -> 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::>(), + ["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 = ["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()][..] + )] + ); + } +} diff --git a/swarm-notify-router/src/dry_run.rs b/swarm-notify-router/src/dry_run.rs index 752806b22..d0c7dd14d 100644 --- a/swarm-notify-router/src/dry_run.rs +++ b/swarm-notify-router/src/dry_run.rs @@ -38,7 +38,7 @@ pub fn run(flows: &[Flow], events: &[(String, Vec)]) -> 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 { diff --git a/swarm-notify-router/src/flow.rs b/swarm-notify-router/src/flow.rs index 4b7e91f08..44a930e9b 100644 --- a/swarm-notify-router/src/flow.rs +++ b/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, +} + /// 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 { + 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 { 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 { +fn emit( + flow: &Flow, + node: &Node, + sink: &TodoSink, + fields: &Fields, + members: &[String], +) -> Vec { 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()], diff --git a/swarm-notify-router/src/main.rs b/swarm-notify-router/src/main.rs index 61bcac658..87c36d984 100644 --- a/swarm-notify-router/src/main.rs +++ b/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, state: kv::Store, flows: stored::Flows, + archetype_flows: archetype_flows::ArchetypeFlows, roster: Option<(Instant, BTreeSet)>, 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() diff --git a/swarm-notify-router/src/route.rs b/swarm-notify-router/src/route.rs index 7be7a321d..fbd575668 100644 --- a/swarm-notify-router/src/route.rs +++ b/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) + ] + ); + } } diff --git a/swarm-queue-client/src/archetypes.rs b/swarm-queue-client/src/archetypes.rs index 6c55c57d2..03de4b7e9 100644 --- a/swarm-queue-client/src/archetypes.rs +++ b/swarm-queue-client/src/archetypes.rs @@ -13,6 +13,7 @@ use std::collections::BTreeMap; use swarm_wire::agent_inference::AgentInference; use swarm_wire::agent_limits::AgentCaps; +use swarm_wire::notify_flows::Flow; /// The KV bucket archetype sources are published into, keyed by source name. pub const BUCKET: &str = "agent-archetypes"; @@ -62,6 +63,11 @@ pub struct ArchetypeMeta { /// none. #[serde(default, skip_serializing_if = "Option::is_none")] pub icon: Option, + /// The notification-router flows every agent of this archetype gets, next + /// to the operator's document and the built-ins. Their todos go to + /// `Recipients::Members`; the router runs them, and absent means none. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub flows: Vec, } /// Open the archetype bucket for writing, creating it if nothing has yet. @@ -144,4 +150,60 @@ mod tests { assert!(wire.contains(r#""prompt":"be brief""#), "{wire}"); assert_eq!(serde_json::from_str::(&wire).unwrap(), record); } + + fn members_flow() -> Flow { + use swarm_wire::notify_flows::{Edge, Node, NodeKind, Recipients, TodoSink}; + use swarm_wire::router_todo::Op; + + Flow { + id: "new-issues".to_owned(), + name: "new issues".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: Recipients::Members, + op: Op::Upsert, + key: "forge:{{ repo }}#{{ number }}:new".to_owned(), + prefix: false, + text: "new issue {{ title }}".to_owned(), + }), + }, + ], + edges: vec![Edge { + from: "src".to_owned(), + to: "tell".to_owned(), + }], + } + } + + /// A record stored before `flows` existed reads as an archetype with none, + /// and an archetype with none is stored without the key. + #[test] + fn a_record_without_flows_parses_and_encodes_without_them() { + let old = r#"{"source":"s","archetypes":{"reviewer":{"description":"reads"}}}"#; + let record: SourceRecord = serde_json::from_str(old).unwrap(); + assert!(record.archetypes["reviewer"].flows.is_empty()); + assert_eq!(serde_json::to_string(&record).unwrap(), old); + } + + /// An archetype's flows survive the bucket, `members` recipient included. + #[test] + fn flows_round_trip_with_the_members_recipient() { + let meta = ArchetypeMeta { + description: "triage".to_owned(), + flows: vec![members_flow()], + ..ArchetypeMeta::default() + }; + let wire = serde_json::to_string(&meta).unwrap(); + assert!(wire.contains(r#""to":"members""#), "{wire}"); + assert_eq!(serde_json::from_str::(&wire).unwrap(), meta); + } } diff --git a/swarm-wire/src/notify_flows.rs b/swarm-wire/src/notify_flows.rs index de80c3929..0ea5d4a8f 100644 --- a/swarm-wire/src/notify_flows.rs +++ b/swarm-wire/src/notify_flows.rs @@ -271,6 +271,11 @@ pub enum Recipients { Field(String), /// This agent, whatever the event. Agent(String), + /// Every agent created from the archetype the flow came from. Only an + /// archetype's own flows take it: the operator's document has no + /// archetype, so [`Flow::validate`] refuses it there, and + /// [`Flow::validate_archetype`] refuses every other recipient. + Members, } /// Why [`Document::decode`] could not read a document. @@ -665,12 +670,29 @@ fn jinja_of_legacy(template: &str) -> String { impl Flow { /// Check the flow is a tree within the caps, and that every node reads - /// only what exists. + /// only what exists. This is the check for the operator's document, which + /// has no archetype, so a [`Recipients::Members`] fails it. /// /// # Errors /// /// Says why not. pub fn validate(&self) -> Result<(), String> { + self.validate_in(Scope::Document) + } + + /// The check for a flow an archetype carries: [`Flow::validate`], except + /// that every todo goes to [`Recipients::Members`] and no node asks for a + /// review. A flow that names an agent, or the agents an event field holds, + /// or asks for a review, belongs in the operator's document. + /// + /// # Errors + /// + /// Says why not. + pub fn validate_archetype(&self) -> Result<(), String> { + self.validate_in(Scope::Archetype) + } + + fn validate_in(&self, scope: Scope) -> Result<(), String> { if !is_id(&self.id) { return Err(format!( "id {:?} is not 1 to 64 letters, digits, - or _", @@ -695,7 +717,7 @@ impl Flow { return Err(format!("node id {} is used twice", node.id)); } node.kind - .validate() + .validate(scope) .map_err(|why| format!("node {}: {why}", node.id))?; } for edge in &self.edges { @@ -766,8 +788,17 @@ impl Flow { } } +/// Where a flow lives, which decides who its todos may go to. +#[derive(Clone, Copy)] +enum Scope { + /// The operator's document, or a built-in. + Document, + /// An archetype's `meta.flows`. + Archetype, +} + impl NodeKind { - fn validate(&self) -> Result<(), String> { + fn validate(&self, scope: Scope) -> Result<(), String> { match self { Self::Source { events } => { if events.is_empty() { @@ -792,7 +823,12 @@ impl NodeKind { )); } } - Self::Todo(sink) => sink.validate()?, + Self::Todo(sink) => sink.validate(scope)?, + Self::RequestReviewer { .. } if matches!(scope, Scope::Archetype) => { + return Err("an archetype's flow cannot request a review: that is the \ + operator's flows document's job" + .to_owned()); + } Self::RequestReviewer { reviewer } => { if !is_login(reviewer) { return Err(format!("reviewer {reviewer:?} is not a forge login")); @@ -804,12 +840,26 @@ impl NodeKind { } impl TodoSink { - fn validate(&self) -> Result<(), String> { - match &self.to { - Recipients::Field(f) if !is_field(f) => { + fn validate(&self, scope: Scope) -> Result<(), String> { + match (&self.to, scope) { + (Recipients::Members, Scope::Archetype) => {} + (Recipients::Members, Scope::Document) => { + return Err("recipients members is for an archetype's flows: the \ + document has no archetype" + .to_owned()); + } + (_, Scope::Archetype) => { + return Err( + "an archetype's flow sends only to its members: a flow to a \ + named agent or to an event field's agents is the operator's flows \ + document's job" + .to_owned(), + ); + } + (Recipients::Field(f), Scope::Document) if !is_field(f) => { return Err(format!("recipients from {f:?}, which is no event field")); } - Recipients::Agent(a) if !is_id(a) => { + (Recipients::Agent(a), Scope::Document) if !is_id(a) => { return Err(format!("recipient {a:?} is not an agent name")); } _ => {} @@ -1681,4 +1731,75 @@ mod tests { }) ); } + + /// A copy of the `custom` flow whose todo goes to `to`. + fn sending_to(id: &str, to: Recipients) -> Flow { + let mut f = custom(id); + let NodeKind::Todo(sink) = &mut f.nodes[2].kind else { + panic!("a todo") + }; + sink.to = to; + f + } + + /// `members` is an archetype flow's recipient: the operator's document and + /// the document check refuse it, an archetype flow takes nothing else. + #[test] + fn members_is_for_an_archetypes_flows_only() { + let members = sending_to("m", Recipients::Members); + // Presence: the same flow passes where `members` belongs. + assert_eq!(members.validate_archetype(), Ok(())); + let why = members.validate().expect_err("no archetype in a document"); + assert!(why.contains("no archetype"), "{why}"); + let doc = Document { + flows: vec![members], + ..Document::default() + }; + assert!(doc.check().is_err()); + assert!(doc.active().0.iter().all(|f| f.id != "m"), "left out"); + + for other in [ + Recipients::Agent("triage".to_owned()), + Recipients::Field("author".to_owned()), + ] { + let f = sending_to("o", other); + assert_eq!(f.validate(), Ok(()), "control: a document takes it"); + let why = f.validate_archetype().expect_err("not for an archetype"); + assert!(why.contains("only to its members"), "{why}"); + } + let mut review = custom("r"); + review.nodes[2].kind = NodeKind::RequestReviewer { + reviewer: "argus".to_owned(), + }; + assert_eq!(review.validate(), Ok(()), "control: a document takes it"); + assert!(review.validate_archetype().is_err()); + for built in builtins() { + assert!(built.validate_archetype().is_err(), "{}", built.id); + } + } + + /// An archetype's flow reads its key and text as the same Jinja templates a + /// document's flow does: an unknown variable or broken syntax is refused. + #[test] + fn an_archetypes_templates_are_checked_like_the_documents() { + let with = |key: &str, text: &str| { + let mut f = sending_to("t", Recipients::Members); + let NodeKind::Todo(sink) = &mut f.nodes[2].kind else { + panic!("a todo") + }; + sink.key = key.to_owned(); + sink.text = text.to_owned(); + f + }; + let ok = with("forge:{{ repo }}#{{ number }}:x", "{{ title }} {{ url }}"); + assert_eq!(ok.validate_archetype(), Ok(()), "control"); + for (key, text) in [ + ("forge:{{ repo }}#{{ numbr }}:x", "t"), + ("k", "{{ nope }}"), + ("k", "{{ title "), + ] { + let f = with(key, text); + assert!(f.validate_archetype().is_err(), "{key} {text}"); + } + } }