feat(shared-topics): GroupPubsubService + shared publish fan-out + SDK handle (D2)

The runtime for shared topics:
- pubsub_repo: fan_out_publish gains `AND t.shared = FALSE` (a shared
  trigger never fires on a per-app publish); new fan_out_shared_publish
  matches `shared = true` pubsub triggers on the resolved owning group,
  each delivery stamped with the writer app_id (M2 model).
- GroupPubsubService trait (shared) + GroupPubsubServiceImpl: resolve the
  owning group (kind='topic') from cx.app_id's chain, require editor+
  (GroupPubsubPublish, fails closed for anon), size-cap, fan out.
- SDK: `pubsub::shared_topic("name")` -> GroupTopicHandle with
  `.publish(subtopic, msg)` / `.publish(msg)`; wired through Services +
  the picloud binary (reuses the one PubsubRepo).
- authz: Capability::GroupPubsubPublish(GroupId), editor+ / script:write.

The owning-group chain walk is the isolation boundary; a sibling-subtree
app never resolves the topic.

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

crates/manager-core/src/pubsub_repo.rs

@@ -12,7 +12,7 @@
 //! becomes a hot path.
 
 use async_trait::async_trait;
-use picloud_shared::{topic_matches, AdminUserId, AppId, ExecutionId};
+use picloud_shared::{topic_matches, AdminUserId, AppId, ExecutionId, GroupId};
 use sqlx::PgPool;
 use uuid::Uuid;
 
@@ -47,6 +47,20 @@ pub trait PubsubRepo: Send + Sync {
         topic: &str,
         event_payload: serde_json::Value,
     ) -> Result<u32, PubsubRepoError>;
+
+    /// §11.6 D2: fan out a SHARED-topic publish to every `shared = true` pubsub
+    /// trigger on `owning_group` (the group resolved from the shared-topic
+    /// declaration), inserting one outbox row each stamped with the WRITER
+    /// `ctx.app_id` — so the handler runs under the publishing app, matching the
+    /// M2 shared-collection trigger model. No chain union (the owning group is
+    /// already resolved) and no suppression (shared triggers aren't suppressible).
+    async fn fan_out_shared_publish(
+        &self,
+        ctx: PublishCtx,
+        owning_group: GroupId,
+        topic: &str,
+        event_payload: serde_json::Value,
+    ) -> Result<u32, PubsubRepoError>;
 }
 
 pub struct PostgresPubsubRepo {
@@ -92,7 +106,7 @@ impl PubsubRepo for PostgresPubsubRepo {
              FROM triggers t \
              JOIN pubsub_trigger_details d ON d.trigger_id = t.id \
              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{ANTIJOIN}",
+             WHERE t.kind = 'pubsub' AND t.enabled = TRUE AND t.shared = FALSE{ANTIJOIN}",
             ANTIJOIN = crate::trigger_repo::TRIGGER_SUPPRESSION_ANTIJOIN,
         ))
         .bind(ctx.app_id.into_inner())
@@ -104,21 +118,7 @@ impl PubsubRepo for PostgresPubsubRepo {
             if !topic_matches(&r.topic_pattern, topic) {
                 continue;
             }
-            sqlx::query(
-                "INSERT INTO outbox ( \
-                    app_id, source_kind, trigger_id, script_id, reply_to, \
-                    payload, origin_principal, trigger_depth, root_execution_id \
-                 ) VALUES ($1, 'pubsub', $2, $3, NULL, $4, $5, $6, $7)",
-            )
-            .bind(ctx.app_id.into_inner())
-            .bind(r.id)
-            .bind(r.script_id)
-            .bind(&event_payload)
-            .bind(ctx.origin_principal.map(AdminUserId::into_inner))
-            .bind(i32::try_from(ctx.trigger_depth.saturating_add(1)).unwrap_or(1))
-            .bind(ctx.root_execution_id.into_inner())
-            .execute(&mut *tx)
-            .await?;
+            insert_pubsub_outbox_row(&mut tx, &ctx, r.id, r.script_id, &event_payload).await?;
             written += 1;
         }
 
@@ -126,4 +126,69 @@ impl PubsubRepo for PostgresPubsubRepo {
         tx.commit().await?;
         Ok(written)
     }
+
+    async fn fan_out_shared_publish(
+        &self,
+        ctx: PublishCtx,
+        owning_group: GroupId,
+        topic: &str,
+        event_payload: serde_json::Value,
+    ) -> Result<u32, PubsubRepoError> {
+        let mut tx = self.pool.begin().await?;
+
+        // §11.6 D2: only `shared = true` pubsub triggers on the RESOLVED owning
+        // group — the shared-topic namespace boundary (a per-app or non-owning
+        // trigger never matches a shared publish). Topic-pattern match in Rust
+        // like the per-app path.
+        let rows: Vec<PubsubTriggerRow> = sqlx::query_as(
+            "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.kind = 'pubsub' AND t.enabled = TRUE AND t.shared = TRUE \
+               AND t.group_id = $1",
+        )
+        .bind(owning_group.into_inner())
+        .fetch_all(&mut *tx)
+        .await?;
+
+        let mut written: u32 = 0;
+        for r in rows {
+            if !topic_matches(&r.topic_pattern, topic) {
+                continue;
+            }
+            insert_pubsub_outbox_row(&mut tx, &ctx, r.id, r.script_id, &event_payload).await?;
+            written += 1;
+        }
+
+        tx.commit().await?;
+        Ok(written)
+    }
+}
+
+/// Insert one pubsub delivery row, stamped with the writer `ctx.app_id` so the
+/// handler runs under the publishing app. Shared by the per-app + shared
+/// fan-outs.
+async fn insert_pubsub_outbox_row(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    ctx: &PublishCtx,
+    trigger_id: Uuid,
+    script_id: Uuid,
+    event_payload: &serde_json::Value,
+) -> Result<(), PubsubRepoError> {
+    sqlx::query(
+        "INSERT INTO outbox ( \
+            app_id, source_kind, trigger_id, script_id, reply_to, \
+            payload, origin_principal, trigger_depth, root_execution_id \
+         ) VALUES ($1, 'pubsub', $2, $3, NULL, $4, $5, $6, $7)",
+    )
+    .bind(ctx.app_id.into_inner())
+    .bind(trigger_id)
+    .bind(script_id)
+    .bind(event_payload)
+    .bind(ctx.origin_principal.map(AdminUserId::into_inner))
+    .bind(i32::try_from(ctx.trigger_depth.saturating_add(1)).unwrap_or(1))
+    .bind(ctx.root_execution_id.into_inner())
+    .execute(&mut **tx)
+    .await?;
+    Ok(())
 }