From 9e312b65ae6b6593c8d3353911053a64f3c79590 Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Wed, 22 Jul 2026 14:46:34 +0300 Subject: [PATCH] chore(deps): update vendored runtime crates (#5081) --- Cargo.lock | 20 +- Cargo.toml | 8 +- app/src-tauri/Cargo.lock | 20 +- src/openhuman/flows/builder_tools.rs | 55 +- src/openhuman/flows/builder_tools_tests.rs | 210 ++++- src/openhuman/flows/ops.rs | 498 +++++++++- src/openhuman/flows/ops_tests.rs | 855 ++++++++++++++++++ src/openhuman/flows/tools.rs | 118 +-- src/openhuman/flows/tools_tests.rs | 72 ++ src/openhuman/tinyagents/middleware.rs | 14 +- src/openhuman/tinyflows/caps.rs | 59 +- tests/json_rpc_e2e.rs | 35 +- ...osio_credentials_state_raw_coverage_e2e.rs | 10 +- vendor/tinyagents | 2 +- vendor/tinychannels | 2 +- vendor/tinyflows | 2 +- vendor/tinyjuice | 2 +- 17 files changed, 1821 insertions(+), 161 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index ea9bc745a..92b1b3bca 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -765,9 +765,9 @@ checksum = "8f1fe948ff07f4bd06c30984e69f5b4899c516a3ef74f34df92a2df2ab535495" [[package]] name = "bytes" -version = "1.11.1" +version = "1.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" dependencies = [ "serde", ] @@ -3910,9 +3910,9 @@ dependencies = [ [[package]] name = "mail-parser" -version = "0.11.3" +version = "0.11.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d8a2420e9ce11c2b0583ca97ddff7ab2398c8a613154e9b72e3bafdbf767f1d7" +checksum = "47785d444be4d32c1709171c6219a90f667c0ad0ffe68b4b179e794f31f4f9e8" dependencies = [ "hashify", ] @@ -4746,7 +4746,7 @@ dependencies = [ "tar", "tempfile", "thiserror 2.0.18", - "tinyagents 1.9.0", + "tinyagents 2.1.0", "tinychannels", "tinycortex", "tinyflows", @@ -7334,15 +7334,13 @@ dependencies = [ [[package]] name = "tinyagents" -version = "1.9.0" +version = "2.0.0" dependencies = [ "async-trait", "bytes", "chrono", "futures", "reqwest 0.12.28", - "rhai", - "rusqlite", "serde", "serde_json", "sha2 0.11.0", @@ -7353,13 +7351,15 @@ dependencies = [ [[package]] name = "tinyagents" -version = "2.0.0" +version = "2.1.0" dependencies = [ "async-trait", "bytes", "chrono", "futures", "reqwest 0.12.28", + "rhai", + "rusqlite", "serde", "serde_json", "sha2 0.11.0", @@ -7453,7 +7453,7 @@ dependencies = [ "serde", "serde_json", "thiserror 2.0.18", - "tinyagents 1.9.0", + "tinyagents 2.1.0", "tracing", ] diff --git a/Cargo.toml b/Cargo.toml index 302573597..837f5f41c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -72,9 +72,9 @@ crate-type = ["rlib"] tinyplace = "2.0" # tinyflows — host-agnostic workflow engine (typed node graph → validate → compile → # run on tinyagents). Powers the "Workflows" feature via the seam in -# `src/openhuman/tinyflows/` + the `flows::` domain. Pulls tinyagents 1.7 transitively -# (same version openhuman already uses — no conflict). Published on crates.io -# and patched below to the vendored submodule. +# `src/openhuman/tinyflows/` + the `flows::` domain. Pulls tinyagents 2.1 transitively +# (same version OpenHuman uses directly, preserving one trait identity). Published +# on crates.io and patched below to the vendored submodule. # # `mock` feature: enables `tinyflows::caps::mock::mock_capabilities()` — the # deterministic in-memory capability bundle the flows `dry_run_workflow` agent @@ -104,7 +104,7 @@ tinyjuice = { version = "0.2.1", default-features = false } # NOT enabled here — the default-ON `flows` feature turns it on via # `tinyagents/repl` (#4797), so a slim build without `flows` sheds `rhai`. # The crate itself can never be dropped: 26+ domains consume tinyagents. -tinyagents = { version = "1.7", features = ["sqlite"] } +tinyagents = { version = "2.1", features = ["sqlite"] } # TinyCortex — Rust core for the memory engine (store/chunks/tree/retrieval/ # queue/ingest/score + long tail), vendored as a git submodule and patched # below to `vendor/tinycortex`. OpenHuman's memory subsystem migrates onto this diff --git a/app/src-tauri/Cargo.lock b/app/src-tauri/Cargo.lock index 8d13a39f8..2609cdc1c 100644 --- a/app/src-tauri/Cargo.lock +++ b/app/src-tauri/Cargo.lock @@ -912,9 +912,9 @@ checksum = "8f1fe948ff07f4bd06c30984e69f5b4899c516a3ef74f34df92a2df2ab535495" [[package]] name = "bytes" -version = "1.11.1" +version = "1.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" dependencies = [ "serde", ] @@ -4603,9 +4603,9 @@ dependencies = [ [[package]] name = "mail-parser" -version = "0.11.3" +version = "0.11.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d8a2420e9ce11c2b0583ca97ddff7ab2398c8a613154e9b72e3bafdbf767f1d7" +checksum = "47785d444be4d32c1709171c6219a90f667c0ad0ffe68b4b179e794f31f4f9e8" dependencies = [ "hashify", ] @@ -5603,7 +5603,7 @@ dependencies = [ "tar", "tempfile", "thiserror 2.0.18", - "tinyagents 1.9.0", + "tinyagents 2.1.0", "tinychannels", "tinycortex", "tinyflows", @@ -8970,15 +8970,13 @@ dependencies = [ [[package]] name = "tinyagents" -version = "1.9.0" +version = "2.0.0" dependencies = [ "async-trait", "bytes", "chrono", "futures", "reqwest 0.12.28", - "rhai", - "rusqlite", "serde", "serde_json", "sha2 0.11.0", @@ -8989,13 +8987,15 @@ dependencies = [ [[package]] name = "tinyagents" -version = "2.0.0" +version = "2.1.0" dependencies = [ "async-trait", "bytes", "chrono", "futures", "reqwest 0.12.28", + "rhai", + "rusqlite", "serde", "serde_json", "sha2 0.11.0", @@ -9084,7 +9084,7 @@ dependencies = [ "serde", "serde_json", "thiserror 2.0.18", - "tinyagents 1.9.0", + "tinyagents 2.1.0", "tracing", ] diff --git a/src/openhuman/flows/builder_tools.rs b/src/openhuman/flows/builder_tools.rs index 5f2aa45e4..170c3c0bc 100644 --- a/src/openhuman/flows/builder_tools.rs +++ b/src/openhuman/flows/builder_tools.rs @@ -547,25 +547,28 @@ impl Tool for EditWorkflowTool { } }; - // Write the applied edit back to the draft (the durable working copy), - // so it survives across turns/reloads even if validation/gates below - // still flag something to fix next. - if let Some(ref draft_id) = write_back_draft { - let edited_json = serde_json::to_value(&edited)?; - if let Err(e) = ops::flows_draft_update( - &self.config, - draft_id, - Some(name.clone()), - Some(edited_json), - None, - ) { - tracing::warn!(target: "flows", %draft_id, error = %e, "[flows] edit_workflow: could not write edit back to draft"); + let write_edit_to_draft = || -> anyhow::Result<()> { + if let Some(ref draft_id) = write_back_draft { + let edited_json = serde_json::to_value(&edited)?; + if let Err(e) = ops::flows_draft_update( + &self.config, + draft_id, + Some(name.clone()), + Some(edited_json), + None, + ) { + tracing::warn!(target: "flows", %draft_id, error = %e, "[flows] edit_workflow: could not write edit back to draft"); + } } - } + Ok(()) + }; // Structural validation of the RESULT — surface every problem at once. let structural = tinyflows::validate::validate_all(&edited); if !structural.is_empty() { + // Preserve the longstanding working-copy contract: an applied edit + // survives for the next repair turn even when structurally invalid. + write_edit_to_draft()?; let messages: Vec = structural.iter().map(ToString::to_string).collect(); tracing::debug!( target: "flows", @@ -579,6 +582,30 @@ impl Tool for EditWorkflowTool { ))); } + // Engine-incompatible topologies are different from ordinary builder + // follow-up errors: persisting one would leave a draft that no current + // save/run path can accept. Reject it before advancing the durable + // working copy, while preserving the established write-back behavior + // for later binding/connection/contract gates. + let compatibility = ops::config_aware_engine_compatibility_errors(&self.config, &edited); + if !compatibility.is_empty() { + tracing::debug!( + target: "flows", + %name, + error_count = compatibility.len(), + "[flows] edit_workflow: the edited graph is engine-incompatible" + ); + return Ok(ToolResult::error(format!( + "The edited graph is incompatible with the current engine:\n\n{}\n\nFix the ops and call edit_workflow again.", + compatibility.join("\n\n") + ))); + } + + // Write the accepted structural edit back to the draft (the durable + // working copy), so it survives across turns/reloads even if a later + // binding/connection/contract gate flags something to fix next. + write_edit_to_draft()?; + // Full builder hard-gate stack + proposal payload (shared with revise). // Thread the persistence-state handles so the payload carries draft_id / // flow_id / persisted:false and can't be misread as a save. diff --git a/src/openhuman/flows/builder_tools_tests.rs b/src/openhuman/flows/builder_tools_tests.rs index 2c3ee04e7..af765479a 100644 --- a/src/openhuman/flows/builder_tools_tests.rs +++ b/src/openhuman/flows/builder_tools_tests.rs @@ -2159,18 +2159,224 @@ async fn edit_workflow_accepts_node_id_aliases_end_to_end() { #[tokio::test] async fn edit_workflow_rejects_a_result_that_is_structurally_invalid() { + use crate::openhuman::flows::DraftOrigin; let tmp = TempDir::new().unwrap(); - let tool = EditWorkflowTool::new(test_config(&tmp)); + let config = test_config(&tmp); + let draft = ops::flows_draft_create( + &config, + None, + "Structural repair".to_string(), + valid_graph(), + DraftOrigin::Chat, + ) + .unwrap() + .value; + let tool = EditWorkflowTool::new(config.clone()); // Removing the only trigger leaves the graph structurally invalid. let result = tool .execute(json!({ - "graph": valid_graph(), + "draft_id": draft.id, "ops": [ { "op": "remove_node", "id": "t" } ] })) .await .unwrap(); assert!(result.is_error); assert!(result.output().contains("trigger"), "{}", result.output()); + let reloaded = ops::flows_draft_get(&config, &draft.id).unwrap().value; + assert!( + reloaded.graph["nodes"] + .as_array() + .unwrap() + .iter() + .all(|node| node["id"] != "t"), + "structurally invalid applied edits remain available for the repair turn" + ); +} + +#[tokio::test] +async fn edit_workflow_rejects_an_engine_incompatible_result() { + use crate::openhuman::flows::DraftOrigin; + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let safe_graph = json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Trigger" }, + { "id": "outer", "kind": "condition", "name": "Outer", "config": { "field": "outer" } }, + { "id": "inner", "kind": "condition", "name": "Inner", "config": { "field": "inner" } }, + { "id": "outer_else", "kind": "output_parser", "name": "Outer else" }, + { "id": "inner_else", "kind": "output_parser", "name": "Inner else" }, + { "id": "a", "kind": "output_parser", "name": "A" }, + { "id": "c", "kind": "output_parser", "name": "C" }, + { "id": "m", "kind": "merge", "name": "Merge" } + ], + "edges": [ + { "from_node": "t", "from_port": "main", "to_node": "outer" }, + { "from_node": "t", "from_port": "main", "to_node": "c" }, + { "from_node": "outer", "from_port": "true", "to_node": "inner" }, + { "from_node": "outer", "from_port": "false", "to_node": "outer_else" }, + { "from_node": "inner", "from_port": "true", "to_node": "a" }, + { "from_node": "inner", "from_port": "false", "to_node": "inner_else" }, + { "from_node": "a", "from_port": "main", "to_node": "m" } + ] + }); + let draft = ops::flows_draft_create( + &config, + None, + "Safe draft".to_string(), + safe_graph.clone(), + DraftOrigin::Chat, + ) + .unwrap() + .value; + let tool = EditWorkflowTool::new(config.clone()); + let result = tool + .execute(json!({ + "draft_id": draft.id, + "ops": [ + { "op": "add_edge", "edge": { "from_node": "c", "from_port": "main", "to_node": "m" } } + ] + })) + .await + .unwrap(); + + assert!(result.is_error, "{}", result.output()); + assert!( + result + .output() + .contains("unsupported_nested_conditional_fan_in"), + "{}", + result.output() + ); + let reloaded = ops::flows_draft_get(&config, &draft.id).unwrap().value; + assert_eq!( + reloaded.graph, safe_graph, + "a rejected edit must not advance the durable draft" + ); +} + +#[tokio::test] +async fn edit_workflow_does_not_persist_an_incompatible_saved_child_reference() { + use crate::openhuman::flows::DraftOrigin; + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let legacy_child = json!({ + "nodes": [ + { "id": "start", "kind": "trigger", "name": "Trigger" }, + { "id": "outer", "kind": "condition", "name": "Outer", "config": { "field": "outer" } }, + { "id": "inner", "kind": "condition", "name": "Inner", "config": { "field": "inner" } }, + { "id": "outer_else", "kind": "output_parser", "name": "Outer else" }, + { "id": "inner_else", "kind": "output_parser", "name": "Inner else" }, + { "id": "a", "kind": "output_parser", "name": "A" }, + { "id": "c", "kind": "output_parser", "name": "C" }, + { "id": "m", "kind": "merge", "name": "Merge" } + ], + "edges": [ + { "from_node": "start", "to_node": "outer" }, + { "from_node": "start", "to_node": "c" }, + { "from_node": "outer", "from_port": "true", "to_node": "inner" }, + { "from_node": "outer", "from_port": "false", "to_node": "outer_else" }, + { "from_node": "inner", "from_port": "true", "to_node": "a" }, + { "from_node": "inner", "from_port": "false", "to_node": "inner_else" }, + { "from_node": "a", "to_node": "m" }, + { "from_node": "c", "to_node": "m" } + ] + }); + let child_graph = ops::migrate_and_deserialize_graph(legacy_child).unwrap(); + tinyflows::validate::validate(&child_graph).unwrap(); + let child = crate::openhuman::flows::store::create_flow( + &config, + "Legacy unsafe child".to_string(), + child_graph, + false, + false, + ) + .unwrap(); + let safe_graph = json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Trigger" }, + { + "id": "child", + "kind": "sub_workflow", + "name": "Child", + "config": { "workflow_id": "=inputs.workflow_id" } + } + ], + "edges": [{ "from_node": "t", "to_node": "child" }] + }); + let draft = ops::flows_draft_create( + &config, + None, + "Safe draft".to_string(), + safe_graph.clone(), + DraftOrigin::Chat, + ) + .unwrap() + .value; + + let result = EditWorkflowTool::new(config.clone()) + .execute(json!({ + "draft_id": draft.id, + "ops": [{ + "op": "update_node_config", + "id": "child", + "config": { "workflow_id": child.id } + }] + })) + .await + .unwrap(); + + assert!(result.is_error, "{}", result.output()); + assert!( + result + .output() + .contains("unsupported_nested_conditional_fan_in"), + "{}", + result.output() + ); + let reloaded = ops::flows_draft_get(&config, &draft.id).unwrap().value; + assert_eq!( + reloaded.graph, safe_graph, + "a rejected saved-child edit must not advance the durable draft" + ); +} + +#[tokio::test] +async fn edit_workflow_preserves_non_engine_gate_edits_in_the_draft() { + use crate::openhuman::flows::DraftOrigin; + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let draft = ops::flows_draft_create( + &config, + None, + "Binding follow-up".to_string(), + unresolvable_binding_graph(), + DraftOrigin::Chat, + ) + .unwrap() + .value; + let tool = EditWorkflowTool::new(config.clone()); + let result = tool + .execute(json!({ + "draft_id": draft.id, + "ops": [ + { "op": "set_node_name", "id": "summarize", "name": "Renamed before binding fix" } + ] + })) + .await + .unwrap(); + + assert!( + result.is_error, + "binding gate should still reject the proposal" + ); + let reloaded = ops::flows_draft_get(&config, &draft.id).unwrap().value; + let renamed = reloaded.graph["nodes"] + .as_array() + .unwrap() + .iter() + .find(|node| node["id"] == "summarize") + .unwrap(); + assert_eq!(renamed["name"], "Renamed before binding fix"); } #[tokio::test] diff --git a/src/openhuman/flows/ops.rs b/src/openhuman/flows/ops.rs index e6edc4074..c33c7c75d 100644 --- a/src/openhuman/flows/ops.rs +++ b/src/openhuman/flows/ops.rs @@ -3,6 +3,7 @@ //! `schemas.rs`'s `handle_*` RPC/CLI handlers, mirroring //! `src/openhuman/cron/ops.rs`. +use std::collections::HashSet; use std::sync::Arc; use chrono::Utc; @@ -40,6 +41,11 @@ const FLOW_RUN_TIMEOUT_SECS: u64 = 600; /// this is a dedicated flows-side TTL, not a reuse of the approval store's. const FLOW_PARKED_TTL_SECS: i64 = 600; +/// Stable host-validation code for a topology that the currently vendored +/// TinyFlows/TinyAgents barrier-relief implementation cannot execute safely. +const UNSUPPORTED_NESTED_CONDITIONAL_FAN_IN: &str = "unsupported_nested_conditional_fan_in"; +const UNSUPPORTED_MAIN_PORT_CONDITIONAL_FAN_IN: &str = "unsupported_main_port_conditional_fan_in"; + // ───────────────────────────────────────────────────────────────────────────── // Phase 2 — autonomy-tier gating of acting flow nodes // ───────────────────────────────────────────────────────────────────────────── @@ -93,9 +99,305 @@ const FLOW_PARKED_TTL_SECS: i64 = 600; pub(crate) fn validate_and_migrate_graph(graph_json: Value) -> Result { let graph = migrate_and_deserialize_graph(graph_json)?; tinyflows::validate::validate(&graph).map_err(|e| e.to_string())?; + ensure_engine_compatible(&graph)?; Ok(graph) } +/// Detects fan-in predecessors controlled by more than one branching decision. +/// +/// TinyFlows lowers every fan-in edge as a waiting edge and registers a +/// barrier relief for conditional predecessors. The current lowering chooses +/// only the first upstream brancher, while TinyAgents cannot prove reachability +/// through a second brancher. Depending on node declaration order, that can +/// either relieve the barrier before the real predecessor runs (silently +/// dropping its data) or leave the fan-in unfired. Fail closed until the +/// vendored engine models nested decisions directly. +/// +/// This intentionally mirrors TinyFlows' topology classification rather than +/// limiting the check to `merge` nodes: any node with multiple incoming edges +/// is lowered as a fan-in barrier. A predecessor reachable from the trigger by +/// `main`-only edges is unconditional and needs no relief, so it is safe. +pub(crate) fn engine_compatibility_errors( + graph: &WorkflowGraph, +) -> Vec { + let mut errors = Vec::new(); + collect_engine_compatibility_errors(graph, 0, &mut errors); + errors +} + +fn collect_engine_compatibility_errors( + graph: &WorkflowGraph, + depth: u64, + errors: &mut Vec, +) { + errors.extend(graph_engine_compatibility_errors(graph)); + if depth >= tinyflows::engine::MAX_SUB_WORKFLOW_DEPTH { + return; + } + + for node in &graph.nodes { + if node.kind != NodeKind::SubWorkflow { + continue; + } + let Some(inline) = node.config.get("workflow") else { + continue; + }; + let Ok(child) = serde_json::from_value::(inline.clone()) else { + // TinyFlows reports malformed inline children as capability errors; + // this gate is specifically for otherwise-deserializable unsafe + // topologies. + continue; + }; + let first_child_error = errors.len(); + collect_engine_compatibility_errors(&child, depth + 1, errors); + for error in &mut errors[first_child_error..] { + error.message = format!("Inline sub_workflow node '{}': {}", node.id, error.message); + } + } +} + +fn graph_engine_compatibility_errors( + graph: &WorkflowGraph, +) -> Vec { + let Some(trigger) = graph.trigger() else { + return Vec::new(); + }; + let mut errors = Vec::new(); + + for fan_in in &graph.nodes { + let incoming: Vec<&str> = graph + .edges + .iter() + .filter(|edge| edge.to_node == fan_in.id) + .map(|edge| edge.from_node.as_str()) + .collect(); + if incoming.len() <= 1 { + continue; + } + + for predecessor in incoming { + // Reaching a router itself unconditionally does not make the edge + // it selects into the fan-in unconditional. Let router + // predecessors reach the port-aware analysis below. + if !is_branching_node(graph, predecessor) + && reaches_on_main_edges(graph, &trigger.id, predecessor, &fan_in.id) + { + continue; + } + + let mut controlling_branchers = 0usize; + let mut controlled_via_main_port = false; + for candidate in &graph.nodes { + let is_router = matches!(candidate.kind, NodeKind::Condition | NodeKind::Switch); + let ports: HashSet<&str> = graph + .edges + .iter() + .filter(|edge| edge.from_node == candidate.id) + .map(|edge| edge.from_port.as_str()) + .collect(); + if ports.len() < 2 && !is_router { + continue; + } + // When the router is itself the incoming predecessor, its + // branch edge must be tested against the fan-in (asking whether + // that edge reaches the router again can never succeed). + let controlled_target = if candidate.id == predecessor { + fan_in.id.as_str() + } else { + predecessor + }; + let reaches_from_port = |port: &str| { + reaches_via_port(graph, &candidate.id, port, controlled_target, &fan_in.id) + }; + let any_port_reaches = ports.iter().any(|port| reaches_from_port(port)); + // A router with one wired output still has unwired runtime + // choices that emit no successor, so that sole edge cannot + // prove unconditional reachability. Router reconvergence is + // only deterministic when every runtime choice is wired: + // both condition outcomes, or a switch fallback. Generic + // multi-port nodes retain their existing all-port behavior. + let routing_choices_are_exhaustive = match candidate.kind { + NodeKind::Condition => ports.contains("true") && ports.contains("false"), + NodeKind::Switch => ports.contains("default"), + _ => true, + }; + let can_prove_all_routing_choices = if is_router { + routing_choices_are_exhaustive + } else { + ports.len() >= 2 + }; + let every_port_deterministically_reaches = can_prove_all_routing_choices + && ports.iter().all(|port| { + reaches_deterministically_via_port( + graph, + &candidate.id, + port, + controlled_target, + &fan_in.id, + ) + }); + // A multi-port node only controls this predecessor when the + // predecessor is reachable from it but not guaranteed by a + // deterministic path on every routing choice. This matches + // TinyAgents' relief proof, which stops at another router. + if any_port_reaches && !every_port_deterministically_reaches { + controlling_branchers += 1; + controlled_via_main_port |= ports.contains("main") && reaches_from_port("main"); + } + } + + let (code, routing_kind) = if controlled_via_main_port { + ( + UNSUPPORTED_MAIN_PORT_CONDITIONAL_FAN_IN, + "a conditional branch labelled 'main'", + ) + } else if controlling_branchers >= 2 { + ( + UNSUPPORTED_NESTED_CONDITIONAL_FAN_IN, + "nested conditional routing", + ) + } else { + continue; + }; + errors.push(crate::openhuman::flows::FlowValidationError { + code: code.to_string(), + message: format!( + "Fan-in node '{}' has predecessor '{}' behind {routing_kind}; \ + this topology is temporarily unsupported because it can silently lose \ + merged data. Flatten the conditional branch or join it before this fan-in.", + fan_in.id, predecessor + ), + node_id: Some(fan_in.id.clone()), + field: None, + }); + } + } + + errors +} + +fn ensure_engine_compatible(graph: &WorkflowGraph) -> Result<(), String> { + match engine_compatibility_errors(graph).into_iter().next() { + Some(error) => Err(format!("{}: {}", error.code, error.message)), + None => Ok(()), + } +} + +/// Host-aware compatibility check, including saved descendants that graph-only +/// validation cannot inspect. Authoring boundaries use it before persistence; +/// execution boundaries use it before compiling a root run/resume or returning +/// a resolver graph, so an unsafe descendant cannot run after earlier effects. +fn ensure_config_aware_engine_compatible( + config: &Config, + graph: &WorkflowGraph, +) -> Result<(), String> { + match config_aware_engine_compatibility_errors(config, graph) + .into_iter() + .next() + { + Some(error) => Err(error), + None => Ok(()), + } +} + +fn reaches_on_main_edges(graph: &WorkflowGraph, from: &str, to: &str, stop: &str) -> bool { + if from == to { + return true; + } + let mut stack: Vec<&str> = if is_branching_node(graph, from) { + Vec::new() + } else { + graph + .edges + .iter() + .filter(|edge| edge.from_node == from && edge.from_port == "main") + .map(|edge| edge.to_node.as_str()) + .collect() + }; + let mut seen = HashSet::new(); + while let Some(node) = stack.pop() { + if node == to { + return true; + } + if node == stop || !seen.insert(node) { + continue; + } + // Port labels are arbitrary. A node with multiple distinct output + // ports is runtime-selective even when one label happens to be `main`, + // so nothing beyond it is unconditionally reachable. + if is_branching_node(graph, node) { + continue; + } + stack.extend( + graph + .edges + .iter() + .filter(|edge| edge.from_node == node && edge.from_port == "main") + .map(|edge| edge.to_node.as_str()), + ); + } + false +} + +fn is_branching_node(graph: &WorkflowGraph, node_id: &str) -> bool { + graph.nodes.iter().any(|node| { + node.id == node_id && matches!(node.kind, NodeKind::Condition | NodeKind::Switch) + }) || graph + .edges + .iter() + .filter(|edge| edge.from_node == node_id) + .map(|edge| edge.from_port.as_str()) + .collect::>() + .len() + >= 2 +} + +fn reaches_via_port( + graph: &WorkflowGraph, + brancher: &str, + port: &str, + target: &str, + stop: &str, +) -> bool { + let mut stack: Vec<&str> = graph + .edges + .iter() + .filter(|edge| edge.from_node == brancher && edge.from_port == port) + .map(|edge| edge.to_node.as_str()) + .collect(); + let mut seen = HashSet::new(); + while let Some(node) = stack.pop() { + if node == target { + return true; + } + if node == stop || !seen.insert(node) { + continue; + } + stack.extend( + graph + .edges + .iter() + .filter(|edge| edge.from_node == node) + .map(|edge| edge.to_node.as_str()), + ); + } + false +} + +fn reaches_deterministically_via_port( + graph: &WorkflowGraph, + brancher: &str, + port: &str, + target: &str, + stop: &str, +) -> bool { + graph + .edges + .iter() + .filter(|edge| edge.from_node == brancher && edge.from_port == port) + .any(|edge| reaches_on_main_edges(graph, &edge.to_node, target, stop)) +} + /// Runs a raw graph JSON value through migration + deserialization **without** /// the structural `validate` step. Splits the two so a caller that wants /// *every* structural error (via `tinyflows::validate::validate_all`) can run @@ -123,10 +425,10 @@ pub(crate) fn to_flow_validation_error( } } -/// The single canonical definition of the builder hard-gate stack: the three +/// The single canonical definition of the builder hard-gate stack: the four /// author-time gates that reject (not warn) a graph an agent must not propose -/// or persist — binding-resolvability, tool-contract, and required-arg -/// resolvability, in increasing cost order. +/// or persist — engine compatibility, binding-resolvability, tool-contract, +/// and required-arg resolvability, in increasing cost order. /// /// Returns an empty `Vec` when the graph passes; otherwise the first failing /// gate's node-level error messages (short-circuiting, so an expensive later @@ -140,6 +442,10 @@ pub(crate) fn to_flow_validation_error( /// `validate_and_migrate_graph` / `validate_all` first) — these gates check /// resolvability/contracts on a compilable graph. pub(crate) async fn run_builder_gates(config: &Config, graph: &WorkflowGraph) -> Vec { + let compatibility_errors = config_aware_engine_compatibility_errors(config, graph); + if !compatibility_errors.is_empty() { + return compatibility_errors; + } // Cheap, sync: a binding guaranteed to resolve null / wrong at runtime. let binding_errors = validate_binding_resolvability(graph); if !binding_errors.is_empty() { @@ -165,6 +471,106 @@ pub(crate) async fn run_builder_gates(config: &Config, graph: &WorkflowGraph) -> validate_required_arg_resolvability(graph).await } +/// Checks literal `workflow_id` children reachable from an authoring candidate. +/// +/// Pure graph validation can recurse through inline children, but resolving a +/// saved child requires the host store. Keep that lookup in the config-aware +/// builder gate so strict RPC and agent-authored proposals/saves cannot bless a +/// parent that is already known to fail at execution. Dynamic `=` expressions, +/// missing ids, and store failures retain their existing runtime diagnostics; +/// this gate only rejects a saved graph whose topology is demonstrably unsafe. +fn referenced_workflow_compatibility_errors(config: &Config, graph: &WorkflowGraph) -> Vec { + let mut pending = vec![(graph.clone(), 0_u64, Vec::::new())]; + // Record the shallowest visit, not just whether an id was seen. The same + // child can be referenced by multiple branches; a deep DFS visit must not + // suppress a later shallower visit that has more depth budget remaining. + let mut visited_depths = std::collections::HashMap::::new(); + + while let Some((current, depth, path)) = pending.pop() { + if depth >= tinyflows::engine::MAX_SUB_WORKFLOW_DEPTH { + continue; + } + + for node in ¤t.nodes { + if node.kind != NodeKind::SubWorkflow { + continue; + } + + let mut child_path = path.clone(); + child_path.push(node.id.clone()); + + let inline = node.config.get("workflow"); + let configured_workflow_id = node + .config + .get("workflow_id") + .and_then(Value::as_str) + .map(str::trim) + .filter(|id| !id.is_empty()); + // Structural validation requires exactly one source and runs before + // this helper. Retain that precedence defensively if a future caller + // passes an invalid graph directly: do not inspect either source as + // though TinyFlows could choose between them at runtime. + if inline.is_some() && configured_workflow_id.is_some() { + continue; + } + + if let Some(inline) = inline { + if let Ok(child) = serde_json::from_value::(inline.clone()) { + pending.push((child, depth + 1, child_path.clone())); + } + continue; + } + + let Some(workflow_id) = configured_workflow_id.filter(|id| !id.starts_with('=')) else { + continue; + }; + let child_depth = depth + 1; + if visited_depths + .get(workflow_id) + .is_some_and(|seen_depth| *seen_depth <= child_depth) + { + continue; + } + visited_depths.insert(workflow_id.to_string(), child_depth); + + let Ok(Some(child)) = load_flow_graph(config, workflow_id) else { + continue; + }; + if let Some(error) = engine_compatibility_errors(&child).into_iter().next() { + return vec![format!( + "Sub_workflow path '{}' references workflow_id '{}' with an unsupported \ + engine topology: {}: {}", + child_path.join(" -> "), + workflow_id, + error.code, + error.message + )]; + } + pending.push((child, child_depth, child_path)); + } + } + + Vec::new() +} + +/// Returns the complete engine-topology gate for a graph in its host context. +/// The graph-only half covers inline descendants; the config-aware half follows +/// literal saved-workflow references. Authoring and execution boundaries share +/// this helper so neither can accept a graph the other must reject. +pub(crate) fn config_aware_engine_compatibility_errors( + config: &Config, + graph: &WorkflowGraph, +) -> Vec { + let direct = engine_compatibility_errors(graph); + if !direct.is_empty() { + return direct + .into_iter() + .map(|error| format!("{}: {}", error.code, error.message)) + .collect(); + } + referenced_workflow_compatibility_errors(config, graph) +} + /// Strict-mode gate for the create/update RPC path (audit F3): validates /// `graph_json` structurally (surfacing every error at once) and then runs the /// same [`run_builder_gates`] the agent tools enforce, returning `Err` with a @@ -198,8 +604,9 @@ pub(crate) async fn strict_gate(config: &Config, graph_json: &Value) -> Result<( /// `graph` and, if it passes, builds the `workflow_proposal` payload the /// propose/revise/edit tools all return. /// -/// The single home for the gate sequence (binding-resolvability → -/// tool-contract → required-arg resolvability) plus summary/warning assembly, +/// The single home for the gate sequence (engine compatibility → +/// binding-resolvability → tool-contract → required-arg resolvability) plus +/// summary/warning assembly, /// so `revise_workflow` and `edit_workflow` cannot drift. `retry_tool` names /// the tool in the "fix … and call `` again" guidance so each caller's /// error text points the agent back at the right tool. @@ -2053,6 +2460,28 @@ pub fn flows_validate(graph_json: Value) -> RpcOutcome Result, String> { let graph = validate_and_migrate_graph(graph_json)?; + ensure_config_aware_engine_compatible(config, &graph)?; // Rule 1: automatic triggers create DISABLED — the user must arm them // explicitly. @@ -2310,6 +2740,21 @@ pub fn load_flow_graph(config: &Config, id: &str) -> Result Result, String> { + let graph = load_flow_graph(config, id)?; + if let Some(graph) = graph.as_ref() { + ensure_config_aware_engine_compatible(config, graph) + .map_err(|error| format!("workflow_id '{id}' is engine-incompatible: {error}"))?; + } + Ok(graph) +} + /// Lists every saved flow. pub async fn flows_list(config: &Config) -> Result>, String> { let flows = store::list_flows(config).map_err(|e| e.to_string())?; @@ -2632,13 +3077,16 @@ pub async fn flows_update( let new_require_approval = require_approval.unwrap_or(existing.require_approval); let graph_changed = graph_json.is_some(); let graph = match graph_json { - Some(raw) => validate_and_migrate_graph(raw)?, + Some(raw) => { + let graph = validate_and_migrate_graph(raw)?; + ensure_config_aware_engine_compatible(config, &graph)?; + graph + } None => { tinyflows::validate::validate(&existing.graph).map_err(|e| e.to_string())?; existing.graph.clone() } }; - // B29 Rule 1 analogue: disarm every manual/none → automatic trigger // transition, unconditionally — see the doc comment above for why this // must NOT gate on the (possibly stale) `existing.enabled` read. @@ -3105,6 +3553,19 @@ pub async fn flows_run( // `store::get_flow` already ran the stored `graph_json` through // `tinyflows::migrate::migrate` before deserializing, so `flow.graph` is // always on the current schema here. + // + // Author-time validation cannot protect definitions persisted by an older + // OpenHuman build. Re-check immediately before compilation so an upgrade + // fails explicitly instead of silently committing incomplete merge data. + if let Err(error) = ensure_config_aware_engine_compatible(config, &flow.graph) { + tracing::warn!( + target: "flows", + flow_id = %flow_id, + %error, + "[flows] flows_run: rejected — unsupported engine topology" + ); + return Err(error); + } let compiled = tinyflows::compiler::compile(&flow.graph).map_err(|e| e.to_string())?; let config_arc = Arc::new(config.clone()); @@ -3385,6 +3846,29 @@ pub async fn flows_resume( )); } + // A pending checkpoint may have been created before this compatibility + // gate shipped, so resume is an independent authoritative boundary. + if let Err(error) = ensure_config_aware_engine_compatible(config, &flow.graph) { + if let Err(rec_err) = store::record_run(config, flow_id, "failed") { + tracing::warn!( + target: "flows", + flow_id = %flow_id, + %thread_id, + error = %rec_err, + "[flows] flows_resume: failed to record compatibility rejection" + ); + } + let observed = current_persisted_steps(config, thread_id); + finish_flow_run_row(config, thread_id, "failed", &observed, &[], Some(&error)); + tracing::warn!( + target: "flows", + flow_id = %flow_id, + %thread_id, + %error, + "[flows] flows_resume: rejected — unsupported engine topology" + ); + return Err(error); + } let compiled = tinyflows::compiler::compile(&flow.graph).map_err(|e| e.to_string())?; let config_arc = Arc::new(config.clone()); let caps = diff --git a/src/openhuman/flows/ops_tests.rs b/src/openhuman/flows/ops_tests.rs index 54163d502..9eccff8e0 100644 --- a/src/openhuman/flows/ops_tests.rs +++ b/src/openhuman/flows/ops_tests.rs @@ -24,6 +24,639 @@ fn trigger_only_graph() -> Value { }) } +fn nested_conditional_fan_in_graph() -> Value { + json!({ + "name": "nested-conditional-fan-in", + "nodes": [ + { "id": "start", "kind": "trigger", "name": "Trigger" }, + { "id": "outer", "kind": "condition", "name": "Outer", "config": { "field": "outer" } }, + { "id": "inner", "kind": "condition", "name": "Inner", "config": { "field": "inner" } }, + { "id": "outer_else", "kind": "output_parser", "name": "Outer else" }, + { "id": "inner_else", "kind": "output_parser", "name": "Inner else" }, + { "id": "a", "kind": "output_parser", "name": "A" }, + { "id": "c", "kind": "output_parser", "name": "C" }, + { "id": "m", "kind": "merge", "name": "Merge" } + ], + "edges": [ + { "from_node": "start", "from_port": "main", "to_node": "outer" }, + { "from_node": "start", "from_port": "main", "to_node": "c" }, + { "from_node": "outer", "from_port": "true", "to_node": "inner" }, + { "from_node": "outer", "from_port": "false", "to_node": "outer_else" }, + { "from_node": "inner", "from_port": "true", "to_node": "a" }, + { "from_node": "inner", "from_port": "false", "to_node": "inner_else" }, + { "from_node": "a", "from_port": "main", "to_node": "m" }, + { "from_node": "c", "from_port": "main", "to_node": "m" } + ] + }) +} + +fn main_port_conditional_fan_in_graph() -> Value { + json!({ + "name": "main-port-conditional-fan-in", + "nodes": [ + { "id": "start", "kind": "trigger", "name": "Trigger" }, + { "id": "route", "kind": "switch", "name": "Route", "config": { "field": "kind" } }, + { "id": "a", "kind": "output_parser", "name": "A" }, + { "id": "other", "kind": "output_parser", "name": "Other" }, + { "id": "c", "kind": "output_parser", "name": "C" }, + { "id": "m", "kind": "merge", "name": "Merge" } + ], + "edges": [ + { "from_node": "start", "from_port": "main", "to_node": "route" }, + { "from_node": "start", "from_port": "main", "to_node": "c" }, + { "from_node": "route", "from_port": "main", "to_node": "a" }, + { "from_node": "route", "from_port": "other", "to_node": "other" }, + { "from_node": "a", "from_port": "main", "to_node": "m" }, + { "from_node": "c", "from_port": "main", "to_node": "m" } + ] + }) +} + +fn referenced_child_graph(workflow_id: &str) -> Value { + json!({ + "name": "parent-with-saved-child", + "nodes": [ + { "id": "start", "kind": "trigger", "name": "Trigger" }, + { + "id": "saved-child", + "kind": "sub_workflow", + "name": "Saved child", + "config": { "workflow_id": workflow_id } + } + ], + "edges": [ + { "from_node": "start", "from_port": "main", "to_node": "saved-child" } + ] + }) +} + +fn structurally_valid_graph(value: Value) -> WorkflowGraph { + let graph = migrate_and_deserialize_graph(value).expect("graph should deserialize"); + tinyflows::validate::validate(&graph).expect("fixture should be structurally valid"); + graph +} + +fn nested_router_reconvergence_graph(inner_kind: &str, inner_ports: &[&str]) -> WorkflowGraph { + let mut edges = vec![ + json!({ "from_node": "start", "from_port": "main", "to_node": "outer" }), + json!({ "from_node": "start", "from_port": "main", "to_node": "c" }), + json!({ "from_node": "outer", "from_port": "true", "to_node": "inner" }), + json!({ "from_node": "outer", "from_port": "false", "to_node": "outer_else" }), + ]; + edges.extend( + inner_ports + .iter() + .map(|port| json!({ "from_node": "inner", "from_port": port, "to_node": "a" })), + ); + edges.extend([ + json!({ "from_node": "a", "from_port": "main", "to_node": "m" }), + json!({ "from_node": "c", "from_port": "main", "to_node": "m" }), + ]); + + structurally_valid_graph(json!({ + "name": "nested-router-reconvergence", + "nodes": [ + { "id": "start", "kind": "trigger", "name": "Trigger" }, + { "id": "outer", "kind": "condition", "name": "Outer", "config": { "field": "outer" } }, + { "id": "inner", "kind": inner_kind, "name": "Inner", "config": { "field": "inner" } }, + { "id": "outer_else", "kind": "output_parser", "name": "Outer else" }, + { "id": "a", "kind": "output_parser", "name": "A" }, + { "id": "c", "kind": "output_parser", "name": "C" }, + { "id": "m", "kind": "merge", "name": "Merge" } + ], + "edges": edges + })) +} + +#[test] +fn engine_compatibility_distinguishes_nested_from_safe_fan_ins() { + let risky = structurally_valid_graph(nested_conditional_fan_in_graph()); + let errors = engine_compatibility_errors(&risky); + assert_eq!(errors.len(), 1); + assert_eq!(errors[0].code, UNSUPPORTED_NESTED_CONDITIONAL_FAN_IN); + assert_eq!(errors[0].node_id.as_deref(), Some("m")); + + let one_level = structurally_valid_graph(json!({ + "name": "one-level-mixed-fan-in", + "nodes": [ + { "id": "start", "kind": "trigger", "name": "Trigger" }, + { "id": "cond", "kind": "condition", "name": "Condition", "config": { "field": "flag" } }, + { "id": "a", "kind": "output_parser", "name": "A" }, + { "id": "other", "kind": "output_parser", "name": "Other" }, + { "id": "c", "kind": "output_parser", "name": "C" }, + { "id": "m", "kind": "merge", "name": "Merge" } + ], + "edges": [ + { "from_node": "start", "from_port": "main", "to_node": "cond" }, + { "from_node": "start", "from_port": "main", "to_node": "c" }, + { "from_node": "cond", "from_port": "true", "to_node": "a" }, + { "from_node": "cond", "from_port": "false", "to_node": "other" }, + { "from_node": "a", "from_port": "main", "to_node": "m" }, + { "from_node": "c", "from_port": "main", "to_node": "m" } + ] + })); + assert!(engine_compatibility_errors(&one_level).is_empty()); + + let nested_without_fan_in = structurally_valid_graph(json!({ + "name": "nested-without-fan-in", + "nodes": [ + { "id": "start", "kind": "trigger", "name": "Trigger" }, + { "id": "outer", "kind": "condition", "name": "Outer", "config": { "field": "outer" } }, + { "id": "inner", "kind": "condition", "name": "Inner", "config": { "field": "inner" } }, + { "id": "outer_else", "kind": "output_parser", "name": "Outer else" }, + { "id": "a", "kind": "output_parser", "name": "A" }, + { "id": "inner_else", "kind": "output_parser", "name": "Inner else" } + ], + "edges": [ + { "from_node": "start", "from_port": "main", "to_node": "outer" }, + { "from_node": "outer", "from_port": "true", "to_node": "inner" }, + { "from_node": "outer", "from_port": "false", "to_node": "outer_else" }, + { "from_node": "inner", "from_port": "true", "to_node": "a" }, + { "from_node": "inner", "from_port": "false", "to_node": "inner_else" } + ] + })); + assert!(engine_compatibility_errors(&nested_without_fan_in).is_empty()); + + let unconditional = structurally_valid_graph(json!({ + "name": "unconditional-fan-in", + "nodes": [ + { "id": "start", "kind": "trigger", "name": "Trigger" }, + { "id": "a", "kind": "output_parser", "name": "A" }, + { "id": "c", "kind": "output_parser", "name": "C" }, + { "id": "m", "kind": "merge", "name": "Merge" } + ], + "edges": [ + { "from_node": "start", "from_port": "main", "to_node": "a" }, + { "from_node": "start", "from_port": "main", "to_node": "c" }, + { "from_node": "a", "from_port": "main", "to_node": "m" }, + { "from_node": "c", "from_port": "main", "to_node": "m" } + ] + })); + assert!(engine_compatibility_errors(&unconditional).is_empty()); +} + +#[test] +fn engine_compatibility_rejects_main_label_on_conditional_fan_in_path() { + let graph = structurally_valid_graph(main_port_conditional_fan_in_graph()); + let errors = engine_compatibility_errors(&graph); + assert_eq!(errors.len(), 1); + assert_eq!(errors[0].code, UNSUPPORTED_MAIN_PORT_CONDITIONAL_FAN_IN); + assert_eq!(errors[0].node_id.as_deref(), Some("m")); + + let reconverged = structurally_valid_graph(json!({ + "name": "main-port-reconverges-before-fan-in", + "nodes": [ + { "id": "start", "kind": "trigger", "name": "Trigger" }, + { "id": "route", "kind": "switch", "name": "Route", "config": { "field": "kind" } }, + { "id": "a", "kind": "output_parser", "name": "A" }, + { "id": "c", "kind": "output_parser", "name": "C" }, + { "id": "m", "kind": "merge", "name": "Merge" } + ], + "edges": [ + { "from_node": "start", "from_port": "main", "to_node": "route" }, + { "from_node": "start", "from_port": "main", "to_node": "c" }, + { "from_node": "route", "from_port": "main", "to_node": "a" }, + { "from_node": "route", "from_port": "default", "to_node": "a" }, + { "from_node": "a", "from_port": "main", "to_node": "m" }, + { "from_node": "c", "from_port": "main", "to_node": "m" } + ] + })); + assert!(engine_compatibility_errors(&reconverged).is_empty()); +} + +#[test] +fn engine_compatibility_requires_exhaustive_router_choices_for_reconvergence() { + let exhaustive_condition = nested_router_reconvergence_graph("condition", &["true", "false"]); + assert!(engine_compatibility_errors(&exhaustive_condition).is_empty()); + + let missing_condition_branch = nested_router_reconvergence_graph("condition", &["true"]); + let errors = engine_compatibility_errors(&missing_condition_branch); + assert_eq!(errors.len(), 1); + assert_eq!(errors[0].code, UNSUPPORTED_NESTED_CONDITIONAL_FAN_IN); + + let exhaustive_switch = nested_router_reconvergence_graph("switch", &["known-case", "default"]); + assert!(engine_compatibility_errors(&exhaustive_switch).is_empty()); + + // Same-port fan-out is unconditional: TinyFlows schedules both `main` + // successors. A side path after an exhaustive router must not make the + // reconverging path look like another conditional choice. + let exhaustive_switch_with_main_fanout = structurally_valid_graph(json!({ + "nodes": [ + { "id": "start", "kind": "trigger", "name": "Trigger" }, + { "id": "outer", "kind": "condition", "name": "Outer", "config": { "field": "outer" } }, + { "id": "inner", "kind": "switch", "name": "Inner", "config": { "field": "inner" } }, + { "id": "outer_else", "kind": "output_parser", "name": "Outer else" }, + { "id": "fanout", "kind": "output_parser", "name": "Fan out" }, + { "id": "a", "kind": "output_parser", "name": "A" }, + { "id": "side", "kind": "output_parser", "name": "Side" }, + { "id": "c", "kind": "output_parser", "name": "C" }, + { "id": "m", "kind": "merge", "name": "Merge" } + ], + "edges": [ + { "from_node": "start", "from_port": "main", "to_node": "outer" }, + { "from_node": "start", "from_port": "main", "to_node": "c" }, + { "from_node": "outer", "from_port": "true", "to_node": "inner" }, + { "from_node": "outer", "from_port": "false", "to_node": "outer_else" }, + { "from_node": "inner", "from_port": "known-case", "to_node": "fanout" }, + { "from_node": "inner", "from_port": "default", "to_node": "fanout" }, + { "from_node": "fanout", "from_port": "main", "to_node": "a" }, + { "from_node": "fanout", "from_port": "main", "to_node": "side" }, + { "from_node": "a", "from_port": "main", "to_node": "m" }, + { "from_node": "c", "from_port": "main", "to_node": "m" } + ] + })); + assert!(engine_compatibility_errors(&exhaustive_switch_with_main_fanout).is_empty()); + + // A switch with only `default` is exhaustive: every input takes that edge, + // so it is an unconditional step even though it has a single wired port. + let default_only_switch = nested_router_reconvergence_graph("switch", &["default"]); + assert!(engine_compatibility_errors(&default_only_switch).is_empty()); + + let missing_switch_default = + nested_router_reconvergence_graph("switch", &["known-case", "other-case"]); + let errors = engine_compatibility_errors(&missing_switch_default); + assert!(!errors.is_empty()); + assert!(errors + .iter() + .all(|error| error.code == UNSUPPORTED_NESTED_CONDITIONAL_FAN_IN)); + // Both the switch's own reconvergence and the downstream merge are unsafe; + // multiple switch ports may also report the same predecessor. Pin the + // affected fan-ins without coupling the test to diagnostic multiplicity. + assert!(errors + .iter() + .any(|error| error.node_id.as_deref() == Some("a"))); + assert!(errors + .iter() + .any(|error| error.node_id.as_deref() == Some("m"))); +} + +#[test] +fn engine_compatibility_rejects_reconvergence_before_nested_router() { + let graph = structurally_valid_graph(json!({ + "name": "reconverged-before-nested-router", + "nodes": [ + { "id": "start", "kind": "trigger", "name": "Trigger" }, + { "id": "outer", "kind": "condition", "name": "Outer", "config": { "field": "outer" } }, + { "id": "inner", "kind": "condition", "name": "Inner", "config": { "field": "inner" } }, + { "id": "a", "kind": "output_parser", "name": "A" }, + { "id": "inner_else", "kind": "output_parser", "name": "Inner else" }, + { "id": "c", "kind": "output_parser", "name": "C" }, + { "id": "m", "kind": "merge", "name": "Merge" } + ], + "edges": [ + { "from_node": "start", "from_port": "main", "to_node": "outer" }, + { "from_node": "start", "from_port": "main", "to_node": "c" }, + { "from_node": "outer", "from_port": "true", "to_node": "inner" }, + { "from_node": "outer", "from_port": "false", "to_node": "inner" }, + { "from_node": "inner", "from_port": "true", "to_node": "a" }, + { "from_node": "inner", "from_port": "false", "to_node": "inner_else" }, + { "from_node": "a", "from_port": "main", "to_node": "m" }, + { "from_node": "c", "from_port": "main", "to_node": "m" } + ] + })); + let errors = engine_compatibility_errors(&graph); + assert_eq!(errors.len(), 1); + assert_eq!(errors[0].code, UNSUPPORTED_NESTED_CONDITIONAL_FAN_IN); +} + +#[test] +fn engine_compatibility_treats_single_wired_router_outputs_as_conditional() { + let graph = structurally_valid_graph(json!({ + "name": "single-wired-nested-router-fan-in", + "nodes": [ + { "id": "start", "kind": "trigger", "name": "Trigger" }, + { "id": "outer", "kind": "switch", "name": "Outer", "config": { "field": "outer" } }, + { "id": "inner", "kind": "condition", "name": "Inner", "config": { "field": "inner" } }, + { "id": "a", "kind": "output_parser", "name": "A" }, + { "id": "c", "kind": "output_parser", "name": "C" }, + { "id": "m", "kind": "merge", "name": "Merge" } + ], + "edges": [ + { "from_node": "start", "from_port": "main", "to_node": "outer" }, + { "from_node": "start", "from_port": "main", "to_node": "c" }, + { "from_node": "outer", "from_port": "case", "to_node": "inner" }, + { "from_node": "inner", "from_port": "true", "to_node": "a" }, + { "from_node": "a", "from_port": "main", "to_node": "m" }, + { "from_node": "c", "from_port": "main", "to_node": "m" } + ] + })); + + let errors = engine_compatibility_errors(&graph); + assert_eq!(errors.len(), 1); + assert_eq!(errors[0].code, UNSUPPORTED_NESTED_CONDITIONAL_FAN_IN); + assert_eq!(errors[0].node_id.as_deref(), Some("m")); +} + +#[test] +fn engine_compatibility_detects_a_router_directly_preceding_fan_in() { + let nested = structurally_valid_graph(json!({ + "name": "direct-nested-router-fan-in", + "nodes": [ + { "id": "start", "kind": "trigger", "name": "Trigger" }, + { "id": "outer", "kind": "switch", "name": "Outer", "config": { "field": "outer" } }, + { "id": "inner", "kind": "condition", "name": "Inner", "config": { "field": "inner" } }, + { "id": "c", "kind": "output_parser", "name": "C" }, + { "id": "m", "kind": "merge", "name": "Merge" } + ], + "edges": [ + { "from_node": "start", "from_port": "main", "to_node": "outer" }, + { "from_node": "start", "from_port": "main", "to_node": "c" }, + { "from_node": "outer", "from_port": "case", "to_node": "inner" }, + { "from_node": "inner", "from_port": "true", "to_node": "m" }, + { "from_node": "c", "from_port": "main", "to_node": "m" } + ] + })); + let errors = engine_compatibility_errors(&nested); + assert_eq!(errors.len(), 1); + assert_eq!(errors[0].code, UNSUPPORTED_NESTED_CONDITIONAL_FAN_IN); + + let main_port = structurally_valid_graph(json!({ + "name": "direct-main-port-router-fan-in", + "nodes": [ + { "id": "start", "kind": "trigger", "name": "Trigger" }, + { "id": "route", "kind": "switch", "name": "Route", "config": { "field": "kind" } }, + { "id": "c", "kind": "output_parser", "name": "C" }, + { "id": "m", "kind": "merge", "name": "Merge" } + ], + "edges": [ + { "from_node": "start", "from_port": "main", "to_node": "route" }, + { "from_node": "start", "from_port": "main", "to_node": "c" }, + { "from_node": "route", "from_port": "main", "to_node": "m" }, + { "from_node": "c", "from_port": "main", "to_node": "m" } + ] + })); + let errors = engine_compatibility_errors(&main_port); + assert_eq!(errors.len(), 1); + assert_eq!(errors[0].code, UNSUPPORTED_MAIN_PORT_CONDITIONAL_FAN_IN); +} + +#[test] +fn engine_compatibility_recurses_through_nested_inline_sub_workflows() { + let unsafe_child = nested_conditional_fan_in_graph(); + let middle = json!({ + "nodes": [ + { "id": "middle-trigger", "kind": "trigger", "name": "Trigger" }, + { + "id": "inner-child", + "kind": "sub_workflow", + "name": "Inner child", + "config": { "workflow": unsafe_child } + } + ], + "edges": [ + { "from_node": "middle-trigger", "from_port": "main", "to_node": "inner-child" } + ] + }); + let parent = structurally_valid_graph(json!({ + "nodes": [ + { "id": "parent-trigger", "kind": "trigger", "name": "Trigger" }, + { + "id": "middle-child", + "kind": "sub_workflow", + "name": "Middle child", + "config": { "workflow": middle } + } + ], + "edges": [ + { "from_node": "parent-trigger", "from_port": "main", "to_node": "middle-child" } + ] + })); + + let errors = engine_compatibility_errors(&parent); + assert_eq!(errors.len(), 1); + assert_eq!(errors[0].code, UNSUPPORTED_NESTED_CONDITIONAL_FAN_IN); + assert!(errors[0].message.contains("middle-child")); + assert!(errors[0].message.contains("inner-child")); +} + +#[test] +fn resolver_lookup_rejects_an_incompatible_saved_child() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let child = store::create_flow( + &config, + "legacy child".to_string(), + structurally_valid_graph(nested_conditional_fan_in_graph()), + false, + false, + ) + .unwrap(); + + let error = load_engine_compatible_flow_graph(&config, &child.id) + .expect_err("resolver lookup must reject an unsafe legacy child"); + assert!( + error.contains(UNSUPPORTED_NESTED_CONDITIONAL_FAN_IN), + "{error}" + ); + assert!(error.contains(&child.id), "{error}"); +} + +#[test] +fn resolver_lookup_rejects_an_incompatible_saved_grandchild() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let grandchild = store::create_flow( + &config, + "legacy unsafe grandchild".to_string(), + structurally_valid_graph(nested_conditional_fan_in_graph()), + false, + false, + ) + .unwrap(); + let child = store::create_flow( + &config, + "saved child".to_string(), + structurally_valid_graph(referenced_child_graph(&grandchild.id)), + false, + false, + ) + .unwrap(); + + let error = load_engine_compatible_flow_graph(&config, &child.id) + .expect_err("resolver lookup must reject an unsafe saved grandchild"); + assert!( + error.contains(UNSUPPORTED_NESTED_CONDITIONAL_FAN_IN), + "{error}" + ); + assert!(error.contains(&child.id), "{error}"); + assert!(error.contains(&grandchild.id), "{error}"); + assert!(error.contains("saved-child"), "{error}"); +} + +#[test] +fn flows_validate_returns_stable_nested_conditional_fan_in_error() { + let outcome = flows_validate(nested_conditional_fan_in_graph()); + assert!(!outcome.value.valid); + assert_eq!(outcome.value.error_details.len(), 1); + assert_eq!( + outcome.value.error_details[0].code, + UNSUPPORTED_NESTED_CONDITIONAL_FAN_IN + ); + assert_eq!(outcome.value.error_details[0].node_id.as_deref(), Some("m")); + assert!(outcome.value.warnings.is_empty()); +} + +#[tokio::test] +async fn flows_run_rejects_legacy_nested_conditional_fan_in_before_execution() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + // Bypass the current author-time gate to simulate a definition persisted + // by an older OpenHuman build. Reads remain supported; execution does not. + let graph = structurally_valid_graph(nested_conditional_fan_in_graph()); + let flow = store::create_flow(&config, "legacy".to_string(), graph, false, true).unwrap(); + + let err = flows_run( + &config, + &flow.id, + json!({ "outer": true, "inner": true }), + FlowRunTrigger::Rpc, + ) + .await + .expect_err("legacy unsafe topology must fail closed"); + assert!(err.contains(UNSUPPORTED_NESTED_CONDITIONAL_FAN_IN), "{err}"); + + let reloaded = flows_get(&config, &flow.id).await.unwrap(); + assert_eq!(reloaded.value.last_status, None); + assert_eq!( + reloaded.value.graph, flow.graph, + "stored graph must be preserved" + ); +} + +#[tokio::test] +async fn flows_run_rejects_an_incompatible_saved_child_before_execution() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let child = store::create_flow( + &config, + "legacy unsafe child".to_string(), + structurally_valid_graph(nested_conditional_fan_in_graph()), + false, + false, + ) + .unwrap(); + let parent = store::create_flow( + &config, + "parent".to_string(), + structurally_valid_graph(referenced_child_graph(&child.id)), + false, + true, + ) + .unwrap(); + + let error = flows_run(&config, &parent.id, json!({}), FlowRunTrigger::Rpc) + .await + .expect_err("an unsafe saved child must fail before root execution starts"); + assert!( + error.contains(UNSUPPORTED_NESTED_CONDITIONAL_FAN_IN), + "{error}" + ); + assert!(error.contains(&child.id), "{error}"); + + let reloaded = flows_get(&config, &parent.id).await.unwrap().value; + assert_eq!(reloaded.last_status, None, "no run should have started"); +} + +#[tokio::test] +async fn flows_update_allows_metadata_only_edits_of_legacy_incompatible_graph() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let graph = structurally_valid_graph(nested_conditional_fan_in_graph()); + let flow = store::create_flow(&config, "legacy".to_string(), graph, false, false).unwrap(); + + let updated = flows_update( + &config, + &flow.id, + Some("renamed legacy".to_string()), + None, + Some(true), + None, + ) + .await + .expect("metadata-only update should preserve access to a legacy graph"); + + assert_eq!(updated.value.name, "renamed legacy"); + assert!(updated.value.require_approval); + assert_eq!(updated.value.graph, flow.graph); +} + +#[tokio::test] +async fn flows_create_rejects_an_incompatible_saved_child_before_persisting() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let child = store::create_flow( + &config, + "legacy unsafe child".to_string(), + structurally_valid_graph(nested_conditional_fan_in_graph()), + false, + false, + ) + .unwrap(); + + let error = flows_create( + &config, + "rejected parent".to_string(), + referenced_child_graph(&child.id), + false, + ) + .await + .expect_err("create must reject an unsafe saved child"); + + assert!( + error.contains(UNSUPPORTED_NESTED_CONDITIONAL_FAN_IN), + "{error}" + ); + assert!(error.contains(&child.id), "{error}"); + let flows = store::list_flows(&config).unwrap(); + assert_eq!(flows.len(), 1, "the rejected parent must not be persisted"); + assert_eq!(flows[0].id, child.id); +} + +#[tokio::test] +async fn flows_update_rejects_an_incompatible_saved_child_before_persisting() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let child = store::create_flow( + &config, + "legacy unsafe child".to_string(), + structurally_valid_graph(nested_conditional_fan_in_graph()), + false, + false, + ) + .unwrap(); + let original_graph = structurally_valid_graph(trigger_only_graph()); + let parent = store::create_flow( + &config, + "safe parent".to_string(), + original_graph.clone(), + false, + true, + ) + .unwrap(); + + let error = flows_update( + &config, + &parent.id, + None, + Some(referenced_child_graph(&child.id)), + None, + None, + ) + .await + .expect_err("update must reject an unsafe saved child"); + + assert!( + error.contains(UNSUPPORTED_NESTED_CONDITIONAL_FAN_IN), + "{error}" + ); + assert!(error.contains(&child.id), "{error}"); + let reloaded = flows_get(&config, &parent.id).await.unwrap().value; + assert_eq!( + reloaded.graph, original_graph, + "the rejected graph update must not be persisted" + ); +} + #[tokio::test] async fn flows_create_rejects_graph_without_trigger() { let tmp = TempDir::new().unwrap(); @@ -922,6 +1555,118 @@ async fn flows_resume_continues_a_paused_run_to_completion() { ); } +#[tokio::test] +async fn flows_resume_marks_an_incompatible_legacy_checkpoint_failed() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let created = flows_create(&config, "gated".to_string(), approval_gated_graph(), false) + .await + .unwrap(); + let run = flows_run( + &config, + &created.value.id, + json!({ "x": 1 }), + FlowRunTrigger::Rpc, + ) + .await + .unwrap(); + let thread_id = run.value["thread_id"].as_str().unwrap().to_string(); + let pending: Vec = + serde_json::from_value(run.value["pending_approvals"].clone()).unwrap(); + + // Simulate a graph persisted before the host compatibility gate existed. + // The store layer intentionally trusts its typed caller; authoring paths + // own validation. + store::update_flow_graph( + &config, + &created.value.id, + created.value.name.clone(), + structurally_valid_graph(nested_conditional_fan_in_graph()), + created.value.require_approval, + None, + None, + ) + .unwrap(); + + let error = flows_resume(&config, &created.value.id, &thread_id, pending, vec![]) + .await + .expect_err("an incompatible checkpoint cannot be resumed safely"); + assert!( + error.contains(UNSUPPORTED_NESTED_CONDITIONAL_FAN_IN), + "{error}" + ); + + let run_row = flows_get_run(&config, &thread_id).await.unwrap().value; + assert_eq!(run_row.status, "failed"); + assert!(run_row.pending_approvals.is_empty()); + assert!( + run_row + .error + .as_deref() + .is_some_and(|value| value.contains(UNSUPPORTED_NESTED_CONDITIONAL_FAN_IN)), + "the terminal run row should retain the rejection reason: {run_row:?}" + ); + let flow = flows_get(&config, &created.value.id).await.unwrap().value; + assert_eq!(flow.last_status.as_deref(), Some("failed")); +} + +#[tokio::test] +async fn flows_resume_marks_a_checkpoint_with_an_incompatible_saved_child_failed() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let created = flows_create(&config, "gated".to_string(), approval_gated_graph(), false) + .await + .unwrap(); + let run = flows_run( + &config, + &created.value.id, + json!({ "x": 1 }), + FlowRunTrigger::Rpc, + ) + .await + .unwrap(); + let thread_id = run.value["thread_id"].as_str().unwrap().to_string(); + let pending: Vec = + serde_json::from_value(run.value["pending_approvals"].clone()).unwrap(); + let child = store::create_flow( + &config, + "legacy unsafe child".to_string(), + structurally_valid_graph(nested_conditional_fan_in_graph()), + false, + false, + ) + .unwrap(); + store::update_flow_graph( + &config, + &created.value.id, + created.value.name.clone(), + structurally_valid_graph(referenced_child_graph(&child.id)), + created.value.require_approval, + None, + None, + ) + .unwrap(); + + let error = flows_resume(&config, &created.value.id, &thread_id, pending, vec![]) + .await + .expect_err("an incompatible saved child cannot be resumed safely"); + assert!( + error.contains(UNSUPPORTED_NESTED_CONDITIONAL_FAN_IN), + "{error}" + ); + assert!(error.contains(&child.id), "{error}"); + + let run_row = flows_get_run(&config, &thread_id).await.unwrap().value; + assert_eq!(run_row.status, "failed"); + assert!(run_row.pending_approvals.is_empty()); + assert!(run_row + .error + .as_deref() + .is_some_and(|value| value.contains(UNSUPPORTED_NESTED_CONDITIONAL_FAN_IN))); + let flow = flows_get(&config, &created.value.id).await.unwrap().value; + assert_eq!(flow.last_status.as_deref(), Some("failed")); +} + #[tokio::test] async fn flows_resume_missing_flow_errors() { let tmp = TempDir::new().unwrap(); @@ -4745,6 +5490,116 @@ async fn strict_gate_passes_a_valid_graph_and_rejects_a_structurally_invalid_one let err = strict_gate(&config, &bad).await.unwrap_err(); assert!(err.contains("structurally invalid"), "{err}"); assert!(err.contains("trigger"), "{err}"); + + // A structurally valid graph must still pass the shared engine gate. + let err = strict_gate(&config, &nested_conditional_fan_in_graph()) + .await + .unwrap_err(); + assert!(err.contains(UNSUPPORTED_NESTED_CONDITIONAL_FAN_IN), "{err}"); +} + +#[tokio::test] +async fn strict_gate_rejects_an_incompatible_saved_child_reference() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let child = store::create_flow( + &config, + "legacy unsafe child".to_string(), + structurally_valid_graph(nested_conditional_fan_in_graph()), + false, + false, + ) + .unwrap(); + + let error = strict_gate(&config, &referenced_child_graph(&child.id)) + .await + .expect_err("strict authoring must reject an incompatible saved child"); + assert!( + error.contains(UNSUPPORTED_NESTED_CONDITIONAL_FAN_IN), + "{error}" + ); + assert!(error.contains(&child.id), "{error}"); + assert!(error.contains("saved-child"), "{error}"); +} + +#[tokio::test] +async fn builder_proposal_rejects_an_incompatible_saved_child_reference() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let child = store::create_flow( + &config, + "legacy unsafe child".to_string(), + structurally_valid_graph(nested_conditional_fan_in_graph()), + false, + false, + ) + .unwrap(); + let parent = structurally_valid_graph(referenced_child_graph(&child.id)); + + let error = build_builder_proposal( + &config, + "propose_workflow", + "parent", + &parent, + false, + false, + None, + None, + None, + ) + .await + .expect_err("a proposal must reject an incompatible saved child"); + assert!( + error.contains(UNSUPPORTED_NESTED_CONDITIONAL_FAN_IN), + "{error}" + ); + assert!(error.contains(&child.id), "{error}"); + assert!(error.contains("saved-child"), "{error}"); +} + +#[test] +fn referenced_child_compatibility_stops_at_saved_workflow_cycles() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow_a = store::create_flow( + &config, + "cycle a".to_string(), + structurally_valid_graph(trigger_only_graph()), + false, + false, + ) + .unwrap(); + let flow_b = store::create_flow( + &config, + "cycle b".to_string(), + structurally_valid_graph(trigger_only_graph()), + false, + false, + ) + .unwrap(); + store::update_flow_graph( + &config, + &flow_a.id, + flow_a.name.clone(), + structurally_valid_graph(referenced_child_graph(&flow_b.id)), + false, + None, + None, + ) + .unwrap(); + store::update_flow_graph( + &config, + &flow_b.id, + flow_b.name.clone(), + structurally_valid_graph(referenced_child_graph(&flow_a.id)), + false, + None, + None, + ) + .unwrap(); + + let candidate = structurally_valid_graph(referenced_child_graph(&flow_a.id)); + assert!(referenced_workflow_compatibility_errors(&config, &candidate).is_empty()); } // ── core-managed drafts (F5) ───────────────────────────────────────────────── diff --git a/src/openhuman/flows/tools.rs b/src/openhuman/flows/tools.rs index 20eb1278d..0544da57a 100644 --- a/src/openhuman/flows/tools.rs +++ b/src/openhuman/flows/tools.rs @@ -23,9 +23,7 @@ use serde_json::{json, Value}; use tinyflows::model::{Node, NodeKind, WorkflowGraph}; use crate::openhuman::config::Config; -use crate::openhuman::flows::ops::{ - validate_and_migrate_graph, validate_binding_resolvability, validate_tool_contracts, -}; +use crate::openhuman::flows::ops::{build_builder_proposal, validate_and_migrate_graph}; use crate::openhuman::tools::traits::{PermissionLevel, Tool, ToolResult}; /// Max characters kept for a `config_hint` before truncation, so a long @@ -183,94 +181,34 @@ impl Tool for ProposeWorkflowTool { } }; - // Enforcing binding-resolvability gate (see - // `ops::validate_binding_resolvability`): reject outright — rather - // than merely warn — a `tool_call` binding that is guaranteed to - // resolve null (or the wrong value) at runtime, so the builder must - // fix the graph before it can even be proposed to the user. - let binding_errors = validate_binding_resolvability(&graph); - if !binding_errors.is_empty() { - tracing::debug!( - target: "flows", - %name, - error_count = binding_errors.len(), - "[flows] propose_workflow: binding-resolvability check rejected the graph" - ); - return Ok(ToolResult::error(format!( - "{}\n\nFix these bindings and call propose_workflow again.", - binding_errors.join("\n\n") - ))); - } - - // Tool-contract enforcement gate (systemic tool-contract fix, Part 2): - // reject a `tool_call` node whose slug isn't a REAL action in the - // live Composio catalog, or whose real required args aren't all - // wired — before the user ever reviews the proposal. - let contract_errors = validate_tool_contracts(&self.config, &graph).await; - if !contract_errors.is_empty() { - tracing::debug!( - target: "flows", - %name, - error_count = contract_errors.len(), - "[flows] propose_workflow: tool-contract check rejected the graph" - ); - return Ok(ToolResult::error(format!( - "{}\n\nFix these tool_call nodes and call propose_workflow again.", - contract_errors.join("\n\n") - ))); - } - - // Required-arg resolvability gate (issue B18): reject outright — - // rather than merely warn — a REQUIRED outbound arg (e.g. - // `GMAIL_SEND_EMAIL.subject`/`.body`) that LOOKS wired but resolves - // to `null` in a sandboxed test run, before the user ever reviews - // the proposal. See `ops::validate_required_arg_resolvability`. - let null_arg_errors = - crate::openhuman::flows::ops::validate_required_arg_resolvability(&graph).await; - if !null_arg_errors.is_empty() { - tracing::debug!( - target: "flows", - %name, - error_count = null_arg_errors.len(), - "[flows] propose_workflow: required-arg resolvability check rejected the graph" - ); - return Ok(ToolResult::error(format!( - "{}\n\nFix these bindings and call propose_workflow again.", - null_arg_errors.join("\n\n") - ))); - } - - let summary = build_summary(&graph); - // Author-time warnings: unfired trigger kinds + unwired REQUIRED - // Composio args (see `ops::graph_wiring_warnings`) — surfaced on the - // proposal so the builder fixes wiring before the user saves. - let mut warnings = crate::openhuman::flows::ops::graph_trigger_warnings(&graph); - warnings.extend( - crate::openhuman::flows::ops::graph_wiring_warnings(&self.config, &graph).await, - ); - let graph_value = serde_json::to_value(&graph)?; - - tracing::info!( - target: "flows", - %name, - node_count = graph.nodes.len(), + // Route every first proposal through the same canonical hard-gate and + // payload builder as revise/edit/save. In particular, this includes + // compatibility checks for literal workflow_id children, which cannot + // be detected by graph-only validation because they require the store. + match build_builder_proposal( + &self.config, + "propose_workflow", + &name, + &graph, require_approval, - warning_count = warnings.len(), - "[flows] propose_workflow: proposal ready for user review" - ); - - Ok(ToolResult::success(serde_json::to_string_pretty(&json!({ - "type": "workflow_proposal", - // A proposal is never a persisted flow — stamp it so the payload - // can't be misread as a save confirmation (WS2 audit). Matches - // `ops::build_builder_proposal`'s unconditional persisted:false. - "persisted": false, - "name": name, - "graph": graph_value, - "require_approval": require_approval, - "summary": summary, - "warnings": warnings, - }))?)) + false, + None, + None, + None, + ) + .await + { + Ok(payload) => Ok(ToolResult::success(serde_json::to_string_pretty(&payload)?)), + Err(error) => { + tracing::debug!( + target: "flows", + %name, + %error, + "[flows] propose_workflow: builder gate rejected the graph" + ); + Ok(ToolResult::error(error)) + } + } } } diff --git a/src/openhuman/flows/tools_tests.rs b/src/openhuman/flows/tools_tests.rs index 59912f2fc..d2133a079 100644 --- a/src/openhuman/flows/tools_tests.rs +++ b/src/openhuman/flows/tools_tests.rs @@ -323,6 +323,78 @@ async fn propose_workflow_rejects_unschemad_agent_binding() { ); } +#[tokio::test] +async fn propose_workflow_rejects_an_incompatible_saved_child_reference() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let tool = ProposeWorkflowTool::new(Arc::clone(&config)); + + // Simulate a legacy child saved before the current TinyFlows engine + // rejected nested conditional fan-in. The parent itself is structurally + // valid, so only the config-aware shared builder gate can catch this. + let legacy_child = json!({ + "nodes": [ + { "id": "start", "kind": "trigger", "name": "Trigger" }, + { "id": "outer", "kind": "condition", "name": "Outer", "config": { "field": "outer" } }, + { "id": "inner", "kind": "condition", "name": "Inner", "config": { "field": "inner" } }, + { "id": "outer_else", "kind": "output_parser", "name": "Outer else" }, + { "id": "inner_else", "kind": "output_parser", "name": "Inner else" }, + { "id": "a", "kind": "output_parser", "name": "A" }, + { "id": "c", "kind": "output_parser", "name": "C" }, + { "id": "m", "kind": "merge", "name": "Merge" } + ], + "edges": [ + { "from_node": "start", "to_node": "outer" }, + { "from_node": "start", "to_node": "c" }, + { "from_node": "outer", "from_port": "true", "to_node": "inner" }, + { "from_node": "outer", "from_port": "false", "to_node": "outer_else" }, + { "from_node": "inner", "from_port": "true", "to_node": "a" }, + { "from_node": "inner", "from_port": "false", "to_node": "inner_else" }, + { "from_node": "a", "to_node": "m" }, + { "from_node": "c", "to_node": "m" } + ] + }); + let child_graph = crate::openhuman::flows::ops::migrate_and_deserialize_graph(legacy_child) + .expect("legacy child should deserialize"); + tinyflows::validate::validate(&child_graph) + .expect("legacy child should remain structurally valid"); + let child = crate::openhuman::flows::store::create_flow( + &config, + "Legacy unsafe child".to_string(), + child_graph, + false, + false, + ) + .unwrap(); + let parent = json!({ + "nodes": [ + { "id": "start", "kind": "trigger", "name": "Trigger" }, + { + "id": "saved-child", + "kind": "sub_workflow", + "name": "Saved child", + "config": { "workflow_id": child.id } + } + ], + "edges": [{ "from_node": "start", "to_node": "saved-child" }] + }); + + let result = tool + .execute(json!({ "name": "Parent", "graph": parent })) + .await + .unwrap(); + + assert!(result.is_error, "must reject unsafe saved child"); + let output = result.output(); + assert!( + output.contains("unsupported_nested_conditional_fan_in"), + "{output}" + ); + assert!(output.contains(&child.id), "{output}"); + assert!(output.contains("saved-child"), "{output}"); + assert!(output.contains("call propose_workflow again"), "{output}"); +} + /// Docs-drift guard (F2): `propose_workflow`'s hand-written description and the /// typed node-kind contracts are two views of the SAME DSL, and they must not /// diverge. If a node kind is added/renamed or a required config field changes diff --git a/src/openhuman/tinyagents/middleware.rs b/src/openhuman/tinyagents/middleware.rs index 366d0223b..cfc9d07f3 100644 --- a/src/openhuman/tinyagents/middleware.rs +++ b/src/openhuman/tinyagents/middleware.rs @@ -3750,9 +3750,10 @@ mod tests { // ── ToolOutputMiddleware: COMPACTION_EXEMPT_TOOLS (workflow proposals) ─── /// A `workflow_proposal` payload with enough uniform-object rows to clear - /// tinyjuice's `MIN_ROWS` (3) and default ~512-byte tabulation floor — - /// i.e. exactly the shape that used to get its `"type"` marker stripped by - /// the `[json table: …]` rewrite before the middleware exemption existed. + /// tinyjuice's `MIN_ROWS` (3) and OpenHuman's default 2 KiB compaction + /// floor — i.e. exactly the shape that used to get its `"type"` marker + /// stripped by the `[json table: …]` rewrite before the middleware + /// exemption existed. fn large_workflow_proposal_json() -> String { let nodes: Vec = (0..20) .map(|i| { @@ -3809,6 +3810,13 @@ mod tests { // NOT in COMPACTION_EXEMPT_TOOLS loses the `"type"` marker. let mw = compaction_enabled_mw(); let payload = large_workflow_proposal_json(); + assert!( + payload.len() + >= crate::openhuman::config::Config::default() + .tokenjuice + .min_bytes_to_compress, + "baseline payload must clear OpenHuman's configured compaction floor" + ); let mut result = tool_result("some_other_tool", &payload); mw.after_tool(&mut ctx(), &(), &mut result).await.unwrap(); assert_ne!( diff --git a/src/openhuman/tinyflows/caps.rs b/src/openhuman/tinyflows/caps.rs index a18deb1e7..f3b9668d9 100644 --- a/src/openhuman/tinyflows/caps.rs +++ b/src/openhuman/tinyflows/caps.rs @@ -3291,7 +3291,7 @@ impl WorkflowResolver for OpenHumanWorkflowResolver { %workflow_id, "[flows] sub_workflow resolver: resolving workflow_id to a saved flow graph" ); - match flows::ops::load_flow_graph(&self.config, workflow_id) { + match flows::ops::load_engine_compatible_flow_graph(&self.config, workflow_id) { Ok(Some(graph)) => { tracing::debug!( target: "flows", @@ -4714,6 +4714,63 @@ mod tests { } } + #[tokio::test] + async fn resolver_rejects_an_engine_incompatible_saved_graph() { + let tmp = tempfile::TempDir::new().unwrap(); + let config = Arc::new(resolver_test_config(&tmp)); + let flow = flows::ops::flows_create( + &config, + "legacy child".to_string(), + serde_json::to_value(trigger_only_graph()).unwrap(), + false, + ) + .await + .unwrap() + .value; + let unsafe_graph = json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Trigger" }, + { "id": "outer", "kind": "condition", "name": "Outer", "config": { "field": "outer" } }, + { "id": "inner", "kind": "condition", "name": "Inner", "config": { "field": "inner" } }, + { "id": "outer_else", "kind": "output_parser", "name": "Outer else" }, + { "id": "inner_else", "kind": "output_parser", "name": "Inner else" }, + { "id": "a", "kind": "output_parser", "name": "A" }, + { "id": "c", "kind": "output_parser", "name": "C" }, + { "id": "m", "kind": "merge", "name": "Merge" } + ], + "edges": [ + { "from_node": "t", "from_port": "main", "to_node": "outer" }, + { "from_node": "t", "from_port": "main", "to_node": "c" }, + { "from_node": "outer", "from_port": "true", "to_node": "inner" }, + { "from_node": "outer", "from_port": "false", "to_node": "outer_else" }, + { "from_node": "inner", "from_port": "true", "to_node": "a" }, + { "from_node": "inner", "from_port": "false", "to_node": "inner_else" }, + { "from_node": "a", "from_port": "main", "to_node": "m" }, + { "from_node": "c", "from_port": "main", "to_node": "m" } + ] + }); + let db = config.workspace_dir.join("flows").join("flows.db"); + rusqlite::Connection::open(db) + .unwrap() + .execute( + "UPDATE flow_definitions SET graph_json = ?1 WHERE id = ?2", + rusqlite::params![unsafe_graph.to_string(), flow.id], + ) + .unwrap(); + + let error = OpenHumanWorkflowResolver { config } + .resolve(&flow.id) + .await + .expect_err("resolver must reject an incompatible legacy child"); + match error { + EngineError::Capability(message) => assert!( + message.contains("unsupported_nested_conditional_fan_in"), + "{message}" + ), + other => panic!("expected a capability error, got: {other:?}"), + } + } + // ── response_fields_from_schema ───────────────────────────────────────── // Direct unit tests for the pure schema-extraction step inside // `composio_response_fields`'s live-fetch loop — cheaper and more diff --git a/tests/json_rpc_e2e.rs b/tests/json_rpc_e2e.rs index ade98ad76..e450b9577 100644 --- a/tests/json_rpc_e2e.rs +++ b/tests/json_rpc_e2e.rs @@ -15002,7 +15002,13 @@ async fn json_rpc_agent_team_live_member_run_roundtrip_inner() { "Task B must reach done" ); assert!( - poll_team_members_status(&rpc_base, &team_id, "idle").await, + poll_team_members_status( + &rpc_base, + &team_id, + &[alice_id.as_str(), bob_id.as_str()], + "idle", + ) + .await, "team members must return to idle" ); @@ -15105,12 +15111,15 @@ async fn poll_team_task_status(rpc_base: &str, team_id: &str, task_id: &str, wan false } -/// Poll `agent_team_get` until every team member reaches `want` status. -/// -/// Task completion and member cleanup are separate ledger writes, so observing -/// a `done` task does not guarantee the member's idle transition is visible in -/// the same snapshot. -async fn poll_team_members_status(rpc_base: &str, team_id: &str, want: &str) -> bool { +/// Poll `agent_team_get` until every team member reaches `want` (or time out). +/// Task completion and the worker's transition back to idle are separate +/// durable operations, so observing a done task does not yet imply idle. +async fn poll_team_members_status( + rpc_base: &str, + team_id: &str, + expected_member_ids: &[&str], + want: &str, +) -> bool { for attempt in 0..160 { tokio::time::sleep(Duration::from_millis(250)).await; let got = post_json_rpc( @@ -15120,13 +15129,17 @@ async fn poll_team_members_status(rpc_base: &str, team_id: &str, want: &str) -> json!({ "teamId": team_id }), ) .await; - let view = assert_no_jsonrpc_error(&got, "agent_team_get member poll"); - let members = view + let members = assert_no_jsonrpc_error(&got, "agent_team_get member poll") .get("team") - .and_then(|team| team.get("members")) + .and_then(|tv| tv.get("members")) .and_then(Value::as_array); if members.is_some_and(|members| { - !members.is_empty() + members.len() == expected_member_ids.len() + && expected_member_ids.iter().all(|expected_id| { + members.iter().any(|member| { + member.get("id").and_then(Value::as_str) == Some(*expected_id) + }) + }) && members .iter() .all(|member| member.get("memberStatus").and_then(Value::as_str) == Some(want)) diff --git a/tests/raw_coverage/composio_credentials_state_raw_coverage_e2e.rs b/tests/raw_coverage/composio_credentials_state_raw_coverage_e2e.rs index 9af8a2d38..5cade5a55 100644 --- a/tests/raw_coverage/composio_credentials_state_raw_coverage_e2e.rs +++ b/tests/raw_coverage/composio_credentials_state_raw_coverage_e2e.rs @@ -307,17 +307,17 @@ async fn round15_composio_agent_tools_backend_cache_and_trigger_history_edges() ); assert_eq!(action_tool.name(), "GMAIL_FETCH_EMAILS"); assert_eq!(action_tool.category().to_string(), "skill"); - let action_contract = action_tool + let contract_result = action_tool .execute(json!({ "query": "from:me" })) .await - .expect("per-action tool contract gate"); - assert!(action_contract.is_error); - assert!(action_contract.text().contains("Required arguments: query")); + .expect("per-action contract gate"); + assert!(contract_result.is_error); + assert!(contract_result.text().contains("Input JSON schema")); let action_result = action_tool .execute(json!({ "query": "from:me" })) .await - .expect("per-action tool execute after contract gate"); + .expect("per-action tool retry"); assert_eq!(action_result.text(), "Fetched 1 inbox message"); let reserved = composio_authorize(&config, "gmail", Some(json!({ "toolkit": "github" }))) diff --git a/vendor/tinyagents b/vendor/tinyagents index 19dc2c438..2583fccc2 160000 --- a/vendor/tinyagents +++ b/vendor/tinyagents @@ -1 +1 @@ -Subproject commit 19dc2c438e1f8b2715a45ba7030bc455e611dcb2 +Subproject commit 2583fccc213a00f2a3d94744ff1e0d1541368f97 diff --git a/vendor/tinychannels b/vendor/tinychannels index d82953832..14057e0d8 160000 --- a/vendor/tinychannels +++ b/vendor/tinychannels @@ -1 +1 @@ -Subproject commit d82953832c6d76259ef4a2cfc87a3f86bd31d7f1 +Subproject commit 14057e0d877d27450a98dd8c145b4f464d7d1308 diff --git a/vendor/tinyflows b/vendor/tinyflows index e5327de9f..56998ec3e 160000 --- a/vendor/tinyflows +++ b/vendor/tinyflows @@ -1 +1 @@ -Subproject commit e5327de9f602c1fbbf72d45150729f45b91fa3a8 +Subproject commit 56998ec3eae655d7619b634d13aad0efaf0ad76a diff --git a/vendor/tinyjuice b/vendor/tinyjuice index 41b5ca87c..e6848ed87 160000 --- a/vendor/tinyjuice +++ b/vendor/tinyjuice @@ -1 +1 @@ -Subproject commit 41b5ca87c4af786f908beb66eec1aae3df25fbdd +Subproject commit e6848ed87d5d661073e9eab87b6b385c3373ee38