Wake agents from verified workflow mentions

.github/workflows/ci.yml

@@ -707,6 +707,14 @@ jobs:
             --run-ignored ignored-only
         env:
           BUZZ_TEST_DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz
+      - name: Workflow wake lifecycle PostgreSQL tests
+        run: |
+          cargo nextest run \
+            --archive-file target/ci/backend-integration-tests.tar.zst \
+            -E 'package(buzz-relay) and test(/^workflow_sink::/)' \
+            --run-ignored ignored-only
+        env:
+          DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz
       - name: Database pressure observability PostgreSQL tests
         # Explicit pool acquisition and advisory-lock metrics require real
         # Postgres and are ignored by the infrastructure-free unit-test job.

crates/buzz-acp/Cargo.toml

@@ -41,6 +41,7 @@ reqwest = { workspace = true }
 # Serialization
 serde = { workspace = true }
 serde_json = { workspace = true }
+serde_yaml = { workspace = true }
 
 # IDs
 uuid = { workspace = true }

crates/buzz-acp/src/config.rs

@@ -1278,6 +1278,7 @@ pub fn resolve_channel_filters(
 ) -> HashMap<Uuid, ChannelFilter> {
     use buzz_core::kind::{
         KIND_STREAM_MESSAGE, KIND_STREAM_REMINDER, KIND_WORKFLOW_APPROVAL_REQUESTED,
+        KIND_WORKFLOW_MENTION_WAKE,
     };
 
     let target_channels: Vec<Uuid> = if let Some(ref overrides) = config.channels_override {
@@ -1297,6 +1298,7 @@ pub fn resolve_channel_filters(
             let kinds = config.kinds_override.clone().unwrap_or_else(|| {
                 vec![
                     KIND_STREAM_MESSAGE,
+                    KIND_WORKFLOW_MENTION_WAKE,
                     KIND_WORKFLOW_APPROVAL_REQUESTED,
                     KIND_STREAM_REMINDER,
                 ]
@@ -1380,6 +1382,7 @@ pub fn resolve_dynamic_channel_filter(
 ) -> Option<ChannelFilter> {
     use buzz_core::kind::{
         KIND_STREAM_MESSAGE, KIND_STREAM_REMINDER, KIND_WORKFLOW_APPROVAL_REQUESTED,
+        KIND_WORKFLOW_MENTION_WAKE,
     };
 
     // In Mentions/All mode, if the operator explicitly constrained channels
@@ -1402,6 +1405,7 @@ pub fn resolve_dynamic_channel_filter(
             kinds: Some(config.kinds_override.clone().unwrap_or_else(|| {
                 vec![
                     KIND_STREAM_MESSAGE,
+                    KIND_WORKFLOW_MENTION_WAKE,
                     KIND_WORKFLOW_APPROVAL_REQUESTED,
                     KIND_STREAM_REMINDER,
                 ]
@@ -1549,13 +1553,37 @@ mod tests {
         for ch in &channels {
             let f = result.get(ch).expect("channel should be present");
             assert!(f.require_mention, "mentions mode requires mention");
-            let kinds = f.kinds.as_ref().expect("should have kinds");
-            assert!(kinds.contains(&buzz_core::kind::KIND_STREAM_MESSAGE));
-            assert!(kinds.contains(&buzz_core::kind::KIND_WORKFLOW_APPROVAL_REQUESTED));
-            assert!(kinds.contains(&buzz_core::kind::KIND_STREAM_REMINDER));
+            assert_eq!(
+                f.kinds,
+                Some(vec![
+                    buzz_core::kind::KIND_STREAM_MESSAGE,
+                    buzz_core::kind::KIND_WORKFLOW_MENTION_WAKE,
+                    buzz_core::kind::KIND_WORKFLOW_APPROVAL_REQUESTED,
+                    buzz_core::kind::KIND_STREAM_REMINDER,
+                ])
+            );
         }
     }
 
+    #[test]
+    fn test_mentions_mode_dynamic_default_kinds_include_workflow_wake() {
+        let config = test_config(SubscribeMode::Mentions);
+        let channel = Uuid::new_v4();
+        let filter = resolve_dynamic_channel_filter(&config, channel, &[])
+            .expect("dynamic channel should be subscribed");
+
+        assert!(filter.require_mention);
+        assert_eq!(
+            filter.kinds,
+            Some(vec![
+                buzz_core::kind::KIND_STREAM_MESSAGE,
+                buzz_core::kind::KIND_WORKFLOW_MENTION_WAKE,
+                buzz_core::kind::KIND_WORKFLOW_APPROVAL_REQUESTED,
+                buzz_core::kind::KIND_STREAM_REMINDER,
+            ])
+        );
+    }
+
     #[test]
     fn test_mentions_mode_custom_kinds() {
         let mut config = test_config(SubscribeMode::Mentions);

crates/buzz-acp/src/lib.rs

@@ -13,6 +13,7 @@ mod queue;
 mod relay;
 mod setup_mode;
 mod usage;
+mod workflow_wake;
 
 pub use usage::TurnUsage;
 
@@ -2017,6 +2018,12 @@ async fn tokio_main() -> Result<()> {
         tracing::warn!("failed to set startup watermark: {e}");
     }
 
+    let workflow_relay_pubkey = relay
+        .rest_client()
+        .relay_signing_pubkey()
+        .await
+        .map_err(|e| anyhow::anyhow!("relay signing identity error: {e}"))?;
+
     tracing::info!("connected to relay at {}", config.relay_url);
 
     relay
@@ -2108,6 +2115,7 @@ async fn tokio_main() -> Result<()> {
                 kinds: config.kinds_override.clone().unwrap_or_else(|| {
                     vec![
                         KIND_STREAM_MESSAGE,
+                        buzz_core::kind::KIND_WORKFLOW_MENTION_WAKE,
                         KIND_WORKFLOW_APPROVAL_REQUESTED,
                         KIND_STREAM_REMINDER,
                     ]
@@ -2631,6 +2639,79 @@ async fn tokio_main() -> Result<()> {
                     match buzz_event {
                         Some(buzz_event) => {
                             let kind_u32 = buzz_event.event.kind.as_u16() as u32;
+                            if workflow_wake::requires_verified_wake(
+                                &buzz_event.event,
+                                workflow_relay_pubkey,
+                            ) {
+                                continue;
+                            }
+
+                            let (buzz_event, admission_author_override) = if kind_u32
+                                == buzz_core::kind::KIND_WORKFLOW_MENTION_WAKE
+                            {
+                                let Some(wake) = workflow_wake::authenticate(
+                                    &buzz_event.event,
+                                    workflow_relay_pubkey,
+                                ) else {
+                                    tracing::warn!("workflow wake authentication failed");
+                                    continue;
+                                };
+                                let authority = match ctx
+                                    .rest_client
+                                    .workflow_wake_authority(wake.run_id(), &wake.message_event_id())
+                                    .await
+                                {
+                                    Ok(authority) => authority,
+                                    Err(error) if error.is_transient() => {
+                                        // HTTP-status failures exhaust bounded retries; body
+                                        // interruptions also return transient after pacing.
+                                        // Transport dedup recorded this relay-signed wake, but
+                                        // dispatch has not occurred. Re-admit it for filtered
+                                        // replay rather than losing it or bypassing verification.
+                                        if let Err(replay_error) = relay
+                                            .replay_event(
+                                                buzz_event.channel_id,
+                                                buzz_event.event.id.to_hex(),
+                                                buzz_event.event.created_at.as_secs(),
+                                            )
+                                            .await
+                                        {
+                                            tracing::warn!(
+                                                %replay_error,
+                                                "failed to arrange workflow wake authority replay"
+                                            );
+                                        }
+                                        tracing::warn!(%error, "workflow wake authority unavailable; replay queued");
+                                        continue;
+                                    }
+                                    Err(error) => {
+                                        // 403/404 and malformed authority bundles are terminal:
+                                        // replays cannot make a rejected or invalid authority safe.
+                                        tracing::warn!(%error, "workflow wake authority rejected");
+                                        continue;
+                                    }
+                                };
+                                let Some((message, signed_author)) = workflow_wake::verify(
+                                    &buzz_event.event,
+                                    authority,
+                                    workflow_relay_pubkey,
+                                    config.keys.public_key(),
+                                    buzz_event.channel_id,
+                                ) else {
+                                    tracing::warn!("workflow wake authority verification failed");
+                                    continue;
+                                };
+                                (
+                                    relay::BuzzEvent {
+                                        channel_id: buzz_event.channel_id,
+                                        event: message,
+                                    },
+                                    Some(signed_author),
+                                )
+                            } else {
+                                (buzz_event, None)
+                            };
+                            let kind_u32 = buzz_event.event.kind.as_u16() as u32;
 
                             if kind_u32 == KIND_MEMBER_ADDED_NOTIFICATION
                                 || kind_u32 == KIND_MEMBER_REMOVED_NOTIFICATION
@@ -2868,7 +2949,8 @@ async fn tokio_main() -> Result<()> {
                             // explicit pubkey list on top, for external people;
                             // it never revokes same-owner team bots.
                             {
-                                let author = buzz_event.event.pubkey.to_hex();
+                                let author = admission_author_override
+                                    .unwrap_or_else(|| buzz_event.event.pubkey.to_hex());
                                 // DM hardening: resolve channel type (fail-closed
                                 // to DM) so allowlist/anyone modes cannot be
                                 // exercised by non-owner authors inside DMs.