feat(workflows): bind owner triggers to exact signed revisions

Reconcile the complete Node A observable behavior onto current main without carrying its historical merge topology.

Co-authored-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz>
Signed-off-by: Logan Johnson <loganj@squareup.com>

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,11 +3336,12 @@ 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>,
@@ -3337,9 +3349,10 @@ impl Db {
         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,
@@ -3347,6 +3360,7 @@ impl Db {
             name,
             definition_json,
             definition_hash,
+            definition_event_id,
         )
         .await
     }
@@ -3542,12 +3556,46 @@ 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> {
@@ -3555,6 +3603,7 @@ impl Db {
             &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.

crates/buzz-db/src/workflow.rs

@@ -177,6 +177,8 @@ pub struct WorkflowRecord {
     pub definition: serde_json::Value,
     /// SHA-256 hash of the canonical definition JSON.
     pub definition_hash: Vec<u8>,
+    /// Exact owner-signed kind:30620 event that materialized this revision.
+    pub definition_event_id: Option<Vec<u8>>,
     /// Current lifecycle status of the workflow definition.
     pub status: WorkflowStatus,
     /// Whether the workflow will fire on matching events.
@@ -201,6 +203,11 @@ pub struct WorkflowRunRecord {
     pub community_id: CommunityId,
     /// The workflow definition that was executed.
     pub workflow_id: Uuid,
+    /// Exact owner-signed kind:30620 revision this run executes.
+    ///
+    /// NULL is retained only for runs created before the revision-binding
+    /// migration; execution and resume paths must fail those rows closed.
+    pub definition_event_id: Option<Vec<u8>>,
     /// Current execution status of this run.
     pub status: RunStatus,
     /// Raw event ID bytes that triggered this run, if any.
@@ -314,7 +321,7 @@ pub async fn create_workflow(
 /// cross-channel overwrite primitive while still making retries idempotent.
 #[allow(clippy::too_many_arguments)]
 pub async fn upsert_workflow(
-    pool: &PgPool,
+    conn: &mut sqlx::PgConnection,
     community_id: CommunityId,
     id: Uuid,
     channel_id: Option<Uuid>,
@@ -322,16 +329,18 @@ pub async fn upsert_workflow(
     name: &str,
     definition_json: &str,
     definition_hash: &[u8],
+    definition_event_id: &[u8],
 ) -> Result<()> {
     let row = sqlx::query(
         r#"
         INSERT INTO workflows
-            (community_id, id, name, owner_pubkey, channel_id, definition, definition_hash, status, enabled)
-        VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7, 'active', TRUE)
+            (community_id, id, name, owner_pubkey, channel_id, definition, definition_hash, definition_event_id, status, enabled)
+        VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7, $8, 'active', TRUE)
         ON CONFLICT (community_id, id) DO UPDATE
         SET name = EXCLUDED.name,
             definition = EXCLUDED.definition,
             definition_hash = EXCLUDED.definition_hash,
+            definition_event_id = EXCLUDED.definition_event_id,
             updated_at = NOW()
         WHERE workflows.owner_pubkey = EXCLUDED.owner_pubkey
           AND workflows.channel_id IS NOT DISTINCT FROM EXCLUDED.channel_id
@@ -345,7 +354,8 @@ pub async fn upsert_workflow(
     .bind(channel_id)
     .bind(definition_json)
     .bind(definition_hash)
-    .fetch_optional(pool)
+    .bind(definition_event_id)
+    .fetch_optional(conn)
     .await?;
 
     if row.is_none() {
@@ -370,7 +380,7 @@ pub async fn get_workflow(
 ) -> Result<WorkflowRecord> {
     let row = sqlx::query(
         r#"
-        SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash,
+        SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, definition_event_id,
                status::text AS status, enabled, created_at, updated_at
         FROM workflows
         WHERE community_id = $1 AND id = $2
@@ -385,6 +395,34 @@ pub async fn get_workflow(
     row_to_workflow_record(row)
 }
 
+/// Fetch and lock one workflow on the caller's transaction.
+///
+/// The shared row lock is held through commit. Definition replacement takes an
+/// update lock on the same row, so callers can validate an exact revision and
+/// create dependent rows without a replacement committing between those steps.
+pub async fn get_workflow_for_share_in_transaction(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    community_id: CommunityId,
+    id: Uuid,
+) -> Result<WorkflowRecord> {
+    let row = sqlx::query(
+        r#"
+        SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, definition_event_id,
+               status::text AS status, enabled, created_at, updated_at
+        FROM workflows
+        WHERE community_id = $1 AND id = $2
+        FOR SHARE
+        "#,
+    )
+    .bind(community_id.as_uuid())
+    .bind(id)
+    .fetch_optional(&mut **tx)
+    .await?
+    .ok_or_else(|| DbError::NotFound(format!("workflow {id}")))?;
+
+    row_to_workflow_record(row)
+}
+
 /// List workflows for a channel, ordered newest first.
 ///
 /// `limit` is capped at [`LIST_MAX_LIMIT`]. Pass `None` to use [`LIST_DEFAULT_LIMIT`].
@@ -401,7 +439,7 @@ pub async fn list_channel_workflows(
 
     let rows = sqlx::query(
         r#"
-        SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash,
+        SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, definition_event_id,
                status::text AS status, enabled, created_at, updated_at
         FROM workflows
         WHERE community_id = $1 AND channel_id = $2
@@ -432,7 +470,7 @@ pub async fn list_enabled_channel_workflows(
 ) -> Result<Vec<WorkflowRecord>> {
     let rows = sqlx::query(
         r#"
-        SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash,
+        SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, definition_event_id,
                status::text AS status, enabled, created_at, updated_at
         FROM workflows
         WHERE community_id = $1
@@ -460,7 +498,7 @@ pub async fn list_enabled_channel_workflows(
 pub async fn list_all_enabled_workflows(pool: &PgPool) -> Result<Vec<WorkflowRecord>> {
     let rows = sqlx::query(
         r#"
-        SELECT w.id, w.community_id, w.name, w.owner_pubkey, w.channel_id, w.definition, w.definition_hash,
+        SELECT w.id, w.community_id, w.name, w.owner_pubkey, w.channel_id, w.definition, w.definition_hash, w.definition_event_id,
                w.status::text AS status, w.enabled, w.created_at, w.updated_at
         FROM workflows w
         JOIN communities c ON c.id = w.community_id
@@ -793,6 +831,39 @@ pub async fn delete_workflow_for_owner(
 
 // -- Workflow Run CRUD --------------------------------------------------------
 
+/// Insert a new workflow run on the caller's transaction.
+///
+/// Command handlers use this with the transaction that persisted the trigger
+/// event so both rows commit or roll back together.
+pub async fn create_workflow_run_in_transaction(
+    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> {
+    let id = Uuid::new_v4();
+
+    sqlx::query(
+        r#"
+        INSERT INTO workflow_runs
+            (community_id, id, workflow_id, definition_event_id, status, trigger_event_id, current_step, execution_trace, trigger_context)
+        VALUES ($1, $2, $3, $4, 'pending', $5, 0, '[]', $6)
+        "#,
+    )
+    .bind(community_id.as_uuid())
+    .bind(id)
+    .bind(workflow_id)
+    .bind(definition_event_id)
+    .bind(trigger_event_id)
+    .bind(trigger_context)
+    .execute(&mut **tx)
+    .await?;
+
+    Ok(id)
+}
+
 /// Insert a new workflow run. Returns the new run's UUID.
 ///
 /// `trigger_context` is the serialized `TriggerContext` for this run. It is stored
@@ -802,6 +873,7 @@ pub async fn create_workflow_run(
     pool: &PgPool,
     community_id: CommunityId,
     workflow_id: Uuid,
+    definition_event_id: &[u8],
     trigger_event_id: Option<&[u8]>,
     trigger_context: Option<&serde_json::Value>,
 ) -> Result<Uuid> {
@@ -810,13 +882,14 @@ pub async fn create_workflow_run(
     sqlx::query(
         r#"
         INSERT INTO workflow_runs
-            (community_id, id, workflow_id, status, trigger_event_id, current_step, execution_trace, trigger_context)
-        VALUES ($1, $2, $3, 'pending', $4, 0, '[]', $5)
+            (community_id, id, workflow_id, definition_event_id, status, trigger_event_id, current_step, execution_trace, trigger_context)
+        VALUES ($1, $2, $3, $4, 'pending', $5, 0, '[]', $6)
         "#,
     )
     .bind(community_id.as_uuid())
     .bind(id)
     .bind(workflow_id)
+    .bind(definition_event_id)
     .bind(trigger_event_id)
     .bind(trigger_context)
     .execute(pool)
@@ -833,7 +906,7 @@ pub async fn get_workflow_run(
 ) -> Result<WorkflowRunRecord> {
     let row = sqlx::query(
         r#"
-        SELECT community_id, id, workflow_id, status::text AS status, trigger_event_id, current_step,
+        SELECT community_id, id, workflow_id, definition_event_id, status::text AS status, trigger_event_id, current_step,
                execution_trace, trigger_context, started_at, completed_at, error_message, error_code, created_at
         FROM workflow_runs
         WHERE community_id = $1 AND id = $2
@@ -865,7 +938,7 @@ pub async fn list_workflow_runs_page(
     let limit = limit.clamp(1, LIST_MAX_LIMIT);
     let rows = sqlx::query(
         r#"
-        SELECT community_id, id, workflow_id, status::text AS status, trigger_event_id, current_step,
+        SELECT community_id, id, workflow_id, definition_event_id, status::text AS status, trigger_event_id, current_step,
                execution_trace, trigger_context, started_at, completed_at, error_message, error_code, created_at
         FROM workflow_runs
         WHERE community_id = $1 AND workflow_id = $2
@@ -1183,6 +1256,7 @@ fn row_to_workflow_record(row: sqlx::postgres::PgRow) -> Result<WorkflowRecord>
         channel_id,
         definition: row.try_get("definition")?,
         definition_hash: row.try_get("definition_hash")?,
+        definition_event_id: row.try_get("definition_event_id")?,
         status,
         enabled,
         created_at: row.try_get("created_at")?,
@@ -1202,6 +1276,7 @@ fn row_to_run_record(row: sqlx::postgres::PgRow) -> Result<WorkflowRunRecord> {
         id,
         community_id: CommunityId::from_uuid(community_id),
         workflow_id,
+        definition_event_id: row.try_get("definition_event_id")?,
         status,
         trigger_event_id: row.try_get("trigger_event_id")?,
         current_step: row.try_get("current_step")?,
@@ -1247,7 +1322,7 @@ pub async fn find_by_owner_and_name(
 ) -> Result<Option<WorkflowRecord>> {
     let row = sqlx::query(
         r#"
-        SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash,
+        SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, definition_event_id,
                status::text AS status, enabled, created_at, updated_at
         FROM workflows
         WHERE community_id = $1 AND owner_pubkey = $2 AND name = $3
@@ -1382,6 +1457,7 @@ mod tests {
             channel_id: Some(channel_id),
             definition: def.clone(),
             definition_hash: vec![0x01, 0x02, 0x03, 0x04],
+            definition_event_id: None,
             status: WorkflowStatus::Active,
             enabled: true,
             created_at: now,
@@ -1412,6 +1488,7 @@ mod tests {
             channel_id: None,
             definition: serde_json::json!({}),
             definition_hash: vec![],
+            definition_event_id: None,
             status: WorkflowStatus::Active,
             enabled: true,
             created_at: now,
@@ -1434,6 +1511,7 @@ mod tests {
             channel_id: None,
             definition: serde_json::json!({}),
             definition_hash: vec![0xAA],
+            definition_event_id: None,
             status: WorkflowStatus::Active,
             enabled: true,
             created_at: now,
@@ -1463,6 +1541,7 @@ mod tests {
                 channel_id: None,
                 definition: serde_json::json!({}),
                 definition_hash: vec![],
+                definition_event_id: None,
                 status: status.clone(),
                 enabled: true,
                 created_at: now,
@@ -1483,6 +1562,7 @@ mod tests {
             channel_id: None,
             definition: serde_json::json!({}),
             definition_hash: vec![],
+            definition_event_id: None,
             status: WorkflowStatus::Active,
             enabled: false,
             created_at: now,
@@ -1505,6 +1585,7 @@ mod tests {
             id,
             community_id: CommunityId::from_uuid(Uuid::new_v4()),
             workflow_id,
+            definition_event_id: Some(vec![0x42; 32]),
             status: RunStatus::Running,
             trigger_event_id: Some(trigger_event_id.clone()),
             current_step: 2,
@@ -1536,6 +1617,7 @@ mod tests {
             id: Uuid::new_v4(),
             community_id: CommunityId::from_uuid(Uuid::new_v4()),
             workflow_id: Uuid::new_v4(),
+            definition_event_id: None,
             status: RunStatus::Pending,
             trigger_event_id: None,
             current_step: 0,
@@ -1560,6 +1642,7 @@ mod tests {
             id: Uuid::new_v4(),
             community_id: CommunityId::from_uuid(Uuid::new_v4()),
             workflow_id: Uuid::new_v4(),
+            definition_event_id: None,
             status: RunStatus::Failed,
             trigger_event_id: None,
             current_step: 1,
@@ -1592,6 +1675,7 @@ mod tests {
             id: Uuid::new_v4(),
             community_id: CommunityId::from_uuid(Uuid::new_v4()),
             workflow_id: Uuid::new_v4(),
+            definition_event_id: None,
             status: RunStatus::Completed,
             trigger_event_id: None,
             current_step: 2,
@@ -1615,6 +1699,7 @@ mod tests {
             id: Uuid::new_v4(),
             community_id: CommunityId::from_uuid(Uuid::new_v4()),
             workflow_id: Uuid::new_v4(),
+            definition_event_id: None,
             status: RunStatus::Pending,
             trigger_event_id: None,
             current_step: 0,
@@ -1963,7 +2048,7 @@ mod tests {
             .expect("claim wins");
 
         // Create the run the won claim is responsible for, then attach it.
-        let run_id = create_workflow_run(&pool, community, workflow_id, None, None)
+        let run_id = create_workflow_run(&pool, community, workflow_id, &[0x42; 32], None, None)
             .await
             .expect("create run ok");
 
@@ -1992,7 +2077,7 @@ mod tests {
 
         // A second attach is a no-op: the `workflow_run_id IS NULL` guard means
         // an already-linked claim is never re-pointed to a different run.
-        let other_run = create_workflow_run(&pool, community, workflow_id, None, None)
+        let other_run = create_workflow_run(&pool, community, workflow_id, &[0x42; 32], None, None)
             .await
             .expect("create second run ok");
         let reattached =
@@ -2129,6 +2214,177 @@ mod tests {
         .expect("insert workflow");
     }
 
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn stale_revision_after_replacement_creates_no_run() {
+        let pool = setup_pool().await;
+        let community = make_community(&pool).await;
+        let workflow_id = Uuid::new_v4();
+        insert_workflow_with_ids(
+            &pool,
+            community,
+            workflow_id,
+            Uuid::new_v4(),
+            "revision-race",
+        )
+        .await;
+        let revision_a = vec![0xa1u8; 32];
+        let revision_b = vec![0xb2u8; 32];
+        sqlx::query(
+            "UPDATE workflows SET definition_event_id = $3 \
+             WHERE community_id = $1 AND id = $2",
+        )
+        .bind(community.as_uuid())
+        .bind(workflow_id)
+        .bind(revision_a.as_slice())
+        .execute(&pool)
+        .await
+        .expect("install revision A");
+
+        // B wins before stale trigger A enters its commit transaction.
+        sqlx::query(
+            "UPDATE workflows SET definition_event_id = $3 \
+             WHERE community_id = $1 AND id = $2",
+        )
+        .bind(community.as_uuid())
+        .bind(workflow_id)
+        .bind(revision_b.as_slice())
+        .execute(&pool)
+        .await
+        .expect("replace with revision B");
+
+        let mut trigger = pool.begin().await.expect("begin stale trigger");
+        let current = get_workflow_for_share_in_transaction(&mut trigger, community, workflow_id)
+            .await
+            .expect("lock current workflow");
+        assert_ne!(
+            current.definition_event_id.as_deref(),
+            Some(revision_a.as_slice())
+        );
+        trigger.rollback().await.expect("reject stale trigger");
+
+        let runs: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM workflow_runs WHERE community_id = $1 AND workflow_id = $2",
+        )
+        .bind(community.as_uuid())
+        .bind(workflow_id)
+        .fetch_one(&pool)
+        .await
+        .expect("count runs after stale rejection");
+        assert_eq!(runs, 0, "stale revision must not create a workflow run");
+    }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn workflow_run_transaction_rolls_back_and_retry_creates_exactly_one_run() {
+        let pool = setup_pool().await;
+        let community = make_community(&pool).await;
+        let workflow_id = Uuid::new_v4();
+        insert_workflow_with_ids(
+            &pool,
+            community,
+            workflow_id,
+            Uuid::new_v4(),
+            "atomic-trigger",
+        )
+        .await;
+        let trigger_event_id = vec![0x7au8; 32];
+        let trigger_pubkey = vec![0x7bu8; 32];
+        let trigger_sig = vec![0x7cu8; 64];
+
+        let mut aborted = pool.begin().await.expect("begin aborted transaction");
+        sqlx::query(
+            "INSERT INTO events \
+             (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at) \
+             VALUES ($1, $2, $3, NOW(), 46020, '[]'::jsonb, '', $4, NOW())",
+        )
+        .bind(community.as_uuid())
+        .bind(trigger_event_id.as_slice())
+        .bind(trigger_pubkey.as_slice())
+        .bind(trigger_sig.as_slice())
+        .execute(&mut *aborted)
+        .await
+        .expect("insert trigger event before abort");
+        create_workflow_run_in_transaction(
+            &mut aborted,
+            community,
+            workflow_id,
+            &[0x42; 32],
+            Some(&trigger_event_id),
+            None,
+        )
+        .await
+        .expect("insert run before abort");
+        aborted.rollback().await.expect("roll back run");
+
+        let after_abort_event: i64 =
+            sqlx::query_scalar("SELECT COUNT(*) FROM events WHERE community_id = $1 AND id = $2")
+                .bind(community.as_uuid())
+                .bind(&trigger_event_id)
+                .fetch_one(&pool)
+                .await
+                .expect("count rolled-back trigger events");
+        assert_eq!(
+            after_abort_event, 0,
+            "aborted trigger must leave no event row"
+        );
+        let after_abort: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM workflow_runs WHERE community_id = $1 AND trigger_event_id = $2",
+        )
+        .bind(community.as_uuid())
+        .bind(&trigger_event_id)
+        .fetch_one(&pool)
+        .await
+        .expect("count rolled-back runs");
+        assert_eq!(after_abort, 0, "aborted trigger must leave no run row");
+
+        let mut retry = pool.begin().await.expect("begin retry transaction");
+        sqlx::query(
+            "INSERT INTO events \
+             (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at) \
+             VALUES ($1, $2, $3, NOW(), 46020, '[]'::jsonb, '', $4, NOW())",
+        )
+        .bind(community.as_uuid())
+        .bind(trigger_event_id.as_slice())
+        .bind(trigger_pubkey.as_slice())
+        .bind(trigger_sig.as_slice())
+        .execute(&mut *retry)
+        .await
+        .expect("insert trigger event on retry");
+        create_workflow_run_in_transaction(
+            &mut retry,
+            community,
+            workflow_id,
+            &[0x42; 32],
+            Some(&trigger_event_id),
+            None,
+        )
+        .await
+        .expect("insert retry run");
+        retry.commit().await.expect("commit retry run");
+
+        let after_retry_event: i64 =
+            sqlx::query_scalar("SELECT COUNT(*) FROM events WHERE community_id = $1 AND id = $2")
+                .bind(community.as_uuid())
+                .bind(&trigger_event_id)
+                .fetch_one(&pool)
+                .await
+                .expect("count committed trigger events");
+        assert_eq!(
+            after_retry_event, 1,
+            "retry must create exactly one event row"
+        );
+        let after_retry: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM workflow_runs WHERE community_id = $1 AND trigger_event_id = $2",
+        )
+        .bind(community.as_uuid())
+        .bind(&trigger_event_id)
+        .fetch_one(&pool)
+        .await
+        .expect("count committed retry runs");
+        assert_eq!(after_retry, 1, "retry must create exactly one run row");
+    }
+
     /// Issue 4 (workflow identity): the same workflow UUID and channel UUID can
     /// exist in communities A and B (PK `(community_id, id)`). A request-scoped
     /// `get_workflow` / `list_enabled_channel_workflows` MUST return only the
@@ -2273,10 +2529,10 @@ mod tests {
         insert_workflow_with_ids(&pool, community_a, workflow_id, channel_id, "wf-A").await;
         insert_workflow_with_ids(&pool, community_b, workflow_id, Uuid::new_v4(), "wf-B").await;
 
-        let run_a = create_workflow_run(&pool, community_a, workflow_id, None, None)
+        let run_a = create_workflow_run(&pool, community_a, workflow_id, &[0x42; 32], None, None)
             .await
             .expect("run A");
-        let run_b = create_workflow_run(&pool, community_b, workflow_id, None, None)
+        let run_b = create_workflow_run(&pool, community_b, workflow_id, &[0x42; 32], None, None)
             .await
             .expect("run B");
 

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

@@ -2111,52 +2111,41 @@ pub async fn workflow_webhook(
         .await
         .map_err(|_| not_found("workflow not found"))?;
 
+    let definition_event_id = workflow
+        .definition_event_id
+        .as_deref()
+        .ok_or_else(|| not_found("workflow not found"))?;
     let run_id = state
         .db
-        .create_workflow_run(community_id, id, None, trigger_ctx_json.as_ref())
+        .create_workflow_run(
+            community_id,
+            id,
+            definition_event_id,
+            None,
+            trigger_ctx_json.as_ref(),
+        )
         .await
         .map_err(|e| super::internal_error(&format!("db error: {e}")))?;
 
     // Spawn workflow execution asynchronously.
     let engine = Arc::clone(&state.workflow_engine);
-    let db = state.db.clone();
-    let def_value = workflow.definition.clone();
     let trigger_ctx_clone = trigger_ctx.clone();
     tokio::spawn(async move {
-        let def: buzz_workflow::WorkflowDef = match serde_json::from_value(def_value) {
-            Ok(d) => d,
-            Err(e) => {
-                tracing::error!("webhook: failed to parse definition: {e}");
-                if let Err(db_err) = db
-                    .update_workflow_run(
-                        community_id,
-                        run_id,
-                        buzz_db::workflow::RunStatus::Failed,
-                        0,
-                        &serde_json::json!([]),
-                        Some(buzz_db::workflow::WorkflowRunFailure {
-                            code: "invalid_definition",
-                            message: &format!("definition parse error: {e}"),
-                        }),
-                    )
-                    .await
-                {
-                    tracing::error!("webhook: failed to mark run as failed: {db_err}");
-                }
-                return;
+        let result = match engine.load_run_definition(community_id, run_id).await {
+            Ok((_, definition)) => {
+                buzz_workflow::executor::execute_from_step(
+                    &engine,
+                    community_id,
+                    run_id,
+                    &definition,
+                    &trigger_ctx_clone,
+                    0,
+                    None,
+                )
+                .await
             }
+            Err(error) => Err((error, buzz_workflow::error::PartialProgress::default())),
         };
-
-        let result = buzz_workflow::executor::execute_from_step(
-            &engine,
-            community_id,
-            run_id,
-            &def,
-            &trigger_ctx_clone,
-            0,
-            None,
-        )
-        .await;
         engine
             .finalize_run(community_id, run_id, result, None)
             .await;

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

@@ -46,6 +46,7 @@ async fn authorize_workflow_read(
     path: &str,
     raw_query: Option<&str>,
     workflow_id: Uuid,
+    allow_immutable_owner: bool,
 ) -> Result<TenantContext, (StatusCode, Json<Value>)> {
     let raw_host = headers
         .get(axum::http::header::HOST)
@@ -97,15 +98,59 @@ async fn authorize_workflow_read(
         .await
         .map_err(|error| internal_error(&format!("workflow channel access lookup: {error}")))?;
     if !accessible.contains(&channel_id) {
-        return Err(api_error(
-            StatusCode::FORBIDDEN,
-            "workflow is not accessible",
-        ));
+        let controls = allow_immutable_owner
+            && (workflow.owner_pubkey == pubkey_bytes
+                || state
+                    .db
+                    .is_agent_owner(tenant.community(), &workflow.owner_pubkey, &pubkey_bytes)
+                    .await
+                    .map_err(|error| internal_error(&format!("workflow owner lookup: {error}")))?);
+        if !controls {
+            return Err(api_error(
+                StatusCode::FORBIDDEN,
+                "workflow is not accessible",
+            ));
+        }
     }
 
     Ok(tenant)
 }
 
+/// `GET /workflows/{workflow_id}/revision` — current signed revision for an
+/// authorized channel reader or the managed agent's immutable human owner.
+///
+/// This narrow endpoint does not grant channel visibility; it returns only the
+/// exact owner-signed definition event needed to construct a revision-bound
+/// manual trigger.
+pub async fn workflow_revision(
+    State(state): State<Arc<AppState>>,
+    Path(workflow_id): Path<Uuid>,
+    headers: HeaderMap,
+) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
+    let path = format!("/workflows/{workflow_id}/revision");
+    let tenant = authorize_workflow_read(&state, &headers, &path, None, workflow_id, true).await?;
+    let workflow = state
+        .db
+        .get_workflow(tenant.community(), workflow_id)
+        .await
+        .map_err(|_| api_error(StatusCode::NOT_FOUND, "workflow not found"))?;
+    let revision = workflow.definition_event_id.as_deref().ok_or_else(|| {
+        api_error(
+            StatusCode::CONFLICT,
+            "owner-signed workflow revision is unavailable",
+        )
+    })?;
+    let event = state
+        .db
+        .get_event_by_id(tenant.community(), revision)
+        .await
+        .map_err(|error| internal_error(&format!("get workflow revision: {error}")))?
+        .ok_or_else(|| api_error(StatusCode::CONFLICT, "workflow revision is unavailable"))?;
+    Ok(Json(serde_json::to_value(event.event).map_err(
+        |error| internal_error(&format!("serialize workflow revision: {error}")),
+    )?))
+}
+
 /// `GET /workflows/{workflow_id}/runs` — one authorized, keyset-paginated page.
 pub async fn workflow_runs(
     State(state): State<Arc<AppState>>,
@@ -129,8 +174,15 @@ pub async fn workflow_runs(
     }
 
     let path = format!("/workflows/{workflow_id}/runs");
-    let tenant =
-        authorize_workflow_read(&state, &headers, &path, raw_query.as_deref(), workflow_id).await?;
+    let tenant = authorize_workflow_read(
+        &state,
+        &headers,
+        &path,
+        raw_query.as_deref(),
+        workflow_id,
+        false,
+    )
+    .await?;
     let mut rows = state
         .db
         .list_workflow_runs_page(
@@ -169,7 +221,7 @@ pub async fn run_approvals(
     headers: HeaderMap,
 ) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
     let path = format!("/workflows/{workflow_id}/runs/{run_id}/approvals");
-    let tenant = authorize_workflow_read(&state, &headers, &path, None, workflow_id).await?;
+    let tenant = authorize_workflow_read(&state, &headers, &path, None, workflow_id, false).await?;
 
     let run = state
         .db

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

@@ -92,12 +92,11 @@ enum PersistResult {
 /// If the event is a duplicate (ON CONFLICT DO NOTHING), the transaction is
 /// rolled back and `PersistResult::Duplicate` is returned — no mutations needed.
 ///
-/// NOTE: Domain mutations (open_dm, upsert_workflow, etc.) execute on the
-/// connection pool, NOT inside this transaction. The pattern is idempotent but
-/// not strictly atomic: if a mutation succeeds but commit fails, the mutation
-/// persists without the event record. On retry, the event INSERT succeeds
-/// (no conflict), and the mutation re-executes — which is safe for idempotent
-/// operations (open_dm, hide_dm, update_approval, upsert_workflow).
+/// NOTE: Most domain mutations still execute on the connection pool rather
+/// than in this transaction. Workflow-definition ingestion is the exception:
+/// its materialized workflow row and exact signed revision are written through
+/// this transaction so the event and revision binding commit atomically.
+/// Other operations remain idempotent but not strictly atomic.
 #[datastore_span(name = "persist_command_event", system = "postgresql")]
 async fn persist_command_event(
     db: &buzz_db::Db,
@@ -736,8 +735,9 @@ async fn handle_workflow_def(
         .map_err(|e| IngestError::Internal(format!("error: json serialize: {e}")))?;
     let hash = compute_definition_hash(&definition_json_final);
 
-    // Persist the command event — returns open transaction
-    let tx = match persist_command_event(&state.db, tenant, event, None).await? {
+    // Persist the command event — returns the transaction that will also own
+    // the materialized workflow revision update.
+    let mut tx = match persist_command_event(&state.db, tenant, event, Some(channel_id)).await? {
         PersistResult::Duplicate => {
             return Ok(IngestResult {
                 event_id: event.id.to_hex(),
@@ -769,6 +769,7 @@ async fn handle_workflow_def(
     state
         .db
         .upsert_workflow(
+            &mut tx,
             community_id,
             workflow_id,
             Some(channel_id),
@@ -776,6 +777,7 @@ async fn handle_workflow_def(
             &workflow_name,
             &definition_json_final,
             &hash,
+            event.id.as_bytes(),
         )
         .await
         .map_err(|e| match e {
@@ -785,17 +787,18 @@ async fn handle_workflow_def(
             other => IngestError::Internal(format!("error: db upsert_workflow: {other}")),
         })?;
 
-    // Drop the trigger-path cache entry so the new/updated definition fires on
-    // the next matching event instead of after the cache TTL.
-    state
-        .workflow_engine
-        .invalidate_channel_workflows(community_id, channel_id);
-
     // Commit the event transaction after the idempotent workflow upsert succeeds.
     tx.commit()
         .await
         .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?;
 
+    // Invalidate only after commit. Invalidating while the new row is still
+    // invisible lets a concurrent trigger refill the cache with the old
+    // definition and retain it until TTL expiry.
+    state
+        .workflow_engine
+        .invalidate_channel_workflows(community_id, channel_id);
+
     // 5. Return response
     let mut resp = serde_json::json!({
         "workflow_id": workflow_id.to_string(),
@@ -811,6 +814,101 @@ async fn handle_workflow_def(
     })
 }
 
+async fn caller_controls_workflow(
+    state: &Arc<AppState>,
+    community_id: CommunityId,
+    workflow_owner: &[u8],
+    caller: &[u8],
+) -> Result<bool, IngestError> {
+    if workflow_owner == caller {
+        return Ok(true);
+    }
+
+    state
+        .db
+        .is_agent_owner(community_id, workflow_owner, caller)
+        .await
+        .map_err(|e| IngestError::Internal(format!("error: workflow owner check: {e}")))
+}
+
+fn exact_tag_value<'a>(event: &'a Event, name: &str) -> Option<&'a str> {
+    let mut values = event.tags.iter().filter_map(|tag| {
+        (tag.kind().to_string() == name)
+            .then(|| tag.content())
+            .flatten()
+    });
+    let value = values.next()?;
+    values.next().is_none().then_some(value)
+}
+
+async fn verify_workflow_revision(
+    state: &Arc<AppState>,
+    mut tx: Option<&mut sqlx::Transaction<'_, sqlx::Postgres>>,
+    community_id: CommunityId,
+    workflow: &buzz_db::workflow::WorkflowRecord,
+    requested_revision: &[u8],
+) -> Result<(), IngestError> {
+    let Some(persisted_revision) = workflow.definition_event_id.as_deref() else {
+        return Err(IngestError::Rejected(
+            "invalid: owner-signed workflow revision is unavailable".into(),
+        ));
+    };
+    if persisted_revision != requested_revision {
+        return Err(IngestError::Rejected(
+            "conflict: workflow revision does not match current definition".into(),
+        ));
+    }
+
+    let stored = match tx.as_mut() {
+        Some(tx) => {
+            state
+                .db
+                .get_event_by_id_in_transaction(tx, community_id, persisted_revision)
+                .await
+        }
+        None => {
+            state
+                .db
+                .get_event_by_id(community_id, persisted_revision)
+                .await
+        }
+    }
+    .map_err(|e| IngestError::Internal(format!("error: workflow revision lookup: {e}")))?
+    .ok_or_else(|| IngestError::Rejected("invalid: signed workflow revision not found".into()))?;
+    let definition_event = &stored.event;
+    let workflow_id = workflow.id.to_string();
+    let workflow_channel_id = workflow.channel_id.map(|id| id.to_string());
+    if definition_event.id.as_bytes() != persisted_revision
+        || !definition_event.verify_id()
+        || !definition_event.verify_signature()
+        || definition_event.kind.as_u16() as u32 != KIND_WORKFLOW_DEF
+        || definition_event.pubkey.to_bytes().as_slice() != workflow.owner_pubkey
+        || exact_tag_value(definition_event, "d") != Some(workflow_id.as_str())
+        || workflow_channel_id.is_none()
+        || exact_tag_value(definition_event, "h") != workflow_channel_id.as_deref()
+        || stored.channel_id != workflow.channel_id
+    {
+        return Err(IngestError::Rejected(
+            "invalid: signed workflow revision binding mismatch".into(),
+        ));
+    }
+
+    let (_, signed_json) = buzz_workflow::WorkflowEngine::parse_yaml(&definition_event.content)
+        .map_err(|_| {
+            IngestError::Rejected("invalid: signed workflow revision is malformed".into())
+        })?;
+    let signed_definition: serde_json::Value =
+        serde_json::from_str(&signed_json).map_err(|_| {
+            IngestError::Rejected("invalid: signed workflow revision is malformed".into())
+        })?;
+    if signed_definition != webhook_secret::strip_secret(&workflow.definition) {
+        return Err(IngestError::Rejected(
+            "invalid: signed workflow revision differs from materialized definition".into(),
+        ));
+    }
+    Ok(())
+}
+
 async fn handle_workflow_trigger(
     tenant: &TenantContext,
     state: &Arc<AppState>,
@@ -819,13 +917,21 @@ async fn handle_workflow_trigger(
 ) -> Result<IngestResult, IngestError> {
     let self_bytes = auth.pubkey().to_bytes().to_vec();
 
-    // 1. Extract workflow reference from `d` tag or `e` tag
-    let workflow_id_str = extract_d_tag(event)
-        .or_else(|| extract_e_tag(event))
-        .ok_or_else(|| {
-            IngestError::Rejected("invalid: missing workflow reference (d or e tag)".into())
-        })?;
-    let workflow_id = Uuid::parse_str(&workflow_id_str)
+    // 1. Bind the command to both the workflow UUID and one exact signed revision.
+    let workflow_id_str = exact_tag_value(event, "d").ok_or_else(|| {
+        IngestError::Rejected("invalid: expected exactly one workflow d tag".into())
+    })?;
+    let revision_hex = exact_tag_value(event, "e").ok_or_else(|| {
+        IngestError::Rejected("invalid: expected exactly one workflow revision e tag".into())
+    })?;
+    let requested_revision = hex::decode(revision_hex)
+        .map_err(|_| IngestError::Rejected("invalid: bad workflow revision event id".into()))?;
+    if requested_revision.len() != 32 {
+        return Err(IngestError::Rejected(
+            "invalid: bad workflow revision event id".into(),
+        ));
+    }
+    let workflow_id = Uuid::parse_str(workflow_id_str)
         .map_err(|_| IngestError::Rejected("invalid: bad workflow_id format".into()))?;
 
     // 2. Validate workflow exists — scoped to the caller's community. The same
@@ -839,14 +945,20 @@ async fn handle_workflow_trigger(
         .await
         .map_err(|_| IngestError::Rejected("invalid: workflow not found".into()))?;
 
-    // 3. Manual triggers execute with the workflow owner's authority, so only
-    // the owner may start them. Channel membership alone is insufficient: a
-    // member could otherwise invoke another user's webhook or message actions.
-    if workflow.owner_pubkey != self_bytes {
+    // 3. Manual triggers execute with the workflow owner's authority. Permit
+    // that principal and, for a managed agent, its immutable human owner.
+    // Channel membership alone remains insufficient.
+    if !caller_controls_workflow(state, community_id, &workflow.owner_pubkey, &self_bytes).await? {
         return Err(IngestError::Rejected(
             "forbidden: not authorized to trigger this workflow".into(),
         ));
     }
+    // Managed-agent ownership is immutable. Carry the authorized workflow
+    // principal across the transaction boundary so no pool-backed ownership
+    // lookup is attempted while the command transaction holds its connection.
+    let authorized_workflow_owner = workflow.owner_pubkey.clone();
+
+    verify_workflow_revision(state, None, community_id, &workflow, &requested_revision).await?;
 
     // SEC-006: manual triggers must honor the workflow's lifecycle state and
     // recheck the owner's *current* channel authority before creating a run.
@@ -876,7 +988,7 @@ async fn handle_workflow_trigger(
     // Persist the command event under the workflow channel even though the
     // trigger event itself only carries the workflow UUID. Storing channel
     // triggers as global events leaks workflow IDs to unrelated relay members.
-    let tx = match persist_command_event(&state.db, tenant, event, workflow.channel_id).await? {
+    let mut tx = match persist_command_event(&state.db, tenant, event, workflow.channel_id).await? {
         PersistResult::Duplicate => {
             return Ok(IngestResult {
                 event_id: event.id.to_hex(),
@@ -887,6 +999,57 @@ async fn handle_workflow_trigger(
         PersistResult::Inserted(tx) => tx,
     };
 
+    // Serialize the final authority check and run commit with channel
+    // membership writers. If revocation commits first we observe no role; if
+    // this lock wins, revocation cannot commit until this run is durable.
+    buzz_db::channel::acquire_channel_membership_lock(&mut tx, community_id, wf_channel_id)
+        .await
+        .map_err(|e| IngestError::Internal(format!("error: membership lock: {e}")))?;
+
+    // Re-read the workflow under a row lock on the same transaction that will
+    // commit the trigger event and run. Definition replacement updates this row,
+    // so it cannot commit between this exact-revision check and our commit. A
+    // replacement that won first is observed here and rejected as stale.
+    let workflow = state
+        .db
+        .get_workflow_for_share_in_transaction(&mut tx, community_id, workflow_id)
+        .await
+        .map_err(|_| IngestError::Rejected("invalid: workflow not found".into()))?;
+    if workflow.owner_pubkey != authorized_workflow_owner {
+        return Err(IngestError::Rejected(
+            "conflict: workflow owner changed while trigger was being processed".into(),
+        ));
+    }
+    verify_workflow_revision(
+        state,
+        Some(&mut tx),
+        community_id,
+        &workflow,
+        &requested_revision,
+    )
+    .await?;
+    if !workflow.enabled || workflow.status != buzz_db::workflow::WorkflowStatus::Active {
+        return Err(IngestError::Rejected(
+            "forbidden: workflow is disabled or inactive".into(),
+        ));
+    }
+    let role = buzz_db::channel::get_member_role_in_transaction(
+        &mut tx,
+        community_id,
+        wf_channel_id,
+        &workflow.owner_pubkey,
+    )
+    .await
+    .map_err(|e| IngestError::Internal(format!("error: owner authority lookup: {e}")))?;
+    if !matches!(
+        (role.as_deref(), def.requires_elevated_authority()),
+        (Some(_), false) | (Some("owner" | "admin"), true)
+    ) {
+        return Err(IngestError::Rejected(
+            "forbidden: not authorized to trigger this workflow".into(),
+        ));
+    }
+
     // 4. Execute: create workflow run
     let mut trigger_ctx = TriggerContext {
         channel_id: workflow
@@ -894,6 +1057,7 @@ async fn handle_workflow_trigger(
             .map(|id| id.to_string())
             .unwrap_or_default(),
         author: hex::encode(&self_bytes),
+        definition_event_id: revision_hex.to_owned(),
         ..Default::default()
     };
     if !event.content.is_empty() {
@@ -912,9 +1076,11 @@ async fn handle_workflow_trigger(
     let event_id_bytes = event.id.as_bytes().to_vec();
     let run_id = state
         .db
-        .create_workflow_run(
+        .create_workflow_run_in_transaction(
+            &mut tx,
             community_id,
             workflow_id,
+            &requested_revision,
             Some(&event_id_bytes),
             trigger_ctx_json.as_ref(),
         )
@@ -928,44 +1094,23 @@ async fn handle_workflow_trigger(
 
     // 5. Spawn workflow execution
     let engine = Arc::clone(&state.workflow_engine);
-    let db = state.db.clone();
-    let def_value = workflow.definition.clone();
     let trigger_ctx_clone = trigger_ctx.clone();
     tokio::spawn(async move {
-        let def: buzz_workflow::WorkflowDef = match serde_json::from_value(def_value) {
-            Ok(d) => d,
-            Err(e) => {
-                tracing::error!("workflow_trigger: failed to parse definition: {e}");
-                if let Err(db_err) = db
-                    .update_workflow_run(
-                        community_id,
-                        run_id,
-                        RunStatus::Failed,
-                        0,
-                        &serde_json::json!([]),
-                        Some(buzz_db::workflow::WorkflowRunFailure {
-                            code: "invalid_definition",
-                            message: &format!("definition parse error: {e}"),
-                        }),
-                    )
-                    .await
-                {
-                    tracing::error!("workflow_trigger: failed to mark run as failed: {db_err}");
-                }
-                return;
+        let result = match engine.load_run_definition(community_id, run_id).await {
+            Ok((_, definition)) => {
+                buzz_workflow::executor::execute_from_step(
+                    &engine,
+                    community_id,
+                    run_id,
+                    &definition,
+                    &trigger_ctx_clone,
+                    0,
+                    None,
+                )
+                .await
             }
+            Err(error) => Err((error, buzz_workflow::error::PartialProgress::default())),
         };
-
-        let result = buzz_workflow::executor::execute_from_step(
-            &engine,
-            community_id,
-            run_id,
-            &def,
-            &trigger_ctx_clone,
-            0,
-            None,
-        )
-        .await;
         engine
             .finalize_run(community_id, run_id, result, None)
             .await;
@@ -1277,15 +1422,38 @@ async fn resume_workflow_after_approval(
     workflow_id: Uuid,
     resume_index: usize,
 ) {
-    let run = match db.get_workflow_run(community_id, run_id).await {
-        Ok(r) => r,
+    let (run, def) = match engine.load_run_definition(community_id, run_id).await {
+        Ok(bound) => bound,
         Err(e) => {
-            tracing::error!("resume_workflow: failed to fetch run {run_id}: {e}");
+            tracing::error!(
+                "resume_workflow: failed to load run-bound definition for {run_id}: {e}"
+            );
+            if let Ok(run) = db.get_workflow_run(community_id, run_id).await {
+                let _ = db
+                    .update_workflow_run(
+                        community_id,
+                        run_id,
+                        RunStatus::Failed,
+                        run.current_step,
+                        &run.execution_trace,
+                        Some(buzz_db::workflow::WorkflowRunFailure {
+                            code: "invalid_definition_revision",
+                            message: &e.to_string(),
+                        }),
+                    )
+                    .await;
+            }
             return;
         }
     };
 
-    // Guard: only resume runs that are actually waiting for approval
+    if run.workflow_id != workflow_id {
+        tracing::error!(
+            "resume_workflow: approval workflow {workflow_id} does not match run {}",
+            run.workflow_id
+        );
+        return;
+    }
     if run.status != RunStatus::WaitingApproval {
         tracing::warn!(
             "resume_workflow: run {run_id} has status '{}', expected 'waiting_approval'",
@@ -1294,40 +1462,6 @@ async fn resume_workflow_after_approval(
         return;
     }
 
-    let workflow = match db.get_workflow(community_id, workflow_id).await {
-        Ok(w) => w,
-        Err(e) => {
-            tracing::error!("resume_workflow: failed to fetch workflow {workflow_id}: {e}");
-            return;
-        }
-    };
-
-    let def: buzz_workflow::WorkflowDef = match serde_json::from_value(workflow.definition.clone())
-    {
-        Ok(d) => d,
-        Err(e) => {
-            tracing::error!("resume_workflow: failed to parse workflow definition: {e}");
-            if let Err(db_err) = db
-                .update_workflow_run(
-                    community_id,
-                    run_id,
-                    RunStatus::Failed,
-                    run.current_step,
-                    &run.execution_trace,
-                    Some(buzz_db::workflow::WorkflowRunFailure {
-                        code: "invalid_definition",
-                        message: &format!("definition parse error: {e}"),
-                    }),
-                )
-                .await
-            {
-                tracing::error!("resume_workflow: failed to mark run as failed: {db_err}");
-            }
-            return;
-        }
-    };
-
-    // Reconstruct step_outputs from execution trace for template resolution
     let mut initial_outputs: std::collections::HashMap<String, serde_json::Value> =
         std::collections::HashMap::new();
     if let Some(trace_arr) = run.execution_trace.as_array() {
@@ -1341,14 +1475,11 @@ async fn resume_workflow_after_approval(
         }
     }
 
-    // Restore trigger context for {{trigger.*}} templates
     let trigger_ctx: TriggerContext = run
         .trigger_context
         .as_ref()
         .and_then(|v| serde_json::from_value(v.clone()).ok())
         .unwrap_or_default();
-
-    // Execute remaining steps
     let existing_trace = run.execution_trace.as_array().cloned();
     let result = buzz_workflow::executor::execute_from_step(
         &engine,
@@ -1374,7 +1505,10 @@ mod tests {
         let url = std::env::var("BUZZ_TEST_DATABASE_URL")
             .or_else(|_| std::env::var("DATABASE_URL"))
             .unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".to_string());
-        let pool = sqlx::PgPool::connect(&url)
+        let pool = sqlx::postgres::PgPoolOptions::new()
+            .max_connections(1)
+            .acquire_timeout(std::time::Duration::from_secs(1))
+            .connect(&url)
             .await
             .expect("connect workflow persistence test database");
         let db = buzz_db::Db::from_pool(pool);
@@ -1390,6 +1524,164 @@ mod tests {
         (db, TenantContext::resolved(community, host))
     }
 
+    async fn manual_trigger_test_context() -> (Arc<AppState>, TenantContext, Keys, Uuid, Event) {
+        use buzz_core::channel::{ChannelType, ChannelVisibility};
+
+        let url = std::env::var("BUZZ_TEST_DATABASE_URL")
+            .or_else(|_| std::env::var("DATABASE_URL"))
+            .unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".to_string());
+        let setup_pool = sqlx::PgPool::connect(&url)
+            .await
+            .expect("connect workflow trigger setup database");
+        let setup_db = buzz_db::Db::from_pool(setup_pool.clone());
+        setup_db
+            .migrate()
+            .await
+            .expect("migrate workflow trigger test database");
+
+        let host = format!("workflow-trigger-{}.example", Uuid::new_v4().simple());
+        let community = setup_db
+            .ensure_configured_community(&host)
+            .await
+            .expect("create workflow trigger test community")
+            .id;
+        let tenant = TenantContext::resolved(community, host.clone());
+        let human = Keys::generate();
+        let agent = Keys::generate();
+        let human_bytes = human.public_key().to_bytes();
+        let agent_bytes = agent.public_key().to_bytes();
+        setup_db
+            .ensure_user(community, &human_bytes)
+            .await
+            .expect("ensure human owner");
+        setup_db
+            .ensure_user(community, &agent_bytes)
+            .await
+            .expect("ensure managed agent");
+        assert!(setup_db
+            .set_agent_owner(community, &agent_bytes, &human_bytes)
+            .await
+            .expect("set immutable agent owner"));
+        let channel = setup_db
+            .create_channel(
+                community,
+                "manual-trigger-pool",
+                ChannelType::Stream,
+                ChannelVisibility::Open,
+                None,
+                &agent_bytes,
+                None,
+            )
+            .await
+            .expect("create workflow channel");
+        let workflow_id = Uuid::new_v4();
+        let definition = EventBuilder::new(
+            Kind::Custom(KIND_WORKFLOW_DEF as u16),
+            concat!(
+                "name: manual-trigger-pool\n",
+                "trigger:\n  on: message_posted\n",
+                "steps:\n  - id: send\n    action: send_message\n    text: done\n",
+            ),
+        )
+        .tags(vec![
+            Tag::parse(["d", workflow_id.to_string().as_str()]).expect("d tag"),
+            Tag::parse(["h", channel.id.to_string().as_str()]).expect("h tag"),
+        ])
+        .sign_with_keys(&agent)
+        .expect("sign workflow definition");
+        let (_, definition_json) = buzz_workflow::WorkflowEngine::parse_yaml(&definition.content)
+            .expect("parse signed workflow definition");
+        let definition_hash = compute_definition_hash(&definition_json);
+        let mut tx = setup_db
+            .begin_transaction()
+            .await
+            .expect("begin workflow seed");
+        buzz_db::event::insert_event_in_transaction(
+            &mut tx,
+            community,
+            &definition,
+            Some(channel.id),
+        )
+        .await
+        .expect("persist signed workflow definition");
+        setup_db
+            .upsert_workflow(
+                &mut tx,
+                community,
+                workflow_id,
+                Some(channel.id),
+                &agent_bytes,
+                "manual-trigger-pool",
+                &definition_json,
+                &definition_hash,
+                definition.id.as_bytes(),
+            )
+            .await
+            .expect("materialize signed workflow");
+        tx.commit().await.expect("commit signed workflow");
+
+        let pool = sqlx::postgres::PgPoolOptions::new()
+            .max_connections(1)
+            .acquire_timeout(std::time::Duration::from_secs(1))
+            .connect(&url)
+            .await
+            .expect("connect one-connection workflow trigger pool");
+        let db = buzz_db::Db::from_pool(pool.clone());
+        let mut config = crate::config::Config::from_env().expect("config from env");
+        config.database_url = url;
+        config.redis_url = "redis://127.0.0.1:1".to_string();
+        config.relay_url = format!("wss://{host}");
+        config.require_relay_membership = false;
+        let redis_pool = deadpool_redis::Config::from_url(&config.redis_url)
+            .create_pool(Some(deadpool_redis::Runtime::Tokio1))
+            .expect("redis pool config");
+        let pubsub = Arc::new(
+            buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone())
+                .await
+                .expect("pubsub manager"),
+        );
+        let audit = buzz_audit::AuditService::new(pool.clone());
+        let auth = buzz_auth::AuthService::new(config.auth.clone());
+        let search = buzz_search::SearchService::new(pool);
+        let workflow_engine = Arc::new(buzz_workflow::WorkflowEngine::new(
+            db.clone(),
+            buzz_workflow::WorkflowConfig::default(),
+        ));
+        let media_storage = buzz_media::MediaStorage::new(&config.media).expect("media storage");
+        let (state, _audit_shutdown) = AppState::new(
+            config,
+            db,
+            redis_pool,
+            audit,
+            pubsub,
+            auth,
+            search,
+            workflow_engine,
+            Keys::generate(),
+            media_storage,
+        );
+        setup_pool.close().await;
+        (Arc::new(state), tenant, human, workflow_id, definition)
+    }
+
+    fn workflow_trigger_event(keys: &Keys, workflow_id: Uuid, revision: &Event) -> Event {
+        EventBuilder::new(Kind::Custom(KIND_WORKFLOW_TRIGGER as u16), "")
+            .tags(vec![
+                Tag::parse(["d", workflow_id.to_string().as_str()]).expect("d tag"),
+                Tag::parse(["e", revision.id.to_hex().as_str()]).expect("revision tag"),
+            ])
+            .sign_with_keys(keys)
+            .expect("sign workflow trigger")
+    }
+
+    fn http_auth(keys: &Keys) -> IngestAuth {
+        IngestAuth::Http {
+            pubkey: keys.public_key(),
+            scopes: vec![buzz_auth::Scope::MessagesWrite],
+            auth_method: super::super::ingest::HttpAuthMethod::Nip98,
+        }
+    }
+
     fn workflow_event(
         keys: &Keys,
         workflow_id: Uuid,
@@ -1425,6 +1717,60 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn human_owner_manual_trigger_completes_with_one_connection() {
+        let (state, tenant, human, workflow_id, revision) = manual_trigger_test_context().await;
+        let trigger = workflow_trigger_event(&human, workflow_id, &revision);
+
+        let result = tokio::time::timeout(
+            std::time::Duration::from_secs(3),
+            handle_workflow_trigger(&tenant, &state, &trigger, &http_auth(&human)),
+        )
+        .await
+        .expect("human-owner trigger must not wait for a second pool connection")
+        .expect("human-owner trigger must succeed");
+        assert!(result.message.contains("\"run_id\""));
+    }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn concurrent_human_owner_manual_triggers_do_not_starve_one_connection_pool() {
+        let (state, tenant, human, workflow_id, revision) = manual_trigger_test_context().await;
+        let triggers = (0..8)
+            .map(|_| workflow_trigger_event(&human, workflow_id, &revision))
+            .collect::<Vec<_>>();
+
+        let results = tokio::time::timeout(std::time::Duration::from_secs(8), async {
+            let mut tasks = tokio::task::JoinSet::new();
+            for trigger in triggers {
+                let state = Arc::clone(&state);
+                let tenant = tenant.clone();
+                let auth = http_auth(&human);
+                tasks.spawn(async move {
+                    handle_workflow_trigger(&tenant, &state, &trigger, &auth).await
+                });
+            }
+            let mut results = Vec::new();
+            while let Some(result) = tasks.join_next().await {
+                results.push(result.expect("trigger task must not panic"));
+            }
+            results
+        })
+        .await
+        .expect("concurrent triggers must drain rather than pool-starve");
+
+        assert_eq!(results.len(), 8);
+        for result in results {
+            assert!(
+                result
+                    .expect("concurrent human-owner trigger must succeed")
+                    .accepted,
+                "manual trigger should be accepted"
+            );
+        }
+    }
+
     #[test]
     fn workflow_revision_parser_accepts_create_and_valid_update() {
         let revision = [0x42; 32];
@@ -1473,6 +1819,10 @@ mod tests {
         let workflow_id = Uuid::new_v4();
         let created_at = Timestamp::now().as_secs();
         let create = workflow_event(&keys, workflow_id, created_at, None, "create");
+        let channel_id = Uuid::parse_str(
+            exact_tag_value(&create, "h").expect("workflow definition channel tag"),
+        )
+        .expect("workflow definition channel UUID");
 
         let missing_revision = hex::encode([0x24; 32]);
         let missing_revision_update = workflow_event(
@@ -1493,15 +1843,22 @@ mod tests {
                 if message == "conflict: workflow revision does not exist"
         ));
 
-        let PersistResult::Inserted(tx) = persist_command_event(&db, &tenant, &create, None)
-            .await
-            .expect("persist create")
+        let PersistResult::Inserted(tx) =
+            persist_command_event(&db, &tenant, &create, Some(channel_id))
+                .await
+                .expect("persist create")
         else {
             panic!("first create must insert");
         };
         tx.commit().await.expect("commit create");
+        let stored_create = db
+            .get_event_by_id(tenant.community(), create.id.as_bytes())
+            .await
+            .expect("load persisted workflow definition")
+            .expect("persisted workflow definition");
+        assert_eq!(stored_create.channel_id, Some(channel_id));
         assert!(matches!(
-            persist_command_event(&db, &tenant, &create, None)
+            persist_command_event(&db, &tenant, &create, Some(channel_id))
                 .await
                 .expect("replay create"),
             PersistResult::Duplicate

crates/buzz-relay/src/router.rs

@@ -75,6 +75,10 @@ pub fn build_router(state: Arc<AppState>) -> Router {
         // Relay-owned third-party GIF metadata proxy (NIP-98 auth).
         .route(api::gifs::SEARCH_PATH, post(api::gifs::search))
         .route(api::gifs::SHARE_PATH, post(api::gifs::share))
+        .route(
+            "/workflows/{workflow_id}/revision",
+            get(api::workflows::workflow_revision),
+        )
         .route(
             "/workflows/{workflow_id}/runs",
             get(api::workflows::workflow_runs),

crates/buzz-sdk/src/builders.rs

@@ -1640,9 +1640,16 @@ pub fn build_workflow_delete(
     build_delete_addressable(KIND_WORKFLOW_DEF, author_pubkey, &workflow_id.to_string())
 }
 
-/// Build a workflow trigger event (kind 46020).
-pub fn build_workflow_trigger(workflow_id: Uuid) -> Result<EventBuilder, SdkError> {
-    let tags = vec![tag(&["d", &workflow_id.to_string()])?];
+/// Build a workflow trigger event (kind 46020) bound to an exact signed revision.
+pub fn build_workflow_trigger(
+    workflow_id: Uuid,
+    definition_event_id: &str,
+) -> Result<EventBuilder, SdkError> {
+    let revision = check_hex_exact(definition_event_id, 64, "definition_event_id")?;
+    let tags = vec![
+        tag(&["d", &workflow_id.to_string()])?,
+        tag(&["e", &revision])?,
+    ];
     Ok(EventBuilder::new(Kind::Custom(KIND_WORKFLOW_TRIGGER as u16), "").tags(tags))
 }
 
@@ -4010,9 +4017,11 @@ mod tests {
     #[test]
     fn workflow_trigger_happy_path() {
         let wid = uuid();
-        let ev = sign(build_workflow_trigger(wid).unwrap());
+        let ev = sign(build_workflow_trigger(wid, &"ab".repeat(32)).unwrap());
         assert_eq!(ev.kind.as_u16(), 46020);
         assert!(has_tag(&ev, "d", &wid.to_string()));
+        assert!(has_tag(&ev, "e", &"ab".repeat(32)));
+        assert!(build_workflow_trigger(wid, "not-an-event-id").is_err());
     }
 
     #[test]

crates/buzz-test-client/tests/e2e_workflow_agent_owner.rs

@@ -0,0 +1,205 @@
+//! End-to-end authorization and revision-binding coverage for agent-owned workflows.
+//!
+//! Run against a local relay with:
+//! `cargo test -p buzz-test-client --test e2e_workflow_agent_owner -- --ignored`
+
+use buzz_sdk::nip_oa;
+use buzz_test_client::BuzzTestClient;
+use nostr::{EventBuilder, Keys, Kind, Tag};
+
+fn relay_url() -> String {
+    std::env::var("RELAY_URL").unwrap_or_else(|_| "ws://localhost:3000".to_string())
+}
+
+fn workflow_yaml(text: &str) -> String {
+    format!(
+        "name: Agent-owned workflow\n\
+         trigger:\n\
+         \x20 on: webhook\n\
+         steps:\n\
+         \x20 - id: notify\n\
+         \x20   action: send_message\n\
+         \x20   text: {text}\n"
+    )
+}
+
+async fn connect_agent_with_owner(agent: &Keys, owner: &Keys) -> BuzzTestClient {
+    let tag_json =
+        nip_oa::compute_auth_tag(owner, &agent.public_key(), "").expect("compute NIP-OA auth tag");
+    let auth_tag = nip_oa::parse_auth_tag(&tag_json).expect("parse NIP-OA auth tag");
+    let mut client = BuzzTestClient::connect_unauthenticated(&relay_url())
+        .await
+        .expect("connect agent");
+    client
+        .authenticate_with_nip_oa(agent, &auth_tag)
+        .await
+        .expect("authenticate agent with NIP-OA");
+    client
+}
+
+fn trigger(keys: &Keys, workflow_id: uuid::Uuid, revision: &str) -> nostr::Event {
+    buzz_sdk::build_workflow_trigger(workflow_id, revision)
+        .expect("build workflow trigger")
+        .sign_with_keys(keys)
+        .expect("sign workflow trigger")
+}
+
+#[tokio::test]
+#[ignore]
+async fn agent_and_human_owner_trigger_exact_revision_but_unrelated_member_cannot() {
+    let owner = Keys::generate();
+    let agent = Keys::generate();
+    let unrelated = Keys::generate();
+    let channel_id = uuid::Uuid::new_v4();
+    let workflow_id = uuid::Uuid::new_v4();
+
+    // NIP-OA authentication materializes the immutable community-scoped
+    // agent→owner relationship used by workflow authorization.
+    let mut agent_client = connect_agent_with_owner(&agent, &owner).await;
+
+    let create_channel = EventBuilder::new(Kind::Custom(9007), "")
+        .tags([
+            Tag::parse(["h", &channel_id.to_string()]).unwrap(),
+            Tag::parse(["name", "workflow-agent-owner-e2e"]).unwrap(),
+            Tag::parse(["channel_type", "stream"]).unwrap(),
+            Tag::parse(["visibility", "open"]).unwrap(),
+        ])
+        .sign_with_keys(&agent)
+        .unwrap();
+    let response = agent_client
+        .send_event(create_channel)
+        .await
+        .expect("create channel");
+    assert!(response.accepted, "channel rejected: {}", response.message);
+
+    let definition =
+        buzz_sdk::build_workflow_def(channel_id, workflow_id, &workflow_yaml("owner-triggered"))
+            .expect("build workflow definition")
+            .sign_with_keys(&agent)
+            .expect("sign workflow definition");
+    let revision = definition.id.to_hex();
+    let response = agent_client
+        .send_event(definition)
+        .await
+        .expect("define workflow");
+    assert!(
+        response.accepted,
+        "definition rejected: {}",
+        response.message
+    );
+
+    let add_unrelated = EventBuilder::new(Kind::Custom(9000), "")
+        .tags([
+            Tag::parse(["h", &channel_id.to_string()]).unwrap(),
+            Tag::parse(["p", &unrelated.public_key().to_hex()]).unwrap(),
+        ])
+        .sign_with_keys(&agent)
+        .expect("sign add-member event");
+    let response = agent_client
+        .send_event(add_unrelated)
+        .await
+        .expect("add unrelated member");
+    assert!(
+        response.accepted,
+        "add member rejected: {}",
+        response.message
+    );
+
+    let response = agent_client
+        .send_event(trigger(&agent, workflow_id, &revision))
+        .await
+        .expect("send agent trigger");
+    assert!(
+        response.accepted,
+        "agent trigger rejected: {}",
+        response.message
+    );
+
+    let mut unrelated_client = BuzzTestClient::connect(&relay_url(), &unrelated)
+        .await
+        .expect("connect unrelated user");
+    let response = unrelated_client
+        .send_event(trigger(&unrelated, workflow_id, &revision))
+        .await
+        .expect("send unrelated trigger");
+    assert!(!response.accepted, "unrelated member triggered workflow");
+    assert!(response.message.contains("not authorized to trigger"));
+
+    let mut owner_client = BuzzTestClient::connect(&relay_url(), &owner)
+        .await
+        .expect("connect human owner");
+    let response = owner_client
+        .send_event(trigger(&owner, workflow_id, &revision))
+        .await
+        .expect("send owner trigger");
+    assert!(
+        response.accepted,
+        "owner trigger rejected: {}",
+        response.message
+    );
+
+    let missing_revision = EventBuilder::new(Kind::Custom(46020), "")
+        .tags([Tag::parse(["d", &workflow_id.to_string()]).unwrap()])
+        .sign_with_keys(&owner)
+        .unwrap();
+    let response = owner_client
+        .send_event(missing_revision)
+        .await
+        .expect("send unbound trigger");
+    assert!(!response.accepted, "trigger without revision was accepted");
+    assert!(response.message.contains("workflow revision e tag"));
+
+    let wrong_revision = "ab".repeat(32);
+    let response = owner_client
+        .send_event(trigger(&owner, workflow_id, &wrong_revision))
+        .await
+        .expect("send wrong-revision trigger");
+    assert!(!response.accepted, "wrong revision was accepted");
+    assert!(response
+        .message
+        .contains("does not match current definition"));
+
+    // A valid old revision becomes stale immediately after an agent-signed update.
+    // NIP-33 revisions use second-granularity timestamps. Advance past the
+    // creation second so this update is newer regardless of the event-ID
+    // tie-breaker for equal timestamps.
+    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+    let update = buzz_sdk::build_workflow_update(
+        channel_id,
+        workflow_id,
+        &workflow_yaml("updated"),
+        &revision,
+    )
+    .expect("build workflow update")
+    .sign_with_keys(&agent)
+    .expect("sign workflow update");
+    let current_revision = update.id.to_hex();
+    let response = agent_client
+        .send_event(update)
+        .await
+        .expect("update workflow");
+    assert!(response.accepted, "update rejected: {}", response.message);
+
+    let response = owner_client
+        .send_event(trigger(&owner, workflow_id, &revision))
+        .await
+        .expect("send stale trigger");
+    assert!(!response.accepted, "stale revision was accepted");
+    assert!(response
+        .message
+        .contains("does not match current definition"));
+
+    let response = owner_client
+        .send_event(trigger(&owner, workflow_id, &current_revision))
+        .await
+        .expect("send current trigger");
+    assert!(
+        response.accepted,
+        "current revision rejected: {}",
+        response.message
+    );
+
+    agent_client.disconnect().await.ok();
+    unrelated_client.disconnect().await.ok();
+    owner_client.disconnect().await.ok();
+}

crates/buzz-workflow/src/executor.rs

@@ -43,6 +43,9 @@ pub struct TriggerContext {
     pub is_reply: bool,
     /// Arbitrary webhook body fields (webhook trigger).
     pub webhook_fields: HashMap<String, String>,
+    /// Exact owner-signed kind:30620 definition revision executed by this run.
+    #[serde(default)]
+    pub definition_event_id: String,
 }
 
 impl TriggerContext {
@@ -1312,6 +1315,7 @@ mod tests {
             message_id: "event-id-hex".to_owned(),
             is_reply: false,
             webhook_fields: HashMap::new(),
+            definition_event_id: String::new(),
         }
     }
 

crates/buzz-workflow/src/lib.rs

@@ -44,7 +44,9 @@ use std::collections::HashMap;
 use std::sync::Arc;
 use std::sync::OnceLock;
 
-use buzz_core::kind::{event_kind_u32, is_workflow_execution_kind, KIND_REACTION};
+use buzz_core::kind::{
+    event_kind_u32, is_workflow_execution_kind, KIND_REACTION, KIND_WORKFLOW_DEF,
+};
 use buzz_core::tenant::CommunityId;
 use buzz_db::workflow::RunStatus;
 use buzz_db::Db;
@@ -122,6 +124,62 @@ impl WorkflowEngine {
         }
     }
 
+    /// Load and verify the exact owner-signed definition bound to a run.
+    ///
+    /// The mutable `workflows` row supplies only immutable identity/channel
+    /// binding. Definition content always comes from the run's signed event;
+    /// legacy runs without a revision fail closed.
+    pub async fn load_run_definition(
+        &self,
+        community_id: CommunityId,
+        run_id: Uuid,
+    ) -> Result<(buzz_db::workflow::WorkflowRunRecord, WorkflowDef), WorkflowError> {
+        let run = self.db.get_workflow_run(community_id, run_id).await?;
+        let revision = run.definition_event_id.as_deref().ok_or_else(|| {
+            WorkflowError::InvalidDefinition(
+                "workflow run has no owner-signed definition revision".into(),
+            )
+        })?;
+        let workflow = self.db.get_workflow(community_id, run.workflow_id).await?;
+        let stored = self
+            .db
+            .get_event_by_id_including_deleted(community_id, revision)
+            .await?
+            .ok_or_else(|| {
+                WorkflowError::InvalidDefinition(
+                    "workflow run definition event is unavailable".into(),
+                )
+            })?;
+        let event = &stored.event;
+        let workflow_id = run.workflow_id.to_string();
+        let channel_id = workflow.channel_id.map(|id| id.to_string());
+        let exact_tag = |name: &str| {
+            let mut values = event.tags.iter().filter_map(|tag| {
+                (tag.kind().to_string() == name)
+                    .then(|| tag.content())
+                    .flatten()
+            });
+            let value = values.next();
+            value.filter(|_| values.next().is_none())
+        };
+        if event.id.as_bytes() != revision
+            || !event.verify_id()
+            || !event.verify_signature()
+            || event_kind_u32(event) != KIND_WORKFLOW_DEF
+            || event.pubkey.to_bytes().as_slice() != workflow.owner_pubkey
+            || exact_tag("d") != Some(workflow_id.as_str())
+            || channel_id.is_none()
+            || exact_tag("h") != channel_id.as_deref()
+            || stored.channel_id != workflow.channel_id
+        {
+            return Err(WorkflowError::InvalidDefinition(
+                "workflow run definition event binding mismatch".into(),
+            ));
+        }
+        let (definition, _) = Self::parse_yaml(&event.content)?;
+        Ok((run, definition))
+    }
+
     /// Drop the cached enabled-workflow list for a channel.
     ///
     /// Must be called after any write to a workflow's trigger eligibility or
@@ -407,6 +465,13 @@ impl WorkflowEngine {
                 .create_workflow_run(
                     community_id,
                     workflow.id,
+                    match workflow.definition_event_id.as_deref() {
+                        Some(revision) => revision,
+                        None => {
+                            tracing::warn!(workflow_id = %workflow.id, "Skipping workflow — signed revision unavailable");
+                            continue;
+                        }
+                    },
                     Some(&trigger_event_id_bytes),
                     Some(&trigger_ctx_json),
                 )
@@ -426,13 +491,22 @@ impl WorkflowEngine {
             );
 
             let engine = Arc::clone(self);
-            let def_clone = def.clone();
             let ctx_clone = trigger_ctx.clone();
 
             tokio::spawn(async move {
-                let result =
-                    executor::execute_run(&engine, community_id, run_id, &def_clone, &ctx_clone)
-                        .await;
+                let result = match engine.load_run_definition(community_id, run_id).await {
+                    Ok((_, definition)) => {
+                        executor::execute_run(
+                            &engine,
+                            community_id,
+                            run_id,
+                            &definition,
+                            &ctx_clone,
+                        )
+                        .await
+                    }
+                    Err(error) => Err((error, PartialProgress::default())),
+                };
                 engine
                     .finalize_run(community_id, run_id, result, None)
                     .await;
@@ -669,6 +743,13 @@ impl WorkflowEngine {
                     .create_workflow_run(
                         community_id,
                         workflow.id,
+                        match workflow.definition_event_id.as_deref() {
+                            Some(revision) => revision,
+                            None => {
+                                tracing::warn!(workflow_id = %workflow.id, "Cron tick: signed revision unavailable");
+                                continue;
+                            }
+                        },
                         None, // no trigger event for cron
                         trigger_ctx_json.as_ref(),
                     )
@@ -721,17 +802,21 @@ impl WorkflowEngine {
                 );
 
                 let engine = Arc::clone(self);
-                let def_clone = def.clone();
                 let ctx_clone = trigger_ctx.clone();
                 tokio::spawn(async move {
-                    let result = executor::execute_run(
-                        &engine,
-                        community_id,
-                        run_id,
-                        &def_clone,
-                        &ctx_clone,
-                    )
-                    .await;
+                    let result = match engine.load_run_definition(community_id, run_id).await {
+                        Ok((_, definition)) => {
+                            executor::execute_run(
+                                &engine,
+                                community_id,
+                                run_id,
+                                &definition,
+                                &ctx_clone,
+                            )
+                            .await
+                        }
+                        Err(error) => Err((error, PartialProgress::default())),
+                    };
                     engine
                         .finalize_run(community_id, run_id, result, None)
                         .await;
@@ -999,6 +1084,7 @@ pub fn build_trigger_context(event: &buzz_core::StoredEvent) -> executor::Trigge
         message_id,
         is_reply: event_is_reply(&event.event),
         webhook_fields: HashMap::new(),
+        definition_event_id: String::new(),
     }
 }
 
@@ -1974,17 +2060,25 @@ steps:
             "enabled": true,
         })
         .to_string();
-        let workflow_id = db
-            .create_workflow(
-                community,
-                Some(channel_id),
-                &member,
-                "sec006-event",
-                &def_json,
-                &[0u8; 32],
-            )
-            .await
-            .expect("create workflow");
+        // Seed through the production insert path (`upsert_workflow`), which
+        // binds a signed definition revision: run creation now skips any
+        // workflow without one (legacy NULL-revision rows fail closed).
+        let workflow_id = Uuid::new_v4();
+        let mut tx = db.begin_transaction().await.expect("begin tx");
+        db.upsert_workflow(
+            &mut tx,
+            community,
+            workflow_id,
+            Some(channel_id),
+            &member,
+            "sec006-event",
+            &def_json,
+            &[0u8; 32],
+            &[0x42u8; 32],
+        )
+        .await
+        .expect("create workflow");
+        tx.commit().await.expect("commit workflow");
 
         let engine = Arc::new(WorkflowEngine::new(db.clone(), WorkflowConfig::default()));
 
@@ -2039,29 +2133,41 @@ steps:
         })
         .to_string();
 
-        // Same definition, two owners: plain member vs channel owner.
-        let wf_member = db
-            .create_workflow(
-                community,
-                Some(channel_id),
-                &member,
-                "hook-member",
-                &def_json,
-                &[0u8; 32],
-            )
-            .await
-            .expect("create member workflow");
-        let wf_owner = db
-            .create_workflow(
-                community,
-                Some(channel_id),
-                &creator,
-                "hook-owner",
-                &def_json,
-                &[1u8; 32],
-            )
-            .await
-            .expect("create owner workflow");
+        // Same definition, two owners: plain member vs channel owner. Seed
+        // through `upsert_workflow` so each row carries a bound signed
+        // revision (run creation skips NULL-revision rows fail-closed).
+        let wf_member = Uuid::new_v4();
+        let mut tx = db.begin_transaction().await.expect("begin member tx");
+        db.upsert_workflow(
+            &mut tx,
+            community,
+            wf_member,
+            Some(channel_id),
+            &member,
+            "hook-member",
+            &def_json,
+            &[0u8; 32],
+            &[0x42u8; 32],
+        )
+        .await
+        .expect("create member workflow");
+        tx.commit().await.expect("commit member workflow");
+        let wf_owner = Uuid::new_v4();
+        let mut tx = db.begin_transaction().await.expect("begin owner tx");
+        db.upsert_workflow(
+            &mut tx,
+            community,
+            wf_owner,
+            Some(channel_id),
+            &creator,
+            "hook-owner",
+            &def_json,
+            &[1u8; 32],
+            &[0x43u8; 32],
+        )
+        .await
+        .expect("create owner workflow");
+        tx.commit().await.expect("commit owner workflow");
 
         let engine = Arc::new(WorkflowEngine::new(db.clone(), WorkflowConfig::default()));
         engine
@@ -2087,4 +2193,89 @@ steps:
             "channel owner's call_webhook workflow fires"
         );
     }
+
+    /// A run is bound to the exact signed revision persisted at creation.
+    /// Replacing the definition (NIP-33) soft-deletes the old kind-30620
+    /// event, but historical runs must keep loading their bound revision —
+    /// `load_run_definition` reads through soft deletion.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn run_definition_loads_after_revision_soft_deleted() {
+        let db = setup_db().await;
+        let owner_keys = nostr::Keys::generate();
+        let owner = owner_keys.public_key().to_bytes().to_vec();
+        let member = nostr::Keys::generate().public_key().to_bytes().to_vec();
+        let (community, channel_id) = setup_channel(&db, &owner, &member).await;
+
+        let workflow_id = Uuid::new_v4();
+        let yaml = concat!(
+            "name: revision-history\n",
+            "trigger:\n  on: message_posted\n",
+            "steps:\n  - id: s1\n    action: send_message\n    text: hi\n",
+        );
+        let definition_event =
+            nostr::EventBuilder::new(nostr::Kind::Custom(KIND_WORKFLOW_DEF as u16), yaml)
+                .tags([
+                    nostr::Tag::parse(["d", &workflow_id.to_string()]).expect("d tag"),
+                    nostr::Tag::parse(["h", &channel_id.to_string()]).expect("h tag"),
+                ])
+                .sign_with_keys(&owner_keys)
+                .expect("sign definition");
+        db.insert_event(community, &definition_event, Some(channel_id))
+            .await
+            .expect("store definition event");
+
+        let (def, _) = WorkflowEngine::parse_yaml(yaml).expect("parse yaml");
+        let def_json = serde_json::to_string(&def).expect("serialize definition");
+        let mut tx = db.begin_transaction().await.expect("begin tx");
+        db.upsert_workflow(
+            &mut tx,
+            community,
+            workflow_id,
+            Some(channel_id),
+            &owner,
+            "revision-history",
+            &def_json,
+            &[0u8; 32],
+            definition_event.id.as_bytes(),
+        )
+        .await
+        .expect("create workflow");
+        tx.commit().await.expect("commit workflow");
+
+        let run_id = db
+            .create_workflow_run(
+                community,
+                workflow_id,
+                definition_event.id.as_bytes(),
+                None,
+                None,
+            )
+            .await
+            .expect("create run");
+
+        let engine = Arc::new(WorkflowEngine::new(db.clone(), WorkflowConfig::default()));
+
+        // Bound revision loads while the event is live.
+        engine
+            .load_run_definition(community, run_id)
+            .await
+            .expect("load definition before deletion");
+
+        // Definition replacement soft-deletes the superseded revision event.
+        assert!(
+            db.soft_delete_event(community, definition_event.id.as_bytes())
+                .await
+                .expect("soft delete revision"),
+            "revision event must exist to be soft-deleted"
+        );
+
+        // The historical run still resolves its exact signed revision.
+        let (run, loaded) = engine
+            .load_run_definition(community, run_id)
+            .await
+            .expect("load definition after soft deletion");
+        assert_eq!(run.id, run_id);
+        assert_eq!(loaded.name, "revision-history");
+    }
 }

desktop/src-tauri/src/commands/workflows.rs

@@ -323,7 +323,13 @@ pub async fn trigger_workflow(
     workflow_id: String,
     state: State<'_, AppState>,
 ) -> Result<WorkflowTriggerWire, String> {
-    let builder = events::build_workflow_trigger(&workflow_id)?;
+    // Resolve the current signed definition at trigger time. The relay binds
+    // authorization and execution to this exact revision and rejects a stale
+    // result if an update races this command.
+    let revision: nostr::Event =
+        get_relay_json(&state, &format!("/workflows/{workflow_id}/revision")).await?;
+    let definition_event_id = revision.id.to_hex();
+    let builder = events::build_workflow_trigger(&workflow_id, &definition_event_id)?;
     let result = submit_event(builder, &state).await?;
     trigger_wire_from_message(workflow_id, &result.message)
 }

desktop/src-tauri/src/events.rs

@@ -769,6 +769,31 @@ pub use workflows::{
 mod tests {
     use super::*;
     use nostr::Keys;
+    #[test]
+    fn workflow_trigger_binds_exact_definition_revision() {
+        let workflow_id = Uuid::new_v4().to_string();
+        let revision = "ab".repeat(32);
+        let event = build_workflow_trigger(&workflow_id, &revision)
+            .expect("build workflow trigger")
+            .sign_with_keys(&Keys::generate())
+            .expect("sign workflow trigger");
+        let tags: Vec<Vec<String>> = event
+            .tags
+            .iter()
+            .map(|tag| tag.as_slice().to_vec())
+            .collect();
+
+        assert_eq!(event.kind, Kind::Custom(46020));
+        assert_eq!(
+            tags,
+            vec![
+                vec!["d".to_string(), workflow_id.clone()],
+                vec!["e".to_string(), revision.clone()],
+            ]
+        );
+        assert!(build_workflow_trigger(&workflow_id, "not-an-event-id").is_err());
+    }
+
     #[test]
     fn channel_builders_reject_hash_only_names() {
         let channel_id = Uuid::new_v4();

desktop/src-tauri/src/events/workflows.rs

@@ -31,9 +31,16 @@ pub fn build_workflow_delete(
     Ok(EventBuilder::new(Kind::Custom(5), "").tags(tags))
 }
 
-/// Kind 46020 — trigger a workflow run by id.
-pub fn build_workflow_trigger(workflow_id: &str) -> Result<EventBuilder, String> {
-    let tags = vec![tag(vec!["d", workflow_id])?];
+/// Kind 46020 — trigger a workflow run by id, bound to one exact definition revision.
+pub fn build_workflow_trigger(
+    workflow_id: &str,
+    definition_event_id: &str,
+) -> Result<EventBuilder, String> {
+    EventId::from_hex(definition_event_id).map_err(|_| "invalid workflow revision".to_string())?;
+    let tags = vec![
+        tag(vec!["d", workflow_id])?,
+        tag(vec!["e", definition_event_id])?,
+    ];
     Ok(EventBuilder::new(Kind::Custom(46020), "").tags(tags))
 }
 

migrations/0033_workflow_definition_event_id.sql

@@ -0,0 +1,6 @@
+-- Bind each materialized workflow row to the exact owner-signed kind:30620
+-- revision that produced it. Existing rows remain nullable until re-saved;
+-- revision-bound execution fails closed when the revision is unavailable.
+ALTER TABLE workflows
+    ADD COLUMN definition_event_id BYTEA
+        CHECK (definition_event_id IS NULL OR octet_length(definition_event_id) = 32);

migrations/0034_workflow_run_definition_event_id.sql

@@ -0,0 +1,5 @@
+-- Bind each new workflow run to the exact signed definition it executes.
+-- Existing rows remain NULL and all resume/execution paths fail them closed.
+ALTER TABLE workflow_runs
+    ADD COLUMN definition_event_id BYTEA
+        CHECK (definition_event_id IS NULL OR octet_length(definition_event_id) = 32);

schema/schema.sql

@@ -368,6 +368,11 @@ CREATE TABLE workflows (
     channel_id      UUID,
     definition      JSONB NOT NULL,
     definition_hash BYTEA NOT NULL,
+    -- Exact owner-signed kind:30620 revision that materialized this row.
+    -- Nullable only for pre-0033 rows; revision-bound execution fails closed until re-saved.
+    definition_event_id BYTEA CHECK (
+        definition_event_id IS NULL OR octet_length(definition_event_id) = 32
+    ),
     status          workflow_status NOT NULL DEFAULT 'active',
     enabled         BOOLEAN NOT NULL DEFAULT TRUE,
     created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
@@ -388,6 +393,9 @@ CREATE TABLE workflow_runs (
     community_id        UUID NOT NULL REFERENCES communities(id),
     id                  UUID NOT NULL DEFAULT gen_random_uuid(),
     workflow_id         UUID NOT NULL,
+    definition_event_id BYTEA CHECK (
+        definition_event_id IS NULL OR octet_length(definition_event_id) = 32
+    ),
     status              run_status NOT NULL DEFAULT 'pending',
     trigger_event_id    BYTEA,
     current_step        INT NOT NULL DEFAULT 0,