fix(shared): F-Q-004 unify error variants — Forbidden on QueueError, rename Unavailable→Backend

Brings QueueError / PubsubError / InvokeError in line with sibling
shape used by KvError / DocsError / FilesError / SecretsError:

- QueueError gains an explicit Forbidden variant (previously authz
  denial was squashed into Rejected("forbidden") in queue_service.rs:68,
  losing the structured variant a 403-translation layer would need).
- QueueError::Unavailable renamed → Backend.
- PubsubError::Unavailable renamed → Backend.
- InvokeError::Unavailable renamed → Backend.
- Call sites updated (queue_service, pubsub_service, invoke_service,
  Noop* stubs, From<PubsubRepoError> impl, one test assertion).
- New unit test verifying authed-denied → QueueError::Forbidden.

AUDIT.md anchor: F-Q-004.

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

crates/manager-core/src/invoke_service.rs

@@ -49,7 +49,7 @@ impl InvokeServiceImpl {
             .scripts
             .get(script_id)
             .await
-            .map_err(|e| InvokeError::Unavailable(e.to_string()))?
+            .map_err(|e| InvokeError::Backend(e.to_string()))?
             .ok_or_else(|| InvokeError::NotFound(format!("id {script_id}")))?;
         if script.app_id != cx.app_id {
             return Err(InvokeError::CrossApp);
@@ -72,7 +72,7 @@ impl InvokeServiceImpl {
             .scripts
             .get_by_name(cx.app_id, name)
             .await
-            .map_err(|e| InvokeError::Unavailable(e.to_string()))?
+            .map_err(|e| InvokeError::Backend(e.to_string()))?
             .ok_or_else(|| InvokeError::NotFound(format!("name {name:?}")))?;
         Ok(ResolvedScript {
             script_id: script.id,
@@ -156,7 +156,7 @@ impl InvokeService for InvokeServiceImpl {
                 root_execution_id: Some(cx.root_execution_id),
             })
             .await
-            .map_err(|e| InvokeError::Unavailable(e.to_string()))?;
+            .map_err(|e| InvokeError::Backend(e.to_string()))?;
         Ok(execution_id)
     }
 }

crates/manager-core/src/pubsub_service.rs

@@ -129,7 +129,7 @@ impl PubsubServiceImpl {
 
 impl From<PubsubRepoError> for PubsubError {
     fn from(e: PubsubRepoError) -> Self {
-        Self::Unavailable(e.to_string())
+        Self::Backend(e.to_string())
     }
 }
 
@@ -227,7 +227,7 @@ impl PubsubService for PubsubServiceImpl {
 
         let (Some(topic_repo), Some(secrets)) = (self.topics.as_ref(), self.secrets.as_ref())
         else {
-            return Err(PubsubError::Unavailable(
+            return Err(PubsubError::Backend(
                 "subscriber tokens are not wired in".into(),
             ));
         };
@@ -253,7 +253,7 @@ impl PubsubService for PubsubServiceImpl {
             let registered = topic_repo
                 .get(cx.app_id, name)
                 .await
-                .map_err(|e| PubsubError::Unavailable(e.to_string()))?;
+                .map_err(|e| PubsubError::Backend(e.to_string()))?;
             if !registered.is_some_and(|t| t.external_subscribable) {
                 return Err(PubsubError::SubscriberToken(format!(
                     "pubsub::subscriber_token: topic {name} is not externally subscribable"
@@ -264,7 +264,7 @@ impl PubsubService for PubsubServiceImpl {
         let key = secrets
             .get_or_create_signing_key(cx.app_id)
             .await
-            .map_err(|e| PubsubError::Unavailable(e.to_string()))?;
+            .map_err(|e| PubsubError::Backend(e.to_string()))?;
         let now = chrono::Utc::now().timestamp();
         let claims = TokenClaims {
             app_id: cx.app_id,
@@ -472,7 +472,7 @@ mod tests {
             .publish_durable(&anon_cx(app), "user.created", serde_json::json!(1))
             .await
             .unwrap_err();
-        assert!(matches!(err, PubsubError::Unavailable(_)));
+        assert!(matches!(err, PubsubError::Backend(_)));
         // Rollback: no partial fan-out survived.
         assert_eq!(repo.written_count(), 0);
     }

crates/manager-core/src/queue_service.rs

@@ -65,7 +65,7 @@ impl QueueService for QueueServiceImpl {
                 Capability::AppQueueEnqueue(cx.app_id),
             )
             .await
-            .map_err(|_| QueueError::Rejected("forbidden".into()))?;
+            .map_err(|_| QueueError::Forbidden)?;
         }
 
         let deliver_after = opts
@@ -88,7 +88,7 @@ impl QueueService for QueueServiceImpl {
                 enqueued_by_principal,
             })
             .await
-            .map_err(|e| QueueError::Unavailable(e.to_string()))
+            .map_err(|e| QueueError::Backend(e.to_string()))
     }
 
     async fn depth(&self, cx: &SdkCallCx, queue_name: &str) -> Result<u64, QueueError> {
@@ -98,7 +98,7 @@ impl QueueService for QueueServiceImpl {
         self.repo
             .depth(cx.app_id, queue_name)
             .await
-            .map_err(|e| QueueError::Unavailable(e.to_string()))
+            .map_err(|e| QueueError::Backend(e.to_string()))
     }
 
     async fn depth_pending(&self, cx: &SdkCallCx, queue_name: &str) -> Result<u64, QueueError> {
@@ -108,7 +108,7 @@ impl QueueService for QueueServiceImpl {
         self.repo
             .depth_pending(cx.app_id, queue_name)
             .await
-            .map_err(|e| QueueError::Unavailable(e.to_string()))
+            .map_err(|e| QueueError::Backend(e.to_string()))
     }
 }
 
@@ -312,6 +312,49 @@ mod tests {
         assert!(captured.deliver_after.is_none());
     }
 
+    struct DenyAllAuthz;
+    #[async_trait]
+    impl AuthzRepo for DenyAllAuthz {
+        async fn membership(
+            &self,
+            _user_id: picloud_shared::UserId,
+            _app_id: AppId,
+        ) -> Result<Option<AppRole>, AuthzError> {
+            Ok(None)
+        }
+    }
+
+    #[tokio::test]
+    async fn authed_principal_denied_returns_forbidden_variant() {
+        use picloud_shared::{InstanceRole, Principal, UserId};
+        let repo = Arc::new(CapturingRepo {
+            last: tokio::sync::Mutex::new(None),
+        });
+        let svc = QueueServiceImpl::new(repo, Arc::new(DenyAllAuthz));
+        let exec = ExecutionId::new();
+        let cx = SdkCallCx {
+            app_id: AppId::new(),
+            script_id: ScriptId::new(),
+            principal: Some(Principal {
+                user_id: UserId::new(),
+                instance_role: InstanceRole::Member,
+                scopes: None,
+                app_binding: None,
+            }),
+            execution_id: exec,
+            request_id: RequestId::new(),
+            trigger_depth: 0,
+            root_execution_id: exec,
+            is_dead_letter_handler: false,
+            event: None,
+        };
+        let err = svc
+            .enqueue(&cx, "x", serde_json::Value::Null, EnqueueOpts::default())
+            .await
+            .unwrap_err();
+        assert!(matches!(err, QueueError::Forbidden));
+    }
+
     #[tokio::test]
     async fn depth_passes_through() {
         let repo = Arc::new(CapturingRepo {