diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 1acd4377b6..399e120e26 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -1439,19 +1439,12 @@ async fn resume_workflow_after_approval( return; } - let workflow = match db.get_workflow(community_id, workflow_id).await { - Ok(w) => w, + let (run, def) = match engine.load_run_definition(community_id, run_id).await { + Ok(loaded) => loaded, Err(e) => { - tracing::error!("resume_workflow: failed to fetch workflow {workflow_id}: {e}"); - return; - } - }; - - let def: buzz_workflow::WorkflowDef = match serde_json::from_value(workflow.definition.clone()) - { - Ok(d) => d, - Err(e) => { - tracing::error!("resume_workflow: failed to parse workflow definition: {e}"); + tracing::error!( + "resume_workflow: failed to load signed definition for run {run_id}: {e}" + ); if let Err(db_err) = db .update_workflow_run( community_id, @@ -1461,7 +1454,7 @@ async fn resume_workflow_after_approval( &run.execution_trace, Some(buzz_db::workflow::WorkflowRunFailure { code: "invalid_definition", - message: &format!("definition parse error: {e}"), + message: &format!("signed run definition unavailable: {e}"), }), ) .await @@ -1472,6 +1465,30 @@ async fn resume_workflow_after_approval( } }; + if run.workflow_id != workflow_id { + tracing::error!( + "resume_workflow: approval workflow {workflow_id} does not match run workflow {}", + run.workflow_id + ); + if let Err(e) = db + .update_workflow_run( + community_id, + run_id, + RunStatus::Failed, + run.current_step, + &run.execution_trace, + Some(buzz_db::workflow::WorkflowRunFailure { + code: "approval_binding_mismatch", + message: "approval does not belong to the workflow run", + }), + ) + .await + { + tracing::error!("resume_workflow: failed to mark mismatched run as failed: {e}"); + } + return; + } + // Reconstruct step_outputs from execution trace for template resolution let mut initial_outputs: std::collections::HashMap = std::collections::HashMap::new(); @@ -1515,6 +1532,34 @@ mod postgres_tests { use super::*; use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + #[derive(Default)] + struct RecordingActionSink { + messages: std::sync::Mutex>, + } + + impl buzz_workflow::ActionSink for RecordingActionSink { + fn send_message( + &self, + _community_id: CommunityId, + _channel_id: &str, + text: &str, + _author_pubkey: &str, + _reply_to: Option<&str>, + ) -> std::pin::Pin< + Box< + dyn std::future::Future> + + Send + + '_, + >, + > { + self.messages + .lock() + .expect("recording action sink lock") + .push(text.to_string()); + Box::pin(async { Ok("recorded-event".to_string()) }) + } + } + async fn persistence_test_context() -> (buzz_db::Db, TenantContext) { let url = std::env::var("BUZZ_TEST_DATABASE_URL") .or_else(|_| std::env::var("DATABASE_URL")) @@ -1597,7 +1642,9 @@ mod postgres_tests { concat!( "name: manual-trigger-pool\n", "trigger:\n on: message_posted\n", - "steps:\n - id: send\n action: send_message\n text: done\n", + "steps:\n", + " - id: approval\n action: request_approval\n from: '@owner'\n message: approve\n", + " - id: send\n action: send_message\n text: done\n", ), ) .tags(vec![ @@ -1790,6 +1837,163 @@ mod postgres_tests { assert!(agent_result.message.contains("\"run_id\"")); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn approval_resume_executes_the_run_bound_signed_revision() { + let (state, tenant, _human, agent, workflow_id, revision_a) = + manual_trigger_test_context().await; + let community_id = tenant.community(); + let db = state.db.clone(); + let trigger_context = serde_json::to_value(TriggerContext { + channel_id: exact_tag_value(&revision_a, "h") + .unwrap_or_default() + .to_string(), + ..TriggerContext::default() + }) + .expect("serialize trigger context"); + let run_id = db + .create_workflow_run( + community_id, + workflow_id, + Some(revision_a.id.as_bytes()), + None, + Some(&trigger_context), + ) + .await + .expect("create revision A run"); + db.update_workflow_run( + community_id, + run_id, + RunStatus::WaitingApproval, + 0, + &serde_json::json!([{ + "step_id": "approval", + "output": {"approved": true} + }]), + None, + ) + .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() + .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"); + + let sink = Arc::new(RecordingActionSink::default()); + state.workflow_engine.set_action_sink(sink.clone()); + resume_workflow_after_approval( + Arc::clone(&state.workflow_engine), + db.clone(), + community_id, + run_id, + workflow_id, + 1, + ) + .await; + + let resumed = db + .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" + ); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn approval_resume_fails_closed_without_a_signed_run_revision() { + let (state, tenant, _human, _agent, workflow_id, _revision) = + manual_trigger_test_context().await; + let community_id = tenant.community(); + let db = state.db.clone(); + let run_id = db + .create_workflow_run(community_id, workflow_id, None, None, None) + .await + .expect("create legacy revisionless run"); + db.update_workflow_run( + community_id, + run_id, + RunStatus::WaitingApproval, + 0, + &serde_json::json!([]), + None, + ) + .await + .expect("suspend revisionless run"); + + resume_workflow_after_approval( + Arc::clone(&state.workflow_engine), + db.clone(), + community_id, + run_id, + workflow_id, + 1, + ) + .await; + + let failed = db + .get_workflow_run(community_id, run_id) + .await + .expect("load failed run"); + assert_eq!(failed.status, RunStatus::Failed); + assert_eq!(failed.error_code.as_deref(), Some("invalid_definition")); + assert!(failed + .error_message + .as_deref() + .is_some_and(|message| message.contains("no owner-signed definition revision"))); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn manual_trigger_rejects_non_owner_and_stale_or_missing_revision() { diff --git a/crates/buzz-test-client/tests/conformance_multitenant.rs b/crates/buzz-test-client/tests/conformance_multitenant.rs index 4c8c8904ac..70740c9889 100644 --- a/crates/buzz-test-client/tests/conformance_multitenant.rs +++ b/crates/buzz-test-client/tests/conformance_multitenant.rs @@ -1778,7 +1778,31 @@ mod workflows { .to_string() } - /// Fire a workflow by id on `http_base`'s community (kind:46020, `d`=id). + /// Fetch the current owner-signed workflow revision from `http_base`'s + /// community. Manual triggers must name this exact event id. + async fn workflow_revision(http_base: &str, keys: &Keys, workflow_id: &str) -> String { + let resp = reqwest::Client::new() + .get(format!("{http_base}/workflows/{workflow_id}/revision")) + .header("X-Pubkey", keys.public_key().to_hex()) + .send() + .await + .unwrap_or_else(|e| panic!("GET workflow revision from {http_base} failed: {e}")); + let status = resp.status(); + let body = resp.text().await.expect("read workflow revision body"); + assert!( + status.is_success(), + "workflow revision lookup against {http_base} failed with {status}: {body}" + ); + let event: serde_json::Value = serde_json::from_str(&body) + .unwrap_or_else(|e| panic!("parse workflow revision JSON: {e} ({body:?})")); + event["id"] + .as_str() + .unwrap_or_else(|| panic!("workflow revision response missing event id: {event}")) + .to_string() + } + + /// Fire an exact workflow revision on `http_base`'s community + /// (kind:46020, `d`=workflow id, `e`=signed revision id). /// Returns a normalized `{accepted, message}` body so the caller can assert /// on the *wire-observable* accept/reject and message. The HTTP bridge maps /// `IngestError::Rejected` to HTTP 400 + `{error}` while the WS door maps the @@ -1788,9 +1812,13 @@ mod workflows { http_base: &str, keys: &Keys, workflow_id: &str, + revision: &str, ) -> serde_json::Value { let event = EventBuilder::new(Kind::Custom(KIND_WORKFLOW_TRIGGER), "") - .tags(vec![Tag::parse(["d", workflow_id]).unwrap()]) + .tags(vec![ + Tag::parse(["d", workflow_id]).unwrap(), + Tag::parse(["e", revision]).unwrap(), + ]) .sign_with_keys(keys) .unwrap(); @@ -1837,9 +1865,11 @@ mod workflows { /// This deliberately removes "not a member of U in B" as an alternate /// cause of the B rejection — K *is* a member of U in B. /// 2. Define a workflow in `U` under **A** (kind:30620). The server - /// generates `W` and returns it. `W` is an A-community row. - /// 3. Fire `W` under host **B** (kind:46020, `d`=W) as K. Must be - /// rejected — `accepted == false` and the generic `workflow not found` + /// generates `W` and returns it. Retrieve its current signed revision + /// `R` through A's authenticated revision endpoint. `W` and `R` are + /// A-community records. + /// 3. Fire `W`/`R` under host **B** (kind:46020, `d`=W, `e`=R) as K. Must + /// be rejected — `accepted == false` and the generic `workflow not found` /// message — because `get_workflow(B_community, W)` finds nothing: `W` /// exists only in A. K's membership of U-in-B is irrelevant; the /// lookup never reaches the membership check. @@ -1893,11 +1923,17 @@ mod workflows { uuid::Uuid::parse_str(&workflow_id).is_ok(), "server-generated workflow_id must be a UUID, got {workflow_id:?}" ); + let revision = workflow_revision(&http_a, &keys, &workflow_id).await; + assert_eq!( + revision.len(), + 64, + "signed workflow revision must be a 32-byte event id" + ); - // (3) Fire W under host B as K. Must fail closed: W is an A-community - // row, and get_workflow(B_community, W) finds nothing. K is a member of - // U in B, so a leak here is the community fence failing, not membership. - let b_resp = trigger_workflow(&http_b, &keys, &workflow_id).await; + // (3) Fire W/R under host B as K. Keep the trigger grammar otherwise + // valid with A's real signed revision so only tenant-scoped workflow + // resolution can explain the rejection. + let b_resp = trigger_workflow(&http_b, &keys, &workflow_id, &revision).await; assert_eq!( b_resp["accepted"].as_bool(), Some(false), @@ -1915,7 +1951,7 @@ mod workflows { // proves the B rejection is community confinement, not an // untriggerable workflow, and exercises the same-community happy path // through the fence under test. - let a_resp = trigger_workflow(&http_a, &keys, &workflow_id).await; + let a_resp = trigger_workflow(&http_a, &keys, &workflow_id, &revision).await; assert_eq!( a_resp["accepted"].as_bool(), Some(true),