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)) diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index b553adaabb..4abd8de2b3 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -423,6 +423,31 @@ impl AcpClient { // Callers MUST still call shutdown().await for guaranteed cleanup. .kill_on_drop(true); + // Scrub harness secrets from the child environment before any persona + // env vars are injected. Without this, the agent subprocess (and every + // MCP server / tool process it in turn spawns) inherits the harness's + // BUZZ_PRIVATE_KEY / BUZZ_API_TOKEN, enabling a compromised agent to + // sign Nostr events as the agent identity independently of the harness. + // The intentional key handoff to MCP servers happens via the + // session/new JSON-RPC params (build_mcp_servers), NOT via env vars. + // + // Only remove keys that actually exist in the parent environment. This + // avoids triggering a Windows-specific Command behavior where calling + // env_remove on a non-existent key still flips the Command into + // "explicit-env" mode, breaking child processes that depend on the + // full inherited environment. + for secret_key in &[ + "BUZZ_PRIVATE_KEY", + "BUZZ_API_TOKEN", + "BUZZ_ACP_PRIVATE_KEY", + "BUZZ_ACP_API_TOKEN", + "NOSTR_PRIVATE_KEY", + ] { + if std::env::var_os(secret_key).is_some() { + cmd.env_remove(secret_key); + } + } + // Per-persona env vars (e.g., GOOSE_PROVIDER, BUZZ_AGENT_PROVIDER). // For most keys, operator precedence wins: skip injection if already set // in the parent environment. diff --git a/crates/buzz-agent/src/config.rs b/crates/buzz-agent/src/config.rs index f06849c7d1..e7baeb6348 100644 --- a/crates/buzz-agent/src/config.rs +++ b/crates/buzz-agent/src/config.rs @@ -798,7 +798,7 @@ impl Config { base_url, anthropic_api_version: env_or("ANTHROPIC_API_VERSION", "2023-06-01"), openai_api, - max_rounds: parse_env("BUZZ_AGENT_MAX_ROUNDS", 0)?, + max_rounds: parse_env("BUZZ_AGENT_MAX_ROUNDS", 50)?, max_output_tokens: parse_env("BUZZ_AGENT_MAX_OUTPUT_TOKENS", 32_768)?, llm_timeout: Duration::from_secs(parse_env("BUZZ_AGENT_LLM_TIMEOUT_SECS", 240)?), tool_timeout: Duration::from_secs(parse_env("BUZZ_AGENT_TOOL_TIMEOUT_SECS", 660)?), diff --git a/crates/buzz-audit/src/service.rs b/crates/buzz-audit/src/service.rs index 913131a591..7b27157ab6 100644 --- a/crates/buzz-audit/src/service.rs +++ b/crates/buzz-audit/src/service.rs @@ -1,4 +1,4 @@ -use chrono::{DateTime, Utc}; +use chrono::{DateTime, Timelike, Utc}; use futures_util::FutureExt as _; use sqlx::{Acquire, PgPool, Row}; use tracing::{debug, instrument, warn}; @@ -100,7 +100,15 @@ impl AuditService { }; let seq = prev_seq + 1; - let created_at: DateTime = Utc::now(); + // Truncate timestamp to microsecond precision before hashing. Postgres + // TIMESTAMPTZ stores only microseconds, so hashing at nanosecond + // precision would cause verify_chain to false-alarm on every entry + // written with sub-microsecond created_at (common on Windows where + // GetSystemTimePreciseAsFileTime has ~100ns resolution). + let now = Utc::now(); + let created_at: DateTime = now + .with_nanosecond(now.nanosecond() / 1000 * 1000) + .unwrap_or(now); let mut audit_entry = AuditEntry { community_id, diff --git a/crates/buzz-cli/src/commands/moderation.rs b/crates/buzz-cli/src/commands/moderation.rs index c53aecaf85..709177684d 100644 --- a/crates/buzz-cli/src/commands/moderation.rs +++ b/crates/buzz-cli/src/commands/moderation.rs @@ -109,6 +109,13 @@ async fn cmd_reports( ) -> Result<(), CliError> { let mut path = format!("/moderation/reports?limit={limit}"); if let Some(s) = status { + // Validate status is a simple alphanumeric identifier to prevent + // query-parameter injection (e.g. --status "open&limit=99999"). + if !s.chars().all(|c| c.is_ascii_alphanumeric()) { + return Err(CliError::Usage(format!( + "invalid status '{s}': must be alphanumeric (e.g. open, resolved, dismissed, escalated)" + ))); + } path.push_str(&format!("&status={s}")); } let resp = client.get_authed(&path).await?; 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-core/src/kind.rs b/crates/buzz-core/src/kind.rs index b912169801..56e686dd62 100644 --- a/crates/buzz-core/src/kind.rs +++ b/crates/buzz-core/src/kind.rs @@ -679,6 +679,11 @@ pub const fn is_command_kind(kind: u32) -> bool { /// Returns `true` if `kind` may only be authored by the relay. /// Client submission of these kinds must be rejected. +/// +/// Includes: +/// - Relay-computed overlays (membership lists, summaries, snapshots) +/// - Relay-signed notifications (member added/removed, identity archive deltas) +/// - Audit entries (relay-only hash-chain log) pub const fn is_relay_only_kind(kind: u32) -> bool { matches!( kind, @@ -688,6 +693,11 @@ pub const fn is_relay_only_kind(kind: u32) -> bool { | KIND_DM_VISIBILITY | KIND_THREAD_SUMMARY | KIND_WINDOW_BOUNDS + | KIND_NIP43_MEMBER_ADDED + | KIND_NIP43_MEMBER_REMOVED + | KIND_IA_ARCHIVED + | KIND_IA_UNARCHIVED + | KIND_AUDIT_ENTRY ) } @@ -781,4 +791,29 @@ mod tests { ); } } + + #[test] + fn relay_only_kinds_include_notifications_and_audit() { + // Overlays and snapshots + assert!(is_relay_only_kind(KIND_NIP43_MEMBERSHIP_LIST)); + assert!(is_relay_only_kind(KIND_CHANNEL_SUMMARY)); + assert!(is_relay_only_kind(KIND_PRESENCE_SNAPSHOT)); + assert!(is_relay_only_kind(KIND_DM_VISIBILITY)); + assert!(is_relay_only_kind(KIND_THREAD_SUMMARY)); + assert!(is_relay_only_kind(KIND_WINDOW_BOUNDS)); + // Relay-signed notifications + assert!(is_relay_only_kind(KIND_NIP43_MEMBER_ADDED)); + assert!(is_relay_only_kind(KIND_NIP43_MEMBER_REMOVED)); + assert!(is_relay_only_kind(KIND_IA_ARCHIVED)); + assert!(is_relay_only_kind(KIND_IA_UNARCHIVED)); + // Audit + assert!(is_relay_only_kind(KIND_AUDIT_ENTRY)); + } + + #[test] + fn client_kinds_are_not_relay_only() { + assert!(!is_relay_only_kind(KIND_STREAM_MESSAGE)); + assert!(!is_relay_only_kind(KIND_REACTION)); + assert!(!is_relay_only_kind(KIND_PROFILE)); + } } 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/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)) diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 97c31c2561..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 { @@ -176,37 +205,14 @@ 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(); 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() { @@ -254,15 +260,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}")))?, ]; @@ -362,6 +372,339 @@ 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, tenant) = self.resolve_state_and_tenant(community_id).await?; + + 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) + }) + } + + 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, tenant) = self.resolve_state_and_tenant(community_id).await?; + + 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) 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 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([e_tag, p_tag, wf_tag]) + .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) + }) + } + + 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, tenant) = self.resolve_state_and_tenant(community_id).await?; + + 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}")) + })?; + + // 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]; + 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) + }) + } + + 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, 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 + // (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)] @@ -676,6 +1019,7 @@ mod integration_tests { &channel.id.to_string(), "heads up @Robby — please take a look", &author_hex, + 0, ) .await .expect("send_message"); 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 0c6002e74e..04a1491c71 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,98 @@ pub trait ActionSink: Send + Sync { channel_id: &str, text: &str, 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 + '_>>; + + /// 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 + '_>>; + + /// 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 + '_>>; + + /// 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 a029b44622..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; @@ -39,6 +40,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 { @@ -448,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, + }), } } @@ -518,6 +539,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, @@ -567,7 +589,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)?; @@ -577,16 +605,111 @@ 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: _ } => { - 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 } => { @@ -597,23 +720,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 { @@ -658,10 +802,101 @@ 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. + // 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, + &token_hash_hex, + message, + from, + &owner_pubkey_hex, + ) + .await + { + // 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, @@ -686,9 +921,157 @@ 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. /// /// Uses `Uuid::new_v4()` which draws from the OS CSPRNG (via the `getrandom` @@ -865,70 +1248,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: @@ -1137,6 +1456,7 @@ async fn execute_steps( std::time::Duration::from_secs(timeout_secs), dispatch_action( &step.id, + i, &resolved_action, engine, community_id, @@ -1226,7 +1546,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..290044907e 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 { @@ -189,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 { @@ -315,6 +326,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) => { @@ -433,6 +463,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) => { @@ -670,6 +709,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 @@ -885,6 +975,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 +1064,7 @@ pub fn build_trigger_context(event: &buzz_core::StoredEvent) -> executor::Trigge emoji, message_id, webhook_fields: HashMap::new(), + workflow_depth, } } @@ -1537,6 +1654,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 diff --git a/crates/buzz-workflow/src/schema.rs b/crates/buzz-workflow/src/schema.rs index 34e8bb1960..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 { @@ -190,6 +212,9 @@ impl WorkflowDef { 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 { @@ -335,17 +360,21 @@ 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", + " - 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(), 7); @@ -354,24 +383,24 @@ mod tests { &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 { .. })); + assert!(matches!(&def.steps[5].action, ActionDef::QueryCount { .. })); + assert!(matches!( + &def.steps[6].action, + ActionDef::QueryMessages { .. } + )); } #[test] @@ -391,6 +420,43 @@ mod tests { assert_eq!(def.steps.len(), 3); } + #[test] + 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 (def, _) = parse_yaml(yaml).expect("send_dm should validate"); + assert!(matches!(def.steps[0].action, ActionDef::SendDm { .. })); + } + + #[test] + 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 (def, _) = parse_yaml(yaml).expect("set_channel_topic should validate"); + assert!(matches!( + def.steps[0].action, + ActionDef::SetChannelTopic { .. } + )); + } + + #[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 = diff --git a/desktop/src-tauri/src/commands/agent_auth.rs b/desktop/src-tauri/src/commands/agent_auth.rs index c23263de7e..0ea416d906 100644 --- a/desktop/src-tauri/src/commands/agent_auth.rs +++ b/desktop/src-tauri/src/commands/agent_auth.rs @@ -261,14 +261,44 @@ fn adapter_terminal_argv( fallback_command: &str, ) -> Result, String> { let meta_command = terminal_auth_meta_command(method)?; - let (command, args): (&str, &[String]) = - match meta_command.as_deref().and_then(|argv| argv.split_first()) { - Some((command, args)) => (command.as_str(), args), - None => match method.command.split_first() { - Some((command, args)) => (command.as_str(), args), - None => (fallback_command, method.args.as_slice()), - }, - }; + let (command, args): (&str, &[String]) = match meta_command + .as_deref() + .and_then(|argv| argv.split_first()) + { + Some((command, args)) => { + // Security: the _meta.terminal-auth.command field is supplied by + // the ACP adapter at runtime. A malicious or typosquatted adapter + // could declare an arbitrary binary here (e.g. /tmp/evil). Reject + // any command that doesn't resolve to the adapter's own installed + // path (fallback_command), preventing arbitrary code execution. + let resolved = resolve_command(command) + .map(|p| p.display().to_string()) + .unwrap_or_default(); + if resolved != fallback_command && command != fallback_command { + return Err(format!( + "{} terminal-auth command '{command}' does not match the adapter's installed path — refusing to execute", + runtime_label + )); + } + (fallback_command, args) + } + None => match method.command.split_first() { + Some((command, args)) => { + // Same check for method.command. + let resolved = resolve_command(command) + .map(|p| p.display().to_string()) + .unwrap_or_default(); + if resolved != fallback_command && command != fallback_command { + return Err(format!( + "{} auth method command '{command}' does not match the adapter's installed path — refusing to execute", + runtime_label + )); + } + (fallback_command, args) + } + None => (fallback_command, method.args.as_slice()), + }, + }; if command.trim().is_empty() { return Err(format!( diff --git a/desktop/src-tauri/tauri.conf.json b/desktop/src-tauri/tauri.conf.json index b75e0f5203..06262242a8 100644 --- a/desktop/src-tauri/tauri.conf.json +++ b/desktop/src-tauri/tauri.conf.json @@ -36,7 +36,7 @@ ], "macOSPrivateApi": true, "security": { - "csp": null + "csp": "default-src 'self' 'unsafe-inline' ipc: http://ipc.localhost; img-src 'self' data: blob: http://127.0.0.1:* https:; media-src 'self' blob: http://127.0.0.1:*; connect-src 'self' ipc: http://ipc.localhost ws://localhost:* wss://localhost:* http://localhost:* https://* ws://* wss://*; font-src 'self' data:; style-src 'self' 'unsafe-inline'" } }, "plugins": { diff --git a/landing/index.html b/landing/index.html new file mode 100644 index 0000000000..b396b30370 --- /dev/null +++ b/landing/index.html @@ -0,0 +1,363 @@ + + + + + +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

+
+
+
+ + + + + + + diff --git a/tools/dev-docs/index.html b/tools/dev-docs/index.html new file mode 100644 index 0000000000..8a5db785b0 --- /dev/null +++ b/tools/dev-docs/index.html @@ -0,0 +1,703 @@ + + + + + +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
+
+ +
+
🎨 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. +
+
+ + +
+

🤖 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"
+
+
+ +
+
+ + + + diff --git a/tools/ops-dashboard/index.html b/tools/ops-dashboard/index.html new file mode 100644 index 0000000000..1fc0395b6f --- /dev/null +++ b/tools/ops-dashboard/index.html @@ -0,0 +1,429 @@ + + + + + +Buzz Ops Dashboard + + + + + +
+ +
+ + +
+ + + + +
+ +
+ +
+
📊
+

Connect to a relay

+

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

+
+ + +
+ + +
+ + + + diff --git a/tools/workflow-builder/index.html b/tools/workflow-builder/index.html new file mode 100644 index 0000000000..86f40e3ad9 --- /dev/null +++ b/tools/workflow-builder/index.html @@ -0,0 +1,560 @@ + + + + + +Buzz Workflow Builder + + + + + +
+
+ +
+ + +
+
+ + + +
+ +
+

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 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

+
+ + +
+
+
+
+

✅ Valid — ready to deploy

+
+ +
+ + +
+ + + +