Add owner-authorized manual workflow triggers

.github/workflows/ci.yml

@@ -707,6 +707,16 @@ jobs:
             --run-ignored ignored-only
         env:
           BUZZ_TEST_DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz
+      - name: Manual workflow trigger PostgreSQL tests
+        # Pin delivery identity, result-preserving replay, authorization, and
+        # exact-revision execution; infra-free unit runs ignore these cases.
+        run: |
+          cargo nextest run \
+            --archive-file target/ci/backend-integration-tests.tar.zst \
+            -E 'package(buzz-relay) and (test(/^handlers::command_executor::tests::/) or test(/revoked_immutable_owner_receives_only_revision_id/))' \
+            --run-ignored ignored-only
+        env:
+          BUZZ_TEST_DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz
       - name: Database pressure observability PostgreSQL tests
         # Explicit pool acquisition and advisory-lock metrics require real
         # Postgres and are ignored by the infrastructure-free unit-test job.

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

@@ -162,44 +162,56 @@ 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
-/// event content (MCP parity). When omitted, the event content is `{}`.
+/// event content (MCP parity). When omitted, the event content is empty.
 pub async fn cmd_trigger_workflow(
     client: &BuzzClient,
     workflow_id: &str,
     inputs: Option<&str>,
 ) -> Result<(), CliError> {
     let wf_uuid = parse_uuid(workflow_id)?;
+    let revision = current_workflow_revision(client, workflow_id).await?;
+
+    let builder = workflow_trigger_builder(wf_uuid, &revision, inputs)?;
+    let event = client.sign_event(builder)?;
+    let resp = client.submit_event(event).await?;
+    println!("{}", normalize_write_response(&resp));
+    Ok(())
+}
 
+fn workflow_trigger_builder(
+    workflow_id: uuid::Uuid,
+    revision: &str,
+    inputs: Option<&str>,
+) -> Result<nostr::EventBuilder, CliError> {
     if let Some(raw) = inputs {
-        // Parse and validate it is a JSON object, then build the event manually
-        // so we can embed the inputs as the event content.
         let parsed: serde_json::Value = serde_json::from_str(raw)
             .map_err(|e| CliError::Usage(format!("--inputs is not valid JSON: {e}")))?;
-        if !parsed.is_object() {
-            return Err(CliError::Usage("--inputs must be a JSON object".into()));
-        }
-        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 builder = EventBuilder::new(
-            Kind::Custom(buzz_sdk::kind::KIND_WORKFLOW_TRIGGER as u16),
-            &content,
-        )
-        .tags(tags);
-        let event = client.sign_event(builder)?;
-        let resp = client.submit_event(event).await?;
-        println!("{}", normalize_write_response(&resp));
+        let object = parsed
+            .as_object()
+            .ok_or_else(|| CliError::Usage("--inputs must be a JSON object".into()))?;
+        buzz_sdk::build_workflow_trigger_with_inputs(workflow_id, revision, object).map_err(sdk_err)
     } else {
-        let builder = buzz_sdk::build_workflow_trigger(wf_uuid).map_err(sdk_err)?;
-        let event = client.sign_event(builder)?;
-        let resp = client.submit_event(event).await?;
-        println!("{}", normalize_write_response(&resp));
+        buzz_sdk::build_workflow_trigger(workflow_id, revision).map_err(sdk_err)
     }
-    Ok(())
 }
 
 /// Approve or deny a workflow step — sign and submit a kind:46030 (grant) or 46031 (deny) event.
@@ -254,3 +266,32 @@ pub async fn dispatch(cmd: crate::WorkflowsCmd, client: &BuzzClient) -> Result<(
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn trigger_builders_preserve_inputs_and_distinguish_same_second_operations() {
+        let keys = nostr::Keys::generate();
+        let workflow_id = uuid::Uuid::new_v4();
+        let revision = "ab".repeat(32);
+        let timestamp = nostr::Timestamp::from(1_700_000_000);
+        for inputs in [None, Some(r#"{"message":"hello"}"#)] {
+            let build = || {
+                workflow_trigger_builder(workflow_id, &revision, inputs)
+                    .unwrap()
+                    .custom_created_at(timestamp)
+                    .sign_with_keys(&keys)
+                    .unwrap()
+            };
+            let first = build();
+            let second = build();
+            assert_ne!(first.id, second.id);
+            assert_eq!(first.content, inputs.unwrap_or(""));
+        }
+        for invalid in ["not-json", "[]", "null", "1"] {
+            assert!(workflow_trigger_builder(workflow_id, &revision, Some(invalid)).is_err());
+        }
+    }
+}

crates/buzz-db/src/store/channel.rs

@@ -21,10 +21,11 @@ pub use buzz_core::channel::{ChannelType, ChannelVisibility, MemberRole};
 // Keep the established channel module paths compatible while membership SQL
 // and invariants live in their dedicated store module.
 pub use crate::channel_members::{
-    add_member, get_accessible_channel_ids, get_accessible_channels, get_bot_members,
-    get_member_count, get_member_counts_bulk, get_member_role, get_members, get_members_bulk,
-    get_users_bulk, is_member, list_large_channel_rosters_needing_reconciliation,
-    lock_member_snapshot, membership_pairs, remove_member, verify_channel_roster_fence_behavior,
+    acquire_channel_membership_lock, add_member, get_accessible_channel_ids,
+    get_accessible_channels, get_bot_members, get_member_count, get_member_counts_bulk,
+    get_member_role, get_member_role_in_transaction, get_members, get_members_bulk, get_users_bulk,
+    is_member, list_large_channel_rosters_needing_reconciliation, lock_member_snapshot,
+    membership_pairs, remove_member, verify_channel_roster_fence_behavior,
     verify_channel_roster_fence_catalog, AccessibleChannel, BotChannelEntry, BotMemberRecord,
     LargeChannelRoster, LockedMemberSnapshot, MemberRecord, UserRecord,
 };

crates/buzz-db/src/store/channel_members.rs

@@ -173,7 +173,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,
@@ -1378,6 +1378,29 @@ impl Db {
     }
 }
 
+/// Get an active member role using the caller's transaction.
+///
+/// Authorization callers must acquire [`acquire_channel_membership_lock`] first
+/// and hold the transaction through the dependent mutation.
+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()?)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/buzz-db/src/store/event.rs

@@ -1026,6 +1026,26 @@ pub async fn get_event_by_id(
     }
 }
 
+/// Fetch a single live event by ID on the caller's transaction.
+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.
 ///
@@ -1349,6 +1369,17 @@ pub async fn insert_event_with_thread_metadata(
 }
 
 impl Db {
+    /// Fetch a live event by ID using 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>> {
+        crate::event::get_event_by_id_in_transaction(tx, community_id, id_bytes).await
+    }
+
     /// Inserts an event. Returns `(StoredEvent, was_inserted)` — `false` on duplicate.
     #[datastore_span(name = "insert_event", system = "postgresql")]
     pub async fn insert_event(

crates/buzz-db/src/store/workflow.rs

@@ -409,6 +409,33 @@ pub async fn get_workflow(
     row_to_workflow_record(row)
 }
 
+/// Fetch and share-lock one workflow on an existing transaction.
+///
+/// Definition replacement updates this row, so holding this lock through run
+/// creation keeps revision validation and the dependent run atomic.
+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`].
@@ -817,6 +844,34 @@ pub async fn delete_workflow_for_owner(
 
 // -- Workflow Run CRUD --------------------------------------------------------
 
+/// Insert a new exact-revision workflow run on the caller's transaction.
+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
@@ -851,6 +906,24 @@ pub async fn create_workflow_run(
     Ok(id)
 }
 
+/// Find the committed result of a manual trigger within its tenant and workflow.
+/// The command event and run commit atomically, so a retry can recover this ID.
+pub async fn get_workflow_run_id_by_trigger(
+    pool: &PgPool,
+    community_id: CommunityId,
+    workflow_id: Uuid,
+    trigger_event_id: &[u8],
+) -> Result<Option<Uuid>> {
+    Ok(sqlx::query_scalar(
+        "SELECT id FROM workflow_runs WHERE community_id = $1 AND workflow_id = $2 AND trigger_event_id = $3",
+    )
+    .bind(community_id.as_uuid())
+    .bind(workflow_id)
+    .bind(trigger_event_id)
+    .fetch_optional(pool)
+    .await?)
+}
+
 /// Fetch a single workflow run by ID, scoped to its community.
 pub async fn get_workflow_run(
     pool: &PgPool,
@@ -1297,6 +1370,56 @@ pub async fn find_by_owner_and_name(
 // -- Run and approval Db API --------------------------------------------------
 
 impl Db {
+    /// Recover the committed run ID for an exact manual-trigger retry.
+    #[datastore_span(name = "get_workflow_run_id_by_trigger", system = "postgresql")]
+    pub async fn get_workflow_run_id_by_trigger(
+        &self,
+        community_id: CommunityId,
+        workflow_id: Uuid,
+        trigger_event_id: &[u8],
+    ) -> Result<Option<Uuid>> {
+        crate::workflow::get_workflow_run_id_by_trigger(
+            &self.pool,
+            community_id,
+            workflow_id,
+            trigger_event_id,
+        )
+        .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<crate::workflow::WorkflowRecord> {
+        crate::workflow::get_workflow_for_share_in_transaction(tx, community_id, id).await
+    }
+
+    /// Create an exact-revision 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> {
+        crate::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(