swarm: the notification router tells a PR's author when CI on it fails or is cancelled

swarm-notify-router/src/dry_run.rs

@@ -24,7 +24,7 @@ pub fn run(flows: &[Flow], events: &[(String, Vec<u8>)]) -> DryRun {
     let mut snapshots: BTreeMap<String, Snapshot> = BTreeMap::new();
     let mut run = 0;
     for (event_type, body) in events {
-        let Some(delivery) = Delivery::parse(body) else {
+        let Some(delivery) = Delivery::from_event(event_type, body) else {
             continue;
         };
         run += 1;

swarm-notify-router/src/event.rs

@@ -26,6 +26,11 @@ pub struct Delivery {
     pub title: String,
     pub url: String,
     pub sender: String,
+    /// The account that opened the issue or PR.
+    pub author: String,
+    /// The commit at the PR's head when the delivery was made; empty for an
+    /// issue. A CI run's delivery carries the PR as it was when the run began.
+    pub head: String,
     pub is_pull: bool,
     pub assignees: BTreeSet<String>,
     /// The PR is a draft: the forge's `pull_request.draft`, true for a title
@@ -82,6 +87,15 @@ impl Delivery {
                 .get("sender")
                 .map(|s| text(s, "login"))
                 .unwrap_or_default(),
+            author: item
+                .get("user")
+                .map(|u| text(u, "login"))
+                .unwrap_or_default(),
+            head: item
+                .pointer("/head/sha")
+                .and_then(Json::as_str)
+                .unwrap_or_default()
+                .to_owned(),
             is_pull,
             assignees,
             draft,
@@ -89,6 +103,30 @@ impl Delivery {
         })
     }
 
+    /// Read the delivery of Forgejo event type `event_type`. A `workflow_run_*`
+    /// delivery carries the webhook that started the run as a JSON string in
+    /// `run.event_payload`; for a run on a PR that is the PR's own delivery, so
+    /// the PR is read from it, `action` is the run's outcome (`failure`,
+    /// `cancelled`) and `sender` is the account that triggered the run. A run
+    /// on anything else (a push to a branch with no PR) names no PR and gives
+    /// `None`.
+    pub fn from_event(event_type: &str, body: &[u8]) -> Option<Self> {
+        let Some(outcome) = event_type.strip_prefix("workflow_run_") else {
+            return Self::parse(body);
+        };
+        let json: Json = serde_json::from_slice(body).ok()?;
+        let run = json.get("run")?;
+        let mut delivery = Self::parse(run.get("event_payload")?.as_str()?.as_bytes())?;
+        if !delivery.is_pull {
+            return None;
+        }
+        outcome.clone_into(&mut delivery.action);
+        if let Some(trigger) = run.pointer("/trigger_user/login").and_then(Json::as_str) {
+            trigger.clone_into(&mut delivery.sender);
+        }
+        Some(delivery)
+    }
+
     /// The state bucket key for this delivery's issue: `<repo id>.<number>`,
     /// numeric so it is always a valid key whatever the repo is named.
     pub fn state_key(&self) -> String {
@@ -122,6 +160,11 @@ pub struct Snapshot {
     /// The agents holding a router todo for this issue.
     #[serde(default)]
     pub holders: BTreeSet<String>,
+    /// The commit at the head of the PR as of the last delivery about the PR
+    /// that was no CI run. A run that began at another commit is superseded
+    /// and tells nobody anything.
+    #[serde(default)]
+    pub head: Option<String>,
 }
 
 /// Whether `event_type` is an assign delivery, whose assignee list is the
@@ -153,6 +196,7 @@ pub fn fields(event_type: &str, delivery: &Delivery, snapshot: &Snapshot) -> Fie
         ("title", text(&delivery.title)),
         ("url", text(&delivery.url)),
         ("sender", text(&delivery.sender)),
+        ("author", text(&delivery.author)),
         ("draft", text(bool_text(delivery.draft))),
         ("undrafted", text(bool_text(delivery.undrafted))),
         (
@@ -266,6 +310,89 @@ mod tests {
         serde_json::to_vec(&body).expect("encodes")
     }
 
+    /// A `workflow_run_*` delivery: the run, holding the webhook that started
+    /// it as a JSON string.
+    fn run_body(started_by: &Json) -> Vec<u8> {
+        serde_json::to_vec(&serde_json::json!({
+            "action": "failure",
+            "run": {
+                "id": 5,
+                "trigger_user": { "login": "atlas" },
+                "event_payload": serde_json::to_string(started_by).expect("encodes"),
+            },
+        }))
+        .expect("encodes")
+    }
+
+    fn pull_payload(author: &str) -> Json {
+        serde_json::json!({
+            "action": "synchronized",
+            "number": 7,
+            "pull_request": {
+                "number": 7, "title": "t", "html_url": "u", "user": { "login": author },
+                "head": { "sha": "abc123" },
+            },
+            "repository": { "id": 1, "full_name": "o/r" },
+            "sender": { "login": "pusher" },
+        })
+    }
+
+    #[test]
+    fn a_run_on_a_pull_request_reads_as_that_pull_request_with_the_runs_outcome() {
+        let body = run_body(&pull_payload("iris"));
+        for (event, outcome) in [
+            ("workflow_run_failure", "failure"),
+            ("workflow_run_cancelled", "cancelled"),
+        ] {
+            let d = Delivery::from_event(event, &body).expect("a PR run");
+            assert!(d.is_pull);
+            assert_eq!((d.repo.as_str(), d.number), ("o/r", 7));
+            assert_eq!(d.action, outcome);
+            assert_eq!(d.author, "iris");
+            assert_eq!(d.head, "abc123", "the commit the run began at");
+            assert_eq!(d.sender, "atlas", "the account that triggered the run");
+            assert_eq!(d.state_key(), "1.7");
+            let f = fields(event, &d, &Snapshot::default());
+            assert_eq!(f.get("author"), Some(&Value::Text("iris".to_owned())));
+            assert_eq!(f.get("kind"), Some(&Value::Text("PR".to_owned())));
+        }
+    }
+
+    #[test]
+    fn a_run_that_names_no_pull_request_is_ignored() {
+        let push = serde_json::json!({
+            "ref": "refs/heads/main",
+            "repository": { "id": 1, "full_name": "o/r" },
+        });
+        assert_eq!(
+            Delivery::from_event("workflow_run_failure", &run_body(&push)),
+            None
+        );
+        let issue: Json =
+            serde_json::from_slice(&issue_assign("assigned", &["atlas"])).expect("json");
+        assert_eq!(
+            Delivery::from_event("workflow_run_failure", &run_body(&issue)),
+            None
+        );
+        assert_eq!(
+            Delivery::from_event("workflow_run_failure", b"not json"),
+            None
+        );
+        assert_eq!(
+            Delivery::from_event("workflow_run_failure", b"{\"run\":{}}"),
+            None
+        );
+    }
+
+    #[test]
+    fn other_events_read_the_body_directly() {
+        let body = pull_body("opened", "t", false, None);
+        assert_eq!(
+            Delivery::from_event("pull_request", &body),
+            Delivery::parse(&body)
+        );
+    }
+
     #[test]
     fn the_forges_draft_flag_marks_a_draft() {
         let d = Delivery::parse(&pull_body("synchronized", "WIP: t", true, None)).expect("parses");
@@ -307,6 +434,7 @@ mod tests {
         let before = Snapshot {
             assignees: Some(BTreeSet::from(["atlas".to_owned(), "iris".to_owned()])),
             holders: BTreeSet::new(),
+            ..Snapshot::default()
         };
         let f = fields("issue_assign", &d, &before);
         assert_eq!(list(&f, "assignees_removed"), ["atlas"]);
@@ -316,6 +444,7 @@ mod tests {
         let before = Snapshot {
             assignees: Some(BTreeSet::from(["iris".to_owned()])),
             holders: BTreeSet::new(),
+            ..Snapshot::default()
         };
         assert_eq!(
             list(&fields("issue_assign", &d, &before), "assignees_added"),

swarm-notify-router/src/main.rs

@@ -216,7 +216,7 @@ impl Router {
         let event_type = forge_events::event_type(&message.subject)
             .ok_or_else(|| Skip::Unusable("not a forge event subject".to_owned()))?
             .to_owned();
-        let Some(delivery) = event::Delivery::parse(&message.payload) else {
+        let Some(delivery) = event::Delivery::from_event(&event_type, &message.payload) else {
             // A delivery that names no issue or PR has nothing to route.
             return Ok(Handled::Ignored);
         };

swarm-notify-router/src/route.rs

@@ -32,6 +32,21 @@ pub fn route(
     flows: &[Flow],
     is_agent: impl Fn(&str) -> bool,
 ) -> Routed {
+    let is_run = event_type.starts_with("workflow_run_");
+    // Forgejo cancels the runs of a PR when it is pushed to, and an old run can
+    // finish after the push cleared the PR's CI todo. Such a run began at a
+    // commit the PR has since moved past, so it tells nobody anything.
+    if is_run
+        && !delivery.head.is_empty()
+        && snapshot.head.as_deref().is_some_and(|h| h != delivery.head)
+    {
+        return Routed {
+            actions: Vec::new(),
+            requests: Vec::new(),
+            skipped: Vec::new(),
+            next: snapshot.clone(),
+        };
+    }
     let fields = fields(event_type, delivery, snapshot);
     let mut actions = Vec::new();
     let mut skipped = Vec::new();
@@ -47,6 +62,11 @@ pub fn route(
         }
     }
     let mut next = snapshot.clone();
+    // A run's delivery carries the PR as it was when the run began, so only
+    // the other deliveries say where the PR's head is now.
+    if !is_run && delivery.is_pull && !delivery.head.is_empty() {
+        next.head = Some(delivery.head.clone());
+    }
     if is_assign(event_type) {
         // A skipped assignee is left out of the snapshot, so the next assign
         // event counts it as added again: an agent assigned before it is in
@@ -61,9 +81,13 @@ pub fn route(
             Op::Upsert => {
                 next.holders.insert(action.agent.clone());
             }
-            Op::Clear => {
+            // Only a prefix clear can take every todo the agent held for the
+            // issue. A clear of one key leaves it a holder: it may still hold
+            // the assignment, which the `closed` flow must find.
+            Op::Clear if action.op.prefix => {
                 cleared.insert(action.agent.clone());
             }
+            Op::Clear => {}
         }
     }
     next.holders.retain(|h| !cleared.contains(h));
@@ -89,6 +113,8 @@ mod tests {
             title: "t".to_owned(),
             url: "u".to_owned(),
             sender: "mara".to_owned(),
+            author: "mara".to_owned(),
+            head: String::new(),
             is_pull: false,
             assignees: assignees.iter().map(|a| (*a).to_owned()).collect(),
             draft: false,
@@ -103,6 +129,135 @@ mod tests {
             .collect()
     }
 
+    fn pull_by(author: &str, action: &str) -> Delivery {
+        Delivery {
+            is_pull: true,
+            author: author.to_owned(),
+            ..delivery(action, &[])
+        }
+    }
+
+    /// A failed run tells the PR's author, a cancelled one rewrites that todo,
+    /// and a push clears it. The author keeps holding the PR, so a later close
+    /// still reaches them.
+    #[test]
+    fn a_failed_run_tells_the_author_and_a_push_clears_it() {
+        let flows = builtins();
+        let is_agent = |a: &str| a == "atlas";
+
+        let failed = route(
+            "workflow_run_failure",
+            &pull_by("atlas", "failure"),
+            &Snapshot::default(),
+            &flows,
+            is_agent,
+        );
+        assert_eq!(agents(&failed), [("atlas", Op::Upsert)]);
+        assert_eq!(failed.actions[0].op.key, "forge:o/r#9:ci");
+        assert!(
+            failed.actions[0]
+                .op
+                .summary
+                .starts_with("CI failed on PR o/r#9")
+        );
+        assert_eq!(failed.next.holders, BTreeSet::from(["atlas".to_owned()]));
+
+        let cancelled = route(
+            "workflow_run_cancelled",
+            &pull_by("atlas", "cancelled"),
+            &failed.next,
+            &flows,
+            is_agent,
+        );
+        assert_eq!(agents(&cancelled), [("atlas", Op::Upsert)]);
+        assert_eq!(cancelled.actions[0].op.key, "forge:o/r#9:ci");
+        assert!(
+            cancelled.actions[0]
+                .op
+                .summary
+                .starts_with("CI was cancelled on PR")
+        );
+
+        let pushed = route(
+            "pull_request_sync",
+            &pull_by("atlas", "synchronized"),
+            &cancelled.next,
+            &flows,
+            is_agent,
+        );
+        let ci_clear: Vec<_> = pushed
+            .actions
+            .iter()
+            .filter(|a| a.op.op == Op::Clear)
+            .collect();
+        assert_eq!(ci_clear.len(), 1);
+        assert_eq!(ci_clear[0].op.key, "forge:o/r#9:ci");
+        assert!(
+            !ci_clear[0].op.prefix,
+            "only the CI todo goes, not the assignment"
+        );
+        assert_eq!(pushed.next.holders, BTreeSet::from(["atlas".to_owned()]));
+    }
+
+    fn at_head(mut d: Delivery, head: &str) -> Delivery {
+        head.clone_into(&mut d.head);
+        d
+    }
+
+    /// The push that clears the CI todo also cancels the runs before it, and
+    /// one of them can report in after the clear. It began at the old head, so
+    /// it must not bring the todo back.
+    #[test]
+    fn a_run_from_before_the_latest_push_tells_nobody() {
+        let flows = builtins();
+        let is_agent = |a: &str| a == "atlas";
+        let run = |event: &str, head: &str, before: &Snapshot| {
+            // The router names a run's action after its outcome.
+            let outcome = event.trim_start_matches("workflow_run_");
+            route(
+                event,
+                &at_head(pull_by("atlas", outcome), head),
+                before,
+                &flows,
+                is_agent,
+            )
+        };
+
+        // Before the router has seen the PR's head it can't tell, so it tells.
+        let unknown = run("workflow_run_cancelled", "old", &Snapshot::default());
+        assert_eq!(agents(&unknown), [("atlas", Op::Upsert)]);
+        assert_eq!(unknown.next.head, None, "a run says nothing about the head");
+
+        let pushed = route(
+            "pull_request_sync",
+            &at_head(pull_by("atlas", "synchronized"), "new"),
+            &unknown.next,
+            &flows,
+            is_agent,
+        );
+        assert_eq!(pushed.next.head.as_deref(), Some("new"));
+
+        let late = run("workflow_run_cancelled", "old", &pushed.next);
+        assert!(late.actions.is_empty() && late.skipped.is_empty());
+        assert_eq!(late.next, pushed.next);
+
+        let current = run("workflow_run_failure", "new", &pushed.next);
+        assert_eq!(agents(&current), [("atlas", Op::Upsert)]);
+    }
+
+    #[test]
+    fn a_failed_run_on_a_pull_request_by_a_non_agent_queues_nothing() {
+        let got = route(
+            "workflow_run_failure",
+            &pull_by("mara", "failure"),
+            &Snapshot::default(),
+            &builtins(),
+            |a| a == "atlas",
+        );
+        assert!(got.actions.is_empty());
+        assert_eq!(got.skipped, ["mara"]);
+    }
+
     /// Assign, reassign, close: the three built-ins in the order an issue
     /// meets them, carrying the snapshot from one event to the next.
     #[test]
@@ -208,6 +363,7 @@ mod tests {
         let before = Snapshot {
             assignees: Some(BTreeSet::from(["atlas".to_owned()])),
             holders: BTreeSet::from(["atlas".to_owned()]),
+            ..Snapshot::default()
         };
         let r = route("pull_request_sync", &pull, &before, &flows, |_| false);
         assert_eq!(
@@ -220,7 +376,10 @@ mod tests {
                 reviewer: "review-classifier".to_owned(),
             }]
         );
-        assert!(r.actions.is_empty() && r.skipped.is_empty());
+        // The push also clears the author's CI todo, but the author is no
+        // agent here, so that goes nowhere.
+        assert!(r.actions.is_empty());
+        assert_eq!(r.skipped, ["mara"]);
         assert_eq!(r.next, before);
 
         // The source gates: another PR event requests nothing.
@@ -271,6 +430,7 @@ mod tests {
         let before = Snapshot {
             assignees: Some(BTreeSet::from(["atlas".to_owned()])),
             holders: BTreeSet::from(["atlas".to_owned()]),
+            ..Snapshot::default()
         };
         let r = route(
             "issue_comment",

swarm-notify-router/src/stored.rs

@@ -149,6 +149,17 @@ mod tests {
 
     use super::*;
 
+    /// The built-in flows, in the order the router runs them.
+    const BUILTIN_IDS: [&str; 7] = [
+        "assigned",
+        "unassigned",
+        "closed",
+        "head-classifier",
+        "ci-failed",
+        "ci-cancelled",
+        "ci-cleared",
+    ];
+
     fn ids(f: &Flows) -> Vec<&str> {
         f.active.iter().map(|f| f.id.as_str()).collect()
     }
@@ -188,28 +199,30 @@ mod tests {
     #[test]
     fn with_no_document_the_router_runs_the_builtins() {
         let mut f = Flows::builtins();
-        assert_eq!(
-            ids(&f),
-            ["assigned", "unassigned", "closed", "head-classifier"]
-        );
+        assert_eq!(ids(&f), BUILTIN_IDS);
         f.apply(Read::Absent);
-        assert_eq!(
-            ids(&f),
-            ["assigned", "unassigned", "closed", "head-classifier"]
-        );
+        assert_eq!(ids(&f), BUILTIN_IDS);
         // An empty document is the same.
         f.apply(at(1, doc(&[], &[])));
-        assert_eq!(
-            ids(&f),
-            ["assigned", "unassigned", "closed", "head-classifier"]
-        );
+        assert_eq!(ids(&f), BUILTIN_IDS);
     }
 
     #[test]
     fn a_document_switches_builtins_off_and_adds_custom_flows() {
         let mut f = Flows::builtins();
         f.apply(at(1, doc(&["unassigned"], &["triage"])));
-        assert_eq!(ids(&f), ["assigned", "closed", "head-classifier", "triage"]);
+        assert_eq!(
+            ids(&f),
+            [
+                "assigned",
+                "closed",
+                "head-classifier",
+                "ci-failed",
+                "ci-cancelled",
+                "ci-cleared",
+                "triage"
+            ]
+        );
     }
 
     #[test]
@@ -217,7 +230,10 @@ mod tests {
         let mut f = Flows::builtins();
         f.apply(at(1, doc(&["closed"], &["triage"])));
         let before = ids(&f).join(",");
-        assert_eq!(before, "assigned,unassigned,head-classifier,triage");
+        assert_eq!(
+            before,
+            "assigned,unassigned,head-classifier,ci-failed,ci-cancelled,ci-cleared,triage"
+        );
         for bad in [
             junk(2, b"not json"),
             junk(3, br#"{"version":2,"flows":[]}"#),
@@ -230,20 +246,14 @@ mod tests {
         // The next good revision replaces them: the control that the bad
         // ones were kept out by being bad, not by the loader being stuck.
         f.apply(at(5, doc(&[], &[])));
-        assert_eq!(
-            ids(&f),
-            ["assigned", "unassigned", "closed", "head-classifier"]
-        );
+        assert_eq!(ids(&f), BUILTIN_IDS);
     }
 
     #[test]
     fn a_bad_first_document_is_the_builtins_not_nothing() {
         let mut f = Flows::builtins();
         f.apply(junk(1, br#"{"version":9}"#));
-        assert_eq!(
-            ids(&f),
-            ["assigned", "unassigned", "closed", "head-classifier"]
-        );
+        assert_eq!(ids(&f), BUILTIN_IDS);
     }
 
     #[test]
@@ -251,10 +261,7 @@ mod tests {
         let mut f = Flows::builtins();
         f.apply(at(1, doc(&["closed"], &["triage"])));
         f.apply(Read::Absent);
-        assert_eq!(
-            ids(&f),
-            ["assigned", "unassigned", "closed", "head-classifier"]
-        );
+        assert_eq!(ids(&f), BUILTIN_IDS);
     }
 
     #[test]
@@ -272,9 +279,6 @@ mod tests {
             revision: 1,
             bytes: serde_json::to_vec(&d).unwrap(),
         });
-        assert_eq!(
-            ids(&f),
-            ["assigned", "unassigned", "closed", "head-classifier"]
-        );
+        assert_eq!(ids(&f), BUILTIN_IDS);
     }
 }