test(acpmux): a gated fake agent exits with its daemon

cmux-tui/crates/acpmux/tests/fake_agent.py

@@ -727,6 +727,17 @@ def late():
                  ("claude-haiku-5-5", "Haiku 5.5"), ("claude-fable-5-1", "Fable 5.1")]
 
 
+def stdin_lines(early):
+    """stdin's lines, starting with the bytes read while waiting at the start gate."""
+    if early:
+        if not early.endswith(b"\n"):
+            early += sys.stdin.buffer.readline()
+        # Split and decoded as stdin is: only at newlines (serde_json leaves U+2028 unescaped).
+        import io
+        yield from io.TextIOWrapper(io.BytesIO(early), encoding=sys.stdin.encoding, errors=sys.stdin.errors)
+    yield from sys.stdin
+
+
 def main():
     global LAST_MCP_SERVERS
     # `--fake-login` (the terminal sign-in): sign in and exit.
@@ -738,13 +749,21 @@ def main():
         import signal
         signal.signal(signal.SIGTERM, signal.SIG_IGN)
     # FAKE_START_GATE=<path>: start (read stdin) only once that file exists: a slow agent start.
+    # It still watches stdin while it waits: once the daemon that started it is gone (EOF), it exits, so a
+    # test that never opens the gate leaves no agent behind (257 of them polled forever on huge-blue, 2026-10-21).
+    # What arrives meanwhile is kept for after the gate.
     gate = os.environ.get("FAKE_START_GATE")
+    early = b""
     if gate:
-        import time
+        import select
         while not os.path.exists(gate):
-            time.sleep(0.02)
+            if select.select([sys.stdin.fileno()], [], [], 0.02)[0]:
+                chunk = os.read(sys.stdin.fileno(), 65536)
+                if not chunk:
+                    return
+                early += chunk
     sessions = 0
-    for line in sys.stdin:
+    for line in stdin_lines(early):
         line = line.strip()
         if not line:
             continue

cmux-tui/crates/acpmux/tests/person_auto_start.rs

@@ -251,6 +251,13 @@ impl Drop for Daemon {
     fn drop(&mut self) {
         // SAFETY: the pid of the child this test spawned and still owns.
         unsafe { libc::kill(self.child.id() as i32, libc::SIGTERM) };
+        // A daemon that does not stop in time is killed, so a failed test
+        // never hangs here holding its agents' stdin open.
+        let deadline = std::time::Instant::now() + Duration::from_secs(10);
+        while matches!(self.child.try_wait(), Ok(None)) && std::time::Instant::now() < deadline {
+            std::thread::sleep(Duration::from_millis(20));
+        }
+        let _ = self.child.kill();
         let _ = self.child.wait();
         let _ = std::fs::remove_dir_all(&self.dir);
     }
@@ -689,6 +696,62 @@ async fn auto_start_a_request_during_the_start_never_rebinds_the_approval() {
     assert_eq!(current_mode(&s["result"]), "bypass", "{s}");
 }
 
+/// The gated harness never outlives the daemon holding its stdin: a test that
+/// fails before it opens the gate leaves no agent polling for it (257 were
+/// left on huge-blue, 2026-10-21). What the daemon sent before the gate opens
+/// is still read after it.
+#[test]
+fn auto_start_a_gated_agent_exits_with_its_daemon_and_keeps_what_came_first() {
+    use std::io::Write;
+    let dir = std::env::temp_dir().join(format!("aas-gate-{}", std::process::id()));
+    let _ = std::fs::remove_dir_all(&dir);
+    std::fs::create_dir_all(&dir).unwrap();
+    let gate = dir.join("gate");
+    let spawn = || {
+        std::process::Command::new("python3")
+            .arg(FAKE)
+            .env("FAKE_START_GATE", &gate)
+            .stdin(std::process::Stdio::piped())
+            .stdout(std::process::Stdio::piped())
+            .stderr(std::process::Stdio::null())
+            .spawn()
+            .unwrap()
+    };
+    let exited = |child: &mut std::process::Child| {
+        let deadline = std::time::Instant::now() + Duration::from_secs(10);
+        loop {
+            if let Ok(Some(_)) = child.try_wait() {
+                return true;
+            }
+            if std::time::Instant::now() > deadline {
+                let _ = child.kill();
+                let _ = child.wait();
+                return false;
+            }
+            std::thread::sleep(Duration::from_millis(20));
+        }
+    };
+
+    let mut waiting = spawn();
+    std::thread::sleep(Duration::from_millis(300));
+    drop(waiting.stdin.take());
+    assert!(exited(&mut waiting), "a gated agent outlived the stdin of its daemon");
+
+    let mut early = spawn();
+    let mut stdin = early.stdin.take().unwrap();
+    let init = json!({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": 1}});
+    writeln!(stdin, "{init}").unwrap();
+    std::thread::sleep(Duration::from_millis(300));
+    std::fs::write(&gate, "").unwrap();
+    let mut line = String::new();
+    BufReader::new(early.stdout.take().unwrap()).read_line(&mut line).unwrap();
+    drop(stdin);
+    assert!(exited(&mut early), "the agent did not exit at EOF after its gate");
+    let _ = std::fs::remove_dir_all(&dir);
+    let reply: Value = serde_json::from_str(&line).unwrap_or_else(|e| panic!("{e}: {line:?}"));
+    assert_eq!(reply["id"], 1, "{reply}");
+}
+
 /// The caller token and its socket are acpmux's own: a harness profile's or
 /// a preset's env that names them is dropped, so an agent never runs with a
 /// token or socket its config chose (`hub/person.rs` (a)).