From 85d67b93b4bf4b0c715609a908eb4b7a8f7d323b Mon Sep 17 00:00:00 2001 From: nocodedltd <208452262+nocodedltd@users.noreply.github.com> Date: Thu, 6 Aug 2026 06:00:34 +0100 Subject: [PATCH 1/4] Harden workflow approvals and ACP publishing Signed-off-by: nocodedltd <208452262+nocodedltd@users.noreply.github.com> --- Cargo.lock | 1 + crates/buzz-acp/src/acp.rs | 93 +++ crates/buzz-acp/src/config.rs | 11 + crates/buzz-acp/src/lib.rs | 3 + crates/buzz-acp/src/pool.rs | 252 ++++++ crates/buzz-db/src/lib.rs | 71 ++ crates/buzz-db/src/migration.rs | 11 +- crates/buzz-db/src/workflow.rs | 328 +++++++- crates/buzz-relay/src/api/bridge.rs | 171 ++++ .../src/handlers/command_executor.rs | 170 +++- crates/buzz-relay/src/main.rs | 27 + crates/buzz-relay/src/router.rs | 12 + crates/buzz-relay/src/workflow_sink.rs | 115 +++ crates/buzz-workflow/Cargo.toml | 1 + .../examples/validate_workflow.rs | 81 ++ crates/buzz-workflow/src/action_sink.rs | 40 + crates/buzz-workflow/src/executor.rs | 420 +++++++++- crates/buzz-workflow/src/lib.rs | 738 +++++++++++++++++- crates/buzz-workflow/src/schema.rs | 79 +- desktop/src-tauri/src/commands/workflows.rs | 95 ++- .../src-tauri/src/commands/workflows_tests.rs | 121 +++ desktop/src-tauri/src/events.rs | 12 +- desktop/src-tauri/src/relay.rs | 19 + .../workflows/ui/WorkflowApprovalCard.tsx | 28 +- .../workflows/ui/WorkflowStepCard.tsx | 7 +- desktop/src/shared/api/tauriWorkflows.ts | 2 + .../api/workflowApprovalContract.test.mjs | 95 +++ desktop/src/shared/api/workflowTypes.ts | 7 + ...oval_request_message_and_step_dispatch.sql | 56 ++ schema/schema.sql | 16 + 30 files changed, 2980 insertions(+), 102 deletions(-) create mode 100644 crates/buzz-workflow/examples/validate_workflow.rs create mode 100644 desktop/src/shared/api/workflowApprovalContract.test.mjs create mode 100644 migrations/0029_approval_request_message_and_step_dispatch.sql diff --git a/Cargo.lock b/Cargo.lock index 73ecb249d4..ccb562de9c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1359,6 +1359,7 @@ dependencies = [ "serde", "serde_json", "serde_yaml", + "sha2 0.11.0", "thiserror 2.0.18", "tokio", "tracing", diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 93109fa94d..2ff9000ba8 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -211,6 +211,16 @@ pub struct AcpClient { /// deltas. Both goose and buzz-agent emit this notification; goose gates /// on client capability advertisement, buzz-agent emits unconditionally. goose_usage: UsageTracker, + /// Text emitted in `agent_message_chunk` updates for the current assistant + /// message. + /// + /// The buffer is reset immediately before every `session/prompt` and when + /// a tool call starts. ACP agents may emit narration before each tool call; + /// clearing at that boundary ensures the authenticated publisher returns + /// only the last assistant answer, not every narration fragment from the + /// entire tool loop. Keeping the buffer in the ACP client means adapters do + /// not need a separate reporting tool merely to return their final answer. + agent_message: String, } /// Recursively merge `overlay` into `base`, with `overlay` winning on scalar/shape @@ -550,9 +560,15 @@ impl AcpClient { steering_supported: false, steer_rx: None, goose_usage: UsageTracker::default(), + agent_message: String::new(), }) } + /// Consume the text emitted by the agent during the most recent prompt. + pub(crate) fn take_agent_message(&mut self) -> String { + std::mem::take(&mut self.agent_message) + } + /// Attach a local observer feed to this ACP client. pub fn set_observer(&mut self, observer: Option, agent_index: usize) { self.observer = observer; @@ -768,6 +784,10 @@ impl AcpClient { idle_timeout: std::time::Duration, max_duration: std::time::Duration, ) -> Result { + // Each prompt owns exactly one response buffer. This also prevents a + // new-session `initial_message` response from leaking into the first + // real channel turn. + self.agent_message.clear(); let params = build_prompt_params(session_id, prompt_blocks); let hard_deadline = tokio::time::Instant::now() + max_duration; self.current_hard_deadline = Some(hard_deadline); @@ -1733,10 +1753,15 @@ impl AcpClient { "agent_message_chunk" => { if let Some(text) = update["content"]["text"].as_str() { tracing::info!(target: "acp::stream", "{text}"); + self.agent_message.push_str(text); } false } "tool_call" => { + // Text before a tool call is intermediate narration. A final + // answer can only be the assistant text emitted after the last + // tool call in the turn, so discard the earlier fragment. + self.agent_message.clear(); let title = update .get("title") .and_then(|v| v.as_str()) @@ -3591,6 +3616,74 @@ mod tests { .expect("spawn cat as inert client") } + #[tokio::test] + async fn agent_message_chunks_are_collected_and_consumed() { + let mut client = spawn_inert_client().await; + for text in ["BUZZ_REENGAGEMENT_", "DRAFT_READY"] { + let update = serde_json::json!({ + "jsonrpc": "2.0", + "method": "session/update", + "params": { + "sessionId": "test-session", + "update": { + "sessionUpdate": "agent_message_chunk", + "content": {"text": text} + } + } + }); + let _ = client.handle_session_update(&update); + } + + assert_eq!(client.take_agent_message(), "BUZZ_REENGAGEMENT_DRAFT_READY"); + assert!(client.take_agent_message().is_empty()); + } + + #[tokio::test] + async fn agent_message_buffer_keeps_only_text_after_last_tool_call() { + let mut client = spawn_inert_client().await; + let narration = serde_json::json!({ + "jsonrpc": "2.0", + "method": "session/update", + "params": { + "sessionId": "test-session", + "update": { + "sessionUpdate": "agent_message_chunk", + "content": {"text": "I will research this now."} + } + } + }); + let tool_call = serde_json::json!({ + "jsonrpc": "2.0", + "method": "session/update", + "params": { + "sessionId": "test-session", + "update": { + "sessionUpdate": "tool_call", + "toolCallId": "tool-1", + "title": "Read CRM", + "kind": "read" + } + } + }); + let final_answer = serde_json::json!({ + "jsonrpc": "2.0", + "method": "session/update", + "params": { + "sessionId": "test-session", + "update": { + "sessionUpdate": "agent_message_chunk", + "content": {"text": "BUZZ_REENGAGEMENT_DRAFT_READY"} + } + } + }); + + let _ = client.handle_session_update(&narration); + let _ = client.handle_session_update(&tool_call); + let _ = client.handle_session_update(&final_answer); + + assert_eq!(client.take_agent_message(), "BUZZ_REENGAGEMENT_DRAFT_READY"); + } + /// Build a `session/update` JSON-RPC notification carrying a /// `session_info_update` with the given `_meta.goose.activeRunId` value. /// Pass `None` to omit the `activeRunId` field entirely. diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index d959685846..b3ecc51811 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -376,6 +376,12 @@ pub struct CliArgs { #[arg(long, env = "BUZZ_ACP_NO_TYPING")] pub no_typing: bool, + /// Publish the final ACP agent message as a signed reply to the triggering + /// Buzz event. Intended for managed agents whose only output surface is + /// their authenticated harness. + #[arg(long, env = "BUZZ_ACP_PUBLISH_FINAL_RESPONSE", default_value_t = false)] + pub publish_final_response: bool, + /// Enable NIP-AE agent core memory injection. /// /// Memory injection is on by default. When enabled, the harness @@ -520,6 +526,9 @@ pub struct Config { pub max_turns_per_session: u32, pub presence_enabled: bool, pub typing_enabled: bool, + /// Whether successful channel turns publish their final ACP response as a + /// signed reply to the triggering event. + pub publish_final_response: bool, /// Whether NIP-AE agent core memory injection is enabled. When false, /// the harness skips the per-session core engram fetch and renders no /// `[Agent Memory — core]` section. On by default; disabled via the @@ -1086,6 +1095,7 @@ impl Config { max_turns_per_session: args.max_turns_per_session, presence_enabled: !args.no_presence, typing_enabled: !args.no_typing, + publish_final_response: args.publish_final_response, memory_enabled: args.memory && !args.no_memory, model, session_title: args @@ -1460,6 +1470,7 @@ mod tests { max_turns_per_session: 0, presence_enabled: true, typing_enabled: true, + publish_final_response: false, memory_enabled: true, model: None, session_title: None, diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 65c9dd6203..eaa2c1b6df 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -1835,6 +1835,7 @@ async fn tokio_main() -> Result<()> { channel_info: pool::ChannelInfoResolver::new(channel_info_map, relay.rest_client()), context_message_limit: config.context_message_limit, max_turns_per_session: config.max_turns_per_session, + publish_final_response: config.publish_final_response, permission_mode: config.permission_mode, agent_keys: config.keys.clone(), agent_owner_pubkey: startup_owner @@ -6196,6 +6197,7 @@ mod build_mcp_servers_tests { max_turns_per_session: 0, presence_enabled: true, typing_enabled: true, + publish_final_response: false, memory_enabled: false, model: None, session_title: None, @@ -6418,6 +6420,7 @@ mod error_outcome_emission_tests { max_turns_per_session: 0, presence_enabled: true, typing_enabled: true, + publish_final_response: false, memory_enabled: false, model: None, session_title: None, diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 8430307d9c..d79e6353bb 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -543,6 +543,9 @@ pub struct PromptContext { pub context_message_limit: u32, /// Max turns per session before proactive rotation. 0 = disabled. pub max_turns_per_session: u32, + /// Publish a successful channel turn's final ACP response as an + /// authenticated reply to the triggering Buzz event. + pub publish_final_response: bool, /// Permission mode to apply after session creation. `Default` = skip. pub permission_mode: PermissionMode, /// Agent identity — used to derive the NIP-AE conversation key at @@ -2139,6 +2142,59 @@ pub async fn run_prompt_task( Ok(stop_reason) => { log_stop_reason(&source, &stop_reason); + let final_response = agent.acp.take_agent_message(); + if ctx.publish_final_response { + if let (PromptSource::Channel(channel_id), Some(channel_batch)) = + (&source, batch.as_ref()) + { + match post_final_agent_response( + &ctx.rest_client, + channel_batch, + &final_response, + ) + .await + { + Ok(event_id) => tracing::info!( + target: "pool::prompt", + channel = %channel_id, + event_id = %event_id, + "published authenticated final agent response" + ), + Err(error) => { + tracing::error!( + target: "pool::prompt", + channel = %channel_id, + "failed to publish authenticated final agent response: {error}" + ); + let usage = agent.acp.take_turn_usage(); + publish_agent_turn_metric( + &ctx, + usage, + observer_channel_id, + &session_id, + &turn_id, + Some(buzz_core::agent_turn_metric::StopReason::Error), + ) + .await; + send_prompt_result( + &result_tx, + &turn_id, + agent, + source, + PromptOutcome::Error(AcpError::AgentError { + code: -32000, + message: format!( + "authenticated final-response publish failed: {error}" + ), + }), + None, + ); + return; + } + } + } + } + let should_rotate = matches!( stop_reason, StopReason::MaxTokens | StopReason::MaxTurnRequests @@ -3909,6 +3965,90 @@ pub(crate) async fn post_failure_notice( } } +/// Publish an ACP adapter's final message as the authenticated agent's reply +/// to the event that triggered the turn. +/// +/// The signed event is constructed once and `RestClient` retries that exact +/// body, so an ambiguous HTTP retry cannot create a second Nostr event ID. +/// A relay rejection is only treated as success when querying the same event +/// ID confirms that the first attempt was already persisted. +async fn post_final_agent_response( + rest: &crate::relay::RestClient, + batch: &FlushBatch, + content: &str, +) -> Result { + const MAX_FINAL_RESPONSE_BYTES: usize = 20_000; + + if content.trim().is_empty() { + return Err("agent returned an empty final response".to_string()); + } + if content.len() > MAX_FINAL_RESPONSE_BYTES { + return Err(format!( + "agent final response is {} bytes; maximum is {MAX_FINAL_RESPONSE_BYTES}", + content.len() + )); + } + + let trigger = batch + .events + .last() + .ok_or_else(|| "channel turn has no triggering event".to_string())?; + let trigger_id = trigger.event.id; + let thread_tags = crate::queue::parse_thread_tags(&trigger.event); + let root_id = thread_tags + .root_event_id + .as_deref() + .and_then(|value| nostr::EventId::from_hex(value).ok()) + .unwrap_or(trigger_id); + let thread_ref = buzz_sdk::ThreadRef { + root_event_id: root_id, + parent_event_id: trigger_id, + }; + let event = buzz_sdk::build_message( + batch.channel_id, + content, + Some(&thread_ref), + &[], + false, + &[], + ) + .map_err(|error| format!("build failed: {error}"))? + .sign_with_keys(&rest.keys) + .map_err(|error| format!("sign failed: {error}"))?; + let event_id = event.id; + + let response = tokio::time::timeout(Duration::from_secs(15), rest.submit_event(&event)) + .await + .map_err(|_| "publish timed out".to_string())? + .map_err(|error| format!("publish failed: {error}"))?; + if response + .get("accepted") + .and_then(serde_json::Value::as_bool) + .unwrap_or(false) + { + return Ok(event_id); + } + + let persisted = tokio::time::timeout( + Duration::from_secs(5), + rest.query(&[nostr::Filter::new().id(event_id)]), + ) + .await + .ok() + .and_then(Result::ok) + .and_then(|value| value.as_array().map(|events| !events.is_empty())) + .unwrap_or(false); + if persisted { + Ok(event_id) + } else { + let message = response + .get("message") + .and_then(serde_json::Value::as_str) + .unwrap_or("relay rejected event"); + Err(format!("relay rejected final response: {message}")) + } +} + /// Best-effort: remove a reaction via a signed kind:5 (NIP-09) deletion event. /// /// Queries kind:7 reactions by our pubkey targeting the event, finds the matching @@ -6536,6 +6676,7 @@ mod tests { ), context_message_limit: 0, max_turns_per_session: 0, + publish_final_response: false, permission_mode: PermissionMode::Default, agent_keys: agent_keys.clone(), agent_owner_pubkey: owner_pubkey, @@ -6902,6 +7043,117 @@ mod tests { ) } + #[tokio::test] + async fn final_agent_response_is_signed_and_replies_to_trigger() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind test HTTP server"); + let base_url = format!("http://{}", listener.local_addr().expect("local addr")); + let (event_tx, event_rx) = tokio::sync::oneshot::channel(); + let server = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.expect("accept request"); + let mut request = Vec::new(); + let mut buf = [0_u8; 4096]; + loop { + let read = socket.read(&mut buf).await.expect("read request"); + if read == 0 { + break; + } + request.extend_from_slice(&buf[..read]); + if let Some(header_end) = request.windows(4).position(|w| w == b"\r\n\r\n") { + let headers = String::from_utf8_lossy(&request[..header_end]); + let content_length = headers + .lines() + .find_map(|line| { + let (name, value) = line.split_once(':')?; + name.eq_ignore_ascii_case("content-length") + .then(|| value.trim().parse::().ok()) + .flatten() + }) + .unwrap_or(0); + if request.len() >= header_end + 4 + content_length { + let body = &request[header_end + 4..header_end + 4 + content_length]; + let event: nostr::Event = + serde_json::from_slice(body).expect("signed event JSON"); + let _ = event_tx.send(event); + break; + } + } + } + let body = r#"{"accepted":true,"message":"saved"}"#; + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + body.len(), + body + ); + socket + .write_all(response.as_bytes()) + .await + .expect("write response"); + }); + + let agent_keys = nostr::Keys::generate(); + let author_keys = nostr::Keys::generate(); + let channel_id = Uuid::new_v4(); + let trigger = EventBuilder::new(Kind::Custom(9), "@Hermes draft this") + .tags([ + Tag::parse(["h", &channel_id.to_string()]).expect("h tag"), + Tag::parse(["p", &agent_keys.public_key().to_hex()]).expect("p tag"), + ]) + .sign_with_keys(&author_keys) + .expect("sign trigger"); + let trigger_id = trigger.id; + let batch = FlushBatch { + channel_id, + events: vec![crate::queue::BatchEvent { + event: trigger, + prompt_tag: "test".to_string(), + received_at: std::time::Instant::now(), + }], + cancelled_events: vec![], + cancel_reason: None, + }; + let rest = crate::relay::RestClient { + http: reqwest::Client::new(), + base_url, + keys: agent_keys.clone(), + auth_tag_json: None, + }; + + let published_id = post_final_agent_response( + &rest, + &batch, + "BUZZ_REENGAGEMENT_DRAFT_READY\nDRAFT_START\nHi John\nDRAFT_END", + ) + .await + .expect("publish final response"); + let published = event_rx.await.expect("captured event"); + server.await.expect("server task"); + + assert_eq!(published.id, published_id); + assert_eq!(published.pubkey, agent_keys.public_key()); + assert!(published.verify().is_ok()); + assert_eq!( + published.content, + "BUZZ_REENGAGEMENT_DRAFT_READY\nDRAFT_START\nHi John\nDRAFT_END" + ); + let tags: Vec> = published + .tags + .iter() + .map(|tag| tag.clone().to_vec()) + .collect(); + assert!(tags + .iter() + .any(|tag| tag == &["h", &channel_id.to_string()])); + assert!( + tags.iter() + .any(|tag| tag == &["e", &trigger_id.to_hex(), "", "reply"]), + "tags: {tags:?}" + ); + } + fn channel_metadata_response(id: Uuid, tags: &[[&str; 2]]) -> serde_json::Value { let mut event_tags = vec![json!(["d", id.to_string()])]; event_tags.extend(tags.iter().map(|[k, v]| json!([k, v]))); diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 9b26876747..3e7b6c6464 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -3843,6 +3843,77 @@ impl Db { workflow::get_approval_by_stored_hash(&self.pool, community_id, token_hash).await } + /// List pending approvals past their deadline, across all communities. + pub async fn list_expired_pending_approvals(&self) -> Result> { + workflow::list_expired_pending_approvals(&self.pool).await + } + + /// Update an approval's status inside an existing transaction. + pub async fn update_approval_by_stored_hash_tx( + &self, + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + community_id: CommunityId, + token_hash: &[u8], + status: workflow::ApprovalStatus, + approver_pubkey: Option<&[u8]>, + note: Option<&str>, + ) -> Result { + workflow::update_approval_by_stored_hash_tx( + tx, + community_id, + token_hash, + status, + approver_pubkey, + note, + ) + .await + } + + /// Claim a step for at-most-once dispatch. + pub async fn claim_step_dispatch( + &self, + community_id: CommunityId, + run_id: uuid::Uuid, + step_id: &str, + ) -> Result { + workflow::claim_step_dispatch(&self.pool, community_id, run_id, step_id).await + } + + /// Record the event a claimed step produced. + pub async fn complete_step_dispatch( + &self, + community_id: CommunityId, + run_id: uuid::Uuid, + step_id: &str, + event_id: &[u8], + ) -> Result<()> { + workflow::complete_step_dispatch(&self.pool, community_id, run_id, step_id, event_id).await + } + + /// Release a claim whose side effect provably did not happen. + pub async fn release_step_dispatch( + &self, + community_id: CommunityId, + run_id: uuid::Uuid, + step_id: &str, + ) -> Result<()> { + workflow::release_step_dispatch(&self.pool, community_id, run_id, step_id).await + } + + /// Atomically claim a waiting run for resumption. True for exactly one caller. + pub async fn claim_run_for_resume( + &self, + community_id: CommunityId, + run_id: uuid::Uuid, + ) -> Result { + workflow::claim_run_for_resume(&self.pool, community_id, run_id).await + } + + /// Granted approvals whose run never resumed (crash-window recovery). + pub async fn list_granted_but_waiting_runs(&self) -> Result> { + workflow::list_granted_but_waiting_runs(&self.pool).await + } + /// Fetch all approvals for a workflow run. pub async fn get_run_approvals( &self, diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index 37f54d0fa2..5680248480 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -561,7 +561,7 @@ mod tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 28); + assert_eq!(migrations.len(), 29); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] @@ -946,6 +946,15 @@ mod tests { long_reactions.contains("ALTER TABLE reactions ALTER COLUMN emoji TYPE VARCHAR(66)") ); assert!(desired_schema.contains("emoji VARCHAR(66) NOT NULL")); + + // Workflow approval package persistence and the exactly-once dispatch + // journal are additive so existing installations receive them without + // changing the consolidated schema checksum. + assert_eq!(migrations[28].version, 29); + let workflow_approval_safety = migrations[28].sql.as_str(); + assert!(workflow_approval_safety.contains("ADD COLUMN IF NOT EXISTS request_message")); + assert!(workflow_approval_safety + .contains("CREATE TABLE IF NOT EXISTS workflow_step_dispatches")); } #[test] diff --git a/crates/buzz-db/src/workflow.rs b/crates/buzz-db/src/workflow.rs index 7a2396c1fd..b9ed25c18c 100644 --- a/crates/buzz-db/src/workflow.rs +++ b/crates/buzz-db/src/workflow.rs @@ -258,8 +258,12 @@ pub struct ApprovalRecord { pub status: ApprovalStatus, /// Compressed public key bytes of the user who acted on this approval. pub approver_pubkey: Option>, - /// Optional note left by the approver. + /// Optional note left by the approver. Distinct from `request_message`: + /// this travels back from the approver, that is what they were shown. pub note: Option, + /// The prompt the approver was shown. `None` for rows created before + /// migration 0027 — render that as "not recorded", never as empty. + pub request_message: Option, /// When this approval request expires. pub expires_at: DateTime, /// When the approval record was created. @@ -937,6 +941,10 @@ pub struct CreateApprovalParams<'a> { pub step_index: i32, /// Who may approve (user mention or role spec). pub approver_spec: &'a str, + /// The rendered prompt shown to the approver — what is actually being + /// approved. Persisted so the approval card can display it; without it a + /// grant is a decision made blind. + pub request_message: &'a str, /// When this approval request expires. pub expires_at: DateTime, } @@ -954,6 +962,7 @@ pub async fn create_approval(pool: &PgPool, params: CreateApprovalParams<'_>) -> step_id, step_index, approver_spec, + request_message, expires_at, } = params; let token_hash = hash_approval_token(token); @@ -961,8 +970,8 @@ pub async fn create_approval(pool: &PgPool, params: CreateApprovalParams<'_>) -> sqlx::query( r#" INSERT INTO workflow_approvals - (community_id, token, workflow_id, run_id, step_id, step_index, approver_spec, status, expires_at) - VALUES ($1, $2, $3, $4, $5, $6, $7, 'pending', $8) + (community_id, token, workflow_id, run_id, step_id, step_index, approver_spec, request_message, status, expires_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, 'pending', $9) "#, ) .bind(community_id.as_uuid()) @@ -972,6 +981,7 @@ pub async fn create_approval(pool: &PgPool, params: CreateApprovalParams<'_>) -> .bind(step_id) .bind(step_index) .bind(approver_spec) + .bind(request_message) .bind(expires_at) .execute(pool) .await?; @@ -1008,7 +1018,8 @@ pub async fn get_approval_by_stored_hash( let row = sqlx::query( r#" SELECT token, workflow_id, run_id, step_id, step_index, approver_spec, - status::text AS status, approver_pubkey, note, expires_at, created_at + status::text AS status, approver_pubkey, note, request_message, + expires_at, created_at FROM workflow_approvals WHERE community_id = $1 AND token = $2 "#, @@ -1022,6 +1033,50 @@ pub async fn get_approval_by_stored_hash( row_to_approval_record(row) } +/// An expired approval gate awaiting the fail-closed sweep. +#[derive(Debug, Clone)] +pub struct ExpiredApproval { + /// Community owning the approval — the sweep is cross-tenant, so every + /// follow-up write must be scoped back to this value. + pub community_id: CommunityId, + /// Stored (hashed) token, ready for `update_approval_by_stored_hash`. + pub token_hash: Vec, + /// The run to fail closed. + pub run_id: Uuid, +} + +/// List pending approvals whose deadline has passed, across all communities. +/// +/// Deliberately cross-tenant: the cron loop that calls this is global, matching +/// `list_all_enabled_workflows`. Each row carries its own `community_id` so the +/// caller never has to infer a tenant. +/// +/// Bounded to keep one slow tick from stalling the loop; anything left over is +/// picked up on the next pass. +pub async fn list_expired_pending_approvals(pool: &PgPool) -> Result> { + let rows = sqlx::query( + r#" + SELECT community_id, token, run_id + FROM workflow_approvals + WHERE status = 'pending' AND expires_at < NOW() + ORDER BY expires_at + LIMIT 500 + "#, + ) + .fetch_all(pool) + .await?; + + rows.into_iter() + .map(|row| { + Ok(ExpiredApproval { + community_id: CommunityId::from_uuid(row.try_get("community_id")?), + token_hash: row.try_get("token")?, + run_id: row.try_get("run_id")?, + }) + }) + .collect() +} + /// Fetch all approval records for a given workflow run. pub async fn get_run_approvals( pool: &PgPool, @@ -1032,7 +1087,8 @@ pub async fn get_run_approvals( let rows = sqlx::query( r#" SELECT token, workflow_id, run_id, step_id, step_index, approver_spec, - status::text AS status, approver_pubkey, note, expires_at, created_at + status::text AS status, approver_pubkey, note, request_message, + expires_at, created_at FROM workflow_approvals WHERE community_id = $1 AND run_id = $2 AND workflow_id = $3 ORDER BY step_index, created_at @@ -1120,6 +1176,261 @@ pub async fn update_approval_by_stored_hash( Ok(affected > 0) } +/// Transaction-scoped variant of [`update_approval_by_stored_hash`]. +/// +/// The approval decision and the command event that authorised it must land in +/// the same transaction. Running the decision on the pool while the event sits +/// in an uncommitted transaction leaves a crash window where an approval reads +/// `granted` with no corresponding kind:46030 event on record — an unattributable +/// state change on a gate whose whole purpose is attribution. +/// +/// The `status = 'pending'` guard is what makes this idempotent: a replayed or +/// concurrent decision affects zero rows and the caller treats that as "already +/// acted on" rather than applying a second decision. +pub async fn update_approval_by_stored_hash_tx( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + community_id: CommunityId, + token_hash: &[u8], + status: ApprovalStatus, + approver_pubkey: Option<&[u8]>, + note: Option<&str>, +) -> Result { + let status_str = status.to_string(); + let affected = sqlx::query( + r#" + UPDATE workflow_approvals + SET status = $1::approval_status, + approver_pubkey = $2, + note = $3, + granted_at = CASE WHEN $4 = 'granted' THEN NOW() ELSE granted_at END, + denied_at = CASE WHEN $5 = 'denied' THEN NOW() ELSE denied_at END + WHERE community_id = $6 AND token = $7 AND status = 'pending' + "#, + ) + .bind(&status_str) + .bind(approver_pubkey) + .bind(note) + .bind(&status_str) + .bind(&status_str) + .bind(community_id.as_uuid()) + .bind(token_hash) + .execute(tx.as_mut()) + .await? + .rows_affected(); + + Ok(affected > 0) +} + +/// Outcome of claiming a step for dispatch. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum StepDispatchClaim { + /// This caller won the claim and must perform the side effect. + Claimed, + /// Already dispatched by an earlier attempt; the recorded event id is + /// returned so the caller can reuse it instead of emitting a second event. + AlreadyDispatched(Vec), + /// Claimed by an attempt that never completed — a process died between + /// claiming and recording the event. Deliberately NOT re-claimable: we + /// cannot tell whether the external side effect happened, and re-running it + /// is the one outcome that must never occur silently. + InFlight, +} + +/// Claim `(community_id, run_id, step_id)` for at-most-once dispatch. +/// +/// The claim is decided by Postgres via `INSERT ... ON CONFLICT DO NOTHING` on +/// the primary key, not by a read — so it is safe across concurrent resumers, +/// process restarts and multiple pods. +/// +/// This is what the run-level CAS cannot provide. The CAS makes resumption +/// start once; this makes each individual step's side effect happen once, which +/// matters because a re-executed `send_message` produces a new signed event and +/// therefore a second external instruction. +pub async fn claim_step_dispatch( + pool: &PgPool, + community_id: CommunityId, + run_id: Uuid, + step_id: &str, +) -> Result { + let inserted = sqlx::query( + r#" + INSERT INTO workflow_step_dispatches (community_id, run_id, step_id) + VALUES ($1, $2, $3) + ON CONFLICT (community_id, run_id, step_id) DO NOTHING + "#, + ) + .bind(community_id.as_uuid()) + .bind(run_id) + .bind(step_id) + .execute(pool) + .await? + .rows_affected(); + + if inserted > 0 { + return Ok(StepDispatchClaim::Claimed); + } + + let row = sqlx::query( + "SELECT event_id FROM workflow_step_dispatches \ + WHERE community_id = $1 AND run_id = $2 AND step_id = $3", + ) + .bind(community_id.as_uuid()) + .bind(run_id) + .bind(step_id) + .fetch_optional(pool) + .await?; + + match row { + Some(r) => { + let event_id: Option> = r.try_get("event_id")?; + Ok(match event_id { + Some(id) => StepDispatchClaim::AlreadyDispatched(id), + None => StepDispatchClaim::InFlight, + }) + } + // The claim row disappeared, normally because the parent run was + // deleted concurrently. Sending without a durable journal entry would + // remove the replay protection this function exists to provide, so + // fail closed rather than treating an unrecorded claim as permission. + None => Err(DbError::NotFound(format!( + "workflow step dispatch claim {run_id}/{step_id}" + ))), + } +} + +/// Record the event a claimed step produced, completing the claim. +pub async fn complete_step_dispatch( + pool: &PgPool, + community_id: CommunityId, + run_id: Uuid, + step_id: &str, + event_id: &[u8], +) -> Result<()> { + sqlx::query( + "UPDATE workflow_step_dispatches \ + SET event_id = $4, completed_at = NOW() \ + WHERE community_id = $1 AND run_id = $2 AND step_id = $3", + ) + .bind(community_id.as_uuid()) + .bind(run_id) + .bind(step_id) + .bind(event_id) + .execute(pool) + .await?; + Ok(()) +} + +/// Release a claim whose side effect provably did not happen. +/// +/// Only safe when the dispatch failed *before* anything left the relay. Used on +/// an error return from the sink, so a transient failure does not permanently +/// wedge the step. +pub async fn release_step_dispatch( + pool: &PgPool, + community_id: CommunityId, + run_id: Uuid, + step_id: &str, +) -> Result<()> { + sqlx::query( + "DELETE FROM workflow_step_dispatches \ + WHERE community_id = $1 AND run_id = $2 AND step_id = $3 AND event_id IS NULL", + ) + .bind(community_id.as_uuid()) + .bind(run_id) + .bind(step_id) + .execute(pool) + .await?; + Ok(()) +} + +/// Atomically claim a `waiting_approval` run for resumption. +/// +/// Returns `true` for exactly one caller. This is a compare-and-set, not a +/// read-then-act check: the single `UPDATE ... WHERE status = 'waiting_approval'` +/// takes a row lock, so a concurrent claimer either blocks and then matches zero +/// rows, or matches zero rows immediately. +/// +/// This is required because resumption has two independent triggers — the task +/// spawned by `handle_approval_grant` and the recovery sweep — and a third if +/// more than one relay pod is running. Two claimers both observing +/// `waiting_approval` and both proceeding would execute every post-gate step +/// twice, which for a `send_message` step means a second, differently-signed +/// event and therefore a second external send instruction. +/// +/// NOTE: this makes resumption *start* at most once. It does not by itself make +/// each action at-most-once across a crash **mid-execution** — a process that +/// dies after emitting a step's event but before its trace is persisted will, +/// once the run is recovered, re-emit that step. Closing that gap needs a +/// durable per-step action journal keyed by `(community_id, run_id, step_id)`. +pub async fn claim_run_for_resume( + pool: &PgPool, + community_id: CommunityId, + run_id: Uuid, +) -> Result { + let affected = sqlx::query( + r#" + UPDATE workflow_runs + SET status = 'running' + WHERE community_id = $1 AND id = $2 AND status = 'waiting_approval' + "#, + ) + .bind(community_id.as_uuid()) + .bind(run_id) + .execute(pool) + .await? + .rows_affected(); + + Ok(affected > 0) +} + +/// Runs that were granted but never resumed — the crash-window recovery set. +/// +/// A grant commits the decision, then resumption is spawned separately. If the +/// process dies in between, the run sits in `waiting_approval` forever against a +/// `granted` approval. This query is the durable backstop: it is derived purely +/// from committed state, so recovery does not depend on any in-memory task +/// having survived. +pub async fn list_granted_but_waiting_runs(pool: &PgPool) -> Result> { + let rows = sqlx::query( + r#" + SELECT a.community_id, a.run_id, a.workflow_id, a.step_index + FROM workflow_approvals a + JOIN workflow_runs r + ON r.community_id = a.community_id AND r.id = a.run_id + WHERE a.status = 'granted' + AND r.status = 'waiting_approval' + ORDER BY a.granted_at + LIMIT 100 + "#, + ) + .fetch_all(pool) + .await?; + + rows.into_iter() + .map(|row| { + Ok(GrantedWaitingRun { + community_id: CommunityId::from_uuid(row.try_get("community_id")?), + run_id: row.try_get("run_id")?, + workflow_id: row.try_get("workflow_id")?, + step_index: row.try_get("step_index")?, + }) + }) + .collect() +} + +/// A granted approval whose run never resumed. See [`list_granted_but_waiting_runs`]. +#[derive(Debug, Clone)] +pub struct GrantedWaitingRun { + /// Community owning the run. + pub community_id: CommunityId, + /// The run to resume. + pub run_id: Uuid, + /// The workflow definition to resume against. + pub workflow_id: Uuid, + /// Index of the gate step; resumption starts at `step_index + 1`. + pub step_index: i32, +} + // -- Row mappers -------------------------------------------------------------- fn row_to_workflow_record(row: sqlx::postgres::PgRow) -> Result { @@ -1189,6 +1500,7 @@ fn row_to_approval_record(row: sqlx::postgres::PgRow) -> Result status, approver_pubkey: row.try_get("approver_pubkey")?, note: row.try_get("note")?, + request_message: row.try_get("request_message")?, expires_at: row.try_get("expires_at")?, created_at: row.try_get("created_at")?, }) @@ -1605,6 +1917,7 @@ mod tests { status: ApprovalStatus::Pending, approver_pubkey: None, note: None, + request_message: Some("Review the outreach package".to_owned()), expires_at, created_at: now, }; @@ -1635,6 +1948,7 @@ mod tests { status: ApprovalStatus::Granted, approver_pubkey: Some(approver_pubkey.clone()), note: Some("Looks good, approved.".to_owned()), + request_message: Some("Review the outreach package".to_owned()), expires_at: now, created_at: now, }; @@ -1658,6 +1972,7 @@ mod tests { status: ApprovalStatus::Denied, approver_pubkey: Some(vec![0xbb; 32]), note: Some("Not ready for production.".to_owned()), + request_message: Some("Review the outreach package".to_owned()), expires_at: now, created_at: now, }; @@ -1679,6 +1994,7 @@ mod tests { status: ApprovalStatus::Pending, approver_pubkey: None, note: None, + request_message: None, expires_at: now, created_at: now, }; @@ -2244,6 +2560,7 @@ mod tests { step_id: "gate", step_index: 0, approver_spec: "@anyone", + request_message: "Review A", expires_at: expires, }, ) @@ -2259,6 +2576,7 @@ mod tests { step_id: "gate", step_index: 0, approver_spec: "@anyone", + request_message: "Review B", expires_at: expires, }, ) diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index a118ff453f..2427fbd1b5 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -15,6 +15,7 @@ use serde_json::Value; use buzz_auth::{LimitType, Nip98ReplayGuard, DEFAULT_REPLAY_TTL_SECS}; use buzz_core::TenantContext; +use uuid::Uuid; use crate::handlers::ingest::{IngestAuth, IngestError}; use crate::state::AppState; @@ -2128,6 +2129,176 @@ fn clamp_limit(requested: Option) -> i64 { .unwrap_or(MODERATION_READ_LIMIT) } +/// Optional `?limit=` for workflow run reads. +#[derive(serde::Deserialize, Default)] +pub struct WorkflowRunsQuery { + limit: Option, +} + +/// Authorize a workflow read: NIP-98, then membership of the workflow's channel. +/// +/// Runs and approvals are not Nostr events, so they carry no `h` tag for the +/// normal channel-scoped read path to key off. Membership of the owning channel +/// is the equivalent boundary: if you can read the room, you can read what the +/// room's workflows did. A workflow with no channel is refused rather than +/// treated as public. +async fn authorize_workflow_read( + state: &Arc, + headers: &HeaderMap, + path: &str, + raw_query: Option<&str>, + workflow_id: Uuid, +) -> Result)> { + let raw_host = headers + .get(axum::http::header::HOST) + .and_then(|v| v.to_str().ok()) + .unwrap_or(""); + let tenant = crate::tenant::bind_community(&state.db, raw_host) + .await + .map_err(|_| { + api_error( + StatusCode::NOT_FOUND, + "relay: no community is configured for this host", + ) + })?; + + let path_with_query = match raw_query { + Some(q) if !q.is_empty() => format!("{path}?{q}"), + _ => path.to_string(), + }; + let url = nip98_expected_url(&state.config.relay_url, &tenant, &path_with_query); + let (pubkey, event_id_bytes) = + verify_bridge_auth(headers, "GET", &url, None, state.config.require_auth_token)?; + check_nip98_replay(state, &tenant, event_id_bytes).await?; + let pubkey_bytes = pubkey.to_bytes().to_vec(); + + let workflow = state + .db + .get_workflow(tenant.community(), workflow_id) + .await + .map_err(|_| api_error(StatusCode::NOT_FOUND, "workflow not found"))?; + + let channel_id = workflow.channel_id.ok_or_else(|| { + api_error( + StatusCode::FORBIDDEN, + "restricted: workflow is not bound to a channel", + ) + })?; + + let is_member = state + .is_member_cached(tenant.community(), channel_id, &pubkey_bytes) + .await + .map_err(|e| internal_error(&format!("membership check: {e}")))?; + + if !is_member { + return Err(api_error( + StatusCode::FORBIDDEN, + "restricted: not a member of the workflow's channel", + )); + } + + Ok(tenant) +} + +fn run_json(r: &buzz_db::workflow::WorkflowRunRecord) -> Value { + serde_json::json!({ + "id": r.id, + "workflow_id": r.workflow_id, + "status": r.status.to_string(), + "current_step": r.current_step, + "execution_trace": r.execution_trace, + "started_at": r.started_at.map(|t| t.timestamp()), + "completed_at": r.completed_at.map(|t| t.timestamp()), + "error_message": r.error_message, + "created_at": r.created_at.timestamp(), + }) +} + +/// Serialize an approval for the desktop. +/// +/// `token` is the **hashed** token, hex-encoded — deliberately not the raw +/// value. It is exactly what a kind:46030/46031 grant puts in its `d` tag, so +/// the client can act on the gate without ever holding the bearer token. +fn approval_json(a: &buzz_db::workflow::ApprovalRecord) -> Value { + serde_json::json!({ + "token": hex::encode(&a.token), + "workflow_id": a.workflow_id, + "run_id": a.run_id, + "step_id": a.step_id, + "step_index": a.step_index, + "approver_spec": a.approver_spec, + "status": a.status.to_string(), + "approver_pubkey": a.approver_pubkey.as_ref().map(hex::encode), + "note": a.note, + // What the approver is being asked to approve. Null for gates created + // before migration 0027 — the client must show that as "not recorded" + // rather than an empty package. + "request_message": a.request_message, + "expires_at": a.expires_at.to_rfc3339(), + "created_at": a.created_at.timestamp(), + }) +} + +/// `GET /workflows/{workflow_id}/runs` — run history (NIP-98 + channel member). +pub async fn workflow_runs( + State(state): State>, + Path(workflow_id): Path, + headers: HeaderMap, + RawQuery(raw_query): RawQuery, + Query(q): Query, +) -> Result, (StatusCode, Json)> { + let wf_uuid = Uuid::parse_str(&workflow_id) + .map_err(|_| api_error(StatusCode::BAD_REQUEST, "invalid workflow id"))?; + + let tenant = authorize_workflow_read( + &state, + &headers, + &format!("/workflows/{workflow_id}/runs"), + raw_query.as_deref(), + wf_uuid, + ) + .await?; + + let limit = q.limit.filter(|n| *n > 0).unwrap_or(50).min(200); + let rows = state + .db + .list_workflow_runs(tenant.community(), wf_uuid, limit) + .await + .map_err(|e| internal_error(&format!("list runs: {e}")))?; + + Ok(Json(Value::Array(rows.iter().map(run_json).collect()))) +} + +/// `GET /workflows/{workflow_id}/runs/{run_id}/approvals` — gates for one run. +pub async fn workflow_run_approvals( + State(state): State>, + Path((workflow_id, run_id)): Path<(String, String)>, + headers: HeaderMap, + RawQuery(raw_query): RawQuery, +) -> Result, (StatusCode, Json)> { + let wf_uuid = Uuid::parse_str(&workflow_id) + .map_err(|_| api_error(StatusCode::BAD_REQUEST, "invalid workflow id"))?; + let run_uuid = Uuid::parse_str(&run_id) + .map_err(|_| api_error(StatusCode::BAD_REQUEST, "invalid run id"))?; + + let tenant = authorize_workflow_read( + &state, + &headers, + &format!("/workflows/{workflow_id}/runs/{run_id}/approvals"), + raw_query.as_deref(), + wf_uuid, + ) + .await?; + + let rows = state + .db + .get_run_approvals(tenant.community(), wf_uuid, run_uuid) + .await + .map_err(|e| internal_error(&format!("list approvals: {e}")))?; + + Ok(Json(Value::Array(rows.iter().map(approval_json).collect()))) +} + /// `GET /moderation/reports` — the moderation queue (NIP-98 + mod-authz). pub async fn moderation_reports( State(state): State>, diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 2d82736807..6414a2e694 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -1070,7 +1070,7 @@ async fn handle_approval_grant( check_approver_spec(&approval.approver_spec, &self_hex)?; // Persist the command event — returns open transaction - let tx = match persist_command_event(state, tenant, event, None).await? { + let mut tx = match persist_command_event(state, tenant, event, None).await? { PersistResult::Duplicate => { return Ok(IngestResult { event_id: event.id.to_hex(), @@ -1081,7 +1081,11 @@ async fn handle_approval_grant( PersistResult::Inserted(tx) => tx, }; - // 5. Execute: update approval status to granted + // 5. Execute: update approval status to granted — INSIDE the same + // transaction as the command event. An approval that reads `granted` + // with no kind:46030 event on record is an unattributable decision on a + // gate whose entire purpose is attribution, so the two commit together + // or not at all. let note = if event.content.is_empty() { None } else { @@ -1090,7 +1094,8 @@ async fn handle_approval_grant( let updated = state .db - .update_approval_by_stored_hash( + .update_approval_by_stored_hash_tx( + &mut tx, tenant.community(), &token_hash, ApprovalStatus::Granted, @@ -1101,12 +1106,14 @@ async fn handle_approval_grant( .map_err(|e| IngestError::Internal(format!("error: db update_approval: {e}")))?; if !updated { + // Rolls back the event insert too — a decision that did not apply must + // not leave a command event implying it did. return Err(IngestError::Rejected( "invalid: approval already acted on (race)".into(), )); } - // Commit: event + approval update succeeded atomically. + // Commit: event + approval update land atomically. tx.commit() .await .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?; @@ -1133,6 +1140,7 @@ async fn handle_approval_grant( serde_json::json!({ "status": "granted", "run_id": run_id.to_string(), + "workflow_id": workflow_id.to_string(), }) ), }) @@ -1181,7 +1189,7 @@ async fn handle_approval_deny( check_approver_spec(&approval.approver_spec, &self_hex)?; // Persist the command event — returns open transaction - let tx = match persist_command_event(state, tenant, event, None).await? { + let mut tx = match persist_command_event(state, tenant, event, None).await? { PersistResult::Duplicate => { return Ok(IngestResult { event_id: event.id.to_hex(), @@ -1192,7 +1200,8 @@ async fn handle_approval_deny( PersistResult::Inserted(tx) => tx, }; - // 5. Execute: update approval status to denied + // 5. Execute: update approval status to denied — same transaction as the + // command event. See handle_approval_grant for why. let note = if event.content.is_empty() { None } else { @@ -1201,7 +1210,8 @@ async fn handle_approval_deny( let updated = state .db - .update_approval_by_stored_hash( + .update_approval_by_stored_hash_tx( + &mut tx, tenant.community(), &token_hash, ApprovalStatus::Denied, @@ -1217,7 +1227,7 @@ async fn handle_approval_deny( )); } - // Commit: event + approval denial succeeded atomically. + // Commit: event + approval denial land atomically. tx.commit() .await .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?; @@ -1270,6 +1280,7 @@ async fn handle_approval_deny( serde_json::json!({ "status": "denied", "run_id": run_id.to_string(), + "workflow_id": approval.workflow_id.to_string(), }) ), }) @@ -1284,6 +1295,30 @@ async fn resume_workflow_after_approval( workflow_id: Uuid, resume_index: usize, ) { + // Atomically claim the run before reading it. Resumption has two + // independent triggers (the task spawned on grant, and the recovery sweep), + // plus one per extra pod. A read-then-act status check is TOCTOU: two + // callers can both observe `waiting_approval` and both execute every + // post-gate step. For a `send_message` step that means two distinct signed + // events and therefore two external send instructions. + // + // The compare-and-set below returns true for exactly one caller; everyone + // else returns here having done nothing. + match db.claim_run_for_resume(community_id, run_id).await { + Ok(true) => {} + Ok(false) => { + tracing::debug!( + run_id = %run_id, + "resume_workflow: run not in waiting_approval (already claimed or resolved) — skipping" + ); + return; + } + Err(e) => { + tracing::error!("resume_workflow: claim failed for run {run_id}: {e}"); + return; + } + } + let run = match db.get_workflow_run(community_id, run_id).await { Ok(r) => r, Err(e) => { @@ -1292,15 +1327,6 @@ async fn resume_workflow_after_approval( } }; - // Guard: only resume runs that are actually waiting for approval - if run.status != RunStatus::WaitingApproval { - tracing::warn!( - "resume_workflow: run {run_id} has status '{}', expected 'waiting_approval'", - run.status - ); - return; - } - let workflow = match db.get_workflow(community_id, workflow_id).await { Ok(w) => w, Err(e) => { @@ -1368,3 +1394,113 @@ async fn resume_workflow_after_approval( .finalize_run(community_id, run_id, result, existing_trace) .await; } + +/// Resume any run left parked against an already-granted approval. +/// +/// The crash window this closes: `handle_approval_grant` commits the decision, +/// then spawns resumption. If the process dies in between, the decision is +/// durable but the resumption is not. Without this sweep the run waits forever +/// and the approver has no way to tell — the gate reads `granted`, so a repeat +/// grant is rejected as "already acted on". +/// +/// Safe to run repeatedly and concurrently with a live resumption: the atomic +/// claim inside `resume_workflow_after_approval` admits exactly one caller. +/// Expired and denied gates are not selected at all, so this can never +/// resurrect a fail-closed run. +/// +/// A process that dies after claiming still leaves the run in `running`, which +/// this query deliberately does not select. Send-message steps now have a +/// durable per-step journal, but other action types do not, so blindly taking +/// over stale running rows could still duplicate a webhook or another side +/// effect. Recovery therefore remains conservative and fail-closed. +pub async fn recover_granted_waiting_runs(state: &Arc) { + let stranded = match state.db.list_granted_but_waiting_runs().await { + Ok(rows) => rows, + Err(e) => { + tracing::error!("approval recovery: list failed: {e}"); + return; + } + }; + + for row in stranded { + tracing::warn!( + run_id = %row.run_id, + "Approval recovery: resuming run granted but never resumed" + ); + resume_workflow_after_approval( + Arc::clone(&state.workflow_engine), + state.db.clone(), + row.community_id, + row.run_id, + row.workflow_id, + row.step_index as usize + 1, + ) + .await; + } +} + +#[cfg(test)] +mod approval_tests { + use super::*; + + const CHARLIE: &str = "1a99c7e0596b98299393c384a3b1959374e483c6658772ce3337ea0474e74b90"; + const AGENT: &str = "212b212c906f32e9922f3d8c6fd0a691439766699b45dcf048b9ce94cb4ed637"; + + #[test] + fn designated_pubkey_may_approve() { + assert!(check_approver_spec(CHARLIE, CHARLIE).is_ok()); + } + + #[test] + fn spec_match_is_case_insensitive_both_ways() { + assert!(check_approver_spec(&CHARLIE.to_uppercase(), CHARLIE).is_ok()); + assert!(check_approver_spec(CHARLIE, &CHARLIE.to_uppercase()).is_ok()); + } + + #[test] + fn other_pubkey_may_not_approve() { + // The security property that makes a pubkey spec worth using: reading + // the gate (and its token) is not the same as being able to pass it. + let err = + check_approver_spec(CHARLIE, AGENT).expect_err("a non-designated key must be refused"); + assert!( + format!("{err:?}").contains("not the designated approver"), + "unexpected error: {err:?}" + ); + } + + #[test] + fn any_spec_admits_any_authenticated_key() { + assert!(check_approver_spec("any", AGENT).is_ok()); + assert!(check_approver_spec("", AGENT).is_ok()); + assert!(check_approver_spec(" ", AGENT).is_ok()); + } + + #[test] + fn mention_and_role_specs_fail_closed() { + // Guards the regression this whole change exists to remove: a gate + // saved with `from: "@charlie"` is unapprovable by anyone, including + // Charlie. Validation now blocks the save, and this is the backstop + // for definitions stored before that check landed. + for spec in [ + "@charlie", + "@release-manager", + "owner", + "admin", + "role:owner", + ] { + let err = + check_approver_spec(spec, CHARLIE).expect_err("unsupported spec must fail closed"); + assert!( + format!("{err:?}").contains("not yet supported"), + "spec '{spec}' gave unexpected error: {err:?}" + ); + } + } + + #[test] + fn malformed_hex_specs_fail_closed() { + assert!(check_approver_spec("1a99c7e0", CHARLIE).is_err()); + assert!(check_approver_spec(&"z".repeat(64), CHARLIE).is_err()); + } +} diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 34dc2dfcf8..7a776060c8 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -625,6 +625,33 @@ async fn main() -> anyhow::Result<()> { let wf_cron = Arc::clone(&workflow_engine); tokio::spawn(async move { wf_cron.run().await }); + // Approval resumption recovery. `handle_approval_grant` commits the + // decision and then spawns resumption; a crash in that gap leaves a run + // parked in `waiting_approval` against an already-`granted` approval, which + // nothing else would ever pick up. This sweep is derived purely from + // committed state, so it does not depend on any task surviving the crash. + // + // Idempotent by construction: `resume_workflow_after_approval` refuses any + // run not in `waiting_approval`, so a resumption already in flight, or one + // that has already finished, is a no-op here. That is also what stops a + // granted run being executed twice. + { + let recovery_state = Arc::clone(&state); + let interval_secs: u64 = std::env::var("BUZZ_APPROVAL_RECOVERY_INTERVAL_SECS") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(60); + tokio::spawn(async move { + loop { + tokio::time::sleep(std::time::Duration::from_secs(interval_secs)).await; + buzz_relay::handlers::command_executor::recover_granted_waiting_runs( + &recovery_state, + ) + .await; + } + }); + } + // Ephemeral channel reaper — archives channels whose TTL deadline has passed. // Runs every 60s, matching the workflow cron loop pattern. The SQL UPDATE // uses `archived_at IS NULL` as a guard, so concurrent runs from multiple diff --git a/crates/buzz-relay/src/router.rs b/crates/buzz-relay/src/router.rs index 400ed1dfe3..24cef8c8bd 100644 --- a/crates/buzz-relay/src/router.rs +++ b/crates/buzz-relay/src/router.rs @@ -117,6 +117,18 @@ pub fn build_router(state: Arc) -> Router { "/moderation/restricted", get(api::bridge::moderation_restricted), ) + // Workflow run/approval reads (NIP-98 auth + channel-membership gate). + // Runs and approvals are DB rows, not Nostr events, so they cannot be + // served over /query — these are the read path the desktop approval + // card needs to select a waiting run and act on its gate. + .route( + "/workflows/{workflow_id}/runs", + get(api::bridge::workflow_runs), + ) + .route( + "/workflows/{workflow_id}/runs/{run_id}/approvals", + get(api::bridge::workflow_run_approvals), + ) // Webhook trigger (secret-authenticated, no NIP-98) .route("/hooks/{id}", post(api::bridge::workflow_webhook)) // Mesh demo echo probe — testbed-only; 404 unless BUZZ_MESH=on and diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 97c31c2561..2125ac07c7 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -362,6 +362,121 @@ impl ActionSink for RelayActionSink { Ok(event_id_hex) }) } + + fn emit_approval_request( + &self, + community_id: CommunityId, + channel_id: &str, + req: buzz_workflow::ApprovalRequest<'_>, + ) -> Pin> + Send + '_>> { + let channel_id = channel_id.to_owned(); + let token = req.token.to_owned(); + let token_hash_hex = req.token_hash_hex.to_owned(); + let step_id = req.step_id.to_owned(); + let approver_spec = req.approver_spec.to_owned(); + let message = req.message.to_owned(); + let run_id = req.run_id; + let workflow_id = req.workflow_id; + let expires_at = req.expires_at; + + Box::pin(async move { + let state = self + .state + .upgrade() + .ok_or_else(|| ActionSinkError::Database("relay is shutting down".into()))?; + + 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); + + let channel_uuid = Uuid::parse_str(&channel_id) + .map_err(|e| ActionSinkError::InvalidInput(format!("invalid UUID: {e}")))?; + let channel_id_canonical = channel_uuid.to_string(); + + // `d` carries the hashed token because that is what an inbound + // kind:46030/46031 grant quotes; `token` carries the raw value the + // approver needs for `buzz workflows approve --token`. + let mut tags = vec![ + Tag::parse(["d", &token_hash_hex]) + .map_err(|e| ActionSinkError::EventBuild(format!("d tag: {e}")))?, + Tag::parse(["token", &token]) + .map_err(|e| ActionSinkError::EventBuild(format!("token tag: {e}")))?, + Tag::parse(["h", &channel_id_canonical]) + .map_err(|e| ActionSinkError::EventBuild(format!("h tag: {e}")))?, + Tag::parse(["run", &run_id.to_string()]) + .map_err(|e| ActionSinkError::EventBuild(format!("run tag: {e}")))?, + Tag::parse(["workflow", &workflow_id.to_string()]) + .map_err(|e| ActionSinkError::EventBuild(format!("workflow tag: {e}")))?, + Tag::parse(["step", &step_id]) + .map_err(|e| ActionSinkError::EventBuild(format!("step tag: {e}")))?, + Tag::parse(["expires_at", &expires_at.timestamp().to_string()]) + .map_err(|e| ActionSinkError::EventBuild(format!("expires tag: {e}")))?, + // Never re-enter the trigger path from an approval request. + Tag::parse(["buzz:workflow", "true"]) + .map_err(|e| ActionSinkError::EventBuild(format!("workflow tag: {e}")))?, + ]; + + // A pubkey spec gets a `p` tag so the designated approver is + // notified (and any agent watching wakes). `any` gets none — + // there is nobody specific to address. + if approver_spec != "any" { + tags.push( + Tag::parse(["p", &approver_spec]) + .map_err(|e| ActionSinkError::EventBuild(format!("approver p tag: {e}")))?, + ); + } + + let kind_u32 = buzz_core::kind::KIND_WORKFLOW_APPROVAL_REQUESTED; + let event = EventBuilder::new(Kind::from(kind_u32 as u16), &message) + .tags(tags) + .sign_with_keys(&state.relay_keypair) + .map_err(|e| ActionSinkError::EventBuild(format!("signing: {e}")))?; + + let event_id_hex = event.id.to_hex(); + + info!( + event_id = %event_id_hex, + channel_id = %channel_id_canonical, + run_id = %run_id, + step = %step_id, + "Workflow approval gate: emitting kind {kind_u32} request" + ); + + let (stored_event, was_inserted) = state + .db + .insert_event_with_thread_metadata( + tenant.community(), + &event, + Some(channel_uuid), + None, + ) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))?; + + if was_inserted { + let relay_pubkey_hex = state.relay_keypair.public_key().to_hex(); + let _ = dispatch_persistent_event( + &tenant, + &state, + &stored_event, + kind_u32, + &relay_pubkey_hex, + None, + ) + .await; + } + + Ok(event_id_hex) + }) + } } #[cfg(test)] diff --git a/crates/buzz-workflow/Cargo.toml b/crates/buzz-workflow/Cargo.toml index d4813e56d4..c7abcf25f2 100644 --- a/crates/buzz-workflow/Cargo.toml +++ b/crates/buzz-workflow/Cargo.toml @@ -11,6 +11,7 @@ description = "YAML-as-code workflow engine for Buzz" buzz-core = { workspace = true } buzz-db = { workspace = true } hex = { workspace = true } +sha2 = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } serde_yaml = { workspace = true } diff --git a/crates/buzz-workflow/examples/validate_workflow.rs b/crates/buzz-workflow/examples/validate_workflow.rs new file mode 100644 index 0000000000..8e2c39e04f --- /dev/null +++ b/crates/buzz-workflow/examples/validate_workflow.rs @@ -0,0 +1,81 @@ +//! Validate a workflow YAML file without publishing it. +//! +//! `parse_yaml` covers schema + definition validation (including the +//! `request_approval.from` rule), but a `message_posted` filter is an evalexpr +//! expression that is only evaluated at trigger time — a malformed one fails +//! silently in production, never firing. This also compiles the filter against +//! a representative context so that class of error surfaces before publishing. +//! +//! cargo run -p buzz-workflow --example validate_workflow -- [author_hex] [text] [reply_author_hex] [reply_text] +//! cargo run -p buzz-workflow --example validate_workflow -- --emit-json [author_hex] [text] [reply_author_hex] [reply_text] + +use buzz_workflow::executor::TriggerContext; +use buzz_workflow::schema::TriggerDef; + +#[tokio::main] +async fn main() { + let mut args: Vec = std::env::args().skip(1).collect(); + let emit_json = args.first().is_some_and(|arg| arg == "--emit-json"); + if emit_json { + args.remove(0); + } + let mut args = args.into_iter(); + let path = args + .next() + .expect("usage: validate_workflow [--emit-json] [author] [text] [reply_author] [reply_text]"); + let author = args.next().unwrap_or_default(); + let text = args.next().unwrap_or_default(); + let reply_to_author = args.next().unwrap_or_default(); + let reply_to_text = args.next().unwrap_or_default(); + + let yaml = std::fs::read_to_string(&path).expect("read yaml"); + + let (def, json) = match buzz_workflow::WorkflowEngine::parse_yaml(&yaml) { + Ok(v) => v, + Err(e) => { + eprintln!("INVALID {path}\n {e}"); + std::process::exit(1); + } + }; + + if !emit_json { + println!("valid {path}"); + println!(" name {}", def.name); + println!(" steps {}", def.steps.len()); + } + + if let TriggerDef::MessagePosted { filter: Some(f) } + | TriggerDef::DiffPosted { filter: Some(f) } = &def.trigger + { + let ctx = TriggerContext { + text: text.clone(), + author: author.clone(), + channel_id: "00000000-0000-0000-0000-000000000000".to_owned(), + timestamp: "1700000000".to_owned(), + emoji: String::new(), + message_id: "0".repeat(64), + reply_to_text, + reply_to_author, + reply_to_message_id: String::new(), + webhook_fields: Default::default(), + }; + match buzz_workflow::executor::evaluate_condition(f, &ctx, &Default::default()).await { + Ok(fires) => { + if !emit_json { + println!(" filter compiles; against the sample context it fires: {fires}"); + } + } + Err(e) => { + eprintln!(" FILTER ERROR: {e}"); + eprintln!( + " A filter that cannot evaluate never fires — this would be a silent no-op." + ); + std::process::exit(2); + } + } + } + + if emit_json { + println!("{json}"); + } +} diff --git a/crates/buzz-workflow/src/action_sink.rs b/crates/buzz-workflow/src/action_sink.rs index 0c6002e74e..ab15bfc63e 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 + '_>>; + + /// Publish the kind:46010 approval-requested event for a suspended run. + /// + /// Without this the run would wait on a token nobody has ever seen. The + /// event carries the raw token so the approver can quote it back via + /// `buzz workflows approve --token`. + /// + /// Note on the raw token in channel-visible content: the token identifies + /// *which* gate is being answered, it is not the authorisation. The relay + /// checks the grant's signing key against `approver_spec` independently, so + /// reading the token does not let a non-approver pass the gate — unless the + /// spec is `"any"`, in which case the token is the only barrier. Prefer a + /// pubkey spec for anything with an external side effect. + fn emit_approval_request( + &self, + community_id: CommunityId, + channel_id: &str, + req: ApprovalRequest<'_>, + ) -> Pin> + Send + '_>>; +} + +/// Parameters for [`ActionSink::emit_approval_request`]. +#[derive(Debug, Clone)] +pub struct ApprovalRequest<'a> { + /// Raw approval token (UUID) the approver quotes back. + pub token: &'a str, + /// Hex-encoded SHA-256 of the token — the `d` tag an inbound grant carries. + pub token_hash_hex: &'a str, + /// The run awaiting approval. + pub run_id: uuid::Uuid, + /// The workflow the run belongs to. + pub workflow_id: uuid::Uuid, + /// Step id of the gate. + pub step_id: &'a str, + /// Approver spec: `"any"` or a 64-char hex pubkey. + pub approver_spec: &'a str, + /// Prompt shown to the approver. + pub message: &'a str, + /// Absolute expiry of the gate. + pub expires_at: chrono::DateTime, } diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index e30541377e..d625d06216 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -37,6 +37,22 @@ pub struct TriggerContext { pub emoji: String, /// Event ID of the triggering message (hex string). pub message_id: String, + /// Content of the persisted event this trigger directly replies to. + /// + /// Empty for top-level events, missing/deleted parents, or lookup failure. + /// This is resolved by the workflow engine from relay-owned thread metadata, + /// not copied from child-event content. + #[serde(default)] + pub reply_to_text: String, + /// Signing pubkey of the persisted event this trigger directly replies to. + /// + /// This deliberately ignores attribution tags: filters that use this field + /// are checking the parent event's cryptographic author. + #[serde(default)] + pub reply_to_author: String, + /// Event ID of the persisted event this trigger directly replies to. + #[serde(default)] + pub reply_to_message_id: String, /// Arbitrary webhook body fields (webhook trigger). pub webhook_fields: HashMap, } @@ -54,6 +70,9 @@ impl TriggerContext { "timestamp" => Some(&self.timestamp), "emoji" => Some(&self.emoji), "message_id" => Some(&self.message_id), + "reply_to_text" => Some(&self.reply_to_text), + "reply_to_author" => Some(&self.reply_to_author), + "reply_to_message_id" => Some(&self.reply_to_message_id), other => self.webhook_fields.get(other).map(|s| s.as_str()), } } @@ -210,6 +229,9 @@ fn apply_filter(value: String, filter: &str) -> Result { /// | `trigger.text` | `trigger_text` | /// | `trigger.author` | `trigger_author` | /// | `trigger.channel_id` | `trigger_channel_id` | +/// | `trigger.reply_to_text` | `trigger_reply_to_text` | +/// | `trigger.reply_to_author` | `trigger_reply_to_author` | +/// | `trigger.reply_to_message_id` | `trigger_reply_to_message_id` | /// | `trigger.timestamp` | `trigger_timestamp` | /// | `trigger.emoji` | `trigger_emoji` | /// | `trigger.message_id` | `trigger_message_id` | @@ -220,6 +242,7 @@ fn apply_filter(value: String, filter: &str) -> Result { /// - `str_contains(haystack, needle)` → bool /// - `str_starts_with(s, prefix)` → bool /// - `str_ends_with(s, suffix)` → bool +/// - `str_trim(s)` → string /// - `str_len(s)` → int pub fn build_eval_context( trigger_ctx: &TriggerContext, @@ -274,6 +297,15 @@ pub fn build_eval_context( ) .map_err(|e| WorkflowError::ConditionError(e.to_string()))?; + ctx.set_function( + "str_trim".into(), + Function::new(|arg| { + let s = arg.as_string()?; + Ok(Value::String(s.trim().to_string())) + }), + ) + .map_err(|e| WorkflowError::ConditionError(e.to_string()))?; + // Register webhook fields first as `trigger_FIELD` so that standard trigger // fields inserted below always take precedence and cannot be spoofed. for (key, val) in &trigger_ctx.webhook_fields { @@ -293,6 +325,15 @@ pub fn build_eval_context( ("trigger_timestamp", trigger_ctx.timestamp.as_str()), ("trigger_emoji", trigger_ctx.emoji.as_str()), ("trigger_message_id", trigger_ctx.message_id.as_str()), + ("trigger_reply_to_text", trigger_ctx.reply_to_text.as_str()), + ( + "trigger_reply_to_author", + trigger_ctx.reply_to_author.as_str(), + ), + ( + "trigger_reply_to_message_id", + trigger_ctx.reply_to_message_id.as_str(), + ), ]; for (name, val) in &trigger_fields { @@ -458,13 +499,39 @@ pub enum StepResult { Completed(JsonValue), /// Step requests suspension (approval gate). Execution must pause. Suspended { - /// Token used to resume or reject this approval gate. - approval_token: String, + /// Everything the caller needs to persist the approval and notify the + /// approver. Carrying this out of the executor (rather than the bare + /// token) is what lets `finalize_run` create a real + /// `workflow_approvals` row instead of guessing at the spec/deadline. + approval: Box, }, /// Step was skipped due to `if:` condition being false. Skipped, } +/// An approval gate that execution suspended on. +/// +/// Produced by the `request_approval` action and consumed by +/// [`crate::WorkflowEngine::finalize_run`], which persists it and emits the +/// kind:46010 request event. The raw `token` is the bearer reference the +/// approver quotes back; only its SHA-256 is stored. +#[derive(Debug, Clone)] +pub struct PendingApproval { + /// Raw approval token (UUID). Stored hashed; surfaced to the approver. + pub token: String, + /// The `id` of the step that suspended. + pub step_id: String, + /// Zero-based index of the suspending step, so resume starts at `+ 1`. + pub step_index: usize, + /// Who may approve. `""`/`"any"` or a 64-char hex pubkey — anything else + /// is rejected at definition-validation time and fails closed here. + pub approver_spec: String, + /// Human-readable prompt shown to the approver. + pub message: String, + /// Absolute deadline after which the gate expires and the run fails. + pub expires_at: chrono::DateTime, +} + fn resolve_send_message_channel( explicit_channel: Option<&str>, trigger_channel: &str, @@ -518,6 +585,7 @@ fn resolve_send_message_channel( /// persist state and stop the execution loop. pub async fn dispatch_action( step_id: &str, + step_index: usize, action: &ActionDef, engine: &WorkflowEngine, community_id: CommunityId, @@ -565,11 +633,90 @@ pub async fn dispatch_action( "SendMessage → {channel_id}: {text}" ); - let event_id = engine - .action_sink()? - .send_message(community_id, &channel_id, text, &owner_pubkey_hex) + // At-most-once dispatch. Claim the step durably before emitting. + // + // The run-level CAS guarantees resumption starts once; it does not + // cover a crash *between* emitting this event and persisting the + // trace entry that records it. On recovery the step would re-run, + // and because a Nostr event is re-signed with a fresh timestamp the + // result is a second, distinct message — for the post-gate step of + // the outreach workflow, a second instruction to send on LinkedIn. + // + // The claim is decided by a primary-key conflict in Postgres, so it + // holds across concurrent resumers, restarts and multiple pods. + use buzz_db::workflow::StepDispatchClaim; + let claim = engine + .db + .claim_step_dispatch(community_id, run_id, step_id) .await - .map_err(WorkflowError::from)?; + .map_err(|e| { + WorkflowError::WebhookError(format!( + "SendMessage: dispatch claim failed for step {step_id}: {e}" + )) + })?; + + let event_id = match claim { + StepDispatchClaim::AlreadyDispatched(id) => { + // Reuse the original event rather than emitting a second + // one. This is the replay path. + let hex_id = hex::encode(&id); + info!( + run_id = %run_id, step = step_id, event_id = %hex_id, + "SendMessage already dispatched — reusing recorded event, not re-sending" + ); + hex_id + } + StepDispatchClaim::InFlight => { + // A previous attempt claimed this step and never recorded an + // event. We cannot know whether the message left the relay, + // so we must not send again. Fail closed and let a human + // decide — a duplicate external send is the worse outcome. + return Err(WorkflowError::WebhookError(format!( + "SendMessage: step {step_id} was claimed by an attempt that did not \ + complete; refusing to re-send because the original dispatch may have \ + succeeded. Inspect workflow_step_dispatches for run {run_id}." + ))); + } + StepDispatchClaim::Claimed => { + let sent = engine + .action_sink()? + .send_message(community_id, &channel_id, text, &owner_pubkey_hex) + .await; + + match sent { + Ok(id) => { + // Record before returning, so a replay finds it. + if let Ok(bytes) = hex::decode(&id) { + if let Err(e) = engine + .db + .complete_step_dispatch(community_id, run_id, step_id, &bytes) + .await + { + warn!( + run_id = %run_id, step = step_id, + "SendMessage: failed to record dispatch: {e}. The step \ + will read as in-flight on replay and refuse to re-send." + ); + } + } + id + } + Err(e) => { + // The sink failed, so nothing left the relay for + // this attempt. Releasing the claim is safe and + // stops a transient error wedging the step forever. + if let Err(rel) = engine + .db + .release_step_dispatch(community_id, run_id, step_id) + .await + { + warn!(run_id = %run_id, step = step_id, "release claim failed: {rel}"); + } + return Err(WorkflowError::from(e)); + } + } + } + }; Ok(StepResult::Completed(serde_json::json!({ "sent": true, @@ -653,18 +800,31 @@ pub async fn dispatch_action( timeout, } => { let timeout_str = timeout.as_deref().unwrap_or("24h"); + let timeout_secs = parse_duration_secs(timeout_str)?; + + // Fail closed on an unsupported approver spec. `validate()` rejects + // these at save time, but a definition stored before that check + // existed can still reach here — suspending on a spec no grant can + // ever satisfy would strand the run until it expired. + let approver_spec = normalize_approver_spec(from)?; + info!( run_id = %run_id, step = step_id, - "RequestApproval from={from} timeout={timeout_str}: {message}" + "RequestApproval timeout={timeout_str}: suspending for approval" ); 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. + let expires_at = chrono::Utc::now() + chrono::Duration::seconds(timeout_secs as i64); Ok(StepResult::Suspended { - approval_token: token, + approval: Box::new(PendingApproval { + token, + step_id: step_id.to_owned(), + step_index, + approver_spec, + message: message.clone(), + expires_at, + }), }) } @@ -699,6 +859,37 @@ fn generate_approval_token(_run_id: Uuid, _step_id: &str) -> String { Uuid::new_v4().to_string() } +/// Canonicalize a `request_approval.from` spec, or reject it. +/// +/// Mirrors `check_approver_spec` in the relay's command executor, which is the +/// enforcement point for an inbound grant. The two must agree: a spec that +/// validates here but not there produces a gate that suspends and can never be +/// approved. +/// +/// Accepted: +/// - `""` / `"any"` — any authenticated user may approve. Normalized to `"any"`. +/// - 64-char hex pubkey — only that key may approve. Normalized to lowercase. +/// +/// Everything else (`@charlie`, `@release-manager`, role names) fails closed. +/// Role-based specs are not implemented relay-side, so accepting one here would +/// strand every run that reached the gate. +pub(crate) fn normalize_approver_spec(spec: &str) -> Result { + let trimmed = spec.trim(); + + if trimmed.is_empty() || trimmed.eq_ignore_ascii_case("any") { + return Ok("any".to_owned()); + } + + if trimmed.len() == 64 && trimmed.chars().all(|c| c.is_ascii_hexdigit()) { + return Ok(trimmed.to_lowercase()); + } + + Err(WorkflowError::InvalidDefinition(format!( + "request_approval.from must be \"any\" or a 64-character hex pubkey; \ + got '{trimmed}', which no approval grant can satisfy" + ))) +} + /// Parse a duration string like "5m", "1h", "30s" into seconds. /// /// Exposed as `pub(crate)` so `schema.rs` can use it for interval validation. @@ -942,7 +1133,7 @@ async fn add_reaction_impl(message_id: &str, emoji: &str) -> Result, + pub pending_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. @@ -959,10 +1150,12 @@ pub struct ExecutionResult { /// 3. Dispatches the action. /// 4. Stores the step output for use by later steps. /// -/// On `RequestApproval`: returns `ExecutionResult` with `approval_token = Some(token)`. -/// Caller must persist the approval record and update the run status. +/// On `RequestApproval`: returns `ExecutionResult` with +/// `pending_approval = Some(..)` and stops. Steps after the gate are not +/// dispatched. The caller persists the approval record and moves the run to +/// `waiting_approval`. /// -/// Returns `ExecutionResult` with `approval_token = None` on normal completion. +/// Returns `ExecutionResult` with `pending_approval = None` on normal completion. /// /// Enforces `engine.config.max_concurrent` via a semaphore — returns /// [`WorkflowError::CapacityExceeded`] immediately if all permits are taken. @@ -1140,6 +1333,7 @@ async fn execute_steps( std::time::Duration::from_secs(timeout_secs), dispatch_action( &step.id, + i, &resolved_action, engine, community_id, @@ -1183,15 +1377,19 @@ async fn execute_steps( })); step_outputs.insert(step.id.clone(), output); } - StepResult::Suspended { approval_token } => { + StepResult::Suspended { approval } => { 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 immediately. Every later step stays unexecuted until an + // authorised grant resumes the run at `step_index + 1`. + trace.push(serde_json::json!({ + "step_id": step.id, + "status": "waiting_approval", + })); return Ok(ExecutionResult { - approval_token: Some(approval_token), + pending_approval: Some(approval), step_index: i, step_outputs, trace, @@ -1209,7 +1407,7 @@ async fn execute_steps( info!(run_id = %run_id, "Workflow run completed"); Ok(ExecutionResult { - approval_token: None, + pending_approval: None, step_index: def.steps.len(), step_outputs, trace, @@ -1229,6 +1427,9 @@ mod tests { timestamp: "1700000000".to_owned(), emoji: "fire".to_owned(), message_id: "event-id-hex".to_owned(), + reply_to_text: "@Hermes REENGAGEMENT_RESEARCH_REQUEST".to_owned(), + reply_to_author: "relay-pubkey".to_owned(), + reply_to_message_id: "parent-event-id-hex".to_owned(), webhook_fields: HashMap::new(), } } @@ -1247,6 +1448,43 @@ mod tests { assert_eq!(out, "By abc123def456"); } + #[test] + fn resolve_reply_parent_context() { + let ctx = make_trigger(); + let out = resolve_template( + "Parent {{trigger.reply_to_author}}: {{trigger.reply_to_text}}", + &ctx, + &HashMap::new(), + ) + .unwrap(); + assert_eq!( + out, + "Parent relay-pubkey: @Hermes REENGAGEMENT_RESEARCH_REQUEST" + ); + } + + #[tokio::test] + async fn condition_can_require_cryptographic_reply_parent() { + let ctx = make_trigger(); + let condition = concat!( + "trigger_reply_to_author == \"relay-pubkey\" && ", + "str_starts_with(trigger_reply_to_text, ", + "\"@Hermes REENGAGEMENT_RESEARCH_REQUEST\")" + ); + assert!(evaluate_condition(condition, &ctx, &HashMap::new()) + .await + .unwrap()); + + let mut post_approval = ctx; + post_approval.reply_to_text = "@Hermes BUZZ_REENGAGEMENT_APPROVED".to_owned(); + assert!( + !evaluate_condition(condition, &post_approval, &HashMap::new()) + .await + .unwrap(), + "a post-approval response must not satisfy the research-response gate" + ); + } + #[test] fn resolve_step_output() { let ctx = make_trigger(); @@ -1659,6 +1897,20 @@ mod tests { assert!(result); } + #[tokio::test] + async fn condition_str_trim_composes_with_ends_with() { + let mut ctx = make_trigger(); + ctx.text = "package\nDRAFT_END \n\t".to_string(); + let result = evaluate_condition( + "str_ends_with(str_trim(trigger_text), \"DRAFT_END\")", + &ctx, + &HashMap::new(), + ) + .await + .unwrap(); + assert!(result); + } + #[tokio::test] async fn condition_str_len() { let ctx = make_trigger(); // text = "P1 incident in production" (25 chars) @@ -1834,4 +2086,132 @@ mod tests { .expect("override should be accepted"); assert_eq!(resolved, override_channel_id.to_string()); } + + // ── WF-08: approval gate suspension ───────────────────────────────────── + + const HEX64: &str = "1a99c7e0596b98299393c384a3b1959374e483c6658772ce3337ea0474e74b90"; + + #[test] + fn approver_spec_normalizes_any_and_empty() { + assert_eq!(normalize_approver_spec("any").unwrap(), "any"); + assert_eq!(normalize_approver_spec("ANY").unwrap(), "any"); + assert_eq!(normalize_approver_spec("").unwrap(), "any"); + assert_eq!(normalize_approver_spec(" ").unwrap(), "any"); + } + + #[test] + fn approver_spec_normalizes_hex_to_lowercase() { + let upper = HEX64.to_uppercase(); + assert_eq!(normalize_approver_spec(&upper).unwrap(), HEX64); + assert_eq!(normalize_approver_spec(HEX64).unwrap(), HEX64); + } + + #[test] + fn approver_spec_rejects_mentions_and_roles() { + // These are exactly the specs the relay's check_approver_spec fails + // closed on. Accepting them here would strand a run at the gate. + for bad in ["@charlie", "@release-manager", "owner", "admin"] { + assert!( + normalize_approver_spec(bad).is_err(), + "spec '{bad}' must be rejected" + ); + } + } + + #[test] + fn approver_spec_rejects_malformed_hex() { + // Right length, wrong alphabet. + let non_hex = "z".repeat(64); + assert!(normalize_approver_spec(&non_hex).is_err()); + // Right alphabet, wrong length. + assert!(normalize_approver_spec("1a99c7e0").is_err()); + assert!(normalize_approver_spec(&format!("{HEX64}00")).is_err()); + } + + /// A three-step definition: act, gate, act. Mirrors the real + /// "draft -> approve -> send" shape. + fn gated_def(from: &str) -> WorkflowDef { + let yaml = format!( + "name: Gated\ntrigger:\n on: message_posted\nsteps:\n - id: draft\n action: send_message\n text: 'draft'\n - id: gate\n action: request_approval\n from: '{from}'\n timeout: '48h'\n message: 'send it?'\n - id: send\n action: send_message\n text: 'sent'\n" + ); + crate::schema::parse_yaml(&yaml) + .expect("fixture should parse") + .0 + } + + #[test] + fn approval_gate_is_the_second_of_three_steps() { + let def = gated_def(HEX64); + assert_eq!(def.steps.len(), 3); + assert_eq!(def.steps[1].id, "gate"); + assert!(matches!( + def.steps[1].action, + ActionDef::RequestApproval { .. } + )); + // The step after the gate is a real side effect. If suspension ever + // failed to stop the loop, this is what would fire unapproved. + assert_eq!(def.steps[2].id, "send"); + } + + #[test] + fn suspended_result_carries_everything_needed_to_persist_the_gate() { + // Build the PendingApproval the way dispatch_action does, and assert the + // fields the DB row and the kind:46010 event are built from. + let expires_at = chrono::Utc::now() + chrono::Duration::seconds(172_800); + let approval = PendingApproval { + token: Uuid::new_v4().to_string(), + step_id: "gate".to_owned(), + step_index: 1, + approver_spec: normalize_approver_spec(HEX64).unwrap(), + message: "send it?".to_owned(), + expires_at, + }; + + // Resume must start at the step *after* the gate, never re-run it. + assert_eq!(approval.step_index + 1, 2); + assert_eq!(approval.approver_spec, HEX64); + // The token is a UUID, which is what the CLI's validate_uuid expects. + assert!(Uuid::parse_str(&approval.token).is_ok()); + + let result = ExecutionResult { + pending_approval: Some(Box::new(approval)), + step_index: 1, + step_outputs: HashMap::new(), + trace: Vec::new(), + }; + assert!( + result.pending_approval.is_some(), + "a suspended run must be distinguishable from a completed one" + ); + assert_eq!( + result.step_index, 1, + "step_index must be the gate, so step 2 stays unexecuted" + ); + } + + #[test] + fn completed_result_has_no_pending_approval() { + let result = ExecutionResult { + pending_approval: None, + step_index: 3, + step_outputs: HashMap::new(), + trace: Vec::new(), + }; + // finalize_run branches on exactly this: None means Completed. + assert!(result.pending_approval.is_none()); + } + + #[test] + fn approval_timeout_parses_to_expected_deadline() { + assert_eq!(parse_duration_secs("48h").unwrap(), 172_800); + assert_eq!(parse_duration_secs("30m").unwrap(), 1_800); + assert_eq!(parse_duration_secs("60s").unwrap(), 60); + assert!(parse_duration_secs("soon").is_err()); + } + + #[test] + fn default_approval_timeout_is_24h() { + // dispatch_action falls back to "24h" when `timeout` is omitted. + assert_eq!(parse_duration_secs("24h").unwrap(), 86_400); + } } diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index e142221169..87d33fb4cf 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -35,7 +35,7 @@ pub mod error; pub mod executor; pub mod schema; -pub use action_sink::{ActionSink, ActionSinkError}; +pub use action_sink::{ActionSink, ActionSinkError, ApprovalRequest}; pub use error::{PartialProgress, WorkflowError}; pub use executor::ExecutionResult; pub use schema::{ActionDef, Step, TriggerDef, WorkflowDef}; @@ -226,30 +226,42 @@ 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 Some(approval) = result.pending_approval { + // WF-08: persist the gate, park the run, then tell the + // approver. Any failure along the way fails the run closed — + // a run left in `running` with no approval row is a run that + // silently never resumes. if let Err(e) = self - .db - .update_workflow_run( + .suspend_run_for_approval( community_id, run_id, - RunStatus::Failed, step_count, &trace_json, - Some("approval gates not yet implemented — see WF-08"), + &approval, ) .await { tracing::error!( run_id = %run_id, - "Failed to update run to Failed (approval gate): {e}" + "Failed to open approval gate, failing run closed: {e}" ); + if let Err(db_err) = self + .db + .update_workflow_run( + community_id, + run_id, + RunStatus::Failed, + step_count, + &trace_json, + Some(&format!("approval gate could not be opened: {e}")), + ) + .await + { + tracing::error!( + run_id = %run_id, + "Failed to update run to Failed (approval gate): {db_err}" + ); + } } } else { tracing::info!(run_id = %run_id, "Workflow run completed"); @@ -298,6 +310,179 @@ impl WorkflowEngine { } } + /// Open an approval gate: persist the request, park the run, notify. + /// + /// Ordering is deliberate. The `workflow_approvals` row and the + /// `waiting_approval` status are both committed *before* the kind:46010 + /// event exists, because that event is the first moment anyone can learn + /// the token. A grant racing in against a half-applied state would hit + /// `resume_workflow_after_approval`'s status guard and bail, stranding the + /// run — so the visible artefact goes last. + async fn suspend_run_for_approval( + &self, + community_id: CommunityId, + run_id: uuid::Uuid, + step_count: i32, + trace_json: &serde_json::Value, + approval: &executor::PendingApproval, + ) -> Result<(), WorkflowError> { + use sha2::{Digest, Sha256}; + + let run = self + .db + .get_workflow_run(community_id, run_id) + .await + .map_err(|e| WorkflowError::InvalidDefinition(format!("run lookup failed: {e}")))?; + + let workflow = self + .db + .get_workflow(community_id, run.workflow_id) + .await + .map_err(|e| { + WorkflowError::InvalidDefinition(format!("workflow lookup failed: {e}")) + })?; + + let channel_id = workflow.channel_id.ok_or_else(|| { + WorkflowError::InvalidDefinition( + "approval gate requires a channel-bound workflow: nowhere to post the request" + .into(), + ) + })?; + + // 1. Persist the gate. `create_approval` hashes the raw token itself. + self.db + .create_approval(buzz_db::workflow::CreateApprovalParams { + community_id, + token: &approval.token, + workflow_id: run.workflow_id, + run_id, + step_id: &approval.step_id, + step_index: approval.step_index as i32, + approver_spec: &approval.approver_spec, + // The fully template-resolved prompt, so the approval card can + // show the exact package the decision applies to. + request_message: &approval.message, + expires_at: approval.expires_at, + }) + .await + .map_err(|e| WorkflowError::InvalidDefinition(format!("create_approval: {e}")))?; + + // 2. Park the run so the resume guard passes and the cron expiry sweep + // can find it. + self.db + .update_workflow_run( + community_id, + run_id, + RunStatus::WaitingApproval, + step_count, + trace_json, + None, + ) + .await + .map_err(|e| WorkflowError::InvalidDefinition(format!("park run: {e}")))?; + + // 3. Publish the request so the approver has a token to quote back. + let token_hash_hex = hex::encode(Sha256::digest(approval.token.as_bytes())); + self.action_sink()? + .emit_approval_request( + community_id, + &channel_id.to_string(), + ApprovalRequest { + token: &approval.token, + token_hash_hex: &token_hash_hex, + run_id, + workflow_id: run.workflow_id, + step_id: &approval.step_id, + approver_spec: &approval.approver_spec, + message: &approval.message, + expires_at: approval.expires_at, + }, + ) + .await?; + + tracing::info!( + run_id = %run_id, + step = %approval.step_id, + expires_at = %approval.expires_at, + "Approval gate opened — run parked in waiting_approval" + ); + + Ok(()) + } + + /// Fail closed every approval gate whose deadline has passed. + /// + /// Called once per cron tick. Marks the approval `expired` and the run + /// `failed`. A gate that times out must never behave like a grant: steps + /// after it stay unexecuted. + async fn expire_stale_approvals(&self) { + let stale = match self.db.list_expired_pending_approvals().await { + Ok(rows) => rows, + Err(e) => { + tracing::error!("Approval expiry sweep: failed to list: {e}"); + return; + } + }; + + for row in stale { + let marked = self + .db + .update_approval_by_stored_hash( + row.community_id, + &row.token_hash, + buzz_db::workflow::ApprovalStatus::Expired, + None, + None, + ) + .await; + + match marked { + // `false` means another actor (a grant or deny landing in the + // same instant) already moved it out of pending — leave the run + // alone, that path owns it. + Ok(false) => continue, + Err(e) => { + tracing::error!(run_id = %row.run_id, "Approval expiry: mark failed: {e}"); + continue; + } + Ok(true) => {} + } + + let run = match self.db.get_workflow_run(row.community_id, row.run_id).await { + Ok(r) => r, + Err(e) => { + tracing::error!(run_id = %row.run_id, "Approval expiry: run lookup: {e}"); + continue; + } + }; + + if run.status != RunStatus::WaitingApproval { + continue; + } + + if let Err(e) = self + .db + .update_workflow_run( + row.community_id, + row.run_id, + RunStatus::Failed, + run.current_step, + &run.execution_trace, + Some("approval gate expired without a decision"), + ) + .await + { + tracing::error!(run_id = %row.run_id, "Approval expiry: fail run: {e}"); + continue; + } + + tracing::warn!( + run_id = %row.run_id, + "Approval gate expired — run failed closed, later steps not executed" + ); + } + } + /// 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. @@ -351,7 +536,44 @@ impl WorkflowEngine { return Ok(()); } - let trigger_ctx = build_trigger_context(event); + let mut trigger_ctx = build_trigger_context(event); + + // Resolve direct-reply provenance from relay-owned thread metadata and + // the persisted parent event. Child events cannot spoof these fields by + // copying text or tags, which lets workflow filters distinguish an + // agent response to one workflow step from a response to another. + // Any lookup failure leaves the fields empty, so provenance-sensitive + // filters fail closed while ordinary top-level workflows still run. + match self + .db + .get_thread_metadata_by_event(community_id, event.event.id.as_bytes()) + .await + { + Ok(Some(metadata)) => { + if let Some(parent_id) = metadata.parent_event_id { + match self.db.get_event_by_id(community_id, &parent_id).await { + Ok(Some(parent)) => { + trigger_ctx.reply_to_text = parent.event.content.clone(); + trigger_ctx.reply_to_author = parent.event.pubkey.to_hex(); + trigger_ctx.reply_to_message_id = parent.event.id.to_hex(); + } + Ok(None) => tracing::debug!( + event_id = %event.event.id.to_hex(), + "Workflow reply parent was missing or deleted" + ), + Err(error) => tracing::warn!( + event_id = %event.event.id.to_hex(), + "Workflow reply parent lookup failed closed: {error}" + ), + } + } + } + Ok(None) => {} + Err(error) => tracing::warn!( + event_id = %event.event.id.to_hex(), + "Workflow thread metadata lookup failed closed: {error}" + ), + } let trigger_ctx_json: serde_json::Value = match serde_json::to_value(&trigger_ctx) { Ok(v) => v, @@ -488,6 +710,11 @@ impl WorkflowEngine { let now = Utc::now(); + // Fail closed on gates whose deadline has passed, before firing + // anything new. Runs the same tick as cron so there is no second + // background task to reason about. + self.expire_stale_approvals().await; + let workflows = match self.db.list_all_enabled_workflows().await { Ok(wf) => wf, Err(e) => { @@ -1010,6 +1237,9 @@ pub fn build_trigger_context(event: &buzz_core::StoredEvent) -> executor::Trigge timestamp: event.event.created_at.as_secs().to_string(), emoji, message_id, + reply_to_text: String::new(), + reply_to_author: String::new(), + reply_to_message_id: String::new(), webhook_fields: HashMap::new(), } } @@ -1901,4 +2131,482 @@ steps: "channel owner's call_webhook workflow fires" ); } + + // ── WF-08: true execution, concurrency and replay ─────────────────────── + + use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering}; + use std::sync::Mutex as StdMutex; + + /// An ActionSink that records every side effect instead of performing one. + /// + /// This is what makes these tests *execution* tests rather than structural + /// ones: the assertion is on what the executor actually dispatched. + #[derive(Default)] + struct RecordingSink { + sent: StdMutex>, + approvals: StdMutex>, + counter: AtomicUsize, + } + + impl RecordingSink { + fn sent_texts(&self) -> Vec { + self.sent + .lock() + .unwrap() + .iter() + .map(|(_, t)| t.clone()) + .collect() + } + fn approval_count(&self) -> usize { + self.approvals.lock().unwrap().len() + } + } + + impl crate::ActionSink for RecordingSink { + fn send_message( + &self, + _community_id: CommunityId, + channel_id: &str, + text: &str, + _author_pubkey: &str, + ) -> std::pin::Pin< + Box< + dyn std::future::Future> + + Send + + '_, + >, + > { + let channel_id = channel_id.to_owned(); + let text = text.to_owned(); + Box::pin(async move { + self.sent.lock().unwrap().push((channel_id, text)); + // Distinct id per call, mirroring the real sink where each + // dispatch is a freshly-signed event. + let n = self.counter.fetch_add(1, AtomicOrdering::SeqCst); + Ok(format!("{:064x}", n + 1)) + }) + } + + fn emit_approval_request( + &self, + _community_id: CommunityId, + _channel_id: &str, + req: crate::ApprovalRequest<'_>, + ) -> std::pin::Pin< + Box< + dyn std::future::Future> + + Send + + '_, + >, + > { + let msg = req.message.to_owned(); + Box::pin(async move { + self.approvals.lock().unwrap().push(msg); + Ok("a".repeat(64)) + }) + } + } + + const APPROVER: &str = "1a99c7e0596b98299393c384a3b1959374e483c6658772ce3337ea0474e74b90"; + + /// draft -> gate -> send. The third step is the one that must never run + /// before an approval. + fn gated_definition_with_timeout(timeout: &str) -> String { + serde_json::json!({ + "name": "gated", + "trigger": {"on": "message_posted"}, + "enabled": true, + "steps": [ + {"id": "draft", "action": "send_message", "text": "DRAFT_PACKAGE"}, + {"id": "gate", "action": "request_approval", "from": APPROVER, + "timeout": timeout, "message": "approve DRAFT_PACKAGE?"}, + {"id": "send", "action": "send_message", "text": "EXTERNAL_SEND"}, + ], + }) + .to_string() + } + + async fn run_to_gate() -> ( + buzz_db::Db, + CommunityId, + Uuid, + Arc, + Arc, + ) { + run_to_gate_with_timeout("48h").await + } + + async fn run_to_gate_with_timeout( + timeout: &str, + ) -> ( + buzz_db::Db, + CommunityId, + Uuid, + Arc, + Arc, + ) { + let db = setup_db().await; + let creator = nostr::Keys::generate().public_key().to_bytes().to_vec(); + let member = nostr::Keys::generate().public_key().to_bytes().to_vec(); + let (community, channel_id) = setup_channel(&db, &creator, &member).await; + + let workflow_id = db + .create_workflow( + community, + Some(channel_id), + &member, + "gated", + &gated_definition_with_timeout(timeout), + &[0u8; 32], + ) + .await + .expect("create workflow"); + + let engine = Arc::new(WorkflowEngine::new(db.clone(), WorkflowConfig::default())); + let sink = Arc::new(RecordingSink::default()); + engine.set_action_sink(sink.clone()); + + engine + .on_event(community, &message_event(channel_id)) + .await + .expect("on_event"); + + // on_event spawns execution; wait for the run to settle. + let mut run_id = None; + for _ in 0..100 { + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + let runs = db + .list_workflow_runs(community, workflow_id, 10) + .await + .expect("runs"); + if let Some(r) = runs.first() { + if r.status != buzz_db::workflow::RunStatus::Running + && r.status != buzz_db::workflow::RunStatus::Pending + { + run_id = Some(r.id); + break; + } + } + } + let run_id = run_id.expect("run should reach a terminal-or-waiting state"); + (db, community, run_id, sink, engine) + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn approval_gate_stops_execution_before_the_sending_step() { + let (db, community, run_id, sink, _engine) = run_to_gate().await; + + let run = db.get_workflow_run(community, run_id).await.expect("run"); + assert_eq!( + run.status, + buzz_db::workflow::RunStatus::WaitingApproval, + "run must park at the gate, not fail or complete" + ); + + let sent = sink.sent_texts(); + assert!( + sent.iter().any(|t| t.contains("DRAFT_PACKAGE")), + "the pre-gate step must have run: {sent:?}" + ); + assert!( + !sent.iter().any(|t| t.contains("EXTERNAL_SEND")), + "THE post-gate side effect must NOT have run before approval: {sent:?}" + ); + assert_eq!( + sink.approval_count(), + 1, + "exactly one approval request emitted" + ); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn gate_persists_the_exact_request_package() { + let (db, community, run_id, _sink, _engine) = run_to_gate().await; + let run = db.get_workflow_run(community, run_id).await.expect("run"); + let approvals = db + .get_run_approvals(community, run.workflow_id, run_id) + .await + .expect("approvals"); + + assert_eq!(approvals.len(), 1); + let a = &approvals[0]; + assert_eq!(a.status, buzz_db::workflow::ApprovalStatus::Pending); + assert_eq!(a.approver_spec, APPROVER); + assert_eq!( + a.request_message.as_deref(), + Some("approve DRAFT_PACKAGE?"), + "the approval card has nothing to show without this" + ); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn only_one_concurrent_resumer_claims_the_run() { + let (db, community, run_id, _sink, _engine) = run_to_gate().await; + + // Two resumers race: the task spawned on grant, and the recovery sweep. + let (a, b) = tokio::join!( + db.claim_run_for_resume(community, run_id), + db.claim_run_for_resume(community, run_id) + ); + let wins = [a.expect("claim a"), b.expect("claim b")] + .iter() + .filter(|w| **w) + .count(); + assert_eq!( + wins, 1, + "exactly one resumer may proceed; a read guard would allow two" + ); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn step_dispatch_is_claimed_at_most_once() { + use buzz_db::workflow::StepDispatchClaim; + let (db, community, run_id, _sink, _engine) = run_to_gate().await; + + let (a, b) = tokio::join!( + db.claim_step_dispatch(community, run_id, "send"), + db.claim_step_dispatch(community, run_id, "send") + ); + let a = a.expect("claim a"); + let b = b.expect("claim b"); + let claimed = [&a, &b] + .iter() + .filter(|c| ***c == StepDispatchClaim::Claimed) + .count(); + assert_eq!(claimed, 1, "only one caller may perform the external send"); + // The loser must not be told to go ahead. + let loser = if a == StepDispatchClaim::Claimed { + b + } else { + a + }; + assert_ne!(loser, StepDispatchClaim::Claimed); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn replay_after_dispatch_reuses_the_original_event() { + use buzz_db::workflow::StepDispatchClaim; + let (db, community, run_id, _sink, _engine) = run_to_gate().await; + + assert_eq!( + db.claim_step_dispatch(community, run_id, "send") + .await + .expect("claim"), + StepDispatchClaim::Claimed + ); + let original = vec![0xABu8; 32]; + db.complete_step_dispatch(community, run_id, "send", &original) + .await + .expect("complete"); + + // Crash recovery replays the step. + match db + .claim_step_dispatch(community, run_id, "send") + .await + .expect("replay") + { + StepDispatchClaim::AlreadyDispatched(id) => assert_eq!( + id, original, + "replay must reuse the original event, never mint a second send" + ), + other => panic!("replay must not re-claim: {other:?}"), + } + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn interrupted_dispatch_refuses_to_resend() { + use buzz_db::workflow::StepDispatchClaim; + let (db, community, run_id, _sink, _engine) = run_to_gate().await; + + // Claimed, then the process died before recording the event. + assert_eq!( + db.claim_step_dispatch(community, run_id, "send") + .await + .expect("claim"), + StepDispatchClaim::Claimed + ); + assert_eq!( + db.claim_step_dispatch(community, run_id, "send") + .await + .expect("replay"), + StepDispatchClaim::InFlight, + "an unresolved dispatch must fail closed, not optimistically re-send" + ); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn denial_leaves_the_sending_step_unexecuted() { + let (db, community, run_id, sink, _engine) = run_to_gate().await; + let run = db.get_workflow_run(community, run_id).await.expect("run"); + let approvals = db + .get_run_approvals(community, run.workflow_id, run_id) + .await + .expect("approvals"); + let token_hash = approvals[0].token.clone(); + + let denier = nostr::Keys::generate().public_key().to_bytes().to_vec(); + assert!(db + .update_approval_by_stored_hash( + community, + &token_hash, + buzz_db::workflow::ApprovalStatus::Denied, + Some(&denier), + Some("no") + ) + .await + .expect("deny")); + + // A denied gate must not be selected by recovery. + let stranded = db.list_granted_but_waiting_runs().await.expect("stranded"); + assert!( + !stranded.iter().any(|r| r.run_id == run_id), + "a denied run must never be picked up for resumption" + ); + assert!( + !sink + .sent_texts() + .iter() + .any(|t| t.contains("EXTERNAL_SEND")), + "denial must never dispatch the post-gate step" + ); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn expiry_fails_closed_and_never_runs_later_steps() { + // A real 1s deadline, waited out. No production helper exists to move an + // approval's expiry, and adding one would put a method that mutates + // approval deadlines into the shipped API for the sake of one test. + let (db, community, run_id, sink, engine) = run_to_gate_with_timeout("1s").await; + tokio::time::sleep(std::time::Duration::from_millis(1500)).await; + + engine.expire_stale_approvals().await; + + let run = db.get_workflow_run(community, run_id).await.expect("run"); + assert_eq!( + run.status, + buzz_db::workflow::RunStatus::Failed, + "an expired gate must fail the run closed" + ); + assert!( + !sink + .sent_texts() + .iter() + .any(|t| t.contains("EXTERNAL_SEND")), + "expiry must never dispatch the post-gate step" + ); + let after = db + .get_run_approvals(community, run.workflow_id, run_id) + .await + .expect("approvals"); + assert_eq!(after[0].status, buzz_db::workflow::ApprovalStatus::Expired); + } + + /// The end-to-end grant path: an authorised decision resumes the run at + /// gate+1, dispatches the external step exactly once, and a subsequent + /// recovery attempt adds nothing. + /// + /// The CAS and journal unit tests prove the primitives in isolation; only + /// this proves the resume *index* and the executor wiring are right — an + /// off-by-one here would either re-run the gate forever or skip the send. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn authorised_grant_resumes_at_gate_plus_one_and_sends_once() { + let (db, community, run_id, sink, engine) = run_to_gate().await; + let run = db.get_workflow_run(community, run_id).await.expect("run"); + let workflow_id = run.workflow_id; + let approvals = db + .get_run_approvals(community, workflow_id, run_id) + .await + .expect("approvals"); + let approval = approvals[0].clone(); + assert_eq!(approval.step_index, 1, "the gate is step index 1"); + + // Mark granted through the same guarded update the relay's grant + // handler uses (status='pending' predicate included). + let approver = hex::decode(APPROVER).expect("hex"); + assert!(db + .update_approval_by_stored_hash( + community, + &approval.token, + buzz_db::workflow::ApprovalStatus::Granted, + Some(&approver), + None, + ) + .await + .expect("grant")); + + // Resume exactly as the relay does: claim, then execute from gate+1. + let resume_index = approval.step_index as usize + 1; + assert!(db + .claim_run_for_resume(community, run_id) + .await + .expect("claim")); + + let workflow = db.get_workflow(community, workflow_id).await.expect("wf"); + let def: crate::schema::WorkflowDef = + serde_json::from_value(workflow.definition.clone()).expect("def"); + let trigger_ctx = crate::executor::TriggerContext::default(); + let result = crate::executor::execute_from_step( + &engine, + community, + run_id, + &def, + &trigger_ctx, + resume_index, + None, + ) + .await; + engine.finalize_run(community, run_id, result, None).await; + + let sent = sink.sent_texts(); + let sends = sent.iter().filter(|t| t.contains("EXTERNAL_SEND")).count(); + assert_eq!( + sends, 1, + "the approved step must dispatch exactly once: {sent:?}" + ); + let drafts = sent.iter().filter(|t| t.contains("DRAFT_PACKAGE")).count(); + assert_eq!( + drafts, 1, + "resume must start after the gate, not replay the draft" + ); + + let run = db.get_workflow_run(community, run_id).await.expect("run"); + assert_eq!(run.status, buzz_db::workflow::RunStatus::Completed); + + // Now the crash-recovery path runs again over the same run. The step + // journal must suppress a second external send. + let stranded = db.list_granted_but_waiting_runs().await.expect("stranded"); + assert!( + !stranded.iter().any(|r| r.run_id == run_id), + "a completed run must not be selected for recovery" + ); + let second = crate::executor::execute_from_step( + &engine, + community, + run_id, + &def, + &trigger_ctx, + resume_index, + None, + ) + .await; + let _ = second; + let sends_after = sink + .sent_texts() + .iter() + .filter(|t| t.contains("EXTERNAL_SEND")) + .count(); + assert_eq!( + sends_after, 1, + "a replayed resume must not issue a second external send" + ); + } } diff --git a/crates/buzz-workflow/src/schema.rs b/crates/buzz-workflow/src/schema.rs index 9bc79aa48b..5c217f05ea 100644 --- a/crates/buzz-workflow/src/schema.rs +++ b/crates/buzz-workflow/src/schema.rs @@ -238,6 +238,27 @@ impl WorkflowDef { } } + // Reject approval gates nobody could ever satisfy. The relay enforces + // the same rule on an inbound grant, so an unsupported spec here would + // produce a run that suspends and then expires 24h later having done + // nothing — the failure mode WF-08 exists to remove. + for step in &self.steps { + if let ActionDef::RequestApproval { from, timeout, .. } = &step.action { + crate::executor::normalize_approver_spec(from).map_err(|e| { + WorkflowError::InvalidDefinition(format!("step '{}': {e}", step.id)) + })?; + + if let Some(t) = timeout { + crate::executor::parse_duration_secs(t).map_err(|_| { + WorkflowError::InvalidDefinition(format!( + "step '{}': invalid approval timeout '{t}': expected a duration like '30m', '24h'", + step.id + )) + })?; + } + } + } + Ok(()) } } @@ -357,7 +378,7 @@ mod tests { " - id: topic\n action: set_channel_topic\n topic: Status active\n", " - id: react\n action: add_reaction\n emoji: white_check_mark\n", " - id: hook\n action: call_webhook\n url: https://hooks.example.com/notify\n method: POST\n", - " - id: approve\n action: request_approval\n from: '@manager'\n message: Approve?\n timeout: 4h\n", + " - id: approve\n action: request_approval\n from: '1a99c7e0596b98299393c384a3b1959374e483c6658772ce3337ea0474e74b90'\n message: Approve?\n timeout: 4h\n", " - id: wait\n action: delay\n duration: 5m\n", ); let (def, _) = parse_yaml(yaml).expect("parse failed"); @@ -393,7 +414,7 @@ mod tests { "name: Deploy Approval\n", "trigger:\n on: webhook\n", "steps:\n", - " - id: request\n action: request_approval\n from: '@engineering-lead'\n", + " - id: request\n action: request_approval\n from: '1a99c7e0596b98299393c384a3b1959374e483c6658772ce3337ea0474e74b90'\n", " message: Approve deploy?\n timeout: 4h\n", " - id: notify_approved\n if: 'steps_request_output_approved == true'\n", " action: send_message\n text: Deploy approved\n", @@ -888,4 +909,58 @@ mod tests { TriggerDef::DiffPosted { filter: Some(_) } )); } + + // ── WF-08: approval gate definition validation ─────────────────────────── + + const HEX64: &str = "1a99c7e0596b98299393c384a3b1959374e483c6658772ce3337ea0474e74b90"; + + fn approval_yaml(from: &str, timeout: &str) -> String { + format!( + "name: Gated\ntrigger:\n on: message_posted\nsteps:\n - id: draft\n action: send_message\n text: 'draft'\n - id: gate\n action: request_approval\n from: '{from}'\n timeout: '{timeout}'\n message: 'ok?'\n - id: send\n action: send_message\n text: 'sent'\n" + ) + } + + #[test] + fn approval_gate_accepts_hex_pubkey_spec() { + let (def, _) = parse_yaml(&approval_yaml(HEX64, "48h")).expect("should parse"); + assert_eq!(def.steps.len(), 3); + } + + #[test] + fn approval_gate_accepts_any_spec() { + parse_yaml(&approval_yaml("any", "1h")).expect("\"any\" is a supported spec"); + } + + #[test] + fn approval_gate_rejects_mention_spec() { + // The exact shape that shipped in the #pipeline workflow. The relay's + // check_approver_spec fails closed on it, so a gate built this way + // suspends and can never be approved. Reject it at save time instead. + let err = + parse_yaml(&approval_yaml("@charlie", "48h")).expect_err("@charlie must be rejected"); + let msg = err.to_string(); + assert!(msg.contains("gate"), "error should name the step: {msg}"); + assert!( + msg.contains("64-character hex pubkey"), + "error should say what is accepted: {msg}" + ); + } + + #[test] + fn approval_gate_rejects_role_spec() { + parse_yaml(&approval_yaml("@release-manager", "24h")) + .expect_err("role specs are not implemented relay-side"); + } + + #[test] + fn approval_gate_rejects_short_hex_spec() { + parse_yaml(&approval_yaml("1a99c7e0", "24h")) + .expect_err("a truncated pubkey must not pass"); + } + + #[test] + fn approval_gate_rejects_invalid_timeout() { + parse_yaml(&approval_yaml(HEX64, "soon")) + .expect_err("unparseable timeout must be rejected at save time"); + } } diff --git a/desktop/src-tauri/src/commands/workflows.rs b/desktop/src-tauri/src/commands/workflows.rs index 1d5f309fb5..d5b3fef8c1 100644 --- a/desktop/src-tauri/src/commands/workflows.rs +++ b/desktop/src-tauri/src/commands/workflows.rs @@ -117,30 +117,24 @@ pub async fn get_workflow( .ok_or_else(|| "workflow not found".to_string()) } +/// Run history for a workflow, newest first. +/// +/// Reads `GET /workflows/{id}/runs` (NIP-98 + channel membership). Runs are DB +/// rows rather than Nostr events, so there is no filter that could fetch them +/// over `/query`. The relay returns exactly the `RawWorkflowRun` field set, so +/// the frontend's `raw.map(fromRawWorkflowRun)` stays safe — the response is +/// always a bare array, never a `{ runs: [...] }` wrapper. #[tauri::command] pub async fn get_workflow_runs( workflow_id: String, limit: Option, - _state: State<'_, AppState>, + state: State<'_, AppState>, ) -> Result, String> { - // TODO(workflow-runs): Run reconstruction is a clearly-scoped follow-up. - // The authoritative run record the frontend's `WorkflowRun` shape needs - // (status / current_step / execution_trace / error_message) lives in the - // relay DB and is not exposed to the desktop client as a single queryable - // record. If the relay starts emitting lifecycle events (46001–46007, …), - // folding that stream into `WorkflowRun` would be another viable design. - // The important bit for this command is that raw lifecycle events are not - // the `RawWorkflowRun` contract. - // - // Until then we return a bare empty array — NOT a raw-event wrapper. The - // frontend wrapper (`getWorkflowRuns`) does `raw.map(fromRawWorkflowRun)`, - // so it must receive an array; the wrapped `{ runs: [...] }` shape would - // make `.map()` throw and crash the detail panel (the same TypeError class - // as the original page bug). Raw lifecycle events also don't carry the - // `id`/`workflow_id`/`status`/… fields `RawWorkflowRun` expects, so an - // empty list is the honest, safe placeholder. - let _ = (workflow_id, limit); - Ok(Vec::new()) + let path = match limit { + Some(n) => format!("/workflows/{workflow_id}/runs?limit={n}"), + None => format!("/workflows/{workflow_id}/runs"), + }; + crate::relay::get_relay_json::>(&state, &path).await } // ── Writes ─────────────────────────────────────────────────────────────────── @@ -250,19 +244,23 @@ pub async fn trigger_workflow( // ── Approvals ──────────────────────────────────────────────────────────────── +/// Approval gates for one run. +/// +/// The `token` field returned here is the **hashed** token, hex-encoded — the +/// exact value a kind:46030/46031 grant carries in its `d` tag. The desktop +/// therefore never handles the raw bearer token; it passes this straight +/// through to [`grant_approval`] / [`deny_approval`]. #[tauri::command] pub async fn get_run_approvals( workflow_id: String, run_id: String, - _state: State<'_, AppState>, + state: State<'_, AppState>, ) -> Result, String> { - // TODO(workflow-runs): Like runs (see `get_workflow_runs`), reconstructing - // approvals into the frontend's `WorkflowApproval` shape from lifecycle - // events (46010/46011/46012) is a clearly-scoped follow-up tracked under - // TODO(workflow-runs). Return a bare empty array so the frontend's - // `getRunApprovals` (`raw.map(fromRawApproval)`) is safe. - let _ = (workflow_id, run_id); - Ok(Vec::new()) + crate::relay::get_relay_json::>( + &state, + &format!("/workflows/{workflow_id}/runs/{run_id}/approvals"), + ) + .await } #[tauri::command] @@ -273,7 +271,7 @@ pub async fn grant_approval( ) -> Result { let builder = events::build_approval_grant(&token, note.as_deref())?; let result = submit_event(builder, &state).await?; - Ok(serde_json::json!({ "event_id": result.event_id })) + Ok(approval_action_response(&token, "granted", &result.message)) } #[tauri::command] @@ -284,7 +282,46 @@ pub async fn deny_approval( ) -> Result { let builder = events::build_approval_deny(&token, note.as_deref())?; let result = submit_event(builder, &state).await?; - Ok(serde_json::json!({ "event_id": result.event_id })) + Ok(approval_action_response(&token, "denied", &result.message)) +} + +/// Build the `RawApprovalActionResponse` the frontend converts with +/// `fromRawApprovalResponse` — `{ token, status, run_id, workflow_id }`. +/// +/// Previously this returned `{ event_id }`, so every field the frontend read +/// was `undefined` and the approval card rendered an action that appeared to +/// succeed while carrying no run reference. The relay's OK message contains +/// `response:{"status":..,"run_id":..,"workflow_id":..}`; `token` is echoed +/// from the caller because the relay never returns it. +/// +/// `fallback_status` is used only when the relay's message cannot be parsed +/// (e.g. the duplicate-event short-circuit, which carries no JSON body); the +/// ids then stay empty strings rather than being invented. +pub(crate) fn approval_action_response(token: &str, fallback_status: &str, message: &str) -> Value { + let parsed = parse_command_response::(message).ok(); + let field = |key: &str| -> String { + parsed + .as_ref() + .and_then(|v| v.get(key)) + .and_then(Value::as_str) + .unwrap_or_default() + .to_string() + }; + let status = { + let s = field("status"); + if s.is_empty() { + fallback_status.to_string() + } else { + s + } + }; + + serde_json::json!({ + "token": token, + "status": status, + "run_id": field("run_id"), + "workflow_id": field("workflow_id"), + }) } // ── Helpers (pure, unit-tested in workflows_tests.rs) ───────────────────────── diff --git a/desktop/src-tauri/src/commands/workflows_tests.rs b/desktop/src-tauri/src/commands/workflows_tests.rs index f07f4b0f42..1cf2e68f6d 100644 --- a/desktop/src-tauri/src/commands/workflows_tests.rs +++ b/desktop/src-tauri/src/commands/workflows_tests.rs @@ -207,3 +207,124 @@ fn runs_and_approvals_serialize_to_bare_empty_array() { "[]" ); } + +#[cfg(test)] +mod approval_contract_tests { + use crate::commands::workflows::approval_action_response; + use crate::events::{build_approval_deny, build_approval_grant}; + use nostr::Keys; + + const HASH: &str = "3b1f8c2a9d4e5061728394a5b6c7d8e9f0a1b2c3d4e5f60718293a4b5c6d7e8f"; + + fn tag_pairs(builder: nostr::EventBuilder) -> Vec<(String, String)> { + let event = builder.sign_with_keys(&Keys::generate()).expect("sign"); + event + .tags + .iter() + .map(|t| { + ( + t.kind().to_string(), + t.content().unwrap_or_default().to_string(), + ) + }) + .collect() + } + + #[test] + fn grant_emits_d_tag_with_the_hashed_token() { + // The relay resolves a gate via extract_d_tag(..).or_else(extract_e_tag) + // and looks it up with get_approval_by_stored_hash. A `t` tag — which is + // what this used to emit — is simply not read, so every desktop grant + // was rejected as "missing approval reference". + let tags = tag_pairs(build_approval_grant(HASH, None).expect("build")); + assert!( + tags.iter().any(|(k, v)| k == "d" && v == HASH), + "grant must carry d=: {tags:?}" + ); + assert!( + !tags.iter().any(|(k, _)| k == "t"), + "a `t` tag is never read by the relay: {tags:?}" + ); + } + + #[test] + fn deny_emits_d_tag_with_the_hashed_token() { + let tags = tag_pairs(build_approval_deny(HASH, None).expect("build")); + assert!(tags.iter().any(|(k, v)| k == "d" && v == HASH)); + assert!(!tags.iter().any(|(k, _)| k == "t")); + } + + #[test] + fn grant_and_deny_carry_the_value_verbatim_never_rehashed() { + // The desktop is handed the already-hashed token by get_run_approvals. + // Hashing it again here would produce a value matching no stored row. + for builder in [ + build_approval_grant(HASH, None).expect("grant"), + build_approval_deny(HASH, None).expect("deny"), + ] { + let tags = tag_pairs(builder); + let d = tags.iter().find(|(k, _)| k == "d").expect("d tag").1.clone(); + assert_eq!(d, HASH, "value must pass through untouched"); + assert_eq!(d.len(), 64); + assert!(d.chars().all(|c| c.is_ascii_hexdigit())); + } + } + + #[test] + fn grant_and_deny_use_distinct_kinds() { + let g = build_approval_grant(HASH, None) + .expect("g") + .sign_with_keys(&Keys::generate()) + .expect("sign"); + let d = build_approval_deny(HASH, None) + .expect("d") + .sign_with_keys(&Keys::generate()) + .expect("sign"); + assert_eq!(g.kind.as_u16(), 46030); + assert_eq!(d.kind.as_u16(), 46031); + } + + #[test] + fn note_becomes_the_event_content() { + let e = build_approval_grant(HASH, Some("looks right")) + .expect("build") + .sign_with_keys(&Keys::generate()) + .expect("sign"); + assert_eq!(e.content, "looks right"); + } + + #[test] + fn approval_response_parses_status_run_and_workflow_ids() { + // The relay's OK message shape. Previously this command returned + // {event_id}, so every field the frontend converter read was undefined + // and the card rendered a successful-looking action with no run bound. + let msg = r#"response:{"status":"granted","run_id":"adc3ad0d-d487-4bed-acd7-31e676b50ce5","workflow_id":"5eb69e19-3e13-45e1-ac27-8858bb1f2a36"}"#; + let v = approval_action_response(HASH, "granted", msg); + assert_eq!(v["token"], HASH); + assert_eq!(v["status"], "granted"); + assert_eq!(v["run_id"], "adc3ad0d-d487-4bed-acd7-31e676b50ce5"); + assert_eq!(v["workflow_id"], "5eb69e19-3e13-45e1-ac27-8858bb1f2a36"); + // No field may be absent — that is what produced `undefined` in the UI. + for k in ["token", "status", "run_id", "workflow_id"] { + assert!(v.get(k).is_some(), "missing key {k}"); + } + } + + #[test] + fn approval_response_parses_denied() { + let msg = r#"response:{"status":"denied","run_id":"r1","workflow_id":"w1"}"#; + let v = approval_action_response(HASH, "denied", msg); + assert_eq!(v["status"], "denied"); + assert_eq!(v["run_id"], "r1"); + } + + #[test] + fn unparseable_message_falls_back_without_inventing_ids() { + // The duplicate-event short-circuit carries no JSON body. The status + // falls back, but ids stay empty rather than being fabricated. + let v = approval_action_response(HASH, "granted", "duplicate: already processed"); + assert_eq!(v["status"], "granted"); + assert_eq!(v["run_id"], ""); + assert_eq!(v["workflow_id"], ""); + } +} diff --git a/desktop/src-tauri/src/events.rs b/desktop/src-tauri/src/events.rs index 777d56d02e..970ec1e2e1 100644 --- a/desktop/src-tauri/src/events.rs +++ b/desktop/src-tauri/src/events.rs @@ -832,15 +832,15 @@ pub fn build_workflow_trigger(workflow_id: &str) -> Result Ok(EventBuilder::new(Kind::Custom(46020), "").tags(tags)) } -/// Kind 46030 — grant an approval token (with optional note). -pub fn build_approval_grant(token: &str, note: Option<&str>) -> Result { - let tags = vec![tag(vec!["t", token])?]; +/// Kind 46030: grant using `d=` with an optional note. +pub fn build_approval_grant(token_hash_hex: &str, note: Option<&str>) -> Result { + let tags = vec![tag(vec!["d", token_hash_hex])?]; Ok(EventBuilder::new(Kind::Custom(46030), note.unwrap_or("")).tags(tags)) } -/// Kind 46031 — deny an approval token (with optional note). -pub fn build_approval_deny(token: &str, note: Option<&str>) -> Result { - let tags = vec![tag(vec!["t", token])?]; +/// Kind 46031: deny using the same hashed `d`-tag contract. +pub fn build_approval_deny(token_hash_hex: &str, note: Option<&str>) -> Result { + let tags = vec![tag(vec!["d", token_hash_hex])?]; Ok(EventBuilder::new(Kind::Custom(46031), note.unwrap_or("")).tags(tags)) } diff --git a/desktop/src-tauri/src/relay.rs b/desktop/src-tauri/src/relay.rs index 71aa21c413..e0b2a548a1 100644 --- a/desktop/src-tauri/src/relay.rs +++ b/desktop/src-tauri/src/relay.rs @@ -309,6 +309,25 @@ pub async fn query_relay( query_relay_at(state, &relay_api_base_url_with_override(state), filters).await } +/// NIP-98 authenticated `GET` against a relay REST path, returning parsed JSON. +/// `path` includes its query string because the relay verifies the full NIP-98 +/// `u` value. Workflow DB rows cannot be fetched with a Nostr `/query` filter. +pub async fn get_relay_json( + state: &AppState, + path: &str, +) -> Result { + crate::relay_admission::wait_for_rate_limit().await; + let url = format!("{}{}", relay_api_base_url_with_override(state), path); + let auth = build_nip98_auth_header(&Method::GET, &url, &[], state)?; + + let response = state.http_client.get(&url).header("Authorization", auth).send().await + .map_err(|e| classify_request_error(&e))?; + if !response.status().is_success() { + return Err(relay_error_message(response).await); + } + parse_json_response(response).await +} + /// Like [`query_relay`] but targets an explicit HTTP API base URL instead of /// the workspace override. Used when a query must hit a specific relay (e.g. /// reconciling an agent's profile on the relay where it was published). diff --git a/desktop/src/features/workflows/ui/WorkflowApprovalCard.tsx b/desktop/src/features/workflows/ui/WorkflowApprovalCard.tsx index 8c39b01f59..9502e86012 100644 --- a/desktop/src/features/workflows/ui/WorkflowApprovalCard.tsx +++ b/desktop/src/features/workflows/ui/WorkflowApprovalCard.tsx @@ -26,6 +26,30 @@ export function WorkflowApprovalCard({ approval }: WorkflowApprovalCardProps) { data-testid="workflow-approval-card" >

Approval Required

+ + {/* + The exact package being approved. Approving from a card that shows only + approver and expiry is approving blind, which for a gate that authorises + an external send defeats the point of the gate. Read-only and + pre-wrapped so the reviewed text is byte-for-byte what was submitted. + */} + {approval.requestMessage ? ( + 
+          {approval.requestMessage}
+
+ ) : ( +

+ No request package was recorded for this gate, so there is nothing to + review. Approval is disabled. Deny it and re-run the workflow. +

+ )} +

Approver: {approval.approverSpec}

@@ -44,7 +68,9 @@ export function WorkflowApprovalCard({ approval }: WorkflowApprovalCardProps) {