feat: migrate workers to namespace-aware iii-sdk 0.22 prereleases

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.8"
+iii-sdk = "=0.22.0-alpha.2"
 tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "signal", "time"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"

approval-gate/src/configuration.rs

@@ -29,7 +29,25 @@ use crate::config::WorkerConfig;
 /// whole-snapshot replaces the inner `Arc` under the write lock.
 pub type ConfigCell = Arc<RwLock<Arc<WorkerConfig>>>;
 
-pub const CONFIG_ID: &str = "approval-gate";
+pub const DEFAULT_CONFIG_ID: &str = "approval-gate";
+
+/// The configuration entry this worker owns.
+///
+/// `III_CONFIG_NAME` when a supervisor set it, else the built-in name. A worker
+/// that hardcodes its id turns that id into a global scarce name: two instances
+/// share one entry and take turns overwriting it, and each write wakes both.
+/// Being told which entry is its own is what lets them differ.
+pub fn config_id() -> &'static str {
+    static ID: std::sync::OnceLock<String> = std::sync::OnceLock::new();
+    ID.get_or_init(|| {
+        std::env::var("III_CONFIG_NAME")
+            .ok()
+            .map(|value| value.trim().to_string())
+            .filter(|value| !value.is_empty())
+            .unwrap_or_else(|| DEFAULT_CONFIG_ID.to_string())
+    })
+    .as_str()
+}
 const CONFIG_FN_ID: &str = "approval::on-config-change";
 const CONFIG_RETRIES: u32 = 3;
 /// Base backoff between configuration RPC retries; multiplied by the
@@ -59,7 +77,7 @@ const FILESYSTEM_ACCESS_WATCH_ON_ERROR: &str = "fail_open";
 /// value, so this is safe to call every boot).
 pub async fn register_config(iii: &IIIClient, seed: Option<&WorkerConfig>) -> Result<(), String> {
     let mut payload = json!({
-        "id": CONFIG_ID,
+        "id": config_id(),
         "name": "Approval Gate",
         "description": "Policy and decision surface settings: the deployment \
                         approval defaults (permission mode for new sessions \
@@ -98,14 +116,14 @@ async fn should_seed_default_value(iii: &IIIClient) -> Result<bool, String> {
 async fn get_config_value(iii: &IIIClient) -> Result<Value, String> {
     try_get_config_value(iii)
         .await?
-        .ok_or_else(|| format!("configuration `{CONFIG_ID}` not found"))
+        .ok_or_else(|| format!("configuration `{config_entry}` not found", config_entry = config_id()))
 }
 
 /// Returns `Ok(None)` when the entry does not exist. The engine's
 /// missing-entry codes vary in case (`function_not_found`,
 /// `STATEMENT_NOT_FOUND`, `NOT_FOUND`), so match case-insensitively.
 async fn try_get_config_value(iii: &IIIClient) -> Result<Option<Value>, String> {
-    match trigger_with_retry(iii, "configuration::get", json!({ "id": CONFIG_ID })).await {
+    match trigger_with_retry(iii, "configuration::get", json!({ "id": config_id() })).await {
         Ok(resp) => Ok(resp.get("value").cloned()),
         Err(e) if e.to_ascii_uppercase().contains("NOT_FOUND") => Ok(None),
         Err(e) => Err(e),
@@ -128,6 +146,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 +176,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",
@@ -258,10 +278,11 @@ pub fn register_config_trigger(iii: &IIIClient, cell: ConfigCell) -> Result<(),
         trigger_type: "configuration".to_string(),
         function_id: CONFIG_FN_ID.to_string(),
         config: json!({
-            "configuration_id": CONFIG_ID,
+            "configuration_id": config_id(),
             "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"] }