af_workflow/

executor.rs

//! Compile a spec branch into a runnable step chain and drive events through it.
//!
//! Port of the execution half of `platform/compiler.py`. The spec DAG is
//! linear day-1 (per R9): one ingress head, then a topologically-ordered chain
//! of step nodes. The ingress node is the event *source* (owned by the runner,
//! next phase); compilation separates it from the steps it feeds.

use std::collections::{HashMap, VecDeque};

use serde_json::{json, Value};

use crate::event::Event;
use crate::node::{StepNode, WorkflowContext};
use crate::recorder::{RunStatus, StepStatus};
use crate::registry::{NodeError, NodeRegistry};
use crate::result::{PreparedAction, StepResult};
use crate::spec::Branch;

/// Why a branch could not be compiled.
#[derive(Debug, thiserror::Error)]
pub enum CompileError {
    /// A node could not be built.
    #[error(transparent)]
    Node(#[from] NodeError),
    /// Branch '`branch_id`' has no ingress node (chain needs a source).
    #[error("branch '{branch_id}' has no ingress node (chain needs a source)")]
    /// The branch has no ingress node.
    NoIngress {
        /// Branch id.
        branch_id: String,
    },
    /// Branch '`branch_id`' has `count` ingress nodes; day-1 supports one.
    #[error("branch '{branch_id}' has {count} ingress nodes; day-1 supports one")]
    /// The branch has more than one ingress node.
    MultipleIngress {
        /// Branch id.
        branch_id: String,
        /// Ingress nodes found.
        count: usize,
    },
    /// Branch '`branch_id`' edges form a cycle (DAG required).
    #[error("branch '{branch_id}' edges form a cycle (DAG required)")]
    /// The branch is not a DAG.
    Cycle {
        /// Branch id.
        branch_id: String,
    },
}

/// A compiled, runnable branch: an ingress source + an ordered step chain.
pub(crate) struct CompiledBranch {
    steps: Vec<CompiledStep>,
}

struct CompiledStep {
    node_id: String,
    node_type: String,
    node: Box<dyn StepNode>,
    fan_out_limit: usize,
    action_capable: bool,
}

const MAX_RUN_FAN_OUT: usize = 1_000;

fn fan_out_limit(config: &Value) -> usize {
    ["count", "levels", "fanout"]
        .into_iter()
        .find_map(|key| config.get(key)?.as_u64())
        .and_then(|value| usize::try_from(value).ok())
        .unwrap_or(MAX_RUN_FAN_OUT)
        .min(MAX_RUN_FAN_OUT)
}

fn is_material(node_type: &str) -> bool {
    node_type.starts_with("execute.") || node_type.starts_with("notify.")
}

fn event_detail(event: &Event) -> Value {
    json!({
        "event_id": event.id,
        "payload": event.payload,
        "metadata": event.metadata,
    })
}

/// Where a run ended.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Terminal {
    /// All events fell out (every branch of the chain Dropped).
    Dropped {
        /// Node that dropped the last event.
        node_id: String,
        /// Why.
        reason: String,
    },
    /// The chain ran to the end with at least one surviving event.
    Completed,
}

/// Summary of one event driven through a compiled branch.
#[derive(Debug, Clone)]
pub struct RunOutcome {
    /// Steps the event passed through.
    pub steps_run: usize,
    /// Where the run ended.
    pub terminal: Terminal,
    /// Events still alive at the end (empty if everything dropped).
    pub survivors: Vec<Event>,
    /// External effects the steps prepared, in emission order.
    pub actions: Vec<PreparedAction>,
    /// At least one event passed a material step or reached the branch end.
    pub matched: bool,
    /// The branch reached a sink or retained an event through its final step.
    pub succeeded: bool,
}

impl CompiledBranch {
    /// Compile a spec branch against a node registry.
    pub(crate) fn compile(branch: &Branch, registry: &NodeRegistry) -> Result<Self, CompileError> {
        let order = topo_order(branch)?;

        // Split ingress head from step chain.
        let ingress_nodes: Vec<&crate::spec::Node> = branch
            .nodes
            .iter()
            .filter(|n| registry.is_ingress(&n.node_type))
            .collect();
        match ingress_nodes.len() {
            0 => {
                return Err(CompileError::NoIngress {
                    branch_id: branch.branch_id.clone(),
                })
            }
            1 => {}
            n => {
                return Err(CompileError::MultipleIngress {
                    branch_id: branch.branch_id.clone(),
                    count: n,
                })
            }
        }
        let by_id: HashMap<&str, &crate::spec::Node> =
            branch.nodes.iter().map(|n| (n.id.as_str(), n)).collect();

        let mut steps = Vec::new();
        for node_id in order {
            let node = by_id[node_id.as_str()];
            if registry.is_ingress(&node.node_type) {
                continue; // ingress is the source, not a step
            }
            let built = registry.build_step(&node.node_type, &node.config)?;
            steps.push(CompiledStep {
                node_id: node.id.clone(),
                node_type: node.node_type.clone(),
                node: built,
                fan_out_limit: fan_out_limit(&node.config),
                action_capable: node.node_type.starts_with("execute.")
                    || registry
                        .capability(&node.node_type)
                        .is_some_and(|manifest| manifest.kind == crate::CapabilityKind::Action),
            });
        }

        Ok(Self { steps })
    }

    /// Drive one event through the step chain.
    ///
    /// `Pass` forwards, `Drop` removes that event, `FanOut` multiplies it. The
    /// run completes when the chain is exhausted or every event has dropped.
    pub(crate) async fn run_event(&self, ctx: &WorkflowContext, event: Event) -> RunOutcome {
        self.run_event_with_continuation_from(ctx, event, 0).await.0
    }

    pub(crate) async fn run_event_from(
        &self,
        ctx: &WorkflowContext,
        event: Event,
        start_step: usize,
    ) -> RunOutcome {
        self.run_event_with_continuation_from(ctx, event, start_step)
            .await
            .0
    }

    pub(crate) async fn run_event_with_continuation_from(
        &self,
        ctx: &WorkflowContext,
        event: Event,
        start_step: usize,
    ) -> (RunOutcome, Option<usize>) {
        let trigger = Value::Object(event.payload.clone());
        let mut run = ctx.recorder.start(&ctx.trigger_kind, trigger).await;
        let mut current = vec![event];
        let mut steps_run = 0;
        let mut material_steps = 0u32;
        let mut actions = Vec::new();

        for (step_index, step) in self.steps.iter().enumerate().skip(start_step) {
            let mut next = Vec::new();
            let mut last_drop: Option<(String, Option<String>)> = None;
            let mut fan_out_error = None;
            for ev in &current {
                match step.node.process(ev, ctx).await {
                    StepResult::Pass(event) => {
                        if is_material(&step.node_type) {
                            run.record_step(
                                &step.node_id,
                                &step.node_type,
                                StepStatus::Ok,
                                None,
                                event_detail(&event),
                            )
                            .await;
                            material_steps += 1;
                        }
                        next.push(event);
                    }
                    StepResult::Drop {
                        reason,
                        exit_reason,
                    } => last_drop = Some((reason, exit_reason)),
                    StepResult::FanOut(evs) => {
                        if evs.len() > step.fan_out_limit
                            || next.len().saturating_add(evs.len()) > MAX_RUN_FAN_OUT
                        {
                            fan_out_error = Some(format!(
                                "fan-out exceeded node limit {} or run limit {MAX_RUN_FAN_OUT}",
                                step.fan_out_limit
                            ));
                            break;
                        }
                        next.extend(evs);
                    }
                    StepResult::Action { event, action } => {
                        if !step.action_capable {
                            fan_out_error = Some(format!(
                                "node '{}' emitted an action without an action capability",
                                step.node_id
                            ));
                            break;
                        }
                        run.record_step(
                            &step.node_id,
                            &step.node_type,
                            StepStatus::Ok,
                            None,
                            event_detail(&event),
                        )
                        .await;
                        material_steps += 1;
                        actions.push(*action);
                        next.push(event);
                    }
                }
            }
            steps_run += 1;
            if let Some(reason) = fan_out_error {
                run.record_step(
                    &step.node_id,
                    &step.node_type,
                    StepStatus::Error,
                    Some("fanout_limit_exceeded"),
                    json!({ "reason": reason }),
                )
                .await;
                run.end(RunStatus::Error, Some("fanout_limit_exceeded"))
                    .await;
                return (
                    RunOutcome {
                        steps_run,
                        terminal: Terminal::Dropped {
                            node_id: step.node_id.clone(),
                            reason,
                        },
                        survivors: Vec::new(),
                        actions: Vec::new(),
                        matched: false,
                        succeeded: false,
                    },
                    None,
                );
            }
            if next.is_empty() {
                let (reason, exit_reason) = last_drop.unwrap_or_else(|| ("dropped".into(), None));

                if step.node_type.starts_with("sink.") || material_steps > 0 {
                    run.end(RunStatus::Ok, Some("natural")).await;
                } else if let Some(code) = exit_reason {
                    let step_status = if code.starts_with("invalid_") {
                        StepStatus::Error
                    } else {
                        StepStatus::Skipped
                    };
                    run.record_step(
                        &step.node_id,
                        &step.node_type,
                        step_status,
                        Some(&code),
                        json!({ "reason": reason }),
                    )
                    .await;
                    run.end(
                        if step_status == StepStatus::Error {
                            RunStatus::Error
                        } else {
                            RunStatus::Skipped
                        },
                        Some(&code),
                    )
                    .await;
                } else {
                    run.mark_filtered(&step.node_id, &step.node_type, &reason)
                        .await;
                    run.end(RunStatus::Skipped, None).await;
                }

                let sink_completed = step.node_type.starts_with("sink.");
                return (
                    RunOutcome {
                        steps_run,
                        terminal: if sink_completed {
                            Terminal::Completed
                        } else {
                            Terminal::Dropped {
                                node_id: step.node_id.clone(),
                                reason,
                            }
                        },
                        survivors: Vec::new(),
                        actions,
                        matched: sink_completed || material_steps > 0,
                        succeeded: sink_completed,
                    },
                    None,
                );
            }
            current = next;
            if !actions.is_empty() {
                run.end(RunStatus::Ok, Some("action_pending")).await;
                return (
                    RunOutcome {
                        steps_run,
                        terminal: Terminal::Completed,
                        survivors: current,
                        actions,
                        matched: true,
                        succeeded: false,
                    },
                    (step_index + 1 < self.steps.len()).then_some(step_index + 1),
                );
            }
        }

        run.end(
            if material_steps > 0 {
                RunStatus::Ok
            } else {
                RunStatus::Skipped
            },
            Some("natural"),
        )
        .await;
        (
            RunOutcome {
                steps_run,
                terminal: Terminal::Completed,
                survivors: current,
                actions,
                matched: true,
                succeeded: true,
            },
            None,
        )
    }
}

/// Kahn topological sort over a branch's edges. Nodes with no edges keep spec
/// order. Errors on a cycle.
fn topo_order(branch: &Branch) -> Result<Vec<String>, CompileError> {
    let ids: Vec<&str> = branch.nodes.iter().map(|n| n.id.as_str()).collect();

    let mut indegree: HashMap<&str, usize> = ids.iter().map(|id| (*id, 0)).collect();
    let mut adj: HashMap<&str, Vec<&str>> = ids.iter().map(|id| (*id, Vec::new())).collect();

    for edge in &branch.edges {
        // Dangling edges are caught by Spec::validate_structure; ignore here.
        if let (Some(successors), Some(indegree)) = (
            adj.get_mut(edge.source.as_str()),
            indegree.get_mut(edge.target.as_str()),
        ) {
            successors.push(&edge.target);
            *indegree += 1;
        }
    }

    // Seed queue in spec order to keep deterministic output.
    let mut queue: VecDeque<&str> = ids.iter().copied().filter(|id| indegree[id] == 0).collect();

    let mut order = Vec::with_capacity(ids.len());
    while let Some(id) = queue.pop_front() {
        order.push(id.to_string());
        for &next in &adj[id] {
            let Some(d) = indegree.get_mut(next) else {
                continue;
            };
            *d -= 1;
            if *d == 0 {
                queue.push_back(next);
            }
        }
    }

    if order.len() != ids.len() {
        return Err(CompileError::Cycle {
            branch_id: branch.branch_id.clone(),
        });
    }
    Ok(order)
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use async_trait::async_trait;
    use serde_json::json;

    use super::*;
    use crate::node::StepNode;
    use crate::spec::{Edge, Node};
    use crate::state::MemoryState;

    struct OverProducingMap;

    struct Pass;

    #[async_trait]
    impl StepNode for Pass {
        async fn process(&self, event: &Event, _: &WorkflowContext) -> StepResult {
            StepResult::Pass(event.clone())
        }
    }

    struct Drop;

    #[async_trait]
    impl StepNode for Drop {
        async fn process(&self, _: &Event, _: &WorkflowContext) -> StepResult {
            StepResult::drop("filtered")
        }
    }

    #[async_trait]
    impl StepNode for OverProducingMap {
        fn produces_fan_out(&self) -> bool {
            true
        }

        async fn process(&self, event: &Event, _: &WorkflowContext) -> StepResult {
            StepResult::FanOut(vec![event.clone(), event.clone(), event.clone()])
        }
    }

    fn build_over_producing(_: &Value) -> Result<Box<dyn StepNode>, NodeError> {
        Ok(Box::new(OverProducingMap))
    }

    #[tokio::test]
    async fn runtime_rejects_more_fanout_than_the_static_declaration() {
        let mut registry = NodeRegistry::empty();
        registry.register_ingress("ingress.event");
        registry.register_step("map.test", build_over_producing);
        registry.register_fan_out("map.test");
        let branch = Branch {
            branch_id: "root".into(),
            nodes: vec![
                Node {
                    id: "in".into(),
                    node_type: "ingress.event".into(),
                    config: json!({}),
                },
                Node {
                    id: "map".into(),
                    node_type: "map.test".into(),
                    config: json!({"count": 2}),
                },
            ],
            edges: vec![Edge {
                source: "in".into(),
                target: "map".into(),
            }],
        };
        let compiled = CompiledBranch::compile(&branch, &registry).unwrap();
        let context = WorkflowContext::new("root", Arc::new(MemoryState::new()));
        let outcome = compiled
            .run_event(&context, Event::from_json(json!({})))
            .await;
        assert!(matches!(
            outcome.terminal,
            Terminal::Dropped { ref reason, .. } if reason.contains("fan-out exceeded")
        ));
        assert!(outcome.survivors.is_empty());
    }

    #[tokio::test]
    async fn a_late_filter_matches_without_claiming_success() {
        let mut registry = NodeRegistry::empty();
        registry.register_ingress("ingress.event");
        registry.register_step("execute.pass", |_| Ok(Box::new(Pass)));
        registry.register_step("filter.drop", |_| Ok(Box::new(Drop)));
        let branch = Branch {
            branch_id: "root".into(),
            nodes: vec![
                Node {
                    id: "in".into(),
                    node_type: "ingress.event".into(),
                    config: json!({}),
                },
                Node {
                    id: "material".into(),
                    node_type: "execute.pass".into(),
                    config: json!({}),
                },
                Node {
                    id: "drop".into(),
                    node_type: "filter.drop".into(),
                    config: json!({}),
                },
            ],
            edges: vec![
                Edge {
                    source: "in".into(),
                    target: "material".into(),
                },
                Edge {
                    source: "material".into(),
                    target: "drop".into(),
                },
            ],
        };
        let outcome = CompiledBranch::compile(&branch, &registry)
            .unwrap()
            .run_event(
                &WorkflowContext::new("root", Arc::new(MemoryState::new())),
                Event::from_json(json!({})),
            )
            .await;
        assert!(outcome.matched);
        assert!(!outcome.succeeded);
        assert!(matches!(outcome.terminal, Terminal::Dropped { .. }));
    }
}