feat(engine): let a host name a run, and key approval reviews on it

CHANGELOG.md

@@ -9,6 +9,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **`RunInput::with_run_id`** — a host names a run with a durable,
+  server-generated id, seeded into the run state as `run.id`. The engine's own
+  run id is process-local and minted fresh on every call (a resume re-executes,
+  so it changes between a run and its own resume), which makes it unusable for
+  anything that must name *this* run across a pause. The `approval` node's
+  `request_id` defaults to `"<run id>:<node id>"` and so is now unique per run
+  **and** stable across resume — the property that makes one human review
+  rather than a fresh card per resume. Seeded outside `run.trigger` on purpose:
+  the trigger is caller-supplied and, on a webhook, attacker-influenced, and the
+  review-de-duplication key must not be.
+
 - **`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,

examples/hitl_review.rs

@@ -24,7 +24,7 @@ async fn main() {
         ApprovalDecision, ApprovalOutcome, ApprovalProvider, ApprovalRequest, Capabilities,
     };
     use tinyflows::compiler::compile;
-    use tinyflows::engine::{resume, run};
+    use tinyflows::engine::{RunInput, resume, run};
     use tinyflows::model::{Edge, Node, NodeKind, WorkflowGraph};
 
     /// A host's review desk: one row per `request_id`, holding the verdict once
@@ -108,7 +108,6 @@ async fn main() {
                     // 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",
@@ -141,11 +140,18 @@ async fn main() {
         approvals: Some(desk.clone()),
         ..mock_capabilities()
     };
+    // The host names the run. This is what gives the review a stable identity
+    // (`request_id` defaults to "<run id>:<node id>"), so the resume below
+    // resolves the card already in front of a person instead of opening a
+    // second one. It is seeded outside the trigger payload on purpose: a
+    // caller-supplied value here would hand an attacker the de-duplication key.
+    let run_id = "run-7f3a";
     let trigger = json!({ "url": "https://example.com/drafts/42" });
+    let input = || RunInput::new(trigger.clone()).with_run_id(run_id);
 
     // 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");
+    let paused = run(&compiled, input(), &caps).await.expect("run");
     println!("--- before the human answers ---");
     println!("pending_approvals: {:?}", paused.pending_approvals);
     println!("desk queue:        {:?}", desk.queue());
@@ -165,7 +171,7 @@ async fn main() {
 
     // 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)
+    let done = resume(&compiled, input(), vec![], &caps)
         .await
         .expect("resume");
     println!("--- after the human answers ---");

src/catalog/contracts/group_03.rs

@@ -166,7 +166,9 @@ pub(super) fn contract_approval() -> NodeKindContract {
                  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 \
+                 A host names the run with `RunInput::with_run_id`, which lands in `run.id`; \
+                 with that seeded this field is optional. Required only 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 \

src/engine.rs

@@ -146,6 +146,31 @@ pub struct RunInput {
     /// array) has nowhere to put them, so smuggling approvals through the
     /// payload cannot work in general.
     pub approvals: Vec<String>,
+    /// A durable, **host-generated** identity for this run, seeded into the run
+    /// state as `run.id`.
+    ///
+    /// The engine keeps its own process-local run id for observability, but
+    /// that one is minted fresh inside every `run` call — and a resume
+    /// re-executes the workflow, so it changes between a run and its own
+    /// resume. Anything that must name *this* run across a pause therefore
+    /// cannot use it, and only the host (which owns run persistence) knows an
+    /// id that survives.
+    ///
+    /// Today's consumer is the [`approval`](crate::nodes::integration::approval)
+    /// node, whose `request_id` defaults to `"<run id>:<node id>"` — the key the
+    /// host's [`ApprovalProvider`](crate::caps::ApprovalProvider) de-duplicates
+    /// reviews on. Unique per run *and* stable across resume is exactly the
+    /// property that makes one human review, rather than a fresh card every
+    /// time the run is looked at.
+    ///
+    /// **Must be server-generated.** It lands in `run.id`, outside the
+    /// caller-supplied trigger payload, precisely so it is not attacker
+    /// influenced; copying a request field into it hands an attacker the
+    /// de-duplication key and, with it, an earlier run's approval.
+    ///
+    /// `None` leaves `run.id` unset, which is what every caller predating this
+    /// meant.
+    pub run_id: Option<String>,
 }
 
 impl RunInput {
@@ -156,6 +181,7 @@ impl RunInput {
             trigger,
             inputs: Map::new(),
             approvals: Vec::new(),
+            run_id: None,
         }
     }
 
@@ -172,6 +198,17 @@ impl RunInput {
         self.approvals = approvals;
         self
     }
+
+    /// Names this run with a durable, host-generated id (see [`Self::run_id`]).
+    ///
+    /// Pass the **same** id when resuming the run: that is what makes a paused
+    /// human review resolve to the one already in front of a person instead of
+    /// opening a second one.
+    #[must_use]
+    pub fn with_run_id(mut self, run_id: impl Into<String>) -> Self {
+        self.run_id = Some(run_id.into());
+        self
+    }
 }
 
 impl From<Value> for RunInput {

src/engine/run_state.rs

@@ -89,6 +89,7 @@ pub(super) async fn build_and_run(
         trigger,
         inputs,
         approvals,
+        run_id: host_run_id,
     } = input.into();
     let resolved_inputs = crate::model::resolve_inputs(&workflow.graph.inputs, &inputs)?;
 
@@ -125,6 +126,15 @@ pub(super) async fn build_and_run(
     // `=inputs.<name>` (and jq programs walking `run` still see it too).
     let mut initial =
         json!({ "run": { "trigger": trigger, "inputs": resolved_inputs, "approvals": approvals } });
+    // The host's durable run identity, seeded OUTSIDE `run.trigger` on purpose:
+    // the trigger is caller-supplied and, on a webhook, attacker-influenced,
+    // while this slot is one a host fills deliberately. Anything keyed on it
+    // (today the `approval` node's `request_id`) therefore rests on an id the
+    // host chose, not one a request could smuggle in. Absent when the caller
+    // named no run, which is every caller predating `RunInput::run_id`.
+    if let Some(host_run_id) = host_run_id {
+        merge(&mut initial, json!({ "run": { "id": host_run_id } }));
+    }
     merge(&mut initial, seed_items);
     // The nesting cap for `sub_workflow` chains, read off the trigger config
     // like every other run-level knob and seeded into the run state so the
@@ -279,7 +289,7 @@ pub async fn resume(
 }
 
 /// Unions `newly_approved` into the run input's `trigger["approvals"]`, leaving
-/// the declared-input values untouched.
+/// the declared-input values and the run's host id untouched.
 ///
 /// Reads defensively: a missing or non-array `approvals` yields an empty
 /// starting set, and a non-object trigger (which carries no fields to preserve)
@@ -295,6 +305,7 @@ pub(super) fn merge_approvals(input: impl Into<RunInput>, newly_approved: Vec<St
         mut trigger,
         inputs,
         approvals: prior,
+        run_id,
     } = input.into();
 
     // Approval **provenance** is preserved rather than flattened, and the two
@@ -350,5 +361,10 @@ pub(super) fn merge_approvals(input: impl Into<RunInput>, newly_approved: Vec<St
         trigger,
         inputs,
         approvals: explicit,
+        // Carried through untouched. A resume is the SAME run continuing, so
+        // dropping its id here would re-key everything derived from it — a
+        // paused human review would open a second card instead of resolving
+        // the one already in front of somebody.
+        run_id,
     }
 }

src/nodes/integration/approval_tests/approval_tests_part_02_tests.rs

@@ -250,6 +250,80 @@ async fn a_resume_decision_withdraws_the_provider_card() {
     );
 }
 
+/// The host-seeded run id is what makes the default `request_id` work: unique
+/// per run, and — crucially — the *same* across a resume, so a paused review
+/// resolves the card already in front of a person.
+#[tokio::test]
+async fn a_host_seeded_run_id_derives_the_request_id_and_survives_a_resume() {
+    use crate::caps::mock::MockApprovals;
+    use crate::engine::{RunInput, resume};
+
+    let graph = wf_raw(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 input = || RunInput::new(json!({ "url": "https://example.com" })).with_run_id("run-7f3a");
+
+    // Nobody has answered: the run pauses at the review.
+    let paused = run(&compiled, input(), &caps).await.expect("run");
+    assert_eq!(paused.pending_approvals, vec!["review".to_string()]);
+
+    // Resuming the SAME run must address the SAME review, not open a second.
+    let resumed = resume(&compiled, input(), vec!["review".to_string()], &caps)
+        .await
+        .expect("resume");
+    assert_eq!(
+        resumed.output["nodes"]["review"]["items"][0]["json"]["request_id"],
+        "run-7f3a:review"
+    );
+    let ids = provider.requested();
+    assert!(
+        ids.iter().all(|id| id == "run-7f3a:review"),
+        "every ask must name one review across the run and its resume, got {ids:?}"
+    );
+}
+
+/// Two runs of the same graph must not share a review — that collision is
+/// precisely what would let a later run inherit an earlier approval.
+#[tokio::test]
+async fn two_runs_of_one_graph_get_distinct_reviews() {
+    use crate::engine::RunInput;
+
+    let graph = wf_raw(json!({ "title": "Publish this?" }));
+    let compiled = compile(&graph).expect("compile");
+    // `mock_capabilities()` wires `MockApprovals::approving()` by default (see
+    // its doc comment), so the review settles inline rather than pausing —
+    // `items[0]` below is the approved decision, not a paused/null node.
+    let caps = mock_capabilities();
+
+    let first = run(
+        &compiled,
+        RunInput::new(json!({})).with_run_id("run-a"),
+        &caps,
+    )
+    .await
+    .expect("first run");
+    let second = run(
+        &compiled,
+        RunInput::new(json!({})).with_run_id("run-b"),
+        &caps,
+    )
+    .await
+    .expect("second run");
+
+    assert_eq!(
+        first.output["nodes"]["review"]["items"][0]["json"]["request_id"],
+        "run-a:review"
+    );
+    assert_eq!(
+        second.output["nodes"]["review"]["items"][0]["json"]["request_id"],
+        "run-b: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.
 ///