From cf322746a7da45f5f7fad29bb655f9ceeb0b7176 Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Sat, 5 Sep 2026 00:22:59 +0100 Subject: [PATCH] fix(maestro): serialize structured shutdown --- crates/maestro-acp/tests/fake_acp_agent.py | 4 ++ crates/maestro-daemon/src/local_api.rs | 1 + crates/maestro-daemon/src/session.rs | 19 ++++++++++ crates/maestro-daemon/tests/acp.rs | 44 ++++++++++++++++++++++ 4 files changed, 68 insertions(+) diff --git a/crates/maestro-acp/tests/fake_acp_agent.py b/crates/maestro-acp/tests/fake_acp_agent.py index 03d4c3f..b1a5a16 100755 --- a/crates/maestro-acp/tests/fake_acp_agent.py +++ b/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": diff --git a/crates/maestro-daemon/src/local_api.rs b/crates/maestro-daemon/src/local_api.rs index 173e932..fc5f4e2 100644 --- a/crates/maestro-daemon/src/local_api.rs +++ b/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, } } diff --git a/crates/maestro-daemon/src/session.rs b/crates/maestro-daemon/src/session.rs index 384192e..4694719 100644 --- a/crates/maestro-daemon/src/session.rs +++ b/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, Option>)> = { 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 { + 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 { let (effective_cwd, worktree_path, worktree_base_repo) = match &req.worktree { Some(spec) => { let WorktreeSpec { diff --git a/crates/maestro-daemon/tests/acp.rs b/crates/maestro-daemon/tests/acp.rs index 6af36cb..1f530a5 100644 --- a/crates/maestro-daemon/tests/acp.rs +++ b/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();