Bind managed-agent workflow triggers to signed revisions

crates/buzz-cli/src/commands/workflows.rs

@@ -162,6 +162,22 @@ pub async fn cmd_delete_workflow(client: &BuzzClient, workflow_id: &str) -> Resu
     Ok(())
 }
 
+async fn current_workflow_revision(
+    client: &BuzzClient,
+    workflow_id: &str,
+) -> Result<String, CliError> {
+    let response = client
+        .get_authed(&format!("/workflows/{workflow_id}/revision"))
+        .await?;
+    let event: serde_json::Value = serde_json::from_str(&response)
+        .map_err(|e| CliError::Other(format!("invalid workflow revision response: {e}")))?;
+    event
+        .get("id")
+        .and_then(|id| id.as_str())
+        .map(str::to_owned)
+        .ok_or_else(|| CliError::NotFound(format!("workflow {workflow_id} not found")))
+}
+
 /// Trigger a workflow — sign and submit a kind:46020 event.
 ///
 /// When `inputs` is provided, it is parsed as a JSON object and used as the
@@ -172,6 +188,7 @@ pub async fn cmd_trigger_workflow(
     inputs: Option<&str>,
 ) -> Result<(), CliError> {
     let wf_uuid = parse_uuid(workflow_id)?;
+    let revision = current_workflow_revision(client, workflow_id).await?;
 
     if let Some(raw) = inputs {
         // Parse and validate it is a JSON object, then build the event manually
@@ -183,8 +200,12 @@ pub async fn cmd_trigger_workflow(
         }
         let content = serde_json::to_string(&parsed).unwrap_or_default();
         use nostr::{EventBuilder, Kind, Tag};
-        let tags = vec![Tag::parse(["d", &wf_uuid.to_string()])
-            .map_err(|e| CliError::Other(format!("tag error: {e}")))?];
+        let tags = vec![
+            Tag::parse(["d", &wf_uuid.to_string()])
+                .map_err(|e| CliError::Other(format!("tag error: {e}")))?,
+            Tag::parse(["e", revision.as_str()])
+                .map_err(|e| CliError::Other(format!("tag error: {e}")))?,
+        ];
         let builder = EventBuilder::new(
             Kind::Custom(buzz_sdk::kind::KIND_WORKFLOW_TRIGGER as u16),
             &content,
@@ -194,7 +215,7 @@ pub async fn cmd_trigger_workflow(
         let resp = client.submit_event(event).await?;
         println!("{}", normalize_write_response(&resp));
     } else {
-        let builder = buzz_sdk::build_workflow_trigger(wf_uuid).map_err(sdk_err)?;
+        let builder = buzz_sdk::build_workflow_trigger(wf_uuid, &revision).map_err(sdk_err)?;
         let event = client.sign_event(builder)?;
         let resp = client.submit_event(event).await?;
         println!("{}", normalize_write_response(&resp));

crates/buzz-db/src/channel.rs

@@ -473,7 +473,7 @@ pub async fn verify_channel_roster_fence_behavior(pool: &sqlx::PgPool) -> Result
 /// Take the per-channel membership lock. MUST be the first statement in the
 /// transaction that then reads roles/owner counts and writes membership, so the
 /// whole check-then-write sequence is atomic against a concurrent one.
-async fn acquire_channel_membership_lock(
+pub async fn acquire_channel_membership_lock(
     tx: &mut Transaction<'_, Postgres>,
     community_id: CommunityId,
     channel_id: Uuid,
@@ -1872,6 +1872,30 @@ pub async fn get_member_role(
     Ok(row.map(|r| r.try_get("role")).transpose()?)
 }
 
+/// Get an active member role using the caller's transaction.
+///
+/// Callers that authorize a workflow run must first acquire
+/// [`acquire_channel_membership_lock`] and hold the transaction through run
+/// commit, serializing authorization with membership revocation.
+pub async fn get_member_role_in_transaction(
+    tx: &mut Transaction<'_, Postgres>,
+    community_id: CommunityId,
+    channel_id: Uuid,
+    pubkey: &[u8],
+) -> Result<Option<String>> {
+    let row = sqlx::query(
+        "SELECT cm.role::text AS role FROM channel_members cm \
+         JOIN channels c ON cm.community_id = c.community_id AND cm.channel_id = c.id AND c.deleted_at IS NULL \
+         WHERE cm.community_id = $1 AND cm.channel_id = $2 AND cm.pubkey = $3 AND cm.removed_at IS NULL",
+    )
+    .bind(community_id.as_uuid())
+    .bind(channel_id)
+    .bind(pubkey)
+    .fetch_optional(&mut **tx)
+    .await?;
+    Ok(row.map(|r| r.try_get("role")).transpose()?)
+}
+
 /// Archive ephemeral channels whose TTL deadline has passed.
 ///
 /// Returns the `(community_id, host, channel_id)` list that was archived. Idempotent — the

crates/buzz-db/src/event.rs

@@ -1032,6 +1032,30 @@ pub async fn get_event_by_id(
     }
 }
 
+/// Fetches a single non-deleted event by ID on the caller's transaction.
+///
+/// Use this when a transaction already owns the connection so the lookup cannot
+/// block waiting for another pool connection or observe a different snapshot.
+pub async fn get_event_by_id_in_transaction(
+    tx: &mut Transaction<'_, Postgres>,
+    community_id: CommunityId,
+    id_bytes: &[u8],
+) -> Result<Option<StoredEvent>> {
+    let row = sqlx::query(
+        "SELECT id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id \
+         FROM events WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL ORDER BY created_at DESC LIMIT 1",
+    )
+    .bind(community_id.as_uuid())
+    .bind(id_bytes)
+    .fetch_optional(&mut **tx)
+    .await?;
+
+    match row {
+        Some(r) => row_to_stored_event(r),
+        None => Ok(None),
+    }
+}
+
 /// Fetches the latest global (non-channel, `channel_id IS NULL`) replaceable event
 /// for a (kind, pubkey) pair.
 ///

crates/buzz-db/src/lib.rs

@@ -1419,6 +1419,17 @@ impl Db {
         event::get_event_by_id(&self.pool, community_id, id_bytes).await
     }
 
+    /// Fetches a single non-deleted event by its raw ID bytes on the caller's transaction.
+    #[datastore_span(name = "get_event_by_id_in_transaction", system = "postgresql")]
+    pub async fn get_event_by_id_in_transaction(
+        &self,
+        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+        community_id: CommunityId,
+        id_bytes: &[u8],
+    ) -> Result<Option<StoredEvent>> {
+        event::get_event_by_id_in_transaction(tx, community_id, id_bytes).await
+    }
+
     /// Fetches a single event by its raw ID bytes, **including soft-deleted rows**.
     #[datastore_span(name = "get_event_by_id_including_deleted", system = "postgresql")]
     pub async fn get_event_by_id_including_deleted(
@@ -3325,28 +3336,31 @@ impl Db {
         .await
     }
 
-    /// Insert or update a workflow using its NIP-33 `d`-tag UUID.
+    /// Atomically insert or update a workflow using its NIP-33 `d`-tag UUID.
     #[allow(clippy::too_many_arguments)]
     #[datastore_span(name = "upsert_workflow", system = "postgresql")]
     pub async fn upsert_workflow(
         &self,
+        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
         community_id: CommunityId,
         id: Uuid,
         channel_id: Option<Uuid>,
         owner_pubkey: &[u8],
         name: &str,
         definition_json: &str,
         definition_hash: &[u8],
+        definition_event_id: &[u8],
     ) -> Result<()> {
         workflow::upsert_workflow(
-            &self.pool,
+            tx.as_mut(),
             community_id,
             id,
             channel_id,
             owner_pubkey,
             name,
             definition_json,
             definition_hash,
+            definition_event_id,
         )
         .await
     }
@@ -3542,19 +3556,54 @@ impl Db {
         workflow::find_by_owner_and_name(&self.pool, community_id, owner_pubkey, name).await
     }
 
+    /// Fetch and share-lock one workflow on an existing transaction.
+    #[datastore_span(name = "get_workflow_for_share_in_transaction", system = "postgresql")]
+    pub async fn get_workflow_for_share_in_transaction(
+        &self,
+        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+        community_id: CommunityId,
+        id: Uuid,
+    ) -> Result<workflow::WorkflowRecord> {
+        workflow::get_workflow_for_share_in_transaction(tx, community_id, id).await
+    }
+
+    /// Create a new workflow run on an existing transaction.
+    #[datastore_span(name = "create_workflow_run_in_transaction", system = "postgresql")]
+    pub async fn create_workflow_run_in_transaction(
+        &self,
+        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+        community_id: CommunityId,
+        workflow_id: Uuid,
+        definition_event_id: &[u8],
+        trigger_event_id: Option<&[u8]>,
+        trigger_context: Option<&serde_json::Value>,
+    ) -> Result<Uuid> {
+        workflow::create_workflow_run_in_transaction(
+            tx,
+            community_id,
+            workflow_id,
+            definition_event_id,
+            trigger_event_id,
+            trigger_context,
+        )
+        .await
+    }
+
     /// Create a new workflow run.
     #[datastore_span(name = "create_workflow_run", system = "postgresql")]
     pub async fn create_workflow_run(
         &self,
         community_id: CommunityId,
         workflow_id: Uuid,
+        definition_event_id: &[u8],
         trigger_event_id: Option<&[u8]>,
         trigger_context: Option<&serde_json::Value>,
     ) -> Result<Uuid> {
         workflow::create_workflow_run(
             &self.pool,
             community_id,
             workflow_id,
+            definition_event_id,
             trigger_event_id,
             trigger_context,
         )

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(), 32);
+        assert_eq!(migrations.len(), 34);
         assert_eq!(migrations[0].version, 1);
         assert_eq!(&*migrations[0].description, "initial schema");
         assert!(migrations[0]
@@ -1072,6 +1072,21 @@ mod tests {
         assert!(roster_fence.contains("snapshot_members IS DISTINCT FROM canonical_members"));
         assert!(roster_fence.contains("ERRCODE = '23514'"));
 
+        // Exact workflow-definition revision persistence is additive and keeps
+        // pre-migration rows nullable until their next signed definition write.
+        assert_eq!(migrations[32].version, 33);
+        let workflow_revision = migrations[32].sql.as_str();
+        assert!(workflow_revision.contains("ADD COLUMN definition_event_id BYTEA"));
+        assert!(workflow_revision.contains("octet_length(definition_event_id) = 32"));
+
+        // Runs durably inherit the exact signed revision they execute. Existing
+        // runs remain nullable and execution/resume must fail them closed.
+        assert_eq!(migrations[33].version, 34);
+        let run_revision = migrations[33].sql.as_str();
+        assert!(run_revision.contains("ALTER TABLE workflow_runs"));
+        assert!(run_revision.contains("ADD COLUMN definition_event_id BYTEA"));
+        assert!(run_revision.contains("octet_length(definition_event_id) = 32"));
+
         // 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.