an agent gets one todo when its pr starts conflicting with the target branch

swarm-controller/src/forge.rs

@@ -46,6 +46,7 @@ use swarm_wire::review_classifier::Commit;
 
 pub mod agent_token;
 mod api;
+mod base_pulls;
 pub mod builder_account;
 pub mod ci_metrics_account;
 pub mod classifier_account;
@@ -2321,10 +2322,12 @@ mod tests {
             ["workflow_run_failure", "workflow_run_cancelled"]
         );
         // The events a stored flow's source may name are the ones the hooks
-        // deliver to the router.
+        // deliver to the router, and the one the controller makes.
         let mut routed = events(DeliveryKind::Issues);
         routed.extend(events(DeliveryKind::PullRequest));
         routed.extend(events(DeliveryKind::WorkflowRun));
+        // Not a hook: the controller publishes it itself (`crate::pr_conflicts`).
+        routed.push(swarm_wire::forge_events::MERGEABILITY_EVENT);
         routed.sort_unstable();
         let mut sources = swarm_wire::notify_flows::SOURCE_EVENTS
             .iter()

swarm-controller/src/forge/base_pulls.rs

@@ -0,0 +1,42 @@
+//! The open pull requests that target one branch.
+
+use anyhow::{Context, Result};
+use forgejo_api::structs::{
+    PullRequest, RepoListPullRequestsQuery, RepoListPullRequestsQueryState,
+};
+
+use super::Client;
+
+impl Client {
+    /// Every open pull request in `owner/repo` whose target branch is `base`,
+    /// as the forge lists them: one call per page of 50.
+    ///
+    /// # Errors
+    ///
+    /// The forge refused or did not answer.
+    pub async fn open_pulls_on(
+        &self,
+        owner: &str,
+        repo: &str,
+        base: &str,
+    ) -> Result<Vec<PullRequest>> {
+        let query = RepoListPullRequestsQuery {
+            state: Some(RepoListPullRequestsQueryState::Open),
+            sort: None,
+            milestone: None,
+            labels: None,
+            poster: None,
+            base: Some(base.to_owned()),
+            head: None,
+        };
+        self.api
+            .call(|api| {
+                Box::pin(
+                    api.repo_list_pull_requests(owner, repo, query.clone())
+                        .all(),
+                )
+            })
+            .await
+            .with_context(|| format!("list open pull requests on {base} in {owner}/{repo}"))
+    }
+}

swarm-controller/src/forge_events.rs

@@ -10,6 +10,9 @@
 //! `pull_request_sync`, …), not `X-Forgejo-Event`, which folds those into
 //! `issues` and `pull_request`. It reaches the subject only once it is one
 //! plain token.
+//!
+//! [`publish_event`] keeps an event this daemon makes, such as the
+//! `pull_request_mergeability` event of `crate::pr_conflicts`, the same way.
 
 use axum::http::HeaderMap;
 use std::time::Duration;
@@ -51,15 +54,42 @@ pub(crate) async fn publish(
     body: axum::body::Bytes,
 ) -> Result<(), NotKept> {
     let subject = subject(headers)?;
+    let mut message = async_nats::jetstream::message::PublishMessage::build().payload(body);
+    if let Some(id) = headers.get(DELIVERY_HEADER).and_then(|v| v.to_str().ok()) {
+        message = message.message_id(id);
+    }
+    keep(state, subject, message).await
+}
+
+/// Keep an event this daemon makes itself, of `event_type`, in the stream, and
+/// wait for the stream's ack. The router reads it like a delivery.
+pub(crate) async fn publish_event(
+    state: &crate::routes::AppState,
+    event_type: &str,
+    body: Vec<u8>,
+) -> Result<(), NotKept> {
+    if !swarm_wire::forge_events::is_subject_token(event_type) {
+        return Err(NotKept::BadEventType);
+    }
+    let message = async_nats::jetstream::message::PublishMessage::build().payload(body.into());
+    keep(
+        state,
+        swarm_wire::forge_events::subject(event_type),
+        message,
+    )
+    .await
+}
+
+async fn keep(
+    state: &crate::routes::AppState,
+    subject: String,
+    message: async_nats::jetstream::message::PublishMessage,
+) -> Result<(), NotKept> {
     let Some(status) = state.status.as_ref() else {
         return Err(NotKept::Queue("no swarm queue is configured".to_owned()));
     };
     let client = status.queue_client();
     swarm_queue_client::ensure_connected(&client).map_err(|e| NotKept::Queue(e.to_string()))?;
-    let mut message = async_nats::jetstream::message::PublishMessage::build().payload(body);
-    if let Some(id) = headers.get(DELIVERY_HEADER).and_then(|v| v.to_str().ok()) {
-        message = message.message_id(id);
-    }
     let kept = async {
         swarm_queue_client::forge_events::open_or_create(&client)
             .await

swarm-controller/src/main.rs

@@ -72,6 +72,7 @@ mod notify_router_api;
 mod observability_grant;
 mod otel_http_client;
 mod pause_hold;
+mod pr_conflicts;
 mod pr_report;
 mod proxy_credential;
 mod pulls;

swarm-controller/src/pr_conflicts.rs

@@ -0,0 +1,517 @@
+//! Telling the notification router when an open pull request stops or starts
+//! merging into its target branch.
+//!
+//! Forgejo sends no delivery when a push to a branch changes whether the pull
+//! requests targeting it merge, and the router holds no forge client. So the
+//! controller watches the pushes it already receives on the `vcs-activity`
+//! hook: after a push to a branch it waits for the forge to recompute, lists
+//! the open pull requests that target the branch, and publishes one
+//! [`MERGEABILITY_EVENT`] per pull request. The router keeps whether each pull
+//! request was already conflicting and acts only on a change.
+//!
+//! The forge's `mergeable` flag is false for a conflict, for an error, for a
+//! draft, and while the check still runs. A draft is left out, and a pull
+//! request counts as conflicting only when two reads one [`RECHECK`] apart both
+//! say false, which a check still running does not survive.
+
+use std::collections::{BTreeSet, HashMap, HashSet};
+use std::future::Future;
+use std::sync::{LazyLock, Mutex, PoisonError};
+use std::time::Duration;
+
+use forgejo_api::structs::PullRequest;
+use serde_json::{Value as Json, json};
+use swarm_wire::forge_events::MERGEABILITY_EVENT;
+
+use crate::routes::AppState;
+
+/// How long after a push the forge gets to recompute before the first read.
+const SETTLE: Duration = Duration::from_secs(30);
+
+/// The wait between reads of a pull request that did not read as mergeable.
+const RECHECK: Duration = Duration::from_secs(30);
+
+/// The most reads one check makes.
+const MAX_READS: usize = 4;
+
+/// One push to a branch, as far as this module needs it.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub(crate) struct BranchPush {
+    /// `owner/name`.
+    pub(crate) repo: String,
+    pub(crate) repo_id: i64,
+    pub(crate) branch: String,
+}
+
+impl BranchPush {
+    /// The push a `push` webhook body describes, or `None` for anything else:
+    /// a tag, a push that deleted its branch, or a body that is no push.
+    pub(crate) fn parse(body: &[u8]) -> Option<Self> {
+        let json: Json = serde_json::from_slice(body).ok()?;
+        let branch = json.get("ref")?.as_str()?.strip_prefix("refs/heads/")?;
+        let deleted = json
+            .get("after")
+            .and_then(Json::as_str)
+            .is_some_and(|after| !after.is_empty() && after.bytes().all(|b| b == b'0'));
+        let repository = json.get("repository")?;
+        if branch.is_empty() || deleted {
+            return None;
+        }
+        Some(Self {
+            repo: repository.get("full_name")?.as_str()?.to_owned(),
+            repo_id: repository.get("id")?.as_i64()?,
+            branch: branch.to_owned(),
+        })
+    }
+
+    fn key(&self) -> (String, String) {
+        (self.repo.clone(), self.branch.clone())
+    }
+}
+
+/// The branches with a check under way, so a burst of pushes to one branch is
+/// one check and not one each.
+#[derive(Default)]
+pub(crate) struct Debounce {
+    /// Whether a push arrived since the check last started.
+    running: Mutex<HashMap<(String, String), bool>>,
+}
+
+impl Debounce {
+    /// A push to `key`. True when no check was running, which the caller
+    /// starts; false when one was, which will run again once it finishes.
+    pub(crate) fn begin(&self, key: (String, String)) -> bool {
+        let mut running = self.running.lock().unwrap_or_else(PoisonError::into_inner);
+        if let Some(pushed_again) = running.get_mut(&key) {
+            *pushed_again = true;
+            return false;
+        }
+        running.insert(key, false);
+        true
+    }
+
+    /// A check of `key` is done. True when a push arrived meanwhile, so the
+    /// check runs again; false when it was the last, and the branch is free.
+    pub(crate) fn finish(&self, key: &(String, String)) -> bool {
+        let mut running = self.running.lock().unwrap_or_else(PoisonError::into_inner);
+        match running.get_mut(key) {
+            Some(pushed_again) if *pushed_again => {
+                *pushed_again = false;
+                true
+            }
+            _ => {
+                running.remove(key);
+                false
+            }
+        }
+    }
+}
+
+static RUNNING: LazyLock<Debounce> = LazyLock::new(Debounce::default);
+
+/// What a check needs of one listed pull request.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub(crate) struct Pull {
+    pub(crate) number: i64,
+    title: String,
+    url: String,
+    author: String,
+    assignees: BTreeSet<String>,
+    head: String,
+    draft: bool,
+    /// The forge's flag: `None` when it did not say.
+    mergeable: Option<bool>,
+}
+
+impl Pull {
+    /// `pull` as listed, or `None` when it names no number or no author, so
+    /// there is nobody to tell.
+    pub(crate) fn from_listed(pull: &PullRequest) -> Option<Self> {
+        let login = |u: &forgejo_api::structs::User| u.login.clone();
+        Some(Self {
+            number: pull.number?,
+            title: pull.title.clone().unwrap_or_default(),
+            url: pull
+                .html_url
+                .as_ref()
+                .map(ToString::to_string)
+                .unwrap_or_default(),
+            author: pull.user.as_ref().and_then(login)?,
+            assignees: pull.assignees.iter().flatten().filter_map(login).collect(),
+            head: pull
+                .head
+                .as_ref()
+                .and_then(|h| h.sha.clone())
+                .unwrap_or_default(),
+            draft: pull.draft.unwrap_or(false),
+            mergeable: pull.mergeable,
+        })
+    }
+
+    /// The body of the event that says `action` (`conflicting` or
+    /// `mergeable`) of this pull request: a `pull_request` delivery the
+    /// router's parser reads.
+    fn event_body(&self, push: &BranchPush, action: &str) -> Vec<u8> {
+        let user = |login: &str| json!({ "login": login });
+        let body = json!({
+            "action": action,
+            "number": self.number,
+            "pull_request": {
+                "number": self.number,
+                "title": self.title,
+                "html_url": self.url,
+                "user": user(&self.author),
+                "assignees": self.assignees.iter().map(|a| user(a)).collect::<Vec<_>>(),
+                "head": { "sha": self.head },
+                "base": { "ref": push.branch },
+                "draft": self.draft,
+                "mergeable": action == "mergeable",
+            },
+            "repository": { "id": push.repo_id, "full_name": push.repo },
+        });
+        serde_json::to_vec(&body).unwrap_or_default()
+    }
+}
+
+/// Read the open pull requests on `push`'s branch with `list`, up to
+/// [`MAX_READS`] times [`RECHECK`] apart, and `publish` what each says: the
+/// action `mergeable` the first read that finds it so, and `conflicting` the
+/// second read in a row that finds the forge's flag false. A pull request that
+/// is a draft, or never settles, gets nothing.
+pub(crate) async fn check<L, LF, P, PF>(push: &BranchPush, recheck: Duration, list: L, publish: P)
+where
+    L: Fn() -> LF,
+    LF: Future<Output = anyhow::Result<Vec<Pull>>>,
+    P: Fn(&'static str, Vec<u8>) -> PF,
+    PF: Future<Output = ()>,
+{
+    let mut after_false = HashSet::new();
+    let mut settled = HashSet::new();
+    for read in 0..MAX_READS {
+        if read > 0 {
+            tokio::time::sleep(recheck).await;
+        }
+        let pulls = match list().await {
+            Ok(pulls) => pulls,
+            Err(e) => {
+                tracing::warn!(repo = %push.repo, branch = %push.branch, error = ?e, "pr conflicts: listing pull requests failed");
+                return;
+            }
+        };
+        for pull in &pulls {
+            if pull.draft || settled.contains(&pull.number) {
+                continue;
+            }
+            let action = match pull.mergeable {
+                Some(true) => "mergeable",
+                Some(false) if after_false.contains(&pull.number) => "conflicting",
+                Some(false) => {
+                    after_false.insert(pull.number);
+                    continue;
+                }
+                None => {
+                    after_false.remove(&pull.number);
+                    continue;
+                }
+            };
+            settled.insert(pull.number);
+            publish(action, pull.event_body(push, action)).await;
+        }
+        if pulls.iter().all(|p| p.draft || settled.contains(&p.number)) {
+            return;
+        }
+    }
+}
+
+/// A push arrived on the `vcs-activity` hook. Starts the check for its branch
+/// unless one is already under way, in which case that one runs again when it
+/// is done.
+pub(crate) fn on_push(state: &AppState, body: &[u8]) {
+    let Some(push) = BranchPush::parse(body) else {
+        return;
+    };
+    if RUNNING.begin(push.key()) {
+        tokio::spawn(watch(state.clone(), push));
+    }
+}
+
+async fn watch(state: AppState, push: BranchPush) {
+    let Some((owner, name)) = push.repo.split_once('/') else {
+        RUNNING.finish(&push.key());
+        return;
+    };
+    loop {
+        tokio::time::sleep(SETTLE).await;
+        if let Some(forge) = state.forge.get() {
+            check(
+                &push,
+                RECHECK,
+                || async {
+                    let listed = forge.open_pulls_on(owner, name, &push.branch).await?;
+                    Ok(listed.iter().filter_map(Pull::from_listed).collect())
+                },
+                |action, body| {
+                    let (state, push) = (&state, &push);
+                    async move {
+                        if let Err(not_kept) =
+                            crate::forge_events::publish_event(state, MERGEABILITY_EVENT, body)
+                                .await
+                        {
+                            tracing::warn!(repo = %push.repo, action, reason = ?not_kept, "pr conflicts: event not kept");
+                        }
+                    }
+                },
+            )
+            .await;
+        }
+        if !RUNNING.finish(&push.key()) {
+            return;
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::{Arc, Mutex};
+
+    fn push() -> BranchPush {
+        BranchPush {
+            repo: "o/r".to_owned(),
+            repo_id: 7,
+            branch: "main".to_owned(),
+        }
+    }
+
+    fn pull(number: i64, mergeable: Option<bool>) -> Pull {
+        Pull {
+            number,
+            title: "t".to_owned(),
+            url: "https://forge/o/r/pulls/1".to_owned(),
+            author: "atlas".to_owned(),
+            assignees: BTreeSet::from(["iris".to_owned()]),
+            head: "abc".to_owned(),
+            draft: false,
+            mergeable,
+        }
+    }
+
+    /// What `check` published for the reads `reads` give in turn, as
+    /// `(action, number)`, and how many reads it made.
+    async fn run(reads: Vec<Vec<Pull>>) -> (Vec<(String, i64)>, usize) {
+        let seen = Arc::new(Mutex::new(Vec::new()));
+        let calls = Arc::new(Mutex::new(0_usize));
+        let reads = Arc::new(reads);
+        check(
+            &push(),
+            Duration::ZERO,
+            || {
+                let (reads, calls) = (Arc::clone(&reads), Arc::clone(&calls));
+                async move {
+                    let mut n = calls.lock().expect("lock");
+                    let read = reads.get(*n).or(reads.last()).cloned().unwrap_or_default();
+                    *n += 1;
+                    Ok(read)
+                }
+            },
+            |action, body| {
+                let seen = Arc::clone(&seen);
+                async move {
+                    let json: Json = serde_json::from_slice(&body).expect("json");
+                    assert_eq!(json["action"], action);
+                    seen.lock()
+                        .expect("lock")
+                        .push((action.to_owned(), json["number"].as_i64().expect("number")));
+                }
+            },
+        )
+        .await;
+        let published = seen.lock().expect("lock").clone();
+        let made = *calls.lock().expect("lock");
+        (published, made)
+    }
+
+    fn act(a: &str, n: i64) -> (String, i64) {
+        (a.to_owned(), n)
+    }
+
+    #[tokio::test]
+    async fn each_pull_request_is_published_by_what_the_forge_says() {
+        let (published, reads) = run(vec![vec![pull(1, Some(true)), pull(2, Some(false))]]).await;
+        assert_eq!(published[0], act("mergeable", 1));
+        assert_eq!(reads, 2, "the conflicting one is read again to confirm it");
+        assert_eq!(published.len(), 2);
+        assert_eq!(published[1], act("conflicting", 2));
+    }
+
+    #[tokio::test]
+    async fn all_mergeable_pull_requests_cost_one_read() {
+        let (published, reads) = run(vec![vec![pull(1, Some(true)), pull(2, Some(true))]]).await;
+        assert_eq!(published, [act("mergeable", 1), act("mergeable", 2)]);
+        assert_eq!(reads, 1);
+        let (none, reads) = run(vec![vec![]]).await;
+        assert!(none.is_empty());
+        assert_eq!(reads, 1);
+    }
+
+    /// A check still running reads as false; it must not become a conflict.
+    #[tokio::test]
+    async fn a_check_still_running_is_not_a_conflict() {
+        let (published, reads) =
+            run(vec![vec![pull(1, Some(false))], vec![pull(1, Some(true))]]).await;
+        assert_eq!(published, [act("mergeable", 1)]);
+        assert_eq!(reads, 2);
+    }
+
+    #[tokio::test]
+    async fn a_pull_request_the_forge_has_not_decided_is_read_again() {
+        let (published, reads) = run(vec![
+            vec![pull(1, None)],
+            vec![pull(1, None)],
+            vec![pull(1, Some(true))],
+        ])
+        .await;
+        assert_eq!(published, [act("mergeable", 1)]);
+        assert_eq!(reads, 3);
+
+        let (published, reads) = run(vec![vec![pull(1, None)]]).await;
+        assert!(published.is_empty(), "never decided, never published");
+        assert_eq!(reads, MAX_READS);
+
+        // False, then undecided, then false is not two false reads in a row.
+        let (published, _) = run(vec![
+            vec![pull(1, Some(false))],
+            vec![pull(1, None)],
+            vec![pull(1, Some(false))],
+            vec![pull(1, None)],
+        ])
+        .await;
+        assert!(published.is_empty());
+    }
+
+    #[tokio::test]
+    async fn a_draft_is_left_alone() {
+        let draft = Pull {
+            draft: true,
+            ..pull(1, Some(false))
+        };
+        let (published, reads) = run(vec![vec![draft.clone()]]).await;
+        assert!(published.is_empty());
+        assert_eq!(reads, 1);
+
+        // Beside a pull request that needs a second read, the draft would
+        // read false twice, as it always does.
+        let (published, reads) = run(vec![vec![draft, pull(2, Some(false))]]).await;
+        assert_eq!(published, [act("conflicting", 2)]);
+        assert_eq!(reads, 2);
+    }
+
+    #[tokio::test]
+    async fn a_failed_listing_publishes_nothing() {
+        let published = Arc::new(Mutex::new(0_usize));
+        let counted = Arc::clone(&published);
+        check(
+            &push(),
+            Duration::ZERO,
+            || async { anyhow::bail!("forge down") },
+            |_, _| {
+                let counted = Arc::clone(&counted);
+                async move {
+                    *counted.lock().expect("lock") += 1;
+                }
+            },
+        )
+        .await;
+        assert_eq!(*published.lock().expect("lock"), 0);
+    }
+
+    /// The body is a `pull_request` delivery in the shape the router parses.
+    #[test]
+    fn the_event_body_names_the_author_the_assignees_and_the_base() {
+        let body = pull(3, Some(false)).event_body(&push(), "conflicting");
+        let json: Json = serde_json::from_slice(&body).expect("json");
+        assert_eq!(json["action"], "conflicting");
+        assert_eq!(json["repository"]["id"], 7);
+        assert_eq!(json["repository"]["full_name"], "o/r");
+        assert_eq!(json["pull_request"]["number"], 3);
+        assert_eq!(json["pull_request"]["user"]["login"], "atlas");
+        assert_eq!(json["pull_request"]["assignees"][0]["login"], "iris");
+        assert_eq!(json["pull_request"]["base"]["ref"], "main");
+        assert_eq!(json["pull_request"]["head"]["sha"], "abc");
+    }
+
+    #[test]
+    fn a_listed_pull_request_is_read_and_one_without_an_author_is_not() {
+        let listed = |user: Json| -> PullRequest {
+            serde_json::from_value(json!({
+                "number": 4,
+                "title": "t",
+                "html_url": "https://forge/o/r/pulls/4",
+                "user": user,
+                "assignees": [{ "login": "iris" }],
+                "head": { "sha": "abc" },
+                "draft": false,
+                "mergeable": false,
+                "labels": [],
+                "requested_reviewers": [],
+                "base": { "ref": "main" },
+                "created_at": "2026-11-07T08:00:00Z",
+                "updated_at": "2026-11-07T09:00:00Z",
+                "additions": 1,
+                "deletions": 1,
+            }))
+            .expect("a pull request")
+        };
+        let got = Pull::from_listed(&listed(json!({ "login": "atlas" }))).expect("read");
+        assert_eq!(got.number, 4);
+        assert_eq!(got.author, "atlas");
+        assert_eq!(got.mergeable, Some(false));
+        assert_eq!(got.assignees, BTreeSet::from(["iris".to_owned()]));
+        assert_eq!(Pull::from_listed(&listed(Json::Null)), None);
+    }
+
+    fn push_body(git_ref: &str, after: &str) -> Vec<u8> {
+        serde_json::to_vec(&json!({
+            "ref": git_ref,
+            "after": after,
+            "repository": { "id": 7, "full_name": "o/r" },
+        }))
+        .expect("json")
+    }
+
+    #[test]
+    fn a_push_to_a_branch_is_read_and_a_tag_or_a_delete_is_not() {
+        assert_eq!(
+            BranchPush::parse(&push_body("refs/heads/feat/x", "abc123")),
+            Some(BranchPush {
+                repo: "o/r".to_owned(),
+                repo_id: 7,
+                branch: "feat/x".to_owned(),
+            })
+        );
+        assert_eq!(
+            BranchPush::parse(&push_body("refs/tags/v1", "abc123")),
+            None
+        );
+        assert_eq!(
+            BranchPush::parse(&push_body("refs/heads/main", "0000000")),
+            None
+        );
+        assert_eq!(BranchPush::parse(b"not json"), None);
+    }
+
+    /// Pushes during a check make it run once more, however many they are.
+    #[test]
+    fn pushes_during_a_check_are_one_more_check() {
+        let d = Debounce::default();
+        let key = |b: &str| ("o/r".to_owned(), b.to_owned());
+        assert!(d.begin(key("main")), "the first push starts a check");
+        assert!(!d.begin(key("main")));
+        assert!(!d.begin(key("main")));
+        assert!(d.begin(key("other")), "another branch has its own");
+        assert!(d.finish(&key("main")), "the pushes since ask for one more");
+        assert!(!d.finish(&key("main")), "and then the branch is free");
+        assert!(d.begin(key("main")), "so the next push starts a new check");
+        assert!(!d.finish(&key("other")));
+    }
+}

swarm-controller/src/webhook.rs

@@ -406,6 +406,9 @@ pub(super) enum DeliveryKind {
     /// hook: that one is repo-scoped to the knowledge repo alone, and this
     /// one is instance-wide (registered via `admin_create_hook`, not
     /// `repo_create_hook`) — different scope, same event name.
+    ///
+    /// A push to a branch also starts the check of the PRs that target it
+    /// (`crate::pr_conflicts`).
     #[strum(serialize = "vcs-activity")]
     VcsActivity,
     /// `pull_request_review_request` events on every repo in the instance,
@@ -647,6 +650,7 @@ pub(super) async fn post_webhook_forge(
                 tokio::spawn(crate::vcs_metrics::count_default_branch_push(
                     activity, forge,
                 ));
+                crate::pr_conflicts::on_push(&state, &body);
             } else {
                 tracing::warn!("webhook: vcs-activity delivery did not parse as a push payload");
             }