greentic_dev/
flow_cmd.rs

1use std::fs;
2use std::path::Path;
3
4use crate::cli::{ConfigFlowModeArg, FlowAddStepArgs};
5use crate::pack_init::{self, PackInitIntent};
6use crate::path_safety::normalize_under_root;
7use anyhow::{Context, Result, anyhow, bail};
8use greentic_flow::flow_bundle::load_and_validate_bundle;
9use serde_json::Value as JsonValue;
10use serde_yaml_bw as serde_yaml;
11use serde_yaml_bw::Mapping as YamlMapping;
12use serde_yaml_bw::Sequence as YamlSequence;
13use serde_yaml_bw::Value as YamlValue;
14use std::fs::OpenOptions;
15use std::io::Write;
16use std::path::PathBuf;
17use std::str::FromStr;
18use std::{io, io::IsTerminal};
19
20pub fn validate(path: &Path, compact_json: bool) -> Result<()> {
21    let root = std::env::current_dir()
22        .context("failed to resolve workspace root")?
23        .canonicalize()
24        .context("failed to canonicalize workspace root")?;
25    let safe = normalize_under_root(&root, path)?;
26    let source = fs::read_to_string(&safe)
27        .with_context(|| format!("failed to read flow definition at {}", safe.display()))?;
28
29    let bundle = load_and_validate_bundle(&source, Some(&safe)).with_context(|| {
30        format!(
31            "flow validation failed for {} using greentic-flow",
32            safe.display()
33        )
34    })?;
35
36    let serialized = if compact_json {
37        serde_json::to_string(&bundle)?
38    } else {
39        serde_json::to_string_pretty(&bundle)?
40    };
41
42    println!("{serialized}");
43    Ok(())
44}
45
46pub fn run_add_step(args: FlowAddStepArgs) -> Result<()> {
47    let root = std::env::current_dir()
48        .context("failed to resolve workspace root")?
49        .canonicalize()
50        .context("failed to canonicalize workspace root")?;
51    let flow_path = root.join("flows").join(format!("{}.ygtc", args.flow_id));
52    if !flow_path.exists() {
53        bail!("flow file not found: {}", flow_path.display());
54    }
55    let flow_src = std::fs::read_to_string(&flow_path)
56        .with_context(|| format!("failed to read {}", flow_path.display()))?;
57    let mut flow_doc: YamlValue = serde_yaml::from_str(&flow_src)
58        .with_context(|| format!("failed to parse {}", flow_path.display()))?;
59
60    let coord = args
61        .coordinate
62        .ok_or_else(|| anyhow!("component coordinate is required (pass --coordinate)"))?;
63
64    // Fetch or use local component bundle
65    let bundle_dir = resolve_component_bundle(&coord, args.profile.as_deref())?;
66    let flows_dir = bundle_dir.join("flows");
67    let custom_flow = flows_dir.join("custom.ygtc");
68    let default_flow = flows_dir.join("default.ygtc");
69    let selected = match args.mode {
70        Some(ConfigFlowModeArg::Custom) => {
71            if custom_flow.exists() {
72                custom_flow
73            } else {
74                default_flow.clone()
75            }
76        }
77        Some(ConfigFlowModeArg::Default) => {
78            if default_flow.exists() {
79                default_flow
80            } else {
81                custom_flow.clone()
82            }
83        }
84        None => {
85            if default_flow.exists() {
86                default_flow
87            } else if custom_flow.exists() {
88                custom_flow
89            } else {
90                bail!("component bundle does not provide flows/default.ygtc or flows/custom.ygtc")
91            }
92        }
93    };
94    if !selected.exists() {
95        bail!("selected config flow missing at {}", selected.display());
96    }
97
98    let output = crate::pack_run::run_config_flow(&selected)
99        .with_context(|| format!("failed to run config flow {}", selected.display()))?;
100    let (node_id, mut node) = parse_config_flow_output(&output)?;
101    let after = args
102        .after
103        .clone()
104        .or_else(|| prompt_routing_target(&flow_doc));
105    if let Some(after) = after.as_deref() {
106        patch_placeholder_routing(&mut node, after);
107    }
108
109    // Update target flow YAML
110    let nodes = flow_doc
111        .as_mapping_mut()
112        .and_then(|m| m.get_mut(YamlValue::String("nodes".to_string(), None)))
113        .and_then(|n| n.as_mapping_mut())
114        .ok_or_else(|| anyhow!("flow missing nodes map"))?;
115    nodes.insert(
116        YamlValue::String(node_id.clone(), None),
117        node_to_yaml(node)?,
118    );
119
120    if let Some(after) = args.after {
121        append_routing(&mut flow_doc, &after, &node_id)?;
122    } else if let Some(after) = after.as_deref() {
123        append_routing(&mut flow_doc, after, &node_id)?;
124    }
125
126    let rendered = serde_yaml::to_string(&flow_doc).context("failed to render updated flow")?;
127    let mut file = OpenOptions::new()
128        .write(true)
129        .truncate(true)
130        .open(&flow_path)
131        .with_context(|| format!("failed to open {} for writing", flow_path.display()))?;
132    file.write_all(rendered.as_bytes())
133        .with_context(|| format!("failed to write {}", flow_path.display()))?;
134
135    println!(
136        "Added node `{}` from config flow {} to {}",
137        node_id,
138        selected.file_name().unwrap_or_default().to_string_lossy(),
139        flow_path.display()
140    );
141    Ok(())
142}
143
144fn prompt_routing_target(flow_doc: &YamlValue) -> Option<String> {
145    if !io::stdout().is_terminal() {
146        return None;
147    }
148    let nodes = flow_doc
149        .as_mapping()
150        .and_then(|m| m.get(YamlValue::String("nodes".to_string(), None)))
151        .and_then(|n| n.as_mapping())?;
152    let mut keys: Vec<String> = nodes
153        .keys()
154        .filter_map(|k| k.as_str().map(|s| s.to_string()))
155        .collect();
156    keys.sort();
157    if keys.is_empty() {
158        return None;
159    }
160
161    println!("Select where to route from (empty to skip):");
162    for (idx, key) in keys.iter().enumerate() {
163        println!("  {}) {}", idx + 1, key);
164    }
165    print!("Choice: ");
166    let _ = io::stdout().flush();
167    let mut buf = String::new();
168    if io::stdin().read_line(&mut buf).is_err() {
169        return None;
170    }
171    let choice = buf.trim();
172    if choice.is_empty() {
173        return None;
174    }
175    if let Ok(idx) = choice.parse::<usize>()
176        && idx >= 1
177        && idx <= keys.len()
178    {
179        return Some(keys[idx - 1].clone());
180    }
181    None
182}
183
184fn patch_placeholder_routing(node: &mut JsonValue, next: &str) {
185    let Some(map) = node.as_object_mut() else {
186        return;
187    };
188    let Some(routing) = map.get_mut("routing") else {
189        return;
190    };
191    let Some(routes) = routing.as_array_mut() else {
192        return;
193    };
194    for entry in routes.iter_mut() {
195        if let Some(JsonValue::String(to)) =
196            entry.as_object_mut().and_then(|route| route.get_mut("to"))
197            && to == "NEXT_NODE_PLACEHOLDER"
198        {
199            *to = next.to_string();
200        }
201    }
202}
203
204fn resolve_component_bundle(coordinate: &str, profile: Option<&str>) -> Result<PathBuf> {
205    let path = PathBuf::from_str(coordinate).unwrap_or_default();
206    if path.exists() {
207        return Ok(path);
208    }
209    let dir = pack_init::run_component_add(coordinate, profile, PackInitIntent::Dev)?;
210    Ok(dir)
211}
212
213pub fn parse_config_flow_output(output: &str) -> Result<(String, JsonValue)> {
214    let value: JsonValue =
215        serde_json::from_str(output).context("config flow output is not valid JSON")?;
216    let obj = value
217        .as_object()
218        .ok_or_else(|| anyhow!("config flow output must be a JSON object"))?;
219    let node_id = obj
220        .get("node_id")
221        .and_then(|v| v.as_str())
222        .ok_or_else(|| anyhow!("config flow output missing node_id"))?
223        .to_string();
224    let node = obj
225        .get("node")
226        .ok_or_else(|| anyhow!("config flow output missing node"))?
227        .clone();
228    if !node.is_object() {
229        bail!("config flow output node must be an object");
230    }
231    Ok((node_id, node))
232}
233
234fn node_to_yaml(node: JsonValue) -> Result<YamlValue> {
235    let yaml_string = serde_yaml::to_string(&node).context("failed to render node to YAML")?;
236    let yaml_value: YamlValue =
237        serde_yaml::from_str(&yaml_string).context("failed to parse rendered YAML")?;
238    Ok(yaml_value)
239}
240
241fn append_routing(flow_doc: &mut YamlValue, from_node: &str, to_node: &str) -> Result<()> {
242    let nodes = flow_doc
243        .as_mapping_mut()
244        .and_then(|m| m.get_mut(YamlValue::String("nodes".to_string(), None)))
245        .and_then(|n| n.as_mapping_mut())
246        .ok_or_else(|| anyhow!("flow missing nodes map"))?;
247    let key = YamlValue::String(from_node.to_string(), None);
248    let Some(node) = nodes.get_mut(&key) else {
249        bail!("node `{from_node}` not found for routing update");
250    };
251    let mapping = node
252        .as_mapping_mut()
253        .ok_or_else(|| anyhow!("node `{from_node}` is not a mapping"))?;
254    let routes_key = YamlValue::String("routing".to_string(), None);
255    let routing = mapping
256        .entry(routes_key)
257        .or_insert(YamlValue::Sequence(YamlSequence::default()));
258    let seq = routing
259        .as_sequence_mut()
260        .ok_or_else(|| anyhow!("routing for `{from_node}` is not a list"))?;
261    let mut entry = YamlMapping::new();
262    entry.insert(
263        YamlValue::String("to".to_string(), None),
264        YamlValue::String(to_node.to_string(), None),
265    );
266    seq.elements.push(YamlValue::Mapping(entry));
267    Ok(())
268}
greentic_dev/flow_cmd.rs

greentic_dev/
flow_cmd.rs
