diff --git a/docs/swarm/notify-router.md b/docs/swarm/notify-router.md index ef9c9d16..74cbf779 100644 --- a/docs/swarm/notify-router.md +++ b/docs/swarm/notify-router.md @@ -64,21 +64,59 @@ The router holds no forge token and no webhook secret, and registers no hook. A flow is a graph of nodes joined by edges: one `source` node naming the forge event types it starts from, `match` nodes that pass when every condition -holds, and `sink` nodes that emit one todo op per recipient. A match node can +holds, and `todo` nodes that emit one todo op per recipient. A match node can have more than one child, and each one runs, so one event can reach several -sinks. A flow must be a tree: one source, one parent for every other node, -every node reachable, no children under a sink. +todos. A flow must be a tree: one source, one parent for every other node, +every node reachable, no children under a `todo`. 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. The router ships the three flows above, each a source → match on `action` → -todo sink. +todo. The code defines them (`swarm_wire::notify_flows::builtins`), so a +build that changes one reaches every swarm. A custom flow has the same type and +the same evaluator, so it can do anything a built-in does. + +**The stored document.** swarm-controller keeps the rest in one document, the +`all` key of the `notify-flows` bucket: the ids of the built-in flows you +switched off, and the custom flows. Only the controller writes it, through +`GET` and `PUT /api/notify/flows`. A `PUT` replaces the whole document in one +compare-and-set. The controller answers 400 unless every flow passes the tree +check above and reads only what exists: a source event from `issues`, +`issue_assign`, `pull_request` or `pull_request_assign`, and condition and +recipient fields from the event's field list. A `PUT` that carries the +`revision` it read gets 409 when another write landed since. Swarm UI's +[`/notifications`](ui.md#notification-flows) page shows the document and +switches built-ins. + +**What the router runs.** It reads the key at most every 10 seconds, when an +event arrives, and runs the built-ins you left on, then the custom flows that +are on and pass the tree check. It logs a custom flow that fails the check, +leaves it out, and runs the rest. + +| the key | the router runs | +| -------------------------------------- | ----------------------- | +| absent, deleted, or its bucket missing | the built-in flows | +| a document it reads | the above | +| a document it can't read | the flows it ran before | +| a read that fails | the flows it ran before | + +"Can't read" means not JSON, not a flows document, or a `version` other than 1. +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. + +**Deploying in any order.** The controller, the router and the UI can be on +different hosts. 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. ## 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), publish on `$SWARM.router.todo.*`, create `ROUTER_TODOS`, its own `notify-router-state` bucket, and `STREAM.INFO` on the agent-status bucket, which lists the agents | -| swarm-controller | publish on `$SWARM.events.forge.*` | -| 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), publish on `$SWARM.router.todo.*`, create `ROUTER_TODOS`, its own `notify-router-state` bucket, read on the `notify-flows` bucket, and `STREAM.INFO` on the agent-status bucket, which lists the agents | +| swarm-controller | publish on `$SWARM.events.forge.*`; 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.*` or `$SWARM.events.forge.*`, so each has one writer. The agent's create subject diff --git a/docs/swarm/ui.md b/docs/swarm/ui.md index d49e0045..106aa794 100644 --- a/docs/swarm/ui.md +++ b/docs/swarm/ui.md @@ -19,6 +19,7 @@ domain, covers host-level detail for one hive. | `/schedules` | the scheduled prompts: add, edit, pause, resume and remove | | `/broadcast` | send one message to several agents at once, picked from a grid of their icons | | `/messages` | the swarm-wide message feed: every message between agents, and to them from the operator, reminders and compaction, with a history window and a filter on sender or recipient | +| `/notifications` | the notification router's flows: a switch for each built-in, and every flow drawn as a tree | | `/issues` | a cross-repo issue report | | `/review` | the review classifier: its recent runs, a try-out on a real PR, and its settings | | `/builds` | the swarm builder's recent builds of agent config PRs, with what nix printed when one failed | @@ -216,6 +217,19 @@ Only an operator can change a schedule here or with `swarmctl schedule`; an agent has no way to add one. The page needs the controller's `/api/schedules` routes. +### Notification flows + +`/notifications` lists the [notification router's flows](notify-router.md#flows) +in two groups. Each **built-in** flow has an on/off switch; switching one off +stops the router running it from its next event, and switching it on again +restores it. Each **custom** flow shows its name and whether it's on. Every +flow draws as an indented tree: the forge events it starts from, the +conditions under it, and the todo each branch ends in. + +The page edits nothing else. Add custom flows with `PUT /api/notify/flows`. +A switch sends the whole flows document with the revision the page read, so if +another change landed since, the controller refuses it and the page shows why. + ### Sending a message to an agent Under an agent's terminal, in the preview and in its own tab, is a **message** diff --git a/swarm-controller/README.md b/swarm-controller/README.md index 40a56120..9daadf03 100644 --- a/swarm-controller/README.md +++ b/swarm-controller/README.md @@ -82,6 +82,11 @@ The swarm's control plane. The swarm UI and `swarmctl` are its clients. `review-classifier-settings` KV bucket, which this daemon alone reads and writes (`GET`/`PUT /api/review-classifier/settings`) → [`docs/swarm/review-classifier.md`](../docs/swarm/review-classifier.md#swarm-level-settings). +- **Notification flows** — which of the notification router's built-in flows are + switched off, and the custom flows, as one document in the `notify-flows` + KV bucket, which this daemon alone writes (`GET`/`PUT /api/notify/flows`); + the router reads it → + [`docs/swarm/notify-router.md`](../docs/swarm/notify-router.md#flows). - **Operator profiles** — each operator's name and the pronouns on their forge profile, for every member of authelia's operator group, as one list under the one key of the `operator-profiles` KV bucket. This daemon alone diff --git a/swarm-controller/src/notify_flows.rs b/swarm-controller/src/notify_flows.rs index 5108ac18..9f303ee3 100644 --- a/swarm-controller/src/notify_flows.rs +++ b/swarm-controller/src/notify_flows.rs @@ -201,8 +201,7 @@ pub(crate) async fn put_flows( let (current, revision) = stored(&store).await.map_err(|e| queue_problem(&e))?; // An unreadable document is replaced only when it is not a version // this build does not know: a newer build's document is kept for it. - if let Stored::Unreadable(why @ DecodeError::UnknownVersion(_)) = ¤t - { + if let Stored::Unreadable(why @ DecodeError::UnknownVersion(_)) = ¤t { return Err(unreadable(why)); } if put.revision.is_some() && put.revision != revision { @@ -269,16 +268,12 @@ mod tests { } } - fn body( - revision: Option, - disabled: &[&str], - flows: Vec, - ) -> Result, JsonRejection> { - Ok(Json(PutFlows { + fn body(revision: Option, disabled: &[&str], flows: Vec) -> Json { + Json(PutFlows { revision, disabled_builtins: disabled.iter().map(|s| (*s).to_owned()).collect(), flows, - })) + }) } #[tokio::test] @@ -295,7 +290,7 @@ mod tests { let Json(saved) = put_flows( State(state.clone()), - body(None, &["unassigned"], vec![custom("triage")]), + Ok(body(None, &["unassigned"], vec![custom("triage")])), ) .await .expect("put"); @@ -307,18 +302,22 @@ mod tests { // The stored bytes are the document the router decodes. let store = open(&client).await.expect("bucket"); - let bytes = store.get(notify_flows::KEY).await.expect("read").expect("stored"); + let bytes = store + .get(notify_flows::KEY) + .await + .expect("read") + .expect("stored"); let doc = Document::decode(&bytes).expect("the router reads it"); assert_eq!(doc.active().0.len(), 3, "2 built-ins and the custom flow"); - // A refused save changes nothing: a flow with a cycle-free orphan. + // A refused save changes nothing: an orphan node, and a built-in that does not exist. let mut orphan = custom("orphan"); orphan.edges.clear(); - let refused = put_flows(State(state.clone()), body(None, &[], vec![orphan])) + let refused = put_flows(State(state.clone()), Ok(body(None, &[], vec![orphan]))) .await .expect_err("an orphan node"); assert_eq!(refused.status, Some(StatusCode::BAD_REQUEST), "{refused:?}"); - let unknown = put_flows(State(state.clone()), body(None, &["nope"], vec![])) + let unknown = put_flows(State(state.clone()), Ok(body(None, &["nope"], vec![]))) .await .expect_err("no such built-in"); assert_eq!(unknown.status, Some(StatusCode::BAD_REQUEST), "{unknown:?}"); @@ -326,14 +325,14 @@ mod tests { assert_eq!(after, saved); // A stale revision is refused; the current one lands. - let stale = put_flows( + let outdated = put_flows( State(state.clone()), - body(Some(revision + 100), &[], vec![]), + Ok(body(Some(revision + 100), &[], vec![])), ) .await .expect_err("stale"); - assert_eq!(stale.status, Some(StatusCode::CONFLICT), "{stale:?}"); - let Json(next) = put_flows(State(state.clone()), body(Some(revision), &[], vec![])) + assert_eq!(outdated.status, Some(StatusCode::CONFLICT), "{outdated:?}"); + let Json(next) = put_flows(State(state.clone()), Ok(body(Some(revision), &[], vec![]))) .await .expect("current revision"); assert!(next.revision > Some(revision)); @@ -350,16 +349,25 @@ mod tests { let store = open(&client).await.expect("bucket"); store - .put(notify_flows::KEY, br#"{"version":2,"flows":[]}"#.to_vec().into()) + .put( + notify_flows::KEY, + br#"{"version":2,"flows":[]}"#.to_vec().into(), + ) .await .expect("seed"); - let got = get_flows(State(state.clone())).await.expect_err("version 2"); + let got = get_flows(State(state.clone())) + .await + .expect_err("version 2"); assert_eq!(got.status, Some(StatusCode::CONFLICT), "{got:?}"); - let refused = put_flows(State(state.clone()), body(None, &[], vec![])) + let refused = put_flows(State(state.clone()), Ok(body(None, &[], vec![]))) .await .expect_err("version 2 is kept"); assert_eq!(refused.status, Some(StatusCode::CONFLICT), "{refused:?}"); - let kept = store.get(notify_flows::KEY).await.expect("read").expect("stored"); + let kept = store + .get(notify_flows::KEY) + .await + .expect("read") + .expect("stored"); assert_eq!(&kept[..], br#"{"version":2,"flows":[]}"#); // Control: garbage that names no version is replaced by a save. @@ -367,7 +375,7 @@ mod tests { .put(notify_flows::KEY, b"garbage".to_vec().into()) .await .expect("seed"); - let _ = put_flows(State(state.clone()), body(None, &["closed"], vec![])) + let _ = put_flows(State(state.clone()), Ok(body(None, &["closed"], vec![]))) .await .expect("replaces garbage"); let Json(view) = get_flows(State(state)).await.expect("readable again"); diff --git a/swarm-controller/src/startup.rs b/swarm-controller/src/startup.rs index 132a688f..f2529e1e 100644 --- a/swarm-controller/src/startup.rs +++ b/swarm-controller/src/startup.rs @@ -31,8 +31,8 @@ use crate::{ agent_control, agent_icon, agent_input, agent_limits, agent_login, agent_permissions, agent_renewal, agent_state_stream, agent_status, agents_md, alerts, archetypes, auth, builder, claude_login, config_pr, forge, forge_account, github_account, inference, issue_report, - linked_accounts, matrix_account, message_stream, notify_flows, pulls, queue_identity, review_classifier, - schedules, status, subagent_term, term_stream, wanted, webhook, + linked_accounts, matrix_account, message_stream, notify_flows, pulls, queue_identity, + review_classifier, schedules, status, subagent_term, term_stream, wanted, webhook, }; /// Where the daemon binds, overridable via `SWARM_CONTROLLER_SOCKET`. diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index 000dbf97..05bd88ee 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -1829,7 +1829,10 @@ mod tests { "$JS.API.STREAM.INFO.KV_agent-status".to_owned(), // The flows document: read it. format!("$JS.API.STREAM.INFO.KV_{flows}"), - format!("$JS.API.DIRECT.GET.KV_{flows}.$KV.{flows}.{}", notify_flows::KEY), + format!( + "$JS.API.DIRECT.GET.KV_{flows}.$KV.{flows}.{}", + notify_flows::KEY + ), ] { assert!(may(&allowed), "the router may not reach {allowed}"); } diff --git a/swarm-notify-router/src/flow.rs b/swarm-notify-router/src/flow.rs index 3eca668d..f0a262d8 100644 --- a/swarm-notify-router/src/flow.rs +++ b/swarm-notify-router/src/flow.rs @@ -124,7 +124,6 @@ pub fn expand(template: &str, fields: &Fields) -> String { out } - #[cfg(test)] mod tests { use swarm_wire::notify_flows::{Edge, builtins}; @@ -230,14 +229,7 @@ mod tests { assert_eq!(a.op.key, "forge:hyperhive/hyperhive#7:"); assert_eq!(a.op.summary, ""); } - assert!( - evaluate( - &closed, - "issues", - &event("reopened", &[], &[], &["atlas"]) - ) - .is_empty() - ); + assert!(evaluate(&closed, "issues", &event("reopened", &[], &[], &["atlas"])).is_empty()); } /// A match node with two children runs both: the branching the tree diff --git a/swarm-notify-router/src/stored.rs b/swarm-notify-router/src/stored.rs index e34f834a..76512783 100644 --- a/swarm-notify-router/src/stored.rs +++ b/swarm-notify-router/src/stored.rs @@ -18,9 +18,9 @@ use std::time::{Duration, Instant}; +use async_nats::jetstream::ErrorCode; use async_nats::jetstream::context::GetStreamErrorKind; use async_nats::jetstream::{self, kv}; -use async_nats::jetstream::ErrorCode; use swarm_wire::notify_flows::{self, Document, Flow}; /// How long a read of the key is reused. @@ -31,7 +31,10 @@ const REFRESH: Duration = Duration::from_secs(10); pub enum Read { /// No key, a deleted key, or no bucket. Absent, - Document { revision: u64, bytes: Vec }, + Document { + revision: u64, + bytes: Vec, + }, Failed(String), } @@ -62,7 +65,9 @@ impl Flows { Read::Failed(why) => self.note(format!("reading the key failed: {why}")), Read::Absent => { if self.revision.take().is_some() { - tracing::info!("notify router: the flows document is gone; running the built-in flows"); + tracing::info!( + "notify router: the flows document is gone; running the built-in flows" + ); } self.active = Document::default().active().0; self.problem = None; diff --git a/swarm-wire/src/notify_flows.rs b/swarm-wire/src/notify_flows.rs index 60505666..df019281 100644 --- a/swarm-wire/src/notify_flows.rs +++ b/swarm-wire/src/notify_flows.rs @@ -42,7 +42,12 @@ pub const MAX_TEXT: usize = 500; /// The forge event types a source node can start from: the ones /// swarm-controller's hooks deliver to the events stream. -pub const SOURCE_EVENTS: &[&str] = &["issues", "issue_assign", "pull_request", "pull_request_assign"]; +pub const SOURCE_EVENTS: &[&str] = &[ + "issues", + "issue_assign", + "pull_request", + "pull_request_assign", +]; /// The fields of an event a condition or a recipient can name. pub const FIELDS: &[&str] = &[ @@ -204,8 +209,8 @@ impl Document { /// Not the expected shape, or a [`Document::version`] this build does not /// know. The reader keeps what it had. pub fn decode(bytes: &[u8]) -> Result { - let value: serde_json::Value = - serde_json::from_slice(bytes).map_err(|e| DecodeError::Malformed(format!("not JSON: {e}")))?; + let value: serde_json::Value = serde_json::from_slice(bytes) + .map_err(|e| DecodeError::Malformed(format!("not JSON: {e}")))?; if let Some(found) = value.get("version") && found != VERSION { @@ -225,15 +230,14 @@ impl Document { return Err(format!("version {} is not {VERSION}", self.version)); } if self.flows.len() > MAX_FLOWS { - return Err(format!( - "{} flows, more than {MAX_FLOWS}", - self.flows.len() - )); + return Err(format!("{} flows, more than {MAX_FLOWS}", self.flows.len())); } let builtin = builtins(); for id in &self.disabled_builtins { if !builtin.iter().any(|b| b.id == *id) { - return Err(format!("disabled_builtins names {id}, which is no built-in flow")); + return Err(format!( + "disabled_builtins names {id}, which is no built-in flow" + )); } } let mut ids: BTreeSet<&str> = builtin.iter().map(|b| b.id.as_str()).collect(); @@ -244,7 +248,8 @@ impl Document { flow.id )); } - flow.validate().map_err(|why| format!("flow {}: {why}", flow.id))?; + flow.validate() + .map_err(|why| format!("flow {}: {why}", flow.id))?; } Ok(()) } @@ -401,7 +406,10 @@ impl NodeKind { } Self::Match { when } => { if let Some(bad) = when.iter().find(|c| !is_field(&c.field)) { - return Err(format!("condition on {:?}, which is no event field", bad.field)); + return Err(format!( + "condition on {:?}, which is no event field", + bad.field + )); } } Self::Todo(sink) => sink.validate()?, @@ -534,7 +542,10 @@ mod tests { } fn builtin(id: &str) -> Flow { - builtins().into_iter().find(|f| f.id == id).expect("a built-in") + builtins() + .into_iter() + .find(|f| f.id == id) + .expect("a built-in") } /// A custom flow: opened issue or PR → a todo for `triage`. @@ -704,7 +715,12 @@ mod tests { ); let mut bad = custom("bad"); bad.edges.pop(); - assert!(doc(&[], vec![bad]).check().unwrap_err().starts_with("flow bad:")); + assert!( + doc(&[], vec![bad]) + .check() + .unwrap_err() + .starts_with("flow bad:") + ); let many = (0..=MAX_FLOWS).map(|i| custom(&format!("f{i}"))).collect(); assert!(doc(&[], many).check().is_err(), "flow count"); let mut old = doc(&[], vec![]);