fix(tools): harden rss dispatch ownership

Compile from_source as supplied bytes, cache programs by module-tree
digest, keep a single production worker path through tools::dispatch,
and fail closed on process shutdown races and malformed JSON payloads.

Author: fffonion

docs/configuration.md

@@ -193,12 +193,11 @@ page bounds.
 ## Coding tools and serial loop
 
 The library `AgentService` worker compiles bundled `rss/agent/main.rss` and
-drives a **serial** native tool loop. RSS builds canonical provider requests
-and dispatches tools only through the native host bridges
-(`agent::provider_call`, `agent::tool_dispatch`). This is not an
-OpenAI-compatible inference path.
+drives a **serial** RSS `tools::dispatch` loop over the generic capability
+host (`agent::provider_call`, filesystem/process/artifact adapters). This is
+not an OpenAI-compatible inference path.
 
-Built-in native tools, in registry order:
+Built-in RSS registry tools, in registry order:
 
 | Name | Toolset | Risk | Notes |
 | --- | --- | --- | --- |

rss/tools/dispatch.rss

@@ -59,21 +59,6 @@ fn call_id(call: map) -> string {
     id
 }
 
-fn json_object_text(text: string) -> bool {
-    let mut ok: bool = false;
-    if text.length >= 2 {
-        if text[0:1] == "{" {
-            let second: string = text[1:2];
-            if second == "\"" || second == "}" || second == " " || second == "\n" || second == "\t" {
-                if text[(text.length - 1):text.length] == "}" {
-                    ok = true;
-                }
-            }
-        }
-    }
-    ok
-}
-
 fn parse_call_arguments(call: map) -> map {
     let mut out: map = {
         ok: true,
@@ -93,19 +78,24 @@ fn parse_call_arguments(call: map) -> map {
         if call.has("arguments_json") {
             if type(call.arguments_json) == "string" {
                 let text: string = call.arguments_json;
-                if text.length > 0 {
-                    if json_object_text(text) {
-                        let decoded: map = json::decode(text);
-                        out.arguments = decoded;
-                    } else {
-                        out = {
-                            ok: false,
-                            arguments: {},
-                            code: "malformed_payload",
-                            message: "tool arguments must be a JSON object"
-                        };
-                    }
+                let parsed: map = agent::parse_json_object(text);
+                if map_bool(parsed, "ok", false) {
+                    out.arguments = types::map_map(parsed, "value");
+                } else {
+                    out = {
+                        ok: false,
+                        arguments: {},
+                        code: "malformed_payload",
+                        message: "tool arguments must be a JSON object"
+                    };
                 }
+            } else {
+                out = {
+                    ok: false,
+                    arguments: {},
+                    code: "malformed_payload",
+                    message: "tool arguments must be a JSON object"
+                };
             }
         }
     }
@@ -128,7 +118,18 @@ fn find_named(registry: array, name: string) -> map {
     let mut found: map = {};
     let mut matches: int = 0;
     let mut i: int = 0;
+    let mut scans: int = 0;
     while i < registry.length {
+        if i < 0 {
+            break;
+        }
+        scans = scans + 1;
+        if scans > 64 {
+            break;
+        }
+        if registry.has(i) == false {
+            break;
+        }
         let mut entry: map = {};
         if type(registry[i]) == "map" {
             let coerced: map = registry[i];
@@ -146,23 +147,57 @@ fn find_named(registry: array, name: string) -> map {
     }
 }
 
+// Duplicate scan is O(n^2) with a hard n<=64 bound. The comparison
+// counter is n*(n-1)/2 worst case; 64 is exact-max, 65 fails closed.
 fn has_duplicate_names(registry: array) -> bool {
     let mut dup: bool = false;
-    let mut i: int = 0;
-    while i < registry.length {
-        let mut entry: map = {};
-        if type(registry[i]) == "map" {
-            let coerced: map = registry[i];
-            entry = coerced;
-        }
-        let name: string = types::map_string(entry, "name", "");
-        if name.length > 0 {
-            let counted: map = find_named(registry, name);
-            if counted.matches > 1 {
+    if registry.length > 64 {
+        dup = true;
+    } else {
+        let mut comparisons: int = 0;
+        let mut i: int = 0;
+        while i < registry.length {
+            if i < 0 {
                 dup = true;
+                break;
+            }
+            if registry.has(i) == false {
+                break;
             }
+            let mut left: map = {};
+            if type(registry[i]) == "map" {
+                let coerced: map = registry[i];
+                left = coerced;
+            }
+            let name: string = types::map_string(left, "name", "");
+            if name.length > 0 {
+                let mut j: int = i + 1;
+                while j < registry.length {
+                    if j < 0 {
+                        dup = true;
+                        break;
+                    }
+                    comparisons = comparisons + 1;
+                    if comparisons > 64 * 64 {
+                        dup = true;
+                        break;
+                    }
+                    if registry.has(j) == false {
+                        break;
+                    }
+                    let mut right: map = {};
+                    if type(registry[j]) == "map" {
+                        let other: map = registry[j];
+                        right = other;
+                    }
+                    if types::map_string(right, "name", "") == name {
+                        dup = true;
+                    }
+                    j = j + 1;
+                }
+            }
+            i = i + 1;
         }
-        i = i + 1;
     }
     dup
 }

src/capabilities/mod.rs

@@ -10,6 +10,8 @@ pub mod types;
 mod confined_io;
 mod hash;
 
+pub(crate) use hash::sha256_hex;
+
 pub use artifacts::{ArtifactCapability, ArtifactLimits, ArtifactRef};
 pub use filesystem::{
     FilesystemCapability, FilesystemLimits, FsDirEntry, FsList, FsMetadata, FsRead, FsWrite,

src/capabilities/process.rs

@@ -7,7 +7,7 @@ use std::{
     collections::HashMap,
     sync::{
         Arc, Mutex, Weak,
-        atomic::{AtomicBool, AtomicUsize, Ordering},
+        atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering},
         mpsc::{self, RecvTimeoutError},
     },
     thread,
@@ -146,9 +146,13 @@ struct ProcessInner {
     host_limits: ProcessLimits,
     root: ConfinedFsRoot,
     table: Mutex<HashMap<String, OwnedProcess>>,
+    closing: AtomicBool,
+    generation: AtomicU64,
     after_running_poll_hook: Mutex<Option<ProcessOpHook>>,
     write_blocked_hook: Mutex<Option<ProcessOpHook>>,
     before_write_cycle_hook: Mutex<Option<ProcessOpHook>>,
+    before_os_spawn_hook: Mutex<Option<ProcessOpHook>>,
+    after_os_spawn_hook: Mutex<Option<ProcessOpHook>>,
     stdin_workers: AtomicUsize,
 }
 
@@ -208,9 +212,13 @@ impl ProcessCapability {
                 host_limits,
                 root,
                 table: Mutex::new(HashMap::new()),
+                closing: AtomicBool::new(false),
+                generation: AtomicU64::new(1),
                 after_running_poll_hook: Mutex::new(None),
                 write_blocked_hook: Mutex::new(None),
                 before_write_cycle_hook: Mutex::new(None),
+                before_os_spawn_hook: Mutex::new(None),
+                after_os_spawn_hook: Mutex::new(None),
                 stdin_workers: AtomicUsize::new(0),
             }),
         })
@@ -239,6 +247,9 @@ impl ProcessCapability {
         stdin: Option<&[u8]>,
     ) -> Result<ProcessSpawn, CapabilityError> {
         let claims = self.authorize(token, CapabilityRisk::Execute)?;
+        if self.is_closing() {
+            return Err(closing_error());
+        }
         if argv.is_empty() {
             return Err(CapabilityError::new(
                 "invalid_request",
@@ -281,16 +292,32 @@ impl ProcessCapability {
         {
             request = request.with_stdin(stdin.to_vec());
         }
+        if self.is_closing() {
+            return Err(closing_error());
+        }
+        fire_hook(&self.inner.before_os_spawn_hook);
+        if self.is_closing() {
+            return Err(closing_error());
+        }
+        let fence = self.inner.generation.load(Ordering::SeqCst);
         let process = BoundedProcess::spawn(request).map_err(map_process_error)?;
+        fire_hook(&self.inner.after_os_spawn_hook);
         let handle = process.lifecycle_handle();
         let write_handle = handle.clone();
         let pid = handle.pid();
         let id = uuid::Uuid::new_v4().simple().to_string();
-        self.inner
-            .table
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner())
-            .insert(
+        {
+            let mut table = self
+                .inner
+                .table
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner());
+            if self.is_closing() || self.inner.generation.load(Ordering::SeqCst) != fence {
+                drop(table);
+                let _ = handle.shutdown();
+                return Err(closing_error());
+            }
+            table.insert(
                 id.clone(),
                 OwnedProcess {
                     owner_key: claims.owner.key(),
@@ -299,6 +326,7 @@ impl ProcessCapability {
                     cancel: cancel.clone(),
                 },
             );
+        }
         let reaper = Arc::new(ProcessReaper {
             inner: Arc::downgrade(&self.inner),
             id: id.clone(),
@@ -466,6 +494,24 @@ impl ProcessCapability {
             .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(hook);
     }
 
+    /// Test barrier: fires immediately before the OS spawn syscall.
+    pub fn set_before_os_spawn_hook(&self, hook: Arc<dyn Fn() + Send + Sync>) {
+        *self
+            .inner
+            .before_os_spawn_hook
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(hook);
+    }
+
+    /// Test barrier: fires after OS spawn and before table insert.
+    pub fn set_after_os_spawn_hook(&self, hook: Arc<dyn Fn() + Send + Sync>) {
+        *self
+            .inner
+            .after_os_spawn_hook
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(hook);
+    }
+
     /// Returns a bounded log window.
     pub fn log(
         &self,
@@ -550,9 +596,13 @@ impl ProcessCapability {
 
     /// Terminates every owned child and drops table entries.
     ///
-    /// Run cleanup must drain committed background residue; `cancel_all`
-    /// kills the process tree but leaves handles observable in the table.
+    /// Closing is irreversible: later spawns refuse insert and terminate any
+    /// OS process created after the fence. Run cleanup must drain committed
+    /// background residue; `cancel_all` kills the process tree but leaves
+    /// handles observable in the table.
     pub fn shutdown_all(&self) {
+        self.inner.closing.store(true, Ordering::SeqCst);
+        self.inner.generation.fetch_add(1, Ordering::SeqCst);
         let owned: Vec<OwnedProcess> = {
             let mut table = self
                 .inner
@@ -566,6 +616,11 @@ impl ProcessCapability {
         }
     }
 
+    /// True after [`Self::shutdown_all`] has started. The fence is irreversible.
+    pub fn is_closing(&self) -> bool {
+        self.inner.closing.load(Ordering::SeqCst)
+    }
+
     fn authorize(&self, token: &str, risk: CapabilityRisk) -> Result<TokenClaims, CapabilityError> {
         match self
             .inner
@@ -578,6 +633,9 @@ impl ProcessCapability {
     }
 
     fn lookup(&self, token: &str, handle: &str) -> Result<OwnedProcess, CapabilityError> {
+        if self.is_closing() {
+            return Err(closing_error());
+        }
         let claims = self.authorize(token, CapabilityRisk::Execute)?;
         let table = self
             .inner
@@ -712,6 +770,10 @@ fn fire_hook(slot: &Mutex<Option<ProcessOpHook>>) {
     }
 }
 
+fn closing_error() -> CapabilityError {
+    CapabilityError::new("capability_unavailable", "process capability is closing")
+}
+
 fn interrupt_write_worker(
     handle: &BoundedProcessHandle,
     worker: thread::JoinHandle<()>,

src/gateway/api_server.rs

@@ -691,7 +691,7 @@ async fn delete_session_handler(
     {
         state
             .service()
-            .cleanup_session_native_dispatch(&session_id_for_cleanup);
+            .cleanup_session_capability_host(&session_id_for_cleanup);
     }
     response
 }