greentic_runner_host/runner/
flow_adapter.rs

1use std::collections::{BTreeMap, BTreeSet};
2use std::str::FromStr;
3
4use anyhow::{Context, Result, anyhow};
5use greentic_flow::model::{FlowDoc, NodeDoc};
6use greentic_types::flow::FlowHasher;
7use greentic_types::{
8    ComponentId, Flow, FlowComponentRef, FlowId, FlowKind, FlowMetadata, InputMapping, Node,
9    NodeId, OutputMapping, Routing, TelemetryHints,
10};
11use indexmap::IndexMap;
12use serde::{Deserialize, Serialize};
13use serde_json::{Map as JsonMap, Value};
14
15#[derive(Debug, Clone, Serialize, Deserialize)]
16pub struct FlowIR {
17    pub id: String,
18    pub flow_type: String,
19    pub start: Option<String>,
20    pub parameters: Value,
21    pub nodes: IndexMap<String, NodeIR>,
22}
23
24#[derive(Debug, Clone, Serialize, Deserialize)]
25pub struct NodeIR {
26    pub component: String,
27    pub payload_expr: Value,
28    pub routes: Vec<RouteIR>,
29}
30
31#[derive(Debug, Clone, Serialize, Deserialize)]
32pub struct RouteIR {
33    pub to: Option<String>,
34    #[serde(default)]
35    pub out: bool,
36}
37
38const FLOW_SCHEMA_VERSION: &str = "1.0";
39
40/// Runner-native flow op-keys that must reach the engine **verbatim** as the
41/// node `component` (NOT rewritten into a `component.exec` wrapper).
42///
43/// This is the single source of truth shared by the runtime-flow load path
44/// (`flow_doc_to_ir`, which already preserves these structurally) and the
45/// legacy flow-JSON load path (`normalize_flow_doc` in `pack.rs`, which would
46/// otherwise misclassify any non-`emit.*` op-key as `component.exec`). The
47/// engine's `From<Node>` dispatch (`runner::engine`) matches exactly these
48/// component strings — keep the two sets in lockstep when adding a node type.
49///
50/// `emit.*` prefixes and the self-contained `mcp:<server>/<tool>` ref form are
51/// handled separately by [`is_native_op_key`] because they are prefix-matched
52/// rather than exact.
53const NATIVE_OP_KEYS: &[&str] = &[
54    "flow.call",
55    "flow.goto",
56    "provider.invoke",
57    "session.wait",
58    "state.get",
59    "state.set",
60    "var.set",
61    "dw.agent",
62    "dw.agent_graph",
63    "sorla.call",
64    "operala.call",
65    "agentic.call",
66    "telco-x.call",
67    "approval.call",
68    "mcp",
69];
70
71/// Whether `key` is a runner-native flow op-key that the engine dispatches
72/// directly (so the loader must preserve it verbatim instead of wrapping it in
73/// a `component.exec` node).
74///
75/// Covers the exact [`NATIVE_OP_KEYS`] plus the `emit.*` builtin family and the
76/// self-contained `mcp:<server>/<tool>` ref form.
77pub(crate) fn is_native_op_key(key: &str) -> bool {
78    key.starts_with("emit.") || key.starts_with("mcp:") || NATIVE_OP_KEYS.contains(&key)
79}
80
81pub fn flow_doc_to_ir(doc: FlowDoc) -> Result<FlowIR> {
82    let mut nodes: IndexMap<String, NodeIR> = IndexMap::new();
83    for (id, node) in doc.nodes {
84        let (component, payload_expr) = extract_node_payload(&node)
85            .with_context(|| format!("missing component for node `{id}`"))?;
86        let routes = parse_routes(node.routing)?;
87        nodes.insert(
88            id.clone(),
89            NodeIR {
90                component,
91                payload_expr,
92                routes,
93            },
94        );
95    }
96
97    Ok(FlowIR {
98        id: doc.id,
99        flow_type: doc.flow_type,
100        start: doc.start,
101        parameters: doc.parameters,
102        nodes,
103    })
104}
105
106fn extract_node_payload(node: &NodeDoc) -> Result<(String, Value)> {
107    if let Some(operation) = node.operation.as_deref() {
108        return Ok((operation.to_string(), node.payload.clone()));
109    }
110    let Some((component, payload)) = node
111        .raw
112        .iter()
113        .next()
114        .map(|(key, value)| (key.clone(), value.clone()))
115    else {
116        return Err(anyhow!("node missing operation payload"));
117    };
118    Ok((component, payload))
119}
120
121pub fn flow_ir_to_flow(flow_ir: FlowIR) -> Result<Flow> {
122    let id = FlowId::from_str(&flow_ir.id)
123        .with_context(|| format!("invalid flow id `{}`", flow_ir.id))?;
124    let kind = map_flow_kind(&flow_ir.flow_type)?;
125
126    let mut entrypoints = BTreeMap::new();
127    if let Some(start) = &flow_ir.start {
128        entrypoints.insert("default".to_string(), Value::String(start.clone()));
129    }
130
131    let nodes = map_nodes(flow_ir.nodes)?;
132    let metadata = build_metadata(flow_ir.start, flow_ir.parameters);
133
134    Ok(Flow {
135        schema_version: FLOW_SCHEMA_VERSION.to_string(),
136        id,
137        kind,
138        entrypoints,
139        nodes,
140        metadata,
141    })
142}
143
144fn map_flow_kind(kind: &str) -> Result<FlowKind> {
145    match kind {
146        "messaging" => Ok(FlowKind::Messaging),
147        "event" | "events" => Ok(FlowKind::Event),
148        "component-config" => Ok(FlowKind::ComponentConfig),
149        "job" => Ok(FlowKind::Job),
150        "http" => Ok(FlowKind::Http),
151        other => Err(anyhow!("unknown flow kind `{other}`")),
152    }
153}
154
155fn map_nodes(nodes: IndexMap<String, NodeIR>) -> Result<IndexMap<NodeId, Node, FlowHasher>> {
156    let mut mapped: IndexMap<NodeId, Node, FlowHasher> = IndexMap::default();
157    for (raw_id, node_ir) in nodes {
158        let node_id =
159            NodeId::from_str(&raw_id).with_context(|| format!("invalid node id `{raw_id}`"))?;
160        let node = map_node(node_id.clone(), node_ir)?;
161        mapped.insert(node_id, node);
162    }
163    Ok(mapped)
164}
165
166fn map_node(node_id: NodeId, node_ir: NodeIR) -> Result<Node> {
167    let component_id = ComponentId::from_str(&node_ir.component).with_context(|| {
168        format!(
169            "invalid component ref `{}` for node {}",
170            node_ir.component,
171            node_id.as_str()
172        )
173    })?;
174    let component = FlowComponentRef {
175        id: component_id,
176        pack_alias: None,
177        operation: None,
178    };
179    let routing = map_routing(&node_ir.routes)?;
180    Ok(Node {
181        id: node_id,
182        component,
183        input: InputMapping {
184            mapping: node_ir.payload_expr,
185        },
186        output: OutputMapping {
187            mapping: Value::Object(JsonMap::new()),
188        },
189        err_map: None,
190        routing,
191        telemetry: TelemetryHints::default(),
192        conversational: false,
193    })
194}
195
196fn map_routing(routes: &[RouteIR]) -> Result<Routing> {
197    if routes.is_empty() {
198        return Ok(Routing::End);
199    }
200
201    if routes.len() == 1 {
202        let route = &routes[0];
203        if route.out || route.to.as_deref() == Some("out") {
204            return Ok(Routing::End);
205        }
206        if let Some(to) = &route.to {
207            let node_id =
208                NodeId::from_str(to).with_context(|| format!("invalid route target `{to}`"))?;
209            return Ok(Routing::Next { node_id });
210        }
211    }
212
213    serde_json::to_value(routes)
214        .map(Routing::Custom)
215        .map_err(|err| anyhow!(err))
216}
217
218fn build_metadata(start: Option<String>, parameters: Value) -> FlowMetadata {
219    let mut extra = JsonMap::new();
220    if let Some(start) = start {
221        extra.insert("start".into(), Value::String(start));
222    }
223    if !parameters.is_null() {
224        extra.insert("parameters".into(), parameters);
225    }
226    FlowMetadata {
227        title: None,
228        description: None,
229        tags: BTreeSet::new(),
230        extra: Value::Object(extra),
231    }
232}
233
234fn parse_routes(raw: Value) -> Result<Vec<RouteIR>> {
235    if raw.is_null() {
236        return Ok(Vec::new());
237    }
238    serde_json::from_value::<Vec<RouteIR>>(raw.clone()).map_err(|err| {
239        anyhow!("failed to parse routes from node routing: {err}; value was {raw:?}")
240    })
241}
242
243#[cfg(test)]
244mod tests {
245    use super::*;
246
247    #[test]
248    fn telco_x_call_is_a_recognized_native_op() {
249        // Wire-ready: the runner recognizes telco-x.call as a native
250        // remote-dispatch node (not an unknown component.exec), alongside the
251        // other dispatch runtimes. The runtime side is not deployed yet.
252        assert!(is_native_op_key("telco-x.call"));
253        assert!(is_native_op_key("operala.call"));
254        assert!(is_native_op_key("sorla.call"));
255        assert!(!is_native_op_key("nope.call"));
256    }
257
258    #[test]
259    fn approval_call_is_a_recognized_native_op() {
260        assert!(is_native_op_key("approval.call"));
261    }
262}
greentic_runner_host/runner/flow_adapter.rs

greentic_runner_host/runner/
flow_adapter.rs
