fix(maestro): own ACP child signals

crates/maestro-acp/src/lib.rs

@@ -156,6 +156,10 @@ enum Command {
     Close,
 }
 
+enum ChildCommand {
+    Signal(libc::c_int),
+}
+
 /// Why the handshake did not produce a session id.
 enum ReadyError {
     /// The adapter answered with an error.
@@ -180,6 +184,7 @@ pub struct AcpSession {
     session_id: String,
     pid: u32,
     commands: mpsc::UnboundedSender<Command>,
+    child_commands: mpsc::UnboundedSender<ChildCommand>,
     updates: Mutex<Option<mpsc::UnboundedReceiver<SessionNotification>>>,
     exit: watch::Receiver<Option<ExitStatus>>,
 }
@@ -221,9 +226,6 @@ impl AcpSession {
     /// End the child: `session/cancel`, wait `cancel_grace`, `SIGTERM`, wait
     /// `term_grace`, `SIGKILL`. Returns once the child has been reaped.
     pub async fn terminate(&self, cancel_grace: Duration, term_grace: Duration) {
-        if self.has_exited() {
-            return;
-        }
         self.cancel();
         if tokio::time::timeout(cancel_grace, self.wait())
             .await
@@ -245,16 +247,14 @@ impl AcpSession {
     }
 
     fn signal(&self, sig: libc::c_int) {
-        signal_pid(self.pid, sig);
+        let _ = self.child_commands.send(ChildCommand::Signal(sig));
     }
 }
 
 impl Drop for AcpSession {
     fn drop(&mut self) {
+        self.signal(libc::SIGTERM);
         let _ = self.commands.send(Command::Close);
-        if !self.has_exited() {
-            self.signal(libc::SIGTERM);
-        }
     }
 }
 
@@ -301,11 +301,8 @@ pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
     tokio::spawn(drain_stderr(stderr, pid, stderr_tail.clone()));
 
     let (exit_tx, exit_rx) = watch::channel(None);
-    tokio::spawn(async move {
+    let (child_commands_tx, child_commands_rx) = mpsc::unbounded_channel();
-        let status = child.wait().await.ok();
+    tokio::spawn(supervise_child(child, pid, child_commands_rx, exit_tx));
-        tracing::debug!(pid, ?status, "{ADAPTER_BIN} exited");
-        let _ = exit_tx.send(status);
-    });
 
     let (updates_tx, updates_rx) = mpsc::unbounded_channel();
     let (commands_tx, commands_rx) = mpsc::unbounded_channel();
@@ -327,6 +324,7 @@ pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
                 session_id,
                 pid,
                 commands: commands_tx,
+                child_commands: child_commands_tx,
                 updates: Mutex::new(Some(updates_rx)),
                 exit: exit_rx,
             });
@@ -356,12 +354,12 @@ pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
     };
     // No half-started adapter survives a failed spawn.
     let _ = commands_tx.send(Command::Close);
-    signal_pid(pid, libc::SIGTERM);
+    let _ = child_commands_tx.send(ChildCommand::Signal(libc::SIGTERM));
     if tokio::time::timeout(TERM_GRACE, wait_exit(&exit_rx))
         .await
         .is_err()
     {
-        signal_pid(pid, libc::SIGKILL);
+        let _ = child_commands_tx.send(ChildCommand::Signal(libc::SIGKILL));
         wait_exit(&exit_rx).await;
     }
     Err(failure)
@@ -373,11 +371,39 @@ async fn wait_exit(exit: &watch::Receiver<Option<ExitStatus>>) -> Option<ExitSta
     *status
 }
 
-fn signal_pid(pid: u32, sig: libc::c_int) {
+async fn supervise_child(
-    // ESRCH just means the child is already gone.
+    mut child: tokio::process::Child,
-    // SAFETY: plain syscall on a pid we spawned and have not reaped.
+    pid: u32,
-    unsafe {
+    mut commands: mpsc::UnboundedReceiver<ChildCommand>,
-        libc::kill(pid as libc::pid_t, sig);
+    exit_tx: watch::Sender<Option<ExitStatus>>,
+) {
+    loop {
+        tokio::select! {
+            status = child.wait() => {
+                let status = status.ok();
+                tracing::debug!(pid, ?status, "{ADAPTER_BIN} exited");
+                let _ = exit_tx.send(status);
+                return;
+            }
+            command = commands.recv() => match command {
+                Some(ChildCommand::Signal(sig)) => {
+                    // This task alone owns `child`. It cannot reap the child
+                    // between receiving this command and sending the signal,
+                    // so its PID cannot be reused in that window.
+                    unsafe { libc::kill(pid as libc::pid_t, sig) };
+                }
+                None => {
+                    // The session vanished without a command receiver. This
+                    // task still owns the unreaped child, so this cannot hit
+                    // a reused PID.
+                    unsafe { libc::kill(pid as libc::pid_t, libc::SIGTERM) };
+                    let status = child.wait().await.ok();
+                    tracing::debug!(pid, ?status, "{ADAPTER_BIN} exited");
+                    let _ = exit_tx.send(status);
+                    return;
+                }
+            },
+        }
     }
 }
 

crates/maestro-acp/tests/fake_acp_agent.py

@@ -20,6 +20,7 @@ Environment knobs, all optional:
   FAKE_ACP_EXIT_AFTER_NEW=1  exit 0 right after answering `session/new`
   FAKE_ACP_EXIT_ON_CANCEL=1  exit 0 on `session/cancel` (default: keep running
                              so the SIGTERM path is exercised)
+  FAKE_ACP_IGNORE_EOF=1      keep running after stdin closes, until signalled
   FAKE_ACP_EMIT_ON_CANCEL=1  emit a final notification before handling cancel
   FAKE_ACP_NO_LOAD=1      advertise `loadSession: false`
   FAKE_ACP_LOAD_REPLAY=n  on `session/load`, replay n `agent_message_chunk`
@@ -135,3 +136,7 @@ for raw in sys.stdin:
     elif rid is not None:
         send({"jsonrpc": "2.0", "id": rid,
               "error": {"code": -32601, "message": f"method not found: {method}"}})
+
+if os.environ.get("FAKE_ACP_IGNORE_EOF") == "1":
+    while True:
+        signal.pause()

crates/maestro-acp/tests/spawn.rs

@@ -69,6 +69,24 @@ impl Fake {
         }
         panic!("log never contained {needle:?}:\n{}", self.log());
     }
+
+    fn pid(&self) -> u32 {
+        self.log()
+            .lines()
+            .find_map(|line| line.strip_prefix("start pid="))
+            .and_then(|rest| rest.split(' ').next())
+            .and_then(|pid| pid.parse().ok())
+            .expect("pid in log")
+    }
+
+    async fn wait_for_exit(&self) {
+        let pid = self.pid();
+        let deadline = Instant::now() + Duration::from_secs(5);
+        while pid_alive(pid) && Instant::now() < deadline {
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        }
+        assert!(!pid_alive(pid), "child {pid} still alive");
+    }
 }
 
 fn pid_alive(pid: u32) -> bool {
@@ -274,6 +292,20 @@ async fn child_exit_before_handshake_reports_exited_early() {
     }
 }
 
+#[tokio::test]
+async fn drop_terminates_and_reaps_the_child() {
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut config = fake.config(cwd.path());
+    config.env.insert("FAKE_ACP_IGNORE_EOF".into(), "1".into());
+    let session = maestro_acp::spawn(config).await.expect("spawn");
+
+    drop(session);
+
+    fake.wait_for_log("signal SIGTERM").await;
+    fake.wait_for_exit().await;
+}
+
 fn chunk_text(n: &maestro_acp::schema::SessionNotification) -> String {
     match &n.update {
         SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content {