diff --git a/Cargo.lock b/Cargo.lock index 2fa13de..75e72fc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1558,6 +1558,7 @@ version = "0.1.0" dependencies = [ "chrono", "futures-util", + "libc", "maestro-protocol", "maestro-testing", "maestro-tui", diff --git a/crates/maestro-e2e/Cargo.toml b/crates/maestro-e2e/Cargo.toml index 56a4d21..899bb82 100644 --- a/crates/maestro-e2e/Cargo.toml +++ b/crates/maestro-e2e/Cargo.toml @@ -13,6 +13,7 @@ path = "src/lib.rs" [dependencies] chrono.workspace = true futures-util.workspace = true +libc.workspace = true maestro-protocol = { workspace = true, features = ["client"] } maestro-testing.workspace = true serde.workspace = true diff --git a/crates/maestro-e2e/src/daemon.rs b/crates/maestro-e2e/src/daemon.rs index ca0e1b9..c7925ec 100644 --- a/crates/maestro-e2e/src/daemon.rs +++ b/crates/maestro-e2e/src/daemon.rs @@ -87,6 +87,14 @@ impl Daemon { } } + /// Send the daemon the same SIGTERM systemd uses, without waiting for its + /// graceful HTTP shutdown to finish. + pub fn send_sigterm(&self) { + let child = self.child.as_ref().expect("daemon is running"); + let result = unsafe { libc::kill(child.id() as libc::pid_t, libc::SIGTERM) }; + assert_eq!(result, 0, "send SIGTERM to daemon"); + } + /// Re-spawn the daemon on the same port + state. The previous tmux /// server (under `TMUX_TMPDIR`) is left running so `recover()` can adopt /// its sessions. diff --git a/crates/maestro-e2e/tests/e2e.rs b/crates/maestro-e2e/tests/e2e.rs index 675bb57..fe39520 100644 --- a/crates/maestro-e2e/tests/e2e.rs +++ b/crates/maestro-e2e/tests/e2e.rs @@ -15,7 +15,7 @@ use std::time::{Duration, Instant}; use futures_util::{SinkExt, StreamExt}; use maestro_e2e::util::wait_for_async; use maestro_e2e::{Daemon, TestClient}; -use maestro_protocol::{CreateSessionRequest, WorktreeSpec}; +use maestro_protocol::{CreateSessionRequest, SessionStatus, SessionView, WorktreeSpec}; use tempfile::TempDir; use tokio_tungstenite::tungstenite::Message as WsMessage; @@ -36,6 +36,61 @@ fn create_req(cwd: &std::path::Path, command: &str) -> CreateSessionRequest { } } +struct FakeAcp { + _dir: TempDir, + env: HashMap, +} + +fn fake_acp(log: &std::path::Path) -> FakeAcp { + let dir = tempfile::tempdir().expect("fake adapter dir"); + let adapter = dir.path().join("claude-code-acp"); + std::fs::copy( + concat!( + env!("CARGO_MANIFEST_DIR"), + "/../maestro-acp/tests/fake_acp_agent.py" + ), + &adapter, + ) + .expect("copy fake adapter"); + let mut permissions = std::fs::metadata(&adapter) + .expect("adapter metadata") + .permissions(); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + permissions.set_mode(0o755); + } + std::fs::set_permissions(&adapter, permissions).expect("make fake adapter executable"); + let env = HashMap::from([ + ( + "PATH".into(), + format!( + "{}:{}", + dir.path().display(), + std::env::var("PATH").unwrap_or_default() + ), + ), + ("FAKE_ACP_LOG".into(), log.to_string_lossy().into_owned()), + ]); + FakeAcp { _dir: dir, env } +} + +fn fake_acp_pid(log: &std::path::Path) -> Option { + std::fs::read_to_string(log).ok()?.lines().find_map(|line| { + line.strip_prefix("start pid=")? + .split(' ') + .next()? + .parse() + .ok() + }) +} + +fn pid_alive(pid: u32) -> bool { + std::fs::read_to_string(format!("/proc/{pid}/stat")) + .map(|stat| !stat.contains(") Z ")) + .unwrap_or(false) +} + /// A worktree spec built through its wire shape. The daemon and clients ship /// together (#87 dropped `cleanup_on_kill` from the struct), so the JSON /// object carries only the fields the lifecycle actually honours. @@ -995,6 +1050,66 @@ async fn daemon_restart_recovers_session() { client.kill_session(s.id).await.expect("kill"); } +#[tokio::test(flavor = "multi_thread")] +async fn sigterm_stops_structured_child_before_events_ws_drains() { + let (mut daemon, client) = bring_up().await; + let cwd = TempDir::new().unwrap(); + let log = cwd.path().join("fake-acp.log"); + let fake = fake_acp(&log); + let session = client + .spawn_session(&CreateSessionRequest { + view: SessionView::Structured, + cwd: cwd.path().to_path_buf(), + command: "claude".into(), + env: fake.env.clone(), + metadata: HashMap::new(), + worktree: None, + }) + .await + .expect("spawn structured session"); + let ws_url = daemon.client().events_url().expect("events URL"); + let (_events_ws, _) = tokio_tungstenite::connect_async(ws_url) + .await + .expect("events WS handshake"); + wait_for_async("fake ACP starts", Duration::from_secs(5), || { + let log = log.clone(); + async move { fake_acp_pid(&log).is_some() } + }) + .await; + let pid = fake_acp_pid(&log).expect("fake ACP pid"); + + daemon.send_sigterm(); + + wait_for_async( + "fake ACP exits after SIGTERM", + Duration::from_secs(5), + || async { !pid_alive(pid) }, + ) + .await; + let sidecar_path = daemon + .state_dir() + .join("sessions") + .join(format!("{}.json", session.id)); + wait_for_async( + "dead structured session sidecar", + Duration::from_secs(5), + || { + let sidecar_path = sidecar_path.clone(); + async move { + std::fs::read(&sidecar_path) + .ok() + .and_then(|json| { + serde_json::from_slice::(&json).ok() + }) + .is_some_and(|saved| saved.status == SessionStatus::Dead) + } + }, + ) + .await; + + daemon.stop(); +} + #[tokio::test(flavor = "multi_thread")] async fn concurrent_spawns_each_get_unique_ids() { let (_daemon, client) = bring_up().await;