diff --git a/crates/maestro-acp/tests/fake_acp_agent.py b/crates/maestro-acp/tests/fake_acp_agent.py index fd8d2ad..30de64e 100755 --- a/crates/maestro-acp/tests/fake_acp_agent.py +++ b/crates/maestro-acp/tests/fake_acp_agent.py @@ -20,6 +20,7 @@ Environment knobs, all optional: 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) + FAKE_ACP_EMIT_ON_CANCEL=1 emit a final notification before handling cancel FAKE_ACP_NO_LOAD=1 advertise `loadSession: false` FAKE_ACP_LOAD_REPLAY=n on `session/load`, replay n `agent_message_chunk` notifications ("replayed ") before answering, @@ -127,6 +128,8 @@ for raw in sys.stdin: send({"jsonrpc": "2.0", "id": rid, "result": {}}) send(chunk("hello after load")) elif method == "session/cancel": + if os.environ.get("FAKE_ACP_EMIT_ON_CANCEL") == "1": + send(chunk("final on cancel")) if os.environ.get("FAKE_ACP_EXIT_ON_CANCEL") == "1": sys.exit(0) elif rid is not None: diff --git a/crates/maestro-daemon/src/session.rs b/crates/maestro-daemon/src/session.rs index 47cea7a..00e9bfb 100644 --- a/crates/maestro-daemon/src/session.rs +++ b/crates/maestro-daemon/src/session.rs @@ -17,7 +17,7 @@ use maestro_protocol::{ ActivityState, CatalogEvent, CreateSessionRequest, ErrorInfo, NameSource, Session, SessionStatus, SessionView, WorktreeSpec, NAME_KEY, NAME_SOURCE_KEY, }; -use tokio::sync::{broadcast, Mutex}; +use tokio::sync::{broadcast, watch, Mutex}; use uuid::Uuid; use crate::panes::Panes; @@ -94,9 +94,18 @@ pub struct SessionRecord { /// 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>, - /// Live notifications after they have been appended to the event log. - /// `None` for terminal sessions and failed structured spawns. - pub acp_events: Option>, + /// Open log plus its post-append live sender. Their shared lock is the + /// replay/live seam used by the structured WebSocket in #138. + acp_events: Option>>, + /// Completion signal for the task draining ACP notifications into the + /// event log. Shutdown waits for it before returning. + acp_event_pump: Option>, +} + +#[derive(Debug)] +struct AcpEventStream { + log: EventLog, + live: broadcast::Sender, } #[derive(Debug, thiserror::Error)] @@ -297,11 +306,28 @@ impl SessionManager { sessions.values().map(|r| r.session.clone()).collect() } - /// Subscribe to live structured-session events. Every event returned here - /// has already been appended to the session's durable log. - pub async fn subscribe_acp_events(&self, id: Uuid) -> Option> { - let sessions = self.inner.sessions.lock().await; - Some(sessions.get(&id)?.acp_events.as_ref()?.subscribe()) + /// Read replay and acquire its live continuation under the same lock. + /// Events are therefore present exactly once across the returned seam. + pub async fn subscribe_acp_events_with_replay( + &self, + id: Uuid, + cursor: EventCursor, + limit: usize, + ) -> Result, broadcast::Receiver)>, SessionError> { + let stream = { + let sessions = self.inner.sessions.lock().await; + let Some(record) = sessions.get(&id) else { + return Ok(None); + }; + let Some(stream) = &record.acp_events else { + return Ok(None); + }; + stream.clone() + }; + let stream = stream.lock().await; + let live = stream.live.subscribe(); + let replay = self.inner.sidecar.read_events(id, cursor, limit)?; + Ok(Some((replay, live))) } /// Read a cursor-bounded slice of the durable structured-session log. @@ -323,12 +349,15 @@ 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) { - let children: Vec<(Uuid, Arc)> = { + let children: Vec<(Uuid, Arc, Option>)> = { 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))) + .filter_map(|(id, r)| { + r.acp + .clone() + .map(|acp| (*id, acp, r.acp_event_pump.clone())) + }) .collect() }; if children.is_empty() { @@ -337,9 +366,14 @@ impl SessionManager { 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)); + .map(|(_, acp, _)| acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)); futures_util::future::join_all(terminations).await; - for (id, _) in children { + let drains = children + .iter() + .filter_map(|(_, _, done)| done.clone()) + .map(wait_for_event_pump); + futures_util::future::join_all(drains).await; + for (id, _, _) in children { mark_session_dead(&self.inner, id).await; } } @@ -527,11 +561,11 @@ impl SessionManager { } }; - let (acp_events, event_log) = if let Some(acp) = &acp { + let acp_events = if let Some(acp) = &acp { match self.inner.sidecar.open_events(id) { Ok(log) => { let (tx, _) = broadcast::channel(EVENTS_CAPACITY); - (Some(tx), Some(log)) + Some(Arc::new(Mutex::new(AcpEventStream { log, live: tx }))) } Err(e) => { acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE) @@ -542,7 +576,7 @@ impl SessionManager { } } } else { - (None, None) + None }; // Capture spawn-history inputs before `req` is partially moved into @@ -615,8 +649,7 @@ impl SessionManager { let event_pump = acp .as_ref() .zip(acp_events.as_ref()) - .zip(event_log) - .map(|((acp, events), log)| (acp.clone(), events.clone(), log)); + .map(|(acp, events)| (acp.clone(), events.clone())); { let mut sessions = self.inner.sessions.lock().await; sessions.insert( @@ -635,11 +668,15 @@ impl SessionManager { awaiting_suppressed: false, acp, acp_events, + acp_event_pump: None, }, ); } - if let Some((acp, events, log)) = event_pump { - start_acp_event_pump(&self.inner, id, &acp, events, log); + if let Some((acp, events)) = event_pump { + let done = start_acp_event_pump(&self.inner, id, &acp, events); + if let Some(record) = self.inner.sessions.lock().await.get_mut(&id) { + record.acp_event_pump = Some(done); + } } let _ = self.inner.events.send(CatalogEvent::SessionAdded { session: session.clone(), @@ -856,6 +893,7 @@ impl SessionManager { // daemon; only tmux-backed sessions can be adopted. acp: None, acp_events: None, + acp_event_pump: None, }, ); let _ = self @@ -880,10 +918,10 @@ impl SessionManager { } let id = session.id; + let mut session_env = sanitize_env(&session.env)?; let fifo_path = self.inner.fifo_dir.join(format!("{id}.fifo")); let bridge = PtyBridge::start_with_replay(fifo_path.clone(), self.inner.replay_bytes).await?; - let mut session_env = sanitize_env(&session.env)?; session_env.insert("MAESTRO_SESSION_ID".into(), id.to_string()); if let Some(url) = self .inner @@ -930,20 +968,21 @@ impl SessionManager { }; session.activity_detail = failure; - let (acp_events, event_log) = if let Some(acp) = &acp { + let acp_events = if let Some(acp) = &acp { match self.inner.sidecar.open_events(id) { Ok(log) => { let (tx, _) = broadcast::channel(EVENTS_CAPACITY); - (Some(tx), Some(log)) + Some(Arc::new(Mutex::new(AcpEventStream { log, live: tx }))) } Err(error) => { acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE) .await; + let _ = std::fs::remove_file(&fifo_path); return Err(SessionError::Sidecar(error)); } } } else { - (None, None) + None }; if let Err(error) = self.inner.sidecar.write(&session) { if let Some(acp) = &acp { @@ -956,8 +995,7 @@ impl SessionManager { let event_pump = acp .as_ref() .zip(acp_events.as_ref()) - .zip(event_log) - .map(|((acp, events), log)| (acp.clone(), events.clone(), log)); + .map(|(acp, events)| (acp.clone(), events.clone())); self.inner.sessions.lock().await.insert( id, @@ -981,10 +1019,14 @@ impl SessionManager { awaiting_suppressed: false, acp, acp_events, + acp_event_pump: None, }, ); - if let Some((acp, events, log)) = event_pump { - start_acp_event_pump(&self.inner, id, &acp, events, log); + if let Some((acp, events)) = event_pump { + let done = start_acp_event_pump(&self.inner, id, &acp, events); + if let Some(record) = self.inner.sessions.lock().await.get_mut(&id) { + record.acp_event_pump = Some(done); + } } let _ = self .inner @@ -1321,34 +1363,49 @@ fn start_acp_event_pump( inner: &Arc, id: Uuid, acp: &Arc, - events: broadcast::Sender, - mut log: EventLog, -) { + events: Arc>, +) -> watch::Receiver { + let (done_tx, done_rx) = watch::channel(false); let Some(mut updates) = acp.take_updates() else { tracing::error!(%id, "structured session notification stream was already taken"); - return; + let _ = done_tx.send(true); + return done_rx; }; let weak_inner = Arc::downgrade(inner); let weak_acp = Arc::downgrade(acp); tokio::spawn(async move { while let Some(notification) = updates.recv().await { - let seq = match log.append(¬ification) { - Ok(seq) => seq, - Err(error) => { - tracing::error!(%id, %error, "structured event log append failed"); - if let Some(acp) = weak_acp.upgrade() { - acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE) - .await; + let append_error = { + let mut events = events.lock().await; + match events.log.append(¬ification) { + Ok(seq) => { + let _ = events.live.send(LoggedEvent { seq, notification }); + None } - if let Some(inner) = weak_inner.upgrade() { - mark_session_dead(&inner, id).await; - } - return; + Err(error) => Some(error), } }; - let _ = events.send(LoggedEvent { seq, notification }); + if let Some(error) = append_error { + tracing::error!(%id, %error, "structured event log append failed"); + if let Some(acp) = weak_acp.upgrade() { + acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE) + .await; + } + if let Some(inner) = weak_inner.upgrade() { + mark_session_dead(&inner, id).await; + } + break; + } } + let _ = done_tx.send(true); }); + done_rx +} + +async fn wait_for_event_pump(mut done: watch::Receiver) { + if !*done.borrow() { + let _ = done.wait_for(|finished| *finished).await; + } } async fn tmux_blocking(f: F) -> T @@ -1742,6 +1799,9 @@ async fn cleanup_record( acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE) .await; } + if let Some(done) = record.acp_event_pump.clone() { + wait_for_event_pump(done).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 index 694015f..1055499 100644 --- a/crates/maestro-daemon/tests/acp.rs +++ b/crates/maestro-daemon/tests/acp.rs @@ -168,11 +168,14 @@ async fn structured_spawn_starts_the_adapter_in_cwd_with_session_env() { .await; // Live delivery happens only after the notification is durable. - let mut updates = fx + let (replay, mut updates) = fx .manager - .subscribe_acp_events(session.id) + .subscribe_acp_events_with_replay(session.id, EventCursor::Since(0), 10) .await + .expect("replay read") .expect("event subscription"); + assert_eq!(replay.len(), 1); + assert!(updates.try_recv().is_err(), "replay duplicated onto live"); unsafe { libc::kill(fake.pid() as libc::pid_t, libc::SIGUSR1); } @@ -251,10 +254,13 @@ async fn recovery_loads_the_acp_session_without_duplicating_history() { }) .await; - let mut live = revived - .subscribe_acp_events(session.id) + let (replay, mut live) = revived + .subscribe_acp_events_with_replay(session.id, EventCursor::Since(0), 10) .await + .expect("replay read") .expect("recovered event subscription"); + assert_eq!(replay.len(), 3); + assert!(live.try_recv().is_err(), "replay duplicated onto live"); let new_pid = fake.pid(); assert_ne!(new_pid, old_pid); unsafe { @@ -279,6 +285,39 @@ async fn recovery_loads_the_acp_session_without_duplicating_history() { revived.kill(session.id, false).await.expect("kill"); } +#[tokio::test] +async fn failed_recovery_cleans_its_fifo() { + 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"); + let old_pid = fake.pid(); + fx.manager.shutdown().await; + wait_for_exit(old_pid).await; + + let events_path = fx + .sidecar_dir() + .join(format!("{}.events.jsonl", session.id)); + std::fs::remove_file(&events_path).unwrap(); + std::fs::create_dir(&events_path).unwrap(); + let revived = SessionManager::new( + Arc::new(fx.tmux.tmux.clone()), + SidecarStore::new(fx.sidecar_dir()), + WorktreeManager::new(None), + fx.fifo_dir(), + ); + + assert!(revived.recover().await.is_err()); + assert!(!fx.fifo_dir().join(format!("{}.fifo", session.id)).exists()); + let failed_pid = fake.pid(); + assert_ne!(failed_pid, old_pid); + wait_for_exit(failed_pid).await; +} + #[tokio::test] async fn missing_adapter_yields_a_failed_session_naming_the_install_command() { let fx = DaemonFixture::new(); @@ -387,6 +426,38 @@ async fn shutdown_reaps_every_adapter_child() { assert_eq!(sidecar.status, SessionStatus::Dead); } +#[tokio::test] +async fn shutdown_drains_final_notifications_to_the_log() { + let fx = DaemonFixture::new(); + let fake = Fake::new(); + let cwd = TempDir::new().unwrap(); + let mut env = fake.env(); + env.insert("FAKE_ACP_EMIT_ON_CANCEL".into(), "1".into()); + + let session = fx + .manager + .create(structured(cwd.path(), env)) + .await + .expect("create"); + wait_for_async(Duration::from_secs(5), || async { + fx.manager + .read_acp_events(session.id, EventCursor::Since(0), 10) + .is_ok_and(|events| events.len() == 1) + }) + .await; + + fx.manager.shutdown().await; + + let stored = fx + .manager + .read_acp_events(session.id, EventCursor::Since(0), 10) + .expect("event log after shutdown"); + assert_eq!( + stored.iter().map(event_text).collect::>(), + ["hello from fake", "final on cancel"] + ); +} + #[tokio::test] async fn adapter_exit_marks_the_session_dead_and_keeps_it_listed() { let fx = DaemonFixture::new();