beam_core/

workflow_definition.rs

use std::collections::{BTreeMap, BTreeSet, HashMap};

use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};
use serde_json::Value;

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct ParamDef {
    #[serde(rename = "type")]
    pub param_type: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub format: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub required: Option<bool>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub default: Option<Value>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub description: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct RetryPolicy {
    pub max_attempts: u64,
    pub backoff: String,
    pub base_ms: u64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub factor: Option<f64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub jitter: Option<bool>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct HumanGate {
    pub stage: String,
    pub prompt: Value,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub approvers: Option<Vec<String>>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub deadline_ms: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub on_timeout: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct LoopTerminate {
    pub node: String,
    pub via: String,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct LoopOutputProjection {
    pub from: String,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct NodeBase {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub description: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub depends: Option<Vec<String>>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub human_gate: Option<HumanGate>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub retry_policy: Option<RetryPolicy>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub timeout_ms: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub max_output_bytes: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub output_schema: Option<Value>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub unsafe_allow_ungated: Option<bool>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct SubagentNode {
    #[serde(flatten)]
    pub base: NodeBase,
    pub bot: String,
    pub prompt: Value,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub working_dir: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub model_overrides: Option<Value>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub tool_policy: Option<Value>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct HostExecutorNode {
    #[serde(flatten)]
    pub base: NodeBase,
    pub executor: String,
    pub input: Value,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct LoopNode {
    #[serde(flatten)]
    pub base: NodeBase,
    pub max_iterations: u64,
    pub body: Vec<String>,
    pub terminate: LoopTerminate,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub output: Option<LoopOutputProjection>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct DecisionNode {
    #[serde(flatten)]
    pub base: NodeBase,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(tag = "type", rename_all = "camelCase")]
pub enum WorkflowNode {
    Subagent(SubagentNode),
    HostExecutor(HostExecutorNode),
    Loop(LoopNode),
    Decision(DecisionNode),
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct WorkflowDefaults {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub retry_policy: Option<RetryPolicy>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub timeout_ms: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub max_output_bytes: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub max_concurrency: Option<u64>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct WorkflowDefinition {
    pub workflow_id: String,
    pub version: u64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub params: Option<BTreeMap<String, ParamDef>>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub defaults: Option<WorkflowDefaults>,
    pub nodes: BTreeMap<String, WorkflowNode>,
}

pub fn parse_workflow_definition(raw: &str) -> Result<WorkflowDefinition> {
    let def: WorkflowDefinition =
        serde_json::from_str(raw).context("failed to parse workflow json")?;
    validate_workflow_definition(&def)?;
    Ok(def)
}

/// Side-effect executors that MUST be gated (humanGate or unsafeAllowUngated).
const SIDE_EFFECT_EXECUTORS: &[&str] = &["feishu-send", "feishu-reply", "beam-schedule"];

fn is_side_effect_executor(executor: &str) -> bool {
    SIDE_EFFECT_EXECUTORS.contains(&executor)
}

pub fn validate_workflow_definition(def: &WorkflowDefinition) -> Result<()> {
    if def.workflow_id.trim().is_empty() {
        anyhow::bail!("workflowId 缺失");
    }
    if def.version == 0 {
        anyhow::bail!("version must be positive");
    }
    if def.nodes.is_empty() {
        anyhow::bail!("Workflow must declare at least one node");
    }
    for node_id in def.nodes.keys() {
        // Node id must match ^[A-Za-z0-9_.-]+$ (non-empty)
        if node_id.is_empty() {
            anyhow::bail!("nodeId '' rejected: must match ^[A-Za-z0-9_.-]+$");
        }
        if !node_id
            .chars()
            .all(|ch| ch.is_ascii_alphanumeric() || ch == '_' || ch == '.' || ch == '-')
        {
            anyhow::bail!(
                "nodeId '{}' rejected: must match ^[A-Za-z0-9_.-]+$",
                node_id
            );
        }
        // Preserve existing path-traversal rejection
        if node_id == "." || node_id == ".." || node_id.contains("..") {
            anyhow::bail!(
                "nodeId '{}' rejected: path-traversal style ids are not allowed",
                node_id
            );
        }
    }
    for (node_id, node) in &def.nodes {
        match node {
            WorkflowNode::HostExecutor(host) => {
                // Side-effect executors must be gated
                if is_side_effect_executor(&host.executor)
                    && host.base.human_gate.is_none()
                    && !host.base.unsafe_allow_ungated.unwrap_or(false)
                {
                    anyhow::bail!(
                        "nodeId '{}': side-effect executor '{}' must have a humanGate or set unsafeAllowUngated: true",
                        node_id,
                        host.executor
                    );
                }
            }
            _ => {}
        }
    }
    for (node_id, node) in &def.nodes {
        for dep in node_depends(node) {
            if !def.nodes.contains_key(dep) {
                anyhow::bail!("Node '{}' depends on unknown node '{}'", node_id, dep);
            }
            if dep == node_id {
                anyhow::bail!("Node '{}' depends on itself", node_id);
            }
        }
    }
    detect_cycles(def)?;

    // Loop definition validation (Task 8.3)
    validate_loop_definitions(def)?;

    // Check for at least one scheduler-visible root node
    // (exclude body nodes and Decision nodes — they're dispatched inside their loop context)
    let body_owner = build_body_owner_map(def);
    let has_root = def.nodes.iter().any(|(node_id, node)| {
        if body_owner.contains_key(node_id) {
            return false;
        }
        if matches!(node, WorkflowNode::Decision(_)) {
            return false;
        }
        node_depends(node).is_empty()
    });
    if !has_root {
        anyhow::bail!(
            "Workflow has no scheduler-visible root node (every non-loop-body, non-decision node has dependencies)"
        );
    }
    Ok(())
}

fn node_depends(node: &WorkflowNode) -> &[String] {
    match node {
        WorkflowNode::Subagent(n) => n.base.depends.as_deref().unwrap_or(&[]),
        WorkflowNode::HostExecutor(n) => n.base.depends.as_deref().unwrap_or(&[]),
        WorkflowNode::Loop(n) => n.base.depends.as_deref().unwrap_or(&[]),
        WorkflowNode::Decision(n) => n.base.depends.as_deref().unwrap_or(&[]),
    }
}

fn detect_cycles(def: &WorkflowDefinition) -> Result<()> {
    #[derive(Clone, Copy, PartialEq, Eq)]
    enum Mark {
        Temp,
        Perm,
    }
    fn visit(
        node_id: &str,
        def: &WorkflowDefinition,
        marks: &mut BTreeMap<String, Mark>,
        stack: &mut BTreeSet<String>,
    ) -> Result<()> {
        if matches!(marks.get(node_id), Some(Mark::Perm)) {
            return Ok(());
        }
        if matches!(marks.get(node_id), Some(Mark::Temp)) {
            anyhow::bail!("Workflow graph contains a cycle at '{}'", node_id);
        }
        marks.insert(node_id.to_string(), Mark::Temp);
        stack.insert(node_id.to_string());
        for dep in node_depends(def.nodes.get(node_id).expect("node must exist")) {
            visit(dep, def, marks, stack)?;
        }
        stack.remove(node_id);
        marks.insert(node_id.to_string(), Mark::Perm);
        Ok(())
    }

    let mut marks = BTreeMap::new();
    let mut stack = BTreeSet::new();
    for node_id in def.nodes.keys() {
        visit(node_id, def, &mut marks, &mut stack)?;
    }
    Ok(())
}

/// Build a map from each body node id → its owning loop node id.
fn build_body_owner_map(def: &WorkflowDefinition) -> HashMap<String, String> {
    let mut owner = HashMap::new();
    for (loop_id, node) in &def.nodes {
        if let WorkflowNode::Loop(loop_node) = node {
            for body_id in &loop_node.body {
                owner.insert(body_id.clone(), loop_id.clone());
            }
        }
    }
    owner
}

/// Validate loop definitions per Task 8.3 rules:
/// 1. body nodes must exist
/// 2. body nodes cannot be loops (no nested loops)
/// 3. terminate.node must exist, be in body, be a human-gated Decision, and
///    terminate.via must be "humanGate"; all Decisions must have a loop owner
/// 4. each loop body can have at most one Decision (the terminate.node)
/// 5. body external deps must appear in loop.depends
/// 6. external nodes cannot depend on loop body nodes
/// 7. output.from must identify an output-producing body node
/// 8. sink loops must declare output.from
fn validate_loop_definitions(def: &WorkflowDefinition) -> Result<()> {
    let body_owner = build_body_owner_map(def);

    // Track which Decision node is owned by which loop (via terminate.node)
    let mut decision_loop_owner: HashMap<String, String> = HashMap::new();

    for (loop_id, node) in &def.nodes {
        let loop_node = match node {
            WorkflowNode::Loop(ln) => ln,
            _ => continue,
        };

        // Rule 1: body nodes must exist in workflow nodes
        for body_id in &loop_node.body {
            if !def.nodes.contains_key(body_id) {
                anyhow::bail!(
                    "loop '{}' body node '{}' not found in workflow nodes",
                    loop_id,
                    body_id
                );
            }
        }

        // Rule 2: body nodes cannot be Loop (no nested loops)
        for body_id in &loop_node.body {
            if matches!(def.nodes.get(body_id), Some(WorkflowNode::Loop(_))) {
                anyhow::bail!(
                    "loop '{}' body node '{}' cannot be a Loop node (nested loops are not supported)",
                    loop_id,
                    body_id
                );
            }
        }

        // Rule 3a: terminate.node must exist
        let term_node_id = &loop_node.terminate.node;
        if !def.nodes.contains_key(term_node_id) {
            anyhow::bail!(
                "loop '{}' terminate.node '{}' not found in workflow nodes",
                loop_id,
                term_node_id
            );
        }

        // Rule 3a: terminate.node must be in the loop body
        if !loop_node.body.contains(term_node_id) {
            anyhow::bail!(
                "loop '{}' terminate.node '{}' must be in the loop body",
                loop_id,
                term_node_id
            );
        }

        // Rule 3a: terminate.node must be a Decision node. Gate details are
        // checked after structural validation to preserve useful diagnostics.
        match def.nodes.get(term_node_id) {
            Some(WorkflowNode::Decision(_)) => {
                // Each Decision can belong to at most one loop
                if let Some(existing_owner) =
                    decision_loop_owner.insert(term_node_id.clone(), loop_id.clone())
                {
                    anyhow::bail!(
                        "Decision node '{}' is used as terminate.node by multiple loops: '{}' and '{}'",
                        term_node_id,
                        existing_owner,
                        loop_id
                    );
                }
            }
            _ => {
                anyhow::bail!(
                    "loop '{}' terminate.node '{}' must be a Decision node, got {:?}",
                    loop_id,
                    term_node_id,
                    std::mem::discriminant(def.nodes.get(term_node_id).unwrap())
                );
            }
        }

        // Rule 4: each loop body can have at most one Decision (the terminate.node)
        for body_id in &loop_node.body {
            if body_id != term_node_id
                && matches!(def.nodes.get(body_id), Some(WorkflowNode::Decision(_)))
            {
                anyhow::bail!(
                    "loop '{}' has Decision node '{}' in body that is not the terminate.node; \
                     each loop body can have at most one Decision node (which must be the terminate.node)",
                    loop_id,
                    body_id
                );
            }
        }

        // Rule 5: body external deps must be declared in loop.depends
        let loop_depends: BTreeSet<&str> = loop_node
            .base
            .depends
            .as_ref()
            .map(|v| v.iter().map(String::as_str).collect())
            .unwrap_or_default();
        for body_id in &loop_node.body {
            if let Some(body_node) = def.nodes.get(body_id) {
                for dep in node_depends(body_node) {
                    if !loop_node.body.contains(dep) {
                        // dep is external to the loop body
                        if !loop_depends.contains(dep.as_str()) {
                            anyhow::bail!(
                                "loop '{}' body node '{}' depends on external node '{}'; \
                                 all external dependencies of body nodes must be declared in the loop's depends",
                                loop_id,
                                body_id,
                                dep
                            );
                        }
                    }
                }
            }
        }
    }

    // Rule 3b: all Decision nodes must be owned by some loop (no standalone Decision)
    for (node_id, node) in &def.nodes {
        if matches!(node, WorkflowNode::Decision(_)) {
            if !decision_loop_owner.contains_key(node_id) {
                anyhow::bail!(
                    "Decision node '{}' is standalone; Decision nodes must be used as a loop's terminate.node and reside in that loop's body",
                    node_id
                );
            }
        }
    }

    // Rule 6: non-body nodes (including loop nodes themselves) cannot depend on
    // loop body nodes.  External nodes that need a loop's result must depend on
    // the loop node itself, not on individual body nodes.
    for (node_id, node) in &def.nodes {
        if body_owner.contains_key(node_id) {
            continue; // skip body nodes themselves
        }
        if matches!(node, WorkflowNode::Decision(_)) {
            continue; // Decision nodes not owned by a loop are already rejected by rule 3b;
            // owned Decision nodes are body nodes (skip above)
        }
        for dep in node_depends(node) {
            if body_owner.contains_key(dep) {
                let owner = body_owner.get(dep).unwrap();
                anyhow::bail!(
                    "node '{}' depends on loop body node '{}'; \
                     nodes must depend on the loop node '{}' instead of its body node",
                    node_id,
                    dep,
                    owner
                );
            }
        }
    }

    // Rule 7: every declared output must come from an output-producing body node.
    for (loop_id, node) in &def.nodes {
        let WorkflowNode::Loop(loop_node) = node else {
            continue;
        };
        let Some(output) = &loop_node.output else {
            continue;
        };
        if !loop_node.body.contains(&output.from) {
            anyhow::bail!(
                "loop '{}' output.from '{}' must identify a node in the loop body",
                loop_id,
                output.from
            );
        }
        if !matches!(
            def.nodes.get(&output.from),
            Some(WorkflowNode::Subagent(_) | WorkflowNode::HostExecutor(_))
        ) {
            anyhow::bail!(
                "loop '{}' output.from '{}' must identify an output-producing Subagent or HostExecutor body node",
                loop_id,
                output.from
            );
        }
    }

    // Rule 8: sink loops must declare output.from
    // A loop is a "sink" if no non-body, non-decision node depends on it.
    let sinks = find_non_body_sinks(def, &body_owner);
    for sink_id in &sinks {
        if let Some(WorkflowNode::Loop(loop_node)) = def.nodes.get(sink_id) {
            if loop_node.output.is_none() {
                anyhow::bail!(
                    "sink loop '{}' must declare output.from (the loop is not depended on by any external node)",
                    sink_id
                );
            }
        }
    }

    // Loop termination is only implemented through a human gate.
    for (loop_id, node) in &def.nodes {
        let WorkflowNode::Loop(loop_node) = node else {
            continue;
        };
        if loop_node.terminate.via != "humanGate" {
            anyhow::bail!(
                "loop '{}' terminate.via must be 'humanGate', got '{}'",
                loop_id,
                loop_node.terminate.via
            );
        }
        let decision = match def.nodes.get(&loop_node.terminate.node) {
            Some(WorkflowNode::Decision(decision)) => decision,
            _ => continue,
        };
        if decision.base.human_gate.is_none() {
            anyhow::bail!(
                "loop '{}' terminate.node '{}' must declare humanGate",
                loop_id,
                loop_node.terminate.node
            );
        }
    }

    Ok(())
}

/// Find sink nodes excluding body nodes and Decision nodes.
/// A node is a sink if no non-body, non-decision node depends on it.
fn find_non_body_sinks(
    def: &WorkflowDefinition,
    body_owner: &HashMap<String, String>,
) -> Vec<String> {
    let mut referenced: BTreeSet<String> = BTreeSet::new();
    for (node_id, node) in &def.nodes {
        if body_owner.contains_key(node_id) {
            continue;
        }
        if matches!(node, WorkflowNode::Decision(_)) {
            continue;
        }
        for dep in node_depends(node) {
            referenced.insert(dep.clone());
        }
    }
    def.nodes
        .iter()
        .filter_map(|(node_id, node)| {
            if body_owner.contains_key(node_id) {
                return None;
            }
            if matches!(node, WorkflowNode::Decision(_)) {
                return None;
            }
            (!referenced.contains(node_id)).then_some(node_id.clone())
        })
        .collect()
}

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

    #[test]
    fn parse_workflow_definition_accepts_basic_graph() {
        let def = parse_workflow_definition(
            r#"{
                "workflowId":"flow-a",
                "version":1,
                "nodes":{
                    "a":{"type":"subagent","bot":"bot-a","prompt":"hi"},
                    "b":{"type":"hostExecutor","executor":"feishu-send","input":1,"depends":["a"],"humanGate":{"stage":"before","prompt":"approve?"}}
                }
            }"#,
        )
        .expect("definition");
        assert_eq!(def.workflow_id, "flow-a");
        assert_eq!(def.nodes.len(), 2);
    }

    // -- Task 1.1: node id validation --

    #[test]
    fn reject_node_id_with_slash() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"node/a":{"type":"subagent","bot":"b","prompt":"p"}}}"#,
        )
        .unwrap_err();
        assert!(err.to_string().contains("nodeId"), "got: {err}");
    }

    #[test]
    fn reject_node_id_with_space() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"node a":{"type":"subagent","bot":"b","prompt":"p"}}}"#,
        )
        .unwrap_err();
        assert!(err.to_string().contains("nodeId"), "got: {err}");
    }

    #[test]
    fn reject_node_id_dotdot() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"..":{"type":"subagent","bot":"b","prompt":"p"}}}"#,
        )
        .unwrap_err();
        assert!(err.to_string().contains("nodeId"), "got: {err}");
    }

    #[test]
    fn reject_node_id_containing_dotdot() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"a..b":{"type":"subagent","bot":"b","prompt":"p"}}}"#,
        )
        .unwrap_err();
        assert!(err.to_string().contains("nodeId"), "got: {err}");
    }

    #[test]
    fn reject_empty_node_id() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"":{"type":"subagent","bot":"b","prompt":"p"}}}"#,
        )
        .unwrap_err();
        assert!(err.to_string().contains("nodeId"), "got: {err}");
    }

    #[test]
    fn accept_node_id_with_dash() {
        let def = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"node-a":{"type":"subagent","bot":"b","prompt":"p"}}}"#,
        )
        .expect("dash ok");
        assert!(def.nodes.contains_key("node-a"));
    }

    #[test]
    fn accept_node_id_with_underscore() {
        let def = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"node_a":{"type":"subagent","bot":"b","prompt":"p"}}}"#,
        )
        .expect("underscore ok");
        assert!(def.nodes.contains_key("node_a"));
    }

    #[test]
    fn accept_node_id_with_dot() {
        let def = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"node.a":{"type":"subagent","bot":"b","prompt":"p"}}}"#,
        )
        .expect("dot ok");
        assert!(def.nodes.contains_key("node.a"));
    }

    // -- Task 1.2: side-effect executor gate validation --

    #[test]
    fn reject_ungated_feishu_send() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"a":{"type":"hostExecutor","executor":"feishu-send","input":1}}}"#,
        )
        .unwrap_err();
        assert!(
            err.to_string().contains("humanGate") || err.to_string().contains("side-effect"),
            "got: {err}"
        );
    }

    #[test]
    fn reject_ungated_feishu_reply() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"a":{"type":"hostExecutor","executor":"feishu-reply","input":1}}}"#,
        )
        .unwrap_err();
        assert!(
            err.to_string().contains("humanGate") || err.to_string().contains("side-effect"),
            "got: {err}"
        );
    }

    #[test]
    fn reject_ungated_beam_schedule() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"a":{"type":"hostExecutor","executor":"beam-schedule","input":1}}}"#,
        )
        .unwrap_err();
        assert!(
            err.to_string().contains("humanGate") || err.to_string().contains("side-effect"),
            "got: {err}"
        );
    }

    #[test]
    fn accept_gated_feishu_send() {
        let def = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"a":{"type":"hostExecutor","executor":"feishu-send","input":1,"humanGate":{"stage":"before","prompt":"ok?"}}}}"#,
        )
        .expect("gated feishu-send ok");
        assert!(def.nodes.contains_key("a"));
    }

    #[test]
    fn accept_unsafe_allow_ungated_feishu_send() {
        let def = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"a":{"type":"hostExecutor","executor":"feishu-send","input":1,"unsafeAllowUngated":true}}}"#,
        )
        .expect("unsafeAllowUngated ok");
        assert!(def.nodes.contains_key("a"));
    }

    #[test]
    fn accept_ungated_non_side_effect_executor() {
        let def = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"a":{"type":"hostExecutor","executor":"custom-tool","input":1}}}"#,
        )
        .expect("non-side-effect ok");
        assert!(def.nodes.contains_key("a"));
    }

    // -- Task 8.2: loop nodes are now accepted (validation deferred to Task 8.3) --

    #[test]
    fn accept_loop_node_with_minimal_body() {
        // Sink loops need both an output-producing body node and a gated decision.
        let def = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"work":{"type":"subagent","bot":"b","prompt":"p"},"d":{"type":"decision","humanGate":{"stage":"approve","prompt":"continue?"},"depends":["work"]},"l":{"type":"loop","maxIterations":3,"body":["work","d"],"terminate":{"node":"d","via":"humanGate"},"output":{"from":"work"}}}}"#,
        )
        .expect("loop accepted");
        assert!(def.nodes.contains_key("l"));
    }

    #[test]
    fn reject_standalone_decision_node() {
        // Task 8.3: standalone Decision nodes (not used as a loop terminate.node) are rejected
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"d":{"type":"decision"}}}"#,
        )
        .unwrap_err();
        assert!(
            err.to_string().contains("standalone"),
            "expected 'standalone' in error, got: {err}"
        );
    }

    #[test]
    fn ordinary_dag_workflow_accepted() {
        let def = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"a":{"type":"subagent","bot":"b","prompt":"p"},"c":{"type":"subagent","bot":"b","prompt":"q","depends":["a"]}}}"#,
        )
        .expect("ordinary DAG ok");
        assert_eq!(def.nodes.len(), 2);
    }

    // -- Task 8.2: code-review-loop workflow now parses successfully --
    #[test]
    fn accept_code_review_loop_workflow_json() {
        let raw = include_str!("../../../workflows/code-review-loop.workflow.json");
        let def = parse_workflow_definition(raw).expect("code-review-loop parsed");
        assert!(def.nodes.contains_key("review-loop"));
        assert!(def.nodes.contains_key("implement"));
        assert!(def.nodes.contains_key("review"));
        assert!(def.nodes.contains_key("reviewDecision"));
    }

    // -- Task 9.2: subagent-approval-feishu-send example parses --
    #[test]
    fn accept_subagent_approval_feishu_send_workflow_json() {
        let raw = include_str!("../../../workflows/subagent-approval-feishu-send.workflow.json");
        let def = parse_workflow_definition(raw).expect("subagent-approval-feishu-send parsed");
        assert_eq!(def.workflow_id, "subagent-approval-feishu-send");
        assert_eq!(def.nodes.len(), 2);
        assert!(def.nodes.contains_key("draft"));
        assert!(def.nodes.contains_key("send"));
        // send depends on draft
        let send_node = def.nodes.get("send").unwrap();
        match send_node {
            // send is a gated feishu-send: humanGate present, unsafeAllowUngated absent
            WorkflowNode::HostExecutor(n) => {
                assert_eq!(n.executor, "feishu-send");
                assert_eq!(
                    n.base.depends.as_deref(),
                    Some(vec!["draft".to_string()].as_slice())
                );
                assert!(n.base.human_gate.is_some(), "send must have humanGate");
                assert!(
                    !n.base.unsafe_allow_ungated.unwrap_or(false),
                    "send must NOT use unsafeAllowUngated"
                );
            }
            _ => panic!("expected hostExecutor"),
        }
    }

    // -- Task 8.3: loop definition validation --

    // Rule 1: body nodes must exist
    #[test]
    fn reject_loop_with_missing_body_node() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"d":{"type":"decision"},"l":{"type":"loop","maxIterations":3,"body":["missing","d"],"terminate":{"node":"d","via":"humanGate"},"output":{"from":"d"}}}}"#,
        )
        .unwrap_err();
        assert!(
            err.to_string().contains("body node") && err.to_string().contains("missing"),
            "expected body node 'missing' error, got: {err}"
        );
    }

    // Rule 2: body node cannot be a loop (no nested loops)
    #[test]
    fn reject_loop_with_nested_loop_body() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"d":{"type":"decision"},"inner":{"type":"loop","maxIterations":2,"body":["d"],"terminate":{"node":"d","via":"humanGate"},"output":{"from":"d"}},"l":{"type":"loop","maxIterations":3,"body":["inner","d"],"terminate":{"node":"d","via":"humanGate"},"output":{"from":"d"}}}}"#,
        )
        .unwrap_err();
        assert!(
            err.to_string().contains("nested loop") || err.to_string().contains("cannot be a Loop"),
            "expected nested loop error, got: {err}"
        );
    }

    // Rule 3a: terminate.node must exist
    #[test]
    fn reject_loop_with_missing_terminate_node() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"d":{"type":"decision"},"l":{"type":"loop","maxIterations":3,"body":["d"],"terminate":{"node":"nonexistent","via":"humanGate"},"output":{"from":"d"}}}}"#,
        )
        .unwrap_err();
        assert!(
            err.to_string().contains("terminate.node") && err.to_string().contains("nonexistent"),
            "expected terminate.node not found error, got: {err}"
        );
    }

    // Rule 3a: terminate.node must be in loop body
    #[test]
    fn reject_loop_with_terminate_node_not_in_body() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"d":{"type":"decision"},"x":{"type":"subagent","bot":"b","prompt":"p"},"l":{"type":"loop","maxIterations":3,"body":["d"],"terminate":{"node":"x","via":"humanGate"}}}}"#,
        )
        .unwrap_err();
        assert!(
            err.to_string().contains("terminate.node") && err.to_string().contains("body"),
            "expected terminate.node must be in body error, got: {err}"
        );
    }

    // Rule 3a: terminate.node must be a Decision node
    #[test]
    fn reject_loop_with_non_decision_terminate_node() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"a":{"type":"subagent","bot":"b","prompt":"p"},"l":{"type":"loop","maxIterations":3,"body":["a"],"terminate":{"node":"a","via":"humanGate"}}}}"#,
        )
        .unwrap_err();
        assert!(
            err.to_string().contains("Decision node"),
            "expected Decision node error, got: {err}"
        );
    }

    #[test]
    fn reject_loop_with_ungated_decision_terminate_node() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"work":{"type":"subagent","bot":"b","prompt":"p"},"d":{"type":"decision","depends":["work"]},"l":{"type":"loop","maxIterations":3,"body":["work","d"],"terminate":{"node":"d","via":"humanGate"},"output":{"from":"work"}}}}"#,
        )
        .unwrap_err();
        assert!(err.to_string().contains("humanGate"), "got: {err}");
    }

    #[test]
    fn reject_loop_with_non_human_gate_terminate_via() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"work":{"type":"subagent","bot":"b","prompt":"p"},"d":{"type":"decision","humanGate":{"stage":"approve","prompt":"continue?"},"depends":["work"]},"l":{"type":"loop","maxIterations":3,"body":["work","d"],"terminate":{"node":"d","via":"automatic"},"output":{"from":"work"}}}}"#,
        )
        .unwrap_err();
        assert!(
            err.to_string().contains("terminate.via") && err.to_string().contains("humanGate"),
            "got: {err}"
        );
    }

    // Rule 3b: Decision node cannot be standalone
    #[test]
    fn reject_decision_not_used_by_any_loop() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"a":{"type":"subagent","bot":"b","prompt":"p"},"d":{"type":"decision"}}}"#,
        )
        .unwrap_err();
        assert!(
            err.to_string().contains("standalone"),
            "expected standalone Decision error, got: {err}"
        );
    }

    // Rule 3b: Decision node must be in body of the loop that uses it
    #[test]
    fn reject_decision_outside_loop_body() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"d":{"type":"decision"},"l":{"type":"loop","maxIterations":3,"body":[],"terminate":{"node":"d","via":"humanGate"}}}}"#,
        )
        .unwrap_err();
        assert!(
            err.to_string().contains("body"),
            "expected body error for terminate.node not in body, got: {err}"
        );
    }

    // Rule 3b: same Decision cannot be used by multiple loops
    #[test]
    fn reject_decision_used_by_multiple_loops() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"d":{"type":"decision"},"l1":{"type":"loop","maxIterations":3,"body":["d"],"terminate":{"node":"d","via":"humanGate"},"output":{"from":"d"}},"l2":{"type":"loop","maxIterations":3,"body":["d"],"terminate":{"node":"d","via":"humanGate"},"output":{"from":"d"}}}}"#,
        )
        .unwrap_err();
        assert!(
            err.to_string().contains("multiple loops"),
            "expected 'multiple loops' error, got: {err}"
        );
    }

    // Rule 4: each loop body can have at most one Decision (the terminate.node)
    #[test]
    fn reject_loop_with_extra_decision_in_body() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"d1":{"type":"decision"},"d2":{"type":"decision"},"l":{"type":"loop","maxIterations":3,"body":["d1","d2"],"terminate":{"node":"d1","via":"humanGate"},"output":{"from":"d1"}}}}"#,
        )
        .unwrap_err();
        assert!(
            err.to_string().contains("at most one Decision"),
            "expected 'at most one Decision' error, got: {err}"
        );
    }

    // Rule 5: body external deps must appear in loop.depends
    #[test]
    fn reject_body_node_with_undeclared_external_dependency() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"ext":{"type":"subagent","bot":"b","prompt":"p"},"d":{"type":"decision"},"inner":{"type":"subagent","bot":"b","prompt":"p","depends":["ext"]},"l":{"type":"loop","maxIterations":3,"body":["inner","d"],"terminate":{"node":"d","via":"humanGate"},"output":{"from":"inner"}}}}"#,
        )
        .unwrap_err();
        assert!(
            err.to_string().contains("external") && err.to_string().contains("depends"),
            "expected external dep must be declared in loop.depends error, got: {err}"
        );
    }

    #[test]
    fn accept_body_node_with_external_dependency_declared_in_loop_depends() {
        let def = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"ext":{"type":"subagent","bot":"b","prompt":"p"},"d":{"type":"decision","humanGate":{"stage":"approve","prompt":"continue?"},"depends":["inner"]},"inner":{"type":"subagent","bot":"b","prompt":"p","depends":["ext"]},"l":{"type":"loop","maxIterations":3,"body":["inner","d"],"depends":["ext"],"terminate":{"node":"d","via":"humanGate"},"output":{"from":"inner"}}}}"#,
        )
        .expect("loop with declared external dep accepted");
        assert!(def.nodes.contains_key("l"));
    }

    // Rule 6: external nodes cannot depend on loop body nodes
    #[test]
    fn reject_external_node_depending_on_loop_body_node() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"d":{"type":"decision"},"inner":{"type":"subagent","bot":"b","prompt":"p"},"ext":{"type":"subagent","bot":"b","prompt":"p","depends":["inner"]},"l":{"type":"loop","maxIterations":3,"body":["inner","d"],"terminate":{"node":"d","via":"humanGate"},"output":{"from":"inner"}}}}"#,
        )
        .unwrap_err();
        assert!(
            err.to_string().contains("depends on loop body node"),
            "expected depends on loop body node error, got: {err}"
        );
    }

    #[test]
    fn accept_external_node_depending_on_loop_node_instead_of_body_node() {
        let def = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"d":{"type":"decision","humanGate":{"stage":"approve","prompt":"continue?"},"depends":["inner"]},"inner":{"type":"subagent","bot":"b","prompt":"p"},"ext":{"type":"subagent","bot":"b","prompt":"p","depends":["l"]},"l":{"type":"loop","maxIterations":3,"body":["inner","d"],"terminate":{"node":"d","via":"humanGate"},"output":{"from":"inner"}}}}"#,
        )
        .expect("external node depends on loop node accepted");
        assert!(def.nodes.contains_key("l"));
    }

    // Rule 6 (cont): loop node itself must not depend on body nodes
    // (either its own body or another loop's body)
    #[test]
    fn reject_loop_depends_on_own_body_node() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"d":{"type":"decision"},"inner":{"type":"subagent","bot":"b","prompt":"p"},"l":{"type":"loop","maxIterations":3,"body":["inner","d"],"depends":["inner"],"terminate":{"node":"d","via":"humanGate"},"output":{"from":"inner"}}}}"#,
        )
        .unwrap_err();
        assert!(
            err.to_string().contains("body node"),
            "expected 'body node' error for loop depending on own body node, got: {err}"
        );
    }

    #[test]
    fn reject_loop_depends_on_another_loop_body_node() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"d1":{"type":"decision"},"a":{"type":"subagent","bot":"b","prompt":"p"},"l1":{"type":"loop","maxIterations":3,"body":["a","d1"],"terminate":{"node":"d1","via":"humanGate"},"output":{"from":"a"}},"d2":{"type":"decision"},"l2":{"type":"loop","maxIterations":3,"body":["d2"],"depends":["a"],"terminate":{"node":"d2","via":"humanGate"},"output":{"from":"d2"}}}}"#,
        )
        .unwrap_err();
        assert!(
            err.to_string().contains("body node") && err.to_string().contains("'l2'"),
            "expected 'body node' error for l2 depending on another loop's body node, got: {err}"
        );
    }

    // Also verify that loop depends on another loop (not body node) is still accepted
    #[test]
    fn accept_loop_depends_on_another_loop_node() {
        let def = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"d1":{"type":"decision","humanGate":{"stage":"approve","prompt":"continue?"},"depends":["a"]},"a":{"type":"subagent","bot":"b","prompt":"p"},"l1":{"type":"loop","maxIterations":3,"body":["a","d1"],"terminate":{"node":"d1","via":"humanGate"},"output":{"from":"a"}},"b":{"type":"subagent","bot":"b","prompt":"q"},"d2":{"type":"decision","humanGate":{"stage":"approve","prompt":"continue?"},"depends":["b"]},"l2":{"type":"loop","maxIterations":3,"body":["b","d2"],"depends":["l1"],"terminate":{"node":"d2","via":"humanGate"},"output":{"from":"b"}}}}"#,
        )
        .expect("loop depends on another loop node accepted");
        assert!(def.nodes.contains_key("l1"));
        assert!(def.nodes.contains_key("l2"));
    }

    // Rule 8: sink loop must declare output.from
    #[test]
    fn reject_sink_loop_without_output_from() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"d":{"type":"decision"},"inner":{"type":"subagent","bot":"b","prompt":"p"},"l":{"type":"loop","maxIterations":3,"body":["inner","d"],"terminate":{"node":"d","via":"humanGate"}}}}"#,
        )
        .unwrap_err();
        assert!(
            err.to_string().contains("output.from"),
            "expected 'output.from' error for sink loop, got: {err}"
        );
    }

    #[test]
    fn accept_sink_loop_with_output_from() {
        let def = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"d":{"type":"decision","humanGate":{"stage":"approve","prompt":"continue?"},"depends":["inner"]},"inner":{"type":"subagent","bot":"b","prompt":"p"},"l":{"type":"loop","maxIterations":3,"body":["inner","d"],"terminate":{"node":"d","via":"humanGate"},"output":{"from":"inner"}}}}"#,
        )
        .expect("sink loop with output.from accepted");
        assert!(def.nodes.contains_key("l"));
    }

    #[test]
    fn reject_sink_loop_with_unknown_output_from() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"inner":{"type":"subagent","bot":"b","prompt":"p"},"d":{"type":"decision","humanGate":{"stage":"approve","prompt":"continue?"},"depends":["inner"]},"l":{"type":"loop","maxIterations":3,"body":["inner","d"],"terminate":{"node":"d","via":"humanGate"},"output":{"from":"missing"}}}}"#,
        )
        .unwrap_err();
        assert!(err.to_string().contains("loop body"), "got: {err}");
    }

    #[test]
    fn reject_sink_loop_with_external_output_from() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"external":{"type":"subagent","bot":"b","prompt":"p"},"inner":{"type":"subagent","bot":"b","prompt":"p"},"d":{"type":"decision","humanGate":{"stage":"approve","prompt":"continue?"},"depends":["inner"]},"l":{"type":"loop","maxIterations":3,"body":["inner","d"],"terminate":{"node":"d","via":"humanGate"},"output":{"from":"external"}}}}"#,
        )
        .unwrap_err();
        assert!(err.to_string().contains("loop body"), "got: {err}");
    }

    #[test]
    fn reject_sink_loop_with_decision_output_from() {
        let err = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"inner":{"type":"subagent","bot":"b","prompt":"p"},"d":{"type":"decision","humanGate":{"stage":"approve","prompt":"continue?"},"depends":["inner"]},"l":{"type":"loop","maxIterations":3,"body":["inner","d"],"terminate":{"node":"d","via":"humanGate"},"output":{"from":"d"}}}}"#,
        )
        .unwrap_err();
        assert!(err.to_string().contains("output-producing"), "got: {err}");
    }

    #[test]
    fn accept_non_sink_loop_without_output_from() {
        // Loop is not a sink because external node 'ext' depends on it
        let def = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"d":{"type":"decision","humanGate":{"stage":"approve","prompt":"continue?"},"depends":["inner"]},"inner":{"type":"subagent","bot":"b","prompt":"p"},"l":{"type":"loop","maxIterations":3,"body":["inner","d"],"terminate":{"node":"d","via":"humanGate"}},"ext":{"type":"subagent","bot":"b","prompt":"p","depends":["l"]}}}"#,
        )
        .expect("non-sink loop without output.from accepted");
        assert!(def.nodes.contains_key("l"));
    }

    // Edge case: loop with depends on external node but no body external deps (valid)
    #[test]
    fn accept_loop_with_explicit_depends_and_no_body_external_deps() {
        let def = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"ext":{"type":"subagent","bot":"b","prompt":"p"},"d":{"type":"decision","humanGate":{"stage":"approve","prompt":"continue?"},"depends":["inner"]},"inner":{"type":"subagent","bot":"b","prompt":"p"},"l":{"type":"loop","maxIterations":3,"body":["inner","d"],"depends":["ext"],"terminate":{"node":"d","via":"humanGate"},"output":{"from":"inner"}}}}"#,
        )
        .expect("loop with explicit depends accepted");
        assert!(def.nodes.contains_key("l"));
    }

    // Regression: multiple loops in same workflow
    #[test]
    fn accept_two_independent_loops() {
        let def = parse_workflow_definition(
            r#"{"workflowId":"f","version":1,"nodes":{"d1":{"type":"decision","humanGate":{"stage":"approve","prompt":"continue?"},"depends":["inner1"]},"inner1":{"type":"subagent","bot":"b","prompt":"p"},"l1":{"type":"loop","maxIterations":3,"body":["inner1","d1"],"terminate":{"node":"d1","via":"humanGate"},"output":{"from":"inner1"}},"d2":{"type":"decision","humanGate":{"stage":"approve","prompt":"continue?"},"depends":["inner2"]},"inner2":{"type":"subagent","bot":"b","prompt":"p"},"l2":{"type":"loop","maxIterations":3,"body":["inner2","d2"],"depends":["l1"],"terminate":{"node":"d2","via":"humanGate"},"output":{"from":"inner2"}}}}"#,
        )
        .expect("two independent loops accepted");
        assert!(def.nodes.contains_key("l1"));
        assert!(def.nodes.contains_key("l2"));
    }
}