Reconcile provable legacy workflow revisions

crates/buzz-db/src/lib.rs

@@ -3330,27 +3330,46 @@ impl Db {
     #[datastore_span(name = "upsert_workflow", system = "postgresql")]
     pub async fn upsert_workflow(
         &self,
+        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
         community_id: CommunityId,
         id: Uuid,
         channel_id: Option<Uuid>,
         owner_pubkey: &[u8],
         name: &str,
         definition_json: &str,
         definition_hash: &[u8],
+        definition_event_id: &[u8],
     ) -> Result<()> {
         workflow::upsert_workflow(
-            &self.pool,
+            tx,
             community_id,
             id,
             channel_id,
             owner_pubkey,
             name,
             definition_json,
             definition_hash,
+            definition_event_id,
         )
         .await
     }
 
+    /// List workflows that have not yet captured an exact signed revision.
+    #[datastore_span(name = "list_legacy_workflows", system = "postgresql")]
+    pub async fn list_legacy_workflows(&self) -> Result<Vec<workflow::WorkflowRecord>> {
+        workflow::list_legacy_workflows(&self.pool).await
+    }
+
+    /// Compare-and-set an exact revision onto an unchanged legacy workflow snapshot.
+    #[datastore_span(name = "bind_legacy_workflow_revision", system = "postgresql")]
+    pub async fn bind_legacy_workflow_revision(
+        &self,
+        workflow: &workflow::WorkflowRecord,
+        definition_event_id: &[u8],
+    ) -> Result<bool> {
+        workflow::bind_legacy_workflow_revision(&self.pool, workflow, definition_event_id).await
+    }
+
     /// Fetch a single workflow by ID, scoped to its community.
     #[datastore_span(name = "get_workflow", system = "postgresql")]
     pub async fn get_workflow(
@@ -3548,13 +3567,15 @@ impl Db {
         &self,
         community_id: CommunityId,
         workflow_id: Uuid,
+        definition_event_id: Option<&[u8]>,
         trigger_event_id: Option<&[u8]>,
         trigger_context: Option<&serde_json::Value>,
     ) -> Result<Uuid> {
         workflow::create_workflow_run(
             &self.pool,
             community_id,
             workflow_id,
+            definition_event_id,
             trigger_event_id,
             trigger_context,
         )

crates/buzz-db/src/migration.rs

@@ -649,7 +649,7 @@ mod tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 34);
+        assert_eq!(migrations.len(), 35);
         assert_eq!(migrations[0].version, 1);
         assert_eq!(&*migrations[0].description, "initial schema");
         assert!(migrations[0]
@@ -1112,6 +1112,24 @@ mod tests {
         assert!(heartbeat_vacuum.contains("vacuum_truncate = false"));
         assert!(desired_schema.contains("vacuum_truncate = false"));
 
+        // Workflow revision capture is additive: nullable 32-byte event IDs on
+        // both the materialized definition and run, with identical fresh-schema
+        // constraints and no backfill hidden in startup migration state.
+        assert_eq!(migrations[34].version, 35);
+        let workflow_revision_binding = migrations[34].sql.as_str();
+        assert!(workflow_revision_binding.contains("ALTER TABLE workflows"));
+        assert!(workflow_revision_binding.contains("ALTER TABLE workflow_runs"));
+        assert!(workflow_revision_binding.contains("octet_length(definition_event_id) = 32"));
+        assert!(!workflow_revision_binding
+            .to_ascii_lowercase()
+            .contains("update "));
+        assert_eq!(
+            desired_schema
+                .matches("octet_length(definition_event_id) = 32")
+                .count(),
+            2
+        );
+
         // pgschema intentionally reconciles DDL, not seed DML or table storage
         // parameters. Its post-apply reconciliation must restore and verify
         // both parts of the live heartbeat contract for fresh bootstraps.