fix(manager-core): F-P-007 concurrent queue dispatch per tick

tick_queue_arm called list_active_queue_consumers() and then iterated
serially, awaiting one queue.claim(app, queue) per consumer. With N
consumers and a 100ms tick the worst-case throughput was
N × (claim + executor) / tick — one slow handler blocked every other
queue's progress on the dispatcher's task.

Replace the for-loop with `futures::stream::iter(consumers)
.for_each_concurrent(QUEUE_DISPATCH_PARALLELISM, …)`. The execution
gate already caps real script-concurrency to its permits, so this just
removes the dispatcher-side serialization.

QUEUE_DISPATCH_PARALLELISM = 32 (matches the default gate). Workspace
gains `futures = 0.3` so `for_each_concurrent` is available.

AUDIT.md anchor: F-P-007.

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

Cargo.lock

@@ -830,6 +830,21 @@ dependencies = [
  "percent-encoding",
 ]
 
+[[package]]
+name = "futures"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8b147ee9d1f6d097cef9ce628cd2ee62288d963e16fb287bd9286455b241382d"
+dependencies = [
+ "futures-channel",
+ "futures-core",
+ "futures-executor",
+ "futures-io",
+ "futures-sink",
+ "futures-task",
+ "futures-util",
+]
+
 [[package]]
 name = "futures-channel"
 version = "0.3.32"
@@ -874,6 +889,17 @@ version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718"
 
+[[package]]
+name = "futures-macro"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "futures-sink"
 version = "0.3.32"
@@ -892,8 +918,10 @@ version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6"
 dependencies = [
+ "futures-channel",
  "futures-core",
  "futures-io",
+ "futures-macro",
  "futures-sink",
  "futures-task",
  "memchr",
@@ -1862,6 +1890,7 @@ dependencies = [
  "chrono-tz",
  "cron",
  "data-encoding",
+ "futures",
  "hex",
  "hmac",
  "lettre",

Cargo.toml

@@ -53,8 +53,10 @@ chrono = { version = "0.4", features = ["serde"] }
 chrono-tz = "0.9"
 cron = "0.12"
 
-# Async traits
+# Async traits + bounded-concurrency stream utilities (queue dispatcher
+# uses `for_each_concurrent`).
 async-trait = "0.1"
+futures = "0.3"
 
 # Rhai scripting. Pinned exactly (`=1.24`) because the `internals`
 # feature surface is not semver-stable — future bumps must be deliberate.

crates/manager-core/Cargo.toml

@@ -14,6 +14,7 @@ picloud-executor-core.workspace = true
 picloud-orchestrator-core.workspace = true
 
 async-trait.workspace = true
+futures.workspace = true
 axum.workspace = true
 rand.workspace = true
 serde.workspace = true

crates/manager-core/src/dispatcher.rs

@@ -79,6 +79,11 @@ const TICK_INTERVAL: Duration = Duration::from_millis(100);
 /// 5-minute cap.
 const ASYNC_EXEC_TIMEOUT: Duration = Duration::from_secs(300);
 
+/// F-P-007: per-tick concurrency cap for queue dispatches. Matches the
+/// default execution-gate concurrency so the gate is what bounds real
+/// parallelism, not the dispatcher's serial-await.
+const QUEUE_DISPATCH_PARALLELISM: usize = 32;
+
 impl Dispatcher {
     /// Spawn the dispatcher loop as a detached `tokio::task`. Also
     /// spawns the v1.1.9 queue visibility-timeout reclaim task. Both
@@ -149,17 +154,30 @@ impl Dispatcher {
     /// attempt one claim per `(app_id, queue_name)`. Each successful
     /// claim runs through the same executor path as outbox-backed
     /// trigger dispatches, so retry / dead-letter wiring is shared.
+    ///
+    /// F-P-007: bound concurrency with `for_each_concurrent` instead of
+    /// serial-await. With N consumers and a 100ms tick the worst-case
+    /// throughput was N × (claim + executor) / tick; concurrent dispatch
+    /// lets one slow handler not block the others. The execution gate
+    /// still caps real parallelism — this just removes the
+    /// dispatcher-side serialization.
     async fn tick_queue_arm(&self) -> Result<(), DispatcherError> {
+        use futures::stream::{self, StreamExt};
         let consumers = self
             .triggers
             .list_active_queue_consumers()
             .await
             .map_err(|e| DispatcherError::ResolveTrigger(e.to_string()))?;
-        for consumer in consumers {
-            if let Err(err) = self.dispatch_one_queue(&consumer).await {
-                tracing::warn!(?err, queue = %consumer.queue_name, "queue dispatch errored");
-            }
-        }
+        // Concurrency cap matches the execution gate's typical default;
+        // overflow attempts return Err(Overloaded) and back off so a
+        // queue burst doesn't hammer the executor.
+        stream::iter(consumers)
+            .for_each_concurrent(QUEUE_DISPATCH_PARALLELISM, |consumer| async move {
+                if let Err(err) = self.dispatch_one_queue(&consumer).await {
+                    tracing::warn!(?err, queue = %consumer.queue_name, "queue dispatch errored");
+                }
+            })
+            .await;
         Ok(())
     }