feat(modules): live chain-union dispatch for group trigger templates (§11 tail T3)

The runtime heart: a descendant app's event now fires its ancestor
groups' trigger templates, resolved LIVE (no materialization).

- trigger_repo: list_matching_kv/docs/files prepend CHAIN_LEVELS_CTE and
  `JOIN chain c ON (t.app_id = c.app_owner OR t.group_id = c.group_owner)`
  (bind $1 = app_id), so each returns the app's own triggers PLUS
  ancestor-group templates in one statement. NULL-comparison semantics
  keep app/group rows from cross-matching; each row matches at exactly one
  chain depth (no dup fan-out).
- pubsub_repo: same union on the publish fan-out query.
- The outbox row stamps the firing app_id + the (group-owned) script_id;
  the dispatcher's existing script_invocable() (app-owned OR invocable via
  chain membership) already runs a group handler under a descendant app —
  the Phase-4 inheriting-app boundary, unchanged.

Glob/ops/topic filtering in Rust is untouched. 404 manager-core unit
tests green; clippy -D clean. Cross-subtree firing is journey-proven in T4.

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

crates/manager-core/src/pubsub_repo.rs

@@ -16,6 +16,8 @@ use picloud_shared::{topic_matches, AdminUserId, AppId, ExecutionId};
 use sqlx::PgPool;
 use uuid::Uuid;
 
+use crate::config_resolver::CHAIN_LEVELS_CTE;
+
 #[derive(Debug, thiserror::Error)]
 pub enum PubsubRepoError {
     #[error("database error: {0}")]
@@ -78,12 +80,18 @@ impl PubsubRepo for PostgresPubsubRepo {
         // Load all enabled pubsub triggers for the app; filter by topic
         // pattern in Rust (keeps the query simple, honours the
         // empty/`*`/prefix semantics without teaching SQL about globs).
-        let rows: Vec<PubsubTriggerRow> = sqlx::query_as(
+        // §11 tail: the chain union picks up the app's own pubsub triggers AND
-            "SELECT t.id, t.script_id, d.topic_pattern \
+        // ancestor-group pubsub TEMPLATES (live, no materialization). The outbox
+        // row below stamps the firing `ctx.app_id`, so a template runs under the
+        // publishing app — the inheriting-app boundary.
+        let rows: Vec<PubsubTriggerRow> = sqlx::query_as(&format!(
+            "{CHAIN_LEVELS_CTE} \
+             SELECT t.id, t.script_id, d.topic_pattern \
              FROM triggers t \
              JOIN pubsub_trigger_details d ON d.trigger_id = t.id \
-             WHERE t.app_id = $1 AND t.kind = 'pubsub' AND t.enabled = TRUE",
+             JOIN chain c ON (t.app_id = c.app_owner OR t.group_id = c.group_owner) \
-        )
+             WHERE t.kind = 'pubsub' AND t.enabled = TRUE"
+        ))
         .bind(ctx.app_id.into_inner())
         .fetch_all(&mut *tx)
         .await?;

crates/manager-core/src/trigger_repo.rs

@@ -14,6 +14,7 @@ use serde::{Deserialize, Serialize};
 use sqlx::PgPool;
 use uuid::Uuid;
 
+use crate::config_resolver::CHAIN_LEVELS_CTE;
 use crate::trigger_config::BackoffShape;
 
 #[derive(Debug, thiserror::Error)]
@@ -1304,15 +1305,17 @@ impl TriggerRepo for PostgresTriggerRepo {
         // happens in Rust so we don't have to teach the query about
         // `*` and `prefix:*`. Sets are tiny in practice (one app's
         // worth of triggers, usually a handful).
-        let rows: Vec<KvMatchRow> = sqlx::query_as(
+        let rows: Vec<KvMatchRow> = sqlx::query_as(&format!(
-            "SELECT t.id, t.script_id, t.dispatch_mode, \
+            "{CHAIN_LEVELS_CTE} \
+             SELECT t.id, t.script_id, t.dispatch_mode, \
                     t.retry_max_attempts, t.retry_backoff, t.retry_base_ms, \
                     t.registered_by_principal, \
                     d.collection_glob, d.ops \
              FROM triggers t \
              JOIN kv_trigger_details d ON d.trigger_id = t.id \
-             WHERE t.app_id = $1 AND t.kind = 'kv' AND t.enabled = TRUE",
+             JOIN chain c ON (t.app_id = c.app_owner OR t.group_id = c.group_owner) \
-        )
+             WHERE t.kind = 'kv' AND t.enabled = TRUE"
+        ))
         .bind(app_id.into_inner())
         .fetch_all(&self.pool)
         .await?;
@@ -1352,15 +1355,17 @@ impl TriggerRepo for PostgresTriggerRepo {
         // ops check into SQL (`WHERE $op = ANY(ops)`) — that would
         // exclude rows with `ops = '{}'` from the results, breaking
         // the empty-array-means-any-op semantic.
-        let rows: Vec<KvMatchRow> = sqlx::query_as(
+        let rows: Vec<KvMatchRow> = sqlx::query_as(&format!(
-            "SELECT t.id, t.script_id, t.dispatch_mode, \
+            "{CHAIN_LEVELS_CTE} \
+             SELECT t.id, t.script_id, t.dispatch_mode, \
                     t.retry_max_attempts, t.retry_backoff, t.retry_base_ms, \
                     t.registered_by_principal, \
                     d.collection_glob, d.ops \
              FROM triggers t \
              JOIN docs_trigger_details d ON d.trigger_id = t.id \
-             WHERE t.app_id = $1 AND t.kind = 'docs' AND t.enabled = TRUE",
+             JOIN chain c ON (t.app_id = c.app_owner OR t.group_id = c.group_owner) \
-        )
+             WHERE t.kind = 'docs' AND t.enabled = TRUE"
+        ))
         .bind(app_id.into_inner())
         .fetch_all(&self.pool)
         .await?;
@@ -1397,15 +1402,17 @@ impl TriggerRepo for PostgresTriggerRepo {
     ) -> Result<Vec<FilesTriggerMatch>, TriggerRepoError> {
         // Mirrors list_matching_kv: pull every enabled files trigger,
         // filter glob + ops in Rust (empty ops array means "any op").
-        let rows: Vec<KvMatchRow> = sqlx::query_as(
+        let rows: Vec<KvMatchRow> = sqlx::query_as(&format!(
-            "SELECT t.id, t.script_id, t.dispatch_mode, \
+            "{CHAIN_LEVELS_CTE} \
+             SELECT t.id, t.script_id, t.dispatch_mode, \
                     t.retry_max_attempts, t.retry_backoff, t.retry_base_ms, \
                     t.registered_by_principal, \
                     d.collection_glob, d.ops \
              FROM triggers t \
              JOIN files_trigger_details d ON d.trigger_id = t.id \
-             WHERE t.app_id = $1 AND t.kind = 'files' AND t.enabled = TRUE",
+             JOIN chain c ON (t.app_id = c.app_owner OR t.group_id = c.group_owner) \
-        )
+             WHERE t.kind = 'files' AND t.enabled = TRUE"
+        ))
         .bind(app_id.into_inner())
         .fetch_all(&self.pool)
         .await?;