docs: notification flows, their API and the notifications page

docs/swarm/notify-router.md

@@ -64,20 +64,58 @@ 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,
+todos. A flow must be a tree: one source, one parent for every other node,
-every node reachable, no children under a sink.
+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/<router client id>/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-notify-router`, in its container | store: `read` on `swarm/services/<router client id>/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.*`                                                                                                                                                                                                                                                                    |
+| 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-<agent>` in the form whose filter is `$SWARM.router.todo.<agent>`, pull from it, and ack what it delivered                                                                                                                                                                                                 |
 
 No other client may publish on `$SWARM.router.todo.*` or

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**

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

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(_)) = &current
+        if let Stored::Unreadable(why @ DecodeError::UnknownVersion(_)) = &current {
-        {
             return Err(unreadable(why));
         }
         if put.revision.is_some() && put.revision != revision {
@@ -269,16 +268,12 @@ mod tests {
         }
     }
 
-    fn body(
+    fn body(revision: Option<u64>, disabled: &[&str], flows: Vec<Flow>) -> Json<PutFlows> {
-        revision: Option<u64>,
+        Json(PutFlows {
-        disabled: &[&str],
-        flows: Vec<Flow>,
-    ) -> Result<Json<PutFlows>, JsonRejection> {
-        Ok(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:?}");
+        assert_eq!(outdated.status, Some(StatusCode::CONFLICT), "{outdated:?}");
-        let Json(next) = put_flows(State(state.clone()), body(Some(revision), &[], vec![]))
+        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");

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,
+    linked_accounts, matrix_account, message_stream, notify_flows, pulls, queue_identity,
-    schedules, status, subagent_term, term_stream, wanted, webhook,
+    review_classifier, schedules, status, subagent_term, term_stream, wanted, webhook,
 };
 
 /// Where the daemon binds, overridable via `SWARM_CONTROLLER_SOCKET`.

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}");
         }

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!(
+        assert!(evaluate(&closed, "issues", &event("reopened", &[], &[], &["atlas"])).is_empty());
-            evaluate(
-                &closed,
-                "issues",
-                &event("reopened", &[], &[], &["atlas"])
-            )
-            .is_empty()
-        );
     }
 
     /// A match node with two children runs both: the branching the tree

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<u8> },
+    Document {
+        revision: u64,
+        bytes: Vec<u8>,
+    },
     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;

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<Self, DecodeError> {
-        let value: serde_json::Value =
+        let value: serde_json::Value = serde_json::from_slice(bytes)
-            serde_json::from_slice(bytes).map_err(|e| DecodeError::Malformed(format!("not JSON: {e}")))?;
+            .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!(
+            return Err(format!("{} flows, more than {MAX_FLOWS}", self.flows.len()));
-                "{} 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![]);