feat(enabled): dispatcher fire-time re-check for disabled triggers/scripts

Close the §4.3 outbox gap: the match-time `enabled` check can't see a
trigger (or its target script) disabled *after* an outbox row was
enqueued, so a pending row would still fire. `resolve_trigger` now folds
`trigger.enabled && script.enabled` into the resolved row, and
`dispatch_one` drops the row at fire time when inactive instead of firing
a stale event.

The queue arm needs no change here: `list_active_queue_consumers` already
re-lists only enabled queue triggers each tick, so a disable is honored
promptly without a stale-row window.

Tested: manager-core lib 366 green; clippy -D warnings clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

crates/manager-core/src/dispatcher.rs

@@ -546,6 +546,7 @@ impl Dispatcher {
         // 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 {
@@ -610,6 +611,23 @@ impl Dispatcher {
             | OutboxSourceKind::Pubsub
             | OutboxSourceKind::Email => {
                 let resolved = self.resolve_trigger(&row).await?;
+                // §4.3 fire-time re-check: if the trigger or its target script
+                // was disabled after this row was enqueued, drop the row rather
+                // than fire a stale event. Closes the outbox gap where the
+                // match-time `enabled` check can't see a later toggle.
+                if !resolved.active {
+                    tracing::debug!(
+                        outbox_id = %row.id,
+                        app_id = %row.app_id,
+                        "trigger or target script disabled since enqueue; dropping row"
+                    );
+                    self.outbox
+                        .delete(row.id)
+                        .await
+                        .map_err(|e| DispatcherError::Outbox(e.to_string()))?;
+                    drop(permit);
+                    return Ok(());
+                }
                 let req = match self.build_exec_request(&row, &resolved).await {
                     Ok(req) => req,
                     Err(err) => {
@@ -725,6 +743,7 @@ impl Dispatcher {
         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_source: script.source,
             script_name: script.name,
@@ -844,6 +863,7 @@ impl Dispatcher {
         let resolved = ResolvedTrigger {
             trigger_kind: TriggerKind::Kv, // placeholder; HTTP doesn't have a kind
             is_dead_letter_handler: false,
+            active: true,
             script_id,
             script_source: script.source,
             script_name: payload.script_name,
@@ -941,6 +961,7 @@ impl Dispatcher {
         let resolved = ResolvedTrigger {
             trigger_kind: TriggerKind::Cron, // placeholder; not used downstream
             is_dead_letter_handler: false,
+            active: true,
             script_id: script.id,
             script_source: script.source,
             script_name: script.name,
@@ -1238,6 +1259,9 @@ impl Dispatcher {
 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,