greentic_dev/

flow_cmd.rs

use std::fs;
use std::path::{Path, PathBuf};

use crate::cli::{ConfigFlowModeArg, FlowAddStepArgs};
use crate::component_add::run_component_add;
use crate::pack_init::PackInitIntent;
use crate::path_safety::normalize_under_root;
use anyhow::{Context, Result, anyhow, bail};
use greentic_flow::flow_bundle::load_and_validate_bundle;
use serde_json::Value as JsonValue;
use serde_yaml_bw as serde_yaml;
use std::io::Write;
use std::str::FromStr;
use std::{io, io::IsTerminal};
use tempfile::NamedTempFile;

use greentic_types::FlowId;
use greentic_types::component::ComponentManifest;

const DEFAULT_CONFIG_FLOW_TYPE: &str = "component-config";

pub fn validate(path: &Path, compact_json: bool) -> Result<()> {
    let root = std::env::current_dir()
        .context("failed to resolve workspace root")?
        .canonicalize()
        .context("failed to canonicalize workspace root")?;
    let safe = normalize_under_root(&root, path)?;
    let source = fs::read_to_string(&safe)
        .with_context(|| format!("failed to read flow definition at {}", safe.display()))?;

    let bundle = load_and_validate_bundle(&source, Some(&safe)).with_context(|| {
        format!(
            "flow validation failed for {} using greentic-flow",
            safe.display()
        )
    })?;

    let serialized = if compact_json {
        serde_json::to_string(&bundle)?
    } else {
        serde_json::to_string_pretty(&bundle)?
    };

    println!("{serialized}");
    Ok(())
}

pub fn render_config_flow_yaml(config_flow_id: &str, graph: &JsonValue) -> Result<String> {
    let mut graph_obj = graph
        .as_object()
        .cloned()
        .ok_or_else(|| anyhow!("config flow `{config_flow_id}` graph is not an object"))?;
    let ty = graph_obj
        .entry("type".to_string())
        .or_insert(JsonValue::String(DEFAULT_CONFIG_FLOW_TYPE.to_string()));
    if !ty.is_string() {
        *ty = JsonValue::String(DEFAULT_CONFIG_FLOW_TYPE.to_string());
    }
    let normalized = JsonValue::Object(graph_obj);
    serde_yaml::to_string(&normalized).context("failed to render config flow graph to YAML")
}

pub fn run_add_step(args: FlowAddStepArgs) -> Result<()> {
    let manifest_path = args
        .manifest
        .clone()
        .unwrap_or_else(|| PathBuf::from("component.manifest.json"));
    if !manifest_path.exists() {
        bail!(
            "component.manifest.json not found at {}. Use --manifest to point to the manifest file.",
            manifest_path.display()
        );
    }
    let manifest_raw = std::fs::read_to_string(&manifest_path)
        .with_context(|| format!("failed to read {}", manifest_path.display()))?;
    let mut manifest_value: JsonValue = serde_json::from_str(&manifest_raw).with_context(|| {
        format!(
            "failed to parse component manifest JSON at {}",
            manifest_path.display()
        )
    })?;
    normalize_manifest(&mut manifest_value);
    let manifest: ComponentManifest =
        serde_json::from_value(manifest_value).with_context(|| {
            format!(
                "failed to parse component manifest JSON at {}",
                manifest_path.display()
            )
        })?;

    let config_flow_id = match args.mode {
        Some(ConfigFlowModeArg::Custom) => "custom".to_string(),
        Some(ConfigFlowModeArg::Default) => "default".to_string(),
        None => args.flow.clone(),
    };
    let config_flow_key = FlowId::from_str(&config_flow_id).map_err(|_| {
        anyhow!(
            "invalid flow identifier `{}`; flow ids must be valid FlowId strings",
            config_flow_id
        )
    })?;
    let Some(config_flow) = manifest.dev_flows.get(&config_flow_key) else {
        bail!(
            "Flow '{}' is missing from manifest.dev_flows. Run `greentic-component flow update` to regenerate config flows.",
            config_flow_id
        );
    };

    let coord = args
        .coordinate
        .ok_or_else(|| anyhow!("component coordinate is required (pass --coordinate)"))?;

    // Ensure the component is available locally (fetch if needed).
    let _bundle_dir = resolve_component_bundle(&coord, args.profile.as_deref())?;

    // Render the dev flow graph to YAML so the existing runner can consume it.
    let config_flow_yaml = render_config_flow_yaml(&config_flow_id, &config_flow.graph)?;
    let mut temp_flow =
        NamedTempFile::new().context("failed to create temporary config flow file")?;
    temp_flow
        .write_all(config_flow_yaml.as_bytes())
        .context("failed to write temporary config flow")?;
    temp_flow.flush()?;

    let pack_flow_path = PathBuf::from("flows").join(format!("{}.ygtc", args.flow_id));
    if !pack_flow_path.exists() {
        bail!(
            "Pack flow '{}' not found at {}",
            args.flow_id,
            pack_flow_path.display()
        );
    }
    let pack_flow_raw = std::fs::read_to_string(&pack_flow_path)
        .with_context(|| format!("failed to read pack flow {}", pack_flow_path.display()))?;
    let mut pack_flow_json: JsonValue = serde_yaml::from_str(&pack_flow_raw)
        .with_context(|| format!("invalid YAML in {}", pack_flow_path.display()))?;

    let output = crate::pack_run::run_config_flow(temp_flow.path())
        .with_context(|| format!("failed to run config flow {}", config_flow_id))?;
    let (node_id, mut node) = parse_config_flow_output(&output)?;
    normalize_node(&node_id, &mut node, &manifest)?;
    let after = args
        .after
        .clone()
        .or_else(|| prompt_routing_target(&pack_flow_json));
    if let Some(after) = after.as_deref() {
        patch_placeholder_routing(&mut node, after);
    }

    let graph_obj = pack_flow_json
        .as_object_mut()
        .ok_or_else(|| anyhow!("pack flow {} is not a mapping", pack_flow_path.display()))?;

    // Update target pack flow JSON
    let nodes = graph_obj
        .get_mut("nodes")
        .and_then(|n| n.as_object_mut())
        .ok_or_else(|| anyhow!("flow `{}` missing nodes map", args.flow_id))?;
    nodes.insert(node_id.clone(), node);

    if let Some(after) = args.after {
        append_routing(graph_obj, &after, &node_id)?;
    } else if let Some(after) = after.as_deref() {
        append_routing(graph_obj, after, &node_id)?;
    }

    let rendered =
        serde_yaml::to_string(&pack_flow_json).context("failed to render updated pack flow")?;
    std::fs::write(&pack_flow_path, rendered)
        .with_context(|| format!("failed to write {}", pack_flow_path.display()))?;

    println!(
        "Added node `{}` from config flow {} to {}",
        node_id,
        config_flow_id,
        pack_flow_path.display()
    );
    Ok(())
}

fn normalize_node(node_id: &str, node: &mut JsonValue, manifest: &ComponentManifest) -> Result<()> {
    let Some(obj) = node.as_object_mut() else {
        return Ok(());
    };

    // If the config flow forgot to populate component.exec details, inject sensible defaults so the
    // generated pack flow can execute.
    if let Some(exec) = obj
        .get_mut("component.exec")
        .and_then(|v| v.as_object_mut())
    {
        exec.entry("component")
            .or_insert_with(|| JsonValue::String(manifest.id.to_string()));

        let op_missing = match exec.get("operation") {
            Some(JsonValue::String(s)) => s.trim().is_empty(),
            None => true,
            _ => false,
        };
        if op_missing {
            let op = manifest
                .operations
                .first()
                .map(|o| o.name.clone())
                .ok_or_else(|| {
                    anyhow!(
                        "config flow emitted component.exec without operation and manifest has none"
                    )
                })?;
            exec.insert("operation".to_string(), JsonValue::String(op.clone()));
            exec.entry("op").or_insert(JsonValue::String(op));
        } else if let Some(JsonValue::String(op)) = exec.get("operation").cloned() {
            exec.entry("op").or_insert(JsonValue::String(op));
        }
    }

    if let Some(tool) = obj.get("tool").and_then(|v| v.as_object()) {
        let Some(comp) = tool.get("component").and_then(|v| v.as_str()) else {
            bail!("config flow output tool is missing `component`");
        };
        // Without an operation, we cannot produce a valid component.exec node.
        // Surface a clear error so users know to regenerate config flows or add an operation.
        bail!(
            "config flow emitted invalid step `{}`: component `{}` has no operations; regenerate config flows with a component version that declares operations",
            node_id,
            comp
        );
    }

    Ok(())
}

fn normalize_manifest(manifest: &mut JsonValue) {
    let Some(obj) = manifest.as_object_mut() else {
        return;
    };
    if let Some(ops) = obj.get_mut("operations")
        && let Some(arr) = ops.as_array_mut()
    {
        let mut normalized = Vec::with_capacity(arr.len());
        for entry in arr.drain(..) {
            if let Some(s) = entry.as_str() {
                let mut map = serde_json::Map::new();
                map.insert("name".to_string(), JsonValue::String(s.to_string()));
                normalized.push(JsonValue::Object(map));
            } else {
                normalized.push(entry);
            }
        }
        *arr = normalized;
    }
}

fn prompt_routing_target(flow_doc: &JsonValue) -> Option<String> {
    if !io::stdout().is_terminal() {
        return None;
    }
    let nodes = flow_doc
        .as_object()
        .and_then(|m| m.get("nodes"))
        .and_then(|n| n.as_object())?;
    let mut keys: Vec<String> = nodes.keys().cloned().collect();
    keys.sort();
    if keys.is_empty() {
        return None;
    }

    println!("Select where to route from (empty to skip):");
    for (idx, key) in keys.iter().enumerate() {
        println!("  {}) {}", idx + 1, key);
    }
    print!("Choice: ");
    let _ = io::stdout().flush();
    let mut buf = String::new();
    if io::stdin().read_line(&mut buf).is_err() {
        return None;
    }
    let choice = buf.trim();
    if choice.is_empty() {
        return None;
    }
    if let Ok(idx) = choice.parse::<usize>()
        && idx >= 1
        && idx <= keys.len()
    {
        return Some(keys[idx - 1].clone());
    }
    None
}

fn patch_placeholder_routing(node: &mut JsonValue, next: &str) {
    let Some(map) = node.as_object_mut() else {
        return;
    };
    let Some(routing) = map.get_mut("routing") else {
        return;
    };
    let Some(routes) = routing.as_array_mut() else {
        return;
    };
    for entry in routes.iter_mut() {
        if let Some(JsonValue::String(to)) =
            entry.as_object_mut().and_then(|route| route.get_mut("to"))
            && to == "NEXT_NODE_PLACEHOLDER"
        {
            *to = next.to_string();
        }
    }
}

fn resolve_component_bundle(coordinate: &str, profile: Option<&str>) -> Result<PathBuf> {
    let path = PathBuf::from_str(coordinate).unwrap_or_default();
    if path.exists() {
        return Ok(path);
    }
    let dir = run_component_add(coordinate, profile, PackInitIntent::Dev)?;
    Ok(dir)
}

pub fn parse_config_flow_output(output: &str) -> Result<(String, JsonValue)> {
    let value: JsonValue =
        serde_json::from_str(output).context("config flow output is not valid JSON")?;
    let obj = value
        .as_object()
        .ok_or_else(|| anyhow!("config flow output must be a JSON object"))?;
    let node_id = obj
        .get("node_id")
        .and_then(|v| v.as_str())
        .ok_or_else(|| anyhow!("config flow output missing node_id"))?
        .to_string();
    let node = obj
        .get("node")
        .ok_or_else(|| anyhow!("config flow output missing node"))?
        .clone();
    if !node.is_object() {
        bail!("config flow output node must be an object");
    }
    Ok((node_id, node))
}

fn append_routing(
    flow_doc: &mut serde_json::Map<String, JsonValue>,
    from_node: &str,
    to_node: &str,
) -> Result<()> {
    let nodes = flow_doc
        .get_mut("nodes")
        .and_then(|n| n.as_object_mut())
        .ok_or_else(|| anyhow!("flow missing nodes map"))?;
    let Some(node) = nodes.get_mut(from_node) else {
        bail!("node `{from_node}` not found for routing update");
    };
    let mapping = node
        .as_object_mut()
        .ok_or_else(|| anyhow!("node `{from_node}` is not an object"))?;
    let routing = mapping
        .entry("routing")
        .or_insert(JsonValue::Array(Vec::new()));
    let seq = routing
        .as_array_mut()
        .ok_or_else(|| anyhow!("routing for `{from_node}` is not a list"))?;
    let mut entry = serde_json::Map::new();
    entry.insert("to".to_string(), JsonValue::String(to_node.to_string()));
    seq.push(JsonValue::Object(entry));
    Ok(())
}