af_workflow/
executor.rs

1//! Compile a spec branch into a runnable step chain and drive events through it.
2//!
3//! Port of the execution half of `platform/compiler.py`. The spec DAG is
4//! linear day-1 (per R9): one ingress head, then a topologically-ordered chain
5//! of step nodes. The ingress node is the event *source* (owned by the runner,
6//! next phase); compilation separates it from the steps it feeds.
7
8use std::collections::{HashMap, HashSet, VecDeque};
9
10use serde_json::{json, Value};
11
12use crate::event::Event;
13use crate::node::{StepNode, WorkflowContext};
14use crate::recorder::{RunStatus, StepStatus};
15use crate::registry::{NodeError, NodeRegistry};
16use crate::result::StepResult;
17use crate::spec::Branch;
18
19#[derive(Debug, thiserror::Error)]
20pub enum CompileError {
21    #[error(transparent)]
22    Node(#[from] NodeError),
23    #[error("branch '{branch_id}' has no ingress node (chain needs a source)")]
24    NoIngress { branch_id: String },
25    #[error("branch '{branch_id}' has {count} ingress nodes; day-1 supports one")]
26    MultipleIngress { branch_id: String, count: usize },
27    #[error("branch '{branch_id}' edges form a cycle (DAG required)")]
28    Cycle { branch_id: String },
29}
30
31/// A compiled, runnable branch: an ingress source + an ordered step chain.
32pub(crate) struct CompiledBranch {
33    steps: Vec<CompiledStep>,
34}
35
36struct CompiledStep {
37    node_id: String,
38    node_type: String,
39    node: Box<dyn StepNode>,
40}
41
42fn is_material(node_type: &str) -> bool {
43    node_type.starts_with("execute.") || node_type.starts_with("notify.")
44}
45
46fn event_detail(event: &Event) -> Value {
47    json!({
48        "event_id": event.id,
49        "payload": event.payload,
50        "metadata": event.metadata,
51    })
52}
53
54/// Where a run ended.
55#[derive(Debug, Clone, PartialEq, Eq)]
56pub enum Terminal {
57    /// All events fell out (every branch of the chain Dropped).
58    Dropped { node_id: String, reason: String },
59    /// The chain ran to the end with at least one surviving event.
60    Completed,
61}
62
63/// Summary of one event driven through a compiled branch.
64#[derive(Debug, Clone)]
65pub struct RunOutcome {
66    pub steps_run: usize,
67    pub terminal: Terminal,
68    /// Events still alive at the end (empty if everything dropped).
69    pub survivors: Vec<Event>,
70}
71
72impl CompiledBranch {
73    /// Compile a spec branch against a node registry.
74    pub(crate) fn compile(branch: &Branch, registry: &NodeRegistry) -> Result<Self, CompileError> {
75        let order = topo_order(branch)?;
76
77        // Split ingress head from step chain.
78        let ingress_nodes: Vec<&crate::spec::Node> = branch
79            .nodes
80            .iter()
81            .filter(|n| registry.is_ingress(&n.node_type))
82            .collect();
83        match ingress_nodes.len() {
84            0 => {
85                return Err(CompileError::NoIngress {
86                    branch_id: branch.branch_id.clone(),
87                })
88            }
89            1 => {}
90            n => {
91                return Err(CompileError::MultipleIngress {
92                    branch_id: branch.branch_id.clone(),
93                    count: n,
94                })
95            }
96        }
97        let by_id: HashMap<&str, &crate::spec::Node> =
98            branch.nodes.iter().map(|n| (n.id.as_str(), n)).collect();
99
100        let mut steps = Vec::new();
101        for node_id in order {
102            let node = by_id[node_id.as_str()];
103            if registry.is_ingress(&node.node_type) {
104                continue; // ingress is the source, not a step
105            }
106            let built = registry.build_step(&node.node_type, &node.config)?;
107            steps.push(CompiledStep {
108                node_id: node.id.clone(),
109                node_type: node.node_type.clone(),
110                node: built,
111            });
112        }
113
114        Ok(Self { steps })
115    }
116
117    /// Drive one event through the step chain.
118    ///
119    /// `Pass` forwards, `Drop` removes that event, `FanOut` multiplies it. The
120    /// run completes when the chain is exhausted or every event has dropped.
121    pub(crate) async fn run_event(&self, ctx: &WorkflowContext, event: Event) -> RunOutcome {
122        let trigger = Value::Object(event.payload.clone());
123        let mut run = ctx.recorder.start(&ctx.trigger_kind, trigger).await;
124        let mut current = vec![event];
125        let mut steps_run = 0;
126        let mut material_steps = 0u32;
127
128        for step in &self.steps {
129            let mut next = Vec::new();
130            let mut last_drop: Option<(String, Option<String>)> = None;
131            for ev in &current {
132                match step.node.process(ev, ctx).await {
133                    StepResult::Pass(event) => {
134                        if is_material(&step.node_type) {
135                            run.record_step(
136                                &step.node_id,
137                                &step.node_type,
138                                StepStatus::Ok,
139                                None,
140                                event_detail(&event),
141                            )
142                            .await;
143                            material_steps += 1;
144                        }
145                        next.push(event);
146                    }
147                    StepResult::Drop {
148                        reason,
149                        exit_reason,
150                    } => last_drop = Some((reason, exit_reason)),
151                    StepResult::FanOut(evs) => next.extend(evs),
152                }
153            }
154            steps_run += 1;
155            if next.is_empty() {
156                let (reason, exit_reason) = last_drop.unwrap_or_else(|| ("dropped".into(), None));
157
158                if step.node_type.starts_with("sink.") || material_steps > 0 {
159                    run.end(RunStatus::Ok, Some("natural")).await;
160                } else if let Some(code) = exit_reason {
161                    let step_status = if code.starts_with("invalid_") {
162                        StepStatus::Error
163                    } else {
164                        StepStatus::Skipped
165                    };
166                    run.record_step(
167                        &step.node_id,
168                        &step.node_type,
169                        step_status,
170                        Some(&code),
171                        json!({ "reason": reason }),
172                    )
173                    .await;
174                    run.end(
175                        if step_status == StepStatus::Error {
176                            RunStatus::Error
177                        } else {
178                            RunStatus::Skipped
179                        },
180                        Some(&code),
181                    )
182                    .await;
183                } else {
184                    run.mark_filtered(&step.node_id, &step.node_type, &reason)
185                        .await;
186                    run.end(RunStatus::Skipped, None).await;
187                }
188
189                return RunOutcome {
190                    steps_run,
191                    terminal: Terminal::Dropped {
192                        node_id: step.node_id.clone(),
193                        reason,
194                    },
195                    survivors: Vec::new(),
196                };
197            }
198            current = next;
199        }
200
201        run.end(
202            if material_steps > 0 {
203                RunStatus::Ok
204            } else {
205                RunStatus::Skipped
206            },
207            Some("natural"),
208        )
209        .await;
210        RunOutcome {
211            steps_run,
212            terminal: Terminal::Completed,
213            survivors: current,
214        }
215    }
216}
217
218/// Kahn topological sort over a branch's edges. Nodes with no edges keep spec
219/// order. Errors on a cycle.
220fn topo_order(branch: &Branch) -> Result<Vec<String>, CompileError> {
221    let ids: Vec<&str> = branch.nodes.iter().map(|n| n.id.as_str()).collect();
222    let id_set: HashSet<&str> = ids.iter().copied().collect();
223
224    let mut indegree: HashMap<&str, usize> = ids.iter().map(|id| (*id, 0)).collect();
225    let mut adj: HashMap<&str, Vec<&str>> = ids.iter().map(|id| (*id, Vec::new())).collect();
226
227    for edge in &branch.edges {
228        // Dangling edges are caught by Spec::validate_structure; ignore here.
229        if id_set.contains(edge.source.as_str()) && id_set.contains(edge.target.as_str()) {
230            adj.get_mut(edge.source.as_str())
231                .unwrap()
232                .push(&edge.target);
233            *indegree.get_mut(edge.target.as_str()).unwrap() += 1;
234        }
235    }
236
237    // Seed queue in spec order to keep deterministic output.
238    let mut queue: VecDeque<&str> = ids.iter().copied().filter(|id| indegree[id] == 0).collect();
239
240    let mut order = Vec::with_capacity(ids.len());
241    while let Some(id) = queue.pop_front() {
242        order.push(id.to_string());
243        for &next in &adj[id] {
244            let d = indegree.get_mut(next).unwrap();
245            *d -= 1;
246            if *d == 0 {
247                queue.push_back(next);
248            }
249        }
250    }
251
252    if order.len() != ids.len() {
253        return Err(CompileError::Cycle {
254            branch_id: branch.branch_id.clone(),
255        });
256    }
257    Ok(order)
258}
af_workflow/executor.rs

af_workflow/
executor.rs
