From 1b1922d896c2e7f869032505db1fb31f172e63ef Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Fri, 4 Sep 2026 06:48:10 -0400 Subject: [PATCH] fix(workflow): reject stale captured manual continuations Signed-off-by: Logan Johnson --- crates/buzz-relay/src/api/workflows.rs | 2 +- .../src/handlers/command_executor.rs | 143 +++++++++++------- crates/buzz-workflow/src/lib.rs | 14 +- 3 files changed, 97 insertions(+), 62 deletions(-) diff --git a/crates/buzz-relay/src/api/workflows.rs b/crates/buzz-relay/src/api/workflows.rs index cccfbe9bbd..06da8fbc84 100644 --- a/crates/buzz-relay/src/api/workflows.rs +++ b/crates/buzz-relay/src/api/workflows.rs @@ -452,7 +452,7 @@ mod postgres_tests { let definition_hash = sha2::Sha256::digest(definition_json.as_bytes()); let mut tx = state .db - .begin_transaction() + .begin_event_write_transaction() .await .expect("begin workflow seed"); buzz_db::event::insert_event_in_transaction( diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 38450f729e..ea4502bee9 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -1666,7 +1666,7 @@ mod postgres_tests { .expect("parse signed workflow definition"); let definition_hash = compute_definition_hash(&definition_json); let mut tx = setup_db - .begin_transaction() + .begin_event_write_transaction() .await .expect("begin workflow seed"); buzz_db::event::insert_event_in_transaction( @@ -1866,7 +1866,23 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] - async fn approval_resume_executes_the_run_bound_signed_revision() { + async fn approval_resume_rejects_replaced_run_revision() { + assert_approval_revision(Some(true)).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn approval_resume_rejects_deleted_run_revision() { + assert_approval_revision(Some(false)).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn approval_resume_accepts_unchanged_run_revision() { + assert_approval_revision(None).await; + } + + async fn assert_approval_revision(change: Option) { let (state, tenant, _human, agent, workflow_id, revision_a) = manual_trigger_test_context().await; let community_id = tenant.community(); @@ -1902,53 +1918,64 @@ mod postgres_tests { .await .expect("suspend revision A run for approval"); - let channel_id = Uuid::parse_str(exact_tag_value(&revision_a, "h").expect("channel tag")) - .expect("channel UUID"); - let revision_b = EventBuilder::new( - Kind::Custom(KIND_WORKFLOW_DEF as u16), - concat!( - "name: revision-b\n", - "trigger:\n on: message_posted\n", - "steps:\n", - " - id: approval\n action: request_approval\n from: '@owner'\n message: approve\n", - " - id: after\n action: send_message\n text: revision B\n", - ), - ) - .tags(vec![ - Tag::parse(["d", workflow_id.to_string().as_str()]).expect("d tag"), - Tag::parse(["h", channel_id.to_string().as_str()]).expect("h tag"), - ]) - .sign_with_keys(&agent) - .expect("sign revision B"); - let (_, definition_b_json) = buzz_workflow::WorkflowEngine::parse_yaml(&revision_b.content) - .expect("parse revision B"); - let definition_b_hash = compute_definition_hash(&definition_b_json); - let mut tx = db - .begin_transaction() + if change == Some(true) { + let channel_id = + Uuid::parse_str(exact_tag_value(&revision_a, "h").expect("channel tag")) + .expect("channel UUID"); + let revision_b = EventBuilder::new( + Kind::Custom(KIND_WORKFLOW_DEF as u16), + concat!( + "name: revision-b\n", + "trigger:\n on: message_posted\n", + "steps:\n", + " - id: approval\n action: request_approval\n from: '@owner'\n message: approve\n", + " - id: after\n action: send_message\n text: revision B\n", + ), + ) + .tags(vec![ + Tag::parse(["d", workflow_id.to_string().as_str()]).expect("d tag"), + Tag::parse(["h", channel_id.to_string().as_str()]).expect("h tag"), + ]) + .custom_created_at(nostr::Timestamp::from(revision_a.created_at.as_secs() + 1)) + .sign_with_keys(&agent) + .expect("sign revision B"); + let (_, definition_b_json) = + buzz_workflow::WorkflowEngine::parse_yaml(&revision_b.content) + .expect("parse revision B"); + let definition_b_hash = compute_definition_hash(&definition_b_json); + let mut tx = db + .begin_event_write_transaction() + .await + .expect("begin revision B update"); + db.replace_parameterized_event_in_transaction( + &mut tx, + community_id, + &revision_b, + &workflow_id.to_string(), + Some(channel_id), + buzz_db::replaceable::ParameterizedReplacePrecondition::Unconditional, + ) .await - .expect("begin revision B update"); - buzz_db::event::insert_event_in_transaction( - &mut tx, - community_id, - &revision_b, - Some(channel_id), - ) - .await - .expect("persist revision B"); - db.upsert_workflow( - &mut tx, - community_id, - workflow_id, - Some(channel_id), - &agent.public_key().to_bytes(), - "revision-b", - &definition_b_json, - &definition_b_hash, - revision_b.id.as_bytes(), - ) - .await - .expect("materialize revision B"); - tx.commit().await.expect("commit revision B update"); + .expect("persist revision B"); + db.upsert_workflow( + &mut tx, + community_id, + workflow_id, + Some(channel_id), + &agent.public_key().to_bytes(), + "revision-b", + &definition_b_json, + &definition_b_hash, + revision_b.id.as_bytes(), + ) + .await + .expect("materialize revision B"); + tx.commit().await.expect("commit revision B update"); + } else if change == Some(false) { + db.soft_delete_event(community_id, revision_a.id.as_bytes()) + .await + .expect("delete captured revision"); + } let sink = Arc::new(RecordingActionSink::default()); state.workflow_engine.set_action_sink(sink.clone()); @@ -1966,15 +1993,17 @@ mod postgres_tests { .get_workflow_run(community_id, run_id) .await .expect("load resumed run"); - assert_eq!(resumed.status, RunStatus::Completed); - assert_eq!( - sink.messages - .lock() - .expect("recorded messages lock") - .as_slice(), - ["done"], - "approval resume must execute revision A, never current revision B" - ); + let messages = sink.messages.lock().expect("recorded messages lock"); + if change.is_some() { + assert_eq!(resumed.status, RunStatus::Failed); + assert!( + messages.is_empty(), + "stale approval must execute neither captured A nor current B" + ); + } else { + assert_eq!(resumed.status, RunStatus::Completed); + assert_eq!(messages.as_slice(), ["done"]); + } } #[tokio::test] diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index fb44b6d568..2197fc1a3f 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -126,9 +126,10 @@ impl WorkflowEngine { /// Load and verify the exact owner-signed definition bound to a run. /// - /// The mutable `workflows` row supplies only immutable identity/channel - /// binding. Definition content always comes from the run's signed event; - /// legacy runs without a revision fail closed. + /// The mutable `workflows` row supplies identity/channel binding and the + /// current revision pointer. Content always comes from the run's signed event; + /// legacy runs without a revision fail closed. Replaced or explicitly deleted + /// definitions cannot authorize pending execution or continuation. pub async fn load_run_definition( &self, community_id: CommunityId, @@ -141,9 +142,14 @@ impl WorkflowEngine { ) })?; let workflow = self.db.get_workflow(community_id, run.workflow_id).await?; + if workflow.definition_event_id.as_deref() != Some(revision) { + return Err(WorkflowError::InvalidDefinition( + "workflow run definition revision is no longer current".into(), + )); + } let stored = self .db - .get_event_by_id_including_deleted(community_id, revision) + .get_event_by_id(community_id, revision) .await? .ok_or_else(|| { WorkflowError::InvalidDefinition(