agents get a todo when assigned to an issue or PR, rewritten when unassigned and cleared on close

A new swarm service, swarm-notify-router, in its own container on the host
that sets deploy.notify-router.enable (defaulted on by allSwarmServices where
the host has the router's store leaf, like the claude login minter).

- swarm-controller registers instance-wide hooks for `issues` +
  `issue_assign` and `pull_request` + `pull_request_assign`, updates the
  event list of a hook already registered with fewer (leaving its active flag as it is), and keeps each
  verified body unchanged in the SWARM_EVENTS stream on
  $SWARM.events.forge.<X-Forgejo-Event-Type>, deduped by X-Forgejo-Delivery.
  A delivery the stream did not keep answers 503.
- swarm-notify-router drains its durable consumer on that stream, runs its
  flows (source -> match -> todo sink, stored as nodes plus edges and checked
  to be a tree) and queues each op on $SWARM.router.todo.<agent> in the
  work-queue stream ROUTER_TODOS. Built-in flows: assigned, unassigned (same
  key, rewritten) and closed (prefix-clears every holder's router todos). A
  per-issue snapshot in its notify-router-state bucket holds the last
  assignees and the todo holders; only agents in the agent-status bucket get
  ops, and an assignee that was not one is left out of the snapshot so the
  next assign event retries it. After the first event that must be retried,
  the rest of the batch is nakked so redelivery keeps stream order.
- hive-agent creates its own durable consumer todo-<agent> and applies each
  op to its todo store as subsystem `router`, then acks; a refused op stops
  the rest of its batch the same way. The store's (subsystem, key) dedup
  makes a redelivery harmless. New Todos::clear_prefix.
- swarm-nats-auth: a --notify-router-client arm; the controller may publish
  $SWARM.events.forge.*; each agent gets three {agent} templates (extended
  consumer create whose filter the server checks, MSG.NEXT, ACK).
- Credentials: a notify-router store leaf minted by glue-bao-tls, a
  swarm-notify-router bao policy (read on its own queue client secret), an
  authelia machine client, the secret publisher entry, a journal forwarder
  route. No manual step.

hive-forge-notify is unchanged, so agents also still get forge's own
notifications.

Refs #5571

swarm-notify-router/Cargo.toml

@@ -0,0 +1,34 @@
+[package]
+name = "swarm-notify-router"
+version.workspace = true
+readme = "README.md"
+edition.workspace = true
+
+[[bin]]
+name = "swarm-notify-router"
+path = "src/main.rs"
+
+[dependencies]
+anyhow.workspace = true
+# `kv` for its state bucket; `jetstream` comes with it.
+async-nats = { workspace = true, features = ["kv"] }
+# StreamExt / TryStreamExt on the pulled batch and the subject listing.
+futures-util.workspace = true
+hive-log.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+# The connect, and the two streams' configs it shares with the controller and
+# the agent harness.
+swarm-queue-client = { workspace = true, features = [
+  "forge-events",
+  "router-todo",
+] }
+# Its queue client secret, read with its own store leaf.
+swarm-secret-client.workspace = true
+# The stream, subject, bucket and op names.
+swarm-wire.workspace = true
+tokio.workspace = true
+tracing.workspace = true
+
+[lints]
+workspace = true

swarm-notify-router/README.md

@@ -0,0 +1,8 @@
+# swarm-notify-router
+
+Routes forge events to agents' todo lists. It drains its consumer on the
+forge events stream that swarm-controller fills from its webhooks, runs its
+flows over each delivery, and queues the todo ops they emit on each agent's
+subject of the `ROUTER_TODOS` stream. Each agent's harness applies them.
+
+Design: [`docs/swarm/notify-router.md`](../docs/swarm/notify-router.md).

swarm-notify-router/src/event.rs

@@ -0,0 +1,233 @@
+//! A forge delivery as the fields a flow reads, and the per-issue snapshot
+//! that fills in what the delivery leaves out.
+//!
+//! Forgejo's `issue_assign` and `pull_request_assign` deliveries carry the
+//! assignees *after* the change and not which one changed. [`Snapshot`] holds
+//! the assignees the router last saw on an assign delivery, so the added and
+//! removed ones are the difference. With no snapshot yet, every current
+//! assignee counts as added and none as removed: an unassign is noticed only
+//! on an issue the router has seen an assign delivery for.
+
+use std::collections::BTreeSet;
+
+use serde::{Deserialize, Serialize};
+use serde_json::Value as Json;
+
+use crate::flow::{Fields, Value};
+
+/// The parts of an `issues` or `pull_request` family delivery the router
+/// uses.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct Delivery {
+    pub action: String,
+    pub repo: String,
+    pub repo_id: i64,
+    pub number: i64,
+    pub title: String,
+    pub url: String,
+    pub sender: String,
+    pub is_pull: bool,
+    pub assignees: BTreeSet<String>,
+}
+
+impl Delivery {
+    /// Read a delivery body, or `None` when it names no issue or PR.
+    pub fn parse(body: &[u8]) -> Option<Self> {
+        let json: Json = serde_json::from_slice(body).ok()?;
+        let (item, is_pull) = match json.get("pull_request") {
+            Some(pr) if pr.is_object() => (pr, true),
+            _ => (json.get("issue")?, false),
+        };
+        let text = |v: &Json, k: &str| {
+            v.get(k)
+                .and_then(Json::as_str)
+                .unwrap_or_default()
+                .to_owned()
+        };
+        let repository = json.get("repository")?;
+        let assignees = item
+            .get("assignees")
+            .and_then(Json::as_array)
+            .map(|a| {
+                a.iter()
+                    .filter_map(|u| u.get("login").and_then(Json::as_str))
+                    .map(str::to_owned)
+                    .collect()
+            })
+            .unwrap_or_default();
+        Some(Self {
+            action: text(&json, "action"),
+            repo: text(repository, "full_name"),
+            repo_id: repository.get("id")?.as_i64()?,
+            number: item.get("number")?.as_i64()?,
+            title: text(item, "title"),
+            url: text(item, "html_url"),
+            sender: json
+                .get("sender")
+                .map(|s| text(s, "login"))
+                .unwrap_or_default(),
+            is_pull,
+            assignees,
+        })
+    }
+
+    /// 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 {
+        format!("{}.{}", self.repo_id, self.number)
+    }
+}
+
+/// What the router keeps per issue in its state bucket.
+#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct Snapshot {
+    /// The assignees after the last assign delivery seen, less those that
+    /// were no agent of the roster then, or `None` before the first.
+    #[serde(default)]
+    pub assignees: Option<BTreeSet<String>>,
+    /// The agents holding a router todo for this issue.
+    #[serde(default)]
+    pub holders: BTreeSet<String>,
+}
+
+/// Whether `event_type` is an assign delivery, whose assignee list is the
+/// set after the change.
+pub fn is_assign(event_type: &str) -> bool {
+    matches!(event_type, "issue_assign" | "pull_request_assign")
+}
+
+/// The fields a flow reads for `delivery`, given the issue's `snapshot`.
+pub fn fields(event_type: &str, delivery: &Delivery, snapshot: &Snapshot) -> Fields {
+    let (added, removed) = if is_assign(event_type) {
+        match &snapshot.assignees {
+            Some(before) => (
+                delivery.assignees.difference(before).cloned().collect(),
+                before.difference(&delivery.assignees).cloned().collect(),
+            ),
+            None => (delivery.assignees.iter().cloned().collect(), Vec::new()),
+        }
+    } else {
+        (Vec::new(), Vec::new())
+    };
+    let text = |s: &str| Value::Text(s.to_owned());
+    Fields::from([
+        ("event", text(event_type)),
+        ("action", text(&delivery.action)),
+        ("kind", text(if delivery.is_pull { "PR" } else { "issue" })),
+        ("repo", text(&delivery.repo)),
+        ("number", Value::Text(delivery.number.to_string())),
+        ("title", text(&delivery.title)),
+        ("url", text(&delivery.url)),
+        ("sender", text(&delivery.sender)),
+        (
+            "assignees",
+            Value::List(delivery.assignees.iter().cloned().collect()),
+        ),
+        ("assignees_added", Value::List(added)),
+        ("assignees_removed", Value::List(removed)),
+        (
+            "todo_holders",
+            Value::List(snapshot.holders.iter().cloned().collect()),
+        ),
+    ])
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn issue_assign(action: &str, assignees: &[&str]) -> Vec<u8> {
+        let users: Vec<Json> = assignees
+            .iter()
+            .map(|a| serde_json::json!({ "login": a }))
+            .collect();
+        serde_json::to_vec(&serde_json::json!({
+            "action": action,
+            "number": 7,
+            "issue": {
+                "number": 7,
+                "title": "router",
+                "html_url": "https://forge/issue-7",
+                "assignees": users,
+            },
+            "repository": { "id": 42, "full_name": "hyperhive/hyperhive" },
+            "sender": { "login": "mara" },
+        }))
+        .expect("encodes")
+    }
+
+    fn list(fields: &Fields, name: &str) -> Vec<String> {
+        match fields.get(name) {
+            Some(Value::List(v)) => v.clone(),
+            other => panic!("{name}: {other:?}"),
+        }
+    }
+
+    #[test]
+    fn an_issue_delivery_reads_back_its_fields() {
+        let d = Delivery::parse(&issue_assign("assigned", &["atlas", "iris"])).expect("parses");
+        assert_eq!(d.repo, "hyperhive/hyperhive");
+        assert_eq!(d.number, 7);
+        assert!(!d.is_pull);
+        assert_eq!(d.state_key(), "42.7");
+        assert_eq!(
+            d.assignees,
+            BTreeSet::from(["atlas".to_owned(), "iris".to_owned()])
+        );
+    }
+
+    #[test]
+    fn a_pull_request_delivery_reads_the_pull_request_object() {
+        let body = serde_json::to_vec(&serde_json::json!({
+            "action": "closed",
+            "number": 7,
+            "pull_request": { "number": 7, "title": "t", "html_url": "u", "assignees": null },
+            "repository": { "id": 1, "full_name": "o/r" },
+            "sender": { "login": "s" },
+        }))
+        .expect("encodes");
+        let d = Delivery::parse(&body).expect("parses");
+        assert!(d.is_pull);
+        assert_eq!(d.number, 7);
+        assert!(d.assignees.is_empty());
+        assert_eq!(Delivery::parse(b"{\"ref\":\"main\"}"), None);
+    }
+
+    #[test]
+    fn the_snapshot_turns_the_after_list_into_added_and_removed() {
+        let d = Delivery::parse(&issue_assign("unassigned", &["iris"])).expect("parses");
+        let before = Snapshot {
+            assignees: Some(BTreeSet::from(["atlas".to_owned(), "iris".to_owned()])),
+            holders: BTreeSet::new(),
+        };
+        let f = fields("issue_assign", &d, &before);
+        assert_eq!(list(&f, "assignees_removed"), ["atlas"]);
+        assert!(list(&f, "assignees_added").is_empty());
+
+        let d = Delivery::parse(&issue_assign("assigned", &["atlas", "iris"])).expect("parses");
+        let before = Snapshot {
+            assignees: Some(BTreeSet::from(["iris".to_owned()])),
+            holders: BTreeSet::new(),
+        };
+        assert_eq!(
+            list(&fields("issue_assign", &d, &before), "assignees_added"),
+            ["atlas"]
+        );
+    }
+
+    #[test]
+    fn with_no_snapshot_every_assignee_is_added_and_none_removed() {
+        let d = Delivery::parse(&issue_assign("assigned", &["atlas"])).expect("parses");
+        let f = fields("issue_assign", &d, &Snapshot::default());
+        assert_eq!(list(&f, "assignees_added"), ["atlas"]);
+        assert!(list(&f, "assignees_removed").is_empty());
+    }
+
+    #[test]
+    fn a_delivery_that_is_not_an_assign_adds_and_removes_nobody() {
+        let d = Delivery::parse(&issue_assign("closed", &["atlas"])).expect("parses");
+        let f = fields("issues", &d, &Snapshot::default());
+        assert!(list(&f, "assignees_added").is_empty());
+        assert!(list(&f, "assignees_removed").is_empty());
+    }
+}

swarm-notify-router/src/flow.rs

@@ -0,0 +1,518 @@
+//! Flows: what the router does with an event.
+//!
+//! A flow is a graph of nodes joined by edges, rooted at one `source` node.
+//! A `match` node passes when every one of its conditions holds, and then
+//! each of its children runs, so one event can reach several branches. A
+//! `sink` node is a leaf and emits one todo op per recipient.
+//!
+//! Flows are trees: one source, one parent for every other node, every node
+//! reachable from the source. [`Flow::validate`] checks that, and the router
+//! runs only flows that pass.
+//!
+//! The built-in flows ([`builtins`]) are ordinary flows defined in code.
+
+use std::collections::{BTreeMap, BTreeSet};
+
+use swarm_wire::router_todo::{Op, RouterTodoOp};
+
+/// The most nodes one flow may have.
+const MAX_NODES: usize = 64;
+
+/// The deepest a node may sit below the source.
+const MAX_DEPTH: usize = 16;
+
+/// One event, as the fields a flow reads.
+pub type Fields = BTreeMap<&'static str, Value>;
+
+/// A field's value.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum Value {
+    Text(String),
+    List(Vec<String>),
+}
+
+#[derive(Clone, Debug)]
+pub struct Flow {
+    pub id: String,
+    pub nodes: Vec<Node>,
+    pub edges: Vec<Edge>,
+}
+
+#[derive(Clone, Debug)]
+pub struct Node {
+    pub id: String,
+    pub kind: NodeKind,
+}
+
+/// A directed edge, parent to child.
+#[derive(Clone, Debug)]
+pub struct Edge {
+    pub from: String,
+    pub to: String,
+}
+
+#[derive(Clone, Debug)]
+pub enum NodeKind {
+    /// The root: the forge event types the flow starts from.
+    Source { events: Vec<String> },
+    /// Passes when every condition holds.
+    Match { when: Vec<Condition> },
+    /// Emits a todo op to each recipient.
+    Todo(TodoSink),
+}
+
+/// `field` equals `value`. A list field holds when one of its items does.
+#[derive(Clone, Debug)]
+pub struct Condition {
+    pub field: String,
+    pub equals: String,
+}
+
+#[derive(Clone, Debug)]
+pub struct TodoSink {
+    /// The field naming the recipients, a list of agent names.
+    pub to: String,
+    pub op: Op,
+    /// The todo key, with `{field}` placeholders.
+    pub key: String,
+    /// For a clear: `key` is a prefix.
+    pub prefix: bool,
+    /// The todo text, with `{field}` placeholders. Unused for a clear.
+    pub text: String,
+}
+
+/// One op the walk emitted, for one recipient.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct Action {
+    pub flow: String,
+    pub node: String,
+    pub agent: String,
+    pub op: RouterTodoOp,
+}
+
+impl Flow {
+    /// Check the flow is a tree within the caps, or say why not.
+    pub fn validate(&self) -> Result<(), String> {
+        if self.nodes.len() > MAX_NODES {
+            return Err(format!("{} nodes, more than {MAX_NODES}", self.nodes.len()));
+        }
+        let mut ids = BTreeSet::new();
+        for node in &self.nodes {
+            if !ids.insert(node.id.as_str()) {
+                return Err(format!("node id {} is used twice", node.id));
+            }
+        }
+        for edge in &self.edges {
+            for end in [&edge.from, &edge.to] {
+                if !ids.contains(end.as_str()) {
+                    return Err(format!("an edge names {end}, which is no node"));
+                }
+            }
+        }
+        let roots: Vec<&Node> = self
+            .nodes
+            .iter()
+            .filter(|n| matches!(n.kind, NodeKind::Source { .. }))
+            .collect();
+        let [root] = roots.as_slice() else {
+            return Err(format!("{} source nodes, not one", roots.len()));
+        };
+        for node in &self.nodes {
+            if matches!(node.kind, NodeKind::Todo(_)) && !self.children(&node.id).is_empty() {
+                return Err(format!("sink {} has children", node.id));
+            }
+        }
+        // From the one root, every node reached exactly once: a node reached
+        // twice has a second parent or sits on a cycle, and one not reached is
+        // cut off.
+        let mut reached = 0;
+        let mut stack = vec![(root.id.as_str(), 0)];
+        while let Some((id, depth)) = stack.pop() {
+            if depth > MAX_DEPTH {
+                return Err(format!("node {id} is deeper than {MAX_DEPTH}"));
+            }
+            reached += 1;
+            if reached > self.nodes.len() {
+                return Err(
+                    "a node is reached twice: it has two parents or sits on a cycle".to_owned(),
+                );
+            }
+            stack.extend(self.children(id).into_iter().map(|c| (c, depth + 1)));
+        }
+        if reached != self.nodes.len() {
+            return Err(format!(
+                "{} of {} nodes are not reachable from the source",
+                self.nodes.len() - reached,
+                self.nodes.len()
+            ));
+        }
+        Ok(())
+    }
+
+    fn children(&self, id: &str) -> Vec<&str> {
+        self.edges
+            .iter()
+            .filter(|e| e.from == id)
+            .map(|e| e.to.as_str())
+            .collect()
+    }
+
+    fn node(&self, id: &str) -> Option<&Node> {
+        self.nodes.iter().find(|n| n.id == id)
+    }
+
+    /// Walk the flow for one event of type `event_type` and return the ops it
+    /// emits. Assumes [`Self::validate`] passed.
+    pub fn evaluate(&self, event_type: &str, fields: &Fields) -> Vec<Action> {
+        let Some(root) = self
+            .nodes
+            .iter()
+            .find(|n| matches!(&n.kind, NodeKind::Source { events } if events.iter().any(|e| e == event_type)))
+        else {
+            return Vec::new();
+        };
+        let mut actions = Vec::new();
+        let mut stack = vec![root.id.as_str()];
+        while let Some(id) = stack.pop() {
+            let Some(node) = self.node(id) else { continue };
+            match &node.kind {
+                NodeKind::Source { .. } => stack.extend(self.children(id)),
+                NodeKind::Match { when } => {
+                    if when.iter().all(|c| holds(c, fields)) {
+                        stack.extend(self.children(id));
+                    }
+                }
+                NodeKind::Todo(sink) => actions.extend(self.emit(node, sink, fields)),
+            }
+        }
+        actions
+    }
+
+    fn emit(&self, node: &Node, sink: &TodoSink, fields: &Fields) -> Vec<Action> {
+        let recipients = match fields.get(sink.to.as_str()) {
+            Some(Value::List(items)) => items.clone(),
+            Some(Value::Text(one)) => vec![one.clone()],
+            None => Vec::new(),
+        };
+        let op = RouterTodoOp {
+            op: sink.op,
+            key: expand(&sink.key, fields),
+            prefix: sink.op == Op::Clear && sink.prefix,
+            summary: match sink.op {
+                Op::Upsert => expand(&sink.text, fields),
+                Op::Clear => String::new(),
+            },
+            source: format!("router:{}", self.id),
+        };
+        recipients
+            .into_iter()
+            .map(|agent| Action {
+                flow: self.id.clone(),
+                node: node.id.clone(),
+                agent,
+                op: op.clone(),
+            })
+            .collect()
+    }
+}
+
+fn holds(condition: &Condition, fields: &Fields) -> bool {
+    match fields.get(condition.field.as_str()) {
+        Some(Value::Text(v)) => *v == condition.equals,
+        Some(Value::List(items)) => items.contains(&condition.equals),
+        None => false,
+    }
+}
+
+/// `template` with each `{field}` replaced by that text field's value. A list
+/// field is joined with `, `; an unknown name is left as written.
+pub fn expand(template: &str, fields: &Fields) -> String {
+    let mut out = String::with_capacity(template.len());
+    let mut rest = template;
+    while let Some(open) = rest.find('{') {
+        out.push_str(&rest[..open]);
+        let after = &rest[open + 1..];
+        let Some(close) = after.find('}') else {
+            out.push_str(&rest[open..]);
+            return out;
+        };
+        let name = &after[..close];
+        match fields.get(name) {
+            Some(Value::Text(v)) => out.push_str(v),
+            Some(Value::List(items)) => out.push_str(&items.join(", ")),
+            None => {
+                out.push('{');
+                out.push_str(name);
+                out.push('}');
+            }
+        }
+        rest = &after[close + 1..];
+    }
+    out.push_str(rest);
+    out
+}
+
+/// The flows the router ships with.
+///
+/// - `assigned`: a newly assigned agent gets a todo for the issue or PR.
+/// - `unassigned`: an agent taken off it gets that todo rewritten to say so.
+///   Both write the same key, so the second replaces the first.
+/// - `closed`: closing it clears every router todo the router queued for it.
+pub fn builtins() -> Vec<Flow> {
+    let assignment_key = "forge:{repo}#{number}:assignment";
+    vec![
+        chain(
+            "assigned",
+            &["issue_assign", "pull_request_assign"],
+            "assigned",
+            TodoSink {
+                to: "assignees_added".to_owned(),
+                op: Op::Upsert,
+                key: assignment_key.to_owned(),
+                prefix: false,
+                text: "you are assigned to {kind} {repo}#{number}: {title} ({url})".to_owned(),
+            },
+        ),
+        chain(
+            "unassigned",
+            &["issue_assign", "pull_request_assign"],
+            "unassigned",
+            TodoSink {
+                to: "assignees_removed".to_owned(),
+                op: Op::Upsert,
+                key: assignment_key.to_owned(),
+                prefix: false,
+                text: "you are not assigned to {kind} {repo}#{number} anymore: {title} ({url})"
+                    .to_owned(),
+            },
+        ),
+        chain(
+            "closed",
+            &["issues", "pull_request"],
+            "closed",
+            TodoSink {
+                to: "todo_holders".to_owned(),
+                op: Op::Clear,
+                key: "forge:{repo}#{number}:".to_owned(),
+                prefix: true,
+                text: String::new(),
+            },
+        ),
+    ]
+}
+
+/// source(`events`) → match(`action` = `action`) → `sink`.
+fn chain(id: &str, events: &[&str], action: &str, sink: TodoSink) -> Flow {
+    let node = |id: &str, kind| Node {
+        id: id.to_owned(),
+        kind,
+    };
+    let edge = |from: &str, to: &str| Edge {
+        from: from.to_owned(),
+        to: to.to_owned(),
+    };
+    Flow {
+        id: id.to_owned(),
+        nodes: vec![
+            node(
+                "source",
+                NodeKind::Source {
+                    events: events.iter().map(|e| (*e).to_owned()).collect(),
+                },
+            ),
+            node(
+                "match",
+                NodeKind::Match {
+                    when: vec![Condition {
+                        field: "action".to_owned(),
+                        equals: action.to_owned(),
+                    }],
+                },
+            ),
+            node("sink", NodeKind::Todo(sink)),
+        ],
+        edges: vec![edge("source", "match"), edge("match", "sink")],
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn text(v: &str) -> Value {
+        Value::Text(v.to_owned())
+    }
+
+    fn list(v: &[&str]) -> Value {
+        Value::List(v.iter().map(|s| (*s).to_owned()).collect())
+    }
+
+    fn event(action: &str, added: &[&str], removed: &[&str], holders: &[&str]) -> Fields {
+        Fields::from([
+            ("action", text(action)),
+            ("kind", text("issue")),
+            ("repo", text("hyperhive/hyperhive")),
+            ("number", text("7")),
+            ("title", text("router")),
+            ("url", text("https://forge/7")),
+            ("assignees_added", list(added)),
+            ("assignees_removed", list(removed)),
+            ("todo_holders", list(holders)),
+        ])
+    }
+
+    fn flow(id: &str) -> Flow {
+        builtins()
+            .into_iter()
+            .find(|f| f.id == id)
+            .expect("a built-in")
+    }
+
+    #[test]
+    fn every_builtin_is_a_valid_tree() {
+        for f in builtins() {
+            assert_eq!(f.validate(), Ok(()), "{}", f.id);
+        }
+    }
+
+    #[test]
+    fn assigning_queues_a_todo_for_each_newly_assigned_agent_only() {
+        let got = flow("assigned").evaluate(
+            "issue_assign",
+            &event("assigned", &["atlas", "iris"], &[], &[]),
+        );
+        let agents: Vec<&str> = got.iter().map(|a| a.agent.as_str()).collect();
+        assert_eq!(agents, ["atlas", "iris"]);
+        assert_eq!(got[0].op.op, Op::Upsert);
+        assert_eq!(got[0].op.key, "forge:hyperhive/hyperhive#7:assignment");
+        assert_eq!(
+            got[0].op.summary,
+            "you are assigned to issue hyperhive/hyperhive#7: router (https://forge/7)"
+        );
+        assert_eq!(got[0].op.source, "router:assigned");
+        // The match gates: an unassign event reaches no assigned sink.
+        assert!(
+            flow("assigned")
+                .evaluate("issue_assign", &event("unassigned", &["atlas"], &[], &[]))
+                .is_empty()
+        );
+        // The source gates: another event type starts nothing.
+        assert!(
+            flow("assigned")
+                .evaluate("issues", &event("assigned", &["atlas"], &[], &[]))
+                .is_empty()
+        );
+    }
+
+    #[test]
+    fn unassigning_rewrites_the_same_key_for_the_removed_agent() {
+        let got = flow("unassigned").evaluate(
+            "pull_request_assign",
+            &event("unassigned", &[], &["argus"], &[]),
+        );
+        assert_eq!(got.len(), 1);
+        assert_eq!(got[0].agent, "argus");
+        assert_eq!(got[0].op.key, "forge:hyperhive/hyperhive#7:assignment");
+        assert!(
+            got[0]
+                .op
+                .summary
+                .starts_with("you are not assigned to issue")
+        );
+    }
+
+    #[test]
+    fn closing_clears_by_prefix_for_every_holder() {
+        let got =
+            flow("closed").evaluate("issues", &event("closed", &[], &[], &["atlas", "argus"]));
+        assert_eq!(got.len(), 2);
+        for a in &got {
+            assert_eq!(a.op.op, Op::Clear);
+            assert!(a.op.prefix);
+            assert_eq!(a.op.key, "forge:hyperhive/hyperhive#7:");
+            assert_eq!(a.op.summary, "");
+        }
+        assert!(
+            flow("closed")
+                .evaluate("issues", &event("reopened", &[], &[], &["atlas"]))
+                .is_empty()
+        );
+    }
+
+    /// A match node with two children runs both: the branching the tree
+    /// allows.
+    #[test]
+    fn a_match_with_two_children_runs_both() {
+        let mut f = flow("assigned");
+        let NodeKind::Todo(sink) = f.nodes[2].kind.clone() else {
+            panic!("the third node is the sink")
+        };
+        f.nodes.push(Node {
+            id: "second".to_owned(),
+            kind: NodeKind::Todo(TodoSink {
+                to: "assignees_removed".to_owned(),
+                ..sink
+            }),
+        });
+        f.edges.push(Edge {
+            from: "match".to_owned(),
+            to: "second".to_owned(),
+        });
+        assert_eq!(f.validate(), Ok(()));
+        let got = f.evaluate(
+            "issue_assign",
+            &event("assigned", &["atlas"], &["argus"], &[]),
+        );
+        let mut nodes: Vec<(&str, &str)> = got
+            .iter()
+            .map(|a| (a.node.as_str(), a.agent.as_str()))
+            .collect();
+        nodes.sort_unstable();
+        assert_eq!(nodes, [("second", "argus"), ("sink", "atlas")]);
+    }
+
+    #[test]
+    fn a_flow_that_is_not_a_tree_is_refused() {
+        let edge = |from: &str, to: &str| Edge {
+            from: from.to_owned(),
+            to: to.to_owned(),
+        };
+        let mut cycle = flow("assigned");
+        cycle.edges[0] = edge("sink", "match");
+        cycle.edges.push(edge("match", "sink"));
+        assert!(cycle.validate().is_err(), "a cycle");
+
+        let mut orphan = flow("assigned");
+        orphan.edges.pop();
+        assert!(orphan.validate().is_err(), "an unreachable sink");
+
+        let mut two_parents = flow("assigned");
+        two_parents.edges.push(edge("source", "sink"));
+        assert!(two_parents.validate().is_err(), "a sink with two parents");
+
+        let mut sink_child = flow("assigned");
+        sink_child.nodes.push(Node {
+            id: "m2".to_owned(),
+            kind: NodeKind::Match { when: Vec::new() },
+        });
+        sink_child.edges.push(edge("sink", "m2"));
+        assert!(sink_child.validate().is_err(), "a sink with a child");
+
+        let mut dangling = flow("assigned");
+        dangling.edges.push(edge("match", "nowhere"));
+        assert!(dangling.validate().is_err(), "an edge to no node");
+
+        let mut two_sources = flow("assigned");
+        two_sources.nodes[1].kind = NodeKind::Source { events: Vec::new() };
+        assert!(two_sources.validate().is_err(), "two sources");
+    }
+
+    #[test]
+    fn a_placeholder_expands_to_its_field_and_an_unknown_one_stays() {
+        let fields = event("assigned", &["a", "b"], &[], &[]);
+        assert_eq!(
+            expand("{repo}#{number} {assignees_added} {nope} {", &fields),
+            "hyperhive/hyperhive#7 a, b {nope} {"
+        );
+    }
+}

swarm-notify-router/src/main.rs

@@ -0,0 +1,295 @@
+//! The swarm notification router.
+//!
+//! Drains its durable consumer on the forge events stream
+//! (`swarm_wire::forge_events`), runs every flow over each delivery
+//! ([`flow`], [`route`]), and queues the todo ops the flows emit on each
+//! recipient agent's subject of `ROUTER_TODOS` (`swarm_wire::router_todo`).
+//! Each agent's harness applies them to its own todo list.
+//!
+//! Per issue it keeps a snapshot in its state bucket ([`event::Snapshot`]):
+//! the assignees it last saw and the agents it queued a todo for. An event is
+//! acked once its ops are queued and the snapshot is written, so a crash
+//! between the two redelivers it, and the stream drops the republished ops
+//! by their message id.
+//!
+//! It connects to the queue as its own OIDC client, whose secret it reads
+//! from the store under its own certificate, as `swarm-claude-login` does.
+//! It holds no forge token and no webhook secret.
+
+use std::collections::BTreeSet;
+use std::path::PathBuf;
+use std::time::{Duration, Instant};
+
+use anyhow::{Context as _, Result, bail};
+use async_nats::jetstream::{self, consumer::pull, kv, message::PublishMessage};
+use futures_util::{StreamExt as _, TryStreamExt as _};
+use swarm_queue_client::{ClientSecret, QueueConfig};
+use swarm_secret_client::stored_secret::StoredSecret;
+use swarm_wire::{forge_events, notify_router, router_todo};
+
+mod event;
+mod flow;
+mod route;
+mod settle;
+
+/// The cert-auth role this service logs in to the store as. `swarm-bao.nix`
+/// writes it, matching the leaf's subject.
+const CERT_ROLE: &str = "swarm-notify-router";
+
+/// The environment prefix `swarm-notify-router.nix` sets.
+const PREFIX: &str = "SWARM_NOTIFY_ROUTER";
+
+/// How many events one pull asks for, and how long it waits for them.
+const BATCH: usize = 16;
+const PULL_WAIT: Duration = Duration::from_secs(30);
+
+/// How long an event may stay unacked before the stream redelivers it.
+const EVENT_ACK_WAIT: Duration = Duration::from_mins(1);
+
+/// How long the router waits after a failed pull or event before it pulls
+/// again, when the failed event comes back first.
+const RETRY_DELAY: Duration = Duration::from_secs(30);
+
+/// How long the list of agents is reused before it is read again.
+const ROSTER_TTL: Duration = Duration::from_mins(1);
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    hive_log::init();
+    swarm_secret_client::install_crypto_provider();
+    let var = |name: &str| {
+        let full = format!("{PREFIX}_{name}");
+        std::env::var(&full).with_context(|| format!("{full} is unset"))
+    };
+    let client_id = var("OIDC_CLIENT_ID")?;
+    let secret =
+        swarm_secret_client::read_with_startup_retry("reading the queue client secret", || {
+            client_secret(&client_id)
+        })
+        .await?;
+    let reread_id = client_id.clone();
+    let secret = StoredSecret::new(client_id.clone(), secret, move || {
+        let client_id = reread_id.clone();
+        async move { client_secret(&client_id).await.map_err(Into::into) }
+    });
+    let queue = swarm_queue_client::connect(QueueConfig {
+        url: var("NATS_URL")?,
+        token_endpoint: var("OIDC_TOKEN_ENDPOINT")?,
+        client_id,
+        client_secret: ClientSecret::Stored(secret),
+        ca_file: std::env::var(format!("{PREFIX}_OIDC_CA_FILE"))
+            .ok()
+            .map(PathBuf::from),
+    })
+    .await?;
+
+    let flows: Vec<flow::Flow> = flow::builtins()
+        .into_iter()
+        .filter(|f| match f.validate() {
+            Ok(()) => true,
+            Err(reason) => {
+                tracing::error!(flow = %f.id, %reason, "notify router: skipping a flow that is not a tree");
+                false
+            }
+        })
+        .collect();
+    let mut router = Router::open(&queue, flows).await?;
+    tracing::info!(
+        stream = forge_events::STREAM,
+        "notify router: routing forge events"
+    );
+    loop {
+        if let Err(e) = router.pull_once().await {
+            tracing::warn!(
+                error = format!("{e:#}"),
+                "notify router: pulling events failed; retrying"
+            );
+            tokio::time::sleep(RETRY_DELAY).await;
+        }
+    }
+}
+
+struct Router {
+    js: jetstream::Context,
+    events: jetstream::consumer::Consumer<pull::Config>,
+    state: kv::Store,
+    flows: Vec<flow::Flow>,
+    roster: Option<(Instant, BTreeSet<String>)>,
+}
+
+impl Router {
+    async fn open(queue: &async_nats::Client, flows: Vec<flow::Flow>) -> Result<Self> {
+        let js = jetstream::new(queue.clone());
+        let events = swarm_queue_client::forge_events::open_or_create(queue)
+            .await
+            .context("opening the forge events stream")?;
+        swarm_queue_client::router_todo::open_or_create(queue)
+            .await
+            .context("opening the router todo stream")?;
+        let events = events
+            .create_consumer(pull::Config {
+                durable_name: Some(forge_events::ROUTER_CONSUMER.to_owned()),
+                description: Some("The notification router".to_owned()),
+                filter_subject: forge_events::subject_wildcard(),
+                ack_wait: EVENT_ACK_WAIT,
+                ..Default::default()
+            })
+            .await
+            .context("creating the router's consumer")?;
+        let state = match js.get_key_value(notify_router::STATE_BUCKET).await {
+            Ok(store) => store,
+            Err(e) => {
+                tracing::info!(reason = %e, "notify router: state bucket not available, creating it");
+                js.create_key_value(kv::Config {
+                    bucket: notify_router::STATE_BUCKET.to_owned(),
+                    description: "Per-issue assignees and todo holders".to_owned(),
+                    history: 1,
+                    ..Default::default()
+                })
+                .await
+                .context("creating the state bucket")?
+            }
+        };
+        Ok(Self {
+            js,
+            events,
+            state,
+            flows,
+            roster: None,
+        })
+    }
+
+    /// Pull one batch and handle each event in stream order.
+    async fn pull_once(&mut self) -> Result<()> {
+        let batch = self
+            .events
+            .batch()
+            .max_messages(BATCH)
+            .expires(PULL_WAIT)
+            .messages()
+            .await
+            .context("requesting a batch")?;
+        let batch = batch.map(|m| m.map_err(|e| anyhow::anyhow!("{e}")));
+        settle::batch(batch, async |m: &jetstream::Message| self.handle(m).await).await
+    }
+
+    async fn handle(&mut self, message: &jetstream::Message) -> Result<(), Skip> {
+        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 {
+            // A delivery that names no issue or PR has nothing to route.
+            return Ok(());
+        };
+        let seq = message
+            .info()
+            .map(|i| i.stream_sequence)
+            .map_err(|e| Skip::Unusable(format!("no stream sequence: {e}")))?;
+        let key = delivery.state_key();
+        let snapshot: event::Snapshot = match self.state.get(&key).await.map_err(retry)? {
+            Some(bytes) => serde_json::from_slice(&bytes).unwrap_or_else(|e| {
+                tracing::warn!(%key, error = %e, "notify router: unreadable snapshot; starting afresh");
+                event::Snapshot::default()
+            }),
+            None => event::Snapshot::default(),
+        };
+        let roster = self.roster().await.map_err(Skip::Retry)?;
+        let routed = route::route(&event_type, &delivery, &snapshot, &self.flows, |a| {
+            roster.contains(a)
+        });
+        for action in &routed.actions {
+            self.queue(seq, action).await.map_err(Skip::Retry)?;
+        }
+        if !routed.skipped.is_empty() {
+            tracing::info!(%event_type, repo = %delivery.repo, number = delivery.number, skipped = ?routed.skipped, "notify router: recipients that are not agents got nothing");
+        }
+        if routed.next != snapshot {
+            let bytes = serde_json::to_vec(&routed.next).expect("a snapshot encodes");
+            self.state.put(&key, bytes.into()).await.map_err(retry)?;
+        }
+        Ok(())
+    }
+
+    /// Queue one op on its agent's subject and wait for the stream's ack.
+    async fn queue(&self, seq: u64, action: &flow::Action) -> Result<()> {
+        let subject = router_todo::subject(&action.agent);
+        let body = serde_json::to_vec(&action.op).expect("an op encodes");
+        let id = format!("{seq}:{}:{}:{}", action.flow, action.node, action.agent);
+        self.js
+            .send_publish(
+                subject.clone(),
+                PublishMessage::build().payload(body.into()).message_id(id),
+            )
+            .await
+            .with_context(|| format!("publishing to {subject}"))?
+            .await
+            .with_context(|| format!("waiting for {subject}'s ack"))?;
+        tracing::info!(agent = %action.agent, flow = %action.flow, key = %action.op.key, op = ?action.op.op, "notify router: todo op queued");
+        Ok(())
+    }
+
+    /// The agents of this swarm: the agent names in the agent-status bucket's
+    /// keys (`<hive>.<agent>`), read from the stream's subject list.
+    async fn roster(&mut self) -> Result<BTreeSet<String>> {
+        if let Some((at, agents)) = &self.roster
+            && at.elapsed() < ROSTER_TTL
+        {
+            return Ok(agents.clone());
+        }
+        let bucket = swarm_wire::agent_status::BUCKET;
+        let stream = self
+            .js
+            .get_stream(format!("KV_{bucket}"))
+            .await
+            .context("opening the agent-status bucket")?;
+        let subjects: Vec<(String, usize)> = stream
+            .info_with_subjects(format!("$KV.{bucket}.>"))
+            .await
+            .context("listing agent-status keys")?
+            .try_collect()
+            .await
+            .context("listing agent-status keys")?;
+        let prefix = format!("$KV.{bucket}.");
+        let agents: BTreeSet<String> = subjects
+            .into_iter()
+            .filter_map(|(subject, _)| {
+                let key = subject.strip_prefix(&prefix)?;
+                swarm_wire::agent_status::split_key(key).map(|(_, agent)| agent.to_owned())
+            })
+            .collect();
+        self.roster = Some((Instant::now(), agents.clone()));
+        Ok(agents)
+    }
+}
+
+/// Why an event was not handled.
+enum Skip {
+    /// It can never be handled; dropped for good.
+    Unusable(String),
+    /// The queue or the store failed; tried again.
+    Retry(anyhow::Error),
+}
+
+fn retry(e: impl std::error::Error + Send + Sync + 'static) -> Skip {
+    Skip::Retry(e.into())
+}
+
+async fn client_secret(client_id: &str) -> Result<String> {
+    use swarm_secret_client::builder::{ClientSecret, client_secret_path};
+
+    let path = client_secret_path(client_id)?;
+    let store = swarm_secret_client::SecretStore::from_env(CERT_ROLE)
+        .await
+        .context("logging in to the secret store")?;
+    let stored: Option<ClientSecret> = store
+        .read_optional(&path)
+        .await
+        .with_context(|| format!("reading {path}"))?;
+    match stored {
+        Some(s) if !s.value.trim().is_empty() => Ok(s.value.trim().to_owned()),
+        _ => bail!(
+            "nothing at {path}: swarm-secret-publish, in authelia's container, has not \
+             published this service's client secret"
+        ),
+    }
+}

swarm-notify-router/src/route.rs

@@ -0,0 +1,194 @@
+//! One event through every flow: the ops to queue and the issue's next
+//! snapshot. No I/O, so the whole decision is testable.
+
+use std::collections::BTreeSet;
+
+use swarm_wire::router_todo::Op;
+
+use crate::event::{Delivery, Snapshot, fields, is_assign};
+use crate::flow::{Action, Flow};
+
+/// What one event comes to.
+#[derive(Debug, PartialEq, Eq)]
+pub struct Routed {
+    /// The ops to queue, one per agent that exists.
+    pub actions: Vec<Action>,
+    /// Recipients that are not agents of this swarm, so got nothing.
+    pub skipped: Vec<String>,
+    pub next: Snapshot,
+}
+
+/// Run `flows` over one delivery of `event_type` on an issue whose snapshot is
+/// `snapshot`. `is_agent` says whether a recipient is an agent of this swarm.
+pub fn route(
+    event_type: &str,
+    delivery: &Delivery,
+    snapshot: &Snapshot,
+    flows: &[Flow],
+    is_agent: impl Fn(&str) -> bool,
+) -> Routed {
+    let fields = fields(event_type, delivery, snapshot);
+    let mut actions = Vec::new();
+    let mut skipped = Vec::new();
+    for action in flows.iter().flat_map(|f| f.evaluate(event_type, &fields)) {
+        if is_agent(&action.agent) {
+            actions.push(action);
+        } else if !skipped.contains(&action.agent) {
+            skipped.push(action.agent);
+        }
+    }
+    let mut next = snapshot.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
+        // the roster gets its todo once it is.
+        let mut assignees = delivery.assignees.clone();
+        assignees.retain(|a| !skipped.contains(a));
+        next.assignees = Some(assignees);
+    }
+    let mut cleared = BTreeSet::new();
+    for action in &actions {
+        match action.op.op {
+            Op::Upsert => {
+                next.holders.insert(action.agent.clone());
+            }
+            Op::Clear => {
+                cleared.insert(action.agent.clone());
+            }
+        }
+    }
+    next.holders.retain(|h| !cleared.contains(h));
+    Routed {
+        actions,
+        skipped,
+        next,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::flow::builtins;
+
+    fn delivery(action: &str, assignees: &[&str]) -> Delivery {
+        Delivery {
+            action: action.to_owned(),
+            repo: "o/r".to_owned(),
+            repo_id: 1,
+            number: 9,
+            title: "t".to_owned(),
+            url: "u".to_owned(),
+            sender: "mara".to_owned(),
+            is_pull: false,
+            assignees: assignees.iter().map(|a| (*a).to_owned()).collect(),
+        }
+    }
+
+    fn agents(a: &Routed) -> Vec<(&str, Op)> {
+        a.actions
+            .iter()
+            .map(|x| (x.agent.as_str(), x.op.op))
+            .collect()
+    }
+
+    /// Assign, reassign, close: the three built-ins in the order an issue
+    /// meets them, carrying the snapshot from one event to the next.
+    #[test]
+    fn an_issue_assigned_reassigned_and_closed_queues_and_clears_its_todos() {
+        let flows = builtins();
+        let is_agent = |a: &str| a != "mara";
+
+        let one = route(
+            "issue_assign",
+            &delivery("assigned", &["atlas", "mara"]),
+            &Snapshot::default(),
+            &flows,
+            is_agent,
+        );
+        assert_eq!(agents(&one), [("atlas", Op::Upsert)]);
+        assert_eq!(one.skipped, ["mara"]);
+        assert_eq!(one.next.holders, BTreeSet::from(["atlas".to_owned()]));
+
+        let two = route(
+            "issue_assign",
+            &delivery("unassigned", &["mara"]),
+            &one.next,
+            &flows,
+            is_agent,
+        );
+        assert_eq!(agents(&two), [("atlas", Op::Upsert)]);
+        assert!(two.actions[0].op.summary.contains("not assigned"));
+
+        let three = route(
+            "issue_assign",
+            &delivery("assigned", &["mara", "iris"]),
+            &two.next,
+            &flows,
+            is_agent,
+        );
+        assert_eq!(agents(&three), [("iris", Op::Upsert)]);
+
+        let closed = route(
+            "issues",
+            &delivery("closed", &["mara", "iris"]),
+            &three.next,
+            &flows,
+            is_agent,
+        );
+        let mut got = agents(&closed);
+        got.sort_unstable_by_key(|(a, _)| *a);
+        assert_eq!(got, [("atlas", Op::Clear), ("iris", Op::Clear)]);
+        assert!(closed.next.holders.is_empty());
+        // A close leaves the assignee snapshot as the last assign set it.
+        assert_eq!(closed.next.assignees, three.next.assignees);
+    }
+
+    /// An agent assigned before it reports status gets nothing then, and its
+    /// todo on the next assign event, once it is in the roster.
+    #[test]
+    fn an_assignee_not_yet_in_the_roster_gets_its_todo_on_the_next_assign() {
+        let flows = builtins();
+        let early = route(
+            "issue_assign",
+            &delivery("assigned", &["nova"]),
+            &Snapshot::default(),
+            &flows,
+            |_| false,
+        );
+        assert!(early.actions.is_empty());
+        assert_eq!(early.skipped, ["nova"]);
+        assert_eq!(early.next.assignees, Some(BTreeSet::new()));
+
+        let later = route(
+            "issue_assign",
+            &delivery("assigned", &["nova", "atlas"]),
+            &early.next,
+            &flows,
+            |_| true,
+        );
+        let mut got = agents(&later);
+        got.sort_unstable_by_key(|(a, _)| *a);
+        assert_eq!(got, [("atlas", Op::Upsert), ("nova", Op::Upsert)]);
+        assert_eq!(
+            later.next.assignees,
+            Some(BTreeSet::from(["atlas".to_owned(), "nova".to_owned()]))
+        );
+    }
+
+    #[test]
+    fn an_event_no_flow_matches_changes_nothing() {
+        let before = Snapshot {
+            assignees: Some(BTreeSet::from(["atlas".to_owned()])),
+            holders: BTreeSet::from(["atlas".to_owned()]),
+        };
+        let r = route(
+            "issue_comment",
+            &delivery("created", &[]),
+            &before,
+            &builtins(),
+            |_| true,
+        );
+        assert!(r.actions.is_empty());
+        assert_eq!(r.next, before);
+    }
+}

swarm-notify-router/src/settle.rs

@@ -0,0 +1,155 @@
+//! Handling one pulled batch in stream order.
+//!
+//! An issue's snapshot diff depends on the order its events are handled in,
+//! so a failed event must not let a later one overtake it. After the first
+//! event that has to be retried, nothing more of the batch is handled: it and
+//! every later event are nakked for immediate redelivery, and the caller waits
+//! before it pulls again, which hands them back in stream order.
+
+use anyhow::Result;
+use async_nats::jetstream::AckKind;
+use futures_util::{Stream, StreamExt as _};
+
+use crate::Skip;
+
+/// One pulled event: what it is called in the log, and how it is acked.
+pub(crate) trait Pulled {
+    fn subject(&self) -> &str;
+    async fn settle(&self, ack: AckKind) -> Result<()>;
+}
+
+impl Pulled for async_nats::jetstream::Message {
+    fn subject(&self) -> &str {
+        &self.subject
+    }
+
+    async fn settle(&self, ack: AckKind) -> Result<()> {
+        self.ack_with(ack)
+            .await
+            .map_err(|e| anyhow::anyhow!("acking an event: {e}"))
+    }
+}
+
+/// Handle each event of `batch` with `handle` and settle it. Returns the
+/// error of the first event that has to be retried, once the rest of the
+/// batch is nakked behind it.
+pub(crate) async fn batch<P: Pulled>(
+    batch: impl Stream<Item = Result<P>>,
+    mut handle: impl AsyncFnMut(&P) -> Result<(), Skip>,
+) -> Result<()> {
+    let mut batch = std::pin::pin!(batch);
+    let mut failed: Option<anyhow::Error> = None;
+    while let Some(event) = batch.next().await {
+        let event = event?;
+        let ack = if failed.is_some() {
+            AckKind::Nak(None)
+        } else {
+            match handle(&event).await {
+                Ok(()) => AckKind::Ack,
+                Err(Skip::Unusable(reason)) => {
+                    tracing::warn!(subject = event.subject(), %reason, "notify router: dropping an unusable event");
+                    AckKind::Term
+                }
+                Err(Skip::Retry(e)) => {
+                    failed = Some(e);
+                    AckKind::Nak(None)
+                }
+            }
+        };
+        event.settle(ack).await?;
+    }
+    match failed {
+        Some(e) => Err(e.context(
+            "an event failed; it and the rest of its batch are redelivered in stream order",
+        )),
+        None => Ok(()),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::cell::RefCell;
+
+    use super::*;
+
+    struct Fake<'a> {
+        name: &'static str,
+        settled: &'a RefCell<Vec<(&'static str, String)>>,
+    }
+
+    impl Pulled for Fake<'_> {
+        fn subject(&self) -> &str {
+            self.name
+        }
+
+        async fn settle(&self, ack: AckKind) -> Result<()> {
+            self.settled
+                .borrow_mut()
+                .push((self.name, format!("{ack:?}")));
+            Ok(())
+        }
+    }
+
+    /// `assigned` fails, so `unassigned` after it is not handled: both come
+    /// back, in order, rather than the unassign overtaking the assign.
+    #[tokio::test]
+    async fn after_a_retry_the_rest_of_the_batch_is_not_handled_and_comes_back() {
+        let settled = RefCell::new(Vec::new());
+        let events = ["opened", "assigned", "unassigned", "closed"].map(|name| {
+            Ok(Fake {
+                name,
+                settled: &settled,
+            })
+        });
+        let mut handled = Vec::new();
+        let result = batch(futures_util::stream::iter(events), async |e: &Fake<'_>| {
+            handled.push(e.name);
+            if e.name == "assigned" {
+                Err(Skip::Retry(anyhow::anyhow!("the queue is down")))
+            } else {
+                Ok(())
+            }
+        })
+        .await;
+        assert!(result.is_err());
+        assert_eq!(handled, ["opened", "assigned"]);
+        let nak = format!("{:?}", AckKind::Nak(None));
+        assert_eq!(
+            settled.into_inner(),
+            [
+                ("opened", format!("{:?}", AckKind::Ack)),
+                ("assigned", nak.clone()),
+                ("unassigned", nak.clone()),
+                ("closed", nak),
+            ]
+        );
+    }
+
+    /// Control: an unusable event is dropped and the batch goes on.
+    #[tokio::test]
+    async fn an_unusable_event_is_dropped_and_the_batch_goes_on() {
+        let settled = RefCell::new(Vec::new());
+        let events = ["bad", "good"].map(|name| {
+            Ok(Fake {
+                name,
+                settled: &settled,
+            })
+        });
+        let result = batch(futures_util::stream::iter(events), async |e: &Fake<'_>| {
+            if e.name == "bad" {
+                Err(Skip::Unusable("no".to_owned()))
+            } else {
+                Ok(())
+            }
+        })
+        .await;
+        assert!(result.is_ok());
+        assert_eq!(
+            settled.into_inner(),
+            [
+                ("bad", format!("{:?}", AckKind::Term)),
+                ("good", format!("{:?}", AckKind::Ack)),
+            ]
+        );
+    }
+}