fix(manager-core): F-S-001 cap kv/docs/pubsub/queue payload sizes (default 256 KB)

Files (per-file cap), secrets (64 KB default), and email (25 MB default)
already enforce limits; kv::set, docs::create/update, pubsub::publish_durable
and queue::enqueue accepted any JSON value straight to a JSONB column with
no size validation. An anonymous public-HTTP script could fill disk via
queue::enqueue or amplify a single publish into N outbox rows × payload bytes.

Adds four new error variants:
- KvError::ValueTooLarge { limit, actual }
- DocsError::ValueTooLarge { limit, actual }
- PubsubError::MessageTooLarge { limit, actual }
- QueueError::PayloadTooLarge { limit, actual }

Each stateful service grows a `max_value_bytes` field with:
- Conservative 256 KB default (DEFAULT_KV_MAX_VALUE_BYTES etc.).
- New `with_max_*` constructor preserving the old `new()` signature.
- Env-knob reader (`*_max_*_from_env()`) — mirrors SecretsConfig::from_env.

Validation runs at the entry point BEFORE authz so an anonymous DoS doesn't
pay a membership lookup per attempt.

Wired via env knobs:
- PICLOUD_KV_MAX_VALUE_BYTES
- PICLOUD_DOCS_MAX_VALUE_BYTES
- PICLOUD_PUBSUB_MAX_MESSAGE_BYTES
- PICLOUD_QUEUE_MAX_PAYLOAD_BYTES

Documented in CLAUDE.md runtime config table.

New unit test in queue_service verifying the cap fires before authz.

AUDIT.md anchor: F-S-001. Depends on F-Q-004 (Backend variant).

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

crates/manager-core/src/queue_service.rs

@@ -15,16 +15,50 @@ use picloud_shared::{EnqueueOpts, QueueError, QueueMessageId, QueueService, SdkC
 use crate::authz::{self, AuthzRepo, Capability};
 use crate::queue_repo::{NewQueueMessage, QueueRepo};
 
+/// Default per-message JSON-encoded payload cap (256 KB). Override with
+/// `PICLOUD_QUEUE_MAX_PAYLOAD_BYTES`.
+pub const DEFAULT_QUEUE_MAX_PAYLOAD_BYTES: usize = 256 * 1024;
+
+/// Read `PICLOUD_QUEUE_MAX_PAYLOAD_BYTES`; invalid values fall back to
+/// the conservative default with a warning.
+#[must_use]
+pub fn queue_max_payload_bytes_from_env() -> usize {
+    if let Ok(v) = std::env::var("PICLOUD_QUEUE_MAX_PAYLOAD_BYTES") {
+        match v.trim().parse::<usize>() {
+            Ok(n) if n > 0 => return n,
+            _ => tracing::warn!(
+                value = %v,
+                "ignoring invalid PICLOUD_QUEUE_MAX_PAYLOAD_BYTES (want a positive integer)"
+            ),
+        }
+    }
+    DEFAULT_QUEUE_MAX_PAYLOAD_BYTES
+}
+
 /// Production impl: authz gate → repo. Trivial wrapper.
 pub struct QueueServiceImpl {
     repo: Arc<dyn QueueRepo>,
     authz: Arc<dyn AuthzRepo>,
+    max_payload_bytes: usize,
 }
 
 impl QueueServiceImpl {
     #[must_use]
     pub fn new(repo: Arc<dyn QueueRepo>, authz: Arc<dyn AuthzRepo>) -> Self {
-        Self { repo, authz }
+        Self::with_max_payload_bytes(repo, authz, DEFAULT_QUEUE_MAX_PAYLOAD_BYTES)
+    }
+
+    #[must_use]
+    pub fn with_max_payload_bytes(
+        repo: Arc<dyn QueueRepo>,
+        authz: Arc<dyn AuthzRepo>,
+        max_payload_bytes: usize,
+    ) -> Self {
+        Self {
+            repo,
+            authz,
+            max_payload_bytes,
+        }
     }
 }
 
@@ -40,6 +74,17 @@ impl QueueService for QueueServiceImpl {
         if queue_name.is_empty() {
             return Err(QueueError::EmptyName);
         }
+        // Reject oversized payloads BEFORE the authz check so an
+        // anonymous public-script DoS doesn't pay the membership lookup.
+        let encoded_len = serde_json::to_vec(&payload)
+            .map(|v| v.len())
+            .map_err(|e| QueueError::Rejected(format!("encode payload: {e}")))?;
+        if encoded_len > self.max_payload_bytes {
+            return Err(QueueError::PayloadTooLarge {
+                limit: self.max_payload_bytes,
+                actual: encoded_len,
+            });
+        }
         let max_attempts = opts.max_attempts.unwrap_or(3);
         if !(1..=20).contains(&max_attempts) {
             return Err(QueueError::InvalidOpts(
@@ -311,6 +356,24 @@ mod tests {
         assert!(captured.deliver_after.is_none());
     }
 
+    #[tokio::test]
+    async fn oversized_payload_is_rejected_before_authz() {
+        let repo = Arc::new(CapturingRepo {
+            last: tokio::sync::Mutex::new(None),
+        });
+        let svc = QueueServiceImpl::with_max_payload_bytes(repo, Arc::new(AlwaysAllowAuthz), 16);
+        let cx = anon_cx();
+        let payload = serde_json::json!({ "blob": "x".repeat(200) });
+        let err = svc
+            .enqueue(&cx, "jobs", payload, EnqueueOpts::default())
+            .await
+            .unwrap_err();
+        assert!(
+            matches!(err, QueueError::PayloadTooLarge { .. }),
+            "expected PayloadTooLarge, got {err:?}"
+        );
+    }
+
     struct DenyAllAuthz;
     #[async_trait]
     impl AuthzRepo for DenyAllAuthz {