feat(triggers): shared dead-letter trigger fan-out (B2)

A group declares a declaratively-authored [[triggers.dead_letter]] shared=true
handler; when a message in the group's SHARED queue is exhausted, the
dispatcher's q_terminal group branch (after persisting to group_dead_letters)
fans out to it via list_matching_shared_dead_letter(owning_group, "queue", …),
each outbox row stamped the WRITER app_id (the consuming app — the M2 shared
-write model), so the handler runs under the consumer. The per-app
list_matching_dead_letter gained AND t.shared = FALSE (the shared flag is the
namespace boundary); the owning-group filter is the isolation boundary.

Adds BundleTrigger::DeadLetter + a DeadLetterTriggerSpec manifest kind (group
+shared only — validate_bundle_for rejects app-owned or non-shared, and exempts
it from the shared-requires-a-collection rule); insert_trigger_tx now accepts
dead_letter and writes dead_letter_trigger_details; current_trigger_identity
matches a group-shared dead_letter so re-apply is a NoOp (app-owned ones stay
diff-invisible). Pinned by tests/shared_dead_letter.rs (owning group matches,
per-app query does not, foreign group does not).

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

crates/manager-core/tests/shared_dead_letter.rs

@@ -0,0 +1,177 @@
+//! §11.6 B2 integration test: SHARED dead-letter triggers.
+//! A group-owned `dead_letter` trigger marked `shared = true` fires when a
+//! message in that group's SHARED queue is exhausted. It matches via the
+//! OWNING-group query (`list_matching_shared_dead_letter`); a descendant app's
+//! per-app dead-letter query (`list_matching_dead_letter`) does NOT match it
+//! (the `shared` flag is the namespace boundary), and a foreign sibling group
+//! does NOT match it (the owning-group filter is the isolation boundary).
+//!
+//! Deterministic: drives the repo match queries directly (no async dispatcher).
+//! Skips when `DATABASE_URL` is unset.
+
+#![allow(clippy::needless_pass_by_value, clippy::too_many_lines)]
+
+use picloud_manager_core::trigger_repo::{PostgresTriggerRepo, TriggerRepo};
+use picloud_shared::{AppId, GroupId};
+use sqlx::postgres::PgPoolOptions;
+use sqlx::PgPool;
+use uuid::Uuid;
+
+async fn pool_or_skip() -> Option<PgPool> {
+    let Ok(url) = std::env::var("DATABASE_URL") else {
+        picloud_test_support::abort_if_db_required("shared_dead_letter");
+        eprintln!("shared_dead_letter: DATABASE_URL unset — skipping");
+        return None;
+    };
+    let pool = PgPoolOptions::new()
+        .max_connections(2)
+        .connect(&url)
+        .await
+        .expect("connect");
+    sqlx::migrate!("./migrations")
+        .run(&pool)
+        .await
+        .expect("migrate");
+    Some(pool)
+}
+
+/// Insert a group-owned dead_letter trigger (`shared` flag) + its details.
+async fn dead_letter_trigger(
+    pool: &PgPool,
+    group_id: Uuid,
+    script: Uuid,
+    admin: Uuid,
+    shared: bool,
+) -> Uuid {
+    let row: (Uuid,) = sqlx::query_as(
+        "INSERT INTO triggers \
+            (app_id, group_id, script_id, kind, enabled, dispatch_mode, \
+             retry_max_attempts, retry_backoff, retry_base_ms, \
+             registered_by_principal, name, shared) \
+         VALUES (NULL, $1, $2, 'dead_letter', TRUE, 'async', 1, 'constant', 0, $3, $4, $5) \
+         RETURNING id",
+    )
+    .bind(group_id)
+    .bind(script)
+    .bind(admin)
+    .bind(Uuid::new_v4().simple().to_string())
+    .bind(shared)
+    .fetch_one(pool)
+    .await
+    .expect("trigger");
+    sqlx::query(
+        "INSERT INTO dead_letter_trigger_details \
+            (trigger_id, source_filter, trigger_id_filter, script_id_filter) \
+         VALUES ($1, NULL, NULL, NULL)",
+    )
+    .bind(row.0)
+    .execute(pool)
+    .await
+    .expect("details");
+    row.0
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn shared_dead_letter_matches_only_owning_group() {
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
+    let sfx = Uuid::new_v4().simple().to_string();
+    let admin = {
+        let r: (Uuid,) = sqlx::query_as(
+            "INSERT INTO admin_users (username, password_hash) VALUES ($1, 'x') RETURNING id",
+        )
+        .bind(format!("dl-{sfx}"))
+        .fetch_one(&pool)
+        .await
+        .unwrap();
+        r.0
+    };
+
+    // Group G with a group-owned handler + a SHARED dead_letter trigger.
+    let g: (Uuid,) = sqlx::query_as("INSERT INTO groups (slug, name) VALUES ($1, $1) RETURNING id")
+        .bind(format!("dl-g-{sfx}"))
+        .fetch_one(&pool)
+        .await
+        .unwrap();
+    let handler: (Uuid,) = sqlx::query_as(
+        "INSERT INTO scripts (name, source, group_id) VALUES ($1, 'x', $2) RETURNING id",
+    )
+    .bind(format!("on-dl-{sfx}"))
+    .bind(g.0)
+    .fetch_one(&pool)
+    .await
+    .unwrap();
+    let shared_trig = dead_letter_trigger(&pool, g.0, handler.0, admin, true).await;
+
+    // Descendant app A under G.
+    let a: (Uuid,) =
+        sqlx::query_as("INSERT INTO apps (slug, name, group_id) VALUES ($1, $1, $2) RETURNING id")
+            .bind(format!("dl-a-{sfx}"))
+            .bind(g.0)
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+
+    // Sibling group S (foreign — not an ancestor of A).
+    let s: (Uuid,) = sqlx::query_as("INSERT INTO groups (slug, name) VALUES ($1, $1) RETURNING id")
+        .bind(format!("dl-s-{sfx}"))
+        .fetch_one(&pool)
+        .await
+        .unwrap();
+
+    let trig = PostgresTriggerRepo::new(pool.clone());
+
+    // 1. The owning group G's shared query returns the trigger.
+    let g_matches = trig
+        .list_matching_shared_dead_letter(GroupId::from(g.0), "queue", None, None)
+        .await
+        .expect("shared match");
+    assert!(
+        g_matches.iter().any(|m| m.trigger_id == shared_trig.into()),
+        "the owning group's shared dead-letter query must match its shared trigger"
+    );
+
+    // 2. The per-app query on descendant A returns NOTHING — a shared group
+    //    template must not match the per-app (`shared = FALSE`) path.
+    let a_matches = trig
+        .list_matching_dead_letter(AppId::from(a.0), "queue", None, None)
+        .await
+        .expect("app match");
+    assert!(
+        !a_matches.iter().any(|m| m.trigger_id == shared_trig.into()),
+        "a per-app dead-letter query must NOT match the group's shared trigger"
+    );
+
+    // 3. A foreign sibling group S returns NOTHING (owning-group isolation).
+    let s_matches = trig
+        .list_matching_shared_dead_letter(GroupId::from(s.0), "queue", None, None)
+        .await
+        .expect("foreign match");
+    assert!(
+        !s_matches.iter().any(|m| m.trigger_id == shared_trig.into()),
+        "a foreign group's shared dead-letter query must NOT match another group's trigger"
+    );
+
+    // Cleanup.
+    let _ = sqlx::query("DELETE FROM triggers WHERE id = $1")
+        .bind(shared_trig)
+        .execute(&pool)
+        .await;
+    let _ = sqlx::query("DELETE FROM apps WHERE id = $1")
+        .bind(a.0)
+        .execute(&pool)
+        .await;
+    let _ = sqlx::query("DELETE FROM scripts WHERE id = $1")
+        .bind(handler.0)
+        .execute(&pool)
+        .await;
+    let _ = sqlx::query("DELETE FROM groups WHERE id = ANY($1)")
+        .bind(vec![g.0, s.0])
+        .execute(&pool)
+        .await;
+    let _ = sqlx::query("DELETE FROM admin_users WHERE id = $1")
+        .bind(admin)
+        .execute(&pool)
+        .await;
+}