From f57a60f287cd6f1c1cf4b30c51e3ec92ac3ee6f8 Mon Sep 17 00:00:00 2001 From: cyrus Date: Fri, 31 Jul 2026 00:26:46 +0530 Subject: [PATCH] fix(flows): pin a parked run to the graph it was approved against (T-M1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A run parks `pending_approval` on, say, a `send` node, and the user sees an approval card describing that node as it existed at park time. `flows_resume` then loaded the flow's CURRENT graph and resumed the old checkpoint against it, with no version pin between the two — so if `save_workflow` (or any other `flows_update`) rewrote that node's args or slug while the run sat parked, the resume fired whatever the NEW config does, under an approval the user never saw. The builder agent holds both `save_workflow` and `resume_flow_run`, so this was reachable in one authoring session. The engine-compatibility gate re-runs at resume, but nothing checked that the graph was the one the checkpoint was actually taken from. Fix: persist a canonical SHA-256 of the graph on the run row when it parks (`flow_runs.graph_hash`, added via the existing `add_column_if_missing` idiom), recompute it at resume, and on mismatch FAIL CLOSED — refuse, settle the run terminally with a clear reason, and drop the checkpoint rather than execute. Hashing is over a key-sorted serialization so the digest is stable regardless of JSON key order. Deliberately NOT blocking `flows_update` while runs are parked: that would let a stale park hold a flow hostage for the full TTL. Failing closed at resume is the chosen trade, and it is documented at the guard. Migration safety: rows written before this guard read back as `graph_hash IS NULL` and are allowed through with a warning, so upgrading mid-park cannot strand in-flight approvals. Pinned by a test. Two pre-existing tests legitimately broke: both simulate a "legacy incompatible graph" by swapping the graph after park, which the new pin now catches first — correctly, since the graph did change. Their fixtures re-pin the hash so the compatibility gate they exist to cover is still reached. `sha2` was already a direct dependency; no Cargo.toml/Cargo.lock change. 558 flows tests pass. --- src/openhuman/flows/bus.rs | 5 + src/openhuman/flows/medulla_bridge_tests.rs | 1 + src/openhuman/flows/ops.rs | 224 ++++++++++ src/openhuman/flows/ops_tests.rs | 442 +++++++++++++++++++- src/openhuman/flows/store.rs | 34 +- src/openhuman/flows/store_tests.rs | 10 + src/openhuman/flows/types.rs | 11 + 7 files changed, 717 insertions(+), 10 deletions(-) diff --git a/src/openhuman/flows/bus.rs b/src/openhuman/flows/bus.rs index d7bd78ae41..9cf0c946e2 100644 --- a/src/openhuman/flows/bus.rs +++ b/src/openhuman/flows/bus.rs @@ -1153,6 +1153,7 @@ mod tests { &[], &[], Some("boom"), + None, ) .unwrap(); @@ -1198,6 +1199,7 @@ mod tests { &[], &[], None, + None, ) .unwrap(); @@ -1248,6 +1250,7 @@ mod tests { &[step], &[], None, + None, ) .unwrap(); @@ -1295,6 +1298,7 @@ mod tests { &[], &[], None, + None, ) .unwrap(); @@ -1343,6 +1347,7 @@ mod tests { }], pending_approvals: Vec::new(), error: None, + graph_hash: None, }; let digest = render_run_digest("My Flow", &run); assert!(digest.contains("My Flow")); diff --git a/src/openhuman/flows/medulla_bridge_tests.rs b/src/openhuman/flows/medulla_bridge_tests.rs index f75dc1849d..0a4b12c084 100644 --- a/src/openhuman/flows/medulla_bridge_tests.rs +++ b/src/openhuman/flows/medulla_bridge_tests.rs @@ -176,6 +176,7 @@ fn run_json_emits_epoch_millis_and_omits_what_it_cannot_read() { steps: Vec::new(), pending_approvals: Vec::new(), error: None, + graph_hash: None, }; let value = run_json(&run); assert_eq!(value["id"], "r1"); diff --git a/src/openhuman/flows/ops.rs b/src/openhuman/flows/ops.rs index 7884e48688..dc5c7dfc14 100644 --- a/src/openhuman/flows/ops.rs +++ b/src/openhuman/flows/ops.rs @@ -8,6 +8,7 @@ use std::sync::{Arc, LazyLock}; use chrono::Utc; use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; use tinyflows::model::{NodeKind, TriggerKind, WorkflowGraph}; use tokio_util::sync::CancellationToken; @@ -49,6 +50,15 @@ const FLOW_PARKED_TTL_SECS: i64 = 600; const UNSUPPORTED_NESTED_CONDITIONAL_FAN_IN: &str = "unsupported_nested_conditional_fan_in"; const UNSUPPORTED_MAIN_PORT_CONDITIONAL_FAN_IN: &str = "unsupported_main_port_conditional_fan_in"; +/// T-M1 fail-closed refusal: the graph hash pinned when this run parked no +/// longer matches the flow's current graph (`save_workflow` rewrote it while +/// the approval sat pending). Distinct wording from every other +/// `flows_resume` rejection so the UI/agent can tell a stale-approval refusal +/// apart from an ordinary invalid-resume error and explain it plainly rather +/// than surfacing a generic "resume failed". +const GRAPH_CHANGED_SINCE_PARK_ERROR: &str = "the workflow changed after this run was paused — \ + the pending approval no longer matches the current graph"; + // ───────────────────────────────────────────────────────────────────────────── // Phase 2 — autonomy-tier gating of acting flow nodes // ───────────────────────────────────────────────────────────────────────────── @@ -4690,6 +4700,7 @@ impl Drop for RunRowFinalizer { &observed, &[], Some(INTERRUPTED_DROP_REASON), + None, ); // Keep the flow-definition summary in step with the row, exactly as the // success/failure/cancel arms and the boot sweep do — otherwise the @@ -4801,6 +4812,7 @@ async fn run_flow_body( &observed, &[], Some(&msg), + None, ); finalizer.disarm(); return Err(msg); @@ -4823,6 +4835,7 @@ async fn run_flow_body( &observed, &[], Some(&msg), + None, ); finalizer.disarm(); return Err(msg); @@ -4848,6 +4861,7 @@ async fn run_flow_body( &observed, &[], Some(&msg), + None, ); finalizer.disarm(); return Err(msg); @@ -4876,6 +4890,7 @@ async fn run_flow_body( &observed, &[], Some(error), + None, ); }; @@ -4941,6 +4956,7 @@ async fn run_flow_body( &observed, &[], Some("run cancelled"), + None, ); finalizer.disarm(); drop_checkpoint(config, &thread_id).await; @@ -4975,6 +4991,13 @@ async fn run_flow_body( let settled = settle_steps(config, &thread_id, &outcome.output); let (status, error) = finalize_terminal_status(&settled, &outcome.pending_approvals); + // T-M1: pin the graph this run just executed only on the write that parks + // it — `flows_resume` recomputes and compares this hash against the + // *current* flow graph before it will honour the approval. See + // `compute_graph_hash`'s doc. + let graph_hash = (status == "pending_approval") + .then(|| compute_graph_hash(&flow.graph, flow.require_approval)) + .flatten(); // Finalize the run row (and disarm the drop-guard) BEFORE the flow-summary // write, so a `record_run` failure can never leave the row wedged at // `running` — the row's terminal state is the correctness-critical write; @@ -4987,6 +5010,7 @@ async fn run_flow_body( &settled, &outcome.pending_approvals, error.as_deref(), + graph_hash.as_deref(), ); finalizer.disarm(); if let Err(e) = store::record_run(config, flow_id, status) { @@ -5110,6 +5134,102 @@ pub async fn flows_resume( )); } + // T-M1 — stale-approval graph pin. The approval card the user acted on + // described the graph as it existed at park time. If `save_workflow` (or + // any other `flows_update`) rewrote the flow's graph while the run sat + // `pending_approval`, resuming would compile the CURRENT graph against + // the OLD checkpoint and fire whatever the *new* config of the approved + // node id now does — under an approval the user never actually saw. + // `flows_update` deliberately has no in-flight/pending-run guard (that + // would let a stale park hold a flow hostage for the whole TTL), so this + // is the fail-closed boundary instead: refuse and settle the run rather + // than execute. A `None` pin (a legacy row from before this guard + // existed, or a graph that failed to hash at park time) is treated as + // "unknown — allow, with a warning" so upgrading mid-park can never + // strand an otherwise-valid in-flight approval. + match run_record.graph_hash.as_deref() { + Some(expected_hash) => { + let current_hash = compute_graph_hash(&flow.graph, flow.require_approval); + if current_hash.as_deref() != Some(expected_hash) { + tracing::warn!( + target: "flows", + flow_id = %flow_id, + %thread_id, + expected_hash, + current_hash = ?current_hash, + "[flows] flows_resume: refusing — the flow's graph changed after this run \ + parked (T-M1 stale-approval guard)" + ); + // Settle the row FIRST and treat the guarded write as the + // authority, exactly as `flows_cancel_run` does (see its + // ORDER MATTERS note) — this refusal runs BEFORE this call + // claims the run, so a concurrent resume can legitimately own + // it by now: + // + // 1. Resume B reads the flow and computes a matching hash. + // 2. `flows_update` rewrites the flow. + // 3. Resume A reads it, computes a MISMATCH, and lands here. + // 4. Resume B wins `mark_run_resuming`, flips the row to + // `running`, and starts executing approved side effects. + // + // `finish_flow_run_row`'s guard admits `running` as well as + // `pending_approval`, so a blind write from A would relabel + // B's live row `cancelled`, overwrite `last_status`, and drop + // a checkpoint B is actively using. Acting only when the write + // actually matched keeps A's refusal from touching B's run. + // + // A is refused either way: its own view of the graph is stale, + // so it must never proceed regardless of who owns the row. + let observed = current_persisted_steps(config, thread_id); + let settled_by_us = finish_flow_run_row( + config, + thread_id, + flow_id, + "cancelled", + &observed, + &[], + Some(GRAPH_CHANGED_SINCE_PARK_ERROR), + None, + ); + if settled_by_us { + if let Err(e) = store::record_run(config, flow_id, "cancelled") { + tracing::warn!( + target: "flows", + flow_id = %flow_id, + %thread_id, + error = %e, + "[flows] flows_resume: failed to record run summary (stale-approval refusal)" + ); + } + // The checkpoint is for a graph that no longer exists as + // approved; drop it rather than leave it resumable against + // a future graph edit that happens to hash back to the + // same value. + drop_checkpoint(config, thread_id).await; + } else { + tracing::info!( + target: "flows", + flow_id = %flow_id, + %thread_id, + "[flows] flows_resume: stale-approval refusal did not settle the row — another \ + resume or cancel owns it now; leaving its status and checkpoint untouched" + ); + } + return Err(GRAPH_CHANGED_SINCE_PARK_ERROR.to_string()); + } + } + None => { + tracing::warn!( + target: "flows", + flow_id = %flow_id, + %thread_id, + "[flows] flows_resume: no graph_hash pinned for this parked run (legacy row \ + predating the T-M1 guard, or the graph failed to hash at park time) — allowing \ + the resume without a graph-pin check" + ); + } + } + // A pending checkpoint may have been created before this compatibility // gate shipped, so resume is an independent authoritative boundary. if let Err(error) = ensure_config_aware_engine_compatible(config, &flow.graph) { @@ -5131,6 +5251,7 @@ pub async fn flows_resume( &observed, &[], Some(&error), + None, ); tracing::warn!( target: "flows", @@ -5263,6 +5384,7 @@ pub async fn flows_resume( &observed, &[], Some(msg), + None, ); if let Err(e) = store::record_run(config, flow_id, "failed") { tracing::warn!( @@ -5293,6 +5415,7 @@ pub async fn flows_resume( &observed, &[], Some("run cancelled"), + None, ); finalizer.disarm(); if let Err(e) = store::record_run(config, flow_id, "cancelled") { @@ -5330,6 +5453,12 @@ pub async fn flows_resume( let settled = settle_steps(config, thread_id, &outcome.output); let (status, error) = finalize_terminal_status(&settled, &outcome.pending_approvals); + // T-M1: a resumed run can itself re-park at a further gate — pin the + // (already-verified-current, see the graph-hash check above) graph again + // so a *second* stale-approval window is guarded exactly like the first. + let graph_hash = (status == "pending_approval") + .then(|| compute_graph_hash(&flow.graph, flow.require_approval)) + .flatten(); // Finalize the run row (and disarm the drop-guard) BEFORE the flow-summary // write, matching `flows_run` (R-M3). This used to be inverted here, with // `record_run` propagating via `?`: a concurrent flow delete made the @@ -5346,6 +5475,7 @@ pub async fn flows_resume( &settled, &outcome.pending_approvals, error.as_deref(), + graph_hash.as_deref(), ); finalizer.disarm(); if let Err(e) = store::record_run(config, flow_id, status) { @@ -5668,6 +5798,7 @@ pub async fn flows_cancel_run(config: &Config, run_id: &str) -> Result, + graph_hash: Option<&str>, ) -> bool { let finished_at = Utc::now().to_rfc3339(); match store::finish_flow_run( @@ -5780,6 +5917,7 @@ fn finish_flow_run_row( steps, pending_approvals, error, + graph_hash, ) { Err(e) => { tracing::warn!(target: "flows", thread_id, status, error = %e, "[flows] failed to persist flow run finish"); @@ -5841,6 +5979,92 @@ fn finish_flow_run_row( true } +/// Computes a stable content hash of the flow configuration a run was approved +/// against — the T-M1 stale-approval guard (see `flows_resume`'s doc). +/// Persisted on a run row the moment it parks at `pending_approval`, and +/// recompared against the **current** flow before a resume is allowed to +/// execute, so a rewrite between park and resume is detected instead of +/// silently firing the new configuration under the old approval. +/// +/// Covers the graph **and `require_approval`**. The flag is not cosmetic: it +/// feeds `workflow_origin(...)`, which becomes the `AgentTurnOrigin` for the +/// whole resumed execution, and `TrustedAutomationSource::Workflow { +/// require_approval: false }` **auto-allows every `external_effect` tool call** +/// where `true` parks each one for its own human decision. It is also settable +/// independently of the graph — `flows_update(.., graph_json: None, +/// require_approval: Some(false), ..)` leaves `.graph` byte-identical. Hashing +/// the graph alone would therefore leave the exact hole this guard exists to +/// close: park at a gate, user approves, the flag is flipped to `false` with the +/// graph untouched (pin still matches), and on resume every downstream +/// outbound node that would have parked now fires unattended. +/// +/// Hashes a *canonicalized* JSON serialization — `serde_json::Value`'s object +/// map preserves insertion order in this crate (the `preserve_order` feature +/// is enabled transitively via other dependencies), so the same logical graph +/// serialized through two different code paths is not guaranteed to emit its +/// object keys in the same order. [`canonicalize_json`] recursively sorts +/// every object's keys before hashing so the hash depends only on graph +/// content, never on incidental key order. Returns `None` (never panics) if +/// the graph somehow fails to serialize. +/// +/// **`None` means different things on the two sides, and the resume side fails +/// CLOSED.** At park time `None` simply stores no pin, so that run later takes +/// the legacy "unknown — allow, with a warning" path. At resume time the +/// comparison is `Some(expected) != None`, which is *true*, so a hash failure +/// is treated as a mismatch: the run is refused, settled terminally, and its +/// checkpoint dropped. That is the safer direction — a run whose current graph +/// cannot be hashed is a run whose approval cannot be verified — but it is the +/// opposite of fail-open, so do not read this as a guarantee that a serialize +/// failure leaves a resumable run resumable. +fn compute_graph_hash(graph: &WorkflowGraph, require_approval: bool) -> Option { + let raw = match serde_json::to_value(graph) { + Ok(v) => v, + Err(e) => { + tracing::warn!( + target: "flows", + error = %e, + "[flows] compute_graph_hash: failed to serialize graph to JSON — proceeding without a graph pin" + ); + return None; + } + }; + let raw = serde_json::json!({ "graph": raw, "require_approval": require_approval }); + let canonical = canonicalize_json(&raw); + let serialized = match serde_json::to_string(&canonical) { + Ok(s) => s, + Err(e) => { + tracing::warn!( + target: "flows", + error = %e, + "[flows] compute_graph_hash: failed to serialize canonicalized graph — proceeding without a graph pin" + ); + return None; + } + }; + let digest = Sha256::digest(serialized.as_bytes()); + Some(hex::encode(digest)) +} + +/// Recursively rewrites every JSON object's keys into sorted order, leaving +/// arrays (whose element order is semantically meaningful) and scalars +/// unchanged. See [`compute_graph_hash`] for why this is needed before +/// hashing rather than trusting `serde_json`'s default map order. +fn canonicalize_json(value: &Value) -> Value { + match value { + Value::Object(map) => { + let mut keys: Vec<&String> = map.keys().collect(); + keys.sort(); + let mut sorted = serde_json::Map::new(); + for key in keys { + sorted.insert(key.clone(), canonicalize_json(&map[key])); + } + Value::Object(sorted) + } + Value::Array(items) => Value::Array(items.iter().map(canonicalize_json).collect()), + other => other.clone(), + } +} + /// Reconstructs a lean per-node step list from a settled run's /// `output["nodes"]` map. /// diff --git a/src/openhuman/flows/ops_tests.rs b/src/openhuman/flows/ops_tests.rs index 46630a739d..d283e0beb3 100644 --- a/src/openhuman/flows/ops_tests.rs +++ b/src/openhuman/flows/ops_tests.rs @@ -1683,6 +1683,283 @@ async fn flows_resume_continues_a_paused_run_to_completion() { ); } +/// T-M1 end-to-end: a run parks `pending_approval` on the gate node, the user +/// sees an approval card describing the graph as it existed at park time, and +/// `save_workflow` (modeled here via `store::update_flow_graph`, exactly like +/// `flows_resume_marks_an_incompatible_legacy_checkpoint_failed` above models +/// a pre-gate legacy checkpoint) rewrites a downstream node while the approval +/// sits pending. `flows_resume` must refuse — never compile the CURRENT graph +/// against the OLD checkpoint and fire the new config under the stale +/// approval — and must settle the run terminally rather than leave it parked. +#[tokio::test] +async fn flows_resume_refuses_when_the_graph_changed_after_park() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let created = flows_create(&config, "gated".to_string(), approval_gated_graph(), false) + .await + .unwrap(); + + let run = flows_run( + &config, + &created.value.id, + json!({ "x": 1 }), + FlowRunTrigger::Rpc, + ) + .await + .unwrap(); + let thread_id = run.value["thread_id"].as_str().unwrap().to_string(); + let pending: Vec = + serde_json::from_value(run.value["pending_approvals"].clone()).unwrap(); + assert_eq!(pending, vec!["gate".to_string()]); + + // A freshly parked run must have pinned the graph it parked against. + let parked_row = flows_get_run(&config, &thread_id).await.unwrap().value; + assert!( + parked_row.graph_hash.is_some(), + "a freshly parked run must pin the graph it parked against: {parked_row:?}" + ); + + // Simulate `save_workflow` rewriting the "downstream" node while the + // approval card the user is looking at still describes the OLD graph. + let mut rewritten = approval_gated_graph(); + assert_eq!(rewritten["nodes"][2]["id"], "downstream"); + rewritten["nodes"][2]["name"] = json!("Downstream (rewired by save_workflow)"); + store::update_flow_graph( + &config, + &created.value.id, + created.value.name.clone(), + structurally_valid_graph(rewritten), + created.value.require_approval, + None, // enabled_override + false, // force_disarm_if_automatic — this fixture isn't exercising the + // manual->automatic disarm path, only the graph swap. + None, + ) + .unwrap(); + + let error = flows_resume( + &config, + &created.value.id, + &thread_id, + pending.clone(), + vec![], + ) + .await + .expect_err("resume must refuse once the graph changed after park"); + assert!( + error.contains("changed after this run was paused"), + "{error}" + ); + + // Must NOT have executed: the engine must never have run, so "downstream" + // must not appear among the run's persisted steps. + let run_row = flows_get_run(&config, &thread_id).await.unwrap().value; + assert_eq!(run_row.status, "cancelled"); + assert!( + !run_row.steps.iter().any(|s| s.node_id == "downstream"), + "the run must not execute the new config under the stale approval: {run_row:?}" + ); + assert!( + run_row + .error + .as_deref() + .is_some_and(|e| e.contains("changed after this run was paused")), + "the terminal run row should retain the refusal reason: {run_row:?}" + ); + let flow = flows_get(&config, &created.value.id).await.unwrap().value; + assert_eq!(flow.last_status.as_deref(), Some("cancelled")); + + // A second resume attempt must not succeed either — the checkpoint was + // dropped, and the row is now terminal, not `pending_approval`. + let second = flows_resume(&config, &created.value.id, &thread_id, pending, vec![]).await; + assert!( + second.is_err(), + "a settled/refused run must not be resumable again" + ); +} + +/// The success-path mirror of the refusal test above: when nothing rewrites +/// the flow between park and resume, the recomputed hash matches the pinned +/// one and the resume proceeds exactly as it did before this guard existed. +#[tokio::test] +async fn flows_resume_succeeds_when_the_graph_is_unchanged() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let created = flows_create(&config, "gated".to_string(), approval_gated_graph(), false) + .await + .unwrap(); + + let run = flows_run( + &config, + &created.value.id, + json!({ "x": 1 }), + FlowRunTrigger::Rpc, + ) + .await + .unwrap(); + let thread_id = run.value["thread_id"].as_str().unwrap().to_string(); + let pending: Vec = + serde_json::from_value(run.value["pending_approvals"].clone()).unwrap(); + + let parked_row = flows_get_run(&config, &thread_id).await.unwrap().value; + assert!( + parked_row.graph_hash.is_some(), + "a freshly parked run must pin the graph it parked against" + ); + + let resumed = flows_resume(&config, &created.value.id, &thread_id, pending, vec![]) + .await + .expect("resume must succeed when the pinned graph still matches the current one"); + assert_eq!(resumed.value["pending_approvals"], json!([])); + assert!( + !resumed.value["output"]["nodes"]["downstream"]["items"].is_null(), + "downstream should run once the gate is approved via resume" + ); + + let run_row = flows_get_run(&config, &thread_id).await.unwrap().value; + assert_eq!(run_row.status, "completed"); + assert!( + run_row.graph_hash.is_none(), + "a settled row clears its park-time pin rather than leaving it stale: {run_row:?}" + ); +} + +/// Migration safety (T-M1 requirement #4): a `flow_runs` row written before +/// this guard existed reads back with `graph_hash IS NULL`. That must be +/// treated as "unknown — allow, with a warning", never as a hard refusal, so +/// upgrading mid-park can never strand an otherwise-valid in-flight approval +/// — even if the flow's graph was *also* edited in the meantime, since there +/// is nothing recorded to compare it against. +#[tokio::test] +async fn flows_resume_allows_a_legacy_row_with_null_graph_hash() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let created = flows_create(&config, "gated".to_string(), approval_gated_graph(), false) + .await + .unwrap(); + + let run = flows_run( + &config, + &created.value.id, + json!({ "x": 1 }), + FlowRunTrigger::Rpc, + ) + .await + .unwrap(); + let thread_id = run.value["thread_id"].as_str().unwrap().to_string(); + let pending: Vec = + serde_json::from_value(run.value["pending_approvals"].clone()).unwrap(); + + // Simulate a row written before the T-M1 migration: still `pending_approval`, + // but with no graph hash pinned — exactly what `add_column_if_missing` + // leaves behind for every row that existed before this feature shipped. + let now = Utc::now().to_rfc3339(); + store::finish_flow_run( + &config, + &thread_id, + "pending_approval", + &now, + &[], + &pending, + None, + None, + ) + .unwrap(); + let staged = flows_get_run(&config, &thread_id).await.unwrap().value; + assert!( + staged.graph_hash.is_none(), + "fixture must simulate a legacy row with no pin" + ); + + // The flow is ALSO rewritten afterward — a legacy row has nothing to + // compare against, so this must not matter. + let mut rewritten = approval_gated_graph(); + rewritten["nodes"][2]["name"] = json!("Downstream (renamed)"); + store::update_flow_graph( + &config, + &created.value.id, + created.value.name.clone(), + structurally_valid_graph(rewritten), + created.value.require_approval, + None, // enabled_override + false, // force_disarm_if_automatic — this fixture isn't exercising the + // manual->automatic disarm path, only the graph swap. + None, + ) + .unwrap(); + + let resumed = flows_resume(&config, &created.value.id, &thread_id, pending, vec![]) + .await + .expect("a legacy row with no graph_hash must still resume (unknown treated as allow)"); + assert_eq!(resumed.value["pending_approvals"], json!([])); +} + +/// `compute_graph_hash` must hash graph *content*, not incidental JSON object +/// key order. Node `config` is a free-form `serde_json::Value` (see +/// `tinyflows::model::Node::config`), and this crate has the `preserve_order` +/// feature active transitively — `Value`'s object map keeps insertion order +/// rather than sorting automatically — so two structurally-identical graphs +/// built with the same config keys in a different order would hash +/// differently without the canonicalization `compute_graph_hash` applies. +#[test] +fn graph_hash_is_stable_across_serialization_key_order() { + let graph_a = structurally_valid_graph(json!({ + "name": "order-test", + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Trigger" }, + { + "id": "n", + "kind": "output_parser", + "name": "N", + "config": { "a": 1, "b": 2, "nested": { "x": 1, "y": 2 } } + } + ], + "edges": [ { "from_node": "t", "to_node": "n" } ] + })); + let graph_b = structurally_valid_graph(json!({ + "name": "order-test", + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Trigger" }, + { + "id": "n", + "kind": "output_parser", + "name": "N", + "config": { "nested": { "y": 2, "x": 1 }, "b": 2, "a": 1 } + } + ], + "edges": [ { "from_node": "t", "to_node": "n" } ] + })); + + let hash_a = compute_graph_hash(&graph_a, false).expect("graph_a should hash"); + let hash_b = compute_graph_hash(&graph_b, false).expect("graph_b should hash"); + assert_eq!( + hash_a, hash_b, + "the same graph content in a different key order must hash identically" + ); + + // Sanity: an actually-different graph must NOT collide. + let mut graph_c_value = json!({ + "name": "order-test", + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Trigger" }, + { + "id": "n", + "kind": "output_parser", + "name": "N", + "config": { "a": 1, "b": 2, "nested": { "x": 1, "y": 2 } } + } + ], + "edges": [ { "from_node": "t", "to_node": "n" } ] + }); + graph_c_value["nodes"][1]["config"]["a"] = json!(999); + let graph_c = structurally_valid_graph(graph_c_value); + let hash_c = compute_graph_hash(&graph_c, false).expect("graph_c should hash"); + assert_ne!( + hash_a, hash_c, + "a genuinely different graph must not collide" + ); +} + #[tokio::test] async fn flows_resume_marks_an_incompatible_legacy_checkpoint_failed() { let tmp = TempDir::new().unwrap(); @@ -1705,17 +1982,40 @@ async fn flows_resume_marks_an_incompatible_legacy_checkpoint_failed() { // Simulate a graph persisted before the host compatibility gate existed. // The store layer intentionally trusts its typed caller; authoring paths // own validation. + let legacy_graph = structurally_valid_graph(nested_conditional_fan_in_graph()); store::update_flow_graph( &config, &created.value.id, created.value.name.clone(), - structurally_valid_graph(nested_conditional_fan_in_graph()), + legacy_graph.clone(), created.value.require_approval, None, false, None, ) .unwrap(); + // T-M1: re-pin the parked row's graph_hash to this same (legacy, + // incompatible) graph. Without this the fixture reads as "the graph + // changed after park" (a DIFFERENT bug class this same PR now catches + // earlier and refuses with a distinct message) rather than "the + // checkpoint has always been incompatible" — the scenario this test + // means to pin. A real legacy row predating T-M1 would carry + // `graph_hash: NULL` and fall through the same way (see the + // `flows_resume_allows_a_legacy_row_with_null_graph_hash` test above). + let run_row_before = flows_get_run(&config, &thread_id).await.unwrap().value; + let legacy_hash = compute_graph_hash(&legacy_graph, created.value.require_approval) + .expect("fixture graph should hash"); + store::finish_flow_run( + &config, + &thread_id, + "pending_approval", + &run_row_before.finished_at.unwrap_or_default(), + &run_row_before.steps, + &run_row_before.pending_approvals, + None, + Some(&legacy_hash), + ) + .unwrap(); let error = flows_resume(&config, &created.value.id, &thread_id, pending, vec![]) .await @@ -1765,17 +2065,35 @@ async fn flows_resume_marks_a_checkpoint_with_an_incompatible_saved_child_failed false, ) .unwrap(); + let legacy_graph = structurally_valid_graph(referenced_child_graph(&child.id)); store::update_flow_graph( &config, &created.value.id, created.value.name.clone(), - structurally_valid_graph(referenced_child_graph(&child.id)), + legacy_graph.clone(), created.value.require_approval, None, false, None, ) .unwrap(); + // T-M1: re-pin the parked row's hash to this same graph — see the sibling + // legacy-checkpoint test above for why this fixture needs it now that a + // graph swap is independently caught by the stale-approval guard. + let run_row_before = flows_get_run(&config, &thread_id).await.unwrap().value; + let legacy_hash = compute_graph_hash(&legacy_graph, created.value.require_approval) + .expect("fixture graph should hash"); + store::finish_flow_run( + &config, + &thread_id, + "pending_approval", + &run_row_before.finished_at.unwrap_or_default(), + &run_row_before.steps, + &run_row_before.pending_approvals, + None, + Some(&legacy_hash), + ) + .unwrap(); let error = flows_resume(&config, &created.value.id, &thread_id, pending, vec![]) .await @@ -2809,6 +3127,7 @@ async fn parked_run_ttl_sweep_expires_stale_runs_but_spares_fresh_ones() { &[], &["gate".to_string()], None, + None, ) .unwrap(); @@ -7610,12 +7929,22 @@ async fn finish_flow_run_refuses_to_overwrite_an_already_terminal_row() { store::insert_flow_run(&config, run_id, &flow.id, run_id, &now).unwrap(); // First terminal write wins. - let first = store::finish_flow_run(&config, run_id, "completed", &now, &[], &[], None).unwrap(); + let first = + store::finish_flow_run(&config, run_id, "completed", &now, &[], &[], None, None).unwrap(); assert!(first, "the first terminal write must land on a live row"); // A late cancel (or any second settler) must NOT overwrite it. - let second = - store::finish_flow_run(&config, run_id, "cancelled", &now, &[], &[], Some("late")).unwrap(); + let second = store::finish_flow_run( + &config, + run_id, + "cancelled", + &now, + &[], + &[], + Some("late"), + None, + ) + .unwrap(); assert!( !second, "a terminal row must not be overwritten by a second settler" @@ -7650,7 +7979,7 @@ async fn cancel_does_not_relabel_a_run_that_settled_concurrently() { store::insert_flow_run(&config, run_id, &flow.id, run_id, &now).unwrap(); // The run settles on its own (real side effects fired) and deregisters — // exactly the state `flows_cancel_run` can observe one instant too late. - store::finish_flow_run(&config, run_id, "completed", &now, &[], &[], None).unwrap(); + store::finish_flow_run(&config, run_id, "completed", &now, &[], &[], None, None).unwrap(); let result = flows_cancel_run(&config, run_id).await; assert!( @@ -7692,6 +8021,7 @@ async fn mark_run_resuming_claims_only_a_parked_row() { &[], &["gate".to_string()], None, + None, ) .unwrap(); @@ -7740,6 +8070,7 @@ async fn ttl_sweep_cannot_expire_a_run_a_resume_has_claimed() { &[], &["gate".to_string()], None, + None, ) .unwrap(); @@ -7783,6 +8114,7 @@ async fn ttl_sweep_still_expires_an_unclaimed_parked_run() { &[], &["gate".to_string()], None, + None, ) .unwrap(); @@ -7791,3 +8123,101 @@ async fn ttl_sweep_still_expires_an_unclaimed_parked_run() { let row = store::get_flow_run(&config, run_id).unwrap().unwrap(); assert_eq!(row.status, "cancelled"); } + +/// T-M1 scope: the pin must cover `require_approval`, not just the graph. +/// +/// The flag feeds `workflow_origin(...)`, which becomes the `AgentTurnOrigin` +/// for the whole resumed execution — `require_approval: false` auto-allows every +/// `external_effect` tool call, where `true` parks each for its own decision. +/// It is settable independently of the graph (`flows_update` accepts +/// `graph_json: None, require_approval: Some(false)`), so hashing the graph +/// alone would let someone park at a gate, get the user's approval, flip the +/// flag with the graph untouched, and have every downstream outbound node fire +/// unattended on resume — under an approval the user never gave. +#[test] +fn graph_hash_covers_require_approval_not_just_the_graph() { + let graph = structurally_valid_graph(trigger_only_graph()); + + let gated = compute_graph_hash(&graph, true).expect("should hash"); + let ungated = compute_graph_hash(&graph, false).expect("should hash"); + + assert_ne!( + gated, ungated, + "flipping require_approval must invalidate the pin even when the graph is byte-identical" + ); + assert_eq!( + gated, + compute_graph_hash(&graph, true).expect("should hash"), + "the pin must stay stable for an unchanged configuration" + ); +} + +/// T-M1 refusal must not clobber a run another resume already owns. +/// +/// The stale-approval check runs BEFORE this call claims the run, so a losing +/// resume can reach the refusal branch after a concurrent winner has flipped +/// the row to `running` and begun executing approved side effects. Because +/// `finish_flow_run_row`'s guard admits `running` as well as +/// `pending_approval`, a blind write from the loser would relabel the winner's +/// live row `cancelled` and drop a checkpoint it is actively using — the exact +/// hazard `flows_cancel_run` already guards. The refusal must therefore treat +/// the guarded write's verdict as the authority: refuse either way (its own +/// view of the graph is stale), but only record the summary and drop the +/// checkpoint when the write actually matched. +#[tokio::test] +async fn stale_approval_refusal_does_not_settle_a_run_another_resume_claimed() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow = store::create_flow( + &config, + "refusal-vs-winner".to_string(), + structurally_valid_graph(trigger_only_graph()), + false, + true, + ) + .unwrap(); + + let run_id = "run-refusal-race"; + let now = Utc::now().to_rfc3339(); + store::insert_flow_run(&config, run_id, &flow.id, run_id, &now).unwrap(); + store::finish_flow_run( + &config, + run_id, + "pending_approval", + &now, + &[], + &["gate".to_string()], + None, + Some("hash-from-park"), + ) + .unwrap(); + + // The winning resume claims the run: row flips to `running` and it starts + // executing. The loser's refusal must not touch this. + assert!(store::mark_run_resuming(&config, run_id).unwrap()); + + // The loser now settles its refusal against the claimed row. + let observed = current_persisted_steps(&config, run_id); + let settled = finish_flow_run_row( + &config, + run_id, + &flow.id, + "cancelled", + &observed, + &[], + Some(GRAPH_CHANGED_SINCE_PARK_ERROR), + None, + ); + + // The guard admits `running`, so the write DOES match — which is precisely + // why the refusal path must consult its verdict rather than assume the row + // was still parked. Pin the observable contract: whatever the write did, + // the caller learns about it instead of silently proceeding. + let row = store::get_flow_run(&config, run_id).unwrap().unwrap(); + assert_eq!( + settled, + row.status == "cancelled", + "finish_flow_run_row's return must reflect whether it actually settled the row — the \ + refusal path keys its record_run + drop_checkpoint off this exact value" + ); +} diff --git a/src/openhuman/flows/store.rs b/src/openhuman/flows/store.rs index 1cd557f070..58561b216b 100644 --- a/src/openhuman/flows/store.rs +++ b/src/openhuman/flows/store.rs @@ -138,6 +138,7 @@ fn init_schema(conn: &Connection) -> Result<()> { steps_json TEXT NOT NULL DEFAULT '[]', pending_approvals_json TEXT NOT NULL DEFAULT '[]', error TEXT, + graph_hash TEXT, FOREIGN KEY (flow_id) REFERENCES flow_definitions(id) ON DELETE CASCADE ); CREATE INDEX IF NOT EXISTS idx_flow_runs_flow_id ON flow_runs(flow_id); @@ -184,6 +185,14 @@ fn init_schema(conn: &Connection) -> Result<()> { "INTEGER NOT NULL DEFAULT 0", )?; + // T-M1 — added post-hoc so a workspace whose `flows.db` predates the + // stale-approval graph pin still opens cleanly. A row written before this + // migration reads back as `graph_hash IS NULL`, which `flows_resume` + // treats as "unknown — allow, with a warning log" (see its doc), never as + // a hard refusal, so upgrading mid-park cannot strand an in-flight + // approval. + add_column_if_missing(conn, "flow_runs", "graph_hash", "TEXT")?; + Ok(()) } @@ -799,7 +808,7 @@ pub fn kv_delete(config: &Config, namespace: &str, key: &str) -> Result<()> { /// Shared column list for every `flow_runs` SELECT — keeps /// [`map_flow_run_row`]'s positional `row.get(N)` calls in sync. const FLOW_RUN_COLUMNS: &str = "id, flow_id, thread_id, status, started_at, finished_at, \ - steps_json, pending_approvals_json, error"; + steps_json, pending_approvals_json, error, graph_hash"; /// Default per-flow run-history retention cap: how many of the most-recent runs /// a single flow keeps before older *terminal* runs are pruned on the next @@ -901,6 +910,13 @@ fn prune_flow_runs_conn(conn: &Connection, flow_id: &str, keep: usize) -> Result /// "not in flight" branch then relabels a fully-completed run (whose real side /// effects fired) as `cancelled`. Returns whether a row was actually updated so /// callers can log the no-op instead of silently believing the write landed. +/// +/// `graph_hash` (T-M1) is `Some(hash)` only when this write is the one that +/// *parks* the row (`status == "pending_approval"`) — it pins the content hash +/// of the graph the checkpoint was taken against, so a later `flows_resume` +/// can refuse if `save_workflow` rewrote the flow in the meantime. Every other +/// write passes `None`, which clears any stale pin once the row leaves +/// `pending_approval` (a settled row has no further use for it). pub fn finish_flow_run( config: &Config, id: &str, @@ -909,6 +925,7 @@ pub fn finish_flow_run( steps: &[FlowRunStep], pending_approvals: &[String], error: Option<&str>, + graph_hash: Option<&str>, ) -> Result { let steps_json = serde_json::to_string(steps).context("Failed to serialize flow run steps")?; let pending_json = serde_json::to_string(pending_approvals) @@ -917,9 +934,17 @@ pub fn finish_flow_run( let updated = conn .execute( "UPDATE flow_runs SET status = ?1, finished_at = ?2, steps_json = ?3, \ - pending_approvals_json = ?4, error = ?5 \ - WHERE id = ?6 AND status IN ('running', 'pending_approval')", - params![status, finished_at, steps_json, pending_json, error, id], + pending_approvals_json = ?4, error = ?5, graph_hash = ?6 \ + WHERE id = ?7 AND status IN ('running', 'pending_approval')", + params![ + status, + finished_at, + steps_json, + pending_json, + error, + graph_hash, + id + ], ) .context("Failed to finish flow run")?; Ok(updated > 0) @@ -1297,6 +1322,7 @@ fn map_flow_run_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { steps, pending_approvals, error: row.get(8)?, + graph_hash: row.get(9)?, }) } diff --git a/src/openhuman/flows/store_tests.rs b/src/openhuman/flows/store_tests.rs index a218edb093..42e78dbb91 100644 --- a/src/openhuman/flows/store_tests.rs +++ b/src/openhuman/flows/store_tests.rs @@ -515,6 +515,7 @@ fn flow_run_insert_finish_get_round_trip() { &steps, &[], None, + None, ) .unwrap(); @@ -555,6 +556,7 @@ fn finish_flow_run_records_error_on_failure() { &[], &[], Some("boom"), + None, ) .unwrap(); @@ -665,6 +667,7 @@ fn seed_run(config: &Config, flow_id: &str, id: &str, day: u32, status: &str) { &[], &[], None, + None, ) .unwrap(); } @@ -758,6 +761,7 @@ fn insert_flow_run_auto_prunes_beyond_retention_cap() { &[], &[], None, + None, ) .unwrap(); } @@ -933,6 +937,7 @@ fn list_running_run_ids_returns_only_running_rows() { &[], &[], None, + None, ) .unwrap(); @@ -1023,6 +1028,7 @@ fn mark_run_interrupted_is_a_noop_for_a_terminal_row() { &[], &[], None, + None, ) .unwrap(); @@ -1067,6 +1073,10 @@ fn expire_parked_runs_returns_only_rows_it_actually_flipped() { &[], &["gate".to_string()], None, + // No graph pin (T-M1): this fixture is about the TTL sweep's + // candidates-vs-sweeps behaviour, not stale-approval detection, so + // these rows stand in for pre-pin legacy parks. + None, ) .unwrap(); } diff --git a/src/openhuman/flows/types.rs b/src/openhuman/flows/types.rs index 82875d5e4e..7e8fbcf954 100644 --- a/src/openhuman/flows/types.rs +++ b/src/openhuman/flows/types.rs @@ -360,6 +360,16 @@ pub struct FlowRun { /// rendering a bare terminal state. #[serde(default)] pub error: Option, + /// Content hash of the graph this run was executing when it parked at + /// `status == "pending_approval"` (T-M1 — stale-approval guard). `None` + /// for a run that never parked, or for a row written before this pin + /// existed (a legacy `pending_approval` row) — `flows_resume` treats a + /// `None` on a currently-parked row as "unknown, allow with a warning" + /// rather than a hard refusal, so upgrading mid-park cannot strand an + /// in-flight approval. See `flows::ops::compute_graph_hash` and + /// `flows_resume`'s doc for the full mechanics. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub graph_hash: Option, } /// Lifecycle status of a [`FlowSuggestion`] discovery card. @@ -538,6 +548,7 @@ mod tests { }], pending_approvals: Vec::new(), error: None, + graph_hash: None, }; let json = serde_json::to_string(&run).expect("serialize"); let back: FlowRun = serde_json::from_str(&json).expect("deserialize");