feat(v1.1.9): InvokeServiceImpl + RouteTable wiring in picloud binary

manager-core/invoke_service.rs implements InvokeService over:
  - ScriptRepository (for Id + Name resolution)
  - RouteTable (for Path resolution via the orchestrator's matcher)
  - OutboxRepo (for invoke_async)

Same-app guard runs at the service entry point: resolve_id() always
verifies resolved.app_id == cx.app_id and returns InvokeError::CrossApp
otherwise. resolve_name() reads cx.app_id when querying (so a wrong
app_id from somewhere else couldn't slip through), but defense-in-depth
won't hurt — future hardening if it surfaces.

resolve_path() runs the matcher against this app's routes only
(RouteTable::snapshot_for_app). Method assumed POST→GET fallback for
v1.1.9 — surfacing method via the SDK is a future addition.

enqueue_async() resolves the target, then writes one outbox row with
source_kind = 'invoke'. The payload carries script_id, script_name,
args, fresh execution_id, root_execution_id (inherited from caller),
trigger_depth + 1. caller principal recorded as origin_principal
(forensic only — the executing script runs with no principal). One
shot — no retry policy on the row; users wrap in retry::with() if
they want retries.

picloud/lib.rs:
  - route_table construction moved BEFORE Services::new so the invoke
    service can hold it. populates from route_repo as before.
  - InvokeServiceImpl wired into Services::new in place of the
    NoopInvokeService placeholder.

Unit tests cover: resolve by Id same-app, cross-app rejected, resolve
by Name finds + not-found, enqueue_async writes Invoke outbox row with
trigger_depth=1.

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

crates/manager-core/src/invoke_service.rs

@@ -0,0 +1,412 @@
+//! `InvokeServiceImpl` — wires `ScriptRepository` + `RouteTable` +
+//! `OutboxRepo` underneath the `picloud_shared::InvokeService` trait.
+//!
+//! Same-app only. The cross-app guard lives at the service entry point:
+//! every resolved script's `app_id` must match `cx.app_id` — otherwise
+//! `InvokeError::CrossApp` (v1.1.x maintains strict isolation).
+//!
+//! `invoke_async` writes a v1.1.9 `OutboxSourceKind::Invoke` row the
+//! dispatcher consumes (commit 10). Runs once per row — no retries.
+//! Callers wrap in `retry::with(...)` if they want retries.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use picloud_orchestrator_core::routing::{matcher, RouteTable};
+use picloud_shared::{
+    ExecutionId, InvokeError, InvokeService, InvokeTarget, ResolvedScript, ScriptId, SdkCallCx,
+};
+
+use crate::outbox_repo::{NewOutboxRow, OutboxRepo, OutboxSourceKind};
+use crate::repo::ScriptRepository;
+
+pub struct InvokeServiceImpl {
+    scripts: Arc<dyn ScriptRepository>,
+    routes: Arc<RouteTable>,
+    outbox: Arc<dyn OutboxRepo>,
+}
+
+impl InvokeServiceImpl {
+    #[must_use]
+    pub fn new(
+        scripts: Arc<dyn ScriptRepository>,
+        routes: Arc<RouteTable>,
+        outbox: Arc<dyn OutboxRepo>,
+    ) -> Self {
+        Self {
+            scripts,
+            routes,
+            outbox,
+        }
+    }
+
+    async fn resolve_id(
+        &self,
+        cx: &SdkCallCx,
+        script_id: ScriptId,
+    ) -> Result<ResolvedScript, InvokeError> {
+        let script = self
+            .scripts
+            .get(script_id)
+            .await
+            .map_err(|e| InvokeError::Unavailable(e.to_string()))?
+            .ok_or_else(|| InvokeError::NotFound(format!("id {script_id}")))?;
+        if script.app_id != cx.app_id {
+            return Err(InvokeError::CrossApp);
+        }
+        Ok(ResolvedScript {
+            script_id: script.id,
+            app_id: script.app_id,
+            source: script.source,
+            updated_at: script.updated_at,
+            name: script.name,
+        })
+    }
+
+    async fn resolve_name(
+        &self,
+        cx: &SdkCallCx,
+        name: &str,
+    ) -> Result<ResolvedScript, InvokeError> {
+        let script = self
+            .scripts
+            .get_by_name(cx.app_id, name)
+            .await
+            .map_err(|e| InvokeError::Unavailable(e.to_string()))?
+            .ok_or_else(|| InvokeError::NotFound(format!("name {name:?}")))?;
+        Ok(ResolvedScript {
+            script_id: script.id,
+            app_id: script.app_id,
+            source: script.source,
+            updated_at: script.updated_at,
+            name: script.name,
+        })
+    }
+
+    async fn resolve_path(
+        &self,
+        cx: &SdkCallCx,
+        path: &str,
+    ) -> Result<ResolvedScript, InvokeError> {
+        // Run the matcher against this app's routes only. Host bucket
+        // is irrelevant for in-app composition — use a synthetic host
+        // that matches the wildcard tier. Method assumed POST since
+        // 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(|| {
+            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
+        // partitions by app already, but resolve_id verifies again).
+        self.resolve_id(cx, m.matched.script_id).await
+    }
+}
+
+#[async_trait]
+impl InvokeService for InvokeServiceImpl {
+    async fn resolve(
+        &self,
+        cx: &SdkCallCx,
+        target: InvokeTarget,
+    ) -> Result<ResolvedScript, InvokeError> {
+        match target {
+            InvokeTarget::Id(id) => self.resolve_id(cx, id).await,
+            InvokeTarget::Name(n) => self.resolve_name(cx, &n).await,
+            InvokeTarget::Path(p) => self.resolve_path(cx, &p).await,
+        }
+    }
+
+    async fn enqueue_async(
+        &self,
+        cx: &SdkCallCx,
+        target: InvokeTarget,
+        args: serde_json::Value,
+    ) -> Result<ExecutionId, InvokeError> {
+        let resolved = self.resolve(cx, target).await?;
+        let execution_id = ExecutionId::new();
+        // Payload carries everything the dispatcher's Invoke arm needs
+        // to build an ExecRequest at fire time.
+        let payload = serde_json::json!({
+            "kind": "invoke",
+            "script_id": resolved.script_id,
+            "script_name": resolved.name,
+            "args": args,
+            "execution_id": execution_id,
+            "caller_request_id": cx.request_id,
+            "root_execution_id": cx.root_execution_id,
+            "trigger_depth": cx.trigger_depth.saturating_add(1),
+        });
+        // Caller's principal is recorded as origin_principal (forensic),
+        // but the executing script runs with no principal — invoke_async
+        // is the fire-and-forget equivalent of a function call, not a
+        // re-authentication.
+        let origin_principal = cx.principal.as_ref().map(|p| p.user_id);
+        self.outbox
+            .insert(NewOutboxRow {
+                app_id: cx.app_id,
+                source_kind: OutboxSourceKind::Invoke,
+                trigger_id: None,
+                script_id: Some(resolved.script_id),
+                reply_to: None,
+                payload,
+                origin_principal,
+                trigger_depth: cx.trigger_depth.saturating_add(1),
+                root_execution_id: Some(cx.root_execution_id),
+            })
+            .await
+            .map_err(|e| InvokeError::Unavailable(e.to_string()))?;
+        Ok(execution_id)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::outbox_repo::{OutboxRepoError, OutboxRow};
+    use crate::repo::{NewScript, ScriptPatch, ScriptRepositoryError};
+    use chrono::Utc;
+    use picloud_shared::{AdminUserId, AppId, ExecutionId, RequestId, Script, ScriptSandbox};
+
+    struct OneScriptRepo {
+        script: Script,
+    }
+    #[async_trait]
+    impl ScriptRepository for OneScriptRepo {
+        async fn get(
+            &self,
+            id: ScriptId,
+        ) -> Result<Option<Script>, ScriptRepositoryError> {
+            Ok(if id == self.script.id {
+                Some(self.script.clone())
+            } else {
+                None
+            })
+        }
+        async fn get_by_name(
+            &self,
+            app_id: AppId,
+            name: &str,
+        ) -> Result<Option<Script>, ScriptRepositoryError> {
+            Ok(if app_id == self.script.app_id && name == self.script.name {
+                Some(self.script.clone())
+            } else {
+                None
+            })
+        }
+        async fn list(&self) -> Result<Vec<Script>, ScriptRepositoryError> {
+            Ok(vec![self.script.clone()])
+        }
+        async fn list_for_app(
+            &self,
+            _app_id: AppId,
+        ) -> Result<Vec<Script>, ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn list_for_user(
+            &self,
+            _user_id: AdminUserId,
+        ) -> Result<Vec<Script>, ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn create(&self, _input: NewScript) -> Result<Script, ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn update(
+            &self,
+            _id: ScriptId,
+            _patch: ScriptPatch,
+        ) -> Result<Script, ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn delete(&self, _id: ScriptId) -> Result<(), ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn count_routes_for_script(
+            &self,
+            _script_id: ScriptId,
+        ) -> Result<i64, ScriptRepositoryError> {
+            Ok(0)
+        }
+        async fn count_triggers_for_script(
+            &self,
+            _script_id: ScriptId,
+        ) -> Result<i64, ScriptRepositoryError> {
+            Ok(0)
+        }
+        async fn list_imports(
+            &self,
+            _script_id: ScriptId,
+        ) -> Result<Vec<Script>, ScriptRepositoryError> {
+            Ok(vec![])
+        }
+    }
+
+    struct CapturingOutbox {
+        last: tokio::sync::Mutex<Option<NewOutboxRow>>,
+    }
+    #[async_trait]
+    impl OutboxRepo for CapturingOutbox {
+        async fn insert(&self, row: NewOutboxRow) -> Result<uuid::Uuid, OutboxRepoError> {
+            let id = uuid::Uuid::new_v4();
+            *self.last.lock().await = Some(row);
+            Ok(id)
+        }
+        async fn claim_due(
+            &self,
+            _claimed_by: &str,
+            _limit: i64,
+        ) -> Result<Vec<OutboxRow>, OutboxRepoError> {
+            Ok(vec![])
+        }
+        async fn delete(&self, _id: uuid::Uuid) -> Result<(), OutboxRepoError> {
+            Ok(())
+        }
+        async fn reschedule(
+            &self,
+            _id: uuid::Uuid,
+            _attempt_count: u32,
+            _next_attempt_at: chrono::DateTime<Utc>,
+        ) -> Result<(), OutboxRepoError> {
+            Ok(())
+        }
+    }
+
+    fn make_script(app_id: AppId, name: &str) -> Script {
+        Script {
+            id: ScriptId::new(),
+            app_id,
+            name: name.into(),
+            description: None,
+            version: 1,
+            source: "0".into(),
+            kind: picloud_shared::ScriptKind::Endpoint,
+            timeout_seconds: 30,
+            memory_limit_mb: 64,
+            sandbox: ScriptSandbox::default(),
+            created_at: Utc::now(),
+            updated_at: Utc::now(),
+        }
+    }
+
+    fn anon_cx(app_id: AppId) -> SdkCallCx {
+        let execution_id = ExecutionId::new();
+        SdkCallCx {
+            app_id,
+            script_id: ScriptId::new(),
+            principal: None,
+            execution_id,
+            request_id: RequestId::new(),
+            trigger_depth: 0,
+            root_execution_id: execution_id,
+            is_dead_letter_handler: false,
+            event: None,
+        }
+    }
+
+    #[tokio::test]
+    async fn resolve_by_id_same_app_succeeds() {
+        let app_id = AppId::new();
+        let script = make_script(app_id, "worker");
+        let svc = InvokeServiceImpl::new(
+            Arc::new(OneScriptRepo {
+                script: script.clone(),
+            }),
+            Arc::new(RouteTable::new()),
+            Arc::new(CapturingOutbox {
+                last: tokio::sync::Mutex::new(None),
+            }),
+        );
+        let cx = anon_cx(app_id);
+        let resolved = svc.resolve(&cx, InvokeTarget::Id(script.id)).await.unwrap();
+        assert_eq!(resolved.script_id, script.id);
+        assert_eq!(resolved.app_id, app_id);
+    }
+
+    #[tokio::test]
+    async fn resolve_by_id_cross_app_rejects() {
+        let app_a = AppId::new();
+        let app_b = AppId::new();
+        let script = make_script(app_a, "worker");
+        let svc = InvokeServiceImpl::new(
+            Arc::new(OneScriptRepo { script: script.clone() }),
+            Arc::new(RouteTable::new()),
+            Arc::new(CapturingOutbox {
+                last: tokio::sync::Mutex::new(None),
+            }),
+        );
+        // 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();
+        assert!(matches!(err, InvokeError::CrossApp));
+    }
+
+    #[tokio::test]
+    async fn resolve_by_name_finds_script_in_caller_app() {
+        let app_id = AppId::new();
+        let script = make_script(app_id, "worker");
+        let svc = InvokeServiceImpl::new(
+            Arc::new(OneScriptRepo { script: script.clone() }),
+            Arc::new(RouteTable::new()),
+            Arc::new(CapturingOutbox {
+                last: tokio::sync::Mutex::new(None),
+            }),
+        );
+        let cx = anon_cx(app_id);
+        let resolved = svc
+            .resolve(&cx, InvokeTarget::Name("worker".into()))
+            .await
+            .unwrap();
+        assert_eq!(resolved.script_id, script.id);
+    }
+
+    #[tokio::test]
+    async fn resolve_by_name_unknown_is_not_found() {
+        let app_id = AppId::new();
+        let script = make_script(app_id, "worker");
+        let svc = InvokeServiceImpl::new(
+            Arc::new(OneScriptRepo { script }),
+            Arc::new(RouteTable::new()),
+            Arc::new(CapturingOutbox {
+                last: tokio::sync::Mutex::new(None),
+            }),
+        );
+        let cx = anon_cx(app_id);
+        let err = svc
+            .resolve(&cx, InvokeTarget::Name("ghost".into()))
+            .await
+            .unwrap_err();
+        assert!(matches!(err, InvokeError::NotFound(_)));
+    }
+
+    #[tokio::test]
+    async fn enqueue_async_writes_outbox_row_with_invoke_kind() {
+        let app_id = AppId::new();
+        let script = make_script(app_id, "worker");
+        let outbox = Arc::new(CapturingOutbox {
+            last: tokio::sync::Mutex::new(None),
+        });
+        let svc = InvokeServiceImpl::new(
+            Arc::new(OneScriptRepo { script: script.clone() }),
+            Arc::new(RouteTable::new()),
+            outbox.clone(),
+        );
+        let cx = anon_cx(app_id);
+        let exec_id = svc
+            .enqueue_async(
+                &cx,
+                InvokeTarget::Id(script.id),
+                serde_json::json!({ "x": 1 }),
+            )
+            .await
+            .unwrap();
+        let row = outbox.last.lock().await.clone().unwrap();
+        assert_eq!(row.app_id, app_id);
+        assert_eq!(row.source_kind, OutboxSourceKind::Invoke);
+        assert_eq!(row.script_id, Some(script.id));
+        assert_eq!(row.trigger_depth, 1);
+        let _ = exec_id;
+    }
+}

crates/manager-core/src/lib.rs

@@ -55,6 +55,7 @@ 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 realtime_authority;

crates/picloud/src/lib.rs

@@ -274,11 +274,27 @@ pub async fn build_app(
             authz.clone(),
         ),
     );
-    // v1.1.9 function composition. Real InvokeService wired in commit 8;
-    // the noop here keeps the binary compiling — without it, scripts
-    // calling `invoke()` will throw a clear error.
-    let invoke: Arc<dyn picloud_shared::InvokeService> =
-        Arc::new(picloud_shared::NoopInvokeService);
+    // 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
+    // routes.
+    let route_table = Arc::new(RouteTable::new());
+    let initial = route_repo.list_all().await?;
+    let compiled = compile_routes(&initial)
+        .map_err(|e| anyhow::anyhow!("failed to compile stored routes: {e}"))?;
+    route_table.replace_all(compiled);
+
+    // v1.1.9 function composition. InvokeServiceImpl resolves targets
+    // by Id, Name (via get_by_name), or Path (via the orchestrator's
+    // RouteTable). invoke_async writes an OutboxSourceKind::Invoke row
+    // that the dispatcher fires through the standard executor path.
+    let invoke: Arc<dyn picloud_shared::InvokeService> = Arc::new(
+        picloud_manager_core::invoke_service::InvokeServiceImpl::new(
+            Arc::new(PostgresScriptRepoHandle(script_repo.clone())),
+            route_table.clone(),
+            outbox_repo.clone(),
+        ),
+    );
     let services = Services::new(
         kv,
         docs,
@@ -296,13 +312,6 @@ pub async fn build_app(
     );
     let engine = Arc::new(Engine::new(Limits::default(), services));
 
-    // Compile the routes table once at startup; admin writes refresh it.
-    let route_table = Arc::new(RouteTable::new());
-    let initial = route_repo.list_all().await?;
-    let compiled = compile_routes(&initial)
-        .map_err(|e| anyhow::anyhow!("failed to compile stored routes: {e}"))?;
-    route_table.replace_all(compiled);
-
     // Same shape for app domains (Host → app_id cache).
     let app_domain_table = Arc::new(AppDomainTable::new());
     let initial_domains = domains_repo.list_all().await?;