feat(engine): observer/deny + sub-workflow-by-id + output_parser validation + cancellation + agent sub-ports (0.3.2→0.5.0)

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`.

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"

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

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<String, WorkflowGraph>,
+}
+
+impl MockWorkflowResolver {
+    /// Registers `graph` under `id`, returning `self` for chaining.
+    #[must_use]
+    pub fn with(mut self, id: impl Into<String>, graph: WorkflowGraph) -> Self {
+        self.workflows.insert(id.into(), graph);
+        self
+    }
+}
+
+#[async_trait]
+impl WorkflowResolver for MockWorkflowResolver {
+    async fn resolve(&self, workflow_id: &str) -> Result<WorkflowGraph> {
+        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),
     }
 }
 

src/caps/mod.rs

@@ -64,6 +64,25 @@ pub trait CodeRunner: Send + Sync {
     async fn run(&self, language: CodeLanguage, source: &str, input: Value) -> Result<Value>;
 }
 
+/// 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<crate::model::WorkflowGraph>;
+}
+
 /// 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<dyn StateStore>,
+    /// 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<dyn WorkflowResolver>,
 }
 
 #[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]