From 38b2681ea541947c74060e5285d38e4b09cd6c13 Mon Sep 17 00:00:00 2001 From: Broc Oppler Date: Tue, 21 Jul 2026 22:27:12 -0700 Subject: [PATCH] feat(workflow): persist and resume approval gates (WF-08) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Runs that hit a request_approval step were marked Failed with 'approval gates not yet implemented' — the executor minted a token and returned Suspended, but finalize_run discarded it, so the grant/deny handlers, resume_workflow_after_approval, and the workflow_approvals CRUD were all dead code. This wires the missing suspend half: - executor: StepResult::Suspended now carries PendingApproval (token, step_id, resolved approver spec, message, timeout). The approver spec is resolved up front — "any", hex pubkey, or npub1…, with or without a leading @ — and unresolvable specs (display names) fail the step loudly instead of minting an approval that check_approver_spec could never authorize. Invalid timeout strings now error at the step too. - finalize_run: replaces the Failed stub with arm-and-announce ordering: create the workflow_approvals row (raw token hashed by buzz-db), move the run to WaitingApproval at the suspended step, then announce via the new ActionSink::emit_approval_requested. Any persistence failure marks the run Failed so a run never waits on an ungrantable gate; the announcement is best-effort once the gate is armed, logged at ERROR on failure. - relay sink: emits kind:46010 signed by the relay keypair — d tag = SHA-256 token hash (the grant handlers' lookup key, now derived from the shared buzz_db::workflow::hash_approval_token), p tag = resolved approver (or workflow owner for "any") so the request lands in the needs-action feed, h tag when the workflow is channel-scoped, raw token + message + run/step metadata in the JSON content. With the row minted and the run in WaitingApproval, the existing grant path (handle_approval_grant → resume_workflow_after_approval → execute_from_step) and deny path (→ Cancelled) operate unchanged. Not covered here, noted for follow-up: kind:46011/46012 lifecycle emission on grant/deny, approval-token multi-tenant conformance rows (currently #[ignore]d pending_lane), and display-name approver resolution via the member directory. Co-Authored-By: Claude Fable 5 Signed-off-by: Broc Oppler --- crates/buzz-db/src/workflow.rs | 6 +- crates/buzz-relay/src/workflow_sink.rs | 124 +++++++++++++++- crates/buzz-workflow/src/action_sink.rs | 40 +++++ crates/buzz-workflow/src/executor.rs | 136 ++++++++++++++--- crates/buzz-workflow/src/lib.rs | 186 ++++++++++++++++++++---- 5 files changed, 447 insertions(+), 45 deletions(-) diff --git a/crates/buzz-db/src/workflow.rs b/crates/buzz-db/src/workflow.rs index 9c02f162c9..8378262432 100644 --- a/crates/buzz-db/src/workflow.rs +++ b/crates/buzz-db/src/workflow.rs @@ -30,7 +30,11 @@ pub const LIST_MAX_LIMIT: i64 = 1000; /// /// Approval tokens are stored hashed so that a DB read does not expose /// the raw token (same pattern as API tokens in buzz-auth). -fn hash_approval_token(token: &str) -> Vec { +/// +/// Public so the relay can derive the same hash when it announces a pending +/// approval (the kind:46010 `d` tag) — keeping every token-hash derivation on +/// this one definition. +pub fn hash_approval_token(token: &str) -> Vec { Sha256::digest(token.as_bytes()).to_vec() } diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 97c31c2561..d269d368e9 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -8,9 +8,9 @@ use std::future::Future; use std::pin::Pin; use std::sync::{Arc, Weak}; -use buzz_core::kind::KIND_STREAM_MESSAGE; +use buzz_core::kind::{KIND_STREAM_MESSAGE, KIND_WORKFLOW_APPROVAL_REQUESTED}; use buzz_core::tenant::CommunityId; -use buzz_workflow::action_sink::{ActionSink, ActionSinkError}; +use buzz_workflow::action_sink::{ActionSink, ActionSinkError, ApprovalRequestNotice}; use chrono::Utc; use nostr::{EventBuilder, Kind, Tag}; use tracing::info; @@ -362,6 +362,126 @@ impl ActionSink for RelayActionSink { Ok(event_id_hex) }) } + + fn emit_approval_requested( + &self, + community_id: CommunityId, + notice: ApprovalRequestNotice, + ) -> Pin> + Send + '_>> { + Box::pin(async move { + let state = self + .state + .upgrade() + .ok_or_else(|| ActionSinkError::Database("relay is shutting down".into()))?; + + // Same tenant-binding rule as send_message: the suspended run + // carries its owning community; never re-derive from config. + let host = state + .db + .lookup_community_host(community_id) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))? + .ok_or_else(|| { + ActionSinkError::Database(format!( + "workflow run community {community_id} is not mapped to a host" + )) + })?; + let tenant = buzz_core::tenant::TenantContext::resolved(community_id, host); + + // `d` tag carries the token *hash* — the exact lookup key + // grant/deny handlers use (`get_approval_by_stored_hash`). The + // raw token travels only in the content payload. + let token_hash_hex = + hex::encode(buzz_db::workflow::hash_approval_token(¬ice.raw_token)); + + let content = serde_json::json!({ + "message": notice.message, + "token": notice.raw_token, + "workflow_id": notice.workflow_id, + "run_id": notice.run_id, + "step_id": notice.step_id, + "expires_at": notice.expires_at.to_rfc3339(), + }) + .to_string(); + + let mut tags = vec![ + Tag::parse(["d", &token_hash_hex]) + .map_err(|e| ActionSinkError::EventBuild(format!("d tag: {e}")))?, + Tag::parse(["p", ¬ice.notify_pubkey_hex]) + .map_err(|e| ActionSinkError::EventBuild(format!("p tag: {e}")))?, + Tag::parse(["buzz:workflow", "true"]) + .map_err(|e| ActionSinkError::EventBuild(format!("workflow tag: {e}")))?, + ]; + if let Some(channel_uuid) = notice.channel_id { + tags.push( + Tag::parse(["h", &channel_uuid.to_string()]) + .map_err(|e| ActionSinkError::EventBuild(format!("h tag: {e}")))?, + ); + } + + let kind = Kind::from(KIND_WORKFLOW_APPROVAL_REQUESTED as u16); + let event = EventBuilder::new(kind, &content) + .tags(tags) + .sign_with_keys(&state.relay_keypair) + .map_err(|e| ActionSinkError::EventBuild(format!("signing: {e}")))?; + + let event_id_hex = event.id.to_hex(); + let event_id_bytes = event.id.as_bytes().to_vec(); + let event_created_at = { + let ts = event.created_at.as_secs() as i64; + chrono::DateTime::from_timestamp(ts, 0).unwrap_or_else(Utc::now) + }; + + info!( + event_id = %event_id_hex, + run_id = %notice.run_id, + step_id = %notice.step_id, + "Workflow RequestApproval: announcing kind {KIND_WORKFLOW_APPROVAL_REQUESTED} event" + ); + + // Approval announcements are top-level events; give them thread + // metadata only when channel-scoped (same shape as send_message). + let thread_meta = + notice + .channel_id + .map(|channel_uuid| buzz_db::event::ThreadMetadataParams { + event_id: &event_id_bytes, + event_created_at, + channel_id: channel_uuid, + parent_event_id: None, + parent_event_created_at: None, + root_event_id: None, + root_event_created_at: None, + depth: 0, + broadcast: false, + }); + + let (stored_event, was_inserted) = state + .db + .insert_event_with_thread_metadata( + tenant.community(), + &event, + notice.channel_id, + thread_meta, + ) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))?; + + if was_inserted { + let _ = dispatch_persistent_event( + &tenant, + &state, + &stored_event, + KIND_WORKFLOW_APPROVAL_REQUESTED, + ¬ice.notify_pubkey_hex, + None, + ) + .await; + } + + Ok(event_id_hex) + }) + } } #[cfg(test)] diff --git a/crates/buzz-workflow/src/action_sink.rs b/crates/buzz-workflow/src/action_sink.rs index 0c6002e74e..450a05848e 100644 --- a/crates/buzz-workflow/src/action_sink.rs +++ b/crates/buzz-workflow/src/action_sink.rs @@ -66,4 +66,44 @@ pub trait ActionSink: Send + Sync { text: &str, author_pubkey: &str, ) -> Pin> + Send + '_>>; + + /// Announce a pending approval as a kind:46010 event (WF-08). + /// + /// The event carries the raw token in its JSON content (approvers hash it + /// into the grant's `d` tag via `buzz workflows approve --token …`), a + /// `d` tag holding the SHA-256 token hash so relay handlers and clients + /// can correlate it with grant/deny events, and a `p` tag for the notify + /// pubkey so the request surfaces in that user's needs-action feed. + /// + /// Returns the event ID hex string on success. + fn emit_approval_requested( + &self, + community_id: CommunityId, + notice: ApprovalRequestNotice, + ) -> Pin> + Send + '_>>; +} + +/// Everything the sink needs to announce a pending approval (kind:46010). +#[derive(Debug, Clone)] +pub struct ApprovalRequestNotice { + /// Channel to post into (the workflow's channel), if the workflow is + /// channel-scoped. `None` emits an unscoped event that still reaches the + /// notify pubkey's feed. + pub channel_id: Option, + /// Raw approval token — never persisted raw; delivered so approvers can + /// present it back to the grant endpoint. + pub raw_token: String, + /// Template-resolved, human-facing approval message. + pub message: String, + /// Pubkey (hex) to `p`-tag: the resolved approver, or the workflow owner + /// when the spec is `"any"`. + pub notify_pubkey_hex: String, + /// The workflow the suspended run belongs to. + pub workflow_id: uuid::Uuid, + /// The suspended run awaiting this approval. + pub run_id: uuid::Uuid, + /// The step that requested approval. + pub step_id: String, + /// When the approval window closes. + pub expires_at: chrono::DateTime, } diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index a029b44622..9120e928dd 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -451,16 +451,34 @@ pub fn resolve_step_templates( } } +/// Suspend metadata carried out of the run loop when a `RequestApproval` +/// step suspends execution (WF-08): everything the engine needs to persist +/// the approval record and announce it — the raw token, the requesting step, +/// the resolved approver, and the human-facing message. +#[derive(Debug, Clone)] +pub struct PendingApproval { + /// Raw approval token. The DB layer hashes it at persist time; it is + /// carried raw here so the kind:46010 announcement can deliver it to + /// approvers (who hash it back into the grant's `d` tag). + pub token: String, + /// The step that requested approval. + pub step_id: String, + /// Approver spec in grant-check form: `"any"` or a 64-char hex pubkey. + pub approver_spec: String, + /// Template-resolved, human-facing approval message. + pub message: String, + /// Approval window in seconds (from `timeout:`, default 24h). + pub timeout_secs: u64, +} + /// Result of dispatching a single step action. #[derive(Debug)] pub enum StepResult { /// Step completed normally. Output is stored in `step_outputs`. Completed(JsonValue), - /// Step requests suspension (approval gate). Execution must pause. - Suspended { - /// Token used to resume or reject this approval gate. - approval_token: String, - }, + /// Step requests suspension (approval gate). Execution must pause and the + /// caller must persist the approval so a later grant can resume the run. + Suspended(PendingApproval), /// Step was skipped due to `if:` condition being false. Skipped, } @@ -653,6 +671,14 @@ pub async fn dispatch_action( timeout, } => { let timeout_str = timeout.as_deref().unwrap_or("24h"); + let timeout_secs = parse_duration_secs(timeout_str)?; + // Resolve the approver before any state is persisted: the relay's + // grant authorization (`check_approver_spec`) accepts only "any" + // or a hex pubkey, so an unresolvable spec would mint an approval + // no grant can ever satisfy. Fail the step loudly instead of + // suspending into a dead end. + let approver_spec = + resolve_approver_spec(from).map_err(WorkflowError::InvalidDefinition)?; info!( run_id = %run_id, step = step_id, "RequestApproval from={from} timeout={timeout_str}: {message}" @@ -660,12 +686,13 @@ pub async fn dispatch_action( let token = generate_approval_token(run_id, step_id); - // TODO (WF-08): create approval record in DB, emit kind:46010. - // For now, return Suspended with the token so the caller can persist state. - - Ok(StepResult::Suspended { - approval_token: token, - }) + Ok(StepResult::Suspended(PendingApproval { + token, + step_id: step_id.to_string(), + approver_spec, + message: message.clone(), + timeout_secs, + })) } Delay { duration } => { @@ -699,6 +726,36 @@ fn generate_approval_token(_run_id: Uuid, _step_id: &str) -> String { Uuid::new_v4().to_string() } +/// Resolve a `request_approval.from` spec into the form the relay's grant +/// authorization (`check_approver_spec`) understands: `"any"` or a lowercase +/// 64-char hex pubkey. +/// +/// Accepted inputs: `""`/`"any"` (case-insensitive), a hex pubkey, an +/// `npub1…` bech32 pubkey, or any of those with a leading `@`. Display-name +/// mentions (e.g. `@release-manager`) cannot be resolved here — the engine +/// has no name directory at suspend time — and are rejected so the run fails +/// visibly instead of minting an approval nobody can ever grant. +pub(crate) fn resolve_approver_spec(from: &str) -> Result { + let spec = from.trim(); + if spec.is_empty() || spec.eq_ignore_ascii_case("any") { + return Ok("any".to_string()); + } + let bare = spec.strip_prefix('@').unwrap_or(spec); + if bare.len() == 64 && bare.bytes().all(|b| b.is_ascii_hexdigit()) { + return Ok(bare.to_ascii_lowercase()); + } + if bare.starts_with("npub1") { + use nostr::prelude::FromBech32; + if let Ok(pk) = nostr::PublicKey::from_bech32(bare) { + return Ok(pk.to_hex()); + } + } + Err(format!( + "request_approval.from '{from}' cannot be resolved to an approver — \ + use \"any\", a 64-char hex pubkey, or an npub1… key" + )) +} + /// Parse a duration string like "5m", "1h", "30s" into seconds. /// /// Exposed as `pub(crate)` so `schema.rs` can use it for interval validation. @@ -939,7 +996,7 @@ async fn add_reaction_impl(message_id: &str, emoji: &str) -> Result, + pub approval: Option, /// Index of the step that suspended (or the total step count on completion). pub step_index: usize, /// Accumulated step outputs at the point of suspension or completion. @@ -1180,15 +1237,15 @@ async fn execute_steps( })); step_outputs.insert(step.id.clone(), output); } - StepResult::Suspended { approval_token } => { + StepResult::Suspended(pending) => { info!( run_id = %run_id, step = %step.id, "Step suspended — awaiting approval (token: )" ); - // Return the token and current state so the caller can persist the - // approval record and update the run's execution trace. + // Return the pending approval and current state so the caller + // can persist the approval record and update the run status. return Ok(ExecutionResult { - approval_token: Some(approval_token), + approval: Some(pending), step_index: i, step_outputs, trace, @@ -1206,7 +1263,7 @@ async fn execute_steps( info!(run_id = %run_id, "Workflow run completed"); Ok(ExecutionResult { - approval_token: None, + approval: None, step_index: def.steps.len(), step_outputs, trace, @@ -1831,4 +1888,49 @@ mod tests { .expect("override should be accepted"); assert_eq!(resolved, override_channel_id.to_string()); } + + #[test] + fn approver_spec_any_and_empty_resolve_to_any() { + assert_eq!(resolve_approver_spec("").unwrap(), "any"); + assert_eq!(resolve_approver_spec(" ").unwrap(), "any"); + assert_eq!(resolve_approver_spec("any").unwrap(), "any"); + assert_eq!(resolve_approver_spec("ANY").unwrap(), "any"); + } + + #[test] + fn approver_spec_hex_pubkey_resolves_lowercased() { + let hex_upper: String = "AB".repeat(32); + let hex_lower = hex_upper.to_ascii_lowercase(); + assert_eq!(resolve_approver_spec(&hex_upper).unwrap(), hex_lower); + // Leading @ is stripped. + assert_eq!( + resolve_approver_spec(&format!("@{hex_upper}")).unwrap(), + hex_lower + ); + } + + #[test] + fn approver_spec_npub_resolves_to_hex() { + use nostr::prelude::ToBech32; + let keys = nostr::Keys::generate(); + let npub = keys.public_key().to_bech32().expect("bech32 encode"); + let expected = keys.public_key().to_hex(); + assert_eq!(resolve_approver_spec(&npub).unwrap(), expected); + assert_eq!( + resolve_approver_spec(&format!("@{npub}")).unwrap(), + expected + ); + } + + #[test] + fn approver_spec_display_names_fail_closed() { + // Display-name mentions can't be resolved to a pubkey at suspend + // time; an approval minted for them could never be granted + // (check_approver_spec accepts only "any"/hex), so they must error. + assert!(resolve_approver_spec("@release-manager").is_err()); + assert!(resolve_approver_spec("release-manager").is_err()); + // Truncated hex and malformed npub fail too. + assert!(resolve_approver_spec(&"ab".repeat(16)).is_err()); + assert!(resolve_approver_spec("npub1notarealkey").is_err()); + } } diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index 93581225ee..f90645f643 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -188,31 +188,15 @@ impl WorkflowEngine { let trace_json = serde_json::Value::Array(full_trace); let step_count = result.step_index as i32; - if result.approval_token.is_some() { - // Approval gates are not yet implemented (WF-08). - // Fail explicitly rather than creating unreachable WaitingApproval rows. - tracing::warn!( - run_id = %run_id, - step_index = result.step_index, - "Workflow hit approval gate — not yet implemented, marking as failed" - ); - if let Err(e) = self - .db - .update_workflow_run( - community_id, - run_id, - RunStatus::Failed, - step_count, - &trace_json, - Some("approval gates not yet implemented — see WF-08"), - ) - .await - { - tracing::error!( - run_id = %run_id, - "Failed to update run to Failed (approval gate): {e}" - ); - } + if let Some(pending) = result.approval { + self.suspend_run_for_approval( + community_id, + run_id, + pending, + step_count, + &trace_json, + ) + .await; } else { tracing::info!(run_id = %run_id, "Workflow run completed"); if let Err(e) = self @@ -260,6 +244,158 @@ impl WorkflowEngine { } } + /// Persist a `RequestApproval` suspension (WF-08): mint the approval row, + /// move the run to `WaitingApproval`, and announce the pending approval as + /// a kind:46010 event. + /// + /// Ordering matters: approval row first, then run status, then the + /// announcement. A grant can only act on a run that is both + /// `WaitingApproval` and has a pending row, so persisting in this order + /// never exposes a half-armed gate. Any persistence failure marks the run + /// Failed — a run must never sit in `WaitingApproval` without a grantable + /// approval row. The announcement itself is best-effort: by then the gate + /// is fully armed and grantable through any surface that knows the token, + /// so an emit failure is logged loudly but does not fail the run. + async fn suspend_run_for_approval( + &self, + community_id: CommunityId, + run_id: uuid::Uuid, + pending: executor::PendingApproval, + step_count: i32, + trace_json: &serde_json::Value, + ) { + match self + .arm_approval_gate(community_id, run_id, &pending, step_count, trace_json) + .await + { + Ok(()) => { + tracing::info!( + run_id = %run_id, + step_id = %pending.step_id, + "Workflow run waiting for approval" + ); + } + Err(reason) => { + tracing::error!( + run_id = %run_id, + step_id = %pending.step_id, + "Failed to arm approval gate — marking run failed: {reason}" + ); + if let Err(e) = self + .db + .update_workflow_run( + community_id, + run_id, + RunStatus::Failed, + step_count, + trace_json, + Some(&format!("approval gate: {reason}")), + ) + .await + { + tracing::error!( + run_id = %run_id, + "Failed to update run to Failed (approval gate): {e}" + ); + } + } + } + } + + /// Fallible core of [`suspend_run_for_approval`]. + async fn arm_approval_gate( + &self, + community_id: CommunityId, + run_id: uuid::Uuid, + pending: &executor::PendingApproval, + step_count: i32, + trace_json: &serde_json::Value, + ) -> Result<(), String> { + let run = self + .db + .get_workflow_run(community_id, run_id) + .await + .map_err(|e| format!("load run: {e}"))?; + let workflow = self + .db + .get_workflow(community_id, run.workflow_id) + .await + .map_err(|e| format!("load workflow {}: {e}", run.workflow_id))?; + + let expires_at = Utc::now() + + chrono::Duration::seconds(i64::try_from(pending.timeout_secs).unwrap_or(i64::MAX)); + + self.db + .create_approval(buzz_db::workflow::CreateApprovalParams { + community_id, + token: &pending.token, + workflow_id: run.workflow_id, + run_id, + step_id: &pending.step_id, + step_index: step_count, + approver_spec: &pending.approver_spec, + expires_at, + }) + .await + .map_err(|e| format!("create approval record: {e}"))?; + + self.db + .update_workflow_run( + community_id, + run_id, + RunStatus::WaitingApproval, + step_count, + trace_json, + None, + ) + .await + .map_err(|e| format!("set run WaitingApproval: {e}"))?; + + // Announce the pending approval (kind:46010). The gate is armed at + // this point, so emit failures must not fail the run — but they do + // hide the token from feeds, so log at ERROR for operator visibility. + let owner_pubkey_hex = hex::encode(&workflow.owner_pubkey); + let notify_pubkey_hex = if pending.approver_spec == "any" { + owner_pubkey_hex + } else { + pending.approver_spec.clone() + }; + match self.action_sink() { + Ok(sink) => { + if let Err(e) = sink + .emit_approval_requested( + community_id, + action_sink::ApprovalRequestNotice { + channel_id: workflow.channel_id, + raw_token: pending.token.clone(), + message: pending.message.clone(), + notify_pubkey_hex, + workflow_id: run.workflow_id, + run_id, + step_id: pending.step_id.clone(), + expires_at, + }, + ) + .await + { + tracing::error!( + run_id = %run_id, + step_id = %pending.step_id, + "Failed to announce pending approval (kind:46010): {e}" + ); + } + } + Err(e) => { + tracing::error!( + run_id = %run_id, + "No action sink — cannot announce pending approval: {e}" + ); + } + } + + Ok(()) + } + /// Called from the event handler post-store hook for every stored event. /// /// Checks whether any workflow in the event's channel has a matching trigger.