//! The triggers-framework dispatcher.
//!
//! Single tokio task that polls the outbox, claims due rows
//! (`FOR UPDATE SKIP LOCKED`), and routes each to the executor.
//! Shares the `ExecutionGate` with sync HTTP — they compete for the
//! same permit budget, matching design notes §2.
//!
//! Outcome handling per design notes §3 and §4:
//!   - reply_to.is_some() (sync HTTP): never retry. Deliver to inbox
//!     (or write `abandoned_executions` if the receiver dropped).
//!   - is_dead_letter_handler == true: never retry, never DL. Failure
//!     just annotates the original DL row with `resolution =
//!     'handler_failed'` and bumps a metric.
//!   - Otherwise on failure: if `attempt_count + 1 < max_attempts`,
//!     reschedule with backoff + jitter. Else, write a `dead_letters`
//!     row and delete from outbox.
//!
//! Depth-limit: `trigger_depth > max_trigger_depth` skips execution
//! entirely (log + metric) and deletes the row — does NOT dead-letter
//! (design notes §4: depth-exceeded means "you built a loop", and
//! dead-lettering would just re-fire the same loop).

use std::sync::Arc;
use std::time::Duration;

use chrono::{DateTime, Utc};
use picloud_executor_core::{
    build_execution_log, ExecError, ExecRequest, ExecResponse, InvocationType,
};
use picloud_orchestrator_core::{ExecutionGate, ExecutorClient};
use picloud_shared::{
    AppId, DeadLetterId, ExecResponseSummary, ExecutionId, ExecutionLogSink, ExecutionSource,
    HttpDispatchPayload, InboxDeliveryOutcome, InboxFailureKind, InboxResolver, InboxResult,
    QueueMessageId, RequestId, Script, ScriptId, ScriptOwner, ScriptSandbox, TriggerEvent,
    TriggerId,
};
use rand::Rng;
use uuid::Uuid;

use crate::abandoned_repo::{AbandonedRepo, NewAbandonedExecution};
use crate::dead_letter_repo::{DeadLetterRepo, NewDeadLetter};
use crate::outbox_repo::{NewOutboxRow, OutboxRepo, OutboxRow, OutboxSourceKind};
use crate::principal_resolver::PrincipalResolver;
use crate::queue_repo::{ClaimedMessage, QueueRepo};
use crate::repo::ScriptRepository;
use crate::trigger_config::{BackoffShape, TriggerConfig};
use crate::trigger_repo::{ActiveQueueConsumer, TriggerKind, TriggerRepo};

/// Bundle the dispatcher reads from. Each handle is `Arc<dyn …>` so
/// tests can substitute in-memory backings.
pub struct Dispatcher {
    pub outbox: Arc<dyn OutboxRepo>,
    pub triggers: Arc<dyn TriggerRepo>,
    pub scripts: Arc<dyn ScriptRepository>,
    pub dead_letters: Arc<dyn DeadLetterRepo>,
    pub abandoned: Arc<dyn AbandonedRepo>,
    pub principals: Arc<dyn PrincipalResolver>,
    pub executor: Arc<dyn ExecutorClient>,
    pub gate: Arc<ExecutionGate>,
    /// G1: records an `execution_logs` row for every trigger run so
    /// background workers (queue / cron / dead-letter / invoke) show up
    /// in `pic logs`, not just synchronous HTTP. Same sink the
    /// orchestrator's data plane writes through.
    pub log_sink: Arc<dyn ExecutionLogSink>,
    pub inbox: Arc<dyn InboxResolver>,
    /// v1.1.9. Reads `queue_messages` for the queue arm + the reclaim
    /// task. None in tests / harnesses that don't exercise queues.
    pub queue: Arc<dyn QueueRepo>,
    /// §11.6 D3. The group shared-queue store. A materialized consumer of a
    /// SHARED group queue template (`consumer.shared_group.is_some()`) claims
    /// from here instead of the per-app `queue`.
    pub group_queue: Arc<dyn crate::group_queue_repo::GroupQueueRepo>,
    pub config: TriggerConfig,
    /// Stable id for this dispatcher instance — written into
    /// `outbox.claimed_by` for forensics. In MVP this is the host's
    /// pid; cluster mode (v1.3+) uses node identity.
    pub instance_id: String,
}

/// How many outbox rows the dispatcher tries to claim per tick.
/// Bounded to keep the working set small even if there's a flood.
const CLAIM_BATCH: i64 = 8;

/// Inputs to `Dispatcher::fan_out_dead_letter`. Pulled out as a struct
/// so the outbox arm (`handle_failure`) and the queue arm
/// (`handle_queue_failure`) can both call it — the queue arm doesn't
/// have an `OutboxRow` to pass.
struct DeadLetterFanOutCtx {
    app_id: picloud_shared::AppId,
    /// The originating event being failed, verbatim. The DL event
    /// nests this under `original` so handler scripts see what
    /// triggered the chain.
    original: TriggerEvent,
    /// Trigger source the DL row was filed under (`"kv"`, `"queue"`,
    /// `"http"`, …). Drives the `source_filter` match on registered
    /// `dead_letter` triggers.
    source: String,
    dead_letter_id: DeadLetterId,
    attempts: u32,
    last_error: String,
    /// Original trigger id (for trigger_id_filter matching).
    trigger_id: Option<picloud_shared::TriggerId>,
    /// Original script id (for script_id_filter matching).
    script_id: Option<ScriptId>,
    first_attempt_at: DateTime<Utc>,
    last_attempt_at: DateTime<Utc>,
    trigger_depth: u32,
    root_execution_id: Option<ExecutionId>,
}

/// Polling cadence. Short enough that fan-out feels instant; long
/// enough that an idle dispatcher doesn't burn cycles.
/// F-Q-009: env-overridable via `PICLOUD_DISPATCHER_TICK_INTERVAL_MS`.
const DEFAULT_TICK_INTERVAL: Duration = Duration::from_millis(100);

/// Hard cap on the wall-clock budget passed to the executor for an
/// async-dispatched script. Sync HTTP gets a per-script timeout via
/// the orchestrator path; async rows don't have one, so we apply a
/// platform-wide ceiling here. Matches `LocalExecutorClient`'s own
/// 5-minute cap.
/// F-Q-009: env-overridable via `PICLOUD_DISPATCHER_ASYNC_EXEC_TIMEOUT_SEC`.
///
/// `pub` so `triggers_api::validate_queue_visibility_timeout` can use
/// the same default for its warn-threshold — otherwise a change here
/// would silently leave the visibility-timeout warner out of sync.
pub const DEFAULT_ASYNC_EXEC_TIMEOUT_SECS: u32 = 300;
const DEFAULT_ASYNC_EXEC_TIMEOUT: Duration =
    Duration::from_secs(DEFAULT_ASYNC_EXEC_TIMEOUT_SECS as u64);

/// Read `PICLOUD_DISPATCHER_TICK_INTERVAL_MS`; invalid values fall back
/// to the documented default with a tracing-warn.
fn tick_interval_from_env() -> Duration {
    if let Ok(v) = std::env::var("PICLOUD_DISPATCHER_TICK_INTERVAL_MS") {
        match v.trim().parse::<u64>() {
            Ok(n) if n > 0 => return Duration::from_millis(n),
            _ => tracing::warn!(
                value = %v,
                "ignoring invalid PICLOUD_DISPATCHER_TICK_INTERVAL_MS (want a positive integer)"
            ),
        }
    }
    DEFAULT_TICK_INTERVAL
}

/// Read `PICLOUD_DISPATCHER_ASYNC_EXEC_TIMEOUT_SEC`; invalid values
/// fall back to the documented default with a tracing-warn.
fn async_exec_timeout_from_env() -> Duration {
    if let Ok(v) = std::env::var("PICLOUD_DISPATCHER_ASYNC_EXEC_TIMEOUT_SEC") {
        match v.trim().parse::<u64>() {
            Ok(n) if n > 0 => return Duration::from_secs(n),
            _ => tracing::warn!(
                value = %v,
                "ignoring invalid PICLOUD_DISPATCHER_ASYNC_EXEC_TIMEOUT_SEC (want a positive integer)"
            ),
        }
    }
    DEFAULT_ASYNC_EXEC_TIMEOUT
}

/// F-P-007: per-tick concurrency cap for queue dispatches. Matches the
/// default execution-gate concurrency so the gate is what bounds real
/// parallelism, not the dispatcher's serial-await.
const QUEUE_DISPATCH_PARALLELISM: usize = 32;

/// Map an outbox row's source kind to the execution-log `source`. The
/// wire strings are identical, so this is total in practice; an unknown
/// kind would default to `Http`.
fn exec_source(row: &OutboxRow) -> ExecutionSource {
    ExecutionSource::from_wire(row.source_kind.as_str()).unwrap_or_default()
}

/// G1: the slice of an `ExecRequest` needed to write an execution-log
/// row, captured before the request is moved into the executor.
struct ExecLogContext {
    app_id: picloud_shared::AppId,
    script_id: ScriptId,
    request_id: RequestId,
    path: String,
    headers: std::collections::BTreeMap<String, String>,
    body: serde_json::Value,
    source: ExecutionSource,
}

impl ExecLogContext {
    fn from_request(req: &ExecRequest, source: ExecutionSource) -> Self {
        Self {
            app_id: req.app_id,
            script_id: req.script_id,
            request_id: req.request_id,
            path: req.path.clone(),
            headers: req.headers.clone(),
            body: req.body.clone(),
            source,
        }
    }
}

impl Dispatcher {
    /// Phase 4 isolation backstop: may `script` run under `app_id`'s context?
    /// True when the script is owned by `app_id` directly, or by a group on
    /// its ancestor chain (an inherited group script). App-owned is the
    /// in-memory fast path — only a group-owned script pays the chain query,
    /// and a query error fails closed (not invocable). This generalizes the
    /// pre-Phase-4 same-app guard (`script.app_id == row.app_id`) to the
    /// hierarchy without changing behavior for app-owned scripts.
    async fn script_invocable(&self, script: &Script, app_id: AppId) -> bool {
        if script.is_owned_by_app(app_id) {
            return true;
        }
        if script.group_id.is_none() {
            return false;
        }
        match self.scripts.is_invocable_by_app(script.id, app_id).await {
            Ok(ok) => ok,
            Err(e) => {
                tracing::error!(
                    error = %e, script_id = %script.id, %app_id,
                    "chain-membership check failed; treating script as not invocable"
                );
                false
            }
        }
    }

    /// Spawn the dispatcher loop as a detached `tokio::task`. Also
    /// spawns the v1.1.9 queue visibility-timeout reclaim task. Both
    /// run for the process lifetime; returned `JoinHandle`s are
    /// dropped on purpose.
    pub fn spawn(self) {
        // Reclaim task: independent cadence (default 30s) so it doesn't
        // contend with the per-100ms dispatcher tick.
        let reclaim_queue = self.queue.clone();
        let reclaim_group_queue = self.group_queue.clone();
        let reclaim_interval =
            Duration::from_millis(u64::from(self.config.queue_reclaim_interval_ms));
        tokio::spawn(async move {
            let mut ticker = tokio::time::interval(reclaim_interval);
            ticker.tick().await;
            loop {
                ticker.tick().await;
                match reclaim_queue.reclaim_visibility_timeouts().await {
                    Ok(0) => {}
                    Ok(n) => tracing::info!(reclaimed = n, "queue visibility-timeout reclaim"),
                    Err(e) => tracing::warn!(?e, "queue reclaim task errored"),
                }
                // §11.6 D3: the group shared-queue store has the same reclaim.
                match reclaim_group_queue.reclaim_visibility_timeouts().await {
                    Ok(0) => {}
                    Ok(n) => {
                        tracing::info!(reclaimed = n, "group-queue visibility-timeout reclaim");
                    }
                    Err(e) => tracing::warn!(?e, "group-queue reclaim task errored"),
                }
            }
        });

        tokio::spawn(async move {
            self.run().await;
        });
    }

    async fn run(self) {
        let mut ticker = tokio::time::interval(tick_interval_from_env());
        // Skip the immediate first fire so we don't race startup.
        ticker.tick().await;
        loop {
            ticker.tick().await;
            if let Err(err) = self.tick().await {
                tracing::warn!(?err, "dispatcher tick errored");
            }
        }
    }

    async fn tick(&self) -> Result<(), DispatcherError> {
        // 1) Outbox arm — KV/docs/pubsub/cron/email/files/dead-letter/http/invoke.
        let rows = self
            .outbox
            .claim_due(&self.instance_id, CLAIM_BATCH)
            .await
            .map_err(|e| DispatcherError::Outbox(e.to_string()))?;
        for row in rows {
            // Process serially within a tick — the outer ticker is the
            // pacing mechanism. Concurrent dispatchers are a cluster-
            // mode concern; v1.1.1 MVP has one.
            if let Err(err) = self.dispatch_one(row).await {
                tracing::warn!(?err, "dispatch one errored");
            }
        }

        // 2) Queue arm (v1.1.9). One claim per active consumer per tick;
        // bounded by the number of registered consumers so a flood in
        // one queue doesn't starve the rest.
        if let Err(err) = self.tick_queue_arm().await {
            tracing::warn!(?err, "queue arm tick errored");
        }
        Ok(())
    }

    /// v1.1.9. Per tick: list every `queue:receive` consumer and
    /// attempt one claim per `(app_id, queue_name)`. Each successful
    /// claim runs through the same executor path as outbox-backed
    /// trigger dispatches, so retry / dead-letter wiring is shared.
    ///
    /// F-P-007: bound concurrency with `for_each_concurrent` instead of
    /// serial-await. With N consumers and a 100ms tick the worst-case
    /// throughput was N × (claim + executor) / tick; concurrent dispatch
    /// lets one slow handler not block the others. The execution gate
    /// still caps real parallelism — this just removes the
    /// dispatcher-side serialization.
    async fn tick_queue_arm(&self) -> Result<(), DispatcherError> {
        use futures::stream::{self, StreamExt};
        let consumers = self
            .triggers
            .list_active_queue_consumers()
            .await
            .map_err(|e| DispatcherError::ResolveTrigger(e.to_string()))?;
        // Concurrency cap matches the execution gate's typical default;
        // overflow attempts return Err(Overloaded) and back off so a
        // queue burst doesn't hammer the executor.
        stream::iter(consumers)
            .for_each_concurrent(QUEUE_DISPATCH_PARALLELISM, |consumer| async move {
                if let Err(err) = self.dispatch_one_queue(&consumer).await {
                    tracing::warn!(?err, queue = %consumer.queue_name, "queue dispatch errored");
                }
            })
            .await;
        Ok(())
    }

    // §11.6 D3: route the four queue store operations to the per-app store or
    // the group shared store, based on whether this consumer was materialized
    // from a SHARED group queue template (`consumer.shared_group`). A group
    // claim is normalized to a `ClaimedMessage` under the CONSUMING app so the
    // rest of the queue arm (handler dispatch, event, logging) is unchanged.
    async fn q_claim(&self, c: &ActiveQueueConsumer) -> Result<Option<ClaimedMessage>, String> {
        match c.shared_group {
            Some(g) => Ok(self
                .group_queue
                .claim(g, &c.queue_name)
                .await
                .map_err(|e| e.to_string())?
                .map(|m| ClaimedMessage {
                    id: m.id,
                    app_id: c.app_id,
                    queue_name: m.collection,
                    payload: m.payload,
                    enqueued_at: m.enqueued_at,
                    attempt: m.attempt,
                    max_attempts: m.max_attempts,
                    claim_token: m.claim_token,
                })),
            None => self
                .queue
                .claim(c.app_id, &c.queue_name)
                .await
                .map_err(|e| e.to_string()),
        }
    }

    async fn q_ack(
        &self,
        c: &ActiveQueueConsumer,
        id: QueueMessageId,
        token: uuid::Uuid,
    ) -> Result<bool, String> {
        match c.shared_group {
            Some(_) => self
                .group_queue
                .ack(id, token)
                .await
                .map_err(|e| e.to_string()),
            None => self.queue.ack(id, token).await.map_err(|e| e.to_string()),
        }
    }

    async fn q_nack(
        &self,
        c: &ActiveQueueConsumer,
        id: QueueMessageId,
        token: uuid::Uuid,
        delay: chrono::Duration,
    ) -> Result<bool, String> {
        match c.shared_group {
            Some(_) => self
                .group_queue
                .nack(id, token, delay)
                .await
                .map_err(|e| e.to_string()),
            None => self
                .queue
                .nack(id, token, delay)
                .await
                .map_err(|e| e.to_string()),
        }
    }

    /// Terminal disposition of a message that can't be processed (script
    /// missing / cross-app / exhausted). Per-app → dead-letter into `dead_letters`
    /// (+ the caller fans out `dead_letter` triggers). Group shared queue →
    /// dead-letter into `group_dead_letters` (Track A M2) and return `None` so the
    /// per-app fan-out is skipped (a competing-consumer message has no single app
    /// to fire per-app handlers under). Returns the dead-letter id only for the
    /// per-app path (drives fan-out).
    async fn q_terminal(
        &self,
        c: &ActiveQueueConsumer,
        claimed: &ClaimedMessage,
        trigger_id: Option<TriggerId>,
        script_id: Option<ScriptId>,
        reason: &str,
    ) -> Option<DeadLetterId> {
        match c.shared_group {
            Some(group_id) => {
                // §11.6 D3: persist the exhausted message to the group dead-letter
                // store instead of dropping it. We return None (not the dl id) so
                // the per-app `fan_out_dead_letter` below is SKIPPED — firing the
                // consuming app's per-app dead_letter handlers on a shared-queue
                // message (competing consumers → nondeterministic app) would be
                // wrong. Fan-out to a *shared* dead_letter trigger is deferred; the
                // row is operator-visible via the group dead-letters admin API.
                match self
                    .group_queue
                    .dead_letter(
                        claimed.id,
                        claimed.claim_token,
                        group_id,
                        &claimed.queue_name,
                        trigger_id,
                        script_id,
                        claimed.attempt,
                        claimed.enqueued_at,
                        reason,
                    )
                    .await
                {
                    Ok(dl_id) => tracing::warn!(
                        reason,
                        queue = %claimed.queue_name,
                        dead_letter_id = %dl_id.into_inner(),
                        "shared-queue message dead-lettered"
                    ),
                    Err(e) => tracing::error!(?e, "shared-queue dead-letter write failed"),
                }
                None
            }
            None => match self
                .queue
                .dead_letter(
                    claimed.id,
                    claimed.claim_token,
                    claimed.app_id,
                    &claimed.queue_name,
                    trigger_id,
                    script_id,
                    claimed.attempt,
                    claimed.enqueued_at,
                    reason,
                )
                .await
            {
                Ok(dl_id) => Some(dl_id),
                Err(e) => {
                    tracing::error!(?e, "queue dead-letter write failed");
                    None
                }
            },
        }
    }

    #[allow(clippy::too_many_lines)]
    async fn dispatch_one_queue(
        &self,
        consumer: &ActiveQueueConsumer,
    ) -> Result<(), DispatcherError> {
        // Atomic claim — None → nothing pending right now for this queue.
        let Some(claimed) = self
            .q_claim(consumer)
            .await
            .map_err(DispatcherError::Outbox)?
        else {
            return Ok(());
        };

        // Gate admission. If saturated, release the claim by nack-
        // immediate (which lets the next tick re-claim). Mirrors the
        // outbox arm.
        let Ok(permit) = self.gate.try_acquire() else {
            let _ = self
                .q_nack(
                    consumer,
                    claimed.id,
                    claimed.claim_token,
                    chrono::Duration::milliseconds(100),
                )
                .await;
            return Ok(());
        };

        // Build the TriggerEvent and ExecRequest. Mirrors build_exec_request
        // but uses the consumer + claimed-message data instead of an
        // outbox row.
        let event = TriggerEvent::Queue {
            queue_name: claimed.queue_name.clone(),
            message: claimed.payload.clone(),
            enqueued_at: claimed.enqueued_at,
            attempt: claimed.attempt,
            message_id: claimed.id.to_string(),
        };

        let script = match self.scripts.get(consumer.script_id).await {
            Ok(Some(s)) => s,
            Ok(None) => {
                tracing::warn!(script_id = %consumer.script_id, "queue trigger script missing; dead-lettering");
                let _ = self
                    .q_terminal(
                        consumer,
                        &claimed,
                        Some(consumer.trigger_id),
                        Some(consumer.script_id),
                        "queue trigger script not found",
                    )
                    .await;
                drop(permit);
                return Ok(());
            }
            Err(e) => return Err(DispatcherError::ResolveTrigger(e.to_string())),
        };

        // Audit 2026-06-11 H-F1 — same-app guard. validate_trigger_target
        // rejects cross-app rows at write time; this is the runtime
        // backstop for hand-edited triggers, partial backup restores, or
        // a script re-pointed across apps without re-validating the
        // existing consumers. Dead-letter the message so the misfire is
        // observable rather than executing one app's script under
        // another app's SdkCallCx.
        // Phase 4: the target must be invocable in the claimed message's app
        // context — owned by that app or an inherited group script on its
        // chain. App-owned scripts take the in-memory fast path; only a group
        // script pays the chain query, and a non-member fails closed
        // (dead-lettered, not silently run under another app's SdkCallCx).
        if !self.script_invocable(&script, claimed.app_id).await {
            tracing::error!(
                script_app = ?script.app_id,
                claim_app = %claimed.app_id,
                script_id = %consumer.script_id,
                "queue consumer script belongs to a different app; dead-lettering"
            );
            let _ = self
                .q_terminal(
                    consumer,
                    &claimed,
                    Some(consumer.trigger_id),
                    Some(consumer.script_id),
                    "queue consumer target belongs to a different app",
                )
                .await;
            drop(permit);
            return Ok(());
        }

        // Fire-time `enabled` re-check (§4.3). The queue arm does not flow
        // through `dispatch_one`'s unified `!active` gate; its only other
        // `enabled` guard is the per-tick `list_active_queue_consumers`
        // snapshot, which is stale for messages already claimed in this
        // tick. A script disabled after the list query but before this
        // claimed message executes must NOT run (`script` here is a fresh
        // read from line ~331, so `enabled` is current). Release the claim
        // (nack) rather than ack/dead-letter: the message stays queued and
        // is processed when the script is re-enabled, and the next tick
        // won't re-claim it because the list filters `s.enabled`. Without
        // this nack the message would sit claimed indefinitely —
        // `reclaim_visibility_timeouts` only reclaims for enabled triggers.
        if !script.enabled {
            tracing::info!(
                script_id = %consumer.script_id,
                trigger_id = %consumer.trigger_id,
                "queue consumer script disabled at fire time; releasing claim"
            );
            if let Err(e) = self
                .q_nack(
                    consumer,
                    claimed.id,
                    claimed.claim_token,
                    chrono::Duration::seconds(1),
                )
                .await
            {
                tracing::warn!(?e, "queue nack on disabled consumer failed");
            }
            drop(permit);
            return Ok(());
        }

        let principal = self
            .principals
            .resolve(consumer.registered_by_principal)
            .await
            .map_err(|e| DispatcherError::ResolveTrigger(e.to_string()))?;

        let execution_id = ExecutionId::new();
        let exec_req = ExecRequest {
            execution_id,
            request_id: RequestId::new(),
            script_id: consumer.script_id,
            script_name: script.name.clone(),
            invocation_type: InvocationType::Function,
            path: "/trigger/queue".into(),
            // Non-HTTP trigger invocation — no request method.
            method: String::new(),
            headers: std::collections::BTreeMap::new(),
            body: serde_json::Value::Null,
            params: std::collections::BTreeMap::new(),
            query: std::collections::BTreeMap::new(),
            rest: String::new(),
            sandbox_overrides: script.sandbox,
            app_id: claimed.app_id,
            script_owner: script.owner(),
            principal: Some(principal),
            trigger_depth: 0,
            root_execution_id: execution_id,
            is_dead_letter_handler: false,
            event: Some(event),
        };
        let identity = picloud_orchestrator_core::ScriptIdentity {
            script_id: consumer.script_id,
            updated_at: script.updated_at,
        };
        // G1: queue consumers dispatch outside the outbox, so they need
        // their own log write. Source is always `queue`; no `reply_to`
        // here, so there's no double-logging concern.
        let log_cx = ExecLogContext::from_request(&exec_req, ExecutionSource::Queue);
        let started = Utc::now();
        let outcome = self
            .executor
            .execute_with_identity(
                identity,
                &script.source,
                exec_req,
                async_exec_timeout_from_env(),
            )
            .await;
        let finished = Utc::now();
        drop(permit);

        self.record_execution_log(&log_cx, &outcome, started, finished)
            .await;

        // Best-effort touch on last_fired_at (a failure here doesn't
        // change ack/nack behavior).
        if let Err(e) = self
            .triggers
            .touch_queue_trigger_last_fired_at(consumer.trigger_id, Utc::now())
            .await
        {
            tracing::debug!(?e, "touch last_fired_at failed");
        }

        match outcome {
            Ok(_) => {
                // Auto-ack on success.
                if let Err(e) = self.q_ack(consumer, claimed.id, claimed.claim_token).await {
                    tracing::warn!(?e, "queue ack failed");
                }
            }
            Err(err) => {
                self.handle_queue_failure(consumer, &claimed, err).await;
            }
        }
        Ok(())
    }

    async fn handle_queue_failure(
        &self,
        consumer: &ActiveQueueConsumer,
        claimed: &ClaimedMessage,
        err: ExecError,
    ) {
        if claimed.attempt < claimed.max_attempts {
            // Retry: clear claim + set deliver_after.
            let delay_ms = compute_backoff(
                claimed.attempt,
                consumer.retry_backoff,
                consumer.retry_base_ms,
                self.config.retry_jitter_pct,
            );
            let delay = chrono::Duration::milliseconds(i64::from(delay_ms));
            if let Err(e) = self
                .q_nack(consumer, claimed.id, claimed.claim_token, delay)
                .await
            {
                tracing::warn!(?e, "queue nack failed");
            }
            return;
        }
        // Exhausted. Dead-letter inline (the helper deletes the queue row +
        // writes a dead_letters row in one transaction). The outbox arm
        // calls `fan_out_dead_letter` after writing its DL row; the queue
        // arm previously omitted this, so registered `dead_letter`
        // handlers filtered on `source = "queue"` sat idle. Fan out the
        // same way here.
        let now = Utc::now();
        let last_error = err.to_string();
        // Per-app → dead-letter (+ fan out below). Shared group queue →
        // dead-lettered into `group_dead_letters` inside q_terminal (Track A M2),
        // which returns None so the per-app fan-out is skipped.
        let dl_id = self
            .q_terminal(
                consumer,
                claimed,
                Some(consumer.trigger_id),
                Some(consumer.script_id),
                &last_error,
            )
            .await;
        if let Some(dead_letter_id) = dl_id {
            let original = TriggerEvent::Queue {
                queue_name: claimed.queue_name.clone(),
                message: claimed.payload.clone(),
                enqueued_at: claimed.enqueued_at,
                attempt: claimed.attempt,
                message_id: claimed.id.to_string(),
            };
            // dead_letter triggers filter on the originating event's
            // wire source ("queue", "kv", "http", …) — same vocabulary
            // as TriggerEvent::source().
            self.fan_out_dead_letter(DeadLetterFanOutCtx {
                app_id: claimed.app_id,
                original,
                source: "queue".to_string(),
                dead_letter_id,
                attempts: claimed.attempt,
                last_error,
                trigger_id: Some(consumer.trigger_id),
                script_id: Some(consumer.script_id),
                first_attempt_at: claimed.enqueued_at,
                last_attempt_at: now,
                // Queue messages always root a depth-1 chain (the queue
                // itself is depth 0; any subsequent DL handler ticks up).
                trigger_depth: 1,
                root_execution_id: None,
            })
            .await;
        }
        // TODO(metrics): bump picloud_queue_dead_letters_total{app_id, queue_name}.
    }

    #[allow(clippy::too_many_lines)]
    async fn dispatch_one(&self, row: OutboxRow) -> Result<(), DispatcherError> {
        // Depth-limit check — design notes §4: loops aren't DL'd.
        if row.trigger_depth > self.config.max_trigger_depth {
            tracing::warn!(
                outbox_id = %row.id,
                app_id = %row.app_id,
                trigger_depth = row.trigger_depth,
                "trigger depth exceeded; dropping row"
            );
            // TODO(metrics): bump `picloud_trigger_depth_exceeded{app_id,trigger_id}`.
            self.outbox
                .delete(row.id)
                .await
                .map_err(|e| DispatcherError::Outbox(e.to_string()))?;
            return Ok(());
        }

        // Gate admission — non-blocking. If the gate is saturated,
        // release the claim by rescheduling so another tick can pick
        // it up. The row stays "due" essentially immediately.
        let Ok(permit) = self.gate.try_acquire() else {
            let next = Utc::now() + chrono::Duration::milliseconds(100);
            self.outbox
                .reschedule(row.id, row.attempt_count, next)
                .await
                .map_err(|e| DispatcherError::Outbox(e.to_string()))?;
            return Ok(());
        };

        // Resolve the trigger config (KV / DL) or pull the HTTP
        // payload directly off the outbox row.
        let (resolved, exec_req) = match row.source_kind {
            OutboxSourceKind::Http => match self.build_http_request(&row).await {
                Ok(pair) => pair,
                Err(err) => {
                    tracing::warn!(outbox_id = %row.id, ?err, "http exec build failed; dropping");
                    self.outbox
                        .delete(row.id)
                        .await
                        .map_err(|e| DispatcherError::Outbox(e.to_string()))?;
                    drop(permit);
                    return Ok(());
                }
            },
            OutboxSourceKind::Invoke => match self.build_invoke_request(&row).await {
                Ok(pair) => pair,
                Err(err) => {
                    tracing::warn!(outbox_id = %row.id, ?err, "invoke exec build failed; dropping");
                    self.outbox
                        .delete(row.id)
                        .await
                        .map_err(|e| DispatcherError::Outbox(e.to_string()))?;
                    drop(permit);
                    return Ok(());
                }
            },
            OutboxSourceKind::Kv
            | OutboxSourceKind::Docs
            | OutboxSourceKind::DeadLetter
            | OutboxSourceKind::Cron
            | OutboxSourceKind::Files
            | OutboxSourceKind::Pubsub
            | OutboxSourceKind::Email => {
                let resolved = self.resolve_trigger(&row).await?;
                let req = match self.build_exec_request(&row, &resolved).await {
                    Ok(req) => req,
                    Err(err) => {
                        tracing::warn!(outbox_id = %row.id, ?err, "exec request build failed; dropping row");
                        self.outbox
                            .delete(row.id)
                            .await
                            .map_err(|e| DispatcherError::Outbox(e.to_string()))?;
                        drop(permit);
                        return Ok(());
                    }
                };
                (resolved, req)
            }
        };

        // §4.3 fire-time re-check, for EVERY outbox source (trigger, async
        // HTTP/202, and invoke): if the target script (or, for triggers, the
        // trigger) was disabled after this row was enqueued, drop it rather
        // than fire a stale event. The match-time `enabled` check can't see a
        // later toggle, so this is the gate that makes a disabled script
        // genuinely non-invocable on the async paths too.
        if !resolved.active {
            tracing::debug!(
                outbox_id = %row.id,
                app_id = %row.app_id,
                "target script/trigger disabled since enqueue; dropping outbox row"
            );
            self.outbox
                .delete(row.id)
                .await
                .map_err(|e| DispatcherError::Outbox(e.to_string()))?;
            drop(permit);
            return Ok(());
        }

        // The gate permit auto-releases when this scope ends or when
        // the executor finishes. We hand control to the executor and
        // wait synchronously here — sync HTTP and dispatcher share the
        // semaphore so this is intentional.
        let source = resolved.script_source.clone();
        let identity = picloud_orchestrator_core::ScriptIdentity {
            script_id: resolved.script_id,
            updated_at: resolved.script_updated_at,
        };
        // G1: capture the request context before `exec_req` is consumed so
        // we can write an execution-log row for this run (see below).
        let log_cx = ExecLogContext::from_request(&exec_req, exec_source(&row));
        let started = Utc::now();
        let outcome = self
            .executor
            .execute_with_identity(identity, &source, exec_req, async_exec_timeout_from_env())
            .await;
        let finished = Utc::now();
        drop(permit);

        // G1: persist an execution-log row so this run shows up in
        // `pic logs`. Skip rows with a synchronous receiver (`reply_to`
        // is set) — those are sync HTTP routes that the orchestrator's
        // inbox path already logs, and logging here too would duplicate
        // them. Trigger rows and fire-and-forget async HTTP (202) have no
        // receiver, so the dispatcher is the only place that can log them.
        if row.reply_to.is_none() {
            self.record_execution_log(&log_cx, &outcome, started, finished)
                .await;
        }

        match outcome {
            Ok(resp) => self.handle_success(&row, &resolved, resp).await,
            Err(err) => self.handle_failure(&row, &resolved, err).await,
        }
    }

    /// G1: best-effort persistence of an execution-log row for a
    /// dispatcher-run script. A sink failure is logged but never changes
    /// ack/nack/retry behavior — the audit trail is not on the hot path
    /// of message-delivery correctness.
    async fn record_execution_log(
        &self,
        cx: &ExecLogContext,
        outcome: &Result<ExecResponse, ExecError>,
        started: DateTime<Utc>,
        finished: DateTime<Utc>,
    ) {
        let log = build_execution_log(
            cx.app_id,
            cx.script_id,
            cx.request_id,
            cx.path.clone(),
            cx.headers.clone(),
            cx.body.clone(),
            cx.source,
            outcome,
            started,
            finished,
        );
        if let Err(e) = self.log_sink.record(log).await {
            tracing::warn!(
                error = %e,
                script_id = %cx.script_id,
                source = cx.source.as_str(),
                "failed to persist trigger execution log"
            );
        }
    }

    async fn resolve_trigger(&self, row: &OutboxRow) -> Result<ResolvedTrigger, DispatcherError> {
        // For KV and DL kinds, the outbox carries `trigger_id`. Use it
        // to look up the trigger row, then resolve the script.
        let Some(trigger_id) = row.trigger_id else {
            return Err(DispatcherError::ResolveTrigger(
                "outbox row missing trigger_id".into(),
            ));
        };
        let trigger = self
            .triggers
            .get(trigger_id)
            .await
            .map_err(|e| DispatcherError::ResolveTrigger(e.to_string()))?
            .ok_or_else(|| {
                DispatcherError::ResolveTrigger(format!("trigger {trigger_id} not found"))
            })?;

        let script = self
            .scripts
            .get(trigger.script_id)
            .await
            .map_err(|e| DispatcherError::ResolveTrigger(e.to_string()))?
            .ok_or_else(|| {
                DispatcherError::ResolveTrigger(format!("script {} not found", trigger.script_id))
            })?;

        // Audit 2026-06-11 H-F1 sibling — same-app guard mirroring
        // build_http_request / build_invoke_request / dispatch_one_queue.
        // `build_exec_request` stamps `ExecRequest.app_id = row.app_id`
        // while sourcing the body from `trigger.script_id`; without this
        // check a hand-edited outbox/trigger row (or a partial restore, or
        // a script re-pointed across apps) could run one app's script under
        // another app's `SdkCallCx.app_id` — the cross-app isolation
        // boundary. Not reachable via the trigger-create or `apply` paths
        // (both resolve the script within the app's own scope), so this is
        // the runtime backstop the other arms already carry.
        // Phase 4: invocable in the outbox row's app context (app-owned or an
        // inherited group script on the chain); a non-member fails closed.
        if !self.script_invocable(&script, row.app_id).await {
            return Err(DispatcherError::ResolveTrigger(
                "trigger outbox target belongs to a different app".into(),
            ));
        }

        Ok(ResolvedTrigger {
            trigger_kind: trigger.kind,
            is_dead_letter_handler: matches!(trigger.kind, TriggerKind::DeadLetter),
            active: trigger.enabled && script.enabled,
            script_id: script.id,
            script_owner: script.owner(),
            script_source: script.source,
            script_name: script.name,
            script_updated_at: script.updated_at,
            sandbox_overrides: script.sandbox,
            registered_by_principal: trigger.registered_by_principal,
            retry_max_attempts: trigger.retry_max_attempts,
            retry_backoff: trigger.retry_backoff,
            retry_base_ms: trigger.retry_base_ms,
        })
    }

    async fn build_exec_request(
        &self,
        row: &OutboxRow,
        resolved: &ResolvedTrigger,
    ) -> Result<ExecRequest, DispatcherError> {
        let trigger_event: TriggerEvent = serde_json::from_value(row.payload.clone())
            .map_err(|e| DispatcherError::ResolveTrigger(format!("decode payload: {e}")))?;

        let principal = self
            .principals
            .resolve(resolved.registered_by_principal)
            .await
            .map_err(|e| DispatcherError::ResolveTrigger(e.to_string()))?;

        let execution_id = ExecutionId::new();
        Ok(ExecRequest {
            execution_id,
            request_id: RequestId::new(),
            script_id: resolved.script_id,
            script_name: resolved.script_name.clone(),
            invocation_type: InvocationType::Function,
            path: format!("/trigger/{}", trigger_event.source()),
            // Non-HTTP trigger invocation — no request method.
            method: String::new(),
            headers: std::collections::BTreeMap::new(),
            body: serde_json::Value::Null,
            params: std::collections::BTreeMap::new(),
            query: std::collections::BTreeMap::new(),
            rest: String::new(),
            sandbox_overrides: resolved.sandbox_overrides,
            app_id: row.app_id,
            script_owner: resolved.script_owner,
            principal: Some(principal),
            trigger_depth: row.trigger_depth,
            root_execution_id: row.root_execution_id.unwrap_or(execution_id),
            is_dead_letter_handler: resolved.is_dead_letter_handler,
            event: Some(trigger_event),
        })
    }

    /// Build an `(ResolvedTrigger, ExecRequest)` for an HTTP outbox
    /// row. HTTP rows don't have a backing `triggers` row (the
    /// `trigger_id` references `routes.id` instead). We pull the
    /// script id off the outbox row, the request shape off the
    /// payload, and synthesize a `ResolvedTrigger` with retry
    /// settings irrelevant for HTTP (sync HTTP is never retried;
    /// async HTTP uses default policy from `TriggerConfig`).
    async fn build_http_request(
        &self,
        row: &OutboxRow,
    ) -> Result<(ResolvedTrigger, ExecRequest), DispatcherError> {
        let Some(script_id) = row.script_id else {
            return Err(DispatcherError::ResolveTrigger(
                "HTTP outbox row missing script_id".into(),
            ));
        };
        let script = self
            .scripts
            .get(script_id)
            .await
            .map_err(|e| DispatcherError::ResolveTrigger(e.to_string()))?
            .ok_or_else(|| {
                DispatcherError::ResolveTrigger(format!("script {script_id} not found"))
            })?;

        // Audit 2026-06-11 H-F1 sibling — same-app guard mirroring
        // build_invoke_request and dispatch_one_queue. A hand-edited
        // outbox row could otherwise execute one app's script under
        // another app's SdkCallCx.
        // Phase 4: invocable in this app's context (see queue/trigger arms).
        if !self.script_invocable(&script, row.app_id).await {
            return Err(DispatcherError::ResolveTrigger(
                "http outbox target belongs to a different app".into(),
            ));
        }

        let payload: HttpDispatchPayload = serde_json::from_value(row.payload.clone())
            .map_err(|e| DispatcherError::ResolveTrigger(format!("decode http payload: {e}")))?;

        let execution_id = ExecutionId::new();
        let req = ExecRequest {
            execution_id,
            request_id: RequestId::new(),
            script_id,
            script_name: payload.script_name.clone(),
            invocation_type: InvocationType::Http,
            path: payload.path.clone(),
            method: payload.method.clone(),
            headers: payload.headers,
            body: payload.body,
            params: payload.params,
            query: payload.query,
            rest: payload.rest,
            sandbox_overrides: script.sandbox,
            app_id: row.app_id,
            script_owner: script.owner(),
            // HTTP outbox rows don't run as the trigger registrant —
            // they run with no principal (public ingress) or the
            // attached one (origin_principal forensic field is not
            // promoted to execution principal in this MVP).
            principal: None,
            trigger_depth: row.trigger_depth,
            root_execution_id: row.root_execution_id.unwrap_or(execution_id),
            is_dead_letter_handler: false,
            event: None,
        };

        let resolved = ResolvedTrigger {
            trigger_kind: TriggerKind::Kv, // placeholder; HTTP doesn't have a kind
            is_dead_letter_handler: false,
            // §4.3: an async-HTTP (202) row whose script was disabled after
            // enqueue is dropped at fire time by the post-match active check.
            active: script.enabled,
            script_id,
            script_owner: script.owner(),
            script_source: script.source,
            script_name: payload.script_name,
            script_updated_at: script.updated_at,
            sandbox_overrides: script.sandbox,
            // HTTP outbox rows don't carry a registered_by_principal
            // — use a sentinel zero UUID since this field isn't used
            // downstream for HTTP (no retries, no inbox principal).
            registered_by_principal: picloud_shared::AdminUserId::from(uuid::Uuid::nil()),
            // Async HTTP uses the platform default retry policy from
            // TriggerConfig. Sync HTTP (reply_to.is_some) never retries
            // regardless.
            retry_max_attempts: self.config.retry_max_attempts,
            retry_backoff: self.config.retry_backoff,
            retry_base_ms: self.config.retry_base_ms,
        };
        Ok((resolved, req))
    }

    /// v1.1.9. Build an `(ResolvedTrigger, ExecRequest)` from an
    /// invoke_async outbox row. The payload was written by
    /// `InvokeService::enqueue_async`. No retry policy — invoke_async
    /// runs once; on failure the row is deleted and a dead_letters row
    /// is written (so the misfire is observable).
    async fn build_invoke_request(
        &self,
        row: &OutboxRow,
    ) -> Result<(ResolvedTrigger, ExecRequest), DispatcherError> {
        // Extract the args + script_id from the payload (InvokeService
        // populates these). Defensive: dead_letter the row if any
        // required field is missing.
        let payload = &row.payload;
        let Some(script_id_str) = payload.get("script_id").and_then(|v| v.as_str()) else {
            return Err(DispatcherError::ResolveTrigger(
                "invoke row missing script_id".into(),
            ));
        };
        let script_id = picloud_shared::ScriptId::from(
            uuid::Uuid::parse_str(script_id_str)
                .map_err(|e| DispatcherError::ResolveTrigger(format!("script_id: {e}")))?,
        );
        let script = self
            .scripts
            .get(script_id)
            .await
            .map_err(|e| DispatcherError::ResolveTrigger(e.to_string()))?
            .ok_or_else(|| {
                DispatcherError::ResolveTrigger(format!("script {script_id} not found"))
            })?;
        // Same-app guard — the script could have been re-assigned or
        // the row could have been hand-edited. Reject mismatches.
        // Phase 4: invocable in this app's context (app-owned or inherited).
        if !self.script_invocable(&script, row.app_id).await {
            return Err(DispatcherError::ResolveTrigger(
                "invoke target belongs to a different app".into(),
            ));
        }

        let args = payload
            .get("args")
            .cloned()
            .unwrap_or(serde_json::Value::Null);
        let trigger_depth = payload
            .get("trigger_depth")
            .and_then(serde_json::Value::as_u64)
            .map_or(row.trigger_depth, |n| u32::try_from(n).unwrap_or(u32::MAX));

        let execution_id = ExecutionId::new();
        let req = ExecRequest {
            execution_id,
            request_id: RequestId::new(),
            script_id: script.id,
            script_name: script.name.clone(),
            invocation_type: InvocationType::Function,
            path: "/invoke_async".into(),
            method: String::new(),
            headers: std::collections::BTreeMap::new(),
            body: args,
            params: std::collections::BTreeMap::new(),
            query: std::collections::BTreeMap::new(),
            rest: String::new(),
            sandbox_overrides: script.sandbox,
            app_id: row.app_id,
            script_owner: script.owner(),
            // invoke_async runs with no principal — same as HTTP outbox.
            principal: None,
            trigger_depth,
            root_execution_id: row.root_execution_id.unwrap_or(execution_id),
            is_dead_letter_handler: false,
            event: None,
        };

        // Build a synthetic ResolvedTrigger so the dispatch_one outer
        // loop can carry on through the same code path. Retry knobs
        // bound to 1 attempt: invoke_async runs ONCE (callers wrap in
        // retry::with for retries).
        let resolved = ResolvedTrigger {
            trigger_kind: TriggerKind::Cron, // placeholder; not used downstream
            is_dead_letter_handler: false,
            // §4.3: a queued invoke() whose target script was disabled after
            // enqueue is dropped at fire time by the post-match active check.
            active: script.enabled,
            script_id: script.id,
            script_owner: script.owner(),
            script_source: script.source,
            script_name: script.name,
            script_updated_at: script.updated_at,
            sandbox_overrides: script.sandbox,
            registered_by_principal: picloud_shared::AdminUserId::from(uuid::Uuid::nil()),
            retry_max_attempts: 1,
            retry_backoff: BackoffShape::Constant,
            retry_base_ms: 0,
        };
        Ok((resolved, req))
    }

    async fn handle_success(
        &self,
        row: &OutboxRow,
        _resolved: &ResolvedTrigger,
        resp: ExecResponse,
    ) -> Result<(), DispatcherError> {
        if let Some(inbox_id) = row.reply_to {
            self.deliver_inbox(row, inbox_id, InboxResult::Success(summarize(&resp)))
                .await;
        }
        self.outbox
            .delete(row.id)
            .await
            .map_err(|e| DispatcherError::Outbox(e.to_string()))?;
        Ok(())
    }

    #[allow(clippy::too_many_lines)]
    async fn handle_failure(
        &self,
        row: &OutboxRow,
        resolved: &ResolvedTrigger,
        err: ExecError,
    ) -> Result<(), DispatcherError> {
        // Sync HTTP: always single-attempt. Always deliver outcome
        // (success-or-failure) to the inbox. Never retry, never DL.
        if let Some(inbox_id) = row.reply_to {
            let (kind, message) = classify_exec_error(&err);
            self.deliver_inbox(
                row,
                inbox_id,
                InboxResult::Failure {
                    kind,
                    message: message.clone(),
                },
            )
            .await;
            self.outbox
                .delete(row.id)
                .await
                .map_err(|e| DispatcherError::Outbox(e.to_string()))?;
            return Ok(());
        }

        // Dead-letter handler: never retry, never DL. Failure
        // annotates the original DL row + bumps a metric.
        if resolved.is_dead_letter_handler {
            tracing::error!(
                outbox_id = %row.id,
                app_id = %row.app_id,
                ?err,
                "dead-letter handler failed; not retrying"
            );
            // TODO(metrics): bump `picloud_dead_letter_handler_failures{app_id}`.
            // Annotate the original DL row (id is `row.payload.dead_letter.id`
            // when the payload is a DeadLetter TriggerEvent). Best-effort:
            // if the payload doesn't decode, just log and move on.
            if let Ok(TriggerEvent::DeadLetter { dead_letter_id, .. }) =
                serde_json::from_value::<TriggerEvent>(row.payload.clone())
            {
                if let Err(e) = self
                    .dead_letters
                    .resolve(dead_letter_id, "handler_failed")
                    .await
                {
                    tracing::warn!(?e, "could not annotate DL row as handler_failed");
                }
            }
            self.outbox
                .delete(row.id)
                .await
                .map_err(|e| DispatcherError::Outbox(e.to_string()))?;
            return Ok(());
        }

        // Async event: retry per policy, then dead-letter.
        let attempt = row.attempt_count + 1;
        if attempt < resolved.retry_max_attempts {
            let delay = compute_backoff(
                attempt,
                resolved.retry_backoff,
                resolved.retry_base_ms,
                self.config.retry_jitter_pct,
            );
            let next = Utc::now() + chrono::Duration::milliseconds(i64::from(delay));
            tracing::info!(
                outbox_id = %row.id,
                attempt,
                max_attempts = resolved.retry_max_attempts,
                retry_in_ms = delay,
                "rescheduling outbox row"
            );
            self.outbox
                .reschedule(row.id, attempt, next)
                .await
                .map_err(|e| DispatcherError::Outbox(e.to_string()))?;
            return Ok(());
        }

        // Exhausted retries → dead-letter.
        let (op, source) = describe_event(row.source_kind, &row.payload);
        let now = Utc::now();
        let dl_id = match self
            .dead_letters
            .insert(NewDeadLetter {
                app_id: row.app_id,
                original_event_id: row.id,
                source: source.clone(),
                op,
                trigger_id: row.trigger_id,
                script_id: Some(resolved.script_id),
                payload: row.payload.clone(),
                attempt_count: attempt,
                first_attempt_at: row.created_at,
                last_attempt_at: now,
                last_error: err.to_string(),
            })
            .await
        {
            Ok(id) => Some(id),
            Err(e) => {
                tracing::error!(?e, "failed to write dead-letter row");
                None
            }
        };

        // v1.1.7 fix: fan the dead-letter out to matching handler triggers.
        // This was missing since v1.1.1 — the row was written but
        // `list_matching_dead_letter` had no production caller, so
        // registered dead_letter handlers never fired. The recursion-stop
        // (a dead-letter handler's own failure is not re-dead-lettered)
        // is upheld by the `is_dead_letter_handler` short-circuit at the
        // top of this function, so this fan-out is only reached for
        // non-handler executions.
        if let Some(dl_id) = dl_id {
            // The DL event nests the original verbatim; if the payload
            // can't be decoded back into a TriggerEvent we can't build
            // the nested `original`, so skip the fan-out (the DL row is
            // still written).
            match serde_json::from_value::<TriggerEvent>(row.payload.clone()) {
                Ok(original) => {
                    self.fan_out_dead_letter(DeadLetterFanOutCtx {
                        app_id: row.app_id,
                        original,
                        source,
                        dead_letter_id: dl_id,
                        attempts: attempt,
                        last_error: err.to_string(),
                        trigger_id: row.trigger_id,
                        script_id: Some(resolved.script_id),
                        first_attempt_at: row.created_at,
                        last_attempt_at: now,
                        trigger_depth: row.trigger_depth,
                        root_execution_id: row.root_execution_id,
                    })
                    .await;
                }
                Err(_) => {
                    tracing::warn!(
                        outbox_id = %row.id,
                        "dead-letter payload is not a TriggerEvent; skipping handler fan-out"
                    );
                }
            }
        }

        self.outbox
            .delete(row.id)
            .await
            .map_err(|e| DispatcherError::Outbox(e.to_string()))?;
        Ok(())
    }

    /// Enqueue one outbox row per matching `dead_letter` trigger so its
    /// handler script runs with the dead-letter event as `ctx.event`.
    /// Best-effort: a lookup/insert failure is logged, not propagated
    /// (the dead-letter row itself is already durably written).
    ///
    /// Called from both the outbox arm (`handle_failure`) and the queue
    /// arm (`handle_queue_failure`). The queue arm previously omitted
    /// this call, so registered `dead_letter` handlers filtered on
    /// `source = "queue"` sat idle even after queue retry-exhaust.
    async fn fan_out_dead_letter(&self, ctx: DeadLetterFanOutCtx) {
        let DeadLetterFanOutCtx {
            app_id,
            original,
            source,
            dead_letter_id,
            attempts,
            last_error,
            trigger_id,
            script_id,
            first_attempt_at,
            last_attempt_at,
            trigger_depth,
            root_execution_id,
        } = ctx;

        let matches = match self
            .triggers
            .list_matching_dead_letter(app_id, &source, trigger_id, script_id)
            .await
        {
            Ok(m) => m,
            Err(e) => {
                tracing::error!(?e, "dead-letter trigger lookup failed");
                return;
            }
        };

        for m in matches {
            let event = TriggerEvent::DeadLetter {
                dead_letter_id,
                original: Box::new(original.clone()),
                attempts,
                last_error: last_error.clone(),
                trigger_id,
                script_id,
                first_attempt_at,
                last_attempt_at,
            };
            let payload = match serde_json::to_value(&event) {
                Ok(p) => p,
                Err(e) => {
                    tracing::error!(?e, "failed to serialize dead-letter event");
                    continue;
                }
            };
            if let Err(e) = self
                .outbox
                .insert(NewOutboxRow {
                    app_id,
                    source_kind: OutboxSourceKind::DeadLetter,
                    trigger_id: Some(m.trigger_id),
                    script_id: Some(m.script_id),
                    reply_to: None,
                    payload,
                    origin_principal: Some(m.registered_by_principal),
                    trigger_depth: trigger_depth.saturating_add(1),
                    root_execution_id,
                })
                .await
            {
                tracing::error!(?e, "failed to enqueue dead-letter handler delivery");
            }
        }
    }

    async fn deliver_inbox(&self, row: &OutboxRow, inbox_id: Uuid, result: InboxResult) {
        match self.inbox.deliver(inbox_id, result.clone()).await {
            InboxDeliveryOutcome::Delivered => {}
            InboxDeliveryOutcome::Abandoned => {
                // Receiver was dropped — record forensic row + bump
                // metric.
                let (status_code, summary) = match &result {
                    InboxResult::Success(s) => (s.status_code, None),
                    InboxResult::Failure { kind, message } => {
                        (failure_kind_to_status(*kind), Some(message.clone()))
                    }
                };
                if let Err(e) = self
                    .abandoned
                    .insert(NewAbandonedExecution {
                        app_id: row.app_id,
                        outbox_id: row.id,
                        script_id: row.script_id,
                        inbox_id,
                        status_code,
                        result_summary: summary,
                    })
                    .await
                {
                    tracing::warn!(?e, "abandoned_executions insert failed");
                }
                // TODO(metrics): bump `picloud_abandoned_executions_total{app_id}`.
            }
        }
    }
}

#[derive(Debug)]
pub struct ResolvedTrigger {
    pub trigger_kind: TriggerKind,
    pub is_dead_letter_handler: bool,
    /// §4.3 fire-time gate: `trigger.enabled && script.enabled` at resolve
    /// time. A row enqueued before either was disabled is dropped, not fired.
    pub active: bool,
    pub script_id: ScriptId,
    pub script_source: String,
    pub script_name: String,
    /// Phase 4b: the resolved script's defining node — the lexical origin
    /// for its `import`s (§5.5). `None` only for a corrupt owner row; the
    /// executor then falls back to `App(app_id)`.
    pub script_owner: Option<ScriptOwner>,
    /// v1.1.3: freshness comparator for the orchestrator's top-level
    /// script cache. The dispatcher hands `(script_id, updated_at)`
    /// in alongside the source so cached ASTs can be reused across
    /// triggered invocations.
    pub script_updated_at: chrono::DateTime<chrono::Utc>,
    pub sandbox_overrides: ScriptSandbox,
    pub registered_by_principal: picloud_shared::AdminUserId,
    pub retry_max_attempts: u32,
    pub retry_backoff: BackoffShape,
    pub retry_base_ms: u32,
}

#[derive(Debug, thiserror::Error)]
pub enum DispatcherError {
    #[error("outbox: {0}")]
    Outbox(String),
    #[error("resolve trigger: {0}")]
    ResolveTrigger(String),
}

fn summarize(resp: &ExecResponse) -> ExecResponseSummary {
    ExecResponseSummary {
        status_code: resp.status_code,
        headers: resp.headers.clone(),
        body: resp.body.clone(),
    }
}

/// Map `ExecError` onto the design-notes §3 status-code table.
fn classify_exec_error(err: &ExecError) -> (InboxFailureKind, String) {
    match err {
        ExecError::Parse(s) | ExecError::InvalidResponse(s) => {
            (InboxFailureKind::Validation, s.clone())
        }
        ExecError::Timeout(_) => (InboxFailureKind::Timeout, err.to_string()),
        ExecError::OperationBudgetExceeded => (InboxFailureKind::OperationBudget, err.to_string()),
        ExecError::Overloaded { .. } => (InboxFailureKind::Overloaded, err.to_string()),
        ExecError::Runtime(s) => (InboxFailureKind::Runtime, s.clone()),
    }
}

fn failure_kind_to_status(k: InboxFailureKind) -> u16 {
    match k {
        InboxFailureKind::Validation => 422,
        InboxFailureKind::Runtime => 502,
        InboxFailureKind::Overloaded => 503,
        InboxFailureKind::Timeout => 504,
        InboxFailureKind::OperationBudget => 507,
        InboxFailureKind::Platform => 500,
    }
}

/// `(op, source)` extracted from the outbox payload. Used to seed the
/// `dead_letters` row when retries exhaust.
///
/// The shape of the payload depends on `source_kind`:
///   * `Http` rows carry an `HttpDispatchPayload` (`method`/`path`).
///   * `Invoke` rows carry an ad-hoc invoke payload (`script_id`/`args`).
///   * Every other kind carries a serialized [`TriggerEvent`], which has
///     `source` and `op` fields at the JSON top level.
///
/// F-S-009 (audit fix): previously this function unconditionally read
/// `source`/`op` from the JSON, which silently produced empty strings
/// for HTTP rows. The DL row then got written with `source = ""` and
/// `from_wire("")` returned `None` on replay, defaulting to
/// `OutboxSourceKind::Kv` — replay rerouted through `resolve_trigger`,
/// which failed because HTTP rows carry no `trigger_id`. Net result:
/// HTTP-async DL replays no-op'd silently.
fn describe_event(source_kind: OutboxSourceKind, payload: &serde_json::Value) -> (String, String) {
    match source_kind {
        OutboxSourceKind::Http => {
            let method = payload
                .get("method")
                .and_then(|v| v.as_str())
                .unwrap_or("?");
            let path = payload.get("path").and_then(|v| v.as_str()).unwrap_or("?");
            (
                format!("{method} {path}"),
                OutboxSourceKind::Http.as_str().to_string(),
            )
        }
        OutboxSourceKind::Invoke => (
            "invoke_async".to_string(),
            OutboxSourceKind::Invoke.as_str().to_string(),
        ),
        OutboxSourceKind::Kv
        | OutboxSourceKind::Docs
        | OutboxSourceKind::DeadLetter
        | OutboxSourceKind::Cron
        | OutboxSourceKind::Files
        | OutboxSourceKind::Pubsub
        | OutboxSourceKind::Email => {
            let source = payload
                .get("source")
                .and_then(|v| v.as_str())
                .unwrap_or("")
                .to_string();
            let op = payload
                .get("op")
                .and_then(|v| v.as_str())
                .unwrap_or("")
                .to_string();
            (op, source)
        }
    }
}

/// Compute backoff (ms) for the given attempt + policy + jitter.
/// Attempt is 1-indexed (first retry = attempt 1).
#[must_use]
pub fn compute_backoff(attempt: u32, backoff: BackoffShape, base_ms: u32, jitter_pct: u32) -> u32 {
    let base_ms = u64::from(base_ms);
    let attempt = u64::from(attempt.saturating_sub(1));
    let raw = match backoff {
        BackoffShape::Constant => base_ms,
        BackoffShape::Linear => base_ms * (attempt + 1),
        // 1x base, 2x base, 4x base, … (saturating).
        BackoffShape::Exponential => base_ms.saturating_mul(1u64 << attempt.min(20)),
    };
    let raw = u32::try_from(raw.min(u64::from(u32::MAX))).unwrap_or(u32::MAX);
    apply_jitter(raw, jitter_pct)
}

fn apply_jitter(raw: u32, pct: u32) -> u32 {
    if pct == 0 {
        return raw;
    }
    let pct = pct.min(100);
    // ±span% — bounded by raw itself so we can't underflow when
    // raw + offset goes below zero.
    let span = u64::from(raw) * u64::from(pct) / 100;
    if span == 0 {
        return raw;
    }
    let span_i64 = i64::try_from(span).unwrap_or(i64::MAX);
    let mut rng = rand::thread_rng();
    let offset = rng.gen_range(-span_i64..=span_i64);
    let signed = i64::from(raw).saturating_add(offset).max(0);
    u32::try_from(signed.min(i64::from(u32::MAX))).unwrap_or(u32::MAX)
}

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

    /// §11.6 D3: a do-nothing group-queue store for dispatcher tests that don't
    /// exercise SHARED queues (every consumer has `shared_group: None`, so these
    /// methods are never reached — they just satisfy the struct field).
    struct NoopGroupQueue;

    #[async_trait::async_trait]
    impl crate::group_queue_repo::GroupQueueRepo for NoopGroupQueue {
        async fn enqueue(
            &self,
            _msg: crate::group_queue_repo::NewGroupQueueMessage,
        ) -> Result<QueueMessageId, crate::group_queue_repo::GroupQueueRepoError> {
            unreachable!("shared queue not exercised")
        }
        async fn claim(
            &self,
            _group_id: picloud_shared::GroupId,
            _collection: &str,
        ) -> Result<
            Option<crate::group_queue_repo::ClaimedGroupMessage>,
            crate::group_queue_repo::GroupQueueRepoError,
        > {
            Ok(None)
        }
        async fn ack(
            &self,
            _id: QueueMessageId,
            _token: Uuid,
        ) -> Result<bool, crate::group_queue_repo::GroupQueueRepoError> {
            Ok(false)
        }
        async fn nack(
            &self,
            _id: QueueMessageId,
            _token: Uuid,
            _delay: chrono::Duration,
        ) -> Result<bool, crate::group_queue_repo::GroupQueueRepoError> {
            Ok(false)
        }
        #[allow(clippy::too_many_arguments)]
        async fn dead_letter(
            &self,
            _id: QueueMessageId,
            _token: Uuid,
            _group_id: picloud_shared::GroupId,
            _collection: &str,
            _trigger_id: Option<picloud_shared::TriggerId>,
            _script_id: Option<ScriptId>,
            _attempt: u32,
            _first_attempt_at: chrono::DateTime<chrono::Utc>,
            _last_error: &str,
        ) -> Result<picloud_shared::DeadLetterId, crate::group_queue_repo::GroupQueueRepoError>
        {
            unreachable!("shared queue not exercised")
        }
        async fn reclaim_visibility_timeouts(
            &self,
        ) -> Result<u64, crate::group_queue_repo::GroupQueueRepoError> {
            Ok(0)
        }
        async fn depth(
            &self,
            _group_id: picloud_shared::GroupId,
            _collection: &str,
        ) -> Result<u64, crate::group_queue_repo::GroupQueueRepoError> {
            Ok(0)
        }
        async fn depth_pending(
            &self,
            _group_id: picloud_shared::GroupId,
            _collection: &str,
        ) -> Result<u64, crate::group_queue_repo::GroupQueueRepoError> {
            Ok(0)
        }
    }

    #[test]
    fn exponential_backoff_doubles_per_attempt() {
        // No jitter (pct=0) for a deterministic check.
        assert_eq!(compute_backoff(1, BackoffShape::Exponential, 1000, 0), 1000);
        assert_eq!(compute_backoff(2, BackoffShape::Exponential, 1000, 0), 2000);
        assert_eq!(compute_backoff(3, BackoffShape::Exponential, 1000, 0), 4000);
        assert_eq!(compute_backoff(4, BackoffShape::Exponential, 1000, 0), 8000);
    }

    #[test]
    fn linear_backoff_scales_with_attempt() {
        assert_eq!(compute_backoff(1, BackoffShape::Linear, 100, 0), 100);
        assert_eq!(compute_backoff(2, BackoffShape::Linear, 100, 0), 200);
        assert_eq!(compute_backoff(5, BackoffShape::Linear, 100, 0), 500);
    }

    #[test]
    fn constant_backoff_returns_base() {
        for attempt in 1..=5 {
            assert_eq!(
                compute_backoff(attempt, BackoffShape::Constant, 750, 0),
                750
            );
        }
    }

    #[test]
    fn jitter_within_pct_of_base() {
        for _ in 0..100 {
            let v = compute_backoff(1, BackoffShape::Constant, 1000, 20);
            // ±20% of 1000 = 800..=1200.
            assert!((800..=1200).contains(&v), "jitter out of range: {v}");
        }
    }

    #[test]
    fn describe_event_http_uses_method_and_path() {
        let payload = serde_json::json!({
            "method": "POST",
            "path": "/webhook",
            "headers": {},
            "body": null,
            "params": {},
            "query": {},
            "rest": "",
            "script_name": "hook",
            "timeout_seconds": 30,
        });
        let (op, source) = describe_event(OutboxSourceKind::Http, &payload);
        assert_eq!(op, "POST /webhook");
        assert_eq!(source, "http");
        // Round-trip via from_wire so the DL replay path resolves to the
        // HTTP branch instead of falling back to Kv.
        assert_eq!(
            OutboxSourceKind::from_wire(&source),
            Some(OutboxSourceKind::Http)
        );
    }

    #[test]
    fn describe_event_invoke_uses_constant_op() {
        let payload = serde_json::json!({
            "script_id": "00000000-0000-0000-0000-000000000000",
            "args": {},
        });
        let (op, source) = describe_event(OutboxSourceKind::Invoke, &payload);
        assert_eq!(op, "invoke_async");
        assert_eq!(source, "invoke");
        assert_eq!(
            OutboxSourceKind::from_wire(&source),
            Some(OutboxSourceKind::Invoke)
        );
    }

    #[test]
    fn describe_event_trigger_event_reads_top_level_fields() {
        let payload = serde_json::json!({
            "source": "kv",
            "op": "set",
            "collection": "users",
            "key": "alice",
            "value": {"name": "Alice"},
        });
        let (op, source) = describe_event(OutboxSourceKind::Kv, &payload);
        assert_eq!(op, "set");
        assert_eq!(source, "kv");
    }

    #[test]
    fn classify_exec_error_covers_every_variant() {
        let parse = classify_exec_error(&ExecError::Parse("nope".into()));
        assert!(matches!(parse.0, InboxFailureKind::Validation));
        let invalid = classify_exec_error(&ExecError::InvalidResponse("bad".into()));
        assert!(matches!(invalid.0, InboxFailureKind::Validation));
        let timeout = classify_exec_error(&ExecError::Timeout(30));
        assert!(matches!(timeout.0, InboxFailureKind::Timeout));
        let budget = classify_exec_error(&ExecError::OperationBudgetExceeded);
        assert!(matches!(budget.0, InboxFailureKind::OperationBudget));
        let runtime = classify_exec_error(&ExecError::Runtime("threw".into()));
        assert!(matches!(runtime.0, InboxFailureKind::Runtime));
        let overload = classify_exec_error(&ExecError::Overloaded {
            retry_after_secs: 1,
        });
        assert!(matches!(overload.0, InboxFailureKind::Overloaded));
    }

    #[test]
    fn failure_kind_status_codes_match_design_notes() {
        assert_eq!(failure_kind_to_status(InboxFailureKind::Validation), 422);
        assert_eq!(failure_kind_to_status(InboxFailureKind::Runtime), 502);
        assert_eq!(failure_kind_to_status(InboxFailureKind::Overloaded), 503);
        assert_eq!(failure_kind_to_status(InboxFailureKind::Timeout), 504);
        assert_eq!(
            failure_kind_to_status(InboxFailureKind::OperationBudget),
            507
        );
        assert_eq!(failure_kind_to_status(InboxFailureKind::Platform), 500);
    }

    // ----------------------------------------------------------------
    // Queue-arm fire-time `enabled` gate (§4.3).
    //
    // `dispatch_one_queue` re-reads the script fresh after claiming a
    // message and, if it has been disabled since the per-tick
    // `list_active_queue_consumers` snapshot, releases the claim (nack)
    // without resolving a principal or executing. This test proves that
    // gate end-to-end with in-memory stubs.
    //
    // Regression property: the principal resolver returns a valid
    // `Principal` and the executor records `executed = true` before
    // erroring, so DELETING the `if !script.enabled` gate makes the flow
    // fall through to resolve → execute, flipping `executed` and failing
    // `assert!(!executed)`. (Verified by temporarily removing the gate.)
    // ----------------------------------------------------------------
    mod queue_enabled_gate {
        use super::*;
        use std::sync::atomic::{AtomicBool, Ordering};
        use std::sync::Arc;

        use async_trait::async_trait;
        use chrono::Utc;
        use picloud_executor_core::{ExecError, ExecRequest, ExecResponse};
        use picloud_orchestrator_core::{ExecutionGate, ExecutorClient};
        use picloud_shared::{
            AdminUserId, AppId, InstanceRole, Principal, Script, ScriptId, ScriptSandbox,
        };
        use uuid::Uuid;

        use crate::abandoned_repo::{AbandonedRepo, NewAbandonedExecution};
        use crate::dead_letter_repo::{DeadLetterRepo, NewDeadLetter};
        use crate::outbox_repo::{NewOutboxRow, OutboxRepo, OutboxRow};
        use crate::principal_resolver::{PrincipalResolver, PrincipalResolverError};
        use crate::queue_repo::{ClaimedMessage, NewQueueMessage, QueueRepo, QueueStats};
        use crate::repo::{NewScript, ScriptPatch, ScriptRepository, ScriptRepositoryError};
        use crate::trigger_config::BackoffShape;
        use crate::trigger_repo::{ActiveQueueConsumer, TriggerRepo};

        // ---- ScriptRepository: only `get` is exercised. ----
        pub(super) struct DisabledScriptRepo {
            pub(super) script: Script,
        }

        #[async_trait]
        impl ScriptRepository for DisabledScriptRepo {
            async fn get(&self, _id: ScriptId) -> Result<Option<Script>, ScriptRepositoryError> {
                Ok(Some(self.script.clone()))
            }
            async fn get_by_name(
                &self,
                _app_id: AppId,
                _name: &str,
            ) -> Result<Option<Script>, ScriptRepositoryError> {
                unimplemented!("not used by this test")
            }
            async fn get_by_name_inherited(
                &self,
                _app_id: AppId,
                _name: &str,
            ) -> Result<Option<Script>, ScriptRepositoryError> {
                unimplemented!("not used by this test")
            }
            async fn is_invocable_by_app(
                &self,
                _script_id: ScriptId,
                app_id: AppId,
            ) -> Result<bool, ScriptRepositoryError> {
                Ok(self.script.is_owned_by_app(app_id))
            }
            async fn list(&self) -> Result<Vec<Script>, ScriptRepositoryError> {
                unimplemented!("not used by this test")
            }
            async fn list_for_app(
                &self,
                _app_id: AppId,
            ) -> Result<Vec<Script>, ScriptRepositoryError> {
                unimplemented!("not used by this test")
            }
            async fn list_for_group(
                &self,
                _group_id: picloud_shared::GroupId,
            ) -> Result<Vec<Script>, ScriptRepositoryError> {
                unimplemented!("not used by this test")
            }
            async fn list_for_user(
                &self,
                _user_id: AdminUserId,
            ) -> Result<Vec<Script>, ScriptRepositoryError> {
                unimplemented!("not used by this test")
            }
            async fn create(&self, _input: NewScript) -> Result<Script, ScriptRepositoryError> {
                unimplemented!("not used by this test")
            }
            async fn update(
                &self,
                _id: ScriptId,
                _patch: ScriptPatch,
            ) -> Result<Script, ScriptRepositoryError> {
                unimplemented!("not used by this test")
            }
            async fn delete(&self, _id: ScriptId) -> Result<(), ScriptRepositoryError> {
                unimplemented!("not used by this test")
            }
            async fn count_routes_for_script(
                &self,
                _script_id: ScriptId,
            ) -> Result<i64, ScriptRepositoryError> {
                unimplemented!("not used by this test")
            }
            async fn count_triggers_for_script(
                &self,
                _script_id: ScriptId,
            ) -> Result<i64, ScriptRepositoryError> {
                unimplemented!("not used by this test")
            }
            async fn list_imports(
                &self,
                _script_id: ScriptId,
            ) -> Result<Vec<Script>, ScriptRepositoryError> {
                unimplemented!("not used by this test")
            }
        }

        // ---- QueueRepo: `claim` returns the message, `nack` records. ----
        struct ClaimNackQueue {
            claimed: ClaimedMessage,
            nacked: Arc<AtomicBool>,
        }

        #[async_trait]
        impl QueueRepo for ClaimNackQueue {
            async fn enqueue(
                &self,
                _msg: NewQueueMessage,
            ) -> Result<picloud_shared::QueueMessageId, crate::queue_repo::QueueRepoError>
            {
                unimplemented!("not used by this test")
            }
            async fn claim(
                &self,
                _app_id: AppId,
                _queue_name: &str,
            ) -> Result<Option<ClaimedMessage>, crate::queue_repo::QueueRepoError> {
                Ok(Some(self.claimed.clone()))
            }
            async fn ack(
                &self,
                _message_id: QueueMessageId,
                _claim_token: Uuid,
            ) -> Result<bool, crate::queue_repo::QueueRepoError> {
                unimplemented!("not used by this test")
            }
            async fn nack(
                &self,
                _message_id: QueueMessageId,
                _claim_token: Uuid,
                _retry_delay: chrono::Duration,
            ) -> Result<bool, crate::queue_repo::QueueRepoError> {
                self.nacked.store(true, Ordering::SeqCst);
                Ok(true)
            }
            async fn reclaim_visibility_timeouts(
                &self,
            ) -> Result<u64, crate::queue_repo::QueueRepoError> {
                unimplemented!("not used by this test")
            }
            async fn depth(
                &self,
                _app_id: AppId,
                _queue_name: &str,
            ) -> Result<u64, crate::queue_repo::QueueRepoError> {
                unimplemented!("not used by this test")
            }
            async fn depth_pending(
                &self,
                _app_id: AppId,
                _queue_name: &str,
            ) -> Result<u64, crate::queue_repo::QueueRepoError> {
                unimplemented!("not used by this test")
            }
            async fn list_for_app(
                &self,
                _app_id: AppId,
            ) -> Result<Vec<(String, QueueStats)>, crate::queue_repo::QueueRepoError> {
                unimplemented!("not used by this test")
            }
            #[allow(clippy::too_many_arguments)]
            async fn dead_letter(
                &self,
                _message_id: QueueMessageId,
                _claim_token: Uuid,
                _app_id: AppId,
                _queue_name: &str,
                _trigger_id: Option<picloud_shared::TriggerId>,
                _script_id: Option<ScriptId>,
                _attempt: u32,
                _first_attempt_at: chrono::DateTime<Utc>,
                _last_error: &str,
            ) -> Result<picloud_shared::DeadLetterId, crate::queue_repo::QueueRepoError>
            {
                unimplemented!("not used by this test")
            }
        }

        // ---- PrincipalResolver: returns a valid Principal (regression). ----
        pub(super) struct OkPrincipals;

        #[async_trait]
        impl PrincipalResolver for OkPrincipals {
            async fn resolve(
                &self,
                user_id: AdminUserId,
            ) -> Result<Principal, PrincipalResolverError> {
                Ok(Principal {
                    user_id,
                    instance_role: InstanceRole::Owner,
                    scopes: None,
                    app_binding: None,
                })
            }
        }

        // ---- ExecutorClient: records execution then errors (regression). ----
        pub(super) struct RecordingExecutor {
            pub(super) executed: Arc<AtomicBool>,
        }

        #[async_trait]
        impl ExecutorClient for RecordingExecutor {
            async fn execute(
                &self,
                _source: &str,
                _req: ExecRequest,
                _timeout: std::time::Duration,
            ) -> Result<ExecResponse, ExecError> {
                self.executed.store(true, Ordering::SeqCst);
                Err(ExecError::Runtime("stub".into()))
            }
            // Intentionally NOT overriding `execute_with_identity`: the
            // default impl forwards to `execute`, which is what gate
            // removal would reach.
        }

        // ---- Remaining deps: never exercised by the disabled path. ----
        pub(super) struct UnusedTriggers;

        #[async_trait]
        impl TriggerRepo for UnusedTriggers {
            async fn create_kv_trigger(
                &self,
                _app_id: AppId,
                _req: crate::trigger_repo::CreateKvTrigger,
            ) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
            {
                unimplemented!("not used by this test")
            }
            async fn create_docs_trigger(
                &self,
                _app_id: AppId,
                _req: crate::trigger_repo::CreateDocsTrigger,
            ) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
            {
                unimplemented!("not used by this test")
            }
            async fn create_dead_letter_trigger(
                &self,
                _app_id: AppId,
                _req: crate::trigger_repo::CreateDeadLetterTrigger,
            ) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
            {
                unimplemented!("not used by this test")
            }
            async fn create_cron_trigger(
                &self,
                _app_id: AppId,
                _req: crate::trigger_repo::CreateCronTrigger,
            ) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
            {
                unimplemented!("not used by this test")
            }
            async fn create_files_trigger(
                &self,
                _app_id: AppId,
                _req: crate::trigger_repo::CreateFilesTrigger,
            ) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
            {
                unimplemented!("not used by this test")
            }
            async fn create_pubsub_trigger(
                &self,
                _app_id: AppId,
                _req: crate::trigger_repo::CreatePubsubTrigger,
            ) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
            {
                unimplemented!("not used by this test")
            }
            async fn create_email_trigger(
                &self,
                _app_id: AppId,
                _req: crate::trigger_repo::CreateEmailTrigger,
            ) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
            {
                unimplemented!("not used by this test")
            }
            async fn email_inbound_target(
                &self,
                _trigger_id: picloud_shared::TriggerId,
            ) -> Result<
                Option<crate::trigger_repo::EmailInboundTarget>,
                crate::trigger_repo::TriggerRepoError,
            > {
                unimplemented!("not used by this test")
            }
            async fn list_for_app(
                &self,
                _app_id: AppId,
            ) -> Result<Vec<crate::trigger_repo::Trigger>, crate::trigger_repo::TriggerRepoError>
            {
                unimplemented!("not used by this test")
            }
            async fn get(
                &self,
                _id: picloud_shared::TriggerId,
            ) -> Result<Option<crate::trigger_repo::Trigger>, crate::trigger_repo::TriggerRepoError>
            {
                unimplemented!("not used by this test")
            }
            async fn delete(
                &self,
                _id: picloud_shared::TriggerId,
            ) -> Result<bool, crate::trigger_repo::TriggerRepoError> {
                unimplemented!("not used by this test")
            }
            async fn list_matching_kv(
                &self,
                _app_id: AppId,
                _collection: &str,
                _op: picloud_shared::KvEventOp,
            ) -> Result<
                Vec<crate::trigger_repo::KvTriggerMatch>,
                crate::trigger_repo::TriggerRepoError,
            > {
                unimplemented!("not used by this test")
            }
            async fn list_matching_docs(
                &self,
                _app_id: AppId,
                _collection: &str,
                _op: picloud_shared::DocsEventOp,
            ) -> Result<
                Vec<crate::trigger_repo::DocsTriggerMatch>,
                crate::trigger_repo::TriggerRepoError,
            > {
                unimplemented!("not used by this test")
            }
            async fn list_matching_files(
                &self,
                _app_id: AppId,
                _collection: &str,
                _op: picloud_shared::FilesEventOp,
            ) -> Result<
                Vec<crate::trigger_repo::FilesTriggerMatch>,
                crate::trigger_repo::TriggerRepoError,
            > {
                unimplemented!("not used by this test")
            }
            async fn list_matching_dead_letter(
                &self,
                _app_id: AppId,
                _source: &str,
                _trigger_id: Option<picloud_shared::TriggerId>,
                _script_id: Option<ScriptId>,
            ) -> Result<
                Vec<crate::trigger_repo::DeadLetterTriggerMatch>,
                crate::trigger_repo::TriggerRepoError,
            > {
                unimplemented!("not used by this test")
            }
            async fn create_queue_trigger(
                &self,
                _app_id: AppId,
                _req: crate::trigger_repo::CreateQueueTrigger,
            ) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
            {
                unimplemented!("not used by this test")
            }
            async fn list_active_queue_consumers(
                &self,
            ) -> Result<Vec<ActiveQueueConsumer>, crate::trigger_repo::TriggerRepoError>
            {
                unimplemented!("not used by this test")
            }
            async fn touch_queue_trigger_last_fired_at(
                &self,
                _trigger_id: picloud_shared::TriggerId,
                _at: chrono::DateTime<Utc>,
            ) -> Result<(), crate::trigger_repo::TriggerRepoError> {
                unimplemented!("not used by this test")
            }
        }

        pub(super) struct UnusedDeadLetters;

        #[async_trait]
        impl DeadLetterRepo for UnusedDeadLetters {
            async fn insert(
                &self,
                _row: NewDeadLetter,
            ) -> Result<picloud_shared::DeadLetterId, crate::dead_letter_repo::DeadLetterRepoError>
            {
                unimplemented!("not used by this test")
            }
            async fn get(
                &self,
                _id: picloud_shared::DeadLetterId,
            ) -> Result<
                Option<crate::dead_letter_repo::DeadLetterRow>,
                crate::dead_letter_repo::DeadLetterRepoError,
            > {
                unimplemented!("not used by this test")
            }
            async fn list_for_app(
                &self,
                _app_id: AppId,
                _unresolved_only: bool,
                _limit: i64,
                _offset: i64,
            ) -> Result<
                Vec<crate::dead_letter_repo::DeadLetterRow>,
                crate::dead_letter_repo::DeadLetterRepoError,
            > {
                unimplemented!("not used by this test")
            }
            async fn unresolved_count(
                &self,
                _app_id: AppId,
            ) -> Result<i64, crate::dead_letter_repo::DeadLetterRepoError> {
                unimplemented!("not used by this test")
            }
            async fn resolve(
                &self,
                _id: picloud_shared::DeadLetterId,
                _reason: &str,
            ) -> Result<(), crate::dead_letter_repo::DeadLetterRepoError> {
                unimplemented!("not used by this test")
            }
            async fn gc(
                &self,
                _older_than: chrono::DateTime<Utc>,
                _limit: i64,
            ) -> Result<u64, crate::dead_letter_repo::DeadLetterRepoError> {
                unimplemented!("not used by this test")
            }
        }

        struct UnusedOutbox;

        #[async_trait]
        impl OutboxRepo for UnusedOutbox {
            async fn insert(
                &self,
                _row: NewOutboxRow,
            ) -> Result<Uuid, crate::outbox_repo::OutboxRepoError> {
                unimplemented!("not used by this test")
            }
            async fn claim_due(
                &self,
                _claimed_by: &str,
                _limit: i64,
            ) -> Result<Vec<OutboxRow>, crate::outbox_repo::OutboxRepoError> {
                unimplemented!("not used by this test")
            }
            async fn delete(&self, _id: Uuid) -> Result<(), crate::outbox_repo::OutboxRepoError> {
                unimplemented!("not used by this test")
            }
            async fn reschedule(
                &self,
                _id: Uuid,
                _attempt_count: u32,
                _next_attempt_at: chrono::DateTime<Utc>,
            ) -> Result<(), crate::outbox_repo::OutboxRepoError> {
                unimplemented!("not used by this test")
            }
        }

        pub(super) struct UnusedAbandoned;

        #[async_trait]
        impl AbandonedRepo for UnusedAbandoned {
            async fn insert(
                &self,
                _row: NewAbandonedExecution,
            ) -> Result<Uuid, crate::abandoned_repo::AbandonedRepoError> {
                unimplemented!("not used by this test")
            }
            async fn gc(
                &self,
                _older_than: chrono::DateTime<Utc>,
                _limit: i64,
            ) -> Result<u64, crate::abandoned_repo::AbandonedRepoError> {
                unimplemented!("not used by this test")
            }
        }

        pub(super) struct UnusedLogSink;

        #[async_trait]
        impl ExecutionLogSink for UnusedLogSink {
            async fn record(
                &self,
                _log: picloud_shared::ExecutionLog,
            ) -> Result<(), picloud_shared::LogSinkError> {
                unimplemented!("not used by this test")
            }
        }

        pub(super) struct UnusedInbox;

        #[async_trait]
        impl InboxResolver for UnusedInbox {
            async fn deliver(
                &self,
                _inbox_id: Uuid,
                _result: picloud_shared::InboxResult,
            ) -> picloud_shared::InboxDeliveryOutcome {
                unimplemented!("not used by this test")
            }
        }

        pub(super) fn disabled_script(app_id: AppId) -> Script {
            Script {
                id: ScriptId::new(),
                app_id: Some(app_id),
                group_id: None,
                name: "worker".into(),
                description: None,
                version: 1,
                source: "0".into(),
                kind: picloud_shared::ScriptKind::Endpoint,
                timeout_seconds: 30,
                memory_limit_mb: 64,
                sandbox: ScriptSandbox::default(),
                enabled: false,
                created_at: Utc::now(),
                updated_at: Utc::now(),
            }
        }

        #[tokio::test]
        async fn disabled_queue_consumer_releases_claim_without_executing() {
            let app_id = AppId::new();
            let script = disabled_script(app_id);

            let claimed = ClaimedMessage {
                id: picloud_shared::QueueMessageId::new(),
                app_id,
                queue_name: "jobs".into(),
                payload: serde_json::Value::Null,
                enqueued_at: Utc::now(),
                attempt: 1,
                max_attempts: 5,
                claim_token: Uuid::new_v4(),
            };

            let consumer = ActiveQueueConsumer {
                trigger_id: picloud_shared::TriggerId::new(),
                app_id,
                script_id: script.id,
                queue_name: "jobs".into(),
                visibility_timeout_secs: 30,
                retry_max_attempts: 5,
                retry_backoff: BackoffShape::Exponential,
                retry_base_ms: 1000,
                registered_by_principal: AdminUserId::new(),
                shared_group: None,
            };

            let nacked = Arc::new(AtomicBool::new(false));
            let executed = Arc::new(AtomicBool::new(false));

            let dispatcher = Dispatcher {
                outbox: Arc::new(UnusedOutbox),
                triggers: Arc::new(UnusedTriggers),
                scripts: Arc::new(DisabledScriptRepo {
                    script: script.clone(),
                }),
                dead_letters: Arc::new(UnusedDeadLetters),
                abandoned: Arc::new(UnusedAbandoned),
                principals: Arc::new(OkPrincipals),
                executor: Arc::new(RecordingExecutor {
                    executed: executed.clone(),
                }),
                gate: Arc::new(ExecutionGate::new(1)),
                log_sink: Arc::new(UnusedLogSink),
                inbox: Arc::new(UnusedInbox),
                queue: Arc::new(ClaimNackQueue {
                    claimed,
                    nacked: nacked.clone(),
                }),
                group_queue: Arc::new(NoopGroupQueue),
                config: TriggerConfig::from_env(),
                instance_id: "test-instance".into(),
            };

            let result = dispatcher.dispatch_one_queue(&consumer).await;

            // 1. The disabled path returns Ok(()).
            assert!(result.is_ok(), "dispatch_one_queue returned {result:?}");
            // 2. The claim was released via nack.
            assert!(
                nacked.load(Ordering::SeqCst),
                "expected nack to release the claim for a disabled consumer"
            );
            // 3. The executor was never reached. If the `if !script.enabled`
            //    gate is deleted, the flow falls through to resolve →
            //    execute and this flips to true.
            assert!(
                !executed.load(Ordering::SeqCst),
                "executor ran for a disabled queue consumer; fire-time gate missing"
            );
        }
    }

    // ----------------------------------------------------------------
    // OUTBOX (async-HTTP) arm fire-time `enabled` gate (§4.3).
    //
    // `dispatch_one` builds an `ExecRequest` from an HTTP outbox row via
    // `build_http_request`, which sets `resolved.active = script.enabled`
    // but does NOT itself reject a disabled script. The unified gate at
    // `if !resolved.active` then drops the row (delete) without executing.
    // This test proves that gate end-to-end for an HTTP-source row whose
    // target script was disabled after the row was enqueued.
    //
    // Regression property (mirrors the queue test): `RecordingExecutor`
    // flips `executed = true` before erroring, so DELETING the
    // `if !resolved.active` gate makes the flow fall through to execute,
    // flipping `executed` and failing `assert!(!executed)`.
    // ----------------------------------------------------------------
    mod outbox_enabled_gate {
        use super::*;
        use std::sync::atomic::{AtomicBool, Ordering};
        use std::sync::{Arc, Mutex};

        use async_trait::async_trait;
        use chrono::Utc;
        use picloud_orchestrator_core::ExecutionGate;
        use picloud_shared::AppId;
        use uuid::Uuid;

        use crate::outbox_repo::{NewOutboxRow, OutboxRepo, OutboxRow, OutboxSourceKind};

        // Reuse the stub trait impls + helpers from the queue test so the
        // two gate tests share one set of in-memory deps.
        use super::queue_enabled_gate::{
            disabled_script, DisabledScriptRepo, OkPrincipals, RecordingExecutor, UnusedAbandoned,
            UnusedDeadLetters, UnusedInbox, UnusedLogSink, UnusedTriggers,
        };

        // QueueRepo is never reached on the outbox arm — a panic stub keeps
        // the surface honest without pulling the queue test's claim stub.
        struct UnusedQueue;

        #[async_trait]
        impl crate::queue_repo::QueueRepo for UnusedQueue {
            async fn enqueue(
                &self,
                _msg: crate::queue_repo::NewQueueMessage,
            ) -> Result<picloud_shared::QueueMessageId, crate::queue_repo::QueueRepoError>
            {
                unimplemented!("not used by this test")
            }
            async fn claim(
                &self,
                _app_id: AppId,
                _queue_name: &str,
            ) -> Result<Option<crate::queue_repo::ClaimedMessage>, crate::queue_repo::QueueRepoError>
            {
                unimplemented!("not used by this test")
            }
            async fn ack(
                &self,
                _message_id: QueueMessageId,
                _claim_token: Uuid,
            ) -> Result<bool, crate::queue_repo::QueueRepoError> {
                unimplemented!("not used by this test")
            }
            async fn nack(
                &self,
                _message_id: QueueMessageId,
                _claim_token: Uuid,
                _retry_delay: chrono::Duration,
            ) -> Result<bool, crate::queue_repo::QueueRepoError> {
                unimplemented!("not used by this test")
            }
            async fn reclaim_visibility_timeouts(
                &self,
            ) -> Result<u64, crate::queue_repo::QueueRepoError> {
                unimplemented!("not used by this test")
            }
            async fn depth(
                &self,
                _app_id: AppId,
                _queue_name: &str,
            ) -> Result<u64, crate::queue_repo::QueueRepoError> {
                unimplemented!("not used by this test")
            }
            async fn depth_pending(
                &self,
                _app_id: AppId,
                _queue_name: &str,
            ) -> Result<u64, crate::queue_repo::QueueRepoError> {
                unimplemented!("not used by this test")
            }
            async fn list_for_app(
                &self,
                _app_id: AppId,
            ) -> Result<
                Vec<(String, crate::queue_repo::QueueStats)>,
                crate::queue_repo::QueueRepoError,
            > {
                unimplemented!("not used by this test")
            }
            #[allow(clippy::too_many_arguments)]
            async fn dead_letter(
                &self,
                _message_id: QueueMessageId,
                _claim_token: Uuid,
                _app_id: AppId,
                _queue_name: &str,
                _trigger_id: Option<picloud_shared::TriggerId>,
                _script_id: Option<picloud_shared::ScriptId>,
                _attempt: u32,
                _first_attempt_at: chrono::DateTime<Utc>,
                _last_error: &str,
            ) -> Result<picloud_shared::DeadLetterId, crate::queue_repo::QueueRepoError>
            {
                unimplemented!("not used by this test")
            }
        }

        // OutboxRepo whose `delete` records the id it was called with; the
        // other methods are never reached on the disabled-drop path.
        struct RecordingOutbox {
            deleted: Arc<Mutex<Option<Uuid>>>,
        }

        #[async_trait]
        impl OutboxRepo for RecordingOutbox {
            async fn insert(
                &self,
                _row: NewOutboxRow,
            ) -> Result<Uuid, crate::outbox_repo::OutboxRepoError> {
                unimplemented!("not used by this test")
            }
            async fn claim_due(
                &self,
                _claimed_by: &str,
                _limit: i64,
            ) -> Result<Vec<OutboxRow>, crate::outbox_repo::OutboxRepoError> {
                unimplemented!("not used by this test")
            }
            async fn delete(&self, id: Uuid) -> Result<(), crate::outbox_repo::OutboxRepoError> {
                *self.deleted.lock().unwrap() = Some(id);
                Ok(())
            }
            async fn reschedule(
                &self,
                _id: Uuid,
                _attempt_count: u32,
                _next_attempt_at: chrono::DateTime<Utc>,
            ) -> Result<(), crate::outbox_repo::OutboxRepoError> {
                unimplemented!("not used by this test")
            }
        }

        #[tokio::test]
        async fn disabled_http_outbox_row_is_dropped_without_executing() {
            let app_id = AppId::new();
            let script = disabled_script(app_id);

            // Minimal valid HttpDispatchPayload (see shared::outbox_writer).
            let payload = serde_json::json!({
                "script_name": "worker",
                "path": "/hook",
                "method": "POST",
                "headers": {},
                "body": null,
                "params": {},
                "query": {},
                "rest": "",
                "timeout_seconds": 30,
            });

            let row_id = Uuid::new_v4();
            let row = OutboxRow {
                id: row_id,
                // Same app_id as the script so the same-app guard passes.
                app_id,
                source_kind: OutboxSourceKind::Http,
                trigger_id: None,
                script_id: Some(script.id),
                reply_to: None,
                payload,
                origin_principal: None,
                trigger_depth: 0,
                root_execution_id: None,
                attempt_count: 0,
                next_attempt_at: Utc::now(),
                created_at: Utc::now(),
            };

            let deleted = Arc::new(Mutex::new(None));
            let executed = Arc::new(AtomicBool::new(false));

            let dispatcher = Dispatcher {
                outbox: Arc::new(RecordingOutbox {
                    deleted: deleted.clone(),
                }),
                triggers: Arc::new(UnusedTriggers),
                scripts: Arc::new(DisabledScriptRepo {
                    script: script.clone(),
                }),
                dead_letters: Arc::new(UnusedDeadLetters),
                abandoned: Arc::new(UnusedAbandoned),
                principals: Arc::new(OkPrincipals),
                executor: Arc::new(RecordingExecutor {
                    executed: executed.clone(),
                }),
                gate: Arc::new(ExecutionGate::new(1)),
                log_sink: Arc::new(UnusedLogSink),
                inbox: Arc::new(UnusedInbox),
                queue: Arc::new(UnusedQueue),
                group_queue: Arc::new(NoopGroupQueue),
                config: TriggerConfig::from_env(),
                instance_id: "test-instance".into(),
            };

            let result = dispatcher.dispatch_one(row).await;

            // 1. The disabled-drop path returns Ok(()).
            assert!(result.is_ok(), "dispatch_one returned {result:?}");
            // 2. The outbox row was deleted with its own id.
            assert_eq!(
                *deleted.lock().unwrap(),
                Some(row_id),
                "expected the disabled HTTP outbox row to be deleted"
            );
            // 3. The executor was never reached. If the `if !resolved.active`
            //    gate at dispatch_one is deleted, the flow falls through to
            //    execute and this flips to true.
            assert!(
                !executed.load(Ordering::SeqCst),
                "executor ran for a disabled HTTP outbox row; fire-time gate missing"
            );
        }
    }
}