From b2b33535ae33c5f9bcecd4bc4109fab5057c88cb Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Fri, 4 Sep 2026 23:42:17 +0100 Subject: [PATCH] fix(maestro): own ACP child signals --- crates/maestro-acp/src/lib.rs | 64 +++++++++++++++------- crates/maestro-acp/tests/fake_acp_agent.py | 5 ++ crates/maestro-acp/tests/spawn.rs | 32 +++++++++++ 3 files changed, 82 insertions(+), 19 deletions(-) diff --git a/crates/maestro-acp/src/lib.rs b/crates/maestro-acp/src/lib.rs index 340426d..277030a 100644 --- a/crates/maestro-acp/src/lib.rs +++ b/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, + child_commands: mpsc::UnboundedSender, updates: Mutex>>, exit: watch::Receiver>, } @@ -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 { tokio::spawn(drain_stderr(stderr, pid, stderr_tail.clone())); let (exit_tx, exit_rx) = watch::channel(None); - tokio::spawn(async move { - let status = child.wait().await.ok(); - tracing::debug!(pid, ?status, "{ADAPTER_BIN} exited"); - let _ = exit_tx.send(status); - }); + let (child_commands_tx, child_commands_rx) = mpsc::unbounded_channel(); + tokio::spawn(supervise_child(child, pid, child_commands_rx, exit_tx)); 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 { 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 { }; // 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, + exit_tx: watch::Sender>, +) { + 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; + } + }, + } } } diff --git a/crates/maestro-acp/tests/fake_acp_agent.py b/crates/maestro-acp/tests/fake_acp_agent.py index 30de64e..03d4c3f 100755 --- a/crates/maestro-acp/tests/fake_acp_agent.py +++ b/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() diff --git a/crates/maestro-acp/tests/spawn.rs b/crates/maestro-acp/tests/spawn.rs index 61775f5..36a8e35 100644 --- a/crates/maestro-acp/tests/spawn.rs +++ b/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 {