fix(workflow): honor enabled: false flag at execution time and persist it

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]

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<Option<bool>> {
+    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<Uuid> {
     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]

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,

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),

crates/buzz-workflow/src/executor.rs

@@ -974,6 +974,15 @@ pub async fn execute_run(
     def: &WorkflowDef,
     trigger_ctx: &TriggerContext,
 ) -> Result<ExecutionResult, (WorkflowError, crate::error::PartialProgress)> {
+    // 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<HashMap<String, JsonValue>>,
 ) -> Result<ExecutionResult, (WorkflowError, crate::error::PartialProgress)> {
+    // 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(|_| {
         (

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.

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';