swarm-controller: serve the notification router's flows at /api/notify/flows

swarm-controller/src/forge.rs

@@ -2113,6 +2113,14 @@ mod tests {
             events(DeliveryKind::PullRequest),
             ["pull_request", "pull_request_assign"]
         );
+        // The events a stored flow's source may name are the ones the hooks
+        // deliver to the router.
+        let mut routed = events(DeliveryKind::Issues);
+        routed.extend(events(DeliveryKind::PullRequest));
+        routed.sort_unstable();
+        let mut sources = swarm_wire::notify_flows::SOURCE_EVENTS.to_vec();
+        sources.sort_unstable();
+        assert_eq!(routed, sources);
     }
 
     #[test]

swarm-controller/src/notify_flows.rs

@@ -230,3 +230,147 @@ pub(crate) async fn put_flows(
         "gave up writing the notification flows after {MAX_ATTEMPTS} conflicting writes"
     )))
 }
+
+#[cfg(test)]
+mod tests {
+    use swarm_wire::notify_flows::{Edge, Node, NodeKind, Recipients, TodoSink};
+    use swarm_wire::router_todo::Op;
+
+    use super::*;
+    use crate::test_nats::TestNats;
+
+    fn custom(id: &str) -> Flow {
+        Flow {
+            id: id.to_owned(),
+            name: "new to triage".to_owned(),
+            enabled: true,
+            nodes: vec![
+                Node {
+                    id: "source".to_owned(),
+                    kind: NodeKind::Source {
+                        events: vec!["issues".to_owned()],
+                    },
+                },
+                Node {
+                    id: "sink".to_owned(),
+                    kind: NodeKind::Todo(TodoSink {
+                        to: Recipients::Agent("triage".to_owned()),
+                        op: Op::Upsert,
+                        key: "forge:{repo}#{number}:triage".to_owned(),
+                        prefix: false,
+                        text: "triage {repo}#{number}".to_owned(),
+                    }),
+                },
+            ],
+            edges: vec![Edge {
+                from: "source".to_owned(),
+                to: "sink".to_owned(),
+            }],
+        }
+    }
+
+    fn body(
+        revision: Option<u64>,
+        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]
+    async fn the_flows_are_read_replaced_and_checked_on_save() {
+        let nats = TestNats::start();
+        let client = nats.client().await;
+        let state = crate::agent_limits::tests::state(&client);
+
+        // Nothing stored: the built-ins alone, none off, no revision.
+        let Json(view) = get_flows(State(state.clone())).await.expect("get");
+        assert_eq!(view.revision, None);
+        assert!(view.disabled_builtins.is_empty() && view.flows.is_empty());
+        assert_eq!(view.builtins.len(), 3);
+
+        let Json(saved) = put_flows(
+            State(state.clone()),
+            body(None, &["unassigned"], vec![custom("triage")]),
+        )
+        .await
+        .expect("put");
+        let revision = saved.revision.expect("a revision");
+        let Json(read) = get_flows(State(state.clone())).await.expect("get");
+        assert_eq!(read, saved);
+        assert_eq!(read.disabled_builtins, ["unassigned"]);
+        assert_eq!(read.flows[0].id, "triage");
+
+        // 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 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.
+        let mut orphan = custom("orphan");
+        orphan.edges.clear();
+        let refused = put_flows(State(state.clone()), 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![]))
+            .await
+            .expect_err("no such built-in");
+        assert_eq!(unknown.status, Some(StatusCode::BAD_REQUEST), "{unknown:?}");
+        let Json(after) = get_flows(State(state.clone())).await.expect("get");
+        assert_eq!(after, saved);
+
+        // A stale revision is refused; the current one lands.
+        let stale = put_flows(
+            State(state.clone()),
+            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![]))
+            .await
+            .expect("current revision");
+        assert!(next.revision > Some(revision));
+        assert!(next.flows.is_empty());
+    }
+
+    /// A document of a version this build does not know is neither served
+    /// nor replaced, and a garbled one is replaced.
+    #[tokio::test]
+    async fn a_document_of_another_version_is_kept_and_a_garbled_one_is_replaced() {
+        let nats = TestNats::start();
+        let client = nats.client().await;
+        let state = crate::agent_limits::tests::state(&client);
+        let store = open(&client).await.expect("bucket");
+
+        store
+            .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");
+        assert_eq!(got.status, Some(StatusCode::CONFLICT), "{got:?}");
+        let refused = put_flows(State(state.clone()), 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");
+        assert_eq!(&kept[..], br#"{"version":2,"flows":[]}"#);
+
+        // Control: garbage that names no version is replaced by a save.
+        store
+            .put(notify_flows::KEY, b"garbage".to_vec().into())
+            .await
+            .expect("seed");
+        let _ = put_flows(State(state.clone()), body(None, &["closed"], vec![]))
+            .await
+            .expect("replaces garbage");
+        let Json(view) = get_flows(State(state)).await.expect("readable again");
+        assert_eq!(view.disabled_builtins, ["closed"]);
+    }
+}

swarm-controller/src/startup.rs

@@ -31,7 +31,7 @@ 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, pulls, queue_identity, review_classifier,
+    linked_accounts, matrix_account, message_stream, notify_flows, pulls, queue_identity, review_classifier,
     schedules, status, subagent_term, term_stream, wanted, webhook,
 };
 
@@ -76,6 +76,7 @@ pub(crate) fn socket_path() -> PathBuf {
         (name = "repos", description = "forge repo browsing (swarm-ui's issue-report page)"),
         (name = "forge", description = "forge site admins (`swarmctl forge make-admin`)"),
         (name = "schedules", description = "scheduled prompts (`swarmctl schedule`)"),
+        (name = "notify", description = "the notification router's flows: built-ins switched off, and custom flows"),
         (name = "review-classifier", description = "trying review classifications against a PR (swarm-ui's try-out page), and the classifier's swarm-wide settings"),
         (name = "pulls", description = "open pull requests, for the try-out page's dropdowns"),
         (name = "alerts", description = "the swarm Grafana's current alerts (swarm-ui's alerts page)"),