feat(workflows): deliver durable managed-agent messages

Atomically bind relay-visible workflow messages to managed-agent inbox rows, publish recipient-only wake hints, and expose authenticated target-scoped claims.

Co-authored-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz>

Signed-off-by: loganj <loganj@squareup.com>

crates/buzz-core/src/filter.rs

@@ -11,10 +11,9 @@ pub fn filters_match(filters: &[Filter], event: &StoredEvent) -> bool {
     filters.iter().any(|f| filter_match_one(f, event))
 }
 
-/// Result-level read authorization for relay-signed events whose content is
-/// private to a single viewer. Currently gates `KIND_DM_VISIBILITY` and
-/// `KIND_AGENT_TURN_METRIC`: the reader MUST equal the event's `#p` tag
-/// (owner). Returns `true` for every other kind.
+/// Result-level read authorization for events whose kind is recipient-private.
+/// For every [`crate::kind::P_GATED_KINDS`] event, the reader MUST equal one of
+/// the event's `#p` tags. Returns `true` for every other kind.
 ///
 /// This guards every delivery surface — WS historical pull (`req.rs`), HTTP
 /// bridge (`bridge.rs`), and live fan-out (`event.rs`) — so a query that
@@ -22,7 +21,7 @@ pub fn filters_match(filters: &[Filter], event: &StoredEvent) -> bool {
 /// a known event id) still cannot read another user's private event.
 pub fn reader_authorized_for_event(event: &nostr::Event, reader_pubkey_hex: &str) -> bool {
     let kind = crate::kind::event_kind_u32(event);
-    if kind != crate::kind::KIND_DM_VISIBILITY && kind != crate::kind::KIND_AGENT_TURN_METRIC {
+    if !crate::kind::P_GATED_KINDS.contains(&kind) {
         return true;
     }
     let p = nostr::SingleLetterTag::lowercase(nostr::Alphabet::P);

crates/buzz-db/src/channel.rs

@@ -1293,6 +1293,8 @@ pub struct UserRecord {
     pub avatar_url: Option<String>,
     /// Optional NIP-05 identifier (e.g. `user@example.com`).
     pub nip05_handle: Option<String>,
+    /// Whether this community-local identity is a managed agent.
+    pub is_agent: bool,
 }
 
 /// A channel record paired with whether the querying user is an active member.
@@ -1437,7 +1439,8 @@ pub async fn get_users_bulk(
         .collect::<Vec<_>>()
         .join(", ");
     let sql = format!(
-        "SELECT pubkey, display_name, avatar_url, nip05_handle \
+        "SELECT pubkey, display_name, avatar_url, nip05_handle, \
+                agent_owner_pubkey IS NOT NULL AS is_agent \
          FROM users WHERE community_id = $1 AND pubkey IN ({placeholders})"
     );
 
@@ -1455,6 +1458,7 @@ pub async fn get_users_bulk(
             display_name: row.try_get("display_name")?,
             avatar_url: row.try_get("avatar_url")?,
             nip05_handle: row.try_get("nip05_handle")?,
+            is_agent: row.try_get("is_agent")?,
         });
     }
     Ok(out)

crates/buzz-relay/src/api/workflows.rs

@@ -6,6 +6,7 @@
 use std::sync::Arc;
 
 use axum::{
+    body::Bytes,
     extract::{Path, Query, RawQuery, State},
     http::{HeaderMap, StatusCode},
     response::Json,
@@ -33,6 +34,17 @@ pub struct RunsQuery {
     limit: Option<i64>,
 }
 
+/// Optional durable workflow-delivery claim selector.
+#[derive(Debug, Deserialize, Default)]
+pub struct DeliveryClaim {
+    delivery_id: Option<Uuid>,
+    run_id: Option<Uuid>,
+    step_id: Option<String>,
+    definition_event_id: Option<String>,
+    message_event_id: Option<String>,
+    channel_id: Option<Uuid>,
+}
+
 fn request_path(path: &str, raw_query: Option<&str>) -> String {
     match raw_query {
         Some(query) if !query.is_empty() => format!("{path}?{query}"),
@@ -195,6 +207,121 @@ pub async fn run_approvals(
     })))
 }
 
+fn decode_event_id(value: &str, field: &str) -> Result<Vec<u8>, (StatusCode, Json<Value>)> {
+    nostr::EventId::from_hex(value)
+        .map(|id| id.as_bytes().to_vec())
+        .map_err(|_| api_error(StatusCode::BAD_REQUEST, &format!("invalid {field}")))
+}
+
+/// `POST /workflows/deliveries/claim` — atomically claim the authenticated
+/// managed agent's requested delivery, or its oldest pending delivery.
+pub async fn claim_delivery(
+    State(state): State<Arc<AppState>>,
+    headers: HeaderMap,
+    body: Bytes,
+) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
+    let raw_host = headers
+        .get(axum::http::header::HOST)
+        .and_then(|value| value.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",
+            )
+        })?;
+    let path = "/workflows/deliveries/claim";
+    let url = bridge::nip98_expected_url(&state.config.relay_url, &tenant, path);
+    let (pubkey, event_id_bytes) = bridge::verify_bridge_auth(
+        &headers,
+        "POST",
+        &url,
+        Some(&body),
+        state.config.require_auth_token,
+    )?;
+    bridge::enforce_http_admission(&state, &tenant, &pubkey).await?;
+    bridge::check_nip98_replay(&state, &tenant, event_id_bytes).await?;
+    let pubkey_bytes = pubkey.to_bytes().to_vec();
+    let auth_tag = headers
+        .get("x-auth-tag")
+        .and_then(|value| value.to_str().ok());
+    super::relay_members::enforce_relay_membership(
+        &state,
+        tenant.community(),
+        &pubkey_bytes,
+        auth_tag,
+    )
+    .await?;
+
+    let request: DeliveryClaim = serde_json::from_slice(&body)
+        .map_err(|_| api_error(StatusCode::BAD_REQUEST, "invalid delivery claim JSON"))?;
+    let expected_fields = [
+        request.run_id.is_some(),
+        request.step_id.is_some(),
+        request.definition_event_id.is_some(),
+        request.message_event_id.is_some(),
+        request.channel_id.is_some(),
+    ];
+    if expected_fields.iter().any(|present| *present)
+        && !expected_fields.iter().all(|present| *present)
+    {
+        return Err(api_error(
+            StatusCode::BAD_REQUEST,
+            "delivery binding fields must be supplied together",
+        ));
+    }
+    let expected = match (
+        request.run_id,
+        request.step_id,
+        request.definition_event_id,
+        request.message_event_id,
+        request.channel_id,
+    ) {
+        (Some(run_id), Some(step_id), Some(definition_id), Some(message_id), Some(channel_id)) => {
+            Some(buzz_db::workflow::WorkflowAgentDeliveryBinding {
+                run_id,
+                step_id,
+                definition_event_id: decode_event_id(&definition_id, "definition_event_id")?,
+                message_event_id: decode_event_id(&message_id, "message_event_id")?,
+                channel_id,
+            })
+        }
+        _ => None,
+    };
+
+    let delivery = state
+        .db
+        .claim_workflow_agent_delivery(
+            tenant.community(),
+            &pubkey_bytes,
+            request.delivery_id,
+            expected.as_ref(),
+        )
+        .await
+        .map_err(|error| internal_error(&format!("claim workflow delivery: {error}")))?;
+    Ok(Json(serde_json::json!({
+        "delivery": delivery.as_ref().map(delivery_json),
+    })))
+}
+
+fn delivery_json(delivery: &buzz_db::workflow::WorkflowAgentDeliveryRecord) -> Value {
+    serde_json::json!({
+        "id": delivery.id,
+        "workflow_id": delivery.workflow_id,
+        "run_id": delivery.run_id,
+        "step_id": delivery.step_id,
+        "definition_event_id": hex::encode(&delivery.definition_event_id),
+        "message_event_id": hex::encode(&delivery.message_event_id),
+        "channel_id": delivery.channel_id,
+        "claim_token": delivery.claim_token,
+        "execution_trace": delivery.execution_trace,
+        "trigger_context": delivery.trigger_context,
+        "created_at": delivery.created_at,
+    })
+}
+
 fn run_json(run: &buzz_db::workflow::WorkflowRunRecord) -> Value {
     serde_json::json!({
         "id": run.id,

crates/buzz-relay/src/handlers/event.rs

@@ -174,6 +174,29 @@ pub async fn filter_fanout_by_access(
         matches
     };
 
+    // p-gated delivery is recipient-only even for channel-less ephemeral events.
+    // Filter authorization on REQ protects historical reads, but live fan-out
+    // must enforce the same rule at this shared send chokepoint so a kindless
+    // subscription cannot observe another recipient's wake or private event.
+    let matches = if buzz_core::kind::P_GATED_KINDS.contains(&event_kind_u32(&stored_event.event)) {
+        matches
+            .into_iter()
+            .filter(|(conn_id, _)| {
+                state
+                    .conn_manager
+                    .pubkey_for_conn(*conn_id)
+                    .is_some_and(|pubkey| {
+                        buzz_core::filter::reader_authorized_for_event(
+                            &stored_event.event,
+                            &hex::encode(pubkey),
+                        )
+                    })
+            })
+            .collect()
+    } else {
+        matches
+    };
+
     let Some(channel_id) = stored_event.channel_id else {
         return matches;
     };

crates/buzz-relay/src/router.rs

@@ -72,6 +72,10 @@ pub fn build_router(state: Arc<AppState>) -> Router {
         .route("/events", post(api::bridge::submit_event))
         .route("/query", post(api::bridge::query_events))
         .route("/count", post(api::bridge::count_events))
+        .route(
+            "/workflows/deliveries/claim",
+            post(api::workflows::claim_delivery),
+        )
         .route(
             "/workflows/{workflow_id}/runs",
             get(api::workflows::workflow_runs),

crates/buzz-relay/src/workflow_sink.rs

@@ -330,22 +330,80 @@ impl ActionSink for RelayActionSink {
                 .await
                 .map_err(|e| ActionSinkError::Database(e.to_string()))?;
             let named_members: Vec<(String, String)> = users
-                .into_iter()
+                .iter()
                 .filter_map(|u| {
-                    let name = u.display_name?;
+                    let name = u.display_name.clone()?;
                     Some((name, nostr::PublicKey::from_slice(&u.pubkey).ok()?.to_hex()))
                 })
                 .collect();
-            for mentioned in resolve_mention_pubkeys(&text, &named_members) {
-                if mentioned == author_pubkey_hex {
+            let managed_agent_pubkeys: std::collections::HashSet<String> = users
+                .iter()
+                .filter(|u| u.is_agent)
+                .filter_map(|u| {
+                    nostr::PublicKey::from_slice(&u.pubkey)
+                        .ok()
+                        .map(|pk| pk.to_hex())
+                })
+                .collect();
+            let mentioned_pubkeys = resolve_mention_pubkeys(&text, &named_members);
+            for mentioned in &mentioned_pubkeys {
+                if mentioned == &author_pubkey_hex {
                     continue;
                 }
                 tags.push(
-                    Tag::parse(["p", &mentioned])
+                    Tag::parse(["p", mentioned.as_str()])
                         .map_err(|e| ActionSinkError::EventBuild(format!("mention p tag: {e}")))?,
                 );
             }
 
+            let definition_event_id = nostr::EventId::from_hex(&context.definition_event_id)
+                .map_err(|e| {
+                    ActionSinkError::InvalidInput(format!(
+                        "invalid workflow definition event id: {e}"
+                    ))
+                })?;
+            let targets: Vec<buzz_db::workflow::WorkflowAgentDeliveryTarget> = mentioned_pubkeys
+                .iter()
+                .filter(|pubkey| {
+                    pubkey.as_str() != author_pubkey_hex
+                        && managed_agent_pubkeys.contains(pubkey.as_str())
+                })
+                .map(|pubkey| {
+                    nostr::PublicKey::from_hex(pubkey)
+                        .map(|parsed| buzz_db::workflow::WorkflowAgentDeliveryTarget {
+                            id: Uuid::new_v4(),
+                            pubkey: parsed.to_bytes().to_vec(),
+                        })
+                        .map_err(|e| {
+                            ActionSinkError::InvalidInput(format!(
+                                "invalid managed-agent pubkey: {e}"
+                            ))
+                        })
+                })
+                .collect::<Result<_, _>>()?;
+
+            // Serialize the durable identity before signing. A retry returns the
+            // already-committed canonical event rather than minting a second
+            // relay-signed message for the same run step. Messages without a
+            // managed-agent target stay on the ordinary visible-message path.
+            let delivery_transaction = if targets.is_empty() {
+                None
+            } else {
+                let (transaction, existing_message) = state
+                    .db
+                    .lock_workflow_agent_delivery_identity(
+                        tenant.community(),
+                        context.run_id,
+                        &context.step_id,
+                    )
+                    .await
+                    .map_err(|e| ActionSinkError::Database(e.to_string()))?;
+                if let Some((event_id, _)) = existing_message {
+                    return Ok(hex::encode(event_id));
+                }
+                Some(transaction)
+            };
+
             let kind = Kind::from(KIND_STREAM_MESSAGE as u16);
             let event = EventBuilder::new(kind, &text)
                 .tags(tags)
@@ -389,20 +447,47 @@ impl ActionSink for RelayActionSink {
                 },
             });
 
-            let (stored_event, was_inserted) = state
-                .db
-                .insert_event_with_thread_metadata(
-                    tenant.community(),
-                    &event,
-                    Some(channel_uuid),
-                    thread_meta,
-                )
-                .await
-                .map_err(|e| ActionSinkError::Database(e.to_string()))?;
+            let (stored_event, created_delivery_ids) =
+                if let Some(transaction) = delivery_transaction {
+                    state
+                        .db
+                        .commit_workflow_agent_deliveries(
+                            transaction,
+                            tenant.community(),
+                            Some(&event),
+                            &event_id_bytes,
+                            event_created_at,
+                            thread_meta,
+                            context.workflow_id,
+                            context.run_id,
+                            &context.step_id,
+                            definition_event_id.as_bytes(),
+                            channel_uuid,
+                            &targets,
+                            &context.execution_trace,
+                            context.trigger_context.as_ref(),
+                        )
+                        .await
+                        .map_err(|e| ActionSinkError::Database(e.to_string()))?
+                } else {
+                    let (stored, was_inserted) = state
+                        .db
+                        .insert_event_with_thread_metadata(
+                            tenant.community(),
+                            &event,
+                            Some(channel_uuid),
+                            thread_meta,
+                        )
+                        .await
+                        .map_err(|e| ActionSinkError::Database(e.to_string()))?;
+                    (was_inserted.then_some(stored), Vec::new())
+                };
 
-            // 5. Post-persist side effects (fan-out, search, audit)
-            //    Only if actually inserted (idempotency guard).
-            if was_inserted {
+            // 5. Post-commit side effects (fan-out, search, audit). The durable
+            // inbox is authoritative; each ephemeral wake carries only its
+            // target and delivery identifier and may be lost or forged without
+            // changing what an authenticated claim can observe.
+            if let Some(stored_event) = stored_event {
                 let _ = dispatch_persistent_event(
                     &tenant,
                     &state,
@@ -428,6 +513,41 @@ impl ActionSink for RelayActionSink {
                 }
             }
 
+            for delivery_id in created_delivery_ids {
+                let Some(target) = targets.iter().find(|target| target.id == delivery_id) else {
+                    continue;
+                };
+                let target_hex = hex::encode(&target.pubkey);
+                let wake = EventBuilder::new(Kind::from(KIND_WORKFLOW_AGENT_WAKE as u16), "")
+                    .tags(vec![
+                        Tag::parse(["p", target_hex.as_str()])
+                            .map_err(|e| ActionSinkError::EventBuild(format!("wake p tag: {e}")))?,
+                        Tag::parse(["delivery", delivery_id.to_string().as_str()]).map_err(
+                            |e| ActionSinkError::EventBuild(format!("wake delivery tag: {e}")),
+                        )?,
+                    ])
+                    .sign_with_keys(&state.relay_keypair)
+                    .map_err(|e| ActionSinkError::EventBuild(format!("wake signing: {e}")))?;
+                state.mark_local_event(tenant.community(), &wake.id);
+                if let Err(error) = state
+                    .pubsub
+                    .publish_event(&tenant, buzz_pubsub::EventTopic::Global, &wake)
+                    .await
+                {
+                    state
+                        .local_event_ids
+                        .invalidate(&(tenant.community(), wake.id.to_bytes()));
+                    tracing::warn!(%delivery_id, %error, "workflow delivery wake publish failed");
+                }
+                let stored_wake = buzz_core::StoredEvent::new(wake, None);
+                crate::handlers::event::fan_out_event_to_local_subscribers(
+                    &state,
+                    tenant.community(),
+                    &stored_wake,
+                )
+                .await;
+            }
+
             Ok(event_id_hex)
         })
     }

crates/buzz-workflow/src/action_sink.rs

@@ -20,6 +20,10 @@ pub struct WorkflowMessageContext {
     pub definition_event_id: String,
     /// Stable step identifier within the signed definition.
     pub step_id: String,
+    /// Private execution state captured before this step.
+    pub execution_trace: serde_json::Value,
+    /// Private trigger snapshot used to verify the delivery.
+    pub trigger_context: Option<serde_json::Value>,
 }
 
 /// Errors from action sink operations.

crates/buzz-workflow/src/executor.rs

@@ -636,6 +636,8 @@ pub async fn dispatch_action(
                         run_id,
                         definition_event_id: trigger_ctx.definition_event_id.clone(),
                         step_id: step_id.to_owned(),
+                        execution_trace: wf_run.execution_trace,
+                        trigger_context: wf_run.trigger_context,
                     };
                     let event_id = engine
                         .action_sink()?