feat(v1.1.9): dispatcher queue arm + visibility-timeout reclaim task

Extends Dispatcher with the v1.1.9 queue path. The queue table IS the
outbox for queue semantics, so the dispatcher polls queue_messages
directly via FOR UPDATE SKIP LOCKED — no outbox indirection.

Per tick (every 100ms):
  1) outbox arm — unchanged
  2) queue arm: list_active_queue_consumers() → per (app_id, queue_name)
     attempt one claim. Bounded by registered-consumer count so one busy
     queue can't starve others.

Per claimed message:
  - Build TriggerEvent::Queue + ExecRequest (executes as the trigger's
    registering principal, matching design notes §4)
  - dispatch through executor.execute_with_identity (reuses AST cache)
  - success → queue.ack(id, claim_token) — DELETE WHERE id AND token
  - throw + attempt < max_attempts → queue.nack(...) — clear claim, set
    deliver_after = NOW() + compute_backoff(attempt, backoff, base_ms, jitter)
  - throw + exhausted → queue.dead_letter(...) atomic move to dead_letters
    + DELETE in one transaction. fan_out_dead_letter on the outbox arm
    fires registered dead_letter handlers off the new row without changes.

Visibility-timeout reclaim task: separate tokio::spawn ticking on
queue_reclaim_interval_ms (default 30000). UPDATE clears claim_token /
claimed_at on rows whose claimed_at exceeds the per-queue
visibility_timeout_secs (joined from queue_trigger_details). A crashed
consumer thus loses its lease and the message becomes claimable again.

Dispatcher gains queue: Arc<dyn QueueRepo>; picloud/lib.rs threads
queue_repo.clone() into the construction.

ActiveQueueConsumer extended with app_id so the queue claim can be
performed without a follow-up trigger lookup; list_active_queue_consumers
SQL extended to SELECT t.app_id.

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

crates/manager-core/src/dispatcher.rs

@@ -37,9 +37,10 @@ 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::{TriggerKind, TriggerRepo};
+use crate::trigger_repo::{ActiveQueueConsumer, TriggerKind, TriggerRepo};
 
 /// Bundle the dispatcher reads from. Each handle is `Arc<dyn …>` so
 /// tests can substitute in-memory backings.
@@ -53,6 +54,9 @@ pub struct Dispatcher {
     pub executor: Arc<dyn ExecutorClient>,
     pub gate: Arc<ExecutionGate>,
     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>,
     pub config: TriggerConfig,
     /// Stable id for this dispatcher instance — written into
     /// `outbox.claimed_by` for forensics. In MVP this is the host's
@@ -76,10 +80,29 @@ const TICK_INTERVAL: Duration = Duration::from_millis(100);
 const ASYNC_EXEC_TIMEOUT: Duration = Duration::from_secs(300);
 
 impl Dispatcher {
-    /// Spawn the dispatcher loop as a detached `tokio::task`. The
+    /// Spawn the dispatcher loop as a detached `tokio::task`. Also
-    /// returned `JoinHandle` is dropped — the loop runs for the
+    /// spawns the v1.1.9 queue visibility-timeout reclaim task. Both
-    /// process lifetime.
+    /// 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_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"),
+                }
+            }
+        });
+
         tokio::spawn(async move {
             self.run().await;
         });
@@ -98,16 +121,12 @@ impl Dispatcher {
     }
 
     async fn tick(&self) -> Result<(), DispatcherError> {
-        // Cheap gate sample so we don't claim rows we can't dispatch.
+        // 1) Outbox arm — KV/docs/pubsub/cron/email/files/dead-letter/http/invoke.
-        // The exact permit budget is reapplied per-row below.
         let rows = self
             .outbox
             .claim_due(&self.instance_id, CLAIM_BATCH)
             .await
             .map_err(|e| DispatcherError::Outbox(e.to_string()))?;
-        if rows.is_empty() {
-            return Ok(());
-        }
         for row in rows {
             // Process serially within a tick — the outer ticker is the
             // pacing mechanism. Concurrent dispatchers are a cluster-
@@ -116,9 +135,199 @@ impl Dispatcher {
                 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.
+    async fn tick_queue_arm(&self) -> Result<(), DispatcherError> {
+        let consumers = self
+            .triggers
+            .list_active_queue_consumers()
+            .await
+            .map_err(|e| DispatcherError::ResolveTrigger(e.to_string()))?;
+        for consumer in consumers {
+            if let Err(err) = self.dispatch_one_queue(&consumer).await {
+                tracing::warn!(?err, queue = %consumer.queue_name, "queue dispatch errored");
+            }
+        }
+        Ok(())
+    }
+
+    async fn dispatch_one_queue(
+        &self,
+        consumer: &ActiveQueueConsumer,
+    ) -> Result<(), DispatcherError> {
+        // Atomic claim — None → nothing pending right now for this queue.
+        let Some(claimed) = self
+            .queue
+            .claim(consumer.app_id, &consumer.queue_name)
+            .await
+            .map_err(|e| DispatcherError::Outbox(e.to_string()))?
+        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
+                .queue
+                .nack(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
+                    .queue
+                    .dead_letter(
+                        claimed.id,
+                        claimed.claim_token,
+                        claimed.app_id,
+                        &claimed.queue_name,
+                        Some(consumer.trigger_id),
+                        Some(consumer.script_id),
+                        claimed.attempt,
+                        claimed.enqueued_at,
+                        "queue trigger script not found",
+                    )
+                    .await;
+                drop(permit);
+                return Ok(());
+            }
+            Err(e) => return Err(DispatcherError::ResolveTrigger(e.to_string())),
+        };
+
+        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(),
+            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,
+            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,
+        };
+        let outcome = self
+            .executor
+            .execute_with_identity(identity, &script.source, exec_req, ASYNC_EXEC_TIMEOUT)
+            .await;
+        drop(permit);
+
+        // 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.queue.ack(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.queue.nack(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 existing
+        // fan_out_dead_letter path on the dispatcher's outbox arm fires
+        // registered dead_letter handlers off the new row.
+        if let Err(e) = self
+            .queue
+            .dead_letter(
+                claimed.id,
+                claimed.claim_token,
+                claimed.app_id,
+                &claimed.queue_name,
+                Some(consumer.trigger_id),
+                Some(consumer.script_id),
+                claimed.attempt,
+                claimed.enqueued_at,
+                &err.to_string(),
+            )
+            .await
+        {
+            tracing::error!(?e, "queue dead-letter write failed");
+        }
+        // TODO(metrics): bump picloud_queue_dead_letters_total{app_id, queue_name}.
+    }
+
     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 {

crates/manager-core/src/trigger_repo.rs

@@ -281,10 +281,11 @@ pub struct CreateQueueTrigger {
 }
 
 /// One consumer the dispatcher's queue arm needs to scan messages for.
-/// Pulled by `list_active_queue_consumers(app_id)` per tick.
+/// Pulled by `list_active_queue_consumers()` per tick.
 #[derive(Debug, Clone)]
 pub struct ActiveQueueConsumer {
     pub trigger_id: TriggerId,
+    pub app_id: AppId,
     pub script_id: ScriptId,
     pub queue_name: String,
     pub visibility_timeout_secs: u32,
@@ -1290,7 +1291,7 @@ impl TriggerRepo for PostgresTriggerRepo {
         &self,
     ) -> Result<Vec<ActiveQueueConsumer>, TriggerRepoError> {
         let rows: Vec<QueueConsumerRow> = sqlx::query_as(
-            "SELECT t.id AS trigger_id, t.script_id, d.queue_name, \
+            "SELECT t.id AS trigger_id, t.app_id, t.script_id, d.queue_name, \
                     d.visibility_timeout_secs, \
                     t.retry_max_attempts, t.retry_backoff, t.retry_base_ms, \
                     t.registered_by_principal \
@@ -1304,6 +1305,7 @@ impl TriggerRepo for PostgresTriggerRepo {
             .into_iter()
             .map(|r| ActiveQueueConsumer {
                 trigger_id: r.trigger_id.into(),
+                app_id: r.app_id.into(),
                 script_id: r.script_id.into(),
                 queue_name: r.queue_name,
                 visibility_timeout_secs: u32::try_from(r.visibility_timeout_secs).unwrap_or(30),
@@ -1561,6 +1563,7 @@ struct QueueDetailRow {
 #[derive(sqlx::FromRow)]
 struct QueueConsumerRow {
     trigger_id: Uuid,
+    app_id: Uuid,
     script_id: Uuid,
     queue_name: String,
     visibility_timeout_secs: i32,

crates/picloud/src/lib.rs

@@ -344,6 +344,7 @@ pub async fn build_app(
         executor: executor.clone(),
         gate,
         inbox: inbox_resolver,
+        queue: queue_repo.clone(),
         config: trigger_config,
         instance_id: format!("picloud-{}", std::process::id()),
     }