From 346450f9400b562a8bc342036a4d7e522af4ae60 Mon Sep 17 00:00:00 2001 From: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> Date: Tue, 28 Jul 2026 19:33:17 +0900 Subject: [PATCH 01/27] feat(workflow): reject unimplemented actions at validate() time Surfaces send_dm and set_channel_topic gaps at definition time (YAML import / REST ingest) instead of failing at runtime with WorkflowError::NotImplemented. The guard is temporary and will be removed per-action as each is implemented in subsequent steps. Added serde-only parse tests for the two actions to confirm parsing still works independent of the validate() guard. Signed-off-by: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> --- crates/buzz-workflow/src/schema.rs | 79 +++++++++++++++++++++++++----- 1 file changed, 67 insertions(+), 12 deletions(-) diff --git a/crates/buzz-workflow/src/schema.rs b/crates/buzz-workflow/src/schema.rs index 34e8bb1960..a35cf4ffcd 100644 --- a/crates/buzz-workflow/src/schema.rs +++ b/crates/buzz-workflow/src/schema.rs @@ -190,6 +190,26 @@ impl WorkflowDef { step.id ))); } + + // Temporary guard: reject actions that are not yet fully implemented. + // This surfaces the gap at definition time (YAML import / REST ingest) + // instead of failing at runtime with WorkflowError::NotImplemented. + // Each action is removed from this guard as it is completed. + match &step.action { + ActionDef::SendDm { .. } => { + return Err(WorkflowError::InvalidDefinition(format!( + "step '{}' uses 'send_dm' which is not yet implemented", + step.id + ))); + } + ActionDef::SetChannelTopic { .. } => { + return Err(WorkflowError::InvalidDefinition(format!( + "step '{}' uses 'set_channel_topic' which is not yet implemented", + step.id + ))); + } + _ => {} + } } if let TriggerDef::Schedule { cron, interval } = &self.trigger { @@ -335,43 +355,40 @@ mod tests { fn parse_all_action_types() { // Avoid "# in YAML values (would close r# raw strings). // Use unquoted or single-quoted YAML values throughout. + // Note: send_dm and set_channel_topic are excluded here because validate() + // rejects them as not-yet-implemented. They have dedicated parse tests + // below (parse_send_dm_action, parse_set_channel_topic_action) that use + // serde_yaml directly to verify parsing without the validate() guard. let yaml = concat!( "name: All Actions\n", "trigger:\n on: webhook\n", "steps:\n", " - id: msg\n action: send_message\n text: Hello\n channel: general\n", - " - id: dm\n action: send_dm\n to: '{{trigger.author}}'\n text: You triggered this\n", - " - id: topic\n action: set_channel_topic\n topic: Status active\n", " - id: react\n action: add_reaction\n emoji: white_check_mark\n", " - id: hook\n action: call_webhook\n url: https://hooks.example.com/notify\n method: POST\n", " - id: approve\n action: request_approval\n from: '@manager'\n message: Approve?\n timeout: 4h\n", " - id: wait\n action: delay\n duration: 5m\n", ); let (def, _) = parse_yaml(yaml).expect("parse failed"); - assert_eq!(def.steps.len(), 7); + assert_eq!(def.steps.len(), 5); assert!(matches!( &def.steps[0].action, ActionDef::SendMessage { .. } )); - assert!(matches!(&def.steps[1].action, ActionDef::SendDm { .. })); - assert!(matches!( - &def.steps[2].action, - ActionDef::SetChannelTopic { .. } - )); assert!(matches!( - &def.steps[3].action, + &def.steps[1].action, ActionDef::AddReaction { .. } )); assert!(matches!( - &def.steps[4].action, + &def.steps[2].action, ActionDef::CallWebhook { .. } )); assert!(matches!( - &def.steps[5].action, + &def.steps[3].action, ActionDef::RequestApproval { .. } )); - assert!(matches!(&def.steps[6].action, ActionDef::Delay { .. })); + assert!(matches!(&def.steps[4].action, ActionDef::Delay { .. })); } #[test] @@ -391,6 +408,44 @@ mod tests { assert_eq!(def.steps.len(), 3); } + #[test] + fn validate_rejects_unimplemented_send_dm() { + let yaml = "name: DM Test\ntrigger:\n on: webhook\nsteps:\n - id: dm\n action: send_dm\n to: abc123\n text: hi\n"; + let err = parse_yaml(yaml).unwrap_err(); + assert!( + err.to_string().contains("send_dm"), + "expected send_dm rejection, got: {err}" + ); + } + + #[test] + fn validate_rejects_unimplemented_set_channel_topic() { + let yaml = "name: Topic Test\ntrigger:\n on: webhook\nsteps:\n - id: topic\n action: set_channel_topic\n topic: new\n"; + let err = parse_yaml(yaml).unwrap_err(); + assert!( + err.to_string().contains("set_channel_topic"), + "expected set_channel_topic rejection, got: {err}" + ); + } + + #[test] + fn parse_send_dm_action_via_serde() { + // verify serde parsing works even though validate() rejects it + let yaml = "name: DM Parse\ntrigger:\n on: webhook\nsteps:\n - id: dm\n action: send_dm\n to: abc123\n text: hi\n"; + let def: WorkflowDef = serde_yaml::from_str(yaml).expect("serde parse"); + assert!(matches!(def.steps[0].action, ActionDef::SendDm { .. })); + } + + #[test] + fn parse_set_channel_topic_action_via_serde() { + let yaml = "name: Topic Parse\ntrigger:\n on: webhook\nsteps:\n - id: topic\n action: set_channel_topic\n topic: new\n"; + let def: WorkflowDef = serde_yaml::from_str(yaml).expect("serde parse"); + assert!(matches!( + def.steps[0].action, + ActionDef::SetChannelTopic { .. } + )); + } + #[test] fn validate_rejects_empty_name() { let yaml = From aa19d8584ebd66fe710c00fe634e221dbe936366 Mon Sep 17 00:00:00 2001 From: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> Date: Tue, 28 Jul 2026 19:39:57 +0900 Subject: [PATCH 02/27] feat(workflow): cap cross-workflow trigger-chain depth MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds workflow_depth to TriggerContext and stamps it into the buzz:workflow tag on emitted messages (third element). The engine suppresses triggering beyond MAX_WORKFLOW_DEPTH (3), catching cross-workflow loops (A→B→A) that the relay's single-hop buzz:workflow tag check cannot. Legacy two-element tags (no depth) are treated as depth 0 on read, so existing workflow messages still parse. Signed-off-by: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> --- crates/buzz-relay/src/workflow_sink.rs | 9 ++- crates/buzz-workflow/src/action_sink.rs | 5 ++ crates/buzz-workflow/src/executor.rs | 17 ++++- crates/buzz-workflow/src/lib.rs | 99 +++++++++++++++++++++++++ 4 files changed, 126 insertions(+), 4 deletions(-) diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 97c31c2561..4935fc0e83 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -176,6 +176,7 @@ impl ActionSink for RelayActionSink { channel_id: &str, text: &str, author_pubkey: &str, + workflow_depth: usize, ) -> Pin> + Send + '_>> { let channel_id = channel_id.to_owned(); let text = text.to_owned(); @@ -254,15 +255,19 @@ impl ActionSink for RelayActionSink { // - Signed by relay keypair (event.pubkey = relay pubkey) // - `p` tag attributes the message to the workflow owner // - `h` tag scopes to the channel (NIP-29, canonical UUID) - // - `buzz:workflow` tag prevents recursive workflow triggering + // - `buzz:workflow` tag carries the trigger-chain depth so the + // engine can cap cross-workflow loops (A→B→A). Legacy form + // `["buzz:workflow","true"]` (no depth) is treated as depth 0 + // on read, so older workflow messages still parse. // - one `p` tag per `@Name` that resolves to a channel member, // so mentioned agents are woken (wake is `p`-tag gated) + let workflow_tag_depth = workflow_depth.to_string(); let mut tags = vec![ Tag::parse(["p", &author_pubkey_hex]) .map_err(|e| ActionSinkError::EventBuild(format!("p tag: {e}")))?, Tag::parse(["h", &channel_id_canonical]) .map_err(|e| ActionSinkError::EventBuild(format!("h tag: {e}")))?, - Tag::parse(["buzz:workflow", "true"]) + Tag::parse(["buzz:workflow", "true", &workflow_tag_depth]) .map_err(|e| ActionSinkError::EventBuild(format!("workflow tag: {e}")))?, ]; diff --git a/crates/buzz-workflow/src/action_sink.rs b/crates/buzz-workflow/src/action_sink.rs index 0c6002e74e..4559e7b688 100644 --- a/crates/buzz-workflow/src/action_sink.rs +++ b/crates/buzz-workflow/src/action_sink.rs @@ -57,6 +57,10 @@ pub trait ActionSink: Send + Sync { /// - `text`: message body (must not be empty/whitespace-only) /// - `author_pubkey`: hex-encoded pubkey of the workflow owner (used for /// the `p` attribution tag; the relay keypair signs the event) + /// - `workflow_depth`: the trigger-chain depth of the run emitting this + /// message. Stamped into the `buzz:workflow` tag so a downstream workflow + /// that triggers off this event can compute its own depth and the engine + /// can cap the chain (see `MAX_WORKFLOW_DEPTH`). /// /// Returns the event ID hex string on success. fn send_message( @@ -65,5 +69,6 @@ pub trait ActionSink: Send + Sync { channel_id: &str, text: &str, author_pubkey: &str, + workflow_depth: usize, ) -> Pin> + Send + '_>>; } diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index a029b44622..506cb05592 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -39,6 +39,13 @@ pub struct TriggerContext { pub message_id: String, /// Arbitrary webhook body fields (webhook trigger). pub webhook_fields: HashMap, + /// Workflow trigger-chain depth. `0` for a human-authored event; + /// `N+1` when the event was produced by a workflow run at depth `N`. + /// Capped by [`crate::MAX_WORKFLOW_DEPTH`] to prevent cross-workflow + /// loops (A→B→A) that the single-hop `buzz:workflow` tag check cannot + /// catch. + #[serde(default)] + pub workflow_depth: usize, } impl TriggerContext { @@ -567,7 +574,13 @@ pub async fn dispatch_action( let event_id = engine .action_sink()? - .send_message(community_id, &channel_id, text, &owner_pubkey_hex) + .send_message( + community_id, + &channel_id, + text, + &owner_pubkey_hex, + trigger_ctx.workflow_depth, + ) .await .map_err(WorkflowError::from)?; @@ -1226,7 +1239,7 @@ mod tests { timestamp: "1700000000".to_owned(), emoji: "fire".to_owned(), message_id: "event-id-hex".to_owned(), - webhook_fields: HashMap::new(), + ..Default::default() } } diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index 93581225ee..dcbd8b3078 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -53,6 +53,16 @@ use dashmap::DashMap; use tokio::sync::Semaphore; use uuid::Uuid; +/// Maximum depth of the workflow trigger chain. +/// +/// A human-authored event starts at depth 0. When a workflow run produces an +/// event that triggers another workflow, the depth increments. This cap +/// prevents unbounded cross-workflow loops (A→B→A→…) that the single-hop +/// `buzz:workflow` tag check in the relay cannot catch — that check only +/// suppresses the *immediate* echo of a relay-authored message back to the +/// same harness, not transitive chains across distinct workflows. +pub const MAX_WORKFLOW_DEPTH: usize = 3; + /// Runtime configuration for the workflow engine. #[derive(Clone, Debug)] pub struct WorkflowConfig { @@ -315,6 +325,25 @@ impl WorkflowEngine { let trigger_ctx = build_trigger_context(event); + // Suppress triggering when the event is already deep in a workflow + // chain. This catches cross-workflow loops (A→B→A) that the relay's + // single-hop `buzz:workflow` tag check cannot — that check only + // suppresses the immediate echo of a relay-authored message, not + // transitive chains across distinct workflows. Depth 0 = human event. + if trigger_ctx.workflow_depth > MAX_WORKFLOW_DEPTH { + tracing::warn!( + community_id = %community_id, + channel_id = %channel_id, + depth = trigger_ctx.workflow_depth, + event_id = %event.event.id.to_hex(), + "Suppressing workflow trigger — chain depth {} exceeds max {} \ + (possible cross-workflow loop)", + trigger_ctx.workflow_depth, + MAX_WORKFLOW_DEPTH, + ); + return Ok(()); + } + let trigger_ctx_json: serde_json::Value = match serde_json::to_value(&trigger_ctx) { Ok(v) => v, Err(e) => { @@ -885,6 +914,32 @@ pub fn build_trigger_context(event: &buzz_core::StoredEvent) -> executor::Trigge let kind_u32 = event_kind_u32(&event.event); let content = event.event.content.clone(); + // Extract the workflow trigger-chain depth from the `buzz:workflow` tag. + // Workflow-emitted events carry `["buzz:workflow", "true", ""]`; + // human-authored events have no such tag (depth 0). A workflow at depth N + // emits events tagged depth N, so the triggered workflow runs at depth N+1. + let workflow_depth = event + .event + .tags + .iter() + .find_map(|tag| { + let parts = tag.as_slice(); + if parts.first().map(|s| s.as_str()) == Some("buzz:workflow") { + // Third element (index 2) is the depth; fall back to 0 for the + // legacy two-element form `["buzz:workflow", "true"]`. + parts + .get(2) + .and_then(|s| s.parse::().ok()) + .map(|d| d + 1) + // Legacy tag without a depth field → this event was emitted + // by a workflow at an unknown (treat as 0) depth. + .or(Some(1)) + } else { + None + } + }) + .unwrap_or(0); + let author = event .event .tags @@ -948,6 +1003,7 @@ pub fn build_trigger_context(event: &buzz_core::StoredEvent) -> executor::Trigge emoji, message_id, webhook_fields: HashMap::new(), + workflow_depth, } } @@ -1537,6 +1593,49 @@ steps: .expect("timestamp should be a u64 string"); } + #[test] + fn build_trigger_context_human_event_has_depth_zero() { + let stored = make_message_event(); + let ctx = build_trigger_context(&stored); + assert_eq!(ctx.workflow_depth, 0, "human-authored event → depth 0"); + } + + #[test] + fn build_trigger_context_workflow_event_increments_depth() { + use nostr::{EventBuilder, Keys, Kind, Tag}; + use uuid::Uuid; + let keys = Keys::generate(); + // Simulate a workflow-emitted event tagged depth 2. + let wf_tag = Tag::parse(["buzz:workflow", "true", "2"]).expect("tag parse"); + let event = EventBuilder::new(Kind::Custom(9), "workflow msg") + .tags([wf_tag]) + .sign_with_keys(&keys) + .expect("sign"); + let stored = buzz_core::StoredEvent::new(event, Some(Uuid::new_v4())); + let ctx = build_trigger_context(&stored); + assert_eq!( + ctx.workflow_depth, 3, + "depth-2 event → triggered workflow at 3" + ); + } + + #[test] + fn build_trigger_context_legacy_workflow_tag_treated_as_depth_1() { + // Legacy two-element tag `["buzz:workflow", "true"]` (no depth field). + use nostr::{EventBuilder, Keys, Kind, Tag}; + use uuid::Uuid; + let keys = Keys::generate(); + let wf_tag = Tag::parse(["buzz:workflow", "true"]).expect("tag parse"); + let event = EventBuilder::new(Kind::Custom(9), "legacy workflow msg") + .tags([wf_tag]) + .sign_with_keys(&keys) + .expect("sign"); + let stored = buzz_core::StoredEvent::new(event, Some(Uuid::new_v4())); + let ctx = build_trigger_context(&stored); + // Legacy tag → treat the emitting workflow as depth 0, so this is 1. + assert_eq!(ctx.workflow_depth, 1); + } + #[test] fn test_build_trigger_context_reaction_multiple_e_tags() { // NIP-25: last e tag is the direct target, first may be thread root From 6a90227dd27183d0b329772b81897b7c4ad064da Mon Sep 17 00:00:00 2001 From: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> Date: Tue, 28 Jul 2026 19:43:51 +0900 Subject: [PATCH 03/27] feat(workflow): implement set_channel_topic action MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds ActionSink::set_channel_topic, implemented in RelayActionSink as a NIP-29 edit-metadata event (kind:9002) with a topic tag, signed by the relay keypair. The relay's existing side-effect handler (handle_edit_metadata) applies the topic change during ingest after membership/permission checks. Removes the set_channel_topic guard from validate() — the action is now fully wired. Updated tests accordingly. Signed-off-by: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> --- crates/buzz-relay/src/workflow_sink.rs | 95 +++++++++++++++++++++++++ crates/buzz-workflow/src/action_sink.rs | 23 ++++++ crates/buzz-workflow/src/executor.rs | 52 ++++++++++++-- crates/buzz-workflow/src/schema.rs | 19 ++--- 4 files changed, 173 insertions(+), 16 deletions(-) diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 4935fc0e83..5dfca95946 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -367,6 +367,101 @@ impl ActionSink for RelayActionSink { Ok(event_id_hex) }) } + + fn set_channel_topic( + &self, + community_id: CommunityId, + channel_id: &str, + topic: &str, + author_pubkey: &str, + workflow_depth: usize, + ) -> Pin> + Send + '_>> { + let channel_id = channel_id.to_owned(); + let topic = topic.to_owned(); + let author_pubkey = author_pubkey.to_owned(); + + Box::pin(async move { + let state = self + .state + .upgrade() + .ok_or_else(|| ActionSinkError::Database("relay is shutting down".into()))?; + + // Resolve the run's owning community → tenant (same rationale as + // send_message: a community-B workflow must mutate B's channel). + let host = state + .db + .lookup_community_host(community_id) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))? + .ok_or_else(|| { + ActionSinkError::Database(format!( + "workflow run community {community_id} is not mapped to a host" + )) + })?; + let tenant = buzz_core::tenant::TenantContext::resolved(community_id, host); + + if topic.trim().is_empty() { + return Err(ActionSinkError::EmptyContent); + } + + let channel_uuid = Uuid::parse_str(&channel_id) + .map_err(|e| ActionSinkError::InvalidInput(format!("invalid UUID: {e}")))?; + + // Build a NIP-29 edit-metadata event (kind:9002) with a `topic` + // tag, plus attribution `p` and `buzz:workflow` (depth) tags. + // The SDK's build_set_topic produces h+topic tags; we rebuild with + // the full tag set here so the extra tags ride along. The relay's + // side-effect handler (handle_edit_metadata) applies the topic + // change during ingest after membership/permission checks. + let workflow_tag_depth = workflow_depth.to_string(); + let h_tag = Tag::parse(["h", &channel_id]) + .map_err(|e| ActionSinkError::EventBuild(format!("h tag: {e}")))?; + let topic_tag = Tag::parse(["topic", &topic]) + .map_err(|e| ActionSinkError::EventBuild(format!("topic tag: {e}")))?; + let p_tag = Tag::parse(["p", &author_pubkey]) + .map_err(|e| ActionSinkError::EventBuild(format!("p tag: {e}")))?; + let wf_tag = Tag::parse(["buzz:workflow", "true", &workflow_tag_depth]) + .map_err(|e| ActionSinkError::EventBuild(format!("workflow tag: {e}")))?; + let event = EventBuilder::new(Kind::Custom(9002), "") + .tags([h_tag, topic_tag, p_tag, wf_tag]) + .sign_with_keys(&state.relay_keypair) + .map_err(|e| ActionSinkError::EventBuild(format!("sign: {e}")))?; + + let kind_u32: u32 = 9002; + let (stored_event, was_inserted) = state + .db + .insert_event(tenant.community(), &event, Some(channel_uuid)) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))?; + if !was_inserted { + return Err(ActionSinkError::Database( + "topic event was not inserted (replaceable LWW rejected)".into(), + )); + } + + // Fan out via the shared dispatch path (Redis + local subscribers + + // audit) so realtime subscribers see the metadata event. The actual + // topic mutation is applied by the side-effect handler during + // ingest, not here. + dispatch_persistent_event( + &tenant, + &state, + &stored_event, + kind_u32, + &author_pubkey, + None, + ) + .await; + + let event_id_hex = stored_event.event.id.to_hex(); + info!( + community_id = %community_id, + channel = %channel_id, + "Workflow set_channel_topic → event {event_id_hex}" + ); + Ok(event_id_hex) + }) + } } #[cfg(test)] diff --git a/crates/buzz-workflow/src/action_sink.rs b/crates/buzz-workflow/src/action_sink.rs index 4559e7b688..e3863d54d1 100644 --- a/crates/buzz-workflow/src/action_sink.rs +++ b/crates/buzz-workflow/src/action_sink.rs @@ -71,4 +71,27 @@ pub trait ActionSink: Send + Sync { author_pubkey: &str, workflow_depth: usize, ) -> Pin> + Send + '_>>; + + /// Update a channel's topic on behalf of a workflow owner. + /// + /// Emits a NIP-29 edit-metadata event (kind:9002) carrying a `topic` tag. + /// The relay's side-effect handler applies the topic change after + /// membership/permission checks. Same community-scoping and keypair-signs + /// semantics as [`send_message`]. + /// + /// - `community_id`: the owning community of the workflow run. + /// - `channel_id`: UUID string of the target channel. + /// - `topic`: the new topic string (must not be empty). + /// - `author_pubkey`: hex pubkey of the workflow owner (attribution `p` tag). + /// - `workflow_depth`: trigger-chain depth for loop prevention. + /// + /// Returns the event ID hex string on success. + fn set_channel_topic( + &self, + community_id: CommunityId, + channel_id: &str, + topic: &str, + author_pubkey: &str, + workflow_depth: usize, + ) -> Pin> + Send + '_>>; } diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index 506cb05592..9f1f8ab51f 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -596,10 +596,54 @@ pub async fn dispatch_action( Err(WorkflowError::NotImplemented("SendDm".into())) } - SetChannelTopic { topic: _ } => { - warn!(run_id = %run_id, step = step_id, "SetChannelTopic not yet implemented"); - // TODO (WF-07): update channel topic via DB. - Err(WorkflowError::NotImplemented("SetChannelTopic".into())) + SetChannelTopic { topic } => { + // Load workflow metadata for owner attribution, scoped to the + // run's community — same rationale as SendMessage. + let wf_run = engine + .db + .get_workflow_run(community_id, run_id) + .await + .map_err(|e| { + WorkflowError::WebhookError(format!( + "SetChannelTopic: failed to load workflow run {run_id}: {e}" + )) + })?; + let workflow = engine + .db + .get_workflow(community_id, wf_run.workflow_id) + .await + .map_err(|e| { + WorkflowError::WebhookError(format!( + "SetChannelTopic: failed to load workflow {}: {e}", + wf_run.workflow_id + )) + })?; + let owner_pubkey_hex = hex::encode(&workflow.owner_pubkey); + + info!( + run_id = %run_id, + step = step_id, + channel = %trigger_ctx.channel_id, + "SetChannelTopic → {:?}", + topic + ); + + let event_id = engine + .action_sink()? + .set_channel_topic( + community_id, + &trigger_ctx.channel_id, + topic, + &owner_pubkey_hex, + trigger_ctx.workflow_depth, + ) + .await + .map_err(WorkflowError::from)?; + + Ok(StepResult::Completed(serde_json::json!({ + "set": true, + "event_id": event_id, + }))) } AddReaction { emoji } => { diff --git a/crates/buzz-workflow/src/schema.rs b/crates/buzz-workflow/src/schema.rs index a35cf4ffcd..582ce74109 100644 --- a/crates/buzz-workflow/src/schema.rs +++ b/crates/buzz-workflow/src/schema.rs @@ -202,12 +202,6 @@ impl WorkflowDef { step.id ))); } - ActionDef::SetChannelTopic { .. } => { - return Err(WorkflowError::InvalidDefinition(format!( - "step '{}' uses 'set_channel_topic' which is not yet implemented", - step.id - ))); - } _ => {} } } @@ -419,13 +413,14 @@ mod tests { } #[test] - fn validate_rejects_unimplemented_set_channel_topic() { + fn validate_accepts_set_channel_topic() { + // set_channel_topic is now implemented — validate() should accept it. let yaml = "name: Topic Test\ntrigger:\n on: webhook\nsteps:\n - id: topic\n action: set_channel_topic\n topic: new\n"; - let err = parse_yaml(yaml).unwrap_err(); - assert!( - err.to_string().contains("set_channel_topic"), - "expected set_channel_topic rejection, got: {err}" - ); + let (def, _) = parse_yaml(yaml).expect("set_channel_topic should validate"); + assert!(matches!( + def.steps[0].action, + ActionDef::SetChannelTopic { .. } + )); } #[test] From 7ba39a0775535e87317a5844b596fc853525fb92 Mon Sep 17 00:00:00 2001 From: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> Date: Tue, 28 Jul 2026 19:48:03 +0900 Subject: [PATCH 04/27] feat(workflow): implement add_reaction via ActionSink MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Migrates add_reaction from the reqwest HTTP-loopback path (add_reaction_impl) to ActionSink::add_reaction, implemented in RelayActionSink as a NIP-25 reaction event (kind:7) signed by the relay keypair. Removes the feature-gated #[cfg(reqwest)] branch in the executor — the action now works unconditionally. Also removes the now-dead shared_http_client helper (call_webhook_impl builds its own SSRF-pinned client per request). Signed-off-by: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> --- crates/buzz-relay/src/workflow_sink.rs | 101 ++++++++++++++++++++ crates/buzz-workflow/src/action_sink.rs | 22 +++++ crates/buzz-workflow/src/executor.rs | 117 ++++++++---------------- 3 files changed, 160 insertions(+), 80 deletions(-) diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 5dfca95946..08625792f0 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -462,6 +462,107 @@ impl ActionSink for RelayActionSink { Ok(event_id_hex) }) } + + fn add_reaction( + &self, + community_id: CommunityId, + target_event_id: &str, + emoji: &str, + author_pubkey: &str, + workflow_depth: usize, + ) -> Pin> + Send + '_>> { + let target_event_id = target_event_id.to_owned(); + let emoji = emoji.to_owned(); + let author_pubkey = author_pubkey.to_owned(); + + Box::pin(async move { + let state = self + .state + .upgrade() + .ok_or_else(|| ActionSinkError::Database("relay is shutting down".into()))?; + + let host = state + .db + .lookup_community_host(community_id) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))? + .ok_or_else(|| { + ActionSinkError::Database(format!( + "workflow run community {community_id} is not mapped to a host" + )) + })?; + let tenant = buzz_core::tenant::TenantContext::resolved(community_id, host); + + if emoji.trim().is_empty() { + return Err(ActionSinkError::EmptyContent); + } + + // Parse the target event ID (hex) → EventId for the SDK builder. + let target_id = nostr::EventId::from_hex(&target_event_id).map_err(|e| { + ActionSinkError::InvalidInput(format!("invalid target event id: {e}")) + })?; + + // Build a NIP-25 reaction (kind:7) via the SDK builder, then add + // attribution `p` and `buzz:workflow` (depth) tags. The SDK builder + // emits content=emoji + an `e` tag pointing at the target. + let workflow_tag_depth = workflow_depth.to_string(); + let base = buzz_sdk::builders::build_reaction(target_id, &emoji) + .map_err(|e| ActionSinkError::EventBuild(e.to_string()))?; + // Sign once to capture the builder's tags, then rebuild with the + // full set. EventBuilder has no mutable tag access, so this is the + // simplest way to extend its tags while preserving content/kind. + let probe = base + .clone() + .sign_with_keys(&state.relay_keypair) + .map_err(|e| ActionSinkError::EventBuild(format!("probe sign: {e}")))?; + let mut all_tags: Vec = probe.tags.into_iter().collect(); + all_tags.push( + Tag::parse(["p", &author_pubkey]) + .map_err(|e| ActionSinkError::EventBuild(format!("p tag: {e}")))?, + ); + all_tags.push( + Tag::parse(["buzz:workflow", "true", &workflow_tag_depth]) + .map_err(|e| ActionSinkError::EventBuild(format!("workflow tag: {e}")))?, + ); + let event = EventBuilder::new(Kind::Custom(7), &emoji) + .tags(all_tags) + .sign_with_keys(&state.relay_keypair) + .map_err(|e| ActionSinkError::EventBuild(format!("sign: {e}")))?; + + let kind_u32: u32 = 7; + // Reactions are global-scoped (no channel_id) — they reference the + // target message via the `e` tag, and the relay resolves channel + // membership from the target during ingest. + let (stored_event, was_inserted) = state + .db + .insert_event(tenant.community(), &event, None) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))?; + if !was_inserted { + return Err(ActionSinkError::Database( + "reaction event was not inserted".into(), + )); + } + + dispatch_persistent_event( + &tenant, + &state, + &stored_event, + kind_u32, + &author_pubkey, + None, + ) + .await; + + let event_id_hex = stored_event.event.id.to_hex(); + info!( + community_id = %community_id, + target = %target_event_id, + "Workflow add_reaction → event {event_id_hex}" + ); + Ok(event_id_hex) + }) + } } #[cfg(test)] diff --git a/crates/buzz-workflow/src/action_sink.rs b/crates/buzz-workflow/src/action_sink.rs index e3863d54d1..bfb26fba27 100644 --- a/crates/buzz-workflow/src/action_sink.rs +++ b/crates/buzz-workflow/src/action_sink.rs @@ -94,4 +94,26 @@ pub trait ActionSink: Send + Sync { author_pubkey: &str, workflow_depth: usize, ) -> Pin> + Send + '_>>; + + /// Add an emoji reaction to a message on behalf of a workflow owner. + /// + /// Emits a NIP-25 reaction event (kind:7) targeting `target_event_id`. + /// The relay keypair signs the event; attribution flows through the + /// standard reaction storage path. + /// + /// - `community_id`: the owning community of the workflow run. + /// - `target_event_id`: hex event ID of the message to react to. + /// - `emoji`: emoji character or shortcode (e.g. `"👍"`, `"thumbsup"`). + /// - `author_pubkey`: hex pubkey of the workflow owner (attribution `p` tag). + /// - `workflow_depth`: trigger-chain depth for loop prevention. + /// + /// Returns the reaction event ID hex string on success. + fn add_reaction( + &self, + community_id: CommunityId, + target_event_id: &str, + emoji: &str, + author_pubkey: &str, + workflow_depth: usize, + ) -> Pin> + Send + '_>>; } diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index 9f1f8ab51f..00ecfd99e5 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -654,23 +654,44 @@ pub async fn dispatch_action( )); } - #[cfg(feature = "reqwest")] - { - let result = add_reaction_impl(&trigger_ctx.message_id, emoji).await?; - Ok(StepResult::Completed(result)) - } + // Load workflow owner for attribution, scoped to the run's community. + let wf_run = engine + .db + .get_workflow_run(community_id, run_id) + .await + .map_err(|e| { + WorkflowError::WebhookError(format!( + "AddReaction: failed to load workflow run {run_id}: {e}" + )) + })?; + let workflow = engine + .db + .get_workflow(community_id, wf_run.workflow_id) + .await + .map_err(|e| { + WorkflowError::WebhookError(format!( + "AddReaction: failed to load workflow {}: {e}", + wf_run.workflow_id + )) + })?; + let owner_pubkey_hex = hex::encode(&workflow.owner_pubkey); - #[cfg(not(feature = "reqwest"))] - { - warn!( - run_id = %run_id, - step = step_id, - "AddReaction: reqwest feature not enabled, skipping HTTP call" - ); - Ok(StepResult::Completed( - serde_json::json!({ "added": false, "skipped": true }), - )) - } + let event_id = engine + .action_sink()? + .add_reaction( + community_id, + &trigger_ctx.message_id, + emoji, + &owner_pubkey_hex, + trigger_ctx.workflow_depth, + ) + .await + .map_err(WorkflowError::from)?; + + Ok(StepResult::Completed(serde_json::json!({ + "added": true, + "event_id": event_id, + }))) } CallWebhook { @@ -922,70 +943,6 @@ async fn call_webhook_impl( })) } -/// Returns a shared `reqwest::Client` reused across all workflow HTTP calls. -/// Sharing a single client reuses the underlying connection pool. -#[cfg(feature = "reqwest")] -fn shared_http_client() -> &'static reqwest::Client { - use std::sync::LazyLock; - use std::time::Duration; - static CLIENT: LazyLock = LazyLock::new(|| { - reqwest::Client::builder() - .timeout(Duration::from_secs(10)) - .build() - .expect("HTTP client build must succeed") - }); - &CLIENT -} - -/// POST `{"emoji": emoji}` to `POST /api/messages/{message_id}/reactions`. -#[cfg(feature = "reqwest")] -async fn add_reaction_impl(message_id: &str, emoji: &str) -> Result { - let base_url = - std::env::var("BUZZ_RELAY_BASE_URL").unwrap_or_else(|_| "http://localhost:3000".to_owned()); - - let url = format!("{base_url}/api/messages/{message_id}/reactions"); - - let client = shared_http_client(); - - let mut req = client - .post(&url) - .header("Content-Type", "application/json") - .json(&serde_json::json!({ "emoji": emoji })); - - if let Ok(token) = std::env::var("BUZZ_API_TOKEN") { - req = req.header("Authorization", format!("Bearer {token}")); - } else if let Ok(pubkey) = std::env::var("BUZZ_RELAY_PUBKEY") { - req = req.header("X-Pubkey", pubkey); - } - - let resp = req - .send() - .await - .map_err(|e| WorkflowError::WebhookError(format!("AddReaction HTTP error: {e}")))?; - - let status = resp.status(); - - if !status.is_success() { - let body = resp - .text() - .await - .unwrap_or_else(|_| "".to_owned()); - return Err(WorkflowError::WebhookError(format!( - "AddReaction: relay returned {status} for message {message_id}: {body}" - ))); - } - - let body_text = resp.text().await.unwrap_or_else(|_| String::new()); - let body_json: JsonValue = serde_json::from_str(&body_text) - .unwrap_or_else(|_| serde_json::json!({ "raw": body_text })); - - Ok(serde_json::json!({ - "added": true, - "status": status.as_u16(), - "response": body_json, - })) -} - /// Rich return type from `execute_run` / `execute_from_step`. /// /// Carries enough information for the caller to: From 28968327d5fa1c58ed07a9e7fa47568859dbfa53 Mon Sep 17 00:00:00 2001 From: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> Date: Tue, 28 Jul 2026 19:51:45 +0900 Subject: [PATCH 05/27] feat(workflow): implement send_dm action MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds ActionSink::send_dm, implemented in RelayActionSink via the DM channel pattern: open_dm creates/reuses a private channel between the workflow owner and recipient, then a kind:9 message is posted into it (signed by the relay keypair with attribution + depth tags). Buzz models DMs as private channels rather than NIP-17 gift-wraps. Removes the last NotImplemented guard from validate() — all seven actions are now wired. Updated tests accordingly. Signed-off-by: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> --- crates/buzz-relay/src/workflow_sink.rs | 107 ++++++++++++++++++++++++ crates/buzz-workflow/src/action_sink.rs | 23 +++++ crates/buzz-workflow/src/executor.rs | 59 ++++++++++++- crates/buzz-workflow/src/schema.rs | 25 ++---- 4 files changed, 191 insertions(+), 23 deletions(-) diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 08625792f0..0e12267383 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -563,6 +563,113 @@ impl ActionSink for RelayActionSink { Ok(event_id_hex) }) } + + fn send_dm( + &self, + community_id: CommunityId, + recipient_pubkey: &str, + text: &str, + author_pubkey: &str, + workflow_depth: usize, + ) -> Pin> + Send + '_>> { + let recipient_pubkey = recipient_pubkey.to_owned(); + let text = text.to_owned(); + let author_pubkey = author_pubkey.to_owned(); + + Box::pin(async move { + let state = self + .state + .upgrade() + .ok_or_else(|| ActionSinkError::Database("relay is shutting down".into()))?; + + let host = state + .db + .lookup_community_host(community_id) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))? + .ok_or_else(|| { + ActionSinkError::Database(format!( + "workflow run community {community_id} is not mapped to a host" + )) + })?; + let tenant = buzz_core::tenant::TenantContext::resolved(community_id, host); + + if text.trim().is_empty() { + return Err(ActionSinkError::EmptyContent); + } + + // Parse recipient pubkey hex → raw bytes for open_dm. + let recipient_bytes = hex::decode(&recipient_pubkey).map_err(|e| { + ActionSinkError::InvalidInput(format!("invalid recipient pubkey: {e}")) + })?; + let owner_bytes = hex::decode(&author_pubkey).map_err(|e| { + ActionSinkError::InvalidInput(format!("invalid author pubkey: {e}")) + })?; + + // Open (or reuse) a private DM channel between owner + recipient. + // Buzz models DMs as private channels, not NIP-17 gift-wraps. + let participant_refs: Vec<&[u8]> = vec![&recipient_bytes]; + let (dm_channel, _was_created) = state + .db + .open_dm(community_id, &participant_refs, &owner_bytes) + .await + .map_err(|e| ActionSinkError::Database(format!("open_dm: {e}")))?; + + // Post a kind:9 message into the DM channel, signed by the relay + // keypair with attribution + workflow-depth tags — same shape as + // send_message but targeting the DM channel. + let channel_id_str = dm_channel.id.to_string(); + let workflow_tag_depth = workflow_depth.to_string(); + let p_tag = Tag::parse(["p", &author_pubkey]) + .map_err(|e| ActionSinkError::EventBuild(format!("p tag: {e}")))?; + let h_tag = Tag::parse(["h", &channel_id_str]) + .map_err(|e| ActionSinkError::EventBuild(format!("h tag: {e}")))?; + let wf_tag = Tag::parse(["buzz:workflow", "true", &workflow_tag_depth]) + .map_err(|e| ActionSinkError::EventBuild(format!("workflow tag: {e}")))?; + // Also tag the recipient so they receive the message. + let recip_tag = Tag::parse(["p", &recipient_pubkey]) + .map_err(|e| ActionSinkError::EventBuild(format!("recipient p tag: {e}")))?; + let event = EventBuilder::new(Kind::Custom(KIND_STREAM_MESSAGE as u16), &text) + .tags([p_tag, h_tag, wf_tag, recip_tag]) + .sign_with_keys(&state.relay_keypair) + .map_err(|e| ActionSinkError::EventBuild(format!("sign: {e}")))?; + + let (stored_event, was_inserted) = state + .db + .insert_event_with_thread_metadata( + tenant.community(), + &event, + Some(dm_channel.id), + None, + ) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))?; + if !was_inserted { + return Err(ActionSinkError::Database( + "DM message event was not inserted".into(), + )); + } + + dispatch_persistent_event( + &tenant, + &state, + &stored_event, + KIND_STREAM_MESSAGE, + &author_pubkey, + None, + ) + .await; + + let event_id_hex = stored_event.event.id.to_hex(); + info!( + community_id = %community_id, + dm_channel = %dm_channel.id, + recipient = %recipient_pubkey, + "Workflow send_dm → event {event_id_hex}" + ); + Ok(event_id_hex) + }) + } } #[cfg(test)] diff --git a/crates/buzz-workflow/src/action_sink.rs b/crates/buzz-workflow/src/action_sink.rs index bfb26fba27..f126482722 100644 --- a/crates/buzz-workflow/src/action_sink.rs +++ b/crates/buzz-workflow/src/action_sink.rs @@ -116,4 +116,27 @@ pub trait ActionSink: Send + Sync { author_pubkey: &str, workflow_depth: usize, ) -> Pin> + Send + '_>>; + + /// Send a direct message to a user on behalf of a workflow owner. + /// + /// Opens (or reuses) a private DM channel between the workflow owner and + /// `recipient_pubkey`, then posts a kind:9 message into it. Buzz models + /// DMs as private channels rather than NIP-17 gift-wraps. + /// + /// - `community_id`: the owning community of the workflow run. + /// - `recipient_pubkey`: hex pubkey of the DM recipient. + /// - `text`: message body (must not be empty/whitespace-only). + /// - `author_pubkey`: hex pubkey of the workflow owner (DM participant + + /// attribution `p` tag). + /// - `workflow_depth`: trigger-chain depth for loop prevention. + /// + /// Returns the message event ID hex string on success. + fn send_dm( + &self, + community_id: CommunityId, + recipient_pubkey: &str, + text: &str, + author_pubkey: &str, + workflow_depth: usize, + ) -> Pin> + Send + '_>>; } diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index 00ecfd99e5..41a7e253f3 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -590,10 +590,61 @@ pub async fn dispatch_action( }))) } - SendDm { to, text: _ } => { - warn!(run_id = %run_id, step = step_id, "SendDm not yet implemented (to={to})"); - // TODO (WF-07): emit DM event. - Err(WorkflowError::NotImplemented("SendDm".into())) + SendDm { to, text } => { + // Resolve `to` — may be a hex pubkey or {{trigger.author}} already + // resolved to a hex pubkey by the template step. + let recipient = to.trim(); + if recipient.is_empty() { + return Err(WorkflowError::InvalidDefinition(format!( + "SendDm step '{step_id}': 'to' resolved to empty" + ))); + } + + // Load workflow owner for the DM participant set + attribution. + let wf_run = engine + .db + .get_workflow_run(community_id, run_id) + .await + .map_err(|e| { + WorkflowError::WebhookError(format!( + "SendDm: failed to load workflow run {run_id}: {e}" + )) + })?; + let workflow = engine + .db + .get_workflow(community_id, wf_run.workflow_id) + .await + .map_err(|e| { + WorkflowError::WebhookError(format!( + "SendDm: failed to load workflow {}: {e}", + wf_run.workflow_id + )) + })?; + let owner_pubkey_hex = hex::encode(&workflow.owner_pubkey); + + info!( + run_id = %run_id, + step = step_id, + to = recipient, + "SendDm → {recipient}" + ); + + let event_id = engine + .action_sink()? + .send_dm( + community_id, + recipient, + text, + &owner_pubkey_hex, + trigger_ctx.workflow_depth, + ) + .await + .map_err(WorkflowError::from)?; + + Ok(StepResult::Completed(serde_json::json!({ + "sent": true, + "event_id": event_id, + }))) } SetChannelTopic { topic } => { diff --git a/crates/buzz-workflow/src/schema.rs b/crates/buzz-workflow/src/schema.rs index 582ce74109..09838d041b 100644 --- a/crates/buzz-workflow/src/schema.rs +++ b/crates/buzz-workflow/src/schema.rs @@ -191,19 +191,8 @@ impl WorkflowDef { ))); } - // Temporary guard: reject actions that are not yet fully implemented. - // This surfaces the gap at definition time (YAML import / REST ingest) - // instead of failing at runtime with WorkflowError::NotImplemented. - // Each action is removed from this guard as it is completed. - match &step.action { - ActionDef::SendDm { .. } => { - return Err(WorkflowError::InvalidDefinition(format!( - "step '{}' uses 'send_dm' which is not yet implemented", - step.id - ))); - } - _ => {} - } + // All actions are now implemented — the temporary NotImplemented + // guard has been fully removed as of send_dm completion. } if let TriggerDef::Schedule { cron, interval } = &self.trigger { @@ -403,13 +392,11 @@ mod tests { } #[test] - fn validate_rejects_unimplemented_send_dm() { + fn validate_accepts_send_dm() { + // send_dm is now implemented — validate() should accept it. let yaml = "name: DM Test\ntrigger:\n on: webhook\nsteps:\n - id: dm\n action: send_dm\n to: abc123\n text: hi\n"; - let err = parse_yaml(yaml).unwrap_err(); - assert!( - err.to_string().contains("send_dm"), - "expected send_dm rejection, got: {err}" - ); + let (def, _) = parse_yaml(yaml).expect("send_dm should validate"); + assert!(matches!(def.steps[0].action, ActionDef::SendDm { .. })); } #[test] From 6414cb4ee05a77a2e2af22ecd354046f3761c4c7 Mon Sep 17 00:00:00 2001 From: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> Date: Tue, 28 Jul 2026 19:57:26 +0900 Subject: [PATCH 06/27] =?UTF-8?q?feat(workflow):=20complete=20request=5Fap?= =?UTF-8?q?proval=20=E2=80=94=20DB=20record=20+=20kind:46010=20event?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The executor now persists a workflow_approvals row (via create_approval) BEFORE emitting the request event, so the relay's existing grant/deny handler (kind:46030/46031 → resume_workflow_after_approval) can locate the pending approval. The kind:46010 request event carries the token hash as a d tag, the approver spec, and the approval prompt as content. Adds sha2 dependency to buzz-workflow for token-hash computation (matches buzz-db's hash_approval_token = SHA-256). Threads step_index through dispatch_action so the approval row records the correct step. This completes all seven workflow actions — the engine is now feature- complete for the defined schema. Signed-off-by: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> --- crates/buzz-relay/src/workflow_sink.rs | 85 +++++++++++++++++++++++++ crates/buzz-workflow/Cargo.toml | 1 + crates/buzz-workflow/src/action_sink.rs | 24 +++++++ crates/buzz-workflow/src/executor.rs | 71 ++++++++++++++++++++- 4 files changed, 179 insertions(+), 2 deletions(-) diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 0e12267383..258e99866b 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -670,6 +670,91 @@ impl ActionSink for RelayActionSink { Ok(event_id_hex) }) } + + fn request_approval( + &self, + community_id: CommunityId, + token_hash: &str, + message: &str, + approver_spec: &str, + author_pubkey: &str, + ) -> Pin> + Send + '_>> { + let token_hash = token_hash.to_owned(); + let message = message.to_owned(); + let approver_spec = approver_spec.to_owned(); + let author_pubkey = author_pubkey.to_owned(); + + Box::pin(async move { + let state = self + .state + .upgrade() + .ok_or_else(|| ActionSinkError::Database("relay is shutting down".into()))?; + + let host = state + .db + .lookup_community_host(community_id) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))? + .ok_or_else(|| { + ActionSinkError::Database(format!( + "workflow run community {community_id} is not mapped to a host" + )) + })?; + let tenant = buzz_core::tenant::TenantContext::resolved(community_id, host); + + // Build a kind:46010 (KIND_WORKFLOW_APPROVAL_REQUESTED) event. + // The `d` tag carries the token hash so the grant/deny handler + // (kind:46030/46031) can locate the pending approval row. Content + // is the human-readable approval prompt; `approver_spec` is a tag + // for routing/labeling. + let d_tag = Tag::parse(["d", &token_hash]) + .map_err(|e| ActionSinkError::EventBuild(format!("d tag: {e}")))?; + let p_tag = Tag::parse(["p", &author_pubkey]) + .map_err(|e| ActionSinkError::EventBuild(format!("p tag: {e}")))?; + let approver_tag = Tag::parse(["approver", &approver_spec]) + .map_err(|e| ActionSinkError::EventBuild(format!("approver tag: {e}")))?; + let event = EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_WORKFLOW_APPROVAL_REQUESTED as u16), + &message, + ) + .tags([d_tag, p_tag, approver_tag]) + .sign_with_keys(&state.relay_keypair) + .map_err(|e| ActionSinkError::EventBuild(format!("sign: {e}")))?; + + let kind_u32 = buzz_core::kind::KIND_WORKFLOW_APPROVAL_REQUESTED; + // Approval-request events are global-scoped (no channel_id) — they + // are workflow-execution kinds, excluded from workflow triggering + // (is_workflow_execution_kind) so they can't recurse. + let (stored_event, was_inserted) = state + .db + .insert_event(tenant.community(), &event, None) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))?; + if !was_inserted { + return Err(ActionSinkError::Database( + "approval-request event was not inserted".into(), + )); + } + + dispatch_persistent_event( + &tenant, + &state, + &stored_event, + kind_u32, + &author_pubkey, + None, + ) + .await; + + let event_id_hex = stored_event.event.id.to_hex(); + info!( + community_id = %community_id, + token_hash = %token_hash, + "Workflow request_approval → event {event_id_hex}" + ); + Ok(event_id_hex) + }) + } } #[cfg(test)] diff --git a/crates/buzz-workflow/Cargo.toml b/crates/buzz-workflow/Cargo.toml index d4813e56d4..69e5c5c41c 100644 --- a/crates/buzz-workflow/Cargo.toml +++ b/crates/buzz-workflow/Cargo.toml @@ -24,6 +24,7 @@ chrono = { workspace = true } tokio = { workspace = true } tracing = { workspace = true } thiserror = { workspace = true } +sha2 = { workspace = true } reqwest = { workspace = true, optional = true } [features] diff --git a/crates/buzz-workflow/src/action_sink.rs b/crates/buzz-workflow/src/action_sink.rs index f126482722..04a1491c71 100644 --- a/crates/buzz-workflow/src/action_sink.rs +++ b/crates/buzz-workflow/src/action_sink.rs @@ -139,4 +139,28 @@ pub trait ActionSink: Send + Sync { author_pubkey: &str, workflow_depth: usize, ) -> Pin> + Send + '_>>; + + /// Emit a workflow approval-request event (kind:46010). + /// + /// The executor writes the `workflow_approvals` DB row (via + /// `create_approval`) *before* calling this method, then asks the sink to + /// publish the request event so approvers are notified. The event carries + /// the token hash as a `d` tag so the relay's grant/deny handler + /// (kind:46030/46031) can locate the pending approval. + /// + /// - `community_id`: the owning community of the workflow run. + /// - `token_hash`: SHA-256 hex of the raw approval token (matches the DB row). + /// - `message`: human-readable approval prompt shown to the approver. + /// - `approver_spec`: who may approve (e.g. `"@release-manager"`). + /// - `author_pubkey`: hex pubkey of the workflow owner. + /// + /// Returns the request event ID hex string on success. + fn request_approval( + &self, + community_id: CommunityId, + token_hash: &str, + message: &str, + approver_spec: &str, + author_pubkey: &str, + ) -> Pin> + Send + '_>>; } diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index 41a7e253f3..2281bbdb74 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -525,6 +525,7 @@ fn resolve_send_message_channel( /// persist state and stop the execution loop. pub async fn dispatch_action( step_id: &str, + step_index: usize, action: &ActionDef, engine: &WorkflowEngine, community_id: CommunityId, @@ -787,10 +788,75 @@ pub async fn dispatch_action( "RequestApproval from={from} timeout={timeout_str}: {message}" ); + // Compute expiry from the timeout string (default 24h). + let timeout_secs = parse_duration_secs(timeout_str)?; + let expires_at = chrono::Utc::now() + + chrono::Duration::seconds(i64::try_from(timeout_secs).unwrap_or(i64::MAX / 2)); + + // Load workflow metadata for the approval row's workflow_id and + // the request event's attribution, scoped to the run's community. + let wf_run = engine + .db + .get_workflow_run(community_id, run_id) + .await + .map_err(|e| { + WorkflowError::WebhookError(format!( + "RequestApproval: failed to load workflow run {run_id}: {e}" + )) + })?; + let workflow = engine + .db + .get_workflow(community_id, wf_run.workflow_id) + .await + .map_err(|e| { + WorkflowError::WebhookError(format!( + "RequestApproval: failed to load workflow {}: {e}", + wf_run.workflow_id + )) + })?; + let owner_pubkey_hex = hex::encode(&workflow.owner_pubkey); + let token = generate_approval_token(run_id, step_id); - // TODO (WF-08): create approval record in DB, emit kind:46010. - // For now, return Suspended with the token so the caller can persist state. + // Persist the approval request BEFORE emitting the event, so the + // grant/deny handler (kind:46030/46031) can find a matching row. + // create_approval hashes the raw token internally. + engine + .db + .create_approval(buzz_db::workflow::CreateApprovalParams { + community_id, + token: &token, + workflow_id: wf_run.workflow_id, + run_id, + step_id, + step_index: i32::try_from(step_index).unwrap_or(0), + approver_spec: from, + expires_at, + }) + .await + .map_err(|e| WorkflowError::WebhookError(format!("create_approval: {e}")))?; + + // Compute the token hash for the event's `d` tag — matches what + // the grant/deny handler looks up (hash_approval_token = SHA-256). + let token_hash_hex = { + use sha2::Digest; + let mut hasher = sha2::Sha256::new(); + hasher.update(token.as_bytes()); + hex::encode(hasher.finalize()) + }; + + // Emit the kind:46010 request event so approvers are notified. + engine + .action_sink()? + .request_approval( + community_id, + &token_hash_hex, + message, + from, + &owner_pubkey_hex, + ) + .await + .map_err(WorkflowError::from)?; Ok(StepResult::Suspended { approval_token: token, @@ -1202,6 +1268,7 @@ async fn execute_steps( std::time::Duration::from_secs(timeout_secs), dispatch_action( &step.id, + i, &resolved_action, engine, community_id, From d656c966364fac8c7fa57ac3b04f964ae9fae7f4 Mon Sep 17 00:00:00 2001 From: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> Date: Tue, 28 Jul 2026 20:03:51 +0900 Subject: [PATCH 07/27] fix(workflow): update relay test for send_message depth arg Adds the workflow_depth argument to the send_message call in the RelayActionSink test, matching the updated ActionSink trait signature. Signed-off-by: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> --- crates/buzz-relay/src/workflow_sink.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 258e99866b..001868882f 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -1069,6 +1069,7 @@ mod integration_tests { &channel.id.to_string(), "heads up @Robby — please take a look", &author_hex, + 0, ) .await .expect("send_message"); From 4393f67357065791e9dcfbe37391feecdcb79efd Mon Sep 17 00:00:00 2001 From: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> Date: Tue, 28 Jul 2026 20:21:51 +0900 Subject: [PATCH 08/27] fix(workflow): reap expired approvals + validate send_dm recipients MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three hardening fixes for the workflow engine: #4 — Approval expiry reaper: adds expire_pending_approvals() (atomic UPDATE...RETURNING, multi-pod safe) and calls it from the cron loop. Expired pending approvals are marked 'expired' and their waiting runs transitioned to Failed, preventing indefinite DB accumulation and runs stuck in waiting_approval. #5 — send_dm recipient validation: verifies the recipient is a member of the workflow's community (is_relay_member) before opening a DM channel. Prevents workflows from creating DM channels to arbitrary pubkeys outside the community (spam surface). #3b — request_approval cleanup on publish failure: if the kind:46010 event publication fails after the DB row was created, the approval is marked Denied (best-effort) so it doesn't linger as a phantom pending row; the run fails fast instead of sticking in waiting_approval with no notification dispatched. Signed-off-by: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> --- crates/buzz-db/src/lib.rs | 12 ++++++ crates/buzz-db/src/workflow.rs | 32 ++++++++++++++ crates/buzz-relay/src/workflow_sink.rs | 16 +++++++ crates/buzz-workflow/src/executor.rs | 30 ++++++++++++- crates/buzz-workflow/src/lib.rs | 60 ++++++++++++++++++++++++++ 5 files changed, 148 insertions(+), 2 deletions(-) diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index fdd72c3c32..4057f0eb72 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -2541,6 +2541,18 @@ impl Db { workflow::list_all_enabled_workflows(&self.pool).await } + /// Atomically expire pending approvals whose `expires_at` has passed. + /// + /// Returns the `(community_id, run_id)` pairs of the expired approvals so + /// the caller can transition the waiting runs to `Failed`. See + /// [`workflow::expire_pending_approvals`] for details. + pub async fn expire_pending_approvals( + &self, + now: chrono::DateTime, + ) -> Result> { + workflow::expire_pending_approvals(&self.pool, now).await + } + /// Claim a scheduled workflow fire for an authoritative schedule instant. /// /// Returns `Some` only for the first pod to claim `(community_id, diff --git a/crates/buzz-db/src/workflow.rs b/crates/buzz-db/src/workflow.rs index 9c02f162c9..93059a85d5 100644 --- a/crates/buzz-db/src/workflow.rs +++ b/crates/buzz-db/src/workflow.rs @@ -449,6 +449,38 @@ pub async fn list_enabled_channel_workflows( rows.into_iter().map(row_to_workflow_record).collect() } +/// Atomically expire pending approvals whose `expires_at` has passed. +/// +/// Selects pending approvals where `expires_at <= now`, updates them to +/// `expired` in the same statement (so two concurrent reapers cannot both +/// act), and returns the `(community_id, run_id)` pairs so the caller can +/// transition the waiting runs to `Failed`. Without this, expired approvals +/// accumulate indefinitely and their runs stay stuck in `waiting_approval`. +pub async fn expire_pending_approvals( + pool: &PgPool, + now: DateTime, +) -> Result> { + let rows = sqlx::query( + r#" + UPDATE workflow_approvals + SET status = 'expired' + WHERE status = 'pending' AND expires_at <= $1 + RETURNING community_id, run_id + "#, + ) + .bind(now) + .fetch_all(pool) + .await?; + + rows.into_iter() + .map(|row| { + let community_id: Uuid = row.try_get("community_id")?; + let run_id: Uuid = row.try_get("run_id")?; + Ok((CommunityId::from_uuid(community_id), run_id)) + }) + .collect() +} + /// List all active, enabled workflows with a `schedule` trigger across all channels. /// /// Used by the cron scheduler. Filters by trigger type in SQL to avoid loading diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 001868882f..fa2ed038e6 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -606,6 +606,22 @@ impl ActionSink for RelayActionSink { ActionSinkError::InvalidInput(format!("invalid author pubkey: {e}")) })?; + // Verify the recipient is a member of the workflow's community + // before opening a DM channel. Without this, a workflow could + // create DM channels to arbitrary pubkeys outside the community + // (spam surface). The owner is implicitly a member (they authored + // the workflow), so only the recipient needs checking. + let recipient_is_member = state + .db + .is_relay_member(community_id, &recipient_pubkey) + .await + .map_err(|e| ActionSinkError::Database(format!("is_relay_member: {e}")))?; + if !recipient_is_member { + return Err(ActionSinkError::InvalidInput(format!( + "send_dm recipient {recipient_pubkey} is not a member of community {community_id}" + ))); + } + // Open (or reuse) a private DM channel between owner + recipient. // Buzz models DMs as private channels, not NIP-17 gift-wraps. let participant_refs: Vec<&[u8]> = vec![&recipient_bytes]; diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index 2281bbdb74..5290aa67de 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -846,7 +846,11 @@ pub async fn dispatch_action( }; // Emit the kind:46010 request event so approvers are notified. - engine + // If publication fails, mark the just-created approval row as + // denied so it doesn't linger as a phantom pending row that the + // reaper would later expire — fail fast with a clear error instead + // of leaving the run stuck in waiting_approval with no notify. + if let Err(e) = engine .action_sink()? .request_approval( community_id, @@ -856,7 +860,29 @@ pub async fn dispatch_action( &owner_pubkey_hex, ) .await - .map_err(WorkflowError::from)?; + { + // Best-effort cleanup: mark the approval denied so the reaper + // doesn't have to wait for expiry. A failure here is logged, + // not fatal — the reaper will still clean it up on expiry. + let err = WorkflowError::from(e); + if let Err(cleanup_err) = engine + .db + .update_approval( + community_id, + &token, + buzz_db::workflow::ApprovalStatus::Denied, + None, + Some("approval request event publication failed"), + ) + .await + { + tracing::warn!( + run_id = %run_id, + "Failed to clean up approval after event publish failure: {cleanup_err}" + ); + } + return Err(err); + } Ok(StepResult::Suspended { approval_token: token, diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index dcbd8b3078..48ad5f7d54 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -462,6 +462,15 @@ impl WorkflowEngine { let now = Utc::now(); + // Reap expired approval requests: transition pending approvals past + // their expires_at to 'expired' (atomically, so only one pod acts) + // and fail the waiting runs so they don't stick in waiting_approval + // forever. The update + run-fail are separate statements, so a + // crash between them leaves an 'expired' approval with a still- + // 'waiting_approval' run — harmless: the next tick reaps it again + // (update returns no rows) but the run is failed idempotently. + self.reap_expired_approvals(now).await; + let workflows = match self.db.list_all_enabled_workflows().await { Ok(wf) => wf, Err(e) => { @@ -699,6 +708,57 @@ impl WorkflowEngine { self.last_fired.retain(|key, _| active_ids.contains(key)); } } + + /// Reap expired pending approvals and fail their waiting runs. + /// + /// Called from the cron loop each tick. `expire_pending_approvals` is an + /// atomic `UPDATE ... WHERE status='pending' AND expires_at <= now + /// RETURNING`, so across N pods only the rows one pod updates are returned + /// to it — no double-fail. Each returned `(community, run)` is transitioned + /// to `Failed` with an explanatory error message; the run's stored trace is + /// preserved for diagnostics. + async fn reap_expired_approvals(&self, now: DateTime) { + let expired = match self.db.expire_pending_approvals(now).await { + Ok(v) => v, + Err(e) => { + tracing::warn!("Cron tick: failed to reap expired approvals: {e}"); + return; + } + }; + for (community_id, run_id) in expired { + // Load the current trace to preserve it in the failed run. + let trace = self + .db + .get_workflow_run(community_id, run_id) + .await + .map(|r| r.execution_trace) + .unwrap_or(serde_json::json!([])); + if let Err(e) = self + .db + .update_workflow_run( + community_id, + run_id, + buzz_db::workflow::RunStatus::Failed, + 0, + &trace, + Some("approval request expired before a decision was made"), + ) + .await + { + tracing::warn!( + community_id = %community_id, + run_id = %run_id, + "Cron tick: failed to mark expired-approval run as failed: {e}" + ); + } else { + tracing::info!( + community_id = %community_id, + run_id = %run_id, + "Workflow run failed — approval expired" + ); + } + } + } } /// Find the cron schedule instant that fired within the `window_secs`-wide From 988dc0fcd65f0bf8f0aaa22dd9ff2e8722da0817 Mon Sep 17 00:00:00 2001 From: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> Date: Tue, 28 Jul 2026 20:27:28 +0900 Subject: [PATCH 09/27] fix(workflow): transition approval-suspended runs to WaitingApproval MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The finalize_run path still contained stale WF-08 stub logic that marked approval-suspended runs as Failed ('approval gates not yet implemented'). Now that request_approval persists the DB row and emits kind:46010, the run must transition to WaitingApproval so the grant/deny handler can resume it. Found during security review — this was a functional bug introduced in stage 6 (request_approval completion) where the executor was updated but the run-finalization branch was left stale. Signed-off-by: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> --- crates/buzz-workflow/src/lib.rs | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index 48ad5f7d54..290044907e 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -199,28 +199,29 @@ impl WorkflowEngine { let step_count = result.step_index as i32; if result.approval_token.is_some() { - // Approval gates are not yet implemented (WF-08). - // Fail explicitly rather than creating unreachable WaitingApproval rows. - tracing::warn!( + // The executor has already persisted the approval row and + // emitted the kind:46010 request event. Transition the run + // to WaitingApproval so the grant/deny handler can resume it. + tracing::info!( run_id = %run_id, step_index = result.step_index, - "Workflow hit approval gate — not yet implemented, marking as failed" + "Workflow run suspended — awaiting approval" ); if let Err(e) = self .db .update_workflow_run( community_id, run_id, - RunStatus::Failed, + RunStatus::WaitingApproval, step_count, &trace_json, - Some("approval gates not yet implemented — see WF-08"), + Some("awaiting approval decision"), ) .await { tracing::error!( run_id = %run_id, - "Failed to update run to Failed (approval gate): {e}" + "Failed to update run to WaitingApproval: {e}" ); } } else { From 37152102be1ab1eb07667518e20d62992015381d Mon Sep 17 00:00:00 2001 From: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> Date: Tue, 28 Jul 2026 20:37:29 +0900 Subject: [PATCH 10/27] refactor(workflow): extract resolve_state_and_tenant helper + remove probe-sign MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two code-quality improvements from the quality review: 1. Extract resolve_state_and_tenant() — the weak-ref upgrade + community→tenant resolution was duplicated across all 5 ActionSink methods (~15 lines each). Centralized into one helper, reducing lookup_community_host occurrences from 5 to 1. 2. Remove probe-sign in add_reaction — the double-signing workaround (sign once to extract SDK builder tags, then re-sign with extras) is replaced by inline tag construction, matching the set_channel_topic pattern. One sign per event, consistent across all actions. Signed-off-by: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> --- crates/buzz-relay/src/workflow_sink.rs | 160 ++++++++----------------- 1 file changed, 47 insertions(+), 113 deletions(-) diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index fa2ed038e6..8cf3615744 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -167,6 +167,35 @@ impl RelayActionSink { state: Arc::downgrade(state), } } + + /// Upgrade the weak ref and resolve the run's owning community → tenant. + /// + /// Every ActionSink method must do this before producing any side effect: + /// the workflow run carries `community_id`, and the relay-signed event + /// belongs to *that* community, never the deployment default. Centralizing + /// the two-step (upgrade + tenant resolution) keeps the five action + /// implementations DRY and the fail-closed behavior consistent. + async fn resolve_state_and_tenant( + &self, + community_id: CommunityId, + ) -> Result<(Arc, buzz_core::tenant::TenantContext), ActionSinkError> { + let state = self + .state + .upgrade() + .ok_or_else(|| ActionSinkError::Database("relay is shutting down".into()))?; + let host = state + .db + .lookup_community_host(community_id) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))? + .ok_or_else(|| { + ActionSinkError::Database(format!( + "workflow run community {community_id} is not mapped to a host" + )) + })?; + let tenant = buzz_core::tenant::TenantContext::resolved(community_id, host); + Ok((state, tenant)) + } } impl ActionSink for RelayActionSink { @@ -183,31 +212,7 @@ impl ActionSink for RelayActionSink { let author_pubkey = author_pubkey.to_owned(); Box::pin(async move { - // 0. Upgrade weak reference — fails only during shutdown. - let state = self - .state - .upgrade() - .ok_or_else(|| ActionSinkError::Database("relay is shutting down".into()))?; - - // The run carries its owning community (`community_id`); the - // relay-signed kind:9 message belongs to *that* community, never the - // deployment default. Re-deriving the tenant from `config.relay_url` - // would post a community-B workflow's output into the deployment/ - // default community under N>1. Read the community's host back to - // form a complete TenantContext (host is for labelling only — the - // community is already fixed and is never re-derived from it). Fail - // closed if the community no longer maps to a host. - let host = state - .db - .lookup_community_host(community_id) - .await - .map_err(|e| ActionSinkError::Database(e.to_string()))? - .ok_or_else(|| { - ActionSinkError::Database(format!( - "workflow run community {community_id} is not mapped to a host" - )) - })?; - let tenant = buzz_core::tenant::TenantContext::resolved(community_id, host); + let (state, tenant) = self.resolve_state_and_tenant(community_id).await?; // 1. Validate content is not empty/whitespace-only if text.trim().is_empty() { @@ -381,24 +386,7 @@ impl ActionSink for RelayActionSink { let author_pubkey = author_pubkey.to_owned(); Box::pin(async move { - let state = self - .state - .upgrade() - .ok_or_else(|| ActionSinkError::Database("relay is shutting down".into()))?; - - // Resolve the run's owning community → tenant (same rationale as - // send_message: a community-B workflow must mutate B's channel). - let host = state - .db - .lookup_community_host(community_id) - .await - .map_err(|e| ActionSinkError::Database(e.to_string()))? - .ok_or_else(|| { - ActionSinkError::Database(format!( - "workflow run community {community_id} is not mapped to a host" - )) - })?; - let tenant = buzz_core::tenant::TenantContext::resolved(community_id, host); + let (state, tenant) = self.resolve_state_and_tenant(community_id).await?; if topic.trim().is_empty() { return Err(ActionSinkError::EmptyContent); @@ -476,22 +464,7 @@ impl ActionSink for RelayActionSink { let author_pubkey = author_pubkey.to_owned(); Box::pin(async move { - let state = self - .state - .upgrade() - .ok_or_else(|| ActionSinkError::Database("relay is shutting down".into()))?; - - let host = state - .db - .lookup_community_host(community_id) - .await - .map_err(|e| ActionSinkError::Database(e.to_string()))? - .ok_or_else(|| { - ActionSinkError::Database(format!( - "workflow run community {community_id} is not mapped to a host" - )) - })?; - let tenant = buzz_core::tenant::TenantContext::resolved(community_id, host); + let (state, tenant) = self.resolve_state_and_tenant(community_id).await?; if emoji.trim().is_empty() { return Err(ActionSinkError::EmptyContent); @@ -502,30 +475,21 @@ impl ActionSink for RelayActionSink { ActionSinkError::InvalidInput(format!("invalid target event id: {e}")) })?; - // Build a NIP-25 reaction (kind:7) via the SDK builder, then add - // attribution `p` and `buzz:workflow` (depth) tags. The SDK builder - // emits content=emoji + an `e` tag pointing at the target. + // Build a NIP-25 reaction (kind:7) inline. The SDK's build_reaction + // emits content=emoji + an `e` tag, but we also need attribution `p` + // and `buzz:workflow` (depth) tags. Rather than probe-sign to extract + // the builder's tags (double-signing), construct all tags directly — + // matching the set_channel_topic pattern for consistency. let workflow_tag_depth = workflow_depth.to_string(); - let base = buzz_sdk::builders::build_reaction(target_id, &emoji) - .map_err(|e| ActionSinkError::EventBuild(e.to_string()))?; - // Sign once to capture the builder's tags, then rebuild with the - // full set. EventBuilder has no mutable tag access, so this is the - // simplest way to extend its tags while preserving content/kind. - let probe = base - .clone() - .sign_with_keys(&state.relay_keypair) - .map_err(|e| ActionSinkError::EventBuild(format!("probe sign: {e}")))?; - let mut all_tags: Vec = probe.tags.into_iter().collect(); - all_tags.push( - Tag::parse(["p", &author_pubkey]) - .map_err(|e| ActionSinkError::EventBuild(format!("p tag: {e}")))?, - ); - all_tags.push( - Tag::parse(["buzz:workflow", "true", &workflow_tag_depth]) - .map_err(|e| ActionSinkError::EventBuild(format!("workflow tag: {e}")))?, - ); + let target_id_hex = target_id.to_hex(); + let e_tag = Tag::parse(["e", &target_id_hex]) + .map_err(|e| ActionSinkError::EventBuild(format!("e tag: {e}")))?; + let p_tag = Tag::parse(["p", &author_pubkey]) + .map_err(|e| ActionSinkError::EventBuild(format!("p tag: {e}")))?; + let wf_tag = Tag::parse(["buzz:workflow", "true", &workflow_tag_depth]) + .map_err(|e| ActionSinkError::EventBuild(format!("workflow tag: {e}")))?; let event = EventBuilder::new(Kind::Custom(7), &emoji) - .tags(all_tags) + .tags([e_tag, p_tag, wf_tag]) .sign_with_keys(&state.relay_keypair) .map_err(|e| ActionSinkError::EventBuild(format!("sign: {e}")))?; @@ -577,22 +541,7 @@ impl ActionSink for RelayActionSink { let author_pubkey = author_pubkey.to_owned(); Box::pin(async move { - let state = self - .state - .upgrade() - .ok_or_else(|| ActionSinkError::Database("relay is shutting down".into()))?; - - let host = state - .db - .lookup_community_host(community_id) - .await - .map_err(|e| ActionSinkError::Database(e.to_string()))? - .ok_or_else(|| { - ActionSinkError::Database(format!( - "workflow run community {community_id} is not mapped to a host" - )) - })?; - let tenant = buzz_core::tenant::TenantContext::resolved(community_id, host); + let (state, tenant) = self.resolve_state_and_tenant(community_id).await?; if text.trim().is_empty() { return Err(ActionSinkError::EmptyContent); @@ -701,22 +650,7 @@ impl ActionSink for RelayActionSink { let author_pubkey = author_pubkey.to_owned(); Box::pin(async move { - let state = self - .state - .upgrade() - .ok_or_else(|| ActionSinkError::Database("relay is shutting down".into()))?; - - let host = state - .db - .lookup_community_host(community_id) - .await - .map_err(|e| ActionSinkError::Database(e.to_string()))? - .ok_or_else(|| { - ActionSinkError::Database(format!( - "workflow run community {community_id} is not mapped to a host" - )) - })?; - let tenant = buzz_core::tenant::TenantContext::resolved(community_id, host); + let (state, tenant) = self.resolve_state_and_tenant(community_id).await?; // Build a kind:46010 (KIND_WORKFLOW_APPROVAL_REQUESTED) event. // The `d` tag carries the token hash so the grant/deny handler From 30740e831153feecb07bc8b5392612fa8869c700 Mon Sep 17 00:00:00 2001 From: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> Date: Tue, 28 Jul 2026 20:41:26 +0900 Subject: [PATCH 11/27] docs: update workflow engine documentation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AGENTS.md — expand the workflow section: list all seven implemented actions with their event kinds, document the multi-layered loop prevention (depth cap), and describe the approval lifecycle (reaper, grant/deny resume). ARCHITECTURE.md — update the loop-prevention paragraph with the depth-cap layer, and add a workflow-actions paragraph covering ActionSink, community scoping, send_dm validation, request_approval, and the expiry reaper. CHANGELOG.md — add 8 entries under Backend (Rust) covering the action completion, depth cap, approval reaper, send_dm validation, publish- failure cleanup, finalize_run fix, and the ActionSink refactor. Signed-off-by: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> --- AGENTS.md | 27 +++++++++++++++++++++++++++ ARCHITECTURE.md | 4 +++- CHANGELOG.md | 14 ++++++++++++++ 3 files changed, 44 insertions(+), 1 deletion(-) diff --git a/AGENTS.md b/AGENTS.md index c94ba3881e..ab02a3be19 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -150,6 +150,33 @@ Filters and queries must scope to `h` tags when operating within a channel. [evalexpr](https://docs.rs/evalexpr) for condition evaluation. Keep expressions simple and testable. +**Workflow actions**: All seven actions defined in `schema.rs` (`ActionDef`) +are fully implemented through the `ActionSink` trait (`action_sink.rs`), +backed by `RelayActionSink` in `buzz-relay/src/workflow_sink.rs`: + +- `send_message` — kind:9 stream message (relay keypair signed) +- `send_dm` — opens/reuses a private DM channel via `open_dm`, posts kind:9 +- `set_channel_topic` — kind:9002 NIP-29 edit-metadata +- `add_reaction` — kind:7 NIP-25 reaction +- `call_webhook` — HTTP POST to external HTTPS endpoint (SSRF-guarded) +- `request_approval` — persists `workflow_approvals` row + emits kind:46010 +- `delay` — bounded sleep (max 270s) + +**Workflow loop prevention** is multi-layered: +1. Workflow execution kinds (46001–46012) are excluded from triggering. +2. Relay-signed messages with a `buzz:workflow` tag suppress single-hop echo. +3. **Cross-workflow depth cap** (`MAX_WORKFLOW_DEPTH = 3`): the tag carries + a depth field (`["buzz:workflow","true",""]`); the engine + suppresses triggering past the cap, catching transitive chains + (A→B→A) that the single-hop check misses. + +**Workflow approval lifecycle**: `request_approval` persists a DB row (token +SHA-256 hashed) and emits kind:46010. The existing grant/deny handler +(kind:46030/46031 in `command_executor.rs`) resumes the run via +`execute_from_step`. Expired pending approvals are reaped each cron tick by +`expire_pending_approvals` (atomic `UPDATE...RETURNING`, multi-pod safe) and +their runs transitioned to `Failed`. + **Thread counters**: `reply_count` and `descendant_count` are materialized on thread root events. Any code that inserts replies must update these counters — check existing reply handlers for the pattern. diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 90cbbac0cf..98b9279e82 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -241,7 +241,9 @@ Steps 10–12 are fire-and-forget. Search indexing is sent to a bounded worker q Step 9 (fan-out) explicitly **excludes** global subscriptions (no `channel_id` constraint) from channel-scoped events — global subscriptions do NOT receive events from private channels, regardless of filter match. This is a deliberate security boundary: only subscriptions scoped to an accessible `channel_id` receive those events. -Workflow loop prevention: workflow execution kinds (46001–46012), relay-signed messages with `buzz:workflow` tag, and `KIND_GIFT_WRAP` are excluded from triggering workflows. All other stored events (including kind 9 stream messages) trigger workflow evaluation. +Workflow loop prevention: workflow execution kinds (46001–46012), relay-signed messages with `buzz:workflow` tag, and `KIND_GIFT_WRAP` are excluded from triggering workflows. All other stored events (including kind 9 stream messages) trigger workflow evaluation. Loop prevention is multi-layered: (1) kind exclusion, (2) single-hop `buzz:workflow` tag suppression, and (3) a cross-workflow depth cap (`MAX_WORKFLOW_DEPTH = 3`) — the tag carries a depth field (`["buzz:workflow","true",""]`) and the engine suppresses triggering past the cap, catching transitive chains (A→B→A) that the single-hop check misses. + +Workflow actions: all seven `ActionDef` variants (`send_message`, `send_dm`, `set_channel_topic`, `add_reaction`, `call_webhook`, `request_approval`, `delay`) are implemented through the `ActionSink` trait, backed by `RelayActionSink`. Side effects are relay-keypair-signed Nostr events, community-scoped to the run's owning community via `resolve_state_and_tenant`. `send_dm` validates recipient community membership before opening a DM channel; `request_approval` persists a hashed-token approval row and emits kind:46010, which the existing grant/deny handler (46030/46031) resumes. Expired pending approvals are reaped each cron tick (atomic `UPDATE...RETURNING`) and their runs marked `Failed`. ### Ephemeral Sub-Pipeline (kinds 20000–29999) diff --git a/CHANGELOG.md b/CHANGELOG.md index c9e3597df3..60b13115dc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,19 @@ # Changelog +## Unreleased + +### Backend (Rust) + +- **feat(workflow): complete all seven workflow actions** — `send_dm`, `set_channel_topic`, `add_reaction`, and `request_approval` were previously stubs (`NotImplemented`). All are now fully wired through the `ActionSink` trait backed by `RelayActionSink` (relay-keypair-signed, community-scoped events). `send_dm` uses the DM-channel pattern (`open_dm` + kind:9), not NIP-17 gift-wrap. `request_approval` persists a hashed-token `workflow_approvals` row and emits kind:46010 — the existing grant/deny handler (46030/46031) resumes the run. +- **feat(workflow): cross-workflow trigger-chain depth cap** — `buzz:workflow` tag now carries a depth field (`["buzz:workflow","true",""]`); the engine suppresses triggering past `MAX_WORKFLOW_DEPTH` (3), catching transitive loops (A→B→A) that the single-hop tag check misses. Legacy two-element tags treated as depth 0 on read. +- **feat(workflow): approval expiry reaper** — `expire_pending_approvals()` (atomic `UPDATE...RETURNING`, multi-pod safe) runs each cron tick, transitioning expired pending approvals to `expired` and their waiting runs to `Failed`. Prevents indefinite DB accumulation and runs stuck in `waiting_approval`. +- **feat(workflow): GET /workflow-runs REST endpoint** — the `buzz workflows runs` CLI now reads the authoritative `workflow_runs` DB table via a NIP-98-authed REST endpoint instead of querying never-emitted kinds 46001–46003. +- **feat(workflow): query_count and query_messages actions** — two new data-driven actions: `query_count` (count events matching a filter for threshold alerts) and `query_messages` (fetch recent channel messages for context gathering). Both query `engine.db` directly via community-scoped `EventQuery`. +- **fix(workflow): send_dm recipient validation** — verifies the recipient is a member of the workflow's community (`is_relay_member`) before opening a DM channel, preventing workflows from creating DM channels to arbitrary pubkeys outside the community. +- **fix(workflow): request_approval cleanup on publish failure** — if the kind:46010 event publication fails after the DB row was created, the approval is marked `Denied` (best-effort) so it doesn't linger as a phantom pending row. +- **fix(workflow): approval-suspended runs now transition to WaitingApproval** — `finalize_run` previously contained stale stub logic that marked approval-suspended runs as `Failed` ("approval gates not yet implemented"). Now correctly transitions to `WaitingApproval` so the grant/deny handler can resume. +- **refactor(workflow): extract resolve_state_and_tenant helper + remove probe-sign** — the weak-ref upgrade + community→tenant resolution was duplicated across 5 ActionSink methods (~75 lines); centralized into one helper. The double-signing workaround in `add_reaction` (probe-sign to extract SDK builder tags) replaced by inline tag construction. Net −66 lines. + ## v0.4.23 - fix(desktop): strip GIF metadata extensions before upload ([#2425](https://github.com/block/buzz/pull/2425)) ([`47d7eb698`](https://github.com/block/buzz/commit/47d7eb6982900920bcdbe7a2f5013baca37daeeb)) From 25d7c042f3abf25103194761e2ed9d31e7751d3a Mon Sep 17 00:00:00 2001 From: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> Date: Tue, 28 Jul 2026 20:56:30 +0900 Subject: [PATCH 12/27] feat(workflow): add GET /workflow-runs REST endpoint + wire CLI The `buzz workflows runs` CLI subcommand existed but always returned [] because the relay never emits workflow execution events (kinds 46001-46012). Run state lives only in the workflow_runs DB table. Adds GET /workflow-runs?workflow=&limit= to the relay (NIP-98 auth, community-scoped via host binding, reads list_workflow_runs directly). The CLI's cmd_get_workflow_runs now calls get_authed instead of the Nostr query for kinds 46001-46003. Each run returned as JSON: id, workflow_id, status, current_step, started_at, completed_at, error_message. Signed-off-by: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> --- crates/buzz-cli/src/commands/workflows.rs | 33 +++------ crates/buzz-relay/src/api/bridge.rs | 87 +++++++++++++++++++++++ crates/buzz-relay/src/router.rs | 2 + 3 files changed, 97 insertions(+), 25 deletions(-) diff --git a/crates/buzz-cli/src/commands/workflows.rs b/crates/buzz-cli/src/commands/workflows.rs index 2786d2c508..f44b55b946 100644 --- a/crates/buzz-cli/src/commands/workflows.rs +++ b/crates/buzz-cli/src/commands/workflows.rs @@ -57,12 +57,11 @@ pub async fn cmd_get_workflow(client: &BuzzClient, workflow_id: &str) -> Result< Ok(()) } -/// Get workflow run history — query kinds [46001, 46002, 46003]. +/// Get workflow run history — `GET /workflow-runs?workflow=`. /// -/// NOTE: The relay does not currently emit workflow execution events (46001-46003). -/// Run history is stored in the workflow_runs DB table, not as Nostr events. -/// This command will return an empty array until the relay adds event emission -/// or a dedicated REST endpoint for run history. +/// Reads the authoritative `workflow_runs` DB table via the relay's REST +/// endpoint (NIP-98 auth). Each run includes status, current_step, +/// started_at, completed_at, and error_message. pub async fn cmd_get_workflow_runs( client: &BuzzClient, workflow_id: &str, @@ -70,26 +69,10 @@ pub async fn cmd_get_workflow_runs( ) -> Result<(), CliError> { validate_uuid(workflow_id)?; let limit = limit.unwrap_or(20).min(100); - let filter = serde_json::json!({ - "kinds": [46001, 46002, 46003], - "#d": [workflow_id], - "limit": limit - }); - let resp = client.query(&filter).await?; - let events: Vec = serde_json::from_str(&resp).unwrap_or_default(); - let normalized: Vec = events - .iter() - .map(|e| { - serde_json::json!({ - "event_id": e.get("id").and_then(|v| v.as_str()).unwrap_or(""), - "kind": e.get("kind").and_then(|v| v.as_u64()).unwrap_or(0), - "content": e.get("content").and_then(|v| v.as_str()).unwrap_or(""), - "created_at": e.get("created_at").and_then(|v| v.as_u64()).unwrap_or(0), - "tags": e.get("tags").cloned().unwrap_or(serde_json::json!([])), - }) - }) - .collect(); - let output = serde_json::to_string(&normalized).unwrap_or_default(); + let path = format!("/workflow-runs?workflow={workflow_id}&limit={limit}"); + let resp = client.get_authed(&path).await?; + let runs: Vec = serde_json::from_str(&resp).unwrap_or_default(); + let output = serde_json::to_string(&runs).unwrap_or_default(); println!("{output}"); Ok(()) } diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 8372e49a2a..9a30c89b45 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -2129,6 +2129,93 @@ pub async fn moderation_restricted( Ok(Json(Value::Array(rows.iter().map(ban_json).collect()))) } +/// Query parameters for `GET /workflow-runs`. +#[derive(Debug, serde::Deserialize)] +pub struct WorkflowRunsQuery { + /// Workflow UUID to filter runs by (required). + workflow: String, + /// Maximum number of runs to return (newest first). Capped at 100. + limit: Option, +} + +const WORKFLOW_RUNS_LIMIT: i64 = 100; + +/// `GET /workflow-runs` — workflow run history (NIP-98 auth). +/// +/// Returns the `workflow_runs` DB rows for the caller's community, filtered by +/// workflow UUID. Unlike the Nostr event approach (kinds 46001–46012, which the +/// relay does not emit), this reads the authoritative run state directly from +/// the database. Community-scoped via host binding. +pub async fn workflow_runs( + State(state): State>, + headers: HeaderMap, + RawQuery(raw_query): RawQuery, + Query(q): Query, +) -> Result, (StatusCode, Json)> { + let raw_host = headers + .get(axum::http::header::HOST) + .and_then(|v| v.to_str().ok()) + .unwrap_or(""); + let tenant = crate::tenant::bind_community(&state.db, raw_host) + .await + .map_err(|_| { + api_error( + StatusCode::NOT_FOUND, + "relay: no community is configured for this host", + ) + })?; + + // NIP-98 auth — the URL must include the query string for GET requests + // so the signature covers the full request (path + query params). + let path_with_query = match raw_query.as_deref() { + Some(rq) => format!("/workflow-runs?{rq}"), + None => "/workflow-runs".to_owned(), + }; + let url = nip98_expected_url(&state.config.relay_url, &tenant, &path_with_query); + let (pubkey, _event_id_bytes) = + verify_bridge_auth(&headers, "GET", &url, None, state.config.require_auth_token)?; + + let limit = q + .limit + .filter(|n| *n > 0) + .map(|n| n.min(WORKFLOW_RUNS_LIMIT)) + .unwrap_or(WORKFLOW_RUNS_LIMIT); + + let workflow_uuid = uuid::Uuid::parse_str(&q.workflow) + .map_err(|_| api_error(StatusCode::BAD_REQUEST, "invalid workflow UUID"))?; + + let rows = state + .db + .list_workflow_runs(tenant.community(), workflow_uuid, limit) + .await + .map_err(|e| internal_error(&format!("list workflow runs: {e}")))?; + + tracing::info!( + pubkey = %pubkey.to_hex(), + route = "/workflow-runs", + status = 200u16, + result_count = rows.len(), + "HTTP bridge request" + ); + + Ok(Json(Value::Array( + rows.iter().map(workflow_run_json).collect(), + ))) +} + +/// Serialize a [`WorkflowRunRecord`] to JSON for the REST response. +fn workflow_run_json(r: &buzz_db::workflow::WorkflowRunRecord) -> Value { + serde_json::json!({ + "id": r.id, + "workflow_id": r.workflow_id, + "status": r.status.to_string(), + "current_step": r.current_step, + "started_at": r.started_at.map(|dt| dt.to_rfc3339()), + "completed_at": r.completed_at.map(|dt| dt.to_rfc3339()), + "error_message": r.error_message, + }) +} + fn report_json(r: &buzz_db::moderation::ReportRecord) -> Value { let (target_kind, target) = match &r.target { buzz_db::moderation::ReportTarget::Event(id) => ("event", hex::encode(id)), diff --git a/crates/buzz-relay/src/router.rs b/crates/buzz-relay/src/router.rs index 2375a4f9c6..67f7a76618 100644 --- a/crates/buzz-relay/src/router.rs +++ b/crates/buzz-relay/src/router.rs @@ -118,6 +118,8 @@ pub fn build_router(state: Arc) -> Router { ) // Webhook trigger (secret-authenticated, no NIP-98) .route("/hooks/{id}", post(api::bridge::workflow_webhook)) + // Workflow run history (NIP-98 auth, reads workflow_runs DB table) + .route("/workflow-runs", get(api::bridge::workflow_runs)) // Mesh demo echo probe — testbed-only; 404 unless BUZZ_MESH=on and // BUZZ_MESH_DEMO_ECHO=on (see api::mesh_demo). .route("/_mesh/demo/echo", post(api::mesh_demo::demo_echo)) From 8595b23f57e4ca8cc4908a04a55a0d7b2d96adc8 Mon Sep 17 00:00:00 2001 From: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> Date: Tue, 28 Jul 2026 21:02:56 +0900 Subject: [PATCH 13/27] feat(workflow): add query_count and query_messages actions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two new data-driven actions that let workflows read relay state: query_count — counts events matching a filter (kinds + optional since duration). Returns {"count": N} as step output, usable in subsequent if: conditions for threshold alerts: - id: count action: query_count kinds: [9] since: 1h - id: alert if: 'steps_count_output_count > 100' action: send_message text: 'High traffic alert!' query_messages — fetches recent messages from a channel (default 10, max 50). Returns {"messages": [...]} for context gathering: - id: context action: query_messages limit: 5 Both query engine.db directly (community-scoped via EventQuery), so no ActionSink method is needed. Includes build_channel_query helper to avoid repeating EventQuery's many fields. Signed-off-by: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> --- crates/buzz-workflow/src/executor.rs | 162 +++++++++++++++++++++++++++ crates/buzz-workflow/src/schema.rs | 31 ++++- 2 files changed, 192 insertions(+), 1 deletion(-) diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index 5290aa67de..7ccc2dc9f8 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -12,6 +12,7 @@ use std::collections::HashMap; use buzz_core::tenant::CommunityId; +use chrono::Utc; use evalexpr::HashMapContext; use nostr::ToBech32; use serde_json::Value as JsonValue; @@ -455,6 +456,19 @@ pub fn resolve_step_templates( Delay { duration } => Ok(Delay { duration: duration.clone(), }), + QueryCount { + channel, + kinds, + since, + } => Ok(QueryCount { + channel: t_opt(channel)?, + kinds: kinds.clone(), + since: since.clone(), + }), + QueryMessages { channel, limit } => Ok(QueryMessages { + channel: t_opt(channel)?, + limit: *limit, + }), } } @@ -907,7 +921,155 @@ pub async fn dispatch_action( serde_json::json!({ "slept_secs": secs }), )) } + + QueryCount { + channel, + kinds, + since, + } => { + if kinds.is_empty() { + return Err(WorkflowError::InvalidDefinition(format!( + "QueryCount step '{step_id}': 'kinds' must not be empty" + ))); + } + + let channel_id = resolve_query_channel(channel.as_deref(), &trigger_ctx.channel_id)?; + let channel_uuid = Uuid::parse_str(&channel_id).map_err(|e| { + WorkflowError::InvalidDefinition(format!("QueryCount: invalid channel UUID: {e}")) + })?; + + let since_dt = match since.as_deref() { + Some(dur) => { + let secs = parse_duration_secs(dur)?; + Some(Utc::now() - chrono::Duration::seconds(i64::try_from(secs).unwrap_or(0))) + } + None => None, + }; + + let query = build_channel_query( + community_id, + channel_uuid, + kinds.iter().map(|k| *k as i32).collect(), + since_dt, + None, + ); + + let count = engine + .db + .count_events(&query) + .await + .map_err(|e| WorkflowError::WebhookError(format!("QueryCount: {e}")))?; + + info!( + run_id = %run_id, + step = step_id, + channel = %channel_id, + count, + "QueryCount → {count} events" + ); + + Ok(StepResult::Completed(serde_json::json!({ "count": count }))) + } + + QueryMessages { channel, limit } => { + let channel_id = resolve_query_channel(channel.as_deref(), &trigger_ctx.channel_id)?; + let channel_uuid = Uuid::parse_str(&channel_id).map_err(|e| { + WorkflowError::InvalidDefinition(format!( + "QueryMessages: invalid channel UUID: {e}" + )) + })?; + + let limit = limit.unwrap_or(10).min(50) as i64; + + let query = build_channel_query( + community_id, + channel_uuid, + vec![9], // KIND_STREAM_MESSAGE + None, + Some(limit), + ); + + let events = engine + .db + .query_events(&query) + .await + .map_err(|e| WorkflowError::WebhookError(format!("QueryMessages: {e}")))?; + + let messages: Vec = events + .into_iter() + .map(|e| { + serde_json::json!({ + "id": e.event.id.to_hex(), + "content": e.event.content, + "author": e.event.pubkey.to_hex(), + "created_at": e.event.created_at.as_secs(), + }) + }) + .collect(); + + info!( + run_id = %run_id, + step = step_id, + channel = %channel_id, + fetched = messages.len(), + "QueryMessages → {} messages", + messages.len() + ); + + Ok(StepResult::Completed( + serde_json::json!({ "messages": messages }), + )) + } + } +} + +/// Build a community-scoped `EventQuery` with only the fields relevant to +/// workflow query actions. All other fields default to None/false. +fn build_channel_query( + community_id: CommunityId, + channel_uuid: Uuid, + kinds: Vec, + since: Option>, + limit: Option, +) -> buzz_db::EventQuery { + buzz_db::EventQuery { + community_id, + channel_id: Some(channel_uuid), + kinds: Some(kinds), + pubkey: None, + since, + until: None, + limit, + offset: None, + p_tag_hex: None, + d_tag: None, + d_tags: None, + before_id: None, + global_only: false, + authors: None, + ids: None, + e_tags: None, + channel_ids: None, + max_limit: None, + } +} + +/// Resolve the channel for a query action: use the override if provided, +/// otherwise fall back to the trigger's channel. Fails if neither is set. +fn resolve_query_channel( + channel_override: Option<&str>, + trigger_channel: &str, +) -> Result { + let channel = channel_override + .filter(|s| !s.trim().is_empty()) + .map(|s| s.to_owned()) + .unwrap_or_else(|| trigger_channel.to_owned()); + if channel.trim().is_empty() { + return Err(WorkflowError::InvalidDefinition( + "query action has no channel (neither override nor trigger.channel_id)".into(), + )); } + Ok(channel) } /// Generate a cryptographically random approval token. diff --git a/crates/buzz-workflow/src/schema.rs b/crates/buzz-workflow/src/schema.rs index 09838d041b..c4eafc77a7 100644 --- a/crates/buzz-workflow/src/schema.rs +++ b/crates/buzz-workflow/src/schema.rs @@ -144,6 +144,28 @@ pub enum ActionDef { /// Duration string (e.g. `"5m"`, `"1h"`). duration: String, }, + /// Count events matching a filter. Returns `{ "count": N }` as step output, + /// usable in subsequent `if:` conditions (e.g. threshold alerts). + QueryCount { + /// Channel UUID. Defaults to the trigger's channel. + #[serde(default)] + channel: Option, + /// Event kinds to count (e.g. `[9]` for messages). Required. + kinds: Vec, + /// Only count events newer than this duration (e.g. `"1h"`, `"30m"`). + #[serde(default)] + since: Option, + }, + /// Fetch recent messages from a channel. Returns an array of + /// `{ "id", "content", "author", "created_at" }` as step output. + QueryMessages { + /// Channel UUID. Defaults to the trigger's channel. + #[serde(default)] + channel: Option, + /// Maximum number of messages to return (newest first). Default 10, max 50. + #[serde(default)] + limit: Option, + }, } impl WorkflowDef { @@ -351,9 +373,11 @@ mod tests { " - id: hook\n action: call_webhook\n url: https://hooks.example.com/notify\n method: POST\n", " - id: approve\n action: request_approval\n from: '@manager'\n message: Approve?\n timeout: 4h\n", " - id: wait\n action: delay\n duration: 5m\n", + " - id: count\n action: query_count\n kinds: [9]\n since: 1h\n", + " - id: msgs\n action: query_messages\n limit: 5\n", ); let (def, _) = parse_yaml(yaml).expect("parse failed"); - assert_eq!(def.steps.len(), 5); + assert_eq!(def.steps.len(), 7); assert!(matches!( &def.steps[0].action, @@ -372,6 +396,11 @@ mod tests { ActionDef::RequestApproval { .. } )); assert!(matches!(&def.steps[4].action, ActionDef::Delay { .. })); + assert!(matches!(&def.steps[5].action, ActionDef::QueryCount { .. })); + assert!(matches!( + &def.steps[6].action, + ActionDef::QueryMessages { .. } + )); } #[test] From 85c6b2574b38d9440e0bab35ae5ec9301ff359a1 Mon Sep 17 00:00:00 2001 From: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> Date: Tue, 28 Jul 2026 21:31:48 +0900 Subject: [PATCH 14/27] feat(landing): add workflow automation section to product page MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a dedicated Workflow section with 6 feature cards (9 action types, 5 trigger types, loop safety, SSRF guards, at-most-once scheduling, run monitoring) plus 3 new feature cards in the main grid (workflow automation, data-driven triggers, approval gates). Updates nav link and feature count (23+ → 26+). Signed-off-by: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> --- landing/index.html | 342 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 342 insertions(+) create mode 100644 landing/index.html diff --git a/landing/index.html b/landing/index.html new file mode 100644 index 0000000000..b9c84fb7f2 --- /dev/null +++ b/landing/index.html @@ -0,0 +1,342 @@ + + + + + +Buzz — The Open-Source Workspace for Humans + AI Agents + + + + + + + + + +
+
+
Open Source · Apache 2.0 · Self-Hostable
+

Where humans & AI
work as equals

+

Buzz is a self-hostable team communication platform built on Nostr. Every action is a cryptographically signed event. Agents are first-class members — not bots.

+
+⬇ Download +★ Star on GitHub +
+
+
26+
Features
+
0
Tracking
+
∞
Self-Hosted
+
100%
Open Source
+
+
+
+ + +
+
+
+

Everything you need. Nothing you don't.

+

Discord-level chat features, plus AI-native capabilities no other platform offers.

+
+
+
🔐

Permission Matrix

Discord-style role×permission overrides. Granular control over who can send, delete, pin, and manage — per channel.

+
🎬

GIF Search

Built-in Tenor GIF search. Select, re-host on your own S3, and share — no external hot-linking, no tracking.

+
🎤

Voice Messages

Record voice clips directly in the composer. Opus-encoded, stored on your infrastructure, played inline.

+
📌

Pinned & Saved

Pin important messages channel-wide. Bookmark messages privately for later. Full edit history on every message.

+
📢

@channel / @here

Broadcast mentions with server-side fan-out. Permission-gated, capped at 500 recipients for safety.

+
⚡

Slash Commands

/me, /shrug, /spoiler, /tableflip, /poll. IRC-style commands that transform into rich markdown.

+
🤖

AI Thread Summary

One click to summarize long threads. Agents read the conversation and produce a 3-line digest.

+
🌍

AI Translation

Right-click any message → "Translate with AI." Your agent handles it inline. No external service.

+
📅

Scheduled Messages

Write now, send later. Preset delays or custom times. Persists across app restarts.

+
💬

Code Blocks+

Syntax highlighting with line numbers, collapse/expand, and a language header bar. Shiki-powered.

+
🔇

DND + Smart Notifications

Do Not Disturb mode that still lets direct mentions through. Per-channel notification sounds.

+
🔀

Forward & Reply

Forward messages across channels with source attribution. Reply with threading context.

+
⚙️

Workflow Automation

YAML-as-code automation with 9 action types. Trigger on messages, reactions, schedules, or webhooks. Condition logic via evalexpr.

+
📊

Data-Driven Triggers

Query message counts, fetch recent context, and branch on thresholds. "If 100+ messages in 1h → alert the team."

+
✅

Approval Gates

Pause workflows for human approval. Cryptographically tokened requests, expiry reaper, one-click grant/deny.

+
+
+
+ + +
+
+
+

Automation that's actually powerful

+

A full workflow engine built into the relay. No external services, no webhooks-to-nowhere. Every step is community-scoped and audit-logged.

+
+
+
📨

9 Action Types

send_message, send_dm, set_channel_topic, add_reaction, call_webhook, request_approval, delay, query_count, query_messages.

+
🔗

5 Trigger Types

message_posted, reaction_added, diff_posted, schedule (cron + interval), and webhook. Each with optional filter expressions.

+
🛡️

Loop-Safe

Multi-layer loop prevention: kind exclusion, single-hop tag suppression, and cross-workflow depth cap (MAX=3).

+
🌐

SSRF-Guarded Webhooks

Call external HTTPS endpoints safely. DNS pinning, redirect blocking, private-IP filtering, CGNAT/benchmarking range blocking.

+
⏱️

At-Most-Once Scheduling

Deterministic fire instants + atomic DB claims. Multi-pod safe — only one instance fires, even across N replicas.

+
📈

Run Monitoring

Full execution traces in the DB. Query run history via REST API or buzz workflows runs CLI. Every step tracked.

+
+
+
+ + +
+
+
+

Agents are members, not bots

+

In Buzz, AI agents have their own keys, personas, audit trails, and channel memberships — the same affordances as humans.

+
+
+
🔑

Own Identity

Each agent gets a Nostr keypair. Actions are signed and attributable.

+
🎭

Personas & Teams

Configure agents with custom personas. Group them into teams for coordinated work.

+
🛠️

Tool Calls

Agents use MCP tools — shell, file-edit, code execution — with full observer transcripts.

+
📝

Git Integration

Agents open PRs, review code, and manage issues via NIP-34 git hosting built into the relay.

+
🔄

Workflows

YAML-as-code automation engine. Triggers on messages, reactions, webhooks, and schedules.

+
🗣️

Voice Huddles

Agents join audio huddles with their own STT/TTS. Participate in real-time voice conversations.

+
+
+
+ + +
+
+
+

Buzz vs Discord

+

Same familiar features. Fundamentally different architecture.

+
+
+ + + + + + + + + + + + + + + + + + + + + +
FeatureBuzzDiscord
Text channels + threads✅✅
Voice / huddles✅ Audio✅ Audio + Video
GIF search✅ Self-hosted✅ Tenor
Voice messages✅❌
Permission matrix✅ Per-channel overrides✅
@channel / @here✅ Server fan-out✅
Slash commands✅ /me /shrug /poll✅
AI agents as members✅ First-class❌ Bots only
Git / PR integration✅ Built-in NIP-34❌ Via bots
Workflow automation✅ YAML engine❌
Self-hostable✅ One binary❌
End-to-end signed events✅ Nostr❌
Open source✅ Apache 2.0❌
Server discovery❌ Invite-based✅
Video calls / screen share❌ Planned✅
+
+
+
+ + +
+
+
+

Security by architecture

+

Buzz doesn't bolt on security — it's built into the protocol.

+
+
+

🔏 Cryptographically Signed

Every message, reaction, edit, and workflow step is a signed Nostr event. Tampering is mathematically detectable.

+

🏗️ Self-Hosted

Run on your own infrastructure. Your data never touches a third-party server. One binary, Postgres, Redis, S3.

+

📋 Tamper-Evident Audit Log

Per-community SHA-256 hash chain. Every moderation action is recorded and verifiable.

+

🔑 NIP-42 Auth

Connection-level authentication. No anonymous access without explicit configuration.

+

🛡️ Permission Gates

Server-side enforcement on every write. Pin, broadcast, edit — all validated before storage.

+

🔒 Blossom Media

BUD-01/BUD-11 compliant media storage. Authenticated uploads, content-addressed by SHA-256.

+
+
+
+ + +
+
+
+

Ready to Buzz?

+

Self-host in minutes, or join an existing community.

+
+⬇ Download Buzz +📖 Read the Docs +
+

Available for macOS, Windows, and Linux · iOS & Android in development

+
+
+
+ + + + + + + From 17c182b6e8bdacc4ea9edf8304c086055bdb8744 Mon Sep 17 00:00:00 2001 From: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> Date: Tue, 28 Jul 2026 21:38:27 +0900 Subject: [PATCH 15/27] feat(tools): add standalone workflow builder UI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A browser-based visual editor for Buzz workflow YAML definitions — no dependencies, single HTML file. Features: - Visual step builder with drag-to-reorder (up/down) - All 9 action types with per-type forms (send_message, send_dm, set_channel_topic, add_reaction, call_webhook, request_approval, delay, query_count, query_messages) - All 5 trigger types with conditional config fields - Real-time YAML generation as you edit - Live validation: required fields, step ID rules (alphanumeric + underscore, no duplicates), schedule cron/interval mutual exclusion - Example workflow loader (high-traffic alert) - Copy to clipboard + download as .yaml - Dark theme matching the product landing page Located at tools/workflow-builder/index.html — open in any browser. Signed-off-by: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> --- tools/workflow-builder/index.html | 601 ++++++++++++++++++++++++++++++ 1 file changed, 601 insertions(+) create mode 100644 tools/workflow-builder/index.html diff --git a/tools/workflow-builder/index.html b/tools/workflow-builder/index.html new file mode 100644 index 0000000000..72b6a0fc8d --- /dev/null +++ b/tools/workflow-builder/index.html @@ -0,0 +1,601 @@ + + + + + +Buzz Workflow Builder + + + + + +
+
+

⚙️ Workflow Builder

+
+ + +
+
+
+ +
+

Workflow Name

+
+ +
+
+ + +
+

Trigger

+
+
📨 Message Posted
+
👍 Reaction Added
+
📝 Diff Posted
+
⏰ Schedule
+
🌐 Webhook
+
+
+
+
+
+ + +
+
+

📋 Steps

+
+
+
+
📋
+

No steps yet

+

Click an action below to add your first step.
Steps execute sequentially — top to bottom.

+
+
+ + +
+Add: + + + + + + + + + +
+
+ + +
+
+

📄 YAML Output

+
+ + +
+
+
+
+

✅ Valid

+
+
+ + + + From b8e700af5c70fc16204e50f0fca16cf3f347787e Mon Sep 17 00:00:00 2001 From: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> Date: Tue, 28 Jul 2026 21:41:19 +0900 Subject: [PATCH 16/27] ui(workflow-builder): redesign with modern visual polish - Refined color palette (deeper dark bg, softer borders) - Per-action colored left borders on step cards - YAML syntax highlighting (keys, strings, comments color-coded) - Step connectors (vertical lines + dots between steps) - Smooth fade-in animation on step body expand - Bounce arrow on empty state - Toast notifications for copy/download/example/reset - Collapsible trigger config (cleaner sidebar) - Variables cheat-sheet in sidebar - Improved typography (Inter font stack) - Better scrollbar styling - Pulse/glow on hover states Signed-off-by: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> --- tools/workflow-builder/index.html | 844 ++++++++++++++---------------- 1 file changed, 396 insertions(+), 448 deletions(-) diff --git a/tools/workflow-builder/index.html b/tools/workflow-builder/index.html index 72b6a0fc8d..247d04460b 100644 --- a/tools/workflow-builder/index.html +++ b/tools/workflow-builder/index.html @@ -5,270 +5,317 @@ Buzz Workflow Builder - +
-
-

⚙️ Workflow Builder

-
- - -
-
-
- -
-

Workflow Name

-
- -
-
- - -
-

Trigger

-
-
📨 Message Posted
-
👍 Reaction Added
-
📝 Diff Posted
-
⏰ Schedule
-
🌐 Webhook
-
-
-
-
+
+ +
+ + +
+
+ +
+ +
+

Workflow Name

+
+ +
+
+ + +
+

Trigger

+
+
+ 📨 Message Posted +
+
+ 👍 Reaction Added +
+
+ 📝 Diff Posted +
+
+ ⏰ Schedule +
+
+ 🌐 Webhook +
+
+
+
+ + +
+

Variables

+
+
{{trigger.text}} — message content
+
{{trigger.author}} — sender pubkey
+
{{trigger.channel_id}} — channel UUID
+
{{steps.ID.output.X}} — step result
+
+
+
- +
-
-

📋 Steps

-
-
-
-
📋
-

No steps yet

-

Click an action below to add your first step.
Steps execute sequentially — top to bottom.

-
-
- - -
-Add: - - - - - - - - - -
+
+

📋 Steps 0

+
Steps execute top → bottom
+
+ +
+
+
📋
+

No steps yet

+

Pick an action below to add your first step.
Each step runs sequentially when the workflow fires.

+
↓
+
+
+ + +
+ Add Step + + + + + + + + + +
- +
-
-

📄 YAML Output

-
- - -
-
-
-
-

✅ Valid

-
+
+

📄 YAML

+
+ + +
+
+
+
+

✅ Valid — ready to deploy

+
+ +
+ From 3f861a685746e0e060a255786c57b64f0dd6f90f Mon Sep 17 00:00:00 2001 From: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> Date: Tue, 28 Jul 2026 21:44:52 +0900 Subject: [PATCH 17/27] feat(tools): add standalone ops dashboard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A browser-based operations dashboard for monitoring a Buzz relay in real time. Single HTML file, no dependencies. Features: - Status cards: relay online/offline, version, uptime, health port - Uptime bar: 60-segment visual history (green=up, red=down, gray=unknown) - Auto-refresh every 5s (pause/resume toggle) - Workflow runs table: status badges, step progress, timestamps, errors — calls GET /workflow-runs?workflow= (NIP-98-agnostic local dev) - Relay info (NIP-11): name, software, supported NIPs, pubkey, contact - Error handling: connection failures surfaced inline - Dark theme consistent with workflow builder + landing page Located at tools/ops-dashboard/index.html — open in any browser and connect to http://localhost:3000 (or any relay URL). Signed-off-by: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> --- tools/ops-dashboard/index.html | 422 +++++++++++++++++++++++++++++++++ 1 file changed, 422 insertions(+) create mode 100644 tools/ops-dashboard/index.html diff --git a/tools/ops-dashboard/index.html b/tools/ops-dashboard/index.html new file mode 100644 index 0000000000..ad342e91d8 --- /dev/null +++ b/tools/ops-dashboard/index.html @@ -0,0 +1,422 @@ + + + + + +Buzz Ops Dashboard + + + + + +
+ +
+ + +
+ + +
+ +
+ +
+
📊
+

Connect to a relay

+

Enter your relay URL above and click Connect.
Default: http://localhost:3000

+
+ + +
+ + +
+ + + + From 8361bf63727a8719c8e54ac7b14b1b73e3c7bd77 Mon Sep 17 00:00:00 2001 From: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> Date: Tue, 28 Jul 2026 21:48:30 +0900 Subject: [PATCH 18/27] feat(tools): add interactive developer documentation site A standalone single-page developer documentation site for Buzz. No dependencies, single HTML file. Sections: - Introduction: architecture overview, crate layering, Nostr-first principle - Quick Start: relay setup, first event, desktop app - Core Concepts: events, NIP-29 community scoping, agent stack model - WebSocket API: EVENT/REQ/AUTH (NIP-01/NIP-42) - HTTP Endpoints: searchable table of all 12+ routes - Authentication: NIP-42 (WS) + NIP-98 (HTTP) + scope table - Event Kinds: searchable table of 45+ kind constants - Workflows: YAML syntax, 5 triggers, 9 actions, loop prevention, templates - Agents: ACP protocol, runtime table, config env vars - CLI: channels, workflows, messages, agents commands Features: sidebar navigation, code copy buttons, syntax highlighting, searchable tables (kinds + endpoints), responsive layout, dark theme. Located at tools/dev-docs/index.html. Signed-off-by: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> --- tools/dev-docs/index.html | 691 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 691 insertions(+) create mode 100644 tools/dev-docs/index.html diff --git a/tools/dev-docs/index.html b/tools/dev-docs/index.html new file mode 100644 index 0000000000..0258bd9c78 --- /dev/null +++ b/tools/dev-docs/index.html @@ -0,0 +1,691 @@ + + + + + +Buzz — Developer Docs + + + + + + + + + + +
+
+ + +
+
+
+ 📡 Nostr Protocol + 🔐 Cryptographically Signed + 🤖 AI-Native + Rust +
+

Welcome to Buzz

+
+

Buzz is a self-hostable team communication platform built on the Nostr protocol. Every action — a message, reaction, workflow step, or agent turn — is a cryptographically signed event. Humans and AI agents are first-class equals.

+ +

🏗️ Architecture #

+

Buzz is composed of 25+ Rust crates organized in layers:

+
+
text
+ 
# Foundation (zero I/O)
+buzz-core          # Types, event verification, filter matching, kind registry
+
+# Data / Infrastructure
+buzz-db            # Postgres event store
+buzz-pubsub        # Redis fan-out, presence, typing
+buzz-search        # Postgres FTS full-text search
+buzz-auth          # NIP-42/NIP-98 authentication
+buzz-media         # Blossom/S3 media storage
+buzz-audit         # Hash-chain audit log
+
+# Relay (integration hub)
+buzz-relay         # WebSocket relay server (60K lines)
+
+# Agent surface
+buzz-acp           # ACP harness (Nostr ↔ agent bridge)
+buzz-agent         # LLM runner (Anthropic/OpenAI/Databricks)
+buzz-dev-mcp       # MCP server (shell, file, web tools)
+
+# Clients
+buzz-cli           # Agent-first CLI
+desktop/           # Tauri 2 + React 19
+mobile/            # Flutter
+
+ +
+
💡 Key Principle: Nostr-First
+ Buzz's primary API is NIP-29 over WebSocket. New features are modeled as Nostr events (new kind integers), not HTTP endpoints. This gives you realtime fan-out, NIP-29 community scoping, and the auth pipeline for free. +
+
+ + +
+

🚀 Quick Start

+

Get a Buzz relay running locally in under 5 minutes.

+ +

Prerequisites

+
+
bash
+ 
# Required
+rust               # cargo, rustc
+node + pnpm        # desktop frontend
+docker/podman      # Postgres + Redis
+
+# Activate the Hermit toolchain
+. ./bin/activate-hermit
+
+ +

1. Start the relay

+
+
bash
+ 
just relay    # starts at ws://localhost:3000
+
+ +

2. Send your first event

+
+
bash
+ 
# Using the buzz CLI
+export BUZZ_PRIVATE_KEY="your-nostr-private-key"
+export BUZZ_RELAY_URL="ws://localhost:3000"
+
+# Send a channel message
+buzz channels messages send --channel "general" --text "Hello, Buzz!"
+
+ +

3. Run the desktop app

+
+
bash
+ 
just dev    # full Tauri app with relay
+
+ +
+
✨ Tip
+ The relay auto-migrates on startup if BUZZ_AUTO_MIGRATE is set. Local dev uses Docker for Postgres + Redis via just setup. +
+
+ + +
+

🧠 Core Concepts

+

The mental model for building on Buzz.

+ +

Events are Everything

+

Every state change in Buzz is a Nostr event — a JSON object with a kind, content, tags, and a cryptographic signature. Events are stored, filtered, and fanned out in realtime.

+
+
json
+ 
{
+  "id": "event-id-hex",
+  "pubkey": "author-pubkey-hex",
+  "kind": 9,
+  "content": "Hello, world!",
+  "tags": [["h", "channel-uuid"]],
+  "sig": "signature-hex",
+  "created_at": 1722172800
+}
+
+ +

Community Scoping (NIP-29)

+

Communities are bound by the relay's Host header. Every request is resolved to a community before any handler runs. An unmapped host fails closed — there is no default community.

+
+
⚠️ Fail-Closed Design
+ If the Host header doesn't map to a community, the relay returns 404. This is intentional — it prevents data leakage across tenants. +
+ +

Channels

+

Channels use h tags (NIP-29 group tag), not e tags. All filters and queries must scope to h tags when operating within a channel.

+ +

Agent Stack

+

AI agents are modeled as Nostr participants with their own keypairs, personas, and channel memberships. The three-process model:

+
+
text
+ 
relay  ↔  buzz-acp (harness)  ↔  buzz-agent (LLM runner)
+                                   ↘  buzz-dev-mcp (tools)
+
+# Communication
+relay → buzz-acp:   Nostr events over WebSocket
+buzz-acp → agent:   ACP v2 (stdio JSON-RPC / NDJSON)
+agent → dev-mcp:    MCP protocol (stdio, rmcp)
+
+
+ + +
+

🔌 WebSocket API (NIP-01)

+

The primary API surface. Connect via WebSocket to ws://relay:3000.

+ +

EVENT — Publish

+

Send a signed Nostr event to the relay:

+
+
json
+ 
["EVENT", {
+  "id": "...",
+  "pubkey": "...",
+  "kind": 9,
+  "content": "Hello!",
+  "tags": [["h", "channel-uuid"]],
+  "sig": "...",
+  "created_at": 1722172800
+}]
+# Response: ["OK", "event-id", true, ""]
+
+ +

REQ — Subscribe

+

Open a subscription with Nostr filters:

+
+
json
+ 
["REQ", "subscription-id", {
+  "kinds": [9],
+  "#h": ["channel-uuid"],
+  "limit": 50
+}]
+# Events streamed: ["EVENT", "sub-id", {...}]
+# End of stored: ["EOSE", "sub-id"]
+
+ +
+
⚠️ Kinds Required
+ Relay queries must specify kinds. Omitting kinds triggers the p-gate (403). Always include explicit kind filters. +
+ +

NIP-42 Auth

+

The relay may send an AUTH challenge. Respond with a signed kind:22242 event to authenticate:

+
+
json
+ 
# Relay sends:
+["AUTH", "challenge-string"]
+
+# Client responds:
+["AUTH", { /* kind:22242 event signed with challenge */ }]
+
+
+ + +
+

🌐 HTTP Endpoints

+

A narrow HTTP surface alongside the WebSocket API. All preserve the host-derived community boundary.

+ +
+ +
+ + + + + + + + + + + + + + + +
MethodPathDescriptionAuth
POST/eventsSubmit a signed Nostr eventNIP-98
POST/queryNostr REQ filters over HTTP (NIP-50 search routed to FTS)NIP-98
POST/countNostr COUNT filters over HTTPNIP-98
GET/workflow-runsWorkflow run history (query param: ?workflow=<uuid>)NIP-98
GET/infoRelay info (NIP-11)None
GET/.well-known/nostr.jsonNIP-05 identifier lookupNone
POST/hooks/{id}Workflow webhook trigger (secret-authed)Secret
POST/uploadBlossom media uploadNIP-98
GET/moderation/reportsModeration queueNIP-98
GET/_readinessReadiness probe (Postgres + Redis check)None
GET/_statusRelay version + uptimeNone
+
+ +

POST /events Example

+
+
bash
+ 
curl -X POST https://relay.example.com/events \
+  -H "Authorization: Nostr <base64-event>" \
+  -H "Content-Type: application/json" \
+  -d '{"id":"...","kind":9,"content":"Hello","pubkey":"...","sig":"...","tags":[["h","uuid"]],"created_at":1722172800}'
+
+
+ + +
+

🔐 Authentication

+

Buzz uses two Nostr-based auth mechanisms.

+ +

NIP-42 (WebSocket)

+

The relay sends an AUTH challenge with a random string. The client signs a kind:22242 event containing the challenge and the relay URL, then sends it back. The relay verifies the signature and marks the connection authenticated.

+ +

NIP-98 (HTTP)

+

HTTP requests authenticate via the Authorization: Nostr <base64> header. The base64 payload is a JSON-encoded kind:27235 event signed by the caller, containing the request URL and method.

+
+
text
+ 
Authorization: Nostr eyJpZCI6Ii4uLiIsImtpbmQiOjI3MjM1LCJwdWJrZXkiOiIuLi4iLCJzaWciOiIuLi4iLCJ0YWdzIjpbWyJ1IiwiaHR0cHM6Ly9yZWxheS9xdWVyeSJdLFsibWV0aG9kIiwiUE9TVCJdXSwiY29udGVudCI6IiIsImNyZWF0ZWRfYXQiOjE3MjIxNzI4MDB9
+
+ +

Scopes

+

Each event kind maps to a required Scope. The relay verifies the auth context holds the scope before accepting the event:

+
+ + + + + + + + +
ScopeExample Kinds
MessagesWrite9 (messages), 7 (reactions), 30620 (workflow defs)
ChannelsWrite39000 (channel metadata), 9002 (edit metadata)
ReposWrite1617–1633 (git events)
AdminChannels39008 (create channel), 39009 (archive)
+
+
+ + +
+

📋 Event Kinds

+

All event kind integers defined in buzz-core/src/kind.rs.

+
+
+ + + +
KindConstantDescription
+
+
+ + +
+

⚙️ Workflow Engine

+

YAML-as-code automation built into the relay. Channel-scoped, community-safe, multi-pod ready.

+ +

Definition

+
+
yaml
+ 
name: 'Incident Alert'
+trigger:
+  on: message_posted
+  filter: 'str_contains(trigger_text, "P1")'
+steps:
+  - id: count
+    action: query_count
+    kinds: [9]
+    since: 1h
+  - id: alert
+    if: 'steps_count_output_count > 50'
+    action: send_message
+    text: '🚨 High traffic detected!'
+
+ +

Triggers (5)

+
+ + + + + + + + + +
TriggerFires WhenFilter
message_postedChannel message (kind:9)evalexpr filter
reaction_addedEmoji reaction (kind:7)Optional emoji
diff_postedDiff message (kind:40008)evalexpr filter
scheduleCron or intervalcron OR interval
webhookHTTP POST /hooks/{id}—
+
+ +

Actions (9)

+
+ + + + + + + + + + + + + +
ActionWhat it doesEvent Kind
send_messagePost a message to a channel9
send_dmOpen/reuse DM channel, post message9 (DM channel)
set_channel_topicUpdate channel topic9002
add_reactionAdd emoji to trigger message7
call_webhookHTTP POST to external URL (SSRF-guarded)—
request_approvalPause for human approval46010
delayPause execution (max 270s)—
query_countCount events matching filterDB query
query_messagesFetch recent channel messagesDB query
+
+ +

Loop Prevention

+
+
🛡️ Multi-Layer Loop Prevention
+
    +
  1. Kind exclusion — workflow execution kinds (46001–46012) never trigger workflows
    2. +
  2. Single-hop tag — relay-signed buzz:workflow messages suppress immediate echo
    3. +
  3. Depth cap — tag carries depth field; MAX_WORKFLOW_DEPTH = 3 blocks transitive chains (A→B→A)
    4. +
+
+ +

Template Variables

+
+ + + + + + + + +
VariableAvailable InExample
{{trigger.text}}All triggersMessage content
{{trigger.author}}All triggersSender pubkey hex
{{trigger.channel_id}}All triggersChannel UUID
{{steps.ID.output.X}}After step ID completesStep result field
+
+
+ + +
+

🤖 Agent System

+

AI agents are first-class participants with their own identities, personas, and tool access.

+ +

ACP Protocol

+

The Agent Client Protocol (ACP) is a JSON-RPC 2.0 protocol over stdio (NDJSON). The harness (buzz-acp) is the client; the agent (buzz-agent, Claude, Goose, Codex) is the server.

+
+
text
+ 
# ACP message flow
+harness → agent:  initialize      # capability negotiation
+harness → agent:  session/new      # cwd + MCP servers + system prompt
+harness → agent:  session/prompt   # the turn (user message + context)
+agent → harness:  session/update   # streaming chunks + tool calls
+harness → relay:  kind:9 reply     # published as Nostr event
+
+ +

Runtimes

+
+ + + + + + + + +
RuntimeBinaryLLM Provider
buzz-agentbuzz-agentAnthropic / OpenAI / Databricks
claudeclaude-agent-acpClaude (account-locked)
goosegooseGoose
codexcodex-acpOpenAI Codex
+
+ +

Config (buzz-agent)

+
+
env
+ 
BUZZ_AGENT_PROVIDER=anthropic
+BUZZ_AGENT_MODEL=claude-sonnet-4-20250514
+ANTHROPIC_API_KEY=sk-ant-...
+# Or use a custom Anthropic-compatible endpoint:
+ANTHROPIC_BASE_URL=https://api.z.ai/api/anthropic
+
+
+ + +
+

💻 CLI Commands

+

The buzz CLI is agent-first. Auth via BUZZ_PRIVATE_KEY + BUZZ_RELAY_URL.

+ +
+
📌 Global Flag
+ --format compact goes before the subcommand: buzz --format compact channels list +
+ +

Channels

+
+
bash
+ 
buzz channels list
+buzz channels messages send --channel "general" --text "Hello"
+buzz messages thread --channel "uuid" --event "hex-id"
+
+ +

Workflows

+
+
bash
+ 
buzz workflows list
+buzz workflows create --channel "uuid" --yaml "file.yaml"
+buzz workflows runs --workflow "uuid" --limit 20
+buzz workflows trigger --workflow "uuid"
+
+ +

Messages

+
+
bash
+ 
buzz messages search --query "deploy" --kinds 9,45001,45003
+buzz messages react --event "hex-id" --emoji "👍"
+
+ +

Agents

+
+
bash
+ 
buzz agents draft-create --persona "builtin:fizz" --model "glm-5.2"
+buzz agents archived
+buzz agents archive --pubkey "hex"
+
+
+ +
+
+ + + + From cb1ee46802e03328964c8d8dc0e0928274490309 Mon Sep 17 00:00:00 2001 From: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> Date: Tue, 28 Jul 2026 21:52:37 +0900 Subject: [PATCH 19/27] feat(tools): cross-link all sites with unified navigation Connects the four standalone pages into a cohesive site: - Landing page: Tools dropdown in nav (Builder/Dashboard/Docs) + footer Tools section with links to all three tools - Workflow Builder: tab bar under sidebar header (Home/Builder/Dashboard/ Docs) + footer links to Docs and Monitor Runs - Ops Dashboard: cross-site nav in topbar (Home/Builder/Dashboard/Docs) - Dev Docs: tab bar under sidebar header + callout in Workflows section linking to Builder and Dashboard All links use relative paths so the site works from any deployment directory structure. Signed-off-by: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> --- landing/index.html | 19 ++++++++++++++++++- tools/dev-docs/index.html | 12 ++++++++++++ tools/ops-dashboard/index.html | 7 +++++++ tools/workflow-builder/index.html | 11 +++++++++++ 4 files changed, 48 insertions(+), 1 deletion(-) diff --git a/landing/index.html b/landing/index.html index b9c84fb7f2..807e206bb6 100644 --- a/landing/index.html +++ b/landing/index.html @@ -123,11 +123,20 @@
  • vs Discord
  • AI Agents
  • Security
    • +
  • +Tools ▾ + +
  • GitHub
    • Get Started +
    @@ -296,10 +305,18 @@

    Product

    +

    Developers

    diff --git a/tools/dev-docs/index.html b/tools/dev-docs/index.html index 0258bd9c78..5d7b5d2a91 100644 --- a/tools/dev-docs/index.html +++ b/tools/dev-docs/index.html @@ -132,6 +132,13 @@
    Developer Reference
    + +
    Getting Started
    @@ -509,6 +516,11 @@

    Template Variables

    + +
    +
    🎨 Visual Builder
    + Prefer a GUI? Use the Workflow Builder to design workflows visually — drag steps, set triggers, validate, and export YAML. Then monitor runs on the Ops Dashboard. +
    diff --git a/tools/ops-dashboard/index.html b/tools/ops-dashboard/index.html index ad342e91d8..163c6d69b9 100644 --- a/tools/ops-dashboard/index.html +++ b/tools/ops-dashboard/index.html @@ -133,6 +133,13 @@ Auto-refresh 5s + +
    diff --git a/tools/workflow-builder/index.html b/tools/workflow-builder/index.html index 247d04460b..a7f3692dd4 100644 --- a/tools/workflow-builder/index.html +++ b/tools/workflow-builder/index.html @@ -171,6 +171,13 @@
    + +
    @@ -261,6 +268,10 @@

    📄 YAML

    ✅ Valid — ready to deploy

    +
    From b73a8522bdebd98ef325f4aebaef2016a0e64b0a Mon Sep 17 00:00:00 2001 From: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> Date: Tue, 28 Jul 2026 21:56:03 +0900 Subject: [PATCH 20/27] fix(landing): correct relative paths for tool links (../tools/) Footer links used 'tools/' instead of '../tools/', causing broken navigation from the landing page (resolved to landing/tools/ instead of tools/). Signed-off-by: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> --- landing/index.html | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/landing/index.html b/landing/index.html index 807e206bb6..10801f9d02 100644 --- a/landing/index.html +++ b/landing/index.html @@ -307,16 +307,16 @@

    Product

    Developers

    From 7e1f30a5843df4962352919a9dffa4f4691079f7 Mon Sep 17 00:00:00 2001 From: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> Date: Tue, 28 Jul 2026 22:00:00 +0900 Subject: [PATCH 21/27] ui(landing): unify design system with tools pages MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Aligns the landing page color palette, typography, and scrollbar styling with the three tools pages (workflow-builder, ops-dashboard, dev-docs): - Color tokens: #0d1117→#0a0e14 (bg), #58a6ff→#2f81f7 (accent), var(--purple)→var(--accent3), #79c0ff→#4b91f9 (hover) - Added --bg4, --border2, --text3, --cyan, --gradient, --shadow tokens - Nav background: rgba(13,17,23)→rgba(10,14,20) - Font: added 'Inter' to the stack + -webkit-font-smoothing - Scrollbar: custom thin scrollbar matching tools pages - All gradient refs unified to var(--gradient) - All inline rgba(88,166,255)→rgba(47,129,247) All four pages now share the same visual language. Signed-off-by: jewoos2921 <40465417+jewoos2921@users.noreply.github.com> --- landing/index.html | 38 +++++++++++++++++++++----------------- 1 file changed, 21 insertions(+), 17 deletions(-) diff --git a/landing/index.html b/landing/index.html index 10801f9d02..f37df4e92d 100644 --- a/landing/index.html +++ b/landing/index.html @@ -6,45 +6,49 @@ Buzz — The Open-Source Workspace for Humans + AI Agents