feat(nodes): approval node kind + ApprovalProvider capability

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.<id>.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<Arc<dyn ApprovalProvider>>` 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.
 

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

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<HashMap<String, Option<ApprovalDecision>>>,
+    }
+
+    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<String> {
+            let mut ids: Vec<String> = 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<ApprovalOutcome> {
+            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"
+    );
+}