feat(flows): improve workflow builder UX and diagnostics (#4574)

src/openhuman/agent_registry/agents/loader.rs

@@ -989,10 +989,15 @@ mod tests {
     #[test]
     fn workflow_builder_is_registered_worker_with_narrow_propose_or_read_scope() {
         // Phase 5a/5b: the workflow-builder must be a Worker-tier leaf whose
-        // tool scope is EXACTLY the propose-or-read belt — no persistence
-        // (flows_create/update/set_enabled), no shell, no file writes, no
-        // channel sends. This pins the "propose, never persist" invariant in
-        // the agent definition itself, not just the tool implementations.
+        // tool scope is EXACTLY the propose-or-read + Composio discovery/connect
+        // + confirmed test-run belt — no persistence (flows_create/update/
+        // set_enabled), no shell, no file writes, no channel sends, and no
+        // composio_execute. It can list toolkits/connections, raise the inline
+        // connect card, and `run_workflow` a flow the user already SAVED to test
+        // it (a real run the prompt gates behind user confirmation), but it can
+        // never persist/enable a flow or perform a raw integration action. This
+        // pins the invariant in the agent definition itself, not just the tool
+        // implementations.
         let def = find("workflow_builder");
         assert_eq!(def.agent_tier, AgentTier::Worker);
         assert_eq!(def.delegate_name.as_deref(), Some("build_workflow"));
@@ -1013,6 +1018,10 @@ mod tests {
                     "list_flow_connections",
                     "search_tool_catalog",
                     "dry_run_workflow",
+                    "run_workflow",
+                    "composio_list_toolkits",
+                    "composio_list_connections",
+                    "composio_connect",
                 ];
                 for required in expected {
                     assert!(

src/openhuman/agent_registry/agents/workflow_builder/agent.toml

@@ -23,9 +23,15 @@ hint = "burst"
 
 [tools]
 # DELIBERATELY NARROW: propose/revise (validate-only) + read (flows, runs,
-# connections, tool catalog) + sandbox dry-run. NO shell, NO file writes, NO
-# channel sends, NO flows_create/update/set_enabled — the agent can never
-# persist or enable a flow.
+# connections, tool catalog) + sandbox dry-run + Composio discovery/connect +
+# a confirmed real test-run of a SAVED flow. NO shell, NO file writes, NO
+# channel sends, NO composio_execute, and NO flows_create/update/set_enabled —
+# the agent can never persist or enable a flow, or perform a real integration
+# action directly. Composio access is limited to LISTING toolkits/connections
+# and raising the inline CONNECT card (an approval-gated OAuth hand-off).
+# `run_workflow` executes a flow the user has ALREADY saved to test it — a real
+# run, but the prompt requires asking the user to confirm first, and the flow's
+# own approval gate still pauses outbound-action nodes.
 named = [
     "propose_workflow",
     "revise_workflow",
@@ -35,4 +41,8 @@ named = [
     "list_flow_connections",
     "search_tool_catalog",
     "dry_run_workflow",
+    "run_workflow",
+    "composio_list_toolkits",
+    "composio_list_connections",
+    "composio_connect",
 ]

src/openhuman/agent_registry/agents/workflow_builder/prompt.md

@@ -8,8 +8,8 @@ user to review and save.
 
 ## The one invariant you must never break: propose, never persist
 
-You **cannot and must not** create, update, enable, disable, or run a real saved
-flow. You have no tool that does — by design. Your only outputs are:
+You **cannot and must not** create, update, enable, or disable a saved flow. You
+have no tool that does — by design. Your authoring outputs are:
 
 - **`propose_workflow`** / **`revise_workflow`** — these *validate* a candidate
   graph and hand back a proposal summary. They **never** save anything.
@@ -21,6 +21,24 @@ Only the user's own "Save & enable" click in the review card persists a flow
 (via `flows_create`, which re-validates server-side). If a user says "just turn
 it on for me", explain that you can only propose it — they confirm the save.
 
+## Testing a saved flow: `run_workflow` (ask first!)
+
+Once the user has **saved** a flow, you can `run_workflow { flow_id }` to test it
+end-to-end. Unlike `dry_run_workflow`, this is a **real run** — real effects can
+fire (the flow's own approval gate still pauses outbound-action nodes, but treat
+it as real). Rules:
+
+1. **Only a saved flow.** `run_workflow` needs a `flow_id`; if the workflow isn't
+   saved yet, tell the user to Save it first (you can't run a draft — use
+   `dry_run_workflow` for a draft wiring check).
+2. **ALWAYS ask for confirmation and wait for an explicit "yes"** before calling
+   `run_workflow`. Say what it will do ("This will run the flow for real and may
+   send/act on live data — run it now?") and only proceed once they agree. Never
+   run a workflow unprompted or as a surprise side effect of another request.
+3. After a run, read the result (status + any nodes paused for approval) and
+   report what happened; if it failed, `get_flow_run` for the steps and propose a
+   fix.
+
 ## Your authoring loop
 
 1. **Understand the trigger and the steps.** What starts the flow? What should
@@ -34,6 +52,37 @@ it on for me", explain that you can only propose it — they confirm the save.
      `http_request` node or tell the user the integration isn't available.
    - `list_flows` / `get_flow` → reuse or clone an existing flow instead of
      duplicating one.
+   - **Missing the integration the workflow needs?** See "Connecting
+     integrations" below — you can help the user link it before you build,
+     rather than dead-ending.
+
+## Connecting integrations
+
+A workflow often needs an app the user hasn't linked yet (a `tool_call` on
+Gmail, Slack, Notion…). You can close that gap yourself instead of telling the
+user to go do it elsewhere:
+
+- **`composio_list_toolkits`** — the catalog of connectable apps (slugs like
+  `gmail`, `slack`, `googlesheets`). Use it to find the right toolkit for what
+  the user described.
+- **`composio_list_connections`** — which toolkits the user has ALREADY
+  connected (mirrors `list_flow_connections`' Composio side). Check here first —
+  never ask someone to connect an app they've already linked.
+- **`composio_connect`** — raises an inline **Connect** card for a toolkit and
+  waits for the user to approve the OAuth hand-off. Call it when the workflow
+  needs an app that isn't in `composio_list_connections` yet. After it returns
+  connected, re-run `list_flow_connections` to pick up the fresh
+  `connection_ref` and put it on the node.
+
+Still bounded: you can **discover and connect** apps, but you have **no** tool to
+*execute* a Composio action (`composio_execute` is deliberately out of scope) and
+**no** tool to persist a flow. Connecting is a setup step in service of a
+proposal — the user still saves the workflow themselves.
+
+Typical setup arc: user asks for a Slack step → `composio_list_connections`
+shows Slack isn't linked → `composio_connect { toolkit: "slack" }` → once
+connected, `list_flow_connections` → build the `tool_call` node with the real
+`connection_ref` + a `search_tool_catalog` slug → dry-run → propose.
 3. **Build the graph** (see the model below).
 4. **Self-check with `dry_run_workflow`** on the draft — catch missing edges,
    wrong ports, unreachable nodes. Fix and re-run.
@@ -57,9 +106,22 @@ A `WorkflowGraph` is `{ name?, nodes: [...], edges: [...] }`.
 
 1. **`trigger`** — the entry point (`config.trigger_kind`, see triggers below).
 2. **`agent`** — an LLM step. `config.prompt` (may use `=` expressions).
-3. **`tool_call`** — a Composio action. `config.slug` **REQUIRED** (from
-   `search_tool_catalog`) + `config.args`; `config.connection_ref` for the
-   account.
+   **If the agent's output feeds a `tool_call`, it MUST declare an output
+   schema** — set `config.output_parser.schema` (a JSON Schema object) — so
+   its emitted item is a structured object whose fields downstream nodes can
+   address (`=nodes.<agent_id>.item.<field>`). Without a schema the agent
+   emits `{text: "..."}` and `=item.to`-style bindings resolve to null.
+3. **`tool_call`** — an action. Two flavours by `config.slug`:
+   - **Composio app action** — `config.slug` = a real action slug (from
+     `search_tool_catalog`, e.g. `GMAIL_SEND_EMAIL`) + `config.connection_ref`
+     for the account. **Wire every REQUIRED arg in `config.args` from a named
+     upstream node** — e.g. an email send needs `to`/`recipient_email`, usually
+     `"to": "=nodes.<upstream_id>.item.email"`. A required arg left unwired (or
+     whose expression misses) now fails BEFORE the provider call — both in
+     `dry_run_workflow` and in real runs — with an error naming the field.
+   - **Native OpenHuman tool** — `config.slug` = `oh:<tool_name>` (e.g.
+     `oh:web_search`) to call one of the assistant's own built-in tools (search,
+     media generation, files, …). No `connection_ref`. Args go in `config.args`.
 4. **`http_request`** — `config.method` + `config.url`, optional `headers` /
    `body`; `config.connection_ref` = an `http_cred:<name>` for auth.
 5. **`code`** — `config.language` (`"javascript"` | `"python"`) + `config.source`.
@@ -85,8 +147,39 @@ the run scope (`.`):
 - A string **without** a leading `=` is a literal. To emit a literal `=`, don't
   start the string with it.
 
+The scope exposes:
+
+- `item` / `items` — the **direct predecessor(s)'** output (first item / all
+  items, in edge order).
+- `run` — run metadata and the trigger payload.
+- `nodes` — **every completed node's output, keyed by node id**:
+  `nodes.<id>.item` (first item) and `nodes.<id>.items` (all items). Use this
+  to reference ANY upstream node — not just the immediate predecessor — and to
+  disambiguate a fan-in node's inputs. Dotted form: `"=nodes.fetch.item.email"`;
+  jq form: `"=.nodes[\"fetch\"].items[0].email"`. Ids (not names) are the key.
+
 Use expressions to thread data between steps (a `transform`'s `set`, an
-`agent`'s `prompt`, a `tool_call`'s `args`).
+`agent`'s `prompt`, a `tool_call`'s `args`). Prefer `=nodes.<id>.…` for
+`tool_call` args so the binding survives graph re-wiring.
+
+**Worked example — agent → Gmail send.** The agent must declare a schema, and
+the tool_call wires each required arg from the agent BY ID:
+
+```json
+{ "id": "extract", "kind": "agent", "config": {
+    "prompt": "=\"Extract the recipient and a reply from: \" + .item.text",
+    "output_parser": { "schema": { "type": "object",
+      "required": ["email", "subject", "body"],
+      "properties": { "email": {"type": "string"}, "subject": {"type": "string"}, "body": {"type": "string"} } } } } }
+{ "id": "send", "kind": "tool_call", "config": {
+    "slug": "GMAIL_SEND_EMAIL", "connection_ref": "composio:gmail:<conn_id>",
+    "args": { "to": "=nodes.extract.item.email",
+              "subject": "=nodes.extract.item.subject",
+              "body": "=nodes.extract.item.body" } } }
+```
+
+Without the schema, `=nodes.extract.item.email` would be null (the agent's
+item would be `{text: ...}`) and the send would fail preflight naming `to`.
 
 ### Trigger kinds — which ones actually fire
 

src/openhuman/flows/builder_tools.rs

@@ -21,6 +21,14 @@
 //! executes against `tinyflows`' deterministic **mock** capabilities so no real
 //! LLM / tool / HTTP / code side effect can fire. Only the user's own
 //! "Save & enable" click (→ `openhuman.flows_create`) writes anything.
+//!
+//! The agent's full tool scope (see `agent_registry/agents/workflow_builder/
+//! agent.toml`) also grants the Composio **discovery/connect** tools —
+//! `composio_list_toolkits`, `composio_list_connections`, `composio_connect`
+//! (defined in `composio/tools.rs`) — so the builder can link an app the
+//! workflow needs before proposing. Those stay within the invariant: connect
+//! is an approval-gated OAuth hand-off, and `composio_execute` (running a real
+//! action) remains deliberately OUT of scope.
 
 use std::sync::Arc;
 
@@ -156,7 +164,10 @@ impl Tool for ReviseWorkflowTool {
         };
 
         let summary = super::tools::build_summary(&graph);
-        let warnings = ops::graph_trigger_warnings(&graph);
+        let mut warnings = ops::graph_trigger_warnings(&graph);
+        // Author-time wiring check: unwired REQUIRED Composio args come back
+        // as warnings naming the field, before the user ever saves.
+        warnings.extend(ops::graph_wiring_warnings(&self.config, &graph).await);
         let graph_value = serde_json::to_value(&graph)?;
 
         tracing::info!(
@@ -613,13 +624,20 @@ impl Tool for SearchToolCatalogTool {
 /// Autonomy-tier gated (issue: Phase 2 node gating): read-only tier refuses,
 /// mirroring the `SecurityPolicy` contract that a read-only session cannot
 /// exercise executable capability even in simulation.
+///
+/// **Wiring preflight:** the mock tool invoker is wrapped in the host's
+/// [`PreflightToolInvoker`](crate::openhuman::tinyflows::caps::PreflightToolInvoker),
+/// so a Composio `tool_call` whose required arg is missing or `=`-resolved to
+/// null fails the dry run with the same actionable, field-naming error a real
+/// run would produce — the echo mocks alone would happily accept a null `to`.
 pub struct DryRunWorkflowTool {
     security: Arc<SecurityPolicy>,
+    config: Arc<Config>,
 }
 
 impl DryRunWorkflowTool {
-    pub fn new(security: Arc<SecurityPolicy>) -> Self {
-        Self { security }
+    pub fn new(security: Arc<SecurityPolicy>, config: Arc<Config>) -> Self {
+        Self { security, config }
     }
 }
 
@@ -719,7 +737,13 @@ impl Tool for DryRunWorkflowTool {
             }
         };
 
-        let caps = tinyflows::caps::mock::mock_capabilities();
+        let mut caps = tinyflows::caps::mock::mock_capabilities();
+        // Wiring preflight over the echo mocks (see the struct doc): required
+        // Composio args must be present and non-null even in the sandbox.
+        caps.tools = std::sync::Arc::new(crate::openhuman::tinyflows::caps::PreflightToolInvoker {
+            config: self.config.clone(),
+            inner: caps.tools.clone(),
+        });
         let run = tinyflows::engine::run(&compiled, input, &caps);
         let outcome = match tokio::time::timeout(
             std::time::Duration::from_secs(DRY_RUN_TIMEOUT_SECS),

src/openhuman/flows/builder_tools_tests.rs

@@ -201,7 +201,10 @@ async fn search_tool_catalog_missing_query_is_error() {
 
 #[test]
 fn dry_run_is_execute_permission() {
-    let tool = DryRunWorkflowTool::new(policy(AutonomyLevel::Supervised));
+    let tool = DryRunWorkflowTool::new(
+        policy(AutonomyLevel::Supervised),
+        test_config(&TempDir::new().unwrap()),
+    );
     assert_eq!(tool.name(), "dry_run_workflow");
     assert_eq!(tool.permission_level(), PermissionLevel::Execute);
     // Mock-backed: no real outbound effect.
@@ -210,7 +213,10 @@ fn dry_run_is_execute_permission() {
 
 #[tokio::test]
 async fn dry_run_refused_under_readonly_tier() {
-    let tool = DryRunWorkflowTool::new(policy(AutonomyLevel::ReadOnly));
+    let tool = DryRunWorkflowTool::new(
+        policy(AutonomyLevel::ReadOnly),
+        test_config(&TempDir::new().unwrap()),
+    );
     let result = tool
         .execute(json!({ "graph": valid_graph() }))
         .await
@@ -221,7 +227,10 @@ async fn dry_run_refused_under_readonly_tier() {
 
 #[tokio::test]
 async fn dry_run_supervised_runs_against_mock_and_labels_sandbox() {
-    let tool = DryRunWorkflowTool::new(policy(AutonomyLevel::Supervised));
+    let tool = DryRunWorkflowTool::new(
+        policy(AutonomyLevel::Supervised),
+        test_config(&TempDir::new().unwrap()),
+    );
     let result = tool
         .execute(json!({ "graph": valid_graph(), "input": { "x": 1 } }))
         .await
@@ -239,10 +248,119 @@ async fn dry_run_supervised_runs_against_mock_and_labels_sandbox() {
 
 #[tokio::test]
 async fn dry_run_invalid_graph_is_error() {
-    let tool = DryRunWorkflowTool::new(policy(AutonomyLevel::Full));
+    let tool = DryRunWorkflowTool::new(
+        policy(AutonomyLevel::Full),
+        test_config(&TempDir::new().unwrap()),
+    );
     let result = tool
         .execute(json!({ "graph": { "nodes": [], "edges": [] } }))
         .await
         .unwrap();
     assert!(result.is_error);
 }
+
+#[tokio::test]
+async fn dry_run_catches_unwired_required_composio_arg() {
+    // Seed the preflight schema cache so no live Composio backend is needed.
+    // NOTE: the cache is process-global and other tests seed the `gmail`
+    // toolkit too — keep every seeding of GMAIL_SEND_EMAIL identical
+    // (`to` + `body`) so test order can't change the outcome.
+    let mut entries = std::collections::HashMap::new();
+    entries.insert(
+        "GMAIL_SEND_EMAIL".to_string(),
+        vec!["to".to_string(), "body".to_string()],
+    );
+    crate::openhuman::tinyflows::caps::seed_required_args_cache("gmail", entries);
+
+    let tmp = TempDir::new().unwrap();
+    let tool = DryRunWorkflowTool::new(policy(AutonomyLevel::Supervised), test_config(&tmp));
+
+    let graph_with = |args: Value| {
+        json!({
+            "nodes": [
+                { "id": "t", "kind": "trigger", "name": "Manual" },
+                { "id": "send", "kind": "tool_call", "name": "Send email",
+                  "config": { "slug": "GMAIL_SEND_EMAIL", "args": args } }
+            ],
+            "edges": [ { "from_node": "t", "to_node": "send" } ]
+        })
+    };
+
+    // `to` is a `=`-expression that misses (trigger input has no `email`):
+    // the dry run must fail BEFORE the (mock) tool call, naming the field.
+    let result = tool
+        .execute(json!({
+            "graph": graph_with(json!({ "to": "=item.email", "body": "hello" })),
+            "input": {}
+        }))
+        .await
+        .unwrap();
+    let out = result.output();
+    assert!(
+        out.contains("`to`") && out.contains("required"),
+        "dry run must name the unwired required arg: {out}"
+    );
+
+    // The same flow with `to` wired from the trigger passes the preflight.
+    let result = tool
+        .execute(json!({
+            "graph": graph_with(json!({ "to": "=item.email", "body": "hello" })),
+            "input": { "email": "a@b.com" }
+        }))
+        .await
+        .unwrap();
+    assert!(!result.is_error, "{}", result.output());
+    let parsed: Value = serde_json::from_str(&result.output()).unwrap();
+    assert_eq!(parsed["sandbox"], true);
+    assert_eq!(
+        parsed["ok"], true,
+        "wired flow must dry-run green: {parsed}"
+    );
+}
+
+#[tokio::test]
+async fn revise_workflow_warns_on_unwired_required_composio_arg() {
+    let mut entries = std::collections::HashMap::new();
+    entries.insert(
+        "GMAIL_SEND_EMAIL".to_string(),
+        vec!["to".to_string(), "body".to_string()],
+    );
+    crate::openhuman::tinyflows::caps::seed_required_args_cache("gmail", entries);
+
+    let tmp = TempDir::new().unwrap();
+    let tool = ReviseWorkflowTool::new(test_config(&tmp));
+    let result = tool
+        .execute(json!({
+            "name": "Send mail",
+            "graph": {
+                "nodes": [
+                    { "id": "t", "kind": "trigger", "name": "Manual" },
+                    { "id": "send", "kind": "tool_call", "name": "Send",
+                      // `body` wired via expression (counts as wired); `to` absent.
+                      "config": { "slug": "GMAIL_SEND_EMAIL",
+                                  "args": { "body": "=item.text" } } }
+                ],
+                "edges": [ { "from_node": "t", "to_node": "send" } ]
+            }
+        }))
+        .await
+        .unwrap();
+
+    assert!(!result.is_error, "{}", result.output());
+    let parsed: Value = serde_json::from_str(&result.output()).unwrap();
+    let warnings = parsed["warnings"].as_array().unwrap();
+    assert!(
+        warnings.iter().any(|w| {
+            let w = w.as_str().unwrap_or_default();
+            w.contains("`to`") && w.contains("send")
+        }),
+        "expected a warning naming node `send` and arg `to`: {warnings:?}"
+    );
+    // `body` is wired (expression) — no warning for it.
+    assert!(
+        !warnings
+            .iter()
+            .any(|w| w.as_str().unwrap_or_default().contains("`body`")),
+        "wired arg must not warn: {warnings:?}"
+    );
+}

src/openhuman/flows/ops.rs

@@ -162,6 +162,56 @@ pub(crate) fn graph_trigger_warnings(graph: &WorkflowGraph) -> Vec<String> {
     )]
 }
 
+/// Author-time wiring warnings for Composio `tool_call` nodes: flags every
+/// **required** arg (per the action's schema, best-effort cached lookup) that
+/// is absent or a literal `null` in `config.args` — the exact mis-wiring that
+/// would later fail the run's required-arg preflight.
+///
+/// Static by design: an arg carrying an `=`-expression counts as wired (only
+/// the runtime preflight can tell whether it resolves), a `=`-derived slug is
+/// skipped (can't know the action), and native `oh:` tools are skipped (no
+/// Composio schema). Best-effort like the runtime preflight — no schema, no
+/// warning, never a block.
+pub(crate) async fn graph_wiring_warnings(config: &Config, graph: &WorkflowGraph) -> Vec<String> {
+    use crate::openhuman::tinyflows::caps::{composio_required_args, missing_required_args};
+
+    let mut warnings = Vec::new();
+    for node in &graph.nodes {
+        if node.kind != tinyflows::model::NodeKind::ToolCall {
+            continue;
+        }
+        let Some(slug) = node.config.get("slug").and_then(Value::as_str) else {
+            continue;
+        };
+        // `=`-derived slugs are resolved at runtime; native tools have no
+        // Composio schema to check against.
+        if slug.starts_with('=') || slug.starts_with("oh:") {
+            continue;
+        }
+        let Some(required) = composio_required_args(config, slug).await else {
+            tracing::debug!(target: "flows", node = %node.id, %slug, "[flows] wiring check: no schema — skipping node");
+            continue;
+        };
+        let args = node.config.get("args").cloned().unwrap_or(Value::Null);
+        for missing in missing_required_args(&required, &args) {
+            tracing::warn!(
+                target: "flows",
+                node = %node.id,
+                %slug,
+                arg = %missing,
+                "[flows] wiring check: required arg not wired"
+            );
+            warnings.push(format!(
+                "Node '{}': required arg `{missing}` of `{slug}` is not wired — set \
+                 args.{missing}, e.g. \"=nodes.<upstream_id>.item.<field>\" (an agent feeding \
+                 this value needs an output schema so its fields are addressable).",
+                node.id
+            ));
+        }
+    }
+    warnings
+}
+
 /// Validates a candidate graph without persisting it — the same
 /// migrate/validate path `flows_create` and `ProposeWorkflowTool` use — and
 /// reports structural errors alongside non-fatal trigger warnings
@@ -1506,6 +1556,7 @@ fn reconstruct_steps(output: &Value) -> Vec<FlowRunStep> {
             // Reconstructed post-hoc: no live status/timing (see FlowRunStep).
             status: None,
             duration_ms: None,
+            diagnostics: Vec::new(),
         })
         .collect()
 }

src/openhuman/flows/ops_tests.rs

@@ -1044,12 +1044,14 @@ async fn observer_persists_each_step_incrementally() {
         status: StepStatus::Success,
         output: json!([{ "json": { "ok": true } }]),
         duration_ms: 7,
+        diagnostics: Vec::new(),
     });
     observer.on_step_finish(&ExecutionStep {
         node_id: "b".to_string(),
         status: StepStatus::Error,
         output: Value::Null,
         duration_ms: 3,
+        diagnostics: Vec::new(),
     });
 
     // The store now holds both live steps with real status + timing — proof of
@@ -1069,6 +1071,7 @@ async fn observer_persists_each_step_incrementally() {
         status: StepStatus::Success,
         output: json!([{ "json": { "ok": true } }]),
         duration_ms: 42,
+        diagnostics: Vec::new(),
     });
     let row = store::get_flow_run(&config, &run_id).unwrap().unwrap();
     assert_eq!(row.steps.len(), 2, "re-firing a node must not duplicate it");

src/openhuman/flows/tools.rs

@@ -177,6 +177,13 @@ impl Tool for ProposeWorkflowTool {
         };
 
         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!(
@@ -184,6 +191,7 @@ impl Tool for ProposeWorkflowTool {
             %name,
             node_count = graph.nodes.len(),
             require_approval,
+            warning_count = warnings.len(),
             "[flows] propose_workflow: proposal ready for user review"
         );
 
@@ -193,10 +201,110 @@ impl Tool for ProposeWorkflowTool {
             "graph": graph_value,
             "require_approval": require_approval,
             "summary": summary,
+            "warnings": warnings,
         }))?))
     }
 }
 
+/// Runs a **saved** workflow by id so the `workflow-builder` agent can *test*
+/// it end-to-end. Unlike [`crate::openhuman::flows::builder_tools::DryRunWorkflowTool`]
+/// (a MOCK sandbox), this is a **real** run — so it is `PermissionLevel::Write`
+/// with `external_effect() == true`. Two safety layers remain: the flow's own
+/// `require_approval` gate still pauses outbound-action nodes mid-run, and the
+/// agent prompt requires it to ASK THE USER for confirmation before ever
+/// calling this. It only runs an already-persisted flow (no `flow_id`, no run).
+pub struct RunFlowTool {
+    config: Arc<Config>,
+}
+
+impl RunFlowTool {
+    pub fn new(config: Arc<Config>) -> Self {
+        Self { config }
+    }
+}
+
+#[async_trait]
+impl Tool for RunFlowTool {
+    fn name(&self) -> &str {
+        "run_workflow"
+    }
+
+    fn description(&self) -> &str {
+        "Run a SAVED workflow by id to TEST it end-to-end. This is a REAL run, not a \
+         simulation — real effects can fire (use dry_run_workflow for a safe MOCK run \
+         instead). It only works on a flow the user has already saved; pass its `flow_id`. \
+         You MUST ask the user to confirm and wait for an explicit 'yes' before calling this \
+         — never run a workflow unprompted. The flow's own approval gate still pauses \
+         outbound-action nodes. Params: { flow_id (required), input? }. Returns the run's \
+         status + any nodes paused for approval."
+    }
+
+    fn parameters_schema(&self) -> Value {
+        json!({
+            "type": "object",
+            "properties": {
+                "flow_id": {
+                    "type": "string",
+                    "description": "Id of the SAVED flow to run (the user must have saved it first)."
+                },
+                "input": {
+                    "description": "Optional trigger input passed to the run (defaults to {})."
+                }
+            },
+            "required": ["flow_id"]
+        })
+    }
+
+    fn permission_level(&self) -> PermissionLevel {
+        // A real run with real effects — gated like a Write-class action.
+        PermissionLevel::Write
+    }
+
+    fn external_effect(&self) -> bool {
+        true
+    }
+
+    async fn execute(&self, args: Value) -> anyhow::Result<ToolResult> {
+        let flow_id =
+            match args.get("flow_id").and_then(Value::as_str).map(str::trim) {
+                Some(id) if !id.is_empty() => id.to_string(),
+                _ => return Ok(ToolResult::error(
+                    "Missing 'flow_id' — run_workflow only works on a SAVED flow. Ask the user \
+                     to Save the workflow first, then run it by id."
+                        .to_string(),
+                )),
+            };
+        let input = args.get("input").cloned().unwrap_or_else(|| json!({}));
+
+        tracing::info!(
+            target: "flows",
+            %flow_id,
+            "[flows] run_workflow: agent-initiated test run starting"
+        );
+
+        match crate::openhuman::flows::ops::flows_run(
+            &self.config,
+            &flow_id,
+            input,
+            crate::openhuman::flows::types::FlowRunTrigger::Rpc,
+        )
+        .await
+        {
+            Ok(outcome) => Ok(ToolResult::success(serde_json::to_string_pretty(&json!({
+                "type": "workflow_run_result",
+                "flow_id": flow_id,
+                "result": outcome.value,
+            }))?)),
+            Err(e) => {
+                tracing::debug!(target: "flows", %flow_id, error = %e, "[flows] run_workflow: failed");
+                Ok(ToolResult::error(format!(
+                    "Could not run flow '{flow_id}': {e}"
+                )))
+            }
+        }
+    }
+}
+
 /// Builds the `{ trigger, steps }` summary surfaced to both the LLM (in the
 /// tool result) and the chat UI's `WorkflowProposalCard`.
 ///

src/openhuman/flows/types.rs

@@ -141,6 +141,13 @@ pub struct FlowRunStep {
     /// observed incrementally. `None` for a step recovered post-hoc.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub duration_ms: Option<u64>,
+    /// Data-binding diagnostics from the engine: each config `=`-expression
+    /// that resolved to `null` during this step, as
+    /// `{ "location": "args.to", "expression": "=item.to" }`. Lets the run
+    /// view point at the exact unresolved wiring. Empty for clean steps and
+    /// for steps recovered post-hoc.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub diagnostics: Vec<serde_json::Value>,
 }
 
 /// A resolvable connection the flows UI / agent picker can attach to a node's

src/openhuman/runtime_node/ops.rs

@@ -6,8 +6,8 @@ use crate::openhuman::agent::host_runtime::{NativeRuntime, RuntimeAdapter};
 use crate::openhuman::config::Config;
 use crate::openhuman::memory::Memory;
 use crate::openhuman::runtime_node::types::{ExecuteToolOutcome, RuntimeToolSummary};
-use crate::openhuman::security::SecurityPolicy;
-use crate::openhuman::tools::{self, Tool, ToolCallOptions, ToolScope};
+use crate::openhuman::security::{CommandClass, SecurityPolicy};
+use crate::openhuman::tools::{self, PermissionLevel, Tool, ToolCallOptions, ToolScope};
 use tracing::{debug, trace};
 
 fn tool_scope_label(scope: ToolScope) -> &'static str {
@@ -30,6 +30,51 @@ fn summarize_tool(tool: &dyn Tool) -> RuntimeToolSummary {
     }
 }
 
+fn classify_shell_tool_call(security: &SecurityPolicy, args: &serde_json::Value) -> CommandClass {
+    let command = args
+        .get("command")
+        .and_then(|value| value.as_str())
+        .unwrap_or("");
+    let mut class = security.classify_command(command);
+    if let Some(declared) = args
+        .get("category")
+        .and_then(|value| value.as_str())
+        .and_then(SecurityPolicy::parse_declared_class)
+    {
+        class = class.max(declared);
+    }
+    class
+}
+
+fn command_class_for_tool(
+    security: &SecurityPolicy,
+    tool: &dyn Tool,
+    args: &serde_json::Value,
+) -> CommandClass {
+    if tool.name() == "shell" {
+        return classify_shell_tool_call(security, args);
+    }
+
+    let permission = tool.permission_level_with_args(args);
+    match permission {
+        PermissionLevel::None | PermissionLevel::ReadOnly => {
+            if tool.external_effect_with_args(args) {
+                CommandClass::Network
+            } else {
+                CommandClass::Read
+            }
+        }
+        PermissionLevel::Write | PermissionLevel::Execute => {
+            if tool.external_effect_with_args(args) {
+                CommandClass::Network
+            } else {
+                CommandClass::Write
+            }
+        }
+        PermissionLevel::Dangerous => CommandClass::Destructive,
+    }
+}
+
 fn build_runtime_tools(config: &Config) -> Result<Vec<Box<dyn Tool>>, String> {
     debug!(
         workspace = %config.workspace_dir.display(),
@@ -104,6 +149,36 @@ pub fn list_tools(config: &Config) -> Result<Vec<RuntimeToolSummary>, String> {
     Ok(summaries)
 }
 
+pub fn classify_tool_call(
+    config: &Config,
+    tool_name: &str,
+    args: &serde_json::Value,
+) -> Result<CommandClass, String> {
+    debug!(tool_name, "[runtime_node::ops] classify_tool_call: start");
+    let security =
+        SecurityPolicy::from_config(&config.autonomy, &config.workspace_dir, &config.action_dir);
+    let tools = build_runtime_tools(config)?;
+    let tool = tools
+        .into_iter()
+        .find(|tool| tool.name() == tool_name)
+        .ok_or_else(|| {
+            debug!(
+                tool_name,
+                "[runtime_node::ops] classify_tool_call: tool not found"
+            );
+            format!("unknown tool `{tool_name}`")
+        })?;
+    let class = command_class_for_tool(&security, tool.as_ref(), args);
+    debug!(
+        tool_name,
+        ?class,
+        permission = %tool.permission_level_with_args(args),
+        external_effect = tool.external_effect_with_args(args),
+        "[runtime_node::ops] classify_tool_call: done"
+    );
+    Ok(class)
+}
+
 pub async fn execute_tool(
     config: &Config,
     tool_name: &str,
@@ -217,6 +292,42 @@ mod tests {
         }
     }
 
+    struct PolicyTool {
+        name: &'static str,
+        permission: PermissionLevel,
+        external_effect: bool,
+    }
+
+    #[async_trait]
+    impl Tool for PolicyTool {
+        fn name(&self) -> &str {
+            self.name
+        }
+
+        fn description(&self) -> &str {
+            "Policy test tool"
+        }
+
+        fn parameters_schema(&self) -> serde_json::Value {
+            json!({"type": "object"})
+        }
+
+        fn permission_level(&self) -> PermissionLevel {
+            self.permission
+        }
+
+        fn external_effect_with_args(&self, _args: &serde_json::Value) -> bool {
+            self.external_effect
+        }
+
+        async fn execute(
+            &self,
+            _args: serde_json::Value,
+        ) -> anyhow::Result<crate::openhuman::skills::types::ToolResult> {
+            Ok(crate::openhuman::skills::types::ToolResult::success("ok"))
+        }
+    }
+
     #[test]
     fn summarize_tool_exposes_metadata() {
         let summary = summarize_tool(&DummyTool);
@@ -232,4 +343,84 @@ mod tests {
         assert_eq!(tool_scope_label(ToolScope::AgentOnly), "agent_only");
         assert_eq!(tool_scope_label(ToolScope::CliRpcOnly), "cli_rpc_only");
     }
+
+    #[test]
+    fn command_class_for_tool_maps_metadata_to_policy_buckets() {
+        let security = SecurityPolicy::default();
+
+        let read = PolicyTool {
+            name: "read_tool",
+            permission: PermissionLevel::ReadOnly,
+            external_effect: false,
+        };
+        assert_eq!(
+            command_class_for_tool(&security, &read, &json!({})),
+            CommandClass::Read
+        );
+
+        let write = PolicyTool {
+            name: "write_tool",
+            permission: PermissionLevel::Write,
+            external_effect: false,
+        };
+        assert_eq!(
+            command_class_for_tool(&security, &write, &json!({})),
+            CommandClass::Write
+        );
+
+        let outbound = PolicyTool {
+            name: "outbound_tool",
+            permission: PermissionLevel::Write,
+            external_effect: true,
+        };
+        assert_eq!(
+            command_class_for_tool(&security, &outbound, &json!({})),
+            CommandClass::Network
+        );
+
+        let dangerous = PolicyTool {
+            name: "dangerous_tool",
+            permission: PermissionLevel::Dangerous,
+            external_effect: true,
+        };
+        assert_eq!(
+            command_class_for_tool(&security, &dangerous, &json!({})),
+            CommandClass::Destructive
+        );
+    }
+
+    #[test]
+    fn command_class_for_shell_uses_command_args() {
+        let security = SecurityPolicy::default();
+        let shell = PolicyTool {
+            name: "shell",
+            permission: PermissionLevel::Execute,
+            external_effect: true,
+        };
+
+        assert_eq!(
+            command_class_for_tool(&security, &shell, &json!({"command": "ls src"})),
+            CommandClass::Read
+        );
+        assert_eq!(
+            command_class_for_tool(&security, &shell, &json!({"command": "touch out.txt"})),
+            CommandClass::Write
+        );
+        assert_eq!(
+            command_class_for_tool(
+                &security,
+                &shell,
+                &json!({"command": "curl https://example.com"})
+            ),
+            CommandClass::Network
+        );
+        assert_eq!(
+            command_class_for_tool(
+                &security,
+                &shell,
+                &json!({"command": "cat Cargo.toml", "category": "destructive"})
+            ),
+            CommandClass::Destructive
+        );
+    }
 }

src/openhuman/tinyflows/caps.rs

@@ -198,12 +198,60 @@ fn escalated_origin_for_prompt(
     }
 }
 
+/// Returns true when an agent-node completion `request` asked for structured
+/// output: an `output_parser.schema` is configured on the node, or the config
+/// sets `response_format: "json"`.
+///
+/// This is the host-side contract for **agent → tool wiring**: downstream
+/// `=item.<field>` bindings only work when the agent's emitted item is a
+/// structured object, so an agent feeding a `tool_call` should declare an
+/// output schema (or `response_format: "json"`).
+fn structured_output_requested(request: &Value) -> bool {
+    let has_schema = request
+        .get("output_parser")
+        .and_then(|p| p.get("schema"))
+        .is_some_and(|s| !s.is_null());
+    let json_format = request.get("response_format").and_then(Value::as_str) == Some("json");
+    has_schema || json_format
+}
+
+/// Best-effort parse of an LLM completion as structured JSON.
+///
+/// Accepts a bare JSON object/array or one wrapped in a markdown code fence
+/// (```json … ``` or ``` … ```). Returns `None` for anything that doesn't
+/// parse to an object or array — scalars pass through the legacy `{text}`
+/// shape instead, since `item.<field>` addressing is meaningless on them.
+pub(crate) fn parse_llm_json(text: &str) -> Option<Value> {
+    let trimmed = text.trim();
+    let candidate = match trimmed.strip_prefix("```") {
+        Some(rest) => {
+            let rest = rest.strip_prefix("json").unwrap_or(rest);
+            match rest.rsplit_once("```") {
+                Some((inner, _)) => inner.trim(),
+                None => trimmed,
+            }
+        }
+        None => trimmed,
+    };
+    let parsed = serde_json::from_str::<Value>(candidate).ok()?;
+    matches!(parsed, Value::Object(_) | Value::Array(_)).then_some(parsed)
+}
+
 /// [`LlmProvider`] adapter over OpenHuman's inference stack
 /// (`src/openhuman/inference/provider/`).
 ///
 /// The `agent` node is single-completion in tinyflows 0.2 (no tool-calling
 /// loop, no sub-ports), so `complete` performs exactly one `provider.chat`
 /// call and returns its result — no agent loop is driven here.
+///
+/// **Structured output**: when the node requested it (an
+/// `output_parser.schema` or `response_format: "json"` in the config), the
+/// completion text is parsed as JSON and the **parsed object** is returned as
+/// the response value — so a downstream node can bind `=item.<field>` (or
+/// `=nodes.<agent_id>.item.<field>`) instead of receiving an opaque
+/// `{text: "..."}` blob. A completion that doesn't parse falls back to the
+/// legacy shape, where the agent node's `output_parser` sub-port can still
+/// coerce it via the schema auto-fix path.
 pub struct OpenHumanLlm {
     pub config: Arc<Config>,
 }
@@ -230,7 +278,10 @@ impl LlmProvider for OpenHumanLlm {
             .and_then(Value::as_u64)
             .and_then(|n| u32::try_from(n).ok());
 
-        let messages: Vec<ChatMessage> = match request.get("messages").and_then(Value::as_array) {
+        let structured = structured_output_requested(&request);
+
+        let mut messages: Vec<ChatMessage> = match request.get("messages").and_then(Value::as_array)
+        {
             Some(entries) if !entries.is_empty() => entries
                 .iter()
                 .filter_map(|entry| {
@@ -253,10 +304,29 @@ impl LlmProvider for OpenHumanLlm {
             }
         };
 
+        // Structured mode: steer the model toward parseable JSON. The schema
+        // (when configured) rides along so the model knows the exact shape.
+        if structured {
+            let mut instruction = "Respond with a single JSON object only — no prose, no \
+                                   markdown code fences."
+                .to_string();
+            if let Some(schema) = request
+                .get("output_parser")
+                .and_then(|p| p.get("schema"))
+                .filter(|s| !s.is_null())
+            {
+                instruction.push_str(&format!(
+                    " The object must match this JSON Schema:\n{schema}"
+                ));
+            }
+            messages.insert(0, ChatMessage::system(instruction));
+        }
+
         tracing::debug!(
             target: "flows",
             role,
             message_count = messages.len(),
+            structured,
             "[flows] llm.complete: dispatching agent-node completion"
         );
 
@@ -277,6 +347,26 @@ impl LlmProvider for OpenHumanLlm {
             .await
             .map_err(|e| EngineError::Capability(e.to_string()))?;
 
+        // Structured mode: surface the parsed object itself so downstream
+        // `=item.<field>` / `=nodes.<id>.item.<field>` bindings work. The
+        // agent node's output_parser sub-port then validates it against the
+        // configured schema (and auto-fixes when it doesn't parse here).
+        if structured {
+            if let Some(parsed) = response.text.as_deref().and_then(parse_llm_json) {
+                tracing::debug!(
+                    target: "flows",
+                    "[flows] llm.complete: structured output parsed from completion text"
+                );
+                return Ok(parsed);
+            }
+            tracing::warn!(
+                target: "flows",
+                "[flows] llm.complete: structured output requested but the completion did not \
+                 parse as JSON — falling back to the {{text}} shape (the output_parser sub-port \
+                 may still coerce it)"
+            );
+        }
+
         Ok(json!({
             "text": response.text,
             "tool_calls": response.tool_calls,
@@ -542,9 +632,220 @@ pub struct OpenHumanTools {
     pub config: Arc<Config>,
 }
 
+/// Prefix marking a `tool_call` node's slug as a NATIVE OpenHuman tool (the
+/// "Tool" node) rather than a Composio action (the "App action" node). e.g.
+/// `oh:web_search`. Native tools run through the same agent tool registry the
+/// assistant uses (`runtime_node::ops::execute_tool`), so a flow can call
+/// search / media generation / file / shell / etc. — the full toolset.
+pub(crate) const NATIVE_TOOL_PREFIX: &str = "oh:";
+
+/// Process-level cache for [`composio_required_args`]: toolkit → (uppercase
+/// action slug → required top-level arg names). One `list_tools` fetch per
+/// toolkit per process; schemas are effectively static within a session.
+static REQUIRED_ARGS_CACHE: std::sync::OnceLock<
+    std::sync::Mutex<
+        std::collections::HashMap<String, std::collections::HashMap<String, Vec<String>>>,
+    >,
+> = std::sync::OnceLock::new();
+
+/// Seeds the required-args cache for a toolkit — test hook so preflight
+/// behavior can be exercised without a live Composio backend.
+#[cfg(test)]
+pub(crate) fn seed_required_args_cache(
+    toolkit: &str,
+    entries: std::collections::HashMap<String, Vec<String>>,
+) {
+    REQUIRED_ARGS_CACHE
+        .get_or_init(Default::default)
+        .lock()
+        .expect("required-args cache poisoned")
+        .insert(toolkit.to_string(), entries);
+}
+
+/// Best-effort lookup of a Composio action's **required** top-level parameter
+/// names, from the toolkit's tool schemas (`parameters.required`).
+///
+/// Returns `None` when the schema is unavailable — unknown toolkit, client
+/// construction failure, or a failed/empty listing — so callers can skip the
+/// preflight rather than block execution on a catalog hiccup. Results are
+/// cached per toolkit for the life of the process.
+pub(crate) async fn composio_required_args(config: &Config, slug: &str) -> Option<Vec<String>> {
+    let toolkit = crate::openhuman::memory_sync::composio::providers::toolkit_from_slug(slug)?;
+    let slug_key = slug.to_ascii_uppercase();
+
+    if let Some(by_slug) = REQUIRED_ARGS_CACHE
+        .get_or_init(Default::default)
+        .lock()
+        .ok()?
+        .get(&toolkit)
+    {
+        return by_slug.get(&slug_key).cloned();
+    }
+
+    tracing::debug!(target: "flows", %toolkit, %slug, "[flows] preflight: fetching tool schemas for toolkit");
+    let resp = crate::openhuman::composio::ops::composio_list_tools(
+        config,
+        Some(vec![toolkit.clone()]),
+        None,
+    )
+    .await
+    .map_err(|e| {
+        tracing::debug!(target: "flows", %toolkit, error = %e, "[flows] preflight: schema fetch failed — skipping check");
+        e
+    })
+    .ok()?;
+
+    let mut by_slug = std::collections::HashMap::new();
+    for tool in &resp.value.tools {
+        let required: Vec<String> = tool
+            .function
+            .parameters
+            .as_ref()
+            .and_then(|p| p.get("required"))
+            .and_then(Value::as_array)
+            .map(|arr| {
+                arr.iter()
+                    .filter_map(|v| v.as_str().map(str::to_string))
+                    .collect()
+            })
+            .unwrap_or_default();
+        by_slug.insert(tool.function.name.to_ascii_uppercase(), required);
+    }
+    let found = by_slug.get(&slug_key).cloned();
+    if let Ok(mut cache) = REQUIRED_ARGS_CACHE.get_or_init(Default::default).lock() {
+        cache.insert(toolkit, by_slug);
+    }
+    found
+}
+
+/// Returns the names in `required` that are absent or `null` in `args`.
+pub(crate) fn missing_required_args(required: &[String], args: &Value) -> Vec<String> {
+    required
+        .iter()
+        .filter(|name| match args.get(name.as_str()) {
+            None => true,
+            Some(v) => v.is_null(),
+        })
+        .cloned()
+        .collect()
+}
+
+/// Required-arg preflight for a Composio `tool_call`: fails **before** the
+/// Composio dispatch when a required arg is missing or resolved to `null`,
+/// with a message that names the field and the likely fix — instead of letting
+/// the raw provider error surface from deep inside the call.
+///
+/// Best-effort by design: when the action's schema cannot be looked up the
+/// check is skipped (never blocks on catalog availability).
+pub(crate) async fn preflight_composio_args(
+    config: &Config,
+    slug: &str,
+    args: &Value,
+) -> Result<()> {
+    let Some(required) = composio_required_args(config, slug).await else {
+        tracing::debug!(target: "flows", %slug, "[flows] preflight: no schema for action — skipping required-arg check");
+        return Ok(());
+    };
+    let missing = missing_required_args(&required, args);
+    if missing.is_empty() {
+        tracing::debug!(target: "flows", %slug, "[flows] preflight: all required args present");
+        return Ok(());
+    }
+    tracing::warn!(target: "flows", %slug, ?missing, "[flows] preflight: required arg(s) missing or null — failing before dispatch");
+    let list = missing
+        .iter()
+        .map(|m| format!("`{m}`"))
+        .collect::<Vec<_>>()
+        .join(", ");
+    let first = &missing[0];
+    Err(EngineError::Capability(format!(
+        "tool_call `{slug}`: required arg(s) {list} missing or resolved to null — wire each from \
+         an upstream node's output, e.g. \"{first}\": \"=nodes.<node_id>.item.<field>\". If the \
+         value comes from an agent node, give that agent an output schema \
+         (config.output_parser.schema) so its fields are addressable."
+    )))
+}
+
+/// A [`ToolInvoker`] decorator that runs the host's Composio required-arg
+/// preflight before delegating to `inner`.
+///
+/// Used by `dry_run_workflow`: the dry-run path executes against tinyflows'
+/// echo mocks, which would happily accept a `null` required arg — wrapping
+/// the mock invoker with this makes the wiring check actually check wiring,
+/// so an unwired required arg fails the dry run with the same actionable
+/// message a real run would produce.
+pub struct PreflightToolInvoker {
+    /// Host config, for the Composio schema lookup.
+    pub config: Arc<Config>,
+    /// The delegate that performs the actual invocation (e.g. the mock).
+    pub inner: Arc<dyn ToolInvoker>,
+}
+
+#[async_trait]
+impl ToolInvoker for PreflightToolInvoker {
+    async fn invoke(&self, slug: &str, args: Value, conn: Option<&str>) -> Result<Value> {
+        if !slug.starts_with(NATIVE_TOOL_PREFIX) {
+            preflight_composio_args(&self.config, slug, &args).await?;
+        }
+        self.inner.invoke(slug, args, conn).await
+    }
+}
+
 #[async_trait]
 impl ToolInvoker for OpenHumanTools {
     async fn invoke(&self, slug: &str, args: Value, conn: Option<&str>) -> Result<Value> {
+        // Native OpenHuman tool path (the "Tool" node): `oh:<tool_name>`. Bypasses
+        // the Composio curation gate (it isn't a Composio slug) but still runs
+        // through the autonomy-tier + approval gates, then dispatches to the
+        // agent tool registry.
+        if let Some(tool_name) = slug.strip_prefix(NATIVE_TOOL_PREFIX) {
+            let tool_name = tool_name.trim();
+            if tool_name.is_empty() {
+                return Err(EngineError::Capability(
+                    "tool_call node: native tool slug is empty (expected `oh:<tool_name>`)"
+                        .to_string(),
+                ));
+            }
+
+            let security = SecurityPolicy::from_config(
+                &self.config.autonomy,
+                &self.config.workspace_dir,
+                &self.config.action_dir,
+            );
+            let class = crate::openhuman::runtime_node::ops::classify_tool_call(
+                &self.config,
+                tool_name,
+                &args,
+            )
+            .map_err(EngineError::Capability)?;
+            let tier_decision = enforce_node_tier_gate(&security, class, "tool_call")?;
+            let summary = crate::openhuman::approval::summarize_action(tool_name, &args);
+            let redacted = crate::openhuman::approval::redact_args(&args);
+            let (outcome, _request_id) =
+                gate_call_for_tier(tier_decision, tool_name, &summary, redacted).await;
+            if let crate::openhuman::approval::GateOutcome::Deny { reason } = outcome {
+                return Err(EngineError::Capability(reason));
+            }
+            tracing::debug!(
+                target: "flows",
+                %tool_name,
+                ?class,
+                ?tier_decision,
+                "[flows] tool_call: dispatching NATIVE OpenHuman tool"
+            );
+            let outcome = crate::openhuman::runtime_node::ops::execute_tool(
+                &self.config,
+                tool_name,
+                args,
+                false,
+            )
+            .await
+            .map_err(EngineError::Capability)?;
+            return serde_json::to_value(&outcome.result).map_err(|e| {
+                EngineError::Capability(format!("could not serialize tool result: {e}"))
+            });
+        }
+
         // Curation + scope gate — hard allowlist (see [`is_curated_flow_tool`]'s
         // doc for why this differs from the general agent tool-call path).
         // Runs before anything else — a rejected slug never reaches the
@@ -561,6 +862,12 @@ impl ToolInvoker for OpenHumanTools {
             )));
         }
 
+        // Required-arg preflight — fail with an actionable, field-naming error
+        // BEFORE the approval gate and the Composio dispatch, so a mis-wired
+        // arg (`=`-expression that resolved to null) never reaches the
+        // provider or asks the user to approve a call that cannot succeed.
+        preflight_composio_args(&self.config, slug, &args).await?;
+
         // Approval gate (see the struct doc). Mirrors
         // `tinyagents/middleware.rs::ApprovalSecurityMiddleware::wrap_tool`'s
         // shape exactly: compute summary/redacted args only when a gate is

src/openhuman/tinyflows/observability.rs

@@ -125,12 +125,27 @@ impl RunObserver for FlowRunObserver {
         // Persist the live step. `duration_ms` is `u128` on the engine side;
         // clamp into `u64` (a node executor taking > 584 million years is not a
         // real concern, but never panic on cast).
+        if !step.diagnostics.is_empty() {
+            tracing::warn!(
+                target: "flows",
+                flow_id = %self.flow_id,
+                run_id = %self.run_id,
+                node = %step.node_id,
+                diagnostics = ?step.diagnostics,
+                "[flows] observer: step reported null-resolved expression(s) — persisting for the run view"
+            );
+        }
         let flow_step = FlowRunStep {
             node_id: step.node_id.clone(),
             output: step.output.clone(),
             port: None,
             status: Some(status.to_string()),
             duration_ms: Some(u64::try_from(step.duration_ms).unwrap_or(u64::MAX)),
+            diagnostics: step
+                .diagnostics
+                .iter()
+                .map(|d| serde_json::to_value(d).unwrap_or(serde_json::Value::Null))
+                .collect(),
         };
         if let Err(e) = upsert_flow_run_step(&self.config, &self.run_id, &flow_step) {
             tracing::warn!(
@@ -191,6 +206,7 @@ mod tests {
             status: StepStatus::Success,
             output: Value::Null,
             duration_ms: 5,
+            diagnostics: Vec::new(),
         });
         observer.on_run_finish(&Run {
             id: "run-1".to_string(),

src/openhuman/tinyflows/tests.rs

@@ -354,3 +354,137 @@ fn http_cred_name_returns_none_for_non_http_cred_ref_or_empty_name() {
     assert_eq!(super::caps::http_cred_name("composio:slack:acct_1"), None);
     assert_eq!(super::caps::http_cred_name("http_cred:"), None);
 }
+
+// ── structured agent output (parse_llm_json) ────────────────────────────
+
+#[test]
+fn parse_llm_json_accepts_bare_and_fenced_objects() {
+    let obj = super::caps::parse_llm_json(r#"{ "to": "a@b.com", "subject": "hi" }"#)
+        .expect("bare object parses");
+    assert_eq!(obj["to"], "a@b.com");
+
+    let fenced = "```json\n{ \"to\": \"a@b.com\" }\n```";
+    let obj = super::caps::parse_llm_json(fenced).expect("fenced object parses");
+    assert_eq!(obj["to"], "a@b.com");
+
+    let fenced_plain = "```\n[1, 2]\n```";
+    assert_eq!(
+        super::caps::parse_llm_json(fenced_plain),
+        Some(serde_json::json!([1, 2]))
+    );
+}
+
+#[test]
+fn parse_llm_json_rejects_prose_and_scalars() {
+    // Prose is not JSON.
+    assert_eq!(super::caps::parse_llm_json("Sure! Here's the email."), None);
+    // Scalars parse as JSON but are not addressable — legacy shape instead.
+    assert_eq!(super::caps::parse_llm_json("42"), None);
+    assert_eq!(super::caps::parse_llm_json("\"just a string\""), None);
+}
+
+// ── tool_call required-arg preflight ─────────────────────────────────────
+
+#[test]
+fn missing_required_args_flags_absent_and_null() {
+    let required = vec!["to".to_string(), "subject".to_string(), "body".to_string()];
+    let args = json!({ "to": null, "subject": "hi" });
+    assert_eq!(
+        super::caps::missing_required_args(&required, &args),
+        vec!["to".to_string(), "body".to_string()]
+    );
+    let full = json!({ "to": "a@b.com", "subject": "hi", "body": "text" });
+    assert!(super::caps::missing_required_args(&required, &full).is_empty());
+}
+
+#[tokio::test]
+async fn preflight_fails_before_dispatch_naming_the_missing_field() {
+    let tmp = TempDir::new().unwrap();
+    let config = test_config(&tmp);
+    // Seed the schema cache so no live Composio backend is needed.
+    let mut entries = std::collections::HashMap::new();
+    entries.insert(
+        "GMAIL_SEND_EMAIL".to_string(),
+        vec!["to".to_string(), "subject".to_string(), "body".to_string()],
+    );
+    super::caps::seed_required_args_cache("gmail", entries);
+
+    // `to` resolved to null (the classic mis-wired agent → tool_call case).
+    let err = super::caps::preflight_composio_args(
+        &config,
+        "GMAIL_SEND_EMAIL",
+        &json!({ "to": null, "subject": "hi", "body": "text" }),
+    )
+    .await
+    .expect_err("null required arg must fail preflight");
+    let msg = err.to_string();
+    assert!(msg.contains("`to`"), "error must name the field: {msg}");
+    assert!(
+        msg.contains("=nodes.<node_id>.item.<field>"),
+        "error must suggest the wiring fix: {msg}"
+    );
+    assert!(
+        msg.contains("output schema"),
+        "error must mention the agent output schema rule: {msg}"
+    );
+
+    // Fully-wired args pass.
+    super::caps::preflight_composio_args(
+        &config,
+        "GMAIL_SEND_EMAIL",
+        &json!({ "to": "a@b.com", "subject": "hi", "body": "text" }),
+    )
+    .await
+    .expect("wired args must pass preflight");
+}
+
+#[tokio::test]
+async fn preflight_skips_when_no_schema_is_available() {
+    let tmp = TempDir::new().unwrap();
+    let config = test_config(&tmp);
+    // A slug whose toolkit is unknown to the curation catalog has no schema
+    // source at all — the preflight must skip, never block.
+    super::caps::preflight_composio_args(&config, "NOT_A_REAL_TOOLKIT_ACTION", &json!({}))
+        .await
+        .expect("preflight must be best-effort when no schema is available");
+}
+
+#[tokio::test]
+async fn preflight_invoker_gates_the_mock_tool_path() {
+    use tinyflows::caps::ToolInvoker as _;
+
+    let tmp = TempDir::new().unwrap();
+    let config = test_config(&tmp);
+    let mut entries = std::collections::HashMap::new();
+    entries.insert("GMAIL_SEND_EMAIL".to_string(), vec!["to".to_string()]);
+    super::caps::seed_required_args_cache("gmail", entries);
+
+    let mock = tinyflows::caps::mock::mock_capabilities();
+    let invoker = super::caps::PreflightToolInvoker {
+        config,
+        inner: mock.tools.clone(),
+    };
+
+    // Unwired required arg: fails with the named field even though the inner
+    // mock would echo anything.
+    let err = invoker
+        .invoke("GMAIL_SEND_EMAIL", json!({ "to": null }), None)
+        .await
+        .expect_err("dry-run preflight must catch the unwired arg");
+    assert!(err.to_string().contains("`to`"));
+
+    // Wired arg: delegates to the mock echo.
+    let ok = invoker
+        .invoke("GMAIL_SEND_EMAIL", json!({ "to": "a@b.com" }), None)
+        .await
+        .expect("wired arg passes through to the mock");
+    assert_eq!(ok["tool"], "GMAIL_SEND_EMAIL");
+
+    // Native `oh:` slugs bypass the Composio preflight (no Composio schema).
+    // The mock echoes them unchecked.
+    let ok = invoker
+        .invoke("oh:web_search", json!({}), None)
+        .await
+        .expect("native slug bypasses composio preflight");
+    assert_eq!(ok["tool"], "oh:web_search");
+}

src/openhuman/tools/ops.rs

@@ -286,7 +286,11 @@ pub fn all_tools_with_runtime(
         Box::new(GetFlowRunTool::new(config.clone())),
         Box::new(ListFlowConnectionsTool::new(config.clone())),
         Box::new(SearchToolCatalogTool::new()),
-        Box::new(DryRunWorkflowTool::new(security.clone())),
+        Box::new(DryRunWorkflowTool::new(security.clone(), config.clone())),
+        // Real end-to-end test run of a SAVED flow (Write / external-effect). The
+        // workflow-builder prompt requires it to ask the user for confirmation
+        // first, and the flow's own approval gate still pauses outbound nodes.
+        Box::new(RunFlowTool::new(config.clone())),
         // Flow Scout discovery: the `flow_discovery` agent's terminal emit
         // sink. Read-only reasoning over the user's data ends by calling
         // `suggest_workflows`, which persists workflow ideas for the Flows page