fix(miniapp): allow concurrent RPCs per worker by narrowing lock scope

src/crates/assembly/core/src/agentic/tools/product_runtime/catalog.rs

@@ -782,7 +782,7 @@ mod tests {
             .allowed_tool_names
             .contains(&"ReviewPlatform".to_string()));
         assert!(manifest
-            .collapsed_tool_names
+            .deferred_tool_names
             .contains(&"ReviewPlatform".to_string()));
         assert!(manifest
             .tool_definitions

src/crates/assembly/core/src/miniapp/js_worker_pool.rs

@@ -19,7 +19,6 @@ use bitfun_services_integrations::miniapp::worker_pool::{
 use serde_json::Value;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
-use tokio::sync::Mutex;
 
 pub struct JsWorkerPool {
     inner: ServiceJsWorkerPool,
@@ -50,7 +49,7 @@ impl JsWorkerPool {
         worker_revision: &str,
         policy_json: &str,
         node_perms: Option<&NodePermissions>,
-    ) -> BitFunResult<Arc<Mutex<JsWorker>>> {
+    ) -> BitFunResult<Arc<JsWorker>> {
         self.inner
             .get_or_spawn(app_id, worker_revision, policy_json, node_perms)
             .await
@@ -65,7 +64,7 @@ impl JsWorkerPool {
         worker_revision: &str,
         policy_json: &str,
         node_perms: Option<&NodePermissions>,
-    ) -> BitFunResult<Arc<Mutex<JsWorker>>> {
+    ) -> BitFunResult<Arc<JsWorker>> {
         self.inner
             .get_or_spawn_with_app_dir(
                 worker_key,

src/crates/services/services-integrations/src/miniapp/worker.rs

@@ -1,13 +1,20 @@
 //! JS Worker — single child process (Bun/Node) with stdin/stderr JSON-RPC.
+//!
+//! Concurrency model: `call` takes `&self` and locks stdin only while writing
+//! the request line; responses are routed by id through the `pending` map, so
+//! multiple RPCs on the same worker proceed independently. An in-flight
+//! counter marks the worker busy — the pool must never idle-reap or LRU-evict
+//! a worker with pending RPCs, since `last_activity` alone goes stale during
+//! long calls that emit no stderr output.
 
 use bitfun_product_domains::miniapp::runtime::DetectedRuntime;
 use serde_json::Value;
 use std::collections::HashMap;
 use std::future::Future;
 use std::path::Path;
 use std::pin::Pin;
-use std::sync::atomic::{AtomicI64, Ordering};
-use std::sync::Arc;
+use std::sync::atomic::{AtomicI64, AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex as StdMutex, MutexGuard};
 use std::time::Duration;
 use tokio::io::{AsyncBufReadExt, BufReader};
 use tokio::process::{Child, ChildStdin};
@@ -31,12 +38,41 @@ pub trait MiniAppWorkerEventSink: Send + Sync {
 
 pub type SharedMiniAppWorkerEventSink = Arc<dyn MiniAppWorkerEventSink>;
 
+fn now_ms() -> i64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_millis() as i64
+}
+
+fn lock_pending(pending: &StdMutex<PendingResponseMap>) -> MutexGuard<'_, PendingResponseMap> {
+    pending.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
+}
+
+/// Removes the pending entry and drops the busy mark on every exit path of
+/// `call`, including timeout and caller cancellation.
+struct InflightGuard {
+    id: String,
+    pending: Arc<StdMutex<PendingResponseMap>>,
+    inflight: Arc<AtomicUsize>,
+    last_activity: Arc<AtomicI64>,
+}
+
+impl Drop for InflightGuard {
+    fn drop(&mut self) {
+        lock_pending(&self.pending).remove(&self.id);
+        self.inflight.fetch_sub(1, Ordering::SeqCst);
+        self.last_activity.store(now_ms(), Ordering::SeqCst);
+    }
+}
+
 /// Single JS Worker process: stdin for requests, stderr for RPC responses, stdout for user logs.
 pub struct JsWorker {
-    _child: Child,
+    child: Mutex<Child>,
     stdin: Mutex<Option<ChildStdin>>,
-    pending: Arc<Mutex<PendingResponseMap>>,
+    pending: Arc<StdMutex<PendingResponseMap>>,
     last_activity: Arc<AtomicI64>,
+    inflight: Arc<AtomicUsize>,
 }
 
 impl JsWorker {
@@ -67,13 +103,8 @@ impl JsWorker {
         let stderr = child.stderr.take().ok_or("No stderr")?;
         let _stdout = child.stdout.take();
 
-        let pending = Arc::new(Mutex::new(PendingResponseMap::new()));
-        let last_activity = Arc::new(AtomicI64::new(
-            std::time::SystemTime::now()
-                .duration_since(std::time::UNIX_EPOCH)
-                .unwrap_or_default()
-                .as_millis() as i64,
-        ));
+        let pending = Arc::new(StdMutex::new(PendingResponseMap::new()));
+        let last_activity = Arc::new(AtomicI64::new(now_ms()));
 
         let pending_clone = pending.clone();
         let last_activity_clone = last_activity.clone();
@@ -84,15 +115,7 @@ impl JsWorker {
                 if line.is_empty() {
                     continue;
                 }
-                let _ =
-                    last_activity_clone.fetch_update(Ordering::SeqCst, Ordering::SeqCst, |_| {
-                        Some(
-                            std::time::SystemTime::now()
-                                .duration_since(std::time::UNIX_EPOCH)
-                                .unwrap_or_default()
-                                .as_millis() as i64,
-                        )
-                    });
+                last_activity_clone.store(now_ms(), Ordering::SeqCst);
                 let msg: Value = match serde_json::from_str(&line) {
                     Ok(v) => v,
                     Err(_) => continue,
@@ -112,8 +135,8 @@ impl JsWorker {
                             .cloned()
                             .ok_or_else(|| "Missing result".to_string())
                     };
-                    let mut guard = pending_clone.lock().await;
-                    if let Some(tx) = guard.remove(&id) {
+                    let tx = lock_pending(&pending_clone).remove(&id);
+                    if let Some(tx) = tx {
                         let _ = tx.send(result);
                     }
                     continue;
@@ -133,17 +156,29 @@ impl JsWorker {
                     .await;
                 }
             }
+
+            // stderr EOF: the process died — fail all in-flight calls now
+            // instead of letting each caller run out its own timeout.
+            let waiters: Vec<PendingResponseSender> = {
+                let mut guard = lock_pending(&pending_clone);
+                guard.drain().map(|(_, tx)| tx).collect()
+            };
+            for tx in waiters {
+                let _ = tx.send(Err("Worker exited".to_string()));
+            }
         });
 
         Ok(Self {
-            _child: child,
+            child: Mutex::new(child),
             stdin: Mutex::new(Some(stdin_handle)),
             pending,
             last_activity,
+            inflight: Arc::new(AtomicUsize::new(0)),
         })
     }
 
     /// Send a JSON-RPC request and wait for the response (with timeout).
+    /// Concurrent calls interleave: the stdin lock covers only the write.
     pub async fn call(
         &self,
         method: &str,
@@ -160,51 +195,112 @@ impl JsWorker {
         let line = serde_json::to_string(&request).map_err(|e| e.to_string())? + "\n";
 
         let (tx, rx) = oneshot::channel();
+        self.inflight.fetch_add(1, Ordering::SeqCst);
+        let _inflight = InflightGuard {
+            id: id.clone(),
+            pending: Arc::clone(&self.pending),
+            inflight: Arc::clone(&self.inflight),
+            last_activity: Arc::clone(&self.last_activity),
+        };
+        lock_pending(&self.pending).insert(id, tx);
+        self.last_activity.store(now_ms(), Ordering::SeqCst);
+
         {
-            let mut guard = self.pending.lock().await;
-            guard.insert(id.clone(), tx);
+            let mut stdin_guard = self.stdin.lock().await;
+            let stdin = stdin_guard.as_mut().ok_or("Worker stdin closed")?;
+            use tokio::io::AsyncWriteExt;
+            stdin
+                .write_all(line.as_bytes())
+                .await
+                .map_err(|e| e.to_string())?;
+            stdin.flush().await.map_err(|e| e.to_string())?;
         }
-        self.last_activity.store(
-            std::time::SystemTime::now()
-                .duration_since(std::time::UNIX_EPOCH)
-                .unwrap_or_default()
-                .as_millis() as i64,
-            Ordering::SeqCst,
-        );
-
-        let mut stdin_guard = self.stdin.lock().await;
-        let stdin = stdin_guard.as_mut().ok_or("Worker stdin closed")?;
-        use tokio::io::AsyncWriteExt;
-        stdin
-            .write_all(line.as_bytes())
-            .await
-            .map_err(|e| e.to_string())?;
-        stdin.flush().await.map_err(|e| e.to_string())?;
-        drop(stdin_guard);
 
         let timeout = Duration::from_millis(timeout_ms);
         match tokio::time::timeout(timeout, rx).await {
             Ok(Ok(Ok(v))) => Ok(v),
             Ok(Ok(Err(e))) => Err(e),
-            Ok(Err(_)) => {
-                let _ = self.pending.lock().await.remove(&id);
-                Err("Worker dropped response".to_string())
-            }
-            Err(_) => {
-                let _ = self.pending.lock().await.remove(&id);
-                Err(format!("Worker call timeout ({}ms)", timeout_ms))
-            }
+            Ok(Err(_)) => Err("Worker dropped response".to_string()),
+            Err(_) => Err(format!("Worker call timeout ({}ms)", timeout_ms)),
         }
     }
 
-    /// Last activity timestamp (millis since epoch).
+    /// Last activity timestamp (millis since epoch). Refreshed on request
+    /// send, on every stderr line, and when an in-flight call settles.
     pub fn last_activity_ms(&self) -> i64 {
         self.last_activity.load(Ordering::SeqCst)
     }
 
-    /// Kill the worker process.
-    pub async fn kill(&mut self) {
-        let _ = self._child.start_kill();
-        let _ = tokio::time::timeout(Duration::from_secs(2), self._child.wait()).await;
+    /// Number of RPCs currently awaiting a response.
+    pub fn inflight_count(&self) -> usize {
+        self.inflight.load(Ordering::SeqCst)
+    }
+
+    /// True while any RPC is in flight. Busy workers must not be reaped or
+    /// evicted: `last_activity` alone can look stale during a long quiet call.
+    pub fn is_busy(&self) -> bool {
+        self.inflight_count() > 0
+    }
+
+    /// Kill the worker process. Safe to call while RPCs are in flight — the
+    /// stderr reader fails them with "Worker exited" once the process dies.
+    pub async fn kill(&self) {
+        let mut child = self.child.lock().await;
+        let _ = child.start_kill();
+        let _ = tokio::time::timeout(Duration::from_secs(2), child.wait()).await;
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn inflight_guard_settles_call_state_on_drop() {
+        let pending: Arc<StdMutex<PendingResponseMap>> =
+            Arc::new(StdMutex::new(PendingResponseMap::new()));
+        let inflight = Arc::new(AtomicUsize::new(1));
+        let last_activity = Arc::new(AtomicI64::new(0));
+
+        let (tx, mut rx) = oneshot::channel();
+        lock_pending(&pending).insert("rpc-1".to_string(), tx);
+
+        drop(InflightGuard {
+            id: "rpc-1".to_string(),
+            pending: Arc::clone(&pending),
+            inflight: Arc::clone(&inflight),
+            last_activity: Arc::clone(&last_activity),
+        });
+
+        assert!(lock_pending(&pending).is_empty());
+        assert_eq!(inflight.load(Ordering::SeqCst), 0);
+        assert!(last_activity.load(Ordering::SeqCst) > 0);
+        // The sender side is gone, so a late response would be dropped.
+        assert!(rx.try_recv().is_err());
+    }
+
+    #[test]
+    fn inflight_guard_leaves_other_pending_calls_untouched() {
+        let pending: Arc<StdMutex<PendingResponseMap>> =
+            Arc::new(StdMutex::new(PendingResponseMap::new()));
+        let inflight = Arc::new(AtomicUsize::new(2));
+        let last_activity = Arc::new(AtomicI64::new(0));
+
+        let (tx1, _rx1) = oneshot::channel();
+        let (tx2, _rx2) = oneshot::channel();
+        lock_pending(&pending).insert("rpc-1".to_string(), tx1);
+        lock_pending(&pending).insert("rpc-2".to_string(), tx2);
+
+        drop(InflightGuard {
+            id: "rpc-1".to_string(),
+            pending: Arc::clone(&pending),
+            inflight: Arc::clone(&inflight),
+            last_activity: Arc::clone(&last_activity),
+        });
+
+        let guard = lock_pending(&pending);
+        assert_eq!(guard.len(), 1);
+        assert!(guard.contains_key("rpc-2"));
+        assert_eq!(inflight.load(Ordering::SeqCst), 1);
     }
 }