Make workflow run history authoritative in Desktop

crates/buzz-db/src/lib.rs

@@ -4051,6 +4051,27 @@ impl Db {
         workflow::list_workflow_runs(&self.pool, community_id, workflow_id, limit).await
     }
 
+    /// List one keyset-paginated page of workflow runs.
+    #[datastore_span(name = "list_workflow_runs_page", system = "postgresql")]
+    pub async fn list_workflow_runs_page(
+        &self,
+        community_id: CommunityId,
+        workflow_id: Uuid,
+        before: Option<chrono::DateTime<chrono::Utc>>,
+        before_id: Option<Uuid>,
+        limit: i64,
+    ) -> Result<Vec<workflow::WorkflowRunRecord>> {
+        workflow::list_workflow_runs_page(
+            &self.pool,
+            community_id,
+            workflow_id,
+            before,
+            before_id,
+            limit,
+        )
+        .await
+    }
+
     /// Update a workflow run's status.
     #[datastore_span(name = "update_workflow_run", system = "postgresql")]
     pub async fn update_workflow_run(
@@ -4060,7 +4081,7 @@ impl Db {
         status: workflow::RunStatus,
         current_step: i32,
         trace: &serde_json::Value,
-        error: Option<&str>,
+        failure: Option<workflow::WorkflowRunFailure<'_>>,
     ) -> Result<()> {
         workflow::update_workflow_run(
             &self.pool,
@@ -4069,7 +4090,7 @@ impl Db {
             status,
             current_step,
             trace,
-            error,
+            failure,
         )
         .await
     }

crates/buzz-db/src/migration.rs

@@ -625,7 +625,7 @@ mod tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 30);
+        assert_eq!(migrations.len(), 31);
         assert_eq!(migrations[0].version, 1);
         assert_eq!(&*migrations[0].description, "initial schema");
         assert!(migrations[0]
@@ -1038,6 +1038,27 @@ mod tests {
         assert!(deletion_recovery.contains("SET LOCAL lock_timeout = '5s'"));
     }
 
+    #[test]
+    fn workflow_run_error_codes_are_additive_and_backfilled_without_parsing_diagnostics() {
+        let mut migrations: Vec<_> = MIGRATOR.iter().collect();
+        migrations.sort_by_key(|migration| migration.version);
+
+        assert_eq!(migrations[30].version, 31);
+        let sql = migrations[30].sql.as_str();
+        assert!(sql.contains("ALTER TABLE workflow_runs ADD COLUMN error_code TEXT"));
+        assert!(sql.contains("SET error_code = 'legacy_unclassified'"));
+        assert!(sql.contains("status IN ('failed', 'cancelled')"));
+        assert!(!sql.contains("error_message LIKE"));
+        assert!(!MIGRATOR
+            .iter()
+            .find(|migration| migration.version == 1)
+            .expect("initial migration")
+            .sql
+            .as_str()
+            .contains("error_code"));
+        assert!(include_str!("../../../schema/schema.sql").contains("error_code          TEXT"));
+    }
+
     #[test]
     fn migration_lint_detects_tables_missing_community_id_by_default() {
         let sql = r#"

crates/buzz-db/src/workflow.rs

@@ -216,8 +216,11 @@ pub struct WorkflowRunRecord {
     pub started_at: Option<DateTime<Utc>>,
     /// When execution finished (success or failure).
     pub completed_at: Option<DateTime<Utc>>,
-    /// Error message if the run failed.
+    /// Redacted human-readable diagnostic for failed or cancelled runs.
     pub error_message: Option<String>,
+    /// Stable machine-readable failure or cancellation classification.
+    /// Kept separate from `error_message` so callers never parse diagnostics.
+    pub error_code: Option<String>,
     /// When the run record was created.
     pub created_at: DateTime<Utc>,
 }
@@ -831,7 +834,7 @@ pub async fn get_workflow_run(
     let row = sqlx::query(
         r#"
         SELECT community_id, id, workflow_id, status::text AS status, trigger_event_id, current_step,
-               execution_trace, trigger_context, started_at, completed_at, error_message, created_at
+               execution_trace, trigger_context, started_at, completed_at, error_message, error_code, created_at
         FROM workflow_runs
         WHERE community_id = $1 AND id = $2
         "#,
@@ -845,34 +848,67 @@ pub async fn get_workflow_run(
     row_to_run_record(row)
 }
 
-/// List runs for a workflow, newest first, up to `limit` rows.
-pub async fn list_workflow_runs(
+/// List runs for a workflow using a stable newest-first keyset.
+///
+/// Rows are ordered by `(created_at DESC, id DESC)`. A cursor is valid only
+/// when both `before` and `before_id` are supplied; callers should pass the
+/// final row from the previous page. `limit` is clamped to the shared list
+/// bounds.
+pub async fn list_workflow_runs_page(
     pool: &PgPool,
     community_id: CommunityId,
     workflow_id: Uuid,
+    before: Option<DateTime<Utc>>,
+    before_id: Option<Uuid>,
     limit: i64,
 ) -> Result<Vec<WorkflowRunRecord>> {
-    let limit = limit.min(1000);
+    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,
-               execution_trace, trigger_context, started_at, completed_at, error_message, created_at
+               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
-        ORDER BY created_at DESC
-        LIMIT $3
+          AND (
+              $3::timestamptz IS NULL
+              OR $4::uuid IS NULL
+              OR (created_at, id) < ($3, $4)
+          )
+        ORDER BY created_at DESC, id DESC
+        LIMIT $5
         "#,
     )
     .bind(community_id.as_uuid())
     .bind(workflow_id)
+    .bind(before)
+    .bind(before_id)
     .bind(limit)
     .fetch_all(pool)
     .await?;
 
     rows.into_iter().map(row_to_run_record).collect()
 }
 
-/// Update run status, current step, execution trace, and optional error message.
+/// List runs for a workflow, newest first, up to `limit` rows.
+pub async fn list_workflow_runs(
+    pool: &PgPool,
+    community_id: CommunityId,
+    workflow_id: Uuid,
+    limit: i64,
+) -> Result<Vec<WorkflowRunRecord>> {
+    list_workflow_runs_page(pool, community_id, workflow_id, None, None, limit).await
+}
+
+/// Structured failure persisted for a workflow run.
+#[derive(Debug, Clone, Copy)]
+pub struct WorkflowRunFailure<'a> {
+    /// Stable machine-readable failure code.
+    pub code: &'a str,
+    /// Human-readable failure detail.
+    pub message: &'a str,
+}
+
+/// Update run status, current step, execution trace, and optional failure.
 ///
 /// Fix C3: `started_at` is set when the NEW status is 'running' and `started_at`
 /// has not yet been stamped (IS NULL). The original code read `status` from the
@@ -885,26 +921,31 @@ pub async fn update_workflow_run(
     status: RunStatus,
     current_step: i32,
     trace: &serde_json::Value,
-    error: Option<&str>,
+    failure: Option<WorkflowRunFailure<'_>>,
 ) -> Result<()> {
     let status_str = status.to_string();
+    let (error_code, error) = failure
+        .map(|failure| (Some(failure.code), Some(failure.message)))
+        .unwrap_or((None, None));
     let affected = sqlx::query(
         r#"
         UPDATE workflow_runs
         SET status        = $1::run_status,
             current_step  = $2,
             execution_trace = $3,
-            error_message = $4,
-            started_at    = CASE WHEN $5 = 'running' AND started_at IS NULL
+            error_code    = $4,
+            error_message = $5,
+            started_at    = CASE WHEN $6 = 'running' AND started_at IS NULL
                                  THEN NOW() ELSE started_at END,
-            completed_at  = CASE WHEN $6 IN ('completed','failed','cancelled')
+            completed_at  = CASE WHEN $7 IN ('completed','failed','cancelled')
                                  THEN NOW() ELSE completed_at END
-        WHERE community_id = $7 AND id = $8
+        WHERE community_id = $8 AND id = $9
         "#,
     )
     .bind(&status_str)
     .bind(current_step)
     .bind(trace)
+    .bind(error_code)
     .bind(error)
     .bind(&status_str) // for started_at CASE
     .bind(&status_str) // for completed_at CASE
@@ -1169,6 +1210,7 @@ fn row_to_run_record(row: sqlx::postgres::PgRow) -> Result<WorkflowRunRecord> {
         started_at: row.try_get("started_at")?,
         completed_at: row.try_get("completed_at")?,
         error_message: row.try_get("error_message")?,
+        error_code: row.try_get("error_code")?,
         created_at: row.try_get("created_at")?,
     })
 }
@@ -1473,6 +1515,7 @@ mod tests {
             started_at: Some(now),
             completed_at: None,
             error_message: None,
+            error_code: None,
             created_at: now,
         };
 
@@ -1501,6 +1544,7 @@ mod tests {
             started_at: None,
             completed_at: None,
             error_message: None,
+            error_code: None,
             created_at: now,
         };
 
@@ -1524,6 +1568,7 @@ mod tests {
             started_at: Some(now),
             completed_at: Some(now),
             error_message: Some("step timeout exceeded".to_owned()),
+            error_code: Some("step_timeout".to_owned()),
             created_at: now,
         };
 
@@ -1555,6 +1600,7 @@ mod tests {
             started_at: Some(now),
             completed_at: Some(now),
             error_message: None,
+            error_code: None,
             created_at: now,
         };
 
@@ -1577,6 +1623,7 @@ mod tests {
             started_at: None,
             completed_at: None,
             error_message: None,
+            error_code: None,
             created_at: now,
         };
 

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

@@ -21,7 +21,7 @@ use crate::state::AppState;
 
 use super::{api_error, internal_error, not_found};
 
-async fn enforce_http_admission(
+pub(crate) async fn enforce_http_admission(
     state: &AppState,
     tenant: &TenantContext,
     pubkey: &nostr::PublicKey,
@@ -1938,7 +1938,10 @@ pub async fn workflow_webhook(
                         buzz_db::workflow::RunStatus::Failed,
                         0,
                         &serde_json::json!([]),
-                        Some(&format!("definition parse error: {e}")),
+                        Some(buzz_db::workflow::WorkflowRunFailure {
+                            code: "invalid_definition",
+                            message: &format!("definition parse error: {e}"),
+                        }),
                     )
                     .await
                 {

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

@@ -9,6 +9,7 @@ pub mod media;
 pub mod mesh_demo;
 pub mod nip05;
 pub mod operator;
+pub mod workflows;
 
 // Re-export imeta helpers used by ingest pipeline.
 pub use crate::handlers::imeta::{validate_imeta_tags, verify_imeta_blobs};