diff --git a/crates/buzz-core/src/filter.rs b/crates/buzz-core/src/filter.rs index 32e3a7ad16..035df28bb4 100644 --- a/crates/buzz-core/src/filter.rs +++ b/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); diff --git a/crates/buzz-db/src/channel.rs b/crates/buzz-db/src/channel.rs index 109a9367d7..739fcf1bdc 100644 --- a/crates/buzz-db/src/channel.rs +++ b/crates/buzz-db/src/channel.rs @@ -1293,6 +1293,8 @@ pub struct UserRecord { pub avatar_url: Option, /// Optional NIP-05 identifier (e.g. `user@example.com`). pub nip05_handle: Option, + /// 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::>() .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) diff --git a/crates/buzz-relay/src/api/workflows.rs b/crates/buzz-relay/src/api/workflows.rs index a3d5a6c729..c530c4bde4 100644 --- a/crates/buzz-relay/src/api/workflows.rs +++ b/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, } +/// Optional durable workflow-delivery claim selector. +#[derive(Debug, Deserialize, Default)] +pub struct DeliveryClaim { + delivery_id: Option, + run_id: Option, + step_id: Option, + definition_event_id: Option, + message_event_id: Option, + channel_id: Option, +} + 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, (StatusCode, Json)> { + 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>, + headers: HeaderMap, + body: Bytes, +) -> Result, (StatusCode, Json)> { + 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, diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index ccba40f328..86b6c2ca65 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/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; }; diff --git a/crates/buzz-relay/src/router.rs b/crates/buzz-relay/src/router.rs index 1dce66e91e..f1bdae2146 100644 --- a/crates/buzz-relay/src/router.rs +++ b/crates/buzz-relay/src/router.rs @@ -72,6 +72,10 @@ pub fn build_router(state: Arc) -> 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), diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index c8519bac38..ea2a961dff 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/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 = 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 = 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::>()?; + + // 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) }) } diff --git a/crates/buzz-workflow/src/action_sink.rs b/crates/buzz-workflow/src/action_sink.rs index e49433f394..ffceccb134 100644 --- a/crates/buzz-workflow/src/action_sink.rs +++ b/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, } /// Errors from action sink operations. diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index d00d3439c0..431a331799 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/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()?