vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

closed flow also queues a closed todo on each agent that held one, after the clear

Browse source
This commit is contained in:
atlas 2026-10-24 13:58:08 +02:00 • committed by mara
commit 0a1e4db0f4
4 changed files with 99 additions and 24 deletions
Download patch file Download diff file Expand all files Collapse all files

14
docs/swarm/notify-router.md
View file

@ -12,17 +12,21 @@ the way it did. An agent can get the same news from both.
## What an agent gets
| Flow | Event | The agent's todo |
| ------------ | ------------------------------------------ | ------------------------------------------------------------------------------------------- |
| `assigned` | someone assigns an agent to an issue or PR | `you are assigned to issue <repo>#<n>: <title> (<url>)`, key `forge:<repo>#<n>:assignment` |
| `unassigned` | someone takes an agent off an issue or PR | the same key rewritten: `you are not assigned to issue <repo>#<n> anymore: <title> (<url>)` |
| `closed` | someone closes or merges the issue or PR | the router clears every router todo it queued for it, on every agent that holds one |
| Flow | Event | The agent's todo |
| ------------ | ------------------------------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `assigned` | someone assigns an agent to an issue or PR | `you are assigned to issue <repo>#<n>: <title> (<url>)`, key `forge:<repo>#<n>:assignment` |
| `unassigned` | someone takes an agent off an issue or PR | the same key rewritten: `you are not assigned to issue <repo>#<n> anymore: <title> (<url>)` |
| `closed` | someone closes or merges the issue or PR | the router clears every router todo it queued for it, then queues `closed` on each agent that held one: `<kind> <repo>#<n> was closed: <title> (<url>)`, key `forge:<repo>#<n>:closed` |
The todos are subsystem `router` in the agent's todo list. A todo under the
same key replaces the one before, so an issue gives an agent one assignment
todo however often it changes hands. A new event reopens a todo the agent
marked done.
The `closed` todo has its own key, apart from the assignment todo. A reopened
issue leaves it in place until the next close replaces it; a new assignment
writes the `assignment` key and leaves it alone.
Only agents get todos. An assignee with no agent of that name gets nothing.
An agent assigned before it first reports its status gets its todo on the
issue's next assign event.

47
swarm-notify-router/src/flow.rs
View file

@ -4,8 +4,9 @@
//! `swarm_wire::notify_flows`; the built-in and the custom flows are the same
//! type and this walk runs both. 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 `todo` node emits one op per recipient. The
//! router runs only flows that pass [`Flow::validate`].
//! reach several branches, in edge order, so a clear listed before an upsert
//! is queued before it. A `todo` node emits one op per recipient. The router
//! runs only flows that pass [`Flow::validate`].
use std::collections::BTreeMap;
@ -46,10 +47,10 @@ pub fn evaluate(flow: &Flow, event_type: &str, fields: &Fields) -> Vec<Action> {
while let Some(id) = stack.pop() {
let Some(node) = flow.node(id) else { continue };
match &node.kind {
NodeKind::Source { .. } => stack.extend(flow.children(id)),
NodeKind::Source { .. } => stack.extend(flow.children(id).into_iter().rev()),
NodeKind::Match { when } => {
if when.iter().all(|c| holds(c, fields)) {
stack.extend(flow.children(id));
stack.extend(flow.children(id).into_iter().rev());
}
}
NodeKind::Todo(sink) => actions.extend(emit(flow, node, sink, fields)),
@ -229,6 +230,7 @@ mod tests {
"issues",
&event("closed", &[], &[], &["atlas", "argus"]),
);
let got: Vec<_> = got.into_iter().filter(|a| a.node == "sink").collect();
assert_eq!(got.len(), 2);
for a in &got {
assert_eq!(a.op.op, Op::Clear);
@ -239,6 +241,43 @@ mod tests {
assert!(evaluate(&closed, "issues", &event("reopened", &[], &[], &["atlas"])).is_empty());
}
/// The clear runs before the `closed` todo, so its prefix does not take the
/// todo with it, and both read the one holders set.
#[test]
fn closing_clears_then_tells_each_holder() {
let closed = flow("closed");
let got = evaluate(
&closed,
"pull_request",
&event("closed", &[], &[], &["atlas", "argus"]),
);
let seen: Vec<(&str, &str, Op)> = got
.iter()
.map(|a| (a.node.as_str(), a.agent.as_str(), a.op.op))
.collect();
assert_eq!(
seen,
[
("sink", "atlas", Op::Clear),
("sink", "argus", Op::Clear),
("inform", "atlas", Op::Upsert),
("inform", "argus", Op::Upsert),
]
);
let tell = &got[2].op;
assert_eq!(tell.key, "forge:hyperhive/hyperhive#7:closed");
assert!(!tell.prefix);
assert_eq!(
tell.summary,
"issue hyperhive/hyperhive#7 was closed: router (https://forge/7)"
);
assert_eq!(tell.source, "router:closed");
// The clear's prefix covers the todo's key: the order is what saves it.
assert!(tell.key.starts_with(&got[0].op.key));
// No holders, no todos.
assert!(evaluate(&closed, "issues", &event("closed", &[], &[], &[])).is_empty());
}
/// A match node with two children runs both: the branching the tree
/// allows.
#[test]

13
swarm-notify-router/src/route.rs
View file

@ -137,9 +137,18 @@ mod tests {
&flows,
is_agent,
);
// Per agent the clear comes before the closed todo that its prefix would take.
let mut got = agents(&closed);
got.sort_unstable_by_key(|(a, _)| *a);
assert_eq!(got, [("atlas", Op::Clear), ("iris", Op::Clear)]);
got.sort_by_key(|(a, _)| *a);
assert_eq!(
got,
[
("atlas", Op::Clear),
("atlas", Op::Upsert),
("iris", Op::Clear),
("iris", Op::Upsert)
]
);
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);

49
swarm-wire/src/notify_flows.rs
View file

@ -628,7 +628,10 @@ impl TodoSink {
/// - `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.
/// - `closed`: closing it clears every router todo the router queued for it,
/// then tells each agent that held one with a `closed` todo. The clear is the
/// first child of the match and runs first: its prefix also covers the
/// `closed` key.
#[must_use]
pub fn builtins() -> Vec<Flow> {
let assignment_key = "forge:{repo}#{number}:assignment";
@ -658,21 +661,41 @@ pub fn builtins() -> Vec<Flow> {
.to_owned(),
},
),
chain(
"closed",
&["issues", "pull_request"],
"closed",
TodoSink {
to: Recipients::Field("todo_holders".to_owned()),
op: Op::Clear,
key: "forge:{repo}#{number}:".to_owned(),
prefix: true,
text: String::new(),
},
),
closed(),
]
}
/// The `closed` flow: the clear, then the todo that says why.
fn closed() -> Flow {
let mut flow = chain(
"closed",
&["issues", "pull_request"],
"closed",
TodoSink {
to: Recipients::Field("todo_holders".to_owned()),
op: Op::Clear,
key: "forge:{repo}#{number}:".to_owned(),
prefix: true,
text: String::new(),
},
);
flow.nodes.push(Node {
id: "inform".to_owned(),
kind: NodeKind::Todo(TodoSink {
to: Recipients::Field("todo_holders".to_owned()),
op: Op::Upsert,
key: "forge:{repo}#{number}:closed".to_owned(),
prefix: false,
text: "{kind} {repo}#{number} was closed: {title} ({url})".to_owned(),
}),
});
flow.edges.push(Edge {
from: "match".to_owned(),
to: "inform".to_owned(),
});
flow
}
/// source(`events`) → match(`action` = `action`) → `sink`.
fn chain(id: &str, events: &[&str], action: &str, sink: TodoSink) -> Flow {
let node = |id: &str, kind| Node {