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/crates/buzz-db/src/workflow.rs b/crates/buzz-db/src/workflow.rs index 7a2396c1fd..8cec761b99 100644 --- a/crates/buzz-db/src/workflow.rs +++ b/crates/buzz-db/src/workflow.rs @@ -26,6 +26,17 @@ pub const LIST_DEFAULT_LIMIT: i64 = 100; /// Hard cap on rows returned by list queries. pub const LIST_MAX_LIMIT: i64 = 1000; +/// 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)) +} + /// 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 +280,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 +294,14 @@ pub async fn create_workflow( definition_hash: &[u8], ) -> Result { let id = Uuid::new_v4(); + // New workflows default to enabled when the definition omits the flag. + let enabled = enabled_from_definition(definition_json)?.unwrap_or(true); 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 +311,7 @@ pub async fn create_workflow( .bind(channel_id) .bind(definition_json) .bind(definition_hash) + .bind(enabled) .execute(pool) .await?; @@ -320,15 +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 = 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 @@ -342,6 +364,7 @@ pub async fn upsert_workflow( .bind(channel_id) .bind(definition_json) .bind(definition_hash) + .bind(enabled) .fetch_optional(pool) .await?; @@ -1449,6 +1472,27 @@ mod tests { assert_eq!(record.status, WorkflowStatus::Active); } + #[test] + fn enabled_from_definition_honors_explicit_flag_and_default() { + 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] + fn enabled_from_definition_rejects_invalid_json() { + assert!(enabled_from_definition("not-json").is_err()); + } + // -- WorkflowRunRecord ---------------------------------------------------- #[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/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..28d1e5375a 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -974,6 +974,15 @@ 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, + // 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()))?; + // Fail fast if all concurrency permits are in use — no queuing. let _permit = engine.run_semaphore.try_acquire().map_err(|_| { ( @@ -1024,6 +1033,14 @@ 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 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()))?; + // 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..86c5c5afbb 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`, 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(()) + } 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. diff --git a/migrations/0028_workflow_enabled_backfill.sql b/migrations/0028_workflow_enabled_backfill.sql new file mode 100644 index 0000000000..ef2eadc809 --- /dev/null +++ b/migrations/0028_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';