Hold a run slot until the session reads its message, and test the sandbox on Linux

Author: gaoyu06

.github/workflows/ci.yml

@@ -21,7 +21,12 @@ jobs:
           components: rustfmt, clippy
 
       - name: Install test dependencies
-        run: sudo apt-get update && sudo apt-get install -y ripgrep
+        run: sudo apt-get update && sudo apt-get install -y ripgrep bubblewrap
+
+      # The sandbox tests run commands under bwrap, which needs unprivileged
+      # user namespaces; Ubuntu 24.04 restricts them through AppArmor.
+      - name: Allow unprivileged user namespaces
+        run: sudo sysctl -w kernel.apparmor_restrict_unprivileged_userns=0
 
       - name: Check formatting
         run: cargo fmt --check

crates/agent-core/src/sandbox.rs

@@ -595,9 +595,9 @@ mod tests {
         assert!(args.contains("--unshare-net"));
     }
 
-    #[cfg(target_os = "macos")]
+    #[cfg(any(target_os = "macos", target_os = "linux"))]
     #[test]
-    fn seatbelt_enforces_the_policy() {
+    fn the_os_sandbox_enforces_the_policy() {
         use std::process::Command;
         let dir = work("seatbelt");
         let outside = real(&env::temp_dir())

crates/agent-core/tests/hosted_engines.rs

@@ -239,7 +239,7 @@ fn host_tools_run_in_the_host_and_host_prompt_reaches_the_model() {
     assert!(assistant_text(&events).contains("<host-marker>brief goes here</host-marker>"));
 }
 
-#[cfg(target_os = "macos")]
+#[cfg(any(target_os = "macos", target_os = "linux"))]
 #[test]
 fn a_sandboxed_engine_runs_commands_inside_and_asks_to_leave() {
     use jucode_agent_core::sandbox::{CommandRule, RuleAction, SandboxMode, SandboxPolicy};

crates/daemon/src/hub.rs

@@ -34,6 +34,10 @@ pub struct Hub {
     sessions: Mutex<HashMap<String, Hosted>>,
     /// Sessions whose engine is running a turn or has queued messages.
     busy: Mutex<HashSet<String>>,
+    /// Delivered messages a session thread has not processed yet. They hold
+    /// a running slot: the engine still reports "ready" until it has read
+    /// the message, and that must not free the slot early.
+    claims: Mutex<HashMap<String, usize>>,
     /// When each session last received a message (ms), for routing.
     last_active: Mutex<HashMap<String, u64>>,
     /// Serializes message delivery (scheduler ticks, sends, tool calls).
@@ -70,6 +74,7 @@ impl Hub {
             version,
             sessions: Mutex::new(HashMap::new()),
             busy: Mutex::new(HashSet::new()),
+            claims: Mutex::new(HashMap::new()),
             last_active: Mutex::new(HashMap::new()),
             delivering: Mutex::new(()),
             next_id: AtomicU64::new(0),
@@ -277,6 +282,7 @@ impl Hub {
             let _ = self.store.record_session_closed(id);
         }
         lock(&self.busy).remove(id);
+        lock(&self.claims).remove(id);
         self.broadcast(&json!({ "type": "session_closed", "session": id }));
     }
 
@@ -315,8 +321,13 @@ impl Hub {
         }
     }
 
-    /// Called by a session thread when its engine starts or finishes work.
+    /// Called by a session thread with its engine's state every tick. A
+    /// session with an unprocessed delivery stays busy.
     pub fn set_busy(&self, session: &str, busy: bool) {
+        let busy = busy
+            || lock(&self.claims)
+                .get(session)
+                .is_some_and(|count| *count > 0);
         let changed = if busy {
             lock(&self.busy).insert(session.to_string())
         } else {
@@ -327,6 +338,17 @@ impl Hub {
         }
     }
 
+    /// The session thread has handed a delivered message to its engine.
+    pub fn release_claim(&self, session: &str) {
+        let mut claims = lock(&self.claims);
+        if let Some(count) = claims.get_mut(session) {
+            *count = count.saturating_sub(1);
+            if *count == 0 {
+                claims.remove(session);
+            }
+        }
+    }
+
     pub fn agents_json(&self) -> Value {
         let records = self.store.sessions();
         let busy = lock(&self.busy).clone();
@@ -610,13 +632,17 @@ impl Hub {
             }
             None => self.create_session(None, Some(&message.to))?,
         };
-        self.forward(
-            &session,
-            json!({ "op": "user_message", "content": delivery_text(message) }),
-        )?;
-        // A delivered message starts (or queues) a run; count it before the
-        // engine reports its status so the next message sees the slot taken.
+        // A delivered message starts (or queues) a run: claim the slot before
+        // forwarding, so the next message sees it taken.
+        *lock(&self.claims).entry(session.clone()).or_default() += 1;
         lock(&self.busy).insert(session.clone());
+        if let Err(error) = self.forward(
+            &session,
+            json!({ "op": "user_message", "content": delivery_text(message), "claimed": true }),
+        ) {
+            self.release_claim(&session);
+            return Err(error);
+        }
         lock(&self.last_active).insert(session.clone(), now());
         self.store
             .record_delivered(&message.id, &session)

crates/daemon/src/session.rs

@@ -157,6 +157,9 @@ fn apply(hub: &Hub, core: &mut AgentCore, id: &str, op: &Value) -> bool {
         }
     }
     let (quit, events) = protocol::apply_op(core, op);
+    if op["claimed"] == true {
+        hub.release_claim(id);
+    }
     for event in events {
         publish(hub, id, event);
     }

crates/daemon/tests/daemon.rs

@@ -795,7 +795,7 @@ fn a_device_pairs_with_a_code_from_the_desktop_until_revoked() {
     assert!(tungstenite::connect(format!("ws://{}/?token={token}", daemon.address)).is_err());
 }
 
-#[cfg(target_os = "macos")]
+#[cfg(any(target_os = "macos", target_os = "linux"))]
 #[test]
 fn an_agent_writes_its_rw_directories_and_not_its_ro_ones() {
     let _guard = setup();
@@ -834,7 +834,7 @@ fn an_agent_writes_its_rw_directories_and_not_its_ro_ones() {
     assert_eq!(fs::read_to_string(dir.join("seen.txt")).unwrap(), "boot ok");
 }
 
-#[cfg(target_os = "macos")]
+#[cfg(any(target_os = "macos", target_os = "linux"))]
 #[test]
 fn an_unattended_escalation_becomes_a_pending_action() {
     let _guard = setup();