buzz/crates/buzz-workflow/src/lib.rs

#![deny(unsafe_code)]
#![warn(missing_docs)]
//! `buzz-workflow` — Workflow engine for Buzz.
//!
//! Channel-scoped automations with sequential execution, variable substitution,
//! conditional logic, and execution traces.
//!
//! ## Architecture
//!
//! - [`WorkflowEngine`] — top-level handle; lives in `AppState`
//! - [`schema`] — YAML/JSON definition types (`WorkflowDef`, `TriggerDef`, `ActionDef`, `Step`)
//! - [`executor`] — sequential execution, template resolution, condition evaluation
//! - [`error`] — [`WorkflowError`] enum
//!
//! ## Usage
//!
//! ```rust,ignore
//! let engine = Arc::new(WorkflowEngine::new(db, WorkflowConfig::default()));
//!
//! // Parse and validate a YAML definition.
//! let (def, json) = WorkflowEngine::parse_yaml(yaml_str)?;
//!
//! // React to an incoming event (called from event handler post-store hook).
//! // The community is the event's server-resolved tenant, threaded from the
//! // relay's bound `TenantContext` — the same workflow UUID can exist in two
//! // communities, so execution is always scoped to its owner.
//! engine.on_event(community_id, &stored_event).await?;
//!
//! // Run the background scheduler (cron triggers).
//! tokio::spawn(async move { engine.run().await });
//! ```

pub mod action_sink;
pub mod error;
pub mod executor;
pub mod schema;

pub use action_sink::{ActionSink, ActionSinkError};
pub use error::{PartialProgress, WorkflowError};
pub use executor::ExecutionResult;
pub use schema::{ActionDef, Step, TriggerDef, WorkflowDef};

use std::collections::HashMap;
use std::sync::Arc;
use std::sync::OnceLock;

use buzz_core::kind::{event_kind_u32, is_workflow_execution_kind, KIND_REACTION};
use buzz_core::tenant::CommunityId;
use buzz_db::workflow::RunStatus;
use buzz_db::Db;
use chrono::{DateTime, Utc};
use dashmap::DashMap;
use tokio::sync::Semaphore;
use uuid::Uuid;

/// Runtime configuration for the workflow engine.
#[derive(Clone, Debug)]
pub struct WorkflowConfig {
    /// Maximum number of concurrently executing workflow runs. Default: 100.
    pub max_concurrent: usize,
    /// Default per-step timeout in seconds. Default: 300 (5 minutes).
    pub default_timeout_secs: u64,
}

impl Default for WorkflowConfig {
    fn default() -> Self {
        Self {
            max_concurrent: 100,
            default_timeout_secs: 300,
        }
    }
}

/// The workflow engine. Clone is cheap (Arc-backed DB pool + semaphore).
pub struct WorkflowEngine {
    pub(crate) db: Db,
    pub(crate) config: WorkflowConfig,
    /// Semaphore enforcing `config.max_concurrent` simultaneous workflow runs.
    pub(crate) run_semaphore: Arc<Semaphore>,
    /// Last-fired timestamps for interval-triggered workflows, keyed by
    /// `(community_id, workflow_id)`. The same workflow UUID can exist in two
    /// communities (the PK is `(community_id, id)`); keying by bare id would let
    /// one community's interval fire suppress the other's for the interval.
    /// In-memory only — lost on restart. Missed fires during downtime are
    /// not replayed (acceptable for MVP).
    pub(crate) last_fired: DashMap<(CommunityId, Uuid), DateTime<Utc>>,
    /// Action sink for executing side-effects (SendMessage, etc.).
    /// Late-initialized via [`set_action_sink`] after `AppState` construction.
    pub(crate) action_sink: OnceLock<Arc<dyn ActionSink>>,
    /// Short-TTL cache for the per-event enabled-workflow lookup, keyed
    /// `(community_id, channel_id)`. Most channels have no workflows, so this
    /// removes one SELECT from nearly every ingested event.
    ///
    /// Consistency: the relay invalidates this cache on its own pod at the two
    /// workflow mutation sites (command upsert, NIP-09 deletion). There is
    /// deliberately no cross-pod invalidation — workflow triggering is not an
    /// access-control fence, so the worst case on another pod is a just-deleted
    /// workflow firing (or a just-created one missing events) for up to the TTL.
    /// The same TTL also bounds the same-pod look-aside race (a stale fill
    /// landing just after an invalidation). Workflow mutations are rare; the
    /// 10s window matches the relay's other moka caches (see `AppState` in
    /// `buzz-relay`).
    pub(crate) workflow_cache:
        moka::sync::Cache<(CommunityId, Uuid), Arc<Vec<buzz_db::workflow::WorkflowRecord>>>,
    /// Workflows whose ban denial was logged at WARN within the last
    /// [`BAN_DENIAL_WARN_INTERVAL`], keyed `(community_id, workflow_id)`. A
    /// banned owner's workflow is denied on every matching event and every
    /// cron tick; this keeps that expected denial from flooding the log.
    pub(crate) ban_denial_warned: moka::sync::Cache<(CommunityId, Uuid), ()>,
}

/// How often a banned owner's workflow denial is logged at WARN. Denials in
/// between log at DEBUG. Other authority failures always log at WARN.
const BAN_DENIAL_WARN_INTERVAL: std::time::Duration = std::time::Duration::from_secs(600);

impl WorkflowEngine {
    /// Create a new `WorkflowEngine`.
    pub fn new(db: Db, config: WorkflowConfig) -> Self {
        let permits = config.max_concurrent.max(1);
        let run_semaphore = Arc::new(Semaphore::new(permits));
        Self {
            db,
            config,
            run_semaphore,
            last_fired: DashMap::new(),
            action_sink: OnceLock::new(),
            workflow_cache: moka::sync::Cache::builder()
                .max_capacity(10_000)
                .time_to_live(std::time::Duration::from_secs(10))
                .build(),
            ban_denial_warned: ban_denial_warn_cache(BAN_DENIAL_WARN_INTERVAL),
        }
    }

    /// Drop the cached enabled-workflow list for a channel.
    ///
    /// Must be called after any write to a workflow's trigger eligibility or
    /// channel binding (currently the relay's command upsert and NIP-09
    /// deletion paths) so same-pod trigger matching sees the change
    /// immediately instead of after the cache TTL.
    pub fn invalidate_channel_workflows(&self, community_id: CommunityId, channel_id: Uuid) {
        self.workflow_cache.invalidate(&(community_id, channel_id));
    }

    /// Fail-closed pre-run authority gate (SEC-006).
    ///
    /// A workflow executes with its **owner's** standing authority long after
    /// the definition was saved, so every run-creation door must recheck the
    /// owner's *current* channel authority immediately before creating a run:
    ///
    /// - the owner must still be an active member of the workflow's channel,
    ///   in a role that can write (a guest cannot);
    /// - if the definition contains an exfiltration-capable action
    ///   (`call_webhook`), the owner must currently hold the `owner` or
    ///   `admin` role;
    /// - the owner, or the human owner of an agent owner, must not be banned
    ///   from the community. Bans leave channel membership in place, so the
    ///   role check alone would keep a banned owner's workflows running.
    ///
    /// Any lookup error denies (fail-closed): a removed owner must never keep
    /// exfiltration authority because a membership read happened to fail.
    /// This is a point-in-time check: it does not cancel runs that already
    /// exist, and a ban that commits after it can race with run creation.
    pub async fn check_owner_authority(
        &self,
        community_id: CommunityId,
        channel_id: Uuid,
        owner_pubkey: &[u8],
        def: &WorkflowDef,
    ) -> Result<(), WorkflowError> {
        let role = self
            .db
            .get_member_role(community_id, channel_id, owner_pubkey)
            .await
            .map_err(|e| {
                WorkflowError::Unauthorized(format!(
                    "owner authority lookup failed (fail-closed): {e}"
                ))
            })?;
        if !owner_authority_allows(role.as_deref(), def.requires_elevated_authority()) {
            return Err(WorkflowError::Unauthorized(
                "workflow owner lacks current channel authority".into(),
            ));
        }
        let restrictions = self
            .db
            .moderation_restriction_state(community_id, owner_pubkey)
            .await
            .map_err(|e| {
                WorkflowError::Unauthorized(format!(
                    "owner restriction lookup failed (fail-closed): {e}"
                ))
            })?;
        if restrictions.banned {
            return Err(WorkflowError::OwnerBanned);
        }
        Ok(())
    }

    /// Log a [`check_owner_authority`](Self::check_owner_authority) denial
    /// that skipped a run. A ban denial logs at WARN at most once per
    /// workflow per [`BAN_DENIAL_WARN_INTERVAL`], and at DEBUG otherwise;
    /// every other denial, including a failed lookup, logs at WARN.
    fn log_owner_denial(
        &self,
        community_id: CommunityId,
        workflow_id: Uuid,
        trigger: &'static str,
        err: &WorkflowError,
    ) {
        if should_warn_owner_denial(&self.ban_denial_warned, community_id, workflow_id, err) {
            tracing::warn!(
                workflow_id = %workflow_id,
                trigger,
                "Skipping workflow — owner authority check failed: {err}"
            );
        } else {
            tracing::debug!(
                workflow_id = %workflow_id,
                trigger,
                "Skipping workflow — owner authority check failed: {err}"
            );
        }
    }

    /// Set the action sink. Called once after `AppState` construction.
    ///
    /// # Panics
    /// Panics if called more than once.
    pub fn set_action_sink(&self, sink: Arc<dyn ActionSink>) {
        if self.action_sink.set(sink).is_err() {
            panic!("action_sink already initialized");
        }
    }

    /// Get the action sink reference.
    ///
    /// Returns `Err(WorkflowError)` if the sink has not been initialized via
    /// [`set_action_sink`]. This avoids a panic if the engine is used before
    /// wiring is complete.
    pub(crate) fn action_sink(&self) -> Result<&dyn ActionSink, WorkflowError> {
        self.action_sink.get().map(|s| s.as_ref()).ok_or_else(|| {
            WorkflowError::InvalidDefinition(
                "action_sink not initialized — call set_action_sink() before executing workflows"
                    .into(),
            )
        })
    }

    /// Parse and validate a YAML workflow definition.
    ///
    /// Returns `(WorkflowDef, canonical_json)` on success. The canonical JSON
    /// is suitable for storage in the `definition` column.
    pub fn parse_yaml(yaml: &str) -> Result<(WorkflowDef, String), WorkflowError> {
        schema::parse_yaml(yaml)
    }

    /// Finalize a workflow run after execution completes or fails.
    ///
    /// This is the **single** place that maps an executor result to a DB status
    /// update. All execution paths (event-triggered, manual trigger/webhook,
    /// approval resume) call this instead of duplicating the 3-way match.
    ///
    /// `existing_trace` is prepended to the executor's trace — used by the
    /// approval-resume path where pre-approval steps already have trace entries.
    pub async fn finalize_run(
        &self,
        community_id: CommunityId,
        run_id: uuid::Uuid,
        result: Result<ExecutionResult, (WorkflowError, PartialProgress)>,
        existing_trace: Option<Vec<serde_json::Value>>,
    ) {
        let prefix = existing_trace.unwrap_or_default();

        match result {
            Ok(result) => {
                let mut full_trace = prefix;
                full_trace.extend(result.trace);
                let trace_json = serde_json::Value::Array(full_trace);
                let step_count = result.step_index as i32;

                if result.approval_token.is_some() {
                    // Approval gates are not yet implemented (WF-08).
                    // Fail explicitly rather than creating unreachable WaitingApproval rows.
                    tracing::warn!(
                        run_id = %run_id,
                        step_index = result.step_index,
                        "Workflow hit approval gate — not yet implemented, marking as failed"
                    );
                    if let Err(e) = self
                        .db
                        .update_workflow_run(
                            community_id,
                            run_id,
                            RunStatus::Failed,
                            step_count,
                            &trace_json,
                            Some(buzz_db::workflow::WorkflowRunFailure {
                                code: "approval_not_supported",
                                message: "approval gates not yet implemented — see WF-08",
                            }),
                        )
                        .await
                    {
                        tracing::error!(
                            run_id = %run_id,
                            "Failed to update run to Failed (approval gate): {e}"
                        );
                    }
                } else {
                    tracing::info!(run_id = %run_id, "Workflow run completed");
                    if let Err(e) = self
                        .db
                        .update_workflow_run(
                            community_id,
                            run_id,
                            RunStatus::Completed,
                            step_count,
                            &trace_json,
                            None,
                        )
                        .await
                    {
                        tracing::error!(
                            run_id = %run_id,
                            "Failed to update run to Completed: {e}"
                        );
                    }
                }
            }
            Err((e, progress)) => {
                tracing::error!(run_id = %run_id, error = %e.log_text(), "Workflow run failed");
                let mut full_trace = prefix;
                full_trace.extend(progress.trace);
                let trace_json = serde_json::Value::Array(full_trace);
                if let Err(db_err) = self
                    .db
                    .update_workflow_run(
                        community_id,
                        run_id,
                        RunStatus::Failed,
                        progress.step_index as i32,
                        &trace_json,
                        Some(buzz_db::workflow::WorkflowRunFailure {
                            code: e.code(),
                            message: &e.to_string(),
                        }),
                    )
                    .await
                {
                    tracing::error!(
                        run_id = %run_id,
                        "Failed to update run to Failed: {db_err}"
                    );
                }
            }
        }
    }

    /// Called from the event handler post-store hook for every stored event.
    ///
    /// Checks whether any workflow in the event's channel has a matching trigger.
    /// Workflow execution events (kinds 46001–46012) are excluded to prevent loops.
    ///
    /// `community_id` is the server-resolved community the event was stored
    /// under — `StoredEvent` does not carry it, and the same channel UUID can
    /// exist in two communities, so the workflow lookup/run-creation must be
    /// scoped to the caller's tenant or community B could trigger community A's
    /// workflow on a colliding channel id.
    ///
    /// The method takes `self: &Arc<Self>` so that the spawned task can hold a
    /// clone of the `Arc` without requiring `'static` on `&self`.
    pub async fn on_event(
        self: &Arc<Self>,
        community_id: CommunityId,
        event: &buzz_core::StoredEvent,
    ) -> Result<(), WorkflowError> {
        let Some(channel_id) = event.channel_id else {
            tracing::debug!(
                event_id = %event.event.id.to_hex(),
                kind = event_kind_u32(&event.event),
                "Skipping workflow trigger — event has no channel_id"
            );
            return Ok(());
        };

        let kind_u32 = event_kind_u32(&event.event);

        // Exclude workflow execution events to prevent infinite loops.
        if is_workflow_execution_kind(kind_u32) {
            return Ok(());
        }

        let cache_key = (community_id, channel_id);
        let workflows = match self.workflow_cache.get(&cache_key) {
            Some(cached) => cached,
            None => {
                let fresh = Arc::new(
                    self.db
                        .list_enabled_channel_workflows(community_id, channel_id)
                        .await
                        .map_err(WorkflowError::from)?,
                );
                self.workflow_cache.insert(cache_key, Arc::clone(&fresh));
                fresh
            }
        };

        if workflows.is_empty() {
            return Ok(());
        }

        let trigger_ctx = build_trigger_context(event);

        let trigger_ctx_json: serde_json::Value = match serde_json::to_value(&trigger_ctx) {
            Ok(v) => v,
            Err(e) => {
                tracing::error!("Failed to serialize trigger context: {e}");
                return Ok(());
            }
        };

        for workflow in workflows.iter() {
            let def: WorkflowDef = match serde_json::from_value(workflow.definition.clone()) {
                Ok(d) => d,
                Err(e) => {
                    tracing::warn!(workflow_id = %workflow.id, "Failed to parse definition: {e}");
                    continue;
                }
            };

            if !def.enabled || !trigger_matches_event(&def.trigger, kind_u32) {
                continue;
            }

            if !should_fire_workflow(&def, &trigger_ctx, workflow.id).await {
                continue;
            }

            // SEC-006: recheck the owner's *current* channel authority
            // immediately before run creation. The cached workflow list can be
            // up to 10s stale, and disable-on-removal can race a concurrent
            // event — this per-fire gate is the authoritative, fail-closed
            // check that a removed (or under-privileged, for exfiltration
            // definitions) owner cannot cause a run.
            if let Err(e) = self
                .check_owner_authority(community_id, channel_id, &workflow.owner_pubkey, &def)
                .await
            {
                self.log_owner_denial(community_id, workflow.id, "event", &e);
                continue;
            }

            let trigger_event_id_bytes = event.event.id.as_bytes().to_vec();
            let run_id = match self
                .db
                .create_workflow_run(
                    community_id,
                    workflow.id,
                    Some(&trigger_event_id_bytes),
                    Some(&trigger_ctx_json),
                )
                .await
            {
                Ok(id) => id,
                Err(e) => {
                    tracing::error!(workflow_id = %workflow.id, "Failed to create run: {e}");
                    continue;
                }
            };

            tracing::debug!(
                workflow_id = %workflow.id,
                run_id = %run_id,
                "Workflow triggered — spawning execution"
            );

            let engine = Arc::clone(self);
            let def_clone = def.clone();
            let ctx_clone = trigger_ctx.clone();

            tokio::spawn(async move {
                let result =
                    executor::execute_run(&engine, community_id, run_id, &def_clone, &ctx_clone)
                        .await;
                engine
                    .finalize_run(community_id, run_id, result, None)
                    .await;
            });
        }

        Ok(())
    }

    /// Interval prefilter: decide whether the interval workflow should fire this
    /// tick, applying the cold-start anchor seed as a side effect.
    ///
    /// `last` is the resolved anchor (in-memory entry if present, else the
    /// durable `latest_scheduled_workflow_fire` read). Returns `true` to proceed
    /// to the durable claim, `false` to suppress this tick.
    ///
    /// Cold-start liveness: a brand-new interval workflow has no in-memory entry
    /// AND no prior claim, so `last` is `None`. `interval_should_fire` then reads
    /// `last = now` and suppresses — correct for the first tick (wait a full
    /// interval), but the in-memory anchor is only written *after* a successful
    /// claim, and no claim is attempted until the prefilter passes. Without
    /// seeding, every subsequent tick repeats with `last = None` and the workflow
    /// suppresses forever. So on the `None` suppress path we seed `now`: the next
    /// tick counts from a real anchor and the workflow fires after one interval.
    /// We seed ONLY when `last` was `None`; when `last` is `Some` we are correctly
    /// mid-interval and must not advance the anchor, or it would never elapse.
    fn interval_prefilter_should_fire(
        &self,
        community_id: CommunityId,
        workflow_id: Uuid,
        dur: &str,
        last: Option<DateTime<Utc>>,
        now: DateTime<Utc>,
    ) -> bool {
        interval_prefilter_should_fire(&self.last_fired, community_id, workflow_id, dur, last, now)
    }

    /// Background loop for scheduled (cron/interval) triggers.
    ///
    /// Ticks every 60 seconds. For each active workflow with a `Schedule`
    /// trigger, checks whether the cron expression or interval has elapsed
    /// and spawns execution if so.
    ///
    /// Uses window-based matching for cron expressions to handle tick drift:
    /// `schedule.after(&(now - 60s)).next() <= now` instead of `includes(now)`.
    ///
    /// Interval tracking is anchored on the durable scheduled-fire claim:
    /// `last_fired` is an in-memory pre-filter, but the
    /// `(community_id, workflow_id, scheduled_for)` claim row is the
    /// at-most-once boundary across pods and restarts. On the first tick after
    /// a restart the interval anchor is seeded from
    /// `latest_scheduled_workflow_fire` so a process bounce cannot double-fire
    /// within an interval.
    pub async fn run(self: &Arc<Self>) {
        tracing::info!("WorkflowEngine cron loop started (60s tick)");

        loop {
            tokio::time::sleep(std::time::Duration::from_secs(60)).await;

            self.cron_tick(Utc::now()).await;
        }
    }

    /// One pass of the cron loop at `now`: fire every enabled schedule
    /// workflow that is due, then prune the interval clock.
    pub(crate) async fn cron_tick(self: &Arc<Self>, now: DateTime<Utc>) {
        let workflows = match self.db.list_all_enabled_workflows().await {
            Ok(wf) => wf,
            Err(e) => {
                tracing::error!("Cron tick: failed to load workflows: {e}");
                return;
            }
        };

        for workflow in &workflows {
            // The same workflow UUID may exist in another community; carry
            // the row's owning community through fire-tracking, run creation,
            // and execution so a fire/run never crosses tenants.
            let community_id = workflow.community_id;
            let def: schema::WorkflowDef = match serde_json::from_value(workflow.definition.clone())
            {
                Ok(d) => d,
                Err(e) => {
                    tracing::warn!(
                        workflow_id = %workflow.id,
                        "Cron tick: failed to parse workflow definition: {e}"
                    );
                    continue;
                }
            };

            if !def.enabled {
                continue;
            }

            // Fix 2: skip workflows with no channel_id — an empty channel_id
            // causes silent downstream failures when the run tries to act on a channel.
            let Some(channel_id) = workflow.channel_id else {
                tracing::warn!(
                    workflow_id = %workflow.id,
                    "Cron tick: skipping schedule workflow with no channel_id"
                );
                continue;
            };

            // Resolve the *deterministic* schedule instant this tick is
            // firing for. `scheduled_for` is computed identically on every
            // pod (cron's own scheduled time, or the interval bucket
            // boundary) so all pods collide on a single durable claim —
            // never `now`, which is per-pod and would let every pod fire.
            let (scheduled_for, trigger_type) = match &def.trigger {
                schema::TriggerDef::Schedule {
                    cron: Some(expr),
                    interval: None,
                } => match cron_fire_instant(expr, now, 60, workflow.id) {
                    Some(instant) => (instant, "cron"),
                    None => continue,
                },
                schema::TriggerDef::Schedule {
                    cron: None,
                    interval: Some(dur),
                } => {
                    // Cheap pre-filter: skip the claim attempt when the
                    // in-memory clock says we're clearly mid-interval. The
                    // durable claim below is the real at-most-once boundary;
                    // this only avoids a DB write every tick. Seed the
                    // anchor from the DB on the first tick after restart so
                    // a process bounce can't double-fire within an interval.
                    let last = match self.last_fired.get(&(community_id, workflow.id)) {
                        Some(t) => Some(*t),
                        None => match self
                            .db
                            .latest_scheduled_workflow_fire(community_id, workflow.id)
                            .await
                        {
                            Ok(anchor) => anchor,
                            Err(e) => {
                                // Fail closed: a missing anchor reads as
                                // last_fired = now in interval_should_fire,
                                // so this tick is suppressed and the next
                                // tick retries. Surface the read failure so
                                // a persistently-unreadable anchor is visible
                                // rather than silently stalling the schedule.
                                tracing::warn!(
                                    community_id = %community_id,
                                    workflow_id = %workflow.id,
                                    "Cron tick: failed to read interval restart anchor, \
                                     suppressing this tick: {e}"
                                );
                                None
                            }
                        },
                    };
                    if !self.interval_prefilter_should_fire(
                        community_id,
                        workflow.id,
                        dur,
                        last,
                        now,
                    ) {
                        continue;
                    }
                    match interval_fire_instant(dur, now, workflow.id) {
                        Some(instant) => (instant, "interval"),
                        None => continue,
                    }
                }
                _ => continue, // Non-schedule triggers handled by on_event()
            };

            // SEC-006: recheck the owner's current channel authority
            // BEFORE the durable claim. Placing the gate after the claim
            // would let a revoked owner's workflow consume the
            // at-most-once fire slot (claims are never re-fired), turning
            // revocation into a denial-of-fire for a later re-enable.
            if let Err(e) = self
                .check_owner_authority(community_id, channel_id, &workflow.owner_pubkey, &def)
                .await
            {
                self.log_owner_denial(community_id, workflow.id, trigger_type, &e);
                continue;
            }

            // Durable at-most-once claim — the cross-pod fire boundary.
            // The loser receives `None` and skips BEFORE any run creation or
            // side effect. `community_id` is the workflow row's own
            // community (server provenance from the scan), never client
            // input; the claim binds `(community_id, workflow_id,
            // scheduled_for)` so a duplicate workflow UUID in another
            // community claims independently.
            match self
                .db
                .claim_scheduled_workflow_fire(community_id, workflow.id, scheduled_for)
                .await
            {
                Ok(Some(_)) => {}
                Ok(None) => {
                    // Another pod (or an earlier tick this pod) already
                    // claimed this instant. Still advance the in-memory
                    // interval clock so we don't re-attempt the claim every
                    // tick for the rest of the interval.
                    if trigger_type == "interval" {
                        self.last_fired.insert((community_id, workflow.id), now);
                    }
                    continue;
                }
                Err(e) => {
                    tracing::error!(
                        workflow_id = %workflow.id,
                        "Cron tick: scheduled-fire claim failed: {e}"
                    );
                    continue;
                }
            }

            // Fix 5: handle serialization errors explicitly rather than silently
            // dropping the trigger context with .ok().
            let trigger_ctx = executor::TriggerContext {
                channel_id: channel_id.to_string(),
                timestamp: now.timestamp().to_string(),
                ..Default::default()
            };
            let trigger_ctx_json = match serde_json::to_value(&trigger_ctx) {
                Ok(v) => Some(v),
                Err(e) => {
                    tracing::error!(
                        workflow_id = %workflow.id,
                        "Cron tick: failed to serialize trigger context: {e}"
                    );
                    continue;
                }
            };

            let run_id = match self
                .db
                .create_workflow_run(
                    community_id,
                    workflow.id,
                    None, // no trigger event for cron
                    trigger_ctx_json.as_ref(),
                )
                .await
            {
                Ok(id) => id,
                Err(e) => {
                    tracing::error!(
                        workflow_id = %workflow.id,
                        "Cron tick: failed to create workflow run: {e}"
                    );
                    // The claim is held but the run failed to create. The
                    // claim row intentionally stays (its `workflow_run_id`
                    // NULL) so this instant is not re-fired: at-most-once is
                    // preserved over exactly-once on transient run-insert
                    // failures.
                    continue;
                }
            };

            // Link the won claim to its run for ops/audit forensics. The
            // claim row already guarantees dedupe; this is best-effort.
            if let Err(e) = self
                .db
                .attach_scheduled_workflow_run(community_id, workflow.id, scheduled_for, run_id)
                .await
            {
                tracing::warn!(
                    workflow_id = %workflow.id,
                    run_id = %run_id,
                    "Cron tick: failed to attach run to scheduled-fire claim: {e}"
                );
            }

            // Update last_fired AFTER a successful claim+insert so that a
            // failure doesn't suppress the next tick for the full interval.
            // Only needed for interval triggers — cron uses window-based
            // matching which already prevents double-fire within the same
            // minute, and the durable claim backstops both.
            if trigger_type == "interval" {
                self.last_fired.insert((community_id, workflow.id), now);
            }

            // Fix 6: log the specific trigger type (cron vs interval).
            tracing::info!(
                workflow_id = %workflow.id,
                run_id = %run_id,
                trigger = trigger_type,
                "Cron trigger fired"
            );

            let engine = Arc::clone(self);
            let def_clone = def.clone();
            let ctx_clone = trigger_ctx.clone();
            tokio::spawn(async move {
                let result =
                    executor::execute_run(&engine, community_id, run_id, &def_clone, &ctx_clone)
                        .await;
                engine
                    .finalize_run(community_id, run_id, result, None)
                    .await;
            });
        }

        // Fix 1: prune stale last_fired entries for workflows that are no longer
        // active/enabled. Without this the DashMap grows monotonically as
        // workflows are deleted or disabled. Keyed by `(community_id, id)` so
        // entries are matched to the same scope they were inserted under.
        let active_ids: std::collections::HashSet<(CommunityId, Uuid)> =
            workflows.iter().map(|w| (w.community_id, w.id)).collect();
        self.last_fired.retain(|key, _| active_ids.contains(key));
    }
}

/// The cache behind [`WorkflowEngine::ban_denial_warned`]: an entry expires
/// `warn_interval` after the WARN that inserted it, so the next ban denial
/// for that workflow warns again.
fn ban_denial_warn_cache(
    warn_interval: std::time::Duration,
) -> moka::sync::Cache<(CommunityId, Uuid), ()> {
    moka::sync::Cache::builder()
        .max_capacity(10_000)
        .time_to_live(warn_interval)
        .build()
}

/// Whether an owner-authority denial for this workflow logs at WARN. A ban
/// denial warns only when no warning for the same workflow is still inside
/// `warned`'s TTL; every other denial always warns.
fn should_warn_owner_denial(
    warned: &moka::sync::Cache<(CommunityId, Uuid), ()>,
    community_id: CommunityId,
    workflow_id: Uuid,
    err: &WorkflowError,
) -> bool {
    if !matches!(err, WorkflowError::OwnerBanned) {
        return true;
    }
    warned
        .entry((community_id, workflow_id))
        .or_insert(())
        .is_fresh()
}

/// Find the cron schedule instant that fired within the `window_secs`-wide
/// window ending at `now`, if any.
///
/// Uses window-based matching: finds the next scheduled time after
/// `(now - window_secs)` and returns it when it falls at or before `now`.
/// This tolerates tick drift gracefully — a 61s tick won't miss a
/// minute-granularity cron expression. The returned instant is the cron's own
/// scheduled time (not `now`), so every pod evaluating the same expression in
/// the same window computes the *same* value — making it a safe, deterministic
/// claim anchor for cross-pod at-most-once firing.
///
/// Returns `None` (and logs a warning) if the expression is invalid or nothing
/// is due in the window.
fn cron_fire_instant(
    expr: &str,
    now: DateTime<Utc>,
    window_secs: i64,
    workflow_id: Uuid,
) -> Option<DateTime<Utc>> {
    let normalized = schema::normalize_cron(expr);
    match normalized.parse::<cron::Schedule>() {
        Ok(sched) => {
            let window_start = now - chrono::Duration::seconds(window_secs);
            sched.after(&window_start).next().filter(|t| *t <= now)
        }
        Err(e) => {
            tracing::warn!(
                workflow_id = %workflow_id,
                "Cron tick: invalid cron expression '{expr}': {e}"
            );
            None
        }
    }
}

/// Quantize `now` to the interval bucket boundary, yielding a deterministic
/// claim anchor that every pod computes identically within the same bucket.
///
/// The boundary is `floor(now / interval) * interval` from the Unix epoch.
/// Because the scheduler ticks every 60s and interval schedules are minutes or
/// longer, bounded cross-pod clock skew keeps all pods inside the same bucket,
/// so they collide on one `(community, workflow, scheduled_for)` claim — only
/// one wins and creates the run. Returns `None` if the duration is unparseable
/// or non-positive (the caller skips firing).
fn interval_fire_instant(
    dur: &str,
    now: DateTime<Utc>,
    workflow_id: Uuid,
) -> Option<DateTime<Utc>> {
    match executor::parse_duration_secs(dur) {
        Ok(interval_secs) if interval_secs > 0 => {
            let secs = interval_secs as i64;
            let bucket = (now.timestamp().div_euclid(secs)) * secs;
            DateTime::from_timestamp(bucket, 0)
        }
        Ok(_) => {
            tracing::warn!(
                workflow_id = %workflow_id,
                "Cron tick: interval duration is zero — skipping"
            );
            None
        }
        Err(e) => {
            tracing::warn!(
                workflow_id = %workflow_id,
                "Cron tick: invalid interval '{dur}': {e}"
            );
            None
        }
    }
}

/// Check whether an interval trigger should fire based on the last-fired time.
///
/// `last_fired` is `None` on the first tick after startup — in that case we
/// default to `now`, which prevents an immediate fire and waits a full interval.
///
/// Returns `false` (and logs a warning) if the duration string is invalid.
fn interval_should_fire(
    dur: &str,
    last_fired: Option<DateTime<Utc>>,
    now: DateTime<Utc>,
    workflow_id: Uuid,
) -> bool {
    match executor::parse_duration_secs(dur) {
        Ok(interval_secs) => {
            // Default to now on first tick — prevents immediate fire after startup.
            let last = last_fired.unwrap_or(now);
            let elapsed = (now - last).num_seconds().unsigned_abs();
            elapsed >= interval_secs
        }
        Err(e) => {
            tracing::warn!(
                workflow_id = %workflow_id,
                "Cron tick: invalid interval '{dur}': {e}"
            );
            false
        }
    }
}

/// Interval prefilter decision + cold-start anchor seed. See the
/// [`WorkflowEngine::interval_prefilter_should_fire`] wrapper for the liveness
/// rationale. Free function over the `last_fired` map so it is unit-testable
/// without a `Db`/Postgres: the only state it touches is the in-memory anchor.
///
/// Returns `true` to fire, `false` to suppress. On the cold-start `None` suppress
/// path it seeds `now` so the next tick has a real anchor; it never advances an
/// existing (`Some`) anchor, which is mid-interval and must elapse on its own.
fn interval_prefilter_should_fire(
    last_fired: &DashMap<(CommunityId, Uuid), DateTime<Utc>>,
    community_id: CommunityId,
    workflow_id: Uuid,
    dur: &str,
    last: Option<DateTime<Utc>>,
    now: DateTime<Utc>,
) -> bool {
    if interval_should_fire(dur, last, now, workflow_id) {
        return true;
    }
    if last.is_none() {
        last_fired.insert((community_id, workflow_id), now);
    }
    false
}

/// Check emoji and filter-expression conditions that determine whether a
/// matched workflow should actually fire. Extracted from `on_event` to keep
/// the per-workflow loop body small.
///
/// Returns `true` if the workflow should fire, `false` to skip.
async fn should_fire_workflow(
    def: &WorkflowDef,
    trigger_ctx: &executor::TriggerContext,
    workflow_id: uuid::Uuid,
) -> bool {
    if let TriggerDef::ReactionAdded {
        emoji: Some(ref expected),
        ..
    } = def.trigger
    {
        if &trigger_ctx.emoji != expected {
            tracing::debug!(
                workflow_id = %workflow_id,
                expected_emoji = %expected,
                actual_emoji = %trigger_ctx.emoji,
                "Reaction emoji mismatch — skipping workflow"
            );
            return false;
        }
    }

    let filter = match &def.trigger {
        TriggerDef::MessagePosted { filter }
        | TriggerDef::ReactionAdded { filter, .. }
        | TriggerDef::DiffPosted { filter } => filter.as_ref(),
        TriggerDef::Schedule { .. } | TriggerDef::Webhook => None,
    };
    if let Some(expr) = filter {
        match executor::evaluate_condition(expr, trigger_ctx, &HashMap::new()).await {
            Ok(true) => {}
            Ok(false) => {
                tracing::debug!(
                    workflow_id = %workflow_id,
                    "Trigger filter evaluated false — skipping workflow"
                );
                return false;
            }
            Err(e) => {
                tracing::warn!(
                    workflow_id = %workflow_id,
                    error_kind = e.code(),
                    "Trigger filter error — skipping workflow"
                );
                return false;
            }
        }
    }

    true
}

/// Build a [`executor::TriggerContext`] from a [`buzz_core::StoredEvent`].
///
/// - `text` — event content (message body or reaction emoji character)
/// - `author` — pubkey hex string
/// - `channel_id` — channel UUID as string (empty if no channel scope)
/// - `timestamp` — Unix timestamp as string
/// - `emoji` — for `KIND_REACTION` events, the content is the emoji; otherwise empty
/// - `message_id` — for reactions, the target message's event ID (from `e` tag);
///   for all other events, the event's own ID
pub fn build_trigger_context(event: &buzz_core::StoredEvent) -> executor::TriggerContext {
    let kind_u32 = event_kind_u32(&event.event);
    let content = event.event.content.clone();

    // Workflow conditions make authorization decisions from `trigger_author`,
    // so it must come from the event signature. An `actor` tag is ordinary
    // signer-controlled metadata and cannot speak for another pubkey.
    let author = event.event.pubkey.to_hex();

    // For reaction events (NIP-25), the content field holds the emoji character
    // or shortcode (e.g. "👍", "+", "-"). Expose it as `emoji`.
    let emoji = if kind_u32 == KIND_REACTION {
        content.clone()
    } else {
        String::new()
    };

    // For reactions (NIP-25), `message_id` should be the target message, not
    // the reaction event itself. NIP-25 stores the target in an `e` tag whose
    // value is a 64-char hex event ID (not a UUID channel reference).
    // Per NIP-25, the last `e` tag is the direct target (earlier ones may be thread roots).
    let message_id = if kind_u32 == KIND_REACTION {
        event
            .event
            .tags
            .iter()
            .rev()
            .find_map(|tag| {
                let key = tag.kind().to_string();
                if key == "e" {
                    tag.content().and_then(|v| {
                        // Distinguish hex event IDs (64 chars) from UUID channel refs.
                        if v.len() == 64 && v.chars().all(|c| c.is_ascii_hexdigit()) {
                            Some(v.to_string())
                        } else {
                            None
                        }
                    })
                } else {
                    None
                }
            })
            // Fallback to the reaction event's own ID if no valid `e` tag found.
            .unwrap_or_else(|| event.event.id.to_hex())
    } else {
        event.event.id.to_hex()
    };

    executor::TriggerContext {
        text: content,
        author,
        channel_id: event
            .channel_id
            .map(|id| id.to_string())
            .unwrap_or_default(),
        timestamp: event.event.created_at.as_secs().to_string(),
        emoji,
        message_id,
        is_reply: event_is_reply(&event.event),
        webhook_fields: HashMap::new(),
    }
}

/// True when an event is a threaded reply — it carries a valid NIP-10 `reply`
/// marker. Delegates to the shared [`buzz_core::nip10`] parser so this stays in
/// lockstep with ingest's `resolve_nip10_thread_meta`: a `root` marker alone is
/// top-level, and a marker with a malformed (non-64-hex) event id is ignored by
/// ingest, so it must not flip `trigger_is_reply` either — else a
/// `trigger_is_reply == false` workflow would skip a message ingest stored as a
/// new top-level post.
fn event_is_reply(event: &nostr::Event) -> bool {
    buzz_core::nip10::parse_thread_markers(&event.tags)
        .reply
        .is_some()
}

/// Pure authority decision for [`WorkflowEngine::check_owner_authority`].
///
/// `role` is the owner's *current* active role in the workflow's channel
/// (`None` = not an active member — removed, left, or never joined).
/// `needs_elevated` is true when the definition contains an
/// exfiltration-capable action (see `WorkflowDef::requires_elevated_authority`).
///
/// Rules:
/// - not a member ⇒ deny, always;
/// - a role that cannot write (`guest`, or a role this build does not
///   know) ⇒ deny: a run acts as its owner, and a guest cannot write;
/// - member or bot ⇒ allowed for ordinary definitions;
/// - elevated definitions ⇒ only `owner` / `admin` roles.
fn owner_authority_allows(role: Option<&str>, needs_elevated: bool) -> bool {
    let Some(role) = role.and_then(|r| r.parse::<buzz_core::channel::MemberRole>().ok()) else {
        return false;
    };
    if needs_elevated {
        role.is_elevated()
    } else {
        role.can_post()
    }
}

/// Returns `true` if the trigger type matches the given event kind.
fn trigger_matches_event(trigger: &TriggerDef, kind_u32: u32) -> bool {
    use buzz_core::kind::{KIND_REACTION, KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_DIFF};
    match trigger {
        TriggerDef::MessagePosted { .. } => kind_u32 == KIND_STREAM_MESSAGE,
        TriggerDef::ReactionAdded { .. } => kind_u32 == KIND_REACTION,
        TriggerDef::DiffPosted { .. } => kind_u32 == KIND_STREAM_MESSAGE_DIFF,
        // Schedule and Webhook triggers are not fired by channel events.
        TriggerDef::Schedule { .. } | TriggerDef::Webhook => false,
    }
}

#[cfg(test)]
mod postgres_tests {
    use super::*;

    #[test]
    fn cron_fire_instant_matches_within_window() {
        // "every minute" cron — should always fire within a 60s window.
        let now = chrono::DateTime::parse_from_rfc3339("2026-06-15T12:00:30Z")
            .unwrap()
            .with_timezone(&Utc);
        let wf_id = Uuid::new_v4();
        // The matched instant is the minute boundary 12:00:00, NOT `now`.
        assert_eq!(
            cron_fire_instant("* * * * *", now, 60, wf_id),
            Some(
                chrono::DateTime::parse_from_rfc3339("2026-06-15T12:00:00Z")
                    .unwrap()
                    .with_timezone(&Utc)
            ),
            "every-minute cron should return the minute boundary as the anchor"
        );
    }

    #[test]
    fn cron_fire_instant_returns_none_for_invalid_expr() {
        let now = Utc::now();
        let wf_id = Uuid::new_v4();
        assert!(
            cron_fire_instant("not-a-cron", now, 60, wf_id).is_none(),
            "invalid cron should return None"
        );
    }

    #[test]
    fn cron_fire_instant_returns_none_outside_window() {
        // Fixed time: 2026-06-15 14:30:00 UTC (a Sunday in June)
        let now = chrono::DateTime::parse_from_rfc3339("2026-06-15T14:30:00Z")
            .unwrap()
            .with_timezone(&Utc);
        let wf_id = Uuid::new_v4();
        // "0 0 1 1 *" = midnight on Jan 1 only — June 15 is definitely outside.
        assert!(
            cron_fire_instant("0 0 1 1 *", now, 60, wf_id).is_none(),
            "Jan-1-only cron should not fire on June 15"
        );
    }

    #[test]
    fn cron_fire_instant_at_exact_minute_boundary() {
        // Fixed time: exactly 09:00:00 UTC. Cron "0 9 * * *" fires at 09:00.
        // Window [08:59:00, 09:00:00] should contain the fire time.
        let now = chrono::DateTime::parse_from_rfc3339("2026-06-15T09:00:00Z")
            .unwrap()
            .with_timezone(&Utc);
        let wf_id = Uuid::new_v4();
        assert_eq!(
            cron_fire_instant("0 9 * * *", now, 60, wf_id),
            Some(now),
            "cron should fire at exact minute boundary, anchored on 09:00:00"
        );
    }

    #[test]
    fn cron_fire_instant_within_drift_window_anchors_on_scheduled_time() {
        // Fixed time: 09:00:45 UTC (45s drift). Cron "0 9 * * *" fires at 09:00.
        // Window [08:59:45, 09:00:45] should still contain 09:00:00. Critically,
        // the anchor is the *scheduled* 09:00:00 — not the drifted `now` — so a
        // second pod ticking at 09:00:50 computes the identical claim key.
        let now = chrono::DateTime::parse_from_rfc3339("2026-06-15T09:00:45Z")
            .unwrap()
            .with_timezone(&Utc);
        let wf_id = Uuid::new_v4();
        assert_eq!(
            cron_fire_instant("0 9 * * *", now, 60, wf_id),
            Some(
                chrono::DateTime::parse_from_rfc3339("2026-06-15T09:00:00Z")
                    .unwrap()
                    .with_timezone(&Utc)
            ),
            "cron anchor must be the scheduled instant, stable across pod tick drift"
        );
    }

    #[test]
    fn cron_fire_instant_returns_none_just_outside_window() {
        // Fixed time: 09:01:01 UTC. Cron "0 9 * * *" fires at 09:00:00.
        // Window [09:00:01, 09:01:01] does NOT contain 09:00:00.
        let now = chrono::DateTime::parse_from_rfc3339("2026-06-15T09:01:01Z")
            .unwrap()
            .with_timezone(&Utc);
        let wf_id = Uuid::new_v4();
        assert!(
            cron_fire_instant("0 9 * * *", now, 60, wf_id).is_none(),
            "cron should not fire 61s after the scheduled time"
        );
    }

    #[test]
    fn interval_fire_instant_quantizes_to_bucket_boundary() {
        // Two pods ticking at different sub-interval offsets must compute the
        // *same* bucket boundary so they collide on one claim. 1h interval,
        // epoch-aligned: 12:34:56 and 12:59:01 both floor to 12:00:00.
        let wf_id = Uuid::new_v4();
        let a = chrono::DateTime::parse_from_rfc3339("2026-06-15T12:34:56Z")
            .unwrap()
            .with_timezone(&Utc);
        let b = chrono::DateTime::parse_from_rfc3339("2026-06-15T12:59:01Z")
            .unwrap()
            .with_timezone(&Utc);
        let bucket = chrono::DateTime::parse_from_rfc3339("2026-06-15T12:00:00Z")
            .unwrap()
            .with_timezone(&Utc);
        assert_eq!(interval_fire_instant("1h", a, wf_id), Some(bucket));
        assert_eq!(interval_fire_instant("1h", b, wf_id), Some(bucket));
        // Next hour is a distinct bucket.
        let c = chrono::DateTime::parse_from_rfc3339("2026-06-15T13:00:10Z")
            .unwrap()
            .with_timezone(&Utc);
        let next_bucket = chrono::DateTime::parse_from_rfc3339("2026-06-15T13:00:00Z")
            .unwrap()
            .with_timezone(&Utc);
        assert_eq!(interval_fire_instant("1h", c, wf_id), Some(next_bucket));
    }

    #[test]
    fn interval_fire_instant_returns_none_for_invalid_duration() {
        let now = Utc::now();
        let wf_id = Uuid::new_v4();
        assert!(interval_fire_instant("not-a-duration", now, wf_id).is_none());
    }

    #[test]
    fn interval_should_fire_returns_false_on_first_tick() {
        // When last_fired is None (first tick), defaults to now → elapsed = 0 → false.
        let now = Utc::now();
        let wf_id = Uuid::new_v4();
        assert!(
            !interval_should_fire("1h", None, now, wf_id),
            "first tick should not fire immediately"
        );
    }

    #[test]
    fn interval_should_fire_returns_true_after_interval_elapsed() {
        let wf_id = Uuid::new_v4();
        let now = Utc::now();
        // last_fired was 2 hours ago; interval is 1h → should fire.
        let last = now - chrono::Duration::hours(2);
        assert!(
            interval_should_fire("1h", Some(last), now, wf_id),
            "should fire after interval elapsed"
        );
    }

    #[test]
    fn interval_should_fire_returns_false_before_interval_elapsed() {
        let wf_id = Uuid::new_v4();
        let now = Utc::now();
        // last_fired was 30 minutes ago; interval is 1h → should not fire.
        let last = now - chrono::Duration::minutes(30);
        assert!(
            !interval_should_fire("1h", Some(last), now, wf_id),
            "should not fire before interval elapsed"
        );
    }

    #[test]
    fn interval_should_fire_returns_false_for_invalid_duration() {
        let now = Utc::now();
        let wf_id = Uuid::new_v4();
        assert!(
            !interval_should_fire("not-a-duration", None, now, wf_id),
            "invalid duration should return false"
        );
    }

    #[test]
    fn interval_should_fire_at_exact_boundary() {
        let wf_id = Uuid::new_v4();
        let now = Utc::now();
        // last_fired was exactly 1 hour ago; interval is 1h → should fire (elapsed >= interval).
        let last = now - chrono::Duration::hours(1);
        assert!(
            interval_should_fire("1h", Some(last), now, wf_id),
            "should fire at exact interval boundary"
        );
    }

    // ── Interval cold-start liveness (Max's blocker on the scheduled lane) ──
    // A brand-new interval workflow has no in-memory anchor and no prior durable
    // claim, so the prefilter resolves `last = None`. Without seeding, every tick
    // reads `None`, suppresses, and writes nothing — the workflow never fires.
    // `interval_prefilter_should_fire` must seed `now` on that first suppress so a
    // real anchor exists for the next tick.

    #[test]
    fn interval_cold_start_seeds_anchor_then_fires_after_one_interval() {
        let map: DashMap<(CommunityId, Uuid), DateTime<Utc>> = DashMap::new();
        let community = CommunityId::from_uuid(Uuid::new_v4());
        let wf = Uuid::new_v4();
        let t0 = Utc::now();

        // Tick 1 (cold start): no in-memory entry, DB anchor is None → last = None.
        let fired_1 = interval_prefilter_should_fire(&map, community, wf, "1h", None, t0);
        assert!(!fired_1, "first tick must suppress (wait a full interval)");
        let seeded = map.get(&(community, wf)).map(|v| *v);
        assert_eq!(
            seeded,
            Some(t0),
            "first suppressed tick must seed the anchor to `now`, else it suppresses forever"
        );

        // Tick 2, mid-interval: caller now passes the seeded anchor as `last`.
        let t1 = t0 + chrono::Duration::minutes(30);
        let last = map.get(&(community, wf)).map(|v| *v);
        let fired_2 = interval_prefilter_should_fire(&map, community, wf, "1h", last, t1);
        assert!(!fired_2, "still mid-interval → suppress");
        assert_eq!(
            map.get(&(community, wf)).map(|v| *v),
            Some(t0),
            "mid-interval suppress must NOT advance the anchor (or it would never elapse)"
        );

        // Tick 3, one interval elapsed → fire.
        let t2 = t0 + chrono::Duration::hours(1);
        let last = map.get(&(community, wf)).map(|v| *v);
        let fired_3 = interval_prefilter_should_fire(&map, community, wf, "1h", last, t2);
        assert!(
            fired_3,
            "after one full interval the cold-started workflow must fire"
        );
    }

    #[test]
    fn interval_prefilter_does_not_advance_existing_anchor_on_suppress() {
        // Regression for the inverse bug: if a `Some` anchor were re-seeded to
        // `now` on every suppressed tick, the interval would never elapse.
        let map: DashMap<(CommunityId, Uuid), DateTime<Utc>> = DashMap::new();
        let community = CommunityId::from_uuid(Uuid::new_v4());
        let wf = Uuid::new_v4();
        let now = Utc::now();
        let anchor = now - chrono::Duration::minutes(10); // 10m into a 1h interval
        map.insert((community, wf), anchor);

        let fired = interval_prefilter_should_fire(&map, community, wf, "1h", Some(anchor), now);
        assert!(!fired, "mid-interval suppress");
        assert_eq!(
            map.get(&(community, wf)).map(|v| *v),
            Some(anchor),
            "existing anchor must be preserved exactly, not advanced to now"
        );
    }

    #[test]
    fn interval_prefilter_passes_through_a_due_fire_without_touching_anchor() {
        // When the interval has elapsed the prefilter returns true and leaves the
        // anchor to the post-claim update path (which writes `now` only on a won
        // claim), so the prefilter must not seed here.
        let map: DashMap<(CommunityId, Uuid), DateTime<Utc>> = DashMap::new();
        let community = CommunityId::from_uuid(Uuid::new_v4());
        let wf = Uuid::new_v4();
        let now = Utc::now();
        let anchor = now - chrono::Duration::hours(2); // overdue on a 1h interval

        let fired = interval_prefilter_should_fire(&map, community, wf, "1h", Some(anchor), now);
        assert!(fired, "overdue interval must fire");
        assert!(
            map.get(&(community, wf)).is_none(),
            "a firing tick must not seed via the prefilter; the post-claim path owns the write"
        );
    }

    #[test]
    fn workflow_config_defaults() {
        let cfg = WorkflowConfig::default();
        assert_eq!(cfg.max_concurrent, 100);
        assert_eq!(cfg.default_timeout_secs, 300);
    }

    #[test]
    fn parse_yaml_roundtrip() {
        let yaml = r#"
name: "Test Workflow"
trigger:
  on: message_posted
steps:
  - id: s1
    action: send_message
    text: "Hello {{trigger.author}}"
"#;
        let (def, json) = WorkflowEngine::parse_yaml(yaml).expect("parse failed");
        assert_eq!(def.name, "Test Workflow");

        let reparsed: WorkflowDef = serde_json::from_str(&json).expect("json round-trip");
        assert_eq!(reparsed.name, def.name);
        assert_eq!(reparsed.steps.len(), 1);
    }

    #[test]
    fn trigger_matches_stream_message() {
        let trigger = TriggerDef::MessagePosted { filter: None };
        assert!(trigger_matches_event(
            &trigger,
            buzz_core::kind::KIND_STREAM_MESSAGE
        ));
        assert!(!trigger_matches_event(
            &trigger,
            buzz_core::kind::KIND_REACTION
        ));
    }

    #[test]
    fn trigger_matches_reaction() {
        let trigger = TriggerDef::ReactionAdded {
            emoji: None,
            filter: None,
        };
        assert!(trigger_matches_event(
            &trigger,
            buzz_core::kind::KIND_REACTION
        ));
        assert!(!trigger_matches_event(
            &trigger,
            buzz_core::kind::KIND_STREAM_MESSAGE
        ));
    }

    #[tokio::test]
    async fn reaction_filter_matches_target_message() {
        let yaml = r#"
name: "React to one message"
trigger:
  on: reaction_added
  filter: 'trigger_message_id == "target-message"'
steps:
  - id: wait
    action: delay
    duration: 1s
"#;
        let (def, _) = WorkflowEngine::parse_yaml(yaml).expect("parse failed");
        let mut trigger_ctx = executor::TriggerContext {
            message_id: "target-message".to_owned(),
            ..Default::default()
        };

        assert!(
            should_fire_workflow(&def, &trigger_ctx, Uuid::new_v4()).await,
            "reaction to the selected message should fire"
        );

        trigger_ctx.message_id = "different-message".to_owned();
        assert!(
            !should_fire_workflow(&def, &trigger_ctx, Uuid::new_v4()).await,
            "reaction to a different message should be filtered out"
        );
    }

    #[test]
    fn schedule_trigger_never_matches_events() {
        let trigger = TriggerDef::Schedule {
            cron: Some("0 9 * * 1-5".to_owned()),
            interval: None,
        };
        // Schedule triggers are fired by the cron loop, not by events.
        assert!(!trigger_matches_event(
            &trigger,
            buzz_core::kind::KIND_STREAM_MESSAGE
        ));
        assert!(!trigger_matches_event(
            &trigger,
            buzz_core::kind::KIND_REACTION
        ));
        assert!(!trigger_matches_event(
            &trigger,
            buzz_core::kind::KIND_WORKFLOW_TRIGGERED
        ));
    }

    #[test]
    fn webhook_trigger_never_matches_events() {
        let trigger = TriggerDef::Webhook;
        assert!(!trigger_matches_event(
            &trigger,
            buzz_core::kind::KIND_STREAM_MESSAGE
        ));
        assert!(!trigger_matches_event(&trigger, 0));
    }

    #[test]
    fn message_posted_matches_kind_9_only() {
        let trigger = TriggerDef::MessagePosted { filter: None };
        // Must match KIND_STREAM_MESSAGE = 9.
        assert!(trigger_matches_event(&trigger, 9));
        // Must NOT match reaction (kind 7).
        assert!(!trigger_matches_event(&trigger, 7));
        // Must NOT match forum post (kind 45001).
        assert!(!trigger_matches_event(&trigger, 45001));
        // Must NOT match stream message v2 (kind 40002).
        assert!(!trigger_matches_event(&trigger, 40002));
    }

    #[test]
    fn reaction_added_matches_kind_7_only() {
        let trigger = TriggerDef::ReactionAdded {
            emoji: None,
            filter: None,
        };
        // Must match KIND_REACTION = 7.
        assert!(trigger_matches_event(&trigger, 7));
        // Must NOT match stream message (kind 9).
        assert!(!trigger_matches_event(&trigger, 9));
        // Must NOT match forum post (kind 45001).
        assert!(!trigger_matches_event(&trigger, 45001));
    }

    #[test]
    fn reaction_added_with_emoji_filter_still_matches_kind_7() {
        // The emoji filter is evaluated at execution time, not trigger-matching time.
        // trigger_matches_event only checks the kind number.
        let trigger = TriggerDef::ReactionAdded {
            emoji: Some("thumbsup".to_owned()),
            filter: None,
        };
        assert!(trigger_matches_event(&trigger, 7));
        assert!(!trigger_matches_event(&trigger, 9));
    }

    #[test]
    fn message_posted_with_filter_still_matches_kind_9() {
        // The filter expression is evaluated at execution time, not trigger-matching time.
        let trigger = TriggerDef::MessagePosted {
            filter: Some("str_contains(trigger_text, \"P1\")".to_owned()),
        };
        assert!(trigger_matches_event(&trigger, 9));
        assert!(!trigger_matches_event(&trigger, 7));
    }

    #[test]
    fn workflow_execution_kinds_do_not_match_any_trigger() {
        // Workflow execution events (46001–46012) must never match triggers
        // to prevent infinite loops. The on_event() method filters these out
        // before calling trigger_matches_event, but verify the function itself
        // also returns false for these kinds.
        let msg_trigger = TriggerDef::MessagePosted { filter: None };
        let react_trigger = TriggerDef::ReactionAdded {
            emoji: None,
            filter: None,
        };

        for kind in buzz_core::kind::KIND_WORKFLOW_TRIGGERED
            ..=buzz_core::kind::KIND_WORKFLOW_APPROVAL_DENIED
        {
            assert!(
                !trigger_matches_event(&msg_trigger, kind),
                "message_posted should not match workflow execution kind {kind}"
            );
            assert!(
                !trigger_matches_event(&react_trigger, kind),
                "reaction_added should not match workflow execution kind {kind}"
            );
        }
    }

    #[test]
    fn trigger_matches_event_kind_zero_matches_nothing() {
        // Kind 0 is a profile event — no trigger should match it.
        let msg_trigger = TriggerDef::MessagePosted { filter: None };
        let react_trigger = TriggerDef::ReactionAdded {
            emoji: None,
            filter: None,
        };
        let sched_trigger = TriggerDef::Schedule {
            cron: None,
            interval: Some("1h".to_owned()),
        };
        let webhook_trigger = TriggerDef::Webhook;

        assert!(!trigger_matches_event(&msg_trigger, 0));
        assert!(!trigger_matches_event(&react_trigger, 0));
        assert!(!trigger_matches_event(&sched_trigger, 0));
        assert!(!trigger_matches_event(&webhook_trigger, 0));
    }

    #[test]
    fn diff_posted_matches_kind_40008_only() {
        let trigger = TriggerDef::DiffPosted { filter: None };
        assert!(trigger_matches_event(&trigger, 40008));
        assert!(!trigger_matches_event(&trigger, 9));
        assert!(!trigger_matches_event(&trigger, 7));
    }

    #[test]
    fn message_posted_does_not_match_kind_40008() {
        let trigger = TriggerDef::MessagePosted { filter: None };
        assert!(!trigger_matches_event(&trigger, 40008));
        assert!(trigger_matches_event(&trigger, 9));
    }

    #[test]
    fn workflow_config_custom_values() {
        let cfg = WorkflowConfig {
            max_concurrent: 50,
            default_timeout_secs: 600,
        };
        assert_eq!(cfg.max_concurrent, 50);
        assert_eq!(cfg.default_timeout_secs, 600);
    }

    fn make_message_event() -> buzz_core::StoredEvent {
        use nostr::{EventBuilder, Keys, Kind};
        use uuid::Uuid;
        let keys = Keys::generate();
        let event = EventBuilder::new(Kind::Custom(9), "hello world")
            .tags([])
            .sign_with_keys(&keys)
            .expect("sign");
        buzz_core::StoredEvent::new(event, Some(Uuid::new_v4()))
    }

    /// Create a reaction event with an `e` tag pointing to a target message.
    fn make_reaction_event() -> (buzz_core::StoredEvent, String) {
        use nostr::{EventBuilder, Keys, Kind, Tag};
        use uuid::Uuid;
        let keys = Keys::generate();
        // Create a dummy target message ID (64-char hex).
        let target_keys = Keys::generate();
        let target_event = EventBuilder::new(Kind::Custom(9), "target msg")
            .tags([])
            .sign_with_keys(&target_keys)
            .expect("sign target");
        let target_id_hex = target_event.id.to_hex();
        // NIP-25: reaction references the target via an `e` tag.
        let e_tag = Tag::parse(["e", &target_id_hex]).expect("tag parse");
        let event = EventBuilder::new(Kind::Reaction, "👍")
            .tags([e_tag])
            .sign_with_keys(&keys)
            .expect("sign");
        (
            buzz_core::StoredEvent::new(event, Some(Uuid::new_v4())),
            target_id_hex,
        )
    }

    #[test]
    fn build_trigger_context_message_event() {
        let stored = make_message_event();
        let ctx = build_trigger_context(&stored);

        assert_eq!(ctx.text, "hello world");
        assert_eq!(ctx.author, stored.event.pubkey.to_hex());
        assert_eq!(ctx.channel_id, stored.channel_id.unwrap().to_string());
        assert_eq!(ctx.timestamp, stored.event.created_at.as_secs().to_string());
        assert_eq!(ctx.message_id, stored.event.id.to_hex());
        // Non-reaction events have empty emoji.
        assert_eq!(ctx.emoji, "");
        assert!(ctx.webhook_fields.is_empty());
        // A top-level message (no e-tags) is not a reply.
        assert!(!ctx.is_reply);
    }

    #[test]
    fn build_trigger_context_is_reply_true_for_threaded_message() {
        use nostr::{EventBuilder, Keys, Kind, Tag};
        use uuid::Uuid;
        let root = Keys::generate();
        let root_event = EventBuilder::new(Kind::Custom(9), "root")
            .tags([])
            .sign_with_keys(&root)
            .expect("sign root");
        let root_hex = root_event.id.to_hex();

        let keys = Keys::generate();
        let event = EventBuilder::new(Kind::Custom(9), "a threaded reply")
            .tags([
                Tag::parse(["e", &root_hex, "", "root"]).expect("root tag"),
                Tag::parse(["e", &root_hex, "", "reply"]).expect("reply tag"),
            ])
            .sign_with_keys(&keys)
            .expect("sign");
        let stored = buzz_core::StoredEvent::new(event, Some(Uuid::new_v4()));
        let ctx = build_trigger_context(&stored);
        assert!(ctx.is_reply, "message with reply/root e-tags is a reply");
    }

    #[test]
    fn build_trigger_context_is_reply_true_for_reply_only_marker() {
        // A NIP-10 `reply` marker without a `root` marker (the fallback ingest
        // treats as `root == reply`) is still a threaded reply.
        use nostr::{EventBuilder, Keys, Kind, Tag};
        use uuid::Uuid;
        let parent = Keys::generate();
        let parent_event = EventBuilder::new(Kind::Custom(9), "parent")
            .sign_with_keys(&parent)
            .expect("sign parent");
        let keys = Keys::generate();
        let event = EventBuilder::new(Kind::Custom(9), "reply only")
            .tags([Tag::parse(["e", &parent_event.id.to_hex(), "", "reply"]).expect("reply tag")])
            .sign_with_keys(&keys)
            .expect("sign");
        let stored = buzz_core::StoredEvent::new(event, Some(Uuid::new_v4()));
        let ctx = build_trigger_context(&stored);
        assert!(ctx.is_reply, "a lone `reply` marker is a reply");
    }

    #[test]
    fn build_trigger_context_is_reply_false_for_root_only_marker() {
        // Ingest treats `(root=Some, reply=None)` as top-level, so
        // `event_is_reply` must too — otherwise `trigger_is_reply == false`
        // would skip a message the relay stored as a new top-level post.
        use nostr::{EventBuilder, Keys, Kind, Tag};
        use uuid::Uuid;
        let root = Keys::generate();
        let root_event = EventBuilder::new(Kind::Custom(9), "root")
            .sign_with_keys(&root)
            .expect("sign root");
        let keys = Keys::generate();
        let event = EventBuilder::new(Kind::Custom(9), "root marker only")
            .tags([Tag::parse(["e", &root_event.id.to_hex(), "", "root"]).expect("root tag")])
            .sign_with_keys(&keys)
            .expect("sign");
        let stored = buzz_core::StoredEvent::new(event, Some(Uuid::new_v4()));
        let ctx = build_trigger_context(&stored);
        assert!(
            !ctx.is_reply,
            "a lone `root` marker is top-level to ingest, not a reply"
        );
    }

    #[test]
    fn build_trigger_context_is_reply_false_for_unmarked_e_tag() {
        // A bare `e` tag with no NIP-10 marker (e.g. a plain mention/quote) is
        // not treated as a thread reply — only `reply`/`root` markers count.
        use nostr::{EventBuilder, Keys, Kind, Tag};
        use uuid::Uuid;
        let other = Keys::generate();
        let other_event = EventBuilder::new(Kind::Custom(9), "other")
            .tags([])
            .sign_with_keys(&other)
            .expect("sign");
        let keys = Keys::generate();
        let event = EventBuilder::new(Kind::Custom(9), "quotes another")
            .tags([Tag::parse(["e", &other_event.id.to_hex()]).expect("bare e tag")])
            .sign_with_keys(&keys)
            .expect("sign");
        let stored = buzz_core::StoredEvent::new(event, Some(Uuid::new_v4()));
        let ctx = build_trigger_context(&stored);
        assert!(!ctx.is_reply, "unmarked e-tag must not count as a reply");
    }

    #[test]
    fn build_trigger_context_is_reply_false_for_malformed_reply_id() {
        // Ingest gates a marker on a valid 64-hex event id; a malformed reply
        // id is not a thread link, so ingest stores the event top-level. The
        // predicate must agree, or `trigger_is_reply == false` would skip it.
        use nostr::{EventBuilder, Keys, Kind, Tag};
        use uuid::Uuid;
        let keys = Keys::generate();
        let event = EventBuilder::new(Kind::Custom(9), "malformed reply marker")
            .tags([Tag::parse(["e", "bad", "", "reply"]).expect("reply tag")])
            .sign_with_keys(&keys)
            .expect("sign");
        let stored = buzz_core::StoredEvent::new(event, Some(Uuid::new_v4()));
        let ctx = build_trigger_context(&stored);
        assert!(
            !ctx.is_reply,
            "a malformed reply id is ignored by ingest, so it is top-level"
        );
    }

    #[test]
    fn build_trigger_context_is_reply_false_for_valid_root_malformed_reply() {
        // A valid `root` marker but a malformed `reply` id: ingest ignores the
        // reply and stores the event as root-only, i.e. top-level. The predicate
        // must not flip to reply on the malformed marker.
        use nostr::{EventBuilder, Keys, Kind, Tag};
        use uuid::Uuid;
        let root = Keys::generate();
        let root_event = EventBuilder::new(Kind::Custom(9), "root")
            .sign_with_keys(&root)
            .expect("sign root");
        let keys = Keys::generate();
        let event = EventBuilder::new(Kind::Custom(9), "valid root, malformed reply")
            .tags([
                Tag::parse(["e", &root_event.id.to_hex(), "", "root"]).expect("root tag"),
                Tag::parse(["e", "bad", "", "reply"]).expect("reply tag"),
            ])
            .sign_with_keys(&keys)
            .expect("sign");
        let stored = buzz_core::StoredEvent::new(event, Some(Uuid::new_v4()));
        let ctx = build_trigger_context(&stored);
        assert!(
            !ctx.is_reply,
            "a valid root with a malformed reply id is top-level to ingest"
        );
    }

    #[test]
    fn build_trigger_context_reaction_event() {
        let (stored, target_id_hex) = make_reaction_event();
        let ctx = build_trigger_context(&stored);

        // For reactions, content IS the emoji.
        assert_eq!(ctx.text, "👍");
        assert_eq!(ctx.emoji, "👍");
        assert_eq!(ctx.author, stored.event.pubkey.to_hex());
        // message_id should be the TARGET message, not the reaction event itself.
        assert_eq!(ctx.message_id, target_id_hex);
        assert_ne!(ctx.message_id, stored.event.id.to_hex());
        assert!(ctx.webhook_fields.is_empty());
    }

    #[test]
    fn build_trigger_context_no_channel_id() {
        use nostr::{EventBuilder, Keys, Kind};
        let keys = Keys::generate();
        let event = EventBuilder::new(Kind::Custom(9), "msg")
            .tags([])
            .sign_with_keys(&keys)
            .expect("sign");
        // channel_id = None (global/DM event)
        let stored = buzz_core::StoredEvent::new(event, None);
        let ctx = build_trigger_context(&stored);

        assert_eq!(ctx.channel_id, "");
        assert_eq!(ctx.text, "msg");
    }

    #[test]
    fn build_trigger_context_author_is_hex_pubkey() {
        let stored = make_message_event();
        let ctx = build_trigger_context(&stored);
        // Pubkey hex is 64 lowercase hex characters.
        assert_eq!(ctx.author.len(), 64);
        assert!(ctx.author.chars().all(|c| c.is_ascii_hexdigit()));
    }

    #[test]
    fn build_trigger_context_ignores_actor_tag() {
        use nostr::{EventBuilder, Keys, Kind, Tag};

        let signer = Keys::generate();
        let impersonated = Keys::generate();
        let event = EventBuilder::new(Kind::Custom(9), "forged actor")
            .tags([Tag::parse(["actor", &impersonated.public_key().to_hex()]).expect("actor tag")])
            .sign_with_keys(&signer)
            .expect("sign");
        let stored = buzz_core::StoredEvent::new(event, Some(uuid::Uuid::new_v4()));

        let ctx = build_trigger_context(&stored);

        assert_eq!(ctx.author, signer.public_key().to_hex());
        assert_ne!(ctx.author, impersonated.public_key().to_hex());
    }

    #[test]
    fn build_trigger_context_message_id_is_hex() {
        let stored = make_message_event();
        let ctx = build_trigger_context(&stored);
        // Event ID hex is 64 lowercase hex characters.
        assert_eq!(ctx.message_id.len(), 64);
        assert!(ctx.message_id.chars().all(|c| c.is_ascii_hexdigit()));
    }

    #[test]
    fn build_trigger_context_timestamp_is_numeric_string() {
        let stored = make_message_event();
        let ctx = build_trigger_context(&stored);
        // Timestamp must parse as a u64.
        ctx.timestamp
            .parse::<u64>()
            .expect("timestamp should be a u64 string");
    }

    #[test]
    fn test_build_trigger_context_reaction_multiple_e_tags() {
        // NIP-25: last e tag is the direct target, first may be thread root
        use nostr::{EventBuilder, EventId, Keys, Kind, Tag};
        use uuid::Uuid;

        let keys = Keys::generate();
        let thread_root_id = EventId::all_zeros();
        let direct_target_id = EventId::from_byte_array([0x42; 32]);

        let event = EventBuilder::new(Kind::Reaction, "👍")
            .tags([
                Tag::parse(["e", &thread_root_id.to_hex()]).unwrap(),
                Tag::parse(["e", &direct_target_id.to_hex()]).unwrap(),
            ])
            .sign_with_keys(&keys)
            .expect("sign");

        let stored = buzz_core::StoredEvent::new(event, Some(Uuid::new_v4()));
        let ctx = build_trigger_context(&stored);

        // Should pick the LAST e tag (direct target), not the first (thread root)
        assert_eq!(ctx.message_id, direct_target_id.to_hex());
    }

    // -- SEC-006: owner authority decision --------------------------------

    #[test]
    fn owner_authority_denies_non_members_always() {
        assert!(!owner_authority_allows(None, false));
        assert!(!owner_authority_allows(None, true));
    }

    #[test]
    fn owner_authority_allows_any_member_for_ordinary_definitions() {
        assert!(owner_authority_allows(Some("member"), false));
        assert!(owner_authority_allows(Some("bot"), false));
        assert!(owner_authority_allows(Some("admin"), false));
        assert!(owner_authority_allows(Some("owner"), false));
    }

    #[test]
    fn owner_authority_denies_roles_that_cannot_write() {
        for elevated in [false, true] {
            assert!(!owner_authority_allows(None, elevated));
            assert!(!owner_authority_allows(Some("guest"), elevated));
            assert!(!owner_authority_allows(Some("superuser"), elevated));
        }
    }

    #[test]
    fn owner_authority_requires_elevated_role_for_exfiltration_definitions() {
        assert!(!owner_authority_allows(Some("member"), true));
        assert!(owner_authority_allows(Some("admin"), true));
        assert!(owner_authority_allows(Some("owner"), true));
    }

    #[test]
    fn requires_elevated_authority_detects_call_webhook() {
        let (plain, _) = WorkflowEngine::parse_yaml(concat!(
            "name: plain\n",
            "trigger:\n  on: message_posted\n",
            "steps:\n  - id: s1\n    action: send_message\n    text: hi\n",
        ))
        .expect("parse plain");
        assert!(!plain.requires_elevated_authority());

        let (hook, _) = WorkflowEngine::parse_yaml(concat!(
            "name: hook\n",
            "trigger:\n  on: message_posted\n",
            "steps:\n  - id: s1\n    action: send_message\n    text: hi\n",
            "  - id: s2\n    action: call_webhook\n    url: https://example.com/x\n",
        ))
        .expect("parse hook");
        assert!(hook.requires_elevated_authority());
    }

    // -- SEC-006: event-path regression (requires Postgres) ----------------

    fn test_database_url() -> String {
        std::env::var("BUZZ_TEST_DATABASE_URL")
            .or_else(|_| std::env::var("DATABASE_URL"))
            // Local-only test default; this is not a production credential.
            .unwrap_or_else(|_| {
                let local_test_database = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1
                local_test_database.to_owned()
            })
    }

    async fn setup_db() -> buzz_db::Db {
        buzz_db::Db::new(&buzz_db::DbConfig {
            database_url: test_database_url(),
            ..Default::default()
        })
        .await
        .expect("connect test DB")
    }

    /// Create a community, a channel owned by `creator`, and add `member` as a
    /// plain member. Returns `(community, channel)`.
    async fn setup_channel(db: &buzz_db::Db, creator: &[u8], member: &[u8]) -> (CommunityId, Uuid) {
        let host = format!("sec006-{}.example", Uuid::new_v4().simple());
        let community = match db
            .create_community_with_owner(&host, &hex::encode(creator), &nostr::Keys::generate())
            .await
            .map(|write| write.outcome)
            .expect("create community")
        {
            buzz_db::CreateCommunityWithOwnerResult::Created(rec) => rec.id,
            other => panic!("unexpected community create result: {other:?}"),
        };
        db.ensure_user(community, creator)
            .await
            .expect("creator user");
        db.ensure_user(community, member)
            .await
            .expect("member user");
        let channel_id = Uuid::new_v4();
        db.create_channel_with_id(
            community,
            channel_id,
            &format!("ch-{}", channel_id.simple()),
            buzz_db::channel::ChannelType::Stream,
            buzz_db::channel::ChannelVisibility::Open,
            None,
            creator,
            None,
        )
        .await
        .expect("create channel");
        db.add_member(
            community,
            channel_id,
            member,
            buzz_db::channel::MemberRole::Member,
            Some(creator),
        )
        .await
        .expect("add member");
        (community, channel_id)
    }

    fn message_event(channel_id: Uuid) -> buzz_core::StoredEvent {
        let keys = nostr::Keys::generate();
        let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "hello")
            .sign_with_keys(&keys)
            .expect("sign");
        buzz_core::StoredEvent::new(event, Some(channel_id))
    }

    /// The event path must stop creating runs the moment the workflow's owner
    /// loses channel membership — even while the workflow row is still
    /// `enabled` (the disable-on-removal side effect is a separate, relay-side
    /// write; this gate must hold on its own).
    #[tokio::test]
    #[ignore = "requires Postgres"]
    async fn on_event_denies_run_after_owner_removed() {
        let db = setup_db().await;
        let creator = nostr::Keys::generate().public_key().to_bytes().to_vec();
        let member = nostr::Keys::generate().public_key().to_bytes().to_vec();
        let (community, channel_id) = setup_channel(&db, &creator, &member).await;

        let def_json = serde_json::json!({
            "name": "sec006-event",
            "trigger": {"on": "message_posted"},
            "steps": [{"id": "s1", "action": "send_message", "text": "hi"}],
            "enabled": true,
        })
        .to_string();
        let workflow_id = db
            .create_workflow(
                community,
                Some(channel_id),
                &member,
                "sec006-event",
                &def_json,
                &[0u8; 32],
            )
            .await
            .expect("create workflow");

        let engine = Arc::new(WorkflowEngine::new(db.clone(), WorkflowConfig::default()));

        // Owner is an active member: the event fires the workflow.
        engine
            .on_event(community, &message_event(channel_id))
            .await
            .expect("on_event while member");
        let runs = db
            .list_workflow_runs(community, workflow_id, 10)
            .await
            .expect("list runs");
        assert_eq!(runs.len(), 1, "member owner's workflow must fire");

        // Remove the owner (actor = channel creator, an owner-role member).
        db.remove_member(community, channel_id, &member, &creator)
            .await
            .expect("remove member");

        // Workflow row is still enabled — only the authority gate stands.
        engine
            .on_event(community, &message_event(channel_id))
            .await
            .expect("on_event after removal");
        let runs = db
            .list_workflow_runs(community, workflow_id, 10)
            .await
            .expect("list runs after removal");
        assert_eq!(
            runs.len(),
            1,
            "no new run may be created after the owner lost membership"
        );
    }

    /// A ban leaves the owner's channel membership in place, so the event
    /// path must still stop creating runs while the owner is banned, and
    /// resume once the ban is lifted. Cron and webhook runs pass through the
    /// same `check_owner_authority` gate.
    #[tokio::test]
    #[ignore = "requires Postgres"]
    async fn on_event_denies_run_while_owner_banned() {
        let db = setup_db().await;
        let creator = nostr::Keys::generate().public_key().to_bytes().to_vec();
        let member = nostr::Keys::generate().public_key().to_bytes().to_vec();
        let (community, channel_id) = setup_channel(&db, &creator, &member).await;

        let def_json = serde_json::json!({
            "name": "ban-event",
            "trigger": {"on": "message_posted"},
            "steps": [{"id": "s1", "action": "send_message", "text": "hi"}],
            "enabled": true,
        })
        .to_string();
        let workflow_id = db
            .create_workflow(
                community,
                Some(channel_id),
                &member,
                "ban-event",
                &def_json,
                &[0u8; 32],
            )
            .await
            .expect("create workflow");
        let engine = Arc::new(WorkflowEngine::new(db.clone(), WorkflowConfig::default()));
        let run_count = || async {
            db.list_workflow_runs(community, workflow_id, 10)
                .await
                .expect("list runs")
                .len()
        };

        db.ban_community_member(community, &member, &creator, None, None)
            .await
            .expect("ban owner");
        engine
            .on_event(community, &message_event(channel_id))
            .await
            .expect("on_event while banned");
        assert_eq!(
            run_count().await,
            0,
            "a banned owner's workflow must not fire"
        );

        assert!(db
            .unban_community_member(community, &member, &creator)
            .await
            .expect("unban owner"));
        engine
            .on_event(community, &message_event(channel_id))
            .await
            .expect("on_event after unban");
        assert_eq!(
            run_count().await,
            1,
            "an unbanned owner's workflow fires again"
        );
    }

    /// Exfiltration-capable definitions (call_webhook) require the owner to
    /// currently hold an elevated role — a plain member's workflow must not
    /// fire even though the owner is still an active channel member.
    #[tokio::test]
    #[ignore = "requires Postgres"]
    async fn on_event_denies_webhook_definition_for_plain_member_owner() {
        let db = setup_db().await;
        let creator = nostr::Keys::generate().public_key().to_bytes().to_vec();
        let member = nostr::Keys::generate().public_key().to_bytes().to_vec();
        let (community, channel_id) = setup_channel(&db, &creator, &member).await;

        let def_json = serde_json::json!({
            "name": "sec006-hook",
            "trigger": {"on": "message_posted"},
            "steps": [{"id": "s1", "action": "call_webhook", "url": "https://example.com/x"}],
            "enabled": true,
        })
        .to_string();

        // Same definition, two owners: plain member vs channel owner.
        let wf_member = db
            .create_workflow(
                community,
                Some(channel_id),
                &member,
                "hook-member",
                &def_json,
                &[0u8; 32],
            )
            .await
            .expect("create member workflow");
        let wf_owner = db
            .create_workflow(
                community,
                Some(channel_id),
                &creator,
                "hook-owner",
                &def_json,
                &[1u8; 32],
            )
            .await
            .expect("create owner workflow");

        let engine = Arc::new(WorkflowEngine::new(db.clone(), WorkflowConfig::default()));
        engine
            .on_event(community, &message_event(channel_id))
            .await
            .expect("on_event");

        let member_runs = db
            .list_workflow_runs(community, wf_member, 10)
            .await
            .expect("member runs");
        assert!(
            member_runs.is_empty(),
            "plain member's call_webhook workflow must not fire"
        );
        let owner_runs = db
            .list_workflow_runs(community, wf_owner, 10)
            .await
            .expect("owner runs");
        assert_eq!(
            owner_runs.len(),
            1,
            "channel owner's call_webhook workflow fires"
        );
    }

    #[test]
    fn ban_denial_warns_once_per_interval_while_other_denials_always_warn() {
        let interval = std::time::Duration::from_millis(500);
        let warned = ban_denial_warn_cache(interval);
        let community = CommunityId::from_uuid(Uuid::new_v4());
        let (wf_a, wf_b) = (Uuid::new_v4(), Uuid::new_v4());
        let banned = WorkflowError::OwnerBanned;
        let lookup_failed = WorkflowError::Unauthorized(
            "owner restriction lookup failed (fail-closed): boom".into(),
        );

        assert!(should_warn_owner_denial(&warned, community, wf_a, &banned));
        assert!(
            !should_warn_owner_denial(&warned, community, wf_a, &banned),
            "a repeated ban denial inside the interval logs below WARN"
        );
        assert!(
            should_warn_owner_denial(&warned, community, wf_b, &banned),
            "each workflow gets its own first warning"
        );
        assert!(
            should_warn_owner_denial(&warned, community, wf_a, &lookup_failed),
            "a failed lookup always warns"
        );
        assert!(
            should_warn_owner_denial(&warned, community, wf_a, &lookup_failed),
            "a failed lookup is never rate-limited"
        );

        std::thread::sleep(interval * 2);
        assert!(
            should_warn_owner_denial(&warned, community, wf_a, &banned),
            "a ban denial warns again once the interval has passed"
        );
        assert!(
            !should_warn_owner_denial(&warned, community, wf_a, &banned),
            "and is suppressed again for the next interval"
        );
    }

    fn schedule_def(name: &str) -> String {
        serde_json::json!({
            "name": name,
            "trigger": {"on": "schedule", "cron": "* * * * *"},
            "steps": [{"id": "s1", "action": "send_message", "text": "tick"}],
            "enabled": true,
        })
        .to_string()
    }

    /// A cron tick must skip a banned owner's workflow and an agent-owned
    /// workflow whose human owner is banned, without consuming either
    /// workflow's scheduled-fire claim: once unbanned, a tick for the same
    /// instant fires them. An unbanned workflow on the same tick still fires.
    ///
    /// `cron_tick` scans every enabled workflow in the database. The
    /// PostgreSQL lane gives each test process its own database
    /// (`scripts/postgres-test-wrapper.sh`), and the assertions read only
    /// this test's workflows, so another test's schedules cannot affect it.
    #[tokio::test]
    #[ignore = "requires Postgres"]
    async fn cron_tick_skips_banned_owners_without_consuming_the_claim() {
        let db = setup_db().await;
        let creator = nostr::Keys::generate().public_key().to_bytes().to_vec();
        let member = nostr::Keys::generate().public_key().to_bytes().to_vec();
        let (community, channel_id) = setup_channel(&db, &creator, &member).await;

        let human = nostr::Keys::generate().public_key().to_bytes().to_vec();
        let agent = nostr::Keys::generate().public_key().to_bytes().to_vec();
        db.ensure_user(community, &human).await.expect("human user");
        db.ensure_user(community, &agent).await.expect("agent user");
        assert!(db
            .set_agent_owner(community, &agent, &human)
            .await
            .expect("set agent owner"));
        db.add_member(
            community,
            channel_id,
            &agent,
            buzz_db::channel::MemberRole::Member,
            Some(&creator),
        )
        .await
        .expect("add agent member");

        let mut workflows = Vec::new();
        for (owner, name, secret) in [
            (&member, "cron-banned-owner", [0u8; 32]),
            (&agent, "cron-banned-agent-owner", [1u8; 32]),
            (&creator, "cron-control", [2u8; 32]),
        ] {
            let id = db
                .create_workflow(
                    community,
                    Some(channel_id),
                    owner,
                    name,
                    &schedule_def(name),
                    &secret,
                )
                .await
                .expect("create workflow");
            workflows.push(id);
        }
        let [banned_wf, agent_wf, control_wf] =
            <[Uuid; 3]>::try_from(workflows).expect("three workflows created");
        let engine = Arc::new(WorkflowEngine::new(db.clone(), WorkflowConfig::default()));
        let run_count = |workflow_id| {
            let db = db.clone();
            async move {
                db.list_workflow_runs(community, workflow_id, 10)
                    .await
                    .expect("list runs")
                    .len()
            }
        };

        db.ban_community_member(community, &member, &creator, None, None)
            .await
            .expect("ban owner");
        db.ban_community_member(community, &human, &creator, None, None)
            .await
            .expect("ban agent's human owner");
        let now = Utc::now();
        engine.cron_tick(now).await;
        assert_eq!(run_count(banned_wf).await, 0, "banned owner must not fire");
        assert_eq!(
            run_count(agent_wf).await,
            0,
            "an agent whose human owner is banned must not fire"
        );
        assert_eq!(
            run_count(control_wf).await,
            1,
            "an unbanned workflow on the same tick still fires"
        );

        for pubkey in [&member, &human] {
            assert!(db
                .unban_community_member(community, pubkey, &creator)
                .await
                .expect("unban"));
        }
        engine.cron_tick(now).await;
        assert_eq!(
            run_count(banned_wf).await,
            1,
            "the denied tick must not have consumed the owner's claim"
        );
        assert_eq!(
            run_count(agent_wf).await,
            1,
            "the denied tick must not have consumed the agent's claim"
        );
        assert_eq!(
            run_count(control_wf).await,
            1,
            "the control's claim for this instant is already taken"
        );
    }

    /// The event path fails closed when the owner's restriction state cannot
    /// be read, and the same event fires once the read succeeds.
    #[tokio::test]
    #[ignore = "requires Postgres"]
    async fn on_event_denies_run_when_restriction_lookup_fails() {
        use sqlx::postgres::PgConnectOptions;

        let db = setup_db().await;
        let creator = nostr::Keys::generate().public_key().to_bytes().to_vec();
        let member = nostr::Keys::generate().public_key().to_bytes().to_vec();
        let (community, channel_id) = setup_channel(&db, &creator, &member).await;
        let def_json = serde_json::json!({
            "name": "lookup-fails",
            "trigger": {"on": "message_posted"},
            "steps": [{"id": "s1", "action": "send_message", "text": "hi"}],
            "enabled": true,
        })
        .to_string();
        let workflow_id = db
            .create_workflow(
                community,
                Some(channel_id),
                &member,
                "lookup-fails",
                &def_json,
                &[0u8; 32],
            )
            .await
            .expect("create workflow");

        // Shadow `community_bans` with a view that errors on read; every
        // other table still resolves to `public`.
        let schema = format!("ban_lookup_fails_{}", Uuid::new_v4().simple());
        sqlx::raw_sql(sqlx::AssertSqlSafe(format!(
            "CREATE SCHEMA {schema}; \
             CREATE VIEW {schema}.community_bans AS \
             SELECT * FROM public.community_bans WHERE 1 / (SELECT count(*) * 0 FROM public.communities) = 1;"
        )))
        .execute(db.pool())
        .await
        .expect("create failing restriction schema");
        let options = test_database_url()
            .parse::<PgConnectOptions>()
            .expect("database url")
            .options([("search_path", format!("{schema}, public"))]);
        let failing_pool = sqlx::PgPool::connect_with(options)
            .await
            .expect("failing restriction pool");
        let failing_engine = Arc::new(WorkflowEngine::new(
            buzz_db::Db::from_pool(failing_pool.clone()),
            WorkflowConfig::default(),
        ));
        let run_count = || async {
            db.list_workflow_runs(community, workflow_id, 10)
                .await
                .expect("list runs")
                .len()
        };

        failing_engine
            .on_event(community, &message_event(channel_id))
            .await
            .expect("on_event with failing lookup");
        assert_eq!(
            run_count().await,
            0,
            "a failed restriction lookup must not create a run"
        );

        let engine = Arc::new(WorkflowEngine::new(db.clone(), WorkflowConfig::default()));
        engine
            .on_event(community, &message_event(channel_id))
            .await
            .expect("on_event with healthy lookup");
        assert_eq!(run_count().await, 1, "a readable, unbanned owner fires");

        failing_pool.close().await;
        sqlx::raw_sql(sqlx::AssertSqlSafe(format!("DROP SCHEMA {schema} CASCADE")))
            .execute(db.pool())
            .await
            .expect("drop failing restriction schema");
    }

    /// Condition errors quote the message they evaluated. The trigger-filter
    /// WARN, the step-condition WARN and the run-failure ERROR log a fixed
    /// code; the run keeps the full diagnostic for its owner.
    #[tokio::test(flavor = "current_thread")]
    #[ignore = "requires Postgres"]
    async fn condition_errors_log_a_fixed_code_without_message_text() {
        #[derive(Clone, Default)]
        struct Captured(Arc<std::sync::Mutex<Vec<u8>>>);
        impl std::io::Write for Captured {
            fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
                self.0.lock().unwrap().extend_from_slice(bytes);
                Ok(bytes.len())
            }
            fn flush(&mut self) -> std::io::Result<()> {
                Ok(())
            }
        }

        let db = setup_db().await;
        let creator = nostr::Keys::generate().public_key().to_bytes().to_vec();
        let member = nostr::Keys::generate().public_key().to_bytes().to_vec();
        let (community, channel_id) = setup_channel(&db, &creator, &member).await;
        let mut workflow_ids = Vec::new();
        for (name, trigger, steps) in [
            (
                "filter-error",
                serde_json::json!({"on": "message_posted", "filter": "trigger_text > 5"}),
                serde_json::json!([{"id": "s1", "action": "send_message", "text": "hi"}]),
            ),
            (
                "step-error",
                serde_json::json!({"on": "message_posted"}),
                serde_json::json!([{
                    "id": "guarded", "if": "trigger_text > 5",
                    "action": "send_message", "text": "hi",
                }]),
            ),
        ] {
            let def = serde_json::json!({
                "name": name, "trigger": trigger, "steps": steps, "enabled": true,
            })
            .to_string();
            let hash = [workflow_ids.len() as u8; 32];
            let id = db
                .create_workflow(community, Some(channel_id), &member, name, &def, &hash)
                .await
                .expect("create workflow");
            workflow_ids.push(id);
        }
        let (filter_wf, step_wf) = (workflow_ids[0], workflow_ids[1]);

        let marker = format!("wf-marker-{}", Uuid::new_v4().simple());
        let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), &marker)
            .sign_with_keys(&nostr::Keys::generate())
            .expect("sign");
        let event = buzz_core::StoredEvent::new(event, Some(channel_id));

        let out = Captured::default();
        let writer = out.clone();
        let _guard = tracing::subscriber::set_default(
            tracing_subscriber::fmt()
                .json()
                .flatten_event(true)
                .with_max_level(tracing::Level::DEBUG)
                .with_writer(move || writer.clone())
                .finish(),
        );
        let engine = Arc::new(WorkflowEngine::new(db.clone(), WorkflowConfig::default()));
        engine.on_event(community, &event).await.expect("on_event");
        let run = tokio::time::timeout(std::time::Duration::from_secs(10), async {
            loop {
                let runs = db
                    .list_workflow_runs(community, step_wf, 10)
                    .await
                    .expect("list runs");
                if let Some(run) = runs.into_iter().find(|r| r.status == RunStatus::Failed) {
                    break run;
                }
                tokio::time::sleep(std::time::Duration::from_millis(20)).await;
            }
        })
        .await
        .expect("the step-condition run must fail");
        let filter_runs = db
            .list_workflow_runs(community, filter_wf, 10)
            .await
            .expect("list filter runs");
        assert!(
            filter_runs.is_empty(),
            "a failed filter must skip the workflow"
        );

        let raw = String::from_utf8(out.0.lock().unwrap().clone()).unwrap();
        let lines: Vec<serde_json::Value> = raw
            .lines()
            .map(|line| serde_json::from_str(line).unwrap())
            .collect();
        let line = |message: &str| {
            lines
                .iter()
                .find(|l| l["message"] == message)
                .unwrap_or_else(|| panic!("no {message:?} line in {raw}"))
        };
        assert_eq!(
            line("Trigger filter error — skipping workflow")["error_kind"],
            "condition_evaluation_failed"
        );
        assert_eq!(
            line("Condition error")["error_kind"],
            "condition_evaluation_failed"
        );
        assert_eq!(
            line("Workflow run failed")["error"],
            "condition_evaluation_failed"
        );
        assert!(!raw.contains(&marker), "marker leaked: {raw}");
        assert!(
            run.error_message
                .as_deref()
                .is_some_and(|m| m.contains(&marker)),
            "the stored run error keeps the diagnostic: {:?}",
            run.error_message
        );
    }
}