af-workflow 0.2.0

//! 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, HashSet, 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::StepResult;
use crate::spec::Branch;

#[derive(Debug, thiserror::Error)]
pub enum CompileError {
    #[error(transparent)]
    Node(#[from] NodeError),
    #[error("branch '{branch_id}' has no ingress node (chain needs a source)")]
    NoIngress { branch_id: String },
    #[error("branch '{branch_id}' has {count} ingress nodes; day-1 supports one")]
    MultipleIngress { branch_id: String, count: usize },
    #[error("branch '{branch_id}' edges form a cycle (DAG required)")]
    Cycle { 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>,
}

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_id: String, 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 {
    pub steps_run: usize,
    pub terminal: Terminal,
    /// Events still alive at the end (empty if everything dropped).
    pub survivors: Vec<Event>,
}

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,
            });
        }

        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 {
        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;

        for step in &self.steps {
            let mut next = Vec::new();
            let mut last_drop: Option<(String, Option<String>)> = 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) => next.extend(evs),
                }
            }
            steps_run += 1;
            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;
                }

                return RunOutcome {
                    steps_run,
                    terminal: Terminal::Dropped {
                        node_id: step.node_id.clone(),
                        reason,
                    },
                    survivors: Vec::new(),
                };
            }
            current = next;
        }

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

/// 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 id_set: HashSet<&str> = ids.iter().copied().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 id_set.contains(edge.source.as_str()) && id_set.contains(edge.target.as_str()) {
            adj.get_mut(edge.source.as_str())
                .unwrap()
                .push(&edge.target);
            *indegree.get_mut(edge.target.as_str()).unwrap() += 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 d = indegree.get_mut(next).unwrap();
            *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)
}