naps62/maestro
Archived
1
0
Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(maestro): persist live ACP events

Browse Source 
Own the ACP update receiver inside SessionManager, append each notification before broadcasting it, and recover structured sidecars by loading their stored ACP session id without duplicating replayed history.
This commit is contained in:
Miguel Palhas
2026-09-04 23:20:48 +01:00
parent 61776b4e0a
commit d11098a478
2 changed files with 344 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/maestro-daemon/src/session.rs
+211 -11
View File
@@ -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<Arc<AcpSession>>,
/// Live notifications after they have been appended to the event log.
/// `None` for terminal sessions and failed structured spawns.
pub acp_events: Option<broadcast::Sender<LoggedEvent>>,
}
#[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<broadcast::Receiver<LoggedEvent>> {
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<tokio::sync::mpsc::UnboundedReceiver<maestro_acp::schema::SessionNotification>>
{
let sessions = self.inner.sessions.lock().await;
sessions.get(&id)?.acp.as_ref()?.take_updates()
cursor: EventCursor,
limit: usize,
) -> Result<Vec<LoggedEvent>, 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<Inner>,
id: Uuid,
acp: &Arc<AcpSession>,
events: broadcast::Sender<LoggedEvent>,
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(&notification) {
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<T, F>(f: F) -> T
where
F: FnOnce() -> T + Send + 'static,
crates/maestro-daemon/tests/acp.rs
+133 -10
View File
@@ -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<String, String>) -> 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::<Vec<_>>(),
["hello from fake", "tick 1", "hello after load", "tick 1"]
);
revived.kill(session.id, false).await.expect("kill");
}
#[tokio::test]