Compare commits

...
5 Commits
Author SHA1 Message Date
Miguel PalhasandClaude Opus 4.6 b339fecad3 feat(maestro): route maestro send through ACP for structured sessions
ci / rust (push) Waiting to run
ci / web (push) Waiting to run
ci / flake (push) Waiting to run
Structured sessions were rejected by `send_input` with NotTerminal.
Now `run_send` checks the session view and routes to
`send_structured_prompt` for structured sessions.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-09-08 13:43:01 +01:00
Miguel PalhasandClaude Opus 4.6 3c10a3a08a fix(maestro): skip Starting sessions in liveness watcher
ci / web (push) Failing after 21s
ci / flake (push) Successful in 3m30s
ci / rust (push) Successful in 6m8s
A structured session still booting its adapter has no ACP handle and
no tmux pane, so the liveness pass misclassified it as a vanished
terminal session and reaped it. Claude Code adapters take long enough
to boot that the 2s liveness tick fires before the handshake finishes.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-09-08 07:25:48 +01:00
Miguel Palhas fd79c8116a Merge remote-tracking branch 'origin/main' into tuba
ci / web (push) Failing after 7s
ci / flake (push) Successful in 3m15s
ci / rust (push) Successful in 5m54s
2026-09-07 21:49:31 +01:00
Miguel PalhasandClaude Opus 4.6 cb8b93f178 test(maestro): update tests for async structured spawn
Tests now assert `Starting` after create and use `wait_for_ready`
before checking post-handshake state. Unsupported command test
expects `BadRequest` instead of a persisted Dead session.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-09-07 21:28:22 +01:00
Miguel PalhasandClaude Opus 4.6 8d35f7f8c8 feat(maestro): async structured session spawn
Register a structured session immediately with `activity: Starting`
and run the ACP handshake in a background task. The HTTP caller gets
an instant response; the user can begin typing a prompt while the
adapter boots.

- Add `ActivityState::Starting` to the protocol
- Open the event log eagerly so the structured WS can connect during
  boot (empty transcript + idle state)
- `queue_acp_prompt` stashes a prompt while `Starting`; the background
  task drains it on successful handshake
- `apply_structured_activity` skips overwriting `Starting`
- `subscribe_acp_projection` allows connecting during `Starting`
- TUI/CLI map `Starting` to idle colour / "starting" label

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-09-07 21:08:08 +01:00
13 changed files with 470 additions and 268 deletions
+2 -2
View File
@@ -41,8 +41,8 @@ pub enum Command {
Attach(AttachArgs), Attach(AttachArgs),
/// Kill a session by ID. /// Kill a session by ID.
Kill(KillArgs), Kill(KillArgs),
/// Type text into a running session, as if it had been typed on an /// Send text to a running session. Terminal sessions receive raw
/// attached terminal. /// keystrokes; structured sessions receive an ACP prompt.
Send(SendArgs), Send(SendArgs),
/// Install or update the agent activity hook (#106). /// Install or update the agent activity hook (#106).
Hooks(HooksArgs), Hooks(HooksArgs),
+28 -15
View File
@@ -130,9 +130,12 @@ pub fn resolve_session(sessions: &[Session], needle: &str) -> Result<uuid::Uuid,
/// `maestro send`. Answers with the session that was typed into, so the /// `maestro send`. Answers with the session that was typed into, so the
/// caller can name it once the daemon has accepted the text. /// caller can name it once the daemon has accepted the text.
/// ///
/// A bare ID skips the catalog fetch; anything else needs the list to match /// Terminal sessions get raw keystrokes via `send_input`; structured sessions
/// against. Text comes from the arguments, or from stdin when there is none — /// get an ACP prompt via `send_structured_prompt` (`--no-enter` is ignored
/// the shape a heredoc or a pipe wants. /// there — ACP prompts are always complete messages).
///
/// Text comes from the arguments, or from stdin when there is none — the
/// shape a heredoc or a pipe wants.
#[allow(clippy::result_large_err)] #[allow(clippy::result_large_err)]
pub async fn run_send( pub async fn run_send(
client: &DaemonClient, client: &DaemonClient,
@@ -155,19 +158,28 @@ pub async fn run_send(
return Err(CliError::NothingToSend); return Err(CliError::NothingToSend);
} }
let id = match uuid::Uuid::parse_str(args.session.trim()) { // We always need the session list: even a bare UUID needs the view field
Ok(id) => id, // to decide which endpoint to hit.
Err(_) => { let sessions = client.list_sessions().await?;
let sessions = client.list_sessions().await?; let id = resolve_session(&sessions, &args.session)?;
resolve_session(&sessions, &args.session)? let view = sessions
} .iter()
}; .find(|s| s.id == id)
.map(|s| s.view)
.unwrap_or(SessionView::Terminal);
let req = SendInputRequest { match view {
text, SessionView::Structured => {
submit: !args.no_enter, client.send_structured_prompt(id, &text).await?;
}; }
client.send_input(id, &req).await?; SessionView::Terminal => {
let req = SendInputRequest {
text,
submit: !args.no_enter,
};
client.send_input(id, &req).await?;
}
}
Ok(id) Ok(id)
} }
@@ -216,6 +228,7 @@ fn print_session_table(out: &mut impl Write, sessions: &[Session], info: &Daemon
ActivityState::Idle => "idle", ActivityState::Idle => "idle",
ActivityState::AwaitingInput => "awaiting", ActivityState::AwaitingInput => "awaiting",
ActivityState::Error => "error", ActivityState::Error => "error",
ActivityState::Starting => "starting",
}; };
// Show the live foreground command alongside the spawn command when // Show the live foreground command alongside the spawn command when
// the daemon has observed one that differs (NAP-340). // the daemon has observed one that differs (NAP-340).
@@ -2,7 +2,7 @@
//! was unit-tested, but nothing exercised the client, the error decoding, or //! was unit-tested, but nothing exercised the client, the error decoding, or
//! the unreachable-daemon path without compiling and forking a real daemon. //! the unreachable-daemon path without compiling and forking a real daemon.
use maestro_protocol::{DaemonClient, DaemonInfo}; use maestro_protocol::{DaemonClient, DaemonInfo, SessionView};
use maestro_testing::{session, MockDaemon}; use maestro_testing::{session, MockDaemon};
use uuid::Uuid; use uuid::Uuid;
@@ -202,6 +202,35 @@ async fn send_resolves_a_session_by_name() {
assert_eq!(sent, id); assert_eq!(sent, id);
} }
/// A structured session routes through the ACP prompt endpoint, not the
/// terminal input one — `--no-enter` is irrelevant (ACP prompts are
/// complete messages).
#[tokio::test]
async fn send_to_structured_session_uses_acp_prompt() {
let mock = MockDaemon::start().await;
let id = Uuid::from_u128(13);
let mut s = session(id, "/tmp", "claude");
s.view = SessionView::Structured;
mock.add_session(s);
let args = maestro_cli::cli::SendArgs {
session: id.to_string(),
text: vec!["fix".into(), "the".into(), "tests".into()],
no_enter: false,
};
let sent = maestro_cli::sessions::run_send(&mock.client(), &args, &mut std::io::empty())
.await
.expect("send");
assert_eq!(sent, id);
// Terminal input must stay empty — the message went through ACP.
assert!(mock.inputs().is_empty());
let prompts = mock.structured_prompts();
assert_eq!(prompts.len(), 1);
assert_eq!(prompts[0].0, id);
assert_eq!(prompts[0].1.prompt, "fix the tests");
}
#[tokio::test] #[tokio::test]
async fn send_to_an_unknown_session_fails_without_typing_anything() { async fn send_to_an_unknown_session_fails_without_typing_anything() {
let mock = MockDaemon::start().await; let mock = MockDaemon::start().await;
+4 -2
View File
@@ -180,7 +180,7 @@ fn dir_error_status(code: &str) -> StatusCode {
fn session_error_status(code: &str) -> StatusCode { fn session_error_status(code: &str) -> StatusCode {
match code { match code {
"not_found" => StatusCode::NOT_FOUND, "not_found" => StatusCode::NOT_FOUND,
"invalid_env" => StatusCode::BAD_REQUEST, "invalid_env" | "bad_request" => StatusCode::BAD_REQUEST,
// A guard refusal is the daemon doing its job, not a failure (#79). // A guard refusal is the daemon doing its job, not a failure (#79).
"kill_guarded" => StatusCode::CONFLICT, "kill_guarded" => StatusCode::CONFLICT,
// The session outlived its pane: nothing to type into, and nothing // The session outlived its pane: nothing to type into, and nothing
@@ -483,7 +483,9 @@ async fn structured_prompt(
Json(body): Json<StructuredPromptRequest>, Json(body): Json<StructuredPromptRequest>,
) -> Response { ) -> Response {
match s.manager.queue_acp_prompt(id, body.prompt).await { match s.manager.queue_acp_prompt(id, body.prompt).await {
Ok(QueueAcpPrompt::Queued) => StatusCode::NO_CONTENT.into_response(), Ok(QueueAcpPrompt::Queued) | Ok(QueueAcpPrompt::Pending) => {
StatusCode::NO_CONTENT.into_response()
}
Ok(QueueAcpPrompt::InFlight) => error_response( Ok(QueueAcpPrompt::InFlight) => error_response(
StatusCode::CONFLICT, StatusCode::CONFLICT,
"turn_in_flight", "turn_in_flight",
+208 -160
View File
@@ -6,12 +6,9 @@
//! handler interact with tmux. //! handler interact with tmux.
use std::collections::{HashMap, HashSet}; use std::collections::{HashMap, HashSet};
use std::future::Future;
use std::path::PathBuf; use std::path::PathBuf;
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Weak}; use std::sync::{Arc, Weak};
use std::task::{Context, Poll};
use std::time::Duration; use std::time::Duration;
use chrono::Utc; use chrono::Utc;
@@ -125,6 +122,9 @@ pub struct SessionRecord {
/// When `true`, the permission pump auto-responds `AllowAlways` to every /// When `true`, the permission pump auto-responds `AllowAlways` to every
/// ACP permission request without surfacing it to clients. /// ACP permission request without surfacing it to clients.
pub auto_approve: bool, pub auto_approve: bool,
/// A prompt submitted while the adapter was still booting (`Starting`).
/// The background handshake task drains this on success.
pub pending_prompt: Option<String>,
} }
impl SessionRecord { impl SessionRecord {
@@ -134,6 +134,13 @@ impl SessionRecord {
&& self.acp.is_some() && self.acp.is_some()
&& self.acp_events.is_some() && self.acp_events.is_some()
} }
/// The session is structured and still booting its adapter.
fn is_starting_structured(&self) -> bool {
self.session.view == SessionView::Structured
&& self.session.activity == ActivityState::Starting
&& self.acp.is_none()
}
} }
#[derive(Debug)] #[derive(Debug)]
@@ -208,6 +215,8 @@ pub enum SessionError {
NotTerminal(Uuid), NotTerminal(Uuid),
#[error("daemon is shutting down")] #[error("daemon is shutting down")]
ShuttingDown, ShuttingDown,
#[error("bad request: {0}")]
BadRequest(String),
/// The kill was refused (#79): the session's worktree holds unsaved work. /// The kill was refused (#79): the session's worktree holds unsaved work.
/// Nothing happened — the session keeps running and the tree is untouched, /// Nothing happened — the session keeps running and the tree is untouched,
/// so it stays the operator's handle on their own unfinished work. /// so it stays the operator's handle on their own unfinished work.
@@ -227,6 +236,8 @@ pub enum BeginAcpTurn {
#[derive(Debug)] #[derive(Debug)]
pub enum QueueAcpPrompt { pub enum QueueAcpPrompt {
Queued, Queued,
/// The adapter is still booting; the prompt is held until it is ready.
Pending,
InFlight, InFlight,
Unavailable, Unavailable,
Failed(String), Failed(String),
@@ -253,6 +264,7 @@ impl SessionError {
SessionError::NotRunning(_) => "not_running", SessionError::NotRunning(_) => "not_running",
SessionError::NotTerminal(_) => "not_terminal", SessionError::NotTerminal(_) => "not_terminal",
SessionError::ShuttingDown => "shutting_down", SessionError::ShuttingDown => "shutting_down",
SessionError::BadRequest(_) => "bad_request",
SessionError::KillGuarded(_) => "kill_guarded", SessionError::KillGuarded(_) => "kill_guarded",
}; };
let unsaved_work = match self { let unsaved_work = match self {
@@ -320,18 +332,6 @@ struct StructuredLifecycle {
} }
impl StructuredLifecycle { impl StructuredLifecycle {
fn admit<'a>(
&'a self,
shutting_down: &AtomicBool,
) -> Result<StructuredCreate<'a>, SessionError> {
let _gate = self.gate.lock().expect("structured lifecycle gate");
if shutting_down.load(Ordering::Acquire) {
return Err(SessionError::ShuttingDown);
}
self.in_flight.fetch_add(1, Ordering::AcqRel);
Ok(StructuredCreate { lifecycle: self })
}
async fn wait_until_settled(&self) { async fn wait_until_settled(&self) {
loop { loop {
let settled = self.settled.notified(); let settled = self.settled.notified();
@@ -343,30 +343,18 @@ impl StructuredLifecycle {
} }
} }
struct StructuredCreate<'a> { /// Release the structured-lifecycle in-flight counter that `create` acquires
lifecycle: &'a StructuredLifecycle, /// before spawning the background ACP handshake task. Called when the task
} /// finishes (success or failure) or when `create_inner` fails before the task
/// is spawned.
impl Drop for StructuredCreate<'_> { fn drop_structured_lifecycle_permit(inner: &Inner) {
fn drop(&mut self) { if inner
if self.lifecycle.in_flight.fetch_sub(1, Ordering::AcqRel) == 1 { .structured_lifecycle
self.lifecycle.settled.notify_one(); .in_flight
} .fetch_sub(1, Ordering::AcqRel)
} == 1
} {
inner.structured_lifecycle.settled.notify_one();
struct AdmittedStructuredCreate<'a> {
create: Pin<Box<dyn Future<Output = Result<Session, SessionError>> + Send + 'a>>,
/// Field order matters: dropping a cancelled create rolls back its inner
/// future before this permit releases shutdown's in-flight barrier.
_permit: StructuredCreate<'a>,
}
impl Future for AdmittedStructuredCreate<'_> {
type Output = Result<Session, SessionError>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
self.create.as_mut().poll(cx)
} }
} }
@@ -660,7 +648,9 @@ impl SessionManager {
let stream = { let stream = {
let sessions = self.inner.sessions.lock().await; let sessions = self.inner.sessions.lock().await;
let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?; let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
if !record.is_live_structured() { // Allow connecting while the adapter is still booting
// (`Starting`) — the event stream exists, it is just empty.
if !record.is_live_structured() && !record.is_starting_structured() {
return Ok(None); return Ok(None);
} }
let Some(stream) = &record.acp_events else { let Some(stream) = &record.acp_events else {
@@ -745,8 +735,14 @@ impl SessionManager {
text: String, text: String,
) -> Result<QueueAcpPrompt, SessionError> { ) -> Result<QueueAcpPrompt, SessionError> {
let (acp, stream) = { let (acp, stream) = {
let sessions = self.inner.sessions.lock().await; let mut sessions = self.inner.sessions.lock().await;
let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?; let record = sessions.get_mut(&id).ok_or(SessionError::NotFound(id))?;
// While the adapter is booting, stash the prompt for the
// background handshake task to drain on success.
if record.is_starting_structured() {
record.pending_prompt = Some(text);
return Ok(QueueAcpPrompt::Pending);
}
if !record.is_live_structured() { if !record.is_live_structured() {
return Ok(QueueAcpPrompt::Unavailable); return Ok(QueueAcpPrompt::Unavailable);
} }
@@ -1217,17 +1213,42 @@ impl SessionManager {
} }
/// Spawn a new session. On success returns the freshly-created `Session`. /// Spawn a new session. On success returns the freshly-created `Session`.
///
/// For structured sessions, the ACP handshake runs in a background task
/// after the session is registered — the HTTP caller gets an instant
/// response with `activity: Starting`. The `StructuredLifecycle` gate is
/// held across the background task so `shutdown()` waits for it.
pub async fn create(&self, req: CreateSessionRequest) -> Result<Session, SessionError> { pub async fn create(&self, req: CreateSessionRequest) -> Result<Session, SessionError> {
if req.view == SessionView::Structured { if req.view == SessionView::Structured {
let permit = self // Acquire the lifecycle gate synchronously (rejects if shutting
// down). The background handshake task calls
// `drop_structured_lifecycle_permit` when it finishes; we must
// not use the RAII `StructuredCreate` because it would drop at
// the end of `create_inner`, before the handshake completes.
let _gate = self
.inner .inner
.structured_lifecycle .structured_lifecycle
.admit(&self.inner.shutting_down)?; .gate
return AdmittedStructuredCreate { .lock()
create: Box::pin(self.create_inner(req)), .expect("structured lifecycle gate");
_permit: permit, if self.inner.shutting_down.load(Ordering::Acquire) {
return Err(SessionError::ShuttingDown);
} }
.await; self.inner
.structured_lifecycle
.in_flight
.fetch_add(1, Ordering::AcqRel);
drop(_gate);
return match self.create_inner(req).await {
Ok(session) => Ok(session),
Err(e) => {
// create_inner failed before spawning the background
// task, so release the permit here.
drop_structured_lifecycle_permit(&self.inner);
Err(e)
}
};
} }
self.create_inner(req).await self.create_inner(req).await
} }
@@ -1289,32 +1310,39 @@ impl SessionManager {
{ {
session_env.insert("MAESTRO_URL".into(), url); session_env.insert("MAESTRO_URL".into(), url);
} }
// A structured session (#134) gets no tmux pane: the agent is a // ── Structured path: validate adapter, open event log, register
// child of the daemon, driven over ACP. An adapter that cannot come // the session immediately with `Starting`, then handshake in the
// up still yields a session — Dead, with the reason on it — so the // background so the HTTP caller gets an instant response.
// operator sees what to fix instead of a silently missing session. //
let (acp, acp_failure) = match req.view { // ── Terminal path: unchanged — tmux spawn is fast enough to stay
SessionView::Structured => match structured_adapter(&req.command) { // synchronous.
Err(error) => (None, Some(error)), // Structured: validate adapter, open event log eagerly. Terminal:
Ok(adapter) => match maestro_acp::spawn( // spawn the tmux pane synchronously (it is fast).
SpawnConfig::new( let structured_spawn = match req.view {
&effective_cwd, SessionView::Structured => {
session_env // Fail fast on an unsupported command before touching disk.
.iter() let adapter = structured_adapter(&req.command).map_err(SessionError::BadRequest)?;
.map(|(k, v)| (k.clone(), v.clone())) // Open the event log eagerly so the structured WS can
.collect(), // connect before the adapter is up.
) let acp_events = match self.inner.sidecar.open_events(id) {
.adapter(adapter), Ok(log) => {
) let (tx, _) = broadcast::channel(EVENTS_CAPACITY);
.await let (projected_tx, _) = broadcast::channel(EVENTS_CAPACITY);
{ Arc::new(Mutex::new(AcpEventStream {
Ok(acp) => (Some(Arc::new(acp)), None), log,
Err(e) => { live: tx,
tracing::warn!(%id, error = %e, "structured session: adapter failed to start"); projection: Projection::default(),
(None, Some(e.to_string())) projected_live: projected_tx,
pending_approvals: HashMap::new(),
operation: Arc::new(Mutex::new(())),
dispatch: AcpDispatchState::default(),
activity_sequence: 0,
}))
} }
}, Err(e) => return Err(SessionError::Sidecar(e)),
}, };
Some((adapter, acp_events))
}
SessionView::Terminal => { SessionView::Terminal => {
let spawned = { let spawned = {
let tmux = self.inner.tmux.clone(); let tmux = self.inner.tmux.clone();
@@ -1328,53 +1356,10 @@ impl SessionManager {
if let Err(e) = spawned { if let Err(e) = spawned {
return Err(SessionError::Tmux(e)); return Err(SessionError::Tmux(e));
} }
(None, None) None
} }
}; };
let acp_events = if let Some(acp) = &acp {
let persisted =
match self
.inner
.sidecar
.read_events(id, EventCursor::Since(0), usize::MAX)
{
Ok(events) => events,
Err(error) => {
acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
.await;
return Err(SessionError::Sidecar(error));
}
};
match self.inner.sidecar.open_events(id) {
Ok(log) => {
let (tx, _) = broadcast::channel(EVENTS_CAPACITY);
let (projected_tx, _) = broadcast::channel(EVENTS_CAPACITY);
let mut projection = Projection::default();
for event in &persisted {
apply_persisted_event(&mut projection, event);
}
Some(Arc::new(Mutex::new(AcpEventStream {
log,
live: tx,
projection,
projected_live: projected_tx,
pending_approvals: HashMap::new(),
operation: Arc::new(Mutex::new(())),
dispatch: AcpDispatchState::default(),
activity_sequence: 0,
})))
}
Err(e) => {
acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
.await;
return Err(SessionError::Sidecar(e));
}
}
} else {
None
};
// Capture spawn-history inputs before `req` is partially moved into // Capture spawn-history inputs before `req` is partially moved into
// `session`. `history_cwd` is the *chosen* working dir (the request // `session`. `history_cwd` is the *chosen* working dir (the request
// `cwd`), not a generated worktree checkout path — that's what the // `cwd`), not a generated worktree checkout path — that's what the
@@ -1384,10 +1369,13 @@ impl SessionManager {
resolve_name(&mut metadata); resolve_name(&mut metadata);
let worktree_branch = req.worktree.as_ref().map(|w| w.branch.clone()); let worktree_branch = req.worktree.as_ref().map(|w| w.branch.clone());
let is_structured = structured_spawn.is_some();
let events_arc = structured_spawn.as_ref().map(|(_, e)| e.clone());
let session = Session { let session = Session {
id, id,
cwd: effective_cwd, cwd: effective_cwd.clone(),
command: req.command, command: req.command.clone(),
env: req.env, env: req.env,
metadata, metadata,
worktree: worktree_path.as_ref().zip(worktree_base_repo.as_ref()).map( worktree: worktree_path.as_ref().zip(worktree_base_repo.as_ref()).map(
@@ -1396,67 +1384,41 @@ impl SessionManager {
base_repo: base_repo.clone(), base_repo: base_repo.clone(),
}, },
), ),
status: if acp_failure.is_some() { status: SessionStatus::Running,
SessionStatus::Dead
} else {
SessionStatus::Running
},
created_at: Utc::now(), created_at: Utc::now(),
// A terminal spawn begins active until its watcher sees otherwise;
// a structured spawn has no turn yet and begins idle.
foreground: None, foreground: None,
activity: match (req.view, acp_failure.is_some()) { activity: if is_structured {
(_, true) => ActivityState::Error, ActivityState::Starting
(SessionView::Structured, false) => ActivityState::Idle, } else {
(SessionView::Terminal, false) => ActivityState::Active, ActivityState::Active
}, },
activity_detail: acp_failure, activity_detail: None,
last_activity_at: Utc::now(), last_activity_at: Utc::now(),
activity_since: chrono::Utc::now(), activity_since: chrono::Utc::now(),
current_cwd: None, current_cwd: None,
// The watcher fills this in on its first blocked-work pass; a
// fresh tree has nothing in the way anyway.
blocked_work: None, blocked_work: None,
last_focused_at: None, last_focused_at: None,
acp_session_id: acp.as_ref().map(|a| a.session_id().to_string()), acp_session_id: None,
view: req.view, view: req.view,
}; };
if let Err(e) = self.inner.sidecar.write(&session) { if let Err(e) = self.inner.sidecar.write(&session) {
// The process already started; end it so no ghost survives a if !is_structured {
// failed sidecar write, then unwind the FIFO + worktree. let tmux = self.inner.tmux.clone();
match &acp { let _ = tmux_blocking(move || tmux.kill_session(id)).await;
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;
}
} }
return Err(SessionError::Sidecar(e)); return Err(SessionError::Sidecar(e));
} }
let event_pump = acp
.as_ref()
.zip(acp_events.as_ref())
.map(|(acp, events)| (acp.clone(), events.clone()));
{ {
let mut sessions = self.inner.sessions.lock().await; let mut sessions = self.inner.sessions.lock().await;
let event_pump = event_pump.map(|(acp, events)| {
let done = start_acp_event_pump(&self.inner, id, &acp, events.clone());
start_acp_permission_pump(&self.inner, id, &acp, events, req.auto_approve);
done
});
sessions.insert( sessions.insert(
id, id,
SessionRecord { SessionRecord {
session: session.clone(), session: session.clone(),
fifo_path, fifo_path,
worktree_path, worktree_path: worktree_path.clone(),
worktree_base_repo, worktree_base_repo: worktree_base_repo.clone(),
bridge, bridge,
reported_activity: None, reported_activity: None,
settled: false, settled: false,
@@ -1464,11 +1426,12 @@ impl SessionManager {
naming: false, naming: false,
focused_since: None, focused_since: None,
awaiting_suppressed: false, awaiting_suppressed: false,
acp, acp: None,
acp_events, acp_events: events_arc,
structured_activity_watermark: None, structured_activity_watermark: None,
acp_event_pump: event_pump, acp_event_pump: None,
auto_approve: req.auto_approve, auto_approve: req.auto_approve,
pending_prompt: None,
}, },
); );
} }
@@ -1497,6 +1460,78 @@ impl SessionManager {
} }
pending.commit(); pending.commit();
// ── Background ACP handshake for structured sessions ──
//
// The session is registered and visible to clients. The adapter
// spawn + handshake runs in a background task; on completion it
// fills in `acp`, starts the event/permission pumps, and
// transitions activity from `Starting` to `Idle` (or to `Error`
// on failure). The `StructuredLifecycle` permit is held until the
// task finishes so `shutdown()` waits for it.
if let Some((adapter, events)) = structured_spawn {
let inner = self.inner.clone();
let env: HashMap<String, String> = session_env
.iter()
.map(|(k, v)| (k.clone(), v.clone()))
.collect();
let cwd = effective_cwd;
let auto_approve = req.auto_approve;
tokio::spawn(async move {
let result = maestro_acp::spawn(SpawnConfig::new(&cwd, env).adapter(adapter)).await;
match result {
Ok(acp_session) => {
let acp = Arc::new(acp_session);
let done = start_acp_event_pump(&inner, id, &acp, events.clone());
start_acp_permission_pump(&inner, id, &acp, events.clone(), auto_approve);
// Drain any prompt submitted while Starting.
let pending_prompt = {
let mut sessions = inner.sessions.lock().await;
let Some(record) = sessions.get_mut(&id) else {
// Session was killed while we were booting.
acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
.await;
drop_structured_lifecycle_permit(&inner);
return;
};
record.acp = Some(acp.clone());
record.acp_event_pump = Some(done);
record.session.acp_session_id = Some(acp.session_id().to_string());
record.session.activity = ActivityState::Idle;
record.session.activity_since = Utc::now();
let _ = inner.sidecar.write(&record.session);
let _ = inner.events.send(CatalogEvent::SessionUpdated {
session: record.session.clone(),
});
record.pending_prompt.take()
};
if let Some(prompt) = pending_prompt {
let manager = SessionManager {
inner: inner.clone(),
};
let _ = manager.queue_acp_prompt(id, prompt).await;
}
}
Err(e) => {
tracing::warn!(%id, error = %e, "structured session: adapter failed to start");
let mut sessions = inner.sessions.lock().await;
if let Some(record) = sessions.get_mut(&id) {
record.session.status = SessionStatus::Dead;
record.session.activity = ActivityState::Error;
record.session.activity_detail = Some(e.to_string());
record.session.activity_since = Utc::now();
let _ = inner.sidecar.write(&record.session);
let _ = inner.events.send(CatalogEvent::SessionUpdated {
session: record.session.clone(),
});
}
}
}
drop_structured_lifecycle_permit(&inner);
});
}
Ok(session) Ok(session)
} }
@@ -1682,6 +1717,7 @@ impl SessionManager {
structured_activity_watermark: None, structured_activity_watermark: None,
acp_event_pump: None, acp_event_pump: None,
auto_approve: false, auto_approve: false,
pending_prompt: None,
}, },
); );
let _ = self let _ = self
@@ -1862,6 +1898,7 @@ impl SessionManager {
// Recovery cannot know the original preset's flag; the // Recovery cannot know the original preset's flag; the
// operator re-enables it by restarting with the preset. // operator re-enables it by restarting with the preset.
auto_approve: false, auto_approve: false,
pending_prompt: None,
}, },
); );
if let Some((acp, events)) = event_pump { if let Some((acp, events)) = event_pump {
@@ -2321,6 +2358,11 @@ async fn apply_structured_activity(
if record.session.view != SessionView::Structured { if record.session.view != SessionView::Structured {
return; return;
} }
// While the adapter is still booting, the background handshake task
// owns the activity state — the activity pass must not overwrite it.
if record.session.activity == ActivityState::Starting {
return;
}
let current_events = record.acp_events.as_ref(); let current_events = record.acp_events.as_ref();
match (expected_events, current_events) { match (expected_events, current_events) {
(Some((expected, sequence)), Some(current)) if Arc::ptr_eq(expected, current) => { (Some((expected, sequence)), Some(current)) if Arc::ptr_eq(expected, current) => {
@@ -2586,6 +2628,12 @@ async fn run_liveness_pass(inner: &Arc<Inner>) {
let mut tmux_backed = Vec::new(); let mut tmux_backed = Vec::new();
let mut acp_backed = Vec::new(); let mut acp_backed = Vec::new();
for (id, record) in running { for (id, record) in running {
if record.is_starting_structured() {
// The adapter hasn't finished its handshake yet — no ACP
// handle and no tmux pane. Skip; the background task owns
// this session's lifecycle until the handshake settles.
continue;
}
match &record.acp { match &record.acp {
Some(acp) => acp_backed.push((*id, acp.clone())), Some(acp) => acp_backed.push((*id, acp.clone())),
None => tmux_backed.push(*id), None => tmux_backed.push(*id),
+135 -81
View File
@@ -130,6 +130,25 @@ async fn wait_for_exit(pid: u32) {
assert!(!pid_alive(pid), "pid {pid} still alive"); assert!(!pid_alive(pid), "pid {pid} still alive");
} }
/// Wait for a session's background ACP handshake to complete (activity
/// transitions out of `Starting`). Returns the refreshed session.
async fn wait_for_ready(manager: &SessionManager, id: uuid::Uuid) -> maestro_protocol::Session {
let deadline = Instant::now() + Duration::from_secs(10);
loop {
let session = manager
.snapshot()
.await
.into_iter()
.find(|s| s.id == id)
.expect("session exists");
if session.activity != ActivityState::Starting {
return session;
}
assert!(Instant::now() < deadline, "handshake timed out");
tokio::time::sleep(Duration::from_millis(20)).await;
}
}
#[tokio::test] #[tokio::test]
async fn structured_spawn_starts_the_adapter_in_cwd_with_session_env() { async fn structured_spawn_starts_the_adapter_in_cwd_with_session_env() {
let fx = DaemonFixture::new(); let fx = DaemonFixture::new();
@@ -138,7 +157,7 @@ async fn structured_spawn_starts_the_adapter_in_cwd_with_session_env() {
let mut env = fake.env(); let mut env = fake.env();
env.insert("FAKE_ACP_ECHO_ENV".into(), "MAESTRO_SESSION_ID".into()); env.insert("FAKE_ACP_ECHO_ENV".into(), "MAESTRO_SESSION_ID".into());
let session = fx let created = fx
.manager .manager
.create(structured_command( .create(structured_command(
cwd.path(), cwd.path(),
@@ -148,8 +167,12 @@ async fn structured_spawn_starts_the_adapter_in_cwd_with_session_env() {
.await .await
.expect("create"); .expect("create");
assert_eq!(session.view, SessionView::Structured); assert_eq!(created.view, SessionView::Structured);
assert_eq!(session.status, SessionStatus::Running); assert_eq!(created.status, SessionStatus::Running);
assert_eq!(created.activity, ActivityState::Starting);
assert!(created.acp_session_id.is_none());
let session = wait_for_ready(&fx.manager, created.id).await;
assert_eq!(session.activity, ActivityState::Idle); assert_eq!(session.activity, ActivityState::Idle);
let acp_id = session.acp_session_id.clone().expect("acp session id"); let acp_id = session.acp_session_id.clone().expect("acp session id");
assert!(acp_id.starts_with("fake-"), "{acp_id}"); assert!(acp_id.starts_with("fake-"), "{acp_id}");
@@ -164,7 +187,7 @@ async fn structured_spawn_starts_the_adapter_in_cwd_with_session_env() {
"{log}" "{log}"
); );
assert!( assert!(
log.contains(&format!("env MAESTRO_SESSION_ID={}", session.id)), log.contains(&format!("env MAESTRO_SESSION_ID={}", created.id)),
"{log}" "{log}"
); );
@@ -232,6 +255,7 @@ async fn structured_activity_tracks_turn_completion_and_failure() {
.create(structured_command(cwd.path(), env, "codex")) .create(structured_command(cwd.path(), env, "codex"))
.await .await
.unwrap(); .unwrap();
let session = wait_for_ready(&fx.manager, session.id).await;
assert!(matches!( assert!(matches!(
fx.manager fx.manager
@@ -295,11 +319,12 @@ async fn projection_snapshot_and_live_stream_share_one_fold() {
let fx = DaemonFixture::new(); let fx = DaemonFixture::new();
let fake = Fake::new(); let fake = Fake::new();
let cwd = TempDir::new().unwrap(); let cwd = TempDir::new().unwrap();
let session = fx let created = fx
.manager .manager
.create(structured(cwd.path(), fake.env())) .create(structured(cwd.path(), fake.env()))
.await .await
.expect("create"); .expect("create");
let session = wait_for_ready(&fx.manager, created.id).await;
wait_for_async(Duration::from_secs(5), || async { wait_for_async(Duration::from_secs(5), || async {
fx.manager fx.manager
@@ -397,6 +422,7 @@ async fn queued_prompt_is_a_durable_user_transcript_row() {
.create(structured(cwd.path(), fake.env())) .create(structured(cwd.path(), fake.env()))
.await .await
.expect("create"); .expect("create");
let session = wait_for_ready(&fx.manager, session.id).await;
assert!(matches!( assert!(matches!(
fx.manager fx.manager
@@ -442,6 +468,7 @@ async fn stale_turn_completion_cannot_end_the_next_turn() {
.create(structured(cwd.path(), fake.env())) .create(structured(cwd.path(), fake.env()))
.await .await
.unwrap(); .unwrap();
wait_for_ready(&fx.manager, session.id).await;
let maestro_daemon::session::BeginAcpTurn::Started(_, first) = let maestro_daemon::session::BeginAcpTurn::Started(_, first) =
fx.manager.begin_acp_turn(session.id).await.unwrap() fx.manager.begin_acp_turn(session.id).await.unwrap()
@@ -485,6 +512,7 @@ async fn cancelled_prompt_that_never_settles_fails_the_session() {
.create(structured(cwd.path(), env)) .create(structured(cwd.path(), env))
.await .await
.unwrap(); .unwrap();
wait_for_ready(&fx.manager, session.id).await;
assert!(matches!( assert!(matches!(
fx.manager fx.manager
@@ -523,6 +551,7 @@ async fn cancellation_settlement_allows_the_next_prompt_permission() {
.create(structured(cwd.path(), env)) .create(structured(cwd.path(), env))
.await .await
.unwrap(); .unwrap();
wait_for_ready(&fx.manager, session.id).await;
assert!(matches!( assert!(matches!(
fx.manager fx.manager
@@ -582,11 +611,12 @@ async fn recovery_loads_the_acp_session_without_duplicating_history() {
let mut env = fake.env(); let mut env = fake.env();
env.insert("FAKE_ACP_LOAD_REPLAY".into(), "2".into()); env.insert("FAKE_ACP_LOAD_REPLAY".into(), "2".into());
let session = fx let created = fx
.manager .manager
.create(structured_command(cwd.path(), env, "codex")) .create(structured_command(cwd.path(), env, "codex"))
.await .await
.expect("create"); .expect("create");
let session = wait_for_ready(&fx.manager, created.id).await;
let acp_id = session.acp_session_id.clone().expect("acp id"); let acp_id = session.acp_session_id.clone().expect("acp id");
wait_for_async(Duration::from_secs(5), || async { wait_for_async(Duration::from_secs(5), || async {
fx.manager fx.manager
@@ -711,6 +741,7 @@ async fn shutdown_preserves_a_pristine_structured_worktree_for_recovery() {
.await .await
.expect("create"); .expect("create");
assert_eq!(session.cwd, worktree); assert_eq!(session.cwd, worktree);
wait_for_ready(&fx.manager, session.id).await;
push_all_to_origin(repo.path(), origin.path()); push_all_to_origin(repo.path(), origin.path());
fx.manager.shutdown().await; fx.manager.shutdown().await;
@@ -759,6 +790,7 @@ async fn failed_recovery_cleans_its_fifo() {
.create(structured(cwd.path(), fake.env())) .create(structured(cwd.path(), fake.env()))
.await .await
.expect("create"); .expect("create");
wait_for_ready(&fx.manager, session.id).await;
let old_pid = fake.pid(); let old_pid = fake.pid();
fx.manager.shutdown().await; fx.manager.shutdown().await;
wait_for_exit(old_pid).await; wait_for_exit(old_pid).await;
@@ -792,12 +824,15 @@ async fn missing_adapter_yields_a_failed_session_naming_the_install_command() {
empty.path().to_string_lossy().into_owned(), empty.path().to_string_lossy().into_owned(),
)]); )]);
let session = fx let created = fx
.manager .manager
.create(structured_command(cwd.path(), env, "codex")) .create(structured_command(cwd.path(), env, "codex"))
.await .await
.expect("a failed session, not an error"); .expect("create registers a Starting session");
assert_eq!(created.activity, ActivityState::Starting);
// The background handshake fails because the adapter binary is missing.
let session = wait_for_ready(&fx.manager, created.id).await;
assert_eq!(session.status, SessionStatus::Dead); assert_eq!(session.status, SessionStatus::Dead);
assert_eq!(session.activity, ActivityState::Error); assert_eq!(session.activity, ActivityState::Error);
let detail = session.activity_detail.clone().expect("actionable detail"); let detail = session.activity_detail.clone().expect("actionable detail");
@@ -810,13 +845,13 @@ async fn missing_adapter_yields_a_failed_session_naming_the_install_command() {
// It is a real session: listed, on disk, and dismissable like any other // It is a real session: listed, on disk, and dismissable like any other
// dead one. // dead one.
assert!(fx.manager.has(session.id).await); assert!(fx.manager.has(created.id).await);
assert!(fx assert!(fx
.sidecar_dir() .sidecar_dir()
.join(format!("{}.json", session.id)) .join(format!("{}.json", created.id))
.exists()); .exists());
fx.manager.kill(session.id, false).await.expect("dismiss"); fx.manager.kill(created.id, false).await.expect("dismiss");
assert!(!fx.manager.has(session.id).await); assert!(!fx.manager.has(created.id).await);
} }
#[tokio::test] #[tokio::test]
@@ -825,20 +860,20 @@ async fn unsupported_structured_command_fails_without_launching_claude() {
let fake = Fake::new(); let fake = Fake::new();
let cwd = TempDir::new().unwrap(); let cwd = TempDir::new().unwrap();
let session = fx let err = fx
.manager .manager
.create(structured_command(cwd.path(), fake.env(), "opencode")) .create(structured_command(cwd.path(), fake.env(), "opencode"))
.await .await
.expect("failed structured session is persisted"); .expect_err("unsupported command is rejected at request validation");
assert_eq!(session.status, SessionStatus::Dead); assert!(
let detail = session.activity_detail.expect("selection error"); matches!(err, maestro_daemon::SessionError::BadRequest(ref msg) if msg.contains("Claude Code or Codex")),
assert!(detail.contains("Claude Code or Codex"), "{detail}"); "{err:?}"
);
assert!( assert!(
fake.log().is_empty(), fake.log().is_empty(),
"unsupported command launched an adapter" "unsupported command launched an adapter"
); );
fx.manager.kill(session.id, false).await.expect("dismiss");
} }
#[tokio::test] #[tokio::test]
@@ -853,6 +888,7 @@ async fn kill_sends_cancel_then_sigterm_and_removes_the_sidecar() {
.create(structured(cwd.path(), fake.env())) .create(structured(cwd.path(), fake.env()))
.await .await
.expect("create"); .expect("create");
wait_for_ready(&fx.manager, session.id).await;
let pid = fake.pid(); let pid = fake.pid();
assert!(pid_alive(pid)); assert!(pid_alive(pid));
@@ -895,6 +931,8 @@ async fn shutdown_reaps_every_adapter_child() {
.create(structured(cwd.path(), b.env())) .create(structured(cwd.path(), b.env()))
.await .await
.expect("create b"); .expect("create b");
wait_for_ready(&fx.manager, sa.id).await;
wait_for_ready(&fx.manager, sb.id).await;
let (pa, pb) = (a.pid(), b.pid()); let (pa, pb) = (a.pid(), b.pid());
assert!(pid_alive(pa) && pid_alive(pb)); assert!(pid_alive(pa) && pid_alive(pb));
@@ -929,10 +967,13 @@ async fn dismissal_during_shutdown_does_not_restore_recovery_artifacts() {
.create(structured(cwd.path(), env)) .create(structured(cwd.path(), env))
.await .await
.expect("create"); .expect("create");
fx.manager let blocker_session = fx
.manager
.create(structured(cwd.path(), blocker.env())) .create(structured(cwd.path(), blocker.env()))
.await .await
.expect("create blocker"); .expect("create blocker");
wait_for_ready(&fx.manager, session.id).await;
wait_for_ready(&fx.manager, blocker_session.id).await;
let manager = fx.manager.clone(); let manager = fx.manager.clone();
let shutdown = tokio::spawn(async move { manager.shutdown().await }); let shutdown = tokio::spawn(async move { manager.shutdown().await });
@@ -984,6 +1025,7 @@ async fn shutdown_does_not_recover_an_adapter_that_already_exited() {
.create(structured(cwd.path(), env)) .create(structured(cwd.path(), env))
.await .await
.expect("create"); .expect("create");
wait_for_ready(&fx.manager, session.id).await;
wait_for_exit(fake.pid()).await; wait_for_exit(fake.pid()).await;
fx.manager.shutdown().await; fx.manager.shutdown().await;
@@ -1017,23 +1059,31 @@ async fn shutdown_waits_for_structured_creation_and_rejects_new_creates() {
let mut env = fake.env(); let mut env = fake.env();
env.insert("FAKE_ACP_DELAY_SESSION_NEW_MS".into(), "500".into()); env.insert("FAKE_ACP_DELAY_SESSION_NEW_MS".into(), "500".into());
let create_cwd = cwd.path().to_path_buf(); // create() returns immediately; the handshake runs in the background
let manager = fx.manager.clone(); // and holds a lifecycle permit.
let create = tokio::spawn(async move { manager.create(structured(&create_cwd, env)).await }); let session = fx
.manager
.create(structured(cwd.path(), env))
.await
.expect("create");
// Wait until the adapter has received session/new (handshake in progress).
wait_for_async(Duration::from_secs(5), || async { wait_for_async(Duration::from_secs(5), || async {
fake.log().contains("recv session/new") fake.log().contains("recv session/new")
}) })
.await; .await;
// Shutdown must wait for the in-flight background handshake.
let manager = fx.manager.clone(); let manager = fx.manager.clone();
let shutdown = tokio::spawn(async move { manager.shutdown().await }); let shutdown = tokio::spawn(async move { manager.shutdown().await });
tokio::time::sleep(Duration::from_millis(50)).await; tokio::time::sleep(Duration::from_millis(50)).await;
assert!( assert!(
!shutdown.is_finished(), !shutdown.is_finished(),
"shutdown returned before create finished" "shutdown returned before handshake finished"
); );
let session = create.await.expect("create task").expect("create"); // The handshake completes in the background (500ms delay).
wait_for_ready(&fx.manager, session.id).await;
let pid = fake.pid(); let pid = fake.pid();
shutdown.await.expect("shutdown task"); shutdown.await.expect("shutdown task");
@@ -1066,27 +1116,37 @@ async fn structured_handshakes_run_independently() {
gate.to_string_lossy().into_owned(), gate.to_string_lossy().into_owned(),
); );
let first_cwd = cwd.path().to_path_buf(); // Both create() calls return immediately with Starting; the first
let manager = fx.manager.clone(); // adapter's handshake is gated while the second runs freely.
let first = let first = fx
tokio::spawn(async move { manager.create(structured(&first_cwd, blocked_env)).await }); .manager
wait_for_async(Duration::from_secs(5), || async { .create(structured(cwd.path(), blocked_env))
blocked.log().contains("recv session/new") .await
}) .expect("first create");
.await; let second = fx
.manager
.create(structured(cwd.path(), ready.env()))
.await
.expect("second create");
let second = tokio::time::timeout( // The second handshake completes independently of the first.
Duration::from_secs(5), let second_ready = wait_for_ready(&fx.manager, second.id).await;
fx.manager.create(structured(cwd.path(), ready.env())), assert_eq!(second_ready.activity, ActivityState::Idle);
)
.await // The first handshake is still blocked.
.expect("second handshake was blocked by the first") let first_snap = fx
.expect("second create"); .manager
assert_eq!(second.status, SessionStatus::Running); .snapshot()
assert!(!first.is_finished(), "the gate must still block the first"); .await
.into_iter()
.find(|s| s.id == first.id)
.unwrap();
assert_eq!(first_snap.activity, ActivityState::Starting);
std::fs::write(&gate, b"release").unwrap(); std::fs::write(&gate, b"release").unwrap();
let first = first.await.expect("first task").expect("first create"); let first_ready = wait_for_ready(&fx.manager, first.id).await;
assert_eq!(first_ready.activity, ActivityState::Idle);
fx.manager.kill(first.id, false).await.expect("kill first"); fx.manager.kill(first.id, false).await.expect("kill first");
fx.manager fx.manager
.kill(second.id, false) .kill(second.id, false)
@@ -1114,16 +1174,18 @@ async fn shutdown_waits_for_every_admitted_structured_create() {
gate_b.to_string_lossy().into_owned(), gate_b.to_string_lossy().into_owned(),
); );
let create_a = { // Both create() calls return immediately; the background handshake
let manager = fx.manager.clone(); // tasks hold lifecycle permits and are blocked on their gates.
let cwd = cwd.path().to_path_buf(); let session_a = fx
tokio::spawn(async move { manager.create(structured(&cwd, env_a)).await }) .manager
}; .create(structured(cwd.path(), env_a))
let create_b = { .await
let manager = fx.manager.clone(); .expect("create a");
let cwd = cwd.path().to_path_buf(); let session_b = fx
tokio::spawn(async move { manager.create(structured(&cwd, env_b)).await }) .manager
}; .create(structured(cwd.path(), env_b))
.await
.expect("create b");
wait_for_async(Duration::from_secs(5), || async { wait_for_async(Duration::from_secs(5), || async {
a.log().contains("recv session/new") && b.log().contains("recv session/new") a.log().contains("recv session/new") && b.log().contains("recv session/new")
}) })
@@ -1143,14 +1205,14 @@ async fn shutdown_waits_for_every_admitted_structured_create() {
assert!(!shutdown.is_finished()); assert!(!shutdown.is_finished());
std::fs::write(&gate_a, b"release").unwrap(); std::fs::write(&gate_a, b"release").unwrap();
let session_a = create_a.await.expect("create a task").expect("create a"); wait_for_ready(&fx.manager, session_a.id).await;
assert!( assert!(
!shutdown.is_finished(), !shutdown.is_finished(),
"shutdown ignored the second create" "shutdown ignored the second create"
); );
std::fs::write(&gate_b, b"release").unwrap(); std::fs::write(&gate_b, b"release").unwrap();
let session_b = create_b.await.expect("create b task").expect("create b"); wait_for_ready(&fx.manager, session_b.id).await;
shutdown.await.expect("shutdown task"); shutdown.await.expect("shutdown task");
wait_for_exit(a.pid()).await; wait_for_exit(a.pid()).await;
@@ -1166,7 +1228,7 @@ async fn shutdown_waits_for_every_admitted_structured_create() {
} }
#[tokio::test] #[tokio::test]
async fn cancelling_a_structured_handshake_rolls_back_and_settles_admission() { async fn killing_a_starting_session_rolls_back_and_settles_admission() {
let fx = DaemonFixture::new(); let fx = DaemonFixture::new();
let fake = Fake::new(); let fake = Fake::new();
let repo = TempDir::new().unwrap(); let repo = TempDir::new().unwrap();
@@ -1192,40 +1254,30 @@ async fn cancelling_a_structured_handshake_rolls_back_and_settles_admission() {
auto_approve: false, auto_approve: false,
}; };
let create = { // create() returns immediately with Starting; the handshake is gated.
let manager = fx.manager.clone(); let session = fx.manager.create(request).await.expect("create");
tokio::spawn(async move { manager.create(request).await }) assert_eq!(session.activity, ActivityState::Starting);
};
wait_for_async(Duration::from_secs(5), || async { wait_for_async(Duration::from_secs(5), || async {
fake.log().contains("recv session/new") fake.log().contains("recv session/new")
}) })
.await; .await;
let id = fake
.log()
.lines()
.find_map(|line| line.strip_prefix("env MAESTRO_SESSION_ID="))
.and_then(|id| uuid::Uuid::parse_str(id).ok())
.expect("session id in adapter log");
let pid = fake.pid(); let pid = fake.pid();
create.abort(); // Kill the session while its handshake is still in progress.
let shutdown = { // force=true because the worktree has an unpushed branch commit.
let manager = fx.manager.clone(); fx.manager.kill(session.id, true).await.expect("kill");
tokio::spawn(async move { manager.shutdown().await })
}; // Release the gate so the background task can finish and settle.
assert!(create.await.unwrap_err().is_cancelled()); std::fs::write(&gate, b"release").unwrap();
tokio::time::timeout(Duration::from_secs(5), shutdown)
// Shutdown must not hang — the lifecycle permit is settled.
tokio::time::timeout(Duration::from_secs(5), fx.manager.shutdown())
.await .await
.expect("cancelled admission kept shutdown waiting") .expect("shutdown hung on a settled permit");
.expect("shutdown task");
wait_for_exit(pid).await; wait_for_exit(pid).await;
wait_for_async(Duration::from_secs(5), || async {
!fx.fifo_dir().join(format!("{id}.fifo")).exists()
&& !repo.path().join("cancelled-create").exists()
})
.await;
assert!(fx.manager.snapshot().await.is_empty()); assert!(fx.manager.snapshot().await.is_empty());
let id = id.to_string(); let id = session.id.to_string();
assert!( assert!(
std::fs::read_dir(fx.sidecar_dir()) std::fs::read_dir(fx.sidecar_dir())
.into_iter() .into_iter()
@@ -1235,7 +1287,7 @@ async fn cancelling_a_structured_handshake_rolls_back_and_settles_admission() {
.file_name() .file_name()
.to_string_lossy() .to_string_lossy()
.starts_with(&id)), .starts_with(&id)),
"cancelled create left a sidecar artifact" "killed Starting session left a sidecar artifact"
); );
} }
@@ -1252,6 +1304,7 @@ async fn shutdown_drains_final_notifications_to_the_log() {
.create(structured(cwd.path(), env)) .create(structured(cwd.path(), env))
.await .await
.expect("create"); .expect("create");
wait_for_ready(&fx.manager, session.id).await;
wait_for_async(Duration::from_secs(5), || async { wait_for_async(Duration::from_secs(5), || async {
fx.manager fx.manager
.read_acp_events(session.id, EventCursor::Since(0), 10) .read_acp_events(session.id, EventCursor::Since(0), 10)
@@ -1282,6 +1335,7 @@ async fn adapter_exit_marks_the_session_dead_and_keeps_it_listed() {
.create(structured(cwd.path(), fake.env())) .create(structured(cwd.path(), fake.env()))
.await .await
.expect("create"); .expect("create");
wait_for_ready(&fx.manager, session.id).await;
fx.manager.start_liveness_watcher(Duration::from_millis(50)); fx.manager.start_liveness_watcher(Duration::from_millis(50));
// The agent dies out from under the daemon: the liveness pass, not tmux, // The agent dies out from under the daemon: the liveness pass, not tmux,
+11 -2
View File
@@ -154,7 +154,8 @@ async fn structured_session(h: &Harness, fake: &FakeAcp) -> Uuid {
} }
async fn structured_session_with_env(h: &Harness, env: HashMap<String, String>) -> Uuid { async fn structured_session_with_env(h: &Harness, env: HashMap<String, String>) -> Uuid {
h.manager let id = h
.manager
.create(CreateSessionRequest { .create(CreateSessionRequest {
view: SessionView::Structured, view: SessionView::Structured,
cwd: "/tmp".into(), cwd: "/tmp".into(),
@@ -166,7 +167,9 @@ async fn structured_session_with_env(h: &Harness, env: HashMap<String, String>)
}) })
.await .await
.unwrap() .unwrap()
.id .id;
wait_for_session_activity(h, id, ActivityState::Idle).await;
id
} }
async fn wait_for_session_activity(h: &Harness, id: Uuid, expected: ActivityState) { async fn wait_for_session_activity(h: &Harness, id: Uuid, expected: ActivityState) {
@@ -1494,6 +1497,12 @@ async fn events_ws_streams_structured_activity_without_hook_overrides() {
let maestro_protocol::CatalogEvent::SessionAdded { session } = next_event(&mut ws).await else { let maestro_protocol::CatalogEvent::SessionAdded { session } = next_event(&mut ws).await else {
panic!("expected structured SessionAdded") panic!("expected structured SessionAdded")
}; };
assert_eq!(session.activity, ActivityState::Starting);
// The background handshake completes and sends SessionUpdated → Idle.
let maestro_protocol::CatalogEvent::SessionUpdated { session } = next_event(&mut ws).await
else {
panic!("expected SessionUpdated after handshake")
};
assert_eq!(session.activity, ActivityState::Idle); assert_eq!(session.activity, ActivityState::Idle);
let client = reqwest::Client::new(); let client = reqwest::Client::new();
+7 -1
View File
@@ -175,6 +175,10 @@ pub enum ActivityState {
AwaitingInput, AwaitingInput,
/// An agent reported a failure, or a structured session/turn failed. /// An agent reported a failure, or a structured session/turn failed.
Error, Error,
/// The ACP adapter is booting; the session is registered but cannot accept
/// prompts yet. Older clients that do not know this variant degrade it to
/// `Idle` via the lenient deserializer — acceptable.
Starting,
} }
impl<'de> Deserialize<'de> for ActivityState { impl<'de> Deserialize<'de> for ActivityState {
@@ -187,6 +191,7 @@ impl<'de> Deserialize<'de> for ActivityState {
"Active" => Self::Active, "Active" => Self::Active,
"AwaitingInput" => Self::AwaitingInput, "AwaitingInput" => Self::AwaitingInput,
"Error" => Self::Error, "Error" => Self::Error,
"Starting" => Self::Starting,
// Known or not, any other spelling degrades to the neutral default // Known or not, any other spelling degrades to the neutral default
_ => Self::Idle, _ => Self::Idle,
}) })
@@ -225,9 +230,10 @@ where
"Idle" => Ok(ActivityState::Idle), "Idle" => Ok(ActivityState::Idle),
"AwaitingInput" => Ok(ActivityState::AwaitingInput), "AwaitingInput" => Ok(ActivityState::AwaitingInput),
"Error" => Ok(ActivityState::Error), "Error" => Ok(ActivityState::Error),
"Starting" => Ok(ActivityState::Starting),
other => Err(D::Error::unknown_variant( other => Err(D::Error::unknown_variant(
other, other,
&["Active", "Idle", "AwaitingInput", "Error"], &["Active", "Idle", "AwaitingInput", "Error", "Starting"],
)), )),
} }
} }
+38 -1
View File
@@ -22,7 +22,8 @@ use chrono::Utc;
use futures_util::{SinkExt, StreamExt}; use futures_util::{SinkExt, StreamExt};
use maestro_protocol::{ use maestro_protocol::{
routes, ApiError, CatalogEvent, CreateSessionRequest, DaemonClient, DaemonInfo, DirListing, routes, ApiError, CatalogEvent, CreateSessionRequest, DaemonClient, DaemonInfo, DirListing,
GitRoot, RecentDirs, SendInputRequest, Session, SessionStatus, PROTOCOL_VERSION, GitRoot, RecentDirs, SendInputRequest, Session, SessionStatus, StructuredPromptRequest,
PROTOCOL_VERSION,
}; };
use tokio::net::TcpListener; use tokio::net::TcpListener;
use tokio::sync::{broadcast, oneshot}; use tokio::sync::{broadcast, oneshot};
@@ -60,6 +61,8 @@ struct Inner {
forced_error: Option<(u16, ApiError)>, forced_error: Option<(u16, ApiError)>,
/// Everything `POST /api/sessions/:id/input` was asked to type, in order. /// Everything `POST /api/sessions/:id/input` was asked to type, in order.
inputs: Vec<(Uuid, SendInputRequest)>, inputs: Vec<(Uuid, SendInputRequest)>,
/// Every ACP prompt sent to a structured session, in order.
structured_prompts: Vec<(Uuid, StructuredPromptRequest)>,
} }
#[derive(Clone)] #[derive(Clone)]
@@ -92,6 +95,10 @@ impl MockDaemon {
.route(routes::SESSIONS, get(list_sessions).post(create_session)) .route(routes::SESSIONS, get(list_sessions).post(create_session))
.route(routes::patterns::SESSION, delete(kill_session)) .route(routes::patterns::SESSION, delete(kill_session))
.route(routes::patterns::SESSION_INPUT, post(send_input)) .route(routes::patterns::SESSION_INPUT, post(send_input))
.route(
routes::patterns::STRUCTURED_PROMPT,
post(structured_prompt),
)
.route(routes::INFO, get(daemon_info)) .route(routes::INFO, get(daemon_info))
.route(routes::DIRS, get(list_dir)) .route(routes::DIRS, get(list_dir))
.route(routes::RECENT_DIRS, get(recent_dirs)) .route(routes::RECENT_DIRS, get(recent_dirs))
@@ -178,6 +185,16 @@ impl MockDaemon {
self.state.inner.lock().expect("poisoned").inputs.clone() self.state.inner.lock().expect("poisoned").inputs.clone()
} }
/// Every ACP prompt this daemon received, oldest first.
pub fn structured_prompts(&self) -> Vec<(Uuid, StructuredPromptRequest)> {
self.state
.inner
.lock()
.expect("poisoned")
.structured_prompts
.clone()
}
pub fn session_count(&self) -> usize { pub fn session_count(&self) -> usize {
self.state.inner.lock().expect("poisoned").sessions.len() self.state.inner.lock().expect("poisoned").sessions.len()
} }
@@ -266,6 +283,26 @@ async fn send_input(
StatusCode::NO_CONTENT.into_response() StatusCode::NO_CONTENT.into_response()
} }
async fn structured_prompt(
State(s): State<AppState>,
Path(id): Path<Uuid>,
Json(req): Json<StructuredPromptRequest>,
) -> Response {
if let Some(r) = forced(&s) {
return r;
}
let mut guard = s.inner.lock().expect("poisoned");
if !guard.sessions.iter().any(|x| x.id == id) {
return (
StatusCode::NOT_FOUND,
Json(ApiError::new("not_found", "no such session on this daemon")),
)
.into_response();
}
guard.structured_prompts.push((id, req));
StatusCode::NO_CONTENT.into_response()
}
async fn daemon_info(State(s): State<AppState>) -> Response { async fn daemon_info(State(s): State<AppState>) -> Response {
if let Some(r) = forced(&s) { if let Some(r) = forced(&s) {
return r; return r;
+3 -2
View File
@@ -82,7 +82,7 @@ impl Surface {
pub fn activity(&self, state: ActivityState) -> Color { pub fn activity(&self, state: ActivityState) -> Color {
match state { match state {
ActivityState::Active => self.active, ActivityState::Active => self.active,
ActivityState::Idle => self.idle, ActivityState::Idle | ActivityState::Starting => self.idle,
ActivityState::AwaitingInput => self.awaiting, ActivityState::AwaitingInput => self.awaiting,
ActivityState::Error => self.error, ActivityState::Error => self.error,
} }
@@ -928,9 +928,10 @@ pub fn contrast_ratio(a: Color, b: Color) -> Option<f64> {
mod tests { mod tests {
use super::*; use super::*;
const STATES: [ActivityState; 4] = [ const STATES: [ActivityState; 5] = [
ActivityState::Active, ActivityState::Active,
ActivityState::Idle, ActivityState::Idle,
ActivityState::Starting,
ActivityState::AwaitingInput, ActivityState::AwaitingInput,
ActivityState::Error, ActivityState::Error,
]; ];
+2
View File
@@ -979,6 +979,7 @@ mod tests {
for state in [ for state in [
ActivityState::Active, ActivityState::Active,
ActivityState::Idle, ActivityState::Idle,
ActivityState::Starting,
ActivityState::AwaitingInput, ActivityState::AwaitingInput,
ActivityState::Error, ActivityState::Error,
] { ] {
@@ -1009,6 +1010,7 @@ mod tests {
for state in [ for state in [
ActivityState::Active, ActivityState::Active,
ActivityState::Idle, ActivityState::Idle,
ActivityState::Starting,
ActivityState::AwaitingInput, ActivityState::AwaitingInput,
ActivityState::Error, ActivityState::Error,
] { ] {
+1
View File
@@ -160,6 +160,7 @@ The state word is never spelled out — the colour carries it.
| ---------------- | --------- | ---------------- | | ---------------- | --------- | ---------------- |
| `Active` | `--live` | `active` | | `Active` | `--live` | `active` |
| `Idle` | `--muted` | `idle` | | `Idle` | `--muted` | `idle` |
| `Starting` | `--muted` | `idle` |
| `AwaitingInput` | `--warn` | `awaiting` | | `AwaitingInput` | `--warn` | `awaiting` |
| `Error` | `--danger`| `error` | | `Error` | `--danger`| `error` |
+1 -1
View File
@@ -11,7 +11,7 @@ export type SessionStatus = "Running" | "Dead";
/** Daemon-inferred activity dimension, orthogonal to `SessionStatus` (NAP-340). /** Daemon-inferred activity dimension, orthogonal to `SessionStatus` (NAP-340).
* Absent on payloads from daemons that predate the field. `Error` is only set * Absent on payloads from daemons that predate the field. `Error` is only set
* when an agent reports failure (#105); the daemon never infers it. */ * when an agent reports failure (#105); the daemon never infers it. */
export type ActivityState = "Active" | "Idle" | "AwaitingInput" | "Error"; export type ActivityState = "Active" | "Idle" | "AwaitingInput" | "Error" | "Starting";
export interface Session { export interface Session {
id: Uuid; id: Uuid;