fix(maestro): serialize structured shutdown

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/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

@@ -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 {
@@ -198,6 +201,9 @@ struct Inner {
     /// 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 {
@@ -241,6 +247,7 @@ impl SessionManager {
                 viewports: Mutex::new(Viewports::new()),
                 next_viewport: AtomicU64::new(0),
                 shutting_down: AtomicBool::new(false),
+                structured_lifecycle: Mutex::new(()),
             }),
         }
     }
@@ -356,6 +363,7 @@ impl SessionManager {
         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
@@ -465,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 {

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();