diff --git a/swarm-controller/src/forge.rs b/swarm-controller/src/forge.rs index e93ecd8f..6b2854f9 100644 --- a/swarm-controller/src/forge.rs +++ b/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] diff --git a/swarm-controller/src/notify_flows.rs b/swarm-controller/src/notify_flows.rs index ad03c915..5108ac18 100644 --- a/swarm-controller/src/notify_flows.rs +++ b/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, + disabled: &[&str], + flows: Vec, + ) -> Result, 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"]); + } +} diff --git a/swarm-controller/src/startup.rs b/swarm-controller/src/startup.rs index b6eca2a1..132a688f 100644 --- a/swarm-controller/src/startup.rs +++ b/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)"),