feat(group-quota): per-group shared-collection quotas (M3)

Global env-var ceilings enforced in the group write path (mirrors the
per-value caps): PICLOUD_GROUP_KV_MAX_ROWS / _DOCS_MAX_ROWS (per-group row
count, checked only on a NEW key/doc — updates exempt) and
_FILES_MAX_TOTAL_BYTES (per-group total blob bytes). New group_quota env
helpers; count_rows/total_bytes repo methods (trait-default 0 for non-Postgres);
QuotaExceeded error variants on the three group error enums; a with_max_rows
test builder. Pinned by a group_kv_service unit test (3rd key rejected, update
of an existing key at the cap allowed).

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

crates/manager-core/src/group_kv_service.rs

@@ -13,11 +13,10 @@
 //!    (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.
+//! 4. Per-value size cap (reuses `PICLOUD_KV_MAX_VALUE_BYTES`) + a per-group
+//!    row quota (§11.6 M3, `PICLOUD_GROUP_KV_MAX_ROWS`).
+//! 5. §11.6 M2 shared-collection triggers: a write fires `shared = true`
+//!    triggers on the owning group, under the writer app (`emit_shared`).
 
 use std::sync::Arc;
 
@@ -40,6 +39,10 @@ pub struct GroupKvServiceImpl {
     resolver: Arc<dyn GroupCollectionResolver>,
     authz: Arc<dyn AuthzRepo>,
     max_value_bytes: usize,
+    /// §11.6 per-group quota: max total keys across the group's shared-KV
+    /// collections (`PICLOUD_GROUP_KV_MAX_ROWS`). Checked only when inserting a
+    /// NEW key.
+    max_rows: u64,
     /// §11.6: fires `shared = true` triggers on a shared-collection write.
     /// Defaults to the noop emitter; the host wires the outbox emitter via
     /// [`Self::with_events`].
@@ -64,6 +67,14 @@ impl GroupKvServiceImpl {
         self
     }
 
+    /// Override the per-group row quota (§11.6). Mainly for tests; the host uses
+    /// the env-var default.
+    #[must_use]
+    pub fn with_max_rows(mut self, max_rows: u64) -> Self {
+        self.max_rows = max_rows;
+        self
+    }
+
     #[must_use]
     pub fn with_max_value_bytes(
         repo: Arc<dyn GroupKvRepo>,
@@ -73,6 +84,7 @@ impl GroupKvServiceImpl {
     ) -> Self {
         Self {
             events: Arc::new(NoopEventEmitter),
+            max_rows: crate::group_quota::group_kv_max_rows_from_env(),
             repo,
             resolver,
             authz,
@@ -196,6 +208,17 @@ impl GroupKvService for GroupKvServiceImpl {
         // per-app path — the has+set is non-transactional but triggers are
         // fire-and-forget).
         let existed = self.repo.has(group_id, collection, key).await?;
+        // §11.6 quota: a NEW key must fit under the group's row ceiling; an
+        // update of an existing key is net-zero and exempt.
+        if !existed {
+            let count = self.repo.count_rows(group_id).await?;
+            if count >= self.max_rows {
+                return Err(GroupKvError::QuotaExceeded {
+                    limit: usize::try_from(self.max_rows).unwrap_or(usize::MAX),
+                    actual: usize::try_from(count).unwrap_or(usize::MAX),
+                });
+            }
+        }
         self.repo
             .set(group_id, collection, key, value.clone())
             .await?;
@@ -342,6 +365,11 @@ mod tests {
                 next_cursor: None,
             })
         }
+
+        async fn count_rows(&self, group_id: GroupId) -> Result<u64, GroupKvRepoError> {
+            let data = self.data.lock().await;
+            Ok(data.keys().filter(|(g, _, _)| *g == group_id).count() as u64)
+        }
     }
 
     /// Maps `(app_id, lowercased name) -> owning group`. Anything absent is
@@ -417,6 +445,49 @@ mod tests {
         )
     }
 
+    #[tokio::test]
+    async fn per_group_row_quota_rejects_new_keys_but_allows_updates() {
+        // §11.6: a group's shared-KV store has a row ceiling; a NEW key over the
+        // cap is rejected, but updating an existing key (net-zero) still works.
+        let app = AppId::new();
+        let group = GroupId::new();
+        let mut resolver = FakeResolver::default();
+        resolver.map.insert((app, "catalog".into()), group);
+        let kv = GroupKvServiceImpl::new(
+            Arc::new(InMemoryGroupKvRepo::default()),
+            Arc::new(resolver),
+            Arc::new(DenyingAuthzRepo),
+        )
+        .with_max_rows(2);
+        let cx = cx_with(app, Some(owner()));
+
+        kv.set(&cx, "catalog", "a", serde_json::json!(1))
+            .await
+            .unwrap();
+        kv.set(&cx, "catalog", "b", serde_json::json!(2))
+            .await
+            .unwrap();
+        // A third distinct key exceeds the cap.
+        let err = kv
+            .set(&cx, "catalog", "c", serde_json::json!(3))
+            .await
+            .unwrap_err();
+        assert!(
+            matches!(
+                err,
+                GroupKvError::QuotaExceeded {
+                    limit: 2,
+                    actual: 2
+                }
+            ),
+            "got {err:?}"
+        );
+        // Updating an existing key is exempt (net-zero).
+        kv.set(&cx, "catalog", "a", serde_json::json!(11))
+            .await
+            .expect("update of an existing key must be allowed at the cap");
+    }
+
     #[tokio::test]
     async fn unrelated_app_gets_collection_not_shared() {
         // app_a's chain declares "catalog"; app_b's does not.