use super::*;
fn retry_repair_prompt_builder_from_dict(
dict: Option<&crate::value::DictMap>,
) -> Option<EqIgnored<VmValue>> {
dict.and_then(|d| d.get("retry_policy"))
.and_then(|policy| policy.as_dict())
.and_then(|policy| policy.get("repair_prompt_builder"))
.filter(|value| !matches!(value, VmValue::Nil))
.cloned()
.map(EqIgnored)
}
pub fn parse_workflow_node_value(value: &VmValue, label: &str) -> Result<WorkflowNode, VmError> {
let mut node: WorkflowNode =
crate::orchestration::parse_json_payload(vm_value_to_json(value), label)?;
let dict = value.as_dict();
node.raw_tools = dict.and_then(|d| d.get("tools")).cloned();
node.raw_auto_compact = dict.and_then(|d| d.get("auto_compact")).cloned();
node.raw_model_policy = dict.and_then(|d| d.get("model_policy")).cloned();
node.raw_context_assembler = dict.and_then(|d| d.get("context_assembler")).cloned();
node.raw_verify = dict
.and_then(|d| d.get("verify"))
.filter(|value| {
matches!(
value,
VmValue::Closure(_) | VmValue::BuiltinRef(_) | VmValue::BuiltinRefId(_)
)
})
.cloned();
node.raw_executor = dict
.and_then(|d| d.get("executor"))
.filter(|value| {
matches!(
value,
VmValue::Closure(_) | VmValue::BuiltinRef(_) | VmValue::BuiltinRefId(_)
)
})
.cloned();
node.retry_policy.repair_prompt_builder = retry_repair_prompt_builder_from_dict(dict);
Ok(node)
}
pub fn parse_workflow_node_json(
json: serde_json::Value,
label: &str,
) -> Result<WorkflowNode, VmError> {
crate::orchestration::parse_json_payload(json, label)
}
pub fn parse_workflow_edge_json(
json: serde_json::Value,
label: &str,
) -> Result<WorkflowEdge, VmError> {
crate::orchestration::parse_json_payload(json, label)
}
pub fn normalize_workflow_value(value: &VmValue) -> Result<WorkflowGraph, VmError> {
let mut graph: WorkflowGraph = crate::orchestration::parse_json_value(value)?;
let as_dict = value.as_dict().cloned().unwrap_or_default();
if let Some(nodes) = as_dict.get("nodes").and_then(VmValue::as_dict) {
for (id, value) in nodes.iter() {
graph.nodes.insert(
id.to_string(),
parse_workflow_node_value(value, "orchestration")?,
);
}
}
if graph.nodes.is_empty() {
for key in ["act", "verify", "repair"] {
if let Some(node_value) = as_dict.get(key) {
let mut node = parse_workflow_node_value(node_value, "orchestration")?;
let raw_node = node_value.as_dict().cloned().unwrap_or_default();
node.id = Some(key.to_string());
if node.kind.is_empty() {
node.kind = if key == "verify" {
"verify".to_string()
} else {
"stage".to_string()
};
}
if node.model_policy.provider.is_none() {
node.model_policy.provider = as_dict
.get("provider")
.map(|value| value.display())
.filter(|value| !value.is_empty());
}
if node.model_policy.model.is_none() {
node.model_policy.model = as_dict
.get("model")
.map(|value| value.display())
.filter(|value| !value.is_empty());
}
if node.model_policy.model_tier.is_none() {
node.model_policy.model_tier = as_dict
.get("model_tier")
.or_else(|| as_dict.get("tier"))
.map(|value| value.display())
.filter(|value| !value.is_empty());
}
if node.model_policy.temperature.is_none() {
node.model_policy.temperature = as_dict.get("temperature").and_then(|value| {
if let VmValue::Float(number) = value {
Some(*number)
} else {
value.as_int().map(|number| number as f64)
}
});
}
if node.model_policy.max_tokens.is_none() {
node.model_policy.max_tokens =
as_dict.get("max_tokens").and_then(|value| value.as_int());
}
if node.mode.is_none() {
node.mode = as_dict
.get("mode")
.map(|value| value.display())
.filter(|value| !value.is_empty());
}
if node.done_sentinel.is_none() {
node.done_sentinel = as_dict
.get("done_sentinel")
.map(|value| value.display())
.filter(|value| !value.is_empty());
}
if key == "verify"
&& node.verify.is_none()
&& (raw_node.contains_key("assert_text")
|| raw_node.contains_key("command")
|| raw_node.contains_key("expect_status")
|| raw_node.contains_key("expect_text"))
{
node.verify = Some(serde_json::json!({
"assert_text": raw_node.get("assert_text").map(vm_value_to_json),
"command": raw_node.get("command").map(vm_value_to_json),
"expect_status": raw_node.get("expect_status").map(vm_value_to_json),
"expect_text": raw_node.get("expect_text").map(vm_value_to_json),
}));
}
graph.nodes.insert(key.to_string(), node);
}
}
if graph.entry.is_empty() && graph.nodes.contains_key("act") {
graph.entry = "act".to_string();
}
if graph.edges.is_empty() && graph.nodes.contains_key("act") {
if graph.nodes.contains_key("verify") {
graph.edges.push(WorkflowEdge {
from: "act".to_string(),
to: "verify".to_string(),
branch: None,
label: None,
});
}
if graph.nodes.contains_key("repair") {
graph.edges.push(WorkflowEdge {
from: "verify".to_string(),
to: "repair".to_string(),
branch: Some("failed".to_string()),
label: None,
});
graph.edges.push(WorkflowEdge {
from: "repair".to_string(),
to: "verify".to_string(),
branch: Some("retry".to_string()),
label: None,
});
}
}
}
if graph.type_name.is_empty() {
graph.type_name = "workflow_graph".to_string();
}
if graph.id.is_empty() {
graph.id = new_id("workflow");
}
if graph.version == 0 {
graph.version = 1;
}
if graph.entry.is_empty() {
graph.entry = graph
.nodes
.keys()
.next()
.cloned()
.unwrap_or_else(|| "act".to_string());
}
for (node_id, node) in &mut graph.nodes {
if node.id.is_none() {
node.id = Some(node_id.clone());
}
if node.kind.is_empty() {
node.kind = "stage".to_string();
}
if node.join_policy.strategy.is_empty() {
node.join_policy.strategy = "all".to_string();
}
if node.reduce_policy.strategy.is_empty() {
node.reduce_policy.strategy = "concat".to_string();
}
if node.output_contract.output_kinds.is_empty() {
node.output_contract.output_kinds = vec![match node.kind.as_str() {
"verify" => "verification_result".to_string(),
"reduce" => node
.reduce_policy
.output_kind
.clone()
.unwrap_or_else(|| "summary".to_string()),
"map" => node
.map_policy
.output_kind
.clone()
.unwrap_or_else(|| "artifact".to_string()),
"escalation" => "plan".to_string(),
_ => "artifact".to_string(),
}];
}
if node.retry_policy.max_attempts == 0 {
node.retry_policy.max_attempts = 1;
}
}
Ok(graph)
}
pub fn validate_workflow(
graph: &WorkflowGraph,
ceiling: Option<&CapabilityPolicy>,
) -> WorkflowValidationReport {
let mut errors = Vec::new();
let mut warnings = Vec::new();
if !graph.nodes.contains_key(&graph.entry) {
errors.push(format!("entry node does not exist: {}", graph.entry));
}
let node_ids: BTreeSet<String> = graph.nodes.keys().cloned().collect();
for edge in &graph.edges {
if !node_ids.contains(&edge.from) {
errors.push(format!("edge.from references unknown node: {}", edge.from));
}
if !node_ids.contains(&edge.to) {
errors.push(format!("edge.to references unknown node: {}", edge.to));
}
}
let reachable_nodes = reachable_nodes(graph);
for node_id in &node_ids {
if !reachable_nodes.contains(node_id) {
warnings.push(format!("node is unreachable: {node_id}"));
}
}
for (node_id, node) in &graph.nodes {
let incoming = graph
.edges
.iter()
.filter(|edge| edge.to == *node_id)
.count();
let outgoing: Vec<&WorkflowEdge> = graph
.edges
.iter()
.filter(|edge| edge.from == *node_id)
.collect();
if let Some(min_inputs) = node.input_contract.min_inputs {
if let Some(max_inputs) = node.input_contract.max_inputs {
if min_inputs > max_inputs {
errors.push(format!(
"node {node_id}: input contract min_inputs exceeds max_inputs"
));
}
}
}
match node.kind.as_str() {
"condition" => {
let has_true = outgoing
.iter()
.any(|edge| edge.branch.as_deref() == Some("true"));
let has_false = outgoing
.iter()
.any(|edge| edge.branch.as_deref() == Some("false"));
if !has_true || !has_false {
errors.push(format!(
"node {node_id}: condition nodes require both 'true' and 'false' branch edges"
));
}
}
"fork" if outgoing.len() < 2 => {
errors.push(format!(
"node {node_id}: fork nodes require at least two outgoing edges"
));
}
"join" if incoming < 2 => {
warnings.push(format!(
"node {node_id}: join node has fewer than two incoming edges"
));
}
"map"
if node.map_policy.items.is_empty()
&& node.map_policy.item_artifact_kind.is_none()
&& node.input_contract.input_kinds.is_empty() =>
{
errors.push(format!(
"node {node_id}: map nodes require items, item_artifact_kind, or input_contract.input_kinds"
));
}
"reduce" if node.input_contract.input_kinds.is_empty() => {
warnings.push(format!(
"node {node_id}: reduce node has no input_contract.input_kinds; it will consume all available artifacts"
));
}
_ => {}
}
}
if let Some(ceiling) = ceiling {
if let Err(error) = ceiling.intersect(&graph.capability_policy) {
errors.push(error);
}
for (node_id, node) in &graph.nodes {
if let Err(error) = ceiling.intersect(&node.capability_policy) {
errors.push(format!("node {node_id}: {error}"));
}
}
}
for diagnostic in crate::tool_surface::validate_workflow_graph(graph) {
let message = format!("{}: {}", diagnostic.code, diagnostic.message);
match diagnostic.severity {
crate::tool_surface::ToolSurfaceSeverity::Error => errors.push(message),
crate::tool_surface::ToolSurfaceSeverity::Warning => warnings.push(message),
}
}
WorkflowValidationReport {
valid: errors.is_empty(),
errors,
warnings,
reachable_nodes: reachable_nodes.into_iter().collect(),
}
}
fn reachable_nodes(graph: &WorkflowGraph) -> BTreeSet<String> {
let mut seen = BTreeSet::new();
let mut stack = vec![graph.entry.clone()];
while let Some(node_id) = stack.pop() {
if !seen.insert(node_id.clone()) {
continue;
}
for edge in graph.edges.iter().filter(|edge| edge.from == node_id) {
stack.push(edge.to.clone());
}
}
seen
}