af_workflow/

validator.rs

//! Static spec validator. Port of `platform/validator.py`.
//!
//! Runs before compilation. A spec that validates is guaranteed to compile to a
//! well-formed workflow under the chassis safety invariants. Each violation
//! carries a `rule_id` so spec authors and tests can pinpoint the failure.
//!
//! Ported rules (structural + safety, all decidable from the spec + node-type
//! prefixes, no business nodes required):
//!
//! - **R1** — each branch is a DAG (no cycles)
//! - **R4** — each branch has ≥1 ingress
//! - **R5** — no edge crosses a branch boundary
//! - **R11** — a `map.*` fan-out reaching `execute.*` must declare a STATIC bound
//!   within `MAX_MAP_FANOUT`, and a product-declared side-effect guard must
//!   dominate the `map.*` node itself
//! - **R9** — fan-in allowed; structural fan-out forbidden for every node type
//!   except `map.*`, which exists to fan out (non-sink
//!   out-degree exactly 1)
//! - **R1a** — every non-ingress node is reachable from some ingress
//! - **R3a** — side-effect guard config cannot reference `$event.*`
//! - **R2** — every `execute.*` is dominated by a product-declared side-effect
//!   guard. Dominance, not reachability: EVERY path from an entry to the execute
//!   must carry a guard.
//! - **R4'** — no fan-out-capable node upstream of `execute.*`
//! - **R7** — taint (`transform.ask_llm` / LLM-backed `tool_invoke`) cannot
//!   reach a side-effect guard
//! - **R3** — registered per-node config schemas are checked before compile
//! - **R6** — explicitly cross-branch state cannot feed execution decisions

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

use serde::Serialize;
use serde_json::Value;

use crate::registry::NodeRegistry;
use crate::spec::{Branch, Node, Spec};

/// A single rule violation.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct Violation {
    /// Rule identifier (`R1`..`R11`).
    pub rule_id: &'static str,
    /// Workflow branch this record belongs to.
    pub branch_id: String,
    /// Human-readable message.
    pub message: String,
}

impl std::fmt::Display for Violation {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "[{}] branch '{}': {}",
            self.rule_id, self.branch_id, self.message
        )
    }
}

/// Hard ceiling on a `map.*` node's declared fan-out.
///
/// Bounded fan-out is only meaningfully bounded if the bound is small enough that a
/// mistake cannot become an incident.
const MAX_MAP_FANOUT: u64 = 64;

/// Config keys a `map.*` node may declare its fan-out with.
const MAP_FANOUT_KEYS: [&str; 3] = ["count", "levels", "fanout"];
const MAX_BRANCHES: usize = 32;
const MAX_NODES: usize = 256;
const MAX_EDGES: usize = 512;

/// Validate a spec against the chassis rules. Returns every violation found
/// (not just the first) so authors can fix in one pass.
pub fn validate(spec: &Spec, registry: &NodeRegistry) -> Result<(), Vec<Violation>> {
    let mut v = Vec::new();

    let node_count = spec.branches.iter().map(|branch| branch.nodes.len()).sum();
    let edge_count = spec.branches.iter().map(|branch| branch.edges.len()).sum();
    for (actual, limit, kind) in [
        (spec.branches.len(), MAX_BRANCHES, "branches"),
        (node_count, MAX_NODES, "nodes"),
        (edge_count, MAX_EDGES, "edges"),
    ] {
        if actual > limit {
            v.push(Violation {
                rule_id: "R13",
                branch_id: "__spec__".into(),
                message: format!("spec has {actual} {kind}, above the limit of {limit}"),
            });
        }
    }

    // R5 first: cross-branch edges would corrupt every per-branch analysis.
    let branch_of: HashMap<&str, &str> = spec
        .branches
        .iter()
        .flat_map(|b| {
            b.nodes
                .iter()
                .map(move |n| (n.id.as_str(), b.branch_id.as_str()))
        })
        .collect();
    for branch in &spec.branches {
        for edge in &branch.edges {
            for endpoint in [&edge.source, &edge.target] {
                if let Some(owner) = branch_of.get(endpoint.as_str()) {
                    if *owner != branch.branch_id {
                        v.push(Violation {
                            rule_id: "R5",
                            branch_id: branch.branch_id.clone(),
                            message: format!(
                                "edge endpoint '{endpoint}' lives in branch '{owner}'"
                            ),
                        });
                    }
                }
            }
        }
    }

    for branch in &spec.branches {
        let g = Graph::build(branch, registry);
        g.check_r1_dag(&mut v);
        g.check_r4_has_ingress(&mut v);
        g.check_r9_fan_out_shape(&mut v);
        g.check_r11_bounded_map_fanout(&mut v);
        g.check_r1a_reachable(&mut v);
        g.check_r3a_guard_no_event(&mut v);
        g.check_r2_execute_guards(&mut v);
        g.check_required_action_guards(&mut v);
        g.check_r4prime_fan_out_upstream_execute(&mut v);
        g.check_r7_taint(&mut v);
        g.check_r6_cross_branch_state(&mut v);
        g.check_r3_config_schema(&mut v);
    }

    if v.is_empty() {
        Ok(())
    } else {
        Err(v)
    }
}

/// Per-branch graph view with the lookups the rules need.
struct Graph<'a> {
    branch_id: &'a str,
    nodes: &'a [Node],
    by_id: HashMap<&'a str, &'a Node>,
    /// adjacency: node id → successors
    succ: HashMap<&'a str, Vec<&'a str>>,
    /// reverse adjacency: node id → predecessors
    pred: HashMap<&'a str, Vec<&'a str>>,
    registry: &'a NodeRegistry,
}

impl<'a> Graph<'a> {
    fn build(branch: &'a Branch, registry: &'a NodeRegistry) -> Self {
        let by_id: HashMap<&str, &Node> = branch.nodes.iter().map(|n| (n.id.as_str(), n)).collect();
        let mut succ: HashMap<&str, Vec<&str>> = branch
            .nodes
            .iter()
            .map(|n| (n.id.as_str(), Vec::new()))
            .collect();
        let mut pred: HashMap<&str, Vec<&str>> = branch
            .nodes
            .iter()
            .map(|n| (n.id.as_str(), Vec::new()))
            .collect();
        for e in &branch.edges {
            if let (Some(successors), Some(predecessors)) = (
                succ.get_mut(e.source.as_str()),
                pred.get_mut(e.target.as_str()),
            ) {
                successors.push(&e.target);
                predecessors.push(&e.source);
            }
        }
        Self {
            branch_id: &branch.branch_id,
            nodes: &branch.nodes,
            by_id,
            succ,
            pred,
            registry,
        }
    }

    fn is_ingress(&self, n: &Node) -> bool {
        self.registry.is_ingress(&n.node_type)
    }
    fn is_sink(&self, n: &Node) -> bool {
        n.node_type.starts_with("sink.")
    }

    fn violation(&self, rule_id: &'static str, message: String) -> Violation {
        Violation {
            rule_id,
            branch_id: self.branch_id.to_string(),
            message,
        }
    }

    // R1 — DAG (Kahn; if not all nodes drain, there is a cycle).
    fn check_r1_dag(&self, out: &mut Vec<Violation>) {
        let mut indeg: HashMap<&str, usize> = self
            .nodes
            .iter()
            .map(|n| (n.id.as_str(), self.pred[n.id.as_str()].len()))
            .collect();
        let mut q: VecDeque<&str> = indeg
            .iter()
            .filter(|(_, d)| **d == 0)
            .map(|(id, _)| *id)
            .collect();
        let mut seen = 0;
        while let Some(id) = q.pop_front() {
            seen += 1;
            for &s in &self.succ[id] {
                let Some(d) = indeg.get_mut(s) else { continue };
                *d -= 1;
                if *d == 0 {
                    q.push_back(s);
                }
            }
        }
        if seen != self.nodes.len() {
            out.push(self.violation("R1", "edges form a cycle (DAG required)".into()));
        }
    }

    // R4 — at least one ingress.
    fn check_r4_has_ingress(&self, out: &mut Vec<Violation>) {
        if !self.nodes.iter().any(|n| self.is_ingress(n)) {
            out.push(self.violation("R4", "branch has no ingress node".into()));
        }
    }

    // R9 — every non-sink node has out-degree exactly 1, EXCEPT `map.*`.
    //
    // Fan-in (in-degree > 1) has always been allowed. `map.*` is the one node family whose
    // purpose is to fan out, so it is exempt here and constrained by R11 instead: a static
    // bound and side-effect authorization above it. Every
    // other type stays forbidden — an accidental fan-out is still a bug.
    fn check_r9_fan_out_shape(&self, out: &mut Vec<Violation>) {
        for n in self.nodes {
            if self.is_sink(n) || Self::is_map(&n.node_type) {
                continue;
            }
            let outdeg = self.succ[n.id.as_str()].len();
            if outdeg > 1 {
                out.push(self.violation(
                    "R9",
                    format!(
                        "node '{}' has out-degree {outdeg}; structural fan-out is forbidden \
                         (only map.* may fan out)",
                        n.id
                    ),
                ));
            }
        }
    }

    /// `map.*` — the fan-out node family.
    fn is_map(node_type: &str) -> bool {
        node_type.starts_with("map.")
    }

    /// A `map.*` node's declared fan-out, if it is a static literal within bounds.
    ///
    /// Must be a plain integer in the spec: a template (`$event.levels`) or a missing key
    /// means the width is decided at runtime by data, and then "bounded" is a promise
    /// nothing enforces.
    fn static_fanout(node: &Node) -> Result<u64, String> {
        let found = MAP_FANOUT_KEYS
            .iter()
            .find_map(|k| node.config.get(*k).map(|v| (*k, v)));
        let Some((key, value)) = found else {
            return Err(format!(
                "must declare its fan-out with one of {MAP_FANOUT_KEYS:?} as a literal integer"
            ));
        };
        let Some(n) = value.as_u64() else {
            return Err(format!(
                "config.{key} must be a literal integer (found {value}); a runtime-decided \
                 width cannot be bounded"
            ));
        };
        if n == 0 {
            return Err(format!(
                "config.{key} is 0; a fan-out of zero cannot reach execute"
            ));
        }
        if n > MAX_MAP_FANOUT {
            return Err(format!(
                "config.{key} is {n}, above the ceiling of {MAX_MAP_FANOUT}"
            ));
        }
        Ok(n)
    }

    /// All descendants of `node_id` (transitive successors).
    fn descendants(&self, node_id: &str) -> HashSet<&'a str> {
        let mut seen = HashSet::new();
        let mut q: VecDeque<&str> = self.succ.get(node_id).cloned().unwrap_or_default().into();
        while let Some(id) = q.pop_front() {
            if let Some((&kid, _)) = self.by_id.get_key_value(id) {
                if seen.insert(kid) {
                    for &sc in &self.succ[id] {
                        q.push_back(sc);
                    }
                }
            }
        }
        seen
    }

    // R11 — a map.* fan-out reaching execute.* must be statically bounded, and a
    // product-declared side-effect guard must dominate the map node itself.
    fn check_r11_bounded_map_fanout(&self, out: &mut Vec<Violation>) {
        let dom = self.dominators();
        for n in self.nodes {
            if !Self::is_map(&n.node_type) {
                continue;
            }
            // Only fan-outs that can reach a side effect are constrained.
            let reaches_execute = self
                .descendants(&n.id)
                .iter()
                .any(|id| self.by_id[id].node_type.starts_with("execute."));
            if !reaches_execute {
                continue;
            }

            if let Err(why) = Self::static_fanout(n) {
                out.push(self.violation(
                    "R11",
                    format!("map node '{}' reaches execute.* and {why}", n.id),
                ));
            }

            let dominated_by_guard = dom.get(n.id.as_str()).is_some_and(|ds| {
                ds.iter().filter(|id| **id != n.id.as_str()).any(|id| {
                    self.registry
                        .is_side_effect_guard(&self.by_id[id].node_type)
                })
            });
            if !dominated_by_guard {
                out.push(self.violation(
                    "R11",
                    format!(
                        "map node '{}' fans out to execute.* but is not dominated by \
                         a product-declared side-effect guard; authorization must happen \
                         before fan-out",
                        n.id
                    ),
                ));
            }
        }
    }

    // R1a — every non-ingress node reachable from some ingress.
    fn check_r1a_reachable(&self, out: &mut Vec<Violation>) {
        let mut reached: HashSet<&str> = HashSet::new();
        let mut q: VecDeque<&str> = self
            .nodes
            .iter()
            .filter(|n| self.is_ingress(n))
            .map(|n| n.id.as_str())
            .collect();
        for id in &q {
            reached.insert(id);
        }
        while let Some(id) = q.pop_front() {
            for &s in &self.succ[id] {
                if reached.insert(s) {
                    q.push_back(s);
                }
            }
        }
        for n in self.nodes {
            if !self.is_ingress(n) && !reached.contains(n.id.as_str()) {
                out.push(self.violation(
                    "R1a",
                    format!("node '{}' is not reachable from any ingress", n.id),
                ));
            }
        }
    }

    // R3a — authorization policy must not be rewritten by event data.
    fn check_r3a_guard_no_event(&self, out: &mut Vec<Violation>) {
        for n in self.nodes {
            if self.registry.is_side_effect_guard(&n.node_type)
                && config_references_event(&n.config)
            {
                out.push(self.violation(
                    "R3a",
                    format!(
                        "side-effect guard '{}' config references $event.* templates",
                        n.id
                    ),
                ));
            }
        }
    }

    /// All ancestors of `node_id` (transitive predecessors).
    fn ancestors(&self, node_id: &str) -> HashSet<&'a str> {
        let mut seen = HashSet::new();
        let mut q: VecDeque<&str> = self.pred.get(node_id).cloned().unwrap_or_default().into();
        while let Some(id) = q.pop_front() {
            // Re-key to the 'a-lifetime id stored in by_id.
            if let Some((&kid, _)) = self.by_id.get_key_value(id) {
                if seen.insert(kid) {
                    for &p in &self.pred[id] {
                        q.push_back(p);
                    }
                }
            }
        }
        seen
    }

    /// Nodes with no predecessor — where execution can enter this graph.
    ///
    /// Usually one (a branch has exactly one `ingress.*` head, R4), but a graph
    /// assembled from several branches has one entry per branch, and that is precisely
    /// the case R2 has to get right.
    fn entries(&self) -> Vec<&'a str> {
        self.nodes
            .iter()
            .map(|n| n.id.as_str())
            .filter(|id| self.pred.get(*id).is_none_or(|p| p.is_empty()))
            .filter_map(|id| self.by_id.get_key_value(id).map(|(&k, _)| k))
            .collect()
    }

    /// The DOMINATORS of every node: `d` dominates `n` when EVERY path from an entry to
    /// `n` passes through `d`.
    ///
    /// Standard iterative dataflow: `Dom(n) = {n} ∪ ⋂ Dom(p)` over predecessors, with
    /// entries dominated only by themselves, iterated to a fixed point. Multiple entries
    /// are handled by treating them as successors of one virtual root, which is what
    /// makes a fan-in graph come out right: a node is dominated only by what is on ALL
    /// incoming paths, not by whatever happens to be reachable from it backwards.
    fn dominators(&self) -> HashMap<&'a str, HashSet<&'a str>> {
        let all: HashSet<&'a str> = self.by_id.keys().copied().collect();
        let entries: HashSet<&'a str> = self.entries().into_iter().collect();

        let mut dom: HashMap<&'a str, HashSet<&'a str>> = HashMap::new();
        for &id in &all {
            if entries.contains(id) {
                dom.insert(id, HashSet::from([id]));
            } else {
                dom.insert(id, all.clone());
            }
        }

        // Bounded by the node count: each pass can only shrink a set, and a set cannot
        // shrink below one element.
        let mut changed = true;
        let mut guard = all.len() + 1;
        while changed && guard > 0 {
            changed = false;
            guard -= 1;
            for &id in &all {
                if entries.contains(id) {
                    continue;
                }
                let preds = self.pred.get(id).cloned().unwrap_or_default();
                let mut next: Option<HashSet<&'a str>> = None;
                for p in preds {
                    let Some((&pk, _)) = self.by_id.get_key_value(p) else {
                        continue;
                    };
                    let pd = &dom[pk];
                    next = Some(match next {
                        None => pd.clone(),
                        Some(acc) => acc.intersection(pd).copied().collect(),
                    });
                }
                let mut next = next.unwrap_or_default();
                next.insert(id);
                if next != dom[id] {
                    dom.insert(id, next);
                    changed = true;
                }
            }
        }
        dom
    }

    // R2 — execute.* dominated by a product-declared side-effect guard.
    //
    // DOMINANCE, not reachability. This used to ask whether a gate appeared anywhere
    // among an execute node's ancestors, which is a strictly weaker question: in a
    // fan-in graph one branch can reach the execute WITHOUT passing a gate while another
    // branch supplies the gate, and the ancestor set contains it either way. The check
    // passed and the guard was bypassable — the whole point of R2 is that it cannot be.
    // `dominators` asks whether EVERY path carries the guard.
    fn check_r2_execute_guards(&self, out: &mut Vec<Violation>) {
        let dom = self.dominators();
        for n in self.nodes {
            if !n.node_type.starts_with("execute.") {
                continue;
            }
            let Some(doms) = dom.get(n.id.as_str()) else {
                continue;
            };
            let guarded = doms.iter().filter(|id| **id != n.id.as_str()).any(|id| {
                self.registry
                    .is_side_effect_guard(&self.by_id[id].node_type)
            });
            if !guarded {
                out.push(self.violation(
                    "R2",
                    format!(
                        "execute node '{}' is not dominated by a product-declared side-effect guard",
                        n.id
                    ),
                ));
            }
        }
    }

    // R12 — capability-declared guards must dominate every action path.
    fn check_required_action_guards(&self, out: &mut Vec<Violation>) {
        let dom = self.dominators();
        for node in self.nodes {
            let Some(manifest) = self.registry.capability(&node.node_type) else {
                continue;
            };
            let Some(dominators) = dom.get(node.id.as_str()) else {
                continue;
            };
            for required in &manifest.required_guards {
                let present = dominators.iter().any(|id| {
                    *id != node.id.as_str()
                        && self.registry.guard_kind(&self.by_id[id].node_type) == Some(*required)
                });
                if !present {
                    out.push(self.violation(
                        "R12",
                        format!(
                            "action '{}' ({}) is not dominated by required {:?} guard",
                            node.id, node.node_type, required
                        ),
                    ));
                }
            }
        }
    }

    // R4' — no fan-out-capable node upstream of execute.*
    fn check_r4prime_fan_out_upstream_execute(&self, out: &mut Vec<Violation>) {
        for n in self.nodes {
            if !n.node_type.starts_with("execute.") {
                continue;
            }
            for anc in self.ancestors(&n.id) {
                let at = &self.by_id[anc].node_type;
                // `map.*` is deliberately exempt: R11 constrains it with a static bound
                // and side-effect authorization above it. R4' catches every other fan-out-capable node
                // upstream of an execute — the accidental-amplification case it was
                // written for.
                if Self::is_map(at) {
                    continue;
                }
                if self.registry.is_fan_out_capable(at) {
                    out.push(self.violation(
                        "R4'",
                        format!(
                            "fan-out-capable node '{anc}' ({at}) is upstream of execute '{}'",
                            n.id
                        ),
                    ));
                }
            }
        }
    }

    fn check_r3_config_schema(&self, out: &mut Vec<Violation>) {
        for node in self.nodes {
            let schema = self.registry.schema(&node.node_type);
            if let Some(manifest) = self.registry.capability(&node.node_type) {
                // Template expressions are type-checked after resolution by the node factory.
                // Literal configs are checked here from the catalog's exact manifest.
                if !contains_template(&node.config) {
                    if let Err(error) = jsonschema::validator_for(&manifest.config_schema)
                        .and_then(|validator| validator.validate(&node.config))
                    {
                        out.push(self.violation(
                            "R3",
                            format!("node '{}' ({}) config: {error}", node.id, node.node_type),
                        ));
                    }
                }
            }
            let Some(schema) = schema else {
                continue;
            };
            for field in &schema.fields {
                match node.config.get(&field.key) {
                    None if field.required => out.push(self.violation(
                        "R3",
                        format!(
                            "node '{}' ({}) is missing config key '{}'",
                            node.id, node.node_type, field.key
                        ),
                    )),
                    Some(value) if !field_type_matches(value, field.ty) => {
                        out.push(self.violation(
                            "R3",
                            format!(
                                "node '{}' ({}) config '{}' must be {:?}",
                                node.id, node.node_type, field.key, field.ty
                            ),
                        ));
                    }
                    _ => {}
                }
            }
            if let Some(config) = node.config.as_object() {
                for key in config.keys().filter(|key| {
                    !schema
                        .fields
                        .iter()
                        .any(|field| field.key.as_str() == key.as_str())
                }) {
                    out.push(self.violation(
                        "R3",
                        format!(
                            "node '{}' ({}) has unknown config key '{key}'",
                            node.id, node.node_type
                        ),
                    ));
                }
            }
        }
    }

    fn check_r6_cross_branch_state(&self, out: &mut Vec<Violation>) {
        for node in self.nodes {
            if !(node.node_type.starts_with("execute.")
                || self.registry.is_side_effect_guard(&node.node_type))
            {
                continue;
            }
            for ancestor in self.ancestors(&node.id) {
                let source = &self.by_id[ancestor].node_type;
                if is_cross_branch_state(source) {
                    out.push(self.violation(
                        "R6",
                        format!(
                            "cross-branch state node '{ancestor}' ({source}) is upstream of '{}' ({})",
                            node.id, node.node_type
                        ),
                    ));
                }
            }
        }
    }

    // R7 — taint cannot decide side-effect authorization.
    fn check_r7_taint(&self, out: &mut Vec<Violation>) {
        for n in self.nodes {
            if !self.registry.is_side_effect_guard(&n.node_type) {
                continue;
            }
            for anc in self.ancestors(&n.id) {
                if is_taint_source(self.by_id[anc]) {
                    out.push(self.violation(
                        "R7",
                        format!(
                            "tainted output of '{anc}' ({}) reaches side-effect guard '{}'",
                            self.by_id[anc].node_type, n.id
                        ),
                    ));
                }
            }
        }
    }
}

fn contains_template(value: &Value) -> bool {
    match value {
        Value::String(value) => value.starts_with('$'),
        Value::Array(values) => values.iter().any(contains_template),
        Value::Object(values) => values.values().any(contains_template),
        _ => false,
    }
}

fn field_type_matches(value: &Value, field_type: crate::registry::FieldType) -> bool {
    use crate::registry::FieldType;

    if matches!(value, Value::String(template) if template.starts_with('$')) {
        return true;
    }
    match field_type {
        FieldType::String => value.is_string(),
        FieldType::Number => value.is_number(),
        FieldType::Bool => value.is_boolean(),
        FieldType::Array => value.is_array(),
        FieldType::Object => value.is_object(),
        FieldType::Any => true,
    }
}

fn is_cross_branch_state(node_type: &str) -> bool {
    matches!(
        node_type,
        "transform.state_publish_cross_branch"
            | "transform.state_read_cross_branch"
            | "transform.state_append_cross_branch"
    )
}

/// A node whose output is untrusted (LLM-derived). `transform.ask_llm` always;
/// `transform.tool_invoke` unless explicitly marked `"llm_backed": false`.
fn is_taint_source(node: &Node) -> bool {
    match node.node_type.as_str() {
        "transform.ask_llm" => true,
        // Default to tainted when the key is absent.
        "transform.tool_invoke" => node
            .config
            .get("llm_backed")
            .and_then(|v| v.as_bool())
            .unwrap_or(true),
        _ => false,
    }
}

/// Whether any string anywhere in a config value is an `$event.` template.
fn config_references_event(config: &Value) -> bool {
    match config {
        Value::String(s) => s.starts_with("$event."),
        Value::Array(a) => a.iter().any(config_references_event),
        Value::Object(o) => o.values().any(config_references_event),
        _ => false,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::spec::Spec;

    fn reg() -> NodeRegistry {
        let mut r = NodeRegistry::with_builtins();
        r.register_step("transform.ask_llm", |_| {
            Err(crate::registry::NodeError::UnknownType(
                "ask_llm stub".into(),
            ))
        });
        r.register_step("guard.permission", |_| {
            Err(crate::registry::NodeError::UnknownType("guard stub".into()))
        });
        r.register_side_effect_guard("guard.permission");
        r
    }

    fn parse(json: &str) -> Spec {
        Spec::from_json(json).unwrap()
    }

    #[test]
    fn clean_linear_spec_passes() {
        let spec = parse(
            r#"{"spec_id":"ok","version":"1.0","branches":[{"branch_id":"__root__",
              "nodes":[
                {"id":"t","type":"ingress.cron","config":{}},
                {"id":"l","type":"sink.log","config":{"message":"x"}}],
              "edges":[{"source":"t","target":"l"}]}]}"#,
        );
        assert!(validate(&spec, &reg()).is_ok());
    }

    #[test]
    fn builtin_schema_errors_trip_r3() {
        let spec = parse(
            r#"{"spec_id":"schema","version":"1.0","branches":[{"branch_id":"__root__",
              "nodes":[
                {"id":"t","type":"ingress.cron","config":{}},
                {"id":"s","type":"transform.state_set","config":{"key":3}}],
              "edges":[{"source":"t","target":"s"}]}]}"#,
        );
        let errors = validate(&spec, &reg()).unwrap_err();
        assert_eq!(
            errors.iter().filter(|error| error.rule_id == "R3").count(),
            2
        );
    }

    #[test]
    fn cross_branch_state_cannot_feed_execution_decisions() {
        let spec = parse(
            r#"{"spec_id":"cross","version":"1.0","branches":[{"branch_id":"__root__",
              "nodes":[
                {"id":"t","type":"ingress.cron","config":{}},
                {"id":"shared","type":"transform.state_read_cross_branch","config":{"key":"x","into":"x"}},
                {"id":"guard","type":"guard.permission","config":{}},
                {"id":"log","type":"sink.log","config":{"message":"done"}}],
              "edges":[{"source":"t","target":"shared"},{"source":"shared","target":"guard"},{"source":"guard","target":"log"}]}]}"#,
        );
        let errors = validate(&spec, &reg()).unwrap_err();
        assert!(errors.iter().any(|error| error.rule_id == "R6"));
    }

    #[test]
    fn cycle_trips_r1() {
        let spec = parse(
            r#"{"spec_id":"c","version":"1.0","branches":[{"branch_id":"__root__",
              "nodes":[
                {"id":"t","type":"ingress.cron","config":{}},
                {"id":"a","type":"transform.set_fields","config":{"fields":{}}},
                {"id":"b","type":"transform.set_fields","config":{"fields":{}}}],
              "edges":[{"source":"t","target":"a"},{"source":"a","target":"b"},
                       {"source":"b","target":"a"}]}]}"#,
        );
        let errs = validate(&spec, &reg()).unwrap_err();
        assert!(errs.iter().any(|e| e.rule_id == "R1"));
    }

    #[test]
    fn fan_out_trips_r9() {
        let spec = parse(
            r#"{"spec_id":"f","version":"1.0","branches":[{"branch_id":"__root__",
              "nodes":[
                {"id":"t","type":"ingress.cron","config":{}},
                {"id":"a","type":"sink.log","config":{"message":"a"}},
                {"id":"b","type":"sink.log","config":{"message":"b"}}],
              "edges":[{"source":"t","target":"a"},{"source":"t","target":"b"}]}]}"#,
        );
        let errs = validate(&spec, &reg()).unwrap_err();
        assert!(errs.iter().any(|e| e.rule_id == "R9"));
    }

    #[test]
    fn execute_without_a_product_guard_trips_r2() {
        let spec = parse(
            r#"{"spec_id":"e","version":"1.0","branches":[{"branch_id":"__root__",
              "nodes":[
                {"id":"t","type":"ingress.cron","config":{}},
                {"id":"x","type":"execute.publish","config":{}}],
              "edges":[{"source":"t","target":"x"}]}]}"#,
        );
        let errs = validate(&spec, &reg()).unwrap_err();
        assert_eq!(errs.iter().filter(|e| e.rule_id == "R2").count(), 1);
    }

    #[test]
    fn funds_action_requires_every_declared_guard_on_every_path() {
        let mut registry = reg();
        for (node_type, kind) in [
            ("guard.freshness", crate::GuardKind::Freshness),
            ("guard.reservation", crate::GuardKind::Reservation),
        ] {
            registry.register_step(node_type, |_| {
                Err(crate::NodeError::UnknownType("guard stub".into()))
            });
            registry.register_guard(node_type, kind);
        }
        registry
            .register_capability(crate::CapabilityManifest::action(
                "execute.funds",
                "1",
                "digest",
                crate::Effect::Funds,
                crate::IdempotencyMode::ReconcileBeforeRetry,
                true,
            ))
            .unwrap();
        registry.register_step("execute.funds", |_| {
            Err(crate::NodeError::UnknownType("action stub".into()))
        });
        let missing = parse(
            r#"{"spec_id":"funds","version":"1","branches":[{"branch_id":"__root__",
              "nodes":[
                {"id":"in","type":"ingress.event","config":{}},
                {"id":"auth","type":"guard.permission","config":{}},
                {"id":"action","type":"execute.funds","config":{}}],
              "edges":[{"source":"in","target":"auth"},{"source":"auth","target":"action"}]}]}"#,
        );
        let errors = validate(&missing, &registry).unwrap_err();
        assert_eq!(
            errors.iter().filter(|error| error.rule_id == "R12").count(),
            2
        );

        let complete = parse(
            r#"{"spec_id":"funds","version":"1","branches":[{"branch_id":"__root__",
              "nodes":[
                {"id":"in","type":"ingress.event","config":{}},
                {"id":"auth","type":"guard.permission","config":{}},
                {"id":"fresh","type":"guard.freshness","config":{}},
                {"id":"reserve","type":"guard.reservation","config":{}},
                {"id":"action","type":"execute.funds","config":{}}],
              "edges":[{"source":"in","target":"auth"},{"source":"auth","target":"fresh"},{"source":"fresh","target":"reserve"},{"source":"reserve","target":"action"}]}]}"#,
        );
        assert!(validate(&complete, &registry).is_ok());
    }

    #[test]
    fn a_guard_on_only_one_incoming_path_still_trips_r2() {
        let spec = parse(
            r#"{"spec_id":"bypass","version":"1.0","branches":[{"branch_id":"__root__",
              "nodes":[
                {"id":"t1","type":"ingress.cron","config":{}},
                {"id":"t2","type":"ingress.cron","config":{}},
                {"id":"g","type":"guard.permission","config":{}},
                {"id":"x","type":"execute.publish","config":{}}],
              "edges":[
                {"source":"t1","target":"g"},
                {"source":"g","target":"x"},
                {"source":"t2","target":"x"}]}]}"#,
        );
        let errs = validate(&spec, &reg()).unwrap_err();
        let r2: Vec<&str> = errs
            .iter()
            .filter(|e| e.rule_id == "R2")
            .map(|e| e.message.as_str())
            .collect();
        assert_eq!(r2.len(), 1, "an unguarded path must trip R2: {r2:?}");
    }

    #[test]
    fn a_guard_on_every_incoming_path_satisfies_r2() {
        let spec = parse(
            r#"{"spec_id":"gated","version":"1.0","branches":[{"branch_id":"__root__",
              "nodes":[
                {"id":"t1","type":"ingress.cron","config":{}},
                {"id":"t2","type":"ingress.cron","config":{}},
                {"id":"j","type":"transform.map","config":{}},
                {"id":"g","type":"guard.permission","config":{}},
                {"id":"x","type":"execute.publish","config":{}}],
              "edges":[
                {"source":"t1","target":"j"},
                {"source":"t2","target":"j"},
                {"source":"j","target":"g"},
                {"source":"g","target":"x"}]}]}"#,
        );
        let r2 = match validate(&spec, &reg()) {
            Ok(()) => Vec::new(),
            Err(errs) => errs
                .iter()
                .filter(|e| e.rule_id == "R2")
                .map(|e| e.message.clone())
                .collect(),
        };
        assert!(
            r2.is_empty(),
            "fan-in where every path is guarded must satisfy R2, got: {r2:?}"
        );
    }

    #[test]
    fn a_bounded_map_fanout_under_shared_authorization_is_accepted() {
        let spec = parse(
            r#"{"spec_id":"grid","version":"1.0","branches":[{"branch_id":"__root__",
              "nodes":[
                {"id":"t","type":"ingress.cron","config":{}},
                {"id":"g","type":"guard.permission","config":{}},
                {"id":"m","type":"map.batch","config":{"count":3}},
                {"id":"x1","type":"execute.publish","config":{}},
                {"id":"x2","type":"execute.publish","config":{}}],
              "edges":[
                {"source":"t","target":"g"},
                {"source":"g","target":"m"},
                {"source":"m","target":"x1"},
                {"source":"m","target":"x2"}]}]}"#,
        );
        let errs = match validate(&spec, &reg()) {
            Ok(()) => Vec::new(),
            Err(e) => e,
        };
        let relevant: Vec<&str> = errs
            .iter()
            .filter(|e| matches!(e.rule_id, "R9" | "R11" | "R4'" | "R2"))
            .map(|e| e.message.as_str())
            .collect();
        assert!(
            relevant.is_empty(),
            "bounded fan-out under shared authorization must validate, got: {relevant:?}"
        );
    }

    #[test]
    fn per_leg_authorization_trips_r11() {
        let spec = parse(
            r#"{"spec_id":"grid_bad","version":"1.0","branches":[{"branch_id":"__root__",
              "nodes":[
                {"id":"t","type":"ingress.cron","config":{}},
                {"id":"m","type":"map.batch","config":{"count":2}},
                {"id":"ga","type":"guard.permission","config":{}},
                {"id":"gb","type":"guard.permission","config":{}},
                {"id":"x1","type":"execute.publish","config":{}},
                {"id":"x2","type":"execute.publish","config":{}}],
              "edges":[
                {"source":"t","target":"m"},
                {"source":"m","target":"ga"},
                {"source":"m","target":"gb"},
                {"source":"ga","target":"x1"},
                {"source":"gb","target":"x2"}]}]}"#,
        );
        let errs = validate(&spec, &reg()).unwrap_err();
        assert!(
            errs.iter()
                .any(|e| e.rule_id == "R11"
                    && e.message.contains("product-declared side-effect guard")),
            "per-leg authorization must trip R11, got: {:?}",
            errs.iter()
                .map(|e| (&e.rule_id, &e.message))
                .collect::<Vec<_>>()
        );
        assert!(
            !errs.iter().any(|e| e.rule_id == "R2"),
            "per-leg guards satisfy R2 — R11 must catch the fan-out placement"
        );
    }

    #[test]
    fn an_unbounded_map_fanout_trips_r11() {
        // A width decided by runtime data is not bounded, whatever the config looks like.
        for cfg in [
            r#"{}"#,
            r#"{"levels":"$event.levels"}"#,
            r#"{"levels":0}"#,
            r#"{"levels":65}"#,
        ] {
            let spec = parse(&format!(
                r#"{{"spec_id":"g","version":"1.0","branches":[{{"branch_id":"__root__",
                  "nodes":[
                    {{"id":"t","type":"ingress.cron","config":{{}}}},
                    {{"id":"g","type":"guard.permission","config":{{}}}},
                    {{"id":"m","type":"map.batch","config":{cfg}}},
                    {{"id":"x","type":"execute.publish","config":{{}}}}],
                  "edges":[
                    {{"source":"t","target":"g"}},
                    {{"source":"g","target":"m"}},
                    {{"source":"m","target":"x"}}]}}]}}"#
            ));
            let errs = validate(&spec, &reg()).unwrap_err();
            assert!(
                errs.iter().any(|e| e.rule_id == "R11"),
                "config {cfg} must trip R11, got: {:?}",
                errs.iter().map(|e| &e.rule_id).collect::<Vec<_>>()
            );
        }
    }

    #[test]
    fn a_map_that_reaches_no_execute_is_unconstrained() {
        // Fan-out with no side effect downstream needs no authorization rule.
        let spec = parse(
            r#"{"spec_id":"notify_fan","version":"1.0","branches":[{"branch_id":"__root__",
              "nodes":[
                {"id":"t","type":"ingress.cron","config":{}},
                {"id":"m","type":"map.recipients","config":{}},
                {"id":"a","type":"sink.log","config":{"message":"a"}},
                {"id":"b","type":"sink.log","config":{"message":"b"}}],
              "edges":[
                {"source":"t","target":"m"},
                {"source":"m","target":"a"},
                {"source":"m","target":"b"}]}]}"#,
        );
        let errs = match validate(&spec, &reg()) {
            Ok(()) => Vec::new(),
            Err(e) => e,
        };
        assert!(
            !errs.iter().any(|e| e.rule_id == "R11" || e.rule_id == "R9"),
            "a map with no execute downstream must be unconstrained, got: {:?}",
            errs.iter()
                .map(|e| (&e.rule_id, &e.message))
                .collect::<Vec<_>>()
        );
    }

    #[test]
    fn non_map_fan_out_is_still_forbidden() {
        // The relaxation is scoped to map.*; an accidental fan-out anywhere else is still
        // a bug, and R9 must keep saying so.
        let spec = parse(
            r#"{"spec_id":"oops","version":"1.0","branches":[{"branch_id":"__root__",
              "nodes":[
                {"id":"t","type":"ingress.cron","config":{}},
                {"id":"d","type":"transform.map","config":{}},
                {"id":"a","type":"sink.log","config":{"message":"a"}},
                {"id":"b","type":"sink.log","config":{"message":"b"}}],
              "edges":[
                {"source":"t","target":"d"},
                {"source":"d","target":"a"},
                {"source":"d","target":"b"}]}]}"#,
        );
        let errs = validate(&spec, &reg()).unwrap_err();
        assert!(
            errs.iter().any(|e| e.rule_id == "R9"),
            "transform.map is not map.* — fan-out there must still trip R9"
        );
    }

    #[test]
    fn llm_taint_into_a_side_effect_guard_trips_r7() {
        let spec = parse(
            r#"{"spec_id":"t7","version":"1.0","branches":[{"branch_id":"__root__",
              "nodes":[
                {"id":"t","type":"ingress.cron","config":{}},
                {"id":"ask","type":"transform.ask_llm","config":{}},
                {"id":"g","type":"guard.permission","config":{}}],
              "edges":[{"source":"t","target":"ask"},{"source":"ask","target":"g"}]}]}"#,
        );
        let errs = validate(&spec, &reg()).unwrap_err();
        assert!(errs.iter().any(|e| e.rule_id == "R7"));
    }

    #[test]
    fn guard_config_with_event_template_trips_r3a() {
        let spec = parse(
            r#"{"spec_id":"t3a","version":"1.0","branches":[{"branch_id":"__root__",
              "nodes":[
                {"id":"t","type":"ingress.cron","config":{}},
                {"id":"g","type":"guard.permission","config":{"scope":"$event.x"}}],
              "edges":[{"source":"t","target":"g"}]}]}"#,
        );
        let errs = validate(&spec, &reg()).unwrap_err();
        assert!(errs.iter().any(|e| e.rule_id == "R3a"));
    }
}

#[cfg(test)]
mod authoring_catalog_tests {
    use super::*;
    use crate::{CapabilityManifest, Effect, IdempotencyMode};
    fn spec(config: Value) -> Spec {
        serde_json::from_value(serde_json::json!({"spec_id":"authoring","version":"1","branches":[{"branch_id":"root","nodes":[{"id":"in","type":"ingress.event","config":{}},{"id":"custom","type":"custom.node","config":config}],"edges":[{"source":"in","target":"custom"}]}]})).unwrap()
    }
    #[test]
    fn manifest_config_schema_and_builtin_unknown_keys_share_r3() {
        let mut registry = NodeRegistry::with_builtins();
        let mut manifest = CapabilityManifest::action(
            "custom.node",
            "1",
            "digest",
            Effect::Pure,
            IdempotencyMode::Native,
            false,
        );
        manifest.config_schema = serde_json::json!({"type":"object","required":["count"],"properties":{"count":{"type":"integer"}},"additionalProperties":false});
        registry.register_capability(manifest).unwrap();
        assert!(validate(&spec(serde_json::json!({"count":1})), &registry).is_ok());
        for bad in [
            serde_json::json!({}),
            serde_json::json!({"count":"one"}),
            serde_json::json!({"count":1,"unknown":true}),
        ] {
            assert!(validate(&spec(bad), &registry)
                .unwrap_err()
                .iter()
                .any(|error| error.rule_id == "R3"));
        }
        assert!(validate(
            &spec(serde_json::json!({"count":"$event.count"})),
            &registry
        )
        .is_ok());
        let builtin:Spec=serde_json::from_value(serde_json::json!({"spec_id":"builtin","version":"1","branches":[{"branch_id":"root","nodes":[{"id":"in","type":"ingress.event","config":{}},{"id":"sink","type":"sink.log","config":{"message":"ok","unknown":true}}],"edges":[{"source":"in","target":"sink"}]}]})).unwrap();
        assert!(validate(&builtin, &registry)
            .unwrap_err()
            .iter()
            .any(|error| error.message.contains("unknown config key")));
        let schemas = registry.authoring_schemas();
        assert!(schemas.windows(2).all(|pair| pair[0].0 < pair[1].0));
        assert_eq!(
            schemas.iter().find(|(id, _)| id == "sink.log").unwrap().1["additionalProperties"],
            false
        );
    }
}