diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 28d80a884fd..1c17a8d85d5 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -707,6 +707,14 @@ jobs: --run-ignored ignored-only env: BUZZ_TEST_DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz + - name: Workflow wake lifecycle PostgreSQL tests + run: | + cargo nextest run \ + --archive-file target/ci/backend-integration-tests.tar.zst \ + -E 'package(buzz-relay) and test(/^workflow_sink::/)' \ + --run-ignored ignored-only + env: + DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz - name: Database pressure observability PostgreSQL tests # Explicit pool acquisition and advisory-lock metrics require real # Postgres and are ignored by the infrastructure-free unit-test job. diff --git a/Cargo.lock b/Cargo.lock index d436016fd62..34f13b0a874 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -846,6 +846,7 @@ dependencies = [ "rustls", "serde", "serde_json", + "serde_yaml", "sha2 0.11.0", "thiserror 2.0.18", "tokio", diff --git a/crates/buzz-acp/Cargo.toml b/crates/buzz-acp/Cargo.toml index d047849806f..6dffb85bc19 100644 --- a/crates/buzz-acp/Cargo.toml +++ b/crates/buzz-acp/Cargo.toml @@ -41,6 +41,7 @@ reqwest = { workspace = true } # Serialization serde = { workspace = true } serde_json = { workspace = true } +serde_yaml = { workspace = true } # IDs uuid = { workspace = true } diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index 2d7b2128320..fdf88597cf2 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -1278,6 +1278,7 @@ pub fn resolve_channel_filters( ) -> HashMap { use buzz_core::kind::{ KIND_STREAM_MESSAGE, KIND_STREAM_REMINDER, KIND_WORKFLOW_APPROVAL_REQUESTED, + KIND_WORKFLOW_MENTION_WAKE, }; let target_channels: Vec = if let Some(ref overrides) = config.channels_override { @@ -1297,6 +1298,7 @@ pub fn resolve_channel_filters( let kinds = config.kinds_override.clone().unwrap_or_else(|| { vec![ KIND_STREAM_MESSAGE, + KIND_WORKFLOW_MENTION_WAKE, KIND_WORKFLOW_APPROVAL_REQUESTED, KIND_STREAM_REMINDER, ] @@ -1380,6 +1382,7 @@ pub fn resolve_dynamic_channel_filter( ) -> Option { use buzz_core::kind::{ KIND_STREAM_MESSAGE, KIND_STREAM_REMINDER, KIND_WORKFLOW_APPROVAL_REQUESTED, + KIND_WORKFLOW_MENTION_WAKE, }; // In Mentions/All mode, if the operator explicitly constrained channels @@ -1402,6 +1405,7 @@ pub fn resolve_dynamic_channel_filter( kinds: Some(config.kinds_override.clone().unwrap_or_else(|| { vec![ KIND_STREAM_MESSAGE, + KIND_WORKFLOW_MENTION_WAKE, KIND_WORKFLOW_APPROVAL_REQUESTED, KIND_STREAM_REMINDER, ] @@ -1549,13 +1553,37 @@ mod tests { for ch in &channels { let f = result.get(ch).expect("channel should be present"); assert!(f.require_mention, "mentions mode requires mention"); - let kinds = f.kinds.as_ref().expect("should have kinds"); - assert!(kinds.contains(&buzz_core::kind::KIND_STREAM_MESSAGE)); - assert!(kinds.contains(&buzz_core::kind::KIND_WORKFLOW_APPROVAL_REQUESTED)); - assert!(kinds.contains(&buzz_core::kind::KIND_STREAM_REMINDER)); + assert_eq!( + f.kinds, + Some(vec![ + buzz_core::kind::KIND_STREAM_MESSAGE, + buzz_core::kind::KIND_WORKFLOW_MENTION_WAKE, + buzz_core::kind::KIND_WORKFLOW_APPROVAL_REQUESTED, + buzz_core::kind::KIND_STREAM_REMINDER, + ]) + ); } } + #[test] + fn test_mentions_mode_dynamic_default_kinds_include_workflow_wake() { + let config = test_config(SubscribeMode::Mentions); + let channel = Uuid::new_v4(); + let filter = resolve_dynamic_channel_filter(&config, channel, &[]) + .expect("dynamic channel should be subscribed"); + + assert!(filter.require_mention); + assert_eq!( + filter.kinds, + Some(vec![ + buzz_core::kind::KIND_STREAM_MESSAGE, + buzz_core::kind::KIND_WORKFLOW_MENTION_WAKE, + buzz_core::kind::KIND_WORKFLOW_APPROVAL_REQUESTED, + buzz_core::kind::KIND_STREAM_REMINDER, + ]) + ); + } + #[test] fn test_mentions_mode_custom_kinds() { let mut config = test_config(SubscribeMode::Mentions); diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 25c6e549052..24575d5b835 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -13,6 +13,7 @@ mod queue; mod relay; mod setup_mode; mod usage; +mod workflow_wake; pub use usage::TurnUsage; @@ -2017,6 +2018,12 @@ async fn tokio_main() -> Result<()> { tracing::warn!("failed to set startup watermark: {e}"); } + let workflow_relay_pubkey = relay + .rest_client() + .relay_signing_pubkey() + .await + .map_err(|e| anyhow::anyhow!("relay signing identity error: {e}"))?; + tracing::info!("connected to relay at {}", config.relay_url); relay @@ -2108,6 +2115,7 @@ async fn tokio_main() -> Result<()> { kinds: config.kinds_override.clone().unwrap_or_else(|| { vec![ KIND_STREAM_MESSAGE, + buzz_core::kind::KIND_WORKFLOW_MENTION_WAKE, KIND_WORKFLOW_APPROVAL_REQUESTED, KIND_STREAM_REMINDER, ] @@ -2631,6 +2639,79 @@ async fn tokio_main() -> Result<()> { match buzz_event { Some(buzz_event) => { let kind_u32 = buzz_event.event.kind.as_u16() as u32; + if workflow_wake::requires_verified_wake( + &buzz_event.event, + workflow_relay_pubkey, + ) { + continue; + } + + let (buzz_event, admission_author_override) = if kind_u32 + == buzz_core::kind::KIND_WORKFLOW_MENTION_WAKE + { + let Some(wake) = workflow_wake::authenticate( + &buzz_event.event, + workflow_relay_pubkey, + ) else { + tracing::warn!("workflow wake authentication failed"); + continue; + }; + let authority = match ctx + .rest_client + .workflow_wake_authority(wake.run_id(), &wake.message_event_id()) + .await + { + Ok(authority) => authority, + Err(error) if error.is_transient() => { + // HTTP-status failures exhaust bounded retries; body + // interruptions also return transient after pacing. + // Transport dedup recorded this relay-signed wake, but + // dispatch has not occurred. Re-admit it for filtered + // replay rather than losing it or bypassing verification. + if let Err(replay_error) = relay + .replay_event( + buzz_event.channel_id, + buzz_event.event.id.to_hex(), + buzz_event.event.created_at.as_secs(), + ) + .await + { + tracing::warn!( + %replay_error, + "failed to arrange workflow wake authority replay" + ); + } + tracing::warn!(%error, "workflow wake authority unavailable; replay queued"); + continue; + } + Err(error) => { + // 403/404 and malformed authority bundles are terminal: + // replays cannot make a rejected or invalid authority safe. + tracing::warn!(%error, "workflow wake authority rejected"); + continue; + } + }; + let Some((message, signed_author)) = workflow_wake::verify( + &buzz_event.event, + authority, + workflow_relay_pubkey, + config.keys.public_key(), + buzz_event.channel_id, + ) else { + tracing::warn!("workflow wake authority verification failed"); + continue; + }; + ( + relay::BuzzEvent { + channel_id: buzz_event.channel_id, + event: message, + }, + Some(signed_author), + ) + } else { + (buzz_event, None) + }; + let kind_u32 = buzz_event.event.kind.as_u16() as u32; if kind_u32 == KIND_MEMBER_ADDED_NOTIFICATION || kind_u32 == KIND_MEMBER_REMOVED_NOTIFICATION @@ -2868,7 +2949,8 @@ async fn tokio_main() -> Result<()> { // explicit pubkey list on top, for external people; // it never revokes same-owner team bots. { - let author = buzz_event.event.pubkey.to_hex(); + let author = admission_author_override + .unwrap_or_else(|| buzz_event.event.pubkey.to_hex()); // DM hardening: resolve channel type (fail-closed // to DM) so allowlist/anyone modes cannot be // exercised by non-owner authors inside DMs. diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index 6188e57a11d..d06caef07b4 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -359,7 +359,7 @@ impl RestClient { Ok(resp) if is_retriable_status(resp.status()) => { let status = resp.status(); tracing::warn!("{method} {path} returned retriable HTTP {status}"); - last_err = Some(RelayError::Http(format!( + last_err = Some(RelayError::TransientHttp(format!( "{method} {path} returned HTTP {status}" ))); } @@ -372,14 +372,15 @@ impl RestClient { } Err(e) if e.is_timeout() || e.is_connect() => { tracing::warn!("{method} {path} network error: {e}"); - last_err = Some(RelayError::Http(e.to_string())); + last_err = Some(RelayError::TransientHttp(e.to_string())); } Err(e) => return Err(RelayError::Http(e.to_string())), } } - Err(last_err - .unwrap_or_else(|| RelayError::Http(format!("{method} {path} failed after retries")))) + Err(last_err.unwrap_or_else(|| { + RelayError::TransientHttp(format!("{method} {path} failed after retries")) + })) } /// POST with NIP-98 auth and retry. Re-signs on each attempt. @@ -410,6 +411,65 @@ impl RestClient { .await } + /// Fetch the relay's advertised signing identity from NIP-11 `/info`. + pub async fn relay_signing_pubkey(&self) -> Result { + let url = format!("{}/info", self.base_url.trim_end_matches('/')); + let value: Value = self + .http + .get(&url) + .header("Accept", "application/nostr+json") + .send() + .await + .map_err(|error| RelayError::Http(error.to_string()))? + .json() + .await + .map_err(|error| RelayError::Http(error.to_string()))?; + let relay_self = value + .get("self") + .and_then(Value::as_str) + .ok_or_else(|| RelayError::Http("relay did not advertise a signing identity".into()))?; + nostr::PublicKey::from_hex(relay_self) + .map_err(|error| RelayError::Http(format!("invalid relay signing identity: {error}"))) + } + + async fn bridge_get(&self, path: &str) -> Result { + let url = format!("{}{}", self.base_url, path); + let auth_tag_header = self.auth_tag_json.clone(); + self.request_with_retry("GET", path, || { + let auth = self.nip98_header("GET", &url, None).unwrap_or_default(); + let mut request = self.http.get(&url).header("Authorization", auth); + if let Some(ref tag) = auth_tag_header { + request = request.header("x-auth-tag", tag); + } + request.send() + }) + .await + } + + /// Fetch one exact workflow-wake authority bundle. + pub async fn workflow_wake_authority( + &self, + run_id: uuid::Uuid, + message_id: &nostr::EventId, + ) -> Result { + let path = format!("/workflow-wakes/{run_id}/{}", message_id.to_hex()); + // A successful status is not a complete authority response. Read the + // body separately so an interrupted transfer remains recoverable while + // a complete but malformed authority document stays terminal. + let response = self.bridge_get(&path).await?; + let body = match response.bytes().await { + Ok(body) => body, + Err(error) => { + // Replay retries this read through the normal verification path. + // Pace body failures too: headers may have arrived immediately, + // bypassing request_with_retry's backoff entirely. + tokio::time::sleep(jittered_duration(REST_RETRY_BASE_DELAYS[0])).await; + return Err(RelayError::TransientHttp(error.to_string())); + } + }; + serde_json::from_slice(&body).map_err(|error| RelayError::Http(error.to_string())) + } + /// Query events via the HTTP bridge: `POST /query` with NIP-98 auth. /// /// Accepts a slice of `nostr::Filter` (serialized as JSON array). @@ -545,10 +605,24 @@ pub enum RelayError { #[error("HTTP error: {0}")] Http(String), + /// A request exhausted its bounded retry budget after only transient + /// failures. Callers may safely schedule delayed recovery; all other HTTP + /// errors, including 403/404 and malformed bodies, are terminal. + #[error("transient HTTP error: {0}")] + TransientHttp(String), + #[error("Unexpected message: {0}")] UnexpectedMessage(String), } +impl RelayError { + /// Whether retry exhaustion, rather than an authority denial or malformed + /// response, caused this failure. + pub fn is_transient(&self) -> bool { + matches!(self, Self::TransientHttp(_)) + } +} + impl From for RelayError { fn from(e: nostr::event::builder::Error) -> Self { RelayError::AuthFailed(e.to_string()) @@ -609,6 +683,15 @@ enum RelayCommand { PublishEvent { event: Box }, /// Floor `since` for membership notification replay; events before startup are never re-delivered. SetStartupWatermark { ts: u64 }, + /// Re-admit an event which reached the harness but could not be safely + /// processed. Removing its transport dedup entry and replaying its channel + /// lets a transient harness-side dependency failure recover without + /// dispatching an unverified event. + ReplayEvent { + channel_id: Uuid, + event_id: String, + created_at: u64, + }, } type WsStream = WebSocketStream>; @@ -902,6 +985,27 @@ impl HarnessRelay { self.event_rx.recv().await.flatten() } + /// Arrange replay of an event that failed harness-side admission. + /// + /// This is intentionally narrower than general event retry: it preserves + /// the subscription's exact filter and reuses transport dedup/replay rather + /// than manufacturing a local event or bypassing verification. + pub async fn replay_event( + &self, + channel_id: Uuid, + event_id: String, + created_at: u64, + ) -> Result<(), RelayError> { + self.cmd_tx + .send(RelayCommand::ReplayEvent { + channel_id, + event_id, + created_at, + }) + .await + .map_err(|_| RelayError::ConnectionClosed) + } + /// Publish a signed event to the relay via the background WebSocket task. /// /// Blocks until the command channel has capacity. For ephemeral events @@ -1216,6 +1320,19 @@ impl BgState { } } + /// Undo transport admission for an event whose harness-side verification + /// dependency failed. The event was never dispatched, so it must become + /// eligible for the existing replay path. The timestamp is retained as a + /// replay floor even though `last_seen` already advanced when it arrived. + fn replay_event(&mut self, channel_id: Uuid, event_id: String, created_at: u64) { + self.seen_ids.remove(&event_id); + self.channel_dropped_since + .entry(channel_id) + .and_modify(|since| *since = (*since).min(created_at)) + .or_insert(created_at); + self.proactive_resubscribe_needed = true; + } + /// Clear all per-channel state for a channel that is being unsubscribed. /// Prevents stale replay on re-subscribe and avoids unbounded state growth /// for channels that are removed and never re-added. @@ -1366,6 +1483,11 @@ fn apply_command_to_state(state: &mut BgState, cmd: RelayCommand) { state.membership_last_seen = Some(ts); } } + RelayCommand::ReplayEvent { + channel_id, + event_id, + created_at, + } => state.replay_event(channel_id, event_id, created_at), // Observer telemetry frames are durable: park them (bounded, visible // overflow) so they are delivered by the post-reconnect drain. Other // ephemeral publishes (typing indicators) are meaningless while @@ -1604,6 +1726,14 @@ async fn execute_connected_command( debug!("startup watermark set to {ts}"); true } + RelayCommand::ReplayEvent { + channel_id, + event_id, + created_at, + } => { + state.replay_event(channel_id, event_id, created_at); + true + } // Control-flow commands — callers handle these before dispatching. RelayCommand::Shutdown | RelayCommand::Reconnect => { debug_assert!( @@ -3246,33 +3376,68 @@ async fn wait_for_reconnect( /// history. On reconnect (`since` is `Some`) subtracts [`SINCE_SKEW_SECS`]. /// /// Returns `true` if the REQ was successfully written to the WebSocket. -async fn send_subscribe( - ws: &mut WsStream, - _state: &BgState, +fn build_channel_req( + sub_id: &str, channel_id: Uuid, agent_pubkey_hex: &str, - since: Option, + since_ts: u64, filter: &ChannelFilter, -) -> bool { - let sub_id = channel_sub_id(channel_id); - +) -> Value { let mut req_filter = serde_json::Map::new(); - // kinds — omit entirely for wildcard subscriptions. - if let Some(ref kinds) = filter.kinds { + // The recipient-gated wake kind always gets its own exact #p filter. This + // preserves `--no-mention-filter` for ordinary channel events without + // weakening wake recipient gating or causing the relay to reject the mixed + // subscription. + let wake_kind = buzz_core::kind::KIND_WORKFLOW_MENTION_WAKE; + let includes_wake = filter + .kinds + .as_ref() + .is_some_and(|kinds| kinds.contains(&wake_kind)); + let normal_kinds = filter.kinds.as_ref().map(|kinds| { + kinds + .iter() + .copied() + .filter(|kind| *kind != wake_kind) + .collect::>() + }); + + if let Some(kinds) = normal_kinds.as_ref().filter(|kinds| !kinds.is_empty()) { req_filter.insert("kinds".into(), json!(kinds)); } - - // #h — always present (channel scope). req_filter.insert("#h".into(), json!([channel_id.to_string()])); - - // #p — only when require_mention is true. if filter.require_mention { req_filter.insert("#p".into(), json!([agent_pubkey_hex])); } + req_filter.insert("since".into(), json!(since_ts)); + + let mut req_filters = Vec::new(); + if normal_kinds.as_ref().is_none_or(|kinds| !kinds.is_empty()) { + req_filters.push(Value::Object(req_filter)); + } + if includes_wake { + let mut wake_filter = serde_json::Map::new(); + wake_filter.insert("kinds".into(), json!([wake_kind])); + wake_filter.insert("#h".into(), json!([channel_id.to_string()])); + wake_filter.insert("#p".into(), json!([agent_pubkey_hex])); + wake_filter.insert("since".into(), json!(since_ts)); + req_filters.push(Value::Object(wake_filter)); + } + + let mut req = vec![json!("REQ"), json!(sub_id)]; + req.extend(req_filters); + Value::Array(req) +} - // since — on first subscribe use current time to skip history; on reconnect - // subtract skew buffer to catch events missed during the disconnect window. +async fn send_subscribe( + ws: &mut WsStream, + _state: &BgState, + channel_id: Uuid, + agent_pubkey_hex: &str, + since: Option, + filter: &ChannelFilter, +) -> bool { + let sub_id = channel_sub_id(channel_id); let since_ts = match since { Some(ts) => ts.saturating_sub(SINCE_SKEW_SECS), None => std::time::SystemTime::now() @@ -3280,9 +3445,7 @@ async fn send_subscribe( .unwrap_or_default() .as_secs(), }; - req_filter.insert("since".into(), json!(since_ts)); - - let req = json!(["REQ", sub_id, Value::Object(req_filter)]); + let req = build_channel_req(&sub_id, channel_id, agent_pubkey_hex, since_ts, filter); match serde_json::to_string(&req) { Ok(text) => { @@ -3746,6 +3909,7 @@ pub(crate) fn parse_relay_message(text: &str) -> Result bool { match err { RelayError::Http(_) | RelayError::Json(_) | RelayError::UnexpectedMessage(_) => true, + RelayError::TransientHttp(_) => false, RelayError::WebSocket(e) => is_terminal_ws_error(e.as_ref()), RelayError::AuthFailed(message) => is_terminal_auth_failure(message), RelayError::NoAuthChallenge | RelayError::ConnectionClosed | RelayError::Timeout => false, @@ -4084,6 +4248,104 @@ async fn wait_for_any_ok( mod tests { use super::*; + #[test] + fn default_mentions_builds_complete_recipient_gated_subscription_shape() { + let channel = Uuid::new_v4(); + let agent = nostr::Keys::generate().public_key().to_hex(); + let req = build_channel_req( + "sub", + channel, + &agent, + 123, + &ChannelFilter { + kinds: Some(vec![ + buzz_core::kind::KIND_STREAM_MESSAGE, + buzz_core::kind::KIND_WORKFLOW_MENTION_WAKE, + buzz_core::kind::KIND_WORKFLOW_APPROVAL_REQUESTED, + buzz_core::kind::KIND_STREAM_REMINDER, + ]), + require_mention: true, + }, + ); + let req = req.as_array().expect("REQ array"); + + assert_eq!(req.len(), 4); + assert_eq!( + req[2]["kinds"], + json!([ + buzz_core::kind::KIND_STREAM_MESSAGE, + buzz_core::kind::KIND_WORKFLOW_APPROVAL_REQUESTED, + buzz_core::kind::KIND_STREAM_REMINDER, + ]) + ); + assert_eq!(req[2]["#p"], json!([agent])); + assert_eq!(req[2]["#h"], json!([channel.to_string()])); + assert_eq!( + req[3]["kinds"], + json!([buzz_core::kind::KIND_WORKFLOW_MENTION_WAKE]) + ); + assert_eq!(req[3]["#p"], json!([agent])); + assert_eq!(req[3]["#h"], json!([channel.to_string()])); + } + + #[test] + fn durable_workflow_wake_is_requested_on_reconnect() { + let channel = Uuid::new_v4(); + let agent = nostr::Keys::generate().public_key().to_hex(); + let req = build_channel_req( + "sub", + channel, + &agent, + 456, + &ChannelFilter { + kinds: Some(vec![buzz_core::kind::KIND_WORKFLOW_MENTION_WAKE]), + require_mention: true, + }, + ); + let req = req.as_array().expect("REQ array"); + + assert_eq!(req.len(), 3); + assert_eq!( + req[2]["kinds"], + json!([buzz_core::kind::KIND_WORKFLOW_MENTION_WAKE]) + ); + assert_eq!(req[2]["#p"], json!([agent])); + assert_eq!(req[2]["#h"], json!([channel.to_string()])); + assert_eq!(req[2]["since"], json!(456)); + } + + #[test] + fn workflow_wake_uses_exact_recipient_filter_when_mentions_are_disabled() { + let channel = Uuid::new_v4(); + let agent = nostr::Keys::generate().public_key().to_hex(); + let req = build_channel_req( + "sub", + channel, + &agent, + 123, + &ChannelFilter { + kinds: Some(vec![ + buzz_core::kind::KIND_STREAM_MESSAGE, + buzz_core::kind::KIND_WORKFLOW_MENTION_WAKE, + ]), + require_mention: false, + }, + ); + let filters = req.as_array().expect("REQ array"); + assert_eq!(filters.len(), 4); + assert_eq!( + filters[2]["kinds"], + json!([buzz_core::kind::KIND_STREAM_MESSAGE]) + ); + assert!(filters[2].get("#p").is_none()); + assert_eq!( + filters[3]["kinds"], + json!([buzz_core::kind::KIND_WORKFLOW_MENTION_WAKE]) + ); + assert_eq!(filters[3]["#p"], json!([agent])); + assert_eq!(filters[3]["#h"], json!([channel.to_string()])); + } + #[test] fn relay_ws_to_http_plain() { assert_eq!( @@ -4438,6 +4700,10 @@ mod tests { assert!(result.is_err()); } + mod workflow_wake_recovery_tests { + include!("workflow_wake_recovery_tests.rs"); + } + #[test] fn subscription_id_starts_with_ch_prefix() { let uuid = Uuid::new_v4(); @@ -4874,6 +5140,33 @@ mod tests { ); } + #[test] + fn workflow_wake_authority_failure_reopens_transport_dedup_for_replay() { + let mut state = BgState::new(); + let channel_id = Uuid::new_v4(); + let keys = nostr::Keys::generate(); + let wake = make_test_event(&keys, 1_000); + let event_id = wake.id.to_hex(); + + // Normal transport delivery claims the ID and advances its watermark. + assert!(state.record_event(channel_id, &wake)); + assert!(!state.record_event(channel_id, &wake)); + + // A failure before authenticated authority verification must not lose + // the wake: make its exact ID eligible and replay from its timestamp. + state.replay_event(channel_id, event_id.clone(), wake.created_at.as_secs()); + assert!(!state.seen_ids.contains(&event_id)); + assert_eq!( + state.channel_since(&channel_id), + Some(wake.created_at.as_secs()) + ); + assert!(state.proactive_resubscribe_needed); + assert!(state.record_event(channel_id, &wake)); + + // Once replayed, ordinary dedup resumes; this does not admit duplicates. + assert!(!state.record_event(channel_id, &wake)); + } + /// Test 8: channel_dropped_since records the OLDEST dropped timestamp. /// /// Simulates the backpressure path directly on BgState: @@ -5212,6 +5505,11 @@ mod tests { let cases: Vec<(&str, RelayError, bool)> = vec![ // ── outer RelayError variants ── ("Http: bad URL", RelayError::Http("bad url".into()), true), + ( + "TransientHttp: exhausted retry budget", + RelayError::TransientHttp("timeout".into()), + false, + ), ( "Json: malformed relay frame", RelayError::Json(serde_json::from_str::<()>("not json").unwrap_err()), diff --git a/crates/buzz-acp/src/workflow_wake.rs b/crates/buzz-acp/src/workflow_wake.rs new file mode 100644 index 00000000000..9908e33b804 --- /dev/null +++ b/crates/buzz-acp/src/workflow_wake.rs @@ -0,0 +1,485 @@ +//! Fail-closed verification of relay-signed workflow mention wakes. + +use buzz_core::kind::{KIND_STREAM_MESSAGE, KIND_WORKFLOW_DEF}; +use buzz_core::workflow_wake::WorkflowMentionWake; +use nostr::{Event, PublicKey}; +use serde::Deserialize; +use uuid::Uuid; + +#[derive(Debug, Deserialize)] +struct WorkflowAuthority { + steps: Vec, +} + +#[derive(Debug, Deserialize)] +struct WorkflowAuthorityStep { + id: String, + action: String, + #[serde(default)] + channel: Option, +} + +/// Exact public authority bundle returned by the authenticated relay read. +#[derive(Debug, Deserialize)] +pub struct WorkflowWakeAuthority { + /// Exact run ID. + pub run_id: Uuid, + /// Exact workflow channel. + pub channel_id: Uuid, + /// Workflow ID named by the signed definition. + pub workflow_id: Uuid, + /// Exact signed definition revision ID. + pub definition_event_id: String, + /// Workflow owner authenticated against relay workflow state. + pub workflow_owner: String, + /// Owner-signed workflow definition. + pub definition: Event, + /// Relay-signed visible message. + pub message: Event, +} + +/// Return whether a relay-signed workflow message must be dispatched only +/// through its separately verified wake. +pub fn requires_verified_wake(event: &Event, relay_pubkey: PublicKey) -> bool { + event.pubkey == relay_pubkey + && event.kind.as_u16() as u32 == KIND_STREAM_MESSAGE + && single_tag(event, "workflow-run").is_some() + && single_tag(event, "workflow-definition").is_some() + && single_tag(event, "workflow-step").is_some() +} + +/// Authenticate and parse a relay-signed workflow wake before any authority lookup. +pub fn authenticate(wake_event: &Event, relay_pubkey: PublicKey) -> Option { + if wake_event.pubkey != relay_pubkey || wake_event.verify().is_err() { + return None; + } + WorkflowMentionWake::parse(wake_event).ok() +} + +/// Verify every authority edge and return the visible message plus its signed author principal. +pub fn verify( + wake_event: &Event, + authority: WorkflowWakeAuthority, + relay_pubkey: PublicKey, + agent_pubkey: PublicKey, + subscription_channel: Uuid, +) -> Option<(Event, String)> { + let wake = authenticate(wake_event, relay_pubkey)?; + if wake.recipient() != agent_pubkey + || authority.workflow_owner != authority.definition.pubkey.to_hex() + || wake.run_id() != authority.run_id + || wake.channel_id() != subscription_channel + || wake.channel_id() != authority.channel_id + || wake.definition_event_id().to_hex() != authority.definition_event_id + || wake.message_event_id() != authority.message.id + { + return None; + } + + let definition = authority.definition; + if definition.verify().is_err() + || definition.kind.as_u16() as u32 != KIND_WORKFLOW_DEF + || definition.id != wake.definition_event_id() + || !exact_tag(&definition, "d", &authority.workflow_id.to_string()) + { + return None; + } + let channel = single_tag(&definition, "h")?; + if channel != authority.channel_id.to_string() { + return None; + } + let message = authority.message; + if message.verify().is_err() + || message.pubkey != relay_pubkey + || message.kind.as_u16() as u32 != KIND_STREAM_MESSAGE + || !exact_tag(&message, "h", channel) + || !contains_tag(&message, "p", &agent_pubkey.to_hex()) + || !exact_tag(&message, "workflow-run", &authority.run_id.to_string()) + || !exact_tag(&message, "workflow-definition", &definition.id.to_hex()) + { + return None; + } + let step_id = single_tag(&message, "workflow-step")?; + let workflow: WorkflowAuthority = serde_yaml::from_str(&definition.content).ok()?; + let step = workflow.steps.iter().find(|step| step.id == step_id)?; + if step.action != "send_message" { + return None; + } + if let Some(target) = step + .channel + .as_deref() + .map(str::trim) + .filter(|value| !value.is_empty()) + { + // The endpoint's relay-signed message is the authority for a resolved + // template target. A definition stores templates, while its execution + // resolves them from per-run state unavailable to ACP; comparing raw + // text would reject valid targets. Literal targets remain an independent + // constraint and are compared as UUIDs so noncanonical spelling works. + if !target.contains("{{") { + let target = Uuid::parse_str(target).ok()?; + let message_channel = Uuid::parse_str(channel).ok()?; + if target != message_channel { + return None; + } + } + } + Some((message, definition.pubkey.to_hex())) +} + +fn single_tag<'a>(event: &'a Event, name: &str) -> Option<&'a str> { + let mut matches = event.tags.iter().filter_map(|tag| { + let values = tag.as_slice(); + (values.len() == 2 && values[0] == name).then(|| values[1].as_str()) + }); + let value = matches.next()?; + matches.next().is_none().then_some(value) +} + +fn exact_tag(event: &Event, name: &str, value: &str) -> bool { + single_tag(event, name).is_some_and(|actual| actual.eq_ignore_ascii_case(value)) +} + +fn contains_tag(event: &Event, name: &str, value: &str) -> bool { + event.tags.iter().any(|tag| { + let values = tag.as_slice(); + values.len() == 2 && values[0] == name && values[1].eq_ignore_ascii_case(value) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use buzz_core::kind::KIND_WORKFLOW_MENTION_WAKE; + use nostr::{EventBuilder, Keys, Kind, Tag}; + + struct Fixture { + relay: Keys, + agent: Keys, + owner: Keys, + channel: Uuid, + run: Uuid, + workflow: Uuid, + definition: Event, + message: Event, + wake: Event, + } + + impl Fixture { + fn new(definition_content: &str, target_channel: Option) -> Self { + let relay = Keys::generate(); + let agent = Keys::generate(); + let owner = Keys::generate(); + let channel = Uuid::new_v4(); + let run = Uuid::new_v4(); + let workflow = Uuid::new_v4(); + let definition = EventBuilder::new( + Kind::Custom(KIND_WORKFLOW_DEF as u16), + definition_content + .replace("$CHANNEL", &target_channel.unwrap_or(channel).to_string()), + ) + .tags([ + Tag::parse(["d", &workflow.to_string()]).expect("d tag"), + Tag::parse(["h", &channel.to_string()]).expect("h tag"), + ]) + .sign_with_keys(&owner) + .expect("definition"); + let message = EventBuilder::new(Kind::Custom(KIND_STREAM_MESSAGE as u16), "do work") + .tags([ + Tag::parse(["h", &channel.to_string()]).expect("h tag"), + Tag::parse(["p", &agent.public_key().to_hex()]).expect("p tag"), + Tag::parse(["workflow-run", &run.to_string()]).expect("run tag"), + Tag::parse(["workflow-definition", &definition.id.to_hex()]) + .expect("definition tag"), + Tag::parse(["workflow-step", "notify"]).expect("step tag"), + ]) + .sign_with_keys(&relay) + .expect("message"); + let wake = WorkflowMentionWake::new( + agent.public_key(), + channel, + run, + definition.id, + message.id, + ) + .sign(&relay) + .expect("wake"); + Self { + relay, + agent, + owner, + channel, + run, + workflow, + definition, + message, + wake, + } + } + + fn valid() -> Self { + Self::new( + "name: wake\ntrigger:\n on: message_posted\nsteps:\n - id: notify\n action: send_message\n text: do work\n channel: $CHANNEL\n", + None, + ) + } + + fn authority(&self) -> WorkflowWakeAuthority { + WorkflowWakeAuthority { + run_id: self.run, + channel_id: self.channel, + workflow_id: self.workflow, + definition_event_id: self.definition.id.to_hex(), + workflow_owner: self.owner.public_key().to_hex(), + definition: self.definition.clone(), + message: self.message.clone(), + } + } + + fn verify(&self, authority: WorkflowWakeAuthority) -> Option<(Event, String)> { + super::verify( + &self.wake, + authority, + self.relay.public_key(), + self.agent.public_key(), + self.channel, + ) + } + } + + #[test] + fn workflow_message_is_ineligible_for_direct_dispatch() { + let fixture = Fixture::valid(); + assert!(requires_verified_wake( + &fixture.message, + fixture.relay.public_key() + )); + assert!(!requires_verified_wake( + &fixture.message, + Keys::generate().public_key() + )); + + let ordinary = EventBuilder::new(Kind::Custom(KIND_STREAM_MESSAGE as u16), "ordinary") + .tags([ + Tag::parse(["h", &fixture.channel.to_string()]).expect("h tag"), + Tag::parse(["p", &fixture.agent.public_key().to_hex()]).expect("p tag"), + ]) + .sign_with_keys(&fixture.relay) + .expect("ordinary message"); + assert!(!requires_verified_wake( + &ordinary, + fixture.relay.public_key() + )); + } + + #[test] + fn rejects_forged_wake_before_authority_lookup() { + let fixture = Fixture::valid(); + let forged = WorkflowMentionWake::new( + fixture.agent.public_key(), + fixture.channel, + fixture.run, + fixture.definition.id, + fixture.message.id, + ) + .sign(&Keys::generate()) + .expect("forged wake"); + + assert!(authenticate(&forged, fixture.relay.public_key()).is_none()); + assert!(authenticate(&fixture.wake, fixture.relay.public_key()).is_some()); + } + + #[test] + fn accepts_exact_authority_and_returns_signed_owner() { + let fixture = Fixture::valid(); + let (message, author) = fixture.verify(fixture.authority()).expect("verified"); + assert_eq!(message.id, fixture.message.id); + assert_eq!(author, fixture.owner.public_key().to_hex()); + } + + #[test] + fn rejects_wrong_wake_signer_or_recipient() { + let fixture = Fixture::valid(); + assert!(super::verify( + &fixture.wake, + fixture.authority(), + Keys::generate().public_key(), + fixture.agent.public_key(), + fixture.channel, + ) + .is_none()); + assert!(super::verify( + &fixture.wake, + fixture.authority(), + fixture.relay.public_key(), + Keys::generate().public_key(), + fixture.channel, + ) + .is_none()); + } + + #[test] + fn rejects_mismatched_run_revision_message_channel_and_owner() { + let fixture = Fixture::valid(); + let mut authority = fixture.authority(); + authority.run_id = Uuid::new_v4(); + assert!(fixture.verify(authority).is_none()); + + let mut authority = fixture.authority(); + authority.definition_event_id = EventBuilder::text_note("other") + .sign_with_keys(&Keys::generate()) + .expect("event") + .id + .to_hex(); + assert!(fixture.verify(authority).is_none()); + + let mut authority = fixture.authority(); + authority.message = EventBuilder::text_note("other") + .sign_with_keys(&fixture.relay) + .expect("event"); + assert!(fixture.verify(authority).is_none()); + + let mut authority = fixture.authority(); + authority.channel_id = Uuid::new_v4(); + assert!(fixture.verify(authority).is_none()); + + let mut authority = fixture.authority(); + authority.workflow_owner = Keys::generate().public_key().to_hex(); + assert!(fixture.verify(authority).is_none()); + } + + #[test] + fn rejects_malformed_or_non_send_message_instruction() { + let malformed = Fixture::new("not: [valid", None); + assert!(malformed.verify(malformed.authority()).is_none()); + + let other_action = Fixture::new( + "name: wake\ntrigger:\n on: message_posted\nsteps:\n - id: notify\n action: add_reaction\n emoji: thumbsup\n", + None, + ); + assert!(other_action.verify(other_action.authority()).is_none()); + } + + #[test] + fn rejects_wrong_step_or_target_channel() { + let fixture = Fixture::valid(); + let wrong_step_message = + EventBuilder::new(Kind::Custom(KIND_STREAM_MESSAGE as u16), "do work") + .tags([ + Tag::parse(["h", &fixture.channel.to_string()]).expect("h tag"), + Tag::parse(["p", &fixture.agent.public_key().to_hex()]).expect("p tag"), + Tag::parse(["workflow-run", &fixture.run.to_string()]).expect("run tag"), + Tag::parse(["workflow-definition", &fixture.definition.id.to_hex()]) + .expect("definition tag"), + Tag::parse(["workflow-step", "missing"]).expect("step tag"), + ]) + .sign_with_keys(&fixture.relay) + .expect("message"); + let wrong_step_wake = WorkflowMentionWake::new( + fixture.agent.public_key(), + fixture.channel, + fixture.run, + fixture.definition.id, + wrong_step_message.id, + ) + .sign(&fixture.relay) + .expect("wake"); + let mut authority = fixture.authority(); + authority.message = wrong_step_message; + assert!(super::verify( + &wrong_step_wake, + authority, + fixture.relay.public_key(), + fixture.agent.public_key(), + fixture.channel, + ) + .is_none()); + + let wrong_target = Fixture::new( + "name: wake\ntrigger:\n on: message_posted\nsteps:\n - id: notify\n action: send_message\n text: do work\n channel: $CHANNEL\n", + Some(Uuid::new_v4()), + ); + assert!(wrong_target.verify(wrong_target.authority()).is_none()); + } + + #[test] + fn accepts_template_target_using_relay_signed_resolved_message_channel() { + let fixture = Fixture::new( + "name: wake\ntrigger:\n on: message_posted\nsteps:\n - id: notify\n action: send_message\n text: do work\n channel: '{{trigger.channel_id}}'\n", + None, + ); + assert!(fixture.verify(fixture.authority()).is_some()); + } + + #[test] + fn accepts_noncanonical_literal_uuid_target() { + let fixture = Fixture::new( + "name: wake\ntrigger:\n on: message_posted\nsteps:\n - id: notify\n action: send_message\n text: do work\n channel: $CHANNEL\n", + None, + ); + let noncanonical = fixture.channel.simple().to_string(); + let definition = EventBuilder::new( + Kind::Custom(KIND_WORKFLOW_DEF as u16), + format!("name: wake\ntrigger:\n on: message_posted\nsteps:\n - id: notify\n action: send_message\n text: do work\n channel: {noncanonical}\n"), + ) + .tags([ + Tag::parse(["d", &fixture.workflow.to_string()]).expect("d tag"), + Tag::parse(["h", &fixture.channel.to_string()]).expect("h tag"), + ]) + .sign_with_keys(&fixture.owner) + .expect("definition"); + let message = EventBuilder::new(Kind::Custom(KIND_STREAM_MESSAGE as u16), "do work") + .tags([ + Tag::parse(["h", &fixture.channel.to_string()]).expect("h tag"), + Tag::parse(["p", &fixture.agent.public_key().to_hex()]).expect("p tag"), + Tag::parse(["workflow-run", &fixture.run.to_string()]).expect("run tag"), + Tag::parse(["workflow-definition", &definition.id.to_hex()]) + .expect("definition tag"), + Tag::parse(["workflow-step", "notify"]).expect("step tag"), + ]) + .sign_with_keys(&fixture.relay) + .expect("message"); + let wake = WorkflowMentionWake::new( + fixture.agent.public_key(), + fixture.channel, + fixture.run, + definition.id, + message.id, + ) + .sign(&fixture.relay) + .expect("wake"); + let authority = WorkflowWakeAuthority { + definition_event_id: definition.id.to_hex(), + definition, + message, + ..fixture.authority() + }; + assert!(super::verify( + &wake, + authority, + fixture.relay.public_key(), + fixture.agent.public_key(), + fixture.channel, + ) + .is_some()); + } + + #[test] + fn malformed_literal_target_remains_rejected() { + let fixture = Fixture::new( + "name: wake\ntrigger:\n on: message_posted\nsteps:\n - id: notify\n action: send_message\n text: do work\n channel: not-a-channel\n", + None, + ); + assert!(fixture.verify(fixture.authority()).is_none()); + } + + #[test] + fn wake_kind_remains_identifier_only() { + let fixture = Fixture::valid(); + assert_eq!( + fixture.wake.kind.as_u16() as u32, + KIND_WORKFLOW_MENTION_WAKE + ); + assert!(fixture.wake.content.is_empty()); + } +} diff --git a/crates/buzz-acp/src/workflow_wake_recovery_tests.rs b/crates/buzz-acp/src/workflow_wake_recovery_tests.rs new file mode 100644 index 00000000000..82a54215e15 --- /dev/null +++ b/crates/buzz-acp/src/workflow_wake_recovery_tests.rs @@ -0,0 +1,263 @@ +// Exhausted HTTP authority retry through the real relay command/replay loop. +use super::*; +use buzz_core::kind::{KIND_STREAM_MESSAGE, KIND_WORKFLOW_DEF, KIND_WORKFLOW_MENTION_WAKE}; +use buzz_core::workflow_wake::WorkflowMentionWake; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +#[tokio::test] +async fn exhausted_authority_failure_replays_exact_wake_and_verifies_before_dispatch() { + assert_authority_recovery(AuthorityFailure::Status).await; +} + +#[tokio::test] +async fn truncated_authority_body_replays_before_dispatch() { + assert_authority_recovery(AuthorityFailure::TruncatedBody).await; +} + +#[tokio::test] +async fn stalled_authority_body_replays_before_dispatch() { + assert_authority_recovery(AuthorityFailure::StalledBody).await; +} + +#[derive(Clone, Copy)] +enum AuthorityFailure { + Status, + TruncatedBody, + StalledBody, +} + +async fn assert_authority_recovery(failure: AuthorityFailure) { + let agent = Keys::generate(); + let owner = Keys::generate(); + let relay_key = Keys::generate(); + let channel = Uuid::new_v4(); + let run = Uuid::new_v4(); + let workflow = Uuid::new_v4(); + let definition = EventBuilder::new(Kind::Custom(KIND_WORKFLOW_DEF as u16), + "name: wake\ntrigger:\n on: message_posted\nsteps:\n - id: notify\n action: send_message\n text: work\n") + .tags([Tag::parse(["d", &workflow.to_string()]).unwrap(), + Tag::parse(["h", &channel.to_string()]).unwrap()]) + .sign_with_keys(&owner).unwrap(); + let message = EventBuilder::new(Kind::Custom(KIND_STREAM_MESSAGE as u16), "work") + .tags([ + Tag::parse(["h", &channel.to_string()]).unwrap(), + Tag::public_key(agent.public_key()), + Tag::parse(["workflow-run", &run.to_string()]).unwrap(), + Tag::parse(["workflow-definition", &definition.id.to_hex()]).unwrap(), + Tag::parse(["workflow-step", "notify"]).unwrap(), + ]) + .sign_with_keys(&relay_key) + .unwrap(); + let wake = + WorkflowMentionWake::new(agent.public_key(), channel, run, definition.id, message.id) + .sign(&relay_key) + .unwrap(); + let body = json!({"run_id":run, "channel_id":channel, "workflow_id":workflow, + "definition_event_id":definition.id.to_hex(), "workflow_owner":owner.public_key().to_hex(), + "definition":definition, "message":message}) + .to_string(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + // Status failures exhaust the HTTP retry budget. Body failures arrive + // after successful headers and must independently reopen transport replay. + let failures = if matches!(failure, AuthorityFailure::Status) { + 4 + } else { + 1 + }; + let http_server = tokio::spawn(async move { + for index in 0..=failures { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut request = [0u8; 4096]; + let len = stream.read(&mut request).await.unwrap(); + let request = String::from_utf8_lossy(&request[..len]); + assert!(request.starts_with(&format!("GET /workflow-wakes/{run}/"))); + if index < failures { + match failure { + AuthorityFailure::Status => { + stream.write_all(b"HTTP/1.1 503 test\r\nContent-Length: 0\r\nConnection: close\r\n\r\n").await.unwrap(); + } + AuthorityFailure::TruncatedBody | AuthorityFailure::StalledBody => { + stream.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 1000\r\nConnection: close\r\n\r\n{").await.unwrap(); + if matches!(failure, AuthorityFailure::StalledBody) { + tokio::spawn(async move { + tokio::time::sleep(Duration::from_secs(5)).await; + drop(stream); + }); + } + } + } + } else { + stream.write_all(format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).as_bytes()).await.unwrap(); + } + } + }); + let http = reqwest::Client::builder() + .timeout(Duration::from_secs(1)) + .build() + .unwrap(); + let rest = RestClient { + http: http.clone(), + base_url: format!("http://{address}"), + keys: agent.clone(), + auth_tag_json: None, + }; + let (ws, mut server) = test_ws_pair().await; + let (event_tx, event_rx) = mpsc::channel(16); + let (observer_control_tx, observer_control_rx) = mpsc::channel(16); + let (cmd_tx, cmd_rx) = mpsc::channel(16); + let bg = tokio::spawn(run_background_task( + ws, + VecDeque::new(), + event_tx, + observer_control_tx, + cmd_rx, + agent.clone(), + "ws://unused".into(), + agent.public_key().to_hex(), + None, + )); + let mut harness = HarnessRelay { + event_rx, + observer_control_rx: Some(observer_control_rx), + cmd_tx, + http, + relay_url: "ws://unused".into(), + keys: agent.clone(), + auth_tag: None, + bg_handle: Some(bg), + }; + harness + .subscribe_channel_from( + channel, + ChannelFilter { + kinds: Some(vec![KIND_WORKFLOW_MENTION_WAKE]), + require_mention: false, + }, + Some(wake.created_at.as_secs()), + ) + .await + .unwrap(); + let initial = next_data_frame(&mut server).await; + let frame = json!(["EVENT", channel_sub_id(channel), wake]).to_string(); + server + .send(Message::Text(frame.clone().into())) + .await + .unwrap(); + let received = timeout(Duration::from_secs(2), harness.next_event()) + .await + .unwrap() + .unwrap(); + let authenticated = + crate::workflow_wake::authenticate(&received.event, relay_key.public_key()).unwrap(); + let error = rest + .workflow_wake_authority(authenticated.run_id(), &authenticated.message_event_id()) + .await + .expect_err("authority transfer fails"); + assert!(error.is_transient()); + assert!( + harness.event_rx.try_recv().is_err(), + "no fabricated event on lookup failure" + ); + harness + .replay_event( + channel, + received.event.id.to_hex(), + received.event.created_at.as_secs(), + ) + .await + .unwrap(); + let replay = next_data_frame(&mut server).await; + assert_eq!(replay[0], "REQ"); + assert_eq!(replay[1], initial[1]); + assert_eq!(replay[2]["kinds"], json!([KIND_WORKFLOW_MENTION_WAKE])); + assert_eq!(replay[2]["#h"], json!([channel.to_string()])); + assert_eq!(replay[2]["#p"], json!([agent.public_key().to_hex()])); + assert!(replay[2]["since"].as_u64().unwrap() <= wake.created_at.as_secs()); + server + .send(Message::Text(frame.clone().into())) + .await + .unwrap(); + let replayed = timeout(Duration::from_secs(2), harness.next_event()) + .await + .unwrap() + .unwrap(); + assert_eq!(replayed.event.id, wake.id); + let authority = rest + .workflow_wake_authority(run, &message.id) + .await + .unwrap(); + let (verified, principal) = crate::workflow_wake::verify( + &replayed.event, + authority, + relay_key.public_key(), + agent.public_key(), + channel, + ) + .expect("full authority verified"); + assert_eq!(verified.id, message.id); + assert_eq!(principal, owner.public_key().to_hex()); + server.send(Message::Text(frame.into())).await.unwrap(); + assert!( + timeout(Duration::from_millis(100), harness.next_event()) + .await + .is_err(), + "normal dedup resumes" + ); + http_server.await.unwrap(); + harness.shutdown().await; +} + +async fn next_data_frame(server: &mut WebSocketStream) -> Value { + timeout(Duration::from_secs(2), async { + loop { + match server.next().await.expect("websocket open").expect("frame") { + Message::Text(text) => return serde_json::from_str(&text).expect("JSON frame"), + Message::Ping(payload) => server.send(Message::Pong(payload)).await.expect("pong"), + other => panic!("unexpected frame {other:?}"), + } + } + }) + .await + .expect("data frame before timeout") +} + +#[tokio::test] +async fn complete_malformed_authority_body_is_terminal() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + for body in ["{", "{}"] { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut request = [0u8; 4096]; + assert!(stream.read(&mut request).await.unwrap() > 0); + stream + .write_all( + format!( + "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ) + .as_bytes(), + ) + .await + .unwrap(); + } + }); + let rest = RestClient { + http: reqwest::Client::new(), + base_url: format!("http://{address}"), + keys: Keys::generate(), + auth_tag_json: None, + }; + for _ in 0..2 { + let error = rest + .workflow_wake_authority(Uuid::new_v4(), &nostr::EventId::all_zeros()) + .await + .unwrap_err(); + assert!( + !error.is_transient(), + "complete malformed authority must not replay" + ); + } + server.await.unwrap(); +} diff --git a/crates/buzz-core/src/filter.rs b/crates/buzz-core/src/filter.rs index 32e3a7ad16b..8333fdb4acf 100644 --- a/crates/buzz-core/src/filter.rs +++ b/crates/buzz-core/src/filter.rs @@ -11,10 +11,10 @@ pub fn filters_match(filters: &[Filter], event: &StoredEvent) -> bool { filters.iter().any(|f| filter_match_one(f, event)) } -/// Result-level read authorization for relay-signed events whose content is -/// private to a single viewer. Currently gates `KIND_DM_VISIBILITY` and -/// `KIND_AGENT_TURN_METRIC`: the reader MUST equal the event's `#p` tag -/// (owner). Returns `true` for every other kind. +/// Result-level read authorization for events whose content or envelope is +/// private to a single viewer. Currently gates `KIND_DM_VISIBILITY`, +/// `KIND_AGENT_TURN_METRIC`, and `KIND_WORKFLOW_MENTION_WAKE`: the reader MUST +/// equal the event's `#p` tag (owner/recipient). Returns `true` for every other kind. /// /// This guards every delivery surface — WS historical pull (`req.rs`), HTTP /// bridge (`bridge.rs`), and live fan-out (`event.rs`) — so a query that @@ -22,7 +22,10 @@ pub fn filters_match(filters: &[Filter], event: &StoredEvent) -> bool { /// a known event id) still cannot read another user's private event. pub fn reader_authorized_for_event(event: &nostr::Event, reader_pubkey_hex: &str) -> bool { let kind = crate::kind::event_kind_u32(event); - if kind != crate::kind::KIND_DM_VISIBILITY && kind != crate::kind::KIND_AGENT_TURN_METRIC { + if kind != crate::kind::KIND_DM_VISIBILITY + && kind != crate::kind::KIND_AGENT_TURN_METRIC + && kind != crate::kind::KIND_WORKFLOW_MENTION_WAKE + { return true; } let p = nostr::SingleLetterTag::lowercase(nostr::Alphabet::P); diff --git a/crates/buzz-core/src/kind.rs b/crates/buzz-core/src/kind.rs index 3c6f1d5913d..e40552accf1 100644 --- a/crates/buzz-core/src/kind.rs +++ b/crates/buzz-core/src/kind.rs @@ -139,7 +139,11 @@ pub const AUTHOR_ONLY_KINDS: &[u32] = &[ /// /// Used by `filter_can_match_result_gated_kinds` to force the per-event /// fallback path in COUNT rather than the fast SQL `count_events()`. -pub const RESULT_GATED_KINDS: &[u32] = &[KIND_DM_VISIBILITY, KIND_AGENT_TURN_METRIC]; +pub const RESULT_GATED_KINDS: &[u32] = &[ + KIND_DM_VISIBILITY, + KIND_AGENT_TURN_METRIC, + KIND_WORKFLOW_MENTION_WAKE, +]; /// Kinds whose stored events have `#p`-bound read access — readable only by /// subscribers whose pubkey appears in the event's `#p` tag. @@ -149,7 +153,7 @@ pub const RESULT_GATED_KINDS: &[u32] = &[KIND_DM_VISIBILITY, KIND_AGENT_TURN_MET /// `#p` values exactly equal the authenticated reader's pubkey. For stored /// (non-ephemeral) kinds in this set, the storage layer additionally writes a /// NULL `search_tsv` so the event is unsearchable through NIP-50 FTS -/// (`schema/schema.sql` and `migrations/0001_initial_schema.sql` — drift +/// (`schema/schema.sql` and the forward FTS migrations — drift /// caught by `p_gated_persistent_kinds_have_storage_null_tsvector` in /// `crates/buzz-search/tests/fts_integration.rs`). /// @@ -158,6 +162,7 @@ pub const RESULT_GATED_KINDS: &[u32] = &[KIND_DM_VISIBILITY, KIND_AGENT_TURN_MET /// storage-layer search defense does not apply to them. pub const P_GATED_KINDS: &[u32] = &[ KIND_AGENT_OBSERVER_FRAME, + KIND_WORKFLOW_MENTION_WAKE, KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, KIND_GIFT_WRAP, @@ -467,6 +472,8 @@ pub const KIND_PAIRING: u32 = 24134; pub const KIND_TYPING_INDICATOR: u32 = 20002; /// Ephemeral: owner-scoped encrypted agent observer telemetry and control frame. pub const KIND_AGENT_OBSERVER_FRAME: u32 = 24200; +/// Durable relay-signed identifier-only workflow mention wake. +pub const KIND_WORKFLOW_MENTION_WAKE: u32 = 44620; /// Ephemeral: huddle emoji reaction burst. Channel-scoped to the ephemeral /// huddle channel with an `h` tag; never stored in the timeline. pub const KIND_HUDDLE_REACTION: u32 = 24810; @@ -698,6 +705,7 @@ pub const ALL_KINDS: &[u32] = &[ KIND_BLOSSOM_AUTH, KIND_PAIRING, KIND_AGENT_OBSERVER_FRAME, + KIND_WORKFLOW_MENTION_WAKE, KIND_HTTP_AUTH, KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_V2, @@ -836,6 +844,7 @@ pub const fn is_relay_only_kind(kind: u32) -> bool { | KIND_DM_VISIBILITY | KIND_THREAD_SUMMARY | KIND_WINDOW_BOUNDS + | KIND_WORKFLOW_MENTION_WAKE ) } @@ -908,8 +917,9 @@ mod tests { } #[test] - fn nip43_membership_snapshot_is_relay_only() { + fn relay_generated_kinds_are_relay_only() { assert!(is_relay_only_kind(KIND_NIP43_MEMBERSHIP_LIST)); + assert!(is_relay_only_kind(KIND_WORKFLOW_MENTION_WAKE)); assert!(!is_relay_only_kind(KIND_NIP43_LEAVE_REQUEST)); } diff --git a/crates/buzz-core/src/lib.rs b/crates/buzz-core/src/lib.rs index 36dc772da3b..574abc9e889 100644 --- a/crates/buzz-core/src/lib.rs +++ b/crates/buzz-core/src/lib.rs @@ -42,6 +42,8 @@ pub mod relay; pub mod tenant; /// Schnorr signature and event ID verification. pub mod verification; +/// Identifier-only workflow mention wake hints. +pub mod workflow_wake; pub use error::VerificationError; pub use event::StoredEvent; diff --git a/crates/buzz-core/src/workflow_wake.rs b/crates/buzz-core/src/workflow_wake.rs new file mode 100644 index 00000000000..ace3bd8714b --- /dev/null +++ b/crates/buzz-core/src/workflow_wake.rs @@ -0,0 +1,327 @@ +//! Identifier-only wake hints for verified workflow mentions. +//! +//! A wake grants no instruction authority. Receivers must authenticate the +//! relay and fetch the exact run-bound workflow definition and visible message +//! before dispatching anything. + +use nostr::{Event, EventBuilder, EventId, Keys, Kind, PublicKey, Tag}; +use thiserror::Error; +use uuid::Uuid; + +use crate::kind::KIND_WORKFLOW_MENTION_WAKE; + +/// A relay-signed, durable hint that a workflow message mentioned one agent. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct WorkflowMentionWake { + recipient: PublicKey, + channel_id: Uuid, + run_id: Uuid, + definition_event_id: EventId, + message_event_id: EventId, +} + +impl WorkflowMentionWake { + /// Construct an identifier-only wake. + pub const fn new( + recipient: PublicKey, + channel_id: Uuid, + run_id: Uuid, + definition_event_id: EventId, + message_event_id: EventId, + ) -> Self { + Self { + recipient, + channel_id, + run_id, + definition_event_id, + message_event_id, + } + } + + /// Sign the canonical empty-content event with the relay identity. + pub fn sign(self, relay_keys: &Keys) -> Result { + EventBuilder::new(Kind::Custom(KIND_WORKFLOW_MENTION_WAKE as u16), "") + .tags(self.canonical_tags()?) + .sign_with_keys(relay_keys) + .map_err(|error| WorkflowMentionWakeError::Signing(error.to_string())) + } + + /// Parse the exact canonical wire shape. Unknown, duplicate, or malformed + /// identity tags are rejected rather than ignored. + pub fn parse(event: &Event) -> Result { + if event.kind.as_u16() as u32 != KIND_WORKFLOW_MENTION_WAKE { + return Err(WorkflowMentionWakeError::WrongKind(event.kind.as_u16())); + } + if !event.content.is_empty() { + return Err(WorkflowMentionWakeError::NonEmptyContent); + } + if event.tags.len() != 5 { + return Err(WorkflowMentionWakeError::WrongTagCount(event.tags.len())); + } + + let tags: Vec<&[String]> = event.tags.iter().map(|tag| tag.as_slice()).collect(); + let recipient = parse_single(&tags, "p", PublicKey::from_hex)?; + let channel_id = parse_single(&tags, "h", |value| value.parse::())?; + let run_id = parse_single(&tags, "run", |value| { + value + .parse::() + .ok() + .filter(|id| !id.is_nil()) + .ok_or(()) + })?; + let definition_event_id = parse_single(&tags, "definition", EventId::from_hex)?; + let message_event_id = parse_single(&tags, "message", EventId::from_hex)?; + + let wake = Self::new( + recipient, + channel_id, + run_id, + definition_event_id, + message_event_id, + ); + let canonical = [ + vec!["p".to_string(), wake.recipient.to_hex()], + vec!["h".to_string(), wake.channel_id.to_string()], + vec!["run".to_string(), wake.run_id.to_string()], + vec!["definition".to_string(), wake.definition_event_id.to_hex()], + vec!["message".to_string(), wake.message_event_id.to_hex()], + ]; + if tags + .iter() + .zip(canonical.iter()) + .any(|(actual, expected)| *actual != expected.as_slice()) + { + return Err(WorkflowMentionWakeError::NonCanonicalTags); + } + Ok(wake) + } + + /// Intended recipient. + pub const fn recipient(self) -> PublicKey { + self.recipient + } + + /// Workflow channel carrying the visible generated message. + pub const fn channel_id(self) -> Uuid { + self.channel_id + } + + /// Exact workflow run. + pub const fn run_id(self) -> Uuid { + self.run_id + } + + /// Exact signed workflow-definition revision selected by the run. + pub const fn definition_event_id(self) -> EventId { + self.definition_event_id + } + + /// Exact visible workflow message to dispatch after verification. + pub const fn message_event_id(self) -> EventId { + self.message_event_id + } + + fn canonical_tags(self) -> Result, WorkflowMentionWakeError> { + [ + vec!["p".to_string(), self.recipient.to_hex()], + vec!["h".to_string(), self.channel_id.to_string()], + vec!["run".to_string(), self.run_id.to_string()], + vec!["definition".to_string(), self.definition_event_id.to_hex()], + vec!["message".to_string(), self.message_event_id.to_hex()], + ] + .into_iter() + .map(|values| { + Tag::parse(values).map_err(|error| WorkflowMentionWakeError::Tag(error.to_string())) + }) + .collect() + } +} + +fn parse_single( + tags: &[&[String]], + name: &'static str, + parse: impl FnOnce(&str) -> Result, +) -> Result { + let matches: Vec<_> = tags + .iter() + .filter(|tag| tag.first().map(String::as_str) == Some(name)) + .collect(); + if matches.len() != 1 || matches[0].len() != 2 { + return Err(WorkflowMentionWakeError::InvalidTag(name)); + } + parse(&matches[0][1]).map_err(|_| WorkflowMentionWakeError::InvalidTag(name)) +} + +/// Invalid workflow mention wake. +#[derive(Debug, Error, PartialEq, Eq)] +pub enum WorkflowMentionWakeError { + /// Event kind is not the workflow mention wake kind. + #[error("wrong workflow mention wake kind: {0}")] + WrongKind(u16), + /// Wake content must be empty. + #[error("workflow mention wake content must be empty")] + NonEmptyContent, + /// Wake must contain exactly the five canonical identity tags. + #[error("wrong workflow mention wake tag count: {0}")] + WrongTagCount(usize), + /// A required identity tag is missing, duplicated, malformed, or has extra fields. + #[error("invalid workflow mention wake {0} tag")] + InvalidTag(&'static str), + /// Tags are not in canonical order or contain a non-canonical representation. + #[error("workflow mention wake tags are not canonical")] + NonCanonicalTags, + /// Canonical tag construction failed. + #[error("workflow mention wake tag construction failed: {0}")] + Tag(String), + /// Event signing failed. + #[error("workflow mention wake signing failed: {0}")] + Signing(String), +} + +#[cfg(test)] +mod tests { + use super::*; + + fn ids() -> (Keys, PublicKey, Uuid, EventId, EventId) { + let relay = Keys::generate(); + let recipient = Keys::generate().public_key(); + let run = Uuid::new_v4(); + let definition = EventBuilder::text_note("definition") + .sign_with_keys(&Keys::generate()) + .expect("sign definition") + .id; + let message = EventBuilder::text_note("message") + .sign_with_keys(&relay) + .expect("sign message") + .id; + (relay, recipient, run, definition, message) + } + + fn custom_event(content: &str, tags: Vec>) -> Event { + EventBuilder::new(Kind::Custom(KIND_WORKFLOW_MENTION_WAKE as u16), content) + .tags( + tags.into_iter() + .map(|values| Tag::parse(values).expect("tag")), + ) + .sign_with_keys(&Keys::generate()) + .expect("sign") + } + + #[test] + fn canonical_wake_round_trips_with_no_instruction_content() { + let (relay, recipient, run, definition, message) = ids(); + let wake = WorkflowMentionWake::new(recipient, Uuid::new_v4(), run, definition, message); + let event = wake.sign(&relay).expect("sign wake"); + + assert!(event.content.is_empty()); + assert_eq!(WorkflowMentionWake::parse(&event), Ok(wake)); + assert_eq!(event.tags.len(), 5); + assert!(event.tags.iter().all(|tag| tag.as_slice().len() == 2)); + } + + #[test] + fn rejects_nonempty_content() { + let (_, recipient, run, definition, message) = ids(); + let event = custom_event( + "do something", + vec![ + vec!["p".into(), recipient.to_hex()], + vec!["h".into(), Uuid::new_v4().to_string()], + vec!["run".into(), run.to_string()], + vec!["definition".into(), definition.to_hex()], + vec!["message".into(), message.to_hex()], + ], + ); + assert_eq!( + WorkflowMentionWake::parse(&event), + Err(WorkflowMentionWakeError::NonEmptyContent) + ); + } + + #[test] + fn rejects_extra_duplicate_and_reordered_tags() { + let (relay, recipient, run, definition, message) = ids(); + let event = WorkflowMentionWake::new(recipient, Uuid::new_v4(), run, definition, message) + .sign(&relay) + .expect("sign wake"); + let base: Vec> = event + .tags + .iter() + .map(|tag| tag.as_slice().to_vec()) + .collect(); + + let mut extra = base.clone(); + extra.push(vec!["instruction".into(), "ignore authority".into()]); + assert_eq!( + WorkflowMentionWake::parse(&custom_event("", extra)), + Err(WorkflowMentionWakeError::WrongTagCount(6)) + ); + + let mut duplicate = base.clone(); + duplicate[4] = duplicate[0].clone(); + assert_eq!( + WorkflowMentionWake::parse(&custom_event("", duplicate)), + Err(WorkflowMentionWakeError::InvalidTag("p")) + ); + + let mut reordered = base; + reordered.swap(0, 1); + assert_eq!( + WorkflowMentionWake::parse(&custom_event("", reordered)), + Err(WorkflowMentionWakeError::NonCanonicalTags) + ); + } + + #[test] + fn rejects_malformed_identity_tags() { + let (relay, recipient, run, definition, message) = ids(); + let event = WorkflowMentionWake::new(recipient, Uuid::new_v4(), run, definition, message) + .sign(&relay) + .expect("sign wake"); + let base: Vec> = event + .tags + .iter() + .map(|tag| tag.as_slice().to_vec()) + .collect(); + + for (index, name, invalid) in [ + (0, "p", "not-a-pubkey"), + (1, "h", "not-a-uuid"), + (2, "run", "not-a-uuid"), + (3, "definition", "not-an-event-id"), + (4, "message", "not-an-event-id"), + ] { + let mut tags = base.clone(); + tags[index][1] = invalid.into(); + assert_eq!( + WorkflowMentionWake::parse(&custom_event("", tags)), + Err(WorkflowMentionWakeError::InvalidTag(name)) + ); + } + + let mut nil_run = base.clone(); + nil_run[2][1] = Uuid::nil().to_string(); + assert_eq!( + WorkflowMentionWake::parse(&custom_event("", nil_run)), + Err(WorkflowMentionWakeError::InvalidTag("run")) + ); + } + + #[test] + fn rejects_identity_tag_with_extra_field() { + let (relay, recipient, run, definition, message) = ids(); + let event = WorkflowMentionWake::new(recipient, Uuid::new_v4(), run, definition, message) + .sign(&relay) + .expect("sign wake"); + let mut tags: Vec> = event + .tags + .iter() + .map(|tag| tag.as_slice().to_vec()) + .collect(); + tags[0].push("marker".into()); + assert_eq!( + WorkflowMentionWake::parse(&custom_event("", tags)), + Err(WorkflowMentionWakeError::InvalidTag("p")) + ); + } +} diff --git a/crates/buzz-db/src/event.rs b/crates/buzz-db/src/event.rs index 136bcce26b5..83da003f3a2 100644 --- a/crates/buzz-db/src/event.rs +++ b/crates/buzz-db/src/event.rs @@ -830,7 +830,8 @@ pub async fn soft_delete_event( event_id: &[u8], ) -> Result { let result = sqlx::query( - "UPDATE events SET deleted_at = NOW() WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL", + "UPDATE events SET deleted_at = COALESCE(deleted_at, NOW()), workflow_revision_superseded = false \ + WHERE community_id = $1 AND id = $2 AND (deleted_at IS NULL OR workflow_revision_superseded)", ) .bind(community_id.as_uuid()) .bind(event_id) @@ -910,7 +911,8 @@ pub async fn soft_delete_event_and_update_thread( let mut tx = pool.begin().await?; let result = sqlx::query( - "UPDATE events SET deleted_at = NOW() WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL", + "UPDATE events SET deleted_at = COALESCE(deleted_at, NOW()), workflow_revision_superseded = false \ + WHERE community_id = $1 AND id = $2 AND (deleted_at IS NULL OR workflow_revision_superseded)", ) .bind(community_id.as_uuid()) .bind(event_id) @@ -1032,6 +1034,30 @@ pub async fn get_event_by_id( } } +/// Fetch a captured workflow revision, including positively identified supersession. +/// Explicit deletion and legacy rows with unknown deletion reasons remain revoked. +/// This does not authorize access; callers must verify the run and current membership. +pub async fn get_workflow_revision( + pool: &PgPool, + community_id: CommunityId, + id_bytes: &[u8], +) -> Result> { + let row = sqlx::query( + "SELECT id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id \ + FROM events WHERE community_id = $1 AND id = $2 AND kind = $3 \ + AND (deleted_at IS NULL OR workflow_revision_superseded) ORDER BY created_at DESC LIMIT 1", + ) + .bind(community_id.as_uuid()) + .bind(id_bytes) + .bind(buzz_core::kind::KIND_WORKFLOW_DEF as i32) + .fetch_optional(pool) + .await?; + match row { + Some(row) => row_to_stored_event(row), + None => Ok(None), + } +} + /// Fetches the latest global (non-channel, `channel_id IS NULL`) replaceable event /// for a (kind, pubkey) pair. /// diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 753a3169ca3..748ca7d137d 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -80,6 +80,7 @@ pub mod usage; pub mod user; /// Workflow, run, and approval persistence. pub mod workflow; +mod workflow_delivery; pub use community::{ ArchivedCommunityRecord, CommunityRecord, CreateCommunityWithOwnerResult, diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index be337cd630e..9faedc43bcc 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -689,7 +689,7 @@ mod tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 40); + assert_eq!(migrations.len(), 42); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] @@ -898,7 +898,7 @@ mod tests { assert!(migrations[32].sql.as_str().contains("search_tsv")); assert!(!migrations[0].sql.as_str().contains("30179")); assert!(include_str!("../../../schema/schema.sql") - .contains("kind IN (1059, 30179, 30300, 30350, 30622, 44100, 44101, 44200)")); + .contains("kind IN (1059, 30179, 30300, 30350, 30622, 44100, 44101, 44200, 44620)")); // Public push-gateway authority is intentionally deployment-global and // durable: immediate revocation and hostile-relay admission cannot be @@ -1181,6 +1181,13 @@ mod tests { 2 ); + // Durable workflow wakes are recipient-gated and must remain outside + // full-text search on both fresh and brownfield databases. + assert_eq!(migrations[40].version, 41); + let workflow_wake_fts = migrations[40].sql.as_str(); + assert!(workflow_wake_fts.contains("kind = 44620")); + assert!(desired_schema.contains("44200, 44620")); + // pgschema intentionally reconciles DDL, not seed DML or table storage // parameters. Its post-apply reconciliation must restore and verify // both parts of the live heartbeat contract for fresh bootstraps. diff --git a/crates/buzz-db/src/replaceable.rs b/crates/buzz-db/src/replaceable.rs index 9b575b6ea18..e8a81796913 100644 --- a/crates/buzz-db/src/replaceable.rs +++ b/crates/buzz-db/src/replaceable.rs @@ -270,6 +270,9 @@ async fn replace_parameterized_event_in_transaction_impl( let statement = if hard_delete_superseded { "DELETE FROM events \ WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL" + } else if kind_i32 == buzz_core::kind::KIND_WORKFLOW_DEF as i32 { + "UPDATE events SET deleted_at = NOW(), workflow_revision_superseded = true \ + WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL" } else { "UPDATE events SET deleted_at = NOW() \ WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL" diff --git a/crates/buzz-db/src/workflow_delivery.rs b/crates/buzz-db/src/workflow_delivery.rs new file mode 100644 index 00000000000..71beb62b2ea --- /dev/null +++ b/crates/buzz-db/src/workflow_delivery.rs @@ -0,0 +1,57 @@ +//! Atomic workflow output persistence and captured-revision reads. +use crate::{event, insert_mentions_in_transaction, Db, Result}; +use buzz_core::{tenant::CommunityId, StoredEvent}; +use uuid::Uuid; + +impl Db { + /// Atomically persist a visible event, its thread metadata/mentions, and all + /// required notifications. No caller may publish any row until this commits. + /// Cancellation or any insert failure rolls the entire bundle back. + pub async fn insert_event_with_notifications( + &self, + community_id: CommunityId, + event: &nostr::Event, + channel_id: Uuid, + thread_meta: Option>, + notifications: &[nostr::Event], + ) -> Result> { + let mut tx = self.begin_transaction().await?; + self.deletion_store() + .guard_transaction(&mut tx, community_id) + .await?; + let mut stored = Vec::with_capacity(1 + notifications.len()); + let message = event::insert_event_with_thread_metadata_tx( + &mut tx, + community_id, + event, + Some(channel_id), + thread_meta, + ) + .await?; + insert_mentions_in_transaction(&mut tx, community_id, event, Some(channel_id)).await?; + stored.push(message); + for notification in notifications { + let row = event::insert_event_in_transaction( + &mut tx, + community_id, + notification, + Some(channel_id), + ) + .await?; + insert_mentions_in_transaction(&mut tx, community_id, notification, Some(channel_id)) + .await?; + stored.push(row); + } + tx.commit().await?; + Ok(stored) + } + + /// Read a captured workflow definition without reviving explicitly deleted revisions. + pub async fn get_workflow_revision( + &self, + community_id: CommunityId, + id_bytes: &[u8], + ) -> Result> { + event::get_workflow_revision(&self.pool, community_id, id_bytes).await + } +} diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 57b65320758..5cd5a6fdea5 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -168,7 +168,7 @@ async fn check_nip98_replay_with_guard( "NIP-98 replay guard failed; rejecting request fail-closed" ); Err(api_error( - StatusCode::UNAUTHORIZED, + StatusCode::SERVICE_UNAVAILABLE, "NIP-98: replay check unavailable", )) } @@ -1233,7 +1233,14 @@ async fn query_events_authed( // Defense-in-depth: never deliver a result-gated event (e.g. kind:44200 // or kind:30622) to a non-owner via the feed path, even though feed SQL // kind allowlists already exclude these kinds. - if !buzz_core::filter::reader_authorized_for_event(&se.event, &authed_pubkey_hex) { + if !crate::handlers::req::event_visible_to_reader( + state, + tenant.community(), + &se.event, + &pubkey_bytes, + ) + .await + { continue; } if let Ok(v) = serde_json::to_value(&se.event) { @@ -1301,7 +1308,14 @@ async fn query_events_authed( // Defense-in-depth: never deliver a result-gated event (e.g. kind:44200 // or kind:30622) to a non-owner via the thread path, even though // requires_h_channel_scope already excludes these kinds from thread metadata. - if !buzz_core::filter::reader_authorized_for_event(&se.event, &authed_pubkey_hex) { + if !crate::handlers::req::event_visible_to_reader( + state, + tenant.community(), + &se.event, + &pubkey_bytes, + ) + .await + { continue; } thread_row_ids.push(se.event.id.to_hex()); @@ -1326,10 +1340,13 @@ async fn query_events_authed( for se in aux_events { if !seen_aux.insert(se.event.id) || !event_in_accessible_channel(&se, &accessible_channels) - || !buzz_core::filter::reader_authorized_for_event( + || !crate::handlers::req::event_visible_to_reader( + state, + tenant.community(), &se.event, - &authed_pubkey_hex, + &pubkey_bytes, ) + .await { continue; } @@ -1446,7 +1463,14 @@ async fn query_events_authed( // Also enforces author-only kinds (30300/30350) and the persona // shared-gate (kind:30175 without ["shared","true"]). Single call // covers all three gated event classes. - if !crate::handlers::req::event_visible_to_reader(&se.event, &pubkey_bytes) { + if !crate::handlers::req::event_visible_to_reader( + state, + tenant.community(), + &se.event, + &pubkey_bytes, + ) + .await + { continue; } if let Ok(v) = serde_json::to_value(&se.event) { @@ -1718,9 +1742,13 @@ async fn count_events_authed( continue; } if !crate::handlers::req::event_visible_to_reader( + state, + tenant.community(), &se.event, &pubkey_bytes, - ) { + ) + .await + { continue; } total += 1; @@ -1788,9 +1816,13 @@ async fn count_events_authed( continue; } if !crate::handlers::req::event_visible_to_reader( + state, + tenant.community(), &se.event, &pubkey_bytes, - ) { + ) + .await + { continue; } total += 1; @@ -1971,7 +2003,14 @@ async fn handle_bridge_search( // branch cannot currently return unshared persona content — but the // check here ensures that a future FTS allowlist change cannot silently // reopen the bypass. - if !crate::handlers::req::event_visible_to_reader(&stored.event, pubkey_bytes) { + if !crate::handlers::req::event_visible_to_reader( + state, + tenant.community(), + &stored.event, + pubkey_bytes, + ) + .await + { continue; } // Dedup across filters. @@ -2759,7 +2798,8 @@ mod tests { /// This test does not require Redis — it injects a guard that always /// returns `Err`, exercising the `Err =>` arm in /// `check_nip98_replay_with_guard` directly. Bites if the arm is changed - /// to admit (`Ok(())` / `Ok(true)`) instead of returning 401. + /// to admit (`Ok(())` / `Ok(true)`) instead of returning retryable 503. + /// A dependency outage is not a replay or invalid-credential verdict. #[tokio::test] async fn nip98_replay_check_fails_closed_when_guard_errors() { use buzz_auth::AuthError; @@ -2767,23 +2807,29 @@ mod tests { use std::future::Future; use std::pin::Pin; - struct AlwaysErrGuard; - impl Nip98ReplayGuard for AlwaysErrGuard { + struct TestGuard { + unavailable: bool, + } + impl Nip98ReplayGuard for TestGuard { fn try_mark_in_scope<'a>( &'a self, _scope: &'a str, _event_id: &'a EventId, _ttl_secs: u64, ) -> Pin> + Send + 'a>> { - Box::pin(async { - Err(AuthError::Internal( - "simulated Redis pool acquire failure".into(), - )) + Box::pin(async move { + if self.unavailable { + Err(AuthError::Internal( + "simulated Redis pool acquire failure".into(), + )) + } else { + Ok(false) + } }) } } - let guard = AlwaysErrGuard; + let guard = TestGuard { unavailable: true }; let tenant = fresh_tenant("relay-a.example"); let event_id_bytes = fresh_nip98_event_id_bytes(); @@ -2792,8 +2838,8 @@ mod tests { .expect_err("guard error MUST fail closed, never admit"); assert_eq!( status, - StatusCode::UNAUTHORIZED, - "fail-closed must return 401" + StatusCode::SERVICE_UNAVAILABLE, + "fail-closed dependency failure must remain retryable" ); let msg = body .get("error") @@ -2804,6 +2850,14 @@ mod tests { "fail-closed body must carry the unavailable signal so callers can \ distinguish unavailability from replay; got body = {body:?}" ); + let (status, _) = check_nip98_replay_with_guard( + &TestGuard { unavailable: false }, + &tenant, + event_id_bytes, + ) + .await + .expect_err("a real replay stays denied"); + assert_eq!(status, StatusCode::UNAUTHORIZED); } /// Build a signed NIP-98 event JSON string for `url` + `method`, mirroring diff --git a/crates/buzz-relay/src/api/workflows.rs b/crates/buzz-relay/src/api/workflows.rs index a3d5a6c729e..1467ff14b82 100644 --- a/crates/buzz-relay/src/api/workflows.rs +++ b/crates/buzz-relay/src/api/workflows.rs @@ -40,6 +40,45 @@ fn request_path(path: &str, raw_query: Option<&str>) -> String { } } +fn ensure_channel_access( + accessible: &[Uuid], + channel_id: Uuid, + error: &'static str, +) -> Result<(), (StatusCode, Json)> { + if accessible.contains(&channel_id) { + Ok(()) + } else { + Err(api_error(StatusCode::FORBIDDEN, error)) + } +} + +async fn enforce_current_channel_read( + state: &Arc, + tenant: &TenantContext, + headers: &HeaderMap, + pubkey: &nostr::PublicKey, + channel_id: Uuid, + error: &'static str, +) -> Result<(), (StatusCode, Json)> { + let pubkey_bytes = pubkey.to_bytes().to_vec(); + let auth_tag = headers + .get("x-auth-tag") + .and_then(|value| value.to_str().ok()); + super::relay_members::enforce_relay_membership( + state, + tenant.community(), + &pubkey_bytes, + auth_tag, + ) + .await?; + + let accessible = state + .get_accessible_channel_ids_cached(tenant.community(), &pubkey_bytes) + .await + .map_err(|error| internal_error(&format!("workflow channel access lookup: {error}")))?; + ensure_channel_access(&accessible, channel_id, error) +} + async fn authorize_workflow_read( state: &Arc, headers: &HeaderMap, @@ -67,18 +106,6 @@ async fn authorize_workflow_read( bridge::enforce_http_admission(state, &tenant, &pubkey).await?; bridge::check_nip98_replay(state, &tenant, event_id_bytes).await?; - let pubkey_bytes = pubkey.to_bytes().to_vec(); - let auth_tag = headers - .get("x-auth-tag") - .and_then(|value| value.to_str().ok()); - super::relay_members::enforce_relay_membership( - state, - tenant.community(), - &pubkey_bytes, - auth_tag, - ) - .await?; - let workflow = state .db .get_workflow(tenant.community(), workflow_id) @@ -92,16 +119,15 @@ async fn authorize_workflow_read( let channel_id = workflow .channel_id .ok_or_else(|| api_error(StatusCode::FORBIDDEN, "workflow is not channel-scoped"))?; - let accessible = state - .get_accessible_channel_ids_cached(tenant.community(), &pubkey_bytes) - .await - .map_err(|error| internal_error(&format!("workflow channel access lookup: {error}")))?; - if !accessible.contains(&channel_id) { - return Err(api_error( - StatusCode::FORBIDDEN, - "workflow is not accessible", - )); - } + enforce_current_channel_read( + state, + &tenant, + headers, + &pubkey, + channel_id, + "workflow is not accessible", + ) + .await?; Ok(tenant) } @@ -226,6 +252,160 @@ fn approval_json(approval: &buzz_db::workflow::ApprovalRecord) -> Value { }) } +// A missing/revoked authority is terminal; unavailable storage is not. Keep +// this distinction at the endpoint so ACP's bounded transport retries can work. +pub(crate) fn wake_lookup_error(error: buzz_db::DbError) -> (StatusCode, Json) { + use buzz_db::DbError; + let status = match &error { + DbError::NotFound(_) => StatusCode::NOT_FOUND, + DbError::Sqlx(sqlx::Error::Io(_) | sqlx::Error::PoolTimedOut | sqlx::Error::PoolClosed) => { + StatusCode::SERVICE_UNAVAILABLE + } + DbError::Sqlx(sqlx::Error::Database(error)) + if error.code().is_some_and(|code| { + code.starts_with("08") + || matches!( + code.as_ref(), + "40001" + | "40P01" + | "53300" + | "55P03" + | "57014" + | "57P01" + | "57P02" + | "57P03" + ) + }) => + { + StatusCode::SERVICE_UNAVAILABLE + } + _ => StatusCode::INTERNAL_SERVER_ERROR, + }; + tracing::warn!(%error, %status, "workflow wake authority lookup failed"); + api_error(status, "workflow wake authority unavailable") +} + +/// `GET /workflow-wakes/{run_id}/{message_id}` — exact authority bundle for one wake. +pub async fn workflow_wake_authority( + State(state): State>, + Path((run_id, message_id)): Path<(Uuid, String)>, + headers: HeaderMap, +) -> Result, (StatusCode, Json)> { + let path = format!("/workflow-wakes/{run_id}/{message_id}"); + let raw_host = headers + .get(axum::http::header::HOST) + .and_then(|value| value.to_str().ok()) + .unwrap_or(""); + let tenant = crate::tenant::bind_community(&state.db, raw_host) + .await + .map_err(|_| api_error(StatusCode::NOT_FOUND, "workflow wake not found"))?; + let url = bridge::nip98_expected_url(&state.config.relay_url, &tenant, &path); + let (recipient, auth_event_id) = + bridge::verify_bridge_auth(&headers, "GET", &url, None, state.config.require_auth_token)?; + bridge::enforce_http_admission(&state, &tenant, &recipient).await?; + bridge::check_nip98_replay(&state, &tenant, auth_event_id).await?; + + let run = state + .db + .get_workflow_run(tenant.community(), run_id) + .await + .map_err(wake_lookup_error)?; + let workflow = state + .db + .get_workflow(tenant.community(), run.workflow_id) + .await + .map_err(wake_lookup_error)?; + let definition_id = run + .definition_event_id + .as_deref() + .filter(|id| id.len() == 32) + .ok_or_else(|| api_error(StatusCode::NOT_FOUND, "workflow wake not found"))?; + let message_id = nostr::EventId::from_hex(&message_id) + .map_err(|_| api_error(StatusCode::BAD_REQUEST, "invalid message id"))?; + let definition = state + .db + .get_workflow_revision(tenant.community(), definition_id) + .await + .map_err(wake_lookup_error)? + .ok_or_else(|| api_error(StatusCode::NOT_FOUND, "workflow wake not found"))?; + let message = state + .db + .get_event_by_id(tenant.community(), message_id.as_bytes()) + .await + .map_err(wake_lookup_error)? + .ok_or_else(|| api_error(StatusCode::NOT_FOUND, "workflow wake not found"))?; + + let recipient_hex = recipient.to_hex(); + let exact_tag = |event: &nostr::Event, name: &str, value: &str| { + event.tags.iter().any(|tag| { + let values = tag.as_slice(); + values.len() == 2 && values[0] == name && values[1].eq_ignore_ascii_case(value) + }) + }; + let message_channel = message + .channel_id + .ok_or_else(|| api_error(StatusCode::NOT_FOUND, "workflow wake not found"))?; + let auth_tag = headers + .get("x-auth-tag") + .and_then(|value| value.to_str().ok()); + super::relay_members::enforce_relay_membership( + &state, + tenant.community(), + &recipient.to_bytes(), + auth_tag, + ) + .await + .map_err(|(status, body)| { + // This shared boundary exposes database lookup failures as 500, while + // explicit roster/authentication denials retain their terminal status. + let status = if status == StatusCode::INTERNAL_SERVER_ERROR { + StatusCode::SERVICE_UNAVAILABLE + } else { + status + }; + (status, body) + })?; + if !state + .db + .is_member(tenant.community(), message_channel, &recipient.to_bytes()) + .await + .map_err(wake_lookup_error)? + { + return Err(api_error( + StatusCode::FORBIDDEN, + "workflow wake not accessible", + )); + } + if workflow.owner_pubkey != definition.event.pubkey.to_bytes() + || workflow.channel_id != Some(message_channel) + || !exact_tag(&definition.event, "h", &message_channel.to_string()) + || !exact_tag(&definition.event, "d", &run.workflow_id.to_string()) + || !exact_tag(&message.event, "h", &message_channel.to_string()) + || !message.event.tags.iter().any(|tag| { + let values = tag.as_slice(); + values.len() == 2 && values[0] == "p" && values[1].eq_ignore_ascii_case(&recipient_hex) + }) + || !exact_tag(&message.event, "workflow-run", &run_id.to_string()) + || !exact_tag( + &message.event, + "workflow-definition", + &definition.event.id.to_hex(), + ) + { + return Err(api_error(StatusCode::NOT_FOUND, "workflow wake not found")); + } + + Ok(Json(serde_json::json!({ + "run_id": run.id, + "channel_id": message_channel, + "workflow_id": run.workflow_id, + "definition_event_id": definition.event.id.to_hex(), + "workflow_owner": hex::encode(&workflow.owner_pubkey), + "definition": definition.event, + "message": message.event, + }))) +} + #[cfg(test)] mod tests { use super::*; @@ -242,6 +422,43 @@ mod tests { ); } + #[test] + fn wake_lookup_failure_is_not_authority_revocation() { + use buzz_db::DbError; + for error in [ + sqlx::Error::PoolClosed, + sqlx::Error::PoolTimedOut, + sqlx::Error::Io(std::io::Error::from(std::io::ErrorKind::ConnectionReset)), + ] { + assert_eq!( + wake_lookup_error(DbError::Sqlx(error)).0, + StatusCode::SERVICE_UNAVAILABLE + ); + } + assert_eq!( + wake_lookup_error(DbError::NotFound("run".into())).0, + StatusCode::NOT_FOUND + ); + assert_eq!( + wake_lookup_error(DbError::InvalidData("bad row".into())).0, + StatusCode::INTERNAL_SERVER_ERROR + ); + assert_eq!( + wake_lookup_error(DbError::Sqlx(sqlx::Error::ColumnNotFound("bad".into()))).0, + StatusCode::INTERNAL_SERVER_ERROR + ); + } + + #[test] + fn channel_access_is_required_at_authority_read_time() { + let channel = Uuid::new_v4(); + assert!(ensure_channel_access(&[channel], channel, "revoked").is_ok()); + + let (status, _) = ensure_channel_access(&[], channel, "revoked") + .expect_err("removed member must not retain authority read access"); + assert_eq!(status, StatusCode::FORBIDDEN); + } + #[test] fn approval_wire_does_not_expose_hash_as_token() { let approval = buzz_db::workflow::ApprovalRecord { diff --git a/crates/buzz-relay/src/handlers/count.rs b/crates/buzz-relay/src/handlers/count.rs index 938674301e7..bd9e606b5ec 100644 --- a/crates/buzz-relay/src/handlers/count.rs +++ b/crates/buzz-relay/src/handlers/count.rs @@ -38,8 +38,9 @@ pub async fn handle_count( } }; - // P-gated kinds (gift wraps, member notifications, observer frames) require - // the caller's own pubkey in the #p tag — same enforcement as WS REQ handler. + // Result-gated kinds (DM visibility, agent metrics, and workflow wakes) + // require the caller's own pubkey in the #p tag for explicit-kind filters. + // Kindless filters remain valid and are restricted by per-event gates. let authed_pubkey_hex = hex::encode(&pubkey_bytes); if !super::req::p_gated_filters_authorized(&filters, &authed_pubkey_hex) { conn.send(RelayMessage::closed( @@ -226,7 +227,14 @@ pub async fn handle_count( { continue; } - if !event_visible_to_reader(&se.event, &pubkey_bytes) { + if !event_visible_to_reader( + &state, + conn.tenant.community(), + &se.event, + &pubkey_bytes, + ) + .await + { continue; } total += 1; @@ -299,7 +307,14 @@ pub async fn handle_count( { continue; } - if !event_visible_to_reader(&se.event, &pubkey_bytes) { + if !event_visible_to_reader( + &state, + conn.tenant.community(), + &se.event, + &pubkey_bytes, + ) + .await + { continue; } total += 1; diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index ccba40f3282..c2a446ef32c 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -174,6 +174,48 @@ pub async fn filter_fanout_by_access( matches }; + let owner_only_kind = event_kind_u32(&stored_event.event); + // Result-gated delivery (DM visibility, agent metrics, and workflow wakes) + // must reach only the exact #p owner/recipient, including kindless channel + // wildcard subscriptions. + let matches = if buzz_core::kind::RESULT_GATED_KINDS.contains(&owner_only_kind) { + matches + .into_iter() + .filter(|(conn_id, _)| { + state + .conn_manager + .pubkey_for_conn(*conn_id) + .is_some_and(|pk| { + buzz_core::filter::reader_authorized_for_event( + &stored_event.event, + &hex::encode(pk), + ) + }) + }) + .collect() + } else { + matches + }; + + if owner_only_kind == buzz_core::kind::KIND_WORKFLOW_MENTION_WAKE { + let mut allowed = Vec::with_capacity(matches.len()); + for (conn_id, sub_id) in matches { + let Some(pubkey) = state.conn_manager.pubkey_for_conn(conn_id) else { + continue; + }; + if super::req::event_visible_to_reader( + state, + community_id, + &stored_event.event, + &pubkey, + ) + .await + { + allowed.push((conn_id, sub_id)); + } + } + return allowed; + } let Some(channel_id) = stored_event.channel_id else { return matches; }; @@ -454,40 +496,13 @@ async fn dispatch_persistent_event_inner( return 0; } }; - // For viewer-private events (kind:30622 DM visibility, kind:44200 agent turn - // metrics), live fan-out must reach only the owner — a kindless `ids:[…]` - // subscription can otherwise match it. Pull paths (HTTP /query, WS historical) - // are gated separately by reader_authorized_for_event. - let owner_only_kind = kind_u32 == buzz_core::kind::KIND_DM_VISIBILITY - || kind_u32 == buzz_core::kind::KIND_AGENT_TURN_METRIC; - let private_event_owner: Option = owner_only_kind - .then(|| { - let p = nostr::SingleLetterTag::lowercase(nostr::Alphabet::P); - stored_event - .event - .tags - .filter(nostr::TagKind::SingleLetter(p)) - .find_map(|t| t.content().map(|s| s.to_string())) - }) - .flatten(); - // Author-only delivery gating (NIP-ER reminders) is enforced centrally in - // filter_fanout_by_access, applied to `matches` above before this loop. The - // DM visibility owner gate is an additional delivery fence, so build shared - // frames only after applying it to the already access-filtered recipient set. + // Result-gated delivery (DM visibility, agent metrics, and workflow wakes) + // is enforced centrally in filter_fanout_by_access, applied to `matches` + // above before this loop. Build shared frames only after every recipient + // has passed that chokepoint. let recipients: Vec<_> = matches .iter() - .filter_map(|(target_conn_id, sub_id)| { - if let Some(ref owner_hex) = private_event_owner { - let is_owner = state - .conn_manager - .pubkey_for(*target_conn_id) - .is_some_and(|pk| hex::encode(pk) == *owner_hex); - if !is_owner { - return None; - } - } - Some((*target_conn_id, sub_id.as_str())) - }) + .map(|(target_conn_id, sub_id)| (*target_conn_id, sub_id.as_str())) .collect(); let frames = fanout_frame_cache(recipients.iter().map(|(_, sub_id)| *sub_id), &event_json); let drop_count = send_fanout_frames(state, recipients, &frames); @@ -1995,6 +2010,7 @@ mod tests { use std::sync::atomic::AtomicU8; use std::sync::Arc; + use buzz_core::workflow_wake::WorkflowMentionWake; use buzz_core::StoredEvent; use nostr::{EventBuilder, Keys, Kind}; use tokio::sync::{mpsc, Mutex}; @@ -2167,6 +2183,52 @@ mod tests { assert_eq!(out, matches); } + #[tokio::test] + async fn workflow_wake_fails_closed_when_membership_cannot_be_established() { + let state = test_state().await; + let community_id = buzz_core::tenant::CommunityId::from_uuid(Uuid::nil()); + let channel_id = Uuid::new_v4(); + state + .channel_visibility_cache + .insert((community_id, channel_id), "open".to_string()); + + let recipient_keys = Keys::generate(); + let other_keys = Keys::generate(); + let relay_keys = Keys::generate(); + let definition = EventBuilder::text_note("definition") + .sign_with_keys(&Keys::generate()) + .expect("sign definition"); + let message = EventBuilder::text_note("message") + .sign_with_keys(&relay_keys) + .expect("sign message"); + let wake = WorkflowMentionWake::new( + recipient_keys.public_key(), + channel_id, + Uuid::new_v4(), + definition.id, + message.id, + ) + .sign(&relay_keys) + .expect("sign wake"); + let stored = StoredEvent::new(wake, Some(channel_id)); + + let recipient = register_conn( + &state, + Some(recipient_keys.public_key().to_bytes().to_vec()), + ); + let other = register_conn(&state, Some(other_keys.public_key().to_bytes().to_vec())); + let unauthed = register_conn(&state, None); + let matches = vec![ + (recipient, "recipient".to_string()), + (other, "other".to_string()), + (unauthed, "unauthed".to_string()), + ]; + + let out = filter_fanout_by_access(&state, community_id, &stored, matches, None).await; + + assert!(out.is_empty()); + } + #[tokio::test] async fn private_channel_keeps_member_drops_non_member_and_unknown() { let state = test_state().await; diff --git a/crates/buzz-relay/src/handlers/req.rs b/crates/buzz-relay/src/handlers/req.rs index d299cc045fa..99c208aa313 100644 --- a/crates/buzz-relay/src/handlers/req.rs +++ b/crates/buzz-relay/src/handlers/req.rs @@ -208,23 +208,20 @@ pub async fn handle_req( return; } - // Applied BEFORE the NIP-50 search branch so that an authenticated member - // cannot use `{"search":"...","kinds":[30174]}` (or similar for p-gated - // kinds) to harvest indexed-but-globally-stored sensitive events. Search - // hits are looked up by event id and returned without the per-filter - // post-check the historical-delivery branch applies, so the gate must run - // here, up front. Only applies to GLOBAL subscriptions (channel_id = None): - // channel-scoped subs can never receive globally-stored events because of - // the fan_out() invariant in subscription.rs. + // Applied BEFORE the NIP-50 search branch so an explicit sensitive-kind + // filter cannot harvest indexed private events. Kindless channel filters + // remain valid NIP-01 wildcards; every returned event is independently + // checked at the shared result gate, and live fan-out uses the same check. + let authed_pubkey_hex = hex::encode(&pubkey_bytes); + if !p_gated_filters_authorized(&filters, &authed_pubkey_hex) { + conn.send(RelayMessage::closed( + &sub_id, + "restricted: p-gated events require #p matching your pubkey", + )); + return; + } + if channel_id.is_none() { - let authed_pubkey_hex = hex::encode(&pubkey_bytes); - if !p_gated_filters_authorized(&filters, &authed_pubkey_hex) { - conn.send(RelayMessage::closed( - &sub_id, - "restricted: p-gated events require #p matching your pubkey", - )); - return; - } if !engram_filters_authorized(&filters, &authed_pubkey_hex) { conn.send(RelayMessage::closed( &sub_id, @@ -448,7 +445,14 @@ pub async fn handle_req( // Also enforces author-only kinds (30300/30350) and the persona // shared-gate (kind:30175 without ["shared","true"]). Single call // covers all three gated event classes. - if !event_visible_to_reader(&stored.event, &pubkey_bytes) { + if !event_visible_to_reader( + &state, + conn.tenant.community(), + &stored.event, + &pubkey_bytes, + ) + .await + { continue; } @@ -781,7 +785,14 @@ async fn handle_search_req( } // Result-level gate: covers author-only, persona shared-gate, // and result-gated kinds in one call. - if !event_visible_to_reader(&stored.event, reader_pubkey_bytes) { + if !event_visible_to_reader( + state, + tenant.community(), + &stored.event, + reader_pubkey_bytes, + ) + .await + { continue; } // Dedup AFTER acceptance — an event that fails filter A's constraints @@ -1182,10 +1193,22 @@ fn extract_channel_id_from_filters(filters: &[Filter]) -> Option { pub(crate) fn p_gated_filters_authorized(filters: &[Filter], authed_pubkey_hex: &str) -> bool { let p_tag = nostr::SingleLetterTag::lowercase(nostr::Alphabet::P); filters.iter().all(|filter| { - let can_match_p_gated = filter.kinds.as_ref().is_none_or(|ks| { - ks.iter() - .any(|kind| P_GATED_KINDS.contains(&(kind.as_u16() as u32))) - }); + // Kindless full-text searches cannot surface p-gated rows, and a + // kindless channel filter is safe to register: global p-gated events + // cannot enter its channel index, while channel-scoped workflow wakes + // are removed by the shared per-event recipient gate. Preserve the + // stricter rule for global wildcards and explicit p-gated kinds. + let has_channel_scope = filter + .generic_tags + .get(&nostr::SingleLetterTag::lowercase(nostr::Alphabet::H)) + .is_some_and(|values| !values.is_empty()); + let can_match_p_gated = filter.kinds.as_ref().map_or_else( + || filter.search.is_none() && !has_channel_scope, + |ks| { + ks.iter() + .any(|kind| P_GATED_KINDS.contains(&(kind.as_u16() as u32))) + }, + ); if !can_match_p_gated { return true; } @@ -1329,6 +1352,14 @@ pub(crate) fn result_gated_count_safe_for_pushdown( filter: &Filter, authed_pubkey_hex: &str, ) -> bool { + // Recipient pinning alone cannot prove current channel membership for wakes. + if filter.kinds.as_ref().is_none_or(|kinds| { + kinds + .iter() + .any(|kind| u32::from(kind.as_u16()) == buzz_core::kind::KIND_WORKFLOW_MENTION_WAKE) + }) { + return false; + } let p_tag = nostr::SingleLetterTag::lowercase(nostr::Alphabet::P); filter .generic_tags @@ -1365,7 +1396,27 @@ pub(crate) fn is_author_only_event(event: &nostr::Event, requester_pubkey_bytes: /// Call this from every read surface — both WS (REQ/COUNT/fan-out) and HTTP /// (NIP-98 `/query`, `/count`, FTS search) — instead of inlining the three /// individual predicates at each site. -pub(crate) fn event_visible_to_reader(event: &nostr::Event, requester_pubkey_bytes: &[u8]) -> bool { +pub(crate) async fn event_visible_to_reader( + state: &AppState, + community: buzz_core::tenant::CommunityId, + event: &nostr::Event, + requester_pubkey_bytes: &[u8], +) -> bool { + if u32::from(event.kind.as_u16()) == buzz_core::kind::KIND_WORKFLOW_MENTION_WAKE { + let Ok(wake) = buzz_core::workflow_wake::WorkflowMentionWake::parse(event) else { + return false; + }; + // Open-channel readability is not wake authority. Read the writer, + // not a cached membership snapshot, at every delivery/count boundary. + if !state + .db + .is_member(community, wake.channel_id(), requester_pubkey_bytes) + .await + .unwrap_or(false) + { + return false; + } + } if is_author_only_event(event, requester_pubkey_bytes) { return false; } @@ -1870,6 +1921,20 @@ mod tests { assert_eq!(extract_channel_id_from_filters(&filters), Some(channel_id)); } + /// A channel-scoped kindless wildcard remains admissible. P-gated global + /// kinds cannot enter the channel subscription index, and the only + /// channel-scoped p-gated kind (workflow wake) is result-gated per recipient. + #[test] + fn channel_wildcard_preserves_all_mode_without_weakening_wakes() { + let authed = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; + let channel = uuid::Uuid::new_v4(); + assert!(p_gated_filters_authorized( + &[filter_with_channel(channel)], + authed + )); + assert!(!p_gated_filters_authorized(&[Filter::new()], authed)); + } + #[test] fn test_search_filter_detection() { let search_filter = Filter::new().search("hello world"); @@ -2291,6 +2356,13 @@ mod tests { assert!(engram_filters_authorized(&[f], &agent)); } + #[test] + fn p_gate_allows_kindless_search_because_p_gated_rows_are_unsearchable() { + let (agent, _, _) = three_pubkeys(); + let f = Filter::new().search("ordinary-channel-search"); + assert!(p_gated_filters_authorized(&[f], &agent)); + } + #[test] fn p_gate_rejects_bare_kind_search_filter_for_gift_wrap() { // P-gated kinds (observer frames, member notifications) are indexed diff --git a/crates/buzz-relay/src/router.rs b/crates/buzz-relay/src/router.rs index dd0fde6fdcd..8e8d4c56646 100644 --- a/crates/buzz-relay/src/router.rs +++ b/crates/buzz-relay/src/router.rs @@ -75,6 +75,10 @@ pub fn build_router(state: Arc) -> Router { // Relay-owned third-party GIF metadata proxy (NIP-98 auth). .route(api::gifs::SEARCH_PATH, post(api::gifs::search)) .route(api::gifs::SHARE_PATH, post(api::gifs::share)) + .route( + "/workflow-wakes/{run_id}/{message_id}", + get(api::workflows::workflow_wake_authority), + ) .route( "/workflows/{workflow_id}/runs", get(api::workflows::workflow_runs), diff --git a/crates/buzz-relay/src/workflow_delivery_tests.rs b/crates/buzz-relay/src/workflow_delivery_tests.rs new file mode 100644 index 00000000000..3d045bc2191 --- /dev/null +++ b/crates/buzz-relay/src/workflow_delivery_tests.rs @@ -0,0 +1,535 @@ +//! Real-storage regressions for workflow wake lifecycle boundaries. +use super::integration_tests::test_state_with_redis; +use super::*; +use axum::{ + extract::{Path, State}, + http::{HeaderMap, StatusCode}, +}; +use buzz_core::{ + channel::{ChannelType, ChannelVisibility, MemberRole}, + tenant::CommunityId, +}; +use buzz_db::CreateCommunityWithOwnerResult; +use nostr::{Event, Keys, Timestamp}; + +struct Fixture { + state: Arc, + community: CommunityId, + host: String, + channel: Uuid, + owner: Keys, + agent: Keys, + workflow: Uuid, +} +impl Fixture { + async fn new() -> Self { + let state = test_state_with_redis( + std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://localhost:6379".into()), + ) + .await; + let owner = Keys::generate(); + let agent = Keys::generate(); + let host = format!("wake-{}.example", Uuid::new_v4().simple()); + let community = match state + .db + .create_community_with_owner(&host, &owner.public_key().to_hex()) + .await + .expect("community") + { + CreateCommunityWithOwnerResult::Created(record) => record.id, + other => panic!("unexpected {other:?}"), + }; + state + .db + .ensure_user(community, &owner.public_key().to_bytes()) + .await + .expect("owner user"); + let channel = state + .db + .create_channel( + community, + "wake", + ChannelType::Stream, + ChannelVisibility::Open, + None, + &owner.public_key().to_bytes(), + None, + ) + .await + .expect("channel") + .id; + state + .db + .ensure_user(community, &agent.public_key().to_bytes()) + .await + .expect("agent"); + state + .db + .update_user_profile( + community, + &agent.public_key().to_bytes(), + Some("Worker"), + None, + None, + None, + ) + .await + .expect("name"); + state + .db + .add_member( + community, + channel, + &agent.public_key().to_bytes(), + MemberRole::Bot, + Some(&owner.public_key().to_bytes()), + ) + .await + .expect("member"); + Self { + state, + community, + host, + channel, + owner, + agent, + workflow: Uuid::new_v4(), + } + } + fn connection( + &self, + ) -> ( + Arc, + tokio::sync::mpsc::Receiver, + ) { + use crate::connection::{AuthState, ConnectionState}; + use std::{collections::HashMap, sync::atomic::AtomicU8}; + use tokio::sync::{mpsc, Mutex, RwLock}; + let (send_tx, rx) = mpsc::channel(16); + let (ctrl_tx, _ctrl_rx) = mpsc::channel(4); + let conn = Arc::new(ConnectionState { + conn_id: Uuid::new_v4(), + tenant: buzz_core::TenantContext::resolved(self.community, &self.host), + remote_addr: "127.0.0.1:1234".parse().expect("address"), + auth_state: RwLock::new(AuthState::Authenticated(buzz_auth::AuthContext { + pubkey: self.agent.public_key(), + scopes: vec![], + channel_ids: None, + auth_method: buzz_auth::AuthMethod::Nip42, + agent_owner_pubkey: None, + })), + subscriptions: Arc::new(Mutex::new(HashMap::new())), + send_tx, + ctrl_tx, + cancel: tokio_util::sync::CancellationToken::new(), + backpressure_count: Arc::new(AtomicU8::new(0)), + grace_limit: 3, + }); + self.state.conn_manager.register( + conn.conn_id, + conn.send_tx.clone(), + conn.ctrl_tx.clone(), + None, + conn.cancel.clone(), + self.community, + conn.backpressure_count.clone(), + conn.subscriptions.clone(), + conn.grace_limit, + ); + self.state + .conn_manager + .set_authenticated_pubkey(conn.conn_id, self.agent.public_key().to_bytes().to_vec()); + (conn, rx) + } + fn headers(&self) -> HeaderMap { + let mut headers = HeaderMap::new(); + headers.insert("host", self.host.parse().expect("host")); + headers.insert( + "x-pubkey", + self.agent.public_key().to_hex().parse().expect("pubkey"), + ); + headers + } + async fn revision(&self, timestamp: u64) -> Event { + let definition = "name: wake\ntrigger:\n on: message_posted\nsteps:\n - id: notify\n action: send_message\n text: '@Worker work'\n"; + let event = EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_WORKFLOW_DEF as u16), + definition, + ) + .custom_created_at(Timestamp::from(timestamp)) + .tags([ + Tag::parse(["d", &self.workflow.to_string()]).expect("d"), + Tag::parse(["h", &self.channel.to_string()]).expect("h"), + ]) + .sign_with_keys(&self.owner) + .expect("definition"); + let mut tx = self.state.db.begin_transaction().await.expect("tx"); + self.state + .db + .replace_parameterized_event_in_transaction( + &mut tx, + self.community, + &event, + &self.workflow.to_string(), + Some(self.channel), + buzz_db::replaceable::ParameterizedReplacePrecondition::Unconditional, + ) + .await + .expect("replace"); + self.state + .db + .upsert_workflow( + &mut tx, + self.community, + self.workflow, + Some(self.channel), + &self.owner.public_key().to_bytes(), + "wake", + "{}", + &[0; 32], + event.id.as_bytes(), + ) + .await + .expect("materialize"); + tx.commit().await.expect("commit"); + event + } + async fn authority( + &self, + run: Uuid, + message: &str, + ) -> Result, (StatusCode, axum::Json)> { + crate::api::workflows::workflow_wake_authority( + State(self.state.clone()), + Path((run, message.to_owned())), + self.headers(), + ) + .await + } +} + +fn next_frame( + rx: &mut tokio::sync::mpsc::Receiver, +) -> serde_json::Value { + let axum::extract::ws::Message::Text(text) = rx.try_recv().expect("frame") else { + panic!("expected text frame"); + }; + serde_json::from_str(&text).expect("frame JSON") +} + +#[tokio::test] +#[ignore = "requires Postgres and Redis"] +async fn captured_revision_survives_replacement_but_not_revocation() { + let f = Fixture::new().await; + let a = f.revision(Timestamp::now().as_secs()).await; + let run = f + .state + .db + .create_workflow_run(f.community, f.workflow, Some(a.id.as_bytes()), None, None) + .await + .expect("run"); + let message = RelayActionSink::new(&f.state) + .send_message( + WorkflowMessageContext { + community_id: f.community, + run_id: run, + step_id: "notify".into(), + definition_event_id: Some(a.id.as_bytes().to_vec()), + }, + &f.channel.to_string(), + "@Worker work", + &f.owner.public_key().to_hex(), + None, + ) + .await + .expect("message"); + f.revision(a.created_at.as_secs() + 1).await; + assert!(f + .state + .db + .get_event_by_id(f.community, a.id.as_bytes()) + .await + .expect("live read") + .is_none()); + let authority = f + .authority(run, &message) + .await + .expect("captured authority"); + assert_eq!(authority.0["definition"]["id"], a.id.to_hex()); + let deletion = EventBuilder::new(Kind::EventDeletion, "revoke captured revision") + .tags([Tag::event(a.id)]) + .sign_with_keys(&f.owner) + .expect("signed deletion"); + let result = crate::handlers::ingest::ingest_event( + &f.state, + &buzz_core::TenantContext::resolved(f.community, &f.host), + deletion, + crate::handlers::ingest::IngestAuth::Nip42 { + pubkey: f.owner.public_key(), + scopes: vec![buzz_auth::Scope::MessagesWrite], + channel_ids: None, + conn_id: Uuid::new_v4(), + }, + ) + .await + .expect("explicit revocation through authenticated deletion ingress"); + assert!(result.accepted); + assert_eq!( + f.authority(run, &message).await.expect_err("revoked").0, + StatusCode::NOT_FOUND + ); +} + +#[tokio::test] +#[ignore = "requires Postgres and Redis"] +async fn removed_open_channel_member_cannot_read_or_count_wakes() { + let f = Fixture::new().await; + let a = f.revision(Timestamp::now().as_secs()).await; + let run = f + .state + .db + .create_workflow_run(f.community, f.workflow, Some(a.id.as_bytes()), None, None) + .await + .expect("run"); + let message = RelayActionSink::new(&f.state) + .send_message( + WorkflowMessageContext { + community_id: f.community, + run_id: run, + step_id: "notify".into(), + definition_event_id: Some(a.id.as_bytes().to_vec()), + }, + &f.channel.to_string(), + "@Worker work", + &f.owner.public_key().to_hex(), + None, + ) + .await + .expect("message"); + let _ = f.authority(run, &message).await.expect("member authority"); + let filter = serde_json::json!({"kinds":[buzz_core::kind::KIND_WORKFLOW_MENTION_WAKE], "#p":[f.agent.public_key().to_hex()], "#h":[f.channel.to_string()]}); + let body = + axum::body::Bytes::from(serde_json::to_vec(&serde_json::json!([filter])).expect("body")); + let before = + crate::api::bridge::query_events(State(f.state.clone()), f.headers(), body.clone()) + .await + .expect("query"); + assert_eq!(before.0.as_array().expect("events").len(), 1); + // Exercise the actual WS handlers and live send path, retaining the same + // connection/subscription across removal to expose stale access state. + let (conn, mut frames) = f.connection(); + let filters: Vec = serde_json::from_slice(&body).expect("filters"); + crate::handlers::req::handle_req( + "wakes".into(), + filters.clone(), + conn.clone(), + f.state.clone(), + ) + .await; + assert_eq!(next_frame(&mut frames)[0], "EVENT"); + assert_eq!(next_frame(&mut frames)[0], "EOSE"); + crate::handlers::count::handle_count( + "count".into(), + filters.clone(), + conn.clone(), + f.state.clone(), + ) + .await; + assert_eq!(next_frame(&mut frames)[2]["count"], 1); + let wake: Event = serde_json::from_value(before.0[0].clone()).expect("wake"); + let stored = buzz_core::StoredEvent::new(wake, Some(f.channel)); + crate::handlers::event::fan_out_event_to_local_subscribers(&f.state, f.community, &stored) + .await; + assert_eq!(next_frame(&mut frames)[0], "EVENT"); + f.state + .db + .remove_member( + f.community, + f.channel, + &f.agent.public_key().to_bytes(), + &f.owner.public_key().to_bytes(), + ) + .await + .expect("remove"); + assert!(f + .state + .db + .get_accessible_channel_ids(f.community, &f.agent.public_key().to_bytes()) + .await + .expect("open readability") + .contains(&f.channel)); + assert_eq!( + f.authority(run, &message) + .await + .expect_err("not membership") + .0, + StatusCode::FORBIDDEN + ); + let after = crate::api::bridge::query_events(State(f.state.clone()), f.headers(), body.clone()) + .await + .expect("query after removal"); + assert!(after.0.as_array().expect("events").is_empty()); + let count = crate::api::bridge::count_events(State(f.state.clone()), f.headers(), body) + .await + .expect("count after removal"); + assert_eq!(count.0["count"], 0); + crate::handlers::event::fan_out_event_to_local_subscribers(&f.state, f.community, &stored) + .await; + assert!( + frames.try_recv().is_err(), + "stale subscription must not deliver" + ); + crate::handlers::req::handle_req( + "wakes".into(), + filters.clone(), + conn.clone(), + f.state.clone(), + ) + .await; + assert_eq!(next_frame(&mut frames)[0], "EOSE", "no historical EVENT"); + crate::handlers::count::handle_count("count".into(), filters, conn, f.state.clone()).await; + assert_eq!(next_frame(&mut frames)[2]["count"], 0); + assert!(frames.try_recv().is_err()); +} + +#[tokio::test] +#[ignore = "requires Postgres and Redis"] +async fn notification_failure_rolls_back_message_mentions_and_thread_metadata() { + let f = Fixture::new().await; + let message = EventBuilder::new(Kind::Custom(KIND_STREAM_MESSAGE as u16), "@Worker work") + .tags([ + Tag::parse(["h", &f.channel.to_string()]).expect("h"), + Tag::public_key(f.agent.public_key()), + ]) + .sign_with_keys(&f.state.relay_keypair) + .expect("message"); + let wake = WorkflowMentionWake::new( + f.agent.public_key(), + f.channel, + Uuid::new_v4(), + message.id, + message.id, + ) + .sign(&f.state.relay_keypair) + .expect("wake"); + // The second notification fails inside the transaction, after the message, + // metadata, mentions and first recipient have been written. + let rejected = EventBuilder::new(Kind::Custom(22242), "auth cannot persist") + .sign_with_keys(&f.state.relay_keypair) + .expect("rejected event"); + let meta = || buzz_db::event::ThreadMetadataParams { + event_id: message.id.as_bytes(), + event_created_at: chrono::DateTime::from_timestamp(message.created_at.as_secs() as i64, 0) + .expect("ts"), + channel_id: f.channel, + parent_event_id: None, + parent_event_created_at: None, + root_event_id: None, + root_event_created_at: None, + depth: 0, + broadcast: false, + }; + assert!(f + .state + .db + .insert_event_with_notifications( + f.community, + &message, + f.channel, + Some(meta()), + &[wake.clone(), rejected] + ) + .await + .is_err()); + for event in [&message, &wake] { + assert!(f + .state + .db + .get_event_by_id(f.community, event.id.as_bytes()) + .await + .expect("rollback read") + .is_none()); + } + assert!(f + .state + .db + .get_thread_metadata_by_event(f.community, message.id.as_bytes()) + .await + .expect("metadata rollback") + .is_none()); + let mut tx = f.state.db.begin_transaction().await.expect("read mentions"); + let mentions: i64 = sqlx::query_scalar( + "SELECT count(*) FROM event_mentions WHERE community_id=$1 AND event_id=$2", + ) + .bind(f.community.as_uuid()) + .bind(message.id.as_bytes().as_slice()) + .fetch_one(&mut *tx) + .await + .expect("mentions"); + assert_eq!(mentions, 0); + tx.rollback().await.expect("read rollback"); + // Commit without any fan-out: a new historical read still recovers all rows. + let second = WorkflowMentionWake::new( + f.owner.public_key(), + f.channel, + Uuid::new_v4(), + message.id, + message.id, + ) + .sign(&f.state.relay_keypair) + .expect("second wake"); + let rows = f + .state + .db + .insert_event_with_notifications( + f.community, + &message, + f.channel, + Some(meta()), + &[wake.clone(), second.clone()], + ) + .await + .expect("commit bundle"); + assert_eq!(rows.len(), 3); + for event in [&message, &wake, &second] { + assert!(f + .state + .db + .get_event_by_id(f.community, event.id.as_bytes()) + .await + .expect("replay read") + .is_some()); + } +} + +#[tokio::test] +#[ignore = "requires Postgres and Redis"] +async fn storage_timeout_is_retryable_but_missing_authority_is_terminal() { + use crate::api::workflows::wake_lookup_error; + let f = Fixture::new().await; + let mut tx = f.state.db.begin_transaction().await.expect("transaction"); + sqlx::query("SET LOCAL statement_timeout = '10ms'") + .execute(&mut *tx) + .await + .expect("set timeout"); + let error = sqlx::query("SELECT pg_sleep(1)") + .execute(&mut *tx) + .await + .expect_err("statement timeout"); + assert_eq!( + error.as_database_error().and_then(|e| e.code()).as_deref(), + Some("57014") + ); + assert_eq!( + wake_lookup_error(error.into()).0, + StatusCode::SERVICE_UNAVAILABLE + ); + tx.rollback().await.expect("rollback"); + let error = f + .state + .db + .get_workflow_run(f.community, Uuid::new_v4()) + .await + .expect_err("missing run"); + assert_eq!(wake_lookup_error(error).0, StatusCode::NOT_FOUND); +} diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 8ce23a2e8ea..684dc285a36 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -9,8 +9,8 @@ use std::pin::Pin; use std::sync::{Arc, Weak}; use buzz_core::kind::KIND_STREAM_MESSAGE; -use buzz_core::tenant::CommunityId; -use buzz_workflow::action_sink::{ActionSink, ActionSinkError}; +use buzz_core::workflow_wake::WorkflowMentionWake; +use buzz_workflow::action_sink::{ActionSink, ActionSinkError, WorkflowMessageContext}; use chrono::Utc; use nostr::{EventBuilder, Kind, Tag}; use tracing::info; @@ -172,12 +172,18 @@ impl RelayActionSink { impl ActionSink for RelayActionSink { fn send_message( &self, - community_id: CommunityId, + context: WorkflowMessageContext, channel_id: &str, text: &str, author_pubkey: &str, reply_to: Option<&str>, ) -> Pin> + Send + '_>> { + let WorkflowMessageContext { + community_id, + run_id, + step_id, + definition_event_id, + } = context; let channel_id = channel_id.to_owned(); let text = text.to_owned(); let author_pubkey = author_pubkey.to_owned(); @@ -313,9 +319,8 @@ impl ActionSink for RelayActionSink { } // Resolve `@Name` mentions to channel-member pubkeys and append a - // `p` tag for each (skipping the author, already tagged above). A - // resolution failure must not drop the message, so log and proceed - // with the base tags. + // `p` tag for each (skipping the author, already tagged above). + // Fail before persistence if resolution cannot establish recipients. let members = state .db .get_members(tenant.community(), channel_uuid) @@ -334,14 +339,43 @@ impl ActionSink for RelayActionSink { Some((name, nostr::PublicKey::from_slice(&u.pubkey).ok()?.to_hex())) }) .collect(); + let mut mentioned_pubkeys = Vec::new(); for mentioned in resolve_mention_pubkeys(&text, &named_members) { if mentioned == author_pubkey_hex { continue; } + let mentioned_pubkey = nostr::PublicKey::from_hex(&mentioned) + .map_err(|e| ActionSinkError::EventBuild(format!("mention pubkey: {e}")))?; tags.push( Tag::parse(["p", &mentioned]) .map_err(|e| ActionSinkError::EventBuild(format!("mention p tag: {e}")))?, ); + mentioned_pubkeys.push(mentioned_pubkey); + } + + let definition_event_id = definition_event_id + .as_deref() + .map(nostr::EventId::from_slice) + .transpose() + .map_err(|e| { + ActionSinkError::InvalidInput(format!("invalid definition event id: {e}")) + })?; + if let Some(definition_event_id) = definition_event_id { + tags.push( + Tag::parse(["workflow-run", &run_id.to_string()]).map_err(|e| { + ActionSinkError::EventBuild(format!("workflow run tag: {e}")) + })?, + ); + tags.push( + Tag::parse(["workflow-definition", &definition_event_id.to_hex()]).map_err( + |e| ActionSinkError::EventBuild(format!("workflow definition tag: {e}")), + )?, + ); + tags.push( + Tag::parse(["workflow-step", &step_id]).map_err(|e| { + ActionSinkError::EventBuild(format!("workflow step tag: {e}")) + })?, + ); } let kind = Kind::from(KIND_STREAM_MESSAGE as u16); @@ -387,31 +421,44 @@ impl ActionSink for RelayActionSink { }, }); - let (stored_event, was_inserted) = state + // Build every wake before writing. The message, thread counters, + // mentions, and all recipients commit together; replay can recover + // committed wakes even if the relay dies before publishing them. + let wakes = build_workflow_wakes( + &state.relay_keypair, + channel_uuid, + run_id, + definition_event_id, + event.id, + mentioned_pubkeys, + )?; + let stored = state .db - .insert_event_with_thread_metadata( + .insert_event_with_notifications( tenant.community(), &event, - Some(channel_uuid), + channel_uuid, thread_meta, + &wakes, ) .await - .map_err(|e| ActionSinkError::Database(e.to_string()))?; + .map_err(|error| ActionSinkError::Database(error.to_string()))?; + let was_inserted = stored.first().is_some_and(|(_, inserted)| *inserted); + for (stored_event, inserted) in &stored { + if *inserted { + let _ = dispatch_persistent_event( + &tenant, + &state, + stored_event, + u32::from(stored_event.event.kind.as_u16()), + &author_pubkey_hex, + None, + ) + .await; + } + } - // 5. Post-persist side effects (fan-out, search, audit) - // Only if actually inserted (idempotency guard). if was_inserted { - let _ = dispatch_persistent_event( - &tenant, - &state, - &stored_event, - kind_u32, - &author_pubkey_hex, - None, - ) - .await; - - // A threaded reply changed its thread's counters — push a fresh // relay-signed kind:39005 so subscribed clients update badge // counts without refetching the head window, exactly as the // ingest path does after a reply insert. Fan-out-only and @@ -431,6 +478,33 @@ impl ActionSink for RelayActionSink { } } +fn build_workflow_wakes( + relay_keys: &nostr::Keys, + channel_id: Uuid, + run_id: Uuid, + definition_event_id: Option, + message_event_id: nostr::EventId, + recipients: Vec, +) -> Result, ActionSinkError> { + let Some(definition_event_id) = definition_event_id else { + return Ok(Vec::new()); + }; + recipients + .into_iter() + .map(|recipient| { + WorkflowMentionWake::new( + recipient, + channel_id, + run_id, + definition_event_id, + message_event_id, + ) + .sign(relay_keys) + .map_err(|error| ActionSinkError::EventBuild(format!("workflow wake: {error}"))) + }) + .collect() +} + #[cfg(test)] mod tests { use super::*; @@ -444,6 +518,62 @@ mod tests { std::iter::repeat_n(nibble, 64).collect() } + #[test] + fn legacy_message_without_revision_emits_no_wake() { + let relay = nostr::Keys::generate(); + let recipient = nostr::Keys::generate().public_key(); + let message = nostr::EventBuilder::text_note("message") + .sign_with_keys(&relay) + .expect("message"); + let wakes = build_workflow_wakes( + &relay, + Uuid::new_v4(), + Uuid::new_v4(), + None, + message.id, + vec![recipient], + ) + .expect("legacy wake build"); + assert!(wakes.is_empty()); + } + + #[test] + fn revision_bound_message_emits_one_identifier_wake_per_recipient() { + let relay = nostr::Keys::generate(); + let recipients = [ + nostr::Keys::generate().public_key(), + nostr::Keys::generate().public_key(), + ]; + let channel = Uuid::new_v4(); + let run = Uuid::new_v4(); + let definition = nostr::EventBuilder::text_note("definition") + .sign_with_keys(&nostr::Keys::generate()) + .expect("definition"); + let message = nostr::EventBuilder::text_note("message") + .sign_with_keys(&relay) + .expect("message"); + let wakes = build_workflow_wakes( + &relay, + channel, + run, + Some(definition.id), + message.id, + recipients.to_vec(), + ) + .expect("wake build"); + assert_eq!(wakes.len(), recipients.len()); + for (event, recipient) in wakes.iter().zip(recipients) { + let wake = WorkflowMentionWake::parse(event).expect("canonical wake"); + assert!(event.content.is_empty()); + assert_eq!(event.pubkey, relay.public_key()); + assert_eq!(wake.recipient(), recipient); + assert_eq!(wake.channel_id(), channel); + assert_eq!(wake.run_id(), run); + assert_eq!(wake.definition_event_id(), definition.id); + assert_eq!(wake.message_event_id(), message.id); + } + } + #[test] fn resolves_exact_member_name() { let members = vec![m("Robby", &pk('a'))]; @@ -626,7 +756,7 @@ mod tests { } #[cfg(test)] -mod integration_tests { +pub(crate) mod integration_tests { //! Regression test for `e3661764` / `7899c1a8`: a workflow `send_message` //! that mentions a channel member by name (`@Name`) must emit a `p` tag for //! that member so ACP agent wake (`event_mentions_agent`, p-tag gated) fires. @@ -639,10 +769,15 @@ mod integration_tests { use std::sync::Arc; /// Real-PG state mirroring `handlers::event::tests::test_state_with_redis_url`. - async fn test_state() -> Arc { + pub(crate) async fn test_state() -> Arc { + test_state_with_redis("redis://127.0.0.1:1".to_string()).await + } + + pub(crate) async fn test_state_with_redis(redis_url: String) -> Arc { let mut config = crate::config::Config::from_env().expect("default config loads"); config.require_relay_membership = false; - config.redis_url = "redis://127.0.0.1:1".to_string(); + config.require_auth_token = false; + config.redis_url = redis_url; let pool = sqlx::PgPool::connect_lazy(&config.database_url).expect("lazy pg pool"); let db = buzz_db::Db::from_pool(pool.clone()); let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) @@ -739,7 +874,12 @@ mod integration_tests { let sink = RelayActionSink::new(&state); let event_id_hex = sink .send_message( - community, + WorkflowMessageContext { + community_id: community, + run_id: Uuid::new_v4(), + step_id: "test-step".into(), + definition_event_id: None, + }, &channel.id.to_string(), "heads up @Robby — please take a look", &author_hex, @@ -815,7 +955,12 @@ mod integration_tests { // 1. A top-level workflow message becomes the thread root. let root_hex = sink .send_message( - community, + WorkflowMessageContext { + community_id: community, + run_id: Uuid::new_v4(), + step_id: "test-step".into(), + definition_event_id: None, + }, &channel.id.to_string(), "root message", &author_hex, @@ -827,7 +972,12 @@ mod integration_tests { // 2. A reply_in_thread message threads onto it. let reply_hex = sink .send_message( - community, + WorkflowMessageContext { + community_id: community, + run_id: Uuid::new_v4(), + step_id: "test-step".into(), + definition_event_id: None, + }, &channel.id.to_string(), "threaded reply", &author_hex, @@ -969,7 +1119,12 @@ mod integration_tests { // A workflow reply onto the metadata-less nested parent. let reply_hex = RelayActionSink::new(&state) .send_message( - community, + WorkflowMessageContext { + community_id: community, + run_id: Uuid::new_v4(), + step_id: "test-step".into(), + definition_event_id: None, + }, &channel_hex, "workflow reply", &author_hex, @@ -1050,7 +1205,12 @@ mod integration_tests { let root_only_reply_hex = RelayActionSink::new(&state) .send_message( - community, + WorkflowMessageContext { + community_id: community, + run_id: Uuid::new_v4(), + step_id: "test-step".into(), + definition_event_id: None, + }, &channel_hex, "workflow reply to root-only parent", &author_hex, @@ -1112,7 +1272,12 @@ mod integration_tests { let unknown = nostr::Keys::generate().public_key().to_hex(); let err = RelayActionSink::new(&state) .send_message( - community, + WorkflowMessageContext { + community_id: community, + run_id: Uuid::new_v4(), + step_id: "test-step".into(), + definition_event_id: None, + }, &channel.id.to_string(), "orphan reply", &author_hex, @@ -1126,3 +1291,7 @@ mod integration_tests { ); } } + +#[cfg(test)] +#[path = "workflow_delivery_tests.rs"] +mod workflow_delivery_tests; diff --git a/crates/buzz-search/tests/fts_integration.rs b/crates/buzz-search/tests/fts_integration.rs index 175a01aaaa3..8ca027ebee1 100644 --- a/crates/buzz-search/tests/fts_integration.rs +++ b/crates/buzz-search/tests/fts_integration.rs @@ -30,8 +30,14 @@ const MIGRATION_0008_SQL: &str = const MIGRATION_0014_SQL: &str = include_str!("../../../migrations/0014_push_lease_fts.sql"); const MIGRATION_0033_SQL: &str = include_str!("../../../migrations/0033_private_managed_agent_fts.sql"); +const MIGRATION_0041_SQL: &str = + include_str!("../../../migrations/0041_workflow_mention_wake_fts.sql"); async fn setup() -> (PgPool, String) { + setup_with_search_policy(true).await +} + +async fn setup_with_search_policy(apply_fresh_allowlist: bool) -> (PgPool, String) { let url = std::env::var("BUZZ_TEST_DATABASE_URL").unwrap_or_else(|_| TEST_DB_URL.to_string()); let schema = format!("fts_test_{}", Uuid::new_v4().simple()); // Connect to the default schema first to create the test schema. @@ -77,15 +83,20 @@ async fn setup() -> (PgPool, String) { pool.execute(MIGRATION_0007_SQL) .await .expect("apply 0007 migration"); - pool.execute(MIGRATION_0008_SQL) - .await - .expect("apply 0008 migration"); + if apply_fresh_allowlist { + pool.execute(MIGRATION_0008_SQL) + .await + .expect("apply 0008 migration"); + } pool.execute(MIGRATION_0014_SQL) .await .expect("apply 0014 migration"); pool.execute(MIGRATION_0033_SQL) .await .expect("apply 0033 migration"); + pool.execute(MIGRATION_0041_SQL) + .await + .expect("apply 0036 migration"); (pool, schema) } @@ -1414,8 +1425,8 @@ async fn author_only_kinds_are_storage_level_unsearchable() { /// search entry point could surface tokenized content from these kinds. The /// L1 NULL tsvector is the unbreakable backstop: `@@` mathematically cannot /// match NULL. This test catches the drift where someone adds a persistent -/// kind to `P_GATED_KINDS` without the matching `schema/schema.sql` + -/// `migrations/0001_initial_schema.sql` skip-set update. +/// kind to `P_GATED_KINDS` without the matching desired schema and forward +/// migration exclusion. /// /// Ephemeral kinds (20000–29999) are skipped: they are never stored, so the /// storage-layer defense does not apply to them regardless of the schema @@ -1428,7 +1439,10 @@ async fn author_only_kinds_are_storage_level_unsearchable() { #[tokio::test] #[ignore = "requires Postgres"] async fn p_gated_persistent_kinds_have_storage_null_tsvector() { - let (pool, schema) = setup().await; + // Exercise the brownfield negative skip-set. The fresh-install positive + // allowlist would make every unknown kind unsearchable and let a missing + // per-kind migration pass vacuously. + let (pool, schema) = setup_with_search_policy(false).await; let c = mk_community(&pool, "p-gated-tripwire.example").await; let token = "pgated_tripwire_marker_qwerty"; diff --git a/crates/buzz-test-client/tests/e2e_relay.rs b/crates/buzz-test-client/tests/e2e_relay.rs index b119d267740..d1f6f0db9a0 100644 --- a/crates/buzz-test-client/tests/e2e_relay.rs +++ b/crates/buzz-test-client/tests/e2e_relay.rs @@ -222,40 +222,54 @@ async fn test_connect_and_authenticate() { #[tokio::test] #[ignore] -async fn test_client_submitted_nip43_membership_snapshots_are_rejected() { +async fn test_client_submitted_relay_only_events_are_rejected() { let url = relay_url(); let keys = Keys::generate(); - // Prove this actor can submit a normal event so the rejection below is + // Prove this actor can submit a normal event so the rejections below are // specifically the relay-only invariant, not a broader authorization failure. - create_test_channel(&keys).await; - let forged = EventBuilder::new(Kind::Custom(13_534), "") - .tags([Tag::parse(["member", &keys.public_key().to_hex(), "owner"]).unwrap()]) - .sign_with_keys(&keys) - .expect("sign forged membership snapshot"); + let channel_id = create_test_channel(&keys).await; + let forged_events = [ + EventBuilder::new(Kind::Custom(13_534), "") + .tags([Tag::parse(["member", &keys.public_key().to_hex(), "owner"]).unwrap()]) + .sign_with_keys(&keys) + .expect("sign forged membership snapshot"), + EventBuilder::new(Kind::Custom(44_620), "") + .tags([ + Tag::parse(["p", &keys.public_key().to_hex()]).unwrap(), + Tag::parse(["h", &channel_id.to_string()]).unwrap(), + Tag::parse(["run", &Uuid::new_v4().to_string()]).unwrap(), + Tag::parse(["definition", &"11".repeat(32)]).unwrap(), + Tag::parse(["message", &"22".repeat(32)]).unwrap(), + ]) + .sign_with_keys(&keys) + .expect("sign forged workflow wake"), + ]; let mut ws = BuzzTestClient::connect(&url, &keys).await.expect("connect"); - let ok = ws - .send_event(forged.clone()) - .await - .expect("submit forged snapshot via websocket"); - assert!(!ok.accepted, "forged WebSocket snapshot must be rejected"); - assert_eq!(ok.message, "restricted: relay-only kind"); + for forged in forged_events { + let ok = ws + .send_event(forged.clone()) + .await + .expect("submit forged relay-only event via websocket"); + assert!(!ok.accepted, "forged WebSocket event must be rejected"); + assert_eq!(ok.message, "restricted: relay-only kind"); + + let response = reqwest::Client::new() + .post(format!("{}/events", relay_http_url())) + .header("X-Pubkey", keys.public_key().to_hex()) + .header("Content-Type", "application/json") + .body(serde_json::to_string(&forged).unwrap()) + .send() + .await + .expect("submit forged relay-only event via HTTP"); + assert_eq!(response.status(), reqwest::StatusCode::BAD_REQUEST); + let body = response.text().await.expect("read HTTP rejection"); + assert!( + body.contains("restricted: relay-only kind"), + "unexpected HTTP rejection: {body}" + ); + } ws.disconnect().await.expect("disconnect"); - - let response = reqwest::Client::new() - .post(format!("{}/events", relay_http_url())) - .header("X-Pubkey", keys.public_key().to_hex()) - .header("Content-Type", "application/json") - .body(serde_json::to_string(&forged).unwrap()) - .send() - .await - .expect("submit forged snapshot via HTTP"); - assert_eq!(response.status(), reqwest::StatusCode::BAD_REQUEST); - let body = response.text().await.expect("read HTTP rejection"); - assert!( - body.contains("restricted: relay-only kind"), - "unexpected HTTP rejection: {body}" - ); } #[tokio::test] diff --git a/crates/buzz-workflow/src/action_sink.rs b/crates/buzz-workflow/src/action_sink.rs index 079c27a913d..ebbc485665f 100644 --- a/crates/buzz-workflow/src/action_sink.rs +++ b/crates/buzz-workflow/src/action_sink.rs @@ -7,6 +7,21 @@ use std::future::Future; use std::pin::Pin; use buzz_core::tenant::CommunityId; +use uuid::Uuid; + +/// Workflow authority carried into a message side effect. +#[derive(Debug, Clone)] +pub struct WorkflowMessageContext { + /// Server-resolved community that owns the workflow run. + pub community_id: CommunityId, + /// Exact workflow run driving this side effect. + pub run_id: Uuid, + /// Exact signed-definition step being executed. + pub step_id: String, + /// Exact signed definition selected when the run was created, or `None` + /// for a legacy run that must not emit an automatic wake. + pub definition_event_id: Option>, +} /// Errors from action sink operations. #[derive(Debug, thiserror::Error)] @@ -48,11 +63,8 @@ impl From for crate::WorkflowError { pub trait ActionSink: Send + Sync { /// Post a message to a channel on behalf of a workflow owner. /// - /// - `community_id`: the server-resolved community that owns the workflow - /// run driving this side effect. The relay-signed message is published - /// under *this* community, never the deployment/default tenant — the run - /// carries its owning community so a workflow in community B posts into B - /// even though the side effect has no inbound connection to bind. + /// - `context`: exact workflow authority driving this side effect, including + /// the server-resolved community and optional signed-definition revision /// - `channel_id`: UUID string of the target channel /// - `text`: message body (must not be empty/whitespace-only) /// - `author_pubkey`: hex-encoded pubkey of the workflow owner (used for @@ -64,7 +76,7 @@ pub trait ActionSink: Send + Sync { /// Returns the event ID hex string on success. fn send_message( &self, - community_id: CommunityId, + context: WorkflowMessageContext, channel_id: &str, text: &str, author_pubkey: &str, diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index 5c712dcff7c..bbd2c222f49 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -18,6 +18,7 @@ use serde_json::Value as JsonValue; use tracing::{debug, info, warn}; use uuid::Uuid; +use crate::action_sink::WorkflowMessageContext; use crate::error::WorkflowError; use crate::schema::{ActionDef, Step, WorkflowDef}; use crate::WorkflowEngine; @@ -625,7 +626,12 @@ pub async fn dispatch_action( let event_id = engine .action_sink()? .send_message( - community_id, + WorkflowMessageContext { + community_id, + run_id, + step_id: step_id.to_owned(), + definition_event_id: wf_run.definition_event_id.clone(), + }, &channel_id, text, &owner_pubkey_hex, diff --git a/migrations/0041_workflow_mention_wake_fts.sql b/migrations/0041_workflow_mention_wake_fts.sql new file mode 100644 index 00000000000..a0c42c95edc --- /dev/null +++ b/migrations/0041_workflow_mention_wake_fts.sql @@ -0,0 +1,32 @@ +-- Kind:44620 is a durable, recipient-gated workflow mention wake. Its canonical +-- content is empty, but keep the storage-level full-text-search backstop aligned +-- with every persistent P_GATED_KINDS member, including on brownfield databases +-- that retain the legacy negative skip-set. +-- +-- Preserve the database's existing search policy for every other kind. As with +-- 0014 and 0033, replacing this generated column rewrites the events table and +-- rebuilds the GIN index under an ACCESS EXCLUSIVE lock. +DO $$ +DECLARE + existing_expression TEXT; +BEGIN + SELECT pg_get_expr(d.adbin, d.adrelid) + INTO existing_expression + FROM pg_attrdef d + JOIN pg_attribute a + ON a.attrelid = d.adrelid + AND a.attnum = d.adnum + WHERE d.adrelid = 'events'::regclass + AND a.attname = 'search_tsv'; + + IF existing_expression IS NULL THEN + RAISE EXCEPTION 'events.search_tsv generated expression not found'; + END IF; + + ALTER TABLE events DROP COLUMN search_tsv; + EXECUTE format( + 'ALTER TABLE events ADD COLUMN search_tsv TSVECTOR GENERATED ALWAYS AS (CASE WHEN kind = 44620 THEN NULL::tsvector ELSE (%s) END) STORED', + existing_expression + ); + CREATE INDEX idx_events_search_tsv ON events USING GIN (search_tsv); +END $$; diff --git a/migrations/0042_workflow_superseded_authority.sql b/migrations/0042_workflow_superseded_authority.sql new file mode 100644 index 00000000000..808ccf7ad72 --- /dev/null +++ b/migrations/0042_workflow_superseded_authority.sql @@ -0,0 +1,3 @@ +-- Supersession retains captured workflow authority; explicit deletion revokes it. +-- Do not infer a deletion reason for historical rows: unknown stays fail-closed. +ALTER TABLE events ADD COLUMN workflow_revision_superseded BOOLEAN NOT NULL DEFAULT false; diff --git a/schema/schema.sql b/schema/schema.sql index b0656ed8959..da1c2632d35 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -219,9 +219,9 @@ CREATE TABLE events ( -- Privacy: encrypted/private routing wrappers and p-gated membership notices -- must never be discoverable through NIP-50 full-text search. NULL tsvector -- never matches `@@`. - -- Keep in sync with migrations (final state: 0001 + 0005 + 0014 + 0033). + -- Keep in sync with migrations (final state: 0001 + 0005 + 0014 + 0033 + 0036). search_tsv TSVECTOR GENERATED ALWAYS AS ( - CASE WHEN kind IN (1059, 30179, 30300, 30350, 30622, 44100, 44101, 44200) THEN NULL::tsvector + CASE WHEN kind IN (1059, 30179, 30300, 30350, 30622, 44100, 44101, 44200, 44620) THEN NULL::tsvector ELSE to_tsvector('simple', content) END ) STORED, @@ -229,6 +229,7 @@ CREATE TABLE events ( received_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), channel_id UUID, deleted_at TIMESTAMPTZ, + workflow_revision_superseded BOOLEAN NOT NULL DEFAULT false, d_tag TEXT, not_before BIGINT, delivered_at BIGINT,