From 5d5a7ab886b8ca3d734ef19f75a538f3ddff1409 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Tue, 25 Aug 2026 22:21:00 -0400 Subject: [PATCH 1/4] feat(workflows): expose authenticated delivery lifecycle API Co-authored-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> Signed-off-by: Logan Johnson --- crates/buzz-db/src/lib.rs | 56 +++ crates/buzz-relay/src/api/workflows.rs | 671 ++++++++++++++++++++++++- crates/buzz-relay/src/router.rs | 16 + 3 files changed, 742 insertions(+), 1 deletion(-) diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index e4dc5f44ed..d477a7ece2 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -3620,6 +3620,62 @@ impl Db { workflow::get_workflow_run(&self.pool, community_id, id).await } + /// Claim one pending workflow delivery for the authenticated target. + #[datastore_span(name = "claim_workflow_agent_delivery", system = "postgresql")] + pub async fn claim_workflow_agent_delivery( + &self, + community_id: CommunityId, + target_pubkey: &nostr::PublicKey, + delivery_id: Option, + expected: Option<&buzz_core::workflow_delivery::WorkflowDeliveryBinding>, + lease_seconds: i64, + ) -> Result< + Option<( + workflow::WorkflowDeliveryLease, + workflow::WorkflowAgentDeliveryRecord, + )>, + > { + workflow::claim_workflow_agent_delivery( + &self.pool, + community_id, + target_pubkey, + delivery_id, + expected, + lease_seconds, + ) + .await + } + + /// Extend a live workflow delivery lease. + #[datastore_span(name = "renew_workflow_agent_delivery", system = "postgresql")] + pub async fn renew_workflow_agent_delivery( + &self, + lease: &workflow::WorkflowDeliveryLease, + lease_seconds: i64, + ) -> Result { + workflow::renew_workflow_agent_delivery(&self.pool, lease, lease_seconds).await + } + + /// Settle a workflow delivery under its current fenced lease. + #[datastore_span(name = "finish_workflow_agent_delivery", system = "postgresql")] + pub async fn finish_workflow_agent_delivery( + &self, + lease: &workflow::WorkflowDeliveryLease, + outcome: workflow::WorkflowDeliveryOutcome, + ) -> Result { + workflow::finish_workflow_agent_delivery(&self.pool, lease, outcome).await + } + + /// Fetch one workflow delivery within its server-resolved community. + #[datastore_span(name = "get_workflow_agent_delivery", system = "postgresql")] + pub async fn get_workflow_agent_delivery( + &self, + community_id: CommunityId, + delivery_id: buzz_core::workflow_delivery::WorkflowDeliveryId, + ) -> Result> { + workflow::get_workflow_agent_delivery(&self.pool, community_id, delivery_id).await + } + /// List runs for a workflow. #[datastore_span(name = "list_workflow_runs", system = "postgresql")] pub async fn list_workflow_runs( diff --git a/crates/buzz-relay/src/api/workflows.rs b/crates/buzz-relay/src/api/workflows.rs index 70d4bfcbd8..1779cdfea5 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, @@ -15,7 +16,10 @@ use serde::Deserialize; use serde_json::Value; use uuid::Uuid; -use buzz_core::TenantContext; +use buzz_core::{ + workflow_delivery::{WorkflowDeliveryBinding, WorkflowDeliveryCause, WorkflowDeliveryId}, + TenantContext, +}; use crate::{ api::{api_error, bridge, internal_error}, @@ -247,6 +251,545 @@ pub async fn run_approvals( }))) } +const DEFAULT_DELIVERY_LEASE_SECONDS: i64 = 60; +const MIN_DELIVERY_LEASE_SECONDS: i64 = 15; +const MAX_DELIVERY_LEASE_SECONDS: i64 = 300; + +/// Full immutable selector for a specific durable delivery. +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct DeliveryBindingRequest { + community_id: Uuid, + workflow_id: Uuid, + run_id: Uuid, + step_id: String, + target_pubkey: String, + definition_event_id: String, + message_event_id: String, + cause: DeliveryCauseRequest, +} + +/// Trigger authority identity carried by a delivery binding. +#[derive(Debug, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub enum DeliveryCauseRequest { + /// Exact signed event which caused the run. + Event { + /// Trigger event identifier. + event_id: String, + }, + /// Exact scheduled firing slot which caused the run. + Schedule { + /// Authoritative Unix-second schedule slot. + scheduled_for_unix_seconds: i64, + }, + /// Opaque server-side webhook invocation identity. + Webhook { + /// Invocation identifier; no webhook secret or payload. + invocation_id: Uuid, + }, +} + +/// Request to claim a specific delivery or poll the oldest pending delivery. +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ClaimDeliveryRequest { + #[serde(default)] + delivery_id: Option, + #[serde(default)] + expected: Option, + #[serde(default = "default_delivery_lease_seconds")] + lease_seconds: i64, +} + +/// Fenced lease capability presented to read, renew, or finish a delivery. +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct DeliveryLeaseRequest { + lease_generation: i64, + binding: DeliveryBindingRequest, +} + +/// Request to extend a delivery lease. +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct RenewDeliveryRequest { + lease_generation: i64, + binding: DeliveryBindingRequest, + #[serde(default = "default_delivery_lease_seconds")] + lease_seconds: i64, +} + +/// Terminal disposition for a delivery. +#[derive(Debug, Deserialize, Clone, Copy)] +#[serde(rename_all = "snake_case")] +pub enum DeliveryDisposition { + /// Delivery work completed successfully. + Finished, + /// Delivery work failed permanently. + Failed, +} + +/// Request to settle a delivery under its fenced lease. +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct FinishDeliveryRequest { + lease_generation: i64, + binding: DeliveryBindingRequest, + disposition: DeliveryDisposition, +} + +fn default_delivery_lease_seconds() -> i64 { + DEFAULT_DELIVERY_LEASE_SECONDS +} + +fn validate_lease_seconds(value: i64) -> Result)> { + if !(MIN_DELIVERY_LEASE_SECONDS..=MAX_DELIVERY_LEASE_SECONDS).contains(&value) { + return Err(api_error( + StatusCode::BAD_REQUEST, + "lease_seconds must be between 15 and 300", + )); + } + Ok(value) +} + +fn parse_event_id(value: &str, field: &str) -> Result)> { + nostr::EventId::from_hex(value) + .map_err(|_| api_error(StatusCode::BAD_REQUEST, &format!("invalid {field}"))) +} + +fn parse_delivery_binding( + request: DeliveryBindingRequest, + tenant: &TenantContext, + authenticated: nostr::PublicKey, +) -> Result)> { + if request.community_id != *tenant.community().as_uuid() { + return Err(api_error(StatusCode::NOT_FOUND, "delivery not found")); + } + let target = nostr::PublicKey::from_hex(&request.target_pubkey) + .map_err(|_| api_error(StatusCode::BAD_REQUEST, "invalid target_pubkey"))?; + if target != authenticated { + return Err(api_error(StatusCode::NOT_FOUND, "delivery not found")); + } + let cause = match request.cause { + DeliveryCauseRequest::Event { event_id } => { + WorkflowDeliveryCause::Event(parse_event_id(&event_id, "cause.event_id")?) + } + DeliveryCauseRequest::Schedule { + scheduled_for_unix_seconds, + } => WorkflowDeliveryCause::Schedule { + scheduled_for_unix_seconds, + }, + DeliveryCauseRequest::Webhook { invocation_id } => { + WorkflowDeliveryCause::Webhook { invocation_id } + } + }; + WorkflowDeliveryBinding::new( + tenant.community(), + request.workflow_id, + request.run_id, + request.step_id, + target, + parse_event_id(&request.definition_event_id, "definition_event_id")?, + parse_event_id(&request.message_event_id, "message_event_id")?, + cause, + ) + .map_err(|_| api_error(StatusCode::BAD_REQUEST, "invalid delivery binding")) +} + +async fn authorize_delivery_request( + state: &Arc, + headers: &HeaderMap, + path: &str, + body: &[u8], +) -> Result<(TenantContext, nostr::PublicKey), (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, "delivery not found"))?; + let url = bridge::nip98_expected_url(&state.config.relay_url, &tenant, path); + let (pubkey, event_id) = bridge::verify_bridge_auth_with_options( + headers, + "POST", + &url, + Some(body), + state.config.require_auth_token, + true, + )?; + bridge::enforce_http_admission(state, &tenant, &pubkey).await?; + bridge::check_nip98_replay(state, &tenant, event_id).await?; + let pubkey_bytes = pubkey.to_bytes(); + 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 (_, owner) = state + .db + .get_agent_channel_policy(tenant.community(), &pubkey_bytes) + .await + .map_err(|error| internal_error(&format!("delivery target lookup: {error}")))? + .ok_or_else(|| api_error(StatusCode::NOT_FOUND, "delivery not found"))?; + if owner.is_none() { + return Err(api_error(StatusCode::NOT_FOUND, "delivery not found")); + } + Ok((tenant, pubkey)) +} + +async fn validate_binding_channel( + state: &Arc, + tenant: &TenantContext, + binding: &WorkflowDeliveryBinding, +) -> Result<(), (StatusCode, Json)> { + let message = state + .db + .get_event_by_id(tenant.community(), binding.message_event_id().as_bytes()) + .await + .map_err(|error| internal_error(&format!("delivery message lookup: {error}")))? + .ok_or_else(|| api_error(StatusCode::NOT_FOUND, "delivery not found"))?; + let Some(channel_id) = message.channel_id else { + return Err(api_error(StatusCode::NOT_FOUND, "delivery not found")); + }; + let workflow = state + .db + .get_workflow(tenant.community(), binding.workflow_id()) + .await + .map_err(|_| api_error(StatusCode::NOT_FOUND, "delivery not found"))?; + if workflow.channel_id != Some(channel_id) { + return Err(api_error(StatusCode::NOT_FOUND, "delivery not found")); + } + Ok(()) +} + +async fn current_bound_delivery( + state: &Arc, + tenant: &TenantContext, + target: nostr::PublicKey, + delivery_id: WorkflowDeliveryId, + lease_generation: i64, + requested: DeliveryBindingRequest, + require_live_claim: bool, +) -> Result)> { + let delivery = state + .db + .get_workflow_agent_delivery(tenant.community(), delivery_id) + .await + .map_err(|error| internal_error(&format!("read workflow delivery: {error}")))? + .ok_or_else(|| api_error(StatusCode::NOT_FOUND, "delivery not found"))?; + let binding = parse_delivery_binding(requested, tenant, target)?; + let live_claim = delivery.status == buzz_db::workflow::WorkflowDeliveryStatus::Claimed + && delivery + .lease_until + .is_some_and(|until| until >= Utc::now()); + if delivery.binding != binding + || delivery.lease_generation != lease_generation + || (require_live_claim && !live_claim) + { + return Err(api_error( + StatusCode::CONFLICT, + "delivery lease is not current", + )); + } + // Resolve message/channel authority only after the caller has demonstrated + // the complete delivery binding. Otherwise this endpoint becomes an event + // existence oracle for arbitrary managed-agent identities. + validate_binding_channel(state, tenant, &binding).await?; + Ok(delivery) +} + +fn lease_for( + tenant: &TenantContext, + target: nostr::PublicKey, + delivery_id: WorkflowDeliveryId, + lease_generation: i64, +) -> buzz_db::workflow::WorkflowDeliveryLease { + buzz_db::workflow::WorkflowDeliveryLease { + community_id: tenant.community(), + delivery_id, + target_pubkey: target, + lease_generation, + // C's mutation queries fence on community/id/target/generation and the + // database clock. This field is returned state, not client authority. + lease_until: Utc::now(), + } +} + +/// `POST /workflows/agent-deliveries/claim` — claim only as the immutable +/// managed-agent target. A wake can supply identifiers but grants no authority. +pub async fn claim_agent_delivery( + State(state): State>, + headers: HeaderMap, + body: Bytes, +) -> Result, (StatusCode, Json)> { + let path = "/workflows/agent-deliveries/claim"; + let (tenant, target) = authorize_delivery_request(&state, &headers, path, &body).await?; + let request: ClaimDeliveryRequest = serde_json::from_slice(&body) + .map_err(|_| api_error(StatusCode::BAD_REQUEST, "invalid delivery claim JSON"))?; + let lease_seconds = validate_lease_seconds(request.lease_seconds)?; + if request.delivery_id.is_none() && request.expected.is_some() { + return Err(api_error( + StatusCode::BAD_REQUEST, + "expected binding requires delivery_id", + )); + } + let delivery_id = request.delivery_id.map(WorkflowDeliveryId::from_uuid); + let expected = request + .expected + .map(|expected| parse_delivery_binding(expected, &tenant, target)) + .transpose()?; + if let Some(binding) = expected.as_ref() { + validate_binding_channel(&state, &tenant, binding).await?; + } + let claimed = state + .db + .claim_workflow_agent_delivery( + tenant.community(), + &target, + delivery_id, + expected.as_ref(), + lease_seconds, + ) + .await + .map_err(|error| internal_error(&format!("claim workflow delivery: {error}")))?; + let Some((lease, delivery)) = claimed else { + return Err(api_error(StatusCode::NOT_FOUND, "delivery not found")); + }; + delivery_response(&state, &tenant, &target, &lease, &delivery).await +} + +/// `POST /workflows/agent-deliveries/{id}/read` — return private execution +/// inputs only to the exact target holding the current live fencing token. +pub async fn read_agent_delivery( + State(state): State>, + Path(delivery_id): Path, + headers: HeaderMap, + body: Bytes, +) -> Result, (StatusCode, Json)> { + let path = format!("/workflows/agent-deliveries/{delivery_id}/read"); + let (tenant, target) = authorize_delivery_request(&state, &headers, &path, &body).await?; + let request: DeliveryLeaseRequest = serde_json::from_slice(&body) + .map_err(|_| api_error(StatusCode::BAD_REQUEST, "invalid delivery read JSON"))?; + let id = WorkflowDeliveryId::from_uuid(delivery_id); + let delivery = current_bound_delivery( + &state, + &tenant, + target, + id, + request.lease_generation, + request.binding, + true, + ) + .await?; + let lease = lease_for(&tenant, target, id, request.lease_generation); + delivery_response(&state, &tenant, &target, &lease, &delivery).await +} + +/// `POST /workflows/agent-deliveries/{id}/renew` — extend the current fenced lease. +pub async fn renew_agent_delivery( + State(state): State>, + Path(delivery_id): Path, + headers: HeaderMap, + body: Bytes, +) -> Result, (StatusCode, Json)> { + let path = format!("/workflows/agent-deliveries/{delivery_id}/renew"); + let (tenant, target) = authorize_delivery_request(&state, &headers, &path, &body).await?; + let request: RenewDeliveryRequest = serde_json::from_slice(&body) + .map_err(|_| api_error(StatusCode::BAD_REQUEST, "invalid delivery renewal JSON"))?; + let lease_seconds = validate_lease_seconds(request.lease_seconds)?; + let id = WorkflowDeliveryId::from_uuid(delivery_id); + current_bound_delivery( + &state, + &tenant, + target, + id, + request.lease_generation, + request.binding, + true, + ) + .await?; + let lease = lease_for(&tenant, target, id, request.lease_generation); + match state + .db + .renew_workflow_agent_delivery(&lease, lease_seconds) + .await + .map_err(|error| internal_error(&format!("renew workflow delivery: {error}")))? + { + buzz_db::workflow::WorkflowDeliveryRenewOutcome::Renewed(lease_until) => Ok(Json( + serde_json::json!({"renewed": true, "lease_until": lease_until}), + )), + buzz_db::workflow::WorkflowDeliveryRenewOutcome::LeaseLost => Err(api_error( + StatusCode::CONFLICT, + "delivery lease is not current", + )), + } +} + +/// `POST /workflows/agent-deliveries/{id}/finish` — settle once, with stable +/// acknowledgement only when a terminal replay has the identical disposition. +pub async fn finish_agent_delivery( + State(state): State>, + Path(delivery_id): Path, + headers: HeaderMap, + body: Bytes, +) -> Result, (StatusCode, Json)> { + let path = format!("/workflows/agent-deliveries/{delivery_id}/finish"); + let (tenant, target) = authorize_delivery_request(&state, &headers, &path, &body).await?; + let request: FinishDeliveryRequest = serde_json::from_slice(&body) + .map_err(|_| api_error(StatusCode::BAD_REQUEST, "invalid delivery finish JSON"))?; + let id = WorkflowDeliveryId::from_uuid(delivery_id); + current_bound_delivery( + &state, + &tenant, + target, + id, + request.lease_generation, + request.binding, + false, + ) + .await?; + let requested = match request.disposition { + DeliveryDisposition::Finished => buzz_db::workflow::WorkflowDeliveryOutcome::Finished, + DeliveryDisposition::Failed => buzz_db::workflow::WorkflowDeliveryOutcome::Failed, + }; + let lease = lease_for(&tenant, target, id, request.lease_generation); + match state + .db + .finish_workflow_agent_delivery(&lease, requested) + .await + .map_err(|error| internal_error(&format!("finish workflow delivery: {error}")))? + { + buzz_db::workflow::WorkflowDeliveryFinishOutcome::Settled(_) => Ok(Json( + serde_json::json!({"finished": true, "replayed": false}), + )), + buzz_db::workflow::WorkflowDeliveryFinishOutcome::AlreadyTerminal(status) + if terminal_matches(status, request.disposition) => + { + Ok(Json( + serde_json::json!({"finished": true, "replayed": true}), + )) + } + buzz_db::workflow::WorkflowDeliveryFinishOutcome::AlreadyTerminal(_) + | buzz_db::workflow::WorkflowDeliveryFinishOutcome::LeaseLost => Err(api_error( + StatusCode::CONFLICT, + "delivery lease is not current", + )), + } +} + +fn terminal_matches( + status: buzz_db::workflow::WorkflowDeliveryStatus, + disposition: DeliveryDisposition, +) -> bool { + matches!( + (status, disposition), + ( + buzz_db::workflow::WorkflowDeliveryStatus::Finished, + DeliveryDisposition::Finished + ) | ( + buzz_db::workflow::WorkflowDeliveryStatus::Failed, + DeliveryDisposition::Failed + ) + ) +} + +async fn delivery_response( + state: &Arc, + tenant: &TenantContext, + target: &nostr::PublicKey, + lease: &buzz_db::workflow::WorkflowDeliveryLease, + delivery: &buzz_db::workflow::WorkflowAgentDeliveryRecord, +) -> Result, (StatusCode, Json)> { + if delivery.binding.community_id() != tenant.community() + || delivery.binding.target_pubkey() != *target + || delivery.id != lease.delivery_id + || delivery.lease_generation != lease.lease_generation + { + return Err(api_error(StatusCode::NOT_FOUND, "delivery not found")); + } + let run = state + .db + .get_workflow_run(tenant.community(), delivery.binding.run_id()) + .await + .map_err(|error| internal_error(&format!("delivery run lookup: {error}")))?; + if run.workflow_id != delivery.binding.workflow_id() + || run.definition_event_id.as_deref() + != Some(delivery.binding.definition_event_id().as_bytes()) + { + return Err(api_error( + StatusCode::CONFLICT, + "delivery binding is unavailable", + )); + } + let definition = state + .db + .get_event_by_id( + tenant.community(), + delivery.binding.definition_event_id().as_bytes(), + ) + .await + .map_err(|error| internal_error(&format!("delivery definition lookup: {error}")))? + .ok_or_else(|| api_error(StatusCode::CONFLICT, "delivery binding is unavailable"))?; + let message = state + .db + .get_event_by_id( + tenant.community(), + delivery.binding.message_event_id().as_bytes(), + ) + .await + .map_err(|error| internal_error(&format!("delivery message lookup: {error}")))? + .ok_or_else(|| api_error(StatusCode::CONFLICT, "delivery binding is unavailable"))?; + Ok(Json(serde_json::json!({ + "delivery": delivery_json(delivery, lease), + "definition_event": definition.event, + "message_event": message.event, + }))) +} + +fn delivery_json( + delivery: &buzz_db::workflow::WorkflowAgentDeliveryRecord, + lease: &buzz_db::workflow::WorkflowDeliveryLease, +) -> Value { + let binding = &delivery.binding; + serde_json::json!({ + "id": delivery.id.as_uuid(), + "community_id": binding.community_id().as_uuid(), + "workflow_id": binding.workflow_id(), + "run_id": binding.run_id(), + "step_id": binding.step_id(), + "target_pubkey": binding.target_pubkey().to_hex(), + "definition_event_id": binding.definition_event_id().to_hex(), + "message_event_id": binding.message_event_id().to_hex(), + "cause": delivery_cause_json(binding.cause()), + "lease_generation": lease.lease_generation, + "lease_until": lease.lease_until, + }) +} + +fn delivery_cause_json(cause: &WorkflowDeliveryCause) -> Value { + match cause { + WorkflowDeliveryCause::Event(event_id) => { + serde_json::json!({"kind": "event", "event_id": event_id.to_hex()}) + } + WorkflowDeliveryCause::Schedule { + scheduled_for_unix_seconds, + } => serde_json::json!({ + "kind": "schedule", + "scheduled_for_unix_seconds": scheduled_for_unix_seconds, + }), + WorkflowDeliveryCause::Webhook { invocation_id } => { + serde_json::json!({"kind": "webhook", "invocation_id": invocation_id}) + } + } +} + fn run_json(run: &buzz_db::workflow::WorkflowRunRecord) -> Value { serde_json::json!({ "id": run.id, @@ -294,6 +837,132 @@ mod tests { ); } + #[test] + fn delivery_wire_excludes_private_run_state() { + let community = buzz_core::tenant::CommunityId::from_uuid(Uuid::new_v4()); + let target = nostr::Keys::generate().public_key(); + let binding = WorkflowDeliveryBinding::new( + community, + Uuid::new_v4(), + Uuid::new_v4(), + "notify".to_owned(), + target, + nostr::EventId::from_byte_array([0x11; 32]), + nostr::EventId::from_byte_array([0x22; 32]), + WorkflowDeliveryCause::Webhook { + invocation_id: Uuid::new_v4(), + }, + ) + .expect("valid delivery binding"); + let delivery = buzz_db::workflow::WorkflowAgentDeliveryRecord { + id: WorkflowDeliveryId::from_uuid(Uuid::new_v4()), + binding, + status: buzz_db::workflow::WorkflowDeliveryStatus::Claimed, + lease_generation: 4, + lease_until: Some(Utc::now()), + claimed_at: Some(Utc::now()), + finished_at: None, + created_at: Utc::now(), + }; + let lease = buzz_db::workflow::WorkflowDeliveryLease { + community_id: community, + delivery_id: delivery.id, + target_pubkey: target, + lease_generation: 4, + lease_until: Utc::now(), + }; + + let wire = serde_json::json!({ + "delivery": delivery_json(&delivery, &lease), + "definition_event": {"content": "signed definition"}, + "message_event": {"content": "visible message"}, + }); + assert!(wire.get("execution_trace").is_none()); + assert!(wire.get("trigger_context").is_none()); + assert!(wire.get("webhook_fields").is_none()); + assert_eq!(wire["delivery"]["lease_generation"], 4); + } + + #[test] + fn delivery_binding_rejects_cross_tenant_and_cross_target() { + let community = buzz_core::tenant::CommunityId::from_uuid(Uuid::new_v4()); + let tenant = TenantContext::resolved(community, "agent.example"); + let authenticated = nostr::Keys::generate().public_key(); + let other = nostr::Keys::generate().public_key(); + let request = |community_id, target_pubkey: String| DeliveryBindingRequest { + community_id, + workflow_id: Uuid::new_v4(), + run_id: Uuid::new_v4(), + step_id: "notify".to_owned(), + target_pubkey, + definition_event_id: "11".repeat(32), + message_event_id: "22".repeat(32), + cause: DeliveryCauseRequest::Webhook { + invocation_id: Uuid::new_v4(), + }, + }; + + let (status, _) = parse_delivery_binding( + request(Uuid::new_v4(), authenticated.to_hex()), + &tenant, + authenticated, + ) + .expect_err("body tenant cannot override host-resolved tenant"); + assert_eq!(status, StatusCode::NOT_FOUND); + + let (status, _) = parse_delivery_binding( + request(*community.as_uuid(), other.to_hex()), + &tenant, + authenticated, + ) + .expect_err("authenticated target must match the binding"); + assert_eq!(status, StatusCode::NOT_FOUND); + } + + #[test] + fn malformed_binding_and_lease_bounds_fail_closed() { + let community = buzz_core::tenant::CommunityId::from_uuid(Uuid::new_v4()); + let tenant = TenantContext::resolved(community, "agent.example"); + let authenticated = nostr::Keys::generate().public_key(); + let malformed = DeliveryBindingRequest { + community_id: *community.as_uuid(), + workflow_id: Uuid::new_v4(), + run_id: Uuid::new_v4(), + step_id: "notify".to_owned(), + target_pubkey: authenticated.to_hex(), + definition_event_id: "not-an-event-id".to_owned(), + message_event_id: "22".repeat(32), + cause: DeliveryCauseRequest::Webhook { + invocation_id: Uuid::new_v4(), + }, + }; + let (status, _) = parse_delivery_binding(malformed, &tenant, authenticated) + .expect_err("malformed signed authority must fail closed"); + assert_eq!(status, StatusCode::BAD_REQUEST); + assert!(validate_lease_seconds(MIN_DELIVERY_LEASE_SECONDS - 1).is_err()); + assert!(validate_lease_seconds(MAX_DELIVERY_LEASE_SECONDS + 1).is_err()); + } + + #[test] + fn terminal_replay_requires_the_same_disposition() { + assert!(terminal_matches( + buzz_db::workflow::WorkflowDeliveryStatus::Finished, + DeliveryDisposition::Finished, + )); + assert!(terminal_matches( + buzz_db::workflow::WorkflowDeliveryStatus::Failed, + DeliveryDisposition::Failed, + )); + assert!(!terminal_matches( + buzz_db::workflow::WorkflowDeliveryStatus::Finished, + DeliveryDisposition::Failed, + )); + assert!(!terminal_matches( + buzz_db::workflow::WorkflowDeliveryStatus::Failed, + DeliveryDisposition::Finished, + )); + } + #[test] fn approval_wire_does_not_expose_hash_as_token() { let approval = buzz_db::workflow::ApprovalRecord { diff --git a/crates/buzz-relay/src/router.rs b/crates/buzz-relay/src/router.rs index 0d9487ec3d..5c1cab02da 100644 --- a/crates/buzz-relay/src/router.rs +++ b/crates/buzz-relay/src/router.rs @@ -87,6 +87,22 @@ pub fn build_router(state: Arc) -> Router { "/workflows/{workflow_id}/runs/{run_id}/approvals", get(api::workflows::run_approvals), ) + .route( + "/workflows/agent-deliveries/claim", + post(api::workflows::claim_agent_delivery), + ) + .route( + "/workflows/agent-deliveries/{delivery_id}/read", + post(api::workflows::read_agent_delivery), + ) + .route( + "/workflows/agent-deliveries/{delivery_id}/renew", + post(api::workflows::renew_agent_delivery), + ) + .route( + "/workflows/agent-deliveries/{delivery_id}/finish", + post(api::workflows::finish_agent_delivery), + ) .route( "/operator/communities", get(api::operator::list_owned_communities).post(api::operator::provision_community), From aa73cf53d0b0728062891307a26f780e55196f56 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Tue, 25 Aug 2026 22:44:05 -0400 Subject: [PATCH 2/4] test(workflows): prove authenticated delivery routes Co-authored-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> Signed-off-by: Logan Johnson --- crates/buzz-relay/src/api/workflows.rs | 477 +++++++++++++++++++++++++ 1 file changed, 477 insertions(+) diff --git a/crates/buzz-relay/src/api/workflows.rs b/crates/buzz-relay/src/api/workflows.rs index 1779cdfea5..e50cc3f600 100644 --- a/crates/buzz-relay/src/api/workflows.rs +++ b/crates/buzz-relay/src/api/workflows.rs @@ -823,7 +823,484 @@ fn approval_json(approval: &buzz_db::workflow::ApprovalRecord) -> Value { #[cfg(test)] mod tests { + use std::sync::Arc; + use super::*; + use axum::{ + body::{to_bytes, Body}, + http::{header, Request}, + }; + use base64::Engine; + use buzz_core::workflow_delivery::WorkflowDeliveryBinding; + use buzz_db::{ + channel::{ChannelType, ChannelVisibility}, + workflow::WorkflowAgentDelivery, + }; + use nostr::{EventBuilder, Keys, Kind, Tag}; + use sha2::{Digest, Sha256}; + use sqlx::PgPool; + use tower::ServiceExt; + + use crate::{router::build_router, state::AppState}; + + const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1 + + struct AlwaysFreshReplayGuard; + + impl buzz_auth::Nip98ReplayGuard for AlwaysFreshReplayGuard { + fn try_mark_in_scope<'a>( + &'a self, + _scope: &'a str, + _event_id: &'a nostr::EventId, + _ttl_secs: u64, + ) -> std::pin::Pin< + Box> + Send + 'a>, + > { + Box::pin(async { Ok(true) }) + } + } + + struct DeliveryRouteFixture { + state: Arc, + pool: PgPool, + host: String, + community: buzz_core::CommunityId, + target: Keys, + other_target: Keys, + binding: WorkflowDeliveryBinding, + delivery_id: WorkflowDeliveryId, + } + + fn nip98_auth_header(keys: &Keys, url: &str, body: &[u8]) -> String { + let hash: [u8; 32] = Sha256::digest(body).into(); + let tags = vec![ + Tag::parse(["u", url]).expect("u tag"), + Tag::parse(["method", "POST"]).expect("method tag"), + Tag::parse(["payload", hex::encode(hash).as_str()]).expect("payload tag"), + ]; + let event = EventBuilder::new(Kind::HttpAuth, "") + .tags(tags) + .sign_with_keys(keys) + .expect("sign NIP-98 event"); + let encoded = base64::engine::general_purpose::STANDARD + .encode(serde_json::to_vec(&event).expect("serialize NIP-98 event")); + format!("Nostr {encoded}") + } + + async fn post_delivery_json( + fixture: &DeliveryRouteFixture, + keys: &Keys, + path: &str, + body: Value, + ) -> axum::response::Response { + let body = body.to_string(); + let url = format!("https://{}{path}", fixture.host); + let auth = nip98_auth_header(keys, &url, body.as_bytes()); + build_router(Arc::clone(&fixture.state)) + .oneshot( + Request::builder() + .method("POST") + .uri(path) + .header(header::HOST, &fixture.host) + .header(header::AUTHORIZATION, auth) + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from(body)) + .expect("request"), + ) + .await + .expect("response") + } + + async fn response_json(response: axum::response::Response) -> Value { + let body = to_bytes(response.into_body(), 1024 * 1024) + .await + .expect("read response body"); + serde_json::from_slice(&body).expect("response JSON") + } + + fn binding_json(binding: &WorkflowDeliveryBinding) -> Value { + let cause = match binding.cause() { + WorkflowDeliveryCause::Event(event_id) => { + serde_json::json!({"kind": "event", "event_id": event_id.to_hex()}) + } + WorkflowDeliveryCause::Schedule { + scheduled_for_unix_seconds, + } => serde_json::json!({ + "kind": "schedule", + "scheduled_for_unix_seconds": scheduled_for_unix_seconds, + }), + WorkflowDeliveryCause::Webhook { invocation_id } => { + serde_json::json!({"kind": "webhook", "invocation_id": invocation_id}) + } + }; + serde_json::json!({ + "community_id": binding.community_id().as_uuid(), + "workflow_id": binding.workflow_id(), + "run_id": binding.run_id(), + "step_id": binding.step_id(), + "target_pubkey": binding.target_pubkey().to_hex(), + "definition_event_id": binding.definition_event_id().to_hex(), + "message_event_id": binding.message_event_id().to_hex(), + "cause": cause, + }) + } + + async fn delivery_route_fixture() -> Option { + let mut config = crate::config::Config::from_env().ok()?; + let database_url = std::env::var("BUZZ_TEST_DATABASE_URL") + .or_else(|_| std::env::var("DATABASE_URL")) + .unwrap_or_else(|_| TEST_DB_URL.to_owned()); + config.database_url = database_url.clone(); + config.redis_url = "redis://127.0.0.1:6379".to_owned(); + let host = format!("workflow-delivery-{}.example", Uuid::new_v4().simple()); + config.relay_url = format!("wss://{host}"); + config.require_relay_membership = false; + + let pool = PgPool::connect(&database_url).await.ok()?; + let db = buzz_db::Db::from_pool(pool.clone()); + db.migrate().await.ok()?; + let community = db.ensure_configured_community(&host).await.ok()?.id; + let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) + .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + .ok()?; + let pubsub = Arc::new( + buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) + .await + .ok()?, + ); + let audit = buzz_audit::AuditService::new(pool.clone()); + let auth = buzz_auth::AuthService::new(config.auth.clone()); + let search = buzz_search::SearchService::new(pool.clone()); + let workflow_engine = Arc::new(buzz_workflow::WorkflowEngine::new( + db.clone(), + buzz_workflow::WorkflowConfig::default(), + )); + let media_storage = buzz_media::MediaStorage::new(&config.media).ok()?; + let (mut state, _audit_shutdown) = AppState::new( + config, + db.clone(), + redis_pool, + audit, + pubsub, + auth, + search, + workflow_engine, + Keys::generate(), + media_storage, + ); + state.nip98_replay = Arc::new(AlwaysFreshReplayGuard); + + let owner = Keys::generate(); + let target = Keys::generate(); + let other_target = Keys::generate(); + for keys in [&owner, &target, &other_target] { + db.ensure_user(community, keys.public_key().as_bytes()) + .await + .ok()?; + } + db.set_agent_owner( + community, + target.public_key().as_bytes(), + owner.public_key().as_bytes(), + ) + .await + .ok()?; + db.set_agent_owner( + community, + other_target.public_key().as_bytes(), + owner.public_key().as_bytes(), + ) + .await + .ok()?; + let channel = db + .create_channel( + community, + "delivery-route-test", + ChannelType::Stream, + ChannelVisibility::Private, + None, + owner.public_key().as_bytes(), + Some(3600), + ) + .await + .ok()?; + let workflow_id = db + .create_workflow( + community, + Some(channel.id), + owner.public_key().as_bytes(), + "delivery-route-test", + r#"{"name":"delivery-route-test","on":{"manual":{}},"steps":[]}"#, + &[0x44; 32], + ) + .await + .ok()?; + let definition = EventBuilder::new(Kind::Custom(30620), "signed definition") + .tag(Tag::parse(["d", workflow_id.to_string().as_str()]).ok()?) + .sign_with_keys(&owner) + .ok()?; + let message = EventBuilder::new(Kind::Custom(9), "private execution input") + .tags([ + Tag::parse(["h", channel.id.to_string().as_str()]).ok()?, + Tag::parse([ + "p", + target.public_key().to_hex().as_str(), + "", + buzz_core::workflow_delivery::WORKFLOW_DELIVERY_TARGET_MARKER, + ]) + .ok()?, + ]) + .sign_with_keys(&owner) + .ok()?; + db.insert_event(community, &definition, Some(channel.id)) + .await + .ok()?; + db.insert_event(community, &message, Some(channel.id)) + .await + .ok()?; + sqlx::query( + "UPDATE workflows SET definition_event_id = $3 WHERE community_id = $1 AND id = $2", + ) + .bind(community.as_uuid()) + .bind(workflow_id) + .bind(definition.id.as_bytes().as_slice()) + .execute(&pool) + .await + .ok()?; + let run_id = db + .create_workflow_run( + community, + workflow_id, + definition.id.as_bytes(), + Some(message.id.as_bytes()), + None, + ) + .await + .ok()?; + let binding = WorkflowDeliveryBinding::new( + community, + workflow_id, + run_id, + "notify", + target.public_key(), + definition.id, + message.id, + WorkflowDeliveryCause::Event(message.id), + ) + .ok()?; + let delivery_id = WorkflowDeliveryId::from_uuid(Uuid::new_v4()); + let (transaction, existing) = buzz_db::workflow::lock_workflow_agent_delivery_identity( + &pool, community, run_id, "notify", + ) + .await + .ok()?; + if existing { + return None; + } + let stored_message = db + .get_event_by_id(community, message.id.as_bytes()) + .await + .ok()??; + buzz_db::workflow::commit_workflow_agent_deliveries( + transaction, + community, + DateTime::::from_timestamp(stored_message.event.created_at.as_secs() as i64, 0) + .expect("message timestamp"), + &[WorkflowAgentDelivery { + id: delivery_id, + binding: binding.clone(), + }], + ) + .await + .ok()?; + + Some(DeliveryRouteFixture { + state: Arc::new(state), + pool, + host, + community, + target, + other_target, + binding, + delivery_id, + }) + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn authenticated_delivery_routes_complete_the_fenced_lifecycle() { + let Some(fixture) = delivery_route_fixture().await else { + eprintln!("skipping: Postgres unavailable"); + return; + }; + let binding = binding_json(&fixture.binding); + let claim = post_delivery_json( + &fixture, + &fixture.target, + "/workflows/agent-deliveries/claim", + serde_json::json!({ + "delivery_id": fixture.delivery_id.as_uuid(), + "expected": binding, + "lease_seconds": 60, + }), + ) + .await; + assert_eq!(claim.status(), StatusCode::OK); + let claimed = response_json(claim).await; + assert_eq!(claimed["delivery"]["lease_generation"], 1); + assert_eq!(claimed["definition_event"]["content"], "signed definition"); + assert_eq!( + claimed["message_event"]["content"], + "private execution input" + ); + assert!(claimed.get("execution_trace").is_none()); + + let lease = serde_json::json!({ + "lease_generation": 1, + "binding": binding_json(&fixture.binding), + }); + let path = format!("/workflows/agent-deliveries/{}/read", fixture.delivery_id); + let read = post_delivery_json(&fixture, &fixture.target, &path, lease.clone()).await; + assert_eq!(read.status(), StatusCode::OK); + + let path = format!("/workflows/agent-deliveries/{}/renew", fixture.delivery_id); + let renew = post_delivery_json( + &fixture, + &fixture.target, + &path, + serde_json::json!({ + "lease_generation": 1, + "binding": binding_json(&fixture.binding), + "lease_seconds": 60, + }), + ) + .await; + assert_eq!(renew.status(), StatusCode::OK); + assert_eq!(response_json(renew).await["renewed"], true); + + let path = format!("/workflows/agent-deliveries/{}/finish", fixture.delivery_id); + let finish_body = serde_json::json!({ + "lease_generation": 1, + "binding": binding_json(&fixture.binding), + "disposition": "finished", + }); + let finish = + post_delivery_json(&fixture, &fixture.target, &path, finish_body.clone()).await; + assert_eq!(finish.status(), StatusCode::OK); + assert_eq!(response_json(finish).await["replayed"], false); + let replay = post_delivery_json(&fixture, &fixture.target, &path, finish_body).await; + assert_eq!(replay.status(), StatusCode::OK); + assert_eq!(response_json(replay).await["replayed"], true); + let conflicting = post_delivery_json( + &fixture, + &fixture.target, + &path, + serde_json::json!({ + "lease_generation": 1, + "binding": binding_json(&fixture.binding), + "disposition": "failed", + }), + ) + .await; + assert_eq!(conflicting.status(), StatusCode::CONFLICT); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn authenticated_delivery_routes_reject_wrong_authority_and_stale_fences() { + let Some(fixture) = delivery_route_fixture().await else { + eprintln!("skipping: Postgres unavailable"); + return; + }; + let claim_path = "/workflows/agent-deliveries/claim"; + let wrong_target = post_delivery_json( + &fixture, + &fixture.other_target, + claim_path, + serde_json::json!({ + "delivery_id": fixture.delivery_id.as_uuid(), + "expected": binding_json(&fixture.binding), + "lease_seconds": 60, + }), + ) + .await; + assert_eq!(wrong_target.status(), StatusCode::NOT_FOUND); + + let mut wrong_tenant = binding_json(&fixture.binding); + wrong_tenant["community_id"] = serde_json::json!(Uuid::new_v4()); + let response = post_delivery_json( + &fixture, + &fixture.target, + claim_path, + serde_json::json!({ + "delivery_id": fixture.delivery_id.as_uuid(), + "expected": wrong_tenant, + "lease_seconds": 60, + }), + ) + .await; + assert_eq!(response.status(), StatusCode::NOT_FOUND); + + let mut wrong_binding = binding_json(&fixture.binding); + wrong_binding["step_id"] = serde_json::json!("other-step"); + let response = post_delivery_json( + &fixture, + &fixture.target, + claim_path, + serde_json::json!({ + "delivery_id": fixture.delivery_id.as_uuid(), + "expected": wrong_binding, + "lease_seconds": 60, + }), + ) + .await; + assert_eq!(response.status(), StatusCode::NOT_FOUND); + + let claim = post_delivery_json( + &fixture, + &fixture.target, + claim_path, + serde_json::json!({ + "delivery_id": fixture.delivery_id.as_uuid(), + "expected": binding_json(&fixture.binding), + "lease_seconds": 60, + }), + ) + .await; + assert_eq!(claim.status(), StatusCode::OK); + let read_path = format!("/workflows/agent-deliveries/{}/read", fixture.delivery_id); + let stale_generation = post_delivery_json( + &fixture, + &fixture.target, + &read_path, + serde_json::json!({ + "lease_generation": 0, + "binding": binding_json(&fixture.binding), + }), + ) + .await; + assert_eq!(stale_generation.status(), StatusCode::CONFLICT); + + sqlx::query( + "UPDATE workflow_agent_deliveries SET lease_until = NOW() - INTERVAL '1 second' WHERE community_id = $1 AND id = $2", + ) + .bind(fixture.community.as_uuid()) + .bind(fixture.delivery_id.as_uuid()) + .execute(&fixture.pool) + .await + .expect("expire delivery lease"); + let expired = post_delivery_json( + &fixture, + &fixture.target, + &read_path, + serde_json::json!({ + "lease_generation": 1, + "binding": binding_json(&fixture.binding), + }), + ) + .await; + assert_eq!(expired.status(), StatusCode::CONFLICT); + } #[test] fn request_path_preserves_signed_query_verbatim() { From 557c6a355b0dc7af88043179ddd0cbcdbbe69647 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Wed, 26 Aug 2026 12:04:20 -0400 Subject: [PATCH 3/4] feat(workflows): sign delivery verification receipts Co-authored-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> Signed-off-by: Logan Johnson --- crates/buzz-core/src/kind.rs | 6 + crates/buzz-core/src/workflow_delivery.rs | 261 +++++++++++++++++- crates/buzz-db/src/lib.rs | 67 +++++ crates/buzz-db/src/migration.rs | 15 +- crates/buzz-db/src/workflow.rs | 98 +++++++ crates/buzz-relay/src/api/bridge.rs | 21 ++ crates/buzz-relay/src/api/workflows.rs | 80 +++++- .../0036_workflow_webhook_invocations.sql | 20 ++ schema/schema.sql | 19 ++ 9 files changed, 583 insertions(+), 4 deletions(-) create mode 100644 migrations/0036_workflow_webhook_invocations.sql diff --git a/crates/buzz-core/src/kind.rs b/crates/buzz-core/src/kind.rs index 2a5145ba2a..35c5c2a906 100644 --- a/crates/buzz-core/src/kind.rs +++ b/crates/buzz-core/src/kind.rs @@ -440,6 +440,11 @@ pub const KIND_WINDOW_BOUNDS: u32 = 39006; /// Ephemeral, relay-authored hint that a durable workflow delivery is ready. pub const KIND_WORKFLOW_AGENT_WAKE: u32 = 24620; +/// Relay-signed proof binding a durable workflow delivery to its public inputs. +/// +/// Returned by the authenticated delivery API; not accepted as client-authored +/// state and not stored in the event log. +pub const KIND_WORKFLOW_DELIVERY_RECEIPT: u32 = 24621; /// Workflow definition (parameterized replaceable, d=workflow_uuid). pub const KIND_WORKFLOW_DEF: u32 = 30620; @@ -696,6 +701,7 @@ pub const ALL_KINDS: &[u32] = &[ KIND_THREAD_SUMMARY, KIND_WINDOW_BOUNDS, KIND_WORKFLOW_AGENT_WAKE, + KIND_WORKFLOW_DELIVERY_RECEIPT, KIND_PRESENCE_UPDATE, KIND_TYPING_INDICATOR, KIND_HUDDLE_REACTION, diff --git a/crates/buzz-core/src/workflow_delivery.rs b/crates/buzz-core/src/workflow_delivery.rs index 3339038e9d..8e061af8c0 100644 --- a/crates/buzz-core/src/workflow_delivery.rs +++ b/crates/buzz-core/src/workflow_delivery.rs @@ -6,11 +6,15 @@ use std::fmt; -use nostr::{Event, EventBuilder, EventId, Kind, PublicKey, Tag}; +use nostr::{Event, EventBuilder, EventId, Keys, Kind, PublicKey, Tag}; +use sha2::{Digest, Sha256}; use thiserror::Error; use uuid::Uuid; -use crate::{kind::KIND_WORKFLOW_AGENT_WAKE, tenant::CommunityId}; +use crate::{ + kind::{KIND_WORKFLOW_AGENT_WAKE, KIND_WORKFLOW_DELIVERY_RECEIPT}, + tenant::CommunityId, +}; /// The target class admitted by the durable workflow delivery protocol. pub const WORKFLOW_DELIVERY_TARGET: &str = "message-v1"; @@ -201,6 +205,135 @@ pub fn message_v1_targets(event: &Event) -> Result, WorkflowDeliv Ok(targets) } +/// Relay-signed proof that the relay produced one public workflow delivery. +/// +/// The receipt contains references and hashes only. Trigger context, webhook +/// fields, step outputs, execution traces, and secrets are deliberately absent. +/// Its timestamp is copied from the visible message, making signing +/// deterministic for one binding/message/key tuple. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WorkflowDeliveryReceipt { + delivery_id: WorkflowDeliveryId, + binding: WorkflowDeliveryBinding, + message_content_sha256: [u8; 32], +} + +impl WorkflowDeliveryReceipt { + /// Construct a receipt after validating the visible event against the binding. + pub fn new( + delivery_id: WorkflowDeliveryId, + binding: WorkflowDeliveryBinding, + message: &Event, + ) -> Result { + binding.validate_message_event(message)?; + Ok(Self { + delivery_id, + binding, + message_content_sha256: Sha256::digest(message.content.as_bytes()).into(), + }) + } + + /// Return the durable delivery identity. + pub const fn delivery_id(&self) -> WorkflowDeliveryId { + self.delivery_id + } + + /// Return the immutable delivery binding proved by this receipt. + pub const fn binding(&self) -> &WorkflowDeliveryBinding { + &self.binding + } + + /// Return the SHA-256 identity of the visible message content. + pub const fn message_content_sha256(&self) -> [u8; 32] { + self.message_content_sha256 + } + + /// Sign the canonical receipt with the relay identity. + pub fn sign(&self, relay_keys: &Keys, message: &Event) -> Result { + self.binding.validate_message_event(message)?; + if <[u8; 32]>::from(Sha256::digest(message.content.as_bytes())) + != self.message_content_sha256 + { + return Err(WorkflowDeliveryError::MessageContentHashMismatch); + } + EventBuilder::new(Kind::Custom(KIND_WORKFLOW_DELIVERY_RECEIPT as u16), "") + .tags(self.canonical_tags()?) + .custom_created_at(message.created_at) + .sign_with_keys(relay_keys) + .map_err(|error| WorkflowDeliveryError::ReceiptSigning(error.to_string())) + } + + /// Verify a receipt's relay signature, canonical shape, binding, and public message. + pub fn verify( + event: &Event, + relay_pubkey: PublicKey, + delivery_id: WorkflowDeliveryId, + binding: &WorkflowDeliveryBinding, + message: &Event, + ) -> Result { + if event.kind.as_u16() != KIND_WORKFLOW_DELIVERY_RECEIPT as u16 { + return Err(WorkflowDeliveryError::WrongReceiptKind(event.kind.as_u16())); + } + if event.pubkey != relay_pubkey || event.verify().is_err() { + return Err(WorkflowDeliveryError::InvalidReceiptSignature); + } + if !event.content.is_empty() { + return Err(WorkflowDeliveryError::ReceiptHasContent); + } + binding.validate_message_event(message)?; + if event.created_at != message.created_at { + return Err(WorkflowDeliveryError::NonCanonicalReceipt); + } + let expected = Self::new(delivery_id, binding.clone(), message)?; + let expected_tags = expected.canonical_tags()?; + if event.tags.as_slice() != expected_tags.as_slice() { + return Err(WorkflowDeliveryError::ReceiptBindingMismatch); + } + Ok(expected) + } + + fn canonical_tags(&self) -> Result, WorkflowDeliveryError> { + let mut tags = vec![ + parse_tag(["delivery", &self.delivery_id.to_string()])?, + parse_tag(["community", &self.binding.community_id().to_string()])?, + parse_tag(["workflow", &self.binding.workflow_id().to_string()])?, + parse_tag(["run", &self.binding.run_id().to_string()])?, + parse_tag(["step", self.binding.step_id()])?, + parse_tag(["p", &self.binding.target_pubkey().to_hex()])?, + parse_tag(["definition", &self.binding.definition_event_id().to_hex()])?, + parse_tag([ + "message", + &self.binding.message_event_id().to_hex(), + &sha256_hex(&self.message_content_sha256), + ])?, + ]; + tags.push(cause_tag(self.binding.cause())?); + Ok(tags) + } +} + +fn sha256_hex(digest: &[u8; 32]) -> String { + use std::fmt::Write as _; + + let mut encoded = String::with_capacity(64); + for byte in digest { + let _ = write!(encoded, "{byte:02x}"); + } + encoded +} + +fn cause_tag(cause: &WorkflowDeliveryCause) -> Result { + match cause { + WorkflowDeliveryCause::Event(event_id) => parse_tag(["cause", "event", &event_id.to_hex()]), + WorkflowDeliveryCause::Schedule { + scheduled_for_unix_seconds, + } => parse_tag(["cause", "schedule", &scheduled_for_unix_seconds.to_string()]), + WorkflowDeliveryCause::Webhook { invocation_id } => { + parse_tag(["cause", "webhook", &invocation_id.to_string()]) + } + } +} + /// An ephemeral identifier-only wake hint for a durable delivery. #[derive(Debug, Clone, PartialEq, Eq)] pub struct WorkflowDeliveryWake { @@ -300,6 +433,27 @@ pub enum WorkflowDeliveryError { /// A `message-v1` recipient appeared more than once. #[error("workflow delivery message has duplicate message-v1 target")] DuplicateMessageV1Target, + /// A receipt was not kind 24621. + #[error("workflow delivery receipt must be kind 24621, got {0}")] + WrongReceiptKind(u16), + /// A receipt was not signed by the expected relay or had an invalid signature. + #[error("workflow delivery receipt signature is invalid")] + InvalidReceiptSignature, + /// A receipt carried content rather than references only. + #[error("workflow delivery receipt content must be empty")] + ReceiptHasContent, + /// A receipt did not use the message timestamp. + #[error("workflow delivery receipt is not canonical")] + NonCanonicalReceipt, + /// A receipt's exact canonical tags disagreed with the expected binding. + #[error("workflow delivery receipt binding does not match")] + ReceiptBindingMismatch, + /// The visible message content did not match the receipt hash. + #[error("workflow delivery receipt message content hash does not match")] + MessageContentHashMismatch, + /// Nostr could not sign the receipt. + #[error("workflow delivery receipt signing failed: {0}")] + ReceiptSigning(String), /// A wake was not kind 24620. #[error("workflow delivery wake must be kind 24620, got {0}")] WrongWakeKind(u16), @@ -643,6 +797,109 @@ mod tests { ); } + #[test] + fn receipt_binds_every_identity_and_contains_no_private_inputs() { + let relay = Keys::generate(); + let original = binding(); + let message = EventBuilder::new(Kind::Custom(9), "rendered secret result") + .tags([original.message_v1_target_tag().unwrap()]) + .sign_with_keys(&relay) + .unwrap(); + let binding = WorkflowDeliveryBinding::new( + original.community_id(), + original.workflow_id(), + original.run_id(), + original.step_id(), + original.target_pubkey(), + original.definition_event_id(), + message.id, + original.cause().clone(), + ) + .unwrap(); + let delivery_id = WorkflowDeliveryId::from_uuid(Uuid::new_v4()); + let receipt = WorkflowDeliveryReceipt::new(delivery_id, binding.clone(), &message).unwrap(); + let signed = receipt.sign(&relay, &message).unwrap(); + + assert!(signed.content.is_empty()); + let wire = serde_json::to_string(&signed).unwrap(); + assert!(!wire.contains("rendered secret result")); + assert!(!wire.contains("trigger_context")); + assert!(!wire.contains("execution_trace")); + assert_eq!( + WorkflowDeliveryReceipt::verify( + &signed, + relay.public_key(), + delivery_id, + &binding, + &message, + ) + .unwrap(), + receipt + ); + + let other_target = Keys::generate().public_key(); + let tampered = WorkflowDeliveryBinding::new( + binding.community_id(), + binding.workflow_id(), + binding.run_id(), + binding.step_id(), + other_target, + binding.definition_event_id(), + binding.message_event_id(), + binding.cause().clone(), + ) + .unwrap(); + assert_eq!( + WorkflowDeliveryReceipt::verify( + &signed, + relay.public_key(), + delivery_id, + &tampered, + &message, + ), + Err(WorkflowDeliveryError::MissingMessageV1Target) + ); + assert_eq!( + WorkflowDeliveryReceipt::verify( + &signed, + Keys::generate().public_key(), + delivery_id, + &binding, + &message, + ), + Err(WorkflowDeliveryError::InvalidReceiptSignature) + ); + } + + #[test] + fn receipt_rejects_message_content_and_signature_tampering() { + let relay = Keys::generate(); + let original = binding(); + let message = EventBuilder::new(Kind::Custom(9), "visible") + .tags([original.message_v1_target_tag().unwrap()]) + .sign_with_keys(&relay) + .unwrap(); + let binding = WorkflowDeliveryBinding::new( + original.community_id(), + original.workflow_id(), + original.run_id(), + original.step_id(), + original.target_pubkey(), + original.definition_event_id(), + message.id, + original.cause().clone(), + ) + .unwrap(); + let id = WorkflowDeliveryId::from_uuid(Uuid::new_v4()); + let receipt = WorkflowDeliveryReceipt::new(id, binding.clone(), &message).unwrap(); + let mut signed = receipt.sign(&relay, &message).unwrap(); + signed.content = "private input".to_owned(); + assert_eq!( + WorkflowDeliveryReceipt::verify(&signed, relay.public_key(), id, &binding, &message), + Err(WorkflowDeliveryError::InvalidReceiptSignature) + ); + } + #[test] fn wake_round_trips_and_contains_no_authority_beyond_identifier_and_target() { let target = Keys::generate().public_key(); diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index d477a7ece2..e0a1a0279e 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -3437,6 +3437,23 @@ impl Db { workflow::latest_scheduled_workflow_fire(&self.pool, community_id, workflow_id).await } + /// Read the run linked by one durable scheduled-fire claim. + #[datastore_span(name = "get_scheduled_workflow_fire_run", system = "postgresql")] + pub async fn get_scheduled_workflow_fire_run( + &self, + community_id: CommunityId, + workflow_id: Uuid, + scheduled_for: chrono::DateTime, + ) -> Result> { + workflow::get_scheduled_workflow_fire_run( + &self.pool, + community_id, + workflow_id, + scheduled_for, + ) + .await + } + /// Attach the workflow run id created from a won scheduled-fire claim. #[datastore_span(name = "attach_scheduled_workflow_run", system = "postgresql")] pub async fn attach_scheduled_workflow_run( @@ -3456,6 +3473,56 @@ impl Db { .await } + /// Persist a payload-free webhook invocation authority. + pub async fn create_workflow_webhook_invocation( + &self, + community_id: CommunityId, + workflow_id: Uuid, + invocation_id: Uuid, + ) -> Result<()> { + workflow::create_workflow_webhook_invocation( + &self.pool, + community_id, + workflow_id, + invocation_id, + ) + .await + } + + /// Link a webhook invocation authority to the run it caused. + pub async fn attach_workflow_webhook_invocation_run( + &self, + community_id: CommunityId, + workflow_id: Uuid, + invocation_id: Uuid, + workflow_run_id: Uuid, + ) -> Result { + workflow::attach_workflow_webhook_invocation_run( + &self.pool, + community_id, + workflow_id, + invocation_id, + workflow_run_id, + ) + .await + } + + /// Read the run linked by one opaque webhook invocation. + pub async fn get_workflow_webhook_invocation_run( + &self, + community_id: CommunityId, + workflow_id: Uuid, + invocation_id: Uuid, + ) -> Result> { + workflow::get_workflow_webhook_invocation_run( + &self.pool, + community_id, + workflow_id, + invocation_id, + ) + .await + } + /// Delete old scheduled workflow fire claims before a retention cutoff. #[datastore_span(name = "prune_scheduled_workflow_fires_before", system = "postgresql")] pub async fn prune_scheduled_workflow_fires_before( diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index e5fd8ca7fb..41acbe980e 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -645,7 +645,7 @@ mod tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 35); + assert_eq!(migrations.len(), 36); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] @@ -1106,6 +1106,16 @@ mod tests { desired_schema.contains("attach_community_write_fence('workflow_agent_deliveries')") ); + // Payload-free durable authority binds each opaque webhook invocation + // to its tenant, workflow, and eventual run. + assert_eq!(migrations[35].version, 36); + let webhook_invocations = migrations[35].sql.as_str(); + assert!(webhook_invocations.contains("CREATE TABLE workflow_webhook_invocations")); + assert!(webhook_invocations.contains("PRIMARY KEY (community_id, invocation_id)")); + assert!(webhook_invocations + .contains("attach_community_write_fence('workflow_webhook_invocations')")); + assert!(desired_schema.contains("CREATE TABLE workflow_webhook_invocations")); + // Fresh desired-state bootstrap must install the identical executable // fence as migration 0032. CI and isolated relay startup use schema.sql // without running migrations, so drift reopens rolling-deploy races. @@ -1584,6 +1594,9 @@ mod tests { // Migration 0035 attaches the fence to the durable managed-agent // delivery inbox after 0029; the desired-state schema carries it inline. expected_fences.insert("workflow_agent_deliveries".to_owned()); + // Migration 0036 likewise adds the payload-free webhook invocation + // authority after 0029. + expected_fences.insert("workflow_webhook_invocations".to_owned()); assert_eq!( expected_fences, schema.fence_attachments, "write-fence attachment targets differ after recovery policy" diff --git a/crates/buzz-db/src/workflow.rs b/crates/buzz-db/src/workflow.rs index 32461830f8..27b959dc2a 100644 --- a/crates/buzz-db/src/workflow.rs +++ b/crates/buzz-db/src/workflow.rs @@ -599,6 +599,32 @@ pub async fn latest_scheduled_workflow_fire( row.try_get("scheduled_for").map_err(Into::into) } +/// Read the durable authority binding for one scheduled workflow fire. +/// +/// Returns the linked run only from the tenant/workflow/slot primary key; a +/// missing or not-yet-attached claim cannot authorize delivery. +pub async fn get_scheduled_workflow_fire_run( + pool: &PgPool, + community_id: CommunityId, + workflow_id: Uuid, + scheduled_for: DateTime, +) -> Result> { + sqlx::query_scalar( + r#" + SELECT workflow_run_id + FROM scheduled_workflow_fires + WHERE community_id = $1 AND workflow_id = $2 AND scheduled_for = $3 + "#, + ) + .bind(community_id.as_uuid()) + .bind(workflow_id) + .bind(scheduled_for) + .fetch_optional(pool) + .await + .map(|value| value.flatten()) + .map_err(Into::into) +} + /// Link a won scheduled-fire claim to the workflow run it created. /// /// This is for ops/audit forensics only; the claim row remains the dedupe @@ -632,6 +658,78 @@ pub async fn attach_scheduled_workflow_run( Ok(result.rows_affected() == 1) } +/// Persist a payload-free webhook invocation authority before run creation. +pub async fn create_workflow_webhook_invocation( + pool: &PgPool, + community_id: CommunityId, + workflow_id: Uuid, + invocation_id: Uuid, +) -> Result<()> { + sqlx::query( + r#" + INSERT INTO workflow_webhook_invocations + (community_id, invocation_id, workflow_id) + SELECT community_id, $3, id + FROM workflows + WHERE community_id = $1 AND id = $2 + "#, + ) + .bind(community_id.as_uuid()) + .bind(workflow_id) + .bind(invocation_id) + .execute(pool) + .await?; + Ok(()) +} + +/// Link a durable webhook invocation to the run it caused. +pub async fn attach_workflow_webhook_invocation_run( + pool: &PgPool, + community_id: CommunityId, + workflow_id: Uuid, + invocation_id: Uuid, + workflow_run_id: Uuid, +) -> Result { + let result = sqlx::query( + r#" + UPDATE workflow_webhook_invocations + SET workflow_run_id = $4 + WHERE community_id = $1 AND workflow_id = $2 AND invocation_id = $3 + AND workflow_run_id IS NULL + "#, + ) + .bind(community_id.as_uuid()) + .bind(workflow_id) + .bind(invocation_id) + .bind(workflow_run_id) + .execute(pool) + .await?; + Ok(result.rows_affected() == 1) +} + +/// Read a webhook invocation's durable tenant/workflow/run binding. +pub async fn get_workflow_webhook_invocation_run( + pool: &PgPool, + community_id: CommunityId, + workflow_id: Uuid, + invocation_id: Uuid, +) -> Result> { + sqlx::query_scalar( + r#" + SELECT workflow_run_id + FROM workflow_webhook_invocations + WHERE community_id = $1 AND workflow_id = $2 AND invocation_id = $3 + "#, + ) + .bind(community_id.as_uuid()) + .bind(workflow_id) + .bind(invocation_id) + .fetch_optional(pool) + .await + .map(|value| value.flatten()) + .map_err(Into::into) +} + /// Delete old scheduled workflow fire claims for retention. /// /// Schedule claim rows are correctness metadata, but they grow with every fire. diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 50f6a74c10..d18c443303 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -2111,6 +2111,16 @@ pub async fn workflow_webhook( .await .map_err(|_| not_found("workflow not found"))?; + // Mint durable opaque authority before run creation. The record contains + // no request body or secret; a NULL run link remains honest if insertion + // of the run later fails. + let webhook_invocation_id = uuid::Uuid::new_v4(); + state + .db + .create_workflow_webhook_invocation(community_id, id, webhook_invocation_id) + .await + .map_err(|e| super::internal_error(&format!("persist webhook invocation: {e}")))?; + let definition_event_id = workflow .definition_event_id .as_deref() @@ -2127,6 +2137,17 @@ pub async fn workflow_webhook( .await .map_err(|e| super::internal_error(&format!("db error: {e}")))?; + let attached = state + .db + .attach_workflow_webhook_invocation_run(community_id, id, webhook_invocation_id, run_id) + .await + .map_err(|e| super::internal_error(&format!("attach webhook invocation: {e}")))?; + if !attached { + return Err(super::internal_error( + "webhook invocation authority was not attachable", + )); + } + // Spawn workflow execution asynchronously. let engine = Arc::clone(&state.workflow_engine); let trigger_ctx_clone = trigger_ctx.clone(); diff --git a/crates/buzz-relay/src/api/workflows.rs b/crates/buzz-relay/src/api/workflows.rs index e50cc3f600..a4130c52d5 100644 --- a/crates/buzz-relay/src/api/workflows.rs +++ b/crates/buzz-relay/src/api/workflows.rs @@ -17,7 +17,9 @@ use serde_json::Value; use uuid::Uuid; use buzz_core::{ - workflow_delivery::{WorkflowDeliveryBinding, WorkflowDeliveryCause, WorkflowDeliveryId}, + workflow_delivery::{ + WorkflowDeliveryBinding, WorkflowDeliveryCause, WorkflowDeliveryId, WorkflowDeliveryReceipt, + }, TenantContext, }; @@ -700,6 +702,66 @@ fn terminal_matches( ) } +async fn validate_delivery_cause_authority( + state: &Arc, + tenant: &TenantContext, + binding: &WorkflowDeliveryBinding, +) -> Result<(), (StatusCode, Json)> { + let authoritative_run = match binding.cause() { + WorkflowDeliveryCause::Event(event_id) => { + let event = state + .db + .get_event_by_id(tenant.community(), event_id.as_bytes()) + .await + .map_err(|error| internal_error(&format!("delivery cause lookup: {error}")))? + .ok_or_else(|| { + api_error(StatusCode::CONFLICT, "delivery authority is unavailable") + })?; + if event.event.verify().is_err() { + return Err(api_error( + StatusCode::CONFLICT, + "delivery authority is unavailable", + )); + } + return Ok(()); + } + WorkflowDeliveryCause::Schedule { + scheduled_for_unix_seconds, + } => { + let scheduled_for = + chrono::DateTime::::from_timestamp(*scheduled_for_unix_seconds, 0) + .ok_or_else(|| { + api_error(StatusCode::CONFLICT, "delivery authority is unavailable") + })?; + state + .db + .get_scheduled_workflow_fire_run( + tenant.community(), + binding.workflow_id(), + scheduled_for, + ) + .await + .map_err(|error| internal_error(&format!("schedule authority lookup: {error}")))? + } + WorkflowDeliveryCause::Webhook { invocation_id } => state + .db + .get_workflow_webhook_invocation_run( + tenant.community(), + binding.workflow_id(), + *invocation_id, + ) + .await + .map_err(|error| internal_error(&format!("webhook authority lookup: {error}")))?, + }; + if authoritative_run != Some(binding.run_id()) { + return Err(api_error( + StatusCode::CONFLICT, + "delivery authority is unavailable", + )); + } + Ok(()) +} + async fn delivery_response( state: &Arc, tenant: &TenantContext, @@ -746,10 +808,16 @@ async fn delivery_response( .await .map_err(|error| internal_error(&format!("delivery message lookup: {error}")))? .ok_or_else(|| api_error(StatusCode::CONFLICT, "delivery binding is unavailable"))?; + validate_delivery_cause_authority(state, tenant, &delivery.binding).await?; + let receipt = + WorkflowDeliveryReceipt::new(delivery.id, delivery.binding.clone(), &message.event) + .and_then(|receipt| receipt.sign(&state.relay_keypair, &message.event)) + .map_err(|error| internal_error(&format!("sign delivery receipt: {error}")))?; Ok(Json(serde_json::json!({ "delivery": delivery_json(delivery, lease), "definition_event": definition.event, "message_event": message.event, + "receipt_event": receipt, }))) } @@ -1154,6 +1222,16 @@ mod tests { "private execution input" ); assert!(claimed.get("execution_trace").is_none()); + assert!(claimed.get("trigger_context").is_none()); + assert!(claimed.get("webhook_fields").is_none()); + let receipt: nostr::Event = + serde_json::from_value(claimed["receipt_event"].clone()).expect("signed receipt event"); + assert_eq!(receipt.pubkey, fixture.state.relay_keypair.public_key()); + assert!(receipt.verify().is_ok()); + assert!(receipt.content.is_empty()); + assert!(!claimed["receipt_event"] + .to_string() + .contains("private execution input")); let lease = serde_json::json!({ "lease_generation": 1, diff --git a/migrations/0036_workflow_webhook_invocations.sql b/migrations/0036_workflow_webhook_invocations.sql new file mode 100644 index 0000000000..650ca8c154 --- /dev/null +++ b/migrations/0036_workflow_webhook_invocations.sql @@ -0,0 +1,20 @@ +-- Durable, payload-free authority for webhook-triggered workflow runs. +-- The body and secret stay in transient trigger context; this table retains +-- only an opaque invocation identity and its tenant/workflow/run binding. +CREATE TABLE workflow_webhook_invocations ( + community_id UUID NOT NULL REFERENCES communities(id), + invocation_id UUID NOT NULL, + workflow_id UUID NOT NULL, + workflow_run_id UUID, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (community_id, invocation_id), + FOREIGN KEY (community_id, workflow_id) + REFERENCES workflows (community_id, id) ON DELETE CASCADE, + FOREIGN KEY (community_id, workflow_run_id) + REFERENCES workflow_runs (community_id, id) ON DELETE NO ACTION +); + +CREATE INDEX idx_workflow_webhook_invocations_created_at + ON workflow_webhook_invocations (created_at); + +SELECT attach_community_write_fence('workflow_webhook_invocations'); diff --git a/schema/schema.sql b/schema/schema.sql index ba897133cf..a91bcd9c43 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -536,6 +536,24 @@ CREATE INDEX idx_workflow_agent_deliveries_lease ON workflow_agent_deliveries (lease_until) WHERE status = 'claimed'; +-- ── Workflow webhook invocations ───────────────────────────────────────────── +-- Payload-free durable authority linking an opaque webhook call to its run. +CREATE TABLE workflow_webhook_invocations ( + community_id UUID NOT NULL REFERENCES communities(id), + invocation_id UUID NOT NULL, + workflow_id UUID NOT NULL, + workflow_run_id UUID, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (community_id, invocation_id), + FOREIGN KEY (community_id, workflow_id) + REFERENCES workflows (community_id, id) ON DELETE CASCADE, + FOREIGN KEY (community_id, workflow_run_id) + REFERENCES workflow_runs (community_id, id) ON DELETE NO ACTION +); + +CREATE INDEX idx_workflow_webhook_invocations_created_at + ON workflow_webhook_invocations (created_at); + -- ── API tokens ──────────────────────────────────────────────────────────────── -- Conformance: "API tokens and NIP-98 replay". token_hash uniqueness scoped to -- (community_id, token_hash); channel claims reference channels in same community. @@ -1810,6 +1828,7 @@ SELECT attach_community_write_fence('reactions'); SELECT attach_community_write_fence('relay_invites'); SELECT attach_community_write_fence('relay_members'); SELECT attach_community_write_fence('scheduled_workflow_fires'); +SELECT attach_community_write_fence('workflow_webhook_invocations'); SELECT attach_community_write_fence('subscriptions'); SELECT attach_community_write_fence('thread_metadata'); SELECT attach_community_write_fence('users'); From 813cd532cc50a622d2f1ccffc22cd132cdfa268c Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Wed, 26 Aug 2026 16:30:15 -0400 Subject: [PATCH 4/4] fix(buzz-db): catalog workflow webhook authority Co-authored-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> Signed-off-by: Logan Johnson --- crates/buzz-db/src/deletion.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/crates/buzz-db/src/deletion.rs b/crates/buzz-db/src/deletion.rs index 8816c91f59..ab2e32d710 100644 --- a/crates/buzz-db/src/deletion.rs +++ b/crates/buzz-db/src/deletion.rs @@ -82,12 +82,14 @@ pub const EXPECTED_SCOPED_TABLES: &[&str] = &[ "workflow_agent_deliveries", "workflow_approvals", "workflow_runs", + "workflow_webhook_invocations", "workflows", ]; /// Foreign-key-safe child-before-parent order for the PostgreSQL purge. pub const PURGE_SCOPED_TABLES: &[&str] = &[ "workflow_agent_deliveries", + "workflow_webhook_invocations", "workflow_approvals", "scheduled_workflow_fires", "workflow_runs",