areev_run_core/
plan.rs

1//! The validated plan graph — load-time validation V1–V6's pure parts
2//! (governed-agents §6.1) over an OMS §8.4 Workflow grain, plus the derived
3//! structure the scheduler walks: indexed nodes and edges, adjacency in
4//! declaration order, and the Tarjan condensation V5 needs.
5//!
6//! Store-dependent validation stays driver-side: V3 (binding hashes resolve
7//! to Tool Definitions), V7 (resolution freeze into the manifest), V8 (price
8//! table coverage). Everything here is a pure function of the Workflow's
9//! content — a foreign OMS file gets exactly the same verdict on every host.
10
11use crate::cond::{self, Cond};
12use crate::error::{Result, RunError};
13use areev_core::types::Workflow;
14use std::collections::BTreeMap;
15
16/// One edge, indexed and with its condition pre-parsed.
17#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
18pub struct PlanEdge {
19    pub src: usize,
20    pub dst: usize,
21    /// The raw condition string (journaled in decision records) and its
22    /// parsed built-in form. Hosts installing their own evaluator get the
23    /// raw string; the built-in gets the parse.
24    pub cond_raw: Option<String>,
25    pub cond: Option<Cond>,
26    pub max_cycles: Option<u32>,
27}
28
29/// A validated, indexed plan. Immutable once built — the runtime never
30/// mutates a plan, only its own scheduler state.
31#[derive(Debug, Clone)]
32pub struct PlanGraph {
33    pub nodes: Vec<String>,
34    pub edges: Vec<PlanEdge>,
35    /// Per node: out-edge indices in **declaration order** — the canonical
36    /// evaluation order (§6.2).
37    pub out_edges: Vec<Vec<usize>>,
38    pub in_edges: Vec<Vec<usize>>,
39    /// Per node: SCC id (condensation vertex).
40    pub scc_of: Vec<usize>,
41    /// Per edge: true iff it is a DFS back-edge in the entry-rooted
42    /// traversal (its destination is an ancestor on the recursion path
43    /// when the edge is walked) — i.e. it closes a cycle. Removing exactly
44    /// the back-edges always leaves an acyclic graph (a standard DFS
45    /// edge-classification theorem), which is what lets the scheduler
46    /// treat a cycle's re-entry point as ANY node, not only the plan's
47    /// entry: a back-edge into node `i` cannot possibly have resolved
48    /// before `i`'s own first run, so `step.rs::refresh_readiness` excludes
49    /// it from `i`'s generation-0 AND-join (RUN issue #33).
50    pub cycle_edge: Vec<bool>,
51    /// Per node: re-attempt budget (`retries[n] = n` re-attempts, so up to
52    /// n+1 executions — §6.3's pinned reading of the field).
53    pub retries: Vec<u32>,
54    /// Per node: the raw binding string from the Workflow (a Tool Definition
55    /// hash for Bound nodes). Resolution to grains is the driver's V3/V7.
56    pub bindings: Vec<Option<String>>,
57}
58
59impl PlanGraph {
60    /// Build and validate. Every violation is an error, never a warning —
61    /// V1/V2 shape errors surface as `RUN-E019`, unbounded cycles as
62    /// `RUN-E002`, unreachable nodes as `RUN-E003`, bad conditions as
63    /// `RUN-E005`.
64    pub fn build(wf: &Workflow) -> Result<Self> {
65        // V1 — nodes: non-empty list, unique ids, sane bytes.
66        if wf.nodes.is_empty() {
67            return Err(RunError::InvalidPlan { why: "workflow has no nodes".into() });
68        }
69        let mut index: BTreeMap<&str, usize> = BTreeMap::new();
70        for (i, n) in wf.nodes.iter().enumerate() {
71            if n.trim().is_empty() || n.len() > 256 || n.chars().any(|c| c.is_control()) {
72                return Err(RunError::InvalidPlan {
73                    why: format!("node id {n:?} is empty, oversized, or has control chars"),
74                });
75            }
76            if index.insert(n.as_str(), i).is_some() {
77                return Err(RunError::InvalidPlan { why: format!("duplicate node id '{n}'") });
78            }
79        }
80
81        // V2 — edge referential integrity (re-validated here even though the
82        // local ADD path checks it: OMS files arrive from other
83        // implementations). V4 — conditions parse.
84        let mut edges = Vec::with_capacity(wf.edges.len());
85        for e in &wf.edges {
86            let (Some(&src), Some(&dst)) = (index.get(e.src.as_str()), index.get(e.dst.as_str()))
87            else {
88                return Err(RunError::InvalidPlan {
89                    why: format!("edge {} -> {} references an unknown node", e.src, e.dst),
90                });
91            };
92            let parsed = match &e.cond {
93                Some(raw) => Some(cond::parse(raw).map_err(|why| RunError::InvalidCondition {
94                    edge: format!("{} -> {}", e.src, e.dst),
95                    why,
96                })?),
97                None => None,
98            };
99            edges.push(PlanEdge {
100                src,
101                dst,
102                cond_raw: e.cond.clone(),
103                cond: parsed,
104                max_cycles: e.max_cycles,
105            });
106        }
107
108        // Shape checks on the auxiliary maps (the pure half of V3).
109        for key in wf.bindings.keys() {
110            if !index.contains_key(key.as_str()) {
111                return Err(RunError::InvalidPlan {
112                    why: format!("bindings names unknown node '{key}'"),
113                });
114            }
115        }
116        for key in wf.retries.keys() {
117            if !index.contains_key(key.as_str()) {
118                return Err(RunError::InvalidPlan {
119                    why: format!("retries names unknown node '{key}'"),
120                });
121            }
122        }
123
124        let n = wf.nodes.len();
125        let mut out_edges = vec![Vec::new(); n];
126        let mut in_edges = vec![Vec::new(); n];
127        for (i, e) in edges.iter().enumerate() {
128            out_edges[e.src].push(i);
129            in_edges[e.dst].push(i);
130        }
131
132        // V6 — reachability from the entry (nodes[0], OMS §8.4). Unreachable
133        // nodes are almost always authoring typos; erroring here catches
134        // them before the first journal write.
135        let mut reachable = vec![false; n];
136        let mut stack = vec![0usize];
137        reachable[0] = true;
138        while let Some(v) = stack.pop() {
139            for &ei in &out_edges[v] {
140                let d = edges[ei].dst;
141                if !reachable[d] {
142                    reachable[d] = true;
143                    stack.push(d);
144                }
145            }
146        }
147        let unreachable: Vec<String> = (0..n)
148            .filter(|&i| !reachable[i])
149            .map(|i| wf.nodes[i].clone())
150            .collect();
151        if !unreachable.is_empty() {
152            return Err(RunError::Unreachable { nodes: unreachable });
153        }
154
155        // V5 — every cycle carries a bound: for each SCC that contains a
156        // cycle (size > 1, or a self-loop), at least one intra-SCC edge must
157        // set max_cycles. Tarjan, iterative (plans are small, but a deep
158        // chain must not blow the stack).
159        let (scc_of, cycle_edge) = tarjan_scc(n, &out_edges, &edges);
160        let mut scc_nodes: BTreeMap<usize, Vec<usize>> = BTreeMap::new();
161        for (v, &c) in scc_of.iter().enumerate() {
162            scc_nodes.entry(c).or_default().push(v);
163        }
164        for (c, members) in &scc_nodes {
165            let cyclic = members.len() > 1
166                || members.iter().any(|&v| {
167                    out_edges[v].iter().any(|&ei| edges[ei].dst == v)
168                });
169            if !cyclic {
170                continue;
171            }
172            let bounded = edges.iter().any(|e| {
173                scc_of[e.src] == *c && scc_of[e.dst] == *c && e.max_cycles.is_some()
174            });
175            if !bounded {
176                return Err(RunError::UnboundedCycle {
177                    nodes: members.iter().map(|&v| wf.nodes[v].clone()).collect(),
178                });
179            }
180        }
181
182        let retries = wf
183            .nodes
184            .iter()
185            .map(|id| wf.retries.get(id).copied().unwrap_or(0))
186            .collect();
187        let bindings = wf.nodes.iter().map(|id| wf.bindings.get(id).cloned()).collect();
188
189        Ok(PlanGraph {
190            nodes: wf.nodes.clone(),
191            edges,
192            out_edges,
193            in_edges,
194            scc_of,
195            cycle_edge,
196            retries,
197            bindings,
198        })
199    }
200
201    /// Terminal nodes: no out-edges (OMS §8.4).
202    pub fn terminals(&self) -> Vec<usize> {
203        (0..self.nodes.len())
204            .filter(|&v| self.out_edges[v].is_empty())
205            .collect()
206    }
207}
208
209/// Iterative Tarjan SCC, plus the entry-rooted DFS back-edge classification
210/// (see `PlanGraph::cycle_edge`'s doc). Returns (SCC id per vertex, back-edge
211/// flag per edge); SCC ids are arbitrary but deterministic for a given
212/// graph.
213///
214/// The back-edge flag rides the same traversal via one extra array,
215/// `in_frame`: true while a vertex is on the *current DFS recursion path*
216/// (pushed when a frame opens, cleared when it closes) — a strictly
217/// narrower set than `on_stack`, which also holds vertices whose frame
218/// already closed but whose SCC hasn't (the cross-edge case, e.g. a
219/// diamond's second incoming arm into an SCC member). An edge whose target
220/// is `in_frame` is a genuine ancestor edge (a back-edge); an edge whose
221/// target is merely `on_stack` but not `in_frame` is a cross/forward edge
222/// and must stay a normal AND-join requirement.
223fn tarjan_scc(n: usize, out_edges: &[Vec<usize>], edges: &[PlanEdge]) -> (Vec<usize>, Vec<bool>) {
224    const UNSET: usize = usize::MAX;
225    let mut idx = vec![UNSET; n];
226    let mut low = vec![0usize; n];
227    let mut on_stack = vec![false; n];
228    let mut in_frame = vec![false; n];
229    let mut scc_of = vec![UNSET; n];
230    let mut back_edge = vec![false; edges.len()];
231    let mut stack: Vec<usize> = Vec::new();
232    let mut next_index = 0usize;
233    let mut next_scc = 0usize;
234
235    // Explicit DFS frames: (vertex, next out-edge position).
236    for root in 0..n {
237        if idx[root] != UNSET {
238            continue;
239        }
240        let mut frames: Vec<(usize, usize)> = vec![(root, 0)];
241        idx[root] = next_index;
242        low[root] = next_index;
243        next_index += 1;
244        stack.push(root);
245        on_stack[root] = true;
246        in_frame[root] = true;
247
248        while let Some(&mut (v, ref mut ei_pos)) = frames.last_mut() {
249            if *ei_pos < out_edges[v].len() {
250                let ei = out_edges[v][*ei_pos];
251                *ei_pos += 1;
252                let w = edges[ei].dst;
253                if idx[w] == UNSET {
254                    idx[w] = next_index;
255                    low[w] = next_index;
256                    next_index += 1;
257                    stack.push(w);
258                    on_stack[w] = true;
259                    in_frame[w] = true;
260                    frames.push((w, 0));
261                } else {
262                    if on_stack[w] {
263                        low[v] = low[v].min(idx[w]);
264                    }
265                    if in_frame[w] {
266                        back_edge[ei] = true;
267                    }
268                }
269            } else {
270                frames.pop();
271                in_frame[v] = false;
272                if let Some(&mut (parent, _)) = frames.last_mut() {
273                    low[parent] = low[parent].min(low[v]);
274                }
275                if low[v] == idx[v] {
276                    loop {
277                        let w = stack.pop().expect("tarjan stack invariant");
278                        on_stack[w] = false;
279                        scc_of[w] = next_scc;
280                        if w == v {
281                            break;
282                        }
283                    }
284                    next_scc += 1;
285                }
286            }
287        }
288    }
289    (scc_of, back_edge)
290}
291
292#[cfg(test)]
293mod tests {
294    use super::*;
295
296    fn wf(nodes: &[&str]) -> Workflow {
297        Workflow::new(nodes.iter().map(|s| s.to_string()).collect())
298    }
299
300    #[test]
301    fn a_diamond_validates() {
302        let w = wf(&["a", "b", "c", "d"])
303            .edge("a", "b")
304            .edge("a", "c")
305            .edge("b", "d")
306            .edge("c", "d");
307        let p = PlanGraph::build(&w).unwrap();
308        assert_eq!(p.terminals(), vec![3]);
309        assert_eq!(p.in_edges[3].len(), 2, "d is an AND-join");
310        // Four distinct SCCs — no cycles.
311        let distinct: std::collections::BTreeSet<_> = p.scc_of.iter().collect();
312        assert_eq!(distinct.len(), 4);
313    }
314
315    #[test]
316    fn bounded_cycles_pass_unbounded_are_refused() {
317        // Multi-node cycle with a bound on the back-edge: fine.
318        let mut e = wf(&["a", "b"]).edge("a", "b");
319        e.edges.push(areev_core::types::WorkflowEdge {
320            src: "b".into(),
321            dst: "a".into(),
322            cond: None,
323            max_cycles: Some(3),
324        });
325        assert!(PlanGraph::build(&e).is_ok());
326
327        // Same cycle, no bound anywhere: RUN-E002.
328        let mut u = wf(&["a", "b"]).edge("a", "b");
329        u.edges.push(areev_core::types::WorkflowEdge {
330            src: "b".into(),
331            dst: "a".into(),
332            cond: None,
333            max_cycles: None,
334        });
335        let err = PlanGraph::build(&u).unwrap_err();
336        assert_eq!(err.code(), "RUN-E002");
337
338        // A self-loop is a cycle too.
339        let mut s = wf(&["a", "z"]).edge("a", "z");
340        s.edges.push(areev_core::types::WorkflowEdge {
341            src: "a".into(),
342            dst: "a".into(),
343            cond: None,
344            max_cycles: None,
345        });
346        assert_eq!(PlanGraph::build(&s).unwrap_err().code(), "RUN-E002");
347    }
348
349    #[test]
350    fn unreachable_nodes_are_refused() {
351        let w = wf(&["a", "b", "orphan"]).edge("a", "b");
352        let err = PlanGraph::build(&w).unwrap_err();
353        assert_eq!(err.code(), "RUN-E003");
354        assert!(err.to_string().contains("orphan"));
355    }
356
357    #[test]
358    fn shape_violations_are_refused_with_reasons() {
359        // Duplicate node id.
360        assert!(PlanGraph::build(&wf(&["a", "a"])).is_err());
361        // Empty node list.
362        assert!(PlanGraph::build(&wf(&[])).is_err());
363        // Edge to an unknown node.
364        let w = wf(&["a"]).edge("a", "ghost");
365        assert!(PlanGraph::build(&w).unwrap_err().to_string().contains("ghost"));
366        // bindings / retries naming unknown nodes.
367        let b = wf(&["a"]).bind("ghost", "deadbeef");
368        assert!(PlanGraph::build(&b).unwrap_err().to_string().contains("ghost"));
369        let r = wf(&["a"]).retry("ghost", 2);
370        assert!(PlanGraph::build(&r).unwrap_err().to_string().contains("ghost"));
371    }
372
373    #[test]
374    fn bad_conditions_name_their_edge() {
375        let w = wf(&["a", "b"]).cond_edge("a", "b", "x === 1");
376        let err = PlanGraph::build(&w).unwrap_err();
377        assert_eq!(err.code(), "RUN-E005");
378        assert!(err.to_string().contains("a -> b"), "{err}");
379    }
380
381    #[test]
382    fn retries_and_bindings_index_by_node_position() {
383        let w = wf(&["a", "b"]).edge("a", "b").retry("b", 4).bind("a", "cafe");
384        let p = PlanGraph::build(&w).unwrap();
385        assert_eq!(p.retries, vec![0, 4]);
386        assert_eq!(p.bindings[0].as_deref(), Some("cafe"));
387        assert_eq!(p.bindings[1], None);
388    }
389}
areev_run_core/plan.rs

areev_run_core/
plan.rs
