test/docs(shared-queues): journey + docs (D3.3)

CLI journey (shared_queues): authoring + `pic triggers ls --group` shared
column + the two validation rejections (undeclared queue collection; shared
on an app), mirroring the shared_topics/shared_triggers norm; the live
competing-consumer + dispatcher behaviour is pinned by the deterministic
manager-core tests. Docs: CLAUDE.md + design doc §11.6 record D3 as shipped.

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

crates/picloud-cli/tests/cli.rs

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

crates/picloud-cli/tests/shared_queues.rs

@@ -0,0 +1,159 @@
+//! §11.6 D3 — shared durable QUEUES, declarative authoring end to end via `pic`.
+//! A group declares a shared `queue` collection + a `[[triggers.queue]]
+//! shared = true` consumer over it; the template applies, `pic triggers ls
+//! --group` shows `shared = true`; a shared queue consumer over an UNDECLARED
+//! collection is rejected, and `shared = true` on an app queue is rejected.
+//!
+//! The live behaviour — competing per-descendant consumers claim one group
+//! store exactly once (SKIP LOCKED), the dispatcher drains the group store — is
+//! pinned deterministically by `manager-core/tests/group_queue.rs` +
+//! `stateful_templates.rs`; driving the async dispatcher from a journey is
+//! deliberately avoided (codebase norm, mirroring `shared_topics`).
+
+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_queue_consumer_applies_lists_and_validates() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let group = common::unique_slug("shq-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 queue.
+    let dir = manifest_dir();
+    fs::write(
+        dir.path().join("scripts/on-task.rhai"),
+        r#"log::info("shared task"); "ok""#,
+    )
+    .unwrap();
+    common::pic_as(&env)
+        .args(["scripts", "deploy"])
+        .arg(dir.path().join("scripts/on-task.rhai"))
+        .args(["--group", &group, "--name", "on-task"])
+        .assert()
+        .success();
+    let _gs = ScriptGuard::new(
+        &env.url,
+        &env.token,
+        &group_script_id(&env, &group, "on-task"),
+    );
+
+    // Group declares a shared `queue` collection `tasks` + a shared queue
+    // consumer over it.
+    let gmanifest = format!(
+        "[group]\nslug = \"{group}\"\nname = \"ShQG\"\n\
+         collections = [{{ name = \"tasks\", kind = \"queue\" }}]\n\n\
+         [[triggers.queue]]\nscript = \"on-task\"\nqueue_name = \"tasks\"\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 queue consumer.
+    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-task"))
+        .unwrap_or_else(|| panic!("no on-task trigger row:\n{ls}"));
+    assert!(
+        row.contains(&"true"),
+        "the shared column must read true for the shared queue consumer:\n{ls}"
+    );
+
+    // A shared queue consumer over a queue the group does NOT declare shared is
+    // rejected at apply.
+    let bad_group = format!(
+        "[group]\nslug = \"{group}\"\nname = \"ShQG\"\n\
+         collections = [{{ name = \"tasks\", kind = \"queue\" }}]\n\n\
+         [[triggers.queue]]\nscript = \"on-task\"\nqueue_name = \"undeclared\"\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 queue consumer over an undeclared queue must be rejected"
+    );
+
+    // `shared = true` on an APP queue trigger is rejected.
+    let app = common::unique_slug("shq-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.queue]]\nscript = \"app-h\"\nqueue_name = \"tasks\"\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 queue consumer on an app is rejected (apps don't own shared queues)"
+    );
+    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

@@ -1205,8 +1205,18 @@ Resolved items now live inline next to their topic. What genuinely remains:
    > (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 +
+   > **Shipped — D3 shared durable QUEUES (competing consumers).** A group declares a `queue` shared
+   > collection; any subtree app enqueues into ONE group-keyed store (`group_queue_messages`, `0065`) via
+   > `queue::shared_collection("name").enqueue(...)` (editor+, `GroupQueueEnqueue`). A group
+   > `[[triggers.queue]] shared = true` consumer **materializes** a consumer copy per descendant app (the
+   > M5 one-consumer-slot check is skipped for shared), and all copies claim the shared store with `FOR
+   > UPDATE SKIP LOCKED` — at-most-once across the subtree, horizontally scaled, each handler under its own
+   > `cx.app_id`. The dispatcher's queue arm routes claim/ack/nack/terminal to the group store when the
+   > consumer's source template is a shared queue (`ActiveQueueConsumer.shared_group`). Pinned by
+   > `tests/group_queue.rs` + `stateful_templates.rs` + the `shared_queues` journey. **Deferred:** a group
+   > dead-letter store (an exhausted shared-queue message is dropped-with-warning).
+   >
+   > **Deferred (documented gaps):** shared-topic external SSE subscription; 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