feat(v1.1.9): QueueService trait + Postgres impl + Services bundle wiring

- shared/services.rs: Services::new gains queue: Arc<dyn QueueService>
  positionally after users (mirrors v1.1.8's append of users).
  with_noop_services adds NoopQueueService.
- manager-core/queue_service.rs: QueueServiceImpl wraps QueueRepo with
  script-as-gate authz on AppQueueEnqueue. enqueue clamps max_attempts
  to [1,20] and delay_ms to [0, 86_400_000ms]. depth/depth_pending are
  read-only — no authz check (scripts in the app can see their own
  queue depths). cx.principal threads through as enqueued_by_principal
  (forensic only).
- manager-core/authz.rs: AppQueueEnqueue(AppId) capability — script:write
  scope, granted to editor+ (same trust shape as AppPubsubPublish).
- picloud/lib.rs: wires PostgresQueueRepo + QueueServiceImpl into
  Services::new alongside the existing v1.1.7+ services.
- 11 sdk test binaries + manager-core/realtime_authority.rs updated to
  pass NoopQueueService to Services::new.

Unit tests cover empty queue_name reject, max_attempts clamping
(0/21 → invalid), delay_ms negative-reject, anonymous principal skips
authz, depth/depth_pending pass-through.

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

crates/manager-core/src/queue_service.rs

@@ -0,0 +1,325 @@
+//! `QueueServiceImpl` — wires `QueueRepo` underneath the
+//! `picloud_shared::QueueService` trait scripts see via the Rhai bridge.
+//!
+//! Mirrors the other stateful services: script-as-gate authz
+//! (`AppQueueEnqueue`, skipped when `cx.principal` is `None`), with the
+//! backend doing the actual write. No `ServiceEventEmitter` here —
+//! enqueue is observable via `queue:receive` triggers (commit 6).
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use chrono::Utc;
+use picloud_shared::{EnqueueOpts, QueueError, QueueMessageId, QueueService, SdkCallCx};
+
+use crate::authz::{self, AuthzRepo, Capability};
+use crate::queue_repo::{NewQueueMessage, QueueRepo};
+
+/// Production impl: authz gate → repo. Trivial wrapper.
+pub struct QueueServiceImpl {
+    repo: Arc<dyn QueueRepo>,
+    authz: Arc<dyn AuthzRepo>,
+}
+
+impl QueueServiceImpl {
+    #[must_use]
+    pub fn new(repo: Arc<dyn QueueRepo>, authz: Arc<dyn AuthzRepo>) -> Self {
+        Self { repo, authz }
+    }
+}
+
+#[async_trait]
+impl QueueService for QueueServiceImpl {
+    async fn enqueue(
+        &self,
+        cx: &SdkCallCx,
+        queue_name: &str,
+        payload: serde_json::Value,
+        opts: EnqueueOpts,
+    ) -> Result<QueueMessageId, QueueError> {
+        if queue_name.is_empty() {
+            return Err(QueueError::EmptyName);
+        }
+        let max_attempts = opts.max_attempts.unwrap_or(3);
+        if !(1..=20).contains(&max_attempts) {
+            return Err(QueueError::InvalidOpts(
+                "queue::enqueue: max_attempts must be in [1, 20]".into(),
+            ));
+        }
+        if let Some(delay) = opts.delay_ms {
+            // 24h cap on delay; matches what a cron tick would do anyway.
+            if !(0..=86_400_000).contains(&delay) {
+                return Err(QueueError::InvalidOpts(
+                    "queue::enqueue: delay_ms must be in [0, 86_400_000]".into(),
+                ));
+            }
+        }
+
+        // Script-as-gate authz: anonymous public-HTTP scripts skip the
+        // check (cx.principal is None); authenticated callers must hold
+        // AppQueueEnqueue.
+        if let Some(principal) = cx.principal.as_ref() {
+            authz::require(
+                &*self.authz,
+                principal,
+                Capability::AppQueueEnqueue(cx.app_id),
+            )
+            .await
+            .map_err(|_| QueueError::Rejected("forbidden".into()))?;
+        }
+
+        let deliver_after = opts
+            .delay_ms
+            .and_then(chrono::Duration::try_milliseconds)
+            .map(|d| Utc::now() + d);
+
+        // Forensic only — never used for authz at consume-time (the
+        // queue:receive trigger runs as the registering principal per
+        // design notes §4).
+        let enqueued_by_principal = cx.principal.as_ref().map(|p| p.user_id);
+
+        self.repo
+            .enqueue(NewQueueMessage {
+                app_id: cx.app_id,
+                queue_name: queue_name.to_string(),
+                payload,
+                deliver_after,
+                max_attempts,
+                enqueued_by_principal,
+            })
+            .await
+            .map_err(|e| QueueError::Unavailable(e.to_string()))
+    }
+
+    async fn depth(&self, cx: &SdkCallCx, queue_name: &str) -> Result<u64, QueueError> {
+        if queue_name.is_empty() {
+            return Err(QueueError::EmptyName);
+        }
+        self.repo
+            .depth(cx.app_id, queue_name)
+            .await
+            .map_err(|e| QueueError::Unavailable(e.to_string()))
+    }
+
+    async fn depth_pending(&self, cx: &SdkCallCx, queue_name: &str) -> Result<u64, QueueError> {
+        if queue_name.is_empty() {
+            return Err(QueueError::EmptyName);
+        }
+        self.repo
+            .depth_pending(cx.app_id, queue_name)
+            .await
+            .map_err(|e| QueueError::Unavailable(e.to_string()))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::authz::AuthzError;
+    use picloud_shared::{AppId, AppRole, ExecutionId, RequestId, ScriptId};
+
+    struct AlwaysAllowAuthz;
+    #[async_trait]
+    impl AuthzRepo for AlwaysAllowAuthz {
+        async fn membership(
+            &self,
+            _user_id: picloud_shared::UserId,
+            _app_id: AppId,
+        ) -> Result<Option<AppRole>, AuthzError> {
+            Ok(Some(AppRole::Editor))
+        }
+    }
+
+    struct CapturingRepo {
+        last: tokio::sync::Mutex<Option<NewQueueMessage>>,
+    }
+    #[async_trait]
+    impl QueueRepo for CapturingRepo {
+        async fn enqueue(
+            &self,
+            msg: NewQueueMessage,
+        ) -> Result<QueueMessageId, crate::queue_repo::QueueRepoError> {
+            let id = QueueMessageId::new();
+            *self.last.lock().await = Some(msg);
+            Ok(id)
+        }
+        async fn claim(
+            &self,
+            _app_id: AppId,
+            _queue_name: &str,
+        ) -> Result<Option<crate::queue_repo::ClaimedMessage>, crate::queue_repo::QueueRepoError>
+        {
+            Ok(None)
+        }
+        async fn ack(
+            &self,
+            _message_id: QueueMessageId,
+            _claim_token: uuid::Uuid,
+        ) -> Result<bool, crate::queue_repo::QueueRepoError> {
+            Ok(true)
+        }
+        async fn nack(
+            &self,
+            _message_id: QueueMessageId,
+            _claim_token: uuid::Uuid,
+            _retry_delay: chrono::Duration,
+        ) -> Result<bool, crate::queue_repo::QueueRepoError> {
+            Ok(true)
+        }
+        async fn reclaim_visibility_timeouts(
+            &self,
+        ) -> Result<u64, crate::queue_repo::QueueRepoError> {
+            Ok(0)
+        }
+        async fn depth(
+            &self,
+            _app_id: AppId,
+            _queue_name: &str,
+        ) -> Result<u64, crate::queue_repo::QueueRepoError> {
+            Ok(42)
+        }
+        async fn depth_pending(
+            &self,
+            _app_id: AppId,
+            _queue_name: &str,
+        ) -> Result<u64, crate::queue_repo::QueueRepoError> {
+            Ok(7)
+        }
+        async fn list_for_app(
+            &self,
+            _app_id: AppId,
+        ) -> Result<Vec<(String, crate::queue_repo::QueueStats)>, crate::queue_repo::QueueRepoError>
+        {
+            Ok(vec![])
+        }
+        async fn dead_letter(
+            &self,
+            _message_id: QueueMessageId,
+            _claim_token: uuid::Uuid,
+            _app_id: AppId,
+            _queue_name: &str,
+            _trigger_id: Option<picloud_shared::TriggerId>,
+            _script_id: Option<ScriptId>,
+            _attempt: u32,
+            _first_attempt_at: chrono::DateTime<chrono::Utc>,
+            _last_error: &str,
+        ) -> Result<picloud_shared::DeadLetterId, crate::queue_repo::QueueRepoError> {
+            Ok(picloud_shared::DeadLetterId::new())
+        }
+    }
+
+    fn anon_cx() -> SdkCallCx {
+        let exec = ExecutionId::new();
+        SdkCallCx {
+            app_id: AppId::new(),
+            script_id: ScriptId::new(),
+            principal: None,
+            execution_id: exec,
+            request_id: RequestId::new(),
+            trigger_depth: 0,
+            root_execution_id: exec,
+            is_dead_letter_handler: false,
+            event: None,
+        }
+    }
+
+    #[tokio::test]
+    async fn empty_queue_name_rejects() {
+        let repo = Arc::new(CapturingRepo {
+            last: tokio::sync::Mutex::new(None),
+        });
+        let svc = QueueServiceImpl::new(repo.clone(), Arc::new(AlwaysAllowAuthz));
+        let cx = anon_cx();
+        let err = svc
+            .enqueue(&cx, "", serde_json::Value::Null, EnqueueOpts::default())
+            .await
+            .unwrap_err();
+        assert!(matches!(err, QueueError::EmptyName));
+    }
+
+    #[tokio::test]
+    async fn invalid_max_attempts_rejects() {
+        let repo = Arc::new(CapturingRepo {
+            last: tokio::sync::Mutex::new(None),
+        });
+        let svc = QueueServiceImpl::new(repo, Arc::new(AlwaysAllowAuthz));
+        let cx = anon_cx();
+        let err = svc
+            .enqueue(
+                &cx,
+                "x",
+                serde_json::Value::Null,
+                EnqueueOpts {
+                    delay_ms: None,
+                    max_attempts: Some(0),
+                },
+            )
+            .await
+            .unwrap_err();
+        assert!(matches!(err, QueueError::InvalidOpts(_)));
+        let err = svc
+            .enqueue(
+                &cx,
+                "x",
+                serde_json::Value::Null,
+                EnqueueOpts {
+                    delay_ms: None,
+                    max_attempts: Some(21),
+                },
+            )
+            .await
+            .unwrap_err();
+        assert!(matches!(err, QueueError::InvalidOpts(_)));
+    }
+
+    #[tokio::test]
+    async fn invalid_delay_rejects() {
+        let repo = Arc::new(CapturingRepo {
+            last: tokio::sync::Mutex::new(None),
+        });
+        let svc = QueueServiceImpl::new(repo, Arc::new(AlwaysAllowAuthz));
+        let cx = anon_cx();
+        let err = svc
+            .enqueue(
+                &cx,
+                "x",
+                serde_json::Value::Null,
+                EnqueueOpts {
+                    delay_ms: Some(-1),
+                    max_attempts: None,
+                },
+            )
+            .await
+            .unwrap_err();
+        assert!(matches!(err, QueueError::InvalidOpts(_)));
+    }
+
+    #[tokio::test]
+    async fn anon_principal_skips_authz_and_writes() {
+        let repo = Arc::new(CapturingRepo {
+            last: tokio::sync::Mutex::new(None),
+        });
+        let svc = QueueServiceImpl::new(repo.clone(), Arc::new(AlwaysAllowAuthz));
+        let cx = anon_cx();
+        let payload = serde_json::json!({ "x": 1 });
+        svc.enqueue(&cx, "jobs", payload.clone(), EnqueueOpts::default())
+            .await
+            .unwrap();
+        let captured = repo.last.lock().await.clone().unwrap();
+        assert_eq!(captured.queue_name, "jobs");
+        assert_eq!(captured.payload, payload);
+        assert_eq!(captured.max_attempts, 3);
+        assert!(captured.deliver_after.is_none());
+    }
+
+    #[tokio::test]
+    async fn depth_passes_through() {
+        let repo = Arc::new(CapturingRepo {
+            last: tokio::sync::Mutex::new(None),
+        });
+        let svc = QueueServiceImpl::new(repo, Arc::new(AlwaysAllowAuthz));
+        let cx = anon_cx();
+        assert_eq!(svc.depth(&cx, "any").await.unwrap(), 42);
+        assert_eq!(svc.depth_pending(&cx, "any").await.unwrap(), 7);
+    }
+}