feat(v1.1.9): TriggerEvent::Queue + TriggerKind::Queue + types

Add the data-shape pieces for v1.1.9's queue surface; behavior lands in
later commits. Each piece compiles independently.

- shared/trigger_event.rs: TriggerEvent::Queue variant (queue_name,
  message, enqueued_at, attempt, message_id). source() returns "queue".
  Surfaced to scripts as ctx.event.queue with op = "receive".
- manager-core/trigger_repo.rs: TriggerKind::Queue + TriggerDetails::Queue +
  CreateQueueTrigger + ActiveQueueConsumer + QueueDetailRow + QueueConsumerRow.
  PostgresTriggerRepo gains create_queue_trigger (advisory-lock-then-SELECT
  enforces one-consumer-per-queue), list_active_queue_consumers (dispatcher
  scan), touch_queue_trigger_last_fired_at. hydrate_one hydrates the Queue arm.
- manager-core/outbox_repo.rs: OutboxSourceKind::Invoke for invoke_async()
  outbox rows.
- manager-core/dispatcher.rs: placeholder OutboxSourceKind::Invoke arm (logs +
  drops the row) so the workspace compiles; real arm lands in commit 10.
- manager-core/trigger_config.rs: queue_reclaim_interval_ms (30000) +
  queue_default_visibility_timeout_secs (30) env-overridable knobs.
- executor-core/engine.rs: trigger_event_to_dynamic handles Queue → builds
  ctx.event.queue map.
- manager-core/triggers_api.rs: in-memory mock TriggerRepo gains the three
  new methods (returns Default-ish values for tests).

Unit tests: TriggerEvent::Queue serde round-trip, TriggerKind::Queue wire
round-trip, advisory_lock_key stability per (app_id, queue_name) pair.

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

crates/manager-core/src/dispatcher.rs

@@ -163,6 +163,22 @@ impl Dispatcher {
                     return Ok(());
                 }
             },
+            OutboxSourceKind::Invoke => {
+                // Wired in commit 10 (invoke_async() — function composition).
+                // Until then a malformed row is impossible to write (no
+                // producer), but defensively drop the row so a manual
+                // INSERT doesn't loop the dispatcher.
+                tracing::warn!(
+                    outbox_id = %row.id,
+                    "OutboxSourceKind::Invoke not wired yet; dropping row"
+                );
+                self.outbox
+                    .delete(row.id)
+                    .await
+                    .map_err(|e| DispatcherError::Outbox(e.to_string()))?;
+                drop(permit);
+                return Ok(());
+            }
             OutboxSourceKind::Kv
             | OutboxSourceKind::Docs
             | OutboxSourceKind::DeadLetter

crates/manager-core/src/outbox_repo.rs

@@ -33,6 +33,10 @@ pub enum OutboxSourceKind {
     Pubsub,
     /// v1.1.7. Inbound email POSTed to the webhook receiver.
     Email,
+    /// v1.1.9. Fire-and-forget function composition via
+    /// `invoke_async(target, args)`. The dispatcher fires once per row;
+    /// callers wrap in `retry::with` if they want retries.
+    Invoke,
 }
 
 impl OutboxSourceKind {
@@ -47,6 +51,7 @@ impl OutboxSourceKind {
             Self::Files => "files",
             Self::Pubsub => "pubsub",
             Self::Email => "email",
+            Self::Invoke => "invoke",
         }
     }
 
@@ -61,6 +66,7 @@ impl OutboxSourceKind {
             "files" => Some(Self::Files),
             "pubsub" => Some(Self::Pubsub),
             "email" => Some(Self::Email),
+            "invoke" => Some(Self::Invoke),
             _ => None,
         }
     }

crates/manager-core/src/trigger_config.rs

@@ -61,6 +61,18 @@ pub struct TriggerConfig {
     /// real-world cron precision is per-minute, so a 30s tick is fine.
     /// Floored at 1s by the scheduler.
     pub cron_tick_interval_ms: u32,
+
+    /// Queue visibility-timeout reclaim task cadence, in ms (v1.1.9).
+    /// Default 30 000. The reclaim task clears claims on
+    /// `queue_messages` whose `claimed_at` exceeds the per-queue
+    /// `visibility_timeout_secs`, so a crashed consumer doesn't lose
+    /// the message.
+    pub queue_reclaim_interval_ms: u32,
+
+    /// Default per-queue visibility-timeout in seconds (v1.1.9). Used
+    /// at queue-trigger creation when the request omits an explicit
+    /// value. Default 30. The per-trigger column overrides this.
+    pub queue_default_visibility_timeout_secs: u32,
 }
 
 impl TriggerConfig {
@@ -75,6 +87,8 @@ impl TriggerConfig {
             dead_letter_retention_days: 30,
             abandoned_retention_days: 7,
             cron_tick_interval_ms: 30_000,
+            queue_reclaim_interval_ms: 30_000,
+            queue_default_visibility_timeout_secs: 30,
         }
     }
 
@@ -101,6 +115,14 @@ impl TriggerConfig {
             &mut c.cron_tick_interval_ms,
             "PICLOUD_CRON_TICK_INTERVAL_MS",
         );
+        load_u32(
+            &mut c.queue_reclaim_interval_ms,
+            "PICLOUD_QUEUE_RECLAIM_INTERVAL_MS",
+        );
+        load_u32(
+            &mut c.queue_default_visibility_timeout_secs,
+            "PICLOUD_QUEUE_DEFAULT_VISIBILITY_TIMEOUT_SECS",
+        );
         c
     }
 }

crates/manager-core/src/trigger_repo.rs

@@ -59,6 +59,9 @@ pub enum TriggerKind {
     Pubsub,
     /// v1.1.7. Inbound email via the webhook receiver.
     Email,
+    /// v1.1.9. Durable queue consumer — exactly one consumer per
+    /// `(app_id, queue_name)`.
+    Queue,
 }
 
 impl TriggerKind {
@@ -72,6 +75,7 @@ impl TriggerKind {
             Self::Files => "files",
             Self::Pubsub => "pubsub",
             Self::Email => "email",
+            Self::Queue => "queue",
         }
     }
 
@@ -85,6 +89,7 @@ impl TriggerKind {
             "files" => Some(Self::Files),
             "pubsub" => Some(Self::Pubsub),
             "email" => Some(Self::Email),
+            "queue" => Some(Self::Queue),
             _ => None,
         }
     }
@@ -145,6 +150,15 @@ pub enum TriggerDetails {
     /// surfaced (it's encrypted at rest); we expose only whether one is
     /// configured so the admin UI can show "signed" vs "unsigned".
     Email { has_inbound_secret: bool },
+    /// v1.1.9. Queue consumer. The queue name is the per-queue grouping
+    /// key; visibility timeout is the per-trigger override; last_fired_at
+    /// is updated by the dispatcher when a message is claimed.
+    Queue {
+        queue_name: String,
+        visibility_timeout_secs: u32,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        last_fired_at: Option<DateTime<Utc>>,
+    },
 }
 
 /// Create payload for a KV trigger. Defaults applied at the admin
@@ -251,6 +265,35 @@ pub struct CreateEmailTrigger {
     pub registered_by_principal: AdminUserId,
 }
 
+/// Create payload for a queue:receive trigger (v1.1.9). One consumer
+/// per `(app_id, queue_name)` is enforced inside `create_queue_trigger`
+/// via a pg_advisory_xact_lock + SELECT.
+#[derive(Debug, Clone)]
+pub struct CreateQueueTrigger {
+    pub script_id: ScriptId,
+    pub queue_name: String,
+    pub visibility_timeout_secs: u32,
+    pub dispatch_mode: TriggerDispatchMode,
+    pub retry_max_attempts: u32,
+    pub retry_backoff: BackoffShape,
+    pub retry_base_ms: u32,
+    pub registered_by_principal: AdminUserId,
+}
+
+/// One consumer the dispatcher's queue arm needs to scan messages for.
+/// Pulled by `list_active_queue_consumers(app_id)` per tick.
+#[derive(Debug, Clone)]
+pub struct ActiveQueueConsumer {
+    pub trigger_id: TriggerId,
+    pub script_id: ScriptId,
+    pub queue_name: String,
+    pub visibility_timeout_secs: u32,
+    pub retry_max_attempts: u32,
+    pub retry_backoff: BackoffShape,
+    pub retry_base_ms: u32,
+    pub registered_by_principal: AdminUserId,
+}
+
 /// What the inbound-email webhook receiver needs to verify + dispatch a
 /// POST. Returned by `email_inbound_target`; `None` when the trigger
 /// doesn't exist or isn't `kind = 'email'`.
@@ -412,6 +455,34 @@ pub trait TriggerRepo: Send + Sync {
         trigger_id: Option<TriggerId>,
         script_id: Option<ScriptId>,
     ) -> Result<Vec<DeadLetterTriggerMatch>, TriggerRepoError>;
+
+    /// v1.1.9. Create a queue:receive trigger. Enforces exactly one
+    /// consumer per `(app_id, queue_name)` via `pg_advisory_xact_lock`
+    /// + SELECT-then-INSERT (a partial unique index across the parent
+    /// app_id + detail queue_name can't be enforced in a single index).
+    /// Returns `TriggerRepoError::Invalid` if a consumer already exists.
+    async fn create_queue_trigger(
+        &self,
+        app_id: AppId,
+        req: CreateQueueTrigger,
+    ) -> Result<Trigger, TriggerRepoError>;
+
+    /// v1.1.9. Dispatcher's queue arm: list every enabled queue:receive
+    /// trigger across all apps. The dispatcher walks these per tick and
+    /// attempts a claim per queue. Bounded by the number of registered
+    /// consumers, which is small.
+    async fn list_active_queue_consumers(
+        &self,
+    ) -> Result<Vec<ActiveQueueConsumer>, TriggerRepoError>;
+
+    /// v1.1.9. Update `last_fired_at` on a queue trigger after a
+    /// successful claim. Best-effort — a failure is logged but doesn't
+    /// fail the dispatch.
+    async fn touch_queue_trigger_last_fired_at(
+        &self,
+        trigger_id: TriggerId,
+        at: DateTime<Utc>,
+    ) -> Result<(), TriggerRepoError>;
 }
 
 // ----------------------------------------------------------------------------
@@ -1113,6 +1184,167 @@ impl TriggerRepo for PostgresTriggerRepo {
             })
             .collect())
     }
+
+    async fn create_queue_trigger(
+        &self,
+        app_id: AppId,
+        req: CreateQueueTrigger,
+    ) -> Result<Trigger, TriggerRepoError> {
+        if req.queue_name.is_empty() {
+            return Err(TriggerRepoError::Invalid(
+                "queue_name must not be empty".into(),
+            ));
+        }
+        if !(5..=3600).contains(&req.visibility_timeout_secs) {
+            return Err(TriggerRepoError::Invalid(
+                "visibility_timeout_secs must be in [5, 3600]".into(),
+            ));
+        }
+
+        let mut tx = self.pool.begin().await?;
+
+        // Per-(app_id, queue_name) advisory lock serializes concurrent
+        // creates so the SELECT-then-INSERT below is atomic. The lock
+        // is per-transaction (xact_lock) and automatically released on
+        // commit/rollback.
+        let lock_key = advisory_lock_key(app_id, &req.queue_name);
+        sqlx::query("SELECT pg_advisory_xact_lock($1)")
+            .bind(lock_key)
+            .execute(&mut *tx)
+            .await?;
+
+        // Reject duplicate consumer (one-per-queue invariant).
+        let existing: Option<(Uuid,)> = sqlx::query_as(
+            "SELECT t.id FROM triggers t \
+             JOIN queue_trigger_details d ON d.trigger_id = t.id \
+             WHERE t.app_id = $1 AND t.kind = 'queue' \
+               AND d.queue_name = $2",
+        )
+        .bind(app_id.into_inner())
+        .bind(&req.queue_name)
+        .fetch_optional(&mut *tx)
+        .await?;
+        if existing.is_some() {
+            return Err(TriggerRepoError::Invalid(format!(
+                "queue '{}' already has a consumer trigger; remove the existing one first",
+                req.queue_name
+            )));
+        }
+
+        let parent: TriggerRow = sqlx::query_as(
+            "INSERT INTO triggers ( \
+                app_id, script_id, kind, enabled, dispatch_mode, \
+                retry_max_attempts, retry_backoff, retry_base_ms, \
+                registered_by_principal \
+             ) VALUES ($1, $2, 'queue', TRUE, $3, $4, $5, $6, $7) \
+             RETURNING id, app_id, script_id, kind, enabled, dispatch_mode, \
+                       retry_max_attempts, retry_backoff, retry_base_ms, \
+                       registered_by_principal, created_at, updated_at",
+        )
+        .bind(app_id.into_inner())
+        .bind(req.script_id.into_inner())
+        .bind(req.dispatch_mode.as_str())
+        .bind(i32::try_from(req.retry_max_attempts).unwrap_or(3))
+        .bind(req.retry_backoff.as_str())
+        .bind(i32::try_from(req.retry_base_ms).unwrap_or(1000))
+        .bind(req.registered_by_principal.into_inner())
+        .fetch_one(&mut *tx)
+        .await?;
+
+        sqlx::query(
+            "INSERT INTO queue_trigger_details \
+                (trigger_id, queue_name, visibility_timeout_secs) \
+             VALUES ($1, $2, $3)",
+        )
+        .bind(parent.id)
+        .bind(&req.queue_name)
+        .bind(i32::try_from(req.visibility_timeout_secs).unwrap_or(30))
+        .execute(&mut *tx)
+        .await?;
+
+        tx.commit().await?;
+
+        Ok(Trigger {
+            id: parent.id.into(),
+            app_id: parent.app_id.into(),
+            script_id: parent.script_id.into(),
+            kind: TriggerKind::Queue,
+            enabled: parent.enabled,
+            dispatch_mode: dispatch_from_str(&parent.dispatch_mode),
+            retry_max_attempts: u32::try_from(parent.retry_max_attempts).unwrap_or(3),
+            retry_backoff: BackoffShape::from_wire(&parent.retry_backoff)
+                .unwrap_or(BackoffShape::Exponential),
+            retry_base_ms: u32::try_from(parent.retry_base_ms).unwrap_or(1000),
+            registered_by_principal: parent.registered_by_principal.into(),
+            created_at: parent.created_at,
+            updated_at: parent.updated_at,
+            details: TriggerDetails::Queue {
+                queue_name: req.queue_name,
+                visibility_timeout_secs: req.visibility_timeout_secs,
+                last_fired_at: None,
+            },
+        })
+    }
+
+    async fn list_active_queue_consumers(
+        &self,
+    ) -> Result<Vec<ActiveQueueConsumer>, TriggerRepoError> {
+        let rows: Vec<QueueConsumerRow> = sqlx::query_as(
+            "SELECT t.id AS trigger_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 \
+             FROM triggers t \
+             JOIN queue_trigger_details d ON d.trigger_id = t.id \
+             WHERE t.kind = 'queue' AND t.enabled = TRUE",
+        )
+        .fetch_all(&self.pool)
+        .await?;
+        Ok(rows
+            .into_iter()
+            .map(|r| ActiveQueueConsumer {
+                trigger_id: r.trigger_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),
+                retry_max_attempts: u32::try_from(r.retry_max_attempts).unwrap_or(3),
+                retry_backoff: BackoffShape::from_wire(&r.retry_backoff)
+                    .unwrap_or(BackoffShape::Exponential),
+                retry_base_ms: u32::try_from(r.retry_base_ms).unwrap_or(1000),
+                registered_by_principal: r.registered_by_principal.into(),
+            })
+            .collect())
+    }
+
+    async fn touch_queue_trigger_last_fired_at(
+        &self,
+        trigger_id: TriggerId,
+        at: DateTime<Utc>,
+    ) -> Result<(), TriggerRepoError> {
+        sqlx::query(
+            "UPDATE queue_trigger_details SET last_fired_at = $2 WHERE trigger_id = $1",
+        )
+        .bind(trigger_id.into_inner())
+        .bind(at)
+        .execute(&self.pool)
+        .await?;
+        Ok(())
+    }
+}
+
+/// Stable per-(app_id, queue_name) i64 derived from the UUID bytes +
+/// the queue name. Used as the key for `pg_advisory_xact_lock` so
+/// concurrent `create_queue_trigger` calls for the same queue serialize.
+fn advisory_lock_key(app_id: AppId, queue_name: &str) -> i64 {
+    use std::collections::hash_map::DefaultHasher;
+    use std::hash::{Hash, Hasher};
+    let mut h = DefaultHasher::new();
+    app_id.into_inner().hash(&mut h);
+    queue_name.hash(&mut h);
+    // Postgres advisory locks key on bigint; reinterpret the u64 as i64.
+    #[allow(clippy::cast_possible_wrap)]
+    let key = h.finish() as i64;
+    key
 }
 
 #[allow(clippy::too_many_lines)]
@@ -1223,6 +1455,20 @@ async fn hydrate_one(pool: &PgPool, parent: TriggerRow) -> Result<Trigger, Trigg
                 has_inbound_secret: row.inbound_secret_encrypted.is_some(),
             }
         }
+        TriggerKind::Queue => {
+            let row: QueueDetailRow = sqlx::query_as(
+                "SELECT queue_name, visibility_timeout_secs, last_fired_at \
+                 FROM queue_trigger_details WHERE trigger_id = $1",
+            )
+            .bind(parent.id)
+            .fetch_one(pool)
+            .await?;
+            TriggerDetails::Queue {
+                queue_name: row.queue_name,
+                visibility_timeout_secs: u32::try_from(row.visibility_timeout_secs).unwrap_or(30),
+                last_fired_at: row.last_fired_at,
+            }
+        }
     };
 
     Ok(Trigger {
@@ -1305,6 +1551,25 @@ struct EmailDetailRow {
     inbound_secret_encrypted: Option<Vec<u8>>,
 }
 
+#[derive(sqlx::FromRow)]
+struct QueueDetailRow {
+    queue_name: String,
+    visibility_timeout_secs: i32,
+    last_fired_at: Option<DateTime<Utc>>,
+}
+
+#[derive(sqlx::FromRow)]
+struct QueueConsumerRow {
+    trigger_id: Uuid,
+    script_id: Uuid,
+    queue_name: String,
+    visibility_timeout_secs: i32,
+    retry_max_attempts: i32,
+    retry_backoff: String,
+    retry_base_ms: i32,
+    registered_by_principal: Uuid,
+}
+
 #[derive(sqlx::FromRow)]
 struct EmailInboundRow {
     app_id: Uuid,
@@ -1365,4 +1630,31 @@ mod tests {
         assert!(collection_matches("widgets", "widgets"));
         assert!(!collection_matches("widgets", "Widgets"));
     }
+
+    #[test]
+    fn trigger_kind_queue_round_trips_through_wire() {
+        assert_eq!(TriggerKind::Queue.as_str(), "queue");
+        assert_eq!(TriggerKind::from_wire("queue"), Some(TriggerKind::Queue));
+    }
+
+    #[test]
+    fn advisory_lock_key_is_stable_per_app_queue_pair() {
+        let app_id_a: AppId = uuid::Uuid::nil().into();
+        let app_id_b: AppId = uuid::Uuid::from_u128(1).into();
+        // Same inputs → same key.
+        assert_eq!(
+            advisory_lock_key(app_id_a, "jobs"),
+            advisory_lock_key(app_id_a, "jobs")
+        );
+        // Different app → different key (extremely high probability).
+        assert_ne!(
+            advisory_lock_key(app_id_a, "jobs"),
+            advisory_lock_key(app_id_b, "jobs")
+        );
+        // Different queue → different key.
+        assert_ne!(
+            advisory_lock_key(app_id_a, "jobs"),
+            advisory_lock_key(app_id_a, "other")
+        );
+    }
 }

crates/manager-core/src/triggers_api.rs

@@ -940,6 +940,47 @@ mod tests {
         ) -> Result<Vec<DeadLetterTriggerMatch>, TriggerRepoError> {
             Ok(vec![])
         }
+        async fn create_queue_trigger(
+            &self,
+            app_id: AppId,
+            req: crate::trigger_repo::CreateQueueTrigger,
+        ) -> Result<Trigger, TriggerRepoError> {
+            let now = Utc::now();
+            let id = TriggerId::new();
+            let trigger = Trigger {
+                id,
+                app_id,
+                script_id: req.script_id,
+                kind: TriggerKind::Queue,
+                enabled: true,
+                dispatch_mode: req.dispatch_mode,
+                retry_max_attempts: req.retry_max_attempts,
+                retry_backoff: req.retry_backoff,
+                retry_base_ms: req.retry_base_ms,
+                registered_by_principal: req.registered_by_principal,
+                created_at: now,
+                updated_at: now,
+                details: TriggerDetails::Queue {
+                    queue_name: req.queue_name,
+                    visibility_timeout_secs: req.visibility_timeout_secs,
+                    last_fired_at: None,
+                },
+            };
+            self.inner.lock().await.insert(id, trigger.clone());
+            Ok(trigger)
+        }
+        async fn list_active_queue_consumers(
+            &self,
+        ) -> Result<Vec<crate::trigger_repo::ActiveQueueConsumer>, TriggerRepoError> {
+            Ok(vec![])
+        }
+        async fn touch_queue_trigger_last_fired_at(
+            &self,
+            _trigger_id: TriggerId,
+            _at: chrono::DateTime<chrono::Utc>,
+        ) -> Result<(), TriggerRepoError> {
+            Ok(())
+        }
     }
 
     struct InMemoryAppRepo {