test/docs(shared-topics): dispatch test + journey + docs (D2)

Deterministic manager-core test (tests/shared_topics.rs) proves the
namespace boundary both ways: a shared publish fans out only to the
group's shared pubsub trigger (not a per-app or non-shared group
template), and a per-app publish never hits the shared trigger. CLI
journey (shared_topics) covers authoring + `pic triggers ls --group`
shared column + the two validation rejections (undeclared topic; shared
on an app), mirroring the shared_triggers norm. Docs: CLAUDE.md + design
doc §11.6 record D1 + D2 as shipped.

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

crates/manager-core/tests/shared_topics.rs

@@ -0,0 +1,225 @@
+//! §11.6 D2 integration test: SHARED-topic pubsub fan-out.
+//!
+//! A group-owned `shared = true` pubsub trigger watches the group's shared
+//! topic. A SHARED publish (`fan_out_shared_publish` on the owning group) fans
+//! out to it — but NOT to a per-app pubsub trigger or a non-shared group
+//! template of the same pattern. Conversely a per-app publish
+//! (`fan_out_publish`) fans out to the app's own + inherited non-shared group
+//! templates but NEVER the shared trigger (`shared` is the namespace boundary).
+//!
+//! Deterministic: drives the repo fan-out directly (no async dispatcher). Skips
+//! when `DATABASE_URL` is unset.
+
+#![allow(clippy::too_many_lines)]
+
+use picloud_manager_core::pubsub_repo::{PostgresPubsubRepo, PublishCtx, PubsubRepo};
+use picloud_shared::{AppId, ExecutionId, GroupId};
+use sqlx::postgres::PgPoolOptions;
+use sqlx::PgPool;
+use uuid::Uuid;
+
+async fn pool_or_skip() -> Option<PgPool> {
+    let Ok(url) = std::env::var("DATABASE_URL") else {
+        eprintln!("shared_topics: DATABASE_URL unset — skipping");
+        return None;
+    };
+    let pool = PgPoolOptions::new()
+        .max_connections(2)
+        .connect(&url)
+        .await
+        .expect("connect");
+    sqlx::migrate!("./migrations")
+        .run(&pool)
+        .await
+        .expect("migrate");
+    Some(pool)
+}
+
+/// Insert a pubsub trigger (owner + `shared` flag) + its details; returns id.
+async fn pubsub_trigger(
+    pool: &PgPool,
+    app_id: Option<Uuid>,
+    group_id: Option<Uuid>,
+    script: Uuid,
+    admin: Uuid,
+    shared: bool,
+    pattern: &str,
+) -> Uuid {
+    let row: (Uuid,) = sqlx::query_as(
+        "INSERT INTO triggers \
+            (app_id, group_id, script_id, kind, enabled, dispatch_mode, \
+             retry_max_attempts, retry_backoff, retry_base_ms, \
+             registered_by_principal, name, shared) \
+         VALUES ($1, $2, $3, 'pubsub', TRUE, 'async', 3, 'exponential', 1000, $4, $5, $6) \
+         RETURNING id",
+    )
+    .bind(app_id)
+    .bind(group_id)
+    .bind(script)
+    .bind(admin)
+    .bind(Uuid::new_v4().simple().to_string())
+    .bind(shared)
+    .fetch_one(pool)
+    .await
+    .unwrap();
+    sqlx::query("INSERT INTO pubsub_trigger_details (trigger_id, topic_pattern) VALUES ($1, $2)")
+        .bind(row.0)
+        .bind(pattern)
+        .execute(pool)
+        .await
+        .unwrap();
+    row.0
+}
+
+async fn outbox_count(pool: &PgPool, trigger_id: Uuid) -> i64 {
+    let (n,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM outbox WHERE trigger_id = $1")
+        .bind(trigger_id)
+        .fetch_one(pool)
+        .await
+        .unwrap();
+    n
+}
+
+fn ctx(app: Uuid) -> PublishCtx {
+    PublishCtx {
+        app_id: AppId::from(app),
+        origin_principal: None,
+        trigger_depth: 0,
+        root_execution_id: ExecutionId::new(),
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn shared_topic_fan_out_respects_the_namespace_boundary() {
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
+    let sfx = Uuid::new_v4().simple().to_string();
+    let admin: (Uuid,) = sqlx::query_as(
+        "INSERT INTO admin_users (username, password_hash) VALUES ($1, 'x') RETURNING id",
+    )
+    .bind(format!("st-{sfx}"))
+    .fetch_one(&pool)
+    .await
+    .unwrap();
+    let g: (Uuid,) = sqlx::query_as("INSERT INTO groups (slug, name) VALUES ($1, $1) RETURNING id")
+        .bind(format!("st-g-{sfx}"))
+        .fetch_one(&pool)
+        .await
+        .unwrap();
+    let handler: (Uuid,) = sqlx::query_as(
+        "INSERT INTO scripts (name, source, group_id) VALUES ($1, 'x', $2) RETURNING id",
+    )
+    .bind(format!("onmsg-{sfx}"))
+    .bind(g.0)
+    .fetch_one(&pool)
+    .await
+    .unwrap();
+    let app: (Uuid,) =
+        sqlx::query_as("INSERT INTO apps (slug, name, group_id) VALUES ($1, $1, $2) RETURNING id")
+            .bind(format!("st-a-{sfx}"))
+            .bind(g.0)
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+    let app_script: (Uuid,) = sqlx::query_as(
+        "INSERT INTO scripts (name, source, app_id) VALUES ($1, 'x', $2) RETURNING id",
+    )
+    .bind(format!("appmsg-{sfx}"))
+    .bind(app.0)
+    .fetch_one(&pool)
+    .await
+    .unwrap();
+
+    // Three triggers, all watching `events.*`:
+    let shared = pubsub_trigger(&pool, None, Some(g.0), handler.0, admin.0, true, "events.*").await;
+    let group_tmpl = pubsub_trigger(
+        &pool,
+        None,
+        Some(g.0),
+        handler.0,
+        admin.0,
+        false,
+        "events.*",
+    )
+    .await;
+    let per_app = pubsub_trigger(
+        &pool,
+        Some(app.0),
+        None,
+        app_script.0,
+        admin.0,
+        false,
+        "events.*",
+    )
+    .await;
+
+    let repo = PostgresPubsubRepo::new(pool.clone());
+    let payload = serde_json::json!({"topic": "events.created", "message": 1});
+
+    // SHARED publish → only the shared trigger fires.
+    let n = repo
+        .fan_out_shared_publish(
+            ctx(app.0),
+            GroupId::from(g.0),
+            "events.created",
+            payload.clone(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(n, 1, "shared publish matched exactly the shared trigger");
+    assert_eq!(outbox_count(&pool, shared).await, 1, "shared trigger fired");
+    assert_eq!(
+        outbox_count(&pool, group_tmpl).await,
+        0,
+        "a non-shared group template must NOT fire on a shared publish"
+    );
+    assert_eq!(
+        outbox_count(&pool, per_app).await,
+        0,
+        "a per-app trigger must NOT fire on a shared publish"
+    );
+
+    // PER-APP publish → the app's own + inherited non-shared template fire, but
+    // NEVER the shared trigger.
+    repo.fan_out_publish(ctx(app.0), "events.created", payload)
+        .await
+        .unwrap();
+    assert_eq!(
+        outbox_count(&pool, shared).await,
+        1,
+        "the shared trigger must NOT fire on a per-app publish (still 1 from before)"
+    );
+    assert_eq!(
+        outbox_count(&pool, group_tmpl).await,
+        1,
+        "the inherited non-shared group template fires on a per-app publish"
+    );
+    assert_eq!(
+        outbox_count(&pool, per_app).await,
+        1,
+        "the per-app trigger fires on a per-app publish"
+    );
+
+    // Cleanup.
+    let _ = sqlx::query("DELETE FROM triggers WHERE registered_by_principal = $1")
+        .bind(admin.0)
+        .execute(&pool)
+        .await;
+    let _ = sqlx::query("DELETE FROM apps WHERE group_id = $1")
+        .bind(g.0)
+        .execute(&pool)
+        .await;
+    let _ = sqlx::query("DELETE FROM scripts WHERE group_id = $1")
+        .bind(g.0)
+        .execute(&pool)
+        .await;
+    let _ = sqlx::query("DELETE FROM groups WHERE id = $1")
+        .bind(g.0)
+        .execute(&pool)
+        .await;
+    let _ = sqlx::query("DELETE FROM admin_users WHERE id = $1")
+        .bind(admin.0)
+        .execute(&pool)
+        .await;
+}

crates/picloud-cli/tests/cli.rs

@@ -43,6 +43,7 @@ mod routes;
 mod scripts;
 mod sealed;
 mod secrets;
+mod shared_topics;
 mod shared_triggers;
 mod staleness;
 mod stateful_templates;

crates/picloud-cli/tests/shared_topics.rs

@@ -0,0 +1,159 @@
+//! §11.6 D2 — shared TOPICS + shared pub/sub triggers, declarative authoring
+//! end to end via `pic`. A group declares a shared `topic` collection + a
+//! `[[triggers.pubsub]] shared = true` handler watching it; the template
+//! applies, `pic triggers ls --group` shows `shared = true`; a shared pubsub
+//! trigger on an UNDECLARED topic is rejected, and `shared = true` on an app
+//! pubsub trigger is rejected.
+//!
+//! The live FIRING (a `pubsub::shared_topic(...).publish(...)` matches the
+//! shared trigger, a per-app publish does not) is pinned deterministically by
+//! `manager-core/tests/shared_topics.rs` — driving the async dispatcher from a
+//! journey is deliberately avoided (codebase norm, mirroring `shared_triggers`).
+
+use std::fs;
+
+use tempfile::TempDir;
+
+use crate::common;
+use crate::common::cleanup::{AppGuard, GroupGuard, ScriptGuard};
+
+fn manifest_dir() -> TempDir {
+    let dir = TempDir::new().expect("tempdir");
+    fs::create_dir_all(dir.path().join("scripts")).expect("scripts dir");
+    dir
+}
+
+fn group_script_id(env: &common::TestEnv, group: &str, name: &str) -> String {
+    let ls = common::pic_as(env)
+        .args(["scripts", "ls", "--group", group])
+        .output()
+        .expect("scripts ls");
+    let table = String::from_utf8(ls.stdout).unwrap();
+    table
+        .lines()
+        .map(common::cells)
+        .find(|c| c.get(2) == Some(&name))
+        .and_then(|c| c.first().map(|s| (*s).to_string()))
+        .unwrap_or_else(|| panic!("group script `{name}` not found:\n{table}"))
+}
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn shared_topic_trigger_applies_lists_and_validates() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let group = common::unique_slug("shtop-grp");
+
+    let _g = GroupGuard::new(&env.url, &env.token, &group);
+    common::pic_as(&env)
+        .args(["groups", "create", &group])
+        .assert()
+        .success();
+
+    // Group handler for the shared topic.
+    let dir = manifest_dir();
+    fs::write(
+        dir.path().join("scripts/on-evt.rhai"),
+        r#"log::info("shared topic event"); "ok""#,
+    )
+    .unwrap();
+    common::pic_as(&env)
+        .args(["scripts", "deploy"])
+        .arg(dir.path().join("scripts/on-evt.rhai"))
+        .args(["--group", &group, "--name", "on-evt"])
+        .assert()
+        .success();
+    let _gs = ScriptGuard::new(
+        &env.url,
+        &env.token,
+        &group_script_id(&env, &group, "on-evt"),
+    );
+
+    // Group declares a shared `topic` collection `events` + a shared pubsub
+    // trigger watching `events.*`.
+    let gmanifest = format!(
+        "[group]\nslug = \"{group}\"\nname = \"ShTopG\"\n\
+         collections = [{{ name = \"events\", kind = \"topic\" }}]\n\n\
+         [[triggers.pubsub]]\nscript = \"on-evt\"\ntopic_pattern = \"events.*\"\nshared = true\n"
+    );
+    let gpath = dir.path().join("group.toml");
+    fs::write(&gpath, &gmanifest).unwrap();
+    common::pic_as(&env)
+        .args(["apply", "--file"])
+        .arg(&gpath)
+        .assert()
+        .success();
+
+    // `pic triggers ls --group` shows shared = true for the pubsub trigger.
+    let ls = String::from_utf8(
+        common::pic_as(&env)
+            .args(["triggers", "ls", "--group", &group])
+            .output()
+            .unwrap()
+            .stdout,
+    )
+    .unwrap();
+    let row = ls
+        .lines()
+        .map(common::cells)
+        .find(|c| c.contains(&"on-evt"))
+        .unwrap_or_else(|| panic!("no on-evt trigger row:\n{ls}"));
+    assert!(
+        row.contains(&"true"),
+        "the shared column must read true for the shared pubsub trigger:\n{ls}"
+    );
+
+    // A shared pubsub trigger whose topic root is NOT a declared shared topic is
+    // rejected at apply.
+    let bad_group = format!(
+        "[group]\nslug = \"{group}\"\nname = \"ShTopG\"\n\
+         collections = [{{ name = \"events\", kind = \"topic\" }}]\n\n\
+         [[triggers.pubsub]]\nscript = \"on-evt\"\ntopic_pattern = \"other.created\"\nshared = true\n"
+    );
+    fs::write(&gpath, &bad_group).unwrap();
+    let out = common::pic_as(&env)
+        .args(["apply", "--file"])
+        .arg(&gpath)
+        .output()
+        .expect("apply bad");
+    assert!(
+        !out.status.success(),
+        "a shared pubsub trigger on an undeclared topic must be rejected"
+    );
+
+    // `shared = true` on an APP pubsub trigger is rejected.
+    let app = common::unique_slug("shtop-app");
+    let _a = AppGuard::new(&env.url, &env.token, &app);
+    common::pic_as(&env)
+        .args(["apps", "create", &app])
+        .assert()
+        .success();
+    fs::write(
+        dir.path().join("scripts/app-h.rhai"),
+        r#"log::info("app"); "ok""#,
+    )
+    .unwrap();
+    let amanifest = format!(
+        "[app]\nslug = \"{app}\"\nname = \"App\"\n\n\
+         [[scripts]]\nname = \"app-h\"\nfile = \"scripts/app-h.rhai\"\n\n\
+         [[triggers.pubsub]]\nscript = \"app-h\"\ntopic_pattern = \"events.*\"\nshared = true\n"
+    );
+    let apath = dir.path().join("app.toml");
+    fs::write(&apath, &amanifest).unwrap();
+    let out = common::pic_as(&env)
+        .args(["apply", "--file"])
+        .arg(&apath)
+        .output()
+        .expect("apply app");
+    assert!(
+        !out.status.success(),
+        "a shared pubsub trigger on an app is rejected (apps don't own shared topics)"
+    );
+    let err = String::from_utf8_lossy(&out.stderr);
+    assert!(
+        err.to_lowercase().contains("shared"),
+        "the rejection must mention shared:\n{err}"
+    );
+}

docs/design/groups-and-project-tool.md

@@ -1193,8 +1193,21 @@ Resolved items now live inline next to their topic. What genuinely remains:
    > kind, or an undeclared collection. Read-only `shared` column in `pic triggers ls --group`; pinned by
    > `tests/shared_triggers.rs` + the `shared_triggers` journey.
    >
-   > **Deferred (documented gaps):** shared **pubsub** triggers (no shared topic store yet); **topics/queue**
-   > shared collections (trigger-centric, same gap); per-group total-size quotas + write-rate limits;
+   > **Shipped — D2 shared TOPICS + shared pub/sub triggers.** A group declares a **storeless** `topic`
+   > shared collection (`group_collections.kind` widened to `topic`+`queue`, `0064`); a
+   > `[[triggers.pubsub]] shared = true` handler watches it. Scripts publish via the explicit
+   > `pubsub::shared_topic("events").publish("created", msg)` handle → `GroupPubsubServiceImpl` resolves the
+   > owning group (kind `topic`) from `cx.app_id`'s chain, requires editor+ (`GroupPubsubPublish`, fails
+   > closed on anon — a publish is a shared write), and fans out via `PubsubRepo::fan_out_shared_publish` to
+   > `shared = true` pubsub triggers on that group, each outbox row stamped the WRITER `app_id` (M2 model).
+   > The per-app `fan_out_publish` gained `AND t.shared = FALSE`, and `validate_bundle_for` requires the
+   > topic pattern's ROOT segment (`events.*` → `events`) be a declared `kind='topic'` collection
+   > (group-only) — the `shared` flag + owning-group chain walk are the isolation boundary. No message store
+   > (topics are events, not persistence). Pinned by `tests/shared_topics.rs` + the `shared_topics` journey.
+   >
+   > **Deferred (documented gaps):** shared-topic external SSE subscription; **queue** shared collections
+   > (D3, competing per-descendant consumers over a group-keyed store); per-group total-size quotas +
+   > write-rate limits;
    > CAS/`set_if`; an operator admin API for shared blobs (scripts use the SDK; `pic collections ls` shows
    > the marker — matches KV/docs); app-declared collections. Multi-node tree-apply leans on the runtime
    > backstop for no-op edges, as elsewhere.