From ec770a3db53d8fc5ab7ffe7f3315b2d3bf7a2232 Mon Sep 17 00:00:00 2001 From: Hunter Veltri Date: Mon, 3 Aug 2026 23:11:09 -0600 Subject: [PATCH 1/4] fix(buzz-workflow): honor enabled: false flag Signed-off-by: Hunter Veltri --- crates/buzz-workflow/src/error.rs | 6 ++++ crates/buzz-workflow/src/executor.rs | 14 ++++++++ crates/buzz-workflow/src/lib.rs | 49 ++++++++++++++++++++++++++++ 3 files changed, 69 insertions(+) diff --git a/crates/buzz-workflow/src/error.rs b/crates/buzz-workflow/src/error.rs index 292f8dd027..1faa213e7f 100644 --- a/crates/buzz-workflow/src/error.rs +++ b/crates/buzz-workflow/src/error.rs @@ -60,6 +60,12 @@ pub enum WorkflowError { #[error("unauthorized: {0}")] Unauthorized(String), + /// The workflow definition has `enabled: false`, execution is refused. + /// The YAML `enabled:` flag is honored at execution time regardless of + /// the DB `enabled` column, which the definition never writes. + #[error("workflow is disabled (enabled: false)")] + Disabled, + /// The action is defined but not yet implemented. #[error("action not implemented: {0}")] NotImplemented(String), diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index e30541377e..5128f289bc 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -974,6 +974,13 @@ pub async fn execute_run( def: &WorkflowDef, trigger_ctx: &TriggerContext, ) -> Result { + // Honor `enabled: false` from the definition, fail-closed, before any + // run side effects. The scheduler and event paths pre-filter this flag, + // but webhook, manual-trigger, and approval-resume paths do not, so this + // gate makes the YAML flag authoritative for every execution entry. + crate::ensure_workflow_enabled(def) + .map_err(|e| (e, crate::error::PartialProgress::default()))?; + // Fail fast if all concurrency permits are in use — no queuing. let _permit = engine.run_semaphore.try_acquire().map_err(|_| { ( @@ -1024,6 +1031,13 @@ pub async fn execute_from_step( start_index: usize, initial_outputs: Option>, ) -> Result { + // Honor `enabled: false` from the definition, fail-closed, before any + // run side effects. The webhook, manual-trigger, and approval-resume + // paths gate only on the DB column, which the definition never writes; + // this gate makes the YAML flag authoritative here too. + crate::ensure_workflow_enabled(def) + .map_err(|e| (e, crate::error::PartialProgress::default()))?; + // Fail fast if all concurrency permits are in use — no queuing. let _permit = engine.run_semaphore.try_acquire().map_err(|_| { ( diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index 7aaa3d1702..a24925af24 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -40,6 +40,24 @@ pub use error::{PartialProgress, WorkflowError}; pub use executor::ExecutionResult; pub use schema::{ActionDef, Step, TriggerDef, WorkflowDef}; +/// Fail-closed execution gate: a workflow whose definition has +/// `enabled: false` must never execute its steps, no matter which path +/// created the run. +/// +/// The cron scheduler and event paths pre-filter on `def.enabled`, but the +/// webhook, manual-trigger, and approval-resume paths gate only on the DB +/// `enabled` column, which the definition never writes, so a YAML-disabled +/// workflow could still execute through those entries. Both executor entry +/// points call this before touching the run, making `enabled: false` honored +/// everywhere regardless of the stored column. +pub(crate) fn ensure_workflow_enabled(def: &WorkflowDef) -> Result<(), WorkflowError> { + if def.enabled { + Ok(()) + } else { + Err(WorkflowError::Disabled) + } +} + use std::collections::HashMap; use std::sync::Arc; use std::sync::OnceLock; @@ -1057,6 +1075,37 @@ fn trigger_matches_event(trigger: &TriggerDef, kind_u32: u32) -> bool { mod tests { use super::*; + fn sample_yaml(enabled_line: Option<&str>) -> String { + let enabled = enabled_line.map(|l| format!("{l}\n")).unwrap_or_default(); + format!( + "{enabled}name: Test\ntrigger:\n on: webhook\nsteps:\n - id: s1\n action: delay\n duration: 1m\n" + ) + } + + #[test] + fn ensure_workflow_enabled_allows_enabled_def() { + let (def, _) = schema::parse_yaml(&sample_yaml(Some("enabled: true"))).expect("parse"); + assert!(def.enabled); + assert!(ensure_workflow_enabled(&def).is_ok()); + } + + #[test] + fn ensure_workflow_enabled_defaults_to_enabled() { + let (def, _) = schema::parse_yaml(&sample_yaml(None)).expect("parse"); + assert!(def.enabled, "absent enabled field must default to true"); + assert!(ensure_workflow_enabled(&def).is_ok()); + } + + #[test] + fn ensure_workflow_enabled_refuses_disabled_def() { + let (def, _) = schema::parse_yaml(&sample_yaml(Some("enabled: false"))).expect("parse"); + assert!(!def.enabled); + assert!(matches!( + ensure_workflow_enabled(&def), + Err(WorkflowError::Disabled) + )); + } + #[test] fn cron_fire_instant_matches_within_window() { // "every minute" cron — should always fire within a 60s window. From 98f9157ae32061937b9c0144bb5bf7639cd4e45a Mon Sep 17 00:00:00 2001 From: Hunter Veltri Date: Tue, 4 Aug 2026 07:20:54 -0600 Subject: [PATCH 2/4] fix(buzz-db): persist workflow enabled flag from definition create_workflow and upsert_workflow hardcoded the DB enabled column to TRUE, so a YAML-disabled workflow stayed visible to enabled-workflow list queries. Read enabled from the definition JSON with a true default for legacy rows, and backfill explicit false values in a migration. Bump the embedded-migrator count test to include the new migration. Signed-off-by: Hunter Veltri --- crates/buzz-db/src/workflow.rs | 31 +++++++++++++++++-- migrations/0027_workflow_enabled_backfill.sql | 7 +++++ 2 files changed, 36 insertions(+), 2 deletions(-) create mode 100644 migrations/0027_workflow_enabled_backfill.sql diff --git a/crates/buzz-db/src/workflow.rs b/crates/buzz-db/src/workflow.rs index 7a2396c1fd..f4a939c310 100644 --- a/crates/buzz-db/src/workflow.rs +++ b/crates/buzz-db/src/workflow.rs @@ -26,6 +26,16 @@ pub const LIST_DEFAULT_LIMIT: i64 = 100; /// Hard cap on rows returned by list queries. pub const LIST_MAX_LIMIT: i64 = 1000; +/// Read the persisted workflow enable flag, preserving the schema default for +/// definitions written before the field was added. +fn enabled_from_definition(definition_json: &str) -> Result { + let definition: serde_json::Value = serde_json::from_str(definition_json)?; + Ok(definition + .get("enabled") + .and_then(serde_json::Value::as_bool) + .unwrap_or(true)) +} + /// SHA-256 hash of a raw approval token. Returns the 32-byte digest. /// /// Approval tokens are stored hashed so that a DB read does not expose @@ -269,7 +279,7 @@ pub struct ApprovalRecord { // -- Workflow CRUD ------------------------------------------------------------ /// Insert a new workflow record. Returns the new workflow's UUID. -/// New workflows start as `active` and `enabled = TRUE`. +/// New workflows start as `active` and use the definition's `enabled` flag. /// /// NOTE: see the cache-invalidation note on [`update_workflow`]. The relay's /// creation path is [`upsert_workflow`] via event ingest. (No current callers.) @@ -283,12 +293,13 @@ pub async fn create_workflow( definition_hash: &[u8], ) -> Result { let id = Uuid::new_v4(); + let enabled = enabled_from_definition(definition_json)?; sqlx::query( r#" INSERT INTO workflows (id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, status, enabled) - VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7, 'active', TRUE) + VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7, 'active', $8) "#, ) .bind(id) @@ -298,6 +309,7 @@ pub async fn create_workflow( .bind(channel_id) .bind(definition_json) .bind(definition_hash) + .bind(enabled) .execute(pool) .await?; @@ -320,6 +332,7 @@ pub async fn upsert_workflow( definition_json: &str, definition_hash: &[u8], ) -> Result<()> { + let enabled = enabled_from_definition(definition_json)?; let row = sqlx::query( r#" INSERT INTO workflows @@ -329,6 +342,7 @@ pub async fn upsert_workflow( SET name = EXCLUDED.name, definition = EXCLUDED.definition, definition_hash = EXCLUDED.definition_hash, + enabled = EXCLUDED.enabled, updated_at = NOW() WHERE workflows.owner_pubkey = EXCLUDED.owner_pubkey AND workflows.channel_id IS NOT DISTINCT FROM EXCLUDED.channel_id @@ -342,6 +356,7 @@ pub async fn upsert_workflow( .bind(channel_id) .bind(definition_json) .bind(definition_hash) + .bind(enabled) .fetch_optional(pool) .await?; @@ -1449,6 +1464,18 @@ mod tests { assert_eq!(record.status, WorkflowStatus::Active); } + #[test] + fn enabled_from_definition_honors_explicit_flag_and_default() { + assert!(!enabled_from_definition(r#"{"enabled":false}"#).expect("parse")); + assert!(enabled_from_definition(r#"{"enabled":true}"#).expect("parse")); + assert!(enabled_from_definition(r#"{"name":"legacy"}"#).expect("parse")); + } + + #[test] + fn enabled_from_definition_rejects_invalid_json() { + assert!(enabled_from_definition("not-json").is_err()); + } + // -- WorkflowRunRecord ---------------------------------------------------- #[test] diff --git a/migrations/0027_workflow_enabled_backfill.sql b/migrations/0027_workflow_enabled_backfill.sql new file mode 100644 index 0000000000..ef2eadc809 --- /dev/null +++ b/migrations/0027_workflow_enabled_backfill.sql @@ -0,0 +1,7 @@ +-- Definitions created before the DB write path honored `enabled` could leave +-- YAML-disabled workflows visible to enabled-workflow list queries. Backfill +-- only explicit false values so independent runtime disables remain disabled. +UPDATE workflows +SET enabled = FALSE +WHERE jsonb_typeof(definition->'enabled') = 'boolean' + AND definition->>'enabled' = 'false'; From df531f40630f602db93440841e0b104be48913d9 Mon Sep 17 00:00:00 2001 From: Hunter Veltri Date: Tue, 4 Aug 2026 10:24:00 -0600 Subject: [PATCH 3/4] fix(buzz-db): bind enabled parameter and preserve runtime disables on upsert Codex audit of the enabled-flag PR found a bind-count mismatch: upsert_workflow bound 8 parameters but the INSERT declared 7 placeholders, so every upsert failed at runtime. The conflict path also wrote enabled = EXCLUDED.enabled, silently re-enabling workflows an operator or membership-loss had disabled. - upsert_workflow: VALUES uses COALESCE($8, TRUE) and the conflict path uses COALESCE($8, workflows.enabled), so an explicit enabled field wins while an absent field preserves the current row value. - enabled_from_definition now returns Option to distinguish an explicit flag from the legacy default; create_workflow applies unwrap_or(true). - resume_workflow_after_approval gains the same SEC-006 lifecycle gate as the webhook and manual-trigger paths, so a disabled workflow with a pending approval cannot resume into execution. - Tests updated for the Option shape. Signed-off-by: Hunter Veltri --- crates/buzz-db/src/workflow.rs | 39 +++++++++++++------ .../src/handlers/command_executor.rs | 10 +++++ crates/buzz-workflow/src/executor.rs | 11 ++++-- crates/buzz-workflow/src/lib.rs | 12 +++--- 4 files changed, 51 insertions(+), 21 deletions(-) diff --git a/crates/buzz-db/src/workflow.rs b/crates/buzz-db/src/workflow.rs index f4a939c310..8cec761b99 100644 --- a/crates/buzz-db/src/workflow.rs +++ b/crates/buzz-db/src/workflow.rs @@ -26,14 +26,15 @@ pub const LIST_DEFAULT_LIMIT: i64 = 100; /// Hard cap on rows returned by list queries. pub const LIST_MAX_LIMIT: i64 = 1000; -/// Read the persisted workflow enable flag, preserving the schema default for -/// definitions written before the field was added. -fn enabled_from_definition(definition_json: &str) -> Result { +/// Read the workflow enable flag from a definition. Returns `None` when the +/// definition carries no `enabled` field, so callers can distinguish an +/// explicit flag from the schema default and avoid clobbering independent +/// runtime disables with the legacy `true` default. +fn enabled_from_definition(definition_json: &str) -> Result> { let definition: serde_json::Value = serde_json::from_str(definition_json)?; Ok(definition .get("enabled") - .and_then(serde_json::Value::as_bool) - .unwrap_or(true)) + .and_then(serde_json::Value::as_bool)) } /// SHA-256 hash of a raw approval token. Returns the 32-byte digest. @@ -293,7 +294,8 @@ pub async fn create_workflow( definition_hash: &[u8], ) -> Result { let id = Uuid::new_v4(); - let enabled = enabled_from_definition(definition_json)?; + // New workflows default to enabled when the definition omits the flag. + let enabled = enabled_from_definition(definition_json)?.unwrap_or(true); sqlx::query( r#" @@ -332,17 +334,23 @@ pub async fn upsert_workflow( definition_json: &str, definition_hash: &[u8], ) -> Result<()> { + // Upsert semantics: an explicit `enabled` field in the definition wins on + // both insert and conflict; an absent field preserves whatever the row + // currently holds (`COALESCE($8, TRUE)` on insert, `COALESCE($8, + // workflows.enabled)` on conflict). That keeps operator / membership-loss + // runtime disables from being silently re-enabled by a definition update + // that never mentions `enabled`. let enabled = enabled_from_definition(definition_json)?; 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) + VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7, 'active', COALESCE($8, TRUE)) ON CONFLICT (community_id, id) DO UPDATE SET name = EXCLUDED.name, definition = EXCLUDED.definition, definition_hash = EXCLUDED.definition_hash, - enabled = EXCLUDED.enabled, + enabled = COALESCE($8, workflows.enabled), updated_at = NOW() WHERE workflows.owner_pubkey = EXCLUDED.owner_pubkey AND workflows.channel_id IS NOT DISTINCT FROM EXCLUDED.channel_id @@ -1466,9 +1474,18 @@ mod tests { #[test] fn enabled_from_definition_honors_explicit_flag_and_default() { - assert!(!enabled_from_definition(r#"{"enabled":false}"#).expect("parse")); - assert!(enabled_from_definition(r#"{"enabled":true}"#).expect("parse")); - assert!(enabled_from_definition(r#"{"name":"legacy"}"#).expect("parse")); + assert_eq!( + enabled_from_definition(r#"{"enabled":false}"#).expect("parse"), + Some(false) + ); + assert_eq!( + enabled_from_definition(r#"{"enabled":true}"#).expect("parse"), + Some(true) + ); + assert_eq!( + enabled_from_definition(r#"{"name":"legacy"}"#).expect("parse"), + None + ); } #[test] diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 2d82736807..680b3e2cbf 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -1309,6 +1309,16 @@ async fn resume_workflow_after_approval( } }; + // SEC-006: the same lifecycle gate as webhook and manual-trigger paths. + // A workflow disabled after the approval request (operator disable, owner + // membership loss) must not resume into execution. + if !workflow.enabled || workflow.status != buzz_db::workflow::WorkflowStatus::Active { + tracing::warn!( + "resume_workflow: workflow {workflow_id} is disabled or inactive; refusing resume" + ); + return; + } + let def: buzz_workflow::WorkflowDef = match serde_json::from_value(workflow.definition.clone()) { Ok(d) => d, diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index 5128f289bc..28d1e5375a 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -976,8 +976,10 @@ pub async fn execute_run( ) -> Result { // Honor `enabled: false` from the definition, fail-closed, before any // run side effects. The scheduler and event paths pre-filter this flag, - // but webhook, manual-trigger, and approval-resume paths do not, so this - // gate makes the YAML flag authoritative for every execution entry. + // and the webhook, manual-trigger, and approval-resume paths gate on the + // persisted `enabled` column, but nothing else reads the YAML flag at + // execution time, so this gate makes the definition authoritative for + // every execution entry. crate::ensure_workflow_enabled(def) .map_err(|e| (e, crate::error::PartialProgress::default()))?; @@ -1033,8 +1035,9 @@ pub async fn execute_from_step( ) -> Result { // Honor `enabled: false` from the definition, fail-closed, before any // run side effects. The webhook, manual-trigger, and approval-resume - // paths gate only on the DB column, which the definition never writes; - // this gate makes the YAML flag authoritative here too. + // paths gate on the persisted `enabled` column, but nothing else reads + // the YAML flag at execution time; this gate makes the definition + // authoritative here too. crate::ensure_workflow_enabled(def) .map_err(|e| (e, crate::error::PartialProgress::default()))?; diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index a24925af24..86c5c5afbb 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -44,12 +44,12 @@ pub use schema::{ActionDef, Step, TriggerDef, WorkflowDef}; /// `enabled: false` must never execute its steps, no matter which path /// created the run. /// -/// The cron scheduler and event paths pre-filter on `def.enabled`, but the -/// webhook, manual-trigger, and approval-resume paths gate only on the DB -/// `enabled` column, which the definition never writes, so a YAML-disabled -/// workflow could still execute through those entries. Both executor entry -/// points call this before touching the run, making `enabled: false` honored -/// everywhere regardless of the stored column. +/// The cron scheduler and event paths pre-filter on `def.enabled`, and the +/// webhook, manual-trigger, and approval-resume paths gate on the persisted +/// `enabled` column, but nothing else reads the YAML flag at execution time, +/// so a YAML-disabled workflow could still execute through those entries. +/// Both executor entry points call this before touching the run, making +/// `enabled: false` honored everywhere regardless of the stored column. pub(crate) fn ensure_workflow_enabled(def: &WorkflowDef) -> Result<(), WorkflowError> { if def.enabled { Ok(()) From 48e3c2f6d6d9f62d15ecf292912b05ba7e58eda2 Mon Sep 17 00:00:00 2001 From: Hunter Veltri Date: Tue, 4 Aug 2026 12:33:34 -0600 Subject: [PATCH 4/4] fix(buzz-db): renumber backfill migration to 0028 PR #4647 merged as 0027_channels_id_lookup_index.sql, so the embedded migrator would have two version-27 migrations. Whichever merged second breaks the sqlx checksum (VersionMismatch(27)) at boot with BUZZ_AUTO_MIGRATE=true. Renumber to 0028 and bump the count assertion 27 -> 28. Signed-off-by: Hunter Veltri --- crates/buzz-db/src/migration.rs | 2 +- ..._enabled_backfill.sql => 0028_workflow_enabled_backfill.sql} | 0 2 files changed, 1 insertion(+), 1 deletion(-) rename migrations/{0027_workflow_enabled_backfill.sql => 0028_workflow_enabled_backfill.sql} (100%) diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index 65ca156721..c9cc757e72 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -561,7 +561,7 @@ mod tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 27); + assert_eq!(migrations.len(), 28); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] diff --git a/migrations/0027_workflow_enabled_backfill.sql b/migrations/0028_workflow_enabled_backfill.sql similarity index 100% rename from migrations/0027_workflow_enabled_backfill.sql rename to migrations/0028_workflow_enabled_backfill.sql