feat(shared-triggers): author + persist + validate shared templates (M2.4)

`shared = true` on a group [[triggers.kv|docs|files]] flows manifest → Bundle
→ insert_trigger_tx (new shared column), mirroring sealed: shared lives on the
Trigger DTO + is part of the apply diff identity (bundle + current sides) so a
toggle re-applies. validate_bundle_for rejects shared on an app owner, on a
non-collection kind (pubsub/cron/etc.), and on a concrete collection the group
does not declare as a shared collection of that kind. Emission (M2.3) now has
authored triggers to match.

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

crates/manager-core/src/apply_service.rs

@@ -183,6 +183,9 @@ pub enum BundleTrigger {
         /// [`BundleTrigger::sealed`].
         #[serde(default)]
         sealed: bool,
+        /// §11.6: shared-collection group trigger. See [`BundleTrigger::shared`].
+        #[serde(default)]
+        shared: bool,
     },
     Docs {
         script: String,
@@ -195,6 +198,8 @@ pub enum BundleTrigger {
         retry_max_attempts: Option<u32>,
         #[serde(default)]
         sealed: bool,
+        #[serde(default)]
+        shared: bool,
     },
     Files {
         script: String,
@@ -207,6 +212,8 @@ pub enum BundleTrigger {
         retry_max_attempts: Option<u32>,
         #[serde(default)]
         sealed: bool,
+        #[serde(default)]
+        shared: bool,
     },
     Cron {
         script: String,
@@ -291,6 +298,39 @@ impl BundleTrigger {
         }
     }
 
+    /// §11.6: whether this template watches a SHARED collection. Only the
+    /// collection-scoped event kinds (kv/docs/files) can be shared; pubsub +
+    /// the app-only kinds are never shared.
+    #[must_use]
+    pub fn shared(&self) -> bool {
+        match self {
+            Self::Kv { shared, .. } | Self::Docs { shared, .. } | Self::Files { shared, .. } => {
+                *shared
+            }
+            Self::Cron { .. } | Self::Pubsub { .. } | Self::Email { .. } | Self::Queue { .. } => {
+                false
+            }
+        }
+    }
+
+    /// The watched collection glob for the collection-scoped kinds (kv/docs/
+    /// files); `None` for the others.
+    #[must_use]
+    pub fn collection_glob(&self) -> Option<&str> {
+        match self {
+            Self::Kv {
+                collection_glob, ..
+            }
+            | Self::Docs {
+                collection_glob, ..
+            }
+            | Self::Files {
+                collection_glob, ..
+            } => Some(collection_glob),
+            _ => None,
+        }
+    }
+
     /// Stable semantic identity (the per-kind tuple), used for the diff.
     #[must_use]
     pub fn identity(&self) -> String {
@@ -303,9 +343,10 @@ impl BundleTrigger {
                 collection_glob,
                 ops,
                 sealed,
+                shared,
                 ..
             } => format!(
-                "kv|{script}|{collection_glob}|{}|{sealed}",
+                "kv|{script}|{collection_glob}|{}|{sealed}|{shared}",
                 sorted_csv(ops.iter().copied().map(KvEventOp::as_str))
             ),
             Self::Docs {
@@ -313,9 +354,10 @@ impl BundleTrigger {
                 collection_glob,
                 ops,
                 sealed,
+                shared,
                 ..
             } => format!(
-                "docs|{script}|{collection_glob}|{}|{sealed}",
+                "docs|{script}|{collection_glob}|{}|{sealed}|{shared}",
                 sorted_csv(ops.iter().copied().map(DocsEventOp::as_str))
             ),
             Self::Files {
@@ -323,9 +365,10 @@ impl BundleTrigger {
                 collection_glob,
                 ops,
                 sealed,
+                shared,
                 ..
             } => format!(
-                "files|{script}|{collection_glob}|{}|{sealed}",
+                "files|{script}|{collection_glob}|{}|{sealed}|{shared}",
                 sorted_csv(ops.iter().copied().map(FilesEventOp::as_str))
             ),
             Self::Cron {
@@ -730,6 +773,40 @@ impl ApplyService {
                             .into(),
                     ));
                 }
+                // §11.6: a `shared` trigger watches the group's SHARED collection.
+                // Only the collection-scoped kinds can be shared (pubsub has no
+                // shared topic store yet), and the watched collection must be one
+                // the SAME group declares shared, of the matching kind.
+                if t.shared() {
+                    let kind = t.kind_str();
+                    if !matches!(
+                        t,
+                        BundleTrigger::Kv { .. }
+                            | BundleTrigger::Docs { .. }
+                            | BundleTrigger::Files { .. }
+                    ) {
+                        return Err(ApplyError::Invalid(format!(
+                            "a `shared` trigger must be a kv/docs/files kind; `{kind}` \
+                             has no shared collection store"
+                        )));
+                    }
+                    let glob = t.collection_glob().unwrap_or_default();
+                    // A concrete (non-glob) collection must be declared shared of
+                    // the same kind on this group. A wildcard glob can't be
+                    // statically resolved, so it's allowed (runtime match).
+                    let is_wildcard = glob.contains('*');
+                    let declared = bundle
+                        .collections
+                        .iter()
+                        .any(|c| c.kind == kind && c.name.eq_ignore_ascii_case(glob));
+                    if !is_wildcard && !declared {
+                        return Err(ApplyError::Invalid(format!(
+                            "a `shared` {kind} trigger watches collection `{glob}`, \
+                             which this group does not declare as a shared {kind} \
+                             collection (add it to `collections`)"
+                        )));
+                    }
+                }
             }
         }
         // §11.6: shared collections are owned by GROUPS. Reject them on an app
@@ -766,6 +843,15 @@ impl ApplyService {
                         .into(),
                 ));
             }
+            // §11.6: `shared` watches a group's SHARED collection — meaningless
+            // for an app (apps don't own shared collections).
+            if bundle.triggers.iter().any(BundleTrigger::shared) {
+                return Err(ApplyError::Invalid(
+                    "an app trigger cannot be `shared` — a shared-collection \
+                     trigger is owned by the group that declares the collection"
+                        .into(),
+                ));
+            }
         }
         // §11 tail M1: both an app and a group may declare suppressions — an
         // app declines an inherited template for itself, a group for its whole
@@ -993,6 +1079,8 @@ impl ApplyService {
                     // §11 tail: a sealed group template is non-suppressible
                     // (rejected on an app owner in `validate_bundle_for`).
                     bt.sealed(),
+                    // §11.6: a shared-collection group trigger.
+                    bt.shared(),
                     &details,
                 )
                 .await
@@ -3267,29 +3355,31 @@ fn validate_email_secrets_present(
 /// as NoOp, but `diff_triggers` separately refuses to emit a Delete for it.
 fn current_trigger_identity(t: &Trigger, name_by_id: &HashMap<ScriptId, String>) -> Option<String> {
     let script = name_by_id.get(&t.script_id)?;
-    // §11 tail: `sealed` is part of the identity for the event kinds (mirroring
+    // §11 tail: `sealed` (and §11.6 `shared`) are part of the identity for the
-    // `BundleTrigger::identity`), so toggling it on a group template re-diffs.
+    // event kinds (mirroring `BundleTrigger::identity`), so toggling either on a
+    // group template re-diffs.
     let sealed = t.sealed;
+    let shared = t.shared;
     match &t.details {
         TriggerDetails::Kv {
             collection_glob,
             ops,
         } => Some(format!(
-            "kv|{script}|{collection_glob}|{}|{sealed}",
+            "kv|{script}|{collection_glob}|{}|{sealed}|{shared}",
             sorted_csv(ops.iter().copied().map(KvEventOp::as_str))
         )),
         TriggerDetails::Docs {
             collection_glob,
             ops,
         } => Some(format!(
-            "docs|{script}|{collection_glob}|{}|{sealed}",
+            "docs|{script}|{collection_glob}|{}|{sealed}|{shared}",
             sorted_csv(ops.iter().copied().map(DocsEventOp::as_str))
         )),
         TriggerDetails::Files {
             collection_glob,
             ops,
         } => Some(format!(
-            "files|{script}|{collection_glob}|{}|{sealed}",
+            "files|{script}|{collection_glob}|{}|{sealed}|{shared}",
             sorted_csv(ops.iter().copied().map(FilesEventOp::as_str))
         )),
         TriggerDetails::Cron {
@@ -4013,6 +4103,7 @@ mod tests {
             kind,
             enabled: true,
             sealed: false,
+            shared: false,
             dispatch_mode: TriggerDispatchMode::Async,
             retry_max_attempts: 3,
             retry_backoff: BackoffShape::Exponential,
@@ -4093,6 +4184,7 @@ mod tests {
             dispatch_mode: None,
             retry_max_attempts: None,
             sealed: false,
+            shared: false,
         })
         .is_err());
         // A non-empty glob passes.
@@ -4103,6 +4195,7 @@ mod tests {
             dispatch_mode: None,
             retry_max_attempts: None,
             sealed: false,
+            shared: false,
         })
         .is_ok());
         // A malformed pubsub topic_pattern (mid-pattern wildcard) is rejected;
@@ -4421,6 +4514,7 @@ mod tests {
             dispatch_mode: None,
             retry_max_attempts: None,
             sealed: false,
+            shared: false,
         }];
         let p = compute_diff(&current, &b);
         assert!(
@@ -4458,6 +4552,7 @@ mod tests {
             dispatch_mode: None,
             retry_max_attempts: None,
             sealed: true,
+            shared: false,
         }];
         let p = compute_diff(&current, &b);
         assert!(

crates/manager-core/src/trigger_repo.rs

@@ -67,6 +67,10 @@ pub struct Trigger {
     /// so toggling it re-materializes the row.
     #[serde(default)]
     pub sealed: bool,
+    /// §11.6: `true` for a shared-collection group trigger (watches a shared
+    /// collection, not per-app ones). Also part of the apply diff identity.
+    #[serde(default)]
+    pub shared: bool,
     pub dispatch_mode: TriggerDispatchMode,
     pub retry_max_attempts: u32,
     pub retry_backoff: BackoffShape,
@@ -623,6 +627,9 @@ pub(crate) async fn insert_trigger_tx(
     // §11 tail: `true` for a sealed (non-suppressible) group trigger template.
     // Always `false` for an app-owned trigger.
     sealed: bool,
+    // §11.6: `true` for a shared-collection group trigger (watches a shared
+    // collection, not per-app ones). Always `false` for an app-owned trigger.
+    shared: bool,
     details: &TriggerDetails,
 ) -> Result<TriggerId, TriggerRepoError> {
     let kind = match details {
@@ -675,8 +682,8 @@ pub(crate) async fn insert_trigger_tx(
         "INSERT INTO triggers ( \
             app_id, group_id, script_id, kind, enabled, dispatch_mode, \
             retry_max_attempts, retry_backoff, retry_base_ms, \
-            registered_by_principal, sealed \
+            registered_by_principal, sealed, shared \
-         ) VALUES ($1, $2, $3, $4, TRUE, $5, $6, $7, $8, $9, $10) RETURNING id",
+         ) VALUES ($1, $2, $3, $4, TRUE, $5, $6, $7, $8, $9, $10, $11) RETURNING id",
     )
     .bind(owner_app_id)
     .bind(owner_group_id)
@@ -688,6 +695,7 @@ pub(crate) async fn insert_trigger_tx(
     .bind(i32::try_from(retry_base_ms).unwrap_or(1000))
     .bind(registered_by.into_inner())
     .bind(sealed)
+    .bind(shared)
     .fetch_one(&mut **tx)
     .await?;
     let tid = row.0;
@@ -888,6 +896,7 @@ impl TriggerRepo for PostgresTriggerRepo {
             kind: TriggerKind::Kv,
             enabled: parent.enabled,
             sealed: parent.sealed,
+            shared: parent.shared,
             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)
@@ -956,6 +965,7 @@ impl TriggerRepo for PostgresTriggerRepo {
             kind: TriggerKind::Docs,
             enabled: parent.enabled,
             sealed: parent.sealed,
+            shared: parent.shared,
             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)
@@ -1019,6 +1029,7 @@ impl TriggerRepo for PostgresTriggerRepo {
             kind: TriggerKind::DeadLetter,
             enabled: parent.enabled,
             sealed: parent.sealed,
+            shared: parent.shared,
             dispatch_mode: dispatch_from_str(&parent.dispatch_mode),
             retry_max_attempts: u32::try_from(parent.retry_max_attempts).unwrap_or(1),
             retry_backoff: BackoffShape::from_wire(&parent.retry_backoff)
@@ -1088,6 +1099,7 @@ impl TriggerRepo for PostgresTriggerRepo {
             kind: TriggerKind::Cron,
             enabled: parent.enabled,
             sealed: parent.sealed,
+            shared: parent.shared,
             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)
@@ -1157,6 +1169,7 @@ impl TriggerRepo for PostgresTriggerRepo {
             kind: TriggerKind::Files,
             enabled: parent.enabled,
             sealed: parent.sealed,
+            shared: parent.shared,
             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)
@@ -1221,6 +1234,7 @@ impl TriggerRepo for PostgresTriggerRepo {
             kind: TriggerKind::Pubsub,
             enabled: parent.enabled,
             sealed: parent.sealed,
+            shared: parent.shared,
             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)
@@ -1283,6 +1297,7 @@ impl TriggerRepo for PostgresTriggerRepo {
             kind: TriggerKind::Email,
             enabled: parent.enabled,
             sealed: parent.sealed,
+            shared: parent.shared,
             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)
@@ -1323,7 +1338,7 @@ impl TriggerRepo for PostgresTriggerRepo {
 
     async fn list_for_app(&self, app_id: AppId) -> Result<Vec<Trigger>, TriggerRepoError> {
         let parents: Vec<TriggerRow> = sqlx::query_as(
-            "SELECT id, app_id, script_id, name, kind, enabled, sealed, dispatch_mode, \
+            "SELECT id, app_id, script_id, name, kind, enabled, sealed, shared, dispatch_mode, \
                     retry_max_attempts, retry_backoff, retry_base_ms, \
                     registered_by_principal, created_at, updated_at \
              FROM triggers WHERE app_id = $1 ORDER BY created_at DESC",
@@ -1351,7 +1366,7 @@ impl TriggerRepo for PostgresTriggerRepo {
 
     async fn list_for_group(&self, group_id: GroupId) -> Result<Vec<Trigger>, TriggerRepoError> {
         let parents: Vec<TriggerRow> = sqlx::query_as(
-            "SELECT id, app_id, group_id, script_id, name, kind, enabled, sealed, dispatch_mode, \
+            "SELECT id, app_id, group_id, script_id, name, kind, enabled, sealed, shared, dispatch_mode, \
                     retry_max_attempts, retry_backoff, retry_base_ms, \
                     registered_by_principal, created_at, updated_at \
              FROM triggers WHERE group_id = $1 ORDER BY created_at DESC",
@@ -1377,7 +1392,7 @@ impl TriggerRepo for PostgresTriggerRepo {
         // rows are all app-owned) — else a template would hydrate with both
         // owners None, breaking the exactly-one invariant in memory.
         let parent: Option<TriggerRow> = sqlx::query_as(
-            "SELECT id, app_id, group_id, script_id, name, kind, enabled, sealed, dispatch_mode, \
+            "SELECT id, app_id, group_id, script_id, name, kind, enabled, sealed, shared, dispatch_mode, \
                     retry_max_attempts, retry_backoff, retry_base_ms, \
                     registered_by_principal, created_at, updated_at \
              FROM triggers WHERE id = $1",
@@ -1726,6 +1741,7 @@ impl TriggerRepo for PostgresTriggerRepo {
             kind: TriggerKind::Queue,
             enabled: parent.enabled,
             sealed: parent.sealed,
+            shared: parent.shared,
             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)
@@ -1936,6 +1952,7 @@ async fn hydrate_one(pool: &PgPool, parent: TriggerRow) -> Result<Trigger, Trigg
         kind,
         enabled: parent.enabled,
         sealed: parent.sealed,
+        shared: parent.shared,
         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)
@@ -1989,6 +2006,9 @@ struct TriggerRow {
     // queries that must see a true value SELECT it explicitly.
     #[sqlx(default)]
     sealed: bool,
+    // §11.6: same `default` treatment as `sealed`.
+    #[sqlx(default)]
+    shared: bool,
     dispatch_mode: String,
     retry_max_attempts: i32,
     retry_backoff: String,

crates/manager-core/src/triggers_api.rs

@@ -909,6 +909,7 @@ mod tests {
                 kind: crate::trigger_repo::TriggerKind::Kv,
                 enabled: true,
                 sealed: false,
+                shared: false,
                 dispatch_mode: req.dispatch_mode,
                 retry_max_attempts: req.retry_max_attempts,
                 retry_backoff: req.retry_backoff,
@@ -940,6 +941,7 @@ mod tests {
                 kind: crate::trigger_repo::TriggerKind::Docs,
                 enabled: true,
                 sealed: false,
+                shared: false,
                 dispatch_mode: req.dispatch_mode,
                 retry_max_attempts: req.retry_max_attempts,
                 retry_backoff: req.retry_backoff,
@@ -971,6 +973,7 @@ mod tests {
                 kind: crate::trigger_repo::TriggerKind::DeadLetter,
                 enabled: true,
                 sealed: false,
+                shared: false,
                 dispatch_mode: TriggerDispatchMode::Async,
                 retry_max_attempts: 1,
                 retry_backoff: BackoffShape::Constant,
@@ -1003,6 +1006,7 @@ mod tests {
                 kind: TriggerKind::Email,
                 enabled: true,
                 sealed: false,
+                shared: false,
                 dispatch_mode: TriggerDispatchMode::Async,
                 retry_max_attempts: 3,
                 retry_backoff: BackoffShape::Exponential,
@@ -1050,6 +1054,7 @@ mod tests {
                 kind: crate::trigger_repo::TriggerKind::Cron,
                 enabled: true,
                 sealed: false,
+                shared: false,
                 dispatch_mode: req.dispatch_mode,
                 retry_max_attempts: req.retry_max_attempts,
                 retry_backoff: req.retry_backoff,
@@ -1082,6 +1087,7 @@ mod tests {
                 kind: crate::trigger_repo::TriggerKind::Files,
                 enabled: true,
                 sealed: false,
+                shared: false,
                 dispatch_mode: req.dispatch_mode,
                 retry_max_attempts: req.retry_max_attempts,
                 retry_backoff: req.retry_backoff,
@@ -1113,6 +1119,7 @@ mod tests {
                 kind: crate::trigger_repo::TriggerKind::Pubsub,
                 enabled: true,
                 sealed: false,
+                shared: false,
                 dispatch_mode: req.dispatch_mode,
                 retry_max_attempts: req.retry_max_attempts,
                 retry_backoff: req.retry_backoff,
@@ -1192,6 +1199,7 @@ mod tests {
                 kind: TriggerKind::Queue,
                 enabled: true,
                 sealed: false,
+                shared: false,
                 dispatch_mode: req.dispatch_mode,
                 retry_max_attempts: req.retry_max_attempts,
                 retry_backoff: req.retry_backoff,

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

@@ -161,8 +161,9 @@ pub async fn run(app_ident: &str, dir: &Path, force: bool, mode: OutputMode) ->
                     ops: d.ops,
                     dispatch_mode,
                     retry_max_attempts,
-                    // app-owned triggers are never sealed (group-template only).
+                    // app-owned triggers are never sealed/shared (group-only).
                     sealed: false,
+                    shared: false,
                 });
             }
             "docs" => {
@@ -174,6 +175,7 @@ pub async fn run(app_ident: &str, dir: &Path, force: bool, mode: OutputMode) ->
                     dispatch_mode,
                     retry_max_attempts,
                     sealed: false,
+                    shared: false,
                 });
             }
             "files" => {
@@ -185,6 +187,7 @@ pub async fn run(app_ident: &str, dir: &Path, force: bool, mode: OutputMode) ->
                     dispatch_mode,
                     retry_max_attempts,
                     sealed: false,
+                    shared: false,
                 });
             }
             "cron" => {

crates/picloud-cli/src/manifest.rs

@@ -419,6 +419,10 @@ pub struct KvTriggerSpec {
     /// non-suppressible (event kinds only; group-only). Omitted ⇒ unsealed.
     #[serde(default, skip_serializing_if = "is_false")]
     pub sealed: bool,
+    /// §11.6: `shared = true` on a `[group]` template makes it watch the group's
+    /// SHARED collection (not per-app ones); group-only. Omitted ⇒ per-app.
+    #[serde(default, skip_serializing_if = "is_false")]
+    pub shared: bool,
 }
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
@@ -434,6 +438,9 @@ pub struct DocsTriggerSpec {
     /// See [`KvTriggerSpec::sealed`].
     #[serde(default, skip_serializing_if = "is_false")]
     pub sealed: bool,
+    /// See [`KvTriggerSpec::shared`].
+    #[serde(default, skip_serializing_if = "is_false")]
+    pub shared: bool,
 }
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
@@ -449,6 +456,9 @@ pub struct FilesTriggerSpec {
     /// See [`KvTriggerSpec::sealed`].
     #[serde(default, skip_serializing_if = "is_false")]
     pub sealed: bool,
+    /// See [`KvTriggerSpec::shared`].
+    #[serde(default, skip_serializing_if = "is_false")]
+    pub shared: bool,
 }
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
@@ -626,6 +636,7 @@ mod tests {
                     dispatch_mode: Some(DispatchMode::Async),
                     retry_max_attempts: Some(5),
                     sealed: false,
+                    shared: false,
                 }],
                 ..ManifestTriggers::default()
             },