greentic_flow/

lib.rs

//! Downstream runtimes must set the current tenant telemetry context via
//! `greentic_types::telemetry::set_current_tenant_ctx` before executing flows
//! (for example, prior to `FlowEngine::run` in the host runner).
#![deny(unsafe_code)]
#![allow(clippy::result_large_err)]

pub mod add_step;
pub mod answers;
pub mod cache;
pub mod component_catalog;
pub mod component_schema;
pub mod component_setup;
pub mod config_flow;
pub mod contracts;
pub mod error;
pub mod flow_bundle;
pub mod flow_ir;
pub mod flow_meta;
pub mod i18n;
pub mod info;
pub mod ir;
pub mod json_output;
pub mod lint;
pub mod loader;
pub mod model;
pub mod path_safety;
pub mod qa_runner;
pub mod questions;
pub mod questions_schema;
pub mod registry;
pub mod resolve;
pub mod resolve_summary;
pub mod schema_mode;
pub mod schema_validate;
pub mod splice;
pub mod template;
pub mod util;
pub mod wizard;
pub mod wizard_ops;
pub mod wizard_state;

pub use flow_bundle::{
    ComponentPin, FlowBundle, NodeRef, blake3_hex, canonicalize_json, extract_component_pins,
    load_and_validate_bundle, load_and_validate_bundle_with_flow,
};
pub use json_output::{JsonDiagnostic, LintJsonOutput, lint_to_stdout_json};
pub use splice::{NEXT_NODE_PLACEHOLDER, splice_node_after};

/// Metadata key under which compiled flows expose their flow-level slot schema.
pub const SLOT_SCHEMA_METADATA_KEY: &str = "greentic.slot_schema";

use crate::{error::Result, model::FlowDoc};
use greentic_types::{
    ComponentId, Flow, FlowComponentRef, FlowId, FlowKind, FlowMetadata, InputMapping, Node,
    NodeId, OutputMapping, Routing, TelemetryHints, flow::FlowHasher,
};
use indexmap::IndexMap;
use serde_json::Value;
use std::collections::{BTreeMap, BTreeSet, HashSet};
use std::path::Path;

/// Map a YAML flow type string to [`FlowKind`].
pub fn map_flow_type(flow_type: &str) -> Result<FlowKind> {
    match flow_type {
        "messaging" => Ok(FlowKind::Messaging),
        "event" | "events" => Ok(FlowKind::Event),
        "component-config" => Ok(FlowKind::ComponentConfig),
        "job" => Ok(FlowKind::Job),
        "http" => Ok(FlowKind::Http),
        other => Err(crate::error::FlowError::UnknownFlowType {
            flow_type: other.to_string(),
            location: crate::error::FlowErrorLocation::at_path("type"),
        }),
    }
}

/// Compile a validated [`FlowDoc`] into the canonical [`Flow`] model.
pub fn compile_flow(doc: FlowDoc) -> Result<Flow> {
    let FlowDoc {
        id,
        title,
        description,
        flow_type,
        start,
        parameters,
        tags,
        schema_version,
        mut entrypoints,
        meta: _,
        slot_schema,
        nodes: node_docs,
    } = doc;

    let kind = map_flow_type(&flow_type)?;
    let known_nodes: HashSet<String> = node_docs.keys().cloned().collect();
    if let Some(entry) = start
        .clone()
        .or_else(|| known_nodes.contains("in").then(|| "in".to_string()))
        .or_else(|| node_docs.keys().next().cloned())
    {
        entrypoints
            .entry("default".to_string())
            .or_insert_with(|| Value::String(entry));
    }

    let mut nodes: IndexMap<NodeId, Node, FlowHasher> = IndexMap::default();
    for (node_id_str, node_doc) in node_docs {
        let node_id = NodeId::new(node_id_str.as_str()).map_err(|e| {
            crate::error::FlowError::InvalidIdentifier {
                kind: "node",
                value: node_id_str.clone(),
                detail: e.to_string(),
                location: crate::error::FlowErrorLocation::at_path(format!("nodes.{node_id_str}")),
            }
        })?;
        let routing = compile_routing(&node_doc.routing, &known_nodes, node_id_str.as_str())?;
        let telemetry = node_doc
            .telemetry
            .map(|t| TelemetryHints {
                span_name: t.span_name,
                attributes: t.attributes,
                sampling: t.sampling,
            })
            .unwrap_or_default();
        let mut op_key: Option<String> = None;
        let mut op_sibling: Option<String> = None;
        let mut payload: Option<Value> = None;
        let mut input_mapping: Option<Value> = None;
        let mut output_mapping: Option<Value> = None;
        let mut err_mapping: Option<Value> = None;
        for (k, v) in node_doc.raw {
            match k.as_str() {
                "in_map" => {
                    input_mapping = Some(v);
                    continue;
                }
                "out_map" | "output" => {
                    output_mapping = Some(v);
                    continue;
                }
                "err_map" => {
                    err_mapping = Some(v);
                    continue;
                }
                "operation" => {
                    op_sibling = v.as_str().map(str::to_string);
                    continue;
                }
                _ => {}
            }
            op_key = Some(k);
            payload = Some(v);
        }
        let operation = op_key.ok_or_else(|| crate::error::FlowError::Internal {
            message: format!("node '{node_id_str}' missing operation key"),
            location: crate::error::FlowErrorLocation::at_path(format!("nodes.{node_id_str}")),
        })?;
        let is_mcp = operation.as_str() == crate::ir::MCP_COMPONENT;
        let is_builtin =
            matches!(operation.as_str(), "questions" | "template") || operation.starts_with("dw.");
        let is_legacy = schema_version.unwrap_or(1) < 2;
        let (component_id, op_field) = if is_mcp {
            // MCP nodes lower to the literal `mcp` component. `server`, `tool`,
            // `arguments` and `output` stay in the node payload (carried via the
            // input mapping below), so the component string remains a valid
            // greentic_types::ComponentId and survives pack/runtime load.
            (crate::ir::MCP_COMPONENT.to_string(), op_sibling)
        } else if is_builtin || is_legacy {
            (operation, op_sibling)
        } else {
            (
                "component.exec".to_string(),
                Some(op_sibling.unwrap_or(operation)),
            )
        };
        let node = Node {
            id: node_id.clone(),
            component: FlowComponentRef {
                id: ComponentId::new(&component_id).unwrap(),
                pack_alias: None,
                operation: op_field,
            },
            input: InputMapping {
                mapping: input_mapping
                    .or(payload)
                    .unwrap_or_else(|| Value::Object(Default::default())),
            },
            output: OutputMapping {
                mapping: output_mapping.unwrap_or_else(|| Value::Object(Default::default())),
            },
            err_map: err_mapping.map(|mapping| OutputMapping { mapping }),
            routing,
            telemetry,
        };
        nodes.insert(node_id, node);
    }

    let flow_id =
        FlowId::new(id.as_str()).map_err(|e| crate::error::FlowError::InvalidIdentifier {
            kind: "flow",
            value: id.clone(),
            detail: e.to_string(),
            location: crate::error::FlowErrorLocation::at_path("id"),
        })?;

    let entrypoints_map: BTreeMap<String, Value> = entrypoints.into_iter().collect();

    let mut extra = parameters;
    if let Some(ss) = slot_schema {
        validate_slot_schema_names(&ss)?;
        match extra {
            Value::Object(ref mut map) => {
                map.insert(SLOT_SCHEMA_METADATA_KEY.to_string(), ss);
            }
            Value::Null => {
                let mut map = serde_json::Map::new();
                map.insert(SLOT_SCHEMA_METADATA_KEY.to_string(), ss);
                extra = Value::Object(map);
            }
            _ => {
                tracing::warn!(
                    flow_id = %id,
                    "slot_schema present but parameters is not an object; skipping forward into extra"
                );
            }
        }
    }

    Ok(Flow {
        schema_version: "flow-v1".to_string(),
        id: flow_id,
        kind,
        entrypoints: entrypoints_map,
        nodes,
        metadata: FlowMetadata {
            title,
            description,
            tags: tags.into_iter().collect::<BTreeSet<_>>(),
            extra,
        },
    })
}

/// Reject duplicate slot names in `slot_schema`. The JSON schema enforces
/// structure (types, required fields, allOf conditionals) but cannot check
/// name uniqueness across array items — `uniqueItems` compares whole objects,
/// not individual fields.
fn validate_slot_schema_names(value: &Value) -> Result<()> {
    use crate::error::{FlowError, FlowErrorLocation, SchemaErrorDetail};

    let Some(slots) = value.as_array() else {
        return Ok(()); // non-array is caught by the JSON schema
    };

    let mut seen_names: HashSet<&str> = HashSet::with_capacity(slots.len());

    for (i, slot) in slots.iter().enumerate() {
        let Some(name) = slot.get("name").and_then(Value::as_str) else {
            continue; // missing/empty name is caught by the JSON schema
        };
        if !seen_names.insert(name) {
            let path = format!("slot_schema[{i}]");
            let message = format!("{path}: duplicate slot name '{name}'");
            let loc_path = format!("{path}/name");
            return Err(FlowError::Schema {
                message: message.clone(),
                details: vec![SchemaErrorDetail {
                    message,
                    location: FlowErrorLocation::at_path(&loc_path),
                }],
                location: FlowErrorLocation::at_path(loc_path),
            });
        }

        // Reject patterns that are not valid regular expressions.
        if let Some(pat) = slot.get("pattern").and_then(Value::as_str)
            && let Err(e) = regex::Regex::new(pat)
        {
            let path = format!("slot_schema[{i}]");
            let message = format!("{path}: invalid regex pattern for slot '{name}': {e}");
            let loc_path = format!("{path}/pattern");
            return Err(FlowError::Schema {
                message: message.clone(),
                details: vec![SchemaErrorDetail {
                    message,
                    location: FlowErrorLocation::at_path(&loc_path),
                }],
                location: FlowErrorLocation::at_path(loc_path),
            });
        }
    }

    Ok(())
}

/// Compile YGTC YAML text into [`Flow`].
pub fn compile_ygtc_str(src: &str) -> Result<Flow> {
    let doc = loader::load_ygtc_from_str(src)?;
    compile_flow(doc)
}

/// Compile a YGTC file into [`Flow`].
pub fn compile_ygtc_file(path: &Path) -> Result<Flow> {
    let doc = loader::load_ygtc_from_path(path)?;
    compile_flow(doc)
}

fn compile_routing(raw: &Value, nodes: &HashSet<String>, node_id: &str) -> Result<Routing> {
    #[derive(serde::Deserialize)]
    struct RouteDoc {
        #[serde(default)]
        to: Option<String>,
        #[serde(default)]
        out: Option<bool>,
        #[serde(default)]
        status: Option<String>,
        #[serde(default)]
        reply: Option<bool>,
        #[serde(default)]
        condition: Option<String>,
    }

    let routes: Vec<RouteDoc> = if raw.is_null() {
        Vec::new()
    } else if let Some(shorthand) = raw.as_str() {
        match shorthand {
            "out" => vec![RouteDoc {
                to: None,
                out: Some(true),
                status: None,
                reply: None,
                condition: None,
            }],
            "reply" => vec![RouteDoc {
                to: None,
                out: None,
                status: None,
                reply: Some(true),
                condition: None,
            }],
            other => {
                return Err(crate::error::FlowError::Routing {
                    node_id: node_id.to_string(),
                    message: format!("invalid routing shorthand '{other}'"),
                    location: crate::error::FlowErrorLocation::at_path(format!(
                        "nodes.{node_id}.routing"
                    )),
                });
            }
        }
    } else {
        serde_json::from_value(raw.clone()).map_err(|e| crate::error::FlowError::Routing {
            node_id: node_id.to_string(),
            message: e.to_string(),
            location: crate::error::FlowErrorLocation::at_path(format!("nodes.{node_id}.routing")),
        })?
    };

    // Any route with a condition expression → preserve as Custom routing
    if routes.iter().any(|r| r.condition.is_some()) {
        // Validate all target nodes exist
        for route in &routes {
            if let Some(to) = &route.to
                && !nodes.contains(to)
            {
                return Err(crate::error::FlowError::MissingNode {
                    target: to.clone(),
                    node_id: node_id.to_string(),
                    location: crate::error::FlowErrorLocation::at_path(format!(
                        "nodes.{node_id}.routing"
                    )),
                });
            }
        }
        return Ok(Routing::Custom(raw.clone()));
    }

    if routes.len() == 1 {
        let route = &routes[0];
        let has_condition = raw
            .as_array()
            .and_then(|routes| routes.first())
            .and_then(Value::as_object)
            .is_some_and(|route| route.contains_key("condition"));
        let is_out = route.out.unwrap_or(false);
        if has_condition {
            return Ok(Routing::Custom(raw.clone()));
        }
        if route.reply.unwrap_or(false) {
            return Ok(Routing::Reply);
        }
        if route.status.is_some() {
            // A single status route is still conditional; preserve it as custom routing
            // instead of silently treating it as an unconditional next-hop.
            return Ok(Routing::Custom(raw.clone()));
        }
        if let Some(to) = &route.to {
            if to == "out" || is_out {
                return Ok(Routing::End);
            }
            if !nodes.contains(to) {
                return Err(crate::error::FlowError::MissingNode {
                    target: to.clone(),
                    node_id: node_id.to_string(),
                    location: crate::error::FlowErrorLocation::at_path(format!(
                        "nodes.{node_id}.routing"
                    )),
                });
            }
            return Ok(Routing::Next {
                node_id: NodeId::new(to.as_str()).map_err(|e| {
                    crate::error::FlowError::InvalidIdentifier {
                        kind: "node",
                        value: to.clone(),
                        detail: e.to_string(),
                        location: crate::error::FlowErrorLocation::at_path(format!(
                            "nodes.{node_id}.routing"
                        )),
                    }
                })?,
            });
        }
        if is_out {
            return Ok(Routing::End);
        }
    }

    if routes.is_empty() {
        return Ok(Routing::End);
    }

    // Attempt to build a Branch when multiple status routes are present.
    if routes.len() >= 2 {
        use std::collections::BTreeMap;
        let has_condition = raw.as_array().is_some_and(|routes| {
            routes.iter().any(|route| {
                route
                    .as_object()
                    .is_some_and(|route| route.contains_key("condition"))
            })
        });
        if has_condition {
            return Ok(Routing::Custom(raw.clone()));
        }
        let mut on_status: BTreeMap<String, NodeId> = BTreeMap::new();
        let mut default: Option<NodeId> = None;
        let mut any_status = false;
        for route in &routes {
            if route.reply.unwrap_or(false) || route.out.unwrap_or(false) {
                return Ok(Routing::Custom(raw.clone()));
            }
            let to = match &route.to {
                Some(t) => t,
                None => return Ok(Routing::Custom(raw.clone())),
            };
            if !nodes.contains(to) {
                return Err(crate::error::FlowError::MissingNode {
                    target: to.clone(),
                    node_id: node_id.to_string(),
                    location: crate::error::FlowErrorLocation::at_path(format!(
                        "nodes.{node_id}.routing"
                    )),
                });
            }
            let to_id = NodeId::new(to.as_str()).map_err(|e| {
                crate::error::FlowError::InvalidIdentifier {
                    kind: "node",
                    value: to.clone(),
                    detail: e.to_string(),
                    location: crate::error::FlowErrorLocation::at_path(format!(
                        "nodes.{node_id}.routing"
                    )),
                }
            })?;
            if let Some(status) = &route.status {
                any_status = true;
                on_status.insert(status.clone(), to_id);
            } else {
                default = Some(to_id);
            }
        }
        if any_status {
            return Ok(Routing::Branch { on_status, default });
        }
        if let Some(default) = default {
            return Ok(Routing::Branch {
                on_status,
                default: Some(default),
            });
        }
    }

    Ok(Routing::Custom(raw.clone()))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::loader::load_ygtc_from_str;
    use serde_json::json;
    use tempfile::tempdir;

    #[test]
    fn map_flow_type_supports_known_aliases() {
        assert_eq!(map_flow_type("messaging").unwrap(), FlowKind::Messaging);
        assert_eq!(map_flow_type("events").unwrap(), FlowKind::Event);
        assert_eq!(
            map_flow_type("component-config").unwrap(),
            FlowKind::ComponentConfig
        );
        assert!(matches!(
            map_flow_type("unknown").unwrap_err(),
            crate::error::FlowError::UnknownFlowType { .. }
        ));
    }

    #[test]
    fn compile_flow_builds_entrypoints_and_branch_routing() {
        let yaml = r#"id: demo
type: messaging
nodes:
  start:
    qa.process: {}
    routing:
      - status: ok
        to: done
      - to: fallback
  done:
    template: "ok"
    routing: out
  fallback:
    template: "fallback"
    routing: reply
"#;

        let flow = compile_ygtc_str(yaml).expect("compile flow");
        assert_eq!(flow.entrypoints.get("default"), Some(&json!("start")));
        match flow
            .nodes
            .get(&NodeId::new("start").unwrap())
            .unwrap()
            .routing
            .clone()
        {
            Routing::Branch { on_status, default } => {
                assert_eq!(on_status.get("ok").unwrap().as_str(), "done");
                assert_eq!(default.unwrap().as_str(), "fallback");
            }
            other => panic!("expected branch routing, got {other:?}"),
        }
    }

    #[test]
    fn compile_ygtc_file_reports_invalid_routing_targets() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("bad.ygtc");
        std::fs::write(
            &path,
            r#"id: demo
type: messaging
nodes:
  start:
    qa.process: {}
    routing:
      - to: missing
"#,
        )
        .unwrap();

        let err = compile_ygtc_file(&path).expect_err("missing routing target should fail");
        assert!(matches!(err, crate::error::FlowError::MissingNode { .. }));
    }

    #[test]
    fn compile_flow_rejects_invalid_routing_shorthand() {
        let err = load_ygtc_from_str(
            r#"id: demo
type: messaging
nodes:
  start:
    qa.process: {}
    routing: invalid
"#,
        )
        .expect_err("invalid shorthand should fail during load");
        // The JSON schema now catches the invalid routing value ("invalid" is
        // not in the `oneOf` enum) before the Rust routing validator runs.
        assert!(
            matches!(
                err,
                crate::error::FlowError::Routing { .. } | crate::error::FlowError::Schema { .. }
            ),
            "expected Routing or Schema error, got: {err}"
        );
    }

    #[test]
    fn compile_preserves_single_conditional_route_as_custom() {
        let flow = compile_ygtc_str(
            r#"
id: default
schema_version: 2
type: messaging
start: greet

nodes:
  greet:
    emit.response:
      text: hi
    routing:
      - condition: in.input.text != ""
        to: ask

  ask:
    emit.response:
      text: ok
    routing:
      - out: true
"#,
        )
        .expect("flow should compile");

        let greet_id = NodeId::new("greet").expect("valid node id");
        let greet = flow.nodes.get(&greet_id).expect("greet node exists");
        assert!(matches!(greet.routing, Routing::Custom(_)));
    }

    #[test]
    fn compile_preserves_multi_route_conditions_as_custom() {
        let flow = compile_ygtc_str(
            r#"
id: default
schema_version: 2
type: messaging
start: ask

nodes:
  ask:
    provider.invoke:
      provider_type: llm.openai.compat.ollama
      op: chat
    routing:
      - condition: error != ""
        to: send_error
      - to: send_response

  send_response:
    emit.response:
      text: ok
    routing:
      - out: true

  send_error:
    emit.response:
      text: bad
    routing:
      - out: true
"#,
        )
        .expect("flow should compile");

        let ask_id = NodeId::new("ask").expect("valid node id");
        let ask = flow.nodes.get(&ask_id).expect("ask node exists");
        assert!(matches!(ask.routing, Routing::Custom(_)));
    }

    #[test]
    fn compile_flow_forwards_slot_schema_into_metadata_extra() {
        let slot_schema = json!([
            { "name": "city", "slot_type": "string", "pattern": "^[A-Z].+" },
            { "name": "color", "slot_type": "enum", "enum_values": ["red", "blue"] }
        ]);
        let doc = crate::model::FlowDoc {
            id: "test".to_string(),
            title: None,
            description: None,
            flow_type: "messaging".to_string(),
            start: None,
            parameters: json!({"custom_key": 42}),
            tags: vec![],
            schema_version: None,
            entrypoints: Default::default(),
            meta: None,
            slot_schema: Some(slot_schema.clone()),
            nodes: {
                let mut m = indexmap::IndexMap::new();
                m.insert(
                    "start".to_string(),
                    crate::model::NodeDoc {
                        routing: json!("out"),
                        raw: {
                            let mut r = indexmap::IndexMap::new();
                            r.insert("template".to_string(), json!("hi"));
                            r
                        },
                        ..Default::default()
                    },
                );
                m
            },
        };

        let flow = compile_flow(doc).expect("compile_flow should succeed");
        assert_eq!(
            flow.metadata.extra[SLOT_SCHEMA_METADATA_KEY], slot_schema,
            "slot_schema must be forwarded into metadata.extra"
        );
        assert_eq!(
            flow.metadata.extra["custom_key"],
            json!(42),
            "original parameters must be preserved"
        );
    }

    /// Helper: build a `FlowDoc` with a `component.exec` node and the given
    /// `slot_schema`. This exercises the path that bypasses JSON-schema
    /// validation in the loader.
    fn component_exec_doc_with_slots(slot_schema: Option<Value>) -> crate::model::FlowDoc {
        crate::model::FlowDoc {
            id: "exec-test".to_string(),
            title: None,
            description: None,
            flow_type: "messaging".to_string(),
            start: None,
            parameters: json!({}),
            tags: vec![],
            schema_version: None,
            entrypoints: Default::default(),
            meta: None,
            slot_schema,
            nodes: {
                let mut m = indexmap::IndexMap::new();
                m.insert(
                    "run".to_string(),
                    crate::model::NodeDoc {
                        routing: json!("out"),
                        raw: {
                            let mut r = indexmap::IndexMap::new();
                            r.insert("component.exec".to_string(), json!("some.component"));
                            r
                        },
                        ..Default::default()
                    },
                );
                m
            },
        }
    }

    #[test]
    fn compile_rejects_duplicate_slot_names() {
        let doc = component_exec_doc_with_slots(Some(json!([
            { "name": "city", "slot_type": "string", "pattern": "^.+" },
            { "name": "city", "slot_type": "number" }
        ])));
        let msg = compile_flow(doc)
            .expect_err("compile_flow should reject duplicate slot names")
            .to_string();
        assert!(
            msg.contains("duplicate"),
            "error should mention 'duplicate': {msg}"
        );
    }

    #[test]
    fn compile_rejects_invalid_regex_in_slot_pattern() {
        let doc = component_exec_doc_with_slots(Some(json!([
            { "name": "city", "slot_type": "string", "pattern": "(" }
        ])));
        let msg = compile_flow(doc)
            .expect_err("compile_flow should reject invalid regex")
            .to_string();
        assert!(
            msg.contains("invalid regex pattern"),
            "error should mention 'invalid regex pattern': {msg}"
        );
    }

    #[test]
    fn compile_accepts_valid_slot_schema_via_component_exec() {
        let doc = component_exec_doc_with_slots(Some(json!([
            { "name": "city", "slot_type": "string", "pattern": "^[A-Z].+" },
            { "name": "color", "slot_type": "enum", "enum_values": ["red", "blue"] },
            { "name": "count", "slot_type": "number" },
            { "name": "active", "slot_type": "boolean" },
            { "name": "when", "slot_type": "date" }
        ])));
        let flow = compile_flow(doc).expect("valid slot_schema should compile");
        assert!(
            flow.metadata.extra.get(SLOT_SCHEMA_METADATA_KEY).is_some(),
            "slot_schema must be present in metadata.extra"
        );
    }

    #[test]
    fn routing_shorthand_validator_rejects_invalid_string_directly() {
        let nodes = HashSet::from(["start".to_string()]);
        let err = compile_routing(&json!("invalid"), &nodes, "start")
            .expect_err("invalid shorthand must be rejected");
        assert!(
            matches!(err, crate::error::FlowError::Routing { .. }),
            "expected Routing error, got: {err}"
        );
    }
}