diff --git a/crates/maestro-daemon/src/session.rs b/crates/maestro-daemon/src/session.rs index 4f9f0d6..6f291b3 100644 --- a/crates/maestro-daemon/src/session.rs +++ b/crates/maestro-daemon/src/session.rs @@ -1321,9 +1321,7 @@ impl SessionManager { let structured_spawn = match req.view { SessionView::Structured => { // Fail fast on an unsupported command before touching disk. - let adapter = structured_adapter(&req.command).map_err(|msg| { - SessionError::BadRequest(msg) - })?; + let adapter = structured_adapter(&req.command).map_err(SessionError::BadRequest)?; // Open the event log eagerly so the structured WS can // connect before the adapter is up. let acp_events = match self.inner.sidecar.open_events(id) { @@ -1480,40 +1478,26 @@ impl SessionManager { 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; + 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, - ); + 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; + 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.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); diff --git a/crates/maestro-daemon/tests/acp.rs b/crates/maestro-daemon/tests/acp.rs index ab4fcec..5eca410 100644 --- a/crates/maestro-daemon/tests/acp.rs +++ b/crates/maestro-daemon/tests/acp.rs @@ -130,6 +130,25 @@ async fn wait_for_exit(pid: u32) { 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] async fn structured_spawn_starts_the_adapter_in_cwd_with_session_env() { 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(); env.insert("FAKE_ACP_ECHO_ENV".into(), "MAESTRO_SESSION_ID".into()); - let session = fx + let created = fx .manager .create(structured_command( cwd.path(), @@ -148,8 +167,12 @@ async fn structured_spawn_starts_the_adapter_in_cwd_with_session_env() { .await .expect("create"); - assert_eq!(session.view, SessionView::Structured); - assert_eq!(session.status, SessionStatus::Running); + assert_eq!(created.view, SessionView::Structured); + 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); let acp_id = session.acp_session_id.clone().expect("acp session 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}" ); assert!( - log.contains(&format!("env MAESTRO_SESSION_ID={}", session.id)), + log.contains(&format!("env MAESTRO_SESSION_ID={}", created.id)), "{log}" ); @@ -232,6 +255,7 @@ async fn structured_activity_tracks_turn_completion_and_failure() { .create(structured_command(cwd.path(), env, "codex")) .await .unwrap(); + let session = wait_for_ready(&fx.manager, session.id).await; assert!(matches!( fx.manager @@ -295,11 +319,12 @@ async fn projection_snapshot_and_live_stream_share_one_fold() { let fx = DaemonFixture::new(); let fake = Fake::new(); let cwd = TempDir::new().unwrap(); - let session = fx + let created = fx .manager .create(structured(cwd.path(), fake.env())) .await .expect("create"); + let session = wait_for_ready(&fx.manager, created.id).await; wait_for_async(Duration::from_secs(5), || async { fx.manager @@ -397,6 +422,7 @@ async fn queued_prompt_is_a_durable_user_transcript_row() { .create(structured(cwd.path(), fake.env())) .await .expect("create"); + let session = wait_for_ready(&fx.manager, session.id).await; assert!(matches!( fx.manager @@ -442,6 +468,7 @@ async fn stale_turn_completion_cannot_end_the_next_turn() { .create(structured(cwd.path(), fake.env())) .await .unwrap(); + wait_for_ready(&fx.manager, session.id).await; let maestro_daemon::session::BeginAcpTurn::Started(_, first) = 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)) .await .unwrap(); + wait_for_ready(&fx.manager, session.id).await; assert!(matches!( fx.manager @@ -523,6 +551,7 @@ async fn cancellation_settlement_allows_the_next_prompt_permission() { .create(structured(cwd.path(), env)) .await .unwrap(); + wait_for_ready(&fx.manager, session.id).await; assert!(matches!( fx.manager @@ -582,11 +611,12 @@ async fn recovery_loads_the_acp_session_without_duplicating_history() { let mut env = fake.env(); env.insert("FAKE_ACP_LOAD_REPLAY".into(), "2".into()); - let session = fx + let created = fx .manager .create(structured_command(cwd.path(), env, "codex")) .await .expect("create"); + let session = wait_for_ready(&fx.manager, created.id).await; let acp_id = session.acp_session_id.clone().expect("acp id"); wait_for_async(Duration::from_secs(5), || async { fx.manager @@ -711,6 +741,7 @@ async fn shutdown_preserves_a_pristine_structured_worktree_for_recovery() { .await .expect("create"); assert_eq!(session.cwd, worktree); + wait_for_ready(&fx.manager, session.id).await; push_all_to_origin(repo.path(), origin.path()); fx.manager.shutdown().await; @@ -759,6 +790,7 @@ async fn failed_recovery_cleans_its_fifo() { .create(structured(cwd.path(), fake.env())) .await .expect("create"); + wait_for_ready(&fx.manager, session.id).await; let old_pid = fake.pid(); fx.manager.shutdown().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(), )]); - let session = fx + let created = fx .manager .create(structured_command(cwd.path(), env, "codex")) .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.activity, ActivityState::Error); 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 // dead one. - assert!(fx.manager.has(session.id).await); + assert!(fx.manager.has(created.id).await); assert!(fx .sidecar_dir() - .join(format!("{}.json", session.id)) + .join(format!("{}.json", created.id)) .exists()); - fx.manager.kill(session.id, false).await.expect("dismiss"); - assert!(!fx.manager.has(session.id).await); + fx.manager.kill(created.id, false).await.expect("dismiss"); + assert!(!fx.manager.has(created.id).await); } #[tokio::test] @@ -825,20 +860,20 @@ async fn unsupported_structured_command_fails_without_launching_claude() { let fake = Fake::new(); let cwd = TempDir::new().unwrap(); - let session = fx + let err = fx .manager .create(structured_command(cwd.path(), fake.env(), "opencode")) .await - .expect("failed structured session is persisted"); + .expect_err("unsupported command is rejected at request validation"); - assert_eq!(session.status, SessionStatus::Dead); - let detail = session.activity_detail.expect("selection error"); - assert!(detail.contains("Claude Code or Codex"), "{detail}"); + assert!( + matches!(err, maestro_daemon::SessionError::BadRequest(ref msg) if msg.contains("Claude Code or Codex")), + "{err:?}" + ); assert!( fake.log().is_empty(), "unsupported command launched an adapter" ); - fx.manager.kill(session.id, false).await.expect("dismiss"); } #[tokio::test] @@ -853,6 +888,7 @@ async fn kill_sends_cancel_then_sigterm_and_removes_the_sidecar() { .create(structured(cwd.path(), fake.env())) .await .expect("create"); + wait_for_ready(&fx.manager, session.id).await; let pid = fake.pid(); assert!(pid_alive(pid)); @@ -895,6 +931,8 @@ async fn shutdown_reaps_every_adapter_child() { .create(structured(cwd.path(), b.env())) .await .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()); 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)) .await .expect("create"); - fx.manager + let blocker_session = fx + .manager .create(structured(cwd.path(), blocker.env())) .await .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 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)) .await .expect("create"); + wait_for_ready(&fx.manager, session.id).await; wait_for_exit(fake.pid()).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(); env.insert("FAKE_ACP_DELAY_SESSION_NEW_MS".into(), "500".into()); - let create_cwd = cwd.path().to_path_buf(); - let manager = fx.manager.clone(); - let create = tokio::spawn(async move { manager.create(structured(&create_cwd, env)).await }); + // create() returns immediately; the handshake runs in the background + // and holds a lifecycle permit. + 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 { fake.log().contains("recv session/new") }) .await; + // Shutdown must wait for the in-flight background handshake. let manager = fx.manager.clone(); let shutdown = tokio::spawn(async move { manager.shutdown().await }); tokio::time::sleep(Duration::from_millis(50)).await; assert!( !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(); shutdown.await.expect("shutdown task"); @@ -1066,27 +1116,37 @@ async fn structured_handshakes_run_independently() { gate.to_string_lossy().into_owned(), ); - let first_cwd = cwd.path().to_path_buf(); - let manager = fx.manager.clone(); - let first = - tokio::spawn(async move { manager.create(structured(&first_cwd, blocked_env)).await }); - wait_for_async(Duration::from_secs(5), || async { - blocked.log().contains("recv session/new") - }) - .await; + // Both create() calls return immediately with Starting; the first + // adapter's handshake is gated while the second runs freely. + let first = fx + .manager + .create(structured(cwd.path(), blocked_env)) + .await + .expect("first create"); + let second = fx + .manager + .create(structured(cwd.path(), ready.env())) + .await + .expect("second create"); - let second = tokio::time::timeout( - Duration::from_secs(5), - fx.manager.create(structured(cwd.path(), ready.env())), - ) - .await - .expect("second handshake was blocked by the first") - .expect("second create"); - assert_eq!(second.status, SessionStatus::Running); - assert!(!first.is_finished(), "the gate must still block the first"); + // The second handshake completes independently of the first. + let second_ready = wait_for_ready(&fx.manager, second.id).await; + assert_eq!(second_ready.activity, ActivityState::Idle); + + // The first handshake is still blocked. + let first_snap = fx + .manager + .snapshot() + .await + .into_iter() + .find(|s| s.id == first.id) + .unwrap(); + assert_eq!(first_snap.activity, ActivityState::Starting); 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(second.id, false) @@ -1114,16 +1174,18 @@ async fn shutdown_waits_for_every_admitted_structured_create() { gate_b.to_string_lossy().into_owned(), ); - let create_a = { - let manager = fx.manager.clone(); - let cwd = cwd.path().to_path_buf(); - tokio::spawn(async move { manager.create(structured(&cwd, env_a)).await }) - }; - let create_b = { - let manager = fx.manager.clone(); - let cwd = cwd.path().to_path_buf(); - tokio::spawn(async move { manager.create(structured(&cwd, env_b)).await }) - }; + // Both create() calls return immediately; the background handshake + // tasks hold lifecycle permits and are blocked on their gates. + let session_a = fx + .manager + .create(structured(cwd.path(), env_a)) + .await + .expect("create a"); + let session_b = fx + .manager + .create(structured(cwd.path(), env_b)) + .await + .expect("create b"); wait_for_async(Duration::from_secs(5), || async { 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()); 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!( !shutdown.is_finished(), "shutdown ignored the second create" ); 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"); wait_for_exit(a.pid()).await; @@ -1166,7 +1228,7 @@ async fn shutdown_waits_for_every_admitted_structured_create() { } #[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 fake = Fake::new(); let repo = TempDir::new().unwrap(); @@ -1192,40 +1254,30 @@ async fn cancelling_a_structured_handshake_rolls_back_and_settles_admission() { auto_approve: false, }; - let create = { - let manager = fx.manager.clone(); - tokio::spawn(async move { manager.create(request).await }) - }; + // create() returns immediately with Starting; the handshake is gated. + let session = fx.manager.create(request).await.expect("create"); + assert_eq!(session.activity, ActivityState::Starting); wait_for_async(Duration::from_secs(5), || async { fake.log().contains("recv session/new") }) .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(); - create.abort(); - let shutdown = { - let manager = fx.manager.clone(); - tokio::spawn(async move { manager.shutdown().await }) - }; - assert!(create.await.unwrap_err().is_cancelled()); - tokio::time::timeout(Duration::from_secs(5), shutdown) + // Kill the session while its handshake is still in progress. + // force=true because the worktree has an unpushed branch commit. + fx.manager.kill(session.id, true).await.expect("kill"); + + // Release the gate so the background task can finish and settle. + std::fs::write(&gate, b"release").unwrap(); + + // Shutdown must not hang — the lifecycle permit is settled. + tokio::time::timeout(Duration::from_secs(5), fx.manager.shutdown()) .await - .expect("cancelled admission kept shutdown waiting") - .expect("shutdown task"); + .expect("shutdown hung on a settled permit"); + 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()); - let id = id.to_string(); + let id = session.id.to_string(); assert!( std::fs::read_dir(fx.sidecar_dir()) .into_iter() @@ -1235,7 +1287,7 @@ async fn cancelling_a_structured_handshake_rolls_back_and_settles_admission() { .file_name() .to_string_lossy() .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)) .await .expect("create"); + wait_for_ready(&fx.manager, session.id).await; wait_for_async(Duration::from_secs(5), || async { fx.manager .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())) .await .expect("create"); + wait_for_ready(&fx.manager, session.id).await; fx.manager.start_liveness_watcher(Duration::from_millis(50)); // The agent dies out from under the daemon: the liveness pass, not tmux, diff --git a/crates/maestro-daemon/tests/local_api.rs b/crates/maestro-daemon/tests/local_api.rs index 725f745..023bf02 100644 --- a/crates/maestro-daemon/tests/local_api.rs +++ b/crates/maestro-daemon/tests/local_api.rs @@ -154,7 +154,8 @@ async fn structured_session(h: &Harness, fake: &FakeAcp) -> Uuid { } async fn structured_session_with_env(h: &Harness, env: HashMap) -> Uuid { - h.manager + let id = h + .manager .create(CreateSessionRequest { view: SessionView::Structured, cwd: "/tmp".into(), @@ -166,7 +167,9 @@ async fn structured_session_with_env(h: &Harness, env: HashMap) }) .await .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) { @@ -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 { 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); let client = reqwest::Client::new(); diff --git a/crates/maestro-tui/src/palette.rs b/crates/maestro-tui/src/palette.rs index 1564153..5dfb169 100644 --- a/crates/maestro-tui/src/palette.rs +++ b/crates/maestro-tui/src/palette.rs @@ -928,9 +928,10 @@ pub fn contrast_ratio(a: Color, b: Color) -> Option { mod tests { use super::*; - const STATES: [ActivityState; 4] = [ + const STATES: [ActivityState; 5] = [ ActivityState::Active, ActivityState::Idle, + ActivityState::Starting, ActivityState::AwaitingInput, ActivityState::Error, ]; diff --git a/crates/maestro-tui/src/sidebar.rs b/crates/maestro-tui/src/sidebar.rs index c91e94f..6d02373 100644 --- a/crates/maestro-tui/src/sidebar.rs +++ b/crates/maestro-tui/src/sidebar.rs @@ -979,6 +979,7 @@ mod tests { for state in [ ActivityState::Active, ActivityState::Idle, + ActivityState::Starting, ActivityState::AwaitingInput, ActivityState::Error, ] { @@ -1009,6 +1010,7 @@ mod tests { for state in [ ActivityState::Active, ActivityState::Idle, + ActivityState::Starting, ActivityState::AwaitingInput, ActivityState::Error, ] { diff --git a/docs/SIDEBAR.md b/docs/SIDEBAR.md index 4ccc9eb..4524d2a 100644 --- a/docs/SIDEBAR.md +++ b/docs/SIDEBAR.md @@ -160,6 +160,7 @@ The state word is never spelled out — the colour carries it. | ---------------- | --------- | ---------------- | | `Active` | `--live` | `active` | | `Idle` | `--muted` | `idle` | +| `Starting` | `--muted` | `idle` | | `AwaitingInput` | `--warn` | `awaiting` | | `Error` | `--danger`| `error` | diff --git a/web/src/types.ts b/web/src/types.ts index eb944ba..da1afa8 100644 --- a/web/src/types.ts +++ b/web/src/types.ts @@ -11,7 +11,7 @@ export type SessionStatus = "Running" | "Dead"; /** Daemon-inferred activity dimension, orthogonal to `SessionStatus` (NAP-340). * Absent on payloads from daemons that predate the field. `Error` is only set * 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 { id: Uuid;