feat(v1.1.6): realtime channels + v1.1.5 follow-ups + version bumps

Server-side realtime SSE on per-app pub/sub topics, plus the three
v1.1.5 follow-ups and the version bumps.

Realtime:
- topics registry (0021) + admin endpoints + Capability::AppTopicManage
  (-> app:admin; no new scope).
- GET /realtime/topics/{topic} SSE endpoint (orchestrator-core data
  plane): Host -> app, RealtimeAuthority gate (404 missing/internal,
  401 bad/absent token), broadcast::Receiver stream + heartbeat.
- RealtimeBroadcaster / RealtimeEvent / RealtimeAuthority traits
  (picloud-shared); InProcessBroadcaster + GC (orchestrator-core);
  DB-backed RealtimeAuthorityImpl (manager-core). Publish path fans out
  to in-process subscribers after the durable outbox commit (best-effort,
  panic-isolated).
- HMAC subscriber tokens (subscriber_token.rs) + app_secrets table (0022)
  + pubsub::subscriber_token SDK (schema 1.6 -> 1.7). TTL clamp + env
  overrides.
- Dashboard Topics tab (register/list/edit/delete, prominent external
  badge, flip confirmation).

v1.1.5 follow-ups:
- Empty blobs accepted (NewFile/FileUpdate::validate) + round-trip test.
- Orphan *.tmp.* sweeper (spawn_files_orphan_sweep).
- Dispatcher e2e tests, one per trigger kind (DATABASE_URL-gated).

Versions: workspace 1.1.6, SDK 1.7, dashboard 0.12.0. Schema-snapshot
golden re-blessed.

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

crates/manager-core/src/app_secrets_repo.rs

@@ -0,0 +1,91 @@
+//! `AppSecretsRepo` — per-app secret material (v1.1.6).
+//!
+//! Today this holds only the HMAC signing key for realtime subscriber
+//! tokens. The key is generated lazily (32 random bytes) on the first
+//! `pubsub::subscriber_token` call for an app and never changes
+//! thereafter in v1.1.6 (no rotation API yet — rotation is the
+//! key-invalidation mechanism, deferred). The key is never exposed to
+//! scripts: the SDK mints tokens, it never returns the key.
+//!
+//! This table is the natural home for v1.1.7's encrypted per-app
+//! secrets work.
+
+use async_trait::async_trait;
+use picloud_shared::AppId;
+use rand::RngCore;
+use sqlx::PgPool;
+
+/// Length of a freshly-generated realtime signing key.
+pub const SIGNING_KEY_LEN: usize = 32;
+
+#[derive(Debug, thiserror::Error)]
+pub enum AppSecretsRepoError {
+    #[error("database error: {0}")]
+    Db(#[from] sqlx::Error),
+}
+
+#[async_trait]
+pub trait AppSecretsRepo: Send + Sync {
+    /// Fetch the app's realtime signing key, generating + persisting one
+    /// (32 random bytes) if absent. Idempotent under concurrency: a
+    /// racing creator's `ON CONFLICT DO NOTHING` insert is a no-op and
+    /// the existing key is returned.
+    async fn get_or_create_signing_key(
+        &self,
+        app_id: AppId,
+    ) -> Result<Vec<u8>, AppSecretsRepoError>;
+
+    /// Fetch the signing key if it exists, WITHOUT creating one. The SSE
+    /// verify path uses this: a missing key means no token was ever
+    /// minted for the app, so any presented token must be rejected.
+    async fn signing_key(&self, app_id: AppId) -> Result<Option<Vec<u8>>, AppSecretsRepoError>;
+}
+
+pub struct PostgresAppSecretsRepo {
+    pool: PgPool,
+}
+
+impl PostgresAppSecretsRepo {
+    #[must_use]
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+}
+
+#[async_trait]
+impl AppSecretsRepo for PostgresAppSecretsRepo {
+    async fn get_or_create_signing_key(
+        &self,
+        app_id: AppId,
+    ) -> Result<Vec<u8>, AppSecretsRepoError> {
+        let mut fresh = vec![0u8; SIGNING_KEY_LEN];
+        rand::thread_rng().fill_bytes(&mut fresh);
+
+        // Insert-if-absent then read: the racing-creator's insert is a
+        // no-op, and the SELECT always returns the winning key.
+        sqlx::query(
+            "INSERT INTO app_secrets (app_id, realtime_signing_key) \
+             VALUES ($1, $2) ON CONFLICT (app_id) DO NOTHING",
+        )
+        .bind(app_id.into_inner())
+        .bind(&fresh)
+        .execute(&self.pool)
+        .await?;
+
+        let key: (Vec<u8>,) =
+            sqlx::query_as("SELECT realtime_signing_key FROM app_secrets WHERE app_id = $1")
+                .bind(app_id.into_inner())
+                .fetch_one(&self.pool)
+                .await?;
+        Ok(key.0)
+    }
+
+    async fn signing_key(&self, app_id: AppId) -> Result<Option<Vec<u8>>, AppSecretsRepoError> {
+        let row: Option<(Vec<u8>,)> =
+            sqlx::query_as("SELECT realtime_signing_key FROM app_secrets WHERE app_id = $1")
+                .bind(app_id.into_inner())
+                .fetch_optional(&self.pool)
+                .await?;
+        Ok(row.map(|r| r.0))
+    }
+}

crates/manager-core/src/authz.rs

@@ -97,6 +97,12 @@ pub enum Capability {
     /// to `app:admin` on API keys. Public-HTTP scripts (principal None)
     /// fail this check — managing dead letters is an admin act.
     AppDeadLetterManage(AppId),
+    /// Register / list / update / delete externally-subscribable topics
+    /// for this app (v1.1.6). Maps to `app:admin` on API keys —
+    /// externalizing a topic is an app-configuration act with security
+    /// weight (it opens an internal pub/sub topic to outside SSE
+    /// subscribers). Granted to `app_admin`+.
+    AppTopicManage(AppId),
 }
 
 impl Capability {
@@ -123,7 +129,8 @@ impl Capability {
             | Self::AppFilesWrite(id)
             | Self::AppPubsubPublish(id)
             | Self::AppManageTriggers(id)
-            | Self::AppDeadLetterManage(id) => Some(id),
+            | Self::AppDeadLetterManage(id)
+            | Self::AppTopicManage(id) => Some(id),
         }
     }
 
@@ -150,9 +157,10 @@ impl Capability {
             | Self::AppPubsubPublish(_) => Scope::ScriptWrite,
             Self::AppWriteRoute(_) => Scope::RouteWrite,
             Self::AppManageDomains(_) => Scope::DomainManage,
-            Self::AppAdmin(_) | Self::AppManageTriggers(_) | Self::AppDeadLetterManage(_) => {
-                Scope::AppAdmin
-            }
+            Self::AppAdmin(_)
+            | Self::AppManageTriggers(_)
+            | Self::AppDeadLetterManage(_)
+            | Self::AppTopicManage(_) => Scope::AppAdmin,
             Self::AppLogRead(_) => Scope::LogRead,
         }
     }
@@ -316,6 +324,7 @@ const fn role_satisfies(role: AppRole, cap: Capability) -> bool {
                 | Capability::AppAdmin(_)
                 | Capability::AppManageTriggers(_)
                 | Capability::AppDeadLetterManage(_)
+                | Capability::AppTopicManage(_)
         );
     match role {
         AppRole::Viewer => in_viewer,
@@ -659,6 +668,35 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn topic_manage_requires_app_admin() {
+        let repo = InMemoryAuthzRepo::default();
+        let app = AppId::new();
+        // Maps to the app:admin scope, not a new one.
+        assert_eq!(
+            Capability::AppTopicManage(app).required_scope(),
+            Scope::AppAdmin
+        );
+
+        // Member with only Editor role cannot manage topics.
+        let p = principal(InstanceRole::Member);
+        repo.grant(p.user_id, app, AppRole::Editor).await;
+        assert_eq!(
+            can(&repo, &p, Capability::AppTopicManage(app))
+                .await
+                .unwrap(),
+            Decision::Deny,
+        );
+
+        // App-admin role can.
+        let admin = principal(InstanceRole::Member);
+        repo.grant(admin.user_id, app, AppRole::AppAdmin).await;
+        assert!(can(&repo, &admin, Capability::AppTopicManage(app))
+            .await
+            .unwrap()
+            .is_allow());
+    }
+
     #[test]
     fn capability_app_id_extraction() {
         let app = AppId::new();

crates/manager-core/src/files_service.rs

@@ -633,20 +633,36 @@ mod tests {
             .await
             .unwrap_err();
         assert!(matches!(err, FilesError::MissingField("content_type")));
-        // data
-        let err = files
+        // Empty `data` is NO LONGER rejected (v1.1.6 relaxation) — see
+        // `empty_file_round_trips`.
+    }
+
+    #[tokio::test]
+    async fn empty_file_round_trips() {
+        // v1.1.6: a zero-byte blob is a valid stored state (sentinels,
+        // placeholders). Create with empty data, then read it back.
+        let files = svc();
+        let cx = anon_cx(AppId::new());
+        let id = files
             .create(
                 &cx,
                 "c",
                 NewFile {
-                    name: "f".into(),
-                    content_type: "text/plain".into(),
+                    name: "empty.bin".into(),
+                    content_type: "application/octet-stream".into(),
                     data: vec![],
                 },
             )
             .await
-            .unwrap_err();
-        assert!(matches!(err, FilesError::MissingField("data")));
+            .expect("empty file create should succeed");
+        let bytes = files.get(&cx, "c", &id.to_string()).await.unwrap();
+        assert_eq!(bytes, Some(Vec::new()));
+        let meta = files
+            .head(&cx, "c", &id.to_string())
+            .await
+            .unwrap()
+            .expect("metadata present");
+        assert_eq!(meta.size, 0);
     }
 
     #[tokio::test]

crates/manager-core/src/files_sweep.rs

@@ -0,0 +1,185 @@
+//! Orphan `*.tmp.*` blob sweeper (v1.1.6, v1.1.5 follow-up).
+//!
+//! The files repo writes blobs atomically: it streams into a
+//! `<id>.tmp.<pid>-<seq>` temp file, fsyncs, then renames to the final
+//! `<id>` path. A crash between create and rename leaves an orphan temp
+//! file that is never read and never reclaimed. This sweeper deletes
+//! those: every `PICLOUD_FILES_ORPHAN_SWEEP_INTERVAL_SEC` (default 6h) it
+//! walks `<root>/files/` and unlinks any `*.tmp.*` file older than
+//! `PICLOUD_FILES_ORPHAN_TMP_TTL_SEC` (default 1h).
+//!
+//! Deliberately bounded: it does NOT cross-check on-disk files against DB
+//! rows (the full reconciling sweeper is v1.3+). It only targets the temp
+//! files, which are unambiguously orphans once past the TTL — no live
+//! writer keeps one around for an hour.
+
+use std::path::{Path, PathBuf};
+use std::time::{Duration, SystemTime};
+
+const ENV_INTERVAL: &str = "PICLOUD_FILES_ORPHAN_SWEEP_INTERVAL_SEC";
+const ENV_TMP_TTL: &str = "PICLOUD_FILES_ORPHAN_TMP_TTL_SEC";
+const DEFAULT_INTERVAL_SECS: u64 = 21_600; // 6h
+const DEFAULT_TMP_TTL_SECS: u64 = 3_600; // 1h
+
+/// Marker that identifies a temp blob (`<id>.tmp.<pid>-<seq>`). A final
+/// blob is named just `<id>` (a UUID), so it never contains this.
+const TMP_MARKER: &str = ".tmp.";
+
+#[derive(Debug, Default, Clone, Copy)]
+pub struct SweepStats {
+    pub dirs_walked: u64,
+    pub files_deleted: u64,
+    pub bytes_reclaimed: u64,
+}
+
+/// Spawn the periodic orphan sweep. Spawned at startup alongside the
+/// cron scheduler and the realtime/cache GC tasks.
+pub fn spawn_files_orphan_sweep(files_root: PathBuf) {
+    let interval = Duration::from_secs(read_secs(ENV_INTERVAL, DEFAULT_INTERVAL_SECS));
+    let ttl = Duration::from_secs(read_secs(ENV_TMP_TTL, DEFAULT_TMP_TTL_SECS));
+    tokio::spawn(async move {
+        let mut ticker = tokio::time::interval(interval);
+        ticker.tick().await; // skip the immediate first fire
+        loop {
+            ticker.tick().await;
+            let root = files_root.clone();
+            // Blocking filesystem walk off the async worker.
+            let stats = tokio::task::spawn_blocking(move || sweep_orphan_tmp_files(&root, ttl))
+                .await
+                .unwrap_or_default();
+            tracing::info!(
+                dirs_walked = stats.dirs_walked,
+                files_deleted = stats.files_deleted,
+                bytes_reclaimed = stats.bytes_reclaimed,
+                "files orphan sweep complete"
+            );
+        }
+    });
+}
+
+/// Walk `<files_root>/files/` and delete `*.tmp.*` files older than
+/// `ttl`. Missing root is not an error (returns zeroed stats). Pure +
+/// synchronous so it's unit-testable without a runtime.
+#[must_use]
+pub fn sweep_orphan_tmp_files(files_root: &Path, ttl: Duration) -> SweepStats {
+    let mut stats = SweepStats::default();
+    let blobs_dir = files_root.join("files");
+    if !blobs_dir.is_dir() {
+        return stats;
+    }
+    let now = SystemTime::now();
+    walk(&blobs_dir, ttl, now, &mut stats);
+    stats
+}
+
+fn walk(dir: &Path, ttl: Duration, now: SystemTime, stats: &mut SweepStats) {
+    stats.dirs_walked += 1;
+    let Ok(entries) = std::fs::read_dir(dir) else {
+        return;
+    };
+    for entry in entries.flatten() {
+        let Ok(ft) = entry.file_type() else {
+            continue;
+        };
+        let path = entry.path();
+        if ft.is_dir() {
+            walk(&path, ttl, now, stats);
+            continue;
+        }
+        if !ft.is_file() {
+            continue;
+        }
+        if !entry.file_name().to_string_lossy().contains(TMP_MARKER) {
+            continue;
+        }
+        let Ok(meta) = entry.metadata() else {
+            continue;
+        };
+        let age = meta
+            .modified()
+            .ok()
+            .and_then(|m| now.duration_since(m).ok())
+            .unwrap_or(Duration::ZERO);
+        if age >= ttl {
+            let size = meta.len();
+            if std::fs::remove_file(&path).is_ok() {
+                stats.files_deleted += 1;
+                stats.bytes_reclaimed += size;
+            }
+        }
+    }
+}
+
+fn read_secs(key: &str, default: u64) -> u64 {
+    match std::env::var(key) {
+        Err(_) => default,
+        Ok(v) => match v.parse::<u64>() {
+            Ok(n) if n > 0 => n,
+            _ => {
+                tracing::warn!(env = key, value = %v, "invalid; using default");
+                default
+            }
+        },
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::atomic::{AtomicU64, Ordering};
+
+    static SEQ: AtomicU64 = AtomicU64::new(0);
+
+    fn tmp_root() -> PathBuf {
+        let n = SEQ.fetch_add(1, Ordering::Relaxed);
+        let dir =
+            std::env::temp_dir().join(format!("picloud-sweep-test-{}-{n}", std::process::id()));
+        std::fs::create_dir_all(dir.join("files").join("ab")).unwrap();
+        dir
+    }
+
+    fn touch(path: &Path) {
+        std::fs::write(path, b"x").unwrap();
+    }
+
+    #[test]
+    fn deletes_old_tmp_files() {
+        let root = tmp_root();
+        let tmp = root.join("files/ab/uuid.tmp.123-0");
+        touch(&tmp);
+        // ttl 0 → any tmp file counts as old.
+        let stats = sweep_orphan_tmp_files(&root, Duration::ZERO);
+        assert_eq!(stats.files_deleted, 1);
+        assert!(!tmp.exists());
+        assert!(stats.bytes_reclaimed >= 1);
+    }
+
+    #[test]
+    fn keeps_young_tmp_files() {
+        let root = tmp_root();
+        let tmp = root.join("files/ab/uuid.tmp.123-0");
+        touch(&tmp);
+        // Large TTL → the just-created file is too young to reap.
+        let stats = sweep_orphan_tmp_files(&root, Duration::from_secs(3600));
+        assert_eq!(stats.files_deleted, 0);
+        assert!(tmp.exists());
+    }
+
+    #[test]
+    fn keeps_non_tmp_files() {
+        let root = tmp_root();
+        let blob = root.join("files/ab/0123456789abcdef");
+        touch(&blob);
+        let stats = sweep_orphan_tmp_files(&root, Duration::ZERO);
+        assert_eq!(stats.files_deleted, 0);
+        assert!(blob.exists());
+    }
+
+    #[test]
+    fn missing_root_does_not_panic() {
+        let root = std::env::temp_dir().join("picloud-sweep-nonexistent-xyz");
+        let stats = sweep_orphan_tmp_files(&root, Duration::ZERO);
+        assert_eq!(stats.files_deleted, 0);
+        assert_eq!(stats.dirs_walked, 0);
+    }
+}

crates/manager-core/src/lib.rs

@@ -16,6 +16,7 @@ pub mod app_domain_repo;
 pub mod app_members_api;
 pub mod app_members_repo;
 pub mod app_repo;
+pub mod app_secrets_repo;
 pub mod apps_api;
 pub mod auth;
 pub mod auth_api;
@@ -33,6 +34,7 @@ pub mod docs_service;
 pub mod files_api;
 pub mod files_repo;
 pub mod files_service;
+pub mod files_sweep;
 pub mod gc;
 pub mod http_service;
 pub mod kv_repo;
@@ -45,12 +47,15 @@ pub mod outbox_repo;
 pub mod principal_resolver;
 pub mod pubsub_repo;
 pub mod pubsub_service;
+pub mod realtime_authority;
 pub mod repo;
 pub mod route_admin;
 pub mod route_repo;
 pub mod sandbox;
 pub mod scheduler;
 pub mod ssrf;
+pub mod topic_repo;
+pub mod topics_api;
 pub mod trigger_config;
 pub mod trigger_repo;
 pub mod triggers_api;
@@ -81,6 +86,9 @@ pub use app_members_repo::{
     PostgresAppMembersRepository,
 };
 pub use app_repo::{resolve_app, AppLookup, AppRepository, PostgresAppRepository};
+pub use app_secrets_repo::{
+    AppSecretsRepo, AppSecretsRepoError, PostgresAppSecretsRepo, SIGNING_KEY_LEN,
+};
 pub use apps_api::{apps_router, AppsState};
 pub use auth_api::auth_router;
 pub use auth_bootstrap::{
@@ -104,6 +112,7 @@ pub use docs_service::DocsServiceImpl;
 pub use files_api::{files_admin_router, FilesAdminState};
 pub use files_repo::{FilesConfig, FilesRepo, FilesRepoError, FsFilesRepo};
 pub use files_service::FilesServiceImpl;
+pub use files_sweep::{spawn_files_orphan_sweep, sweep_orphan_tmp_files, SweepStats};
 pub use gc::{spawn_abandoned_gc, spawn_dead_letter_gc};
 pub use http_service::{HttpConfig, HttpServiceImpl};
 pub use kv_repo::{KvRepo, KvRepoError, PostgresKvRepo};
@@ -116,7 +125,8 @@ pub use outbox_repo::{
 };
 pub use principal_resolver::{AdminPrincipalResolver, PrincipalResolver, PrincipalResolverError};
 pub use pubsub_repo::{PostgresPubsubRepo, PublishCtx, PubsubRepo, PubsubRepoError};
-pub use pubsub_service::PubsubServiceImpl;
+pub use pubsub_service::{PubsubServiceImpl, SubscriberTokenConfig};
+pub use realtime_authority::RealtimeAuthorityImpl;
 pub use repo::{
     ExecutionLogRepository, NewScript, PostgresExecutionLogRepository, PostgresScriptRepository,
     RepoResolver, ScriptPatch, ScriptRepository, ScriptRepositoryError,
@@ -124,6 +134,8 @@ pub use repo::{
 pub use route_admin::{compile_routes, route_admin_router, RouteAdminState};
 pub use route_repo::{NewRoute, PostgresRouteRepository, RouteRepository};
 pub use sandbox::{CeilingError, SandboxCeiling};
+pub use topic_repo::{PostgresTopicRepo, Topic, TopicAuthMode, TopicRepo, TopicRepoError};
+pub use topics_api::{topics_router, TopicsApiError, TopicsState};
 pub use trigger_config::{BackoffShape, TriggerConfig};
 pub use trigger_repo::{
     collection_matches, CreateDeadLetterTrigger, CreateDocsTrigger, CreateFilesTrigger,

crates/manager-core/src/pubsub_service.rs

@@ -11,20 +11,106 @@
 use std::sync::Arc;
 
 use async_trait::async_trait;
-use picloud_shared::{PubsubError, PubsubService, SdkCallCx, TriggerEvent};
+use picloud_shared::subscriber_token::{self, TokenClaims};
+use picloud_shared::{
+    PubsubError, PubsubService, RealtimeBroadcaster, RealtimeEvent, SdkCallCx, TriggerEvent,
+};
 
+use crate::app_secrets_repo::AppSecretsRepo;
 use crate::authz::{self, AuthzRepo, Capability};
 use crate::pubsub_repo::{PublishCtx, PubsubRepo, PubsubRepoError};
+use crate::topic_repo::TopicRepo;
+
+/// TTL bounds (seconds) for `pubsub::subscriber_token`. Env-overridable
+/// via `PICLOUD_SUBSCRIBER_TOKEN_TTL_{MIN,MAX,DEFAULT}_SEC`.
+#[derive(Debug, Clone, Copy)]
+pub struct SubscriberTokenConfig {
+    pub min_ttl: i64,
+    pub max_ttl: i64,
+    pub default_ttl: i64,
+}
+
+impl SubscriberTokenConfig {
+    #[must_use]
+    pub const fn conservative() -> Self {
+        Self {
+            min_ttl: 10,
+            max_ttl: 86_400,
+            default_ttl: 3_600,
+        }
+    }
+
+    /// Load from env, falling back to the conservative defaults for any
+    /// missing / invalid value.
+    #[must_use]
+    pub fn from_env() -> Self {
+        let mut c = Self::conservative();
+        load_i64(&mut c.min_ttl, "PICLOUD_SUBSCRIBER_TOKEN_TTL_MIN_SEC");
+        load_i64(&mut c.max_ttl, "PICLOUD_SUBSCRIBER_TOKEN_TTL_MAX_SEC");
+        load_i64(
+            &mut c.default_ttl,
+            "PICLOUD_SUBSCRIBER_TOKEN_TTL_DEFAULT_SEC",
+        );
+        c
+    }
+}
+
+impl Default for SubscriberTokenConfig {
+    fn default() -> Self {
+        Self::conservative()
+    }
+}
+
+fn load_i64(dst: &mut i64, key: &str) {
+    if let Ok(v) = std::env::var(key) {
+        match v.parse::<i64>() {
+            Ok(n) if n > 0 => *dst = n,
+            _ => tracing::warn!(env = key, value = %v, "ignoring invalid token-ttl value"),
+        }
+    }
+}
 
 pub struct PubsubServiceImpl {
     repo: Arc<dyn PubsubRepo>,
     authz: Arc<dyn AuthzRepo>,
+    // Realtime extras (v1.1.6) — optional so the existing two-arg
+    // constructor (and its unit tests) keep working without them. The
+    // production binary attaches them via `with_realtime`.
+    realtime: Option<Arc<dyn RealtimeBroadcaster>>,
+    topics: Option<Arc<dyn TopicRepo>>,
+    secrets: Option<Arc<dyn AppSecretsRepo>>,
+    token_config: SubscriberTokenConfig,
 }
 
 impl PubsubServiceImpl {
     #[must_use]
     pub fn new(repo: Arc<dyn PubsubRepo>, authz: Arc<dyn AuthzRepo>) -> Self {
-        Self { repo, authz }
+        Self {
+            repo,
+            authz,
+            realtime: None,
+            topics: None,
+            secrets: None,
+            token_config: SubscriberTokenConfig::conservative(),
+        }
+    }
+
+    /// Attach the v1.1.6 realtime surface: the in-process broadcaster
+    /// (publish fan-out to SSE subscribers), the topic registry +
+    /// app-secrets repo (subscriber-token minting), and the TTL config.
+    #[must_use]
+    pub fn with_realtime(
+        mut self,
+        broadcaster: Arc<dyn RealtimeBroadcaster>,
+        topics: Arc<dyn TopicRepo>,
+        secrets: Arc<dyn AppSecretsRepo>,
+        token_config: SubscriberTokenConfig,
+    ) -> Self {
+        self.realtime = Some(broadcaster);
+        self.topics = Some(topics);
+        self.secrets = Some(secrets);
+        self.token_config = token_config;
+        self
     }
 
     async fn check_publish(&self, cx: &SdkCallCx) -> Result<(), PubsubError> {
@@ -60,12 +146,15 @@ impl PubsubService for PubsubServiceImpl {
         }
         self.check_publish(cx).await?;
 
-        // `published_at` is stamped on the manager side so every
-        // delivery agrees on one instant.
+        // `published_at` is stamped once on the manager side so every
+        // delivery path — durable triggers AND the realtime broadcast —
+        // agrees on one instant. The message is cloned into the trigger
+        // event so the realtime path can reuse the original.
+        let published_at = chrono::Utc::now();
         let event = TriggerEvent::Pubsub {
             topic: topic.to_string(),
-            message,
-            published_at: chrono::Utc::now(),
+            message: message.clone(),
+            published_at,
         };
         let payload = serde_json::to_value(&event)
             .map_err(|e| PubsubError::Rejected(format!("event serialize: {e}")))?;
@@ -76,12 +165,115 @@ impl PubsubService for PubsubServiceImpl {
             trigger_depth: cx.trigger_depth,
             root_execution_id: cx.root_execution_id,
         };
+        // Order (design notes §8): transactional outbox fan-out + commit
+        // FIRST; only then the best-effort realtime broadcast. If the
+        // fan-out fails, the publish throws and no broadcast happens. If
+        // the broadcast fails after a committed fan-out, trigger
+        // deliveries still happen and only SSE subscribers miss this
+        // event (no replay in v1.1.6).
+        //
         // No matching triggers → 0 rows written, publish still succeeds.
         self.repo
             .fan_out_publish(publish_ctx, topic, payload)
             .await?;
+
+        // Non-transactional, best-effort fan-out to in-process SSE
+        // subscribers. Run on a child task so a panicking broadcaster
+        // (or a future cluster-mode resolver fault) becomes a warn log,
+        // never a failed publish — the durable deliveries already
+        // committed above.
+        if let Some(realtime) = self.realtime.clone() {
+            let app_id = cx.app_id;
+            let topic_owned = topic.to_string();
+            let realtime_event = RealtimeEvent {
+                topic: topic_owned.clone(),
+                message,
+                published_at,
+            };
+            let handle = tokio::spawn(async move {
+                realtime.publish(app_id, &topic_owned, realtime_event).await;
+            });
+            if let Err(e) = handle.await {
+                tracing::warn!(error = %e, "realtime broadcast failed; publish unaffected");
+            }
+        }
         Ok(())
     }
+
+    async fn mint_subscriber_token(
+        &self,
+        cx: &SdkCallCx,
+        topics: Vec<String>,
+        ttl_seconds: Option<i64>,
+    ) -> Result<String, PubsubError> {
+        // Anonymous (public-HTTP) scripts can't mint — that would bypass
+        // the token-minting authz boundary.
+        let Some(principal) = cx.principal.as_ref() else {
+            return Err(PubsubError::SubscriberToken(
+                "pubsub::subscriber_token: requires an authenticated principal \
+                 (a script on a public route cannot mint tokens)"
+                    .into(),
+            ));
+        };
+        // Minting reuses the existing pub/sub publish capability (no new
+        // scope — the seven-scope commitment holds).
+        authz::require(
+            &*self.authz,
+            principal,
+            Capability::AppPubsubPublish(cx.app_id),
+        )
+        .await
+        .map_err(|_| PubsubError::Forbidden)?;
+
+        let (Some(topic_repo), Some(secrets)) = (self.topics.as_ref(), self.secrets.as_ref())
+        else {
+            return Err(PubsubError::Unavailable(
+                "subscriber tokens are not wired in".into(),
+            ));
+        };
+
+        if topics.is_empty() {
+            return Err(PubsubError::SubscriberToken(
+                "pubsub::subscriber_token: topics list must not be empty".into(),
+            ));
+        }
+
+        let ttl = ttl_seconds.unwrap_or(self.token_config.default_ttl);
+        if ttl < self.token_config.min_ttl || ttl > self.token_config.max_ttl {
+            return Err(PubsubError::SubscriberToken(format!(
+                "pubsub::subscriber_token: ttl_seconds must be between {} and {}",
+                self.token_config.min_ttl, self.token_config.max_ttl
+            )));
+        }
+
+        // Every requested topic must be registered as externally
+        // subscribable in this app — fail fast rather than mint a token
+        // that won't work.
+        for name in &topics {
+            let registered = topic_repo
+                .get(cx.app_id, name)
+                .await
+                .map_err(|e| PubsubError::Unavailable(e.to_string()))?;
+            if !registered.map(|t| t.external_subscribable).unwrap_or(false) {
+                return Err(PubsubError::SubscriberToken(format!(
+                    "pubsub::subscriber_token: topic {name} is not externally subscribable"
+                )));
+            }
+        }
+
+        let key = secrets
+            .get_or_create_signing_key(cx.app_id)
+            .await
+            .map_err(|e| PubsubError::Unavailable(e.to_string()))?;
+        let now = chrono::Utc::now().timestamp();
+        let claims = TokenClaims {
+            app_id: cx.app_id,
+            topics,
+            exp: now.saturating_add(ttl),
+            iat: now,
+        };
+        Ok(subscriber_token::sign(&key, &claims))
+    }
 }
 
 // ----------------------------------------------------------------------------
@@ -317,4 +509,218 @@ mod tests {
             .await
             .unwrap();
     }
+
+    // ------------------------------------------------------------------
+    // v1.1.6 realtime broadcast + subscriber-token minting
+    // ------------------------------------------------------------------
+
+    use crate::app_secrets_repo::{AppSecretsRepo, AppSecretsRepoError};
+    use crate::topic_repo::{Topic, TopicAuthMode, TopicRepo, TopicRepoError};
+    use picloud_orchestrator_core::InProcessBroadcaster;
+    use picloud_shared::{RealtimeBroadcaster, RealtimeEvent};
+
+    /// Topic repo fake: returns the configured topics as registered +
+    /// externally-subscribable (unless absent).
+    struct FakeTopicRepo(Vec<String>);
+    #[async_trait]
+    impl TopicRepo for FakeTopicRepo {
+        async fn create(
+            &self,
+            _: AppId,
+            _: &str,
+            _: bool,
+            _: TopicAuthMode,
+        ) -> Result<Topic, TopicRepoError> {
+            unimplemented!()
+        }
+        async fn list(&self, _: AppId) -> Result<Vec<Topic>, TopicRepoError> {
+            unimplemented!()
+        }
+        async fn get(&self, _: AppId, name: &str) -> Result<Option<Topic>, TopicRepoError> {
+            Ok(self.0.iter().any(|t| t == name).then(|| Topic {
+                name: name.to_string(),
+                external_subscribable: true,
+                auth_mode: TopicAuthMode::Token,
+                created_at: chrono::Utc::now(),
+                updated_at: chrono::Utc::now(),
+            }))
+        }
+        async fn update(
+            &self,
+            _: AppId,
+            _: &str,
+            _: Option<bool>,
+            _: Option<TopicAuthMode>,
+        ) -> Result<Option<Topic>, TopicRepoError> {
+            unimplemented!()
+        }
+        async fn delete(&self, _: AppId, _: &str) -> Result<bool, TopicRepoError> {
+            unimplemented!()
+        }
+    }
+
+    #[derive(Default)]
+    struct FakeSecrets;
+    #[async_trait]
+    impl AppSecretsRepo for FakeSecrets {
+        async fn get_or_create_signing_key(
+            &self,
+            _: AppId,
+        ) -> Result<Vec<u8>, AppSecretsRepoError> {
+            Ok(vec![42u8; 32])
+        }
+        async fn signing_key(&self, _: AppId) -> Result<Option<Vec<u8>>, AppSecretsRepoError> {
+            Ok(Some(vec![42u8; 32]))
+        }
+    }
+
+    /// Broadcaster that panics on publish — proves a broadcast fault
+    /// can't fail the publish.
+    struct PanicBroadcaster;
+    #[async_trait]
+    impl RealtimeBroadcaster for PanicBroadcaster {
+        async fn subscribe(
+            &self,
+            _: AppId,
+            _: &str,
+        ) -> Result<tokio::sync::broadcast::Receiver<RealtimeEvent>, picloud_shared::BroadcasterError>
+        {
+            unimplemented!()
+        }
+        async fn publish(&self, _: AppId, _: &str, _: RealtimeEvent) {
+            panic!("boom");
+        }
+        async fn drop_topic(&self, _: AppId, _: &str) {}
+    }
+
+    fn realtime_svc(
+        repo: Arc<dyn PubsubRepo>,
+        broadcaster: Arc<dyn RealtimeBroadcaster>,
+        topics: Vec<String>,
+    ) -> PubsubServiceImpl {
+        PubsubServiceImpl::new(repo, Arc::new(EditorAuthzRepo)).with_realtime(
+            broadcaster,
+            Arc::new(FakeTopicRepo(topics)),
+            Arc::new(FakeSecrets),
+            SubscriberTokenConfig::conservative(),
+        )
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn publish_broadcasts_to_in_process_subscribers() {
+        let app = AppId::new();
+        let broadcaster = Arc::new(InProcessBroadcaster::new(16));
+        let mut rx = broadcaster.subscribe(app, "chat").await.unwrap();
+        let svc = realtime_svc(
+            Arc::new(InMemoryPubsubRepo::new(vec![])),
+            broadcaster,
+            vec![],
+        );
+        svc.publish_durable(&anon_cx(app), "chat", serde_json::json!({ "hi": 1 }))
+            .await
+            .unwrap();
+        let ev = rx.recv().await.unwrap();
+        assert_eq!(ev.topic, "chat");
+        assert_eq!(ev.message, serde_json::json!({ "hi": 1 }));
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn panicking_broadcaster_does_not_fail_publish() {
+        let app = AppId::new();
+        let svc = realtime_svc(
+            Arc::new(InMemoryPubsubRepo::new(vec![])),
+            Arc::new(PanicBroadcaster),
+            vec![],
+        );
+        // The outbox fan-out committed; the broadcast panic is swallowed.
+        svc.publish_durable(&anon_cx(app), "chat", serde_json::json!(1))
+            .await
+            .expect("publish must succeed despite broadcast panic");
+    }
+
+    fn mint_svc(topics: Vec<String>) -> PubsubServiceImpl {
+        realtime_svc(
+            Arc::new(InMemoryPubsubRepo::new(vec![])),
+            Arc::new(picloud_shared::NoopRealtimeBroadcaster),
+            topics,
+        )
+    }
+
+    #[tokio::test]
+    async fn mint_returns_token_scoped_to_topics() {
+        let app = AppId::new();
+        let svc = mint_svc(vec!["chat".into(), "notify".into()]);
+        let token = svc
+            .mint_subscriber_token(
+                &member_cx(app),
+                vec!["chat".into(), "notify".into()],
+                Some(120),
+            )
+            .await
+            .unwrap();
+        // Verify with the fake key; claims carry the topics + expiry.
+        let claims = subscriber_token::verify(&[42u8; 32], &token, chrono::Utc::now().timestamp())
+            .expect("token verifies");
+        assert_eq!(claims.app_id, app);
+        assert!(claims.allows_topic("chat") && claims.allows_topic("notify"));
+        assert!(claims.exp > claims.iat);
+    }
+
+    #[tokio::test]
+    async fn mint_anonymous_principal_throws() {
+        let app = AppId::new();
+        let svc = mint_svc(vec!["chat".into()]);
+        let err = svc
+            .mint_subscriber_token(&anon_cx(app), vec!["chat".into()], None)
+            .await
+            .unwrap_err();
+        assert!(matches!(err, PubsubError::SubscriberToken(_)));
+    }
+
+    #[tokio::test]
+    async fn mint_empty_topics_throws() {
+        let app = AppId::new();
+        let svc = mint_svc(vec!["chat".into()]);
+        let err = svc
+            .mint_subscriber_token(&member_cx(app), vec![], None)
+            .await
+            .unwrap_err();
+        assert!(matches!(err, PubsubError::SubscriberToken(_)));
+    }
+
+    #[tokio::test]
+    async fn mint_ttl_below_min_and_above_max_throw() {
+        let app = AppId::new();
+        let svc = mint_svc(vec!["chat".into()]);
+        for bad in [Some(5), Some(90_000)] {
+            let err = svc
+                .mint_subscriber_token(&member_cx(app), vec!["chat".into()], bad)
+                .await
+                .unwrap_err();
+            assert!(
+                matches!(err, PubsubError::SubscriberToken(_)),
+                "ttl {bad:?}"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn mint_unregistered_topic_throws_with_message() {
+        let app = AppId::new();
+        // "chat" registered; "secret" is not.
+        let svc = mint_svc(vec!["chat".into()]);
+        let err = svc
+            .mint_subscriber_token(&member_cx(app), vec!["chat".into(), "secret".into()], None)
+            .await
+            .unwrap_err();
+        match err {
+            PubsubError::SubscriberToken(msg) => {
+                assert!(
+                    msg.contains("topic secret is not externally subscribable"),
+                    "got: {msg}"
+                );
+            }
+            other => panic!("expected SubscriberToken, got {other:?}"),
+        }
+    }
 }

crates/manager-core/src/realtime_authority.rs

@@ -0,0 +1,338 @@
+//! `RealtimeAuthorityImpl` — the DB-backed SSE subscribe gate (v1.1.6).
+//!
+//! Backs the [`picloud_shared::RealtimeAuthority`] trait the SSE handler
+//! in orchestrator-core calls. All `topics`-table reads and signing-key
+//! material stay inside this impl so the data-plane crate never touches
+//! the key.
+//!
+//! Verdict mapping (see [`SubscribeDenied`]):
+//!   * topic missing OR not externally subscribable → `NotFound` (404).
+//!     Both collapse to 404 so the endpoint can't probe internal topics.
+//!   * `auth_mode = 'public'` → allow.
+//!   * `auth_mode = 'token'` → verify the HMAC token (present, signed by
+//!     this app's key, unexpired, scoped to this topic) → allow, else
+//!     `Unauthorized` (401, generic — never says which check failed).
+//!
+//! Signing keys never change in v1.1.6 (no rotation API), so a small
+//! in-memory cache avoids a per-subscribe DB read once an app's key has
+//! been seen. The cache is purely an optimization — a cold miss reads
+//! the row.
+
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+
+use async_trait::async_trait;
+use picloud_shared::{subscriber_token, AppId, RealtimeAuthority, SubscribeDenied};
+
+use crate::app_secrets_repo::AppSecretsRepo;
+use crate::topic_repo::{TopicAuthMode, TopicRepo};
+
+pub struct RealtimeAuthorityImpl {
+    topics: Arc<dyn TopicRepo>,
+    secrets: Arc<dyn AppSecretsRepo>,
+    key_cache: Mutex<HashMap<AppId, Vec<u8>>>,
+}
+
+impl RealtimeAuthorityImpl {
+    #[must_use]
+    pub fn new(topics: Arc<dyn TopicRepo>, secrets: Arc<dyn AppSecretsRepo>) -> Self {
+        Self {
+            topics,
+            secrets,
+            key_cache: Mutex::new(HashMap::new()),
+        }
+    }
+
+    /// Fetch the app's signing key, consulting the cache first. Returns
+    /// `None` when the app has no key (no token ever minted) — which the
+    /// caller maps to `Unauthorized`.
+    async fn signing_key(&self, app_id: AppId) -> Result<Option<Vec<u8>>, SubscribeDenied> {
+        if let Ok(cache) = self.key_cache.lock() {
+            if let Some(k) = cache.get(&app_id) {
+                return Ok(Some(k.clone()));
+            }
+        }
+        let key = self
+            .secrets
+            .signing_key(app_id)
+            .await
+            .map_err(|e| SubscribeDenied::Backend(e.to_string()))?;
+        if let Some(ref k) = key {
+            if let Ok(mut cache) = self.key_cache.lock() {
+                cache.insert(app_id, k.clone());
+            }
+        }
+        Ok(key)
+    }
+}
+
+#[async_trait]
+impl RealtimeAuthority for RealtimeAuthorityImpl {
+    async fn authorize_subscribe(
+        &self,
+        app_id: AppId,
+        topic: &str,
+        token: Option<&str>,
+    ) -> Result<(), SubscribeDenied> {
+        let registered = self
+            .topics
+            .get(app_id, topic)
+            .await
+            .map_err(|e| SubscribeDenied::Backend(e.to_string()))?;
+
+        // Missing topic AND internal-only topic both 404 — don't leak
+        // which internal topics exist.
+        let Some(t) = registered.filter(|t| t.external_subscribable) else {
+            return Err(SubscribeDenied::NotFound);
+        };
+
+        match t.auth_mode {
+            TopicAuthMode::Public => Ok(()),
+            TopicAuthMode::Token => {
+                let token = token.ok_or(SubscribeDenied::Unauthorized)?;
+                let key = self
+                    .signing_key(app_id)
+                    .await?
+                    .ok_or(SubscribeDenied::Unauthorized)?;
+                let now = chrono::Utc::now().timestamp();
+                let claims = subscriber_token::verify(&key, token, now)
+                    .map_err(|_| SubscribeDenied::Unauthorized)?;
+                // Per-app key already makes a cross-app token fail the
+                // signature check; this is belt-and-suspenders.
+                if claims.app_id != app_id || !claims.allows_topic(topic) {
+                    return Err(SubscribeDenied::Unauthorized);
+                }
+                Ok(())
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::app_secrets_repo::AppSecretsRepoError;
+    use crate::topic_repo::{Topic, TopicRepoError};
+    use chrono::Utc;
+    use picloud_shared::subscriber_token::{sign, TokenClaims};
+
+    struct FakeTopics(Vec<(AppId, Topic)>);
+    #[async_trait]
+    impl TopicRepo for FakeTopics {
+        async fn create(
+            &self,
+            _: AppId,
+            _: &str,
+            _: bool,
+            _: TopicAuthMode,
+        ) -> Result<Topic, TopicRepoError> {
+            unimplemented!()
+        }
+        async fn list(&self, _: AppId) -> Result<Vec<Topic>, TopicRepoError> {
+            unimplemented!()
+        }
+        async fn get(&self, app_id: AppId, name: &str) -> Result<Option<Topic>, TopicRepoError> {
+            Ok(self
+                .0
+                .iter()
+                .find(|(a, t)| *a == app_id && t.name == name)
+                .map(|(_, t)| t.clone()))
+        }
+        async fn update(
+            &self,
+            _: AppId,
+            _: &str,
+            _: Option<bool>,
+            _: Option<TopicAuthMode>,
+        ) -> Result<Option<Topic>, TopicRepoError> {
+            unimplemented!()
+        }
+        async fn delete(&self, _: AppId, _: &str) -> Result<bool, TopicRepoError> {
+            unimplemented!()
+        }
+    }
+
+    struct FakeSecrets(AppId, Vec<u8>);
+    #[async_trait]
+    impl AppSecretsRepo for FakeSecrets {
+        async fn get_or_create_signing_key(
+            &self,
+            _: AppId,
+        ) -> Result<Vec<u8>, AppSecretsRepoError> {
+            Ok(self.1.clone())
+        }
+        async fn signing_key(&self, app_id: AppId) -> Result<Option<Vec<u8>>, AppSecretsRepoError> {
+            Ok((app_id == self.0).then(|| self.1.clone()))
+        }
+    }
+
+    fn topic(name: &str, external: bool, mode: TopicAuthMode) -> Topic {
+        Topic {
+            name: name.to_string(),
+            external_subscribable: external,
+            auth_mode: mode,
+            created_at: Utc::now(),
+            updated_at: Utc::now(),
+        }
+    }
+
+    fn authority(
+        topics: Vec<(AppId, Topic)>,
+        key_app: AppId,
+        key: Vec<u8>,
+    ) -> RealtimeAuthorityImpl {
+        RealtimeAuthorityImpl::new(
+            Arc::new(FakeTopics(topics)),
+            Arc::new(FakeSecrets(key_app, key)),
+        )
+    }
+
+    #[tokio::test]
+    async fn missing_topic_is_not_found() {
+        let app = AppId::new();
+        let a = authority(vec![], app, vec![0u8; 32]);
+        assert_eq!(
+            a.authorize_subscribe(app, "ghost", None).await,
+            Err(SubscribeDenied::NotFound)
+        );
+    }
+
+    #[tokio::test]
+    async fn internal_only_topic_is_not_found() {
+        let app = AppId::new();
+        let a = authority(
+            vec![(app, topic("internal", false, TopicAuthMode::Public))],
+            app,
+            vec![0u8; 32],
+        );
+        assert_eq!(
+            a.authorize_subscribe(app, "internal", None).await,
+            Err(SubscribeDenied::NotFound)
+        );
+    }
+
+    #[tokio::test]
+    async fn public_topic_allows_without_token() {
+        let app = AppId::new();
+        let a = authority(
+            vec![(app, topic("news", true, TopicAuthMode::Public))],
+            app,
+            vec![0u8; 32],
+        );
+        assert!(a.authorize_subscribe(app, "news", None).await.is_ok());
+    }
+
+    #[tokio::test]
+    async fn token_topic_without_token_is_unauthorized() {
+        let app = AppId::new();
+        let a = authority(
+            vec![(app, topic("chat", true, TopicAuthMode::Token))],
+            app,
+            vec![7u8; 32],
+        );
+        assert_eq!(
+            a.authorize_subscribe(app, "chat", None).await,
+            Err(SubscribeDenied::Unauthorized)
+        );
+    }
+
+    #[tokio::test]
+    async fn token_topic_with_valid_token_allows() {
+        let app = AppId::new();
+        let key = vec![9u8; 32];
+        let a = authority(
+            vec![(app, topic("chat", true, TopicAuthMode::Token))],
+            app,
+            key.clone(),
+        );
+        let token = sign(
+            &key,
+            &TokenClaims {
+                app_id: app,
+                topics: vec!["chat".into()],
+                iat: Utc::now().timestamp(),
+                exp: Utc::now().timestamp() + 60,
+            },
+        );
+        assert!(a
+            .authorize_subscribe(app, "chat", Some(&token))
+            .await
+            .is_ok());
+    }
+
+    #[tokio::test]
+    async fn token_for_other_topic_is_unauthorized() {
+        let app = AppId::new();
+        let key = vec![9u8; 32];
+        let a = authority(
+            vec![(app, topic("chat", true, TopicAuthMode::Token))],
+            app,
+            key.clone(),
+        );
+        let token = sign(
+            &key,
+            &TokenClaims {
+                app_id: app,
+                topics: vec!["other".into()],
+                iat: Utc::now().timestamp(),
+                exp: Utc::now().timestamp() + 60,
+            },
+        );
+        assert_eq!(
+            a.authorize_subscribe(app, "chat", Some(&token)).await,
+            Err(SubscribeDenied::Unauthorized)
+        );
+    }
+
+    #[tokio::test]
+    async fn expired_token_is_unauthorized() {
+        let app = AppId::new();
+        let key = vec![9u8; 32];
+        let a = authority(
+            vec![(app, topic("chat", true, TopicAuthMode::Token))],
+            app,
+            key.clone(),
+        );
+        let token = sign(
+            &key,
+            &TokenClaims {
+                app_id: app,
+                topics: vec!["chat".into()],
+                iat: Utc::now().timestamp() - 120,
+                exp: Utc::now().timestamp() - 60,
+            },
+        );
+        assert_eq!(
+            a.authorize_subscribe(app, "chat", Some(&token)).await,
+            Err(SubscribeDenied::Unauthorized)
+        );
+    }
+
+    #[tokio::test]
+    async fn token_signed_by_other_app_key_is_unauthorized() {
+        let app_a = AppId::new();
+        let app_b = AppId::new();
+        let key_a = vec![1u8; 32];
+        let key_b = vec![2u8; 32];
+        // Authority for app B; its key is key_b.
+        let a = authority(
+            vec![(app_b, topic("chat", true, TopicAuthMode::Token))],
+            app_b,
+            key_b,
+        );
+        // Token signed by app A's key, claiming app A.
+        let token = sign(
+            &key_a,
+            &TokenClaims {
+                app_id: app_a,
+                topics: vec!["chat".into()],
+                iat: Utc::now().timestamp(),
+                exp: Utc::now().timestamp() + 60,
+            },
+        );
+        assert_eq!(
+            a.authorize_subscribe(app_b, "chat", Some(&token)).await,
+            Err(SubscribeDenied::Unauthorized)
+        );
+    }
+}

crates/manager-core/src/topic_repo.rs

@@ -0,0 +1,212 @@
+//! `TopicRepo` — CRUD for the `topics` table (v1.1.6).
+//!
+//! This table holds ONLY topics that have been explicitly externalized
+//! for SSE subscription (design notes §5). Internal-only pub/sub topics
+//! stay implicit — they never get a row here, and the publish path never
+//! consults this table. The two readers are the topic admin endpoints
+//! ([`crate::topics_api`]) and the SSE subscribe authorization
+//! ([`crate::realtime_authority`]).
+
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use picloud_shared::AppId;
+use serde::{Deserialize, Serialize};
+use sqlx::PgPool;
+
+/// External-subscriber auth gate for a topic. `'public'` + `'token'` in
+/// v1.1.6; `'session'` (v1.1.8) and `'script'` (v1.2) extend the DB
+/// CHECK constraint and this enum later.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum TopicAuthMode {
+    Public,
+    Token,
+}
+
+impl TopicAuthMode {
+    #[must_use]
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::Public => "public",
+            Self::Token => "token",
+        }
+    }
+
+    fn from_db(s: &str) -> Result<Self, TopicRepoError> {
+        match s {
+            "public" => Ok(Self::Public),
+            "token" => Ok(Self::Token),
+            other => Err(TopicRepoError::Backend(format!(
+                "unknown auth_mode in DB: {other}"
+            ))),
+        }
+    }
+}
+
+/// A registered, externally-subscribable topic row.
+#[derive(Debug, Clone, Serialize)]
+pub struct Topic {
+    pub name: String,
+    pub external_subscribable: bool,
+    pub auth_mode: TopicAuthMode,
+    pub created_at: DateTime<Utc>,
+    pub updated_at: DateTime<Utc>,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum TopicRepoError {
+    #[error("a topic named {0:?} already exists in this app")]
+    AlreadyExists(String),
+    #[error("database error: {0}")]
+    Db(#[from] sqlx::Error),
+    #[error("topic backend error: {0}")]
+    Backend(String),
+}
+
+#[async_trait]
+pub trait TopicRepo: Send + Sync {
+    /// Register a topic. Errors `AlreadyExists` on PK conflict.
+    async fn create(
+        &self,
+        app_id: AppId,
+        name: &str,
+        external_subscribable: bool,
+        auth_mode: TopicAuthMode,
+    ) -> Result<Topic, TopicRepoError>;
+
+    /// List every registered topic in the app, ordered by name.
+    async fn list(&self, app_id: AppId) -> Result<Vec<Topic>, TopicRepoError>;
+
+    /// Fetch one topic by name, `None` if not registered.
+    async fn get(&self, app_id: AppId, name: &str) -> Result<Option<Topic>, TopicRepoError>;
+
+    /// Update `external_subscribable` and/or `auth_mode` (each `None`
+    /// leaves the column unchanged). `None` return = no such topic.
+    async fn update(
+        &self,
+        app_id: AppId,
+        name: &str,
+        external_subscribable: Option<bool>,
+        auth_mode: Option<TopicAuthMode>,
+    ) -> Result<Option<Topic>, TopicRepoError>;
+
+    /// Unregister a topic. Returns `true` if a row was removed.
+    async fn delete(&self, app_id: AppId, name: &str) -> Result<bool, TopicRepoError>;
+}
+
+#[derive(sqlx::FromRow)]
+struct TopicRow {
+    name: String,
+    external_subscribable: bool,
+    auth_mode: String,
+    created_at: DateTime<Utc>,
+    updated_at: DateTime<Utc>,
+}
+
+impl TopicRow {
+    fn into_topic(self) -> Result<Topic, TopicRepoError> {
+        Ok(Topic {
+            auth_mode: TopicAuthMode::from_db(&self.auth_mode)?,
+            name: self.name,
+            external_subscribable: self.external_subscribable,
+            created_at: self.created_at,
+            updated_at: self.updated_at,
+        })
+    }
+}
+
+const SELECT_COLS: &str = "name, external_subscribable, auth_mode, created_at, updated_at";
+
+pub struct PostgresTopicRepo {
+    pool: PgPool,
+}
+
+impl PostgresTopicRepo {
+    #[must_use]
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+}
+
+#[async_trait]
+impl TopicRepo for PostgresTopicRepo {
+    async fn create(
+        &self,
+        app_id: AppId,
+        name: &str,
+        external_subscribable: bool,
+        auth_mode: TopicAuthMode,
+    ) -> Result<Topic, TopicRepoError> {
+        let row: Option<TopicRow> = sqlx::query_as(&format!(
+            "INSERT INTO topics (app_id, name, external_subscribable, auth_mode) \
+             VALUES ($1, $2, $3, $4) \
+             ON CONFLICT (app_id, name) DO NOTHING \
+             RETURNING {SELECT_COLS}"
+        ))
+        .bind(app_id.into_inner())
+        .bind(name)
+        .bind(external_subscribable)
+        .bind(auth_mode.as_str())
+        .fetch_optional(&self.pool)
+        .await?;
+        match row {
+            Some(r) => r.into_topic(),
+            None => Err(TopicRepoError::AlreadyExists(name.to_string())),
+        }
+    }
+
+    async fn list(&self, app_id: AppId) -> Result<Vec<Topic>, TopicRepoError> {
+        let rows: Vec<TopicRow> = sqlx::query_as(&format!(
+            "SELECT {SELECT_COLS} FROM topics WHERE app_id = $1 ORDER BY name"
+        ))
+        .bind(app_id.into_inner())
+        .fetch_all(&self.pool)
+        .await?;
+        rows.into_iter().map(TopicRow::into_topic).collect()
+    }
+
+    async fn get(&self, app_id: AppId, name: &str) -> Result<Option<Topic>, TopicRepoError> {
+        let row: Option<TopicRow> = sqlx::query_as(&format!(
+            "SELECT {SELECT_COLS} FROM topics WHERE app_id = $1 AND name = $2"
+        ))
+        .bind(app_id.into_inner())
+        .bind(name)
+        .fetch_optional(&self.pool)
+        .await?;
+        row.map(TopicRow::into_topic).transpose()
+    }
+
+    async fn update(
+        &self,
+        app_id: AppId,
+        name: &str,
+        external_subscribable: Option<bool>,
+        auth_mode: Option<TopicAuthMode>,
+    ) -> Result<Option<Topic>, TopicRepoError> {
+        // COALESCE leaves a column untouched when its bind is NULL.
+        let row: Option<TopicRow> = sqlx::query_as(&format!(
+            "UPDATE topics SET \
+                external_subscribable = COALESCE($3, external_subscribable), \
+                auth_mode = COALESCE($4, auth_mode), \
+                updated_at = NOW() \
+             WHERE app_id = $1 AND name = $2 \
+             RETURNING {SELECT_COLS}"
+        ))
+        .bind(app_id.into_inner())
+        .bind(name)
+        .bind(external_subscribable)
+        .bind(auth_mode.map(|m| m.as_str()))
+        .fetch_optional(&self.pool)
+        .await?;
+        row.map(TopicRow::into_topic).transpose()
+    }
+
+    async fn delete(&self, app_id: AppId, name: &str) -> Result<bool, TopicRepoError> {
+        let res = sqlx::query("DELETE FROM topics WHERE app_id = $1 AND name = $2")
+            .bind(app_id.into_inner())
+            .bind(name)
+            .execute(&self.pool)
+            .await?;
+        Ok(res.rows_affected() > 0)
+    }
+}

crates/manager-core/src/topics_api.rs

@@ -0,0 +1,629 @@
+//! `/api/v1/admin/apps/{id}/topics*` — topic registration admin
+//! endpoints (v1.1.6).
+//!
+//! These manage the `topics` table: the explicit registry of which
+//! pub/sub topics are externally subscribable over SSE (design notes
+//! §5). Internal-only topics never appear here.
+//!
+//!   * `POST   /apps/{id}/topics`            — register a topic.
+//!   * `GET    /apps/{id}/topics`            — list registered topics.
+//!   * `PATCH  /apps/{id}/topics/{name}`     — flip external/auth_mode.
+//!   * `DELETE /apps/{id}/topics/{name}`     — unregister + disconnect.
+//!
+//! The PATCH endpoint is deliberately its OWN surface (not folded into a
+//! generic topic update) so every change to externally-subscribable
+//! status is a discrete, watchable/auditable API call (§5 commitment).
+//!
+//! Create / update / delete are gated by `AppTopicManage` (→ `app:admin`
+//! scope); list is gated by the existing `AppRead`. DELETE also drops
+//! the topic's in-process broadcast channel so live SSE subscribers
+//! disconnect cleanly.
+
+use std::sync::Arc;
+
+use axum::extract::{Path, State};
+use axum::http::StatusCode;
+use axum::response::{IntoResponse, Json, Response};
+use axum::routing::{get, patch};
+use axum::{Extension, Router};
+use picloud_shared::{AppId, Principal, RealtimeBroadcaster};
+use serde::Deserialize;
+use serde_json::json;
+
+use crate::app_repo::AppRepository;
+use crate::authz::{require, AuthzDenied, AuthzError, AuthzRepo, Capability};
+use crate::topic_repo::{Topic, TopicAuthMode, TopicRepo, TopicRepoError};
+
+#[derive(Clone)]
+pub struct TopicsState {
+    pub topics: Arc<dyn TopicRepo>,
+    pub apps: Arc<dyn AppRepository>,
+    pub authz: Arc<dyn AuthzRepo>,
+    pub broadcaster: Arc<dyn RealtimeBroadcaster>,
+}
+
+pub fn topics_router(state: TopicsState) -> Router {
+    Router::new()
+        .route("/apps/{app_id}/topics", get(list_topics).post(create_topic))
+        .route(
+            "/apps/{app_id}/topics/{name}",
+            patch(update_topic).delete(delete_topic),
+        )
+        .with_state(state)
+}
+
+#[derive(Debug, Deserialize)]
+pub struct CreateTopicRequest {
+    pub name: String,
+    #[serde(default)]
+    pub external_subscribable: bool,
+    #[serde(default = "default_auth_mode")]
+    pub auth_mode: TopicAuthMode,
+}
+
+const fn default_auth_mode() -> TopicAuthMode {
+    TopicAuthMode::Public
+}
+
+#[derive(Debug, Deserialize)]
+pub struct UpdateTopicRequest {
+    #[serde(default)]
+    pub external_subscribable: Option<bool>,
+    #[serde(default)]
+    pub auth_mode: Option<TopicAuthMode>,
+}
+
+/// Topic names are concrete (external pattern subscription is v1.2), so
+/// reject empties and `*` wildcards at registration.
+fn validate_topic_name(name: &str) -> Result<(), TopicsApiError> {
+    if name.trim().is_empty() {
+        return Err(TopicsApiError::Invalid(
+            "topic name must not be empty".into(),
+        ));
+    }
+    if name.contains('*') {
+        return Err(TopicsApiError::Invalid(
+            "topic name must be a concrete topic, not a pattern (no '*')".into(),
+        ));
+    }
+    Ok(())
+}
+
+async fn create_topic(
+    State(s): State<TopicsState>,
+    Extension(principal): Extension<Principal>,
+    Path(app_id): Path<AppId>,
+    Json(input): Json<CreateTopicRequest>,
+) -> Result<(StatusCode, Json<Topic>), TopicsApiError> {
+    ensure_app_exists(&*s.apps, app_id).await?;
+    require(
+        s.authz.as_ref(),
+        &principal,
+        Capability::AppTopicManage(app_id),
+    )
+    .await?;
+    validate_topic_name(&input.name)?;
+    let topic = s
+        .topics
+        .create(
+            app_id,
+            input.name.trim(),
+            input.external_subscribable,
+            input.auth_mode,
+        )
+        .await?;
+    Ok((StatusCode::CREATED, Json(topic)))
+}
+
+#[derive(Debug, serde::Serialize)]
+struct ListTopicsResponse {
+    topics: Vec<Topic>,
+}
+
+async fn list_topics(
+    State(s): State<TopicsState>,
+    Extension(principal): Extension<Principal>,
+    Path(app_id): Path<AppId>,
+) -> Result<Json<ListTopicsResponse>, TopicsApiError> {
+    ensure_app_exists(&*s.apps, app_id).await?;
+    require(s.authz.as_ref(), &principal, Capability::AppRead(app_id)).await?;
+    let topics = s.topics.list(app_id).await?;
+    Ok(Json(ListTopicsResponse { topics }))
+}
+
+async fn update_topic(
+    State(s): State<TopicsState>,
+    Extension(principal): Extension<Principal>,
+    Path((app_id, name)): Path<(AppId, String)>,
+    Json(input): Json<UpdateTopicRequest>,
+) -> Result<Json<Topic>, TopicsApiError> {
+    ensure_app_exists(&*s.apps, app_id).await?;
+    require(
+        s.authz.as_ref(),
+        &principal,
+        Capability::AppTopicManage(app_id),
+    )
+    .await?;
+    let topic = s
+        .topics
+        .update(app_id, &name, input.external_subscribable, input.auth_mode)
+        .await?
+        .ok_or(TopicsApiError::NotFound)?;
+    Ok(Json(topic))
+}
+
+async fn delete_topic(
+    State(s): State<TopicsState>,
+    Extension(principal): Extension<Principal>,
+    Path((app_id, name)): Path<(AppId, String)>,
+) -> Result<StatusCode, TopicsApiError> {
+    ensure_app_exists(&*s.apps, app_id).await?;
+    require(
+        s.authz.as_ref(),
+        &principal,
+        Capability::AppTopicManage(app_id),
+    )
+    .await?;
+    if !s.topics.delete(app_id, &name).await? {
+        return Err(TopicsApiError::NotFound);
+    }
+    // Disconnect any live SSE subscribers for the now-unregistered topic.
+    s.broadcaster.drop_topic(app_id, &name).await;
+    Ok(StatusCode::NO_CONTENT)
+}
+
+async fn ensure_app_exists(apps: &dyn AppRepository, app_id: AppId) -> Result<(), TopicsApiError> {
+    apps.get_by_id(app_id)
+        .await
+        .map_err(|e| TopicsApiError::Backend(e.to_string()))?
+        .ok_or(TopicsApiError::AppNotFound)?;
+    Ok(())
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum TopicsApiError {
+    #[error("app not found")]
+    AppNotFound,
+    #[error("topic not found")]
+    NotFound,
+    #[error("{0}")]
+    AlreadyExists(String),
+    #[error("invalid request: {0}")]
+    Invalid(String),
+    #[error("forbidden")]
+    Forbidden,
+    #[error("authorization repo error: {0}")]
+    AuthzRepo(String),
+    #[error("topics backend: {0}")]
+    Backend(String),
+}
+
+impl From<AuthzDenied> for TopicsApiError {
+    fn from(d: AuthzDenied) -> Self {
+        match d {
+            AuthzDenied::Denied => Self::Forbidden,
+            AuthzDenied::Repo(e) => Self::AuthzRepo(e.to_string()),
+        }
+    }
+}
+
+impl From<AuthzError> for TopicsApiError {
+    fn from(e: AuthzError) -> Self {
+        Self::AuthzRepo(e.to_string())
+    }
+}
+
+impl From<TopicRepoError> for TopicsApiError {
+    fn from(e: TopicRepoError) -> Self {
+        match e {
+            TopicRepoError::AlreadyExists(name) => {
+                Self::AlreadyExists(format!("a topic named {name:?} already exists in this app"))
+            }
+            other => Self::Backend(other.to_string()),
+        }
+    }
+}
+
+impl IntoResponse for TopicsApiError {
+    fn into_response(self) -> Response {
+        let (status, body) = match &self {
+            Self::AppNotFound | Self::NotFound => {
+                (StatusCode::NOT_FOUND, json!({ "error": self.to_string() }))
+            }
+            Self::AlreadyExists(_) => (StatusCode::CONFLICT, json!({ "error": self.to_string() })),
+            Self::Invalid(_) => (
+                StatusCode::UNPROCESSABLE_ENTITY,
+                json!({ "error": self.to_string() }),
+            ),
+            Self::Forbidden => (StatusCode::FORBIDDEN, json!({ "error": self.to_string() })),
+            Self::AuthzRepo(e) => {
+                tracing::error!(error = %e, "topics admin authz repo error");
+                (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    json!({ "error": "internal error" }),
+                )
+            }
+            Self::Backend(e) => {
+                tracing::error!(error = %e, "topics admin backend error");
+                (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    json!({ "error": "internal error" }),
+                )
+            }
+        };
+        (status, Json(body)).into_response()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    //! In-memory handler tests: capability enforcement, the
+    //! `external_subscribable` default, the flip being its own endpoint,
+    //! cross-app isolation, and DELETE disconnecting subscribers. The
+    //! Postgres repo is exercised by the schema + integration suites.
+
+    use super::*;
+    use crate::repo::ScriptRepositoryError;
+    use async_trait::async_trait;
+    use chrono::Utc;
+    use picloud_shared::{
+        AdminUserId, App, AppRole, BroadcasterError, InstanceRole, RealtimeEvent, UserId,
+    };
+    use std::collections::HashMap;
+    use std::sync::Mutex as StdMutex;
+    use tokio::sync::Mutex;
+
+    #[derive(Default)]
+    struct InMemoryTopicRepo {
+        inner: Mutex<HashMap<(AppId, String), Topic>>,
+    }
+
+    #[async_trait]
+    impl TopicRepo for InMemoryTopicRepo {
+        async fn create(
+            &self,
+            app_id: AppId,
+            name: &str,
+            external_subscribable: bool,
+            auth_mode: TopicAuthMode,
+        ) -> Result<Topic, TopicRepoError> {
+            let mut g = self.inner.lock().await;
+            if g.contains_key(&(app_id, name.to_string())) {
+                return Err(TopicRepoError::AlreadyExists(name.to_string()));
+            }
+            let now = Utc::now();
+            let t = Topic {
+                name: name.to_string(),
+                external_subscribable,
+                auth_mode,
+                created_at: now,
+                updated_at: now,
+            };
+            g.insert((app_id, name.to_string()), t.clone());
+            Ok(t)
+        }
+        async fn list(&self, app_id: AppId) -> Result<Vec<Topic>, TopicRepoError> {
+            let g = self.inner.lock().await;
+            let mut v: Vec<Topic> = g
+                .iter()
+                .filter(|((a, _), _)| *a == app_id)
+                .map(|(_, t)| t.clone())
+                .collect();
+            v.sort_by(|a, b| a.name.cmp(&b.name));
+            Ok(v)
+        }
+        async fn get(&self, app_id: AppId, name: &str) -> Result<Option<Topic>, TopicRepoError> {
+            Ok(self
+                .inner
+                .lock()
+                .await
+                .get(&(app_id, name.to_string()))
+                .cloned())
+        }
+        async fn update(
+            &self,
+            app_id: AppId,
+            name: &str,
+            external_subscribable: Option<bool>,
+            auth_mode: Option<TopicAuthMode>,
+        ) -> Result<Option<Topic>, TopicRepoError> {
+            let mut g = self.inner.lock().await;
+            let Some(t) = g.get_mut(&(app_id, name.to_string())) else {
+                return Ok(None);
+            };
+            if let Some(e) = external_subscribable {
+                t.external_subscribable = e;
+            }
+            if let Some(m) = auth_mode {
+                t.auth_mode = m;
+            }
+            t.updated_at = Utc::now();
+            Ok(Some(t.clone()))
+        }
+        async fn delete(&self, app_id: AppId, name: &str) -> Result<bool, TopicRepoError> {
+            Ok(self
+                .inner
+                .lock()
+                .await
+                .remove(&(app_id, name.to_string()))
+                .is_some())
+        }
+    }
+
+    struct InMemoryAppRepo(AppId);
+    #[async_trait]
+    impl AppRepository for InMemoryAppRepo {
+        async fn list(&self) -> Result<Vec<App>, ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn list_for_user(&self, _: AdminUserId) -> Result<Vec<App>, ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn get_by_id(&self, id: AppId) -> Result<Option<App>, ScriptRepositoryError> {
+            if id != self.0 {
+                return Ok(None);
+            }
+            let now = Utc::now();
+            Ok(Some(App {
+                id,
+                slug: "test".into(),
+                name: "test".into(),
+                description: None,
+                created_at: now,
+                updated_at: now,
+            }))
+        }
+        async fn get_by_slug(&self, _: &str) -> Result<Option<App>, ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn get_by_slug_or_history(
+            &self,
+            _: &str,
+        ) -> Result<Option<crate::app_repo::AppLookup>, ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn slug_in_history(&self, _: &str) -> Result<Option<App>, ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn create(
+            &self,
+            _: &str,
+            _: &str,
+            _: Option<&str>,
+        ) -> Result<App, ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn create_with_takeover(
+            &self,
+            _: &str,
+            _: &str,
+            _: Option<&str>,
+        ) -> Result<App, ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn update(
+            &self,
+            _: AppId,
+            _: Option<&str>,
+            _: Option<Option<&str>>,
+        ) -> Result<App, ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn rename_slug(
+            &self,
+            _: AppId,
+            _: &str,
+            _: bool,
+        ) -> Result<App, ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn delete(&self, _: AppId) -> Result<(), ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn delete_cascade(&self, _: AppId) -> Result<(), ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn count_scripts_in_app(&self, _: AppId) -> Result<i64, ScriptRepositoryError> {
+            unimplemented!()
+        }
+    }
+
+    /// Grants `AppAdmin` only for `granted_app`; denies elsewhere — used
+    /// for the cross-app isolation test.
+    struct PerAppAuthzRepo {
+        granted_app: AppId,
+    }
+    #[async_trait]
+    impl AuthzRepo for PerAppAuthzRepo {
+        async fn membership(
+            &self,
+            _: UserId,
+            app_id: AppId,
+        ) -> Result<Option<AppRole>, AuthzError> {
+            Ok((app_id == self.granted_app).then_some(AppRole::AppAdmin))
+        }
+    }
+
+    struct DenyAuthzRepo;
+    #[async_trait]
+    impl AuthzRepo for DenyAuthzRepo {
+        async fn membership(&self, _: UserId, _: AppId) -> Result<Option<AppRole>, AuthzError> {
+            Ok(None)
+        }
+    }
+
+    #[derive(Default)]
+    struct RecordingBroadcaster {
+        dropped: StdMutex<Vec<(AppId, String)>>,
+    }
+    #[async_trait]
+    impl RealtimeBroadcaster for RecordingBroadcaster {
+        async fn subscribe(
+            &self,
+            _: AppId,
+            _: &str,
+        ) -> Result<tokio::sync::broadcast::Receiver<RealtimeEvent>, BroadcasterError> {
+            unimplemented!()
+        }
+        async fn publish(&self, _: AppId, _: &str, _: RealtimeEvent) {}
+        async fn drop_topic(&self, app_id: AppId, topic: &str) {
+            self.dropped
+                .lock()
+                .unwrap()
+                .push((app_id, topic.to_string()));
+        }
+    }
+
+    fn member() -> Principal {
+        Principal {
+            user_id: AdminUserId::new(),
+            instance_role: InstanceRole::Member,
+            scopes: None,
+            app_binding: None,
+        }
+    }
+
+    fn state(app_id: AppId, authz: Arc<dyn AuthzRepo>) -> (TopicsState, Arc<RecordingBroadcaster>) {
+        let bc = Arc::new(RecordingBroadcaster::default());
+        let state = TopicsState {
+            topics: Arc::new(InMemoryTopicRepo::default()),
+            apps: Arc::new(InMemoryAppRepo(app_id)),
+            authz,
+            broadcaster: bc.clone(),
+        };
+        (state, bc)
+    }
+
+    #[tokio::test]
+    async fn register_defaults_external_subscribable_false() {
+        let app = AppId::new();
+        let (s, _) = state(app, Arc::new(PerAppAuthzRepo { granted_app: app }));
+        let (status, Json(topic)) = create_topic(
+            State(s),
+            Extension(member()),
+            Path(app),
+            Json(CreateTopicRequest {
+                name: "chat".into(),
+                external_subscribable: false,
+                auth_mode: TopicAuthMode::Public,
+            }),
+        )
+        .await
+        .unwrap();
+        assert_eq!(status, StatusCode::CREATED);
+        assert!(!topic.external_subscribable);
+        assert_eq!(topic.name, "chat");
+    }
+
+    #[tokio::test]
+    async fn flip_requires_app_admin_role() {
+        let app = AppId::new();
+        // Topic exists; the caller has no role → PATCH is forbidden.
+        let (s, _) = state(app, Arc::new(DenyAuthzRepo));
+        s.topics
+            .create(app, "chat", false, TopicAuthMode::Public)
+            .await
+            .unwrap();
+        let err = update_topic(
+            State(s),
+            Extension(member()),
+            Path((app, "chat".to_string())),
+            Json(UpdateTopicRequest {
+                external_subscribable: Some(true),
+                auth_mode: None,
+            }),
+        )
+        .await
+        .unwrap_err();
+        assert!(matches!(err, TopicsApiError::Forbidden));
+    }
+
+    #[tokio::test]
+    async fn flip_is_its_own_endpoint_and_toggles_external() {
+        // The PATCH handler is a distinct surface from create; flipping
+        // external_subscribable false→true is a single discrete call.
+        let app = AppId::new();
+        let (s, _) = state(app, Arc::new(PerAppAuthzRepo { granted_app: app }));
+        s.topics
+            .create(app, "chat", false, TopicAuthMode::Public)
+            .await
+            .unwrap();
+        let Json(updated) = update_topic(
+            State(s),
+            Extension(member()),
+            Path((app, "chat".to_string())),
+            Json(UpdateTopicRequest {
+                external_subscribable: Some(true),
+                auth_mode: Some(TopicAuthMode::Token),
+            }),
+        )
+        .await
+        .unwrap();
+        assert!(updated.external_subscribable);
+        assert_eq!(updated.auth_mode, TopicAuthMode::Token);
+    }
+
+    #[tokio::test]
+    async fn delete_disconnects_subscribers() {
+        let app = AppId::new();
+        let (s, bc) = state(app, Arc::new(PerAppAuthzRepo { granted_app: app }));
+        s.topics
+            .create(app, "chat", true, TopicAuthMode::Public)
+            .await
+            .unwrap();
+        let status = delete_topic(
+            State(s),
+            Extension(member()),
+            Path((app, "chat".to_string())),
+        )
+        .await
+        .unwrap();
+        assert_eq!(status, StatusCode::NO_CONTENT);
+        assert_eq!(
+            bc.dropped.lock().unwrap().as_slice(),
+            &[(app, "chat".to_string())]
+        );
+    }
+
+    #[tokio::test]
+    async fn cross_app_admin_cannot_manage_other_app() {
+        let app_a = AppId::new();
+        let app_b = AppId::new();
+        // Caller is admin of app A only; both apps exist via separate state.
+        let authz = Arc::new(PerAppAuthzRepo { granted_app: app_a });
+        // App-B-scoped state, but the caller only has A's grant.
+        let (s, _) = state(app_b, authz);
+        let err = create_topic(
+            State(s),
+            Extension(member()),
+            Path(app_b),
+            Json(CreateTopicRequest {
+                name: "chat".into(),
+                external_subscribable: true,
+                auth_mode: TopicAuthMode::Public,
+            }),
+        )
+        .await
+        .unwrap_err();
+        assert!(matches!(err, TopicsApiError::Forbidden));
+    }
+
+    #[tokio::test]
+    async fn pattern_name_rejected() {
+        let app = AppId::new();
+        let (s, _) = state(app, Arc::new(PerAppAuthzRepo { granted_app: app }));
+        let err = create_topic(
+            State(s),
+            Extension(member()),
+            Path(app),
+            Json(CreateTopicRequest {
+                name: "user.*".into(),
+                external_subscribable: true,
+                auth_mode: TopicAuthMode::Public,
+            }),
+        )
+        .await
+        .unwrap_err();
+        assert!(matches!(err, TopicsApiError::Invalid(_)));
+    }
+}