feat(shared-queues): materialized competing consumers + dispatcher branch (D3.2)

The consumption side of shared durable queues:
- Thread `shared` through BundleTrigger::Queue + QueueTriggerSpec + the
  trigger identity; validate_bundle_for requires a shared queue on a group
  to name a declared kind='queue' collection (a shared queue on an app is
  rejected by the existing app-owner shared guard).
- materialize: a shared queue template materializes a consumer per
  descendant (the M5 one-consumer-slot skip is bypassed for shared —
  competing consumers are intended; each descendant gets one copy).
- dispatcher: ActiveQueueConsumer gains shared_group (from the materialized
  copy's source template via LEFT JOIN); dispatch_one_queue +
  handle_queue_failure route claim/ack/nack/terminal to the group store
  when shared_group is Some, via q_claim/q_ack/q_nack/q_terminal helpers. A
  group claim is normalized to a ClaimedMessage under the consuming app so
  the handler path is unchanged; the reclaim task also drains the group
  store. Exhausted shared messages are dropped (no group dead-letter store
  yet — documented deferral).

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

crates/manager-core/src/apply_service.rs

@@ -260,6 +260,10 @@ pub enum BundleTrigger {
         dispatch_mode: Option<TriggerDispatchMode>,
         #[serde(default)]
         retry_max_attempts: Option<u32>,
+        /// §11.6 D3: `true` for a shared-QUEUE group consumer — materializes a
+        /// competing consumer per descendant app, all claiming the group store.
+        #[serde(default)]
+        shared: bool,
     },
 }
 
@@ -313,8 +317,9 @@ impl BundleTrigger {
             Self::Kv { shared, .. }
             | Self::Docs { shared, .. }
             | Self::Files { shared, .. }
-            | Self::Pubsub { shared, .. } => *shared,
+            | Self::Pubsub { shared, .. }
-            Self::Cron { .. } | Self::Email { .. } | Self::Queue { .. } => false,
+            | Self::Queue { shared, .. } => *shared,
+            Self::Cron { .. } | Self::Email { .. } => false,
         }
     }
 
@@ -390,7 +395,9 @@ impl BundleTrigger {
                 ..
             } => format!("pubsub|{script}|{topic_pattern}|{sealed}|{shared}"),
             Self::Email { script, .. } => format!("email|{script}"),
-            Self::Queue { queue_name, .. } => format!("queue|{queue_name}"),
+            Self::Queue {
+                queue_name, shared, ..
+            } => format!("queue|{queue_name}|{shared}"),
         }
     }
 
@@ -787,9 +794,10 @@ impl ApplyService {
                         BundleTrigger::Pubsub { topic_pattern, .. } => {
                             (topic_pattern.split('.').next().unwrap_or(""), "topic")
                         }
+                        BundleTrigger::Queue { queue_name, .. } => (queue_name.as_str(), "queue"),
                         _ => {
                             return Err(ApplyError::Invalid(format!(
-                                "a `shared` trigger must be a kv/docs/files/pubsub kind; \
+                                "a `shared` trigger must be a kv/docs/files/pubsub/queue kind; \
                                  `{}` has no shared collection store",
                                 t.kind_str()
                             )));
@@ -3416,7 +3424,7 @@ fn current_trigger_identity(t: &Trigger, name_by_id: &HashMap<ScriptId, String>)
             Some(format!("pubsub|{script}|{topic_pattern}|{sealed}|{shared}"))
         }
         TriggerDetails::Email { .. } => Some(format!("email|{script}")),
-        TriggerDetails::Queue { queue_name, .. } => Some(format!("queue|{queue_name}")),
+        TriggerDetails::Queue { queue_name, .. } => Some(format!("queue|{queue_name}|{shared}")),
         TriggerDetails::DeadLetter { .. } => None,
     }
 }
@@ -4254,6 +4262,7 @@ mod tests {
             visibility_timeout_secs: vis,
             dispatch_mode: None,
             retry_max_attempts: None,
+            shared: false,
         };
         assert!(validate_trigger_shape(&queue(Some(10))).is_err());
         assert!(validate_trigger_shape(&queue(Some(
@@ -4715,6 +4724,7 @@ mod tests {
             visibility_timeout_secs: None,
             dispatch_mode: None,
             retry_max_attempts: None,
+            shared: false,
         }];
         let p = compute_diff(&current, &b);
         assert!(

crates/manager-core/src/dispatcher.rs

@@ -31,7 +31,8 @@ use picloud_orchestrator_core::{ExecutionGate, ExecutorClient};
 use picloud_shared::{
     AppId, DeadLetterId, ExecResponseSummary, ExecutionId, ExecutionLogSink, ExecutionSource,
     HttpDispatchPayload, InboxDeliveryOutcome, InboxFailureKind, InboxResolver, InboxResult,
-    RequestId, Script, ScriptId, ScriptOwner, ScriptSandbox, TriggerEvent,
+    QueueMessageId, RequestId, Script, ScriptId, ScriptOwner, ScriptSandbox, TriggerEvent,
+    TriggerId,
 };
 use rand::Rng;
 use uuid::Uuid;
@@ -65,6 +66,10 @@ pub struct Dispatcher {
     /// 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
@@ -225,6 +230,7 @@ impl Dispatcher {
         // 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 {
@@ -237,6 +243,14 @@ impl Dispatcher {
                     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"),
+                }
             }
         });
 
@@ -313,6 +327,126 @@ impl Dispatcher {
         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 (+ the caller
+    /// fans out `dead_letter` triggers). Group shared queue → drop the row
+    /// (no group dead-letter store yet — documented D3 deferral) + warn.
+    /// 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(_) => {
+                if let Err(e) = self
+                    .group_queue
+                    .drop_exhausted(claimed.id, claimed.claim_token)
+                    .await
+                {
+                    tracing::warn!(?e, "shared-queue drop failed");
+                }
+                tracing::warn!(
+                    reason,
+                    queue = %claimed.queue_name,
+                    "shared-queue message dropped (no group dead-letter store yet)"
+                );
+                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,
@@ -320,10 +454,9 @@ impl Dispatcher {
     ) -> Result<(), DispatcherError> {
         // Atomic claim — None → nothing pending right now for this queue.
         let Some(claimed) = self
-            .queue
+            .q_claim(consumer)
-            .claim(consumer.app_id, &consumer.queue_name)
             .await
-            .map_err(|e| DispatcherError::Outbox(e.to_string()))?
+            .map_err(DispatcherError::Outbox)?
         else {
             return Ok(());
         };
@@ -333,8 +466,8 @@ impl Dispatcher {
         // outbox arm.
         let Ok(permit) = self.gate.try_acquire() else {
             let _ = self
-                .queue
+                .q_nack(
-                .nack(
+                    consumer,
                     claimed.id,
                     claimed.claim_token,
                     chrono::Duration::milliseconds(100),
@@ -359,16 +492,11 @@ impl Dispatcher {
             Ok(None) => {
                 tracing::warn!(script_id = %consumer.script_id, "queue trigger script missing; dead-lettering");
                 let _ = self
-                    .queue
+                    .q_terminal(
-                    .dead_letter(
+                        consumer,
-                        claimed.id,
+                        &claimed,
-                        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;
@@ -398,16 +526,11 @@ impl Dispatcher {
                 "queue consumer script belongs to a different app; dead-lettering"
             );
             let _ = self
-                .queue
+                .q_terminal(
-                .dead_letter(
+                    consumer,
-                    claimed.id,
+                    &claimed,
-                    claimed.claim_token,
-                    claimed.app_id,
-                    &claimed.queue_name,
                     Some(consumer.trigger_id),
                     Some(consumer.script_id),
-                    claimed.attempt,
-                    claimed.enqueued_at,
                     "queue consumer target belongs to a different app",
                 )
                 .await;
@@ -434,8 +557,8 @@ impl Dispatcher {
                 "queue consumer script disabled at fire time; releasing claim"
             );
             if let Err(e) = self
-                .queue
+                .q_nack(
-                .nack(
+                    consumer,
                     claimed.id,
                     claimed.claim_token,
                     chrono::Duration::seconds(1),
@@ -515,7 +638,7 @@ impl Dispatcher {
         match outcome {
             Ok(_) => {
                 // Auto-ack on success.
-                if let Err(e) = self.queue.ack(claimed.id, claimed.claim_token).await {
+                if let Err(e) = self.q_ack(consumer, claimed.id, claimed.claim_token).await {
                     tracing::warn!(?e, "queue ack failed");
                 }
             }
@@ -542,8 +665,7 @@ impl Dispatcher {
             );
             let delay = chrono::Duration::milliseconds(i64::from(delay_ms));
             if let Err(e) = self
-                .queue
+                .q_nack(consumer, claimed.id, claimed.claim_token, delay)
-                .nack(claimed.id, claimed.claim_token, delay)
                 .await
             {
                 tracing::warn!(?e, "queue nack failed");
@@ -558,27 +680,18 @@ impl Dispatcher {
         // same way here.
         let now = Utc::now();
         let last_error = err.to_string();
-        let dl_id = match self
+        // Per-app → dead-letter (+ fan out below). Shared group queue → dropped
-            .queue
+        // inside q_terminal (no group dead-letter store yet), returns None so
-            .dead_letter(
+        // the fan-out is skipped.
-                claimed.id,
+        let dl_id = self
-                claimed.claim_token,
+            .q_terminal(
-                claimed.app_id,
+                consumer,
-                &claimed.queue_name,
+                claimed,
                 Some(consumer.trigger_id),
                 Some(consumer.script_id),
-                claimed.attempt,
-                claimed.enqueued_at,
                 &last_error,
             )
-            .await
+            .await;
-        {
-            Ok(dl_id) => Some(dl_id),
-            Err(e) => {
-                tracing::error!(?e, "queue dead-letter write failed");
-                None
-            }
-        };
         if let Some(dead_letter_id) = dl_id {
             let original = TriggerEvent::Queue {
                 queue_name: claimed.queue_name.clone(),
@@ -1512,6 +1625,72 @@ fn apply_jitter(raw: u32, pct: u32) -> u32 {
 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)
+        }
+        async fn drop_exhausted(
+            &self,
+            _id: QueueMessageId,
+            _token: Uuid,
+        ) -> Result<bool, crate::group_queue_repo::GroupQueueRepoError> {
+            Ok(false)
+        }
+        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.
@@ -1778,14 +1957,14 @@ mod tests {
             }
             async fn ack(
                 &self,
-                _message_id: picloud_shared::QueueMessageId,
+                _message_id: QueueMessageId,
                 _claim_token: Uuid,
             ) -> Result<bool, crate::queue_repo::QueueRepoError> {
                 unimplemented!("not used by this test")
             }
             async fn nack(
                 &self,
-                _message_id: picloud_shared::QueueMessageId,
+                _message_id: QueueMessageId,
                 _claim_token: Uuid,
                 _retry_delay: chrono::Duration,
             ) -> Result<bool, crate::queue_repo::QueueRepoError> {
@@ -1820,7 +1999,7 @@ mod tests {
             #[allow(clippy::too_many_arguments)]
             async fn dead_letter(
                 &self,
-                _message_id: picloud_shared::QueueMessageId,
+                _message_id: QueueMessageId,
                 _claim_token: Uuid,
                 _app_id: AppId,
                 _queue_name: &str,
@@ -2205,6 +2384,7 @@ mod tests {
                 retry_backoff: BackoffShape::Exponential,
                 retry_base_ms: 1000,
                 registered_by_principal: AdminUserId::new(),
+                shared_group: None,
             };
 
             let nacked = Arc::new(AtomicBool::new(false));
@@ -2229,6 +2409,7 @@ mod tests {
                     claimed,
                     nacked: nacked.clone(),
                 }),
+                group_queue: Arc::new(NoopGroupQueue),
                 config: TriggerConfig::from_env(),
                 instance_id: "test-instance".into(),
             };
@@ -2310,14 +2491,14 @@ mod tests {
             }
             async fn ack(
                 &self,
-                _message_id: picloud_shared::QueueMessageId,
+                _message_id: QueueMessageId,
                 _claim_token: Uuid,
             ) -> Result<bool, crate::queue_repo::QueueRepoError> {
                 unimplemented!("not used by this test")
             }
             async fn nack(
                 &self,
-                _message_id: picloud_shared::QueueMessageId,
+                _message_id: QueueMessageId,
                 _claim_token: Uuid,
                 _retry_delay: chrono::Duration,
             ) -> Result<bool, crate::queue_repo::QueueRepoError> {
@@ -2354,7 +2535,7 @@ mod tests {
             #[allow(clippy::too_many_arguments)]
             async fn dead_letter(
                 &self,
-                _message_id: picloud_shared::QueueMessageId,
+                _message_id: QueueMessageId,
                 _claim_token: Uuid,
                 _app_id: AppId,
                 _queue_name: &str,
@@ -2461,6 +2642,7 @@ mod tests {
                 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(),
             };

crates/manager-core/src/materialize.rs

@@ -44,6 +44,10 @@ struct ShouldRow {
     effective_app_id: Uuid,
     template_id: Uuid,
     kind: String,
+    /// §11.6 D3: a `shared` queue template's copies drain the GROUP store as
+    /// COMPETING consumers, so the one-consumer-per-(app, queue) slot check is
+    /// skipped for them (each descendant intentionally gets a consumer).
+    shared: bool,
 }
 
 /// Reconcile all materialized stateful-template copies. Idempotent; safe to call
@@ -79,7 +83,7 @@ pub async fn rematerialize_stateful_templates(pool: &PgPool) -> Result<Vec<Strin
              FROM groups g JOIN app_chain ac ON g.id = ac.next_group \
              WHERE ac.depth < 64 AND g.parent_id IS NOT NULL \
          ) \
-         SELECT DISTINCT ac.effective_app_id, t.id AS template_id, t.kind \
+         SELECT DISTINCT ac.effective_app_id, t.id AS template_id, t.kind, t.shared \
          FROM app_chain ac \
          JOIN triggers t ON t.group_id = ac.owner_group \
          WHERE t.group_id IS NOT NULL AND t.enabled = TRUE AND t.kind = ANY($1)",
@@ -118,8 +122,14 @@ pub async fn rematerialize_stateful_templates(pool: &PgPool) -> Result<Vec<Strin
         if existing_set.contains(&(r.effective_app_id, r.template_id)) {
             continue;
         }
-        if let Some(w) =
+        if let Some(w) = materialize_one(
-            materialize_one(&mut tx, r.effective_app_id, r.template_id, &r.kind).await?
+            &mut tx,
+            r.effective_app_id,
+            r.template_id,
+            &r.kind,
+            r.shared,
+        )
+        .await?
         {
             warnings.push(w);
         }
@@ -138,11 +148,14 @@ async fn materialize_one(
     app_id: Uuid,
     template_id: Uuid,
     kind: &str,
+    shared: bool,
 ) -> Result<Option<String>, sqlx::Error> {
-    // §M5.4: a queue copy would violate the one-consumer-per-(app_id, queue_name)
+    // §M5.4: a per-app queue copy would violate the one-consumer-per-(app_id,
-    // invariant if the app already has a consumer (hand-authored or another
+    // queue_name) invariant if the app already has a consumer on that queue —
-    // template) on that queue — skip with a warning.
+    // skip with a warning. §11.6 D3: a SHARED queue copy drains the group store
-    if kind == "queue" {
+    // as a competing consumer (a different store), so the slot check is skipped
+    // — every descendant intentionally gets a consumer.
+    if kind == "queue" && !shared {
         let taken: Option<(Uuid,)> = sqlx::query_as(
             "SELECT t.id FROM triggers t \
              JOIN queue_trigger_details d ON d.trigger_id = t.id \

crates/manager-core/src/trigger_repo.rs

@@ -337,6 +337,11 @@ pub struct ActiveQueueConsumer {
     pub retry_backoff: BackoffShape,
     pub retry_base_ms: u32,
     pub registered_by_principal: AdminUserId,
+    /// §11.6 D3: `Some(group)` when this consumer is a materialized copy of a
+    /// SHARED group queue template — it drains `group_queue_messages(group,
+    /// queue_name)` (competing consumers) instead of the per-app store. `None`
+    /// for an ordinary per-app or non-shared-template consumer.
+    pub shared_group: Option<GroupId>,
 }
 
 /// What the inbound-email webhook receiver needs to verify + dispatch a
@@ -1792,13 +1797,19 @@ impl TriggerRepo for PostgresTriggerRepo {
         &self,
     ) -> Result<Vec<ActiveQueueConsumer>, TriggerRepoError> {
         let rows: Vec<QueueConsumerRow> = sqlx::query_as(
+            // §11.6 D3: LEFT JOIN the SOURCE template of a materialized copy; if
+            // that template is a SHARED group queue, `shared_group` is its
+            // owning group and this consumer drains the group store instead of
+            // the per-app one.
             "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 \
+                    t.registered_by_principal, tmpl.group_id AS shared_group \
              FROM triggers t \
              JOIN queue_trigger_details d ON d.trigger_id = t.id \
              JOIN scripts s ON s.id = t.script_id \
+             LEFT JOIN triggers tmpl \
+                    ON tmpl.id = t.materialized_from AND tmpl.shared = TRUE \
              WHERE t.kind = 'queue' AND t.enabled = TRUE AND s.enabled = TRUE \
                AND t.app_id IS NOT NULL",
         )
@@ -1817,6 +1828,7 @@ impl TriggerRepo for PostgresTriggerRepo {
                     .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(),
+                shared_group: r.shared_group.map(Into::into),
             })
             .collect())
     }
@@ -2096,6 +2108,9 @@ struct QueueConsumerRow {
     retry_backoff: String,
     retry_base_ms: i32,
     registered_by_principal: Uuid,
+    // §11.6 D3: the owning group of a SHARED queue template this consumer was
+    // materialized from; NULL for a per-app / non-shared consumer.
+    shared_group: Option<Uuid>,
 }
 
 #[derive(sqlx::FromRow)]

crates/manager-core/tests/group_queue.rs

@@ -8,7 +8,7 @@
 //! Deterministic: drives `GroupQueueRepo` directly (no dispatcher). Skips when
 //! `DATABASE_URL` is unset.
 
-#![allow(clippy::too_many_lines)]
+#![allow(clippy::too_many_lines, clippy::many_single_char_names)]
 
 use std::collections::HashSet;
 
@@ -52,8 +52,8 @@ async fn competing_consumers_claim_each_message_exactly_once() {
     let repo = PostgresGroupQueueRepo::new(pool.clone());
 
     // Enqueue 50 distinct messages into the shared `tasks` queue.
-    const N: usize = 50;
+    let n_msgs: usize = 50;
-    for i in 0..N {
+    for i in 0..n_msgs {
         repo.enqueue(NewGroupQueueMessage {
             group_id: group,
             collection: "tasks".into(),
@@ -70,14 +70,9 @@ async fn competing_consumers_claim_each_message_exactly_once() {
     // Every claimed payload id must be unique — no message delivered twice.
     let claim_loop = |repo: PostgresGroupQueueRepo, group: GroupId| async move {
         let mut got: Vec<i64> = Vec::new();
-        loop {
+        while let Some(msg) = repo.claim(group, "tasks").await.unwrap() {
-            match repo.claim(group, "tasks").await.unwrap() {
+            got.push(msg.payload["i"].as_i64().unwrap());
-                Some(msg) => {
+            assert!(repo.ack(msg.id, msg.claim_token).await.unwrap(), "ack ok");
-                    got.push(msg.payload["i"].as_i64().unwrap());
-                    assert!(repo.ack(msg.id, msg.claim_token).await.unwrap(), "ack ok");
-                }
-                None => break,
-            }
         }
         got
     };
@@ -96,10 +91,10 @@ async fn competing_consumers_claim_each_message_exactly_once() {
     let unique: HashSet<i64> = all.iter().copied().collect();
     assert_eq!(
         all.len(),
-        N,
+        n_msgs,
         "every message delivered exactly once (no dupes)"
     );
-    assert_eq!(unique.len(), N, "all N distinct ids covered");
+    assert_eq!(unique.len(), n_msgs, "all distinct ids covered");
     assert_eq!(
         repo.depth(group, "tasks").await.unwrap(),
         0,

crates/picloud-cli/src/cmds/pull.rs

@@ -220,6 +220,8 @@ pub async fn run(app_ident: &str, dir: &Path, force: bool, mode: OutputMode) ->
                     visibility_timeout_secs: Some(d.visibility_timeout_secs),
                     dispatch_mode,
                     retry_max_attempts,
+                    // app-owned triggers are never shared (group-only).
+                    shared: false,
                 });
             }
             // `email` is skipped: the server stores the sealed secret value,

crates/picloud-cli/src/manifest.rs

@@ -516,6 +516,11 @@ pub struct QueueTriggerSpec {
     pub dispatch_mode: Option<DispatchMode>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub retry_max_attempts: Option<u32>,
+    /// §11.6 D3: `true` for a shared-QUEUE group consumer over a declared
+    /// `kind = "queue"` collection — competing per-descendant consumers drain
+    /// one group-owned store. Group-only.
+    #[serde(default, skip_serializing_if = "is_false")]
+    pub shared: bool,
 }
 
 /// `[secrets] names = [...]` — declares which secrets the app expects.

crates/picloud/src/lib.rs

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