diff --git a/CHANGELOG.md b/CHANGELOG.md index cd6cb897..17d7edf5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,44 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **`approval` node kind + the `ApprovalProvider` capability** — a + human-in-the-loop review step that carries what is being reviewed (a URL, a + draft, any payload) and routes on the answer: `approved` / `rejected` ports, + the verdict (reviewer, comment, any edit they made) emitted as an item and + readable anywhere as `=nodes..decision.approved`. Reaches the human + through the new optional `caps::ApprovalProvider`, whose `decide` is + create-or-fetch on a stable `request_id` so a resume or a poll never notifies + the reviewer twice. Hosts that wire no provider still get a working node: it + pauses the run and is settled with `engine::resume`, exactly like a + `requires_approval` gate. Waits by suspending by default, with an optional + bounded `wait_mode: "poll"`; `on_reject` (`route` / `error` / `drop`) and + `on_timeout` (`error` / `reject` / `route`) decide what a "no" does. + Note for hosts that build `caps::Capabilities` with a struct literal: it gains + an `approvals: Option>` field, so add + `approvals: None` (or a provider) to keep compiling. + +### Changed + +- **Approval provenance is preserved across a resume.** `RunInput::approvals` + (surfaced as `run.approvals`) now carries only the ids a host explicitly + authorised — those passed to `engine::resume` plus any already on that + channel. It is no longer seeded from `run.trigger.approvals`. + + The trigger is the payload a caller submits, so folding it into the + authorised list meant a caller-written id was promoted to trusted by the + first resume. `run.trigger.approvals` still receives the union and the + `requires_approval` gate still reads it, so that documented channel is + unchanged; what changed is that trigger-origin ids no longer cross into the + explicit one. A host that relied on `RunInput::approvals` echoing ids it had + written into the trigger payload must now pass them to `engine::resume` (or + `RunInput::with_approvals`) instead. + + The `approval` node reads only the explicit channel, and a verdict object in + a resume value must name its review (`node_id` / `request_id`) — an + unaddressed `{"approved": true}` is ignored rather than settling whichever + review reads it, since one resume value is delivered to every interrupted + node. + - **`tinyflows::testkit` — testing, mocking, and live debugging for workflows.** Behind the default-off `testkit` feature; adds no dependencies. diff --git a/README.md b/README.md index 147c2a5c..983853b2 100644 --- a/README.md +++ b/README.md @@ -73,6 +73,11 @@ Rust 2024 · MSRV 1.85 · `#![forbid(unsafe_code)]` · GPL-3.0-or-later. approves and continues. A host can also drive durable, cross-process resume by injecting a `Checkpointer` via `engine::run_with_checkpointer` / `resume_with_checkpointer`. +- Human review as a graph step: an `approval` node carries what is being + reviewed (a URL, a draft, any payload), reaches the human through the + host-implemented `caps::ApprovalProvider`, and routes the verdict — reviewer, + comment, any edit they made — on its `approved` / `rejected` ports. With no + provider injected it degrades to the pause-and-resume gate above. - Observability via `tracing` plus a `RunObserver` hook and `Run` / `ExecutionStep` records. @@ -177,7 +182,7 @@ cargo run --example hello_workflow --features mock ## Examples -The crate ships seven runnable examples under [`examples/`](examples/). Each is +The crate ships eight runnable examples under [`examples/`](examples/). Each is gated on the `mock` cargo feature, so run them with: ```sh @@ -192,6 +197,7 @@ cargo run --example --features mock | `capability_pipeline` | A linear `http_request → code → agent → tool_call` pipeline through the host capability traits (mocked). | | `error_handling` | Per-node `retry` plus `on_error: "route"` recovering a failing node via its `error` port. | | `hitl_approval` | A `requires_approval` gate pauses the run (`pending_approvals`), then `run_resumable(...).resume(...)` continues from the checkpoint. | +| `hitl_review` | An `approval` node against a host-implemented `ApprovalProvider`: the run suspends, a "human" approves with an edit, and the resume takes the `approved` branch. | | `jq_expressions` | The jaq-backed jq engine in a `transform` node (e.g. `=.item.prices | add`). | Omitting `--features mock` is harmless: the demo body is @@ -316,7 +322,8 @@ Run all of them in one go: ```sh for ex in hello_workflow conditional_branch parallel_and_merge \ - capability_pipeline error_handling hitl_approval jq_expressions; do + capability_pipeline error_handling hitl_approval hitl_review \ + jq_expressions; do cargo run --example "$ex" --features mock done ``` diff --git a/examples/hitl_review.rs b/examples/hitl_review.rs new file mode 100644 index 00000000..2a931c39 --- /dev/null +++ b/examples/hitl_review.rs @@ -0,0 +1,189 @@ +#![forbid(unsafe_code)] + +//! Human review as a **step in the graph**: an `approval` node hands a URL to a +//! host-implemented review surface, the run pauses while nobody has answered, +//! and the branch it takes afterwards depends on what the human said. +//! +//! `DeskReview` below stands in for whatever real surface a host has — a Slack +//! card, an inbox row, a web queue. It shows the two things the +//! [`ApprovalProvider`](tinyflows::caps::ApprovalProvider) contract asks for: +//! **create-or-fetch** on `request_id`, so re-asking never notifies the reviewer +//! twice, and a decision that can carry the human's own edit. +//! +//! Run: cargo run --example hitl_review --features mock +#[cfg(feature = "mock")] +#[tokio::main(flavor = "current_thread")] +async fn main() { + use std::collections::HashMap; + use std::sync::{Arc, Mutex}; + + use async_trait::async_trait; + use serde_json::{Value, json}; + use tinyflows::caps::mock::mock_capabilities; + use tinyflows::caps::{ + ApprovalDecision, ApprovalOutcome, ApprovalProvider, ApprovalRequest, Capabilities, + }; + use tinyflows::compiler::compile; + use tinyflows::engine::{resume, run}; + use tinyflows::model::{Edge, Node, NodeKind, WorkflowGraph}; + + /// A host's review desk: one row per `request_id`, holding the verdict once + /// a human has left one. + #[derive(Default)] + struct DeskReview { + rows: Mutex>>, + } + + impl DeskReview { + /// What a human does later, from the host's own UI. + fn answer(&self, request_id: &str, decision: ApprovalDecision) { + self.rows + .lock() + .expect("lock") + .insert(request_id.to_string(), Some(decision)); + } + + /// Every review this desk has been asked to run, decided or not. + fn queue(&self) -> Vec { + let mut ids: Vec = self.rows.lock().expect("lock").keys().cloned().collect(); + ids.sort(); + ids + } + } + + #[async_trait] + impl ApprovalProvider for DeskReview { + async fn decide( + &self, + request: &ApprovalRequest, + ) -> tinyflows::error::Result { + let mut rows = self.rows.lock().expect("lock"); + // Create-or-fetch: the row is keyed on `request_id`, so the run + // asking again after a resume finds THIS review rather than opening + // a second one and pinging the reviewer twice. + let row = rows.entry(request.request_id.clone()).or_insert_with(|| { + println!( + "[desk] new review {:?}: {} -> {}", + request.request_id, + request.title.as_deref().unwrap_or("(untitled)"), + request.subject.value + ); + None + }); + Ok(match row.clone() { + Some(decision) => ApprovalOutcome::Decided(decision), + None => ApprovalOutcome::Pending, + }) + } + } + + fn node(id: &str, kind: NodeKind, config: Value) -> Node { + Node { + id: id.into(), + kind, + type_version: 1, + name: id.into(), + config, + ports: vec![], + position: None, + } + } + fn edge(from: &str, port: &str, to: &str) -> Edge { + Edge { + from_node: from.into(), + from_port: port.into(), + to_node: to.into(), + to_port: "main".into(), + } + } + + // trigger -> review -> publish (on `approved`) / revise (on `rejected`). + let graph = WorkflowGraph { + nodes: vec![ + node("trigger", NodeKind::Trigger, Value::Null), + node( + "review", + NodeKind::Approval, + json!({ + // A real host would key this on the run id (e.g. + // `"=run.id"`) rather than a literal, so two runs of this + // graph never collide on the same review. + "request_id": "hitl-review-example", + "title": "Publish this post?", + "prompt": "Approving publishes it to the public feed.", + "subject_kind": "url", + "subject": "=item.url", + "assignees": ["editor@example.com"], + }), + ), + node( + "publish", + NodeKind::Transform, + json!({ "set": { "published": "=item.subject" } }), + ), + node( + "revise", + NodeKind::Transform, + json!({ "set": { "revise_because": "=item.comment" } }), + ), + ], + edges: vec![ + edge("trigger", "main", "review"), + edge("review", "approved", "publish"), + edge("review", "rejected", "revise"), + ], + ..Default::default() + }; + + let compiled = compile(&graph).expect("compile"); + let desk = Arc::new(DeskReview::default()); + let caps = Capabilities { + approvals: Some(desk.clone()), + ..mock_capabilities() + }; + let trigger = json!({ "url": "https://example.com/drafts/42" }); + + // 1) Nobody has answered, so the run suspends at the review. Nothing is + // burned while the card sits in someone's queue. + let paused = run(&compiled, trigger.clone(), &caps).await.expect("run"); + println!("--- before the human answers ---"); + println!("pending_approvals: {:?}", paused.pending_approvals); + println!("desk queue: {:?}", desk.queue()); + + // 2) The human approves — and edits the URL on the way through, which the + // host reports as the decision's payload. + let request_id = desk.queue().first().cloned().expect("one open review"); + desk.answer( + &request_id, + ApprovalDecision { + approved: true, + decided_by: Some("editor@example.com".into()), + comment: Some("fixed the slug".into()), + payload: Some(json!("https://example.com/drafts/42?utm=newsletter")), + }, + ); + + // 3) Resuming re-asks the desk, which now has the verdict. Note the review + // id is unchanged, so the reviewer is never asked a second time. + let done = resume(&compiled, trigger, vec![], &caps) + .await + .expect("resume"); + println!("--- after the human answers ---"); + println!("pending_approvals: {:?}", done.pending_approvals); + println!( + "review port: {}", + done.output["nodes"]["review"]["port"] + ); + println!( + "published: {}", + done.output["nodes"]["publish"]["items"][0]["json"]["published"] + ); + println!("desk queue: {:?}", desk.queue()); +} + +#[cfg(not(feature = "mock"))] +fn main() { + eprintln!( + "this example needs the mock capabilities: cargo run --example hitl_review --features mock" + ); +} diff --git a/src/caps/approval.rs b/src/caps/approval.rs new file mode 100644 index 00000000..c9878ed6 --- /dev/null +++ b/src/caps/approval.rs @@ -0,0 +1,170 @@ +//! Human-in-the-loop review: the [`ApprovalProvider`] capability an `approval` +//! node reaches the human through. +//! +//! # Why this is a capability at all +//! +//! An `approval` node presents *something* — a URL, a block of text, a +//! generated payload — to a person and waits for an approve/reject. Everything +//! about how that reaches a human is the host's: a Slack card, an inbox row, a +//! web review queue, a phone notification. The crate stays out of it and asks +//! only two things of the host, both expressed here: **register/fetch** a +//! review request, and (optionally) **cancel** one it will never wait on again. +//! +//! # The idempotency contract (load-bearing) +//! +//! [`ApprovalProvider::decide`] is a **create-or-fetch** call keyed on +//! [`ApprovalRequest::request_id`]: the first call with a given id creates the +//! review, every later call with that same id returns the state of *that* +//! review and must not create a second one. +//! +//! This is not a nicety. A node that pauses the run with +//! [`NodeControl::Interrupt`](crate::nodes::NodeControl::Interrupt) has its +//! state update discarded, so on resume it re-runs from the top and calls +//! `decide` again — and a polling node calls it once per poll. A provider that +//! created a fresh review per call would spam the human with a new card every +//! time the run looked at the world. The node derives a stable `request_id` from +//! the run and node id (or takes one from config), so honouring it is enough. +//! +//! # Optional, like [`MemoryProvider`](crate::caps::MemoryProvider) +//! +//! Hosts that wire no provider leave [`Capabilities::approvals`](crate::caps::Capabilities::approvals) +//! `None`. An `approval` node then falls back to the engine's existing +//! pause/resume channel: it interrupts the run naming itself on +//! [`RunOutcome::pending_approvals`](crate::engine::RunOutcome::pending_approvals), +//! and the host settles it out of band with +//! [`engine::resume`](crate::engine::resume). That fallback is deliberate — +//! a host that already has a review surface bolted onto run resumption should +//! not have to implement a trait to use this node. + +use async_trait::async_trait; +use serde_json::Value; + +use crate::error::Result; + +/// What the human is being asked to look at. +/// +/// `kind` is an **opaque, host-defined** rendering hint (the model layer never +/// interprets it), following the `scope` precedent in +/// [`MemoryProvider`](crate::caps::MemoryProvider). The conventional values are +/// `"url"`, `"text"`, `"markdown"`, and `"json"`; a host is free to define more +/// and a host that renders everything the same way may ignore it entirely. +#[derive(Debug, Clone, PartialEq)] +pub struct ApprovalSubject { + /// Rendering hint — `"url"` / `"text"` / `"markdown"` / `"json"` by + /// convention, host-defined in general. + pub kind: String, + /// The thing itself: the URL string, the prose, the payload. + pub value: Value, +} + +/// One review request handed to the host. +#[derive(Debug, Clone, PartialEq)] +pub struct ApprovalRequest { + /// Stable identity of *this* review, and the key the create-or-fetch + /// contract is built on (see the module docs). Derived from the run and + /// node id unless the node's `config.request_id` overrides it, so it is the + /// same string across an interrupt/resume and across every poll. + pub request_id: String, + /// The node asking, for a host that wants to link the review back to the + /// graph. + pub node_id: String, + /// The run this review belongs to, when the run state carries one. + pub run_id: Option, + /// Short human-facing headline (`config.title`). + pub title: Option, + /// Fuller ask — what approving actually authorizes (`config.prompt`). + pub prompt: Option, + /// What is being reviewed. + pub subject: ApprovalSubject, + /// Opaque host-resolved reviewer handles (user ids, emails, a channel, a + /// role name). The crate never interprets these. + pub assignees: Vec, + /// Anything else the graph attached for the host's benefit + /// (`config.metadata`), passed through untouched. + pub metadata: Value, +} + +/// A human's verdict on an [`ApprovalRequest`]. +#[derive(Debug, Clone, PartialEq)] +pub struct ApprovalDecision { + /// `true` for approve, `false` for reject. Binary on purpose: an + /// n-way review is a `switch` on data, not an approval gate. + pub approved: bool, + /// Opaque host handle for whoever decided. + pub decided_by: Option, + /// Free-text note the reviewer left, surfaced on the emitted item so a + /// rejection branch can act on the reason. + pub comment: Option, + /// The subject as the human left it, when the host's review surface lets + /// them edit before approving. `None` means "unchanged", and the node emits + /// the subject it sent. + pub payload: Option, +} + +impl ApprovalDecision { + /// An approval with no reviewer, comment, or edit recorded. + #[must_use] + pub fn approved() -> Self { + Self { + approved: true, + decided_by: None, + comment: None, + payload: None, + } + } + + /// A rejection carrying an optional reason. + #[must_use] + pub fn rejected(comment: Option) -> Self { + Self { + approved: false, + decided_by: None, + comment, + payload: None, + } + } +} + +/// Where a review stands when the host is asked. +#[derive(Debug, Clone, PartialEq)] +pub enum ApprovalOutcome { + /// Nobody has decided yet. The node waits — by suspending the run or by + /// polling, per its `wait_mode`. + Pending, + /// A human decided. + Decided(ApprovalDecision), +} + +/// Host-implemented delivery of an approve/reject decision to a human. +/// +/// See the module docs for the create-or-fetch contract every implementation +/// must honour, and for what happens on hosts that wire no provider at all. +#[async_trait] +pub trait ApprovalProvider: Send + Sync { + /// Registers `request` for human review, or — if a review with that + /// `request_id` already exists — returns where that one stands. + /// + /// Must be idempotent on [`ApprovalRequest::request_id`]: this is called + /// again on every poll and after every resume, and a provider that creates a + /// new review per call notifies the human once per call. + /// + /// # Errors + /// Returns an [`EngineError::Capability`](crate::error::EngineError::Capability) + /// when the review cannot be created or read. + async fn decide(&self, request: &ApprovalRequest) -> Result; + + /// Withdraws a review nobody will wait on any more (the node timed out, the + /// run was cancelled), so a stale card does not sit in a human's queue. + /// + /// Best-effort by design: the default implementation does nothing, and the + /// node logs rather than fails when this errors — the run has already + /// decided what to do by the time it is called. + /// + /// # Errors + /// Returns an [`EngineError::Capability`](crate::error::EngineError::Capability) + /// when the host knows the withdrawal failed. + async fn cancel(&self, request_id: &str, reason: &str) -> Result<()> { + let _ = (request_id, reason); + Ok(()) + } +} diff --git a/src/caps/mock.rs b/src/caps/mock.rs index 3f222ba1..640518cf 100644 --- a/src/caps/mock.rs +++ b/src/caps/mock.rs @@ -4,19 +4,20 @@ //! `mock` cargo feature. The mocks are deterministic echoes — enough to exercise //! the engine and the reference workflows without any external services. -use std::sync::Arc; - use async_trait::async_trait; use serde_json::{Value, json}; use crate::caps::{ - AgentRunner, Capabilities, CodeLanguage, CodeRunner, HttpClient, LlmProvider, MemoryProvider, - ShellOutcome, ShellRequest, ShellRunner, ShellScript, StateStore, ToolInvoker, - WorkflowResolver, + AgentRunner, CodeLanguage, CodeRunner, HttpClient, LlmProvider, MemoryProvider, ShellOutcome, + ShellRequest, ShellRunner, ShellScript, StateStore, ToolInvoker, WorkflowResolver, }; use crate::error::{EngineError, Result}; use crate::model::WorkflowGraph; +#[path = "mock_approvals.rs"] +mod mock_approvals; +pub use mock_approvals::MockApprovals; + /// An [`LlmProvider`] that echoes the request back under a `completion` key. #[derive(Debug, Default, Clone)] pub struct MockLlm; @@ -424,62 +425,12 @@ impl WorkflowResolver for MockWorkflowResolver { } } -/// 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. Unlike -/// [`Capabilities::agent`] (which defaults `None`), [`Capabilities::memory`] is -/// wired to [`MockMemory`] by default — a `memory` node must dry-run -/// successfully out of the box; use `Capabilities { memory: None, ..caps }` to -/// exercise the "host wired no memory store" error path instead. -#[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), - shell: Some(Arc::new(MockShell)), - state: Arc::new(MockStateStore::default()), - resolver: Arc::new(resolver), - // No agent registry by default: `agent` nodes use `MockLlm`. Use - // [`mock_capabilities_with_agent`] to exercise the `agent_ref` path. - agent: None, - // Wired by default (unlike `agent`) — see the doc comment above. - memory: Some(Arc::new(MockMemory)), - // The real tokio-backed runner, so `spawn`/`gate` behave under test the - // way they behave for a host that wires nothing. - tasks: Some(Arc::new(crate::caps::TokioTaskRunner::new())), - } -} - -/// Like [`mock_capabilities`], but wires an [`AgentRunner`] so tests can exercise -/// an `agent` node that selects a named agent kind via `agent_ref`. -#[must_use] -pub fn mock_capabilities_with_agent(agent: impl AgentRunner + 'static) -> Capabilities { - Capabilities { - agent: Some(Arc::new(agent)), - ..mock_capabilities() - } -} - -/// Like [`mock_capabilities`], but with a caller-supplied [`MemoryProvider`] in -/// place of the default [`MockMemory`] — for tests that need custom recall / -/// flavour / people / remember / forget behavior. -#[must_use] -pub fn mock_capabilities_with_memory(memory: impl MemoryProvider + 'static) -> Capabilities { - Capabilities { - memory: Some(Arc::new(memory)), - ..mock_capabilities() - } -} +#[path = "mock_builders.rs"] +mod mock_builders; +pub use mock_builders::{ + mock_capabilities, mock_capabilities_with_agent, mock_capabilities_with_approvals, + mock_capabilities_with_memory, mock_capabilities_with_resolver, +}; #[cfg(test)] #[path = "mock_tests.rs"] diff --git a/src/caps/mock_approvals.rs b/src/caps/mock_approvals.rs new file mode 100644 index 00000000..fdae119f --- /dev/null +++ b/src/caps/mock_approvals.rs @@ -0,0 +1,109 @@ +//! [`MockApprovals`], the in-memory [`ApprovalProvider`] used by +//! [`mock_capabilities`](super::mock::mock_capabilities) and by tests that want +//! to drive the `approval` node's approve / reject / pending paths without a +//! real review surface. + +use async_trait::async_trait; + +use crate::caps::{ApprovalDecision, ApprovalOutcome, ApprovalProvider, ApprovalRequest}; +use crate::error::Result; + +/// An [`ApprovalProvider`] that answers every review the same way, without a +/// human anywhere. +/// +/// Defaults to approving, so a graph containing an `approval` node dry-runs +/// end-to-end out of the box (the `MockMemory` precedent). Use +/// [`rejecting`](Self::rejecting) to drive the reject branch and +/// [`pending`](Self::pending) to exercise the waiting path — a suspending node +/// then pauses the run, and a polling one spends its poll budget. +/// +/// Records every `request_id` it has seen so a test can assert the +/// create-or-fetch contract holds (one review per id, however many activations +/// the node had). +#[derive(Debug, Default)] +pub struct MockApprovals { + outcome: MockApprovalOutcome, + seen: std::sync::Mutex>, + cancelled: std::sync::Mutex>, +} + +/// What a [`MockApprovals`] answers with. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +enum MockApprovalOutcome { + /// Approve immediately. + #[default] + Approve, + /// Reject immediately. + Reject, + /// Never decide. + Pending, +} + +impl MockApprovals { + /// A provider that approves every request. + #[must_use] + pub fn approving() -> Self { + Self::default() + } + + /// A provider that rejects every request, with `"mock rejection"` as the + /// reviewer's comment. + #[must_use] + pub fn rejecting() -> Self { + Self { + outcome: MockApprovalOutcome::Reject, + ..Self::default() + } + } + + /// A provider that leaves every request pending forever. + #[must_use] + pub fn pending() -> Self { + Self { + outcome: MockApprovalOutcome::Pending, + ..Self::default() + } + } + + /// Every `request_id` passed to [`ApprovalProvider::decide`], in call order + /// (with repeats — the point is to show repeats are the *same* id). + #[must_use] + pub fn requested(&self) -> Vec { + self.seen.lock().expect("lock").clone() + } + + /// Every `request_id` passed to [`ApprovalProvider::cancel`]. + #[must_use] + pub fn cancelled(&self) -> Vec { + self.cancelled.lock().expect("lock").clone() + } +} + +#[async_trait] +impl ApprovalProvider for MockApprovals { + async fn decide(&self, request: &ApprovalRequest) -> Result { + self.seen + .lock() + .expect("lock") + .push(request.request_id.clone()); + Ok(match self.outcome { + MockApprovalOutcome::Approve => ApprovalOutcome::Decided(ApprovalDecision { + decided_by: Some("mock-reviewer".to_string()), + ..ApprovalDecision::approved() + }), + MockApprovalOutcome::Reject => ApprovalOutcome::Decided(ApprovalDecision { + decided_by: Some("mock-reviewer".to_string()), + ..ApprovalDecision::rejected(Some("mock rejection".to_string())) + }), + MockApprovalOutcome::Pending => ApprovalOutcome::Pending, + }) + } + + async fn cancel(&self, request_id: &str, _reason: &str) -> Result<()> { + self.cancelled + .lock() + .expect("lock") + .push(request_id.to_string()); + Ok(()) + } +} diff --git a/src/caps/mock_builders.rs b/src/caps/mock_builders.rs new file mode 100644 index 00000000..df70160f --- /dev/null +++ b/src/caps/mock_builders.rs @@ -0,0 +1,88 @@ +//! Builds [`Capabilities`] bundles wired to the mock implementations in +//! [`super`] and [`super::mock_approvals`]. +//! +//! Split out of `mock.rs` to keep that file under the repository's +//! line-length limit; these are all thin variations on one bundle, so they +//! belong together rather than split further. + +use std::sync::Arc; + +use crate::caps::{AgentRunner, ApprovalProvider, Capabilities, MemoryProvider, WorkflowResolver}; + +use super::{ + MockApprovals, MockCode, MockHttp, MockLlm, MockMemory, MockShell, MockStateStore, MockTools, + MockWorkflowResolver, +}; + +/// 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. Unlike +/// [`Capabilities::agent`] (which defaults `None`), [`Capabilities::memory`] is +/// wired to [`MockMemory`] by default — a `memory` node must dry-run +/// successfully out of the box; use `Capabilities { memory: None, ..caps }` to +/// exercise the "host wired no memory store" error path instead. +#[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), + shell: Some(Arc::new(MockShell)), + state: Arc::new(MockStateStore::default()), + resolver: Arc::new(resolver), + // No agent registry by default: `agent` nodes use `MockLlm`. Use + // [`mock_capabilities_with_agent`] to exercise the `agent_ref` path. + agent: None, + // Wired by default (unlike `agent`) — see the doc comment above. + memory: Some(Arc::new(MockMemory)), + // The real tokio-backed runner, so `spawn`/`gate` behave under test the + // way they behave for a host that wires nothing. + tasks: Some(Arc::new(crate::caps::TokioTaskRunner::new())), + // Wired by default, like `memory`: a graph with an `approval` node must + // dry-run without the host standing up a review surface first. + approvals: Some(Arc::new(MockApprovals::approving())), + } +} + +/// Like [`mock_capabilities`], but wires an [`AgentRunner`] so tests can exercise +/// an `agent` node that selects a named agent kind via `agent_ref`. +#[must_use] +pub fn mock_capabilities_with_agent(agent: impl AgentRunner + 'static) -> Capabilities { + Capabilities { + agent: Some(Arc::new(agent)), + ..mock_capabilities() + } +} + +/// Like [`mock_capabilities`], but with a caller-supplied [`MemoryProvider`] in +/// place of the default [`MockMemory`] — for tests that need custom recall / +/// flavour / people / remember / forget behavior. +#[must_use] +pub fn mock_capabilities_with_memory(memory: impl MemoryProvider + 'static) -> Capabilities { + Capabilities { + memory: Some(Arc::new(memory)), + ..mock_capabilities() + } +} + +/// Like [`mock_capabilities`], but with a caller-supplied [`ApprovalProvider`] +/// in place of the default approve-everything [`MockApprovals`] — for tests +/// that need a rejection, a pending review, or a host-shaped decision. +#[must_use] +pub fn mock_capabilities_with_approvals( + approvals: impl ApprovalProvider + 'static, +) -> Capabilities { + Capabilities { + approvals: Some(Arc::new(approvals)), + ..mock_capabilities() + } +} diff --git a/src/caps/mod.rs b/src/caps/mod.rs index 9f2d0694..4e4bb409 100644 --- a/src/caps/mod.rs +++ b/src/caps/mod.rs @@ -6,6 +6,7 @@ //! curated Composio tools, `HttpRequestTool`, and sandboxed code runtimes. pub mod agent; +pub mod approval; #[cfg(any(test, feature = "host-caps"))] pub mod host; #[cfg(any(test, feature = "mock"))] @@ -25,6 +26,9 @@ pub use self::agent::{ AgentInput, AgentModelSelection, AgentRunIdentity, AgentRunOutcome, AgentRunRequest, AgentRunner, AgentUsage, ContextBlock, StopReason, ToolDescriptor, }; +pub use self::approval::{ + ApprovalDecision, ApprovalOutcome, ApprovalProvider, ApprovalRequest, ApprovalSubject, +}; pub use self::schema::sample_for_schema; pub use self::shell::{ShellInterpreter, ShellOutcome, ShellRequest, ShellRunner, ShellScript}; pub use self::tasks::{TaskRunner, TaskSpec, TaskState, TokioTaskRunner}; @@ -199,8 +203,8 @@ pub trait StateStore: Send + Sync { /// Construct one per run from the host's concrete implementations. It carries /// every host-injected capability: the always-present [`LlmProvider`], /// [`ToolInvoker`], [`HttpClient`], [`CodeRunner`], [`StateStore`], and -/// [`WorkflowResolver`], plus the optional [`AgentRunner`] and -/// [`MemoryProvider`]. Nodes reach each one through `ctx.caps` during +/// [`WorkflowResolver`], plus the optional [`AgentRunner`], [`MemoryProvider`], +/// and [`ApprovalProvider`]. Nodes reach each one through `ctx.caps` during /// execution. #[derive(Clone)] pub struct Capabilities { @@ -249,6 +253,13 @@ pub struct Capabilities { /// rather than a correctness one, which is exactly why it is worth saying /// out loud here and in the node catalog. pub tasks: Option>, + /// Optional human-review surface for `approval` nodes. `None` on hosts + /// without one, in which case an `approval` node falls back to pausing the + /// run and letting the host settle it through + /// [`engine::resume`](crate::engine::resume) — so the node still works, it + /// just has no way to *push* the request at anybody. See + /// [`ApprovalProvider`] for the create-or-fetch contract. + pub approvals: Option>, } #[cfg(test)] diff --git a/src/catalog.rs b/src/catalog.rs index e99d8e3c..5c0b4662 100644 --- a/src/catalog.rs +++ b/src/catalog.rs @@ -33,7 +33,7 @@ use group_03::*; /// The node kinds, in the canonical order used wherever the DSL is enumerated /// (matches [`NodeKind`](crate::model::NodeKind)'s serde discriminators). -pub const NODE_KINDS: [&str; 21] = [ +pub const NODE_KINDS: [&str; 22] = [ "trigger", "agent", "tool_call", @@ -54,6 +54,7 @@ pub const NODE_KINDS: [&str; 21] = [ "gate", "scatter", "gather", + "approval", "void", ]; @@ -205,6 +206,7 @@ pub fn contract_for(kind: &str) -> Option { "gate" => contract_gate(), "scatter" => contract_scatter(), "gather" => contract_gather(), + "approval" => contract_approval(), "void" => contract_void(), _ => return None, }; diff --git a/src/catalog/contracts/group_03.rs b/src/catalog/contracts/group_03.rs index 09795ca4..8771894c 100644 --- a/src/catalog/contracts/group_03.rs +++ b/src/catalog/contracts/group_03.rs @@ -116,6 +116,131 @@ pub(super) fn contract_gather() -> NodeKindContract { } } +pub(super) fn contract_approval() -> NodeKindContract { + NodeKindContract { + kind: "approval".to_string(), + summary: "Puts a subject — a URL, a draft, a payload — in front of a HUMAN and routes \ + on their approve/reject." + .to_string(), + description: "Not the same thing as the `requires_approval` flag. That flag holds a \ + node back until someone says go, carries nothing, and its answer is \ + invisible to the graph. This kind IS the review: it carries what is being \ + reviewed, and the verdict (who decided, their comment, any edit they made) \ + comes back as data on the `approved` / `rejected` ports. Reaches the human \ + through the host's ApprovalProvider capability; with none injected it \ + pauses the run instead, and the host settles it with engine::resume." + .to_string(), + config_fields: vec![ + ConfigField::optional( + "subject", + "any | \"=expr\"", + "What the human looks at. Defaults to the input item that arrived, which is \ + usually what you want; set it to project one field (\"=item.url\").", + ), + ConfigField::optional( + "subject_kind", + "string", + "Rendering hint for the host's review surface — url | text | markdown | json \ + (default) by convention, host-defined in general.", + ), + ConfigField::optional("title", "string", "Short headline for the review card."), + ConfigField::optional( + "prompt", + "string", + "What approving actually authorizes — the sentence the reviewer decides on.", + ), + ConfigField::optional( + "assignees", + "array", + "Opaque host-resolved reviewer handles (user ids, emails, a channel, a role).", + ), + ConfigField::optional( + "metadata", + "object", + "Anything else the host's review surface wants, passed through untouched.", + ), + ConfigField::optional( + "request_id", + "string", + "Overrides the review's identity (default \":\", derived from \ + whichever of `run.id` / `run.run_id` the host seeds — never read from the \ + caller-supplied trigger payload). \ + Must be stable across resumes: it is the key the host de-duplicates reviews on. \ + Required when no run-scoped id is available — falling back to the bare node id \ + would let a later run of the same graph reuse an earlier run's decision, so the \ + node refuses to guess and fails instead. SECURITY: whichever run id feeds this \ + must be server-generated, never a caller-supplied trigger field forwarded \ + unvalidated — an attacker who can choose it can collide with an earlier run's \ + request_id and inherit its cached decision, skipping review entirely.", + ), + ConfigField::optional( + "wait_mode", + "enum", + "suspend (default) interrupts the run until the host resumes it; poll \ + re-activates the node on an interval, which is only sane for a decision \ + expected within seconds.", + ) + .with_enum(&["suspend", "poll"]), + ConfigField::optional( + "on_reject", + "enum", + "route (default: emit the verdict on the `rejected` port) | error (fail the \ + node) | drop (emit nothing).", + ) + .with_enum(&["route", "error", "drop"]), + ConfigField::optional( + "on_timeout", + "enum", + "When a POLLING review runs out of budget: error (default) | reject (follow \ + on_reject) | route (emit on the `timeout` port).", + ) + .with_enum(&["error", "reject", "route"]), + ConfigField::optional( + "poll_interval_ms", + "number", + "Gap between polls (default 1000). Ignored when suspending.", + ), + ConfigField::optional( + "max_polls", + "number", + "Poll budget before the wait is called spent (default 60). Each poll costs a \ + super-step. Ignored when suspending — a suspended review waits as long as the \ + host lets it.", + ), + ], + ports: PortSpec::new(&["main"], &["approved", "rejected", "timeout", "error"]), + example: json!({ + "id": "review", "kind": "approval", "name": "Approve the post", + "config": { + "request_id": "=run.id", + "title": "Publish this post?", + "prompt": "Approving publishes it to the public feed.", + "subject_kind": "url", + "subject": "=item.preview_url", + "assignees": ["=inputs.owner"] + } + }), + notes: vec![ + "Wire the `approved` port — an approval whose approve branch is unwired reviews \ + something and then does nothing with the answer." + .to_string(), + "A rejection is DATA, not an error, by default: it routes to `rejected` with the \ + reviewer's comment so a recovery branch can revise and ask again. Use \ + on_reject: \"error\" only when a rejection should abort." + .to_string(), + "The emitted item carries `subject` as the human left it — their edit when the \ + host's surface allowed one, otherwise exactly what was sent — plus `approved`, \ + `comment`, `decided_by`, and the original `input`. The verdict is also readable \ + anywhere as \"=nodes..decision.approved\"." + .to_string(), + "`request_id` is what stops a human being notified once per poll and once per \ + resume: the host is asked create-or-fetch on that id. Override it only with \ + something equally stable." + .to_string(), + ], + } +} + pub(super) fn contract_void() -> NodeKindContract { NodeKindContract { kind: "void".to_string(), diff --git a/src/catalog_tests.rs b/src/catalog_tests.rs index 5bcec5e5..e4931ea3 100644 --- a/src/catalog_tests.rs +++ b/src/catalog_tests.rs @@ -23,7 +23,7 @@ fn every_node_kind_has_a_contract() { } } } - assert_eq!(all_contracts().len(), 21); + assert_eq!(all_contracts().len(), 22); } #[test] @@ -48,8 +48,8 @@ fn unknown_kind_has_no_contract() { } #[test] -fn node_kinds_has_21_entries_including_the_async_and_lane_pairs() { - assert_eq!(NODE_KINDS.len(), 21); +fn node_kinds_has_22_entries_including_the_async_and_lane_pairs() { + assert_eq!(NODE_KINDS.len(), 22); assert!(NODE_KINDS.contains(&"shell")); assert!(NODE_KINDS.contains(&"memory")); assert!(NODE_KINDS.contains(&"dedup")); @@ -67,8 +67,9 @@ fn node_kinds_has_21_entries_including_the_async_and_lane_pairs() { assert_eq!(NODE_KINDS[17], "gate"); assert_eq!(NODE_KINDS[18], "scatter"); assert_eq!(NODE_KINDS[19], "gather"); + assert_eq!(NODE_KINDS[20], "approval"); assert!(NODE_KINDS.contains(&"void")); - assert_eq!(NODE_KINDS[20], "void"); + assert_eq!(NODE_KINDS[21], "void"); } #[test] diff --git a/src/engine/build.rs b/src/engine/build.rs index 0118a86a..383b8f10 100644 --- a/src/engine/build.rs +++ b/src/engine/build.rs @@ -1,6 +1,7 @@ use super::*; mod activation; +mod backoff; mod handlers; mod outcome; mod wiring; diff --git a/src/engine/build/activation.rs b/src/engine/build/activation.rs index 92db200f..c9c61e3b 100644 --- a/src/engine/build/activation.rs +++ b/src/engine/build/activation.rs @@ -549,7 +549,10 @@ impl HandlerData { return Ok(NodeResult::Update(items_update(&node.id, &[], None)?)); } let step = remaining.min(BACKOFF_POLL_MS); - futures_timer::Delay::new(std::time::Duration::from_millis(step)).await; + // Yields even when the timer has already fired, so a retry + // backoff never starves the executor it shares — see + // `super::backoff`. + super::backoff::wait_slice(step).await; remaining -= step; } } diff --git a/src/engine/build/backoff.rs b/src/engine/build/backoff.rs new file mode 100644 index 00000000..e2a6708b --- /dev/null +++ b/src/engine/build/backoff.rs @@ -0,0 +1,66 @@ +//! The engine's waiting primitive: one slice of a backoff. +//! +//! Two places in the engine wait without holding the executor: the retry +//! backoff between a failed attempt and the next one, and the `Reenter` backoff +//! a polling node (a `gate`) asks for between activations. Both chop their wait +//! into short slices so a cancel is seen promptly, and both take a slice from +//! here. +//! +//! # Why a wait has to yield, not merely elapse +//! +//! `futures_timer::Delay` arms its timer when it is **constructed** — `new` +//! computes the deadline and pushes it to the global timer thread — and its +//! `poll` returns `Ready` immediately if that thread has already fired it. So a +//! task descheduled between constructing the `Delay` and first polling it for +//! longer than the slice finds the wait already over, and completes it without +//! ever returning `Pending`: a "wait" during which the executor was never given +//! a turn. +//! +//! That is not a cosmetic difference. The engine runs on the caller's executor, +//! and a backoff is the only point at which it hands that executor back. On a +//! single-threaded runtime the background work a `gate` is waiting on — the +//! tasks a `spawn` node started — can *only* progress while the engine is +//! yielded. A backoff that skips the yield therefore returns the gate to a +//! world that has not moved: it sees the same unsettled tickets, spends another +//! poll against its bounded budget, and the run takes a different number of +//! super-steps than an identical run whose backoff did yield. Under enough load +//! a gate could burn its whole poll budget and time out having never once let +//! the tasks it waited on run. +//! +//! Yielding unconditionally makes the wait mean what it says, and makes the +//! number of polls a gate needs a property of the graph rather than of how the +//! OS happened to schedule the process. + +use std::task::Poll; + +/// Hands the executor exactly one turn. +/// +/// Returns `Pending` on its first poll — waking itself first, so it is +/// re-queued behind whatever else is already ready — and `Ready` on its second. +/// Unconditional: it does not consult a timer, which is the whole point. +async fn yield_once() { + let mut yielded = false; + std::future::poll_fn(|cx| { + if yielded { + return Poll::Ready(()); + } + yielded = true; + cx.waker().wake_by_ref(); + Poll::Pending + }) + .await; +} + +/// Waits `ms` milliseconds, always giving the executor at least one turn. +/// +/// The timer is armed before the yield rather than after, so the turn handed +/// over counts toward the wait instead of being added to it. +pub(super) async fn wait_slice(ms: u64) { + let delay = futures_timer::Delay::new(std::time::Duration::from_millis(ms)); + yield_once().await; + delay.await; +} + +#[cfg(test)] +#[path = "backoff_tests.rs"] +mod tests; diff --git a/src/engine/build/backoff_tests.rs b/src/engine/build/backoff_tests.rs new file mode 100644 index 00000000..005f0789 --- /dev/null +++ b/src/engine/build/backoff_tests.rs @@ -0,0 +1,82 @@ +//! Unit tests for the backoff slice primitive. +//! +//! These pin the contract of [`yield_once`], which is what makes a backoff +//! hand the executor a turn whatever the timer did. The bug that contract +//! exists to prevent is an engine-level one and is covered end to end by +//! `a_gate_takes_the_same_number_of_polls_every_run` in `tests/fuzz_resume.rs`; +//! what is asserted here is the piece that can be checked deterministically. + +use std::future::Future; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::task::{Context, Poll, Wake, Waker}; + +use super::{wait_slice, yield_once}; + +/// A waker that counts how many times it was woken. +struct Counting(AtomicUsize); + +impl Wake for Counting { + fn wake(self: Arc) { + self.wake_by_ref(); + } + + fn wake_by_ref(self: &Arc) { + self.0.fetch_add(1, Ordering::SeqCst); + } +} + +/// A yield is `Pending` once, then `Ready` — and it wakes itself, so an +/// executor re-queues it rather than dropping the task. +/// +/// The self-wake is half the contract: a `Pending` that never wakes is a hang, +/// not a yield. +#[test] +fn a_yield_is_pending_once_then_ready() { + let counter = Arc::new(Counting(AtomicUsize::new(0))); + let waker = Waker::from(counter.clone()); + let mut cx = Context::from_waker(&waker); + let mut future = std::pin::pin!(yield_once()); + + assert_eq!( + future.as_mut().poll(&mut cx), + Poll::Pending, + "the first poll of a yield must hand the executor its turn" + ); + assert_eq!( + counter.0.load(Ordering::SeqCst), + 1, + "a yield must wake itself, or the turn it hands over is never taken back" + ); + assert_eq!( + future.as_mut().poll(&mut cx), + Poll::Ready(()), + "a yield hands over exactly one turn, not a stream of them" + ); +} + +/// A wait slice never completes on its first poll, however the timer resolved. +/// +/// This is the property the whole module exists for: a slice whose timer had +/// already fired must still yield. Asserted against a zero-length slice, whose +/// deadline is already in the past the moment it is armed — the case most +/// likely to short-circuit. +#[test] +fn a_wait_slice_never_completes_on_its_first_poll() { + let counter = Arc::new(Counting(AtomicUsize::new(0))); + let waker = Waker::from(counter); + let mut cx = Context::from_waker(&waker); + let mut future = std::pin::pin!(wait_slice(0)); + + assert_eq!( + future.as_mut().poll(&mut cx), + Poll::Pending, + "an already-elapsed slice must still yield, or concurrent work never runs" + ); +} + +/// And it does finish: the yield does not turn a bounded wait into a hang. +#[tokio::test] +async fn a_wait_slice_still_completes() { + wait_slice(1).await; +} diff --git a/src/engine/build/outcome.rs b/src/engine/build/outcome.rs index 68fa4e30..065d4a47 100644 --- a/src/engine/build/outcome.rs +++ b/src/engine/build/outcome.rs @@ -88,8 +88,10 @@ where return Ok(NodeResult::Update(items_update(&node.id, &[], None)?)); } let slice = remaining.min(BACKOFF_POLL_MS); - futures_timer::Delay::new(std::time::Duration::from_millis(slice)) - .await; + // Always yields, so the tasks this node is polling + // for get a turn before it looks again — see + // `super::backoff`. + super::backoff::wait_slice(slice).await; remaining -= slice; } } diff --git a/src/engine/run_state.rs b/src/engine/run_state.rs index a38cbb37..3d869340 100644 --- a/src/engine/run_state.rs +++ b/src/engine/run_state.rs @@ -286,6 +286,10 @@ pub async fn resume( /// is replaced by a fresh object holding just the merged approvals. Declared /// inputs ride along unchanged — a resume re-runs the *same* parameterized /// workflow, so dropping them would silently change what it does. +/// +/// The returned [`RunInput::approvals`] carries **only** what a host explicitly +/// authorised — never an id read out of the trigger payload. See the body for +/// why the two sets are kept apart. pub(super) fn merge_approvals(input: impl Into, newly_approved: Vec) -> RunInput { let RunInput { mut trigger, @@ -293,7 +297,33 @@ pub(super) fn merge_approvals(input: impl Into, newly_approved: Vec = trigger + // Approval **provenance** is preserved rather than flattened, and the two + // sets are deliberately not the same list. + // + // `explicit` is what a host actually authorised: the ids passed to + // `engine::resume`, plus any carried on `RunInput::approvals`. It never + // includes anything read out of the trigger, because the trigger is the + // payload a caller submitted — on a webhook, attacker-supplied. Flattening + // the two meant a caller-written `trigger.approvals` entry was promoted + // into the trusted list by the first resume, so a self-approval that the + // initial run correctly refused went through on the next one. + // + // `trigger.approvals` still receives the union, unchanged, because the + // `requires_approval` gate in `engine::build::activation` reads exactly + // that key and callers are documented as being able to write approvals + // there directly. Narrowing *that* is a separate change to a shared, + // documented channel; this only stops trigger-origin ids laundering + // themselves into the explicit one. + let mut explicit: Vec = Vec::new(); + for id in newly_approved.into_iter().chain(prior) { + if !explicit.contains(&id) { + explicit.push(id); + } + } + + // The union written back to the trigger: whatever the caller already had + // there, plus everything explicitly authorised. + let mut union: Vec = trigger .get("approvals") .and_then(Value::as_array) .map(|existing| { @@ -304,29 +334,21 @@ pub(super) fn merge_approvals(input: impl Into, newly_approved: Vec tinyflows::caps::Capabilities { // The stub binary refuses every outside-world capability; background // work is no exception, so `spawn` degrades to running inline. tasks: None, + // Likewise for human review: with no provider, an `approval` node + // pauses the run and waits for a resume that the stub never sends. + approvals: None, } } diff --git a/src/model/node_kind.rs b/src/model/node_kind.rs index 939e9672..59199ab9 100644 --- a/src/model/node_kind.rs +++ b/src/model/node_kind.rs @@ -90,6 +90,21 @@ pub enum NodeKind { /// Where lanes end: routing to a gather is a plain activation, so N lanes /// converge on one gather rather than each running their own. Gather, + /// Presents a subject — a URL, a block of text, a generated payload — to a + /// **human** for an approve/reject, and routes on their answer. + /// + /// Distinct from the `requires_approval` flag, which gates a node that + /// would otherwise run: this *is* the review step, it carries what is being + /// reviewed, and its verdict is data (who decided, their comment, any edit + /// they made) that the graph can branch on via its `approved` / `rejected` + /// ports. + /// + /// Reaches the human through the optional + /// [`ApprovalProvider`](crate::caps::ApprovalProvider) capability. With none + /// injected it falls back to pausing the run, so the host settles it with + /// [`engine::resume`](crate::engine::resume) the way it already settles an + /// approval gate. + Approval, /// Waits for tickets — from [`NodeKind::Spawn`], or named by expression — /// and emits their results once its release policy is satisfied. /// diff --git a/src/nodes/execution.rs b/src/nodes/execution.rs index c53618bd..f5155933 100644 --- a/src/nodes/execution.rs +++ b/src/nodes/execution.rs @@ -337,6 +337,7 @@ pub(crate) fn executor_for(kind: &NodeKind) -> Box { NodeKind::Gather => Box::new(control_flow::GatherNode), NodeKind::Spawn => Box::new(integration::SpawnNode), NodeKind::Gate => Box::new(integration::GateNode), + NodeKind::Approval => Box::new(integration::ApprovalNode), NodeKind::Loop => Box::new(control_flow::LoopNode), } } diff --git a/src/nodes/integration/approval.rs b/src/nodes/integration/approval.rs new file mode 100644 index 00000000..e1f1f042 --- /dev/null +++ b/src/nodes/integration/approval.rs @@ -0,0 +1,367 @@ +//! The `approval` node: put something in front of a human, route on their answer. +//! +//! A workflow that posts, pays, emails, or deletes wants a person in the loop at +//! exactly one point, holding exactly one thing — a URL to look at, a draft to +//! read, a payload to sign off. That is what this node is: it carries the +//! **subject** of the review, hands it to the host's review surface through the +//! [`ApprovalProvider`](crate::caps::ApprovalProvider) capability, and emits the +//! verdict on its `approved` / `rejected` ports as ordinary data the graph can +//! branch on. +//! +//! # Not the same thing as `requires_approval` +//! +//! The `requires_approval` flag gates a node that would otherwise run: it says +//! "don't execute this until someone says go", carries nothing, and its answer +//! is a yes/no the graph cannot inspect. This node is the review *itself* — +//! addressable, with a subject, a reviewer, a comment, and a rejection branch. +//! Use the flag to hold back a dangerous node; use this kind when the decision +//! is a step in the workflow. +//! +//! # How waiting works +//! +//! Same two shapes as [`gate`](super::gate), and for the same reasons. A human +//! review is measured in minutes-to-days, so `wait_mode: "suspend"` (the +//! **default** here, unlike `gate`) interrupts the run: nothing is burned while +//! the card sits in someone's queue, and the host resumes the run when they +//! answer. `wait_mode: "poll"` re-activates the node on an interval instead, +//! which is right only when the decision is expected within seconds — each poll +//! costs a super-step and a node visit against the run's budgets, so the poll +//! count is bounded here rather than left to the run-level backstop. +//! +//! # Where a decision can come from +//! +//! In priority order, because more than one channel can be live at once: +//! +//! 1. **The resume value** ([`NodeContext::resume`]) — the checkpointed-resume +//! path, which replays from the checkpoint rather than re-running with a +//! merged run input, so it is the *only* channel there. A rejection in the +//! engine's `{"rejected": []}` shape wins over everything, matching +//! how a `requires_approval` gate treats a denial. +//! 2. **The run's approvals list** (`run.trigger.approvals`) — the +//! re-execute resume path, and how `engine::resume` has always delivered an +//! approval. Listing this node's id there approves it. +//! 3. **The host's [`ApprovalProvider`]** — asked once per activation, under +//! the create-or-fetch contract documented on the trait. +//! 4. **Nobody**, on a host that wired no provider: the node simply waits, so +//! it reduces to a pause the host settles through `engine::resume`. + +use async_trait::async_trait; +use serde_json::{Value, json}; + +use crate::caps::{ApprovalOutcome, ApprovalRequest}; +use crate::data::Item; +use crate::error::{EngineError, Result}; +use crate::nodes::{NodeContext, NodeExecutor, NodeOutput, resolve_config_traced}; + +#[path = "approval_request.rs"] +mod approval_request; +use approval_request::{build_request, decided_item, decision_meta, delivered}; + +/// Default gap between polls, in milliseconds. A second, not the `gate`'s 250ms: +/// nothing a human does resolves faster than that. +const DEFAULT_POLL_INTERVAL_MS: u64 = 1_000; + +/// Default ceiling on polls before the wait budget is called spent. With the +/// default interval that is a minute of waiting — deliberately short, because +/// polling is the wrong mode for a long review and the timeout should say so +/// rather than quietly spending a run's whole visit budget. +const DEFAULT_MAX_POLLS: u64 = 60; + +/// The slot key the poll count is recorded under, so it survives a checkpoint +/// the way a `loop` node's iteration does. +const POLLS_KEY: &str = "polls"; + +/// How the node waits for a human. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum WaitMode { + /// Interrupt the run; the host resumes it when the answer lands. The + /// default — a review is a human-timescale wait. + Suspend, + /// Re-activate on an interval and ask again. Only sane for a decision + /// expected within seconds. + Poll, +} + +impl WaitMode { + fn from_config(config: &Value) -> Self { + match config.get("wait_mode").and_then(Value::as_str) { + Some("poll") => Self::Poll, + _ => Self::Suspend, + } + } +} + +/// What a rejection does to the graph. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum OnReject { + /// Emit the verdict on the `rejected` port (default), so a graph can run a + /// recovery branch — notify the author, revise, ask again. + Route, + /// Fail the node, letting the ordinary `on_error` policy take over. + Error, + /// Emit nothing and let the branch end here. + Drop, +} + +impl OnReject { + fn from_config(config: &Value) -> Self { + match config.get("on_reject").and_then(Value::as_str) { + Some("error") => Self::Error, + Some("drop") => Self::Drop, + _ => Self::Route, + } + } +} + +/// What a spent poll budget does. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum OnTimeout { + /// Fail the node (default): nobody answered, and pretending otherwise is + /// how an unreviewed payload ships. + Error, + /// Treat silence as a rejection and follow the `on_reject` policy. + Reject, + /// Emit on the `timeout` port, so escalation is its own branch. + Route, +} + +impl OnTimeout { + fn from_config(config: &Value) -> Self { + match config.get("on_timeout").and_then(Value::as_str) { + Some("reject") => Self::Reject, + Some("route") => Self::Route, + _ => Self::Error, + } + } +} + +/// Presents a subject to a human and routes on approve / reject. +#[derive(Debug, Default, Clone)] +pub struct ApprovalNode; + +/// Reads a positive integer config field, falling back to `default`. +fn positive_u64(config: &Value, key: &str, default: u64) -> u64 { + config + .get(key) + .and_then(Value::as_u64) + .filter(|n| *n > 0) + .unwrap_or(default) +} + +#[async_trait] +impl NodeExecutor for ApprovalNode { + async fn execute(&self, ctx: NodeContext<'_>) -> Result { + let (config, diagnostics) = resolve_config_traced(&ctx); + let request = build_request(&ctx, &config)?; + let input = ctx + .input + .first() + .map(|item| item.json.clone()) + .unwrap_or(Value::Null); + + // Ask the host only when nothing has already settled this review: a + // resume or a listed approval is the answer, and re-asking would put a + // decided review back in front of the provider. + let outcome = match delivered(&ctx, &request) { + Some(decision) => { + // A provider may have a card open for this review from an + // earlier activation's `decide` call (it went `Pending`, or + // this run never asked because the answer already arrived some + // other way). Either way nobody is waiting on the provider's + // card any more, so withdraw it rather than leave a stale entry + // in the host's queue. Best-effort: the run has already decided + // what to do, and a failed withdrawal must not change that. + if let Some(provider) = ctx.caps.approvals.as_ref() { + if let Err(err) = provider + .cancel(&request.request_id, "resolved via resume") + .await + { + tracing::warn!( + node = %ctx.node.id, + request = %request.request_id, + error = %err, + "withdrawing the provider's review after a resume decision failed" + ); + } + } + ApprovalOutcome::Decided(decision) + } + None => match ctx.caps.approvals.as_ref() { + Some(provider) => provider.decide(&request).await?, + // No provider: the node is a pause the host settles out of band + // through `engine::resume`, which is exactly what waiting does. + None => ApprovalOutcome::Pending, + }, + }; + + let decision = match outcome { + ApprovalOutcome::Decided(decision) => decision, + ApprovalOutcome::Pending => return self.wait(&ctx, &config, &request).await, + }; + + tracing::info!( + node = %ctx.node.id, + request = %request.request_id, + approved = decision.approved, + "approval decided" + ); + + let meta = decision_meta(&decision, &request); + if decision.approved { + return Ok(NodeOutput::routed( + vec![decided_item(&request, &decision, input)], + "approved", + ) + .with_meta(meta) + .with_diagnostics(diagnostics)); + } + + Ok(match OnReject::from_config(&config) { + OnReject::Route => { + NodeOutput::routed(vec![decided_item(&request, &decision, input)], "rejected") + .with_meta(meta) + .with_diagnostics(diagnostics) + } + OnReject::Drop => NodeOutput::empty() + .with_meta(meta) + .with_diagnostics(diagnostics), + OnReject::Error => { + return Err(EngineError::Capability(format!( + "approval node {:?}: rejected by {}{}", + ctx.node.id, + decision.decided_by.as_deref().unwrap_or("a reviewer"), + decision + .comment + .as_deref() + .map(|c| format!(" ({c})")) + .unwrap_or_default(), + ))); + } + }) + } +} + +impl ApprovalNode { + /// Nobody has decided yet: suspend the run, or spend a poll. + async fn wait( + &self, + ctx: &NodeContext<'_>, + config: &Value, + request: &ApprovalRequest, + ) -> Result { + if WaitMode::from_config(config) == WaitMode::Suspend { + // Suspending discards this activation's update, so no poll is + // charged — a suspended review is not spending a budget, it is + // waiting for a person. The payload is what the host renders or + // routes if it did not get the request through a provider. + return Ok(NodeOutput::interrupt( + ctx.node.id.clone(), + json!({ + "kind": "approval", + "node": ctx.node.id, + "request_id": request.request_id, + "title": request.title, + "prompt": request.prompt, + "subject": request.subject.value, + "subject_kind": request.subject.kind, + "assignees": request.assignees, + "metadata": request.metadata, + }), + )); + } + + let polls = ctx + .nodes + .get(&ctx.node.id) + .and_then(|slot| slot.get(POLLS_KEY)) + .and_then(Value::as_u64) + .unwrap_or(0); + let max_polls = positive_u64(config, "max_polls", DEFAULT_MAX_POLLS); + let meta = json!({ POLLS_KEY: polls + 1, "request_id": request.request_id }); + + // `polls` counts activations that have already asked the provider + // once (this one included), so re-entering when `polls + 1 < + // max_polls` — rather than `polls < max_polls` — is what makes + // `max_polls` the number of `decide` calls actually charged, not + // `max_polls + 1`: this activation's call is the `polls + 1`-th. + if polls + 1 < max_polls { + let interval = positive_u64(config, "poll_interval_ms", DEFAULT_POLL_INTERVAL_MS); + return Ok(NodeOutput::reenter_after(interval, meta)); + } + + // Budget spent. Whatever happens next, nobody is waiting on this review + // any more, so withdraw it rather than leaving a dead card in a queue. + // Best-effort: the run has already decided what to do, and a failed + // withdrawal must not change that. + if let Some(provider) = ctx.caps.approvals.as_ref() { + if let Err(err) = provider + .cancel(&request.request_id, "approval node timed out") + .await + { + tracing::warn!( + node = %ctx.node.id, + request = %request.request_id, + error = %err, + "withdrawing the timed-out review failed" + ); + } + } + + // The `=nodes..decision.approved` contract other settled reviews + // give the graph applies here too: a downstream guard reading it after + // a timeout must see `false`, not an absent value, and a `rejected` + // recovery branch reading `=item.comment` / `=item.input` must not get + // `null` back just because the review timed out rather than being + // actively declined. + let comment = format!("no decision after {max_polls} polls"); + let timed_out = json!({ + "approved": false, + "timed_out": true, + "request_id": request.request_id, + "subject": request.subject.value, + "subject_kind": request.subject.kind, + "edited": false, + "decided_by": Value::Null, + "comment": comment, + "input": ctx.input.first().map(|item| item.json.clone()).unwrap_or(Value::Null), + }); + let meta = json!({ + POLLS_KEY: polls + 1, + "request_id": request.request_id, + "decision": { + "approved": false, + "timed_out": true, + "decided_by": Value::Null, + "comment": comment, + "request_id": request.request_id, + } + }); + + match OnTimeout::from_config(config) { + OnTimeout::Error => Err(EngineError::Capability(format!( + "approval node {:?}: no decision after {max_polls} polls; raise \ + `max_polls`/`poll_interval_ms`, switch to `wait_mode: \"suspend\"`, or wire a \ + `timeout` port", + ctx.node.id + ))), + OnTimeout::Route => { + Ok(NodeOutput::routed(vec![Item::new(timed_out)], "timeout").with_meta(meta)) + } + OnTimeout::Reject => match OnReject::from_config(config) { + OnReject::Route => { + Ok(NodeOutput::routed(vec![Item::new(timed_out)], "rejected").with_meta(meta)) + } + OnReject::Drop => Ok(NodeOutput::empty().with_meta(meta)), + OnReject::Error => Err(EngineError::Capability(format!( + "approval node {:?}: no decision after {max_polls} polls, and \ + `on_timeout: \"reject\"` with `on_reject: \"error\"` fails the node", + ctx.node.id + ))), + }, + } + } +} + +#[cfg(test)] +#[path = "approval_tests.rs"] +mod tests; diff --git a/src/nodes/integration/approval_request.rs b/src/nodes/integration/approval_request.rs new file mode 100644 index 00000000..fca38399 --- /dev/null +++ b/src/nodes/integration/approval_request.rs @@ -0,0 +1,305 @@ +//! Building an [`ApprovalRequest`] from a node's config, and reading a +//! settled [`ApprovalDecision`] back out of a resume value, the run's +//! approvals list, or a finished review. +//! +//! Split out of `approval.rs` to keep that file under the repository's +//! line-length limit; request-building and decision-reading are one cohesive +//! concern (every function here is pure data shaping, none of it waits or +//! calls a capability), so they belong together rather than split further. + +use serde_json::{Value, json}; + +use crate::caps::{ApprovalDecision, ApprovalRequest, ApprovalSubject}; +use crate::data::Item; +use crate::error::{EngineError, Result}; +use crate::nodes::NodeContext; + +/// The default rendering hint when the graph does not say what the subject is. +const DEFAULT_SUBJECT_KIND: &str = "json"; + +/// The run's host id, read **only** from the run-level slots a host seeds +/// (`run.id`, `run.run_id`) — never from the trigger payload. +/// +/// Whichever of these is populated becomes part of `request_id`, the provider's +/// create-or-fetch key, so this lookup is a trust boundary rather than a +/// convenience. `run.trigger.*` is the payload a caller hands to +/// `engine::run`; for a webhook or any user-facing trigger that payload is +/// attacker-influenced, and reading a run id out of it would let an attacker +/// pick a `request_id` colliding with an earlier run and inherit its cached +/// decision — approving a new, unreviewed subject without a human ever seeing +/// it. `run.id` / `run.run_id` sit outside the trigger, so a host puts a +/// server-generated value there deliberately. +/// +/// A host must still seed a **server-generated** id (or set +/// `config.request_id` from one) and never copy a caller-supplied field into +/// these slots: the crate is host-agnostic and cannot tell trusted run +/// metadata from untrusted, so enforcing that is the host's job, as it is for +/// any identity used as a de-duplication or idempotency key. +fn run_id(ctx: &NodeContext<'_>) -> Option { + ["id", "run_id"] + .iter() + .find_map(|key| ctx.run.get(*key).and_then(Value::as_str)) + .map(str::to_string) +} + +/// Builds the review request from the node's resolved config. +/// +/// The `request_id` is what makes the provider's create-or-fetch contract +/// work, so it must be **stable across activations**: an interrupt discards the +/// activation's state update, so the node cannot remember an id it generated, +/// and anything derived from the clock or a counter would create a fresh review +/// on every resume. Hence run id + node id, or an explicit `config.request_id` +/// for a host that wants to key reviews its own way. +/// +/// Falling back to the bare node id when *neither* is available would let two +/// different runs of the same graph collide on the same `request_id`: since +/// [`ApprovalProvider::decide`](crate::caps::ApprovalProvider::decide) is +/// create-or-fetch, a later run would silently inherit an earlier run's +/// decision and route an unreviewed subject straight through `approved`. So a +/// node with no `config.request_id` and no run-scoped identity is a +/// configuration error, not a degraded default. +pub(super) fn build_request(ctx: &NodeContext<'_>, config: &Value) -> Result { + let run = run_id(ctx); + let request_id = match config.get("request_id").and_then(Value::as_str) { + Some(explicit) => explicit.to_string(), + None => match &run { + Some(run) => format!("{run}:{}", ctx.node.id), + None => { + return Err(EngineError::Capability(format!( + "approval node {:?}: no `request_id` configured and no run-scoped identity \ + available (expected `run.id` or `run.run_id`, which a host seeds outside the \ + caller-supplied trigger payload); set \ + `config.request_id` explicitly or seed a run id, otherwise later runs could \ + reuse an earlier run's decision", + ctx.node.id + ))); + } + }, + }; + + // The subject defaults to the item that arrived, which is the common case: + // a node upstream produced the thing, and the human looks at it. + let value = config + .get("subject") + .cloned() + .or_else(|| ctx.input.first().map(|item| item.json.clone())) + .unwrap_or(Value::Null); + + // `validate::validate_all` only sees `assignees` as authored: a literal + // non-array (a bare string, the natural single-reviewer mistake) or a + // literal empty array are both refused there. An `=`-bound `assignees` + // (e.g. `"=item.reviewers"`) is a string at author time — it passes that + // check by looking like *some* other field entirely — and resolves to + // its real shape only here, at execution time. So the same two refusals + // apply again to the resolved value: present and not an array, or + // present, an array, and empty (or empty of strings) once resolved. Both + // reach the same nobody-reviews-this audience a validated graph should + // never produce. + let assignees = match config.get("assignees") { + None => Vec::new(), + Some(value) => match value.as_array() { + None => { + return Err(EngineError::Capability(format!( + "approval node {:?}: `assignees` resolved to {value}, not an array of strings", + ctx.node.id + ))); + } + Some(values) => { + // Every element must be a string. Dropping the ones that are + // not would route the review to a *different* audience than the + // graph asked for, silently: `["=item.reviewer", 42]` would + // quietly become a one-reviewer list. Losing a reviewer is + // exactly the kind of quiet change a review must not make. + let mut assignees: Vec = Vec::with_capacity(values.len()); + for value in values { + let Some(handle) = value.as_str() else { + return Err(EngineError::Capability(format!( + "approval node {:?}: `assignees` entry {value} is not a string; a \ + reviewer handle that resolved to something else would be dropped \ + and the review routed to a smaller audience than authored", + ctx.node.id + ))); + }; + assignees.push(handle.to_string()); + } + if assignees.is_empty() { + return Err(EngineError::Capability(format!( + "approval node {:?}: `assignees` resolved to an empty array; a review \ + with nobody assigned can never be resolved", + ctx.node.id + ))); + } + assignees + } + }, + }; + + Ok(ApprovalRequest { + request_id, + node_id: ctx.node.id.clone(), + run_id: run, + title: string_field(config, "title"), + prompt: string_field(config, "prompt"), + subject: ApprovalSubject { + kind: config + .get("subject_kind") + .and_then(Value::as_str) + .unwrap_or(DEFAULT_SUBJECT_KIND) + .to_string(), + value, + }, + assignees, + metadata: config.get("metadata").cloned().unwrap_or(Value::Null), + }) +} + +/// A config field read as a string, ignoring a non-string (an unresolved +/// expression that came back `null`, say). +fn string_field(config: &Value, key: &str) -> Option { + config.get(key).and_then(Value::as_str).map(str::to_string) +} + +/// Whether `list` (an array of strings) names this node or its review. +pub(super) fn names(list: Option<&Value>, request: &ApprovalRequest) -> bool { + list.and_then(Value::as_array).is_some_and(|ids| { + ids.iter() + .filter_map(Value::as_str) + .any(|id| id == request.node_id || id == request.request_id) + }) +} + +/// Reads a decision out of a resume value, if it carries one. +/// +/// Accepts the engine's own `{"rejected": […]}` denial shape (checked +/// first, so a denial always beats an approval delivered in the same value), +/// the mirror `{"approved": […]}`, and a full verdict object — either +/// inline or nested under `decision`. +/// +/// The array forms above are always scoped to this request via [`names`], the +/// same way [`super::gate`](crate::nodes::integration::gate) scopes its own +/// `approved` array — required, because several nodes can be interrupted at +/// once and a resume value is not addressed to just one of them. The inline +/// verdict-object form carries no such array to check, so it is accepted +/// unscoped **only** when it does not itself name a different request — +/// matching the same "single-interrupt convenience" precedent +/// `engine::build::activation`'s bare `Value::Bool(true)` case documents, but +/// without silently absorbing a verdict a host explicitly addressed elsewhere. +pub(super) fn decision_from_resume( + resume: &Value, + request: &ApprovalRequest, +) -> Option { + if names(resume.get("rejected"), request) { + return Some(ApprovalDecision::rejected( + resume + .get("comment") + .and_then(Value::as_str) + .map(str::to_string), + )); + } + if names(resume.get("approved"), request) { + return Some(ApprovalDecision::approved()); + } + + // A verdict object must say WHICH review it settles. One resume value is + // delivered to every interrupted node, so an unaddressed `{"approved": + // true}` would settle whichever reviews happen to read it — approving every + // pending review at once, without the sender needing to know a single id. + // An unaddressed verdict is therefore ignored rather than assumed to be + // ours; the array forms carry their ids and stay supported. + let verdict = resume.get("decision").unwrap_or(resume); + let named = verdict + .get("node_id") + .or_else(|| verdict.get("request_id")) + .and_then(Value::as_str)?; + if named != request.node_id && named != request.request_id { + // Addressed to a different node's review; not ours to take. + return None; + } + let approved = verdict.get("approved").and_then(Value::as_bool)?; + Some(ApprovalDecision { + approved, + decided_by: string_field(verdict, "decided_by"), + comment: string_field(verdict, "comment"), + payload: verdict.get("payload").cloned(), + }) +} + +/// The decision already in hand before the host is asked: a resume value, or +/// this node's id on the run's approvals list. +pub(super) fn delivered( + ctx: &NodeContext<'_>, + request: &ApprovalRequest, +) -> Option { + if let Some(decision) = ctx + .resume + .as_ref() + .and_then(|resume| decision_from_resume(resume, request)) + { + return Some(decision); + } + + // The re-execute resume path: `engine::resume` merges newly-approved ids + // into the run input, and they arrive here as the top-level `run.approvals` + // — the **explicit** channel (`RunInput::approvals`), which a host fills + // deliberately. + // + // `run.trigger.approvals` is deliberately NOT read, even though + // `engine::resume` also writes the merged list there. The trigger is the + // payload a caller hands to `engine::run`, so honouring it would let anyone + // who can start a run post `{"approvals": [""]}` and approve their + // own review on the initial execution — skipping the human entirely. A + // review that can be self-approved by its own subject is not a review. + // + // Known residual, and why it is not fixed here: `merge_approvals` seeds its + // starting set from `trigger["approvals"]`, so a trigger-supplied id is + // folded into the explicit list *on a resume*. That is pre-existing engine + // behaviour shared with the `requires_approval` gate in + // `engine::build::activation`, and narrowing it changes resume semantics for + // every gate, not just this node — so it belongs in its own change rather + // than riding along here. The initial-run bypass, which is the reachable- + // without-a-host-action one, is closed. + if names(ctx.run.get("approvals"), request) { + return Some(ApprovalDecision::approved()); + } + None +} + +/// The item a settled review emits. +/// +/// `subject` is what the human actually signed off on — their edit when the +/// host's surface allowed one, otherwise exactly what was sent — so a +/// downstream node reads one field regardless. The original input is kept under +/// `input` so nothing is lost when the subject was a projection of it. +pub(super) fn decided_item( + request: &ApprovalRequest, + decision: &ApprovalDecision, + input: Value, +) -> Item { + Item::new(json!({ + "approved": decision.approved, + "subject": decision + .payload + .clone() + .unwrap_or_else(|| request.subject.value.clone()), + "subject_kind": request.subject.kind, + "edited": decision.payload.is_some(), + "decided_by": decision.decided_by, + "comment": decision.comment, + "request_id": request.request_id, + "input": input, + })) +} + +/// The slot state a settled review records, so `=nodes..decision.approved` +/// resolves from anywhere in the graph — including from a branch that did not +/// receive the emitted item (a `drop`ped rejection has no item at all). +pub(super) fn decision_meta(decision: &ApprovalDecision, request: &ApprovalRequest) -> Value { + json!({ + "decision": { + "approved": decision.approved, + "decided_by": decision.decided_by, + "comment": decision.comment, + "request_id": request.request_id, + } + }) +} diff --git a/src/nodes/integration/approval_tests.rs b/src/nodes/integration/approval_tests.rs new file mode 100644 index 00000000..16f52f33 --- /dev/null +++ b/src/nodes/integration/approval_tests.rs @@ -0,0 +1,76 @@ +use super::approval_request::{decision_from_resume, names}; +use super::*; +use serde_json::json; + +use crate::caps::ApprovalSubject; +use crate::caps::mock::{MockApprovals, mock_capabilities, mock_capabilities_with_approvals}; +use crate::compiler::compile; +use crate::engine::{RunInput, resume, run}; +use crate::model::{Edge, Node, NodeKind, WorkflowGraph}; + +/// A trigger wired into one `approval` node, with `config` on the approval. +/// +/// Fills in a stable `request_id` when the caller's config does not already +/// name one, since without a run-scoped identity `build_request` now refuses +/// to guess one (see `a_missing_request_id_and_run_id_is_a_configuration_error` +/// below for the case that tests the refusal itself). +fn wf(mut config: Value) -> WorkflowGraph { + if let Some(obj) = config.as_object_mut() { + obj.entry("request_id".to_string()) + .or_insert_with(|| json!("review-request")); + } + wf_raw(config) +} + +/// [`wf`] without the `request_id` auto-fill, for tests that need to control +/// exactly what identity information the node config and run carry. +fn wf_raw(config: Value) -> WorkflowGraph { + 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: "review".into(), + kind: NodeKind::Approval, + type_version: 1, + name: "review".into(), + config, + ports: vec![], + position: None, + }, + ], + edges: vec![Edge { + from_node: "t".into(), + from_port: "main".into(), + to_node: "review".into(), + to_port: "main".into(), + }], + ..Default::default() + } +} + +fn request(id: &str) -> ApprovalRequest { + ApprovalRequest { + request_id: id.to_string(), + node_id: "review".to_string(), + run_id: None, + title: None, + prompt: None, + subject: ApprovalSubject { + kind: "url".to_string(), + value: json!("https://example.com/post/1"), + }, + assignees: vec![], + metadata: Value::Null, + } +} + +include!("approval_tests/approval_tests_part_01_tests.rs"); +include!("approval_tests/approval_tests_part_02_tests.rs"); diff --git a/src/nodes/integration/approval_tests/approval_tests_part_01_tests.rs b/src/nodes/integration/approval_tests/approval_tests_part_01_tests.rs new file mode 100644 index 00000000..29f5ed44 --- /dev/null +++ b/src/nodes/integration/approval_tests/approval_tests_part_01_tests.rs @@ -0,0 +1,396 @@ +/// Opposite default to `gate`, and deliberately so: a human review is a +/// minutes-to-days wait, and polling one burns super-steps for nothing. +#[test] +fn wait_mode_defaults_to_suspend() { + assert_eq!(WaitMode::from_config(&json!({})), WaitMode::Suspend); + assert_eq!( + WaitMode::from_config(&json!({ "wait_mode": "poll" })), + WaitMode::Poll + ); + assert_eq!( + WaitMode::from_config(&json!({ "wait_mode": "nonsense" })), + WaitMode::Suspend + ); +} + +/// Failing safe: an unrecognised policy must not drop a rejection on the floor +/// or fail a run that had a perfectly good recovery branch. +#[test] +fn reject_and_timeout_policies_default_conservatively() { + assert_eq!(OnReject::from_config(&json!({})), OnReject::Route); + assert_eq!( + OnReject::from_config(&json!({ "on_reject": "whatever" })), + OnReject::Route + ); + assert_eq!(OnTimeout::from_config(&json!({})), OnTimeout::Error); + assert_eq!( + OnTimeout::from_config(&json!({ "on_timeout": "eventually" })), + OnTimeout::Error + ); +} + +/// The engine's own denial shape wins over an approval delivered in the same +/// resume value — the same precedence a `requires_approval` gate applies. +#[test] +fn resume_denial_beats_an_approval_in_the_same_value() { + let req = request("run-1:review"); + let decision = decision_from_resume( + &json!({ "rejected": ["review"], "approved": ["review"] }), + &req, + ) + .expect("a decision"); + assert!(!decision.approved); +} + +#[test] +fn resume_reads_a_verdict_inline_or_nested() { + let req = request("run-1:review"); + + let inline = decision_from_resume( + &json!({ "node_id": "review", "approved": true, "decided_by": "ada", "comment": "ship it" }), + &req, + ) + .expect("a decision"); + assert!(inline.approved); + assert_eq!(inline.decided_by.as_deref(), Some("ada")); + assert_eq!(inline.comment.as_deref(), Some("ship it")); + + let nested = decision_from_resume( + &json!({ "decision": { "request_id": "run-1:review", "approved": false, "comment": "wrong link" } }), + &req, + ) + .expect("a decision"); + assert!(!nested.approved); + assert_eq!(nested.comment.as_deref(), Some("wrong link")); + + assert!( + decision_from_resume(&json!({ "unrelated": true }), &req).is_none(), + "a resume that says nothing about this review leaves it pending" + ); +} + +/// A verdict object must name the review it settles. +/// +/// One resume value is delivered to **every** interrupted node, so an +/// unaddressed `{"approved": true}` would settle whichever reviews read it — +/// approving every pending review in the run at once, without the sender +/// needing to know a single id. And a verdict naming a *different* review is +/// not an answer for this one. Both are refused; only an explicit match is +/// taken. +#[test] +fn a_verdict_object_must_name_the_review_it_settles() { + let req = request("run-1:review"); + + let unaddressed = decision_from_resume(&json!({ "approved": true }), &req); + assert!( + unaddressed.is_none(), + "an unaddressed verdict must not settle this review: the same resume value \ + reaches every interrupted node, so it would approve all of them at once" + ); + + let unaddressed_nested = + decision_from_resume(&json!({ "decision": { "approved": true } }), &req); + assert!( + unaddressed_nested.is_none(), + "the nested form carries the same risk and gets the same refusal" + ); + + let for_someone_else = decision_from_resume( + &json!({ "approved": true, "node_id": "other-review" }), + &req, + ); + assert!( + for_someone_else.is_none(), + "a verdict explicitly naming a different node must not settle this review" + ); + + let for_someone_elses_request_id = decision_from_resume( + &json!({ "decision": { "approved": true, "request_id": "run-1:other" } }), + &req, + ); + assert!( + for_someone_elses_request_id.is_none(), + "a verdict explicitly naming a different request_id must not settle this review" + ); + + let for_us = decision_from_resume(&json!({ "approved": true, "node_id": "review" }), &req) + .expect("a verdict naming this node's id is ours to take"); + assert!(for_us.approved); +} + +/// A review can be addressed by node id or by its own request id — a host that +/// tracks reviews by request id must be able to resume with that. +#[test] +fn a_review_answers_to_its_node_id_and_its_request_id() { + let req = request("run-1:review"); + assert!(names(Some(&json!(["review"])), &req)); + assert!(names(Some(&json!(["run-1:review"])), &req)); + assert!(!names(Some(&json!(["other"])), &req)); + assert!(!names(None, &req)); +} + +#[test] +fn positive_config_fields_fall_back_on_zero_or_garbage() { + assert_eq!(positive_u64(&json!({}), "max_polls", 7), 7); + assert_eq!(positive_u64(&json!({ "max_polls": 0 }), "max_polls", 7), 7); + assert_eq!(positive_u64(&json!({ "max_polls": 3 }), "max_polls", 7), 3); +} + +#[tokio::test] +async fn an_approved_review_emits_the_subject_on_the_approved_port() { + let graph = wf(json!({ + "title": "Publish this?", + "subject_kind": "url", + "subject": "=item.url", + })); + let compiled = compile(&graph).expect("compile"); + let out = run( + &compiled, + json!({ "url": "https://example.com/post/1" }), + &mock_capabilities(), + ) + .await + .expect("run"); + + let slot = &out.output["nodes"]["review"]; + assert_eq!(slot["port"], "approved"); + let item = &slot["items"][0]["json"]; + assert_eq!(item["approved"], json!(true)); + assert_eq!(item["subject"], "https://example.com/post/1"); + assert_eq!(item["subject_kind"], "url"); + assert_eq!(item["decided_by"], "mock-reviewer"); + assert_eq!( + slot["decision"]["approved"], + json!(true), + "the verdict is addressable as =nodes.review.decision.approved" + ); +} + +#[tokio::test] +async fn the_subject_defaults_to_the_item_that_arrived() { + let graph = wf(json!({ "title": "Look at this" })); + let compiled = compile(&graph).expect("compile"); + let out = run( + &compiled, + json!({ "draft": "hello world" }), + &mock_capabilities(), + ) + .await + .expect("run"); + + assert_eq!( + out.output["nodes"]["review"]["items"][0]["json"]["subject"], + json!({ "draft": "hello world" }) + ); +} + +#[tokio::test] +async fn a_rejection_routes_to_the_rejected_port_with_its_reason() { + let graph = wf(json!({ "title": "Publish this?" })); + let compiled = compile(&graph).expect("compile"); + let out = run( + &compiled, + json!({ "url": "https://example.com" }), + &mock_capabilities_with_approvals(MockApprovals::rejecting()), + ) + .await + .expect("run"); + + let slot = &out.output["nodes"]["review"]; + assert_eq!(slot["port"], "rejected"); + assert_eq!(slot["items"][0]["json"]["approved"], json!(false)); + assert_eq!(slot["items"][0]["json"]["comment"], "mock rejection"); +} + +#[tokio::test] +async fn on_reject_error_fails_the_node_with_the_reviewer_and_reason() { + let graph = wf(json!({ "on_reject": "error" })); + let compiled = compile(&graph).expect("compile"); + let err = run( + &compiled, + Value::Null, + &mock_capabilities_with_approvals(MockApprovals::rejecting()), + ) + .await + .expect_err("a rejection with on_reject: error fails the run"); + + let message = err.to_string(); + assert!(message.contains("mock-reviewer"), "got {message}"); + assert!(message.contains("mock rejection"), "got {message}"); +} + +#[tokio::test] +async fn a_pending_review_pauses_the_run_and_names_itself() { + let graph = wf(json!({ "title": "Publish this?" })); + let compiled = compile(&graph).expect("compile"); + let out = run( + &compiled, + Value::Null, + &mock_capabilities_with_approvals(MockApprovals::pending()), + ) + .await + .expect("run"); + + assert_eq!(out.pending_approvals, vec!["review".to_string()]); +} + +/// The zero-capability path: a host that wired no provider still gets a +/// working node, because waiting *is* the pause it already knows how to resume. +#[tokio::test] +async fn with_no_provider_the_node_reduces_to_a_pause_the_host_resumes() { + let caps = crate::caps::Capabilities { + approvals: None, + ..mock_capabilities() + }; + let graph = wf(json!({ "title": "Publish this?" })); + let compiled = compile(&graph).expect("compile"); + + let paused = run(&compiled, Value::Null, &caps).await.expect("run"); + assert_eq!(paused.pending_approvals, vec!["review".to_string()]); + + let resumed = resume(&compiled, Value::Null, vec!["review".to_string()], &caps) + .await + .expect("resume"); + assert!(resumed.pending_approvals.is_empty()); + assert_eq!(resumed.output["nodes"]["review"]["port"], "approved"); +} + +/// An approval already on the run input settles the review without the host +/// ever being asked — otherwise a resume would re-open a decided review. +#[tokio::test] +async fn a_listed_approval_settles_the_review_without_asking_the_host() { + let graph = wf(json!({ "title": "Publish this?" })); + let compiled = compile(&graph).expect("compile"); + let provider = std::sync::Arc::new(MockApprovals::pending()); + let caps = crate::caps::Capabilities { + approvals: Some(provider.clone()), + ..mock_capabilities() + }; + + let out = run( + &compiled, + RunInput::new(Value::Null).with_approvals(vec!["review".to_string()]), + &caps, + ) + .await + .expect("run"); + + assert_eq!(out.output["nodes"]["review"]["port"], "approved"); + assert!( + provider.requested().is_empty(), + "a settled review must not be handed to the provider again" + ); +} + +#[tokio::test] +async fn polling_spends_a_bounded_budget_then_follows_on_timeout() { + let graph = wf(json!({ + "wait_mode": "poll", + "poll_interval_ms": 1, + "max_polls": 2, + "on_timeout": "route", + })); + let compiled = compile(&graph).expect("compile"); + let provider = std::sync::Arc::new(MockApprovals::pending()); + let caps = crate::caps::Capabilities { + approvals: Some(provider.clone()), + ..mock_capabilities() + }; + + let out = run(&compiled, Value::Null, &caps).await.expect("run"); + + let slot = &out.output["nodes"]["review"]; + assert_eq!(slot["port"], "timeout"); + assert_eq!(slot["items"][0]["json"]["timed_out"], json!(true)); + + // Every activation asked about the SAME review: the create-or-fetch + // contract is what stops a poll loop notifying a human once per poll. + let ids = provider.requested(); + assert_eq!( + ids.len(), + 2, + "max_polls: 2 must charge exactly 2 decide calls, not max_polls + 1, got {ids:?}" + ); + assert!( + ids.iter().all(|id| id == &ids[0]), + "every poll must reuse one request id, got {ids:?}" + ); + assert_eq!( + provider.cancelled(), + ids[..1].to_vec(), + "a timed-out review is withdrawn rather than left in a queue" + ); +} + +/// A provider that stays `Pending` for its first `pending_calls` calls, then +/// decides — the shape a real review surface has (nobody has looked yet, then +/// someone does), which [`MockApprovals::pending`] alone cannot exercise since +/// it never resolves. +struct ResolvesAfter { + pending_calls: usize, + calls: std::sync::atomic::AtomicUsize, +} + +#[async_trait::async_trait] +impl crate::caps::ApprovalProvider for ResolvesAfter { + async fn decide( + &self, + _request: &crate::caps::ApprovalRequest, + ) -> crate::error::Result { + let call = self + .calls + .fetch_add(1, std::sync::atomic::Ordering::SeqCst); + if call < self.pending_calls { + return Ok(crate::caps::ApprovalOutcome::Pending); + } + Ok(crate::caps::ApprovalOutcome::Decided( + crate::caps::ApprovalDecision { + decided_by: Some("late-reviewer".to_string()), + ..crate::caps::ApprovalDecision::approved() + }, + )) + } +} + +/// A poll that finally finds a `Decided` outcome must stop polling and take +/// the verdict — proving the polling loop actually reads what `decide` +/// returns each time, rather than always re-entering until the budget runs +/// out. +#[tokio::test] +async fn a_polling_review_resolves_once_the_provider_decides() { + let graph = wf(json!({ + "wait_mode": "poll", + "poll_interval_ms": 1, + "max_polls": 10, + })); + let compiled = compile(&graph).expect("compile"); + let provider = ResolvesAfter { + pending_calls: 2, + calls: std::sync::atomic::AtomicUsize::new(0), + }; + let caps = mock_capabilities_with_approvals(provider); + + let out = run(&compiled, Value::Null, &caps).await.expect("run"); + + let slot = &out.output["nodes"]["review"]; + assert_eq!( + slot["port"], "approved", + "a late decision must still be taken rather than timing out, got {slot:?}" + ); + assert_eq!(slot["items"][0]["json"]["decided_by"], "late-reviewer"); +} + +#[tokio::test] +async fn on_timeout_error_is_the_default_and_names_the_node() { + let graph = wf(json!({ "wait_mode": "poll", "poll_interval_ms": 1, "max_polls": 1 })); + let compiled = compile(&graph).expect("compile"); + let err = run( + &compiled, + Value::Null, + &mock_capabilities_with_approvals(MockApprovals::pending()), + ) + .await + .expect_err("an unanswered review fails by default"); + assert!(err.to_string().contains("review"), "got {err}"); +} + diff --git a/src/nodes/integration/approval_tests/approval_tests_part_02_tests.rs b/src/nodes/integration/approval_tests/approval_tests_part_02_tests.rs new file mode 100644 index 00000000..ee8f7717 --- /dev/null +++ b/src/nodes/integration/approval_tests/approval_tests_part_02_tests.rs @@ -0,0 +1,353 @@ +/// Without an explicit `request_id` and without a run-scoped identity the node +/// must refuse to guess: falling back to the bare node id would let a later +/// run of the same graph reuse an earlier run's decision through the +/// provider's create-or-fetch contract, and route an unreviewed subject +/// straight through `approved`. +#[tokio::test] +async fn a_missing_request_id_and_run_id_is_a_configuration_error() { + let graph = wf_raw(json!({ "title": "Publish this?" })); + let compiled = compile(&graph).expect("compile"); + let err = run(&compiled, Value::Null, &mock_capabilities()) + .await + .expect_err("no request_id and no run id must be refused"); + let message = err.to_string(); + assert!(message.contains("request_id"), "got {message}"); +} + +/// `validate::validate_all` only sees the config as authored — a +/// one-element `assignees` array of `=`-bindings passes it. If every binding +/// resolves to something that is not a string (here: a field the trigger +/// input never set, so the binding reads `null`), the array a real review +/// would be routed to is empty at execution time, and that is refused too. +#[tokio::test] +async fn assignees_resolving_to_no_strings_at_runtime_is_a_configuration_error() { + let graph = wf(json!({ "assignees": ["=item.missing_reviewer"] })); + let compiled = compile(&graph).expect("compile"); + let err = run(&compiled, json!({}), &mock_capabilities()) + .await + .expect_err("assignees resolving to zero strings must be refused"); + let message = err.to_string(); + assert!(message.contains("assignees"), "got {message}"); +} + +/// A `run_id` in the **trigger payload** must NOT become the review's identity. +/// +/// The trigger is caller-supplied — for a webhook or any user-facing trigger it +/// is attacker-influenced — and `request_id` is the provider's create-or-fetch +/// key. Honouring it would let an attacker name a review that collides with an +/// earlier run's and inherit its cached decision, approving an unreviewed +/// subject with no human involved. So the node refuses rather than deriving an +/// id it cannot trust. +#[tokio::test] +async fn a_run_id_in_the_trigger_payload_is_not_trusted_as_the_review_identity() { + let graph = wf_raw(json!({ "title": "Publish this?" })); + let compiled = compile(&graph).expect("compile"); + let err = run( + &compiled, + json!({ "run_id": "run-42" }), + &mock_capabilities(), + ) + .await + .expect_err("a trigger-supplied run id must not be accepted as a review identity"); + + let message = err.to_string(); + assert!( + message.contains("run-scoped identity"), + "the error must say what is missing, got {message}" + ); + assert!( + !message.contains("run-42"), + "the untrusted value must not be echoed as though it were usable, got {message}" + ); +} + +/// The supported way to identify a review when the host seeds no run id: +/// an explicit `config.request_id`, which the author controls. +#[tokio::test] +async fn an_explicit_request_id_identifies_the_review() { + let graph = wf_raw(json!({ "title": "Publish this?", "request_id": "review-of-post-42" })); + let compiled = compile(&graph).expect("compile"); + let out = run(&compiled, json!({ "url": "https://example.com" }), &mock_capabilities()) + .await + .expect("an explicit request_id makes the review identifiable"); + + assert_eq!( + out.output["nodes"]["review"]["items"][0]["json"]["request_id"], + "review-of-post-42" + ); +} + +/// `on_reject: "drop"` emits nothing — a regression that accidentally emitted +/// an item here would go unnoticed by every other rejection test, which all +/// use `route`. +#[tokio::test] +async fn on_reject_drop_emits_nothing_but_still_records_the_decision() { + let graph = wf(json!({ "on_reject": "drop" })); + let compiled = compile(&graph).expect("compile"); + let out = run( + &compiled, + Value::Null, + &mock_capabilities_with_approvals(MockApprovals::rejecting()), + ) + .await + .expect("run"); + + let slot = &out.output["nodes"]["review"]; + assert_eq!( + slot["items"], + json!([]), + "on_reject: drop must not emit an item" + ); + assert_eq!( + slot["decision"]["approved"], + json!(false), + "the verdict stays addressable as =nodes.review.decision.approved even when dropped" + ); +} + +/// `on_timeout: "reject"` hands the timed-out review to the `on_reject` +/// policy. Covers all three `on_reject` sub-paths so a regression in any one +/// of them is caught here rather than by a host. +#[tokio::test] +async fn on_timeout_reject_follows_the_on_reject_policy() { + // route: timed_out item lands on `rejected`, with the fields a settled + // review always carries (not just `approved`/`timed_out`). + let graph = wf(json!({ + "wait_mode": "poll", + "poll_interval_ms": 1, + "max_polls": 1, + "on_timeout": "reject", + })); + let compiled = compile(&graph).expect("compile"); + let out = run( + &compiled, + Value::Null, + &mock_capabilities_with_approvals(MockApprovals::pending()), + ) + .await + .expect("run"); + let slot = &out.output["nodes"]["review"]; + assert_eq!(slot["port"], "rejected"); + let item = &slot["items"][0]["json"]; + assert_eq!(item["approved"], json!(false)); + assert_eq!(item["timed_out"], json!(true)); + assert_eq!(item["edited"], json!(false)); + assert_eq!(item["decided_by"], Value::Null); + assert!( + item["comment"].as_str().is_some_and(|c| !c.is_empty()), + "a timed-out rejection still carries a comment explaining why, got {item:?}" + ); + assert_eq!( + slot["decision"]["approved"], + json!(false), + "=nodes.review.decision.approved must resolve to false after a timeout, not be absent" + ); + + // drop: nothing emitted. + let graph = wf(json!({ + "wait_mode": "poll", + "poll_interval_ms": 1, + "max_polls": 1, + "on_timeout": "reject", + "on_reject": "drop", + })); + let compiled = compile(&graph).expect("compile"); + let out = run( + &compiled, + Value::Null, + &mock_capabilities_with_approvals(MockApprovals::pending()), + ) + .await + .expect("run"); + assert_eq!(out.output["nodes"]["review"]["items"], json!([])); + + // error: the node fails rather than silently continuing. + let graph = wf(json!({ + "wait_mode": "poll", + "poll_interval_ms": 1, + "max_polls": 1, + "on_timeout": "reject", + "on_reject": "error", + })); + let compiled = compile(&graph).expect("compile"); + let err = run( + &compiled, + Value::Null, + &mock_capabilities_with_approvals(MockApprovals::pending()), + ) + .await + .expect_err("on_timeout: reject with on_reject: error must fail the node"); + assert!(err.to_string().contains("on_reject"), "got {err}"); +} + +/// A decision that carries the reviewer's own edit (`payload`) drives +/// `subject` to that edit and marks `edited: true` — the feature this node +/// exists for. `decision_from_resume` and `decided_item` are exercised +/// separately elsewhere; this checks the two compose correctly, matching what +/// a `{"decision": {..., "payload": ...}}` resume value produces end to end. +#[test] +fn a_decision_with_a_payload_edits_the_subject() { + let req = request("run-1:review"); + let resume_value = json!({ + "decision": { + "node_id": "review", + "approved": true, + "decided_by": "ada", + "payload": "https://example.com/edited", + } + }); + let decision = decision_from_resume(&resume_value, &req).expect("a decision"); + assert!(decision.approved); + assert_eq!(decision.payload, Some(json!("https://example.com/edited"))); + + let item = decided_item( + &req, + &decision, + json!({ "url": "https://example.com/original" }), + ); + let json = item.json; + assert_eq!(json["approved"], json!(true)); + assert_eq!(json["subject"], "https://example.com/edited"); + assert_eq!(json["edited"], json!(true)); + assert_eq!(json["decided_by"], "ada"); + assert_eq!( + json["input"], + json!({ "url": "https://example.com/original" }) + ); +} + +/// When a resume (or a listed approval) settles the review, any provider card +/// opened by an earlier `decide` call must be withdrawn — otherwise the +/// provider's queue keeps a stale entry for a review the run already closed. +#[tokio::test] +async fn a_resume_decision_withdraws_the_provider_card() { + let graph = wf(json!({ "title": "Publish this?" })); + let compiled = compile(&graph).expect("compile"); + let provider = std::sync::Arc::new(MockApprovals::pending()); + let caps = crate::caps::Capabilities { + approvals: Some(provider.clone()), + ..mock_capabilities() + }; + + // First activation asks the provider and gets Pending, opening a card. + let paused = run(&compiled, Value::Null, &caps).await.expect("run"); + assert_eq!(paused.pending_approvals, vec!["review".to_string()]); + assert!( + !provider.requested().is_empty(), + "the provider must have been asked at least once" + ); + assert!(provider.cancelled().is_empty(), "no reason to cancel yet"); + + // A resume delivers the decision directly (bypassing the provider), so the + // node must withdraw the provider's now-stale card rather than leave it. + let resumed = resume(&compiled, Value::Null, vec!["review".to_string()], &caps) + .await + .expect("resume"); + assert_eq!(resumed.output["nodes"]["review"]["port"], "approved"); + assert!( + !provider.cancelled().is_empty(), + "the provider's card must be withdrawn once a resume settles the review" + ); +} + +/// THE self-approval bypass: a caller must not be able to approve their own +/// review by putting the node's id in the trigger payload they submit. +/// +/// `engine::resume` writes the merged approvals into `run.trigger.approvals` +/// as well as the explicit channel, so it was tempting to read both. But the +/// trigger is whatever a caller handed to `engine::run` — on a webhook, that is +/// attacker-supplied — and a review its own subject can approve is not a +/// review. Only the explicit `RunInput::approvals` channel counts. +#[tokio::test] +async fn approvals_in_the_trigger_payload_cannot_self_approve_a_review() { + let graph = wf_raw(json!({ "title": "Publish this?", "request_id": "review-1" })); + let compiled = compile(&graph).expect("compile"); + let provider = std::sync::Arc::new(MockApprovals::pending()); + let caps = crate::caps::Capabilities { + approvals: Some(provider.clone()), + ..mock_capabilities() + }; + + // The attacker's payload names the review — by node id and by request id. + let outcome = run( + &compiled, + json!({ "approvals": ["review", "review-1"] }), + &caps, + ) + .await + .expect("run"); + + assert_eq!( + outcome.pending_approvals, + vec!["review".to_string()], + "a trigger-supplied approvals list must leave the review pending, not settle it" + ); + assert_ne!( + outcome.output["nodes"]["review"]["port"], + json!("approved"), + "the review must not have been approved by its own caller" + ); +} + +/// The counterpart: the explicit host channel still settles the review, so +/// closing the bypass did not break the supported path. +#[tokio::test] +async fn the_explicit_approvals_channel_still_settles_a_review() { + use crate::engine::RunInput; + + let graph = wf_raw(json!({ "title": "Publish this?", "request_id": "review-1" })); + let compiled = compile(&graph).expect("compile"); + let caps = crate::caps::Capabilities { + approvals: Some(std::sync::Arc::new(MockApprovals::pending())), + ..mock_capabilities() + }; + + let outcome = run( + &compiled, + RunInput::new(json!({})).with_approvals(vec!["review".to_string()]), + &caps, + ) + .await + .expect("run"); + + assert_eq!(outcome.output["nodes"]["review"]["port"], "approved"); + assert!(outcome.pending_approvals.is_empty()); +} + +/// The self-approval bypass must not survive a **resume** either. +/// +/// `merge_approvals` writes the merged list back into `run.trigger.approvals` +/// for the pre-existing `requires_approval` gate, and it used to seed its +/// starting set from that same key — which laundered a caller-written id into +/// the explicit channel on the first resume. So a payload the initial run +/// correctly refused was honoured on the next one. Provenance is kept separate +/// now: only ids a host actually authorised reach `run.approvals`. +#[tokio::test] +async fn a_trigger_supplied_approval_is_not_laundered_by_a_resume() { + use crate::engine::resume; + + let graph = wf_raw(json!({ "title": "Publish this?", "request_id": "review-1" })); + let compiled = compile(&graph).expect("compile"); + let caps = crate::caps::Capabilities { + approvals: Some(std::sync::Arc::new(MockApprovals::pending())), + ..mock_capabilities() + }; + // The attacker names the review in the payload they submit. + let attacker_trigger = json!({ "approvals": ["review", "review-1"] }); + + let paused = run(&compiled, attacker_trigger.clone(), &caps) + .await + .expect("run"); + assert_eq!(paused.pending_approvals, vec!["review".to_string()]); + + // A resume that approves NOTHING must not promote the attacker's ids. + let resumed = resume(&compiled, attacker_trigger, vec![], &caps) + .await + .expect("resume"); + assert_eq!( + resumed.pending_approvals, + vec!["review".to_string()], + "a resume must not launder a trigger-supplied id into the trusted channel" + ); + assert_ne!(resumed.output["nodes"]["review"]["port"], json!("approved")); +} diff --git a/src/nodes/integration/mod.rs b/src/nodes/integration/mod.rs index 04a0a432..a6ab24c4 100644 --- a/src/nodes/integration/mod.rs +++ b/src/nodes/integration/mod.rs @@ -6,6 +6,7 @@ pub mod agent; pub(crate) mod agent_request; +pub mod approval; pub mod code; pub(crate) mod envelope; pub mod gate; @@ -19,6 +20,7 @@ pub mod sub_workflow; pub mod tool_call; pub use agent::AgentNode; +pub use approval::ApprovalNode; pub use code::CodeNode; pub use gate::GateNode; pub use http_request::HttpRequestNode; diff --git a/src/testkit/mocks.rs b/src/testkit/mocks.rs index 85245956..30ffada4 100644 --- a/src/testkit/mocks.rs +++ b/src/testkit/mocks.rs @@ -41,18 +41,20 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; -use async_trait::async_trait; -use serde::{Deserialize, Serialize}; -use serde_json::{Value, json}; - -use crate::caps::{ - AgentRunner, Capabilities, CodeLanguage, CodeRunner, HttpClient, LlmProvider, MemoryProvider, - ShellOutcome, ShellRequest, ShellRunner, StateStore, ToolInvoker, WorkflowResolver, - sample_for_schema, -}; +use serde_json::Value; + +use crate::caps::{Capabilities, sample_for_schema}; use crate::error::{EngineError, Result}; use crate::model::WorkflowGraph; +#[path = "mocks_log.rs"] +mod log; +pub use log::{CallLog, CallOutcome, CapCall}; + +#[path = "mocks_double.rs"] +mod double; +use double::Double; + /// Which capability a call went to. /// /// A plain string rather than an enum on the wire, so a recording written by a @@ -74,113 +76,8 @@ pub mod capability { pub const MEMORY: &str = "memory"; /// [`StateStore`](crate::caps::StateStore). pub const STATE: &str = "state"; -} - -/// How one capability call ended. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum CallOutcome { - /// The call returned a value. - Ok(Value), - /// The call failed, with this message. - Err(String), -} - -/// One capability call a run made. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -#[serde(rename_all = "camelCase")] -pub struct CapCall { - /// Position in the run's single call sequence, from 0. - /// - /// One counter across *all* capabilities, so the log says what order things - /// happened in — which per-capability counters cannot. - pub seq: u64, - /// Which capability — see the [`capability`] constants. - pub capability: String, - /// The trait method (`invoke`, `complete`, `request`, …). - pub method: String, - /// The node that made the call. - /// - /// `None` only when the call was made outside a node activation, which no - /// engine path does today. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub node_id: Option, - /// What identifies the target within the capability: a tool slug, an agent - /// ref, an HTTP method and URL, a state key. Empty when the capability has - /// no such notion. - #[serde(default, skip_serializing_if = "String::is_empty")] - pub target: String, - /// The arguments the call was made with. - pub args: Value, - /// What it returned. - pub outcome: CallOutcome, -} - -/// Every capability call a run made, in order. -/// -/// Shared by every double in one [`MockCaps`], so the ordering across -/// capabilities is real rather than assembled afterwards from separate logs. -#[derive(Debug, Default)] -pub struct CallLog { - calls: Mutex>, - next_seq: AtomicU64, -} - -impl CallLog { - /// An empty log. - #[must_use] - pub fn new() -> Self { - Self::default() - } - - /// Append a call, assigning it the next sequence number. - fn record( - &self, - capability: &str, - method: &str, - node_id: Option, - target: String, - args: Value, - outcome: CallOutcome, - ) { - let call = CapCall { - seq: self.next_seq.fetch_add(1, Ordering::SeqCst), - capability: capability.to_string(), - method: method.to_string(), - node_id, - target, - args, - outcome, - }; - self.calls.lock().expect("call log poisoned").push(call); - } - - /// Every call recorded so far, in sequence order. - #[must_use] - pub fn calls(&self) -> Vec { - let mut calls = self.calls.lock().expect("call log poisoned").clone(); - calls.sort_by_key(|call| call.seq); - calls - } - - /// The calls matching a capability and an optional target glob. - /// - /// `capability` is one of the [`capability`] constants; `target` accepts the - /// same `*` globbing the rules do, and `None` matches every target. - #[must_use] - pub fn matching(&self, capability: &str, target: Option<&str>) -> Vec { - self.calls() - .into_iter() - .filter(|call| call.capability == capability) - .filter(|call| target.is_none_or(|glob| glob_matches(glob, &call.target))) - .collect() - } - - /// How many calls match — the count an assertion usually wants. - #[must_use] - pub fn count(&self, capability: &str, target: Option<&str>) -> usize { - self.matching(capability, target).len() - } + /// [`ApprovalProvider`](crate::caps::ApprovalProvider). + pub const APPROVALS: &str = "approvals"; } /// What a matched rule answers with. @@ -345,6 +242,15 @@ pub struct MockCaps { rules: Vec, log: Arc, workflows: HashMap, + /// Backing map for the [`StateStore`](crate::caps::StateStore) impl. + /// + /// Lives here, not on each [`Double`](double::Double), because + /// [`capabilities_for_node`](Self::capabilities_for_node) builds a fresh + /// `Double` per node activation (so the call log can attribute calls to + /// the right node); a per-`Double` map would make state invisible across + /// activations — including a node's own later activation — defeating the + /// one job a state store has. + state: Mutex>, } impl MockCaps { @@ -412,6 +318,19 @@ impl MockCaps { self.rule(capability::SHELL, "*", respond) } + /// Answer human review for the requests whose `request_id` matches + /// `request_id` (a glob). + /// + /// The answer is read loosely, the way [`on_shell`](Self::on_shell)'s is: + /// `{"approved": false, "comment": "…"}` is a rejection, `{"approved": + /// true}` an approval, and `{"status": "pending"}` a review nobody has got + /// to yet — which is how a test exercises a `poll`ing review or the + /// suspend/resume path. `Respond::error` fails the call instead. + #[must_use] + pub fn on_approval(self, request_id: &str, respond: Respond) -> Self { + self.rule(capability::APPROVALS, request_id, respond) + } + /// Restrict the most recently programmed rule to calls made by `node_id`. /// /// This is what per-node mocking looks like: stub one node's tool calls and @@ -469,6 +388,7 @@ impl MockCaps { shell: Some(Arc::new(Double::new(shared.clone(), None))), memory: Some(Arc::new(Double::new(shared.clone(), None))), tasks: Some(Arc::new(crate::caps::TokioTaskRunner::new())), + approvals: Some(Arc::new(Double::new(shared, None))), } } @@ -490,327 +410,13 @@ impl MockCaps { resolver: Arc::new(Double::new(shared.clone(), node.clone())), agent: Some(Arc::new(Double::new(shared.clone(), node.clone()))), shell: Some(Arc::new(Double::new(shared.clone(), node.clone()))), - memory: Some(Arc::new(Double::new(shared.clone(), node))), + memory: Some(Arc::new(Double::new(shared.clone(), node.clone()))), tasks: Some(Arc::new(crate::caps::TokioTaskRunner::new())), + approvals: Some(Arc::new(Double::new(shared, node))), } } } -/// One capability double: it consults the rules, records the call, and answers. -/// -/// A single type implementing every capability trait rather than nine, because -/// each implementation is the same three steps and nine copies of them would -/// drift. -struct Double { - mocks: Arc, - /// The node this double was scoped to, stamped onto every call it logs. - node_id: Option, - /// Backing map for the [`StateStore`] impl, which is the one capability - /// whose whole job is to remember. - state: Mutex>, -} - -impl Double { - fn new(mocks: Arc, node_id: Option) -> Self { - Self { - mocks, - node_id, - state: Mutex::new(HashMap::new()), - } - } - - /// Consult the rules, log whatever happens, and return it. - async fn dispatch( - &self, - capability: &str, - method: &str, - target: String, - request: Value, - default: impl FnOnce(&Value) -> Value, - ) -> Result { - let programmed = self - .mocks - .respond_to(capability, &target, self.node_id.as_deref(), &request) - .await; - let result = match programmed { - Some(result) => result, - None => Ok(default(&request)), - }; - let outcome = match &result { - Ok(value) => CallOutcome::Ok(value.clone()), - Err(err) => CallOutcome::Err(err.to_string()), - }; - self.mocks.log.record( - capability, - method, - self.node_id.clone(), - target, - request, - outcome, - ); - result - } -} - -#[async_trait] -impl LlmProvider for Double { - async fn complete(&self, request: Value, conn: Option<&str>) -> Result { - let conn = conn.map(str::to_string); - self.dispatch( - capability::LLM, - "complete", - String::new(), - request, - |req| json!({ "completion": req, "connection": conn }), - ) - .await - } -} - -#[async_trait] -impl ToolInvoker for Double { - async fn invoke(&self, slug: &str, args: Value, conn: Option<&str>) -> Result { - let slug_owned = slug.to_string(); - let conn = conn.map(str::to_string); - self.dispatch( - capability::TOOLS, - "invoke", - slug.to_string(), - args, - move |args| json!({ "tool": slug_owned, "args": args, "connection": conn }), - ) - .await - } -} - -#[async_trait] -impl HttpClient for Double { - async fn request(&self, request: Value, conn: Option<&str>) -> Result { - // The URL is what a rule globs on; a request without one still matches - // a bare `*`. - let url = request - .get("url") - .and_then(Value::as_str) - .unwrap_or_default() - .to_string(); - let conn = conn.map(str::to_string); - self.dispatch( - capability::HTTP, - "request", - url, - request, - |req| json!({ "status": 200, "request": req, "connection": conn }), - ) - .await - } -} - -#[async_trait] -impl CodeRunner for Double { - async fn run(&self, language: CodeLanguage, source: &str, input: Value) -> Result { - let request = json!({ - "language": format!("{language:?}"), - "source": source, - "input": input, - }); - self.dispatch( - capability::CODE, - "run", - format!("{language:?}"), - request, - |req| json!({ "result": req.get("input").cloned().unwrap_or(Value::Null) }), - ) - .await - } -} - -#[async_trait] -impl ShellRunner for Double { - async fn run(&self, request: ShellRequest) -> Result { - let script = match &request.script { - crate::caps::ShellScript::Inline(source) => source.clone(), - crate::caps::ShellScript::Path(path) => path.clone(), - }; - let encoded = json!({ - "interpreter": request.interpreter.as_str(), - "script": script, - "cwd": request.cwd, - "env": request.env, - "input": request.input, - }); - let value = self - .dispatch( - capability::SHELL, - "run", - script.clone(), - encoded, - move |_req| json!({ "exit_code": 0, "stdout": script, "stderr": "" }), - ) - .await?; - // A programmed value may describe the whole outcome, or just be the - // stdout a test cares about. Accept either rather than making a caller - // spell out an exit code they do not care about. - Ok(ShellOutcome { - exit_code: value - .get("exit_code") - .and_then(Value::as_i64) - .unwrap_or(0) - .try_into() - .unwrap_or(0), - stdout: value - .get("stdout") - .and_then(Value::as_str) - .map(str::to_string) - .unwrap_or_else(|| value.to_string()), - stderr: value - .get("stderr") - .and_then(Value::as_str) - .unwrap_or_default() - .to_string(), - }) - } -} - -#[async_trait] -impl AgentRunner for Double { - async fn run_agent( - &self, - agent_ref: &str, - request: Value, - conn: Option<&str>, - ) -> Result { - let name = agent_ref.to_string(); - let conn = conn.map(str::to_string); - self.dispatch( - capability::AGENT, - "run_agent", - agent_ref.to_string(), - request, - move |req| json!({ "agent": name, "request": req, "connection": conn }), - ) - .await - } -} - -#[async_trait] -impl MemoryProvider for Double { - async fn recall(&self, scope: &str, query: &str, opts: Value) -> Result { - let request = json!({ "scope": scope, "query": query, "opts": opts }); - self.dispatch( - capability::MEMORY, - "recall", - scope.to_string(), - request, - |_| json!({ "results": [] }), - ) - .await - } - - async fn flavour(&self, slug: &str) -> Result { - let request = json!({ "slug": slug }); - self.dispatch( - capability::MEMORY, - "flavour", - slug.to_string(), - request, - |_| json!({ "traits": {} }), - ) - .await - } - - async fn people(&self, query: Option<&str>) -> Result { - let request = json!({ "query": query }); - self.dispatch( - capability::MEMORY, - "people", - String::new(), - request, - |_| json!({ "people": [] }), - ) - .await - } - - async fn remember(&self, scope: &str, key: &str, value: Value) -> Result<()> { - let request = json!({ "scope": scope, "key": key, "value": value }); - self.dispatch( - capability::MEMORY, - "remember", - format!("{scope}/{key}"), - request, - |_| Value::Null, - ) - .await - .map(|_| ()) - } - - async fn forget(&self, scope: &str, key: &str) -> Result<()> { - let request = json!({ "scope": scope, "key": key }); - self.dispatch( - capability::MEMORY, - "forget", - format!("{scope}/{key}"), - request, - |_| Value::Null, - ) - .await - .map(|_| ()) - } -} - -#[async_trait] -impl StateStore for Double { - async fn load(&self, key: &str) -> Result> { - let stored = self - .state - .lock() - .expect("mock state poisoned") - .get(key) - .cloned(); - // Logged like any other call, but the *store* is the source of truth: - // a rule that overrode a load would make a stateful graph unreadable. - self.mocks.log.record( - capability::STATE, - "load", - self.node_id.clone(), - key.to_string(), - json!({ "key": key }), - CallOutcome::Ok(stored.clone().unwrap_or(Value::Null)), - ); - Ok(stored) - } - - async fn store(&self, key: &str, value: Value) -> Result<()> { - self.state - .lock() - .expect("mock state poisoned") - .insert(key.to_string(), value.clone()); - self.mocks.log.record( - capability::STATE, - "store", - self.node_id.clone(), - key.to_string(), - json!({ "key": key, "value": value }), - CallOutcome::Ok(Value::Null), - ); - Ok(()) - } -} - -#[async_trait] -impl WorkflowResolver for Double { - async fn resolve(&self, workflow_id: &str) -> Result { - self.mocks - .workflows - .get(workflow_id) - .cloned() - .ok_or_else(|| { - EngineError::Capability(format!( - "testkit: no workflow registered as {workflow_id:?}" - )) - }) - } -} - #[cfg(test)] #[path = "mocks_tests.rs"] mod tests; diff --git a/src/testkit/mocks_double.rs b/src/testkit/mocks_double.rs new file mode 100644 index 00000000..60150dfa --- /dev/null +++ b/src/testkit/mocks_double.rs @@ -0,0 +1,400 @@ +//! [`Double`], the single type that implements every capability trait for +//! [`MockCaps`](super::MockCaps) by consulting its rules, recording what +//! happened to [`CallLog`](super::log::CallLog), and answering. +//! +//! Split out of `mocks.rs` to keep that file under the repository's +//! line-length limit. + +use std::sync::Arc; + +use async_trait::async_trait; +use serde_json::{Value, json}; + +use crate::caps::{ + AgentRunner, ApprovalDecision, ApprovalOutcome, ApprovalProvider, ApprovalRequest, + CodeLanguage, CodeRunner, HttpClient, LlmProvider, MemoryProvider, ShellOutcome, ShellRequest, + ShellRunner, StateStore, ToolInvoker, WorkflowResolver, +}; +use crate::error::{EngineError, Result}; +use crate::model::WorkflowGraph; + +use super::log::CallOutcome; +use super::{MockCaps, capability}; + +/// One capability double: it consults the rules, records the call, and answers. +/// +/// A single type implementing every capability trait rather than nine, because +/// each implementation is the same three steps and nine copies of them would +/// drift. +pub(super) struct Double { + mocks: Arc, + /// The node this double was scoped to, stamped onto every call it logs. + node_id: Option, +} + +impl Double { + pub(super) fn new(mocks: Arc, node_id: Option) -> Self { + Self { mocks, node_id } + } + + /// Consult the rules, log whatever happens, and return it. + async fn dispatch( + &self, + capability: &str, + method: &str, + target: String, + request: Value, + default: impl FnOnce(&Value) -> Value, + ) -> Result { + let programmed = self + .mocks + .respond_to(capability, &target, self.node_id.as_deref(), &request) + .await; + let result = match programmed { + Some(result) => result, + None => Ok(default(&request)), + }; + let outcome = match &result { + Ok(value) => CallOutcome::Ok(value.clone()), + Err(err) => CallOutcome::Err(err.to_string()), + }; + self.mocks.log().record( + capability, + method, + self.node_id.clone(), + target, + request, + outcome, + ); + result + } +} + +#[async_trait] +impl LlmProvider for Double { + async fn complete(&self, request: Value, conn: Option<&str>) -> Result { + let conn = conn.map(str::to_string); + self.dispatch( + capability::LLM, + "complete", + String::new(), + request, + |req| json!({ "completion": req, "connection": conn }), + ) + .await + } +} + +#[async_trait] +impl ToolInvoker for Double { + async fn invoke(&self, slug: &str, args: Value, conn: Option<&str>) -> Result { + let slug_owned = slug.to_string(); + let conn = conn.map(str::to_string); + self.dispatch( + capability::TOOLS, + "invoke", + slug.to_string(), + args, + move |args| json!({ "tool": slug_owned, "args": args, "connection": conn }), + ) + .await + } +} + +#[async_trait] +impl HttpClient for Double { + async fn request(&self, request: Value, conn: Option<&str>) -> Result { + // The URL is what a rule globs on; a request without one still matches + // a bare `*`. + let url = request + .get("url") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(); + let conn = conn.map(str::to_string); + self.dispatch( + capability::HTTP, + "request", + url, + request, + |req| json!({ "status": 200, "request": req, "connection": conn }), + ) + .await + } +} + +#[async_trait] +impl CodeRunner for Double { + async fn run(&self, language: CodeLanguage, source: &str, input: Value) -> Result { + let request = json!({ + "language": format!("{language:?}"), + "source": source, + "input": input, + }); + self.dispatch( + capability::CODE, + "run", + format!("{language:?}"), + request, + |req| json!({ "result": req.get("input").cloned().unwrap_or(Value::Null) }), + ) + .await + } +} + +#[async_trait] +impl ShellRunner for Double { + async fn run(&self, request: ShellRequest) -> Result { + let script = match &request.script { + crate::caps::ShellScript::Inline(source) => source.clone(), + crate::caps::ShellScript::Path(path) => path.clone(), + }; + let encoded = json!({ + "interpreter": request.interpreter.as_str(), + "script": script, + "cwd": request.cwd, + "env": request.env, + "input": request.input, + }); + let value = self + .dispatch( + capability::SHELL, + "run", + script.clone(), + encoded, + move |_req| json!({ "exit_code": 0, "stdout": script, "stderr": "" }), + ) + .await?; + // A programmed value may describe the whole outcome, or just be the + // stdout a test cares about. Accept either rather than making a caller + // spell out an exit code they do not care about. + Ok(ShellOutcome { + exit_code: value + .get("exit_code") + .and_then(Value::as_i64) + .unwrap_or(0) + .try_into() + .unwrap_or(0), + stdout: value + .get("stdout") + .and_then(Value::as_str) + .map(str::to_string) + .unwrap_or_else(|| value.to_string()), + stderr: value + .get("stderr") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(), + }) + } +} + +#[async_trait] +impl AgentRunner for Double { + async fn run_agent( + &self, + agent_ref: &str, + request: Value, + conn: Option<&str>, + ) -> Result { + let name = agent_ref.to_string(); + let conn = conn.map(str::to_string); + self.dispatch( + capability::AGENT, + "run_agent", + agent_ref.to_string(), + request, + move |req| json!({ "agent": name, "request": req, "connection": conn }), + ) + .await + } +} + +#[async_trait] +impl MemoryProvider for Double { + async fn recall(&self, scope: &str, query: &str, opts: Value) -> Result { + let request = json!({ "scope": scope, "query": query, "opts": opts }); + self.dispatch( + capability::MEMORY, + "recall", + scope.to_string(), + request, + |_| json!({ "results": [] }), + ) + .await + } + + async fn flavour(&self, slug: &str) -> Result { + let request = json!({ "slug": slug }); + self.dispatch( + capability::MEMORY, + "flavour", + slug.to_string(), + request, + |_| json!({ "traits": {} }), + ) + .await + } + + async fn people(&self, query: Option<&str>) -> Result { + let request = json!({ "query": query }); + self.dispatch( + capability::MEMORY, + "people", + String::new(), + request, + |_| json!({ "people": [] }), + ) + .await + } + + async fn remember(&self, scope: &str, key: &str, value: Value) -> Result<()> { + let request = json!({ "scope": scope, "key": key, "value": value }); + self.dispatch( + capability::MEMORY, + "remember", + format!("{scope}/{key}"), + request, + |_| Value::Null, + ) + .await + .map(|_| ()) + } + + async fn forget(&self, scope: &str, key: &str) -> Result<()> { + let request = json!({ "scope": scope, "key": key }); + self.dispatch( + capability::MEMORY, + "forget", + format!("{scope}/{key}"), + request, + |_| Value::Null, + ) + .await + .map(|_| ()) + } +} + +#[async_trait] +impl StateStore for Double { + async fn load(&self, key: &str) -> Result> { + let stored = self + .mocks + .state + .lock() + .expect("mock state poisoned") + .get(key) + .cloned(); + // Logged like any other call, but the *store* is the source of truth: + // a rule that overrode a load would make a stateful graph unreadable. + self.mocks.log().record( + capability::STATE, + "load", + self.node_id.clone(), + key.to_string(), + json!({ "key": key }), + CallOutcome::Ok(stored.clone().unwrap_or(Value::Null)), + ); + Ok(stored) + } + + async fn store(&self, key: &str, value: Value) -> Result<()> { + self.mocks + .state + .lock() + .expect("mock state poisoned") + .insert(key.to_string(), value.clone()); + self.mocks.log().record( + capability::STATE, + "store", + self.node_id.clone(), + key.to_string(), + json!({ "key": key, "value": value }), + CallOutcome::Ok(Value::Null), + ); + Ok(()) + } +} + +#[async_trait] +impl ApprovalProvider for Double { + async fn decide(&self, request: &ApprovalRequest) -> Result { + let encoded = json!({ + "request_id": request.request_id, + "node_id": request.node_id, + "run_id": request.run_id, + "title": request.title, + "prompt": request.prompt, + "subject": { + "kind": request.subject.kind, + "value": request.subject.value, + }, + "assignees": request.assignees, + "metadata": request.metadata, + }); + // An unprogrammed review approves, so a graph that contains one runs end + // to end without a test standing a reviewer up — the same bargain every + // other default here makes. A test that cares about the answer says so + // with `on_approval`. + let value = self + .dispatch( + capability::APPROVALS, + "decide", + request.request_id.clone(), + encoded, + |_| json!({ "approved": true, "decided_by": "testkit" }), + ) + .await?; + // A programmed answer may be the whole verdict or just the bit the test + // cares about, as with `ShellRunner` above — including the bare string + // `"pending"` `on_approval`'s own doc comment advertises as shorthand + // for "nobody has got to this review yet". + let is_pending = value.as_str() == Some("pending") + || value.get("status").and_then(Value::as_str) == Some("pending"); + if is_pending { + return Ok(ApprovalOutcome::Pending); + } + Ok(ApprovalOutcome::Decided(ApprovalDecision { + approved: value + .get("approved") + .and_then(Value::as_bool) + .unwrap_or(true), + decided_by: value + .get("decided_by") + .and_then(Value::as_str) + .map(str::to_string), + comment: value + .get("comment") + .and_then(Value::as_str) + .map(str::to_string), + payload: value.get("payload").cloned(), + })) + } + + async fn cancel(&self, request_id: &str, reason: &str) -> Result<()> { + self.dispatch( + capability::APPROVALS, + "cancel", + request_id.to_string(), + json!({ "request_id": request_id, "reason": reason }), + |_| json!({ "cancelled": true }), + ) + .await + .map(|_| ()) + } +} + +#[async_trait] +impl WorkflowResolver for Double { + async fn resolve(&self, workflow_id: &str) -> Result { + self.mocks + .workflows + .get(workflow_id) + .cloned() + .ok_or_else(|| { + EngineError::Capability(format!( + "testkit: no workflow registered as {workflow_id:?}" + )) + }) + } +} diff --git a/src/testkit/mocks_log.rs b/src/testkit/mocks_log.rs new file mode 100644 index 00000000..6cea3c76 --- /dev/null +++ b/src/testkit/mocks_log.rs @@ -0,0 +1,122 @@ +//! The call log [`MockCaps`](super::MockCaps) and its [`Double`](super::double) +//! write to: [`CallOutcome`], [`CapCall`], and [`CallLog`] itself. +//! +//! Split out of `mocks.rs` (and out of `mocks_double.rs`, which is itself a +//! split of `mocks.rs`) to keep every file under the repository's +//! line-length limit. + +use std::sync::Mutex; +use std::sync::atomic::{AtomicU64, Ordering}; + +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +use super::glob_matches; + +/// How one capability call ended. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum CallOutcome { + /// The call returned a value. + Ok(Value), + /// The call failed, with this message. + Err(String), +} + +/// One capability call a run made. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CapCall { + /// Position in the run's single call sequence, from 0. + /// + /// One counter across *all* capabilities, so the log says what order things + /// happened in — which per-capability counters cannot. + pub seq: u64, + /// Which capability — see the [`capability`](super::capability) constants. + pub capability: String, + /// The trait method (`invoke`, `complete`, `request`, …). + pub method: String, + /// The node that made the call. + /// + /// `None` only when the call was made outside a node activation, which no + /// engine path does today. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub node_id: Option, + /// What identifies the target within the capability: a tool slug, an agent + /// ref, an HTTP method and URL, a state key. Empty when the capability has + /// no such notion. + #[serde(default, skip_serializing_if = "String::is_empty")] + pub target: String, + /// The arguments the call was made with. + pub args: Value, + /// What it returned. + pub outcome: CallOutcome, +} + +/// Every capability call a run made, in order. +/// +/// Shared by every double in one [`MockCaps`](super::MockCaps), so the +/// ordering across capabilities is real rather than assembled afterwards from +/// separate logs. +#[derive(Debug, Default)] +pub struct CallLog { + calls: Mutex>, + next_seq: AtomicU64, +} + +impl CallLog { + /// An empty log. + #[must_use] + pub fn new() -> Self { + Self::default() + } + + /// Append a call, assigning it the next sequence number. + pub(super) fn record( + &self, + capability: &str, + method: &str, + node_id: Option, + target: String, + args: Value, + outcome: CallOutcome, + ) { + let call = CapCall { + seq: self.next_seq.fetch_add(1, Ordering::SeqCst), + capability: capability.to_string(), + method: method.to_string(), + node_id, + target, + args, + outcome, + }; + self.calls.lock().expect("call log poisoned").push(call); + } + + /// Every call recorded so far, in sequence order. + #[must_use] + pub fn calls(&self) -> Vec { + let mut calls = self.calls.lock().expect("call log poisoned").clone(); + calls.sort_by_key(|call| call.seq); + calls + } + + /// The calls matching a capability and an optional target glob. + /// + /// `capability` is one of the [`capability`](super::capability) constants; `target` accepts the + /// same `*` globbing the rules do, and `None` matches every target. + #[must_use] + pub fn matching(&self, capability: &str, target: Option<&str>) -> Vec { + self.calls() + .into_iter() + .filter(|call| call.capability == capability) + .filter(|call| target.is_none_or(|glob| glob_matches(glob, &call.target))) + .collect() + } + + /// How many calls match — the count an assertion usually wants. + #[must_use] + pub fn count(&self, capability: &str, target: Option<&str>) -> usize { + self.matching(capability, target).len() + } +} diff --git a/src/testkit/mocks_tests.rs b/src/testkit/mocks_tests.rs index bb5705ff..becc2dc5 100644 --- a/src/testkit/mocks_tests.rs +++ b/src/testkit/mocks_tests.rs @@ -5,6 +5,8 @@ //! graph to reach them would be testing the engine instead. use super::*; +use crate::caps::{ApprovalOutcome, ApprovalRequest, ApprovalSubject}; +use serde_json::json; fn mocks(build: impl FnOnce(MockCaps) -> MockCaps) -> Arc { Arc::new(build(MockCaps::new())) @@ -282,6 +284,50 @@ async fn the_state_store_really_stores() { assert_eq!(mocks.log().count(capability::STATE, None), 3); } +/// `capabilities_for_node` builds a fresh `Capabilities` bundle — and so a +/// fresh `Double` — on every node activation, so it can attribute calls to +/// the right node. State must not live on that per-activation `Double`, or +/// nothing written by one activation would be visible to the next: not a +/// later activation of the SAME node (a loop reading what an earlier +/// iteration stored), and not a different node reading what an upstream one +/// wrote. +#[tokio::test] +async fn state_persists_across_node_scoped_bundles() { + let mocks = mocks(|m| m); + + // Node "writer"'s first activation stores a value... + mocks + .capabilities_for_node("writer") + .state + .store("k", json!("from writer")) + .await + .expect("store"); + + // ...a later activation of the SAME node must still see it... + assert_eq!( + mocks + .capabilities_for_node("writer") + .state + .load("k") + .await + .expect("load"), + Some(json!("from writer")), + "a node's own later activation must see what an earlier one stored" + ); + + // ...and so must a DIFFERENT node's bundle, entirely. + assert_eq!( + mocks + .capabilities_for_node("reader") + .state + .load("k") + .await + .expect("load"), + Some(json!("from writer")), + "state is one store shared across the whole run, not one per node" + ); +} + #[tokio::test] async fn a_delayed_response_still_answers() { let mocks = mocks(|m| { @@ -312,3 +358,116 @@ async fn an_unregistered_sub_workflow_is_refused_by_name() { .expect_err("an unregistered id should not silently resolve"); assert!(err.to_string().contains("nope"), "got {err}"); } + +fn approval_request(request_id: &str) -> ApprovalRequest { + ApprovalRequest { + request_id: request_id.to_string(), + node_id: "review".to_string(), + run_id: Some("run-1".to_string()), + title: Some("Ship it?".to_string()), + prompt: None, + subject: ApprovalSubject { + kind: "url".to_string(), + value: json!("https://example.com/preview"), + }, + assignees: vec!["reviewer@example.com".to_string()], + metadata: json!({}), + } +} + +#[tokio::test] +async fn an_unprogrammed_review_approves_and_is_logged() { + // Same bargain as every other default here: a graph containing a review + // runs end to end without a test standing a reviewer up, and the call still + // shows in the log so a test can assert the review was *asked for*. + let mocks = mocks(|m| m); + let caps = mocks.capabilities(); + let approvals = caps.approvals.clone().expect("the doubles wire approvals"); + + let outcome = approvals + .decide(&approval_request("run-1:review")) + .await + .expect("an unprogrammed review still answers"); + + match outcome { + ApprovalOutcome::Decided(decision) => { + assert!(decision.approved); + assert_eq!(decision.decided_by.as_deref(), Some("testkit")); + } + ApprovalOutcome::Pending => panic!("the default should decide, not park the run"), + } + + let calls = mocks.log().calls(); + assert_eq!(calls.len(), 1); + assert_eq!(calls[0].capability, capability::APPROVALS); + assert_eq!(calls[0].target, "run-1:review"); + assert_eq!(calls[0].args["subject"]["kind"], json!("url")); +} + +#[tokio::test] +async fn a_programmed_review_can_reject_or_stay_pending() { + // Rules match on `request_id`, which is what lets one test settle one + // review and leave another waiting — the two halves of the wait paths a + // review node has to cope with. + let mocks = mocks(|m| { + m.on_approval( + "run-1:reject*", + Respond::value(json!({ "approved": false, "comment": "not yet" })), + ) + .on_approval( + "run-1:slow*", + Respond::value(json!({ "status": "pending" })), + ) + }); + let caps = mocks.capabilities(); + let approvals = caps.approvals.clone().expect("the doubles wire approvals"); + + let rejected = approvals + .decide(&approval_request("run-1:rejected-review")) + .await + .expect("call"); + match rejected { + ApprovalOutcome::Decided(decision) => { + assert!(!decision.approved); + assert_eq!(decision.comment.as_deref(), Some("not yet")); + } + ApprovalOutcome::Pending => panic!("a programmed verdict should decide"), + } + + let pending = approvals + .decide(&approval_request("run-1:slow-review")) + .await + .expect("call"); + assert_eq!( + pending, + ApprovalOutcome::Pending, + "`status: pending` is how a test exercises a poll or a suspend" + ); + + approvals + .cancel("run-1:slow-review", "run ended") + .await + .expect("cancelling a review is answered too"); + assert_eq!(mocks.log().count(capability::APPROVALS, None), 3); +} + +/// The bare string `"pending"` is documented shorthand for "nobody has got to +/// this review yet" — the same flexibility `on_shell` accepts a bare stdout +/// string for. A rule answering with it must not be read as an (approving) +/// verdict object with no recognized fields. +#[tokio::test] +async fn on_approval_accepts_a_bare_pending_string() { + let mocks = mocks(|m| m.on_approval("run-1:bare*", Respond::value(json!("pending")))); + let caps = mocks.capabilities(); + let approvals = caps.approvals.clone().expect("the doubles wire approvals"); + + let outcome = approvals + .decide(&approval_request("run-1:bare-review")) + .await + .expect("call"); + assert_eq!( + outcome, + ApprovalOutcome::Pending, + "a bare \"pending\" string must not be read as an approving verdict" + ); +} diff --git a/src/validate.rs b/src/validate.rs index 21b0925c..c816be39 100644 --- a/src/validate.rs +++ b/src/validate.rs @@ -414,6 +414,60 @@ pub fn validate_all(graph: &WorkflowGraph) -> Vec { } } + // `approval` node config. These are all closed enums that SELECT BEHAVIOUR, + // so a typo cannot be caught at run time without silently changing what the + // node does: a misspelled `on_reject` would quietly route a rejection that + // was meant to fail the run, and a misspelled `wait_mode` would quietly + // suspend a review the author wanted polled. Refuse them at the door, where + // the message can name the node and the alternatives. + for node in &graph.nodes { + if node.kind != NodeKind::Approval { + continue; + } + + for (key, allowed) in [ + ("wait_mode", &["suspend", "poll"][..]), + ("on_reject", &["route", "error", "drop"][..]), + ("on_timeout", &["error", "reject", "route"][..]), + ] { + let Some(value) = node.config.get(key) else { + continue; + }; + if !value.as_str().is_some_and(|v| allowed.contains(&v)) { + errors.push(ValidationError::InvalidNodeConfig { + node: node.id.clone(), + reason: format!( + "approval node has unknown `{key}` {value} (expected one of {})", + allowed + .iter() + .map(|v| format!("{v:?}")) + .collect::>() + .join(", ") + ), + }); + } + } + + // Reviewer handles are opaque to the crate, but their *shape* is not: + // a bare string here (the natural mistake for a single reviewer) would + // be read as "nobody", and the review would go to an empty audience + // with no error anywhere. An empty array reaches the same audience of + // nobody just as silently, so it is refused for the same reason. + if let Some(assignees) = node.config.get("assignees") { + if !assignees + .as_array() + .is_some_and(|values| !values.is_empty() && values.iter().all(Value::is_string)) + { + errors.push(ValidationError::InvalidNodeConfig { + node: node.id.clone(), + reason: "approval node `assignees` must be a non-empty array of strings (a \ + single reviewer is a one-element array)" + .to_string(), + }); + } + } + } + // `void` node topology checks. The kind asserts exactly one thing — "the // branch ends here, deliberately" — so the two ways to contradict it are // refused rather than absorbed. An outgoing edge would be dead (a leaf diff --git a/src/validate_tests/validate_tests_part_03_tests.rs b/src/validate_tests/validate_tests_part_03_tests.rs index 6ca294f7..a8cffe7c 100644 --- a/src/validate_tests/validate_tests_part_03_tests.rs +++ b/src/validate_tests/validate_tests_part_03_tests.rs @@ -148,3 +148,84 @@ fn validation_error_code_and_node_id_accessors() { None ); } + +/// Builds a trigger -> approval graph carrying `config` on the approval node. +fn approval_graph(config: serde_json::Value) -> WorkflowGraph { + let mut review = node("review", NodeKind::Approval); + review.config = config; + WorkflowGraph { + nodes: vec![node("t", NodeKind::Trigger), review], + ..Default::default() + } +} + +/// Every `approval` behaviour selector is a closed enum, and a typo in one +/// silently changes what the node does rather than failing — so it is refused +/// here instead. +#[test] +fn approval_behaviour_selectors_must_be_known_values() { + for (key, bad) in [ + ("wait_mode", "suspended"), + ("on_reject", "reroute"), + ("on_timeout", "partial"), + ] { + let graph = approval_graph(serde_json::json!({ (key): bad })); + let errors = validate_all(&graph); + assert!( + errors.iter().any(|e| matches!( + e, + ValidationError::InvalidNodeConfig { node, reason } + if node == "review" && reason.contains(key) + )), + "expected {key}={bad} to be refused, got {errors:?}" + ); + } + + assert!( + validate_all(&approval_graph(serde_json::json!({ + "wait_mode": "poll", + "on_reject": "error", + "on_timeout": "route", + }))) + .is_empty(), + "the documented values must all pass" + ); +} + +/// A single reviewer written as a bare string would be read as an empty +/// audience, so the shape is checked even though the handles are opaque. +#[test] +fn approval_assignees_must_be_an_array_of_strings() { + let errors = validate_all(&approval_graph( + serde_json::json!({ "assignees": "ada" }), + )); + assert!( + errors.iter().any(|e| matches!( + e, + ValidationError::InvalidNodeConfig { node, reason } + if node == "review" && reason.contains("assignees") + )), + "got {errors:?}" + ); + assert!( + validate_all(&approval_graph( + serde_json::json!({ "assignees": ["ada"] }) + )) + .is_empty() + ); +} + +/// An empty array reaches the same silent "nobody reviews this" audience a +/// bare string does, so it is refused for the same reason. +#[test] +fn approval_assignees_must_not_be_an_empty_array() { + let errors = validate_all(&approval_graph(serde_json::json!({ "assignees": [] }))); + assert!( + errors.iter().any(|e| matches!( + e, + ValidationError::InvalidNodeConfig { node, reason } + if node == "review" && reason.contains("assignees") + )), + "an empty `assignees` array must be refused, got {errors:?}" + ); +} diff --git a/src/visualization.rs b/src/visualization.rs index f7472bfc..e0fa3b56 100644 --- a/src/visualization.rs +++ b/src/visualization.rs @@ -335,6 +335,7 @@ fn kind_name(kind: &NodeKind) -> &'static str { NodeKind::Scatter => "scatter", NodeKind::Gather => "gather", NodeKind::Gate => "gate", + NodeKind::Approval => "approval", NodeKind::Void => "void", } } @@ -347,6 +348,9 @@ fn node_fill(kind: Option<&NodeKind>) -> Rgb { Some(NodeKind::Agent | NodeKind::ToolCall | NodeKind::HttpRequest | NodeKind::Memory) => { Rgb([220, 252, 231]) } + // A human review reads as a decision point, not as work the machine + // does, so it takes the branching palette rather than the capability one. + Some(NodeKind::Approval) => Rgb([254, 226, 226]), _ => Rgb([241, 245, 249]), } } @@ -359,6 +363,7 @@ fn node_accent(kind: Option<&NodeKind>) -> Rgb { Some(NodeKind::Agent | NodeKind::ToolCall | NodeKind::HttpRequest | NodeKind::Memory) => { Rgb([22, 163, 74]) } + Some(NodeKind::Approval) => Rgb([220, 38, 38]), None => Rgb([220, 38, 38]), _ => Rgb([71, 85, 105]), } diff --git a/tests/fuzz_interception.proptest-regressions b/tests/fuzz_interception.proptest-regressions new file mode 100644 index 00000000..c9ae9417 --- /dev/null +++ b/tests/fuzz_interception.proptest-regressions @@ -0,0 +1,11 @@ +# Seeds for failure cases proptest has generated in the past. It is +# automatically read and these particular cases re-run before any +# novel cases are generated. +# +# It is recommended to check this file in to source control so that +# everyone who runs the test benefits from these saved cases. +cc 5a8255f59206209c6bbdf34e78e14ead0bae17e4af6fdb64842502686224265d # shrinks to shape = Branch(Branch(Spawned { tasks: 3, release: "all", n: 3 }, Spawned { tasks: 1, release: "all", n: 1 }), Spawned { tasks: 2, release: "all", n: 3 }) +cc 201cea089acc9cbb4ed3a0bf98e20f57b1e671b9fb6d8636084a4c5a06abd47c # shrinks to shape = Loop { max_iter: 3, body: Spawned { tasks: 3, release: "all", n: 3 } } +cc 60dcc55b51c47d53d5c569e160e47326be3516ad1444256c44ef6a21f1a3a708 # shrinks to shape = Nested(Branch(Spawned { tasks: 1, release: "all", n: 2 }, Loop { max_iter: 3, body: Spawned { tasks: 3, release: "all", n: 1 } })) +cc c9826493d112ad19cdf5822aac757e9b93306189414049540997db7018e2e39c # shrinks to shape = Loop { max_iter: 3, body: Spawned { tasks: 1, release: "all", n: 1 } } +cc 10f774fdc90ae37d8987e735b588e777bd8c9ae98bb4f5995d03ab995647ca66 # shrinks to shape = Fanout([Linear(2), Loop { max_iter: 2, body: Linear(1) }, Fanout([Linear(1), Linear(1), Spawned { tasks: 2, release: "all", n: 3 }])]) diff --git a/tests/fuzz_resume.rs b/tests/fuzz_resume.rs index 32339ba8..d71a47b0 100644 --- a/tests/fuzz_resume.rs +++ b/tests/fuzz_resume.rs @@ -199,6 +199,64 @@ fn gated_graphs_are_mostly_runnable_and_actually_suspend() { ); } +/// The shape `resuming_is_deterministic` shrank to when it caught a real race. +/// +/// A gate sitting beside concurrent work, with three spawned tasks either side +/// of a branch — the arrangement in which the number of polls a gate needs was +/// decided by OS scheduling rather than by the graph. +fn regression_shape() -> Shape { + Shape::Gate(Box::new(Shape::Branch( + Box::new(Shape::Spawned { + tasks: 3, + release: "all", + n: 3, + }), + Box::new(Shape::Fanout(vec![ + Shape::Linear(2), + Shape::Linear(0), + Shape::Spawned { + tasks: 3, + release: "all", + n: 3, + }, + ])), + ))) +} + +/// **Regression.** A gate needs the same number of polls every time. +/// +/// `resuming_is_deterministic` found this shape producing runs that agreed on +/// every collected item yet disagreed on the gate's `polls` count and on every +/// downstream `_activation_step`. The cause was not the gate: a `Reenter` +/// backoff whose timer had already fired completed without ever returning +/// `Pending`, so on a single-threaded runtime the spawned tasks the gate was +/// waiting on got no turn to run, and the gate spent an extra poll observing a +/// world that had not moved. Whether that happened depended on whether the +/// thread was descheduled for longer than the poll interval — which is why the +/// property test only ever failed on a loaded CI machine. +/// +/// Run as a fixed loop rather than a property, because the shape is already +/// known and the variable being exercised is repetition, not generation. It +/// reproduced within a few hundred iterations under load before the fix. +#[test] +fn a_gate_takes_the_same_number_of_polls_every_run() { + const RUNS: usize = 150; + + let shape = regression_shape(); + let mut baseline: Option = None; + for run_index in 0..RUNS { + let (_, resumed) = run_both_ways(&shape).expect("the regression shape should compile"); + match &baseline { + None => baseline = Some(resumed), + Some(first) => assert_eq!( + first, &resumed, + "run {run_index} of the same shape differed from the first; a gate's poll count \ + or an activation step is being decided by scheduling rather than by the graph" + ), + } + } +} + proptest! { // Deliberately fewer cases than the other fuzz files: each case runs the // same graph at least twice, once through the checkpointer. diff --git a/tests/smoke_all_nodes.rs b/tests/smoke_all_nodes.rs index ce85bc9b..c7ae41be 100644 --- a/tests/smoke_all_nodes.rs +++ b/tests/smoke_all_nodes.rs @@ -376,6 +376,27 @@ async fn smoke_scatter_gather() { assert_eq!(items.len(), 2, "two lanes, two collected results"); } +/// The default mock approvals provider approves, so an `approval` node +/// dry-runs end to end the way a `memory` node does — the point being that a +/// graph containing a review is runnable without the host standing up a review +/// surface first. +#[tokio::test] +async fn smoke_approval() { + smoke_single_node( + NodeKind::Approval, + json!({ + "title": "Ship it?", + "subject_kind": "url", + "subject": "=item.url", + // `request_id` (or a run-scoped id) is required: without one, the + // node refuses to guess an identity a later run could collide on. + "request_id": "smoke-approval", + }), + json!({ "url": "https://example.com/preview" }), + ) + .await; +} + #[tokio::test] async fn smoke_void() { // `smoke_single_node` asserts a non-empty `items` slot, which a void can diff --git a/wiki/Capability-Traits.md b/wiki/Capability-Traits.md index e284fd6c..bae43f3f 100644 --- a/wiki/Capability-Traits.md +++ b/wiki/Capability-Traits.md @@ -21,6 +21,7 @@ examples without any real backend. | `CodeRunner` | `code` | Executes sandboxed user code (`CodeLanguage::JavaScript` / `Python`) with a JSON input. | | `ShellRunner` | `shell` | Runs a shell script (inline or by path) with a working directory and environment, returning its exit code, stdout, and stderr. Optional: `None` refuses `shell` nodes. | | `StateStore` | resumable / stateful workflows | Durable key/value state (`load` / `store`) for a run. | +| `ApprovalProvider` | `approval` | Puts a subject (a URL, a draft, a payload) in front of a human and reports their approve/reject. Optional: with `None`, an `approval` node pauses the run instead and the host settles it via `engine::resume`. | ## Connection references @@ -42,10 +43,21 @@ parameters (e.g. `args: { "text": "=item.name" }`). Values that do not start wit ## The `Capabilities` bundle The engine receives a `Capabilities` struct — the per-run bundle of host -implementations. It bundles all five host capabilities: `llm`, `tools`, `http`, -`code`, and `state` (each an `Arc`). Nodes reach each one through -`ctx.caps` during execution — for example, durable key/value state via -`ctx.caps.state`. +implementations. It bundles six host capabilities: `llm`, `tools`, `http`, +`code`, `state`, and the optional `approvals` (`Option>`) each an `Arc` (or `Option` of one, for +`approvals`). Nodes reach each one through `ctx.caps` during execution — for +example, durable key/value state via `ctx.caps.state`, and a human review via +`ctx.caps.approvals`. A host that builds `Capabilities` with a struct literal +must add an `approvals` field (`None` if it wires no provider) to keep +compiling. + +`ApprovalProvider::decide` is **create-or-fetch**, keyed on +`ApprovalRequest::request_id`: the first call with an id creates the review, and +every later call with that id reports where *that* review stands. This matters +because an interrupt discards the activation's state update, so the node re-asks +after every resume and on every poll — a provider that created a fresh review per +call would notify the reviewer once per call. Durable, cross-process human-in-the-loop resume is available by implementing `Checkpointer` and driving the run via diff --git a/wiki/Node-Catalog.md b/wiki/Node-Catalog.md index 09a51509..a2007516 100644 --- a/wiki/Node-Catalog.md +++ b/wiki/Node-Catalog.md @@ -76,6 +76,7 @@ traits](Capability-Traits). | `code` | Runs sandboxed user code | Config `language` (`javascript`/`python`), `source` — via `CodeRunner` | | `shell` | Runs a shell script, inline or from a file | Config `source` **or** `script_path`, plus `interpreter` (`sh`/`bash`), `cwd`, `env` — via `ShellRunner` | | `output_parser` | Parses/validates an agent's output into a structured shape | May use `LlmProvider` for auto-fixing; can nest as a sub-agent | +| `approval` | Puts a subject in front of a **human** and routes on approve/reject | Out `approved` / `rejected` / `timeout`; config `subject`, `subject_kind`, `title`, `prompt`, `assignees`, `wait_mode`, `on_reject`, `on_timeout` (`error` default / `reject` / `route` — `route` is required to reach the `timeout` port) — via `ApprovalProvider` | | `sub_workflow` | Runs another workflow as a nested sub-graph | Config: exactly one of `workflow` (inline) / `workflow_id`; optional `inputs` map for the child's declared inputs | The capability-backed integration nodes (`agent`, `tool_call`, `http_request`) @@ -89,6 +90,15 @@ Per-node error handling (`on_error` stop/continue/route, `retry`, an `error` port) and approval gating (`requires_approval`) are configured through the same free-form `config`. +`approval` is not the same thing as the `requires_approval` flag. The flag holds +a node back until someone says go; it carries nothing and its answer is invisible +to the graph. The `approval` **kind** is the review itself — it carries what is +being reviewed, and the verdict (approved, who decided, their comment, any edit +they made) comes back as an item on the `approved` / `rejected` ports, readable +anywhere as `=nodes..decision.approved`. It waits by suspending the run by +default (`wait_mode: "suspend"`), which costs nothing while a card sits in +somebody's queue. + ### Per-item fan-out `agent`, `tool_call`, `http_request`, `memory`, and `sub_workflow` can map over