feat(workflow): complete all 7 actions + depth-cap loop prevention + monitoring + query actions

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","<depth>"]`); 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.

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","<depth>"]`) 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)
 

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","<depth>"]`); 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))

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.

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)?),

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> = 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<Utc> = now
+            .with_nanosecond(now.nanosecond() / 1000 * 1000)
+            .unwrap_or(now);
 
         let mut audit_entry = AuditEntry {
             community_id,

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

crates/buzz-cli/src/commands/workflows.rs

@@ -57,39 +57,22 @@ 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=<uuid>`.
 ///
-/// 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,
     limit: Option<u32>,
 ) -> 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::Value> = serde_json::from_str(&resp).unwrap_or_default();
-    let normalized: Vec<serde_json::Value> = 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::Value> = serde_json::from_str(&resp).unwrap_or_default();
+    let output = serde_json::to_string(&runs).unwrap_or_default();
     println!("{output}");
     Ok(())
 }

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));
+    }
 }

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<chrono::Utc>,
+    ) -> Result<Vec<(CommunityId, uuid::Uuid)>> {
+        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,

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<Utc>,
+) -> Result<Vec<(CommunityId, Uuid)>> {
+    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