From 25fd7567c005269ba400bb7a33e9fa05e9bc78e5 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Mon, 24 Aug 2026 19:15:53 -0400 Subject: [PATCH] wip(workflows): preserve durable inbox foundation Checkpoint the Node B persistence and provenance boundary before producer and API integration. Co-authored-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> Signed-off-by: loganj --- crates/buzz-core/src/kind.rs | 12 +- crates/buzz-db/src/lib.rs | 80 +++++++ crates/buzz-db/src/workflow.rs | 226 +++++++++++++++++- crates/buzz-relay/src/workflow_sink.rs | 6 +- crates/buzz-workflow/src/action_sink.rs | 17 ++ crates/buzz-workflow/src/executor.rs | 13 + migrations/0034_workflow_agent_deliveries.sql | 33 +++ schema/schema.sql | 31 +++ 8 files changed, 413 insertions(+), 5 deletions(-) create mode 100644 migrations/0034_workflow_agent_deliveries.sql diff --git a/crates/buzz-core/src/kind.rs b/crates/buzz-core/src/kind.rs index 3c6f1d5913..06fc320ac0 100644 --- a/crates/buzz-core/src/kind.rs +++ b/crates/buzz-core/src/kind.rs @@ -139,7 +139,11 @@ pub const AUTHOR_ONLY_KINDS: &[u32] = &[ /// /// Used by `filter_can_match_result_gated_kinds` to force the per-event /// fallback path in COUNT rather than the fast SQL `count_events()`. -pub const RESULT_GATED_KINDS: &[u32] = &[KIND_DM_VISIBILITY, KIND_AGENT_TURN_METRIC]; +pub const RESULT_GATED_KINDS: &[u32] = &[ + KIND_DM_VISIBILITY, + KIND_AGENT_TURN_METRIC, + KIND_WORKFLOW_AGENT_WAKE, +]; /// Kinds whose stored events have `#p`-bound read access — readable only by /// subscribers whose pubkey appears in the event's `#p` tag. @@ -158,6 +162,7 @@ pub const RESULT_GATED_KINDS: &[u32] = &[KIND_DM_VISIBILITY, KIND_AGENT_TURN_MET /// storage-layer search defense does not apply to them. pub const P_GATED_KINDS: &[u32] = &[ KIND_AGENT_OBSERVER_FRAME, + KIND_WORKFLOW_AGENT_WAKE, KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, KIND_GIFT_WRAP, @@ -439,6 +444,9 @@ pub const KIND_THREAD_SUMMARY: u32 = 39005; pub const KIND_WINDOW_BOUNDS: u32 = 39006; /// Workflow definition (parameterized replaceable, d=workflow_uuid). +/// Ephemeral, relay-authored hint that a durable workflow delivery is ready. +pub const KIND_WORKFLOW_AGENT_WAKE: u32 = 24620; + pub const KIND_WORKFLOW_DEF: u32 = 30620; /// NIP-DV: per-viewer DM visibility snapshot (relay-signed, parameterized @@ -698,6 +706,7 @@ pub const ALL_KINDS: &[u32] = &[ KIND_BLOSSOM_AUTH, KIND_PAIRING, KIND_AGENT_OBSERVER_FRAME, + KIND_WORKFLOW_AGENT_WAKE, KIND_HTTP_AUTH, KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_V2, @@ -836,6 +845,7 @@ pub const fn is_relay_only_kind(kind: u32) -> bool { | KIND_DM_VISIBILITY | KIND_THREAD_SUMMARY | KIND_WINDOW_BOUNDS + | KIND_WORKFLOW_AGENT_WAKE ) } diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 9c6647983b..1c1a879845 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -4164,6 +4164,86 @@ impl Db { .await } + /// Serialize one workflow message identity until its event and inbox rows commit. + pub async fn lock_workflow_agent_delivery_identity( + &self, + community_id: CommunityId, + run_id: Uuid, + step_id: &str, + ) -> Result<( + sqlx::Transaction<'static, sqlx::Postgres>, + Option<(Vec, chrono::DateTime)>, + )> { + workflow::lock_workflow_agent_delivery_identity(&self.pool, community_id, run_id, step_id) + .await + } + + /// Atomically commit one visible workflow message and its durable targets. + #[allow(clippy::too_many_arguments)] + pub async fn commit_workflow_agent_deliveries( + &self, + transaction: sqlx::Transaction<'static, sqlx::Postgres>, + community_id: CommunityId, + event: Option<&nostr::Event>, + message_event_id: &[u8], + message_event_created_at: chrono::DateTime, + thread_meta: Option>, + workflow_id: Uuid, + run_id: Uuid, + step_id: &str, + definition_event_id: &[u8], + channel_id: Uuid, + targets: &[workflow::WorkflowAgentDeliveryTarget], + execution_trace: &serde_json::Value, + trigger_context: Option<&serde_json::Value>, + ) -> Result<(Option, Vec)> { + let result = workflow::commit_workflow_agent_deliveries( + transaction, + community_id, + event, + message_event_id, + message_event_created_at, + thread_meta, + workflow_id, + run_id, + step_id, + definition_event_id, + channel_id, + targets, + execution_trace, + trigger_context, + ) + .await?; + if result.0.is_some() { + if let Some(event) = event { + if let Err(error) = + insert_mentions(&self.pool, community_id, event, Some(channel_id)).await + { + tracing::warn!(event_id = %event.id, %error, "failed to insert workflow message mentions"); + } + } + } + Ok(result) + } + + /// Claim one durable workflow delivery for the authenticated target. + pub async fn claim_workflow_agent_delivery( + &self, + community_id: CommunityId, + target_pubkey: &[u8], + delivery_id: Option, + expected: Option<&workflow::WorkflowAgentDeliveryBinding>, + ) -> Result> { + workflow::claim_workflow_agent_delivery( + &self.pool, + community_id, + target_pubkey, + delivery_id, + expected, + ) + .await + } + /// Update a workflow run's status. #[datastore_span(name = "update_workflow_run", system = "postgresql")] pub async fn update_workflow_run( diff --git a/crates/buzz-db/src/workflow.rs b/crates/buzz-db/src/workflow.rs index f72d076c0a..d4bebfb8fe 100644 --- a/crates/buzz-db/src/workflow.rs +++ b/crates/buzz-db/src/workflow.rs @@ -11,13 +11,15 @@ use std::fmt; use std::str::FromStr; use chrono::{DateTime, Utc}; +use nostr::Event; use sha2::{Digest, Sha256}; -use sqlx::{PgPool, Row}; +use sqlx::{PgPool, Postgres, Row, Transaction}; use uuid::Uuid; -use buzz_core::CommunityId; +use buzz_core::{CommunityId, StoredEvent}; use crate::error::{DbError, Result}; +use crate::event::ThreadMetadataParams; // -- Token hashing ------------------------------------------------------------ @@ -227,6 +229,59 @@ pub struct WorkflowRunRecord { pub created_at: DateTime, } +/// One immutable target of a workflow-generated visible message. +#[derive(Debug, Clone)] +pub struct WorkflowAgentDeliveryTarget { + /// Stable identifier carried by the wake hint and claim request. + pub id: Uuid, + /// Managed-agent pubkey. + pub pubkey: Vec, +} + +/// Immutable binding optionally supplied from an authenticated wake. +#[derive(Debug, Clone)] +pub struct WorkflowAgentDeliveryBinding { + /// Durable workflow run. + pub run_id: Uuid, + /// Stable workflow step. + pub step_id: String, + /// Exact owner-signed definition revision. + pub definition_event_id: Vec, + /// Relay-signed visible message. + pub message_event_id: Vec, + /// Destination channel. + pub channel_id: Uuid, +} + +/// Durable inbox item returned only to its authenticated target. +#[derive(Debug, Clone)] +pub struct WorkflowAgentDeliveryRecord { + /// Delivery identifier. + pub id: Uuid, + /// Owning workflow. + pub workflow_id: Uuid, + /// Durable run. + pub run_id: Uuid, + /// Stable step identifier. + pub step_id: String, + /// Exact signed definition revision. + pub definition_event_id: Vec, + /// Relay-signed visible message. + pub message_event_id: Vec, + /// Destination channel. + pub channel_id: Uuid, + /// Managed-agent recipient. + pub target_pubkey: Vec, + /// One-shot claim token reserved for the completion child node. + pub claim_token: Option, + /// Private execution trace used by ACP verification. + pub execution_trace: serde_json::Value, + /// Private trigger snapshot used by ACP verification. + pub trigger_context: Option, + /// Creation time for ordered polling. + pub created_at: DateTime, +} + /// A winning scheduled workflow fire claim. /// /// The primary identity is `(workflow_id, scheduled_for)`. `community_id` is @@ -1025,6 +1080,173 @@ pub async fn update_workflow_run( Ok(()) } +/// Serialize one `(community, run, step)` delivery identity. +/// +/// The returned transaction holds the advisory lock until the visible message +/// and every target row commit together, preventing retry races from signing +/// two canonical messages for the same step. +pub async fn lock_workflow_agent_delivery_identity( + pool: &PgPool, + community_id: CommunityId, + run_id: Uuid, + step_id: &str, +) -> Result<( + Transaction<'static, Postgres>, + Option<(Vec, DateTime)>, +)> { + let mut transaction = pool.begin().await?; + let identity = format!("{}:{run_id}:{step_id}", community_id.as_uuid()); + sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))") + .bind(identity) + .execute(&mut *transaction) + .await?; + let existing = sqlx::query_as::<_, (Vec, DateTime)>( + "SELECT message_event_id, message_event_created_at \ + FROM workflow_agent_deliveries \ + WHERE community_id = $1 AND run_id = $2 AND step_id = $3 \ + ORDER BY created_at, id LIMIT 1", + ) + .bind(community_id.as_uuid()) + .bind(run_id) + .bind(step_id) + .fetch_optional(&mut *transaction) + .await?; + Ok((transaction, existing)) +} + +/// Atomically persist the canonical visible event and all managed-agent targets. +#[allow(clippy::too_many_arguments)] +pub async fn commit_workflow_agent_deliveries( + mut transaction: Transaction<'static, Postgres>, + community_id: CommunityId, + event: Option<&Event>, + message_event_id: &[u8], + message_event_created_at: DateTime, + thread_meta: Option>, + workflow_id: Uuid, + run_id: Uuid, + step_id: &str, + definition_event_id: &[u8], + channel_id: Uuid, + targets: &[WorkflowAgentDeliveryTarget], + execution_trace: &serde_json::Value, + trigger_context: Option<&serde_json::Value>, +) -> Result<(Option, Vec)> { + let stored_event = if let Some(event) = event { + let (stored, _) = crate::event::insert_event_with_thread_metadata_tx( + &mut transaction, + community_id, + event, + Some(channel_id), + thread_meta, + ) + .await?; + Some(stored) + } else { + None + }; + + let mut created = Vec::new(); + for target in targets { + let affected = sqlx::query( + "INSERT INTO workflow_agent_deliveries \ + (community_id, id, workflow_id, run_id, step_id, definition_event_id, \ + message_event_id, message_event_created_at, channel_id, target_pubkey, \ + execution_trace, trigger_context) \ + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12) \ + ON CONFLICT (community_id, run_id, step_id, target_pubkey) DO NOTHING", + ) + .bind(community_id.as_uuid()) + .bind(target.id) + .bind(workflow_id) + .bind(run_id) + .bind(step_id) + .bind(definition_event_id) + .bind(message_event_id) + .bind(message_event_created_at) + .bind(channel_id) + .bind(&target.pubkey) + .bind(execution_trace) + .bind(trigger_context) + .execute(&mut *transaction) + .await? + .rows_affected(); + if affected == 1 { + created.push(target.id); + } + } + transaction.commit().await?; + Ok((stored_event, created)) +} + +/// Atomically claim one specific delivery, or the oldest pending delivery. +/// +/// Both selection and update are scoped to the authenticated target and +/// community. Optional wake bindings make a forged or stale hint a miss rather +/// than an alternate authority path. +pub async fn claim_workflow_agent_delivery( + pool: &PgPool, + community_id: CommunityId, + target_pubkey: &[u8], + delivery_id: Option, + expected: Option<&WorkflowAgentDeliveryBinding>, +) -> Result> { + let row = sqlx::query( + r#" + WITH candidate AS ( + SELECT community_id, id + FROM workflow_agent_deliveries + WHERE community_id = $1 AND target_pubkey = $2 + AND ($3::uuid IS NULL OR id = $3) + AND ($4::uuid IS NULL OR run_id = $4) + AND ($5::text IS NULL OR step_id = $5) + AND ($6::bytea IS NULL OR definition_event_id = $6) + AND ($7::bytea IS NULL OR message_event_id = $7) + AND ($8::uuid IS NULL OR channel_id = $8) + AND status = 'pending' + ORDER BY created_at, id + FOR UPDATE SKIP LOCKED + LIMIT 1 + ) + UPDATE workflow_agent_deliveries delivery + SET status = 'claimed', claim_token = gen_random_uuid(), + claimed_at = NOW() + FROM candidate + WHERE delivery.community_id = candidate.community_id + AND delivery.id = candidate.id + RETURNING delivery.* + "#, + ) + .bind(community_id.as_uuid()) + .bind(target_pubkey) + .bind(delivery_id) + .bind(expected.map(|binding| binding.run_id)) + .bind(expected.map(|binding| binding.step_id.as_str())) + .bind(expected.map(|binding| binding.definition_event_id.as_slice())) + .bind(expected.map(|binding| binding.message_event_id.as_slice())) + .bind(expected.map(|binding| binding.channel_id)) + .fetch_optional(pool) + .await?; + + row.map(|row| { + Ok(WorkflowAgentDeliveryRecord { + id: row.try_get("id")?, + workflow_id: row.try_get("workflow_id")?, + run_id: row.try_get("run_id")?, + step_id: row.try_get("step_id")?, + definition_event_id: row.try_get("definition_event_id")?, + message_event_id: row.try_get("message_event_id")?, + channel_id: row.try_get("channel_id")?, + target_pubkey: row.try_get("target_pubkey")?, + claim_token: row.try_get("claim_token")?, + execution_trace: row.try_get("execution_trace")?, + trigger_context: row.try_get("trigger_context")?, + created_at: row.try_get("created_at")?, + }) + }) + .transpose() +} + // -- Approval CRUD ------------------------------------------------------------ /// Parameters for creating a new approval request. diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 8ce23a2e8e..c8519bac38 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -8,9 +8,9 @@ use std::future::Future; use std::pin::Pin; use std::sync::{Arc, Weak}; -use buzz_core::kind::KIND_STREAM_MESSAGE; +use buzz_core::kind::{KIND_STREAM_MESSAGE, KIND_WORKFLOW_AGENT_WAKE, KIND_WORKFLOW_DEF}; use buzz_core::tenant::CommunityId; -use buzz_workflow::action_sink::{ActionSink, ActionSinkError}; +use buzz_workflow::action_sink::{ActionSink, ActionSinkError, WorkflowMessageContext}; use chrono::Utc; use nostr::{EventBuilder, Kind, Tag}; use tracing::info; @@ -176,11 +176,13 @@ impl ActionSink for RelayActionSink { channel_id: &str, text: &str, author_pubkey: &str, + context: &WorkflowMessageContext, reply_to: Option<&str>, ) -> Pin> + Send + '_>> { let channel_id = channel_id.to_owned(); let text = text.to_owned(); let author_pubkey = author_pubkey.to_owned(); + let context = context.clone(); let reply_to = reply_to.map(str::to_owned); Box::pin(async move { diff --git a/crates/buzz-workflow/src/action_sink.rs b/crates/buzz-workflow/src/action_sink.rs index 079c27a913..e49433f394 100644 --- a/crates/buzz-workflow/src/action_sink.rs +++ b/crates/buzz-workflow/src/action_sink.rs @@ -7,6 +7,20 @@ use std::future::Future; use std::pin::Pin; use buzz_core::tenant::CommunityId; +use uuid::Uuid; + +/// Immutable provenance for one workflow-generated message delivery. +#[derive(Debug, Clone)] +pub struct WorkflowMessageContext { + /// Workflow whose signed definition authorizes the message. + pub workflow_id: Uuid, + /// Durable run executing the message step. + pub run_id: Uuid, + /// Exact owner-signed kind:30620 definition revision used by the run. + pub definition_event_id: String, + /// Stable step identifier within the signed definition. + pub step_id: String, +} /// Errors from action sink operations. #[derive(Debug, thiserror::Error)] @@ -57,6 +71,8 @@ pub trait ActionSink: Send + Sync { /// - `text`: message body (must not be empty/whitespace-only) /// - `author_pubkey`: hex-encoded pubkey of the workflow owner (used for /// the `p` attribution tag; the relay keypair signs the event) + /// - `context`: immutable workflow/run/revision/step provenance used to + /// bind durable managed-agent deliveries /// - `reply_to`: when `Some(event_id_hex)`, the message is posted as a /// threaded reply to that event (NIP-10 root/reply tags + real thread /// metadata); when `None`, it is a top-level channel message. @@ -68,6 +84,7 @@ pub trait ActionSink: Send + Sync { channel_id: &str, text: &str, author_pubkey: &str, + context: &WorkflowMessageContext, reply_to: Option<&str>, ) -> Pin> + Send + '_>>; } diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index 9aa9d4c3a2..d00d3439c0 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -18,6 +18,7 @@ use serde_json::Value as JsonValue; use tracing::{debug, info, warn}; use uuid::Uuid; +use crate::action_sink::WorkflowMessageContext; use crate::error::WorkflowError; use crate::schema::{ActionDef, Step, WorkflowDef}; use crate::WorkflowEngine; @@ -625,6 +626,17 @@ pub async fn dispatch_action( "SendMessage → {channel_id}: {text}" ); + if trigger_ctx.definition_event_id.is_empty() { + return Err(WorkflowError::InvalidDefinition( + "SendMessage: workflow definition provenance is unavailable".into(), + )); + } + let context = WorkflowMessageContext { + workflow_id: workflow.id, + run_id, + definition_event_id: trigger_ctx.definition_event_id.clone(), + step_id: step_id.to_owned(), + }; let event_id = engine .action_sink()? .send_message( @@ -632,6 +644,7 @@ pub async fn dispatch_action( &channel_id, text, &owner_pubkey_hex, + &context, reply_to, ) .await diff --git a/migrations/0034_workflow_agent_deliveries.sql b/migrations/0034_workflow_agent_deliveries.sql new file mode 100644 index 0000000000..63893af906 --- /dev/null +++ b/migrations/0034_workflow_agent_deliveries.sql @@ -0,0 +1,33 @@ +-- Durable, target-scoped inbox for workflow messages addressed to managed agents. +CREATE TYPE workflow_agent_delivery_status AS ENUM ('pending', 'claimed'); + +CREATE TABLE workflow_agent_deliveries ( + community_id UUID NOT NULL REFERENCES communities(id), + id UUID NOT NULL, + workflow_id UUID NOT NULL, + run_id UUID NOT NULL, + step_id VARCHAR(64) NOT NULL, + definition_event_id BYTEA NOT NULL CHECK (octet_length(definition_event_id) = 32), + message_event_id BYTEA NOT NULL CHECK (octet_length(message_event_id) = 32), + message_event_created_at TIMESTAMPTZ NOT NULL, + channel_id UUID NOT NULL, + target_pubkey BYTEA NOT NULL CHECK (octet_length(target_pubkey) = 32), + status workflow_agent_delivery_status NOT NULL DEFAULT 'pending', + claim_token UUID, + claimed_at TIMESTAMPTZ, + execution_trace JSONB NOT NULL, + trigger_context JSONB, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (community_id, id), + UNIQUE (community_id, run_id, step_id, target_pubkey), + FOREIGN KEY (community_id, workflow_id) REFERENCES workflows (community_id, id) ON DELETE CASCADE, + FOREIGN KEY (community_id, run_id) REFERENCES workflow_runs (community_id, id) ON DELETE CASCADE, + FOREIGN KEY (community_id, message_event_created_at, message_event_id) + REFERENCES events (community_id, created_at, id) ON DELETE CASCADE +); + +CREATE INDEX idx_workflow_agent_deliveries_pending + ON workflow_agent_deliveries (community_id, target_pubkey, created_at) + WHERE status = 'pending'; + +SELECT attach_community_write_fence('workflow_agent_deliveries'); diff --git a/schema/schema.sql b/schema/schema.sql index 734cdb315e..2eab7e137a 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -411,6 +411,36 @@ CREATE TABLE workflow_runs ( CREATE INDEX idx_workflow_runs_workflow ON workflow_runs (community_id, workflow_id); CREATE INDEX idx_workflow_runs_status ON workflow_runs (community_id, status); +-- Durable, target-scoped inbox for workflow messages addressed to managed agents. +CREATE TABLE workflow_agent_deliveries ( + community_id UUID NOT NULL REFERENCES communities(id), + id UUID NOT NULL, + workflow_id UUID NOT NULL, + run_id UUID NOT NULL, + step_id VARCHAR(64) NOT NULL, + definition_event_id BYTEA NOT NULL CHECK (octet_length(definition_event_id) = 32), + message_event_id BYTEA NOT NULL CHECK (octet_length(message_event_id) = 32), + message_event_created_at TIMESTAMPTZ NOT NULL, + channel_id UUID NOT NULL, + target_pubkey BYTEA NOT NULL CHECK (octet_length(target_pubkey) = 32), + status workflow_agent_delivery_status NOT NULL DEFAULT 'pending', + claim_token UUID, + claimed_at TIMESTAMPTZ, + execution_trace JSONB NOT NULL, + trigger_context JSONB, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (community_id, id), + UNIQUE (community_id, run_id, step_id, target_pubkey), + FOREIGN KEY (community_id, workflow_id) REFERENCES workflows (community_id, id) ON DELETE CASCADE, + FOREIGN KEY (community_id, run_id) REFERENCES workflow_runs (community_id, id) ON DELETE CASCADE, + FOREIGN KEY (community_id, message_event_created_at, message_event_id) + REFERENCES events (community_id, created_at, id) ON DELETE CASCADE +); + +CREATE INDEX idx_workflow_agent_deliveries_pending + ON workflow_agent_deliveries (community_id, target_pubkey, created_at) + WHERE status = 'pending'; + -- ── Workflow approvals ──────────────────────────────────────────────────────── -- token-hash lookup scoped: approval token grants cannot act on another -- community's same hash (conformance). @@ -1749,6 +1779,7 @@ SELECT attach_community_write_fence('scheduled_workflow_fires'); SELECT attach_community_write_fence('subscriptions'); SELECT attach_community_write_fence('thread_metadata'); SELECT attach_community_write_fence('users'); +SELECT attach_community_write_fence('workflow_agent_deliveries'); SELECT attach_community_write_fence('workflow_approvals'); SELECT attach_community_write_fence('workflow_runs'); SELECT attach_community_write_fence('workflows');