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");