1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
//! The route map as a directed graph.
//!
//! The graph answers two questions the Tower asks constantly: whether an edge is permitted, and
//! which agents could still be reached from a given set of live runs. The second is what allows
//! an unreachable rendezvous barrier to be abandoned rather than parked forever.
use std::collections::{BTreeMap, BTreeSet, VecDeque};
use crate::agent::AgentName;
use crate::config::Config;
use crate::route::Join;
/// The rendezvous condition attached to a receiving agent.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct JoinSpec {
/// Agents whose flights are parked until the condition is met.
pub upstreams: BTreeSet<AgentName>,
/// The release condition.
pub join: Join,
/// Backstop for a barrier that never becomes unreachable but also never completes.
pub timeout_sec: Option<u64>,
}
/// The route map, indexed for traversal.
#[derive(Debug, Clone, Default)]
pub struct RouteGraph {
edges: BTreeMap<AgentName, BTreeSet<AgentName>>,
/// Edges that open a new itinerary rather than continuing the current one.
///
/// Kept apart from `edges` because a spawn is a permission like any other but a *distance* of
/// a different kind: the receiver starts a fresh chain with fresh Hops, so counting a spawn as
/// one more step of the same chain measures something that does not exist.
spawns: BTreeMap<AgentName, BTreeSet<AgentName>>,
joins: BTreeMap<AgentName, JoinSpec>,
}
impl RouteGraph {
/// Builds a graph from a factory definition.
///
/// Routes that name several senders and several receivers expand to the full cross product,
/// which is what makes `from = ["a", "b"]` with `to = "c"` a rendezvous and `from = "a"` with
/// `to = ["b", "c"]` a fan-out.
#[must_use]
pub fn from_config(config: &Config) -> Self {
let mut graph = Self::default();
for route in &config.routes {
for from in &route.from {
for to in &route.to {
graph
.edges
.entry(from.clone())
.or_default()
.insert(to.clone());
if route.is_spawn() {
graph
.spawns
.entry(from.clone())
.or_default()
.insert(to.clone());
}
}
}
if let Some(join) = route.join {
for to in &route.to {
graph.joins.insert(
to.clone(),
JoinSpec {
upstreams: route.from.iter().cloned().collect(),
join,
timeout_sec: route.timeout_sec,
},
);
}
}
}
graph
}
/// Returns `true` when this edge opens a new itinerary rather than continuing one.
#[must_use]
pub fn is_spawn(&self, from: &AgentName, to: &AgentName) -> bool {
self.spawns.get(from).is_some_and(|tos| tos.contains(to))
}
/// Returns `true` if `from` is permitted to send to `to`.
#[must_use]
pub fn permits(&self, from: &AgentName, to: &AgentName) -> bool {
self.edges.get(from).is_some_and(|tos| tos.contains(to))
}
/// Returns the agents `from` may send to.
pub fn successors(&self, from: &AgentName) -> impl Iterator<Item = &AgentName> {
self.edges.get(from).into_iter().flatten()
}
/// Every agent reached by a spawn edge.
///
/// These begin chains of their own, so for any question about hop depth they are entry points
/// rather than destinations.
pub fn spawn_targets(&self) -> impl Iterator<Item = &AgentName> {
self.spawns.values().flatten()
}
/// Every agent belonging to the workflow that starts at `entry`.
///
/// Unlike [`RouteGraph::reachable_from`] this **does** cross spawn edges. The two questions
/// are different: reachability asks what could still deliver into *this* itinerary, and a
/// spawned chain never can. Workflow membership asks what this way in sets in motion, and a
/// reviewer spawned by a sweep is unarguably part of the sweep.
///
/// Agents shared between workflows appear in both, which is the honest answer — the
/// developer really is in the triage pipeline and the follow-up pipeline.
#[must_use]
pub fn workflow_from(&self, entry: &AgentName) -> BTreeSet<AgentName> {
let mut seen: BTreeSet<AgentName> = BTreeSet::new();
let mut queue = VecDeque::from([entry.clone()]);
seen.insert(entry.clone());
while let Some(current) = queue.pop_front() {
for next in self.successors(¤t) {
if seen.insert(next.clone()) {
queue.push_back(next.clone());
}
}
}
seen
}
/// Every spawn edge, as a sender/receiver pair.
pub fn spawn_edges(&self) -> impl Iterator<Item = (&AgentName, &AgentName)> {
self.spawns
.iter()
.flat_map(|(from, tos)| tos.iter().map(move |to| (from, to)))
}
/// Returns the rendezvous condition guarding `agent`, if any.
#[must_use]
pub fn join_for(&self, agent: &AgentName) -> Option<&JoinSpec> {
self.joins.get(agent)
}
/// Returns every agent reachable from `sources`, including the sources themselves.
///
/// Hops are deliberately ignored, which makes the answer conservative: an agent that Hops
/// would actually prevent from being reached is still reported as reachable. A barrier is
/// therefore never abandoned prematurely, and `timeout_sec` remains the backstop.
pub fn reachable_from<'a>(
&self,
sources: impl IntoIterator<Item = &'a AgentName>,
) -> BTreeSet<AgentName> {
self.distances_from(sources).into_keys().collect()
}
/// Returns every agent reachable from `sources`, each with its distance in edges.
///
/// The sources sit at distance zero. An agent at distance `d` is therefore woken by flight
/// `d + 1` of a chain that began at a source, which is what makes the figure directly
/// comparable against `max_hops` — see [`mod@crate::validate`].
///
/// Shortest paths are an optimistic bound. A factory whose agents loop will spend far more
/// hops than the distance suggests, so this proves an agent *can* be reached, never that a
/// particular itinerary will get there.
pub fn distances_from<'a>(
&self,
sources: impl IntoIterator<Item = &'a AgentName>,
) -> BTreeMap<AgentName, u32> {
let mut seen: BTreeMap<AgentName, u32> = BTreeMap::new();
let mut queue: VecDeque<AgentName> = VecDeque::new();
for source in sources {
if !seen.contains_key(source) {
seen.insert(source.clone(), 0);
queue.push_back(source.clone());
}
}
while let Some(current) = queue.pop_front() {
let depth = seen[¤t];
for next in self.successors(¤t) {
// A spawn edge is a permission, not a step. The receiver starts a fresh itinerary
// with fresh Hops, so counting it as one more hop of this chain measures a
// distance that does not exist — and for barrier abandonment it is worse than
// wrong: a spawned agent runs under a different itinerary and can never deliver
// to this one's barrier, so treating it as reachable keeps a dead barrier parked.
if self.is_spawn(¤t, next) {
continue;
}
if !seen.contains_key(next) {
seen.insert(next.clone(), depth + 1);
queue.push_back(next.clone());
}
}
}
seen
}
}
#[cfg(test)]
mod tests {
use super::*;
fn graph_from(toml: &str) -> RouteGraph {
let config = Config::from_toml(toml, "test.toml").expect("config parses");
RouteGraph::from_config(&config)
}
#[test]
fn edges_are_directed() {
let graph = graph_from(
r#"
[[routes]]
from = "planner"
to = "coder"
"#,
);
assert!(graph.permits(&"planner".into(), &"coder".into()));
assert!(!graph.permits(&"coder".into(), &"planner".into()));
}
#[test]
fn fan_out_expands_to_one_edge_per_target() {
let graph = graph_from(
r#"
[[routes]]
from = "planner"
to = ["probe_a", "probe_b"]
"#,
);
assert!(graph.permits(&"planner".into(), &"probe_a".into()));
assert!(graph.permits(&"planner".into(), &"probe_b".into()));
assert!(graph.join_for(&"probe_a".into()).is_none());
}
#[test]
fn join_route_records_every_upstream() {
let graph = graph_from(
r#"
[[routes]]
from = ["probe_a", "probe_b"]
to = "collector"
join = "all"
timeout_sec = 60
"#,
);
let spec = graph.join_for(&"collector".into()).expect("join recorded");
assert_eq!(spec.join, Join::All);
assert_eq!(spec.timeout_sec, Some(60));
assert!(spec.upstreams.contains(&"probe_a".into()));
assert!(spec.upstreams.contains(&"probe_b".into()));
}
#[test]
fn reachability_follows_edges_transitively() {
let graph = graph_from(
r#"
[[routes]]
from = "planner"
to = "probe_a"
[[routes]]
from = "probe_a"
to = "collector"
[[routes]]
from = "orphan"
to = "elsewhere"
"#,
);
let reachable = graph.reachable_from([&AgentName::from("planner")]);
assert!(reachable.contains(&"planner".into()));
assert!(reachable.contains(&"collector".into()));
assert!(!reachable.contains(&"orphan".into()));
}
#[test]
fn reachability_terminates_on_cycles() {
let graph = graph_from(
r#"
[[routes]]
from = "a"
to = "b"
[[routes]]
from = "b"
to = "a"
"#,
);
let reachable = graph.reachable_from([&AgentName::from("a")]);
assert_eq!(reachable.len(), 2);
}
#[test]
fn distance_counts_edges_from_the_source() {
let graph = graph_from(
r#"
[[routes]]
from = "planner"
to = "probe_a"
[[routes]]
from = "probe_a"
to = "collector"
"#,
);
let distances = graph.distances_from([&AgentName::from("planner")]);
assert_eq!(distances[&AgentName::from("planner")], 0);
assert_eq!(distances[&AgentName::from("probe_a")], 1);
assert_eq!(distances[&AgentName::from("collector")], 2);
}
#[test]
fn distance_is_the_shortest_path_not_the_longest() {
// `collector` is two edges away the long way round and one edge away directly. Hops are
// spent along whichever path an agent actually chooses, so the short answer is a bound
// and not a prediction.
let graph = graph_from(
r#"
[[routes]]
from = "planner"
to = ["probe_a", "collector"]
[[routes]]
from = "probe_a"
to = "collector"
"#,
);
let distances = graph.distances_from([&AgentName::from("planner")]);
assert_eq!(distances[&AgentName::from("collector")], 1);
}
#[test]
fn distance_omits_agents_no_edge_leads_to() {
let graph = graph_from(
r#"
[[routes]]
from = "planner"
to = "probe_a"
[[routes]]
from = "orphan"
to = "elsewhere"
"#,
);
let distances = graph.distances_from([&AgentName::from("planner")]);
assert!(!distances.contains_key(&AgentName::from("orphan")));
}
}