feat(acp): deliver durable workflow messages

Drive the existing ACP lifecycle through the fenced delivery API and verify
only signed public authority artifacts before dispatch. Keep wake events
identifier-only and preserve the signed workflow owner as the admission
principal.

Co-authored-by: Larry <larry@buzz.local>
Signed-off-by: Logan Johnson <loganj@squareup.com>

crates/buzz-acp/src/config.rs

@@ -1277,7 +1277,8 @@ pub fn resolve_channel_filters(
     rules: &[SubscriptionRule],
 ) -> HashMap<Uuid, ChannelFilter> {
     use buzz_core::kind::{
-        KIND_STREAM_MESSAGE, KIND_STREAM_REMINDER, KIND_WORKFLOW_APPROVAL_REQUESTED,
+        KIND_STREAM_MESSAGE, KIND_STREAM_REMINDER, KIND_WORKFLOW_AGENT_WAKE,
+        KIND_WORKFLOW_APPROVAL_REQUESTED,
     };
 
     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_AGENT_WAKE,
                     KIND_WORKFLOW_APPROVAL_REQUESTED,
                     KIND_STREAM_REMINDER,
                 ]
@@ -1379,7 +1381,8 @@ pub fn resolve_dynamic_channel_filter(
     rules: &[crate::filter::SubscriptionRule],
 ) -> Option<ChannelFilter> {
     use buzz_core::kind::{
-        KIND_STREAM_MESSAGE, KIND_STREAM_REMINDER, KIND_WORKFLOW_APPROVAL_REQUESTED,
+        KIND_STREAM_MESSAGE, KIND_STREAM_REMINDER, KIND_WORKFLOW_AGENT_WAKE,
+        KIND_WORKFLOW_APPROVAL_REQUESTED,
     };
 
     // 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_AGENT_WAKE,
                     KIND_WORKFLOW_APPROVAL_REQUESTED,
                     KIND_STREAM_REMINDER,
                 ]

crates/buzz-acp/src/pool.rs

@@ -828,7 +828,7 @@ impl AgentPool {
         &mut self,
         channel_id: Uuid,
         request: SteerRequest,
-    ) -> Result<(), SteerError> {
+    ) -> Result<String, SteerError> {
         let meta = self
             .task_map
             .values_mut()
@@ -839,7 +839,8 @@ impl AgentPool {
             .as_ref()
             .ok_or_else(|| SteerError::Transport("steer_tx not installed".into()))?;
         tx.try_send(request)
-            .map_err(|e| SteerError::Transport(e.to_string()))
+            .map_err(|e| SteerError::Transport(e.to_string()))?;
+        Ok(meta.turn_id.clone())
     }
 
     /// Durably associate a successful steer with the exact ACP session that

crates/buzz-acp/src/queue.rs

@@ -658,16 +658,19 @@ impl EventQueue {
     /// Returns the event IDs of dropped events so the caller can clean up
     /// any reactions (👀) that were added at queue-push time.
     pub fn drain_channel(&mut self, channel_id: Uuid) -> Vec<String> {
-        let ids = self
-            .queues
-            .remove(&channel_id)
-            .map(|q| q.into_iter().map(|e| e.event.id.to_hex()).collect())
-            .unwrap_or_default();
+        let mut ids = Vec::new();
+        if let Some(events) = self.queues.remove(&channel_id) {
+            ids.extend(events.into_iter().map(|event| event.event.id.to_hex()));
+        }
+        if let Some(events) = self.cancelled_batches.remove(&channel_id) {
+            ids.extend(events.into_iter().map(|event| event.event.id.to_hex()));
+        }
+        if let Some(events) = self.withheld_native_steer.remove(&channel_id) {
+            ids.extend(events.into_iter().map(|event| event.event.id.to_hex()));
+        }
         self.retry_after.remove(&channel_id);
         self.retry_counts.remove(&channel_id);
-        self.cancelled_batches.remove(&channel_id);
         self.cancel_reasons.remove(&channel_id);
-        self.withheld_native_steer.remove(&channel_id);
         // Preserve in_flight_channels AND in_flight_deadlines: the in-flight
         // task will eventually complete (calling mark_complete) or the deadline
         // will expire (auto-cleaning the channel). Removing deadlines without
@@ -769,6 +772,31 @@ impl EventQueue {
         }
     }
 
+    /// Whether an event remains under local queue ownership, including a
+    /// cancel-merge batch or a pending native steer.
+    pub fn contains_event(&self, channel_id: Uuid, event_id: &str) -> bool {
+        self.queues.get(&channel_id).is_some_and(|events| {
+            events
+                .iter()
+                .any(|event| event.event.id.to_hex() == event_id)
+        }) || self
+            .cancelled_batches
+            .get(&channel_id)
+            .is_some_and(|events| {
+                events
+                    .iter()
+                    .any(|event| event.event.id.to_hex() == event_id)
+            })
+            || self
+                .withheld_native_steer
+                .get(&channel_id)
+                .is_some_and(|events| {
+                    events
+                        .iter()
+                        .any(|event| event.event.id.to_hex() == event_id)
+                })
+    }
+
     /// Drop a specific event by id from both the side table and the main
     /// queue.
     ///
@@ -4048,6 +4076,29 @@ mod tests {
         assert_eq!(pending_count(&q), 0);
     }
 
+    #[test]
+    fn test_drain_channel_reports_cancelled_and_withheld_events() {
+        let mut q = EventQueue::new(DedupMode::Queue);
+        let ch = Uuid::new_v4();
+        let cancelled = make_queued(ch, "cancelled");
+        let withheld = make_queued(ch, "withheld");
+        let cancelled_id = cancelled.event.id.to_hex();
+        let withheld_id = withheld.event.id.to_hex();
+
+        q.push(cancelled);
+        let batch = q.flush_next().expect("cancelled batch");
+        q.requeue_as_cancelled(batch, CancelReason::Steer);
+        q.mark_complete(ch);
+        q.push(withheld);
+        assert!(q.mark_native_steer_pending(ch, &withheld_id));
+
+        let drained = q.drain_channel(ch);
+        assert!(drained.contains(&cancelled_id));
+        assert!(drained.contains(&withheld_id));
+        assert!(!q.contains_event(ch, &cancelled_id));
+        assert!(!q.contains_event(ch, &withheld_id));
+    }
+
     #[test]
     fn test_drain_channel_empty_returns_empty() {
         let mut q = EventQueue::new(DedupMode::Queue);

crates/buzz-acp/src/relay.rs

@@ -494,6 +494,17 @@ impl RestClient {
             .map_err(|e| RelayError::Http(e.to_string()))
     }
 
+    /// POST an authenticated JSON document to an arbitrary relay API path.
+    pub async fn post_json(&self, path: &str, value: &Value) -> Result<Value, RelayError> {
+        let body = serde_json::to_vec(value)
+            .map_err(|error| RelayError::Http(format!("JSON serialize error: {error}")))?;
+        let response = self.bridge_post(path, &body).await?;
+        response
+            .json()
+            .await
+            .map_err(|error| RelayError::Http(error.to_string()))
+    }
+
     /// Submit a signed event via the HTTP bridge: `POST /events` with NIP-98 auth.
     ///
     /// The event must already be signed. Returns the relay response JSON.
@@ -635,6 +646,9 @@ pub struct HarnessRelay {
     keys: Keys,
     /// Optional NIP-OA auth tag for relay membership delegation.
     auth_tag: Option<nostr::Tag>,
+    /// Relay signing identity advertised by NIP-11 `self`. Relay-authored
+    /// workflow messages are trusted only when their signature matches this key.
+    relay_self: Option<String>,
     /// Handle to the background task (for clean shutdown).
     /// Wrapped in `Option` so `shutdown()` can take ownership without conflicting
     /// with `Drop` (which only has `&mut self`).
@@ -688,6 +702,11 @@ impl HarnessRelay {
         agent_pubkey_hex: &str,
         auth_tag: Option<nostr::Tag>,
     ) -> Result<Self, RelayError> {
+        let http = reqwest::Client::builder()
+            .timeout(std::time::Duration::from_secs(10))
+            .connect_timeout(std::time::Duration::from_secs(5))
+            .build()
+            .map_err(|e| RelayError::Http(format!("failed to build HTTP client: {e}")))?;
         // Perform the initial connection and auth handshake, retrying
         // transient failures (dropped handshake, timeout) with bounded
         // jittered backoff. A terminal error (bad URL, bad auth tag,
@@ -696,6 +715,18 @@ impl HarnessRelay {
         let (ws, handshake_buffer) =
             retry_initial_connect(|| do_connect(relay_url, keys, auth_tag.as_ref())).await?;
 
+        // NIP-11 is the authority binding between this configured endpoint and
+        // its relay signing key. Fetch it after the successful connection so a
+        // transient startup outage cannot permanently disable workflow handling.
+        // Missing, malformed, or unreachable info still fails closed for workflow
+        // delegation without preventing ordinary ACP use.
+        let relay_self = fetch_relay_self(&http, relay_url).await;
+        if relay_self.is_none() {
+            warn!(
+                "could not fetch relay NIP-11 identity after connecting; workflow delegation is disabled until restart"
+            );
+        }
+
         let (event_tx, event_rx) = mpsc::channel::<Option<BuzzEvent>>(event_channel_capacity());
         let (observer_control_tx, observer_control_rx) =
             mpsc::channel::<Event>(event_channel_capacity());
@@ -725,14 +756,11 @@ impl HarnessRelay {
             event_rx,
             observer_control_rx: Some(observer_control_rx),
             cmd_tx,
-            http: reqwest::Client::builder()
-                .timeout(std::time::Duration::from_secs(10))
-                .connect_timeout(std::time::Duration::from_secs(5))
-                .build()
-                .map_err(|e| RelayError::Http(format!("failed to build HTTP client: {e}")))?,
+            http,
             relay_url: relay_url.to_string(),
             keys: keys.clone(),
             auth_tag,
+            relay_self,
             bg_handle: Some(bg_handle),
         })
     }
@@ -801,6 +829,11 @@ impl HarnessRelay {
         Ok(map)
     }
 
+    /// Return the configured endpoint's NIP-11 relay signing identity.
+    pub fn relay_self(&self) -> Option<&str> {
+        self.relay_self.as_deref()
+    }
+
     /// Build a [`RestClient`] that shares this relay's HTTP credentials.
     ///
     /// The returned client is cheap to clone (wraps `reqwest::Client` which is
@@ -3556,6 +3589,64 @@ async fn send_auth_response(
 /// `ws://host:port` → `http://host:port`
 /// `wss://host:port` → `https://host:port`
 /// Trailing slashes are stripped.
+#[derive(serde::Deserialize)]
+struct RelayInformationDocument {
+    #[serde(default, rename = "self")]
+    relay_self: Option<String>,
+}
+
+async fn fetch_relay_self(http: &reqwest::Client, relay_url: &str) -> Option<String> {
+    // Keep NIP-11 retrieval bounded. The initial connection already proved the
+    // endpoint reachable; these retries cover a short-lived HTTP-side outage.
+    for (attempt, delay) in std::iter::once(None)
+        .chain(
+            STARTUP_CONNECT_BACKOFFS
+                .iter()
+                .take(2)
+                .map(|delay| Some(*delay)),
+        )
+        .enumerate()
+    {
+        if let Some(delay) = delay {
+            tokio::time::sleep(jittered_duration(delay)).await;
+        }
+
+        let Some(response) = http
+            .get(relay_ws_to_http(relay_url))
+            .header("Accept", "application/nostr+json")
+            .send()
+            .await
+            .ok()
+        else {
+            warn!("NIP-11 relay identity fetch attempt {attempt} failed");
+            continue;
+        };
+        if !response.status().is_success() {
+            warn!(
+                "NIP-11 relay identity fetch attempt {attempt} returned HTTP {}",
+                response.status()
+            );
+            continue;
+        }
+        let Some(value) = response
+            .json::<RelayInformationDocument>()
+            .await
+            .ok()
+            .and_then(|document| document.relay_self)
+            .map(|value| value.to_ascii_lowercase())
+        else {
+            warn!("NIP-11 relay identity fetch attempt {attempt} returned invalid metadata");
+            continue;
+        };
+        if let Ok(key) = nostr::PublicKey::from_hex(&value) {
+            return Some(key.to_hex());
+        }
+        warn!("NIP-11 relay identity fetch attempt {attempt} returned an invalid public key");
+    }
+
+    None
+}
+
 pub(crate) fn relay_ws_to_http(url: &str) -> String {
     url.replace("wss://", "https://")
         .replace("ws://", "http://")

crates/buzz-acp/src/verifier.rs

@@ -3,8 +3,8 @@
 //! The relay returns the exact signed workflow definition, the visible message,
 //! and a narrowly scoped relay-signed receipt. This module verifies those
 //! public artifacts without receiving private trigger context, webhook fields,
-//! prior-step outputs, execution traces, or secrets. It performs no I/O and has
-//! no production caller until the capability-gated runtime wires it in.
+//! prior-step outputs, execution traces, or secrets. It performs no I/O; the
+//! capability-gated runtime supplies the authenticated API response directly.
 
 use buzz_core::kind::{KIND_STREAM_MESSAGE, KIND_WORKFLOW_DEF};
 use buzz_core::tenant::CommunityId;

crates/buzz-workflow/src/lib.rs

@@ -1120,6 +1120,42 @@ fn owner_authority_allows(role: Option<&str>, needs_elevated: bool) -> bool {
     }
 }
 
+/// Validate a relay-provided schedule cause against the signed definition.
+///
+/// Freshness/skew is a separate policy; this checks only that the slot is an
+/// exact cron occurrence or interval boundary.
+pub fn schedule_cause_matches(def: &WorkflowDef, slot: DateTime<Utc>) -> bool {
+    match &def.trigger {
+        schema::TriggerDef::Schedule {
+            cron: Some(expr),
+            interval: None,
+        } => {
+            let Ok(schedule) = schema::normalize_cron(expr).parse::<cron::Schedule>() else {
+                return false;
+            };
+            let previous = slot - chrono::Duration::seconds(1);
+            schedule.after(&previous).next() == Some(slot)
+        }
+        schema::TriggerDef::Schedule {
+            cron: None,
+            interval: Some(duration),
+        } => executor::parse_duration_secs(duration)
+            .ok()
+            .is_some_and(|seconds| seconds > 0 && slot.timestamp().rem_euclid(seconds as i64) == 0),
+        _ => false,
+    }
+}
+
+/// Validate that a signed source event semantically satisfies a workflow trigger.
+pub async fn trigger_matches_signed_event(
+    def: &WorkflowDef,
+    trigger_ctx: &executor::TriggerContext,
+    kind_u32: u32,
+) -> bool {
+    trigger_matches_event(&def.trigger, kind_u32)
+        && should_fire_workflow(def, trigger_ctx, Uuid::nil()).await
+}
+
 /// Returns `true` if the trigger type matches the given event kind.
 fn trigger_matches_event(trigger: &TriggerDef, kind_u32: u32) -> bool {
     use buzz_core::kind::{KIND_REACTION, KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_DIFF};