fix(queue): carry the target namespace from enqueue through to delivery

acp/Cargo.toml

@@ -23,7 +23,7 @@ name = "iii-acp"
 path = "src/main.rs"
 
 [dependencies]
-iii-sdk = "=0.21.6"
+iii-sdk = "=0.22.0-alpha.3"
 tokio = { version = "1", features = ["macros", "rt-multi-thread", "io-std", "io-util", "sync", "time", "signal"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"

acp/src/handler.rs

@@ -704,6 +704,9 @@ fn register_event_subscriber(
         function_id: fn_id,
         config: json!({ "stream_name": AGENT_EVENTS_STREAM }),
         metadata: None,
+        // Resolve the trigger's target in this worker's namespace so it matches
+        // where the function was registered (None => engine default).
+        namespace: iii.namespace(),
     }) {
         Ok(t) => Some(t),
         Err(e) => {

approval-gate/Cargo.toml

@@ -15,7 +15,7 @@ name = "approval_gate"
 path = "src/lib.rs"
 
 [dependencies]
-iii-sdk = "=0.21.6"
+iii-sdk = "=0.22.0-alpha.3"
 tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "signal", "time"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"

approval-gate/src/configuration.rs

@@ -128,6 +128,7 @@ pub fn bind_hook(iii: &IIIClient) {
             "on_error": HOOK_ON_ERROR,
         }),
         metadata: None,
+        namespace: iii.namespace(),
     }) {
         Ok(_) => tracing::info!(
             trigger_type = "harness::hook::pre-trigger",
@@ -157,6 +158,7 @@ pub fn bind_filesystem_access_watch_hook(iii: &IIIClient) {
             "on_error": FILESYSTEM_ACCESS_WATCH_ON_ERROR,
         }),
         metadata: None,
+        namespace: iii.namespace(),
     }) {
         Ok(_) => tracing::info!(
             trigger_type = "harness::hook::post-trigger",
@@ -262,6 +264,7 @@ pub fn register_config_trigger(iii: &IIIClient, cell: ConfigCell) -> Result<(),
             "event_types": ["configuration:updated"],
         }),
         metadata: None,
+        namespace: iii.namespace(),
     })?;
     Ok(())
 }

approval-gate/src/events.rs

@@ -253,15 +253,36 @@ impl Emitter {
     ) {
         for binding in set.snapshot() {
             if binding_matches(&binding.filter, session_id, session_metadata) {
-                let res = self
-                    .iii
-                    .trigger(TriggerRequest {
-                        function_id: binding.function_id.clone(),
-                        payload: payload.clone(),
-                        action: Some(TriggerAction::Void),
-                        timeout_ms: None,
-                    })
-                    .await;
+                let request = TriggerRequest {
+                    function_id: binding.function_id.clone(),
+                    payload: payload.clone(),
+                    action: Some(TriggerAction::Void),
+                    timeout_ms: None,
+                };
+                let route_ns = self.iii.namespace().filter(|_| {
+                    !matches!(
+                        binding.function_id.split("::").next(),
+                        Some(
+                            "state"
+                                | "stream"
+                                | "queue"
+                                | "pubsub"
+                                | "configuration"
+                                | "cron"
+                                | "http"
+                                | "engine"
+                                | "sandbox"
+                                | "log"
+                                | "secret"
+                                | "kv"
+                                | "iii"
+                        )
+                    )
+                });
+                let res = match route_ns {
+                    Some(ns) => self.iii.trigger(request.namespace(ns)).await,
+                    None => self.iii.trigger(request).await,
+                };
                 if let Err(e) = res {
                     tracing::warn!(
                         function_id = %binding.function_id,
@@ -314,6 +335,7 @@ mod tests {
             function_id: function_id.into(),
             config,
             metadata: None,
+            namespace: None,
         }
     }
 

approval-gate/src/harness.rs

@@ -10,13 +10,16 @@ use iii_sdk::IIIClient;
 use serde_json::{json, Value};
 
 pub async fn function_resolve(iii: &IIIClient, payload: Value) -> Result<Value, Error> {
-    iii.trigger(TriggerRequest {
+    let request = TriggerRequest {
         function_id: "harness::function::resolve".into(),
         payload,
         action: None,
         timeout_ms: None,
-    })
-    .await
+    };
+    match iii.namespace() {
+        Some(ns) => iii.trigger(request.namespace(ns)).await,
+        None => iii.trigger(request).await,
+    }
 }
 
 fn parse_roots(reply: &Value) -> Vec<String> {
@@ -36,14 +39,16 @@ fn parse_roots(reply: &Value) -> Vec<String> {
 /// error on "no grants" — only on transport/RPC failure, e.g. an older
 /// harness that doesn't implement this control plane yet).
 pub async fn filesystem_grants(iii: &IIIClient, session_id: &str) -> Result<Vec<String>, Error> {
-    let reply = iii
-        .trigger(TriggerRequest {
-            function_id: "harness::filesystem::grants".into(),
-            payload: json!({ "session_id": session_id }),
-            action: None,
-            timeout_ms: None,
-        })
-        .await?;
+    let request = TriggerRequest {
+        function_id: "harness::filesystem::grants".into(),
+        payload: json!({ "session_id": session_id }),
+        action: None,
+        timeout_ms: None,
+    };
+    let reply = match iii.namespace() {
+        Some(ns) => iii.trigger(request.namespace(ns)).await?,
+        None => iii.trigger(request).await?,
+    };
     Ok(parse_roots(&reply))
 }
 
@@ -55,13 +60,15 @@ pub async fn filesystem_grant(
     session_id: &str,
     root: &str,
 ) -> Result<Vec<String>, Error> {
-    let reply = iii
-        .trigger(TriggerRequest {
-            function_id: "harness::filesystem::grant".into(),
-            payload: json!({ "session_id": session_id, "root": root }),
-            action: None,
-            timeout_ms: None,
-        })
-        .await?;
+    let request = TriggerRequest {
+        function_id: "harness::filesystem::grant".into(),
+        payload: json!({ "session_id": session_id, "root": root }),
+        action: None,
+        timeout_ms: None,
+    };
+    let reply = match iii.namespace() {
+        Some(ns) => iii.trigger(request.namespace(ns)).await?,
+        None => iii.trigger(request).await?,
+    };
     Ok(parse_roots(&reply))
 }

approval-gate/src/main.rs

@@ -86,6 +86,7 @@ fn bind_best_effort(
         function_id: function_id.to_string(),
         config,
         metadata: None,
+        namespace: iii.namespace(),
     });
     match res {
         Ok(_) => tracing::info!(trigger_type, function_id, "trigger binding requested"),

approval-gate/src/session.rs

@@ -6,11 +6,14 @@ use iii_sdk::IIIClient;
 use serde_json::{json, Value};
 
 pub async fn get(iii: &IIIClient, session_id: &str) -> Result<Value, Error> {
-    iii.trigger(TriggerRequest {
+    let request = TriggerRequest {
         function_id: "session::get".into(),
         payload: json!({ "session_id": session_id }),
         action: None,
         timeout_ms: None,
-    })
-    .await
+    };
+    match iii.namespace() {
+        Some(ns) => iii.trigger(request.namespace(ns)).await,
+        None => iii.trigger(request).await,
+    }
 }

approval-gate/src/testkit/engine.rs

@@ -459,13 +459,15 @@ pub async fn boot(engine: &Engine, opts: BootOpts) -> TestStack {
         function_id: "recorder::on_created".to_string(),
         config: json!({}),
         metadata: None,
+        namespace: iii.namespace(),
     })
     .expect("bind pending_created");
     iii.register_trigger(RegisterTriggerInput {
         trigger_type: PENDING_RESOLVED.to_string(),
         function_id: "recorder::on_resolved".to_string(),
         config: json!({}),
         metadata: None,
+        namespace: iii.namespace(),
     })
     .expect("bind pending_resolved");
 

bridge/Cargo.toml

@@ -18,7 +18,7 @@ name = "bridge"
 path = "src/main.rs"
 
 [dependencies]
-iii-sdk = "=0.21.6"
+iii-sdk = "=0.22.0-alpha.3"
 async-trait = "0.1"
 tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "signal", "time"] }
 serde = { version = "1", features = ["derive"] }

bridge/src/boot.rs

@@ -122,7 +122,7 @@ pub async fn start(iii: Arc<IIIClient>, config: BridgeConfig) -> anyhow::Result<
 /// `bridge.invoke` + `bridge.invoke_async` on the local engine — exact
 /// function ids and descriptions from the builtin (mod.rs:96-191).
 ///
-/// Both use `new_async_with_bad_request` with the typed [`functions::
+/// Both use `new_async` with the typed [`functions::
 /// InvokeInput`]: the SDK auto-extracts a real request schema instead of the
 /// permissive `AnyValue` a `Value` closure param would emit, while
 /// [`functions::invoke_bad_request`] keeps owning the `deserialization_error`
@@ -133,33 +133,27 @@ pub fn register_bridge_functions(iii: &Arc<IIIClient>, remote: RemoteCell) {
         let caller = caller.clone();
         iii.register_function(
             functions::INVOKE_FN,
-            RegisterFunction::new_async_with_bad_request(
-                move |req: functions::InvokeInput| {
-                    let caller = caller.clone();
-                    async move {
-                        functions::handle_invoke_typed(caller.as_ref(), req)
-                            .await
-                            .map_err(Error::from)
-                    }
-                },
-                functions::invoke_bad_request,
-            )
-            .description("Invoke a function on the remote III instance"),
-        );
-    }
-    iii.register_function(
-        functions::INVOKE_ASYNC_FN,
-        RegisterFunction::new_async_with_bad_request(
-            move |req: functions::InvokeInput| {
+            RegisterFunction::new_async(move |req: functions::InvokeInput| {
                 let caller = caller.clone();
                 async move {
-                    functions::handle_invoke_async_typed(caller.as_ref(), req)
+                    functions::handle_invoke_typed(caller.as_ref(), req)
                         .await
                         .map_err(Error::from)
                 }
-            },
-            functions::invoke_bad_request,
-        )
+            })
+            .description("Invoke a function on the remote III instance"),
+        );
+    }
+    iii.register_function(
+        functions::INVOKE_ASYNC_FN,
+        RegisterFunction::new_async(move |req: functions::InvokeInput| {
+            let caller = caller.clone();
+            async move {
+                functions::handle_invoke_async_typed(caller.as_ref(), req)
+                    .await
+                    .map_err(Error::from)
+            }
+        })
         .description("Fire-and-forget invoke on the remote III instance"),
     );
 }

bridge/src/configuration.rs

@@ -205,6 +205,7 @@ pub async fn register_config_trigger(iii: &Arc<IIIClient>, boot: &BootHandle) ->
             "event_types": ["configuration:updated"],
         }),
         metadata: None,
+        namespace: iii.namespace(),
     })?;
     Ok(())
 }

browser/Cargo.toml

@@ -15,7 +15,7 @@ name = "browser"
 path = "src/lib.rs"
 
 [dependencies]
-iii-sdk = "=0.21.6"
+iii-sdk = "=0.22.0-alpha.3"
 arc-swap = "1"
 tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "signal", "time", "process"] }
 serde = { version = "1", features = ["derive"] }