chore(v1.1.9): clippy clean — workspace -D warnings

Surfaced + fixed during the F3 attestation:

- executor-core/sdk/queue.rs: drop redundant .map(|()| ()) call sites;
  enqueue_blocking discards QueueMessageId via .map(|_id| ()) (intentional)
- executor-core/sdk/retry.rs: hoist use std::collections::hash_map::DefaultHasher
  + use std::hash::Hasher to the top of the file (clippy::items_after_statements);
  replace `as i64` casts with i64::try_from + clear comment about jitter
  bound (clippy::cast_possible_wrap)
- executor-core/sdk/invoke.rs: move _LIMITS_IS_COPY const before #[cfg(test)]
  mod tests (clippy::items_after_test_module)
- manager-core/dispatcher.rs: dispatch_one_queue gains #[allow(too_many_lines)]
  (it's the queue tick's whole logic — split makes it less readable than
  the lint); .map(...).unwrap_or(default) → .map_or(default, ...)
- picloud/lib.rs: Limits { trigger_depth_max, ..Limits::default() } via
  struct-update instead of let-mut-assign (clippy::field_reassign_with_default)
- tests: r#"..."# → r"..." where there are no `"` inside (clippy::needless_raw_string_hashes);
  combine InvokeTarget::Name | InvokeTarget::Path arms with identical bodies
  in sdk_invoke.rs (clippy::match_same_arms)

cargo fmt --all -- --check 2>&1 | tail -3: no output (exit 0)
cargo clippy --workspace --all-targets --all-features -- -D warnings:
  Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.41s
Workspace lib smoke: 432 passing (306 manager-core + 74 executor-core + 34 shared + 18 orchestrator-core).
DB-gated E2E (against docker compose postgres on localhost:15432):
  queue_e2e: 4 ok / invoke_e2e: 4 ok / retry_e2e: 3 ok / migration_queue_messages: 4 ok / schema_snapshot: 1 ok

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

crates/executor-core/src/engine.rs

@@ -195,7 +195,13 @@ impl Engine {
         );
         engine.set_module_resolver(resolver);
         let self_engine = self.self_arc();
-        sdk::register_all(&mut engine, &self.services, cx, effective_limits, self_engine);
+        sdk::register_all(
+            &mut engine,
+            &self.services,
+            cx,
+            effective_limits,
+            self_engine,
+        );
 
         let mut scope = Scope::new();
         scope.push_constant("ctx", build_ctx_map(&req));

crates/executor-core/src/sdk/invoke.rs

@@ -257,7 +257,9 @@ fn args_to_json(value: &Dynamic) -> Result<Json, Box<EvalAltResult>> {
         return Ok(serde_json::Number::from_f64(f).map_or(Json::Null, Json::Number));
     }
     if value.is_string() {
-        return Ok(Json::String(value.clone().into_string().unwrap_or_default()));
+        return Ok(Json::String(
+            value.clone().into_string().unwrap_or_default(),
+        ));
     }
     if let Some(arr) = value.clone().try_cast::<Array>() {
         let mut out = Vec::with_capacity(arr.len());
@@ -276,6 +278,14 @@ fn args_to_json(value: &Dynamic) -> Result<Json, Box<EvalAltResult>> {
     Ok(Json::String(value.to_string()))
 }
 
+/// `Limits` is `Copy`; passed by value into `register` so closures take
+/// owned copies. Hint for readers wondering why it doesn't need `Arc`.
+#[allow(dead_code)]
+const _LIMITS_IS_COPY: fn() = || {
+    fn assert_copy<T: Copy>() {}
+    assert_copy::<Limits>();
+};
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -320,11 +330,3 @@ mod tests {
         assert_eq!(j, serde_json::json!({ "x": 42 }));
     }
 }
-
-/// `Limits` is `Copy`; passed by value into `register` so closures take
-/// owned copies. Hint for readers wondering why it doesn't need `Arc`.
-#[allow(dead_code)]
-const _LIMITS_IS_COPY: fn() = || {
-    fn assert_copy<T: Copy>() {}
-    assert_copy::<Limits>();
-};

crates/executor-core/src/sdk/queue.rs

@@ -37,7 +37,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
             "enqueue",
             move |name: &str, message: Dynamic| -> Result<(), Box<EvalAltResult>> {
                 let json = message_to_json(&message)?;
-                enqueue_blocking(&svc, &cx, name, json, EnqueueOpts::default()).map(|_| ())
+                enqueue_blocking(&svc, &cx, name, json, EnqueueOpts::default())
             },
         );
     }
@@ -52,7 +52,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
             move |name: &str, message: Dynamic, opts: Map| -> Result<(), Box<EvalAltResult>> {
                 let json = message_to_json(&message)?;
                 let opts = parse_opts(&opts)?;
-                enqueue_blocking(&svc, &cx, name, json, opts).map(|_| ())
+                enqueue_blocking(&svc, &cx, name, json, opts)
             },
         );
     }
@@ -111,7 +111,7 @@ fn enqueue_blocking(
     let name = name.to_string();
     handle
         .block_on(async move { svc.enqueue(&cx, &name, payload, opts).await })
-        .map(|_| ())
+        .map(|_id| ())
         .map_err(|err| -> Box<EvalAltResult> {
             EvalAltResult::ErrorRuntime(format!("queue: {err}").into(), rhai::Position::NONE).into()
         })
@@ -187,7 +187,9 @@ fn message_to_json(value: &Dynamic) -> Result<Json, Box<EvalAltResult>> {
         return Ok(serde_json::Number::from_f64(f).map_or(Json::Null, Json::Number));
     }
     if value.is_string() {
-        return Ok(Json::String(value.clone().into_string().unwrap_or_default()));
+        return Ok(Json::String(
+            value.clone().into_string().unwrap_or_default(),
+        ));
     }
     if let Some(arr) = value.clone().try_cast::<Array>() {
         let mut out = Vec::with_capacity(arr.len());

crates/executor-core/src/sdk/retry.rs

@@ -31,6 +31,8 @@
 //! `invoke()`, the inner call gets a fresh cx with `trigger_depth + 1`
 //! via the invoke bridge — retry doesn't see or interfere with that.
 
+use std::collections::hash_map::DefaultHasher;
+use std::hash::Hasher;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -81,11 +83,7 @@ impl BackoffShape {
     }
 }
 
-pub(super) fn register(
+pub(super) fn register(engine: &mut RhaiEngine, _services: &Services, _cx: Arc<SdkCallCx>) {
-    engine: &mut RhaiEngine,
-    _services: &Services,
-    _cx: Arc<SdkCallCx>,
-) {
     // Custom type so scripts can pass Policy values around.
     engine.register_type_with_name::<Policy>("Policy");
 
@@ -125,9 +123,10 @@ pub(super) fn register(
     // not `with` because `with` is a Rhai reserved keyword.
     module.set_native_fn(
         "run",
-        move |ctx: NativeCallContext, policy: Policy, fn_ptr: FnPtr| -> Result<Dynamic, Box<EvalAltResult>> {
+        move |ctx: NativeCallContext,
-            retry_run(&ctx, &policy, &fn_ptr)
+              policy: Policy,
-        },
+              fn_ptr: FnPtr|
+              -> Result<Dynamic, Box<EvalAltResult>> { retry_run(&ctx, &policy, &fn_ptr) },
     );
 
     engine.register_static_module("retry", module.into());
@@ -147,20 +146,16 @@ fn build_policy(opts: rhai::Map) -> Result<Policy, Box<EvalAltResult>> {
         p.max_attempts = n.clamp(1, 20);
     }
     if let Some(v) = opts.get("backoff") {
-        let s = v
+        let s = v.clone().into_string().map_err(|_| -> Box<EvalAltResult> {
-            .clone()
+            EvalAltResult::ErrorRuntime(
-            .into_string()
+                "retry::policy: backoff must be a string".into(),
-            .map_err(|_| -> Box<EvalAltResult> {
+                rhai::Position::NONE,
-                EvalAltResult::ErrorRuntime(
+            )
-                    "retry::policy: backoff must be a string".into(),
+            .into()
-                    rhai::Position::NONE,
+        })?;
-                )
-                .into()
-            })?;
         p.backoff = BackoffShape::from_wire(&s).ok_or_else(|| -> Box<EvalAltResult> {
             EvalAltResult::ErrorRuntime(
-                "retry::policy: backoff must be \"exponential\" | \"linear\" | \"constant\""
+                "retry::policy: backoff must be \"exponential\" | \"linear\" | \"constant\"".into(),
-                    .into(),
                 rhai::Position::NONE,
             )
             .into()
@@ -251,13 +246,16 @@ fn apply_jitter(raw: u32, pct: u32) -> u32 {
     // Deterministic-enough jitter without pulling rand into a hot path:
     // pick a small offset from the high bits of attempt's hash. Random
     // distribution doesn't matter for retry — bounded ± is what we want.
-    use std::collections::hash_map::DefaultHasher;
-    use std::hash::Hasher;
     let mut h = DefaultHasher::new();
     h.write_u64(span);
     h.write_u32(raw);
     let r = h.finish();
-    let offset = (r % (2 * span + 1)) as i64 - span as i64;
+    // span is bounded by raw*100/100 = raw ≤ u32::MAX, so 2*span+1 fits
+    // in i64 without wrapping. `cast_signed` makes the intent explicit
+    // for clippy::cast_possible_wrap.
+    let modded = r % (2 * span + 1);
+    let offset =
+        i64::try_from(modded).unwrap_or(i64::MAX) - i64::try_from(span).unwrap_or(i64::MAX);
     let signed = i64::from(raw).saturating_add(offset).max(0);
     u32::try_from(signed.min(i64::from(u32::MAX))).unwrap_or(u32::MAX)
 }

crates/executor-core/tests/sdk_invoke.rs

@@ -45,8 +45,9 @@ impl InvokeService for FakeInvokeService {
         let entries = self.scripts.lock().unwrap().clone();
         let hit = entries.iter().find(|(id, _app, name, _)| match &target {
             InvokeTarget::Id(t) => t == id,
-            InvokeTarget::Name(t) => t == name,
+            // Test stashes route paths in the `name` column, so Path
-            InvokeTarget::Path(t) => t == name, // tests stash route paths as names
+            // and Name match the same field.
+            InvokeTarget::Name(t) | InvokeTarget::Path(t) => t == name,
         });
         let (id, app, name, source): (ScriptId, AppId, String, String) = hit
             .ok_or_else(|| InvokeError::NotFound(target.describe()))?
@@ -241,7 +242,10 @@ async fn invoke_async_returns_execution_id_string() {
         .unwrap();
     // Body is a string — a UUID — surfaced as Json::String.
     let id = resp.body.as_str().expect("body should be a string");
-    assert!(uuid::Uuid::parse_str(id).is_ok(), "expected UUID, got: {id}");
+    assert!(
+        uuid::Uuid::parse_str(id).is_ok(),
+        "expected UUID, got: {id}"
+    );
     let payloads = svc.async_payloads.lock().unwrap().clone();
     assert_eq!(payloads.len(), 1);
     assert_eq!(payloads[0], json!({ "x": 1 }));

crates/executor-core/tests/sdk_queue.rs

@@ -46,11 +46,7 @@ impl QueueService for RecordingQueue {
         Ok(*self.depth.lock().unwrap())
     }
 
-    async fn depth_pending(
+    async fn depth_pending(&self, _cx: &SdkCallCx, _queue_name: &str) -> Result<u64, QueueError> {
-        &self,
-        _cx: &SdkCallCx,
-        _queue_name: &str,
-    ) -> Result<u64, QueueError> {
         Ok(*self.depth_pending.lock().unwrap())
     }
 }

crates/executor-core/tests/sdk_retry.rs

@@ -8,10 +8,10 @@ use std::sync::{Arc, Mutex};
 
 use picloud_executor_core::{Engine, ExecRequest, InvocationType, Limits};
 use picloud_shared::{
-    AppId, ExecutionId, NoopDeadLetterService, NoopDocsService, NoopEmailService,
+    AppId, ExecutionId, NoopDeadLetterService, NoopDocsService, NoopEmailService, NoopEventEmitter,
-    NoopEventEmitter, NoopFilesService, NoopHttpService, NoopInvokeService, NoopKvService,
+    NoopFilesService, NoopHttpService, NoopInvokeService, NoopKvService, NoopModuleSource,
-    NoopModuleSource, NoopPubsubService, NoopQueueService, NoopSecretsService, NoopUsersService,
+    NoopPubsubService, NoopQueueService, NoopSecretsService, NoopUsersService, RequestId, ScriptId,
-    RequestId, ScriptId, ScriptSandbox, Services,
+    ScriptSandbox, Services,
 };
 use serde_json::Value;
 
@@ -61,11 +61,11 @@ fn baseline_request(app_id: AppId) -> ExecRequest {
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn retry_run_succeeds_first_try() {
     let engine = build_engine();
-    let src = r#"
+    let src = r"
         let p = retry::policy(#{ max_attempts: 3, base_ms: 1 });
         let v = retry::run(p, || 42);
         #{ statusCode: 200, body: v }
-    "#
+    "
     .to_string();
     let req = baseline_request(AppId::new());
     let resp = tokio::task::spawn_blocking(move || engine.execute(&src, req))

crates/manager-core/src/auth_api.rs

@@ -92,11 +92,7 @@ async fn login(State(state): State<AuthState>, Json(input): Json<LoginRequest>)
     // canonical row used in the response DTO.
     let (stored_hash, user_id, is_active) = match creds {
         Some(c) => (c.password_hash, Some(c.id), c.is_active),
-        None => (
+        None => (crate::auth::TIMING_FLAT_DUMMY_HASH.to_string(), None, false),
-            crate::auth::TIMING_FLAT_DUMMY_HASH.to_string(),
-            None,
-            false,
-        ),
     };
 
     let password_ok = verify_password(&stored_hash, &input.password);

crates/manager-core/src/dispatcher.rs

@@ -163,6 +163,7 @@ impl Dispatcher {
         Ok(())
     }
 
+    #[allow(clippy::too_many_lines)]
     async fn dispatch_one_queue(
         &self,
         consumer: &ActiveQueueConsumer,
@@ -183,7 +184,11 @@ impl Dispatcher {
         let Ok(permit) = self.gate.try_acquire() else {
             let _ = self
                 .queue
-                .nack(claimed.id, claimed.claim_token, chrono::Duration::milliseconds(100))
+                .nack(
+                    claimed.id,
+                    claimed.claim_token,
+                    chrono::Duration::milliseconds(100),
+                )
                 .await;
             return Ok(());
         };
@@ -299,7 +304,11 @@ impl Dispatcher {
                 self.config.retry_jitter_pct,
             );
             let delay = chrono::Duration::milliseconds(i64::from(delay_ms));
-            if let Err(e) = self.queue.nack(claimed.id, claimed.claim_token, delay).await {
+            if let Err(e) = self
+                .queue
+                .nack(claimed.id, claimed.claim_token, delay)
+                .await
+            {
                 tracing::warn!(?e, "queue nack failed");
             }
             return;
@@ -621,12 +630,14 @@ impl Dispatcher {
             ));
         }
 
-        let args = payload.get("args").cloned().unwrap_or(serde_json::Value::Null);
+        let args = payload
+            .get("args")
+            .cloned()
+            .unwrap_or(serde_json::Value::Null);
         let trigger_depth = payload
             .get("trigger_depth")
             .and_then(serde_json::Value::as_u64)
-            .map(|n| u32::try_from(n).unwrap_or(u32::MAX))
+            .map_or(row.trigger_depth, |n| u32::try_from(n).unwrap_or(u32::MAX));
-            .unwrap_or(row.trigger_depth);
 
         let execution_id = ExecutionId::new();
         let req = ExecRequest {

crates/manager-core/src/invoke_service.rs

@@ -94,9 +94,8 @@ impl InvokeServiceImpl {
         // most route-resolution invokes target write endpoints; a more
         // exact API would surface method via the SDK in a future bump.
         let app_routes = self.routes.snapshot_for_app(cx.app_id);
-        let m = matcher::r#match(&app_routes, "invoke.local", "POST", path).or_else(|| {
+        let m = matcher::r#match(&app_routes, "invoke.local", "POST", path)
-            matcher::r#match(&app_routes, "invoke.local", "GET", path)
+            .or_else(|| matcher::r#match(&app_routes, "invoke.local", "GET", path));
-        });
         let m = m.ok_or_else(|| InvokeError::NotFound(format!("path {path:?}")))?;
         // Resolve the script the matched route bound. Cross-app
         // re-check via the script row (defense in depth — RouteTable
@@ -175,10 +174,7 @@ mod tests {
     }
     #[async_trait]
     impl ScriptRepository for OneScriptRepo {
-        async fn get(
+        async fn get(&self, id: ScriptId) -> Result<Option<Script>, ScriptRepositoryError> {
-            &self,
-            id: ScriptId,
-        ) -> Result<Option<Script>, ScriptRepositoryError> {
             Ok(if id == self.script.id {
                 Some(self.script.clone())
             } else {
@@ -190,19 +186,18 @@ mod tests {
             app_id: AppId,
             name: &str,
         ) -> Result<Option<Script>, ScriptRepositoryError> {
-            Ok(if app_id == self.script.app_id && name == self.script.name {
+            Ok(
-                Some(self.script.clone())
+                if app_id == self.script.app_id && name == self.script.name {
-            } else {
+                    Some(self.script.clone())
-                None
+                } else {
-            })
+                    None
+                },
+            )
         }
         async fn list(&self) -> Result<Vec<Script>, ScriptRepositoryError> {
             Ok(vec![self.script.clone()])
         }
-        async fn list_for_app(
+        async fn list_for_app(&self, _app_id: AppId) -> Result<Vec<Script>, ScriptRepositoryError> {
-            &self,
-            _app_id: AppId,
-        ) -> Result<Vec<Script>, ScriptRepositoryError> {
             unimplemented!()
         }
         async fn list_for_user(
@@ -331,7 +326,9 @@ mod tests {
         let app_b = AppId::new();
         let script = make_script(app_a, "worker");
         let svc = InvokeServiceImpl::new(
-            Arc::new(OneScriptRepo { script: script.clone() }),
+            Arc::new(OneScriptRepo {
+                script: script.clone(),
+            }),
             Arc::new(RouteTable::new()),
             Arc::new(CapturingOutbox {
                 last: tokio::sync::Mutex::new(None),
@@ -339,7 +336,10 @@ mod tests {
         );
         // Caller is in app_b but the script belongs to app_a.
         let cx = anon_cx(app_b);
-        let err = svc.resolve(&cx, InvokeTarget::Id(script.id)).await.unwrap_err();
+        let err = svc
+            .resolve(&cx, InvokeTarget::Id(script.id))
+            .await
+            .unwrap_err();
         assert!(matches!(err, InvokeError::CrossApp));
     }
 
@@ -348,7 +348,9 @@ mod tests {
         let app_id = AppId::new();
         let script = make_script(app_id, "worker");
         let svc = InvokeServiceImpl::new(
-            Arc::new(OneScriptRepo { script: script.clone() }),
+            Arc::new(OneScriptRepo {
+                script: script.clone(),
+            }),
             Arc::new(RouteTable::new()),
             Arc::new(CapturingOutbox {
                 last: tokio::sync::Mutex::new(None),
@@ -389,7 +391,9 @@ mod tests {
             last: tokio::sync::Mutex::new(None),
         });
         let svc = InvokeServiceImpl::new(
-            Arc::new(OneScriptRepo { script: script.clone() }),
+            Arc::new(OneScriptRepo {
+                script: script.clone(),
+            }),
             Arc::new(RouteTable::new()),
             outbox.clone(),
         );

crates/manager-core/src/lib.rs

@@ -45,6 +45,7 @@ pub mod files_service;
 pub mod files_sweep;
 pub mod gc;
 pub mod http_service;
+pub mod invoke_service;
 pub mod kv_repo;
 pub mod kv_service;
 pub mod log_sink;
@@ -55,7 +56,6 @@ pub mod outbox_repo;
 pub mod principal_resolver;
 pub mod pubsub_repo;
 pub mod pubsub_service;
-pub mod invoke_service;
 pub mod queue_repo;
 pub mod queue_service;
 pub mod queues_api;

crates/manager-core/src/queue_repo.rs

@@ -114,8 +114,7 @@ pub trait QueueRepo: Send + Sync {
     async fn depth(&self, app_id: AppId, queue_name: &str) -> Result<u64, QueueRepoError>;
 
     /// `queue::depth_pending(name)` — currently claimable rows.
-    async fn depth_pending(&self, app_id: AppId, queue_name: &str)
+    async fn depth_pending(&self, app_id: AppId, queue_name: &str) -> Result<u64, QueueRepoError>;
-        -> Result<u64, QueueRepoError>;
 
     /// Dashboard read-only view: every distinct queue name in `app_id`
     /// with its aggregate counts.
@@ -217,13 +216,11 @@ impl QueueRepo for PostgresQueueRepo {
         message_id: QueueMessageId,
         claim_token: Uuid,
     ) -> Result<bool, QueueRepoError> {
-        let res = sqlx::query(
+        let res = sqlx::query("DELETE FROM queue_messages WHERE id = $1 AND claim_token = $2")
-            "DELETE FROM queue_messages WHERE id = $1 AND claim_token = $2",
+            .bind(message_id.into_inner())
-        )
+            .bind(claim_token)
-        .bind(message_id.into_inner())
+            .execute(&self.pool)
-        .bind(claim_token)
+            .await?;
-        .execute(&self.pool)
-        .await?;
         Ok(res.rows_affected() == 1)
     }
 
@@ -275,11 +272,7 @@ impl QueueRepo for PostgresQueueRepo {
         Ok(u64::try_from(n).unwrap_or(0))
     }
 
-    async fn depth_pending(
+    async fn depth_pending(&self, app_id: AppId, queue_name: &str) -> Result<u64, QueueRepoError> {
-        &self,
-        app_id: AppId,
-        queue_name: &str,
-    ) -> Result<u64, QueueRepoError> {
         let (n,): (i64,) = sqlx::query_as(
             "SELECT COUNT(*) FROM queue_messages \
              WHERE app_id = $1 AND queue_name = $2 \

crates/manager-core/src/trigger_repo.rs

@@ -1323,13 +1323,11 @@ impl TriggerRepo for PostgresTriggerRepo {
         trigger_id: TriggerId,
         at: DateTime<Utc>,
     ) -> Result<(), TriggerRepoError> {
-        sqlx::query(
+        sqlx::query("UPDATE queue_trigger_details SET last_fired_at = $2 WHERE trigger_id = $1")
-            "UPDATE queue_trigger_details SET last_fired_at = $2 WHERE trigger_id = $1",
+            .bind(trigger_id.into_inner())
-        )
+            .bind(at)
-        .bind(trigger_id.into_inner())
+            .execute(&self.pool)
-        .bind(at)
+            .await?;
-        .execute(&self.pool)
-        .await?;
         Ok(())
     }
 }

crates/manager-core/tests/migration_queue_messages.rs

@@ -31,7 +31,9 @@ async fn pool_or_skip() -> Option<PgPool> {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn queue_messages_table_exists_with_expected_columns() {
-    let Some(pool) = pool_or_skip().await else { return };
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
     let rows: Vec<(String, String, String)> = sqlx::query_as(
         "SELECT column_name, data_type, is_nullable \
          FROM information_schema.columns \
@@ -63,10 +65,8 @@ async fn queue_messages_table_exists_with_expected_columns() {
     }
 
     // Verify nullability on the right columns.
-    let nullable: std::collections::HashMap<String, String> = rows
+    let nullable: std::collections::HashMap<String, String> =
-        .into_iter()
+        rows.into_iter().map(|(n, _, nl)| (n, nl)).collect();
-        .map(|(n, _, nl)| (n, nl))
-        .collect();
     assert_eq!(nullable["app_id"], "NO");
     assert_eq!(nullable["queue_name"], "NO");
     assert_eq!(nullable["payload"], "NO");
@@ -77,7 +77,9 @@ async fn queue_messages_table_exists_with_expected_columns() {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn queue_trigger_details_table_exists() {
-    let Some(pool) = pool_or_skip().await else { return };
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
     let rows: Vec<(String,)> = sqlx::query_as(
         "SELECT column_name FROM information_schema.columns \
          WHERE table_name = 'queue_trigger_details' \
@@ -87,7 +89,12 @@ async fn queue_trigger_details_table_exists() {
     .await
     .expect("read");
     let names: Vec<&str> = rows.iter().map(|(n,)| n.as_str()).collect();
-    for required in ["trigger_id", "queue_name", "visibility_timeout_secs", "last_fired_at"] {
+    for required in [
+        "trigger_id",
+        "queue_name",
+        "visibility_timeout_secs",
+        "last_fired_at",
+    ] {
         assert!(
             names.contains(&required),
             "queue_trigger_details should have {required}, columns are {names:?}"
@@ -97,7 +104,9 @@ async fn queue_trigger_details_table_exists() {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn queue_widens_trigger_kind_and_outbox_source_kind() {
-    let Some(pool) = pool_or_skip().await else { return };
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
     // The triggers.kind constraint must accept 'queue'.
     sqlx::query(
         "DO $$ BEGIN PERFORM 1 FROM pg_constraint WHERE conname = 'triggers_kind_check'; END $$;",
@@ -133,7 +142,9 @@ async fn queue_widens_trigger_kind_and_outbox_source_kind() {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn queue_messages_dispatch_index_is_partial() {
-    let Some(pool) = pool_or_skip().await else { return };
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
     let (defn,): (String,) = sqlx::query_as(
         "SELECT indexdef FROM pg_indexes \
          WHERE indexname = 'idx_queue_messages_dispatch'",

crates/picloud/src/lib.rs

@@ -264,16 +264,14 @@ pub async fn build_app(
     // v1.1.9 durable per-app queues. Producers write to queue_messages
     // via QueueService; the dispatcher's queue arm (commit 6) consumes
     // claimed messages and fires the registered queue:receive trigger.
-    let queue_repo: Arc<dyn picloud_manager_core::queue_repo::QueueRepo> =
+    let queue_repo: Arc<dyn picloud_manager_core::queue_repo::QueueRepo> = Arc::new(
-        Arc::new(picloud_manager_core::queue_repo::PostgresQueueRepo::new(
+        picloud_manager_core::queue_repo::PostgresQueueRepo::new(pool.clone()),
-            pool.clone(),
+    );
-        ));
+    let queue: Arc<dyn picloud_shared::QueueService> =
-    let queue: Arc<dyn picloud_shared::QueueService> = Arc::new(
+        Arc::new(picloud_manager_core::queue_service::QueueServiceImpl::new(
-        picloud_manager_core::queue_service::QueueServiceImpl::new(
             queue_repo.clone(),
             authz.clone(),
-        ),
+        ));
-    );
     // Route table created early (before Services) so InvokeServiceImpl
     // can use it for path resolution. It's populated below from the
     // route_repo, then re-populated whenever the admin layer writes
@@ -312,8 +310,10 @@ pub async fn build_app(
     );
     // v1.1.9: keep the invoke depth bound aligned with the dispatcher's
     // trigger-depth bound (same counter under the hood).
-    let mut engine_limits = Limits::default();
+    let engine_limits = Limits {
-    engine_limits.trigger_depth_max = trigger_config.max_trigger_depth;
+        trigger_depth_max: trigger_config.max_trigger_depth,
+        ..Limits::default()
+    };
     let engine = Arc::new(Engine::new(engine_limits, services));
     // v1.1.9: install the back-reference the `invoke` SDK bridge needs
     // for synchronous re-entry. Weak so the Arc-cycle stays loose;

crates/picloud/tests/invoke_e2e.rs

@@ -116,7 +116,9 @@ async fn poll_marker(pool: &PgPool, app_id: &str) -> Option<Value> {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn invoke_by_name_same_app_returns_value() {
-    let Some(pool) = pool_or_skip().await else { return };
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
     let (server, app_id) = server_for(pool.clone(), "invoke-name").await;
 
     // Callee returns 42 — the simplest possible round-trip.
@@ -124,7 +126,7 @@ async fn invoke_by_name_same_app_returns_value() {
         &server,
         &app_id,
         "callee",
-        r#"#{ statusCode: 200, body: 42 }"#,
+        r"#{ statusCode: 200, body: 42 }",
     )
     .await;
 
@@ -146,7 +148,9 @@ async fn invoke_by_name_same_app_returns_value() {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn invoke_cross_app_rejects() {
-    let Some(pool) = pool_or_skip().await else { return };
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
     let (server, app_a) = server_for(pool.clone(), "invoke-xa").await;
     // Create a second app via the SAME server / admin (Owner can manage
     // every app) so the script-create calls don't run into role gates.
@@ -163,7 +167,7 @@ async fn invoke_cross_app_rejects() {
         &server,
         &app_b,
         "callee_b",
-        r#"#{ statusCode: 200, body: 0 }"#,
+        r"#{ statusCode: 200, body: 0 }",
     )
     .await;
 
@@ -193,7 +197,9 @@ async fn invoke_cross_app_rejects() {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn invoke_depth_limit_exceeds_cleanly() {
-    let Some(pool) = pool_or_skip().await else { return };
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
     let (server, app_id) = server_for(pool.clone(), "invoke-depth").await;
 
     // The recurser propagates any throw (no try-catch). At the depth
@@ -235,7 +241,9 @@ async fn invoke_depth_limit_exceeds_cleanly() {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn invoke_async_enqueues_outbox_row() {
-    let Some(pool) = pool_or_skip().await else { return };
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
     let (server, app_id) = server_for(pool.clone(), "invoke-async").await;
 
     // Callee writes a marker the test polls — the dispatcher must fire

crates/picloud/tests/queue_e2e.rs

@@ -127,14 +127,13 @@ async fn poll_marker(pool: &PgPool, app_id: &str) -> Option<Value> {
 }
 
 async fn count_queue_messages(pool: &PgPool, app_id: &str, queue: &str) -> i64 {
-    let (n,): (i64,) = sqlx::query_as(
+    let (n,): (i64,) =
-        "SELECT COUNT(*) FROM queue_messages WHERE app_id = $1 AND queue_name = $2",
+        sqlx::query_as("SELECT COUNT(*) FROM queue_messages WHERE app_id = $1 AND queue_name = $2")
-    )
+            .bind(Uuid::parse_str(app_id).expect("uuid"))
-    .bind(Uuid::parse_str(app_id).expect("uuid"))
+            .bind(queue)
-    .bind(queue)
+            .fetch_one(pool)
-    .fetch_one(pool)
+            .await
-    .await
+            .expect("queue count");
-    .expect("queue count");
     n
 }
 
@@ -170,7 +169,9 @@ async fn enqueue_directly(pool: &PgPool, app_id: &str, queue: &str, payload: Val
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn queue_receive_acks_on_success() {
-    let Some(pool) = pool_or_skip().await else { return };
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
     let (server, app_id) = server_for(pool.clone(), "ack").await;
 
     let script_id = create_script(&server, &app_id, "worker", ACK_HANDLER).await;
@@ -186,9 +187,7 @@ async fn queue_receive_acks_on_success() {
 
     enqueue_directly(&pool, &app_id, "jobs", json!({ "x": 42 })).await;
 
-    let marker = poll_marker(&pool, &app_id)
+    let marker = poll_marker(&pool, &app_id).await.expect("handler fired");
-        .await
-        .expect("handler fired");
     assert_eq!(marker["source"], "queue");
     assert_eq!(marker["queue"]["queue_name"], "jobs");
     assert_eq!(marker["queue"]["message"]["x"], 42);
@@ -199,7 +198,9 @@ async fn queue_receive_acks_on_success() {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn queue_receive_dead_letters_after_max_attempts() {
-    let Some(pool) = pool_or_skip().await else { return };
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
     let (server, app_id) = server_for(pool.clone(), "dl").await;
 
     let script_id = create_script(&server, &app_id, "fail", THROW_HANDLER).await;
@@ -236,7 +237,9 @@ async fn queue_receive_dead_letters_after_max_attempts() {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn queue_one_consumer_per_queue_rejected() {
-    let Some(pool) = pool_or_skip().await else { return };
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
     let (server, app_id) = server_for(pool.clone(), "onecon").await;
 
     let s1 = create_script(&server, &app_id, "first", ACK_HANDLER).await;
@@ -276,7 +279,9 @@ async fn queue_one_consumer_per_queue_rejected() {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn queue_visibility_timeout_reclaim() {
-    let Some(pool) = pool_or_skip().await else { return };
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
     let (server, app_id) = server_for(pool.clone(), "vt").await;
 
     let script_id = create_script(&server, &app_id, "slow", ACK_HANDLER).await;
@@ -317,13 +322,12 @@ async fn queue_visibility_timeout_reclaim() {
     // Otherwise the reclaim path is the only way the message would
     // have been picked up. Poll the claim state.
     for _ in 0..50 {
-        let row: Option<(Option<Uuid>,)> = sqlx::query_as(
+        let row: Option<(Option<Uuid>,)> =
-            "SELECT claim_token FROM queue_messages WHERE id = $1",
+            sqlx::query_as("SELECT claim_token FROM queue_messages WHERE id = $1")
-        )
+                .bind(msg_id)
-        .bind(msg_id)
+                .fetch_optional(&pool)
-        .fetch_optional(&pool)
+                .await
-        .await
+                .expect("read");
-        .expect("read");
         if let Some((token,)) = row {
             if token.is_none() {
                 return; // reclaim succeeded — claim cleared

crates/picloud/tests/retry_e2e.rs

@@ -93,7 +93,9 @@ async fn execute(server: &TestServer, script_id: &str) -> Value {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn retry_run_eventually_succeeds_inside_http_handler() {
-    let Some(pool) = pool_or_skip().await else { return };
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
     let (server, app_id) = server_for(pool, "retry-succ").await;
 
     // attempt counter mutates across retries; the 3rd attempt returns
@@ -114,7 +116,9 @@ async fn retry_run_eventually_succeeds_inside_http_handler() {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn retry_run_surfaces_last_error_after_max_attempts() {
-    let Some(pool) = pool_or_skip().await else { return };
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
     let (server, app_id) = server_for(pool, "retry-surf").await;
 
     let src = r#"
@@ -135,7 +139,9 @@ async fn retry_run_surfaces_last_error_after_max_attempts() {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn retry_on_codes_filters_unmatched_errors() {
-    let Some(pool) = pool_or_skip().await else { return };
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
     let (server, app_id) = server_for(pool, "retry-codes").await;
 
     // attempts is mutated on every entry; we throw a non-matching code
@@ -156,5 +162,9 @@ async fn retry_on_codes_filters_unmatched_errors() {
     "#;
     let script_id = create_script(&server, &app_id, "codes", src).await;
     let body = execute(&server, &script_id).await;
-    assert_eq!(body, json!(1), "non-matching error should surface immediately");
+    assert_eq!(
+        body,
+        json!(1),
+        "non-matching error should surface immediately"
+    );
 }