feat(modules): GroupKvService + kv::shared SDK handle (§11.6 C3)

The runtime for shared group collections. A script reaches a shared store via
the explicit kv::shared("name") handle (distinct GroupKvHandle Rhai type, so
private-vs-shared scope is a deliberate choice). The service resolves the
owning group from cx.app_id's ancestor chain — the script never names a group,
and that walk is the isolation boundary (a foreign app's chain never contains
the owning group → CollectionNotShared).

- shared: GroupKvService trait + GroupKvError + NoopGroupKvService; threaded
  into the Services bundle as a field defaulted to noop, opted into via a new
  with_group_kv() (keeps the ~14 Services::new call sites unchanged).
- manager-core: GroupKvServiceImpl with the reads-open/writes-authed split —
  reads use script_gate (anonymous public scripts skip), writes use the new
  script_gate_require_principal (anonymous fails closed). Owner resolution
  behind a GroupCollectionResolver trait (Postgres impl + injectable fake);
  unit tests cover off-chain CollectionNotShared, anonymous-read-OK/
  anonymous-write-Forbidden, authed-no-role-Forbidden.
- executor-core: GroupKvHandle + kv::shared constructor + get/set/has/delete/
  list (Rhai dispatches by receiver type, reusing the block_on bridge).
- picloud: wire PostgresGroupKvRepo + resolver + GroupKvServiceImpl.

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

crates/executor-core/src/sdk/kv.rs

@@ -28,7 +28,7 @@
 
 use std::sync::Arc;
 
-use picloud_shared::{KvService, SdkCallCx, Services};
+use picloud_shared::{GroupKvService, KvService, SdkCallCx, Services};
 use rhai::{Array, Dynamic, Engine as RhaiEngine, EvalAltResult, Map, Module};
 
 use super::bridge::{block_on, dynamic_to_json, json_to_dynamic};
@@ -42,11 +42,25 @@ pub struct KvHandle {
     cx: Arc<SdkCallCx>,
 }
 
+/// §11.6 shared-collection handle, returned by `kv::shared(name)`. A distinct
+/// Rhai type from `KvHandle` so the method set can diverge (e.g. shared
+/// collections never grow triggers) and a script's choice of private-vs-shared
+/// scope is explicit. Routes through the `GroupKvService`, which resolves the
+/// owning group from `cx.app_id` (the script never names a group).
+#[derive(Clone)]
+pub struct GroupKvHandle {
+    collection: String,
+    service: Arc<dyn GroupKvService>,
+    cx: Arc<SdkCallCx>,
+}
+
 pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<SdkCallCx>) {
     let kv_service = services.kv.clone();
+    let group_kv_service = services.group_kv.clone();
 
-    // `kv::collection(name)` — handle constructor lives in the `kv`
+    // `kv::collection(name)` / `kv::shared(name)` — both constructors live in
-    // static module so the script-visible call is `kv::collection(...)`.
+    // the `kv` static module so the script-visible calls are `kv::collection`
+    // and `kv::shared`.
     let mut module = Module::new();
     {
         let kv_service = kv_service.clone();
@@ -65,6 +79,23 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
             },
         );
     }
+    {
+        let group_kv_service = group_kv_service.clone();
+        let cx = cx.clone();
+        module.set_native_fn(
+            "shared",
+            move |name: &str| -> Result<GroupKvHandle, Box<EvalAltResult>> {
+                if name.is_empty() {
+                    return Err("kv::shared name must not be empty".into());
+                }
+                Ok(GroupKvHandle {
+                    collection: name.to_string(),
+                    service: group_kv_service.clone(),
+                    cx: cx.clone(),
+                })
+            },
+        );
+    }
     engine.register_static_module("kv", module.into());
 
     // Methods on KvHandle — `register_fn` with `&mut KvHandle` first
@@ -77,6 +108,15 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
     register_has(engine);
     register_delete(engine);
     register_list(engine);
+
+    // Same method names on GroupKvHandle — Rhai dispatches by receiver type.
+    engine.register_type_with_name::<GroupKvHandle>("GroupKvHandle");
+
+    register_group_get(engine);
+    register_group_set(engine);
+    register_group_has(engine);
+    register_group_delete(engine);
+    register_group_list(engine);
 }
 
 fn register_get(engine: &mut RhaiEngine) {
@@ -174,3 +214,98 @@ fn list_call(
     );
     Ok(m)
 }
+
+// --- GroupKvHandle methods (§11.6 shared collections) ----------------------
+
+fn register_group_get(engine: &mut RhaiEngine) {
+    engine.register_fn(
+        "get",
+        |handle: &mut GroupKvHandle, key: &str| -> Result<Dynamic, Box<EvalAltResult>> {
+            let h = handle.clone();
+            block_on("kv", async move {
+                h.service.get(&h.cx, &h.collection, key).await
+            })
+            .map(|opt| opt.map_or(Dynamic::UNIT, json_to_dynamic))
+        },
+    );
+}
+
+fn register_group_set(engine: &mut RhaiEngine) {
+    engine.register_fn(
+        "set",
+        |handle: &mut GroupKvHandle, key: &str, value: Dynamic| -> Result<(), Box<EvalAltResult>> {
+            let h = handle.clone();
+            let json = dynamic_to_json(&value);
+            block_on("kv", async move {
+                h.service.set(&h.cx, &h.collection, key, json).await
+            })
+        },
+    );
+}
+
+fn register_group_has(engine: &mut RhaiEngine) {
+    engine.register_fn(
+        "has",
+        |handle: &mut GroupKvHandle, key: &str| -> Result<bool, Box<EvalAltResult>> {
+            let h = handle.clone();
+            block_on("kv", async move {
+                h.service.has(&h.cx, &h.collection, key).await
+            })
+        },
+    );
+}
+
+fn register_group_delete(engine: &mut RhaiEngine) {
+    engine.register_fn(
+        "delete",
+        |handle: &mut GroupKvHandle, key: &str| -> Result<bool, Box<EvalAltResult>> {
+            let h = handle.clone();
+            block_on("kv", async move {
+                h.service.delete(&h.cx, &h.collection, key).await
+            })
+        },
+    );
+}
+
+fn register_group_list(engine: &mut RhaiEngine) {
+    engine.register_fn(
+        "list",
+        |handle: &mut GroupKvHandle| -> Result<Map, Box<EvalAltResult>> {
+            group_list_call(handle, None, 0)
+        },
+    );
+    engine.register_fn(
+        "list",
+        |handle: &mut GroupKvHandle, cursor: &str| -> Result<Map, Box<EvalAltResult>> {
+            group_list_call(handle, Some(cursor.to_string()), 0)
+        },
+    );
+    engine.register_fn(
+        "list",
+        |handle: &mut GroupKvHandle, cursor: &str, limit: i64| -> Result<Map, Box<EvalAltResult>> {
+            let limit = u32::try_from(limit.max(0)).unwrap_or(0);
+            group_list_call(handle, Some(cursor.to_string()), limit)
+        },
+    );
+}
+
+fn group_list_call(
+    handle: &GroupKvHandle,
+    cursor: Option<String>,
+    limit: u32,
+) -> Result<Map, Box<EvalAltResult>> {
+    let h = handle.clone();
+    let page = block_on("kv", async move {
+        h.service
+            .list(&h.cx, &h.collection, cursor.as_deref(), limit)
+            .await
+    })?;
+    let mut m = Map::new();
+    let keys: Array = page.keys.into_iter().map(Dynamic::from).collect();
+    m.insert("keys".into(), keys.into());
+    m.insert(
+        "next_cursor".into(),
+        page.next_cursor.map_or(Dynamic::UNIT, Dynamic::from),
+    );
+    Ok(m)
+}

crates/manager-core/src/authz.rs

@@ -457,6 +457,34 @@ pub async fn script_gate<E>(
     }
 }
 
+/// Like [`script_gate`], but **fails closed on an anonymous principal**: a
+/// script with `cx.principal == None` is rejected with `forbidden()` rather
+/// than skipped. Used for actions that must always be performed by an
+/// authenticated caller even though the surrounding service skips authz for
+/// public scripts — e.g. §11.6 group-collection WRITES (reads stay open via
+/// `script_gate`, writes require an authenticated editor+ on the owning group).
+///
+/// # Errors
+///
+/// Returns `forbidden()` when the principal is absent or the capability is
+/// denied, or `backend(repo_err.to_string())` on a repo error.
+pub async fn script_gate_require_principal<E>(
+    repo: &dyn AuthzRepo,
+    cx: &picloud_shared::SdkCallCx,
+    cap: Capability,
+    forbidden: impl FnOnce() -> E,
+    backend: impl FnOnce(String) -> E,
+) -> Result<(), E> {
+    let Some(principal) = cx.principal.as_ref() else {
+        return Err(forbidden());
+    };
+    match require(repo, principal, cap).await {
+        Ok(()) => Ok(()),
+        Err(AuthzDenied::Denied) => Err(forbidden()),
+        Err(AuthzDenied::Repo(e)) => Err(backend(e.to_string())),
+    }
+}
+
 // ----------------------------------------------------------------------------
 // Layer 1: role-derived grant
 // ----------------------------------------------------------------------------

crates/manager-core/src/group_kv_service.rs

@@ -0,0 +1,450 @@
+//! `GroupKvServiceImpl` — wires `GroupKvRepo` + the group-collection registry
+//! underneath the `picloud_shared::GroupKvService` trait that scripts reach via
+//! the `kv::shared("name")` Rhai handle (§11.6).
+//!
+//! Layers added over the raw repo:
+//!
+//! 1. Empty-collection rejection.
+//! 2. **Owner resolution** (the structural isolation boundary): the collection
+//!    name is resolved to the nearest ancestor group that declares it shared,
+//!    walking the calling app's chain. No declaration on the chain → a clean
+//!    `CollectionNotShared`, BEFORE any authz or storage access.
+//! 3. **Reads-open / writes-authed authz**: reads use `script_gate`
+//!    (anonymous public scripts skip); writes use `script_gate_require_principal`
+//!    (anonymous fails closed — shared mutation always needs an authenticated
+//!    editor+ on the owning group).
+//! 4. Per-value size cap (reuses `PICLOUD_KV_MAX_VALUE_BYTES`).
+//!
+//! No event emission in the MVP: a write to a shared collection has no single
+//! app to attribute a trigger to (the "group trigger has no app to watch"
+//! problem). Documented as a deferral in docs §11.6.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use picloud_shared::{AppId, GroupId, GroupKvError, GroupKvService, KvListPage, SdkCallCx};
+use sqlx::PgPool;
+
+use crate::authz::{self, AuthzRepo, Capability};
+use crate::group_collection_repo;
+use crate::group_kv_repo::{GroupKvRepo, GroupKvRepoError};
+use crate::kv_service::kv_max_value_bytes_from_env;
+
+/// Resolves a shared-collection name to its owning group for a calling app
+/// (nearest ancestor group on the app's chain that declares it). Behind a
+/// trait so unit tests can inject a fake without Postgres.
+#[async_trait]
+pub trait GroupCollectionResolver: Send + Sync {
+    async fn resolve_owning_group(
+        &self,
+        app_id: AppId,
+        name: &str,
+    ) -> Result<Option<GroupId>, sqlx::Error>;
+}
+
+/// Postgres-backed resolver — delegates to [`group_collection_repo`].
+pub struct PostgresGroupCollectionResolver {
+    pool: PgPool,
+}
+
+impl PostgresGroupCollectionResolver {
+    #[must_use]
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+}
+
+#[async_trait]
+impl GroupCollectionResolver for PostgresGroupCollectionResolver {
+    async fn resolve_owning_group(
+        &self,
+        app_id: AppId,
+        name: &str,
+    ) -> Result<Option<GroupId>, sqlx::Error> {
+        group_collection_repo::resolve_owning_group(&self.pool, app_id, name).await
+    }
+}
+
+pub struct GroupKvServiceImpl {
+    repo: Arc<dyn GroupKvRepo>,
+    resolver: Arc<dyn GroupCollectionResolver>,
+    authz: Arc<dyn AuthzRepo>,
+    max_value_bytes: usize,
+}
+
+impl GroupKvServiceImpl {
+    #[must_use]
+    pub fn new(
+        repo: Arc<dyn GroupKvRepo>,
+        resolver: Arc<dyn GroupCollectionResolver>,
+        authz: Arc<dyn AuthzRepo>,
+    ) -> Self {
+        Self::with_max_value_bytes(repo, resolver, authz, kv_max_value_bytes_from_env())
+    }
+
+    #[must_use]
+    pub fn with_max_value_bytes(
+        repo: Arc<dyn GroupKvRepo>,
+        resolver: Arc<dyn GroupCollectionResolver>,
+        authz: Arc<dyn AuthzRepo>,
+        max_value_bytes: usize,
+    ) -> Self {
+        Self {
+            repo,
+            resolver,
+            authz,
+            max_value_bytes,
+        }
+    }
+
+    /// The structural boundary: resolve the collection to its owning group on
+    /// the calling app's chain. `CollectionNotShared` when no ancestor group
+    /// declares it — the same outcome a foreign app gets, by construction.
+    async fn owning_group(
+        &self,
+        cx: &SdkCallCx,
+        collection: &str,
+    ) -> Result<GroupId, GroupKvError> {
+        if collection.is_empty() {
+            return Err(GroupKvError::InvalidCollection);
+        }
+        self.resolver
+            .resolve_owning_group(cx.app_id, collection)
+            .await
+            .map_err(|e| GroupKvError::Backend(e.to_string()))?
+            .ok_or_else(|| GroupKvError::CollectionNotShared(collection.to_string()))
+    }
+
+    async fn check_read(&self, cx: &SdkCallCx, group_id: GroupId) -> Result<(), GroupKvError> {
+        authz::script_gate(
+            &*self.authz,
+            cx,
+            Capability::GroupKvRead(group_id),
+            || GroupKvError::Forbidden,
+            GroupKvError::Backend,
+        )
+        .await
+    }
+
+    async fn check_write(&self, cx: &SdkCallCx, group_id: GroupId) -> Result<(), GroupKvError> {
+        // Fails closed for an anonymous principal: shared mutation always needs
+        // an authenticated editor+ on the owning group.
+        authz::script_gate_require_principal(
+            &*self.authz,
+            cx,
+            Capability::GroupKvWrite(group_id),
+            || GroupKvError::Forbidden,
+            GroupKvError::Backend,
+        )
+        .await
+    }
+}
+
+impl From<GroupKvRepoError> for GroupKvError {
+    fn from(e: GroupKvRepoError) -> Self {
+        Self::Backend(e.to_string())
+    }
+}
+
+#[async_trait]
+impl GroupKvService for GroupKvServiceImpl {
+    async fn get(
+        &self,
+        cx: &SdkCallCx,
+        collection: &str,
+        key: &str,
+    ) -> Result<Option<serde_json::Value>, GroupKvError> {
+        let group_id = self.owning_group(cx, collection).await?;
+        self.check_read(cx, group_id).await?;
+        Ok(self.repo.get(group_id, collection, key).await?)
+    }
+
+    async fn set(
+        &self,
+        cx: &SdkCallCx,
+        collection: &str,
+        key: &str,
+        value: serde_json::Value,
+    ) -> Result<(), GroupKvError> {
+        let group_id = self.owning_group(cx, collection).await?;
+        let encoded_len = serde_json::to_vec(&value)
+            .map(|v| v.len())
+            .map_err(|e| GroupKvError::Backend(format!("encode value: {e}")))?;
+        if encoded_len > self.max_value_bytes {
+            return Err(GroupKvError::ValueTooLarge {
+                limit: self.max_value_bytes,
+                actual: encoded_len,
+            });
+        }
+        self.check_write(cx, group_id).await?;
+        self.repo.set(group_id, collection, key, value).await?;
+        Ok(())
+    }
+
+    async fn delete(
+        &self,
+        cx: &SdkCallCx,
+        collection: &str,
+        key: &str,
+    ) -> Result<bool, GroupKvError> {
+        let group_id = self.owning_group(cx, collection).await?;
+        self.check_write(cx, group_id).await?;
+        Ok(self.repo.delete(group_id, collection, key).await?.is_some())
+    }
+
+    async fn has(&self, cx: &SdkCallCx, collection: &str, key: &str) -> Result<bool, GroupKvError> {
+        let group_id = self.owning_group(cx, collection).await?;
+        self.check_read(cx, group_id).await?;
+        Ok(self.repo.has(group_id, collection, key).await?)
+    }
+
+    async fn list(
+        &self,
+        cx: &SdkCallCx,
+        collection: &str,
+        cursor: Option<&str>,
+        limit: u32,
+    ) -> Result<KvListPage, GroupKvError> {
+        let group_id = self.owning_group(cx, collection).await?;
+        self.check_read(cx, group_id).await?;
+        Ok(self.repo.list(group_id, collection, cursor, limit).await?)
+    }
+}
+
+// ----------------------------------------------------------------------------
+// Tests — in-memory repo + a fake resolver so unit tests don't need Postgres.
+// ----------------------------------------------------------------------------
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::authz::{AuthzError, AuthzRepo};
+    use picloud_shared::{
+        AdminUserId, AppRole, ExecutionId, InstanceRole, Principal, RequestId, ScriptId, UserId,
+    };
+    use std::collections::{BTreeMap, HashMap};
+    use tokio::sync::Mutex;
+
+    #[derive(Default)]
+    struct InMemoryGroupKvRepo {
+        data: Mutex<BTreeMap<(GroupId, String, String), serde_json::Value>>,
+    }
+
+    #[async_trait]
+    impl GroupKvRepo for InMemoryGroupKvRepo {
+        async fn get(
+            &self,
+            group_id: GroupId,
+            collection: &str,
+            key: &str,
+        ) -> Result<Option<serde_json::Value>, GroupKvRepoError> {
+            Ok(self
+                .data
+                .lock()
+                .await
+                .get(&(group_id, collection.to_string(), key.to_string()))
+                .cloned())
+        }
+
+        async fn set(
+            &self,
+            group_id: GroupId,
+            collection: &str,
+            key: &str,
+            value: serde_json::Value,
+        ) -> Result<Option<serde_json::Value>, GroupKvRepoError> {
+            Ok(self
+                .data
+                .lock()
+                .await
+                .insert((group_id, collection.to_string(), key.to_string()), value))
+        }
+
+        async fn delete(
+            &self,
+            group_id: GroupId,
+            collection: &str,
+            key: &str,
+        ) -> Result<Option<serde_json::Value>, GroupKvRepoError> {
+            Ok(self
+                .data
+                .lock()
+                .await
+                .remove(&(group_id, collection.to_string(), key.to_string())))
+        }
+
+        async fn has(
+            &self,
+            group_id: GroupId,
+            collection: &str,
+            key: &str,
+        ) -> Result<bool, GroupKvRepoError> {
+            Ok(self.data.lock().await.contains_key(&(
+                group_id,
+                collection.to_string(),
+                key.to_string(),
+            )))
+        }
+
+        async fn list(
+            &self,
+            group_id: GroupId,
+            collection: &str,
+            _cursor: Option<&str>,
+            limit: u32,
+        ) -> Result<KvListPage, GroupKvRepoError> {
+            let data = self.data.lock().await;
+            let mut keys: Vec<String> = data
+                .iter()
+                .filter(|((g, c, _), _)| *g == group_id && c == collection)
+                .map(|((_, _, k), _)| k.clone())
+                .collect();
+            keys.sort();
+            keys.truncate((limit as usize).max(1));
+            Ok(KvListPage {
+                keys,
+                next_cursor: None,
+            })
+        }
+    }
+
+    /// Maps `(app_id, lowercased name) -> owning group`. Anything absent is
+    /// "not shared with you".
+    #[derive(Default)]
+    struct FakeResolver {
+        map: HashMap<(AppId, String), GroupId>,
+    }
+
+    #[async_trait]
+    impl GroupCollectionResolver for FakeResolver {
+        async fn resolve_owning_group(
+            &self,
+            app_id: AppId,
+            name: &str,
+        ) -> Result<Option<GroupId>, sqlx::Error> {
+            Ok(self.map.get(&(app_id, name.to_lowercase())).copied())
+        }
+    }
+
+    #[derive(Default)]
+    struct DenyingAuthzRepo;
+
+    #[async_trait]
+    impl AuthzRepo for DenyingAuthzRepo {
+        async fn membership(
+            &self,
+            _user_id: UserId,
+            _app_id: AppId,
+        ) -> Result<Option<AppRole>, AuthzError> {
+            Ok(None)
+        }
+    }
+
+    fn cx_with(app_id: AppId, principal: Option<Principal>) -> SdkCallCx {
+        SdkCallCx {
+            app_id,
+            script_id: ScriptId::new(),
+            principal,
+            execution_id: ExecutionId::new(),
+            request_id: RequestId::new(),
+            trigger_depth: 0,
+            root_execution_id: ExecutionId::new(),
+            is_dead_letter_handler: false,
+            event: None,
+        }
+    }
+
+    fn owner() -> Principal {
+        Principal {
+            user_id: AdminUserId::new(),
+            instance_role: InstanceRole::Owner,
+            scopes: None,
+            app_binding: None,
+        }
+    }
+
+    fn member_no_role() -> Principal {
+        Principal {
+            user_id: AdminUserId::new(),
+            instance_role: InstanceRole::Member,
+            scopes: None,
+            app_binding: None,
+        }
+    }
+
+    fn svc(resolver: FakeResolver) -> GroupKvServiceImpl {
+        GroupKvServiceImpl::new(
+            Arc::new(InMemoryGroupKvRepo::default()),
+            Arc::new(resolver),
+            Arc::new(DenyingAuthzRepo),
+        )
+    }
+
+    #[tokio::test]
+    async fn unrelated_app_gets_collection_not_shared() {
+        // app_a's chain declares "catalog"; app_b's does not.
+        let app_a = AppId::new();
+        let app_b = AppId::new();
+        let group = GroupId::new();
+        let mut resolver = FakeResolver::default();
+        resolver.map.insert((app_a, "catalog".into()), group);
+        let kv = svc(resolver);
+
+        // app_a (owner principal) can write+read.
+        let cx_a = cx_with(app_a, Some(owner()));
+        kv.set(&cx_a, "catalog", "k", serde_json::json!(1))
+            .await
+            .unwrap();
+        assert_eq!(
+            kv.get(&cx_a, "catalog", "k").await.unwrap(),
+            Some(serde_json::json!(1))
+        );
+
+        // app_b is off-chain — the name does not resolve.
+        let cx_b = cx_with(app_b, Some(owner()));
+        let err = kv.get(&cx_b, "catalog", "k").await.unwrap_err();
+        assert!(matches!(err, GroupKvError::CollectionNotShared(c) if c == "catalog"));
+    }
+
+    #[tokio::test]
+    async fn reads_open_writes_require_auth() {
+        let app = AppId::new();
+        let group = GroupId::new();
+        let mut resolver = FakeResolver::default();
+        resolver.map.insert((app, "catalog".into()), group);
+        let kv = svc(resolver);
+
+        // Seed a value as owner.
+        let owner_cx = cx_with(app, Some(owner()));
+        kv.set(&owner_cx, "catalog", "k", serde_json::json!("v"))
+            .await
+            .unwrap();
+
+        // Anonymous (principal=None) READ is allowed (reads-open).
+        let anon = cx_with(app, None);
+        assert_eq!(
+            kv.get(&anon, "catalog", "k").await.unwrap(),
+            Some(serde_json::json!("v"))
+        );
+
+        // Anonymous WRITE fails closed.
+        let err = kv
+            .set(&anon, "catalog", "k", serde_json::json!("x"))
+            .await
+            .unwrap_err();
+        assert!(matches!(err, GroupKvError::Forbidden));
+
+        // Authenticated member with no group role: write forbidden.
+        let member = cx_with(app, Some(member_no_role()));
+        let err = kv.delete(&member, "catalog", "k").await.unwrap_err();
+        assert!(matches!(err, GroupKvError::Forbidden));
+    }
+
+    #[tokio::test]
+    async fn empty_collection_rejected_before_resolve() {
+        let kv = svc(FakeResolver::default());
+        let cx = cx_with(AppId::new(), None);
+        let err = kv.get(&cx, "", "k").await.unwrap_err();
+        assert!(matches!(err, GroupKvError::InvalidCollection));
+    }
+}

crates/manager-core/src/lib.rs

@@ -52,6 +52,7 @@ pub mod files_sweep;
 pub mod gc;
 pub mod group_collection_repo;
 pub mod group_kv_repo;
+pub mod group_kv_service;
 pub mod group_members_repo;
 pub mod group_repo;
 pub mod group_scripts_api;
@@ -178,6 +179,10 @@ 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_app_user_token_gc, spawn_dead_letter_gc};
+pub use group_kv_repo::{GroupKvRepo, GroupKvRepoError, PostgresGroupKvRepo};
+pub use group_kv_service::{
+    GroupCollectionResolver, GroupKvServiceImpl, PostgresGroupCollectionResolver,
+};
 pub use group_members_repo::{
     GroupMembersRepository, GroupMembersRepositoryError, GroupMembershipDetail, GroupMembershipRow,
     PostgresGroupMembersRepository,
@@ -190,7 +195,6 @@ pub use group_scripts_api::{group_scripts_router, GroupScriptsApiError, GroupScr
 pub use groups_api::{groups_router, GroupsApiError, GroupsState};
 pub use http_service::{HttpConfig, HttpServiceImpl};
 pub use kv_api::{kv_admin_router, KvAdminState};
-pub use group_kv_repo::{GroupKvRepo, GroupKvRepoError, PostgresGroupKvRepo};
 pub use kv_repo::{KvRepo, KvRepoError, PostgresKvRepo};
 pub use kv_service::KvServiceImpl;
 pub use log_sink::PostgresExecutionLogSink;

crates/picloud/src/lib.rs

@@ -19,22 +19,22 @@ use picloud_manager_core::{
     AppDomainRepository, AppMembersRepository, AppMembersState, AppRepository, ApplyService,
     AppsState, AuthState, AuthzRepo, DeadLetterRepo, DeadLettersState, DevEmailState, Dispatcher,
     DocsServiceImpl, EmailInboundState, EmailServiceImpl, FilesAdminState, FilesConfig,
-    FilesServiceImpl, FsFilesRepo, GroupMembersRepository, GroupRepository, GroupsState,
+    FilesServiceImpl, FsFilesRepo, GroupKvServiceImpl, GroupMembersRepository, GroupRepository,
-    HttpConfig, HttpServiceImpl, InboundNonceDedup, KvAdminState, KvServiceImpl,
+    GroupsState, HttpConfig, HttpServiceImpl, InboundNonceDedup, KvAdminState, KvServiceImpl,
     OutboxEventEmitter, OutboxRepo, PostgresAbandonedRepo, PostgresAdminSessionRepository,
     PostgresAdminUserRepository, PostgresApiKeyRepository, PostgresAppDomainRepository,
     PostgresAppMembersRepository, PostgresAppRepository, PostgresAppSecretsRepo,
     PostgresAppUserInvitationRepo, PostgresAppUserPasswordResetRepo, PostgresAppUserRepository,
     PostgresAppUserRoleRepo, PostgresAppUserSessionRepository, PostgresAppUserVerificationRepo,
     PostgresDeadLetterRepo, PostgresDeadLetterService, PostgresDocsRepo,
-    PostgresExecutionLogRepository, PostgresExecutionLogSink, PostgresGroupMembersRepository,
+    PostgresExecutionLogRepository, PostgresExecutionLogSink, PostgresGroupCollectionResolver,
-    PostgresGroupRepository, PostgresKvRepo, PostgresOutboxRepo, PostgresPubsubRepo,
+    PostgresGroupKvRepo, PostgresGroupMembersRepository, PostgresGroupRepository, PostgresKvRepo,
-    PostgresRouteRepository, PostgresScriptRepository, PostgresSecretsRepo, PostgresTopicRepo,
+    PostgresOutboxRepo, PostgresPubsubRepo, PostgresRouteRepository, PostgresScriptRepository,
-    PostgresTriggerRepo, PostgresVarsRepo, PrincipalResolver, PubsubServiceImpl,
+    PostgresSecretsRepo, PostgresTopicRepo, PostgresTriggerRepo, PostgresVarsRepo,
-    RealtimeAuthorityImpl, RepoResolver, RouteAdminState, RouteRepository, SandboxCeiling,
+    PrincipalResolver, PubsubServiceImpl, RealtimeAuthorityImpl, RepoResolver, RouteAdminState,
-    ScriptRepository, SecretsConfig, SecretsServiceImpl, SecretsState, SubscriberTokenConfig,
+    RouteRepository, SandboxCeiling, ScriptRepository, SecretsConfig, SecretsServiceImpl,
-    TopicRepo, TopicsState, TriggerConfig, TriggerRepo, TriggersState, UsersServiceConfig,
+    SecretsState, SubscriberTokenConfig, TopicRepo, TopicsState, TriggerConfig, TriggerRepo,
-    UsersServiceImpl, VarsApiState, VarsServiceImpl,
+    TriggersState, UsersServiceConfig, UsersServiceImpl, VarsApiState, VarsServiceImpl,
 };
 use picloud_orchestrator_core::realtime::DEFAULT_GC_INTERVAL_SECS;
 use picloud_orchestrator_core::routing::{AppDomainTable, RouteTable};
@@ -158,6 +158,15 @@ pub async fn build_app(
         events.clone(),
         picloud_manager_core::kv_service::kv_max_value_bytes_from_env(),
     ));
+    // §11.6 shared group collections (KV): a separate store keyed by the owning
+    // group, resolved from cx.app_id's chain. Reuses the KV value-size cap.
+    let group_kv: Arc<dyn picloud_shared::GroupKvService> =
+        Arc::new(GroupKvServiceImpl::with_max_value_bytes(
+            Arc::new(PostgresGroupKvRepo::new(pool.clone())),
+            Arc::new(PostgresGroupCollectionResolver::new(pool.clone())),
+            authz.clone(),
+            picloud_manager_core::kv_service::kv_max_value_bytes_from_env(),
+        ));
     let docs: Arc<dyn DocsService> = Arc::new(DocsServiceImpl::with_max_value_bytes(
         docs_repo,
         authz.clone(),
@@ -340,7 +349,8 @@ pub async fn build_app(
         queue,
         invoke,
         vars,
-    );
+    )
+    .with_group_kv(group_kv);
     // v1.1.9: keep the invoke depth bound aligned with the dispatcher's
     // trigger-depth bound (same counter under the hood).
     let engine_limits = Limits {

crates/shared/src/group_kv.rs

@@ -0,0 +1,141 @@
+//! `GroupKvService` — the §11.6 shared group-collection KV contract.
+//!
+//! The cross-app-sharing counterpart to [`crate::KvService`]. A script reaches
+//! it via the explicit `kv::shared("name")` handle (distinct from the
+//! app-private `kv::collection("name")`). Unlike `KvService`, the *owner* of
+//! the data is not `cx.app_id`: the impl resolves the **owning group** by
+//! walking the calling app's ancestor chain for the nearest group that declares
+//! the collection group-shared. That ancestry walk is the isolation boundary —
+//! a foreign app's chain never contains the owning group, so the name does not
+//! resolve. The trait surface therefore takes **no** group id (same discipline
+//! as `KvService` omitting `app_id`): owner derivation stays inside the impl.
+//!
+//! Trust model (§11.6): **reads are open** to any script in the subtree
+//! (anonymous public HTTP included — the operator opted in by declaring the
+//! collection); **writes require an authenticated principal** with editor+ on
+//! the owning group.
+
+use async_trait::async_trait;
+use thiserror::Error;
+
+use crate::{KvListPage, SdkCallCx};
+
+#[async_trait]
+pub trait GroupKvService: Send + Sync {
+    async fn get(
+        &self,
+        cx: &SdkCallCx,
+        collection: &str,
+        key: &str,
+    ) -> Result<Option<serde_json::Value>, GroupKvError>;
+
+    async fn set(
+        &self,
+        cx: &SdkCallCx,
+        collection: &str,
+        key: &str,
+        value: serde_json::Value,
+    ) -> Result<(), GroupKvError>;
+
+    async fn delete(
+        &self,
+        cx: &SdkCallCx,
+        collection: &str,
+        key: &str,
+    ) -> Result<bool, GroupKvError>;
+
+    async fn has(&self, cx: &SdkCallCx, collection: &str, key: &str) -> Result<bool, GroupKvError>;
+
+    async fn list(
+        &self,
+        cx: &SdkCallCx,
+        collection: &str,
+        cursor: Option<&str>,
+        limit: u32,
+    ) -> Result<KvListPage, GroupKvError>;
+}
+
+/// Stub for test bundles that don't exercise shared collections. Every call
+/// errors so accidental use surfaces clearly.
+#[derive(Debug, Default, Clone, Copy)]
+pub struct NoopGroupKvService;
+
+#[async_trait]
+impl GroupKvService for NoopGroupKvService {
+    async fn get(
+        &self,
+        _cx: &SdkCallCx,
+        _collection: &str,
+        _key: &str,
+    ) -> Result<Option<serde_json::Value>, GroupKvError> {
+        Err(GroupKvError::Backend("group kv is not wired in".into()))
+    }
+
+    async fn set(
+        &self,
+        _cx: &SdkCallCx,
+        _collection: &str,
+        _key: &str,
+        _value: serde_json::Value,
+    ) -> Result<(), GroupKvError> {
+        Err(GroupKvError::Backend("group kv is not wired in".into()))
+    }
+
+    async fn delete(
+        &self,
+        _cx: &SdkCallCx,
+        _collection: &str,
+        _key: &str,
+    ) -> Result<bool, GroupKvError> {
+        Err(GroupKvError::Backend("group kv is not wired in".into()))
+    }
+
+    async fn has(
+        &self,
+        _cx: &SdkCallCx,
+        _collection: &str,
+        _key: &str,
+    ) -> Result<bool, GroupKvError> {
+        Err(GroupKvError::Backend("group kv is not wired in".into()))
+    }
+
+    async fn list(
+        &self,
+        _cx: &SdkCallCx,
+        _collection: &str,
+        _cursor: Option<&str>,
+        _limit: u32,
+    ) -> Result<KvListPage, GroupKvError> {
+        Err(GroupKvError::Backend("group kv is not wired in".into()))
+    }
+}
+
+/// Failure modes surfaced to the Rhai bridge.
+#[derive(Debug, Error)]
+pub enum GroupKvError {
+    /// Empty collection name; rejected at the SDK boundary.
+    #[error("collection name must not be empty")]
+    InvalidCollection,
+
+    /// No group on the calling app's ancestor chain declares this collection
+    /// group-shared — the structural "not shared with you" boundary. Also the
+    /// outcome for a foreign app naming another subtree's collection.
+    #[error("collection {0:?} is not a shared group collection for this app")]
+    CollectionNotShared(String),
+
+    /// Caller lacked the required capability. For reads this is only raised
+    /// when `cx.principal.is_some()`; for WRITES it is also raised when the
+    /// principal is absent (writes fail closed — shared mutation always needs
+    /// an authenticated editor+ on the owning group).
+    #[error("forbidden")]
+    Forbidden,
+
+    /// JSON-encoded value exceeded the per-value `max_value_bytes` cap
+    /// (`PICLOUD_KV_MAX_VALUE_BYTES`).
+    #[error("group kv: value too large ({actual} bytes; limit {limit})")]
+    ValueTooLarge { limit: usize, actual: usize },
+
+    /// Anything else — Postgres unavailable, serialization failure, etc.
+    #[error("group kv backend error: {0}")]
+    Backend(String),
+}

crates/shared/src/lib.rs

@@ -16,6 +16,7 @@ pub mod exec_summary;
 pub mod execution_log;
 pub mod files;
 pub mod group;
+pub mod group_kv;
 pub mod http;
 pub mod ids;
 pub mod inbox;
@@ -65,6 +66,7 @@ pub use files::{
     SAFE_RENDER_FALLBACK,
 };
 pub use group::Group;
+pub use group_kv::{GroupKvError, GroupKvService, NoopGroupKvService};
 pub use http::{HttpError, HttpRequest, HttpResponse, HttpService, NoopHttpService};
 pub use ids::{
     AdminUserId, ApiKeyId, AppId, AppUserId, ExecutionId, GroupId, InvitationId, QueueMessageId,

crates/shared/src/services.rs

@@ -20,12 +20,12 @@
 use std::sync::Arc;
 
 use crate::{
-    DeadLetterService, DocsService, EmailService, FilesService, HttpService, InvokeService,
+    DeadLetterService, DocsService, EmailService, FilesService, GroupKvService, HttpService,
-    KvService, ModuleSource, NoopDeadLetterService, NoopDocsService, NoopEmailService,
+    InvokeService, KvService, ModuleSource, NoopDeadLetterService, NoopDocsService,
-    NoopEventEmitter, NoopFilesService, NoopHttpService, NoopInvokeService, NoopKvService,
+    NoopEmailService, NoopEventEmitter, NoopFilesService, NoopGroupKvService, NoopHttpService,
-    NoopModuleSource, NoopPubsubService, NoopQueueService, NoopSecretsService, NoopUsersService,
+    NoopInvokeService, NoopKvService, NoopModuleSource, NoopPubsubService, NoopQueueService,
-    NoopVarsService, PubsubService, QueueService, SecretsService, ServiceEventEmitter,
+    NoopSecretsService, NoopUsersService, NoopVarsService, PubsubService, QueueService,
-    UsersService, VarsService,
+    SecretsService, ServiceEventEmitter, UsersService, VarsService,
 };
 
 /// SDK service bundle. See module docs for the lifecycle and the v1.1.x
@@ -114,6 +114,14 @@ pub struct Services {
     /// (§3). Backed by `config_resolver` over Postgres in the picloud
     /// binary; `NoopVarsService` in tests that don't read config.
     pub vars: Arc<dyn VarsService>,
+
+    /// Shared cross-app group collections (§11.6). Scripts get
+    /// `kv::shared(name)` — read/write KV scoped to the nearest ancestor
+    /// group that declares the collection shared. Backed by Postgres in the
+    /// picloud binary (wired via [`Services::with_group_kv`]); defaults to
+    /// `NoopGroupKvService` so test bundles that don't exercise sharing build
+    /// unchanged.
+    pub group_kv: Arc<dyn GroupKvService>,
 }
 
 impl Services {
@@ -153,9 +161,21 @@ impl Services {
             queue,
             invoke,
             vars,
+            // §11.6 group collections default to noop; the picloud binary opts
+            // in via `with_group_kv`. Keeps the ~14 `Services::new` call sites
+            // (mostly tests) unchanged — a field addition, not a param.
+            group_kv: Arc::new(NoopGroupKvService),
         }
     }
 
+    /// Set the §11.6 shared group-collection service (the picloud binary wires
+    /// the Postgres-backed impl; tests leave the noop default).
+    #[must_use]
+    pub fn with_group_kv(mut self, group_kv: Arc<dyn GroupKvService>) -> Self {
+        self.group_kv = group_kv;
+        self
+    }
+
     /// All-noop bundle for tests that build an `Engine` but don't
     /// exercise the stateful services. Returns the same shape as
     /// `Services::new` so callers can't accidentally rely on a stub