diff --git a/src/openhuman/agent_registry/agents/loader.rs b/src/openhuman/agent_registry/agents/loader.rs index 7fd4eca00d..bea67fa686 100644 --- a/src/openhuman/agent_registry/agents/loader.rs +++ b/src/openhuman/agent_registry/agents/loader.rs @@ -1042,16 +1042,27 @@ mod tests { let expected = [ "propose_workflow", "revise_workflow", + "edit_workflow", + "validate_workflow", "save_workflow", "list_flows", "get_flow", + "get_flow_history", "get_flow_run", "list_flow_connections", "search_tool_catalog", "get_tool_contract", "get_tool_output_sample", "list_agent_profiles", + "list_connectable_toolkits", + "list_node_kinds", + "get_node_kind_contract", "dry_run_workflow", + "list_flow_runs", + "resume_flow_run", + "cancel_flow_run", + "create_workflow", + "duplicate_flow", "run_flow", "composio_list_toolkits", "composio_list_connections", diff --git a/src/openhuman/flows/agents/workflow_builder/prompt.md b/src/openhuman/flows/agents/workflow_builder/prompt.md index 7d8430866f..3a643e3496 100644 --- a/src/openhuman/flows/agents/workflow_builder/prompt.md +++ b/src/openhuman/flows/agents/workflow_builder/prompt.md @@ -13,11 +13,15 @@ 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. -- **`dry_run_workflow`** — runs a *draft* in a **sandbox** against mock +- **`dry_run_workflow`** — runs a graph in a **sandbox** against mock capabilities (deterministic echoes). Nothing real happens: no message is sent, - no code runs, no HTTP fires. Treat its output as a wiring check only. + no code runs, no HTTP fires. Treat its output as a wiring check only. Takes the + graph as any of `draft_id` / `flow_id` / an inline `graph` (precedence + `draft_id` > `flow_id` > `graph`). - **`save_workflow`** — the ONE persistence tool you have, and it only writes to - a flow that **already exists** (you need its `flow_id`). See below. + a flow that **already exists** (you need its `flow_id` as the target). Its + source is a `draft_id` (the usual case after iterating with `edit_workflow`) OR + an inline `graph`. See below. Persisting is otherwise the user's own action, not a tool you have — the one exception is `save_workflow` on an **existing** flow id, and only when the @@ -39,8 +43,9 @@ default. Your arc is: **When the user says "save it":** if you have a `save_workflow` action available — an **existing** `flow_id` plus their explicit ask ("save this", -"yes save it onto flow_X") — just call `save_workflow { flow_id, graph, -name? }` and confirm in one plain line what you saved (trigger, steps, and — +"yes save it onto flow_X") — just call `save_workflow { flow_id, draft_id, +name? }` (pass the `draft_id` you've been iterating on; an inline `graph` +also works) and confirm in one plain line what you saved (trigger, steps, and — if the flow is enabled with a schedule/app_event trigger — that it's now live and will fire on its own). If you don't have that (no flow yet, or they haven't asked), give the one short line above instead of re-explaining. @@ -91,13 +96,19 @@ rather than a general context recall), use `memory_hybrid_search` in its - `search_tool_catalog { query, toolkit? }` → real Composio action **slugs** from the FULL LIVE catalog for ANY named app — connected or not, curated or not (curated matches come back `featured: true` and are - ranked first). **Never hallucinate a slug** — if the catalog has no - match for the app, prefer an `http_request` node or tell the user the - integration isn't available. Each match also carries `required_args` / - `output_fields` / `primary_array_path` — but call `get_tool_contract - { slug }` before you actually WIRE a match: it hands back the exact - required args, the full input/output schema, and the array path a - `split_out` should use (see `tool_call` below). `propose_workflow` / + ranked first; a match may also carry `runtime_gated: true`, meaning that + action is blocked on real runs — prefer a `featured` one instead). + **Prefer ONE short keyword** (e.g. `gmail`, `send email`) for the widest + listing; a multi-word query that finds nothing no longer dead-ends — it + falls back to the nearest per-keyword matches with an explanatory `note`, + so read that note rather than assuming the app is missing. **Never + hallucinate a slug** — if the catalog genuinely has no match, prefer an + `http_request` node or tell the user the integration isn't available. Each + match also carries `required_args` / `output_fields` / `primary_array_path` + — but call `get_tool_contract { slug }` before you actually WIRE a match: it + hands back the exact required args, the full input/output schema, and the + array path a `split_out` should use (see `tool_call` below). + `propose_workflow` / `revise_workflow` / `save_workflow` HARD-REJECT a `tool_call` whose slug isn't real in the live catalog, or that's missing one of its real required args — so grounding here isn't optional polish, it's what @@ -117,13 +128,29 @@ You have a machine-readable belt; use it instead of relying on memory: gotchas. Consult these instead of guessing config shapes (this is the source of truth; the summary below is just orientation). - **Iterate cheaply:** once a draft exists, prefer `edit_workflow { draft_id | - flow_id | graph, ops[] }` (add_node / update_node_config[merge-patch] / - rename_node / add_edge / remove_edge / …) over re-emitting the whole graph - with `revise_workflow` — it's fewer tokens and won't drop a node or mangle an - edge. Edits to a `draft_id` are written back to the shared draft. -- **Check without proposing:** `validate_workflow { graph | flow_id }` runs the - same structural + hard-gate stack and returns every problem at once, so you - can self-verify mid-build without emitting a proposal card. + flow_id | graph, ops[] }` over re-emitting the whole graph with + `revise_workflow` — it's fewer tokens and won't drop a node or mangle an edge. + The op shapes (each is `{ "op": , … }`; `id` also accepts the alias + `node_id`, and `rename_node`'s `new_id` accepts `new_node_id`): + `add_node {node}` · `update_node_config {id, config}` (a JSON merge-patch — a + `null` value deletes that config key) · `set_node_name {id, name}` · + `rename_node {id, new_id}` (rewires edges) · `remove_node {id}` (drops its + edges) · `add_edge {edge}` · `remove_edge {from_node, to_node, from_port?, + to_port?}` · `set_node_position {id, position}`. Ops apply **strictly in array + order**, so to replace a node put its `remove_node` BEFORE the `add_node` (or + just `update_node_config` in place) — an "id already exists" error is almost + always that ordering slip. A bad op's error names the failing op index and the + exact shape that op wanted; fix and call again. + **Persistence:** `edit_workflow` NEVER saves. Editing a `flow_id` **seeds a new + draft** from that flow (the flow itself is untouched) and returns its + `draft_id`; editing a `draft_id` writes back to that same draft. The result + always carries `persisted: false` plus a `next` hint — keep iterating by + passing the returned `draft_id` to `edit_workflow` / `dry_run_workflow`, and + persist only on the user's explicit ask with `save_workflow { flow_id, + draft_id }`. A proposal is never a save. +- **Check without proposing:** `validate_workflow { draft_id | flow_id | graph }` + runs the same structural + hard-gate stack and returns every problem at once, + so you can self-verify mid-build without emitting a proposal card. - **Steer connections:** `list_connectable_toolkits` flags which toolkits are already connected — prefer those; the proposal's `required_connections` enumerates what still needs linking. @@ -300,6 +327,20 @@ A `WorkflowGraph` is `{ name?, nodes: [...], edges: [...] }`. slug isn't a real action in the live Composio catalog for its toolkit — a hallucinated or typo'd slug never makes it past validation, so always ground `config.slug` in a `search_tool_catalog` result first. + - **The `connection_ref` is enforced against the RIGHT toolkit.** + `config.connection_ref` must read `composio::` where + `` matches the slug's toolkit AND `` is one of the user's real + connections **for that toolkit** — get each ref verbatim from + `list_flow_connections`. Copying an id from a DIFFERENT toolkit (e.g. a + TikTok connection id onto a Gmail node) is HARD-REJECTED at + `propose_workflow`/`revise_workflow`/`save_workflow`, naming the correct + ref — so never reuse an id across toolkits. + - **`get_tool_contract` may return a top-level `runtime_gate` warning.** For + an uncurated action of a toolkit that ships a curated catalog, the real + runtime tool gate allows curated actions only, so that action is REJECTED + on every real run. Treat a `runtime_gate` warning as a **hard stop**: go + back to `search_tool_catalog` and pick a `featured: true` action instead of + wiring the gated one. - **Wiring a DOWNSTREAM node off THIS tool's output?** Don't guess the field name (e.g. assuming `GMAIL_FETCH_EMAILS` returns `.messages`) — `get_tool_contract`'s `output_fields` names the action's REAL top-level @@ -585,7 +626,11 @@ names, `dry_run_workflow` again, and repeat until it comes back clean. Only then call `propose_workflow` / `save_workflow`. Don't hand back a proposal you haven't verified just because the turn has run long — the user would rather wait one more tool call than review a graph that silently does -nothing. +nothing. **One exception:** a `null_resolutions` entry flagged `unverifiable: +true` (or an `unverifiable_bindings` list) is a Composio-upstream binding the +sandbox genuinely can't check — confirm it with `get_tool_contract` rather +than re-wiring, and don't loop on it (see "Interpreting dry-run results +honestly" below). ### Interpreting dry-run results honestly @@ -613,13 +658,27 @@ values appear: entry and the dry run returns `ok: false`. **These are real bugs — never dismiss them.** Fix every one before proposing. +3. **Unverifiable Composio-upstream bindings** — a `null_resolutions` entry + may carry `unverifiable: true` and an `upstream_tool_call` when the required + arg binds to the OUTPUT of a Composio `tool_call` node (an early-abort dry + run surfaces the same class as `unverifiable_bindings`). The echo sandbox + can never produce a Composio tool's real output fields, so this null does + **NOT** prove the binding wrong — it is genuinely unknowable here. Do **not** + thrash re-wiring it. Confirm the path against `get_tool_contract`'s + `output_fields` / `primary_array_path` (remember Composio results nest under + `.item.json.data.`), or `get_tool_output_sample { slug, args }` for the real + shape; it's a bug only if the path doesn't match the action's actual output. + The propose/save gate no longer blocks on this class, so a graph whose only + flag is `unverifiable` bindings you've confirmed is fine to propose. + #### Sandbox mock behavior per node type (authoritative — do NOT probe) | Node kind | Sandbox output | Enveloped? | What resolves downstream | |-----------|----------------|------------|---------------------------| | `trigger` | Passthrough — echoes the `input` value (default `{}`) | No | Whatever was passed as `input` | -| `agent` (with `output_parser.schema`) | Typed placeholder per schema property (`string`→`""`, `number`/`integer`→`0`, `boolean`→`false`, `object`→`{}`, `array`→`[]`, `enum`→its first listed value) | Yes | `=nodes..item.json.` → the placeholder (non-null) | -| `agent` (no schema) | `{ "agent": "", "request": {...}, "connection": ... }` | Yes | Only `.json.agent` / `.json.request` / `.json.connection` resolve; any other `.json.` → null | +| `agent` (with `output_parser.schema`) | Typed placeholder per schema property (`string`→`""`, `number`/`integer`→`0`, `boolean`→`false`, `object`→`{}`, `array`→`[]`, `enum`→its first listed value). Applies to **every** agent node — plain or with an `agent_ref` | Yes | `=nodes..item.json.` → the placeholder (non-null) | +| `agent` (no schema, plain — no `agent_ref`) | `{ "completion": , "connection": ... }` (the mock LLM echo) | Yes | Only `.json.completion` / `.json.connection` resolve; any other `.json.` → null | +| `agent` (no schema, with `agent_ref`) | `{ "agent": "", "request": {...}, "connection": ... }` | Yes | Only `.json.agent` / `.json.request` / `.json.connection` resolve; any other `.json.` → null | | `tool_call` | Required Composio args are preflight-checked first (missing/null → dry run fails before the mock even runs), then echoes `{ "tool": "", "args": {...}, "connection": ... }` — NOT a real API response | Yes | `.json.tool` / `.json.args` / `.json.connection` resolve; a response-shaped field (e.g. `.json.data.` for a real Composio call — see "the envelope" above) does **not**, because the mock echo carries no `data` wrapper. That is a mock-shape gap, not a wiring bug — don't "fix" a correctly-wired `.json.data.` binding just because the dry run can't resolve it | | `http_request` | `{ "status": 200, "request": {...}, "connection": ... }` | Yes | `.json.status` → `200`; response-body fields → null | | `code` | `{ "result": }` — the real `source` is NOT executed | **No** | `.item.result` resolves directly (no `.json.` — `code` does not envelope) | diff --git a/src/openhuman/flows/builder_tools.rs b/src/openhuman/flows/builder_tools.rs index 599d90b0ab..21a49b19a6 100644 --- a/src/openhuman/flows/builder_tools.rs +++ b/src/openhuman/flows/builder_tools.rs @@ -64,6 +64,30 @@ use crate::openhuman::tools::traits::{PermissionLevel, Tool, ToolResult}; /// capabilities are non-blocking echoes, so this is a generous safety net. const DRY_RUN_TIMEOUT_SECS: u64 = 30; +/// Comma list of the valid `op` tag values, for the missing-/unknown-`op` +/// parse errors surfaced by [`EditWorkflowTool`]. +const VALID_OP_TYPES: &str = "add_node, update_node_config, set_node_name, rename_node, \ + remove_node, add_edge, remove_edge, set_node_position"; + +/// The expected field shape for a given `op` tag, used in `edit_workflow`'s +/// per-op parse diagnostics so a failing op tells the agent exactly what that +/// op type wants. Returns `None` for an unrecognized tag. +fn edit_op_shape(op: &str) -> Option<&'static str> { + Some(match op { + "add_node" => "{ op, node: { id, kind, name, config? } }", + "update_node_config" => { + "{ op, id, config } (id also accepts alias `node_id`; config is a JSON merge-patch)" + } + "set_node_name" => "{ op, id, name } (id also accepts alias `node_id`)", + "rename_node" => "{ op, id, new_id } (also accept aliases `node_id` / `new_node_id`)", + "remove_node" => "{ op, id } (id also accepts alias `node_id`)", + "add_edge" => "{ op, edge: { from_node, to_node, from_port?, to_port? } }", + "remove_edge" => "{ op, from_node, to_node, from_port?, to_port? }", + "set_node_position" => "{ op, id, position: { x, y } } (id also accepts alias `node_id`)", + _ => return None, + }) +} + // ───────────────────────────────────────────────────────────────────────────── // revise_workflow — iterative refine of an existing draft (proposal only) // ───────────────────────────────────────────────────────────────────────────── @@ -191,6 +215,10 @@ impl Tool for ReviseWorkflowTool { require_approval, true, instruction, + // revise_workflow takes only an inline graph — no draft/flow handle + // to echo. The payload still carries persisted:false unconditionally. + None, + None, ) .await { @@ -240,10 +268,14 @@ impl Tool for EditWorkflowTool { {id, config} (JSON merge-patch — a null value deletes that config key), set_node_name \ {id, name}, rename_node {id, new_id} (rewires edges), remove_node {id} (drops its edges), \ add_edge {edge}, remove_edge {from_node, to_node, from_port?, to_port?}, set_node_position \ - {id, position}. Like propose/revise_workflow this ONLY VALIDATES and returns a proposal \ - for the user to review — it never creates, updates, or enables the flow. If an op fails or \ - the resulting graph is invalid, the error names the failing op / node; fix it and call \ - edit_workflow again." + {id, position}. PERSISTENCE: the applied edit is written to a DRAFT, never onto the saved \ + flow — this tool NEVER saves. Editing a flow_id SEEDS A NEW DRAFT from that flow's graph \ + and returns its `draft_id`; editing a draft_id writes back to that same draft. The result \ + carries `draft_id`, `flow_id` (if any), `persisted: false`, and a `next` hint. To keep \ + iterating pass that `draft_id` (to edit_workflow / dry_run_workflow); to persist, call \ + save_workflow { flow_id, draft_id } when the user asks. If an op fails or the resulting \ + graph is invalid, the error names the failing op / node; fix it and call edit_workflow \ + again." } fn parameters_schema(&self) -> Value { @@ -314,9 +346,16 @@ impl Tool for EditWorkflowTool { .filter(|s| !s.is_empty()); let inline_graph = args.get("graph").filter(|v| !v.is_null()); - // When editing a draft, remember its id so the applied edit is written - // back — the draft is the durable working copy across turns/reloads. + // The applied edit is always written back to a durable DRAFT (the shared + // working copy across turns/reloads). `write_back_draft` is the draft id + // it lands on; `edited_from_flow` is the saved flow this edit derives + // from / would persist onto, if any. The core WS2 fix: editing a bare + // `flow_id` used to persist NOTHING and return NO handle — the edit was + // unreachable and read as "written onto the flow". Now a `flow_id` base + // seeds a NEW draft, so the edit is durable, addressable, and clearly + // NOT the saved flow. let mut write_back_draft: Option = None; + let mut edited_from_flow: Option = None; let (base_graph, default_name) = match (draft_id, flow_id, inline_graph) { (Some(id), _, _) => match ops::flows_draft_get(&self.config, id) { @@ -325,6 +364,9 @@ impl Tool for EditWorkflowTool { match ops::migrate_and_deserialize_graph(draft.graph.clone()) { Ok(graph) => { write_back_draft = Some(draft.id.clone()); + // A draft may already be linked to a saved flow — + // carry that through so the proposal echoes it. + edited_from_flow = draft.flow_id.clone(); (graph, draft.name) } Err(e) => { @@ -341,7 +383,47 @@ impl Tool for EditWorkflowTool { } }, (None, Some(id), _) => match ops::flows_get(&self.config, id).await { - Ok(outcome) => (outcome.value.graph, outcome.value.name), + Ok(outcome) => { + let flow = outcome.value; + // Seed a NEW draft from the saved flow's graph so the edit is + // durable and reachable (the RPC/canvas path uses the same + // `flows_draft_create` op). Linking the draft to `flow.id` + // means a later save_workflow { flow_id, draft_id } knows its + // target. + let graph_json = match serde_json::to_value(&flow.graph) { + Ok(v) => v, + Err(e) => { + return Ok(ToolResult::error(format!( + "Could not serialize flow '{id}' to seed a draft: {e}" + ))); + } + }; + match ops::flows_draft_create( + &self.config, + Some(flow.id.clone()), + flow.name.clone(), + graph_json, + crate::openhuman::flows::DraftOrigin::Chat, + ) { + Ok(created) => { + let new_draft_id = created.value.id.clone(); + tracing::debug!( + target: "flows", + draft_id = %new_draft_id, + flow_id = %flow.id, + "[flows] edit_workflow: seeded a new draft from saved flow (edits live on the draft, NOT the flow)" + ); + write_back_draft = Some(new_draft_id); + edited_from_flow = Some(flow.id.clone()); + (flow.graph, flow.name) + } + Err(e) => { + return Ok(ToolResult::error(format!( + "Could not create a draft to edit flow '{id}': {e}" + ))); + } + } + } Err(e) => { return Ok(ToolResult::error(format!( "Could not load flow '{id}' to edit: {e}" @@ -370,31 +452,46 @@ impl Tool for EditWorkflowTool { } }; - // Parse the ops list. - let ops_value = match args.get("ops") { - Some(v) if v.is_array() => v.clone(), + // Parse the ops list element-by-element so a bad op reports its index, + // its `op` tag, the serde error, AND the expected field shape for THAT + // op type — instead of a bare aggregate "missing field `id`" that names + // neither the failing op nor what it wanted (audit WS4). + let ops_array = match args.get("ops") { + Some(Value::Array(items)) => items.clone(), _ => { return Ok(ToolResult::error( "Missing 'ops' parameter (a non-empty array of structured edits).".to_string(), )); } }; - let graph_ops: Vec = match serde_json::from_value(ops_value) - { - Ok(ops) => ops, - Err(e) => { - return Ok(ToolResult::error(format!( - "Could not parse `ops`: {e}. Each op is {{ \"op\": , ... }} — valid \ - types: add_node, update_node_config, set_node_name, rename_node, \ - remove_node, add_edge, remove_edge, set_node_position." - ))); - } - }; - if graph_ops.is_empty() { + if ops_array.is_empty() { return Ok(ToolResult::error( "`ops` is empty — provide at least one edit.".to_string(), )); } + let mut graph_ops: Vec = Vec::with_capacity(ops_array.len()); + for (index, item) in ops_array.into_iter().enumerate() { + let op_tag = item.get("op").and_then(Value::as_str).map(str::to_string); + match serde_json::from_value::(item) { + Ok(op) => graph_ops.push(op), + Err(e) => { + let shape = match op_tag.as_deref() { + Some(tag) => match edit_op_shape(tag) { + Some(shape) => format!("op `{tag}` expects {shape}"), + None => { + format!("unknown op type `{tag}` — valid types: {VALID_OP_TYPES}") + } + }, + None => format!("missing `op` field — valid types: {VALID_OP_TYPES}"), + }; + tracing::debug!(target: "flows", index, ?op_tag, error = %e, "[flows] edit_workflow: op failed to parse"); + return Ok(ToolResult::error(format!( + "Could not parse op {index}: {e}. Expected {shape}. Each op is \ + {{ \"op\": , ... }}. Fix the ops and call edit_workflow again." + ))); + } + } + } let name = args .get("name") @@ -430,8 +527,22 @@ impl Tool for EditWorkflowTool { Ok(graph) => graph, Err(e) => { tracing::debug!(target: "flows", %name, error = %e, "[flows] edit_workflow: an op failed to apply"); + // Ops apply strictly in array order, so an add_node for an id + // that already exists is almost always an ordering mistake + // (adding before removing the old node). Point at the fix — this + // is the exact 2nd wasted call the WS4 audit caught. + let hint = match (e.op, &e.kind) { + ("add_node", tinyflows::graph_ops::GraphOpErrorKind::NodeIdExists(id)) => { + format!( + "\n\nOps apply strictly in array order. To replace node `{id}`, put a \ + remove_node op for it BEFORE the add_node, or use update_node_config \ + to patch it in place." + ) + } + _ => String::new(), + }; return Ok(ToolResult::error(format!( - "{e}\n\nFix the ops and call edit_workflow again." + "{e}{hint}\n\nFix the ops and call edit_workflow again." ))); } }; @@ -469,6 +580,8 @@ impl Tool for EditWorkflowTool { } // 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. match ops::build_builder_proposal( &self.config, "edit_workflow", @@ -477,12 +590,32 @@ impl Tool for EditWorkflowTool { require_approval, true, instruction, + write_back_draft.clone(), + edited_from_flow.clone(), ) .await { Ok(mut payload) => { - if let Some(draft_id) = write_back_draft { - payload["draft_id"] = json!(draft_id); + // A prominent, one-line pointer at where the edit actually lives + // (the draft) vs. where it does NOT (the saved flow) — the exact + // confusion the WS2 audit caught. Only meaningful when the edit + // landed on a draft (inline-graph edits have no durable handle). + if let Some(draft_id) = write_back_draft.as_deref() { + let next = match edited_from_flow.as_deref() { + Some(flow_id) => format!( + "Edits live on draft {draft_id}, NOT on flow {flow_id}. Iterate with \ + edit_workflow/dry_run_workflow {{ draft_id: \"{draft_id}\" }}, then \ + persist with save_workflow {{ flow_id: \"{flow_id}\", draft_id: \ + \"{draft_id}\" }} when the user asks." + ), + None => format!( + "Edits live on draft {draft_id} (not yet linked to a saved flow). \ + Iterate with edit_workflow/dry_run_workflow {{ draft_id: \ + \"{draft_id}\" }}, then persist with create_workflow, or save_workflow \ + {{ flow_id, draft_id: \"{draft_id}\" }} once a flow exists." + ), + }; + payload["next"] = json!(next); } Ok(ToolResult::success(serde_json::to_string_pretty(&payload)?)) } @@ -524,25 +657,31 @@ impl Tool for ValidateWorkflowTool { fn description(&self) -> &str { "Check a workflow graph WITHOUT proposing or saving it — the same validation the \ propose/revise/edit/save tools run, surfaced on its own so you can verify a draft mid-\ - build. Provide the graph to check (inline `graph`, or `flow_id` for a saved flow). \ - Returns { ok, structurally_valid, errors, error_details:[{code, message, node_id}], \ - gate_errors, warnings }: `errors` lists EVERY structural problem at once; `gate_errors` \ - lists the hard author-gate failures (unresolvable bindings, unreal tool slugs, unwired \ - required args) checked only once the graph is structurally valid; `warnings` are \ - non-fatal. `ok` is true only when there are no errors and no gate_errors. Read-only." + build. Provide the graph to check as exactly one of `draft_id` (a working draft), \ + `flow_id` (a saved flow), or inline `graph` (if several are given, draft_id wins, then \ + flow_id). Returns { ok, structurally_valid, errors, error_details:[{code, message, \ + node_id}], gate_errors, warnings }: `errors` lists EVERY structural problem at once; \ + `gate_errors` lists the hard author-gate failures (unresolvable bindings, unreal tool \ + slugs, unwired required args) checked only once the graph is structurally valid; \ + `warnings` are non-fatal. `ok` is true only when there are no errors and no gate_errors. \ + Read-only." } fn parameters_schema(&self) -> Value { json!({ "type": "object", "properties": { + "draft_id": { + "type": "string", + "description": "A working draft to validate. Provide one of draft_id / flow_id / graph (draft_id wins)." + }, "flow_id": { "type": "string", - "description": "A saved flow to validate. Provide this OR `graph`." + "description": "A saved flow to validate. Provide one of draft_id / flow_id / graph." }, "graph": { "type": "object", - "description": "An inline tinyflows WorkflowGraph to validate. Provide this OR `flow_id`.", + "description": "An inline tinyflows WorkflowGraph to validate. Provide one of draft_id / flow_id / graph.", "properties": { "nodes": { "type": "array" }, "edges": { "type": "array" } @@ -561,7 +700,14 @@ impl Tool for ValidateWorkflowTool { } async fn execute(&self, args: Value) -> anyhow::Result { - // Resolve the graph to check from either a saved flow or an inline graph. + // Resolve the graph to check from exactly one of a working draft, a + // saved flow, or an inline graph — same precedence (draft_id > flow_id > + // graph) as edit_workflow, so the sibling tools accept the same handles. + let draft_id = args + .get("draft_id") + .and_then(Value::as_str) + .map(str::trim) + .filter(|s| !s.is_empty()); let flow_id = args .get("flow_id") .and_then(Value::as_str) @@ -569,8 +715,16 @@ impl Tool for ValidateWorkflowTool { .filter(|s| !s.is_empty()); let inline_graph = args.get("graph").filter(|v| !v.is_null()); - let graph_json = match (flow_id, inline_graph) { - (Some(id), _) => match ops::load_flow_graph(&self.config, id) { + let graph_json = match (draft_id, flow_id, inline_graph) { + (Some(id), _, _) => match ops::flows_draft_get(&self.config, id) { + Ok(outcome) => outcome.value.graph, + Err(e) => { + return Ok(ToolResult::error(format!( + "Could not load draft '{id}' to validate: {e}" + ))); + } + }, + (None, Some(id), _) => match ops::load_flow_graph(&self.config, id) { Ok(Some(graph)) => serde_json::to_value(&graph)?, Ok(None) => { return Ok(ToolResult::error(format!("flow '{id}' not found"))); @@ -581,11 +735,11 @@ impl Tool for ValidateWorkflowTool { ))); } }, - (None, Some(graph)) => graph.clone(), - (None, None) => { + (None, None, Some(graph)) => graph.clone(), + (None, None, None) => { return Ok(ToolResult::error( - "Provide either `flow_id` (a saved flow) or `graph` (an inline graph) to \ - validate." + "Provide one of `draft_id` (a working draft), `flow_id` (a saved flow), or \ + `graph` (an inline graph) to validate." .to_string(), )); } @@ -593,6 +747,7 @@ impl Tool for ValidateWorkflowTool { tracing::debug!( target: "flows", + from_draft = draft_id.is_some(), from_flow = flow_id.is_some(), "[flows] validate_workflow: checking graph (read-only)" ); @@ -1458,6 +1613,75 @@ pub(crate) async fn search_live_catalog( toolkit_filter: Option<&str>, limit: usize, ) -> Vec { + search_catalog(config, query, toolkit_filter, limit) + .await + .results +} + +/// Cap on fallback (per-keyword) matches — a near-miss query must not flood the +/// agent's context with the whole toolkit, so the OR-scored fallback returns at +/// most this many rows regardless of the primary `limit`. +const MAX_FALLBACK_RESULTS: usize = 10; + +/// Outcome of a catalog search: the shaped rows, whether the per-keyword +/// fallback pass fired, and an optional advisory `note` the tool surfaces so an +/// agent never misreads a keyword miss as "the action doesn't exist". +pub(crate) struct CatalogSearchOutcome { + pub results: Vec, + /// True when the per-token OR fallback pass ran (primary AND match was + /// empty for a multi-word query). + pub fallback: bool, + /// Advisory note explaining a near-miss / keyword-based search, if any. + pub note: Option, +} + +/// Shape one live-catalog [`ToolContract`](crate::openhuman::tinyflows::caps::ToolContract) +/// into a search-result row. The SINGLE row-construction site shared by both +/// the primary AND-match path and the per-keyword fallback path, so every row +/// carries the same fields — including WS3's `runtime_gated: true` on an +/// uncurated action of a toolkit that ships a curated-only allowlist. +fn shape_catalog_row( + tool: &crate::openhuman::tinyflows::caps::ToolContract, + toolkit: &str, + toolkit_curated: bool, +) -> Value { + let mut row = json!({ + "slug": tool.slug, + "toolkit": toolkit, + "description": tool.description, + "required_args": tool.required_args, + "output_fields": tool.output_fields, + "primary_array_path": tool.primary_array_path, + "featured": tool.is_curated, + }); + // Compact: only present when true. + if !tool.is_curated && toolkit_curated { + if let Some(obj) = row.as_object_mut() { + obj.insert("runtime_gated".to_string(), Value::Bool(true)); + } + } + row +} + +/// Search the FULL LIVE Composio catalog and return a [`CatalogSearchOutcome`]. +/// +/// Primary pass: case-insensitive AND — an action matches only if EVERY +/// whitespace-separated term substring-matches its slug, toolkit name, or +/// description (curated matches ranked first, stable sort preserves fetch +/// order). When that yields zero rows for a MULTI-WORD query, a per-keyword OR +/// fallback runs: each action is scored by how many query tokens match its +/// slug/toolkit/description, and the top [`MAX_FALLBACK_RESULTS`] (ranked by +/// hit-count desc, then curated first) are returned with an advisory `note`. +/// This is what keeps a natural-language query like "twitter tweet replies +/// lookup" from returning a bare `count: 0` even though `TWITTER_*` actions +/// exist — the agent gets the nearest keyword matches instead of falsely +/// concluding the action is missing. +pub(crate) async fn search_catalog( + config: &Config, + query: &str, + toolkit_filter: Option<&str>, + limit: usize, +) -> CatalogSearchOutcome { use crate::openhuman::memory_sync::composio::providers::agent_ready_toolkits; use crate::openhuman::tinyflows::caps::fetch_live_toolkit_catalog; @@ -1487,10 +1711,26 @@ pub(crate) async fn search_live_catalog( })) .await; + // Drop toolkits whose fetch failed (no backend session / network error) — + // they contribute zero results rather than erroring the whole search. + let fetched: Vec<(String, Vec)> = fetched + .into_iter() + .filter_map(|(tk, catalog)| catalog.map(|c| (tk, c))) + .collect(); + + // Does the scanned scope hold ANY actions at all? Distinguishes "keyword + // miss" (has actions, none matched) from "nothing to search" (empty scope). + let any_actions = fetched.iter().any(|(_, catalog)| !catalog.is_empty()); + + // ── Primary pass: case-insensitive AND across every term ── let mut matches: Vec<(bool, Value)> = Vec::new(); - for (toolkit, catalog) in fetched { - let Some(catalog) = catalog else { continue }; - for tool in &catalog { + for (toolkit, catalog) in &fetched { + // WS3 — a toolkit that ships a curated catalog is a hard curated-only + // allowlist at RUNTIME, so any `featured: false` action of it is + // rejected on every real run. Compute once per toolkit and flag those + // rows so the blocker is visible at search time (transcript failure #2). + let toolkit_curated = ops::toolkit_has_curated_catalog(toolkit); + for tool in catalog { let slug_lc = tool.slug.to_ascii_lowercase(); let desc_lc = tool .description @@ -1505,15 +1745,7 @@ pub(crate) async fn search_live_catalog( } matches.push(( tool.is_curated, - json!({ - "slug": tool.slug, - "toolkit": toolkit, - "description": tool.description, - "required_args": tool.required_args, - "output_fields": tool.output_fields, - "primary_array_path": tool.primary_array_path, - "featured": tool.is_curated, - }), + shape_catalog_row(tool, toolkit, toolkit_curated), )); } } @@ -1522,7 +1754,104 @@ pub(crate) async fn search_live_catalog( // within each group. matches.sort_by_key(|(is_curated, _)| std::cmp::Reverse(*is_curated)); matches.truncate(limit); - matches.into_iter().map(|(_, v)| v).collect() + let primary: Vec = matches.into_iter().map(|(_, v)| v).collect(); + + if !primary.is_empty() { + return CatalogSearchOutcome { + results: primary, + fallback: false, + note: None, + }; + } + + // ── Zero primary hits ── + // Single-token queries keep today's behavior exactly; only attach a light + // advisory note so a lone keyword miss still explains the search is + // keyword-based (task WS5.4, optional). + if terms.len() <= 1 { + let note = if any_actions { + Some(format!( + "No actions matched '{query}'. This search is keyword-based (matches action \ + slug/name/description) — try a different single keyword (e.g. 'gmail' or \ + 'tweets')." + )) + } else { + None + }; + return CatalogSearchOutcome { + results: Vec::new(), + fallback: false, + note, + }; + } + + // ── Fallback pass (multi-word, zero primary hits): per-token OR scoring ── + // Score each action by how many DISTINCT query tokens match its + // slug/toolkit/description; keep the primary path's curated boost as the + // tiebreak. Rows go through the SAME `shape_catalog_row` path as primary. + let mut scored: Vec<(usize, bool, Value)> = Vec::new(); + for (toolkit, catalog) in &fetched { + let toolkit_curated = ops::toolkit_has_curated_catalog(toolkit); + for tool in catalog { + let slug_lc = tool.slug.to_ascii_lowercase(); + let desc_lc = tool + .description + .as_deref() + .unwrap_or_default() + .to_ascii_lowercase(); + let hits = terms + .iter() + .filter(|term| { + slug_lc.contains(*term) || toolkit.contains(*term) || desc_lc.contains(*term) + }) + .count(); + if hits == 0 { + continue; + } + scored.push(( + hits, + tool.is_curated, + shape_catalog_row(tool, toolkit, toolkit_curated), + )); + } + } + + // Most keyword hits first, then curated first; stable sort preserves fetch + // order within a (hits, curated) group. + scored.sort_by_key(|(hits, is_curated, _)| std::cmp::Reverse((*hits, *is_curated))); + scored.truncate(limit.min(MAX_FALLBACK_RESULTS)); + let results: Vec = scored.into_iter().map(|(_, _, v)| v).collect(); + + tracing::debug!( + target: "flows", + query, + fallback = true, + hits = results.len(), + "[flows] search_tool_catalog: primary AND-match empty for a multi-word query — ran per-keyword OR fallback" + ); + + if results.is_empty() { + // Literally zero tokens matched anything: no rows, but a note so the + // agent doesn't read `count: 0` as "action doesn't exist" (task WS5.3). + return CatalogSearchOutcome { + results, + fallback: true, + note: Some(format!( + "No actions matched any keyword in '{query}'. This search is keyword-based \ + (matches action slug/name/description) — retry with a single keyword (e.g. one \ + word like 'gmail' or 'tweets') for a full listing." + )), + }; + } + + CatalogSearchOutcome { + results, + fallback: true, + note: Some(format!( + "No exact match for '{query}'. Showing the nearest per-keyword matches — retry with a \ + single keyword (e.g. one word like 'gmail' or 'tweets') for a full listing." + )), + } } #[async_trait] @@ -1553,7 +1882,7 @@ impl Tool for SearchToolCatalogTool { "properties": { "query": { "type": "string", - "description": "Keywords to match against tool slugs/descriptions (case-insensitive; all terms must match)." + "description": "Keywords to match against tool slugs/descriptions (case-insensitive). All terms must match for an exact hit; a multi-word query with no exact match falls back to the nearest per-keyword matches. For the widest listing, prefer ONE keyword (e.g. 'gmail' or 'tweets')." }, "toolkit": { "type": "string", @@ -1585,12 +1914,24 @@ impl Tool for SearchToolCatalogTool { toolkit = toolkit.unwrap_or("(any)"), "[flows] search_tool_catalog: searching the FULL LIVE Composio catalog (read-only)" ); - let results = search_live_catalog(&self.config, &query, toolkit, MAX_CATALOG_RESULTS).await; - Ok(ToolResult::success(serde_json::to_string_pretty(&json!({ - "query": query, - "count": results.len(), - "results": results, - }))?)) + let outcome = search_catalog(&self.config, &query, toolkit, MAX_CATALOG_RESULTS).await; + // Build with `note` first so an agent reading top-down sees the + // near-miss / keyword-based advisory before the (possibly zero) rows. + // `count` is always the number of returned rows, never a stand-in for + // "no such action" — a fallback carries a non-zero count. + let mut obj = serde_json::Map::new(); + if let Some(note) = outcome.note { + obj.insert("note".to_string(), Value::String(note)); + } + obj.insert("query".to_string(), Value::String(query)); + obj.insert( + "count".to_string(), + Value::Number(outcome.results.len().into()), + ); + obj.insert("results".to_string(), Value::Array(outcome.results)); + Ok(ToolResult::success(serde_json::to_string_pretty( + &Value::Object(obj), + )?)) } } @@ -1705,6 +2046,37 @@ impl Tool for GetToolContractTool { // permanently `None`. let contract = crate::openhuman::tinyflows::caps::apply_probe_override(contract.clone()); + + // WS3 — EARLY runtime-gate warning (transcript failure #2): a + // real-but-uncurated action of a toolkit that ships a curated + // catalog is a hard curated-only allowlist at RUNTIME, so it is + // REJECTED on every real run. The late `validate_workflow` gate + // catches it, but only ~15 tool calls after the agent has built + // and wired the node. Surface the blocker HERE, at contract-fetch + // time (and first in the payload), so the agent never wires it. + if !contract.is_curated && ops::toolkit_has_curated_catalog(&toolkit) { + tracing::debug!( + target: "flows", + %slug, + %toolkit, + "[flows] get_tool_contract: uncurated action of a curated toolkit — attaching runtime_gate warning" + ); + #[derive(serde::Serialize)] + struct ContractWithRuntimeGate { + runtime_gate: &'static str, + #[serde(flatten)] + contract: crate::openhuman::tinyflows::caps::ToolContract, + } + let payload = ContractWithRuntimeGate { + runtime_gate: "This action will be REJECTED on every real run — the \ + runtime tool gate only allows curated actions for this \ + toolkit. Pick a `featured: true` result from \ + search_tool_catalog instead.", + contract, + }; + return Ok(ToolResult::success(serde_json::to_string_pretty(&payload)?)); + } + Ok(ToolResult::success(serde_json::to_string_pretty( &contract, )?)) @@ -2170,6 +2542,79 @@ impl Tool for GetNodeKindContractTool { /// (an unexercised branch can be entirely intentional), and is surfaced on /// both the `ok: true` and `ok: false` result shapes so the caller can /// double-check that node's wiring by hand. +/// Builds one `null_resolutions` diagnostic entry for a `tool_call` node's +/// null-resolved `args.*` config expression. +/// +/// The common case reports `{ node_id, location, expression }` — a wiring +/// mistake the agent should fix. But when the null-resolved expression binds to +/// the output of an upstream Composio `tool_call` node +/// ([`ops::composio_tool_call_upstream_ref`]), the entry is instead marked +/// `unverifiable: true` and carries an honest `suggestion`: the echo sandbox +/// can NEVER produce a Composio tool's real output fields, so this particular +/// null is expected here and does NOT prove the binding wrong (WS6 — the +/// transcript audit where the agent re-wired an already-correct binding three +/// times chasing this exact false negative). The message points at +/// `get_tool_contract` / `get_tool_output_sample` as the real disambiguators. +fn build_null_resolution_entry( + node_id: &str, + diag: &tinyflows::expr::NullResolution, + graph: &WorkflowGraph, +) -> Value { + if let Some(upstream) = crate::openhuman::flows::ops::composio_tool_call_upstream_ref( + &diag.expression, + graph, + node_id, + ) { + let field = diag.location.strip_prefix("args.").unwrap_or("args"); + return json!({ + "node_id": node_id, + "location": diag.location, + "expression": diag.expression, + "unverifiable": true, + "upstream_tool_call": upstream, + "suggestion": format!( + "required arg `{field}` binds to the output of Composio tool_call node \ + `{upstream}` — the SANDBOX only echoes tool calls and can never produce \ + their real output fields, so this binding is UNVERIFIABLE here (not \ + necessarily wrong). Confirm the path against get_tool_contract {{ slug }}'s \ + output_fields / primary_array_path (remember Composio results nest under \ + `.item.json.data.`), or get_tool_output_sample {{ slug, args }} for the \ + real shape. It is a real bug only if the path doesn't match the action's \ + actual output." + ), + }); + } + json!({ + "node_id": node_id, + "location": diag.location, + "expression": diag.expression, + }) +} + +/// Every null-resolved `args.*` config expression that landed on a `tool_call` +/// node, as `null_resolutions` diagnostic entries (see +/// [`build_null_resolution_entry`] for the shape, including the WS6 +/// `unverifiable` Composio-upstream variant). Shared by the settled-run path +/// (which fails the dry run on these) and the errored-run path (which surfaces +/// only the `unverifiable` ones so a stop-policy preflight abort explains +/// itself honestly instead of via the generic required-arg text). +fn tool_call_arg_null_entries( + steps: &[tinyflows::observability::ExecutionStep], + graph: &WorkflowGraph, + tool_call_node_ids: &std::collections::HashSet<&str>, +) -> Vec { + steps + .iter() + .filter(|step| tool_call_node_ids.contains(step.node_id.as_str())) + .flat_map(|step| { + step.diagnostics + .iter() + .filter(|&diag| diag.location == "args" || diag.location.starts_with("args.")) + .map(|diag| build_null_resolution_entry(&step.node_id, diag, graph)) + }) + .collect() +} + pub struct DryRunWorkflowTool { security: Arc, config: Arc, @@ -2188,13 +2633,15 @@ impl Tool for DryRunWorkflowTool { } fn description(&self) -> &str { - "Dry-run a DRAFT workflow graph in a SANDBOX to self-verify it before \ + "Dry-run a workflow graph in a SANDBOX to self-verify it before \ proposing. Compiles the graph and executes it against MOCK capabilities \ — every LLM / tool_call / http_request / code node returns a deterministic \ echo, so NOTHING real happens (no messages sent, no code run). Returns the \ simulated per-node output labeled as sandbox output. Use it to catch \ - wiring/routing mistakes; it does NOT prove real integrations work. Pass \ - the same graph shape as propose_workflow, plus an optional `input`." + wiring/routing mistakes; it does NOT prove real integrations work. Provide \ + the graph as exactly one of `draft_id` (a working draft), `flow_id` (a saved \ + flow), or inline `graph` (draft_id wins, then flow_id), plus an optional \ + `input`." } fn parameters_schema(&self) -> Value { @@ -2203,11 +2650,15 @@ impl Tool for DryRunWorkflowTool { "properties": { "draft_id": { "type": "string", - "description": "A working draft to simulate. Provide this OR `graph`." + "description": "A working draft to simulate. Provide one of draft_id / flow_id / graph (draft_id wins)." + }, + "flow_id": { + "type": "string", + "description": "A saved flow to simulate. Provide one of draft_id / flow_id / graph." }, "graph": { "type": "object", - "description": "The DRAFT tinyflows WorkflowGraph to simulate: { nodes: [...], edges: [...] }. Provide this OR `draft_id`.", + "description": "An inline tinyflows WorkflowGraph to simulate: { nodes: [...], edges: [...] }. Provide one of draft_id / flow_id / graph.", "properties": { "nodes": { "type": "array" }, "edges": { "type": "array" } @@ -2235,31 +2686,48 @@ impl Tool for DryRunWorkflowTool { } async fn execute(&self, args: Value) -> anyhow::Result { - // Graph source: a working draft (draft_id) or an inline graph. - let graph_json = if let Some(draft_id) = args + // Graph source: exactly one of a working draft, a saved flow, or an + // inline graph — same precedence (draft_id > flow_id > graph) as the + // sibling validate/edit tools, so they all accept the same handles. + let draft_id = args .get("draft_id") .and_then(Value::as_str) .map(str::trim) - .filter(|s| !s.is_empty()) - { - match ops::flows_draft_get(&self.config, draft_id) { + .filter(|s| !s.is_empty()); + let flow_id = args + .get("flow_id") + .and_then(Value::as_str) + .map(str::trim) + .filter(|s| !s.is_empty()); + let inline_graph = args.get("graph").filter(|v| !v.is_null()); + + let graph_json = match (draft_id, flow_id, inline_graph) { + (Some(id), _, _) => match ops::flows_draft_get(&self.config, id) { Ok(outcome) => outcome.value.graph, Err(e) => { return Ok(ToolResult::error(format!( - "Could not load draft '{draft_id}' to dry-run: {e}" + "Could not load draft '{id}' to dry-run: {e}" ))); } - } - } else { - match args.get("graph") { - Some(v) if !v.is_null() => v.clone(), - _ => { - return Ok(ToolResult::error( - "Provide `graph` (an inline graph) or `draft_id` (a working draft) to \ - dry-run." - .to_string(), - )); + }, + (None, Some(id), _) => match ops::load_flow_graph(&self.config, id) { + Ok(Some(graph)) => serde_json::to_value(&graph)?, + Ok(None) => { + return Ok(ToolResult::error(format!("flow '{id}' not found"))); + } + Err(e) => { + return Ok(ToolResult::error(format!( + "Could not load flow '{id}' to dry-run: {e}" + ))); } + }, + (None, None, Some(v)) => v.clone(), + (None, None, None) => { + return Ok(ToolResult::error( + "Provide one of `draft_id` (a working draft), `flow_id` (a saved flow), or \ + `graph` (an inline graph) to dry-run." + .to_string(), + )); } }; let input = args.get("input").cloned().unwrap_or_else(|| json!({})); @@ -2299,6 +2767,12 @@ impl Tool for DryRunWorkflowTool { let mut caps = tinyflows::caps::mock::mock_capabilities_with_agent( crate::openhuman::tinyflows::caps::SchemaAwareMockAgentRunner, ); + // Plain agent nodes (no `agent_ref`) never reach the runner above — + // the vendored `agent` node routes them to the `llm` slot instead (see + // `SchemaAwareMockLlm`'s doc). Swap the vendored `MockLlm` echo for the + // schema-aware mock so their `output_parser.schema` is honored too, + // instead of the echo shape failing the sub-port's validation. + caps.llm = std::sync::Arc::new(crate::openhuman::tinyflows::caps::SchemaAwareMockLlm); // 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 { @@ -2344,6 +2818,46 @@ impl Tool for DryRunWorkflowTool { { Ok(Ok(outcome)) => outcome, Ok(Err(e)) => { + // A `stop`-policy `tool_call` whose required arg resolved null + // aborts the WHOLE run here (via `PreflightToolInvoker`), so + // the honest per-field diagnostic never reaches the settled-run + // `null_resolutions` path below. Recover it from the observer: + // if the abort was caused by a required arg bound to an upstream + // Composio `tool_call`'s output, the echo mock simply CAN'T + // produce that field — so surface it as `unverifiable` rather + // than letting the generic "required arg missing/null" text + // (which sent the transcript agent re-wiring a correct binding + // three times) stand alone. WS6. + let unverifiable_bindings: Vec = + tool_call_arg_null_entries(&observer.steps(), &graph, &tool_call_node_ids) + .into_iter() + .filter(|entry| { + entry.get("unverifiable").and_then(Value::as_bool) == Some(true) + }) + .collect(); + if !unverifiable_bindings.is_empty() { + tracing::debug!( + target: "flows", + error = %e, + unverifiable_count = unverifiable_bindings.len(), + "[flows] dry_run_workflow: sandbox run aborted on a Composio-upstream \ + binding the echo mock cannot verify — surfacing it honestly" + ); + return Ok(ToolResult::success(serde_json::to_string_pretty(&json!({ + "sandbox": true, + "ok": false, + "error": e.to_string(), + "unverifiable_bindings": unverifiable_bindings, + "note": "SANDBOX (mock) output — a tool_call node aborted because a \ + required arg binds to the output of an upstream Composio tool_call, \ + which the sandbox can only ECHO (it never produces real tool output \ + fields). See unverifiable_bindings: each MAY already be wired \ + correctly — confirm the path with get_tool_contract {{ slug }} \ + (output_fields / primary_array_path; Composio results nest under \ + .item.json.data.) or get_tool_output_sample {{ slug, args }} instead \ + of re-wiring blindly. No real side effects occurred.", + }))?)); + } tracing::debug!(target: "flows", error = %e, "[flows] dry_run_workflow: sandbox run errored"); return Ok(ToolResult::success(serde_json::to_string_pretty(&json!({ "sandbox": true, @@ -2363,23 +2877,12 @@ impl Tool for DryRunWorkflowTool { // `tool_call` node's `args.*` config path — the class of binding // mistake that "builds" (compiles, dry-runs against echo mocks) but // does nothing at runtime because the wired field never had a value. - let null_resolutions: Vec = observer - .steps() - .iter() - .filter(|step| tool_call_node_ids.contains(step.node_id.as_str())) - .flat_map(|step| { - step.diagnostics - .iter() - .filter(|&diag| diag.location == "args" || diag.location.starts_with("args.")) - .map(|diag| { - json!({ - "node_id": step.node_id, - "location": diag.location, - "expression": diag.expression, - }) - }) - }) - .collect(); + // Each entry is honest about WHY it resolved null: a binding to an + // upstream Composio `tool_call`'s output is flagged `unverifiable` + // (the echo mock can't produce real tool output fields) rather than + // reported as a plain wiring mistake — see [`build_null_resolution_entry`]. + let null_resolutions: Vec = + tool_call_arg_null_entries(&observer.steps(), &graph, &tool_call_node_ids); // Collect every null-resolved `agent`-node `prompt` — execution- // breaking in the same way a null `tool_call` arg is: `prompt` is the @@ -2723,12 +3226,16 @@ impl Tool for SaveWorkflowTool { fn description(&self) -> &str { "Save a workflow graph onto an EXISTING saved flow (by `flow_id`), persisting it. \ - Use this after the user asked you to build/update a workflow and you have \ - dry-run-verified the graph: it validates and writes the graph (and optional new \ - `name`) to that flow. It can NOT create a new flow, and it never changes the \ - flow's enabled state or its approval gate. NOTE: if the flow is enabled and the \ - graph has a schedule/app_event trigger, saving arms it — it will start firing on \ - its own. Always tell the user what you saved. Params: { flow_id, graph, name? }." + This is the ONLY builder tool that writes onto a saved flow — edit/validate/dry_run \ + never do. Use it after the user asked you to build/update a workflow and you have \ + dry-run-verified the graph. The graph source is either `draft_id` (a working draft — \ + the usual case after editing with edit_workflow; draft_id wins if both are given) or \ + an inline `graph`; `flow_id` is always required as the persistence TARGET. It \ + validates and writes the graph (and optional new `name`) to that flow. It can NOT \ + create a new flow, and it never changes the flow's enabled state or its approval \ + gate. NOTE: if the flow is enabled and the graph has a schedule/app_event trigger, \ + saving arms it — it will start firing on its own. Always tell the user what you \ + saved. Params: { flow_id, draft_id? | graph?, name? }." } fn parameters_schema(&self) -> Value { @@ -2737,11 +3244,15 @@ impl Tool for SaveWorkflowTool { "properties": { "flow_id": { "type": "string", - "description": "Id of the EXISTING saved flow to write the graph to." + "description": "Id of the EXISTING saved flow to write the graph to (the persistence target — always required)." + }, + "draft_id": { + "type": "string", + "description": "A working draft whose graph to persist onto the flow. Provide this OR inline `graph`; if both are given, draft_id wins." }, "graph": { "type": "object", - "description": "The full tinyflows WorkflowGraph to persist: { name?, nodes: [...], edges: [...] }. Same shape as propose_workflow.", + "description": "The full tinyflows WorkflowGraph to persist: { name?, nodes: [...], edges: [...] }. Provide this OR `draft_id`. Same shape as propose_workflow.", "properties": { "nodes": { "type": "array" }, "edges": { "type": "array" } @@ -2753,7 +3264,7 @@ impl Tool for SaveWorkflowTool { "description": "Optional new human-readable name for the flow." } }, - "required": ["flow_id", "graph"], + "required": ["flow_id"], "additionalProperties": false }) } @@ -2781,10 +3292,36 @@ impl Tool for SaveWorkflowTool { )) } }; - let graph_json = match args.get("graph") { - Some(v) if !v.is_null() => v.clone(), - _ => return Ok(ToolResult::error("Missing 'graph' parameter".to_string())), - }; + // Graph source: a working draft (the usual post-edit_workflow handle) or + // an inline graph. `flow_id` above is the persistence TARGET, always + // required; the draft only supplies the graph to write. If both a + // draft_id and an inline graph are given, the draft wins (it is the + // durable working copy the agent just iterated on). + let draft_id = args + .get("draft_id") + .and_then(Value::as_str) + .map(str::trim) + .filter(|s| !s.is_empty()); + let graph_json = + if let Some(id) = draft_id { + match ops::flows_draft_get(&self.config, id) { + Ok(outcome) => outcome.value.graph, + Err(e) => { + return Ok(ToolResult::error(format!( + "Could not load draft '{id}' to save: {e}" + ))); + } + } + } else { + match args.get("graph") { + Some(v) if !v.is_null() => v.clone(), + _ => return Ok(ToolResult::error( + "Provide `draft_id` (a working draft) or inline `graph` to save onto the \ + flow." + .to_string(), + )), + } + }; let name = args .get("name") .and_then(Value::as_str) @@ -2878,6 +3415,9 @@ impl Tool for SaveWorkflowTool { } Ok(ToolResult::success(serde_json::to_string_pretty(&json!({ "type": "workflow_saved", + // Explicit counterpart to a proposal's persisted:false — this + // graph IS now written onto the saved flow. + "persisted": true, "flow_id": flow.id, "name": flow.name, "enabled": flow.enabled, diff --git a/src/openhuman/flows/builder_tools_tests.rs b/src/openhuman/flows/builder_tools_tests.rs index fc207c73fe..e6a4d59dd1 100644 --- a/src/openhuman/flows/builder_tools_tests.rs +++ b/src/openhuman/flows/builder_tools_tests.rs @@ -211,6 +211,26 @@ fn seeded_gmail_send_contract() -> ToolContract { } } +/// A minimal seeded contract with NO required args, for WS6 dry-run tests: seeds +/// a bespoke toolkit so the required-arg preflight always passes and the sandbox +/// run settles into the `null_resolutions` path (rather than aborting), letting +/// the test assert the honest Composio-upstream diagnostic deterministically — +/// independent of whatever gmail/slack contracts other tests seed into the +/// process-global cache. +fn seeded_ws6_contract(slug: &str, toolkit: &str) -> ToolContract { + ToolContract { + slug: slug.to_string(), + toolkit: toolkit.to_string(), + description: Some("ws6 test action".to_string()), + required_args: vec![], + input_schema: Some(json!({ "type": "object", "additionalProperties": true })), + output_fields: vec![], + output_schema: None, + primary_array_path: None, + is_curated: true, + } +} + #[tokio::test] async fn search_live_catalog_finds_a_seeded_real_gmail_slug() { seed_live_catalog_cache("gmail", vec![seeded_gmail_send_contract()]); @@ -403,6 +423,279 @@ async fn get_tool_contract_rejects_a_hallucinated_slug() { assert!(result.output().contains("not a real action")); } +// ── WS3: early runtime-gate warnings on uncurated actions ──────────────────── +// +// Transcript failure #2: `get_tool_contract { slug: "TWITTER_USER_LOOKUP_ME" }` +// returned `is_curated: false` with no other signal; the agent built and wired +// the node and only ~15 tool calls later did `validate_workflow` reject it. A +// real-but-uncurated action of a toolkit that ships a curated catalog is a hard +// curated-only allowlist at RUNTIME, so surface the blocker at contract-fetch / +// search time. Uses `spotify` / `telegram` (real curated toolkits unused by +// other tests) so these seeds can't race with the shared `gmail`/`slack` keys. + +fn spotify_curated_action() -> ToolContract { + ToolContract { + slug: "SPOTIFY_START_PLAYBACK".to_string(), + toolkit: "spotify".to_string(), + description: Some("Start playback".to_string()), + required_args: vec![], + input_schema: Some(json!({ "type": "object" })), + output_fields: vec![], + output_schema: None, + primary_array_path: None, + is_curated: true, + } +} + +#[tokio::test] +async fn get_tool_contract_warns_on_an_uncurated_action_of_a_curated_toolkit() { + let uncurated = ToolContract { + slug: "SPOTIFY_OBSCURE_ACTION".to_string(), + is_curated: false, + ..spotify_curated_action() + }; + seed_live_catalog_cache("spotify", vec![spotify_curated_action(), uncurated]); + let tmp = TempDir::new().unwrap(); + let tool = GetToolContractTool::new(test_config(&tmp)); + + // Uncurated action → runtime_gate present, FIRST in the payload, contract intact. + let result = tool + .execute(json!({ "slug": "SPOTIFY_OBSCURE_ACTION" })) + .await + .unwrap(); + assert!(!result.is_error, "{}", result.output()); + let out = result.output(); + assert!(out.contains("runtime_gate"), "{out}"); + assert!(out.contains("REJECTED on every real run"), "{out}"); + let gate_pos = out.find("runtime_gate").expect("runtime_gate key"); + let slug_pos = out.find("\"slug\"").expect("slug key"); + assert!( + gate_pos < slug_pos, + "runtime_gate must serialize first (agents read top-down): {out}" + ); + let parsed: Value = serde_json::from_str(&out).unwrap(); + assert_eq!(parsed["slug"], "SPOTIFY_OBSCURE_ACTION"); + assert_eq!(parsed["is_curated"], false); + + // Curated action of the same toolkit → NO runtime_gate. + let result = tool + .execute(json!({ "slug": "SPOTIFY_START_PLAYBACK" })) + .await + .unwrap(); + assert!(!result.is_error, "{}", result.output()); + assert!( + !result.output().contains("runtime_gate"), + "{}", + result.output() + ); +} + +#[tokio::test] +async fn search_tool_catalog_flags_runtime_gated_uncurated_rows() { + let curated = ToolContract { + slug: "TELEGRAM_SEND_MESSAGE".to_string(), + toolkit: "telegram".to_string(), + description: Some("Send a message".to_string()), + required_args: vec![], + input_schema: None, + output_fields: vec![], + output_schema: None, + primary_array_path: None, + is_curated: true, + }; + let uncurated = ToolContract { + slug: "TELEGRAM_OBSCURE_SEND".to_string(), + is_curated: false, + ..curated.clone() + }; + seed_live_catalog_cache("telegram", vec![curated, uncurated]); + + let config = Config::default(); + let results = search_live_catalog(&config, "send", Some("telegram"), 40).await; + assert_eq!(results.len(), 2, "{results:?}"); + // Curated row: no `runtime_gated` key (only present when true). + let curated_row = results.iter().find(|r| r["featured"] == true).unwrap(); + assert!(curated_row.get("runtime_gated").is_none(), "{curated_row}"); + // Uncurated row of a curated toolkit: `runtime_gated: true`. + let uncurated_row = results.iter().find(|r| r["featured"] == false).unwrap(); + assert_eq!(uncurated_row["runtime_gated"], true); +} + +// ── WS5: per-token fallback ranking for zero-result multi-word queries ─────── +// +// Transcript failure: `search_tool_catalog` behaved like near-exact matching — +// multi-word natural-language queries ("twitter tweet replies lookup") returned +// `count: 0` even though the toolkit HAS matching actions, so the agent falsely +// concluded the action didn't exist. The primary pass is a strict case- +// insensitive AND (every token must match); when that misses for a multi-word +// query, a per-keyword OR fallback now returns the nearest matches + a note. + +fn twt_lookup() -> ToolContract { + ToolContract { + slug: "TWTFALLBACKTEST_TWEET_LOOKUP".to_string(), + toolkit: "twtfallbacktest".to_string(), + description: Some("Look up a tweet".to_string()), + required_args: vec!["id".to_string()], + input_schema: None, + output_fields: vec!["text".to_string()], + output_schema: None, + primary_array_path: None, + is_curated: true, + } +} + +fn twt_replies() -> ToolContract { + ToolContract { + slug: "TWTFALLBACKTEST_LIST_REPLIES".to_string(), + toolkit: "twtfallbacktest".to_string(), + description: Some("List replies to a tweet".to_string()), + required_args: vec!["tweet_id".to_string()], + input_schema: None, + output_fields: vec!["replies".to_string()], + output_schema: None, + primary_array_path: None, + is_curated: true, + } +} + +#[tokio::test] +async fn search_catalog_multiword_miss_falls_back_to_per_keyword() { + seed_live_catalog_cache("twtfallbacktest", vec![twt_lookup(), twt_replies()]); + let config = Config::default(); + // Strict AND misses ("twitter"/"timeline" match nothing) but individual + // tokens ("tweet", "replies", "lookup") hit — so the fallback fires. + let outcome = search_catalog( + &config, + "twitter tweet replies lookup timeline", + Some("twtfallbacktest"), + 40, + ) + .await; + assert!( + outcome.fallback, + "multi-word AND-miss must run the fallback" + ); + assert_eq!(outcome.results.len(), 2, "{:?}", outcome.results); + let note = outcome.note.expect("fallback carries an advisory note"); + assert!( + note.contains("nearest per-keyword"), + "note should explain the near-miss + single-keyword retry: {note}" + ); + // Fallback rows carry the SAME shape as primary rows. + for r in &outcome.results { + assert_eq!(r["toolkit"], "twtfallbacktest"); + assert_eq!(r["featured"], true); + assert!(r["required_args"].is_array()); + } +} + +#[tokio::test] +async fn search_tool_catalog_tool_surfaces_fallback_note_with_nonzero_count() { + seed_live_catalog_cache("twtfallbacktest", vec![twt_lookup(), twt_replies()]); + let tmp = TempDir::new().unwrap(); + let tool = SearchToolCatalogTool::new(test_config(&tmp)); + let result = tool + .execute(json!({ + "query": "twitter tweet replies lookup timeline", + "toolkit": "twtfallbacktest" + })) + .await + .unwrap(); + assert!(!result.is_error, "{}", result.output()); + let parsed: Value = serde_json::from_str(&result.output()).unwrap(); + // `count` reflects the returned rows (non-zero) so an agent never reads a + // fallback as "no such action". + assert_eq!(parsed["count"], 2); + assert!(parsed["results"].as_array().unwrap().len() == 2); + assert!(parsed["note"].as_str().unwrap().contains("No exact match")); +} + +#[tokio::test] +async fn search_catalog_single_word_behavior_unchanged() { + seed_live_catalog_cache("onewordtest", vec![twt_lookup()]); + let config = Config::default(); + // A hit: single-word query returns the primary match, no fallback, no note. + let hit = search_catalog(&config, "tweet", Some("onewordtest"), 40).await; + assert!(!hit.fallback); + assert!(hit.note.is_none()); + assert_eq!(hit.results.len(), 1); + // A miss: single-word query stays empty and does NOT run the fallback. + let miss = search_catalog(&config, "zzznomatchzzz", Some("onewordtest"), 40).await; + assert!( + !miss.fallback, + "single-token miss must not trigger fallback" + ); + assert!(miss.results.is_empty()); +} + +#[tokio::test] +async fn search_catalog_multiword_zero_token_match_returns_note() { + seed_live_catalog_cache("zerotoktest", vec![twt_lookup()]); + let config = Config::default(); + // Multi-word query where NO token matches anything: still a note (not a bare + // count: 0), but zero rows. + let outcome = search_catalog(&config, "qqq www eeeeee", Some("zerotoktest"), 40).await; + assert!(outcome.fallback, "multi-word miss ran the fallback pass"); + assert!(outcome.results.is_empty()); + let note = outcome + .note + .expect("zero-token multi-word miss still gets a note"); + assert!( + note.contains("keyword-based"), + "note should explain the keyword-based search: {note}" + ); +} + +#[tokio::test] +async fn search_catalog_fallback_rows_flag_runtime_gated() { + // Reuse the exact telegram seed of the runtime_gated primary test so a + // concurrent run over the shared cache stays self-consistent; telegram is a + // real curated toolkit, so its uncurated action is `runtime_gated`. + let curated = ToolContract { + slug: "TELEGRAM_SEND_MESSAGE".to_string(), + toolkit: "telegram".to_string(), + description: Some("Send a message".to_string()), + required_args: vec![], + input_schema: None, + output_fields: vec![], + output_schema: None, + primary_array_path: None, + is_curated: true, + }; + let uncurated = ToolContract { + slug: "TELEGRAM_OBSCURE_SEND".to_string(), + is_curated: false, + ..curated.clone() + }; + seed_live_catalog_cache("telegram", vec![curated, uncurated]); + + let config = Config::default(); + // "obscure" hits only the uncurated slug; "lookup"/"replies" hit nothing; + // "telegram" matches the toolkit of both — so strict AND misses and the + // fallback ranks the OBSCURE row first (2 hits) over SEND_MESSAGE (1 hit). + let outcome = search_catalog( + &config, + "telegram obscure lookup replies", + Some("telegram"), + 40, + ) + .await; + assert!(outcome.fallback); + assert_eq!(outcome.results.len(), 2, "{:?}", outcome.results); + let gated = outcome + .results + .iter() + .find(|r| r["featured"] == false) + .expect("uncurated row present"); + assert_eq!(gated["runtime_gated"], true); + let curated_row = outcome + .results + .iter() + .find(|r| r["featured"] == true) + .expect("curated row present"); + assert!(curated_row.get("runtime_gated").is_none()); +} + /// B12: a cached real-output probe overrides `get_tool_contract`'s /// schema-derived `primary_array_path`/`output_fields` — most relevant for a /// slug whose live listing (like every GitHub action, verified live) has NO @@ -587,6 +880,78 @@ async fn dry_run_exercises_agent_ref_node_via_mock_agent_runner() { ); } +#[tokio::test] +async fn dry_run_plain_agent_with_output_parser_schema_is_green() { + // Regression for the transcript false-failure: a builder-generated `agent` + // node carries NO `agent_ref`, so the vendored engine routes it to the + // `llm` slot (not the `AgentRunner`). Before `SchemaAwareMockLlm` the plain + // `MockLlm` echo (`{ completion, connection }`) failed the node's + // `output_parser.schema` sub-port with `output_parser: value failed schema + // validation after auto-fix: missing required property ...`, sinking a + // correctly-built graph. Now the mock LLM synthesizes a schema-valid object, + // and a downstream node binds the typed placeholders (non-null). + let tool = DryRunWorkflowTool::new( + policy(AutonomyLevel::Supervised), + test_config(&TempDir::new().unwrap()), + ); + let graph = json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Schedule", + "config": { "trigger_kind": "schedule" } }, + { "id": "a", "kind": "agent", "name": "Extract", + "config": { "prompt": "extract the fields", + "output_parser": { "schema": { "type": "object", + "required": ["subject", "priority", "recipients"], + "properties": { + "subject": { "type": "string" }, + "priority": { "type": "integer" }, + "recipients": { "type": "array" } + } } } } }, + // Downstream node binds the schema'd agent fields: proves the + // placeholders are addressable and resolve to typed (non-null) + // values, not the vendored echo's opaque `{ completion, ... }`. + { "id": "down", "kind": "transform", "name": "Route", + "config": { "set": { + "subject": "=nodes.a.item.json.subject", + "priority": "=nodes.a.item.json.priority", + "recipients": "=nodes.a.item.json.recipients" } } } + ], + "edges": [ + { "from_node": "t", "to_node": "a" }, + { "from_node": "a", "to_node": "down" } + ] + }); + let result = tool + .execute(json!({ "graph": graph, "input": { "topic": "launch" } })) + .await + .unwrap(); + assert!(!result.is_error, "{}", result.output()); + let out = result.output(); + assert!( + !out.to_lowercase().contains("schema validation"), + "plain agent with a valid schema must not hit the output_parser failure: {out}" + ); + let parsed: Value = serde_json::from_str(&out).unwrap(); + assert_eq!(parsed["sandbox"], true); + assert_eq!( + parsed["ok"], true, + "plain-agent-with-schema dry-run must be green: {parsed}" + ); + // The agent envelope's `json` carries the schema-synthesized placeholders. + // (In the run OUTPUT each Item serializes as `{ json: }`, and the + // agent's value is the `{json,text,raw}` envelope — hence the double hop.) + let agent_json = &parsed["output"]["nodes"]["a"]["items"][0]["json"]["json"]; + assert_eq!(agent_json["subject"], "", "{parsed}"); + assert_eq!(agent_json["priority"], 0, "{parsed}"); + assert_eq!(agent_json["recipients"], json!([]), "{parsed}"); + // The downstream node's bindings resolved to those typed placeholders — + // none of them null. + let down_json = &parsed["output"]["nodes"]["down"]["items"][0]["json"]; + assert!(!down_json["subject"].is_null(), "{parsed}"); + assert_eq!(down_json["priority"], 0, "{parsed}"); + assert_eq!(down_json["recipients"], json!([]), "{parsed}"); +} + #[tokio::test] async fn dry_run_invalid_graph_is_error() { let tool = DryRunWorkflowTool::new( @@ -711,6 +1076,112 @@ async fn dry_run_flags_tool_call_arg_null_resolved_from_unschemad_agent() { ); } +#[tokio::test] +async fn dry_run_flags_composio_upstream_binding_as_unverifiable_not_a_wiring_bug() { + // WS6: `post`'s `body` binds to the OUTPUT of an upstream Composio + // `tool_call` (`get_me`). The echo sandbox renders `get_me` as + // `{tool, args, connection}` and can NEVER produce `.item.json.data.username`, + // so the binding resolves `null` here even when it's wired correctly. The + // dry run still fails (`ok: false` — a null could hide a typo), but the + // diagnostic must be HONEST: mark it `unverifiable` and point at + // get_tool_contract / get_tool_output_sample rather than telling the agent + // its (possibly-correct) wiring is broken — the exact false negative that + // sent the transcript agent re-wiring an already-correct binding 3 times. + // Seed bespoke toolkits (no other test touches `ws6up`/`ws6dl`) with NO + // required args, so the required-arg preflight passes and the run settles + // into the `null_resolutions` path deterministically — independent of the + // process-global catalog cache other tests seed for gmail/slack/etc. + seed_live_catalog_cache("ws6up", vec![seeded_ws6_contract("WS6UP_LOOKUP", "ws6up")]); + seed_live_catalog_cache("ws6dl", vec![seeded_ws6_contract("WS6DL_SEND", "ws6dl")]); + let tool = DryRunWorkflowTool::new( + policy(AutonomyLevel::Supervised), + test_config(&TempDir::new().unwrap()), + ); + let graph = json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Manual" }, + { "id": "get_me", "kind": "tool_call", "name": "Who am I", + "config": { "slug": "WS6UP_LOOKUP", "args": {} } }, + { "id": "post", "kind": "tool_call", "name": "Post", + "config": { "slug": "WS6DL_SEND", + "args": { "recipient_email": "a@b.com", "subject": "hi", + "body": "=nodes.get_me.item.json.data.username" } } } + ], + "edges": [ + { "from_node": "t", "to_node": "get_me" }, + { "from_node": "get_me", "to_node": "post" } + ] + }); + let result = tool.execute(json!({ "graph": graph })).await.unwrap(); + assert!(!result.is_error, "{}", result.output()); + let parsed: Value = serde_json::from_str(&result.output()).unwrap(); + assert_eq!(parsed["ok"], false, "{parsed}"); + let null_resolutions = parsed["null_resolutions"] + .as_array() + .expect("null_resolutions array"); + let entry = null_resolutions + .iter() + .find(|e| e["node_id"] == "post" && e["location"] == "args.body") + .unwrap_or_else(|| panic!("expected a post.body null resolution: {parsed}")); + assert_eq!(entry["unverifiable"], true, "{parsed}"); + assert_eq!(entry["upstream_tool_call"], "get_me", "{parsed}"); + let suggestion = entry["suggestion"].as_str().expect("suggestion string"); + assert!(suggestion.contains("UNVERIFIABLE"), "{suggestion}"); + assert!(suggestion.contains("get_tool_contract"), "{suggestion}"); + assert!( + suggestion.contains("get_tool_output_sample"), + "{suggestion}" + ); +} + +#[tokio::test] +async fn dry_run_keeps_generic_null_text_for_a_non_tool_call_upstream_binding() { + // WS6 contrast: `post`'s arg binds to a `transform` node's output (whose + // real output the echo sandbox DOES produce), and the transform never sets + // the referenced field, so the null IS a genuine wiring bug. This entry must + // stay the plain `{ node_id, location, expression }` shape — no + // `unverifiable` flag — so the honest-uncertainty treatment doesn't leak + // onto real mistakes. + seed_live_catalog_cache("ws6dl", vec![seeded_ws6_contract("WS6DL_SEND", "ws6dl")]); + let tool = DryRunWorkflowTool::new( + policy(AutonomyLevel::Supervised), + test_config(&TempDir::new().unwrap()), + ); + let graph = json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Manual" }, + { "id": "build", "kind": "transform", "name": "Build", + "config": { "set": { "unrelated": "x" } } }, + { "id": "post", "kind": "tool_call", "name": "Post", + "config": { "slug": "WS6DL_SEND", + "args": { "recipient_email": "a@b.com", "subject": "hi", + "body": "=nodes.build.item.json.missing" } } } + ], + "edges": [ + { "from_node": "t", "to_node": "build" }, + { "from_node": "build", "to_node": "post" } + ] + }); + let result = tool.execute(json!({ "graph": graph })).await.unwrap(); + assert!(!result.is_error, "{}", result.output()); + let parsed: Value = serde_json::from_str(&result.output()).unwrap(); + assert_eq!(parsed["ok"], false, "{parsed}"); + let entry = parsed["null_resolutions"] + .as_array() + .expect("null_resolutions array") + .iter() + .find(|e| e["node_id"] == "post" && e["location"] == "args.body") + .unwrap_or_else(|| panic!("expected a post.body null resolution: {parsed}")); + assert!( + entry.get("unverifiable").is_none(), + "a non-tool_call upstream must keep the generic diagnostic: {parsed}" + ); + assert!( + entry.get("suggestion").is_none(), + "generic entry carries no unverifiable suggestion: {parsed}" + ); +} + #[tokio::test] async fn dry_run_passes_when_agent_schema_matches_tool_call_binding() { // The FALSE-POSITIVE-PREVENTION case: `summarize` DOES declare a schema @@ -1501,6 +1972,98 @@ async fn edit_workflow_reports_failing_op_with_guidance() { assert!(out.contains("edit_workflow again"), "{out}"); } +#[tokio::test] +async fn edit_workflow_bad_op_reports_index_type_and_shape() { + let tmp = TempDir::new().unwrap(); + let tool = EditWorkflowTool::new(test_config(&tmp)); + // ops 0 and 1 are well-formed; op 2 is an add_node missing its `node`. + let result = tool + .execute(json!({ + "graph": valid_graph(), + "ops": [ + { "op": "set_node_name", "id": "a", "name": "One" }, + { "op": "set_node_name", "id": "a", "name": "Two" }, + { "op": "add_node", "id": "b" } + ] + })) + .await + .unwrap(); + assert!(result.is_error, "{}", result.output()); + let out = result.output(); + // Names the failing op index, its op type, and the expected shape for it. + assert!(out.contains("op 2"), "{out}"); + assert!(out.contains("add_node"), "{out}"); + assert!(out.contains("node:"), "expected add_node shape in: {out}"); + assert!(out.contains("edit_workflow again"), "{out}"); +} + +#[tokio::test] +async fn edit_workflow_missing_op_field_lists_valid_types() { + let tmp = TempDir::new().unwrap(); + let tool = EditWorkflowTool::new(test_config(&tmp)); + let result = tool + .execute(json!({ + "graph": valid_graph(), + "ops": [ { "id": "a", "name": "No op tag" } ] + })) + .await + .unwrap(); + assert!(result.is_error, "{}", result.output()); + let out = result.output(); + assert!(out.contains("op 0"), "{out}"); + assert!(out.contains("missing `op` field"), "{out}"); + assert!(out.contains("update_node_config"), "{out}"); +} + +#[tokio::test] +async fn edit_workflow_add_node_exists_carries_ordering_hint() { + let tmp = TempDir::new().unwrap(); + let tool = EditWorkflowTool::new(test_config(&tmp)); + // Re-adding an existing node id fails in-order; the hint should point at the + // remove-first / patch-in-place fix. + let result = tool + .execute(json!({ + "graph": valid_graph(), + "ops": [ + { "op": "add_node", "node": { "id": "a", "kind": "merge", "name": "Dup" } } + ] + })) + .await + .unwrap(); + assert!(result.is_error, "{}", result.output()); + let out = result.output(); + assert!(out.contains("already exists"), "{out}"); + assert!(out.contains("array order"), "{out}"); + assert!(out.contains("remove_node"), "{out}"); + assert!(out.contains("update_node_config"), "{out}"); +} + +#[tokio::test] +async fn edit_workflow_accepts_node_id_aliases_end_to_end() { + let tmp = TempDir::new().unwrap(); + let tool = EditWorkflowTool::new(test_config(&tmp)); + // A valid ops array using the `node_id` alias (the natural agent guess) + // applies cleanly through edit_workflow. + let result = tool + .execute(json!({ + "graph": valid_graph(), + "name": "Aliased edit", + "ops": [ + { "op": "update_node_config", "node_id": "a", "config": { "prompt": "aliased" } }, + { "op": "set_node_name", "node_id": "a", "name": "Aliased step" } + ] + })) + .await + .unwrap(); + assert!(!result.is_error, "{}", result.output()); + let parsed: Value = serde_json::from_str(&result.output()).unwrap(); + assert_eq!(parsed["type"], "workflow_proposal"); + let nodes = parsed["graph"]["nodes"].as_array().unwrap(); + let agent = nodes.iter().find(|n| n["id"] == "a").unwrap(); + assert_eq!(agent["config"]["prompt"], "aliased"); + assert_eq!(agent["name"], "Aliased step"); +} + #[tokio::test] async fn edit_workflow_rejects_a_result_that_is_structurally_invalid() { let tmp = TempDir::new().unwrap(); @@ -1724,3 +2287,147 @@ fn phase4_write_tools_have_the_right_permissions() { PermissionLevel::None ); } + +// ── WS2: unified draft_id|flow_id|graph handles + explicit persistence state ── + +#[tokio::test] +async fn edit_workflow_by_flow_id_seeds_a_retrievable_draft_and_marks_unpersisted() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + // A saved flow to edit — editing it must NOT write onto the flow (the WS2 + // bug: a flow_id edit used to persist nothing and return no handle). + let flow = ops::flows_create(&config, "Base flow".to_string(), valid_graph(), false) + .await + .unwrap() + .value; + + let tool = EditWorkflowTool::new(config.clone()); + let result = tool + .execute(json!({ + "flow_id": flow.id, + "ops": [ { "op": "set_node_name", "id": "a", "name": "Renamed step" } ] + })) + .await + .unwrap(); + assert!(!result.is_error, "{}", result.output()); + let parsed: Value = serde_json::from_str(&result.output()).unwrap(); + + // The edit lives on a NEW draft, is explicitly NOT persisted, and echoes the + // flow it derives from plus a `next` hint naming the draft. + assert_eq!(parsed["persisted"], false); + assert_eq!(parsed["flow_id"], flow.id.as_str()); + let draft_id = parsed["draft_id"] + .as_str() + .expect("edit_workflow by flow_id returns a draft_id") + .to_string(); + assert!(parsed["next"].as_str().unwrap().contains(&draft_id)); + + // The draft is retrievable via ops::flows_draft_get and holds the EDITED + // graph, linked back to the source flow. + let draft = ops::flows_draft_get(&config, &draft_id).unwrap().value; + assert_eq!(draft.flow_id.as_deref(), Some(flow.id.as_str())); + let agent = draft.graph["nodes"] + .as_array() + .unwrap() + .iter() + .find(|n| n["id"] == "a") + .unwrap(); + assert_eq!(agent["name"], "Renamed step"); + + // The SAVED flow is untouched — the whole point of WS2. + let saved = ops::flows_get(&config, &flow.id).await.unwrap().value; + let saved_graph = serde_json::to_value(&saved.graph).unwrap(); + let saved_agent = saved_graph["nodes"] + .as_array() + .unwrap() + .iter() + .find(|n| n["id"] == "a") + .unwrap(); + assert_eq!( + saved_agent["name"], "Summarize", + "the flow must not be edited" + ); +} + +#[tokio::test] +async fn dry_run_workflow_by_flow_id_runs_the_saved_flow_graph() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow = ops::flows_create(&config, "Runnable".to_string(), valid_graph(), false) + .await + .unwrap() + .value; + let tool = DryRunWorkflowTool::new(policy(AutonomyLevel::Supervised), config.clone()); + let result = tool.execute(json!({ "flow_id": flow.id })).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); +} + +#[tokio::test] +async fn validate_workflow_by_draft_id_checks_the_draft_graph() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let draft = ops::flows_draft_create( + &config, + None, + "Draft".to_string(), + valid_graph(), + crate::openhuman::flows::DraftOrigin::Chat, + ) + .unwrap() + .value; + let tool = ValidateWorkflowTool::new(config.clone()); + let result = tool.execute(json!({ "draft_id": draft.id })).await.unwrap(); + assert!(!result.is_error, "{}", result.output()); + let parsed: Value = serde_json::from_str(&result.output()).unwrap(); + assert_eq!(parsed["ok"], true); + assert_eq!(parsed["structurally_valid"], true); +} + +#[tokio::test] +async fn save_workflow_by_draft_id_persists_the_draft_graph_onto_the_flow() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + // A flow seeded with a bare 1-node graph. + let flow_id = seed_flow(&config, "Blank flow").await; + // A draft holding the richer 2-node valid graph, linked to that flow. + let draft = ops::flows_draft_create( + &config, + Some(flow_id.clone()), + "Draft".to_string(), + valid_graph(), + crate::openhuman::flows::DraftOrigin::Chat, + ) + .unwrap() + .value; + + let tool = SaveWorkflowTool::new(config.clone()); + let result = tool + .execute(json!({ "flow_id": flow_id, "draft_id": draft.id })) + .await + .unwrap(); + assert!(!result.is_error, "{}", result.output()); + let parsed: Value = serde_json::from_str(&result.output()).unwrap(); + assert_eq!(parsed["type"], "workflow_saved"); + assert_eq!(parsed["persisted"], true); + assert_eq!(parsed["node_count"], 2); + + // The draft's graph really landed on the flow. + let saved = ops::flows_get(&config, &flow_id).await.unwrap().value; + assert_eq!(saved.graph.nodes.len(), 2); +} + +#[tokio::test] +async fn revise_workflow_proposal_is_marked_unpersisted() { + let tmp = TempDir::new().unwrap(); + let tool = ReviseWorkflowTool::new(test_config(&tmp)); + let result = tool + .execute(json!({ "name": "R", "graph": valid_graph() })) + .await + .unwrap(); + assert!(!result.is_error, "{}", result.output()); + let parsed: Value = serde_json::from_str(&result.output()).unwrap(); + assert_eq!(parsed["persisted"], false); +} diff --git a/src/openhuman/flows/ops.rs b/src/openhuman/flows/ops.rs index 44b73438a9..4f398e72c0 100644 --- a/src/openhuman/flows/ops.rs +++ b/src/openhuman/flows/ops.rs @@ -142,6 +142,15 @@ pub(crate) async fn run_builder_gates(config: &Config, graph: &WorkflowGraph) -> if !binding_errors.is_empty() { return binding_errors; } + // Async, live connection list: a tool_call whose `connection_ref` names the + // wrong toolkit for its slug, or a connection id the user doesn't actually + // have (WS3 — the transcript bug where a TIKTOK connection id was wired onto + // Twitter/Gmail nodes and every author-time gate returned ok). Cheap: + // one connection-list fetch, no per-node catalog round trips. + let connection_ref_errors = validate_connection_refs(config, graph).await; + if !connection_ref_errors.is_empty() { + return connection_ref_errors; + } // Async, live catalog: a tool_call whose slug isn't a real Composio action // or whose real required args aren't all wired. let contract_errors = validate_tool_contracts(config, graph).await; @@ -192,11 +201,20 @@ pub(crate) async fn strict_gate(config: &Config, graph_json: &Value) -> Result<( /// the tool in the "fix … and call `` again" guidance so each caller's /// error text points the agent back at the right tool. /// +/// `draft_id` / `flow_id` are OPTIONAL persistence-state context echoed onto +/// the payload (the draft this proposal's edit lives on, and the saved flow it +/// derives from / targets). The payload ALWAYS carries `"persisted": false` so +/// a proposal can never be mistaken for a save confirmation — the exact false +/// belief the WS2 audit caught (an agent read a proposal as "written onto the +/// saved flow"). Actual persistence only happens via `save_workflow` / +/// `create_workflow` / `flows_draft_promote`. +/// /// Returns `Ok(payload)` on success, or `Err(message)` with a /// model-consumable, fix-and-retry error when a gate rejects the graph. The /// caller is responsible for structural validation (`validate_and_migrate_graph` /// / `validate_all`) *before* calling this — these gates assume a compilable /// graph. +#[allow(clippy::too_many_arguments)] pub(crate) async fn build_builder_proposal( config: &Config, retry_tool: &str, @@ -205,6 +223,8 @@ pub(crate) async fn build_builder_proposal( require_approval: bool, revision: bool, instruction: Option, + draft_id: Option, + flow_id: Option, ) -> Result { // The full builder hard-gate stack, run through the single canonical // runner so every proposal/save/strict-RPC path gates identically (F3). @@ -238,6 +258,10 @@ pub(crate) async fn build_builder_proposal( let mut payload = json!({ "type": "workflow_proposal", "revision": revision, + // A proposal is NEVER a persisted flow — it is a candidate the user + // still has to accept/save. Stamp this unconditionally so the payload + // can't be misread as a save confirmation (WS2 audit). + "persisted": false, "name": name, "graph": graph_value, "require_approval": require_approval, @@ -248,6 +272,14 @@ pub(crate) async fn build_builder_proposal( if let Some(instruction) = instruction { payload["instruction"] = json!(instruction); } + // Echo the persistence-state handles so the agent can iterate/persist + // against the right ids (the draft the edit lives on; the flow it targets). + if let Some(draft_id) = draft_id { + payload["draft_id"] = json!(draft_id); + } + if let Some(flow_id) = flow_id { + payload["flow_id"] = json!(flow_id); + } Ok(payload) } @@ -1102,10 +1134,23 @@ pub(crate) fn validate_binding_resolvability(graph: &WorkflowGraph) -> Vec bool { + use crate::openhuman::memory_sync::composio::providers::{catalog_for_toolkit, get_provider}; + get_provider(toolkit) + .and_then(|p| p.curated_tools()) + .or_else(|| catalog_for_toolkit(toolkit)) + .is_some() +} + pub(crate) async fn validate_tool_contracts(config: &Config, graph: &WorkflowGraph) -> Vec { - use crate::openhuman::memory_sync::composio::providers::{ - catalog_for_toolkit, get_provider, toolkit_from_slug, - }; + use crate::openhuman::memory_sync::composio::providers::toolkit_from_slug; use crate::openhuman::tinyflows::caps::{ fetch_live_toolkit_catalog, missing_required_args, unsupported_arg_names, }; @@ -1166,10 +1211,7 @@ pub(crate) async fn validate_tool_contracts(config: &Config, graph: &WorkflowGra // action on a curated toolkit and then fail every run with "tool // not permitted". Hold authoring to the same bar the runtime gate // enforces instead of loosening the runtime gate. - let has_static_catalog = get_provider(&toolkit) - .and_then(|p| p.curated_tools()) - .or_else(|| catalog_for_toolkit(&toolkit)) - .is_some(); + let has_static_catalog = toolkit_has_curated_catalog(&toolkit); if has_static_catalog && !contract.is_curated { tracing::warn!( target: "flows", @@ -1273,6 +1315,248 @@ pub(crate) async fn validate_tool_contracts(config: &Config, graph: &WorkflowGra errors } +// ───────────────────────────────────────────────────────────────────────────── +// Connection-ref gate (WS3): a Composio tool_call's `connection_ref` must name +// a real connected account of the RIGHT toolkit +// ───────────────────────────────────────────────────────────────────────────── +// +// Transcript audit: the user's connections were `twitter → +// composio:twitter:ca_JX6QU88UfSk4`, `gmail → composio:gmail:ca_vX_WA8FsqNmE`, +// `tiktok → composio:tiktok:ca_LPCp3WQpaDma`. The agent wired +// `composio:twitter:ca_LPCp3WQpaDma` and `composio:gmail:ca_LPCp3WQpaDma` (the +// TIKTOK id) onto the Twitter and Gmail tool_call nodes. dry_run / validate / +// propose all returned ok:true — nothing cross-checked the id against the user's +// real connections, nor the ref's toolkit segment against the slug — and it +// would fail on the first real run. This gate closes that gap: it parses the +// ref, enforces the toolkit segment matches the slug (needs no I/O), and — when +// the live connection list is reachable — that the id names a real connected +// account of that toolkit, naming the correct ref when it can. + +/// Parses a `composio::` connection_ref into its `(toolkit, id)` +/// segments. Mirrors [`crate::openhuman::tinyflows::caps::composio_connection_id`]'s +/// rsplit for the id (everything after the LAST `:`), taking everything between +/// the `composio:` prefix and that last `:` as the toolkit. Returns `None` for +/// anything that isn't this shape (missing `composio:` prefix, no `:` after it, +/// or an empty toolkit/id segment). +fn parse_composio_connection_ref(conn_ref: &str) -> Option<(&str, &str)> { + let rest = conn_ref.strip_prefix("composio:")?; + let (toolkit, id) = rest.rsplit_once(':')?; + if toolkit.trim().is_empty() || id.trim().is_empty() { + return None; + } + Some((toolkit.trim(), id.trim())) +} + +/// First connected account `connection_ref` for `toolkit` (case-insensitive) +/// from `conns`, used to name the correct ref in a rejection's "did you mean" +/// hint. `None` when the toolkit has no connection at all. +fn first_connection_ref_for_toolkit(conns: &[FlowConnection], toolkit: &str) -> Option { + conns + .iter() + .find(|c| { + c.toolkit + .as_deref() + .is_some_and(|t| t.eq_ignore_ascii_case(toolkit)) + }) + .map(|c| c.connection_ref.clone()) +} + +/// Hard gate: for every Composio `tool_call` node carrying a `connection_ref`, +/// prove the ref names a real connected account of the SAME toolkit as the +/// slug. Fetches the live connection list once (same source +/// [`flows_list_connections`] reads) and delegates the pure matching to +/// [`validate_connection_refs_against`]. +/// +/// Fail-open on I/O: if the Composio connection list is unreachable (backend +/// outage), the id-existence check is SKIPPED (a `tracing::debug!` records it) +/// so a real connection is never false-rejected during an outage — but the +/// toolkit-mismatch check, which needs no I/O, still runs. +pub(crate) async fn validate_connection_refs( + config: &Config, + graph: &WorkflowGraph, +) -> Vec { + let connections: Option> = + match crate::openhuman::composio::ops::composio_list_connections(config).await { + Ok(outcome) => Some(build_flow_connections( + outcome.value.connections, + Vec::new(), + )), + Err(e) => { + tracing::debug!( + target: "flows", + error = %e, + "[flows] connection-ref check: composio connection list unavailable — \ + skipping id-existence check (fail-open); toolkit-mismatch check still runs" + ); + None + } + }; + validate_connection_refs_against(graph, connections.as_deref()) +} + +/// Pure connection-ref validator (no I/O) so the gate's decision logic is +/// unit-testable without a live Composio backend. `connections` is `Some(list)` +/// when the live connection list was fetched (possibly empty — a genuine "no +/// connections" state), or `None` when it was unavailable (fail-open: the +/// id-existence check is skipped, only the toolkit-mismatch check runs). +fn validate_connection_refs_against( + graph: &WorkflowGraph, + connections: Option<&[FlowConnection]>, +) -> Vec { + use crate::openhuman::memory_sync::composio::providers::toolkit_from_slug; + + let mut errors = Vec::new(); + for node in &graph.nodes { + if node.kind != NodeKind::ToolCall { + continue; + } + let Some(slug) = node.config.get("slug").and_then(Value::as_str) else { + continue; + }; + // `=`-derived slugs resolve at runtime; native `oh:` tools have no + // Composio connection to name. + if slug.starts_with('=') || slug.starts_with("oh:") { + continue; + } + // A MISSING `connection_ref` stays allowed (unchanged): a Composio + // tool_call with no ref runs against the ambient signed-in account and + // the flow prompts for a connection at first run. + let Some(conn_ref) = node.config.get("connection_ref").and_then(Value::as_str) else { + continue; + }; + if conn_ref.trim().is_empty() { + continue; + } + let Some(slug_toolkit) = toolkit_from_slug(slug) else { + continue; + }; + + let Some((ref_toolkit, ref_id)) = parse_composio_connection_ref(conn_ref) else { + tracing::debug!( + target: "flows", + node = %node.id, + %slug, + toolkit = %slug_toolkit, + %conn_ref, + matched = false, + "[flows] connection-ref check: malformed ref — rejecting" + ); + errors.push(format!( + "Node '{}': `connection_ref` `{conn_ref}` is malformed — a Composio account ref \ + must look like `composio::` (e.g. \ + `composio:{slug_toolkit}:`). Call list_flow_connections and copy a \ + `connection_ref` value verbatim.", + node.id + )); + continue; + }; + + // Toolkit segment vs the slug's toolkit — needs no I/O. + if !ref_toolkit.eq_ignore_ascii_case(&slug_toolkit) { + let suggestion = connections + .and_then(|conns| first_connection_ref_for_toolkit(conns, &slug_toolkit)); + tracing::debug!( + target: "flows", + node = %node.id, + %slug, + toolkit = %slug_toolkit, + %ref_toolkit, + %ref_id, + matched = false, + "[flows] connection-ref check: toolkit segment does not match the slug's toolkit — rejecting" + ); + let hint = match suggestion { + Some(r) => format!(" — did you mean `{r}`?"), + None => format!( + " — no `{slug_toolkit}` account is connected; connect one with \ + composio_connect (or ask the user to), then use its `connection_ref`" + ), + }; + errors.push(format!( + "Node '{}': `connection_ref` `{conn_ref}` names the `{ref_toolkit}` toolkit but the \ + tool_call slug `{slug}` is a `{slug_toolkit}` action{hint}.", + node.id + )); + continue; + } + + // Existence check: the id must name a real connected account of this + // toolkit. Skipped (fail-open) when the connection list is unavailable. + let Some(conns) = connections else { + tracing::debug!( + target: "flows", + node = %node.id, + %slug, + toolkit = %slug_toolkit, + %ref_id, + "[flows] connection-ref check: toolkit matches; id-existence check skipped (connections unavailable)" + ); + continue; + }; + // The id must belong to a connection OF THIS TOOLKIT — not merely + // exist somewhere. The transcript bug was a real TIKTOK connection id + // stamped onto a `composio:twitter:` ref: the id exists globally, but + // it is not a Twitter account, so it must still be rejected. + let id_exists = conns.iter().any(|c| { + c.toolkit + .as_deref() + .is_some_and(|t| t.eq_ignore_ascii_case(&slug_toolkit)) + && parse_composio_connection_ref(&c.connection_ref) + .is_some_and(|(_, cid)| cid.eq_ignore_ascii_case(ref_id)) + }); + if id_exists { + tracing::debug!( + target: "flows", + node = %node.id, + %slug, + toolkit = %slug_toolkit, + %ref_id, + matched = true, + "[flows] connection-ref check: ref resolves to a real connected account — ok" + ); + continue; + } + // Unknown id. Name the right ref for this toolkit if one exists. + match first_connection_ref_for_toolkit(conns, &slug_toolkit) { + Some(r) => { + tracing::debug!( + target: "flows", + node = %node.id, + %slug, + toolkit = %slug_toolkit, + %ref_id, + matched = false, + "[flows] connection-ref check: unknown id; toolkit has a different connected account — rejecting" + ); + errors.push(format!( + "Node '{}': `connection_ref` `{conn_ref}` does not match any connected \ + `{slug_toolkit}` account — did you mean `{r}`? Call list_flow_connections and \ + copy a `connection_ref` value verbatim.", + node.id + )); + } + None => { + tracing::debug!( + target: "flows", + node = %node.id, + %slug, + toolkit = %slug_toolkit, + %ref_id, + matched = false, + "[flows] connection-ref check: no connected account for this toolkit — rejecting" + ); + errors.push(format!( + "Node '{}': `connection_ref` `{conn_ref}` names a `{slug_toolkit}` account, but \ + no `{slug_toolkit}` account is connected — connect one with composio_connect \ + (or ask the user to), then use its `connection_ref`.", + node.id + )); + } + } + } + errors +} + // ───────────────────────────────────────────────────────────────────────────── // Required-arg resolvability gate (issue B18) // ───────────────────────────────────────────────────────────────────────────── @@ -1348,13 +1632,19 @@ const REQUIRED_ARG_NULL_CHECK_TIMEOUT_SECS: u64 = 15; /// check only ever adds a diagnostic the sandbox actually observed. pub(crate) async fn validate_required_arg_resolvability(graph: &WorkflowGraph) -> Vec { use crate::openhuman::flows::builder_tools::CapturingObserver; - use crate::openhuman::tinyflows::caps::SchemaAwareMockAgentRunner; + use crate::openhuman::tinyflows::caps::{SchemaAwareMockAgentRunner, SchemaAwareMockLlm}; let Ok(compiled) = tinyflows::compiler::compile(graph) else { return Vec::new(); }; - let caps = tinyflows::caps::mock::mock_capabilities_with_agent(SchemaAwareMockAgentRunner); + let mut caps = tinyflows::caps::mock::mock_capabilities_with_agent(SchemaAwareMockAgentRunner); + // Same fix as `DryRunWorkflowTool`: a plain agent node (no `agent_ref`) + // routes to the `llm` slot, not the runner above, so the vendored `MockLlm` + // echo would fail its `output_parser.schema` sub-port and make this gate + // reject a correct graph (which is why `propose_workflow` was rejecting + // valid graphs). The schema-aware mock LLM honors the schema instead. + caps.llm = Arc::new(SchemaAwareMockLlm); let observer = Arc::new(CapturingObserver::default()); let observer_dyn: Arc = observer.clone(); @@ -1426,6 +1716,34 @@ pub(crate) async fn validate_required_arg_resolvability(graph: &WorkflowGraph) - ); continue; } + // A null bound to the OUTPUT of an upstream Composio `tool_call` + // node is UNVERIFIABLE in this echo sandbox — the mock renders a + // Composio `tool_call` as `{tool, args, connection}` and can NEVER + // produce its real output fields (`.item.json.data.`), so a + // downstream binding to one resolves `null` here even when the + // wiring is perfectly correct. Hard-rejecting it (WS6) would block + // a possibly-correct graph from ever being proposed — the exact + // false-negative the transcript audit caught. Downgrade to a + // debug-logged skip; `dry_run_workflow` remains the surface that + // reports it (as an `unverifiable` diagnostic the agent can act on + // via get_tool_contract / get_tool_output_sample). + if let Some(upstream) = + composio_tool_call_upstream_ref(&diag.expression, graph, &step.node_id) + { + tracing::debug!( + target: "flows", + node = %step.node_id, + %slug, + %field, + upstream = %upstream, + expression = %diag.expression, + "[flows] required-arg resolvability check: arg binds to a Composio \ + tool_call's output — UNVERIFIABLE in the echo sandbox (the mock cannot \ + produce real tool output fields), not rejecting; dry_run_workflow \ + reports it instead" + ); + continue; + } tracing::warn!( target: "flows", node = %step.node_id, @@ -1541,6 +1859,72 @@ fn is_trigger_scoped_expression( predecessors.peek().is_some() && predecessors.all(|e| e.from_node == trigger_id) } +/// If a null-resolved config expression on `node_id` is bound to the OUTPUT of +/// an upstream **Composio `tool_call`** node (a `tool_call` whose `slug` is a +/// real Composio action — not `=`-derived, not native `oh:`), returns that +/// upstream node's id; otherwise `None`. +/// +/// The dry-run / gate sandbox renders a Composio `tool_call` as a deterministic +/// echo (`{tool, args, connection}`) and can NEVER produce its real output +/// fields, so a downstream binding to `.item.json.data.` off such a node +/// resolves `null` in the sandbox **even when the wiring is correct** — the +/// binding is UNVERIFIABLE here, not necessarily broken. Callers use this to +/// tell that honest-uncertainty case apart from a genuinely broken binding +/// (one wired to an `agent` / `transform` / `code` / trigger upstream, whose +/// real output the sandbox DOES produce, so a null there IS a real bug). +/// +/// Handles both addressing forms the engine can trace: +/// - explicit `=nodes....` / `=.nodes[""]...` (parsed via +/// [`explicit_nodes_ref`]), and +/// - implicit `=item...` / `=items...`, resolved against `node_id`'s direct +/// predecessor — but only when there is exactly ONE incoming edge, so an +/// ambiguous fan-in is never mis-attributed to a single upstream node. +/// +/// Anything else (a `=run...` trigger reference, a jq expression not rooted at +/// one of the above, or a reference to a non-`tool_call` / native / dynamic +/// node) returns `None`. +pub(crate) fn composio_tool_call_upstream_ref<'a>( + expr: &str, + graph: &'a WorkflowGraph, + node_id: &str, +) -> Option<&'a str> { + let referenced_id: String = if let Some(id) = explicit_nodes_ref(expr) { + id.to_string() + } else { + let body = expr.strip_prefix('=').unwrap_or(expr).trim(); + let body = body.strip_prefix('.').unwrap_or(body); + let is_item_scoped = body == "item" + || body.starts_with("item.") + || body.starts_with("item[") + || body == "items" + || body.starts_with("items.") + || body.starts_with("items["); + if !is_item_scoped { + return None; + } + let mut preds = graph + .edges + .iter() + .filter(|e| e.to_node == node_id) + .map(|e| e.from_node.as_str()); + let first = preds.next()?; + if preds.next().is_some() { + // Ambiguous fan-in — cannot attribute the null to one upstream node. + return None; + } + first.to_string() + }; + let node = graph.nodes.iter().find(|n| n.id == referenced_id)?; + if node.kind != NodeKind::ToolCall { + return None; + } + let slug = node.config.get("slug").and_then(Value::as_str)?; + if slug.starts_with('=') || slug.starts_with("oh:") { + return None; + } + Some(node.id.as_str()) +} + /// 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 diff --git a/src/openhuman/flows/ops_tests.rs b/src/openhuman/flows/ops_tests.rs index 4349f0523f..27731b6374 100644 --- a/src/openhuman/flows/ops_tests.rs +++ b/src/openhuman/flows/ops_tests.rs @@ -2059,6 +2059,162 @@ async fn validate_tool_contracts_passes_a_fully_wired_real_slug() { assert!(errors.is_empty(), "{errors:?}"); } +// ── validate_connection_refs (WS3) ────────────────────────────────────────── +// +// The transcript bug: the user's connections were twitter → +// `composio:twitter:ca_JX6QU88UfSk4`, gmail → `composio:gmail:ca_vX_WA8FsqNmE`, +// tiktok → `composio:tiktok:ca_LPCp3WQpaDma`. The agent wired +// `composio:twitter:ca_LPCp3WQpaDma` (the TIKTOK id) onto a Twitter node and +// every author-time gate returned ok. These tests exercise the pure matcher so +// no live Composio backend is touched. + +/// Build a composio `FlowConnection` fixture (the exact shape +/// `build_flow_connections` produces). +fn ws3_flow_conn(toolkit: &str, id: &str) -> FlowConnection { + FlowConnection { + connection_ref: format!("composio:{toolkit}:{id}"), + kind: "composio".to_string(), + display: toolkit.to_string(), + toolkit: Some(toolkit.to_string()), + scheme: None, + } +} + +/// The user's real connected set from the transcript. +fn ws3_transcript_connections() -> Vec { + vec![ + ws3_flow_conn("twitter", "ca_JX6QU88UfSk4"), + ws3_flow_conn("gmail", "ca_vX_WA8FsqNmE"), + ws3_flow_conn("tiktok", "ca_LPCp3WQpaDma"), + ] +} + +/// A single tool_call node graph with `slug` + optional `connection_ref`. +fn ws3_tool_call_graph(slug: &str, connection_ref: Option<&str>) -> WorkflowGraph { + let mut config = json!({ "slug": slug, "args": {} }); + if let Some(cr) = connection_ref { + config["connection_ref"] = json!(cr); + } + graph(json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Manual" }, + { "id": "act", "kind": "tool_call", "name": "Act", "config": config } + ], + "edges": [ { "from_node": "t", "to_node": "act" } ] + })) +} + +#[test] +fn connection_refs_reject_the_transcript_wrong_id_naming_the_right_ref() { + // Twitter node carrying the TIKTOK connection id: toolkit segment matches + // (twitter == twitter) but the id belongs to no Twitter account. + let g = ws3_tool_call_graph( + "TWITTER_CREATION_OF_A_POST", + Some("composio:twitter:ca_LPCp3WQpaDma"), + ); + let conns = ws3_transcript_connections(); + let errors = validate_connection_refs_against(&g, Some(&conns)); + assert_eq!(errors.len(), 1, "{errors:?}"); + assert!(errors[0].contains("act"), "{}", errors[0]); + assert!( + errors[0].contains("composio:twitter:ca_JX6QU88UfSk4"), + "must name the correct ref verbatim: {}", + errors[0] + ); + assert!(errors[0].contains("did you mean"), "{}", errors[0]); +} + +#[test] +fn connection_refs_reject_a_toolkit_mismatch_naming_the_right_ref() { + // A literal `composio:tiktok:...` ref stamped onto a Twitter node. + let g = ws3_tool_call_graph( + "TWITTER_CREATION_OF_A_POST", + Some("composio:tiktok:ca_LPCp3WQpaDma"), + ); + let conns = ws3_transcript_connections(); + let errors = validate_connection_refs_against(&g, Some(&conns)); + assert_eq!(errors.len(), 1, "{errors:?}"); + assert!(errors[0].contains("tiktok"), "{}", errors[0]); + assert!( + errors[0].contains("composio:twitter:ca_JX6QU88UfSk4"), + "{}", + errors[0] + ); +} + +#[test] +fn connection_refs_reject_an_unknown_id_when_the_toolkit_has_no_connection() { + // Gmail slug, but no gmail account connected at all → point at composio_connect. + let g = ws3_tool_call_graph("GMAIL_SEND_EMAIL", Some("composio:gmail:ca_missing")); + let conns = vec![ws3_flow_conn("twitter", "ca_JX6QU88UfSk4")]; + let errors = validate_connection_refs_against(&g, Some(&conns)); + assert_eq!(errors.len(), 1, "{errors:?}"); + assert!(errors[0].contains("composio_connect"), "{}", errors[0]); + assert!(!errors[0].contains("did you mean"), "{}", errors[0]); +} + +#[test] +fn connection_refs_pass_the_correct_ref() { + let g = ws3_tool_call_graph( + "TWITTER_CREATION_OF_A_POST", + Some("composio:twitter:ca_JX6QU88UfSk4"), + ); + let conns = ws3_transcript_connections(); + let errors = validate_connection_refs_against(&g, Some(&conns)); + assert!(errors.is_empty(), "{errors:?}"); +} + +#[test] +fn connection_refs_reject_a_malformed_ref() { + let g = ws3_tool_call_graph("GMAIL_SEND_EMAIL", Some("gmail-ca_vX_WA8FsqNmE")); + let conns = ws3_transcript_connections(); + let errors = validate_connection_refs_against(&g, Some(&conns)); + assert_eq!(errors.len(), 1, "{errors:?}"); + assert!(errors[0].contains("malformed"), "{}", errors[0]); +} + +#[test] +fn connection_refs_skip_oh_and_refless_and_expression_nodes() { + // Native oh: tool with a ref → skipped. + let g_oh = ws3_tool_call_graph("oh:memory_search", Some("composio:twitter:whatever")); + assert!( + validate_connection_refs_against(&g_oh, Some(&ws3_transcript_connections())).is_empty() + ); + // Composio tool_call with NO connection_ref stays allowed (prompts at run). + let g_refless = ws3_tool_call_graph("TWITTER_CREATION_OF_A_POST", None); + assert!( + validate_connection_refs_against(&g_refless, Some(&ws3_transcript_connections())) + .is_empty() + ); + // `=`-derived slug → skipped. + let g_expr = ws3_tool_call_graph("=item.slug", Some("composio:twitter:ca_LPCp3WQpaDma")); + assert!( + validate_connection_refs_against(&g_expr, Some(&ws3_transcript_connections())).is_empty() + ); +} + +#[test] +fn connection_refs_fail_open_on_unavailable_connections_but_keep_mismatch() { + // Connections unavailable (None): the id-existence check is SKIPPED — a + // toolkit-matched ref with an unknown id passes rather than false-reject. + let g_ok = ws3_tool_call_graph( + "TWITTER_CREATION_OF_A_POST", + Some("composio:twitter:ca_anything"), + ); + assert!( + validate_connection_refs_against(&g_ok, None).is_empty(), + "unknown id must be skipped when connections are unavailable" + ); + // ...but the toolkit-mismatch check needs no I/O and still fires. + let g_mismatch = ws3_tool_call_graph( + "TWITTER_CREATION_OF_A_POST", + Some("composio:tiktok:ca_LPCp3WQpaDma"), + ); + let errors = validate_connection_refs_against(&g_mismatch, None); + assert_eq!(errors.len(), 1, "{errors:?}"); + assert!(errors[0].contains("tiktok"), "{}", errors[0]); +} + // ── validate_required_arg_resolvability (issue B18) ───────────────────────── // // `validate_tool_contracts`'s `missing_required_args` only proves an arg is @@ -2190,6 +2346,95 @@ async fn validate_required_arg_resolvability_rejects_an_explicit_nodes_reference assert!(errors[0].contains("nodes.build_body"), "{}", errors[0]); } +/// A required tool arg wired to a PLAIN agent node's (`no agent_ref`) +/// `output_parser.schema` field must pass this sandbox gate: the schema-aware +/// mock LLM (wired above via `caps.llm = SchemaAwareMockLlm`) synthesizes a +/// schema-valid completion, so the agent's output-parser sub-port succeeds and +/// the downstream `=nodes..item.json.` binding resolves to a typed +/// placeholder (non-null) instead of the run aborting on a schema-validation +/// failure. Without the mock LLM this gate would sink `propose_workflow`/`save` +/// on a correctly-built graph (the vendored `MockLlm` echo fails the sub-port). +#[tokio::test] +async fn validate_required_arg_resolvability_accepts_a_schema_agent_field_binding() { + let g = graph(json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Manual" }, + { "id": "summarize", "kind": "agent", "name": "Summarize", + "config": { "prompt": "summarize the thread", + "output_parser": { "schema": { "type": "object", + "required": ["channel"], + "properties": { "channel": { "type": "string" } } } } } }, + { "id": "post", "kind": "tool_call", "name": "Post", + "config": { "slug": "SLACK_SEND_MESSAGE", + "args": { "channel": "=nodes.summarize.item.json.channel" } } } + ], + "edges": [ + { "from_node": "t", "to_node": "summarize" }, + { "from_node": "summarize", "to_node": "post" } + ] + })); + let errors = validate_required_arg_resolvability(&g).await; + assert!(errors.is_empty(), "{errors:?}"); +} + +/// WS6: a required arg wired to the OUTPUT of an upstream Composio `tool_call` +/// must NOT be hard-rejected by this gate. The echo sandbox renders a Composio +/// `tool_call` as `{tool, args, connection}` and can never produce its real +/// output fields, so `=nodes..item.json.data.` resolves `null` +/// here even when the wiring is perfectly correct — rejecting it would block a +/// possibly-correct graph from ever being proposed (the transcript false +/// negative). Contrast `..._rejects_an_explicit_nodes_reference` above, where +/// the same explicit-`nodes` form addresses a `code` node (whose real output +/// the sandbox DOES produce) and stays a hard reject. +#[tokio::test] +async fn validate_required_arg_resolvability_downgrades_a_composio_tool_call_upstream_binding() { + let g = graph(json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Manual" }, + { "id": "get_me", "kind": "tool_call", "name": "Who am I", + "config": { "slug": "TWITTER_USER_LOOKUP_ME", "args": {} } }, + { "id": "post", "kind": "tool_call", "name": "Post", + "config": { "slug": "GMAIL_SEND_EMAIL", + "args": { "recipient_email": "a@b.com", "subject": "hi", + "body": "=nodes.get_me.item.json.data.username" } } } + ], + "edges": [ + { "from_node": "t", "to_node": "get_me" }, + { "from_node": "get_me", "to_node": "post" } + ] + })); + let errors = validate_required_arg_resolvability(&g).await; + assert!( + errors.is_empty(), + "a binding to a Composio tool_call's output is UNVERIFIABLE, not a hard reject: {errors:?}" + ); +} + +/// WS6 companion: the implicit `=item...` form of the same case — `post`'s only +/// predecessor is a Composio `tool_call`, so `=item.json.data.username` +/// addresses that node's (echo-only) output and is likewise unverifiable, not a +/// reject. +#[tokio::test] +async fn validate_required_arg_resolvability_downgrades_an_item_scoped_composio_upstream_binding() { + let g = graph(json!({ + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Manual" }, + { "id": "get_me", "kind": "tool_call", "name": "Who am I", + "config": { "slug": "TWITTER_USER_LOOKUP_ME", "args": {} } }, + { "id": "post", "kind": "tool_call", "name": "Post", + "config": { "slug": "GMAIL_SEND_EMAIL", + "args": { "recipient_email": "a@b.com", "subject": "hi", + "body": "=item.json.data.username" } } } + ], + "edges": [ + { "from_node": "t", "to_node": "get_me" }, + { "from_node": "get_me", "to_node": "post" } + ] + })); + let errors = validate_required_arg_resolvability(&g).await; + assert!(errors.is_empty(), "{errors:?}"); +} + /// (Codex feedback on this PR) `notion` ships a static curated catalog /// (`catalog_for_toolkit`), so at RUNTIME `flow_tool_allowed`'s Path A /// hard-rejects any slug `find_curated` doesn't recognize — even a real, diff --git a/src/openhuman/flows/schemas.rs b/src/openhuman/flows/schemas.rs index dac1f64167..feee1c1a7c 100644 --- a/src/openhuman/flows/schemas.rs +++ b/src/openhuman/flows/schemas.rs @@ -1694,6 +1694,7 @@ mod tests { "resume", "cancel_run", "list_runs", + "list_all_runs", "get_run", "prune_runs", "build", @@ -1719,7 +1720,7 @@ mod tests { #[test] fn all_registered_controllers_has_handler_per_schema() { let controllers = all_registered_controllers(); - assert_eq!(controllers.len(), 32); + assert_eq!(controllers.len(), 33); let names: Vec<_> = controllers.iter().map(|c| c.schema.function).collect(); assert_eq!( names, @@ -1738,6 +1739,7 @@ mod tests { "resume", "cancel_run", "list_runs", + "list_all_runs", "get_run", "prune_runs", "build", diff --git a/src/openhuman/flows/tools.rs b/src/openhuman/flows/tools.rs index 8a36cf5e99..20eb1278d8 100644 --- a/src/openhuman/flows/tools.rs +++ b/src/openhuman/flows/tools.rs @@ -261,6 +261,10 @@ impl Tool for ProposeWorkflowTool { 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, diff --git a/src/openhuman/flows/tools_tests.rs b/src/openhuman/flows/tools_tests.rs index 3ce228532e..59912f2fcf 100644 --- a/src/openhuman/flows/tools_tests.rs +++ b/src/openhuman/flows/tools_tests.rs @@ -58,6 +58,9 @@ async fn valid_graph_returns_workflow_proposal_success() { assert_eq!(parsed["type"], "workflow_proposal"); assert_eq!(parsed["name"], "Daily standup summary"); assert_eq!(parsed["graph"]["nodes"].as_array().unwrap().len(), 3); + // A proposal is never a persisted flow — the payload must say so (WS2) so + // an agent can't misread it as a save confirmation. + assert_eq!(parsed["persisted"], false); } #[tokio::test] diff --git a/src/openhuman/tinyflows/caps.rs b/src/openhuman/tinyflows/caps.rs index c3db0c133e..5f6424cc83 100644 --- a/src/openhuman/tinyflows/caps.rs +++ b/src/openhuman/tinyflows/caps.rs @@ -1165,6 +1165,71 @@ impl AgentRunner for SchemaAwareMockAgentRunner { } } +/// A **dry-run-only** [`LlmProvider`] mock that, unlike the vendored crate's +/// `tinyflows::caps::mock::MockLlm`, respects an `agent` node's +/// `config.output_parser.schema` when synthesizing its completion. +/// +/// This closes the OTHER half of the same gap [`SchemaAwareMockAgentRunner`] +/// closes. The vendored `agent` node only routes to an [`AgentRunner`] when the +/// node carries a **non-empty `agent_ref`** AND the host wired an agent registry +/// (`vendor/tinyflows/src/nodes/integration/agent.rs`, `run_turn`: +/// `(Some(agent_ref), Some(runner)) => runner.run_agent(...)`); **every other +/// case** — and builder-generated agent nodes carry NO `agent_ref` — falls back +/// to `ctx.caps.llm.complete(cfg.clone(), conn)`. So in the sandbox those plain +/// agent nodes never reach `SchemaAwareMockAgentRunner` at all: they hit the +/// `llm` slot, which (with the vendored `MockLlm`) echoes +/// `{ "completion": , "connection": }`. The agent node's +/// output-parser sub-port then validates that echo against the declared schema +/// (`schema::parse_and_validate` — it validates the WHOLE completion value, not +/// a `.text` field), no field matches, and it falls to a one-shot LLM auto-fix +/// that the same `MockLlm` also can't satisfy — so the dry run errors with +/// `output_parser: value failed schema validation after auto-fix: missing +/// required property ...` even for a workflow a real run would execute cleanly. +/// This false-failure burned many dry-run cycles for correctly-built graphs. +/// +/// When `request` (the node config the node hands to `complete` — see the +/// `_ => ctx.caps.llm.complete(cfg.clone(), conn)` arm above) carries a non-null +/// `output_parser.schema`, this returns [`placeholder_for_schema`] DIRECTLY. +/// The sub-port receives that already-schema-valid object as its `value` +/// (`validate` returns no errors), so it returns `Ok` WITHOUT ever invoking the +/// auto-fix LLM path — exactly the shape the vendored validator's +/// `type`/`required`/`enum` checks accept, with no real model call. With no +/// schema, it mirrors the vendored `MockLlm` echo shape byte-for-byte +/// (`{ "completion": request, "connection": conn }`) so schema-less agent +/// dry-run behavior — and downstream `=nodes..item.json.completion...` +/// bindings — stay identical to today. +#[derive(Debug, Default, Clone)] +pub struct SchemaAwareMockLlm; + +#[async_trait] +impl LlmProvider for SchemaAwareMockLlm { + async fn complete(&self, request: Value, conn: Option<&str>) -> Result { + let schema = request + .get("output_parser") + .and_then(|parser| parser.get("schema")) + .filter(|schema| !schema.is_null()); + match schema { + Some(schema) => { + let placeholder = placeholder_for_schema(schema); + tracing::debug!( + target: "flows", + "[flows] dry_run: schema-aware mock LLM synthesized a placeholder \ + matching output_parser.schema (plain agent node, no agent_ref)" + ); + Ok(placeholder) + } + None => { + tracing::debug!( + target: "flows", + "[flows] dry_run: schema-aware mock LLM has no output_parser.schema — \ + mirroring the vendored MockLlm echo shape" + ); + Ok(json!({ "completion": request, "connection": conn })) + } + } + } +} + /// Builds a placeholder JSON value satisfying `schema`'s `properties`/`type` /// constraints, for [`SchemaAwareMockAgentRunner`]. Only the shallow, top-level /// `properties` map is populated — enough for the minimal validator in @@ -3543,6 +3608,62 @@ mod tests { assert_eq!(out["request"], request); } + // ── SchemaAwareMockLlm ─────────────────────────────────────────────────── + + #[tokio::test] + async fn schema_aware_mock_llm_mirrors_vendored_echo_without_a_schema() { + // No `output_parser.schema`: byte-identical to the vendored `MockLlm` + // so schema-less agent dry runs (which route to the `llm` slot, not the + // runner) keep today's `{ completion, connection }` shape. + let llm = SchemaAwareMockLlm; + let request = json!({ "prompt": "hi" }); + let out = llm + .complete(request.clone(), Some("conn_1")) + .await + .expect("complete"); + assert_eq!(out["completion"], request); + assert_eq!(out["connection"], "conn_1"); + + let without_conn = llm.complete(request, None).await.expect("complete"); + assert!(without_conn["connection"].is_null()); + } + + #[tokio::test] + async fn schema_aware_mock_llm_synthesizes_a_schema_valid_completion() { + // A plain agent node (no `agent_ref`) hands its config to the `llm` + // slot; the returned object must pass the output-parser sub-port's + // validator directly (no auto-fix hop) for every declared type. + let llm = SchemaAwareMockLlm; + let request = json!({ + "prompt": "extract", + "output_parser": { "schema": { "type": "object", + "required": ["email", "count", "active", "meta", "tags"], + "properties": { + "email": { "type": "string" }, + "count": { "type": "integer" }, + "active": { "type": "boolean" }, + "meta": { "type": "object" }, + "tags": { "type": "array" } + } } } + }); + let out = llm.complete(request, None).await.expect("complete"); + assert_eq!(out["email"], ""); + assert_eq!(out["count"], 0); + assert_eq!(out["active"], false); + assert_eq!(out["meta"], json!({})); + assert_eq!(out["tags"], json!([])); + } + + #[tokio::test] + async fn schema_aware_mock_llm_ignores_null_schema() { + // `output_parser: { schema: null }` is treated as "no schema" — the + // vendored echo shape, same as the runner's null-schema handling. + let llm = SchemaAwareMockLlm; + let request = json!({ "prompt": "hi", "output_parser": { "schema": null } }); + let out = llm.complete(request.clone(), None).await.expect("complete"); + assert_eq!(out["completion"], request); + } + #[test] fn placeholder_for_schema_falls_back_to_type_without_properties() { assert_eq!( diff --git a/vendor/tinyflows b/vendor/tinyflows index f18238904f..e5327de9f6 160000 --- a/vendor/tinyflows +++ b/vendor/tinyflows @@ -1 +1 @@ -Subproject commit f18238904f48c395dce99c0c1d759c3dd1bfb678 +Subproject commit e5327de9f602c1fbbf72d45150729f45b91fa3a8