fix(executor): cap per-execution durable fan-out width

`trigger_depth` bounds how DEEP a trigger/invoke chain runs, but nothing bounded
how WIDE one execution could fan out. A single anonymous request running
`for i in 0..1_000_000 { invoke_async("w", #{}) }` costs a few Rhai ops plus one
cheap outbox INSERT per iteration, so within the op / wall-clock budget it could
write ~10^5 durable rows — each dispatched as its own execution — flooding the
outbox and dispatcher (a durable-amplification DoS).

Add a per-execution ceiling on DURABLE emissions (`invoke_async`,
`pubsub::publish_durable`, `queue::enqueue`, and the shared-topic/-queue
variants), env `PICLOUD_MAX_EMISSIONS_PER_EXECUTION` (default 1000). It's a
thread-local counter wrapped by a re-entrancy-aware `EmissionBudgetScope` around
every `execute_ast`: the OUTERMOST scope resets it, so a fresh dispatched
handler (a new pooled-thread task) gets a full budget while a synchronous
`invoke()` / interceptor re-entry nested in the same call stack SHARES it (fan-
out counted across the whole synchronous chain). The outermost Drop re-zeroes
the counter so a pooled thread never leaks a count into the next task.

Pinned by an `emit_budget` unit test (outermost resets, nested shares, ceiling
trips); the invoke/queue/pubsub/workflow journeys (small counts) stay green.

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

crates/executor-core/src/engine.rs

@@ -280,6 +280,13 @@ impl Engine {
     /// cache hands compiled ASTs in directly; this path skips the
     /// per-call compile.
     pub fn execute_ast(&self, ast: &Arc<AST>, req: ExecRequest) -> Result<ExecResponse, ExecError> {
+        // Audit #2: bound per-execution durable fan-out. The scope is
+        // re-entrancy aware — a fresh dispatched execution (a new task on a
+        // pooled thread) resets the budget, while a synchronous invoke() /
+        // interceptor re-entry nested in the SAME call stack shares it, so
+        // fan-out is counted across the whole synchronous chain. Held to the end
+        // of the call; its Drop re-zeroes the counter on the outermost exit.
+        let _emit_scope = crate::sdk::emit_budget::EmissionBudgetScope::enter();
         let effective_limits = self.limits.with_overrides(&req.sandbox_overrides);
         let logs: Arc<Mutex<Vec<LogEntry>>> = Arc::new(Mutex::new(Vec::new()));
         let mut engine = build_engine(effective_limits, Some(logs.clone()));

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

@@ -0,0 +1,124 @@
+//! Per-execution fan-out ceiling (audit fix #2).
+//!
+//! `trigger_depth` bounds how DEEP a trigger/invoke chain can go, but nothing
+//! bounded how WIDE a single execution could fan out: one anonymous request
+//! running `for i in 0..1_000_000 { invoke_async("w", #{}) }` costs a few Rhai
+//! ops per iteration plus one cheap outbox INSERT, so within the op / wall-clock
+//! budget it could write ~10^5 durable rows — each dispatched as its own
+//! execution — flooding the outbox/queue (a durable amplification DoS).
+//!
+//! This caps the number of DURABLE emissions (`invoke_async`, `publish_durable`,
+//! `enqueue`, and their group-shared variants) a single execution may make.
+//! The counter is a thread-local. [`EmissionBudgetScope`] is created around
+//! EVERY `execute_ast` and is **re-entrancy aware**: only the OUTERMOST scope
+//! on a thread resets the counter. A dispatched handler (queue/trigger/
+//! `invoke_async`/cron/workflow) is a fresh task on a pooled thread, so its
+//! scope is outermost → it resets and gets a full budget. A synchronous
+//! `invoke()` / interceptor re-entry runs a nested `execute_ast` in the SAME
+//! call stack, so its scope is inner → it does NOT reset, and the whole
+//! synchronous chain shares one budget (fan-out counted across it). The
+//! outermost Drop re-zeroes the counter so a pooled thread never leaks a count
+//! into the next task.
+
+use std::cell::Cell;
+use std::sync::LazyLock;
+
+use rhai::EvalAltResult;
+
+use crate::sdk::bridge::runtime_err;
+
+/// Default max durable emissions per execution. Generous for a legit batch
+/// producer, far below the ~10^5 a runaway loop can reach within budget.
+const DEFAULT_MAX_EMISSIONS: u32 = 1000;
+
+static MAX_EMISSIONS: LazyLock<u32> = LazyLock::new(|| {
+    std::env::var("PICLOUD_MAX_EMISSIONS_PER_EXECUTION")
+        .ok()
+        .and_then(|v| v.parse::<u32>().ok())
+        .filter(|&n| n > 0)
+        .unwrap_or(DEFAULT_MAX_EMISSIONS)
+});
+
+thread_local! {
+    static EMISSIONS: Cell<u32> = const { Cell::new(0) };
+    /// Scope nesting depth on this thread. `0` when no execution is active.
+    static ACTIVE: Cell<u32> = const { Cell::new(0) };
+}
+
+/// Charge one durable emission against the current execution's budget. Returns
+/// an `Err` (aborting the emit) once the ceiling is exceeded.
+///
+/// # Errors
+/// Returns a Rhai runtime error when the per-execution emission ceiling is hit.
+pub(super) fn charge_emission(service: &str) -> Result<(), Box<EvalAltResult>> {
+    EMISSIONS.with(|c| {
+        let next = c.get().saturating_add(1);
+        if next > *MAX_EMISSIONS {
+            return Err(runtime_err(&format!(
+                "{service}: per-execution emission limit exceeded (max {} durable \
+                 invoke_async/publish/enqueue calls per execution)",
+                *MAX_EMISSIONS
+            )));
+        }
+        c.set(next);
+        Ok(())
+    })
+}
+
+/// RAII scope wrapping one `execute_ast`. Re-entrancy aware: the OUTERMOST scope
+/// on a thread zeroes the counter on entry and on final exit; nested scopes (a
+/// synchronous re-entry in the same call stack) leave it alone, so the chain
+/// shares one budget.
+pub(crate) struct EmissionBudgetScope;
+
+impl EmissionBudgetScope {
+    pub(crate) fn enter() -> Self {
+        ACTIVE.with(|a| {
+            if a.get() == 0 {
+                EMISSIONS.with(|c| c.set(0));
+            }
+            a.set(a.get() + 1);
+        });
+        Self
+    }
+}
+
+impl Drop for EmissionBudgetScope {
+    fn drop(&mut self) {
+        ACTIVE.with(|a| {
+            let n = a.get().saturating_sub(1);
+            a.set(n);
+            if n == 0 {
+                EMISSIONS.with(|c| c.set(0));
+            }
+        });
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn outermost_scope_resets_but_nested_shares_budget() {
+        let outer = EmissionBudgetScope::enter();
+        for _ in 0..DEFAULT_MAX_EMISSIONS {
+            charge_emission("test").expect("under budget");
+        }
+        // A NESTED scope (synchronous re-entry) must NOT reset — the budget is
+        // already exhausted and stays that way.
+        {
+            let _nested = EmissionBudgetScope::enter();
+            assert!(
+                charge_emission("test").is_err(),
+                "a nested re-entry shares the exhausted budget"
+            );
+        }
+        // Still exhausted after the nested scope drops (outer still active).
+        assert!(charge_emission("test").is_err());
+        drop(outer);
+        // A fresh OUTERMOST scope resets.
+        let _fresh = EmissionBudgetScope::enter();
+        charge_emission("test").expect("reset by a fresh outermost scope");
+    }
+}

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

@@ -85,6 +85,9 @@ pub(super) fn register(
         module.set_native_fn(
             "invoke_async",
             move |target: Dynamic, args: Dynamic| -> Result<String, Box<EvalAltResult>> {
+                // Audit #2: count this durable emission against the per-execution
+                // fan-out ceiling before doing any work.
+                crate::sdk::emit_budget::charge_emission("invoke_async")?;
                 let target = parse_target(target)?;
                 let args_json = args_to_json(&args)?;
                 let svc = svc.clone();

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

@@ -16,6 +16,7 @@ pub mod cx;
 pub mod dead_letters;
 pub mod docs;
 pub mod email;
+pub mod emit_budget;
 pub mod files;
 pub mod http;
 pub mod interceptor;

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

@@ -35,6 +35,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
         module.set_native_fn(
             "publish_durable",
             move |topic: &str, message: Dynamic| -> Result<(), Box<EvalAltResult>> {
+                crate::sdk::emit_budget::charge_emission("pubsub::publish_durable")?;
                 let json = message_to_json(&message)?;
                 let svc = svc.clone();
                 let cx = cx.clone();
@@ -110,6 +111,7 @@ fn shared_publish(
     subtopic: &str,
     message: Dynamic,
 ) -> Result<(), Box<EvalAltResult>> {
+    crate::sdk::emit_budget::charge_emission("pubsub::shared_topic")?;
     let json = message_to_json(&message)?;
     let service = h.service.clone();
     let cx = h.cx.clone();

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

@@ -40,6 +40,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
         module.set_native_fn(
             "enqueue",
             move |name: &str, message: Dynamic| -> Result<(), Box<EvalAltResult>> {
+                crate::sdk::emit_budget::charge_emission("queue::enqueue")?;
                 let json = message_to_json(&message)?;
                 enqueue_blocking(&svc, &cx, name, json, EnqueueOpts::default())
             },
@@ -54,6 +55,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
         module.set_native_fn(
             "enqueue",
             move |name: &str, message: Dynamic, opts: Map| -> Result<(), Box<EvalAltResult>> {
+                crate::sdk::emit_budget::charge_emission("queue::enqueue")?;
                 let json = message_to_json(&message)?;
                 let opts = parse_opts(&opts)?;
                 enqueue_blocking(&svc, &cx, name, json, opts)
@@ -131,6 +133,7 @@ fn group_enqueue(
     message: Dynamic,
     opts: EnqueueOpts,
 ) -> Result<(), Box<EvalAltResult>> {
+    crate::sdk::emit_budget::charge_emission("queue::shared_collection")?;
     let json = message_to_json(&message)?;
     let handle = TokioHandle::try_current().map_err(|e| -> Box<EvalAltResult> {
         EvalAltResult::ErrorRuntime(