feat(quota): per-group total-byte quotas for shared KV + docs (Track A M4)

M3 quotas capped a group's shared KV/docs by ROW count only; a group could still
blow past storage with few-but-huge values. Add the symmetric total-bytes ceiling
(files already had one).

- group_quota: PICLOUD_GROUP_{KV,DOCS}_MAX_TOTAL_BYTES env vars (default 256 MiB)
  + accessors.
- GroupKvRepo/GroupDocsRepo: total_bytes(group) = SUM(octet_length(value::text))
  (default Ok(0) so non-Postgres impls skip the check).
- GroupKv/DocsServiceImpl: check the PROJECTED total (old value's bytes
  subtracted, new added) on every set/create/update, so a same-or-smaller update
  near the cap is still allowed (unlike a naive used+new check). New
  TotalBytesQuotaExceeded errors; +with_max_total_bytes for tests.
- KV set switched has->get to obtain the old value's size for the delta.

Pinned by group_kv_service + group_docs_service per_group_byte_quota_uses_the_
projected_total unit tests (reject over-cap, allow same/smaller update near cap).
No migration. CLAUDE.md env-var table updated.

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

crates/manager-core/src/group_kv_service.rs

@@ -43,6 +43,11 @@ pub struct GroupKvServiceImpl {
     /// collections (`PICLOUD_GROUP_KV_MAX_ROWS`). Checked only when inserting a
     /// NEW key.
     max_rows: u64,
+    /// §11.6 M4 per-group quota: max total stored bytes across the group's
+    /// shared-KV collections (`PICLOUD_GROUP_KV_MAX_TOTAL_BYTES`). Checked on the
+    /// projected total (old value subtracted, new added) so a same/smaller update
+    /// near the cap is still allowed.
+    max_total_bytes: 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`].
@@ -85,6 +90,7 @@ impl GroupKvServiceImpl {
         Self {
             events: Arc::new(NoopEventEmitter),
             max_rows: crate::group_quota::group_kv_max_rows_from_env(),
+            max_total_bytes: crate::group_quota::group_kv_max_total_bytes_from_env(),
             repo,
             resolver,
             authz,
@@ -92,6 +98,13 @@ impl GroupKvServiceImpl {
         }
     }
 
+    /// Override the per-group total-bytes quota (§11.6 M4). Mainly for tests.
+    #[must_use]
+    pub fn with_max_total_bytes(mut self, max_total_bytes: u64) -> Self {
+        self.max_total_bytes = max_total_bytes;
+        self
+    }
+
     /// 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.
@@ -204,11 +217,12 @@ impl GroupKvService for GroupKvServiceImpl {
             });
         }
         self.check_write(cx, group_id).await?;
-        // Determine insert vs update for the event op (best-effort, like the
-        // 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
+        // Determine insert vs update for the event op + quota deltas (best-effort,
+        // like the per-app path — the get+set is non-transactional but triggers
+        // are fire-and-forget and quotas are safety rails, not exact accounting).
+        let existing = self.repo.get(group_id, collection, key).await?;
+        let existed = existing.is_some();
+        // §11.6 row 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?;
@@ -219,6 +233,21 @@ impl GroupKvService for GroupKvServiceImpl {
                 });
             }
         }
+        // §11.6 M4 byte quota: the PROJECTED total (old value's bytes subtracted,
+        // new value's added) must fit under the group's total-bytes ceiling, so a
+        // same-or-smaller update near the cap is still allowed.
+        let old_len = existing
+            .as_ref()
+            .and_then(|v| serde_json::to_vec(v).ok())
+            .map_or(0u64, |b| b.len() as u64);
+        let used = self.repo.total_bytes(group_id).await?;
+        let projected = used.saturating_sub(old_len) + encoded_len as u64;
+        if projected > self.max_total_bytes {
+            return Err(GroupKvError::TotalBytesQuotaExceeded {
+                limit: self.max_total_bytes,
+                actual: projected,
+            });
+        }
         self.repo
             .set(group_id, collection, key, value.clone())
             .await?;
@@ -370,6 +399,15 @@ mod tests {
             let data = self.data.lock().await;
             Ok(data.keys().filter(|(g, _, _)| *g == group_id).count() as u64)
         }
+
+        async fn total_bytes(&self, group_id: GroupId) -> Result<u64, GroupKvRepoError> {
+            let data = self.data.lock().await;
+            Ok(data
+                .iter()
+                .filter(|((g, _, _), _)| *g == group_id)
+                .map(|(_, v)| serde_json::to_vec(v).map_or(0, |b| b.len() as u64))
+                .sum())
+        }
     }
 
     /// Maps `(app_id, lowercased name) -> owning group`. Anything absent is
@@ -488,6 +526,56 @@ mod tests {
             .expect("update of an existing key must be allowed at the cap");
     }
 
+    #[tokio::test]
+    async fn per_group_byte_quota_uses_the_projected_total() {
+        // §11.6 M4: the byte ceiling is checked on the PROJECTED total, so a
+        // write that would push the group over is rejected, but a same/smaller
+        // update near the cap is still allowed (unlike a naive used+new check).
+        let app = AppId::new();
+        let group = GroupId::new();
+        let mut resolver = FakeResolver::default();
+        resolver.map.insert((app, "catalog".into()), group);
+        // A 20-byte ceiling; each `"xxxxx"` value encodes to 7 bytes ("\"xxxxx\"").
+        let kv = GroupKvServiceImpl::new(
+            Arc::new(InMemoryGroupKvRepo::default()),
+            Arc::new(resolver),
+            Arc::new(DenyingAuthzRepo),
+        )
+        .with_max_rows(1000)
+        .with_max_total_bytes(20);
+        let cx = cx_with(app, Some(owner()));
+
+        kv.set(&cx, "catalog", "a", serde_json::json!("xxxxx"))
+            .await
+            .unwrap(); // used = 7
+        kv.set(&cx, "catalog", "b", serde_json::json!("xxxxx"))
+            .await
+            .unwrap(); // used = 14
+                       // A third 7-byte value → projected 21 > 20 → rejected.
+        let err = kv
+            .set(&cx, "catalog", "c", serde_json::json!("xxxxx"))
+            .await
+            .unwrap_err();
+        assert!(
+            matches!(err, GroupKvError::TotalBytesQuotaExceeded { limit: 20, .. }),
+            "got {err:?}"
+        );
+        // A same-size update of an existing key stays within budget (projected
+        // = 14 - 7 + 7 = 14 ≤ 20) — the delta rule, not used+new.
+        kv.set(&cx, "catalog", "a", serde_json::json!("yyyyy"))
+            .await
+            .expect("a same-size update near the cap must be allowed");
+        // Growing an existing value past the cap is still rejected.
+        let err = kv
+            .set(&cx, "catalog", "a", serde_json::json!("zzzzzzzzzzzzzzz"))
+            .await
+            .unwrap_err();
+        assert!(
+            matches!(err, GroupKvError::TotalBytesQuotaExceeded { .. }),
+            "got {err:?}"
+        );
+    }
+
     #[tokio::test]
     async fn unrelated_app_gets_collection_not_shared() {
         // app_a's chain declares "catalog"; app_b's does not.