From b1718559df2d436cff061d0e83b7ff9c825638d1 Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Fri, 4 Sep 2026 22:02:39 +0100 Subject: [PATCH] feat(maestro): spawn a claude-code-acp child per structured session (#134) A Structured spawn gets no tmux pane: the daemon starts the adapter in the session cwd with the session env (including MAESTRO_SESSION_ID and MAESTRO_URL), runs the handshake, and records the ACP session id on the sidecar. An adapter that cannot come up still yields a session, Dead with activity Error and the reason as activity_detail, so a missing adapter reads as the install command in the sidebar rather than as a session that never appeared. The liveness pass checks a structured session's child instead of the tmux list. kill sends session/cancel, waits, then SIGTERMs through the ACP handle. SessionManager::shutdown ends every adapter concurrently and marks those sessions Dead on disk. take_acp_updates hands #135 the raw session/update stream once, with earlier notifications still queued. --- Cargo.lock | 1 + crates/maestro-daemon/Cargo.toml | 2 + crates/maestro-daemon/src/session.rs | 182 +++++++++++++--- crates/maestro-daemon/tests/acp.rs | 296 +++++++++++++++++++++++++++ 4 files changed, 449 insertions(+), 32 deletions(-) create mode 100644 crates/maestro-daemon/tests/acp.rs diff --git a/Cargo.lock b/Cargo.lock index 4a07411..2fa13de 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1533,6 +1533,7 @@ dependencies = [ "futures-util", "include_dir", "libc", + "maestro-acp", "maestro-config", "maestro-hooks", "maestro-names", diff --git a/crates/maestro-daemon/Cargo.toml b/crates/maestro-daemon/Cargo.toml index 0da4fb6..3187617 100644 --- a/crates/maestro-daemon/Cargo.toml +++ b/crates/maestro-daemon/Cargo.toml @@ -28,6 +28,7 @@ dirs.workspace = true futures-util.workspace = true include_dir = { workspace = true, optional = true } libc.workspace = true +maestro-acp.workspace = true mime_guess = { workspace = true, optional = true } maestro-config.workspace = true maestro-hooks = { workspace = true, default-features = false } @@ -42,6 +43,7 @@ tracing-subscriber.workspace = true uuid.workspace = true [dev-dependencies] +libc.workspace = true reqwest.workspace = true tokio-tungstenite.workspace = true urlencoding = "2" diff --git a/crates/maestro-daemon/src/session.rs b/crates/maestro-daemon/src/session.rs index e8f6a4e..fd0bb09 100644 --- a/crates/maestro-daemon/src/session.rs +++ b/crates/maestro-daemon/src/session.rs @@ -12,9 +12,10 @@ use std::sync::Arc; use std::time::Duration; use chrono::Utc; +use maestro_acp::{AcpSession, SpawnConfig}; use maestro_protocol::{ ActivityState, CatalogEvent, CreateSessionRequest, ErrorInfo, NameSource, Session, - SessionStatus, WorktreeSpec, NAME_KEY, NAME_SOURCE_KEY, + SessionStatus, SessionView, WorktreeSpec, NAME_KEY, NAME_SOURCE_KEY, }; use tokio::sync::{broadcast, Mutex}; use uuid::Uuid; @@ -87,6 +88,12 @@ pub struct SessionRecord { /// glow, this holds through focus loss until the session does new work. /// Daemon-local; never rides the wire. pub awaiting_suppressed: bool, + /// The `claude-code-acp` child behind a [`SessionView::Structured`] + /// session (#134): its connection, pid and the raw notification stream. + /// `None` for terminal sessions and for a structured session whose + /// adapter never came up. Shared so a kill or shutdown can terminate it + /// outside the sessions lock. Daemon-local; never rides the wire. + pub acp: Option>, } #[derive(Debug, thiserror::Error)] @@ -287,6 +294,47 @@ impl SessionManager { sessions.values().map(|r| r.session.clone()).collect() } + /// The raw `session/update` stream of a structured session (#134), + /// handed out once; `None` for a terminal session, a session whose + /// adapter never came up, or a stream already taken. Notifications that + /// arrived before the take are still queued. #135 takes it to feed the + /// per-session event log. + pub async fn take_acp_updates( + &self, + id: Uuid, + ) -> Option> + { + let sessions = self.inner.sessions.lock().await; + sessions.get(&id)?.acp.as_ref()?.take_updates() + } + + /// End every adapter child before the daemon exits (#134): each gets + /// `session/cancel`, then SIGTERM, concurrently. The sessions stay in + /// the map, marked Dead and written through to their sidecars, so what + /// 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) { + let children: Vec<(Uuid, Arc)> = { + let sessions = self.inner.sessions.lock().await; + sessions + .iter() + .filter(|(_, r)| r.session.status == SessionStatus::Running) + .filter_map(|(id, r)| r.acp.clone().map(|acp| (*id, acp))) + .collect() + }; + if children.is_empty() { + return; + } + tracing::info!(count = children.len(), "shutdown: ending adapter children"); + let terminations = children + .iter() + .map(|(_, acp)| acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)); + futures_util::future::join_all(terminations).await; + for (id, _) in children { + mark_session_dead(&self.inner, id).await; + } + } + /// Subscribe to a session's PTY broadcast. `None` if the session is /// unknown to this daemon. pub async fn subscribe_pty(&self, id: Uuid) -> Option>> { @@ -317,7 +365,7 @@ impl SessionManager { // replayed from it renders as garbage until something forces a // repaint. A dead session has no pane left to capture, so it // keeps the byte tail. - let alive = rec.session.status == SessionStatus::Running; + let alive = rec.session.status == SessionStatus::Running && rec.acp.is_none(); // The capture shells out to tmux while the replay lock is held, // which parks the reader for the length of one tmux round-trip. // That only happens on attach, and the alternative is a seam in @@ -427,20 +475,48 @@ impl SessionManager { { session_env.insert("MAESTRO_URL".into(), url); } - let spawned = { - let tmux = self.inner.tmux.clone(); - let cwd = effective_cwd.clone(); - let command = req.command.clone(); - let sink = pipe_pane_command(&fifo_path); - let env = session_env.clone(); - tmux_blocking(move || tmux.new_session_piped(id, &cwd, &command, &env, &sink)).await + // A structured session (#134) gets no tmux pane: the agent is a + // child of the daemon, driven over ACP. An adapter that cannot come + // up still yields a session — Dead, with the reason on it — so the + // operator sees what to fix instead of a silently missing session. + let (acp, acp_failure) = match req.view { + SessionView::Structured => { + match maestro_acp::spawn(SpawnConfig::new( + &effective_cwd, + session_env + .iter() + .map(|(k, v)| (k.clone(), v.clone())) + .collect(), + )) + .await + { + Ok(acp) => (Some(Arc::new(acp)), None), + Err(e) => { + tracing::warn!(%id, error = %e, "structured session: adapter failed to start"); + (None, Some(e.to_string())) + } + } + } + SessionView::Terminal => { + let spawned = { + let tmux = self.inner.tmux.clone(); + let cwd = effective_cwd.clone(); + let command = req.command.clone(); + let sink = pipe_pane_command(&fifo_path); + let env = session_env.clone(); + tmux_blocking(move || tmux.new_session_piped(id, &cwd, &command, &env, &sink)) + .await + }; + if let Err(e) = spawned { + // `bridge` drops at the end of this block, closing the + // FIFO reader. + let _ = std::fs::remove_file(&fifo_path); + self.rollback_worktree(&worktree_base_repo, &worktree_path); + return Err(SessionError::Tmux(e)); + } + (None, None) + } }; - if let Err(e) = spawned { - // `bridge` drops at the end of this block, closing the FIFO reader. - let _ = std::fs::remove_file(&fifo_path); - self.rollback_worktree(&worktree_base_repo, &worktree_path); - return Err(SessionError::Tmux(e)); - } // Capture spawn-history inputs before `req` is partially moved into // `session`. `history_cwd` is the *chosen* working dir (the request @@ -463,13 +539,21 @@ impl SessionManager { base_repo: base_repo.clone(), }, ), - status: SessionStatus::Running, + status: if acp_failure.is_some() { + SessionStatus::Dead + } else { + SessionStatus::Running + }, created_at: Utc::now(), // A freshly-spawned session is producing its initial output; the // activity watcher refines `foreground`/`activity` on its next tick. foreground: None, - activity: ActivityState::Active, - activity_detail: None, + activity: if acp_failure.is_some() { + ActivityState::Error + } else { + ActivityState::Active + }, + activity_detail: acp_failure, last_activity_at: Utc::now(), activity_since: chrono::Utc::now(), @@ -478,15 +562,23 @@ impl SessionManager { // fresh tree has nothing in the way anyway. blocked_work: None, last_focused_at: None, - acp_session_id: None, + acp_session_id: acp.as_ref().map(|a| a.session_id().to_string()), view: req.view, }; if let Err(e) = self.inner.sidecar.write(&session) { - // tmux already spawned the session; kill it so no ghost survives a + // The process already started; end it so no ghost survives a // failed sidecar write, then unwind the FIFO + worktree. - let tmux = self.inner.tmux.clone(); - let _ = tmux_blocking(move || tmux.kill_session(id)).await; + match &acp { + Some(acp) => { + acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE) + .await + } + None => { + let tmux = self.inner.tmux.clone(); + let _ = tmux_blocking(move || tmux.kill_session(id)).await; + } + } let _ = std::fs::remove_file(&fifo_path); self.rollback_worktree(&worktree_base_repo, &worktree_path); return Err(SessionError::Sidecar(e)); @@ -508,6 +600,7 @@ impl SessionManager { naming: false, focused_since: None, awaiting_suppressed: false, + acp, }, ); } @@ -577,7 +670,9 @@ impl SessionManager { .worktree_base_repo .clone() .zip(record.worktree_path.clone()), - record.session.status == SessionStatus::Running, + // A structured session has no pane to kill; its child is + // ended through the ACP handle in `cleanup_record`. + record.session.status == SessionStatus::Running && record.acp.is_none(), ), } }; @@ -720,6 +815,9 @@ impl SessionManager { naming: false, focused_since: None, awaiting_suppressed: false, + // A structured session's adapter died with the previous + // daemon; only tmux-backed sessions can be adopted. + acp: None, }, ); let _ = self @@ -1089,21 +1187,36 @@ async fn run_liveness_pass(inner: &Arc) { // be competing to start the tmux server). // Only sessions we still believe are running can transition; a session // already marked Dead must not re-emit an update on every 2s tick. - let known: Vec = { + // A structured session (#134) is not in tmux at all: its liveness is its + // adapter child, checked through the ACP handle instead. + let (tmux_backed, acp_backed): (Vec, Vec<(Uuid, Arc)>) = { let sessions = inner.sessions.lock().await; if sessions.is_empty() { return; } - sessions + let running = sessions .iter() - .filter(|(_, r)| r.session.status == SessionStatus::Running) - .map(|(id, _)| *id) - .collect() + .filter(|(_, r)| r.session.status == SessionStatus::Running); + let mut tmux_backed = Vec::new(); + let mut acp_backed = Vec::new(); + for (id, record) in running { + match &record.acp { + Some(acp) => acp_backed.push((*id, acp.clone())), + None => tmux_backed.push(*id), + } + } + (tmux_backed, acp_backed) }; - if known.is_empty() { - return; + + for (id, acp) in acp_backed { + if acp.has_exited() { + mark_session_dead(inner, id).await; + } } + if tmux_backed.is_empty() { + return; + } let live_ids = { let tmux = inner.tmux.clone(); match tmux_blocking(move || tmux.list_sessions()).await { @@ -1116,7 +1229,7 @@ async fn run_liveness_pass(inner: &Arc) { }; let live: HashSet = live_ids.into_iter().collect(); - for id in known { + for id in tmux_backed { if !live.contains(&id) { mark_session_dead(inner, id).await; } @@ -1322,7 +1435,7 @@ async fn mark_session_dead(inner: &Arc, id: Uuid) { let Some((mut session, worktree)) = flipped else { return; }; - tracing::info!(%id, "session vanished from tmux — marking dead"); + tracing::info!(%id, "session process gone — marking dead"); if let Some(wt) = &worktree { if settle_worktree_on_exit(inner, id, wt).await { @@ -1424,6 +1537,11 @@ async fn cleanup_record( let tmux = inner.tmux.clone(); let _ = tmux_blocking(move || tmux.kill_session(id)).await; } + if let Some(acp) = &record.acp { + // `session/cancel`, a short wait, then SIGTERM (#134). + acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE) + .await; + } let _ = std::fs::remove_file(&record.fifo_path); if let Err(e) = inner.sidecar.delete(id) { tracing::warn!(error = ?e, %id, "cleanup: failed to delete sidecar"); diff --git a/crates/maestro-daemon/tests/acp.rs b/crates/maestro-daemon/tests/acp.rs new file mode 100644 index 0000000..18c3884 --- /dev/null +++ b/crates/maestro-daemon/tests/acp.rs @@ -0,0 +1,296 @@ +//! Structured sessions (#134): a `claude-code-acp` child per session instead +//! of a tmux pane. Runs against the fake adapter in +//! `crates/maestro-acp/tests/fake_acp_agent.py`; nothing here needs Claude. + +mod common; + +use std::collections::HashMap; +use std::path::{Path, PathBuf}; +use std::time::{Duration, Instant}; + +use common::{wait_for_async, DaemonFixture}; +use maestro_acp::{ADAPTER_BIN, INSTALL_COMMAND}; +use maestro_protocol::{ActivityState, CreateSessionRequest, SessionStatus, SessionView}; +use tempfile::TempDir; + +fn fake_agent() -> PathBuf { + Path::new(env!("CARGO_MANIFEST_DIR")).join("../maestro-acp/tests/fake_acp_agent.py") +} + +/// The fake adapter installed as `claude-code-acp` in a temp `bin/`, with +/// the log file it appends to. `env()` is what a spawn request carries. +struct Fake { + dir: TempDir, +} + +impl Fake { + fn new() -> Self { + let dir = TempDir::new().unwrap(); + let bin = dir.path().join("bin"); + std::fs::create_dir_all(&bin).unwrap(); + std::fs::copy(fake_agent(), bin.join(ADAPTER_BIN)).unwrap(); + Self { dir } + } + + fn log_path(&self) -> PathBuf { + self.dir.path().join("agent.log") + } + + fn log(&self) -> String { + std::fs::read_to_string(self.log_path()).unwrap_or_default() + } + + /// Fake `bin/` first so it shadows any real adapter, then the process + /// `PATH` so the script's `python3` still resolves. + fn env(&self) -> HashMap { + let path = format!( + "{}:{}", + self.dir.path().join("bin").display(), + std::env::var("PATH").unwrap_or_default() + ); + HashMap::from([ + ("PATH".to_string(), path), + ( + "FAKE_ACP_LOG".to_string(), + self.log_path().to_string_lossy().into_owned(), + ), + ]) + } + + /// The pid the fake logged at start, once it has. + fn pid(&self) -> u32 { + self.log() + .lines() + .find_map(|l| l.strip_prefix("start pid=")) + .and_then(|rest| rest.split(' ').next()) + .and_then(|p| p.parse().ok()) + .expect("pid in log") + } +} + +fn structured(cwd: &Path, env: HashMap) -> CreateSessionRequest { + CreateSessionRequest { + view: SessionView::Structured, + cwd: cwd.to_path_buf(), + command: "claude".into(), + env, + metadata: HashMap::new(), + worktree: None, + } +} + +fn pid_alive(pid: u32) -> bool { + std::fs::read_to_string(format!("/proc/{pid}/stat")) + .map(|s| !s.contains(") Z ")) + .unwrap_or(false) +} + +async fn wait_for_exit(pid: u32) { + let deadline = Instant::now() + Duration::from_secs(5); + while pid_alive(pid) && Instant::now() < deadline { + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert!(!pid_alive(pid), "pid {pid} still alive"); +} + +#[tokio::test] +async fn structured_spawn_starts_the_adapter_in_cwd_with_session_env() { + let fx = DaemonFixture::new(); + let fake = Fake::new(); + let cwd = TempDir::new().unwrap(); + let mut env = fake.env(); + env.insert("FAKE_ACP_ECHO_ENV".into(), "MAESTRO_SESSION_ID".into()); + + let session = fx + .manager + .create(structured(cwd.path(), env)) + .await + .expect("create"); + + assert_eq!(session.view, SessionView::Structured); + assert_eq!(session.status, SessionStatus::Running); + let acp_id = session.acp_session_id.clone().expect("acp session id"); + assert!(acp_id.starts_with("fake-"), "{acp_id}"); + + // The child ran in the session cwd with the session's env — including + // the identity the daemon injects, so the agent knows which session it + // is (#103). + let log = fake.log(); + let cwd_canon = cwd.path().canonicalize().unwrap(); + assert!( + log.contains(&format!("cwd={}", cwd_canon.display())), + "{log}" + ); + assert!( + log.contains(&format!("env MAESTRO_SESSION_ID={}", session.id)), + "{log}" + ); + + // No tmux pane was created for it: the agent is the daemon's child. + let panes = fx.tmux.tmux.list_sessions().expect("list"); + assert!( + !panes.contains(&session.id), + "structured session got a tmux pane" + ); + + // The sidecar carries the id a respawn will pass to session/load (#135). + let sidecar: maestro_protocol::Session = serde_json::from_slice( + &std::fs::read(fx.sidecar_dir().join(format!("{}.json", session.id))).unwrap(), + ) + .unwrap(); + assert_eq!(sidecar.acp_session_id.as_deref(), Some(acp_id.as_str())); + + // The raw notification stream is reachable for #135, handed out once. + let mut updates = fx + .manager + .take_acp_updates(session.id) + .await + .expect("updates receiver"); + assert!(fx.manager.take_acp_updates(session.id).await.is_none()); + let first = tokio::time::timeout(Duration::from_secs(5), updates.recv()) + .await + .expect("an update within 5s") + .expect("channel open"); + assert_eq!(first.session_id.0.as_ref(), acp_id); + + fx.manager.kill(session.id, false).await.expect("kill"); +} + +#[tokio::test] +async fn missing_adapter_yields_a_failed_session_naming_the_install_command() { + let fx = DaemonFixture::new(); + let empty = TempDir::new().unwrap(); + let cwd = TempDir::new().unwrap(); + let env = HashMap::from([( + "PATH".to_string(), + empty.path().to_string_lossy().into_owned(), + )]); + + let session = fx + .manager + .create(structured(cwd.path(), env)) + .await + .expect("a failed session, not an error"); + + assert_eq!(session.status, SessionStatus::Dead); + assert_eq!(session.activity, ActivityState::Error); + let detail = session.activity_detail.clone().expect("actionable detail"); + assert!(detail.contains(ADAPTER_BIN), "{detail}"); + assert!(detail.contains(INSTALL_COMMAND), "{detail}"); + assert_eq!(session.acp_session_id, None); + + // It is a real session: listed, on disk, and dismissable like any other + // dead one. + assert!(fx.manager.has(session.id).await); + assert!(fx + .sidecar_dir() + .join(format!("{}.json", session.id)) + .exists()); + fx.manager.kill(session.id, false).await.expect("dismiss"); + assert!(!fx.manager.has(session.id).await); +} + +#[tokio::test] +async fn kill_sends_cancel_then_sigterm_and_removes_the_sidecar() { + let fx = DaemonFixture::new(); + let fake = Fake::new(); + let cwd = TempDir::new().unwrap(); + let mut events = fx.manager.events(); + + let session = fx + .manager + .create(structured(cwd.path(), fake.env())) + .await + .expect("create"); + let pid = fake.pid(); + assert!(pid_alive(pid)); + + fx.manager.kill(session.id, false).await.expect("kill"); + + wait_for_exit(pid).await; + let log = fake.log(); + let cancel = log.find("recv session/cancel").expect("cancel logged"); + let term = log.find("signal SIGTERM").expect("SIGTERM logged"); + assert!(cancel < term, "cancel must precede SIGTERM:\n{log}"); + assert!(!fx.manager.has(session.id).await); + assert!(!fx + .sidecar_dir() + .join(format!("{}.json", session.id)) + .exists()); + + let mut removed = false; + while let Ok(ev) = events.try_recv() { + if matches!(ev, maestro_protocol::CatalogEvent::SessionRemoved { id } if id == session.id) { + removed = true; + } + } + assert!(removed, "SessionRemoved never broadcast"); +} + +#[tokio::test] +async fn shutdown_reaps_every_adapter_child() { + let fx = DaemonFixture::new(); + let cwd = TempDir::new().unwrap(); + let a = Fake::new(); + let b = Fake::new(); + + let sa = fx + .manager + .create(structured(cwd.path(), a.env())) + .await + .expect("create a"); + let sb = fx + .manager + .create(structured(cwd.path(), b.env())) + .await + .expect("create b"); + let (pa, pb) = (a.pid(), b.pid()); + assert!(pid_alive(pa) && pid_alive(pb)); + + fx.manager.shutdown().await; + + wait_for_exit(pa).await; + wait_for_exit(pb).await; + assert!(a.log().contains("signal SIGTERM"), "{}", a.log()); + assert!(b.log().contains("signal SIGTERM"), "{}", b.log()); + // Still listed, now Dead: the sidecar is truthful for whoever reads it + // after the restart (#135). + assert_eq!(fx.manager.status(sa.id).await, Some(SessionStatus::Dead)); + assert_eq!(fx.manager.status(sb.id).await, Some(SessionStatus::Dead)); + let sidecar: maestro_protocol::Session = serde_json::from_slice( + &std::fs::read(fx.sidecar_dir().join(format!("{}.json", sa.id))).unwrap(), + ) + .unwrap(); + assert_eq!(sidecar.status, SessionStatus::Dead); +} + +#[tokio::test] +async fn adapter_exit_marks_the_session_dead_and_keeps_it_listed() { + let fx = DaemonFixture::new(); + let fake = Fake::new(); + let cwd = TempDir::new().unwrap(); + + let session = fx + .manager + .create(structured(cwd.path(), fake.env())) + .await + .expect("create"); + fx.manager.start_liveness_watcher(Duration::from_millis(50)); + + // The agent dies out from under the daemon: the liveness pass, not tmux, + // is what notices for a structured session. + let pid = fake.pid(); + unsafe { + libc::kill(pid as libc::pid_t, libc::SIGTERM); + } + wait_for_exit(pid).await; + + wait_for_async(Duration::from_secs(3), || async { + fx.manager.status(session.id).await == Some(SessionStatus::Dead) + }) + .await; + assert!(fx.manager.has(session.id).await); + + fx.manager.kill(session.id, false).await.expect("dismiss"); + assert!(!fx.manager.has(session.id).await); +}