test(maestro): cover signal shutdown drain

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

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.

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<String, String>,
+}
+
+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<u32> {
+    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::<maestro_protocol::Session>(&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;