feat(realtime): external SSE subscription for group shared topics (§11.6 D2 / Track A M6)

Shared TOPICS fanned out only to in-cluster trigger handlers; external clients
could not subscribe (per-app topics already can). Add SSE for shared topics.

- RealtimeBroadcaster gains a parallel (group_id, topic) channel map:
  subscribe_group / publish_group / drop_group_topic (default no-ops so
  NoopRealtimeBroadcaster + test doubles are untouched). InProcessBroadcaster
  implements them with a second map; GC + channel_count span both.
- Route GET /realtime/shared/topics/{topic}: Host->app dispatch (as the per-app
  route), then RealtimeAuthority::authorize_subscribe_shared resolves the OWNING
  GROUP from the app's chain (kind=topic, root segment). Reads-open model — the
  resolution IS the authorization, consistent with in-script shared reads; a
  foreign-subtree app never resolves (404, the isolation boundary). No principal
  machinery needed.
- GroupPubsubServiceImpl::with_realtime bridges a shared-topic publish to the
  owning-group channel (best-effort) after the durable trigger fan-out.
- Host wires the broadcaster into the group pubsub service + the collection
  resolver into the authority.

Auth-model note: chose reads-open (subtree app's Host is the grant) over
"authenticated principal + GroupKvRead" — it's both simpler and faithful to how
shared-collection reads already work. Pinned by realtime broadcaster group-map
tests, realtime_api shared-route tests (404 + stream), and
group_pubsub_service::publish_bridges_to_the_group_broadcaster. No migration.

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

crates/orchestrator-core/src/realtime.rs

@@ -25,7 +25,7 @@ use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
 use async_trait::async_trait;
-use picloud_shared::{AppId, BroadcasterError, RealtimeBroadcaster, RealtimeEvent};
+use picloud_shared::{AppId, BroadcasterError, GroupId, RealtimeBroadcaster, RealtimeEvent};
 use tokio::sync::broadcast;
 
 /// Default per-channel broadcast buffer depth.
@@ -37,6 +37,9 @@ pub const DEFAULT_GC_INTERVAL_SECS: u64 = 60;
 
 pub struct InProcessBroadcaster {
     inner: Mutex<HashMap<(AppId, String), broadcast::Sender<RealtimeEvent>>>,
+    /// §11.6 D2: group SHARED topic channels, keyed by the OWNING group so every
+    /// subtree app subscribing to the same shared topic shares one channel.
+    groups: Mutex<HashMap<(GroupId, String), broadcast::Sender<RealtimeEvent>>>,
     capacity: usize,
 }
 
@@ -45,6 +48,7 @@ impl InProcessBroadcaster {
     pub fn new(capacity: usize) -> Self {
         Self {
             inner: Mutex::new(HashMap::new()),
+            groups: Mutex::new(HashMap::new()),
             capacity: capacity.max(1),
         }
     }
@@ -69,21 +73,29 @@ impl InProcessBroadcaster {
         Self::new(capacity)
     }
 
-    /// Number of live channels in the map (test/observability helper).
+    /// Number of live channels in both maps (test/observability helper).
     #[must_use]
     pub fn channel_count(&self) -> usize {
-        self.inner.lock().map(|g| g.len()).unwrap_or(0)
+        let app = self.inner.lock().map(|g| g.len()).unwrap_or(0);
+        let grp = self.groups.lock().map(|g| g.len()).unwrap_or(0);
+        app + grp
     }
 
-    /// Drop senders with zero receivers. Returns how many were removed.
-    /// Called periodically by [`spawn_realtime_gc`].
+    /// Drop senders with zero receivers across both maps. Returns how many were
+    /// removed. Called periodically by [`spawn_realtime_gc`].
     pub fn gc(&self) -> usize {
-        let Ok(mut g) = self.inner.lock() else {
-            return 0;
-        };
-        let before = g.len();
-        g.retain(|_, tx| tx.receiver_count() > 0);
-        before - g.len()
+        let mut removed = 0;
+        if let Ok(mut g) = self.inner.lock() {
+            let before = g.len();
+            g.retain(|_, tx| tx.receiver_count() > 0);
+            removed += before - g.len();
+        }
+        if let Ok(mut g) = self.groups.lock() {
+            let before = g.len();
+            g.retain(|_, tx| tx.receiver_count() > 0);
+            removed += before - g.len();
+        }
+        removed
     }
 }
 
@@ -123,6 +135,36 @@ impl RealtimeBroadcaster for InProcessBroadcaster {
             g.remove(&(app_id, topic.to_string()));
         }
     }
+
+    async fn subscribe_group(
+        &self,
+        group_id: GroupId,
+        topic: &str,
+    ) -> Result<broadcast::Receiver<RealtimeEvent>, BroadcasterError> {
+        let mut g = self
+            .groups
+            .lock()
+            .map_err(|_| BroadcasterError::Unavailable("group broadcaster map poisoned".into()))?;
+        let tx = g
+            .entry((group_id, topic.to_string()))
+            .or_insert_with(|| broadcast::channel(self.capacity).0);
+        Ok(tx.subscribe())
+    }
+
+    async fn publish_group(&self, group_id: GroupId, topic: &str, event: RealtimeEvent) {
+        let Ok(g) = self.groups.lock() else {
+            return;
+        };
+        if let Some(tx) = g.get(&(group_id, topic.to_string())) {
+            let _ = tx.send(event);
+        }
+    }
+
+    async fn drop_group_topic(&self, group_id: GroupId, topic: &str) {
+        if let Ok(mut g) = self.groups.lock() {
+            g.remove(&(group_id, topic.to_string()));
+        }
+    }
 }
 
 /// Spawn the background GC sweep that drops empty channels every
@@ -239,4 +281,40 @@ mod tests {
         b.publish(app, "ghost", event("ghost", 1)).await;
         assert_eq!(b.channel_count(), 0);
     }
+
+    #[tokio::test]
+    async fn group_channel_fans_out_and_isolates_by_group() {
+        // §11.6 D2: a group shared-topic channel is keyed by the OWNING group —
+        // a publish to group A never reaches a subscriber on group B, and the
+        // group map is separate from the per-app map.
+        let b = InProcessBroadcaster::new(16);
+        let g_a = GroupId::new();
+        let g_b = GroupId::new();
+        let mut rx_a1 = b.subscribe_group(g_a, "events").await.unwrap();
+        let mut rx_a2 = b.subscribe_group(g_a, "events").await.unwrap();
+        let mut rx_b = b.subscribe_group(g_b, "events").await.unwrap();
+        assert_eq!(b.channel_count(), 2, "one channel per (group, topic)");
+
+        b.publish_group(g_a, "events", event("events", 7)).await;
+        // Both of group A's subscribers see it...
+        assert_eq!(rx_a1.recv().await.unwrap().message, json!({ "n": 7 }));
+        assert_eq!(rx_a2.recv().await.unwrap().message, json!({ "n": 7 }));
+        // ...group B's does not (owning-group isolation).
+        assert!(rx_b.try_recv().is_err());
+
+        // GC reclaims both maps.
+        drop((rx_a1, rx_a2, rx_b));
+        assert_eq!(b.gc(), 2);
+        assert_eq!(b.channel_count(), 0);
+    }
+
+    #[tokio::test]
+    async fn drop_group_topic_disconnects_subscribers() {
+        let b = InProcessBroadcaster::new(16);
+        let g = GroupId::new();
+        let mut rx = b.subscribe_group(g, "t").await.unwrap();
+        b.drop_group_topic(g, "t").await;
+        assert!(rx.recv().await.is_err());
+        assert_eq!(b.channel_count(), 0);
+    }
 }