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

feat(maestro): spawn a claude-code-acp child per structured session (#134)

Browse Source 
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.
This commit is contained in:
Miguel Palhas
2026-09-04 22:02:39 +01:00
parent 540a3025b1
commit b1718559df
4 changed files with 449 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/maestro-daemon/src/session.rs
+150 -32
View File
@@ -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<Arc<AcpSession>>,
}
#[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<tokio::sync::mpsc::UnboundedReceiver<maestro_acp::schema::SessionNotification>>
{
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<AcpSession>)> = {
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<broadcast::Receiver<Vec<u8>>> {
@@ -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<Inner>) {
// 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<Uuid> = {
// 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<Uuid>, Vec<(Uuid, Arc<AcpSession>)>) = {
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<Inner>) {
};
let live: HashSet<Uuid> = 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<Inner>, 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");