chore(deps): update vendored runtime crates (#5081)

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",
 ]
 

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

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",
 ]
 

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<String> = 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.

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]

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<WorkflowGraph, String> {
     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<crate::openhuman::flows::FlowValidationError> {
+    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<crate::openhuman::flows::FlowValidationError>,
+) {
+    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::<WorkflowGraph>(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<crate::openhuman::flows::FlowValidationError> {
+    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::<HashSet<_>>()
+        .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<String> {
+    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<String> {
+    let mut pending = vec![(graph.clone(), 0_u64, Vec::<String>::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::<String, u64>::new();
+
+    while let Some((current, depth, path)) = pending.pop() {
+        if depth >= tinyflows::engine::MAX_SUB_WORKFLOW_DEPTH {
+            continue;
+        }
+
+        for node in &current.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::<WorkflowGraph>(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<String> {
+    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 `<tool>` 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<crate::openhuman::flows::
         );
     }
 
+    let error_details = engine_compatibility_errors(&graph);
+    if !error_details.is_empty() {
+        let errors = error_details
+            .iter()
+            .map(|error| error.message.clone())
+            .collect();
+        tracing::debug!(
+            target: "flows",
+            error_count = error_details.len(),
+            "[flows] flows_validate: graph uses an unsupported engine topology"
+        );
+        return RpcOutcome::single_log(
+            FlowValidation {
+                valid: false,
+                errors,
+                error_details,
+                warnings: Vec::new(),
+            },
+            "flow validation failed",
+        );
+    }
+
     let warnings = graph_trigger_warnings(&graph);
     for warning in &warnings {
         tracing::warn!(target: "flows", warning = %warning, "[flows] flows_validate: non-fatal validation warning");
@@ -2196,6 +2625,7 @@ pub async fn flows_create(
     require_approval: bool,
 ) -> Result<RpcOutcome<Flow>, 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<Option<WorkflowGraph
     Ok(graph)
 }
 
+/// Resolver-only saved-graph lookup. Authoring tools use [`load_flow_graph`]
+/// so a legacy draft can still be opened and repaired; execution resolves only
+/// graphs the current engine can run safely.
+pub(crate) fn load_engine_compatible_flow_graph(
+    config: &Config,
+    id: &str,
+) -> Result<Option<WorkflowGraph>, 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<RpcOutcome<Vec<Flow>>, 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 =

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<String> =
+        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<String> =
+        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) ─────────────────────────────────────────────────

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))
+            }
+        }
     }
 }
 

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

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<serde_json::Value> = (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!(

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

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))

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" })))