diff --git a/crates/maestro-daemon/src/session.rs b/crates/maestro-daemon/src/session.rs index fd0bb09..47cea7a 100644 --- a/crates/maestro-daemon/src/session.rs +++ b/crates/maestro-daemon/src/session.rs @@ -22,7 +22,7 @@ use uuid::Uuid; use crate::panes::Panes; use crate::pty::{pipe_pane_command, PtyBridge}; -use crate::sidecar::{SidecarError, SidecarStore}; +use crate::sidecar::{EventCursor, EventLog, LoggedEvent, SidecarError, SidecarStore}; use crate::spawn_history::{RecentDir, SpawnHistory, SpawnHistoryRecord}; use crate::tmux::TmuxError; use crate::viewport::{ViewportId, Viewports}; @@ -94,6 +94,9 @@ 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>, } #[derive(Debug, thiserror::Error)] @@ -294,18 +297,24 @@ 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( + /// 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 a cursor-bounded slice of the durable structured-session log. + pub fn read_acp_events( &self, id: Uuid, - ) -> Option> - { - let sessions = self.inner.sessions.lock().await; - sessions.get(&id)?.acp.as_ref()?.take_updates() + cursor: EventCursor, + limit: usize, + ) -> Result, SessionError> { + self.inner + .sidecar + .read_events(id, cursor, limit) + .map_err(SessionError::Sidecar) } /// End every adapter child before the daemon exits (#134): each gets @@ -518,6 +527,24 @@ impl SessionManager { } }; + let (acp_events, event_log) = if let Some(acp) = &acp { + match self.inner.sidecar.open_events(id) { + Ok(log) => { + let (tx, _) = broadcast::channel(EVENTS_CAPACITY); + (Some(tx), Some(log)) + } + Err(e) => { + acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE) + .await; + let _ = std::fs::remove_file(&fifo_path); + self.rollback_worktree(&worktree_base_repo, &worktree_path); + return Err(SessionError::Sidecar(e)); + } + } + } else { + (None, None) + }; + // Capture spawn-history inputs before `req` is partially moved into // `session`. `history_cwd` is the *chosen* working dir (the request // `cwd`), not a generated worktree checkout path — that's what the @@ -580,10 +607,16 @@ impl SessionManager { } } let _ = std::fs::remove_file(&fifo_path); + let _ = self.inner.sidecar.delete(id); self.rollback_worktree(&worktree_base_repo, &worktree_path); return Err(SessionError::Sidecar(e)); } + let event_pump = acp + .as_ref() + .zip(acp_events.as_ref()) + .zip(event_log) + .map(|((acp, events), log)| (acp.clone(), events.clone(), log)); { let mut sessions = self.inner.sessions.lock().await; sessions.insert( @@ -601,9 +634,13 @@ impl SessionManager { focused_since: None, awaiting_suppressed: false, acp, + acp_events, }, ); } + if let Some((acp, events, log)) = event_pump { + start_acp_event_pump(&self.inner, id, &acp, events, log); + } let _ = self.inner.events.send(CatalogEvent::SessionAdded { session: session.clone(), }); @@ -818,6 +855,7 @@ impl SessionManager { // A structured session's adapter died with the previous // daemon; only tmux-backed sessions can be adopted. acp: None, + acp_events: None, }, ); let _ = self @@ -826,6 +864,134 @@ impl SessionManager { .send(CatalogEvent::SessionAdded { session }); adopted += 1; } + + // Structured sessions have no tmux process to discover. Their + // sidecars are the recovery inventory: when an ACP id exists, start a + // fresh adapter process and load that conversation into it. + for mut session in self + .inner + .sidecar + .load_all()? + .into_iter() + .filter(|session| session.view == SessionView::Structured) + { + if self.inner.sessions.lock().await.contains_key(&session.id) { + continue; + } + + let id = session.id; + 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 + .daemon_url + .lock() + .expect("daemon_url mutex") + .clone() + { + session_env.insert("MAESTRO_URL".into(), url); + } + + let (acp, failure) = match session.acp_session_id.clone() { + Some(acp_session_id) => { + let config = SpawnConfig::new( + &session.cwd, + session_env + .iter() + .map(|(key, value)| (key.clone(), value.clone())) + .collect(), + ) + .load_session(acp_session_id); + match maestro_acp::spawn(config).await { + Ok(acp) => (Some(Arc::new(acp)), None), + Err(error) => { + tracing::warn!(%id, %error, "structured session: adapter failed to resume"); + (None, Some(error.to_string())) + } + } + } + None => ( + None, + Some("structured session has no stored ACP session id".into()), + ), + }; + session.status = if failure.is_some() { + SessionStatus::Dead + } else { + SessionStatus::Running + }; + session.activity = if failure.is_some() { + ActivityState::Error + } else { + ActivityState::Active + }; + session.activity_detail = failure; + + let (acp_events, event_log) = if let Some(acp) = &acp { + match self.inner.sidecar.open_events(id) { + Ok(log) => { + let (tx, _) = broadcast::channel(EVENTS_CAPACITY); + (Some(tx), Some(log)) + } + Err(error) => { + acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE) + .await; + return Err(SessionError::Sidecar(error)); + } + } + } else { + (None, None) + }; + if let Err(error) = self.inner.sidecar.write(&session) { + if let Some(acp) = &acp { + acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE) + .await; + } + let _ = std::fs::remove_file(&fifo_path); + return Err(SessionError::Sidecar(error)); + } + let event_pump = acp + .as_ref() + .zip(acp_events.as_ref()) + .zip(event_log) + .map(|((acp, events), log)| (acp.clone(), events.clone(), log)); + + self.inner.sessions.lock().await.insert( + id, + SessionRecord { + session: session.clone(), + fifo_path, + worktree_path: session + .worktree + .as_ref() + .map(|worktree| worktree.path.clone()), + worktree_base_repo: session + .worktree + .as_ref() + .map(|worktree| worktree.base_repo.clone()), + bridge, + reported_activity: None, + settled: false, + #[cfg(feature = "haiku-autoname")] + naming: false, + focused_since: None, + awaiting_suppressed: false, + acp, + acp_events, + }, + ); + if let Some((acp, events, log)) = event_pump { + start_acp_event_pump(&self.inner, id, &acp, events, log); + } + let _ = self + .inner + .events + .send(CatalogEvent::SessionAdded { session }); + adopted += 1; + } Ok(adopted) } @@ -1151,6 +1317,40 @@ fn sanitize_env( Ok(out) } +fn start_acp_event_pump( + inner: &Arc, + id: Uuid, + acp: &Arc, + events: broadcast::Sender, + mut log: EventLog, +) { + let Some(mut updates) = acp.take_updates() else { + tracing::error!(%id, "structured session notification stream was already taken"); + return; + }; + 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; + } + if let Some(inner) = weak_inner.upgrade() { + mark_session_dead(&inner, id).await; + } + return; + } + }; + let _ = events.send(LoggedEvent { seq, notification }); + } + }); +} + async fn tmux_blocking(f: F) -> T where F: FnOnce() -> T + Send + 'static, diff --git a/crates/maestro-daemon/tests/acp.rs b/crates/maestro-daemon/tests/acp.rs index 18c3884..694015f 100644 --- a/crates/maestro-daemon/tests/acp.rs +++ b/crates/maestro-daemon/tests/acp.rs @@ -6,10 +6,15 @@ mod common; use std::collections::HashMap; use std::path::{Path, PathBuf}; +use std::sync::Arc; use std::time::{Duration, Instant}; use common::{wait_for_async, DaemonFixture}; +use maestro_acp::schema::{ContentBlock, SessionUpdate}; use maestro_acp::{ADAPTER_BIN, INSTALL_COMMAND}; +use maestro_daemon::sidecar::{EventCursor, SidecarStore}; +use maestro_daemon::worktree::WorktreeManager; +use maestro_daemon::SessionManager; use maestro_protocol::{ActivityState, CreateSessionRequest, SessionStatus, SessionView}; use tempfile::TempDir; @@ -61,13 +66,28 @@ impl Fake { 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()) + .filter_map(|line| { + line.strip_prefix("start pid=")? + .split(' ') + .next()? + .parse() + .ok() + }) + .next_back() .expect("pid in log") } } +fn event_text(event: &maestro_daemon::sidecar::LoggedEvent) -> &str { + match &event.notification.update { + SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content { + ContentBlock::Text(text) => &text.text, + other => panic!("unexpected content {other:?}"), + }, + other => panic!("unexpected update {other:?}"), + } +} + fn structured(cwd: &Path, env: HashMap) -> CreateSessionRequest { CreateSessionRequest { view: SessionView::Structured, @@ -140,20 +160,123 @@ async fn structured_spawn_starts_the_adapter_in_cwd_with_session_env() { .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. + wait_for_async(Duration::from_secs(5), || async { + fx.manager + .read_acp_events(session.id, EventCursor::Since(0), 10) + .is_ok_and(|events| !events.is_empty()) + }) + .await; + + // Live delivery happens only after the notification is durable. let mut updates = fx .manager - .take_acp_updates(session.id) + .subscribe_acp_events(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()) + .expect("event subscription"); + unsafe { + libc::kill(fake.pid() as libc::pid_t, libc::SIGUSR1); + } + let live = 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); + .expect("broadcast open"); + assert_eq!(event_text(&live), "tick 1"); + let stored = fx + .manager + .read_acp_events(session.id, EventCursor::Since(live.seq - 1), 1) + .expect("read event logged before broadcast"); + assert_eq!(stored, vec![live]); fx.manager.kill(session.id, false).await.expect("kill"); + assert!(!fx + .sidecar_dir() + .join(format!("{}.events.jsonl", session.id)) + .exists()); +} + +#[tokio::test] +async fn recovery_loads_the_acp_session_without_duplicating_history() { + let fx = DaemonFixture::new(); + let fake = Fake::new(); + let cwd = TempDir::new().unwrap(); + let mut env = fake.env(); + env.insert("FAKE_ACP_LOAD_REPLAY".into(), "2".into()); + + let session = fx + .manager + .create(structured(cwd.path(), env)) + .await + .expect("create"); + let acp_id = session.acp_session_id.clone().expect("acp id"); + 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; + unsafe { + libc::kill(fake.pid() as libc::pid_t, libc::SIGUSR1); + } + 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() == 2) + }) + .await; + + let old_pid = fake.pid(); + fx.manager.shutdown().await; + wait_for_exit(old_pid).await; + + let revived = SessionManager::new( + Arc::new(fx.tmux.tmux.clone()), + SidecarStore::new(fx.sidecar_dir()), + WorktreeManager::new(None), + fx.fifo_dir(), + ); + assert_eq!(revived.recover().await.expect("recover"), 1); + assert_eq!( + revived.status(session.id).await, + Some(SessionStatus::Running) + ); + let log = fake.log(); + assert!( + log.contains(&format!("recv session/load sessionId={acp_id}")), + "{log}" + ); + wait_for_async(Duration::from_secs(5), || async { + revived + .read_acp_events(session.id, EventCursor::Since(0), 10) + .is_ok_and(|events| events.len() == 3) + }) + .await; + + let mut live = revived + .subscribe_acp_events(session.id) + .await + .expect("recovered event subscription"); + let new_pid = fake.pid(); + assert_ne!(new_pid, old_pid); + unsafe { + libc::kill(new_pid as libc::pid_t, libc::SIGUSR1); + } + let event = tokio::time::timeout(Duration::from_secs(5), live.recv()) + .await + .expect("live event after recovery") + .expect("broadcast open"); + assert_eq!(event.seq, 4); + assert_eq!(event_text(&event), "tick 1"); + + let stored = revived + .read_acp_events(session.id, EventCursor::Since(0), 10) + .expect("recovered log"); + assert_eq!(stored.len(), 4, "session/load replay must not be appended"); + assert_eq!( + stored.iter().map(event_text).collect::>(), + ["hello from fake", "tick 1", "hello after load", "tick 1"] + ); + + revived.kill(session.id, false).await.expect("kill"); } #[tokio::test]