use std::collections::{HashMap, HashSet, VecDeque};
use serde_json::Value;
use crate::registry::NodeRegistry;
use crate::spec::{Branch, Node, Spec};
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Violation {
pub rule_id: &'static str,
pub branch_id: String,
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
)
}
}
const MAX_MAP_FANOUT: u64 = 64;
const MAP_FANOUT_KEYS: [&str; 3] = ["count", "levels", "fanout"];
const MAX_BRANCHES: usize = 32;
const MAX_NODES: usize = 256;
const MAX_EDGES: usize = 512;
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}"),
});
}
}
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)
}
}
struct Graph<'a> {
branch_id: &'a str,
nodes: &'a [Node],
by_id: HashMap<&'a str, &'a Node>,
succ: HashMap<&'a str, Vec<&'a str>>,
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,
}
}
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()));
}
}
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()));
}
}
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
),
));
}
}
}
fn is_map(node_type: &str) -> bool {
node_type.starts_with("map.")
}
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)
}
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
}
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;
}
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
),
));
}
}
}
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),
));
}
}
}
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
),
));
}
}
}
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() {
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
}
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()
}
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());
}
}
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
}
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
),
));
}
}
}
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
),
));
}
}
}
}
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;
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 Some(schema) = self.registry.schema(&node.node_type) 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
),
));
}
_ => {}
}
}
}
}
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
),
));
}
}
}
}
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 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"
)
}
fn is_taint_source(node: &Node) -> bool {
match node.node_type.as_str() {
"transform.ask_llm" => true,
"transform.tool_invoke" => node
.config
.get("llm_backed")
.and_then(|v| v.as_bool())
.unwrap_or(true),
_ => false,
}
}
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, ®()).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, ®()).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, ®()).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, ®()).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, ®()).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, ®()).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, ®istry).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, ®istry).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, ®()).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, ®()) {
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, ®()) {
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, ®()).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() {
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, ®()).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() {
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, ®()) {
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() {
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, ®()).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, ®()).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, ®()).unwrap_err();
assert!(errs.iter().any(|e| e.rule_id == "R3a"));
}
}