feat(v1.1.9): dispatcher Invoke arm — invoke_async runs once

Replaces the commit 2 placeholder with the real OutboxSourceKind::Invoke
handler.

build_invoke_request(row) hydrates an (ResolvedTrigger, ExecRequest)
pair from a payload InvokeService::enqueue_async wrote. Validates:
  - script_id present + parses as UUID
  - script exists
  - script.app_id == row.app_id (defensive — re-check the cross-app
    boundary; same-app was already verified at enqueue time)

Synthesized ResolvedTrigger carries retry_max_attempts = 1 so the
existing handle_failure path skips the retry loop. invoke_async runs
exactly once; on throw the row is deleted + a dead_letters row written
(handle_failure already does this when attempt >= max_attempts), so a
caller who wants retry semantics wraps the call site in retry::with.

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

crates/manager-core/src/dispatcher.rs

@@ -372,22 +372,18 @@ impl Dispatcher {
                     return Ok(());
                 }
             },
-            OutboxSourceKind::Invoke => {
+            OutboxSourceKind::Invoke => match self.build_invoke_request(&row).await {
-                // Wired in commit 10 (invoke_async() — function composition).
+                Ok(pair) => pair,
-                // Until then a malformed row is impossible to write (no
+                Err(err) => {
-                // producer), but defensively drop the row so a manual
+                    tracing::warn!(outbox_id = %row.id, ?err, "invoke exec build failed; dropping");
-                // INSERT doesn't loop the dispatcher.
+                    self.outbox
-                tracing::warn!(
+                        .delete(row.id)
-                    outbox_id = %row.id,
+                        .await
-                    "OutboxSourceKind::Invoke not wired yet; dropping row"
+                        .map_err(|e| DispatcherError::Outbox(e.to_string()))?;
-                );
+                    drop(permit);
-                self.outbox
+                    return Ok(());
-                    .delete(row.id)
+                }
-                    .await
+            },
-                    .map_err(|e| DispatcherError::Outbox(e.to_string()))?;
-                drop(permit);
-                return Ok(());
-            }
             OutboxSourceKind::Kv
             | OutboxSourceKind::Docs
             | OutboxSourceKind::DeadLetter
@@ -587,6 +583,94 @@ impl Dispatcher {
         Ok((resolved, req))
     }
 
+    /// v1.1.9. Build an `(ResolvedTrigger, ExecRequest)` from an
+    /// invoke_async outbox row. The payload was written by
+    /// `InvokeService::enqueue_async`. No retry policy — invoke_async
+    /// runs once; on failure the row is deleted and a dead_letters row
+    /// is written (so the misfire is observable).
+    async fn build_invoke_request(
+        &self,
+        row: &OutboxRow,
+    ) -> Result<(ResolvedTrigger, ExecRequest), DispatcherError> {
+        // Extract the args + script_id from the payload (InvokeService
+        // populates these). Defensive: dead_letter the row if any
+        // required field is missing.
+        let payload = &row.payload;
+        let Some(script_id_str) = payload.get("script_id").and_then(|v| v.as_str()) else {
+            return Err(DispatcherError::ResolveTrigger(
+                "invoke row missing script_id".into(),
+            ));
+        };
+        let script_id = picloud_shared::ScriptId::from(
+            uuid::Uuid::parse_str(script_id_str)
+                .map_err(|e| DispatcherError::ResolveTrigger(format!("script_id: {e}")))?,
+        );
+        let script = self
+            .scripts
+            .get(script_id)
+            .await
+            .map_err(|e| DispatcherError::ResolveTrigger(e.to_string()))?
+            .ok_or_else(|| {
+                DispatcherError::ResolveTrigger(format!("script {script_id} not found"))
+            })?;
+        // Same-app guard — the script could have been re-assigned or
+        // the row could have been hand-edited. Reject mismatches.
+        if script.app_id != row.app_id {
+            return Err(DispatcherError::ResolveTrigger(
+                "invoke target belongs to a different app".into(),
+            ));
+        }
+
+        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))
+            .unwrap_or(row.trigger_depth);
+
+        let execution_id = ExecutionId::new();
+        let req = ExecRequest {
+            execution_id,
+            request_id: RequestId::new(),
+            script_id: script.id,
+            script_name: script.name.clone(),
+            invocation_type: InvocationType::Function,
+            path: "/invoke_async".into(),
+            headers: std::collections::BTreeMap::new(),
+            body: args,
+            params: std::collections::BTreeMap::new(),
+            query: std::collections::BTreeMap::new(),
+            rest: String::new(),
+            sandbox_overrides: script.sandbox,
+            app_id: row.app_id,
+            // invoke_async runs with no principal — same as HTTP outbox.
+            principal: None,
+            trigger_depth,
+            root_execution_id: row.root_execution_id.unwrap_or(execution_id),
+            is_dead_letter_handler: false,
+            event: None,
+        };
+
+        // Build a synthetic ResolvedTrigger so the dispatch_one outer
+        // loop can carry on through the same code path. Retry knobs
+        // bound to 1 attempt: invoke_async runs ONCE (callers wrap in
+        // retry::with for retries).
+        let resolved = ResolvedTrigger {
+            trigger_kind: TriggerKind::Cron, // placeholder; not used downstream
+            is_dead_letter_handler: false,
+            script_id: script.id,
+            script_source: script.source,
+            script_name: script.name,
+            script_updated_at: script.updated_at,
+            sandbox_overrides: script.sandbox,
+            registered_by_principal: picloud_shared::AdminUserId::from(uuid::Uuid::nil()),
+            retry_max_attempts: 1,
+            retry_backoff: BackoffShape::Constant,
+            retry_base_ms: 0,
+        };
+        Ok((resolved, req))
+    }
+
     async fn handle_success(
         &self,
         row: &OutboxRow,