fix(maestro): serialize structured shutdown

Cargo.lock

@@ -1558,6 +1558,7 @@ version = "0.1.0"
 dependencies = [
  "chrono",
  "futures-util",
+ "libc",
  "maestro-protocol",
  "maestro-testing",
  "maestro-tui",

crates/maestro-acp/tests/fake_acp_agent.py

@@ -17,6 +17,8 @@ Environment knobs, all optional:
                             signal SIGTERM
   FAKE_ACP_ECHO_ENV       name of one env var to log at start
   FAKE_ACP_INIT_FAIL=1    answer `initialize` with a JSON-RPC error
+  FAKE_ACP_DELAY_SESSION_NEW_MS=n  wait n milliseconds before answering
+                             `session/new`
   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)
@@ -36,6 +38,7 @@ import json
 import os
 import signal
 import sys
+import time
 
 LOG = os.environ.get("FAKE_ACP_LOG")
 
@@ -113,6 +116,7 @@ for raw in sys.stdin:
             "authMethods": [],
         }})
     elif method == "session/new":
+        time.sleep(int(os.environ.get("FAKE_ACP_DELAY_SESSION_NEW_MS", "0")) / 1000)
         send({"jsonrpc": "2.0", "id": rid, "result": {"sessionId": session_id}})
         send(chunk("hello from fake"))
         if os.environ.get("FAKE_ACP_EXIT_AFTER_NEW") == "1":

crates/maestro-daemon/src/cli.rs

@@ -122,21 +122,23 @@ async fn run(config_override: Option<PathBuf>) -> Result<(), Box<dyn Error>> {
     // shutdown, and both must reach the adapter children below (#134). The
     // unit's `KillMode=process` signals only the daemon, so nothing else
     // would end them.
-    let shutdown = async {
+    let shutdown_manager = manager.clone();
+    let shutdown = async move {
         let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
             .expect("install SIGTERM handler");
         tokio::select! {
             _ = tokio::signal::ctrl_c() => tracing::info!("ctrl-c received — shutting down"),
             _ = sigterm.recv() => tracing::info!("SIGTERM received — shutting down"),
         }
+        // Axum waits for upgraded WebSockets to drain. Structured children
+        // must be settled before that wait, while the signal path still owns
+        // the shutdown order (#179).
+        shutdown_manager.shutdown().await;
     };
 
     axum::serve(listener, router)
         .tcp_nodelay(true)
         .with_graceful_shutdown(shutdown)
         .await?;
-    // tmux sessions outlive the daemon by design; structured sessions'
-    // adapter children do not (#134).
-    manager.shutdown().await;
     Ok(())
 }

crates/maestro-daemon/src/local_api.rs

@@ -170,6 +170,7 @@ fn session_error_status(code: &str) -> StatusCode {
         // The session outlived its pane: nothing to type into, and nothing
         // the daemon can do about it.
         "not_running" => StatusCode::CONFLICT,
+        "shutting_down" => StatusCode::SERVICE_UNAVAILABLE,
         _ => StatusCode::BAD_GATEWAY,
     }
 }

crates/maestro-daemon/src/session.rs

@@ -7,7 +7,7 @@
 
 use std::collections::{HashMap, HashSet};
 use std::path::PathBuf;
-use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -127,6 +127,8 @@ pub enum SessionError {
     /// a stray keystroke on an attach socket, which is dropped.
     #[error("session is no longer running: {0}")]
     NotRunning(Uuid),
+    #[error("daemon is shutting down")]
+    ShuttingDown,
     /// The kill was refused (#79): the session's worktree holds unsaved work.
     /// Nothing happened — the session keeps running and the tree is untouched,
     /// so it stays the operator's handle on their own unfinished work.
@@ -144,6 +146,7 @@ impl SessionError {
             SessionError::NotFound(_) => "not_found",
             SessionError::InvalidEnv(_) => "invalid_env",
             SessionError::NotRunning(_) => "not_running",
+            SessionError::ShuttingDown => "shutting_down",
             SessionError::KillGuarded(_) => "kill_guarded",
         };
         let unsaved_work = match self {
@@ -195,6 +198,12 @@ struct Inner {
     /// client that reconnects never collides with the registration its
     /// previous socket has not finished releasing.
     next_viewport: AtomicU64,
+    /// Once daemon shutdown starts, it owns structured-session finalization.
+    /// Background passes must not classify those deliberate exits as natural.
+    shutting_down: AtomicBool,
+    /// Serializes structured creation with shutdown so every adapter child is
+    /// either rejected or included in shutdown's final drain.
+    structured_lifecycle: Mutex<()>,
 }
 
 impl SessionManager {
@@ -237,6 +246,8 @@ impl SessionManager {
                 events,
                 viewports: Mutex::new(Viewports::new()),
                 next_viewport: AtomicU64::new(0),
+                shutting_down: AtomicBool::new(false),
+                structured_lifecycle: Mutex::new(()),
             }),
         }
     }
@@ -349,13 +360,18 @@ impl SessionManager {
     /// is on disk after the restart is truthful. tmux sessions are left
     /// alone — surviving a restart is what they are for.
     pub async fn shutdown(&self) {
+        if self.inner.shutting_down.swap(true, Ordering::AcqRel) {
+            return;
+        }
+        let _lifecycle = self.inner.structured_lifecycle.lock().await;
         let children: Vec<(Uuid, Arc<AcpSession>, Option<watch::Receiver<bool>>)> = {
             let sessions = self.inner.sessions.lock().await;
             sessions
                 .iter()
                 .filter_map(|(id, r)| {
-                    r.acp
-                        .clone()
+                    (r.session.status == SessionStatus::Running)
+                        .then_some(())
+                        .and_then(|()| r.acp.clone())
                         .map(|acp| (*id, acp, r.acp_event_pump.clone()))
                 })
                 .collect()
@@ -457,6 +473,17 @@ impl SessionManager {
 
     /// Spawn a new session. On success returns the freshly-created `Session`.
     pub async fn create(&self, req: CreateSessionRequest) -> Result<Session, SessionError> {
+        if req.view == SessionView::Structured {
+            let _lifecycle = self.inner.structured_lifecycle.lock().await;
+            if self.inner.shutting_down.load(Ordering::Acquire) {
+                return Err(SessionError::ShuttingDown);
+            }
+            return self.create_inner(req).await;
+        }
+        self.create_inner(req).await
+    }
+
+    async fn create_inner(&self, req: CreateSessionRequest) -> Result<Session, SessionError> {
         let (effective_cwd, worktree_path, worktree_base_repo) = match &req.worktree {
             Some(spec) => {
                 let WorktreeSpec {
@@ -1438,6 +1465,9 @@ async fn probe_worktree(
 /// One pass of the liveness watcher: diff the in-memory session set against
 /// the tmux-reported one, and reap anything we still track but tmux has lost.
 async fn run_liveness_pass(inner: &Arc<Inner>) {
+    if inner.shutting_down.load(Ordering::Acquire) {
+        return;
+    }
     // Snapshot the in-memory ids first; if we own no sessions there is
     // nothing to reap and no reason to spawn a tmux subprocess (this also
     // avoids racing with the very first `new-session` that would otherwise
@@ -1497,6 +1527,9 @@ async fn run_liveness_pass(inner: &Arc<Inner>) {
 /// `SessionUpdated` only when `activity` or `foreground` actually changes — so a
 /// steady state produces no diffs. Runs on the existing liveness tick.
 async fn run_activity_pass(inner: &Arc<Inner>) {
+    if inner.shutting_down.load(Ordering::Acquire) {
+        return;
+    }
     let ids: Vec<Uuid> = {
         let sessions = inner.sessions.lock().await;
         if sessions.is_empty() {
@@ -1539,6 +1572,9 @@ async fn run_activity_pass(inner: &Arc<Inner>) {
 
         let updated = {
             let mut sessions = inner.sessions.lock().await;
+            if inner.shutting_down.load(Ordering::Acquire) {
+                return;
+            }
             let Some(rec) = sessions.get_mut(&id) else {
                 continue;
             };
@@ -1613,6 +1649,9 @@ async fn run_activity_pass(inner: &Arc<Inner>) {
 /// error must not read as "nothing blocks". Like the activity pass, this only
 /// emits `SessionUpdated` when the value actually changed.
 async fn run_blocked_work_pass(inner: &Arc<Inner>) {
+    if inner.shutting_down.load(Ordering::Acquire) {
+        return;
+    }
     let targets: Vec<(Uuid, (PathBuf, PathBuf))> = {
         let sessions = inner.sessions.lock().await;
         sessions
@@ -1640,6 +1679,9 @@ async fn run_blocked_work_pass(inner: &Arc<Inner>) {
         };
         let updated = {
             let mut sessions = inner.sessions.lock().await;
+            if inner.shutting_down.load(Ordering::Acquire) {
+                return;
+            }
             let Some(rec) = sessions.get_mut(&id) else {
                 continue;
             };
@@ -1672,6 +1714,9 @@ async fn run_blocked_work_pass(inner: &Arc<Inner>) {
 /// [`SessionManager::recover`] adopts only what tmux still holds. Structured
 /// sessions stopped for daemon shutdown retain their sidecars for ACP recovery.
 async fn mark_session_dead(inner: &Arc<Inner>, id: Uuid) {
+    if inner.shutting_down.load(Ordering::Acquire) {
+        return;
+    }
     mark_session_dead_with_worktree_settlement(inner, id, true).await;
 }
 

crates/maestro-daemon/tests/acp.rs

@@ -498,6 +498,50 @@ async fn shutdown_reaps_every_adapter_child() {
     assert_eq!(sidecar.status, SessionStatus::Dead);
 }
 
+#[tokio::test]
+async fn shutdown_waits_for_structured_creation_and_rejects_new_creates() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_DELAY_SESSION_NEW_MS".into(), "500".into());
+
+    let create_cwd = cwd.path().to_path_buf();
+    let manager = fx.manager.clone();
+    let create = tokio::spawn(async move { manager.create(structured(&create_cwd, env)).await });
+    wait_for_async(Duration::from_secs(5), || async {
+        fake.log().contains("recv session/new")
+    })
+    .await;
+
+    let manager = fx.manager.clone();
+    let shutdown = tokio::spawn(async move { manager.shutdown().await });
+    tokio::time::sleep(Duration::from_millis(50)).await;
+    assert!(
+        !shutdown.is_finished(),
+        "shutdown returned before create finished"
+    );
+
+    let session = create.await.expect("create task").expect("create");
+    let pid = fake.pid();
+    shutdown.await.expect("shutdown task");
+
+    wait_for_exit(pid).await;
+    assert!(fake.log().contains("signal SIGTERM"), "{}", fake.log());
+    assert_eq!(
+        fx.manager.status(session.id).await,
+        Some(SessionStatus::Dead)
+    );
+    assert!(matches!(
+        fx.manager.create(structured(cwd.path(), fake.env())).await,
+        Err(maestro_daemon::SessionError::ShuttingDown)
+    ));
+
+    // A repeated signal must not attempt to terminate the child again.
+    fx.manager.shutdown().await;
+    assert_eq!(fake.log().matches("signal SIGTERM").count(), 1);
+}
+
 #[tokio::test]
 async fn shutdown_drains_final_notifications_to_the_log() {
     let fx = DaemonFixture::new();

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;