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 6009fa1..086f31d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1355,7 +1355,7 @@ dependencies = [ [[package]] name = "tinyflows" -version = "0.3.0" +version = "0.5.0" dependencies = [ "async-trait", "futures-timer", diff --git a/Cargo.toml b/Cargo.toml index 12b5531..f7f3743 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "tinyflows" -version = "0.3.0" +version = "0.5.0" edition = "2024" rust-version = "1.85" license = "GPL-3.0-or-later" diff --git a/README.md b/README.md index a0af000..054d1e9 100644 --- a/README.md +++ b/README.md @@ -210,11 +210,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 16c2591..2d7a72f 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. @@ -221,6 +263,91 @@ pub async fn run_with_observer( checkpointer, 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) +} + +/// 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 })), + CancellationToken::new(), ) .await?; Ok(outcome) @@ -252,6 +379,7 @@ fn build_graph( steps: &Arc>>, checkpointer: Arc>, journal: Option>, + token: CancellationToken, ) -> Result<(CompiledGraph, String)> { let graph = &workflow.graph; @@ -332,9 +460,17 @@ 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); + // 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(); @@ -342,11 +478,33 @@ 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 + // `{ "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 @@ -358,6 +516,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). @@ -376,6 +547,51 @@ 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. 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. + 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")) @@ -387,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!({}) @@ -650,6 +867,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, @@ -658,6 +876,8 @@ async fn build_and_run( checkpointer: Arc>, 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)); @@ -678,12 +898,20 @@ async fn build_and_run( &steps, checkpointer, journal, + token.clone(), )?; let seed_items = items_update(&trigger_id, &[Item::new(input.clone())], None) .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 @@ -731,6 +959,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, )) @@ -787,6 +1018,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 @@ -827,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()))?; @@ -847,6 +1125,7 @@ impl ResumableRun { Ok(RunOutcome { output: execution.state, pending_approvals, + cancelled: false, }) } } @@ -879,6 +1158,8 @@ pub async fn run_resumable( checkpointer, thread_id, None, + None, + CancellationToken::new(), ) .await?; Ok(ResumableRun { @@ -927,6 +1208,8 @@ pub async fn run_with_checkpointer( checkpointer, thread_id.to_string(), None, + None, + CancellationToken::new(), ) .await?; Ok(outcome) @@ -967,13 +1250,16 @@ 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, checkpointer, thread_id, newly_approved, + Vec::new(), None, + &observer, ) .await?; Ok(outcome) @@ -1001,14 +1287,51 @@ 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), + None, + CancellationToken::new(), ) .await?; tracing::debug!( @@ -1038,6 +1361,45 @@ 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, + Vec::new(), + 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`]. +/// +/// `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`]. +#[allow(clippy::too_many_arguments)] +pub async fn resume_with_checkpointer_journaled_observed( + workflow: &CompiledWorkflow, + capabilities: &Capabilities, + checkpointer: Arc>, + thread_id: &str, + newly_approved: Vec, + rejected: Vec, + journal: Arc, + observer: &Arc, ) -> Result { let (outcome, graph_run_ids) = resume_with_checkpointer_inner( workflow, @@ -1045,7 +1407,9 @@ pub async fn resume_with_checkpointer_journaled( checkpointer, thread_id, newly_approved, + rejected, Some(journal), + observer, ) .await?; tracing::debug!( @@ -1063,26 +1427,31 @@ pub async fn resume_with_checkpointer_journaled( /// (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, checkpointer: Arc>, thread_id: &str, newly_approved: Vec, + rejected: 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, + CancellationToken::new(), )?; // Approvals recorded for downstream visibility. On resume the interrupted @@ -1091,12 +1460,24 @@ 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() } } }); + if !rejected.is_empty() { + tracing::info!(?rejected, "resuming with denied approval gate(s)"); + } + // 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, - 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()))?; @@ -1116,6 +1497,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, )) @@ -2714,6 +3098,238 @@ 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 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 @@ -2829,4 +3445,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..82d9cce 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,76 @@ 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) { + // 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) + .and_then(|tools| { + tools + .iter() + .find(|t| t.get("slug").and_then(Value::as_str) == Some(slug)) + }); + 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); + // 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); + 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 +261,199 @@ 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()); + } + + #[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 { + 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")) + ); + } +} 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