From 296b59ed2179a6fc20d9e3a67ce1ab98b20fad9c Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sat, 4 Jul 2026 22:51:18 -0700 Subject: [PATCH 1/8] feat(engine): observer-accepting durable+journaled run/resume variants Add `run_with_checkpointer_journaled_observed` and `resume_with_checkpointer_journaled_observed`, which thread a host-supplied `Arc` through the durable + journaled path instead of hard-coding `NoopObserver`. The private `build_and_run` already threaded an observer end to end; `resume_with_checkpointer_inner` now takes one too. The existing non-observed entry points delegate to the observed variants with a `NoopObserver`, so behavior is unchanged for current callers. This unblocks live per-step run observation in downstream hosts (OpenHuman issue G2): a host observer now sees `on_step_finish` for every non-trigger node as a durable checkpointed run executes. Bump 0.3.0 -> 0.3.1 (additive, non-breaking). Claude-Session: https://claude.ai/code/session_01QpsTTxxjFgDhzuozS28tyU --- Cargo.lock | 2 +- Cargo.toml | 2 +- src/engine.rs | 78 ++++++++++++++++++++++++++++++++++++++++++++++++--- 3 files changed, 76 insertions(+), 6 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 6009fa1..cea1b09 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1355,7 +1355,7 @@ dependencies = [ [[package]] name = "tinyflows" -version = "0.3.0" +version = "0.3.1" dependencies = [ "async-trait", "futures-timer", diff --git a/Cargo.toml b/Cargo.toml index 12b5531..9c19dae 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "tinyflows" -version = "0.3.0" +version = "0.3.1" edition = "2024" rust-version = "1.85" license = "GPL-3.0-or-later" diff --git a/src/engine.rs b/src/engine.rs index 80f788e..b7c2ff2 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -967,6 +967,7 @@ pub async fn resume_with_checkpointer( thread_id: &str, newly_approved: Vec, ) -> Result { + let observer = Arc::new(crate::observability::NoopObserver) as Arc; let (outcome, _run_ids) = resume_with_checkpointer_inner( workflow, capabilities, @@ -974,6 +975,7 @@ pub async fn resume_with_checkpointer( thread_id, newly_approved, None, + &observer, ) .await?; Ok(outcome) @@ -1001,11 +1003,46 @@ pub async fn run_with_checkpointer_journaled( journal: Arc, ) -> Result { let observer = Arc::new(crate::observability::NoopObserver) as Arc; - let (_graph, _thread_id, outcome, graph_run_ids) = build_and_run( + run_with_checkpointer_journaled_observed( workflow, input, capabilities, + checkpointer, + thread_id, + journal, &observer, + ) + .await +} + +/// Like [`run_with_checkpointer_journaled`], but additionally reports live +/// run/step records to the host-supplied `observer` as the run executes +/// ([`RunObserver::on_run_start`] once, [`RunObserver::on_step_finish`] per +/// non-trigger node as it finishes, [`RunObserver::on_run_finish`] once at +/// settle). This is the durable + journaled + observed entry point a host uses +/// when it wants **both** post-run journal export **and** live per-step +/// observation (e.g. incremental run-history persistence and a progress feed). +/// +/// The observer is held as `Arc` and cloned into each node +/// handler, which run across threads, so it must be cheap and non-blocking; see +/// [`RunObserver`]'s contract. +/// +/// # Errors +/// Same as [`run_with_checkpointer_journaled`]. +pub async fn run_with_checkpointer_journaled_observed( + workflow: &CompiledWorkflow, + input: Value, + capabilities: &Capabilities, + checkpointer: Arc>, + thread_id: &str, + journal: Arc, + observer: &Arc, +) -> Result { + let (_graph, _thread_id, outcome, graph_run_ids) = build_and_run( + workflow, + input, + capabilities, + observer, checkpointer, thread_id.to_string(), Some(journal), @@ -1038,6 +1075,36 @@ pub async fn resume_with_checkpointer_journaled( thread_id: &str, newly_approved: Vec, journal: Arc, +) -> Result { + let observer = Arc::new(crate::observability::NoopObserver) as Arc; + resume_with_checkpointer_journaled_observed( + workflow, + capabilities, + checkpointer, + thread_id, + newly_approved, + journal, + &observer, + ) + .await +} + +/// Like [`resume_with_checkpointer_journaled`], but additionally reports live +/// step records to the host-supplied `observer` as the resumed run executes +/// (each node that runs after the interrupt boundary fires +/// [`RunObserver::on_step_finish`]). The durable + journaled + observed resume +/// counterpart to [`run_with_checkpointer_journaled_observed`]. +/// +/// # Errors +/// Same as [`resume_with_checkpointer_journaled`]. +pub async fn resume_with_checkpointer_journaled_observed( + workflow: &CompiledWorkflow, + capabilities: &Capabilities, + checkpointer: Arc>, + thread_id: &str, + newly_approved: Vec, + journal: Arc, + observer: &Arc, ) -> Result { let (outcome, graph_run_ids) = resume_with_checkpointer_inner( workflow, @@ -1046,6 +1113,7 @@ pub async fn resume_with_checkpointer_journaled( thread_id, newly_approved, Some(journal), + observer, ) .await?; tracing::debug!( @@ -1070,16 +1138,18 @@ async fn resume_with_checkpointer_inner( thread_id: &str, newly_approved: Vec, journal: Option>, + observer: &Arc, ) -> Result<(RunOutcome, GraphRunIds)> { - let observer = Arc::new(crate::observability::NoopObserver) as Arc; let steps: Arc>> = Arc::new(Mutex::new(Vec::new())); // Rebuild the identical graph and re-attach the SAME checkpointer, so - // `resume` loads the state persisted under `thread_id`. + // `resume` loads the state persisted under `thread_id`. Node handlers fire + // `observer.on_step_finish` for every node that runs after the interrupt + // boundary, so a host observer sees the resumed steps live. let (compiled, _trigger_id) = build_graph( workflow, capabilities, - &observer, + observer, &steps, checkpointer, journal, From 96635d9812cff48bbe4244563a94eeb7bfe730da Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sat, 4 Jul 2026 23:00:44 -0700 Subject: [PATCH 2/8] =?UTF-8?q?feat(engine):=20deny-on-resume=20=E2=80=94?= =?UTF-8?q?=20route=20rejected=20approval=20gates=20to=20their=20error=20p?= =?UTF-8?q?ort?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extend the durable resume seam so a host can *deny* a human-in-the-loop approval gate, not just approve it. `resume_with_checkpointer_journaled_observed` gains a `rejected: Vec` parameter; denied gate ids are delivered to the interrupted node(s) via a structured resume value `{ "rejected": [...] }` (the bare `true` approval value is kept when nothing is denied, so the approve-only path is unchanged). In `build_graph`, a gate whose id is in the rejected set emits an error item on its `error` port when one is wired (so a recovery sub-graph runs) or fails the run when it has none. Bumps to 0.3.2. Claude-Session: https://claude.ai/code/session_01QpsTTxxjFgDhzuozS28tyU --- Cargo.lock | 2 +- Cargo.toml | 2 +- src/engine.rs | 185 +++++++++++++++++++++++++++++++++++++++++++++++++- 3 files changed, 185 insertions(+), 4 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index cea1b09..c84f40f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1355,7 +1355,7 @@ dependencies = [ [[package]] name = "tinyflows" -version = "0.3.1" +version = "0.3.2" dependencies = [ "async-trait", "futures-timer", diff --git a/Cargo.toml b/Cargo.toml index 9c19dae..d13c79e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "tinyflows" -version = "0.3.1" +version = "0.3.2" edition = "2024" rust-version = "1.85" license = "GPL-3.0-or-later" diff --git a/src/engine.rs b/src/engine.rs index b7c2ff2..3061c61 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -335,6 +335,13 @@ fn build_graph( let is_trigger = node.kind == NodeKind::Trigger; // Successors to fan out to concurrently, or `None` for a non-fan-out node. let fan_out = fan_out_targets(&node.id); + // Whether the node has an outgoing edge on the `error` port. A denied + // approval gate (see the resume-deny path below) routes its error item + // there when present, and fails the run when absent. + let has_error_edge = graph + .edges + .iter() + .any(|e| e.from_node == node.id && e.from_port == "error"); builder = builder.add_node(node.id.clone(), move |state: Value, ctx| { let node = node.clone(); @@ -343,6 +350,10 @@ fn build_graph( let observer = observer.clone(); let steps = steps.clone(); let fan_out = fan_out.clone(); + // The resume value delivered to this node on a checkpointed resume, if + // any. A bare `true` means "approve the interrupted gate"; a structured + // `{ "rejected": [, …] }` denies the named gate(s). + let resume_value = ctx.resume.clone(); // A checkpointed resume (see `ResumableRun::resume`) delivers a resume // value to the interrupted node via `NodeContext::resume`; its presence // approves this gate so the run proceeds forward from the checkpoint. @@ -376,6 +387,50 @@ fn build_graph( .and_then(Value::as_bool) .unwrap_or(false); if requires_approval { + // Human-in-the-loop **denial**. A resume delivered with a + // structured value `{ "rejected": [, …] }` (see + // `resume_with_checkpointer_journaled_observed`) denies the + // named gate rather than approving it: the gate emits an + // error item on its `error` port when one is wired (so a + // recovery branch can handle the rejection), or fails the run + // when it has no `error` port. Checked before the approval + // branch so a denial always wins over the bare-resume approval. + let denied = resume_value + .as_ref() + .and_then(|v| v.get("rejected")) + .and_then(Value::as_array) + .is_some_and(|rejected| { + rejected + .iter() + .filter_map(Value::as_str) + .any(|id| id == node.id) + }); + if denied { + tracing::info!(node = %node.id, has_error_edge, "approval gate denied"); + let item = Item::new(json!({ + "error": { + "message": "approval denied", + "node": node.id, + "denied": true, + } + })); + if has_error_edge { + // Route the denial to the `error` port so a recovery + // sub-graph runs. A plain update (not `emit`) so the + // conditional-edge router keys on the recorded port. + return Ok(NodeResult::Update(items_update( + &node.id, + &[item], + Some("error"), + )?)); + } + // No error branch to route to — fail the run so the denial + // is not silently swallowed. + return Err(TinyAgentsError::Graph(format!( + "approval gate '{}' was denied and has no `error` port to route to", + node.id + ))); + } let approved = state .get("run") .and_then(|run| run.get("trigger")) @@ -974,6 +1029,7 @@ pub async fn resume_with_checkpointer( checkpointer, thread_id, newly_approved, + Vec::new(), None, &observer, ) @@ -1083,6 +1139,7 @@ pub async fn resume_with_checkpointer_journaled( checkpointer, thread_id, newly_approved, + Vec::new(), journal, &observer, ) @@ -1095,6 +1152,12 @@ pub async fn resume_with_checkpointer_journaled( /// [`RunObserver::on_step_finish`]). The durable + journaled + observed resume /// counterpart to [`run_with_checkpointer_journaled_observed`]. /// +/// `newly_approved` gate ids proceed on resume; `rejected` gate ids are **denied** +/// — each denied gate routes an error item to its `error` port (when one is +/// wired) or fails the run (when it has none). Pass an empty `rejected` for the +/// approve-only path; the two sets should be disjoint (a gate is approved or +/// denied, not both). +/// /// # Errors /// Same as [`resume_with_checkpointer_journaled`]. pub async fn resume_with_checkpointer_journaled_observed( @@ -1103,6 +1166,7 @@ pub async fn resume_with_checkpointer_journaled_observed( checkpointer: Arc>, thread_id: &str, newly_approved: Vec, + rejected: Vec, journal: Arc, observer: &Arc, ) -> Result { @@ -1112,6 +1176,7 @@ pub async fn resume_with_checkpointer_journaled_observed( checkpointer, thread_id, newly_approved, + rejected, Some(journal), observer, ) @@ -1137,6 +1202,7 @@ async fn resume_with_checkpointer_inner( checkpointer: Arc>, thread_id: &str, newly_approved: Vec, + rejected: Vec, journal: Option>, observer: &Arc, ) -> Result<(RunOutcome, GraphRunIds)> { @@ -1161,12 +1227,22 @@ async fn resume_with_checkpointer_inner( // (tinyagents ignores it on resume, so the resume value is the real // approval channel). let approvals_update = json!({ - "run": { "trigger": { "approvals": newly_approved } } + "run": { "trigger": { "approvals": newly_approved.clone() } } }); + // The resume value delivered to the interrupted gate(s): a bare `true` + // (backward-compatible bare approval) when nothing was denied, otherwise a + // structured value carrying the denied gate ids so the gate handlers can + // route/fail them (see the `denied` branch in `build_graph`). + let resume_value = if rejected.is_empty() { + json!(true) + } else { + tracing::info!(?rejected, "resuming with denied approval gate(s)"); + json!({ "approved": newly_approved, "rejected": rejected }) + }; let execution = compiled .resume( thread_id, - Command::resume(json!(true)).with_update(approvals_update), + Command::resume(resume_value).with_update(approvals_update), ) .await .map_err(|e| EngineError::Capability(e.to_string()))?; @@ -2747,6 +2823,111 @@ mod tests { ); } + #[tokio::test] + async fn resume_denying_a_gate_routes_to_its_error_port() { + // trigger -> gate{requires_approval}; gate has BOTH a `main` edge (to + // `downstream`) and an `error` edge (to `recover`). Denying the gate on + // resume must route the error item to `recover` and leave `downstream` + // untouched. + let cp: Arc> = Arc::new(InMemoryCheckpointer::::default()); + let mut gate = node("gate", NodeKind::OutputParser); + gate.config = json!({ "requires_approval": true }); + let graph = WorkflowGraph { + nodes: vec![ + node("t", NodeKind::Trigger), + gate, + node("downstream", NodeKind::OutputParser), + node("recover", NodeKind::OutputParser), + ], + edges: vec![ + edge("t", "gate"), + port_edge("gate", "main", "downstream"), + port_edge("gate", "error", "recover"), + ], + ..Default::default() + }; + let compiled = compile(&graph).expect("compile"); + + let caps = mock_capabilities(); + let paused = run_with_checkpointer(&compiled, json!({}), &caps, cp.clone(), "thread-deny") + .await + .expect("run_with_checkpointer"); + assert_eq!(paused.pending_approvals, vec!["gate".to_string()]); + + let caps = mock_capabilities(); + let journal = Arc::new(InMemoryGraphEventJournal::new()); + let observer = Arc::new(crate::observability::NoopObserver) as Arc; + let denied = resume_with_checkpointer_journaled_observed( + &compiled, + &caps, + cp.clone(), + "thread-deny", + Vec::new(), // nothing approved + vec!["gate".to_string()], // the gate is denied + journal, + &observer, + ) + .await + .expect("resume with rejection"); + + assert!( + denied.outcome.pending_approvals.is_empty(), + "a denied gate is settled, not left pending" + ); + assert_eq!( + denied.outcome.output["nodes"]["recover"]["items"][0]["json"]["error"]["node"], + json!("gate"), + "the denied gate must route its error item to the `error`-port recovery node" + ); + assert!( + denied.outcome.output["nodes"]["downstream"].is_null(), + "the main branch must not run when the gate is denied" + ); + } + + #[tokio::test] + async fn resume_denying_a_gate_with_no_error_port_fails_the_run() { + // trigger -> gate{requires_approval} -> downstream, with NO `error` edge. + // Denying the gate must fail the run rather than silently swallow it. + let cp: Arc> = Arc::new(InMemoryCheckpointer::::default()); + let mut gate = node("gate", NodeKind::OutputParser); + gate.config = json!({ "requires_approval": true }); + let graph = WorkflowGraph { + nodes: vec![ + node("t", NodeKind::Trigger), + gate, + node("downstream", NodeKind::OutputParser), + ], + edges: vec![edge("t", "gate"), edge("gate", "downstream")], + ..Default::default() + }; + let compiled = compile(&graph).expect("compile"); + + let caps = mock_capabilities(); + run_with_checkpointer(&compiled, json!({}), &caps, cp.clone(), "thread-deny-fail") + .await + .expect("run_with_checkpointer"); + + let caps = mock_capabilities(); + let journal = Arc::new(InMemoryGraphEventJournal::new()); + let observer = Arc::new(crate::observability::NoopObserver) as Arc; + let result = resume_with_checkpointer_journaled_observed( + &compiled, + &caps, + cp.clone(), + "thread-deny-fail", + Vec::new(), + vec!["gate".to_string()], + journal, + &observer, + ) + .await; + assert!( + result.is_err(), + "denying a gate with no error port must fail the run" + ); + } + #[tokio::test] async fn plain_run_and_resumable_unchanged_by_injectable_checkpointer() { // Regression: the default (non-injectable) `run` and `run_resumable` From 0bfa9e588f0d16ea93bf7297c78913ce8cb4ab36 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 5 Jul 2026 06:19:13 -0700 Subject: [PATCH 3/8] feat(engine): sub-workflow by workflow_id via host WorkflowResolver + docs truth-up MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extend the `sub_workflow` node to run a child either from an inline `workflow` graph (unchanged, back-compat) OR a host-managed `workflow_id` reference. Since the engine is persistence-free, add a new host-injected `WorkflowResolver` capability (mirroring LlmProvider/ToolInvoker/…) that maps `workflow_id` → WorkflowGraph; the node calls `ctx.caps.resolver.resolve`. Cycle/depth handling: every nested sub_workflow run (inline or by id) increments a `run.sub_workflow_depth` counter threaded through a new `run_sub_workflow` entry point; a child exceeding MAX_SUB_WORKFLOW_DEPTH (8) is refused, bounding any indirect cycle (A→B→A). A direct self-reference (a resolved child that references the same workflow_id) is caught statically before it runs. validate() now requires exactly one of `workflow`/`workflow_id`. Docs truth-up: README + CLAUDE.md Status claimed jq/jaq expressions and retry backoff/timeouts were "pending" — they are implemented (src/expr.rs routes to jaq; engine.rs has fixed/exponential backoff + node_timeout_secs). Corrected. Adds engine API (new capability + Capabilities field), so bump 0.3.2 → 0.4.0. Mock resolver + tests: by-id resolves & executes, both-set/neither-set reject, self-reference reject, depth-limit enforced. cargo test green. Claude-Session: https://claude.ai/code/session_01QpsTTxxjFgDhzuozS28tyU --- CLAUDE.md | 9 +- Cargo.lock | 2 +- Cargo.toml | 2 +- README.md | 19 +- src/caps/mock.rs | 43 +++- src/caps/mod.rs | 25 +++ src/engine.rs | 58 ++++++ src/nodes/integration/sub_workflow.rs | 269 ++++++++++++++++++++++++-- src/validate.rs | 77 ++++++++ 9 files changed, 481 insertions(+), 23 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 7af6f50..9e29496 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -76,6 +76,9 @@ data flow), the full node catalog (control-flow + capability-backed), conditiona parallel routing with a merge barrier, per-node error handling (`on_error`/retry/error port), `tracing`/`RunObserver` observability, HITL approval gating + `engine::resume`, opaque `connection_ref`, and schema/`type_version` versioning are all implemented and -tested (unit + reference-workflow e2e; `cargo publish --dry-run` clean). Ahead: full -jq/jaq expressions, retry backoff/timeouts, durable checkpointed replay, and the -OpenHuman host integration (Phase B). See `local/docs/08-roadmap.md`. +tested (unit + reference-workflow e2e; `cargo publish --dry-run` clean). Also done: +full jq/jaq `=`-expressions (`src/expr.rs`, routed to `jaq`), retry backoff +(`fixed`/`exponential`) + per-node timeouts (`node_timeout_secs`), and +sub-workflows by inline graph **or** host `workflow_id` (resolved via the injected +`WorkflowResolver`, depth-bounded). Ahead: durable checkpointed super-step replay +and deeper OpenHuman host integration (Phase B). See `local/docs/08-roadmap.md`. diff --git a/Cargo.lock b/Cargo.lock index c84f40f..cb409b0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1355,7 +1355,7 @@ dependencies = [ [[package]] name = "tinyflows" -version = "0.3.2" +version = "0.4.0" dependencies = [ "async-trait", "futures-timer", diff --git a/Cargo.toml b/Cargo.toml index d13c79e..ebd8f6d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "tinyflows" -version = "0.3.2" +version = "0.4.0" edition = "2024" rust-version = "1.85" license = "GPL-3.0-or-later" diff --git a/README.md b/README.md index afdc6da..c11ad89 100644 --- a/README.md +++ b/README.md @@ -206,11 +206,24 @@ observability, opaque `connection_ref` credentials, and `schema_version` / capabilities, guarded by a reference-workflow e2e suite, and `cargo publish --dry-run` is clean. +Also implemented: + +- **A full jq/jaq expression engine.** Every `=`-prefixed string is a jq program + compiled and executed by [`jaq`](https://crates.io/crates/jaq) with the + evaluation scope as its input (`src/expr.rs`); a bare `=.item.name` dotted path + is served by a fast structural walk, while anything richer (filters, pipes, + `add`, …) routes to jaq. +- **Retry backoff timing and per-node timeouts.** A node's `retry` config takes + `backoff_ms` plus `backoff: "fixed" | "exponential"` (capped at 60 s between + attempts), and a run-level `node_timeout_secs` on the trigger applies a + per-node timeout across the whole run. +- **Sub-workflows by reference.** A `sub_workflow` node runs a child either from + an inline `workflow` graph or from a host-managed `workflow_id`, resolved + through the injected `WorkflowResolver` capability. Nesting is depth-bounded + and direct self-references are rejected. + Not yet: -- A full jq/jaq expression engine — a minimal `=`-dotted-path evaluator ships as - an interim. -- Retry backoff timing and per-node timeouts. - Automatic checkpointed super-step replay. Durable, cross-process resume is already supported by injecting a `Checkpointer` (`engine::run_with_checkpointer` / `resume_with_checkpointer`); only the diff --git a/src/caps/mock.rs b/src/caps/mock.rs index e00705d..4e42929 100644 --- a/src/caps/mock.rs +++ b/src/caps/mock.rs @@ -11,8 +11,10 @@ use serde_json::{Value, json}; use crate::caps::{ Capabilities, CodeLanguage, CodeRunner, HttpClient, LlmProvider, StateStore, ToolInvoker, + WorkflowResolver, }; -use crate::error::Result; +use crate::error::{EngineError, Result}; +use crate::model::WorkflowGraph; /// An [`LlmProvider`] that echoes the request back under a `completion` key. #[derive(Debug, Default, Clone)] @@ -79,15 +81,54 @@ impl StateStore for MockStateStore { } } +/// A [`WorkflowResolver`] backed by an in-memory `workflow_id` → graph map. +/// +/// Empty by default (every `resolve` misses with a capability error), so a run +/// that never uses `sub_workflow`-by-id is unaffected. Register graphs with +/// [`MockWorkflowResolver::with`] to exercise the by-id path in tests. +#[derive(Debug, Default, Clone)] +pub struct MockWorkflowResolver { + workflows: std::collections::HashMap, +} + +impl MockWorkflowResolver { + /// Registers `graph` under `id`, returning `self` for chaining. + #[must_use] + pub fn with(mut self, id: impl Into, graph: WorkflowGraph) -> Self { + self.workflows.insert(id.into(), graph); + self + } +} + +#[async_trait] +impl WorkflowResolver for MockWorkflowResolver { + async fn resolve(&self, workflow_id: &str) -> Result { + self.workflows.get(workflow_id).cloned().ok_or_else(|| { + EngineError::Capability(format!("mock resolver: unknown workflow_id: {workflow_id}")) + }) + } +} + /// Builds a [`Capabilities`] bundle wired entirely to the mock implementations. +/// +/// The bundled [`MockWorkflowResolver`] is empty; use +/// [`mock_capabilities_with_resolver`] to supply one that resolves ids. #[must_use] pub fn mock_capabilities() -> Capabilities { + mock_capabilities_with_resolver(MockWorkflowResolver::default()) +} + +/// Like [`mock_capabilities`], but with a caller-supplied [`WorkflowResolver`] +/// so tests can exercise `sub_workflow`-by-id. +#[must_use] +pub fn mock_capabilities_with_resolver(resolver: impl WorkflowResolver + 'static) -> Capabilities { Capabilities { llm: Arc::new(MockLlm), tools: Arc::new(MockTools), http: Arc::new(MockHttp), code: Arc::new(MockCode), state: Arc::new(MockStateStore::default()), + resolver: Arc::new(resolver), } } diff --git a/src/caps/mod.rs b/src/caps/mod.rs index bbd189c..e900f78 100644 --- a/src/caps/mod.rs +++ b/src/caps/mod.rs @@ -64,6 +64,25 @@ pub trait CodeRunner: Send + Sync { async fn run(&self, language: CodeLanguage, source: &str, input: Value) -> Result; } +/// Resolves a saved workflow graph by its host-managed id. +/// +/// A [`sub_workflow`](crate::nodes::integration) node may reference a child by a +/// host `workflow_id` instead of embedding the child graph inline. The engine is +/// deliberately persistence-free, so it delegates that lookup to this +/// host-injected capability: the embedding application maps `workflow_id` onto +/// whatever store it keeps saved workflows in and hands back the portable +/// [`WorkflowGraph`](crate::model::WorkflowGraph) to run. +#[async_trait] +pub trait WorkflowResolver: Send + Sync { + /// Resolves `workflow_id` to the child [`WorkflowGraph`](crate::model::WorkflowGraph) + /// the referencing `sub_workflow` node should compile and run. + /// + /// # Errors + /// Returns an [`EngineError::Capability`](crate::error::EngineError::Capability) + /// when no workflow with that id exists, or when the host cannot load it. + async fn resolve(&self, workflow_id: &str) -> Result; +} + /// Durable key/value state for a run (used by resumable / stateful workflows). #[async_trait] pub trait StateStore: Send + Sync { @@ -95,6 +114,11 @@ pub struct Capabilities { /// run-ledger-backed store) and nodes access durable state through /// `ctx.caps.state`. pub state: Arc, + /// Resolver for `sub_workflow` nodes that reference a child graph by a + /// host `workflow_id` rather than embedding it inline. The host implements + /// [`WorkflowResolver`] over its saved-workflow store; the engine calls it + /// only when a `sub_workflow` node carries a `workflow_id`. + pub resolver: Arc, } #[cfg(test)] @@ -128,6 +152,7 @@ mod tests { assert!(Arc::ptr_eq(&caps.http, &clone.http)); assert!(Arc::ptr_eq(&caps.code, &clone.code)); assert!(Arc::ptr_eq(&caps.state, &clone.state)); + assert!(Arc::ptr_eq(&caps.resolver, &clone.resolver)); } #[tokio::test] diff --git a/src/engine.rs b/src/engine.rs index 3061c61..83e70dc 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -221,6 +221,52 @@ pub async fn run_with_observer( checkpointer, thread_id, None, + None, + ) + .await?; + Ok(outcome) +} + +/// The maximum nesting depth for `sub_workflow` runs. +/// +/// Each nested `sub_workflow` run (inline **or** by `workflow_id`) increments a +/// `run.sub_workflow_depth` counter; once a child would exceed this bound the +/// `sub_workflow` node refuses to run it. This is the engine's backstop against +/// runaway or cyclic references (e.g. flow A → flow B → flow A by id): the chain +/// is cut after at most this many levels regardless of how the cycle is formed. +/// A direct self-reference is additionally caught statically by the node before +/// any run starts (see [`crate::nodes::integration::SubWorkflowNode`]). +pub const MAX_SUB_WORKFLOW_DEPTH: u64 = 8; + +/// Runs a nested child workflow for a `sub_workflow` node, threading the current +/// nesting `depth` into the child run's `run.sub_workflow_depth`. +/// +/// Behaves like [`run`] (no-op observer, process-local in-memory checkpointer) +/// but seeds the depth counter so a further nested `sub_workflow` inside the +/// child can read it back from `ctx.run` and enforce [`MAX_SUB_WORKFLOW_DEPTH`]. +/// Used only by the `sub_workflow` node's recursive execution. +/// +/// # Errors +/// Same as [`run`]. +pub(crate) async fn run_sub_workflow( + workflow: &CompiledWorkflow, + input: Value, + capabilities: &Capabilities, + depth: u64, +) -> Result { + let checkpointer: Arc> = + Arc::new(InMemoryCheckpointer::::default()); + let thread_id = default_thread_id(workflow)?; + let observer: Arc = Arc::new(crate::observability::NoopObserver); + let (_graph, _thread_id, outcome, _run_ids) = build_and_run( + workflow, + input, + capabilities, + &observer, + checkpointer, + thread_id, + None, + Some(json!({ "sub_workflow_depth": depth })), ) .await?; Ok(outcome) @@ -705,6 +751,7 @@ fn default_thread_id(workflow: &CompiledWorkflow) -> Result { /// /// # Errors /// Same as [`run`]. +#[allow(clippy::too_many_arguments)] async fn build_and_run( workflow: &CompiledWorkflow, input: Value, @@ -713,6 +760,7 @@ async fn build_and_run( checkpointer: Arc>, thread_id: String, journal: Option>, + run_meta_overlay: Option, ) -> Result<(CompiledGraph, String, RunOutcome, GraphRunIds)> { // Process-local, monotonic run id — no time/random source. let run_id = format!("run-{}", NEXT_RUN_ID.fetch_add(1, Ordering::Relaxed)); @@ -739,6 +787,13 @@ async fn build_and_run( .map_err(|e| EngineError::Capability(e.to_string()))?; let mut initial = json!({ "run": { "trigger": input } }); merge(&mut initial, seed_items); + // Optional run-level metadata overlaid onto `run` before the run starts — + // e.g. the `sub_workflow_depth` counter a nested `sub_workflow` run threads + // to bound recursion (see [`run_sub_workflow`]). Merged (not overwritten) so + // it sits alongside `run.trigger`. + if let Some(overlay) = run_meta_overlay { + merge(&mut initial, json!({ "run": overlay })); + } // The run is keyed under the caller-supplied `thread_id` (the default paths // pass the trigger id, preserving prior behavior); this is where the @@ -934,6 +989,7 @@ pub async fn run_resumable( checkpointer, thread_id, None, + None, ) .await?; Ok(ResumableRun { @@ -982,6 +1038,7 @@ pub async fn run_with_checkpointer( checkpointer, thread_id.to_string(), None, + None, ) .await?; Ok(outcome) @@ -1102,6 +1159,7 @@ pub async fn run_with_checkpointer_journaled_observed( checkpointer, thread_id.to_string(), Some(journal), + None, ) .await?; tracing::debug!( diff --git a/src/nodes/integration/sub_workflow.rs b/src/nodes/integration/sub_workflow.rs index 2f9de7c..5f5264f 100644 --- a/src/nodes/integration/sub_workflow.rs +++ b/src/nodes/integration/sub_workflow.rs @@ -1,34 +1,132 @@ //! The `sub_workflow` node: runs another workflow as a nested sub-graph. use async_trait::async_trait; +use serde_json::Value; +use crate::engine::MAX_SUB_WORKFLOW_DEPTH; use crate::error::{EngineError, Result}; +use crate::model::{NodeKind, WorkflowGraph}; use crate::nodes::{NodeContext, NodeExecutor, NodeOutput}; /// Runs another workflow as a nested sub-graph. /// -/// The child [`WorkflowGraph`](crate::model::WorkflowGraph) is embedded under the -/// node's `workflow` config key. The node compiles and runs that child graph via -/// the same [`crate::engine::run`], sharing the host [`Capabilities`](crate::caps::Capabilities) -/// with the parent run, and emits the child's final run state as a single output item. +/// The child [`WorkflowGraph`](crate::model::WorkflowGraph) is supplied one of +/// two ways — **exactly one** of these config keys must be present: +/// +/// - `workflow` — the child graph embedded **inline** as JSON (back-compat; +/// the original behavior). +/// - `workflow_id` — a host-managed **reference** to a saved workflow. The +/// engine is persistence-free, so it resolves the id to a graph through the +/// host-injected [`WorkflowResolver`](crate::caps::WorkflowResolver) +/// (`ctx.caps.resolver`). +/// +/// The resolved child is compiled and run via [`crate::engine::run_sub_workflow`], +/// sharing the host [`Capabilities`](crate::caps::Capabilities) with the parent +/// run, and its final run state is emitted as a single output item. +/// +/// ## Cycle / depth handling +/// +/// Every nested `sub_workflow` run (inline or by id) increments a +/// `run.sub_workflow_depth` counter; a child that would exceed +/// [`MAX_SUB_WORKFLOW_DEPTH`] is refused. This bounds **any** cycle — including +/// indirect ones like flow A → flow B → flow A by id — after at most that many +/// levels. In addition, a **direct self-reference** (a resolved child graph that +/// itself references the same `workflow_id`) is caught statically here before the +/// child ever runs, so the common one-level loop fails fast with a clear error +/// rather than unwinding the full depth budget. #[derive(Debug, Default, Clone)] pub struct SubWorkflowNode; +/// Reads the current nesting depth from the run metadata (`0` at the top level). +fn current_depth(run: &Value) -> u64 { + run.get("sub_workflow_depth") + .and_then(Value::as_u64) + .unwrap_or(0) +} + +/// Deserializes the inline `workflow` config value into a [`WorkflowGraph`]. +fn inline_child(workflow: &Value) -> Result { + serde_json::from_value(workflow.clone()) + .map_err(|e| EngineError::Capability(format!("sub_workflow node: invalid workflow: {e}"))) +} + +/// Rejects a resolved child that references the same `workflow_id` it was loaded +/// under — a direct self-reference (one-level cycle). Deeper cycles are still +/// bounded by the depth counter; this catches the obvious case eagerly. +fn reject_self_reference(child: &WorkflowGraph, workflow_id: &str) -> Result<()> { + let self_ref = child + .nodes + .iter() + .filter(|n| n.kind == NodeKind::SubWorkflow) + .any(|n| n.config.get("workflow_id").and_then(Value::as_str) == Some(workflow_id)); + if self_ref { + return Err(EngineError::Capability(format!( + "sub_workflow node: workflow_id {workflow_id:?} references itself (cycle)" + ))); + } + Ok(()) +} + #[async_trait] impl NodeExecutor for SubWorkflowNode { async fn execute(&self, ctx: NodeContext<'_>) -> Result { - let child_value = ctx.node.config.get("workflow").ok_or_else(|| { - EngineError::Capability("sub_workflow node: missing `workflow` in config".to_string()) - })?; - let child: crate::model::WorkflowGraph = serde_json::from_value(child_value.clone()) - .map_err(|e| { - EngineError::Capability(format!("sub_workflow node: invalid workflow: {e}")) - })?; + let inline = ctx.node.config.get("workflow"); + let workflow_id = ctx + .node + .config + .get("workflow_id") + .and_then(Value::as_str) + .filter(|s| !s.is_empty()); + + // Exactly one of `workflow` / `workflow_id` must be set. + let child: WorkflowGraph = match (inline, workflow_id) { + (Some(_), Some(_)) => { + return Err(EngineError::Capability( + "sub_workflow node: set exactly one of `workflow` (inline) or `workflow_id` \ + (reference), not both" + .to_string(), + )); + } + (None, None) => { + return Err(EngineError::Capability( + "sub_workflow node: missing `workflow` (inline) or `workflow_id` (reference) \ + in config" + .to_string(), + )); + } + (Some(inline_value), None) => { + tracing::debug!(node = %ctx.node.id, "sub_workflow: running inline child graph"); + inline_child(inline_value)? + } + (None, Some(id)) => { + tracing::debug!(node = %ctx.node.id, workflow_id = %id, "sub_workflow: resolving child graph by workflow_id"); + let resolved = ctx.caps.resolver.resolve(id).await?; + reject_self_reference(&resolved, id)?; + resolved + } + }; + + // Depth / cycle guard: bound total nesting regardless of how a cycle is + // formed. The child runs one level deeper than the current run. + let child_depth = current_depth(ctx.run) + 1; + if child_depth > MAX_SUB_WORKFLOW_DEPTH { + return Err(EngineError::Capability(format!( + "sub_workflow node: maximum nesting depth {MAX_SUB_WORKFLOW_DEPTH} exceeded \ + (possible cycle)" + ))); + } + let compiled = crate::compiler::compile(&child)?; let input = serde_json::to_value(ctx.input).map_err(|e| EngineError::Capability(e.to_string()))?; // Box the recursive engine call so the async future type stays sized. - let outcome = Box::pin(crate::engine::run(&compiled, input, ctx.caps)).await?; + let outcome = Box::pin(crate::engine::run_sub_workflow( + &compiled, + input, + ctx.caps, + child_depth, + )) + .await?; Ok(NodeOutput::main(vec![crate::data::Item::new( outcome.output, )])) @@ -40,12 +138,15 @@ mod tests { use serde_json::{Value, json}; use super::SubWorkflowNode; - use crate::caps::mock::mock_capabilities; + use crate::caps::Capabilities; + use crate::caps::mock::{ + MockWorkflowResolver, mock_capabilities, mock_capabilities_with_resolver, + }; use crate::compiler::compile; use crate::engine::run; use crate::error::EngineError; use crate::model::{Edge, Node, NodeKind, WorkflowGraph}; - use crate::nodes::{NodeContext, NodeExecutor}; + use crate::nodes::{NodeContext, NodeExecutor, NodeOutput}; fn node(id: &str, kind: NodeKind) -> Node { Node { @@ -144,4 +245,144 @@ mod tests { "child trigger node should have run and echoed its seeded input" ); } + + /// Executes a lone `sub_workflow` node with the given config, run metadata, + /// and capabilities, returning its raw [`Result`]. + async fn execute_with( + config: Value, + run_meta: Value, + caps: &Capabilities, + ) -> Result { + let mut sw = node("sw", NodeKind::SubWorkflow); + sw.config = config; + let input = vec![]; + let ctx = NodeContext { + node: &sw, + input: &input, + run: &run_meta, + caps, + }; + SubWorkflowNode.execute(ctx).await + } + + #[tokio::test] + async fn both_workflow_and_workflow_id_is_rejected() { + // Exactly one of the two config keys may be set. + let err = execute_err(json!({ + "workflow": { "nodes": [], "edges": [] }, + "workflow_id": "child-1" + })) + .await; + assert!( + matches!(err, EngineError::Capability(ref m) if m.contains("exactly one")), + "expected an exactly-one config error, got: {err:?}" + ); + } + + #[tokio::test] + async fn empty_workflow_id_falls_back_to_missing_config_error() { + // A blank `workflow_id` is treated as absent, so with no inline + // `workflow` either the node reports the missing-config error. + let err = execute_err(json!({ "workflow_id": "" })).await; + assert!( + matches!(err, EngineError::Capability(ref m) if m.contains("missing")), + "expected a missing-config error, got: {err:?}" + ); + } + + #[tokio::test] + async fn sub_workflow_by_id_resolves_via_resolver_and_executes() { + // The saved child is a single trigger node, registered under an id the + // parent references via `workflow_id`. + let child = WorkflowGraph { + nodes: vec![node("ct", NodeKind::Trigger)], + ..Default::default() + }; + let caps = + mock_capabilities_with_resolver(MockWorkflowResolver::default().with("child-1", child)); + + let mut sw = node("sw", NodeKind::SubWorkflow); + sw.config = json!({ "workflow_id": "child-1" }); + let parent = WorkflowGraph { + nodes: vec![node("t", NodeKind::Trigger), sw], + edges: vec![Edge { + from_node: "t".to_string(), + from_port: "main".to_string(), + to_node: "sw".to_string(), + to_port: "main".to_string(), + }], + ..Default::default() + }; + let compiled = compile(&parent).expect("compile parent"); + + let out = run(&compiled, json!({ "hi": 1 }), &caps) + .await + .expect("run parent"); + + // The referenced child was resolved and actually ran. + let child_state = &out.output["nodes"]["sw"]["items"][0]["json"]; + assert_eq!( + child_state["nodes"]["ct"]["items"][0]["json"], + json!([{ "json": { "hi": 1 } }]), + "resolved child trigger node should have run and echoed its seeded input" + ); + // The child ran one nesting level deep. + assert_eq!(child_state["run"]["sub_workflow_depth"], json!(1)); + } + + #[tokio::test] + async fn unknown_workflow_id_surfaces_resolver_error() { + // The default mock resolver knows no ids, so resolution fails. + let caps = mock_capabilities(); + let err = execute_with(json!({ "workflow_id": "nope" }), Value::Null, &caps) + .await + .expect_err("unknown id must error"); + assert!( + matches!(err, EngineError::Capability(ref m) if m.contains("nope")), + "expected the resolver's unknown-id error, got: {err:?}" + ); + } + + #[tokio::test] + async fn direct_self_reference_by_id_is_rejected() { + // The saved child itself references the same id — a one-level cycle, + // caught statically before it runs. + let mut inner = node("inner", NodeKind::SubWorkflow); + inner.config = json!({ "workflow_id": "loop-1" }); + let child = WorkflowGraph { + nodes: vec![node("ct", NodeKind::Trigger), inner], + ..Default::default() + }; + let caps = + mock_capabilities_with_resolver(MockWorkflowResolver::default().with("loop-1", child)); + + let err = execute_with(json!({ "workflow_id": "loop-1" }), Value::Null, &caps) + .await + .expect_err("self-reference must be rejected"); + assert!( + matches!(err, EngineError::Capability(ref m) if m.contains("cycle")), + "expected a cycle rejection, got: {err:?}" + ); + } + + #[tokio::test] + async fn depth_limit_is_enforced() { + // A run already at the maximum nesting depth refuses to descend further, + // even for a trivial resolvable child (bounds indirect cycles). + let child = WorkflowGraph { + nodes: vec![node("ct", NodeKind::Trigger)], + ..Default::default() + }; + let caps = + mock_capabilities_with_resolver(MockWorkflowResolver::default().with("child-1", child)); + + let run_meta = json!({ "sub_workflow_depth": crate::engine::MAX_SUB_WORKFLOW_DEPTH }); + let err = execute_with(json!({ "workflow_id": "child-1" }), run_meta, &caps) + .await + .expect_err("exceeding the depth budget must error"); + assert!( + matches!(err, EngineError::Capability(ref m) if m.contains("depth")), + "expected a depth-limit error, got: {err:?}" + ); + } } diff --git a/src/validate.rs b/src/validate.rs index 682b96d..c63a579 100644 --- a/src/validate.rs +++ b/src/validate.rs @@ -42,6 +42,29 @@ pub fn validate(graph: &WorkflowGraph) -> Result<(), ValidationError> { } } + // Per-kind config checks. A `sub_workflow` node must reference its child + // exactly one way: an inline `workflow` graph OR a `workflow_id` reference, + // never both and never neither (the reference form is resolved at run time + // via the host `WorkflowResolver`). + for node in &graph.nodes { + if node.kind == NodeKind::SubWorkflow { + let has_inline = node.config.get("workflow").is_some(); + let has_ref = node + .config + .get("workflow_id") + .and_then(serde_json::Value::as_str) + .is_some_and(|s| !s.is_empty()); + if has_inline == has_ref { + return Err(ValidationError::InvalidNodeConfig { + node: node.id.clone(), + reason: "sub_workflow requires exactly one of `workflow` (inline) or \ + `workflow_id` (reference)" + .to_string(), + }); + } + } + } + Ok(()) } @@ -188,6 +211,60 @@ mod tests { } } + fn sub_workflow_node(config: serde_json::Value) -> Node { + let mut n = node("sw", NodeKind::SubWorkflow); + n.config = config; + n + } + + fn graph_with_sub_workflow(config: serde_json::Value) -> WorkflowGraph { + WorkflowGraph { + nodes: vec![node("t", NodeKind::Trigger), sub_workflow_node(config)], + ..Default::default() + } + } + + #[test] + fn sub_workflow_accepts_inline_workflow() { + let graph = graph_with_sub_workflow(serde_json::json!({ + "workflow": { "nodes": [], "edges": [] } + })); + assert_eq!(validate(&graph), Ok(())); + } + + #[test] + fn sub_workflow_accepts_workflow_id() { + let graph = graph_with_sub_workflow(serde_json::json!({ "workflow_id": "child-1" })); + assert_eq!(validate(&graph), Ok(())); + } + + #[test] + fn sub_workflow_rejects_both_inline_and_id() { + let graph = graph_with_sub_workflow(serde_json::json!({ + "workflow": { "nodes": [], "edges": [] }, + "workflow_id": "child-1" + })); + assert!(matches!( + validate(&graph), + Err(ValidationError::InvalidNodeConfig { .. }) + )); + } + + #[test] + fn sub_workflow_rejects_neither_inline_nor_id() { + // A blank `workflow_id` counts as absent. + let graph = graph_with_sub_workflow(serde_json::json!({ "workflow_id": "" })); + assert!(matches!( + validate(&graph), + Err(ValidationError::InvalidNodeConfig { .. }) + )); + let graph = graph_with_sub_workflow(serde_json::Value::Null); + assert!(matches!( + validate(&graph), + Err(ValidationError::InvalidNodeConfig { .. }) + )); + } + #[test] fn duplicate_id_is_reported_before_trigger_checks() { // Two agents sharing an id and no trigger: the duplicate-id check runs From 01fe8b079e8b7df94594b9d9862d35ce9b404ac2 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 5 Jul 2026 06:38:30 -0700 Subject: [PATCH 4/8] feat(engine): output_parser schema validation + LLM auto-fix, cooperative cancellation token, agent sub-ports MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 7 engine work (part B): 7.2 — output_parser now validates each input item against its configured `config.schema` (a dependency-free JSON Schema subset: type/required/ properties/items/enum). On failure it makes one LLM auto-fix attempt via the injected LlmProvider and re-validates; persistent failure surfaces a capability error routed per the node's on_error policy. No schema ⇒ identity passthrough (back-compat). Shared validate+fix routine lives in nodes/integration/schema.rs. 7.5 — new CancellationToken (Arc, runtime-agnostic) threaded through the run loop and checked at each node boundary; a cancelled run stops scheduling new node work and returns RunOutcome { cancelled: true }. New public entry points run_cancellable / resume_cancellable; existing signatures unchanged (default paths pass a fresh never-cancelled token). RunOutcome gains a `cancelled` field. 7.1 — agent node sub-ports (config-embedded): a `tools` sub-port surfaces available tools to the model and honors a single-hop model-elected tool_call (offered tools only), and an `output_parser` sub-port validates/repairs the agent's output via the shared schema routine. Plain agent nodes are unchanged. chat_model / memory sub-ports remain documented follow-ups (not stubbed). Version 0.4.0 -> 0.5.0 (new public API + RunOutcome field). 260 lib tests green (+21). Claude-Session: https://claude.ai/code/session_01QpsTTxxjFgDhzuozS28tyU --- Cargo.toml | 2 +- src/engine.rs | 216 +++++++++++++++++- src/nodes/integration/agent.rs | 251 ++++++++++++++++++++- src/nodes/integration/mod.rs | 1 + src/nodes/integration/output_parser.rs | 205 ++++++++++++++++- src/nodes/integration/schema.rs | 299 +++++++++++++++++++++++++ 6 files changed, 958 insertions(+), 16 deletions(-) create mode 100644 src/nodes/integration/schema.rs diff --git a/Cargo.toml b/Cargo.toml index ebd8f6d..f7f3743 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "tinyflows" -version = "0.4.0" +version = "0.5.0" edition = "2024" rust-version = "1.85" license = "GPL-3.0-or-later" diff --git a/src/engine.rs b/src/engine.rs index 83e70dc..319dd82 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -14,7 +14,7 @@ //! predecessor is wired with waiting edges so it runs only once all of them //! finish). -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Instant; @@ -61,6 +61,43 @@ use crate::observability::{ExecutionStep, Run, RunObserver, RunStatus, StepStatu /// time- or random-based so ids stay deterministic within a process. static NEXT_RUN_ID: AtomicU64 = AtomicU64::new(0); +/// A cooperative cancellation signal for a workflow run. +/// +/// Cheap to clone (an [`Arc`] around an atomic flag) and runtime-agnostic — the +/// crate deliberately avoids depending on any executor's cancellation type. Hand +/// a clone to a cancellable entry point ([`run_cancellable`] / +/// [`resume_cancellable`]) and keep another; calling [`cancel`](Self::cancel) +/// from anywhere flips the flag, and the run stops scheduling real node work at +/// the next node boundary, returning a [`RunOutcome`] with +/// [`cancelled`](RunOutcome::cancelled) set. +/// +/// Cancellation is **cooperative and boundary-level**: a node already executing +/// runs to completion; the token is checked before each node runs, so no *new* +/// node work starts after cancellation. This complements (does not replace) a +/// host's hard task-abort — it lets a run wind down cleanly rather than being +/// dropped mid-await. +#[derive(Debug, Clone, Default)] +pub struct CancellationToken(Arc); + +impl CancellationToken { + /// Creates a fresh, un-cancelled token. + #[must_use] + pub fn new() -> Self { + Self::default() + } + + /// Signals cancellation. Idempotent; safe to call from any thread. + pub fn cancel(&self) { + self.0.store(true, Ordering::SeqCst); + } + + /// Whether cancellation has been signalled. + #[must_use] + pub fn is_cancelled(&self) -> bool { + self.0.load(Ordering::SeqCst) + } +} + /// The result of a completed workflow run. #[derive(Debug, Clone)] pub struct RunOutcome { @@ -71,6 +108,11 @@ pub struct RunOutcome { /// whose id was not present in the run input's `approvals` array; its /// downstream did not run. Empty for a fully completed run. pub pending_approvals: Vec, + /// Whether the run observed a cancelled [`CancellationToken`] and wound down + /// early. When `true`, some downstream nodes were skipped (their slots in + /// `output` were not produced), so treat `output` as partial. Always `false` + /// for runs started without a token or that completed before any cancel. + pub cancelled: bool, } /// The `tinyagents`-minted identifiers of the underlying graph run. @@ -222,6 +264,44 @@ pub async fn run_with_observer( thread_id, None, None, + CancellationToken::new(), + ) + .await?; + Ok(outcome) +} + +/// Like [`run`], but observes `token`: cancelling it stops the run from +/// scheduling further node work at the next node boundary, and the returned +/// [`RunOutcome`] has [`cancelled`](RunOutcome::cancelled) set. A node already +/// executing when the token flips finishes; no *new* node work starts after +/// cancellation. All other behavior is identical to [`run`]. +/// +/// This is the clean, engine-level cooperative-cancellation path, complementing a +/// host's hard task-abort: the run winds down and returns a partial outcome rather +/// than being dropped mid-await. +/// +/// # Errors +/// Same as [`run`]. +pub async fn run_cancellable( + workflow: &CompiledWorkflow, + input: Value, + capabilities: &Capabilities, + token: CancellationToken, +) -> Result { + let observer = Arc::new(crate::observability::NoopObserver) as Arc; + let checkpointer: Arc> = + Arc::new(InMemoryCheckpointer::::default()); + let thread_id = default_thread_id(workflow)?; + let (_graph, _thread_id, outcome, _run_ids) = build_and_run( + workflow, + input, + capabilities, + &observer, + checkpointer, + thread_id, + None, + None, + token, ) .await?; Ok(outcome) @@ -267,6 +347,7 @@ pub(crate) async fn run_sub_workflow( thread_id, None, Some(json!({ "sub_workflow_depth": depth })), + CancellationToken::new(), ) .await?; Ok(outcome) @@ -298,6 +379,7 @@ fn build_graph( steps: &Arc>>, checkpointer: Arc>, journal: Option>, + token: CancellationToken, ) -> Result<(CompiledGraph, String)> { let graph = &workflow.graph; @@ -378,6 +460,7 @@ fn build_graph( let caps = capabilities.clone(); let observer = observer.clone(); let steps = steps.clone(); + let token = token.clone(); let is_trigger = node.kind == NodeKind::Trigger; // Successors to fan out to concurrently, or `None` for a non-fan-out node. let fan_out = fan_out_targets(&node.id); @@ -395,6 +478,7 @@ fn build_graph( let caps = caps.clone(); let observer = observer.clone(); let steps = steps.clone(); + let token = token.clone(); let fan_out = fan_out.clone(); // The resume value delivered to this node on a checkpointed resume, if // any. A bare `true` means "approve the interrupted gate"; a structured @@ -415,6 +499,19 @@ fn build_graph( None => NodeResult::Update(update), }; + // Cooperative cancellation, checked at the node boundary before + // any real work. When the run's token is cancelled this node + // becomes a no-op: it emits an empty update on the default port + // and — crucially — does **not** fan out (a plain `Update`, not + // `emit`), so a fan-out node's parallel successors are not + // scheduled. Downstream nodes reached by static edges will hit + // this same check and no-op in turn, so the run winds down without + // starting further node work. The engine reports it as cancelled. + if token.is_cancelled() { + tracing::info!(node = %node.id, "run cancelled; skipping node work"); + return Ok(NodeResult::Update(items_update(&node.id, &[], None)?)); + } + if is_trigger { // The trigger payload is pre-seeded into the state; no-op update // (still fanning out if the trigger has parallel successors). @@ -761,6 +858,7 @@ async fn build_and_run( thread_id: String, journal: Option>, run_meta_overlay: Option, + token: CancellationToken, ) -> Result<(CompiledGraph, String, RunOutcome, GraphRunIds)> { // Process-local, monotonic run id — no time/random source. let run_id = format!("run-{}", NEXT_RUN_ID.fetch_add(1, Ordering::Relaxed)); @@ -781,6 +879,7 @@ async fn build_and_run( &steps, checkpointer, journal, + token.clone(), )?; let seed_items = items_update(&trigger_id, &[Item::new(input.clone())], None) @@ -841,6 +940,9 @@ async fn build_and_run( RunOutcome { output: execution.state, pending_approvals, + // A cancelled token means at least one node boundary short-circuited, + // so surface the run as cancelled (its `output` is partial). + cancelled: token.is_cancelled(), }, graph_run_ids, )) @@ -897,6 +999,47 @@ pub async fn resume( run(workflow, merged_input, capabilities).await } +/// Like [`resume`], but observes `token`: cancelling it winds the resumed run +/// down at the next node boundary and sets [`RunOutcome::cancelled`]. This is the +/// re-run-based resume (the same deterministic replay [`resume`] performs), made +/// cooperatively cancellable. +/// +/// # Errors +/// Same as [`resume`]. +pub async fn resume_cancellable( + workflow: &CompiledWorkflow, + input: Value, + newly_approved: Vec, + capabilities: &Capabilities, + token: CancellationToken, +) -> Result { + let mut approvals: Vec = input + .get("approvals") + .and_then(Value::as_array) + .map(|existing| { + existing + .iter() + .filter_map(Value::as_str) + .map(str::to_string) + .collect() + }) + .unwrap_or_default(); + for id in newly_approved { + if !approvals.contains(&id) { + approvals.push(id); + } + } + + let mut merged_input = input; + if let Value::Object(map) = &mut merged_input { + map.insert("approvals".to_string(), json!(approvals)); + } else { + merged_input = json!({ "approvals": approvals }); + } + + run_cancellable(workflow, merged_input, capabilities, token).await +} + /// A live, resumable workflow run. /// /// Unlike the re-run-based [`resume`], this keeps the compiled `tinyagents` graph @@ -957,6 +1100,7 @@ impl ResumableRun { Ok(RunOutcome { output: execution.state, pending_approvals, + cancelled: false, }) } } @@ -990,6 +1134,7 @@ pub async fn run_resumable( thread_id, None, None, + CancellationToken::new(), ) .await?; Ok(ResumableRun { @@ -1039,6 +1184,7 @@ pub async fn run_with_checkpointer( thread_id.to_string(), None, None, + CancellationToken::new(), ) .await?; Ok(outcome) @@ -1160,6 +1306,7 @@ pub async fn run_with_checkpointer_journaled_observed( thread_id.to_string(), Some(journal), None, + CancellationToken::new(), ) .await?; tracing::debug!( @@ -1277,6 +1424,7 @@ async fn resume_with_checkpointer_inner( &steps, checkpointer, journal, + CancellationToken::new(), )?; // Approvals recorded for downstream visibility. On resume the interrupted @@ -1320,6 +1468,9 @@ async fn resume_with_checkpointer_inner( RunOutcome { output: execution.state, pending_approvals, + // Checkpointed resume does not (yet) thread a caller token; a + // cancellable resume goes through `resume_cancellable`. + cancelled: false, }, graph_run_ids, )) @@ -3031,4 +3182,67 @@ mod tests { assert!(done.pending_approvals.is_empty()); assert!(!done.output["nodes"]["downstream"]["items"].is_null()); } + + #[tokio::test] + async fn uncancelled_token_runs_to_completion() { + // A fresh (never-cancelled) token behaves exactly like `run`. + let graph = WorkflowGraph { + nodes: vec![ + node("t", NodeKind::Trigger), + node("p", NodeKind::OutputParser), + ], + edges: vec![edge("t", "p")], + ..Default::default() + }; + let compiled = compile(&graph).expect("compile"); + let token = CancellationToken::new(); + let outcome = run_cancellable(&compiled, json!({ "n": 1 }), &mock_capabilities(), token) + .await + .expect("run"); + assert!(!outcome.cancelled); + assert_eq!(outcome.output["nodes"]["p"]["items"][0]["json"]["n"], 1); + } + + #[tokio::test] + async fn cancelled_token_stops_run_and_reports_cancelled() { + // trigger -> bad (a tool_call with no `slug`, on_error defaulting to + // `stop`). If `bad` ever executed it would fail the whole run. Cancelling + // the token before the run means `bad` short-circuits at its node + // boundary instead of executing, so the run completes cleanly and reports + // cancelled — proving new node work is not scheduled after cancellation. + let graph = WorkflowGraph { + nodes: vec![ + node("t", NodeKind::Trigger), + node("bad", NodeKind::ToolCall), + ], + edges: vec![edge("t", "bad")], + ..Default::default() + }; + let compiled = compile(&graph).expect("compile"); + let token = CancellationToken::new(); + token.cancel(); + let outcome = run_cancellable(&compiled, json!({ "n": 1 }), &mock_capabilities(), token) + .await + .expect("cancelled run still returns Ok"); + assert!(outcome.cancelled, "outcome should report cancelled"); + // `bad` short-circuited: it emitted an empty item list, not a tool result + // and not a run-ending error. + let items = &outcome.output["nodes"]["bad"]["items"]; + assert!( + items.as_array().is_some_and(|a| a.is_empty()), + "cancelled node should emit no items, got: {items:?}" + ); + } + + #[test] + fn cancellation_token_flips_and_is_shared_across_clones() { + let token = CancellationToken::new(); + let clone = token.clone(); + assert!(!token.is_cancelled()); + assert!(!clone.is_cancelled()); + clone.cancel(); + // Both handles observe the flip — they share one atomic flag. + assert!(token.is_cancelled()); + assert!(clone.is_cancelled()); + } } diff --git a/src/nodes/integration/agent.rs b/src/nodes/integration/agent.rs index e26ef5d..342b9a3 100644 --- a/src/nodes/integration/agent.rs +++ b/src/nodes/integration/agent.rs @@ -1,14 +1,39 @@ -//! The `agent` node: an LLM agent turn. +//! The `agent` node: an LLM agent turn with optional sub-ports. use async_trait::async_trait; +use serde_json::Value; use crate::data::Item; use crate::error::Result; +use crate::nodes::integration::schema; use crate::nodes::{NodeContext, NodeExecutor, NodeOutput}; -/// Runs an LLM agent turn with optional chat-model / memory / tool / -/// output-parser sub-ports (via [`crate::caps::LlmProvider`] and -/// [`crate::caps::ToolInvoker`]). +/// Runs an LLM agent turn, optionally composed with **sub-ports** that attach an +/// output parser and tools to the bare completion. +/// +/// The node config is the completion request handed to the injected +/// [`LlmProvider`](crate::caps::LlmProvider). On top of that, two sub-ports are +/// wired (config-embedded, so a plain agent node with just a prompt still works +/// unchanged): +/// +/// - **tool sub-port** (`config.tools`): the available tools are surfaced to the +/// model in the request. If the model's response elects to call one of the +/// *offered* tools — a `tool_call: { slug, args?, connection_ref? }` object in +/// the response — the agent invokes it once via +/// [`ToolInvoker`](crate::caps::ToolInvoker) and attaches the result under +/// `tool_result`. This is a **single hop** (no unbounded agent loop) — a full +/// multi-turn tool-use loop is a documented follow-up. +/// - **output-parser sub-port** (`config.output_parser`): after the completion +/// (and any tool hop), the resulting value is validated/repaired against +/// `config.output_parser.schema` using the shared [`schema`] routine +/// (validate → one LLM auto-fix → re-validate), honoring +/// `config.output_parser.auto_fix` (default `true`). +/// +/// Sub-ports **not** yet wired (documented follow-ups): a `chat_model` sub-port +/// (attached model selection beyond what the request already carries) and a +/// `memory` sub-port (conversation memory injected into the request / persisted +/// across turns). Those require attached-node wiring and/or `StateStore` plumbing +/// and are deliberately left out rather than stubbed. #[derive(Debug, Default, Clone)] pub struct AgentNode; @@ -19,12 +44,69 @@ impl NodeExecutor for AgentNode { // node's input before treating the config as the completion request. let scope = crate::nodes::expr_scope(&ctx); let cfg = crate::expr::resolve(&ctx.node.config, &scope); - // A3-basic: the node config is the completion request; sub-port wiring is a later refinement. - let conn = cfg - .get("connection_ref") - .and_then(serde_json::Value::as_str); + let conn = cfg.get("connection_ref").and_then(Value::as_str); + + // The node config *is* the completion request; when a `tools` sub-port is + // configured its descriptors ride along in the request so the model can + // elect to call one. let response = ctx.caps.llm.complete(cfg.clone(), conn).await?; - Ok(NodeOutput::main(vec![Item::new(response)])) + + // Tool sub-port (single hop): honor a `tool_call` the model returned, but + // only for a tool that was actually offered in `config.tools`. + let mut value = response; + if let Some(tool_call) = value.get("tool_call").cloned() { + if let Some(slug) = tool_call.get("slug").and_then(Value::as_str) { + let offered = cfg + .get("tools") + .and_then(Value::as_array) + .is_some_and(|tools| { + tools + .iter() + .any(|t| t.get("slug").and_then(Value::as_str) == Some(slug)) + }); + if offered { + tracing::debug!(slug, "agent tool sub-port: invoking model-elected tool"); + let args = tool_call.get("args").cloned().unwrap_or(Value::Null); + let tool_conn = tool_call + .get("connection_ref") + .and_then(Value::as_str) + .or(conn); + let result = ctx.caps.tools.invoke(slug, args, tool_conn).await?; + if let Value::Object(map) = &mut value { + map.insert("tool_result".to_string(), result); + } + } else { + tracing::warn!( + slug, + "agent tool sub-port: model elected an un-offered tool; ignoring" + ); + } + } + } + + // Output-parser sub-port: validate/repair the agent output against a schema. + if let Some(parser) = cfg.get("output_parser").filter(|p| !p.is_null()) { + if let Some(parser_schema) = parser.get("schema").filter(|s| !s.is_null()) { + let auto_fix = parser + .get("auto_fix") + .and_then(Value::as_bool) + .unwrap_or(true); + let parser_conn = parser + .get("connection_ref") + .and_then(Value::as_str) + .or(conn); + value = schema::parse_and_validate( + value, + parser_schema, + auto_fix, + &ctx.caps.llm, + parser_conn, + ) + .await?; + } + } + + Ok(NodeOutput::main(vec![Item::new(value)])) } } @@ -172,4 +254,155 @@ mod tests { assert_eq!(out.items.len(), 1); assert_eq!(out.port, None); } + + // --- sub-ports: tool + output_parser --- + + use crate::caps::{Capabilities, LlmProvider}; + use async_trait::async_trait; + use std::sync::Arc; + + fn caps_with_llm(llm: Arc) -> Capabilities { + let mut caps = mock_capabilities(); + caps.llm = llm; + caps + } + + async fn run_agent(node: &Node, caps: &Capabilities) -> Value { + let input: Vec = vec![]; + let run_meta = Value::Null; + let ctx = NodeContext { + node, + input: &input, + run: &run_meta, + caps, + }; + AgentNode + .execute(ctx) + .await + .expect("execute") + .items + .remove(0) + .json + } + + /// An LLM that returns a fixed `tool_call` directive on the completion call. + struct ToolCallingLlm(Value); + + #[async_trait] + impl LlmProvider for ToolCallingLlm { + async fn complete( + &self, + _request: Value, + _conn: Option<&str>, + ) -> crate::error::Result { + Ok(self.0.clone()) + } + } + + #[tokio::test] + async fn tool_sub_port_invokes_offered_tool_and_attaches_result() { + // The model elects to call an offered tool; the agent invokes it once and + // attaches the (mock) tool output under `tool_result`. + let node = agent_node(json!({ + "prompt": "do it", + "tools": [{ "slug": "slack.post" }] + })); + let llm = Arc::new(ToolCallingLlm(json!({ + "tool_call": { "slug": "slack.post", "args": { "text": "hi" } } + }))); + let value = run_agent(&node, &caps_with_llm(llm)).await; + // Mock ToolInvoker echoes the slug/args it was called with. + assert_eq!(value["tool_result"]["tool"], "slack.post"); + assert_eq!(value["tool_result"]["args"]["text"], "hi"); + } + + #[tokio::test] + async fn tool_sub_port_ignores_unoffered_tool() { + // The model tries to call a tool that was never offered; the agent leaves + // the output untouched (no `tool_result`). + let node = agent_node(json!({ + "prompt": "do it", + "tools": [{ "slug": "slack.post" }] + })); + let llm = Arc::new(ToolCallingLlm(json!({ + "tool_call": { "slug": "danger.delete_all" } + }))); + let value = run_agent(&node, &caps_with_llm(llm)).await; + assert!(value.get("tool_result").is_none()); + } + + /// An LLM that returns an invalid completion, but a schema-valid value when + /// asked to coerce (the auto-fix call carries `task == "coerce_to_schema"`). + struct ParserLlm { + completion: Value, + fixed: Value, + } + + #[async_trait] + impl LlmProvider for ParserLlm { + async fn complete( + &self, + request: Value, + _conn: Option<&str>, + ) -> crate::error::Result { + if request.get("task").and_then(Value::as_str) == Some("coerce_to_schema") { + Ok(json!({ "value": self.fixed.clone() })) + } else { + Ok(self.completion.clone()) + } + } + } + + #[tokio::test] + async fn output_parser_sub_port_repairs_agent_output() { + // The completion is missing a required `name`; the output-parser sub-port + // runs a one-shot auto-fix that supplies it. + let node = agent_node(json!({ + "prompt": "hi", + "output_parser": { "schema": { "type": "object", "required": ["name"] } } + })); + let llm = Arc::new(ParserLlm { + completion: json!({ "wrong": 1 }), + fixed: json!({ "name": "fixed" }), + }); + let value = run_agent(&node, &caps_with_llm(llm)).await; + assert_eq!(value, json!({ "name": "fixed" })); + } + + #[tokio::test] + async fn output_parser_sub_port_errors_when_unfixable() { + let node = agent_node(json!({ + "prompt": "hi", + "output_parser": { "schema": { "type": "object", "required": ["name"] } } + })); + // Completion invalid; "fix" still invalid → the node surfaces an error. + let llm = Arc::new(ParserLlm { + completion: json!({ "wrong": 1 }), + fixed: json!({ "still": "wrong" }), + }); + let input: Vec = vec![]; + let run_meta = Value::Null; + let caps = caps_with_llm(llm); + let ctx = NodeContext { + node: &node, + input: &input, + run: &run_meta, + caps: &caps, + }; + let err = AgentNode + .execute(ctx) + .await + .expect_err("unfixable output must error"); + assert!(matches!(err, crate::error::EngineError::Capability(_))); + } + + #[tokio::test] + async fn plain_agent_without_sub_ports_is_unchanged() { + // Back-compat: no tools / output_parser configured ⇒ the completion is + // emitted verbatim (the mock echoes the request under `completion`). + let node = agent_node(json!({ "prompt": "hi" })); + let value = run_agent(&node, &mock_capabilities()).await; + assert_eq!(value["completion"]["prompt"], "hi"); + assert!(value.get("tool_result").is_none()); + } } diff --git a/src/nodes/integration/mod.rs b/src/nodes/integration/mod.rs index ffb64e4..2d089e3 100644 --- a/src/nodes/integration/mod.rs +++ b/src/nodes/integration/mod.rs @@ -8,6 +8,7 @@ pub mod agent; pub mod code; pub mod http_request; pub mod output_parser; +pub(crate) mod schema; pub mod sub_workflow; pub mod tool_call; diff --git a/src/nodes/integration/output_parser.rs b/src/nodes/integration/output_parser.rs index e5e34b0..8a56dc5 100644 --- a/src/nodes/integration/output_parser.rs +++ b/src/nodes/integration/output_parser.rs @@ -1,23 +1,57 @@ //! The `output_parser` node: structures/validates an agent's output. use async_trait::async_trait; +use serde_json::Value; +use crate::data::Item; use crate::error::Result; +use crate::nodes::integration::schema; use crate::nodes::{NodeContext, NodeExecutor, NodeOutput}; /// Parses / validates an upstream agent's output into a structured shape. /// -/// For now this is an identity passthrough of its input items. Structured -/// schema validation and LLM auto-fixing are later refinements. +/// When the node's `config.schema` holds a (subset-of-)JSON-Schema, each input +/// item's `json` is validated against it; on failure the node makes one LLM +/// auto-fix attempt (via the injected [`crate::caps::LlmProvider`]) and +/// re-validates, emitting the repaired value. A value that still fails — or fails +/// with `config.auto_fix == false` — surfaces a capability error, which the +/// engine routes per the node's `on_error` policy (`stop` / `continue` / +/// `route`). See [`schema`] for the supported schema subset. +/// +/// Config: +/// - `schema` — the JSON Schema to validate against. Omitted / null ⇒ the node is +/// an identity passthrough (back-compat with the pre-validation behavior). +/// - `auto_fix` — whether to attempt the one-shot LLM repair (default `true`). +/// - `connection_ref` — optional opaque credential id for the auto-fix LLM call. #[derive(Debug, Default, Clone)] pub struct OutputParserNode; #[async_trait] impl NodeExecutor for OutputParserNode { async fn execute(&self, ctx: NodeContext<'_>) -> Result { - // A3-basic: pass the upstream items through unchanged. Structured-schema - // validation and LLM auto-fixing are later refinements. - Ok(NodeOutput::main(ctx.input.to_vec())) + let cfg = &ctx.node.config; + // Back-compat: with no schema configured the node is an identity + // passthrough of its input items. + let schema_val = match cfg.get("schema") { + Some(s) if !s.is_null() => s, + _ => return Ok(NodeOutput::main(ctx.input.to_vec())), + }; + let auto_fix = cfg.get("auto_fix").and_then(Value::as_bool).unwrap_or(true); + let conn = cfg.get("connection_ref").and_then(Value::as_str); + + let mut out = Vec::with_capacity(ctx.input.len()); + for item in ctx.input { + let validated = schema::parse_and_validate( + item.json.clone(), + schema_val, + auto_fix, + &ctx.caps.llm, + conn, + ) + .await?; + out.push(Item::new(validated)); + } + Ok(NodeOutput::main(out)) } } @@ -87,4 +121,165 @@ mod tests { async fn empty_input_yields_no_items() { assert!(run_parser(vec![]).await.is_empty()); } + + // --- schema validation + LLM auto-fix --- + + use crate::caps::{Capabilities, LlmProvider}; + use async_trait::async_trait; + use std::sync::Arc; + + /// An LLM that returns a canned corrected value under `value` on auto-fix. + struct FixingLlm(Value); + + #[async_trait] + impl LlmProvider for FixingLlm { + async fn complete(&self, _request: Value, _conn: Option<&str>) -> Result { + Ok(json!({ "value": self.0.clone() })) + } + } + + /// Builds a capabilities bundle whose LLM is `llm`, everything else mocked. + fn caps_with_llm(llm: Arc) -> Capabilities { + let mut caps = mock_capabilities(); + caps.llm = llm; + caps + } + + fn schema_node(config: Value) -> Node { + Node { + id: "p".into(), + kind: NodeKind::OutputParser, + type_version: 1, + name: "p".into(), + config, + ports: vec![], + position: None, + } + } + + async fn run_with_caps( + node: &Node, + input: Vec, + caps: &crate::caps::Capabilities, + ) -> Result> { + let run_meta = Value::Null; + let ctx = NodeContext { + node, + input: &input, + run: &run_meta, + caps, + }; + OutputParserNode.execute(ctx).await.map(|o| o.items) + } + + #[tokio::test] + async fn valid_input_passes_schema_unchanged() { + let node = schema_node(json!({ + "schema": { "type": "object", "required": ["name"] } + })); + let input = vec![Item::new(json!({ "name": "A" }))]; + let caps = mock_capabilities(); + let out = run_with_caps(&node, input.clone(), &caps) + .await + .expect("valid"); + assert_eq!(out, input); + } + + #[tokio::test] + async fn invalid_input_is_repaired_by_auto_fix() { + let node = schema_node(json!({ + "schema": { "type": "object", "required": ["name"] } + })); + let input = vec![Item::new(json!({ "wrong": 1 }))]; + let caps = caps_with_llm(Arc::new(FixingLlm(json!({ "name": "fixed" })))); + let out = run_with_caps(&node, input, &caps).await.expect("auto-fix"); + assert_eq!(out, vec![Item::new(json!({ "name": "fixed" }))]); + } + + #[tokio::test] + async fn unfixable_input_errors() { + // The default mock LLM echoes the request, so its "fix" never satisfies + // the schema — the node surfaces a capability error. + let node = schema_node(json!({ + "schema": { "type": "object", "required": ["name"] } + })); + let input = vec![Item::new(json!({ "wrong": 1 }))]; + let caps = mock_capabilities(); + let err = run_with_caps(&node, input, &caps) + .await + .expect_err("unfixable input must error"); + assert!(matches!(err, crate::error::EngineError::Capability(_))); + } + + #[tokio::test] + async fn auto_fix_disabled_errors_on_invalid() { + let node = schema_node(json!({ + "schema": { "type": "object", "required": ["name"] }, + "auto_fix": false + })); + let input = vec![Item::new(json!({ "wrong": 1 }))]; + let caps = caps_with_llm(Arc::new(FixingLlm(json!({ "name": "fixed" })))); + let err = run_with_caps(&node, input, &caps) + .await + .expect_err("auto_fix=false must error"); + assert!(matches!(err, crate::error::EngineError::Capability(_))); + } + + // --- end-to-end through the engine: invalid input routes per on_error --- + + use crate::compiler::compile; + use crate::engine::run; + use crate::model::{Edge, WorkflowGraph}; + + #[tokio::test] + async fn engine_routes_unfixable_output_parser_error_via_on_error_continue() { + // trigger -> output_parser (schema requires `name`, on_error: continue). + // The seeded trigger item lacks `name`; the echo LLM can't fix it, so the + // failure becomes an error item on the default port rather than failing + // the run. + let graph = WorkflowGraph { + nodes: vec![ + Node { + id: "t".into(), + kind: NodeKind::Trigger, + type_version: 1, + name: "t".into(), + config: Value::Null, + ports: vec![], + position: None, + }, + Node { + id: "p".into(), + kind: NodeKind::OutputParser, + type_version: 1, + name: "p".into(), + config: json!({ + "schema": { "type": "object", "required": ["name"] }, + "on_error": "continue" + }), + ports: vec![], + position: None, + }, + ], + edges: vec![Edge { + from_node: "t".into(), + from_port: "main".into(), + to_node: "p".into(), + to_port: "main".into(), + }], + ..Default::default() + }; + let compiled = compile(&graph).expect("compile"); + let out = run(&compiled, json!({ "wrong": 1 }), &mock_capabilities()) + .await + .expect("run completes via on_error"); + let err = &out.output["nodes"]["p"]["items"][0]["json"]["error"]; + assert_eq!(err["node"], "p"); + assert!( + err["message"] + .as_str() + .unwrap() + .contains("schema validation") + ); + } } diff --git a/src/nodes/integration/schema.rs b/src/nodes/integration/schema.rs new file mode 100644 index 0000000..68c2fe4 --- /dev/null +++ b/src/nodes/integration/schema.rs @@ -0,0 +1,299 @@ +//! Minimal JSON-Schema validation plus a one-shot LLM auto-fix. +//! +//! Shared by the [`output_parser`](super::output_parser) node and the +//! [`agent`](super::agent) node's output-parser sub-port. The validator covers a +//! deliberately small, dependency-free subset of JSON Schema — enough to give the +//! parser real teeth without pulling a heavyweight schema crate (which would also +//! not be a host capability, so it must stay in-crate and light): +//! +//! - `type` — one of `object` / `array` / `string` / `number` / `integer` / +//! `boolean` / `null`, or an array of those (any-of); +//! - `required` — required property names on an object; +//! - `properties` — per-property subschemas (recursed when the property exists); +//! - `items` — a subschema applied to every array element; +//! - `enum` — an explicit set of allowed values. +//! +//! Unknown keywords are ignored (they never fail validation), and a non-object +//! schema (e.g. `true`) accepts anything. Anything richer — `$ref`, `oneOf`, +//! numeric bounds, string patterns — is a documented follow-up. + +use std::sync::Arc; + +use serde_json::{Value, json}; + +use crate::caps::LlmProvider; +use crate::error::{EngineError, Result}; + +/// Validates `value` against the supported subset of JSON Schema, returning a +/// list of human-readable error messages. An empty list means the value is +/// valid. +#[must_use] +pub(crate) fn validate(value: &Value, schema: &Value) -> Vec { + let mut errors = Vec::new(); + validate_at("$", value, schema, &mut errors); + errors +} + +/// Recursively validates `value` at JSON-path `path` against `schema`, appending +/// any failures to `errors`. +fn validate_at(path: &str, value: &Value, schema: &Value, errors: &mut Vec) { + // A non-object schema (e.g. the boolean `true`) accepts anything. + let Some(obj) = schema.as_object() else { + return; + }; + + // `enum`: the value must be one of the listed values (by JSON equality). + if let Some(Value::Array(allowed)) = obj.get("enum") { + if !allowed.iter().any(|a| a == value) { + errors.push(format!( + "{path}: value is not one of the allowed `enum` values" + )); + } + } + + // `type`: a single type name or an array of acceptable type names. + if let Some(ty) = obj.get("type") { + let types: Vec<&str> = match ty { + Value::String(s) => vec![s.as_str()], + Value::Array(a) => a.iter().filter_map(Value::as_str).collect(), + _ => vec![], + }; + if !types.is_empty() && !types.iter().any(|t| type_matches(t, value)) { + errors.push(format!( + "{path}: expected type {types:?}, got `{}`", + type_name(value) + )); + // The value is the wrong shape; deeper structural checks below would + // only produce noise, so stop here for this node. + return; + } + } + + // Object constraints: required properties and per-property subschemas. + if let Value::Object(map) = value { + if let Some(Value::Array(required)) = obj.get("required") { + for name in required.iter().filter_map(Value::as_str) { + if !map.contains_key(name) { + errors.push(format!("{path}: missing required property `{name}`")); + } + } + } + if let Some(Value::Object(props)) = obj.get("properties") { + for (key, subschema) in props { + if let Some(child) = map.get(key) { + validate_at(&format!("{path}.{key}"), child, subschema, errors); + } + } + } + } + + // Array constraints: a single `items` subschema applied to each element. + if let Value::Array(arr) = value { + if let Some(items_schema) = obj.get("items") { + for (i, child) in arr.iter().enumerate() { + validate_at(&format!("{path}[{i}]"), child, items_schema, errors); + } + } + } +} + +/// Whether `value` satisfies the JSON-Schema `type` name `ty`. An unknown type +/// keyword is treated as satisfied (we never fail on keywords we don't model). +fn type_matches(ty: &str, value: &Value) -> bool { + match ty { + "object" => value.is_object(), + "array" => value.is_array(), + "string" => value.is_string(), + "number" => value.is_number(), + // JSON Schema treats `1.0` as an integer; accept whole-valued floats too. + "integer" => { + value.is_i64() || value.is_u64() || value.as_f64().is_some_and(|f| f.fract() == 0.0) + } + "boolean" => value.is_boolean(), + "null" => value.is_null(), + _ => true, + } +} + +/// The JSON type name of `value`, for error messages. +fn type_name(value: &Value) -> &'static str { + match value { + Value::Null => "null", + Value::Bool(_) => "boolean", + Value::Number(_) => "number", + Value::String(_) => "string", + Value::Array(_) => "array", + Value::Object(_) => "object", + } +} + +/// Validates `value` against `schema`; on failure, optionally makes a single LLM +/// auto-fix attempt and re-validates. +/// +/// Returns the validated (possibly repaired) value on success. When validation +/// fails and `auto_fix` is set, the injected [`LlmProvider`] is asked once to +/// coerce the value to the schema — the request carries the `schema`, the +/// offending `value`, and the validation `errors`, and the corrected value is +/// read back from the response's `value` field (or the whole response when it has +/// no such field). If it still fails to validate — or `auto_fix` is off — a +/// [`EngineError::Capability`] describing the failures is returned, which the +/// engine then routes per the node's `on_error` policy. +pub(crate) async fn parse_and_validate( + value: Value, + schema: &Value, + auto_fix: bool, + llm: &Arc, + conn: Option<&str>, +) -> Result { + let errors = validate(&value, schema); + if errors.is_empty() { + return Ok(value); + } + if !auto_fix { + return Err(EngineError::Capability(format!( + "output_parser: value failed schema validation: {}", + errors.join("; ") + ))); + } + + tracing::debug!( + error_count = errors.len(), + "output_parser: schema validation failed; attempting one LLM auto-fix" + ); + + let request = json!({ + "task": "coerce_to_schema", + "schema": schema, + "value": value, + "errors": errors, + }); + let response = llm.complete(request, conn).await?; + // The corrected value is the response's `value` field when present, else the + // whole response body. + let fixed = response + .get("value") + .cloned() + .unwrap_or_else(|| response.clone()); + + let remaining = validate(&fixed, schema); + if remaining.is_empty() { + tracing::debug!("output_parser: LLM auto-fix produced a schema-valid value"); + Ok(fixed) + } else { + Err(EngineError::Capability(format!( + "output_parser: value failed schema validation after auto-fix: {}", + remaining.join("; ") + ))) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use async_trait::async_trait; + + #[test] + fn no_schema_object_accepts_anything() { + // A boolean/non-object schema imposes no constraints. + assert!(validate(&json!({"a": 1}), &Value::Bool(true)).is_empty()); + } + + #[test] + fn type_and_required_and_properties() { + let schema = json!({ + "type": "object", + "required": ["name", "age"], + "properties": { + "name": { "type": "string" }, + "age": { "type": "integer" } + } + }); + assert!(validate(&json!({"name": "A", "age": 3}), &schema).is_empty()); + + let missing = validate(&json!({"name": "A"}), &schema); + assert_eq!(missing.len(), 1); + assert!(missing[0].contains("missing required property `age`")); + + let wrong_type = validate(&json!({"name": "A", "age": "old"}), &schema); + assert!(wrong_type.iter().any(|e| e.contains("expected type"))); + } + + #[test] + fn integer_accepts_whole_floats() { + let schema = json!({ "type": "integer" }); + assert!(validate(&json!(3.0), &schema).is_empty()); + assert!(!validate(&json!(3.5), &schema).is_empty()); + } + + #[test] + fn enum_and_array_items() { + let schema = json!({ + "type": "array", + "items": { "enum": ["a", "b"] } + }); + assert!(validate(&json!(["a", "b", "a"]), &schema).is_empty()); + assert!(!validate(&json!(["a", "z"]), &schema).is_empty()); + } + + /// An LLM that returns a fixed value under `value` on the auto-fix call. + struct FixingLlm(Value); + + #[async_trait] + impl LlmProvider for FixingLlm { + async fn complete(&self, _request: Value, _conn: Option<&str>) -> Result { + Ok(json!({ "value": self.0.clone() })) + } + } + + /// An LLM whose "fix" is still invalid. + struct UselessLlm; + + #[async_trait] + impl LlmProvider for UselessLlm { + async fn complete(&self, _request: Value, _conn: Option<&str>) -> Result { + Ok(json!({ "value": { "still": "wrong" } })) + } + } + + #[tokio::test] + async fn valid_value_passes_without_calling_llm() { + let schema = json!({ "type": "object", "required": ["ok"] }); + let llm: Arc = Arc::new(UselessLlm); + let out = parse_and_validate(json!({ "ok": true }), &schema, true, &llm, None) + .await + .expect("valid value passes"); + assert_eq!(out, json!({ "ok": true })); + } + + #[tokio::test] + async fn invalid_value_is_repaired_by_auto_fix() { + let schema = json!({ "type": "object", "required": ["name"] }); + let llm: Arc = Arc::new(FixingLlm(json!({ "name": "fixed" }))); + let out = parse_and_validate(json!({ "wrong": 1 }), &schema, true, &llm, None) + .await + .expect("auto-fix repairs the value"); + assert_eq!(out, json!({ "name": "fixed" })); + } + + #[tokio::test] + async fn unfixable_value_errors() { + let schema = json!({ "type": "object", "required": ["name"] }); + let llm: Arc = Arc::new(UselessLlm); + let err = parse_and_validate(json!({ "wrong": 1 }), &schema, true, &llm, None) + .await + .expect_err("unfixable value must error"); + assert!(matches!(err, EngineError::Capability(ref m) if m.contains("after auto-fix"))); + } + + #[tokio::test] + async fn auto_fix_disabled_errors_immediately() { + let schema = json!({ "type": "object", "required": ["name"] }); + let llm: Arc = Arc::new(FixingLlm(json!({ "name": "fixed" }))); + let err = parse_and_validate(json!({ "wrong": 1 }), &schema, false, &llm, None) + .await + .expect_err("auto-fix disabled must error"); + assert!( + matches!(err, EngineError::Capability(ref m) if m.contains("failed schema validation")) + ); + } +} From b7c39c4c5bd26f915aaafa6dbb4dd38e525c111c Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 5 Jul 2026 10:46:13 -0700 Subject: [PATCH 5/8] fix(engine): allow too_many_arguments on observed resume fns (clippy -D warnings) The observer parameter pushed resume_with_checkpointer_journaled_observed and resume_with_checkpointer_inner to 8 args; the SDK CI runs `cargo clippy --all-targets -- -D warnings`. Suppress consistently with the existing allow at engine.rs:851. No behavior change. Claude-Session: https://claude.ai/code/session_01QpsTTxxjFgDhzuozS28tyU --- src/engine.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/engine.rs b/src/engine.rs index 39f555f..1be5ea4 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -1365,6 +1365,7 @@ pub async fn resume_with_checkpointer_journaled( /// /// # Errors /// Same as [`resume_with_checkpointer_journaled`]. +#[allow(clippy::too_many_arguments)] pub async fn resume_with_checkpointer_journaled_observed( workflow: &CompiledWorkflow, capabilities: &Capabilities, @@ -1401,6 +1402,7 @@ pub async fn resume_with_checkpointer_journaled_observed( /// (optionally journaled), re-attaches the same `checkpointer`, and resumes /// `thread_id`. Returns the outcome plus the resumed execution's /// `tinyagents`-minted run ids. +#[allow(clippy::too_many_arguments)] async fn resume_with_checkpointer_inner( workflow: &CompiledWorkflow, capabilities: &Capabilities, From dcde8fab5e14f103147498346dee720801f718b3 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 5 Jul 2026 10:56:16 -0700 Subject: [PATCH 6/8] chore(engine): commit Cargo.lock for 0.5.0 version bump Cargo.toml was bumped to 0.5.0 but Cargo.lock's tinyflows entry still read 0.4.0. Sync it so the lockfile matches the manifest. Claude-Session: https://claude.ai/code/session_01QpsTTxxjFgDhzuozS28tyU --- Cargo.lock | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Cargo.lock b/Cargo.lock index cb409b0..086f31d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1355,7 +1355,7 @@ dependencies = [ [[package]] name = "tinyflows" -version = "0.4.0" +version = "0.5.0" dependencies = [ "async-trait", "futures-timer", From 3497de262cdc15e0dc140d67adb2a6c102d00563 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 5 Jul 2026 10:56:17 -0700 Subject: [PATCH 7/8] fix(agent): do not trust model-supplied tool credentials (Codex P1) The agent tool sub-port took `tool_call.connection_ref` straight from the LLM response, so a prompt-injection or malformed model reply could select an arbitrary host credential id for the tool call. Credentials now come from trusted config only: the offered tool descriptor's `connection_ref`, falling back to the node's `connection_ref`. The model-supplied value is ignored. Adds regression tests asserting the offered descriptor's / node's connection wins over a hostile `tool_call.connection_ref`. Claude-Session: https://claude.ai/code/session_01QpsTTxxjFgDhzuozS28tyU --- src/nodes/integration/agent.rs | 61 +++++++++++++++++++++++++++++++--- 1 file changed, 56 insertions(+), 5 deletions(-) diff --git a/src/nodes/integration/agent.rs b/src/nodes/integration/agent.rs index 342b9a3..82d9cce 100644 --- a/src/nodes/integration/agent.rs +++ b/src/nodes/integration/agent.rs @@ -56,18 +56,25 @@ impl NodeExecutor for AgentNode { let mut value = response; if let Some(tool_call) = value.get("tool_call").cloned() { if let Some(slug) = tool_call.get("slug").and_then(Value::as_str) { - let offered = cfg + // Only a tool actually offered in `config.tools` may be invoked; + // keep the matched descriptor so its trusted `connection_ref` wins. + let offered_tool = cfg .get("tools") .and_then(Value::as_array) - .is_some_and(|tools| { + .and_then(|tools| { tools .iter() - .any(|t| t.get("slug").and_then(Value::as_str) == Some(slug)) + .find(|t| t.get("slug").and_then(Value::as_str) == Some(slug)) }); - if offered { + if let Some(offered_tool) = offered_tool { tracing::debug!(slug, "agent tool sub-port: invoking model-elected tool"); let args = tool_call.get("args").cloned().unwrap_or(Value::Null); - let tool_conn = tool_call + // Credentials come from trusted config only: the offered tool + // descriptor's `connection_ref`, else the node's. The model's + // `tool_call.connection_ref` is deliberately NOT trusted — a + // prompt-injection could otherwise elect an arbitrary host + // credential id for the call. + let tool_conn = offered_tool .get("connection_ref") .and_then(Value::as_str) .or(conn); @@ -331,6 +338,50 @@ mod tests { assert!(value.get("tool_result").is_none()); } + #[tokio::test] + async fn tool_sub_port_ignores_model_supplied_connection_ref() { + // Security: a model-supplied `tool_call.connection_ref` must NOT be + // trusted (prompt-injection could otherwise select an arbitrary host + // credential). The credential comes from the offered tool descriptor's + // `connection_ref` when present, else the node's `connection_ref`. + let node = agent_node(json!({ + "prompt": "do it", + "connection_ref": "node_acct", + "tools": [{ "slug": "slack.post", "connection_ref": "trusted_acct" }] + })); + let llm = Arc::new(ToolCallingLlm(json!({ + "tool_call": { + "slug": "slack.post", + "args": { "text": "hi" }, + "connection_ref": "attacker_acct" + } + }))); + let value = run_agent(&node, &caps_with_llm(llm)).await; + // The mock ToolInvoker echoes the `conn` it was invoked with: it must be + // the offered descriptor's trusted id, never the model-supplied one. + assert_eq!(value["tool_result"]["connection"], "trusted_acct"); + } + + #[tokio::test] + async fn tool_sub_port_falls_back_to_node_connection_ref() { + // When the offered tool descriptor carries no `connection_ref`, the node's + // `connection_ref` is used — still never the model-supplied one. + let node = agent_node(json!({ + "prompt": "do it", + "connection_ref": "node_acct", + "tools": [{ "slug": "slack.post" }] + })); + let llm = Arc::new(ToolCallingLlm(json!({ + "tool_call": { + "slug": "slack.post", + "args": { "text": "hi" }, + "connection_ref": "attacker_acct" + } + }))); + let value = run_agent(&node, &caps_with_llm(llm)).await; + assert_eq!(value["tool_result"]["connection"], "node_acct"); + } + /// An LLM that returns an invalid completion, but a schema-valid value when /// asked to coerce (the auto-fix call carries `task == "coerce_to_schema"`). struct ParserLlm { From fccd16b1738f57c3e5466ed558620871682963ed Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 5 Jul 2026 10:56:17 -0700 Subject: [PATCH 8/8] fix(engine): honor per-gate approvals and fan-out denial routing (Codex P1/P2) Two resume-path fixes flagged in review: 1. Keep unlisted approval gates pending (P1). tinyagents ignores the `with_update` state write on resume, so the resume value is the sole approval channel. The handler previously treated any `ctx.resume.is_some()` as approval, so resuming to approve/deny one gate would also run every other interrupted gate. Both resume paths now deliver the explicit `approved` gate-id list (checkpointed path also carries `rejected`), and a gate proceeds only if it is a bare-`true` resume (back-compat) or listed in `approved`; gates in neither stay pending. 2. Route denied fan-outs with a command (P2). When a denied gate's `error` port fans out to >=2 recovery nodes, the node is command-routed and has no conditional router; the denial branch emitted a plain (unrouted) update so no recovery branch ran. It now uses the `emit` fan-out command path, which drives all error-port successors (and still falls back to a plain update for the single/mixed-port case). Adds regression tests for both (a parallel-gate partial approval leaving the other gate pending, and a fan-out error-recovery denial). Claude-Session: https://claude.ai/code/session_01QpsTTxxjFgDhzuozS28tyU --- src/engine.rs | 202 ++++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 178 insertions(+), 24 deletions(-) diff --git a/src/engine.rs b/src/engine.rs index 1be5ea4..2d7a72f 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -485,9 +485,26 @@ fn build_graph( // `{ "rejected": [, …] }` denies the named gate(s). let resume_value = ctx.resume.clone(); // A checkpointed resume (see `ResumableRun::resume`) delivers a resume - // value to the interrupted node via `NodeContext::resume`; its presence - // approves this gate so the run proceeds forward from the checkpoint. - let resumed = ctx.resume.is_some(); + // value to the interrupted node via `NodeContext::resume`. A resume + // approves *this* gate only when it is a bare `true` (backward-compat, + // the single-interrupt case) or when this gate's id is explicitly + // listed in the structured resume value's `approved` array. A + // structured resume that names neither this gate in `approved` nor in + // `rejected` leaves it pending — critical when several parallel gates + // are interrupted and the host resolves only some of them. + let approved_by_resume = match ctx.resume.as_ref() { + Some(Value::Bool(true)) => true, + Some(v) => v + .get("approved") + .and_then(Value::as_array) + .is_some_and(|approved| { + approved + .iter() + .filter_map(Value::as_str) + .any(|id| id == node.id) + }), + None => false, + }; async move { // Wraps a node's partial update: a fan-out node drives all of its // successors via a `Command::goto`, everything else emits a plain @@ -559,13 +576,14 @@ fn build_graph( })); if has_error_edge { // Route the denial to the `error` port so a recovery - // sub-graph runs. A plain update (not `emit`) so the - // conditional-edge router keys on the recorded port. - return Ok(NodeResult::Update(items_update( - &node.id, - &[item], - Some("error"), - )?)); + // sub-graph runs. Use `emit`: when the gate's error-port + // recovery edges fan out (≥2 same-port successors) the + // node is command-routed and has no conditional router to + // key on the recorded port, so the branches must be driven + // directly via a `Command::goto`; a single/mixed-port error + // edge falls back to a plain update the conditional-edge + // router consumes. + return Ok(emit(items_update(&node.id, &[item], Some("error"))?)); } // No error branch to route to — fail the run so the denial // is not silently swallowed. @@ -585,9 +603,10 @@ fn build_graph( .filter_map(Value::as_str) .any(|id| id == node.id) }); - // `resumed` is set when a checkpointed resume delivered a - // resume value to this interrupted gate — that approves it. - if !approved && !resumed { + // `approved_by_resume` is set when a checkpointed resume + // delivered an approval (bare `true`, or this gate listed in + // the structured `approved` array) to this interrupted gate. + if !approved && !approved_by_resume { tracing::info!(node = %node.id, "node paused awaiting approval"); let payload = if node.config.is_null() { json!({}) @@ -1080,13 +1099,19 @@ impl ResumableRun { /// example, when there is no pending checkpoint to resume from). pub async fn resume(&self, newly_approved: Vec) -> Result { let approvals_update = json!({ - "run": { "trigger": { "approvals": newly_approved } } + "run": { "trigger": { "approvals": newly_approved.clone() } } }); + // Deliver the explicit `approved` gate id list as the resume value. + // tinyagents ignores the `with_update` state write on resume, so the + // resume value is the sole approval channel: each interrupted gate + // proceeds only if its id is listed, leaving any other parallel gate + // pending rather than blanket-approving every interrupt with a bare `true`. let execution = self .graph .resume( self.thread_id.as_str(), - Command::resume(json!(true)).with_update(approvals_update), + Command::resume(json!({ "approved": newly_approved })) + .with_update(approvals_update), ) .await .map_err(|e| EngineError::Capability(e.to_string()))?; @@ -1437,16 +1462,18 @@ async fn resume_with_checkpointer_inner( let approvals_update = json!({ "run": { "trigger": { "approvals": newly_approved.clone() } } }); - // The resume value delivered to the interrupted gate(s): a bare `true` - // (backward-compatible bare approval) when nothing was denied, otherwise a - // structured value carrying the denied gate ids so the gate handlers can - // route/fail them (see the `denied` branch in `build_graph`). - let resume_value = if rejected.is_empty() { - json!(true) - } else { + if !rejected.is_empty() { tracing::info!(?rejected, "resuming with denied approval gate(s)"); - json!({ "approved": newly_approved, "rejected": rejected }) - }; + } + // Always deliver a structured resume value carrying the explicit `approved` + // and `rejected` gate id lists. tinyagents ignores the `with_update` state + // write on resume, so this value is the sole approval channel and each + // interrupted gate decides for itself: gates in `approved` proceed, gates in + // `rejected` route to their `error` port (or fail), and gates in neither stay + // pending. This is essential when several parallel gates are interrupted and + // the host resolves only some of them — a bare `true` would blanket-approve + // every interrupt regardless of the host's decision. + let resume_value = json!({ "approved": newly_approved, "rejected": rejected }); let execution = compiled .resume( thread_id, @@ -3176,6 +3203,133 @@ mod tests { ); } + #[tokio::test] + async fn parallel_gates_resume_one_leaves_the_other_pending() { + // trigger fans out to two parallel gates g1 and g2 (both on the `main` + // port), each feeding its own downstream. Resuming with only g1 approved + // must run g1's downstream while g2 — listed in neither `approved` nor + // `rejected` — stays pending and its downstream stays blocked. A bare + // `true` resume value would blanket-approve g2 too and wrongly run d2. + let graph = WorkflowGraph { + nodes: vec![ + node("t", NodeKind::Trigger), + gate("g1"), + gate("g2"), + node("d1", NodeKind::OutputParser), + node("d2", NodeKind::OutputParser), + ], + edges: vec![ + edge("t", "g1"), + edge("t", "g2"), + edge("g1", "d1"), + edge("g2", "d2"), + ], + ..Default::default() + }; + let compiled = compile(&graph).expect("compile"); + let caps = mock_capabilities(); + + // The engine serializes interrupts across the fan-out: g1 is the first + // gate to pend; g2 pends only once g1 is resolved. The invariant this test + // guards is that approving g1 must NOT also approve g2 — a bare `true` + // resume value would blanket-approve every interrupted gate. + let rr = run_resumable(&compiled, json!({}), &caps) + .await + .expect("run_resumable"); + assert_eq!( + rr.outcome().pending_approvals, + vec!["g1".to_string()], + "g1 is the first parallel gate to pend" + ); + + let after_g1 = rr.resume(vec!["g1".to_string()]).await.expect("resume g1"); + assert!( + after_g1.pending_approvals.contains(&"g2".to_string()), + "g2 must stay pending when only g1 is approved (a bare-true resume \ + would wrongly blanket-approve it), got {:?}", + after_g1.pending_approvals + ); + assert!( + after_g1.output["nodes"]["d2"].is_null(), + "g2's downstream must NOT run while g2 is still pending" + ); + + // Resolving g2 too settles the run: no gate remains pending and g2's + // downstream finally runs. + let after_g2 = rr.resume(vec!["g2".to_string()]).await.expect("resume g2"); + assert!( + after_g2.pending_approvals.is_empty(), + "no gate pending once both parallel gates are approved, got {:?}", + after_g2.pending_approvals + ); + assert!( + !after_g2.output["nodes"]["d2"]["items"].is_null(), + "g2's downstream runs once g2 is approved" + ); + } + + #[tokio::test] + async fn resume_denying_a_gate_fans_out_to_multiple_error_recovery_nodes() { + // A denied gate whose `error` port fans out to TWO recovery nodes (≥2 + // edges on the same port) is command-routed and has no conditional router; + // the denial must still drive BOTH recovery branches via the fan-out + // command path rather than a plain (unrouted) update. + let cp: Arc> = Arc::new(InMemoryCheckpointer::::default()); + let mut gate = node("gate", NodeKind::OutputParser); + gate.config = json!({ "requires_approval": true }); + let graph = WorkflowGraph { + nodes: vec![ + node("t", NodeKind::Trigger), + gate, + node("recover_a", NodeKind::OutputParser), + node("recover_b", NodeKind::OutputParser), + ], + edges: vec![ + edge("t", "gate"), + port_edge("gate", "error", "recover_a"), + port_edge("gate", "error", "recover_b"), + ], + ..Default::default() + }; + let compiled = compile(&graph).expect("compile"); + + let caps = mock_capabilities(); + let paused = run_with_checkpointer( + &compiled, + json!({}), + &caps, + cp.clone(), + "thread-fanout-deny", + ) + .await + .expect("run_with_checkpointer"); + assert_eq!(paused.pending_approvals, vec!["gate".to_string()]); + + let caps = mock_capabilities(); + let journal = Arc::new(InMemoryGraphEventJournal::new()); + let observer = Arc::new(crate::observability::NoopObserver) as Arc; + let denied = resume_with_checkpointer_journaled_observed( + &compiled, + &caps, + cp.clone(), + "thread-fanout-deny", + Vec::new(), + vec!["gate".to_string()], + journal, + &observer, + ) + .await + .expect("resume with rejection"); + + for recovery in ["recover_a", "recover_b"] { + assert_eq!( + denied.outcome.output["nodes"][recovery]["items"][0]["json"]["error"]["node"], + json!("gate"), + "both fan-out error-recovery branches must run on denial: {recovery}" + ); + } + } + #[tokio::test] async fn durable_resume_with_journal_surfaces_resume_observations() { // Same durable resume path as above, but with a graph event journal attached