diff --git a/crates/maestro-cli/src/sessions.rs b/crates/maestro-cli/src/sessions.rs index 4a996de..79d40a3 100644 --- a/crates/maestro-cli/src/sessions.rs +++ b/crates/maestro-cli/src/sessions.rs @@ -346,6 +346,7 @@ mod tests { last_focused_at: None, acp_session_id: None, auto_approve: false, + queued_prompts: Vec::new(), } } diff --git a/crates/maestro-daemon/src/local_api.rs b/crates/maestro-daemon/src/local_api.rs index e0eb078..7e597d3 100644 --- a/crates/maestro-daemon/src/local_api.rs +++ b/crates/maestro-daemon/src/local_api.rs @@ -483,13 +483,12 @@ async fn structured_prompt( Json(body): Json, ) -> Response { match s.manager.queue_acp_prompt(id, body.prompt).await { - Ok(QueueAcpPrompt::Queued) | Ok(QueueAcpPrompt::Pending) => { - StatusCode::NO_CONTENT.into_response() - } - Ok(QueueAcpPrompt::InFlight) => error_response( - StatusCode::CONFLICT, - "turn_in_flight", - "the structured session already has an active turn", + Ok(QueueAcpPrompt::Dispatched) => StatusCode::NO_CONTENT.into_response(), + Ok(QueueAcpPrompt::Queued) => StatusCode::ACCEPTED.into_response(), + Ok(QueueAcpPrompt::Full) => error_response( + StatusCode::TOO_MANY_REQUESTS, + "queue_full", + "the prompt queue has reached its capacity", ), Ok(QueueAcpPrompt::Unavailable) => error_response( StatusCode::CONFLICT, diff --git a/crates/maestro-daemon/src/session.rs b/crates/maestro-daemon/src/session.rs index 1e37be5..dbe9cff 100644 --- a/crates/maestro-daemon/src/session.rs +++ b/crates/maestro-daemon/src/session.rs @@ -5,7 +5,7 @@ //! single point through which the API handlers and the per-session WS //! handler interact with tmux. -use std::collections::{HashMap, HashSet}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::path::PathBuf; use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; use std::sync::{Arc, Weak}; @@ -15,8 +15,8 @@ use chrono::Utc; use maestro_acp::{AcpAdapter, AcpSession, PermissionRequest, SpawnConfig}; use maestro_protocol::display::command_tool; use maestro_protocol::{ - ActivityState, CatalogEvent, CreateSessionRequest, ErrorInfo, NameSource, Session, - SessionStatus, SessionView, StructuredReplayEvent, TranscriptDelta, TranscriptRow, + ActivityState, CatalogEvent, CreateSessionRequest, ErrorInfo, NameSource, ReducedState, + Session, SessionStatus, SessionView, StructuredReplayEvent, TranscriptDelta, TranscriptRow, TranscriptRowKind, WorktreeSpec, NAME_KEY, NAME_SOURCE_KEY, }; use tokio::sync::{broadcast, oneshot, watch, Mutex, Notify}; @@ -122,9 +122,6 @@ pub struct SessionRecord { /// When `true`, the permission pump auto-responds `AllowAlways` to every /// ACP permission request without surfacing it to clients. 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, } impl SessionRecord { @@ -149,6 +146,9 @@ struct StructuredActivityWatermark { sequence: u64, } +/// Maximum number of prompts that can be queued while a turn is in flight. +const PROMPT_QUEUE_CAP: usize = 20; + #[derive(Debug)] struct AcpEventStream { log: EventLog, @@ -160,6 +160,10 @@ struct AcpEventStream { /// claimed: it is queued onto ACP before another operation can proceed. operation: Arc>, dispatch: AcpDispatchState, + /// Server-side FIFO of prompts accepted while a turn was in flight or the + /// adapter was still booting. Drained as a single concatenated prompt when + /// the current turn completes. Persisted to the session sidecar. + prompt_queue: VecDeque, activity_sequence: u64, } @@ -168,6 +172,23 @@ impl AcpEventStream { self.activity_sequence += 1; (self.activity_sequence, self.projection.activity()) } + + /// The reduced state with the prompt queue overlaid. Every broadcast and + /// snapshot should use this instead of `projection.state()` directly so + /// clients always see the current queue. + fn state_with_queue(&self) -> ReducedState { + let mut state = self.projection.state(); + state.queued_prompts = self.prompt_queue.iter().cloned().collect(); + state + } + + /// Broadcast a projection update with the prompt queue overlaid into the + /// state. Every `projected_live.send` should go through this so clients + /// never see a stale queue. + fn broadcast(&self, mut update: ProjectionUpdate) { + update.state.queued_prompts = self.prompt_queue.iter().cloned().collect(); + let _ = self.projected_live.send(update); + } } #[derive(Debug, Default)] @@ -235,11 +256,14 @@ pub enum BeginAcpTurn { /// Result of serializing a projected turn claim with its ACP prompt enqueue. #[derive(Debug)] pub enum QueueAcpPrompt { + /// The prompt was dispatched to ACP immediately (no turn was in flight). + Dispatched, + /// The prompt was accepted into the server-side queue and will be + /// dispatched when the current turn completes. Queued, - /// The adapter is still booting; the prompt is held until it is ready. - Pending, - InFlight, Unavailable, + /// The prompt queue has reached its capacity. + Full, Failed(String), } @@ -662,7 +686,7 @@ impl SessionManager { let live = stream.projected_live.subscribe(); Ok(Some(( stream.projection.snapshot(stream.log.last_seq()), - stream.projection.state(), + stream.state_with_queue(), live, ))) } @@ -692,7 +716,7 @@ impl SessionManager { return Ok(false); }; let projected = persist_turn_update(&mut stream, projected); - let _ = stream.projected_live.send(projected); + stream.broadcast(projected); drop(stream); sync_structured_activity(&self.inner, id).await; Ok(true) @@ -717,7 +741,7 @@ impl SessionManager { return Ok(BeginAcpTurn::InFlight); } let (turn, update) = stream.projection.begin_turn(); - let _ = stream.projected_live.send(update); + stream.broadcast(update); drop(stream); sync_structured_activity(&self.inner, id).await; Ok(BeginAcpTurn::Started(acp, turn)) @@ -735,13 +759,28 @@ impl SessionManager { text: String, ) -> Result { let (acp, stream) = { - let mut sessions = self.inner.sessions.lock().await; - 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. + let sessions = self.inner.sessions.lock().await; + let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?; if record.is_starting_structured() { - record.pending_prompt = Some(text); - return Ok(QueueAcpPrompt::Pending); + // Adapter still booting — queue the prompt for drain on boot + // success. The AcpEventStream exists, just no ACP handle yet. + let stream = record + .acp_events + .clone() + .expect("structured session has events at Starting"); + let mut stream = stream.lock().await; + if stream.prompt_queue.len() >= PROMPT_QUEUE_CAP { + return Ok(QueueAcpPrompt::Full); + } + stream.prompt_queue.push_back(text); + let update = ProjectionUpdate { + transcript: None, + state: stream.projection.state(), + }; + stream.broadcast(update); + drop(stream); + self.persist_prompt_queue(id).await; + return Ok(QueueAcpPrompt::Queued); } if !record.is_live_structured() { return Ok(QueueAcpPrompt::Unavailable); @@ -762,12 +801,37 @@ impl SessionManager { } if let Some(turn) = stream.dispatch.queued_turn { if stream.dispatch.cancelled_turn != Some(turn) { - return Ok(QueueAcpPrompt::InFlight); + // Turn is in flight — queue instead of rejecting. + if stream.prompt_queue.len() >= PROMPT_QUEUE_CAP { + return Ok(QueueAcpPrompt::Full); + } + stream.prompt_queue.push_back(text); + let update = ProjectionUpdate { + transcript: None, + state: stream.projection.state(), + }; + stream.broadcast(update); + drop(stream); + self.persist_prompt_queue(id).await; + return Ok(QueueAcpPrompt::Queued); } Some((turn, stream.dispatch.settled.clone())) } else { if stream.projection.turn_active() { - return Ok(QueueAcpPrompt::InFlight); + // Projection says active but dispatch says idle — + // queue rather than rejecting. + if stream.prompt_queue.len() >= PROMPT_QUEUE_CAP { + return Ok(QueueAcpPrompt::Full); + } + stream.prompt_queue.push_back(text); + let update = ProjectionUpdate { + transcript: None, + state: stream.projection.state(), + }; + stream.broadcast(update); + drop(stream); + self.persist_prompt_queue(id).await; + return Ok(QueueAcpPrompt::Queued); } let row = TranscriptRow { id: format!("user:{}", Uuid::new_v4()), @@ -789,9 +853,9 @@ impl SessionManager { let logged = LoggedEvent { seq, event }; let projected = stream.projection.restore_row(row); let _ = stream.live.send(logged); - let _ = stream.projected_live.send(projected); + stream.broadcast(projected); let (turn, update) = stream.projection.begin_turn(); - let _ = stream.projected_live.send(update); + stream.broadcast(update); match acp.prompt(text.clone(), turn) { Ok(done) => { stream.dispatch.queued_turn = Some(turn); @@ -809,7 +873,7 @@ impl SessionManager { if let Ok(true) = self.claim_autoname(id).await { crate::autoname::spawn_naming(self.clone(), id, text); } - return Ok(QueueAcpPrompt::Queued); + return Ok(QueueAcpPrompt::Dispatched); } Err(message) => { cancel_pending_approvals(&mut stream); @@ -820,7 +884,7 @@ impl SessionManager { "newly claimed turn remains current while stream is locked", ); let projected = persist_turn_update(&mut stream, projected); - let _ = stream.projected_live.send(projected); + stream.broadcast(projected); drop(stream); sync_structured_activity(&self.inner, id).await; return Ok(QueueAcpPrompt::Failed(message)); @@ -840,6 +904,27 @@ impl SessionManager { } } + /// Persist the current prompt queue to the session sidecar. + async fn persist_prompt_queue(&self, id: Uuid) { + let stream_arc = { + let sessions = self.inner.sessions.lock().await; + sessions.get(&id).and_then(|r| r.acp_events.clone()) + }; + let Some(stream_arc) = stream_arc else { return }; + let queue: Vec = stream_arc + .lock() + .await + .prompt_queue + .iter() + .cloned() + .collect(); + let mut sessions = self.inner.sessions.lock().await; + if let Some(record) = sessions.get_mut(&id) { + record.session.queued_prompts = queue; + let _ = self.inner.sidecar.write(&record.session); + } + } + async fn finish_queued_acp_turn(&self, id: Uuid, turn: u64, lifecycle: TurnLifecycle) { let stream = self .inner @@ -860,16 +945,33 @@ impl SessionManager { stream.dispatch.cancelled_turn = None; } stream.dispatch.settled.notify_waiters(); - if stream.projection.active_turn() != Some(turn) { - return; - } - cancel_pending_approvals(&mut stream); - if let Some(projected) = stream.projection.turn_lifecycle(lifecycle, turn) { - let projected = persist_turn_update(&mut stream, projected); - let _ = stream.projected_live.send(projected); + if stream.projection.active_turn() == Some(turn) { + cancel_pending_approvals(&mut stream); + if let Some(projected) = stream.projection.turn_lifecycle(lifecycle, turn) { + let projected = persist_turn_update(&mut stream, projected); + stream.broadcast(projected); + } } + // Drain the prompt queue: concatenate all queued prompts into a single + // prompt and dispatch as the next turn. This must run even when the + // projection already ended the turn (e.g. cancel settled via the event + // stream before this function ran). + let queued_text = if stream.prompt_queue.is_empty() { + None + } else { + let texts: Vec = stream.prompt_queue.drain(..).collect(); + Some(texts.join("\n\n")) + }; drop(stream); sync_structured_activity(&self.inner, id).await; + if let Some(text) = queued_text { + self.persist_prompt_queue(id).await; + // Cannot call queue_acp_prompt here (async cycle via the spawn + // inside queue_acp_prompt → finish_queued_acp_turn → here). + // Spawn an independent task instead. + let manager = self.clone(); + drain_prompt_queue(manager, id, text); + } } async fn fail_cancelled_acp_turn( @@ -983,7 +1085,7 @@ impl SessionManager { .turn_lifecycle(TurnLifecycle::Cancel, turn) .expect("active turn remains current while stream is locked"); let projected = persist_turn_update(&mut stream, projected); - let _ = stream.projected_live.send(projected); + stream.broadcast(projected); drop(stream); sync_structured_activity(&self.inner, id).await; if let Some(turn) = cancelled_queued_turn { @@ -1032,7 +1134,7 @@ impl SessionManager { let outcome = permission_outcome(decision, &pending.options); let _ = pending.response.send(outcome); if let Some(update) = stream.projection.resolve_approval(nonce) { - let _ = stream.projected_live.send(update); + stream.broadcast(update); } drop(stream); sync_structured_activity(&self.inner, id).await; @@ -1340,6 +1442,7 @@ impl SessionManager { pending_approvals: HashMap::new(), operation: Arc::new(Mutex::new(())), dispatch: AcpDispatchState::default(), + prompt_queue: VecDeque::new(), activity_sequence: 0, })) } @@ -1405,6 +1508,7 @@ impl SessionManager { acp_session_id: None, view: req.view, auto_approve: req.auto_approve, + queued_prompts: Vec::new(), }; if let Err(e) = self.inner.sidecar.write(&session) { @@ -1436,7 +1540,6 @@ impl SessionManager { structured_activity_watermark: None, acp_event_pump: None, auto_approve: req.auto_approve, - pending_prompt: None, }, ); } @@ -1490,8 +1593,8 @@ impl SessionManager { 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 = { + // Drain any prompts queued while the adapter was Starting. + let queued_text = { let mut sessions = inner.sessions.lock().await; let Some(record) = sessions.get_mut(&id) else { // Session was killed while we were booting. @@ -1505,17 +1608,30 @@ impl SessionManager { record.session.acp_session_id = Some(acp.session_id().to_string()); record.session.activity = ActivityState::Idle; record.session.activity_since = Utc::now(); + let queued = if let Some(ref stream) = record.acp_events { + let mut stream = stream.lock().await; + if stream.prompt_queue.is_empty() { + None + } else { + let texts: Vec = + stream.prompt_queue.drain(..).collect(); + Some(texts.join("\n\n")) + } + } else { + None + }; + record.session.queued_prompts.clear(); let _ = inner.sidecar.write(&record.session); let _ = inner.events.send(CatalogEvent::SessionUpdated { session: record.session.clone(), }); - record.pending_prompt.take() + queued }; - if let Some(prompt) = pending_prompt { + if let Some(text) = queued_text { let manager = SessionManager { inner: inner.clone(), }; - let _ = manager.queue_acp_prompt(id, prompt).await; + let _ = manager.queue_acp_prompt(id, text).await; } } Err(e) => { @@ -1680,6 +1796,7 @@ impl SessionManager { // other field's forward-compat default. view: maestro_protocol::SessionView::Terminal, auto_approve: false, + queued_prompts: Vec::new(), }, }; let fifo_path = self.inner.fifo_dir.join(format!("{id}.fifo")); @@ -1718,7 +1835,6 @@ impl SessionManager { structured_activity_watermark: None, acp_event_pump: None, auto_approve: session.auto_approve, - pending_prompt: None, }, ); let _ = self @@ -1846,6 +1962,7 @@ impl SessionManager { pending_approvals: HashMap::new(), operation: Arc::new(Mutex::new(())), dispatch: AcpDispatchState::default(), + prompt_queue: session.queued_prompts.drain(..).collect(), activity_sequence: 0, }))) } @@ -1893,11 +2010,10 @@ impl SessionManager { focused_since: None, awaiting_suppressed: false, acp, - acp_events, + acp_events: acp_events.clone(), structured_activity_watermark: None, acp_event_pump: None, auto_approve: session.auto_approve, - pending_prompt: None, }, ); if let Some((acp, events)) = event_pump { @@ -1906,6 +2022,23 @@ impl SessionManager { if let Some(record) = self.inner.sessions.lock().await.get_mut(&id) { record.acp_event_pump = Some(done); } + // Drain any prompts that were queued before the daemon restarted. + let queued_text = { + let stream = acp_events + .as_ref() + .expect("recovery with acp implies acp_events"); + let mut stream = stream.lock().await; + if stream.prompt_queue.is_empty() { + None + } else { + let texts: Vec = stream.prompt_queue.drain(..).collect(); + Some(texts.join("\n\n")) + } + }; + if let Some(text) = queued_text { + self.persist_prompt_queue(id).await; + let _ = self.queue_acp_prompt(id, text).await; + } } let _ = self .inner @@ -2264,7 +2397,7 @@ fn start_acp_event_pump( }; let projected = apply_persisted_event(&mut events.projection, &logged); let _ = events.live.send(logged); - let _ = events.projected_live.send(projected); + events.broadcast(projected); None } Err(error) => Some(error), @@ -2434,7 +2567,7 @@ fn start_acp_permission_pump( let outcome = record_permission_request(&mut events, request, response); let fatal = matches!(outcome, RecordOutcome::Fatal); if let RecordOutcome::Live(update) = outcome { - let _ = events.projected_live.send(*update); + events.broadcast(*update); } drop(events); if fatal { @@ -2527,7 +2660,7 @@ fn cancel_pending_approvals(events: &mut AcpEventStream) { .send(maestro_acp::schema::RequestPermissionOutcome::Cancelled); } if let Some(update) = events.projection.clear_approval() { - let _ = events.projected_live.send(update); + events.broadcast(update); } } @@ -2877,6 +3010,18 @@ async fn run_blocked_work_pass(inner: &Arc) { /// Terminal sessions that die naturally do not survive a daemon restart: /// [`SessionManager::recover`] adopts only what tmux still holds. Structured /// sessions stopped for daemon shutdown retain their sidecars for ACP recovery. +/// Spawn an independent task to dispatch the next prompt from the drained +/// queue. Free function with an explicit `Box::pin` future to break the +/// compiler's async-cycle check: `queue_acp_prompt` → spawn → +/// `finish_queued_acp_turn` → here → `queue_acp_prompt`. +fn drain_prompt_queue(manager: SessionManager, id: Uuid, text: String) { + let fut: std::pin::Pin + Send>> = + Box::pin(async move { + let _ = manager.queue_acp_prompt(id, text).await; + }); + tokio::spawn(fut); +} + async fn mark_session_dead(inner: &Arc, id: Uuid) { if inner.shutting_down.load(Ordering::Acquire) { return; @@ -2930,7 +3075,9 @@ async fn mark_session_dead_with_worktree_settlement( .get(&id) .and_then(|record| record.acp_events.clone()); if let Some(stream) = stream { - cancel_pending_approvals(&mut *stream.lock().await); + let mut stream = stream.lock().await; + cancel_pending_approvals(&mut stream); + stream.prompt_queue.clear(); } tracing::info!(%id, "session process gone — marking dead"); @@ -2957,6 +3104,7 @@ async fn mark_session_dead_with_worktree_settlement( return false; }; record.settled = true; + record.session.queued_prompts.clear(); record.session.clone() }; @@ -3120,6 +3268,7 @@ mod tests { pending_approvals: std::collections::HashMap::new(), operation: std::sync::Arc::new(tokio::sync::Mutex::new(())), dispatch: AcpDispatchState::default(), + prompt_queue: std::collections::VecDeque::new(), activity_sequence: 0, } } diff --git a/crates/maestro-daemon/tests/acp.rs b/crates/maestro-daemon/tests/acp.rs index 5eca410..d0a008c 100644 --- a/crates/maestro-daemon/tests/acp.rs +++ b/crates/maestro-daemon/tests/acp.rs @@ -262,7 +262,7 @@ async fn structured_activity_tracks_turn_completion_and_failure() { .queue_acp_prompt(session.id, "fail".into()) .await .unwrap(), - maestro_daemon::session::QueueAcpPrompt::Queued + maestro_daemon::session::QueueAcpPrompt::Dispatched )); assert_eq!( fx.manager @@ -289,7 +289,7 @@ async fn structured_activity_tracks_turn_completion_and_failure() { .queue_acp_prompt(session.id, "succeed".into()) .await .unwrap(), - maestro_daemon::session::QueueAcpPrompt::Queued + maestro_daemon::session::QueueAcpPrompt::Dispatched )); assert_eq!( fx.manager @@ -429,7 +429,7 @@ async fn queued_prompt_is_a_durable_user_transcript_row() { .queue_acp_prompt(session.id, "show my prompt".into()) .await .expect("queue prompt"), - maestro_daemon::session::QueueAcpPrompt::Queued + maestro_daemon::session::QueueAcpPrompt::Dispatched )); let (snapshot, _, _) = fx @@ -519,7 +519,7 @@ async fn cancelled_prompt_that_never_settles_fails_the_session() { .queue_acp_prompt(session.id, "first".into()) .await .unwrap(), - maestro_daemon::session::QueueAcpPrompt::Queued + maestro_daemon::session::QueueAcpPrompt::Dispatched )); assert!(fx.manager.cancel_acp_turn(session.id).await.unwrap()); let second = fx @@ -558,7 +558,7 @@ async fn cancellation_settlement_allows_the_next_prompt_permission() { .queue_acp_prompt(session.id, "first".into()) .await .unwrap(), - maestro_daemon::session::QueueAcpPrompt::Queued + maestro_daemon::session::QueueAcpPrompt::Dispatched )); assert!(fx.manager.cancel_acp_turn(session.id).await.unwrap()); assert!(matches!( @@ -566,7 +566,7 @@ async fn cancellation_settlement_allows_the_next_prompt_permission() { .queue_acp_prompt(session.id, "second".into()) .await .unwrap(), - maestro_daemon::session::QueueAcpPrompt::Queued + maestro_daemon::session::QueueAcpPrompt::Dispatched )); let nonce = tokio::time::timeout(Duration::from_secs(2), async { diff --git a/crates/maestro-daemon/tests/local_api.rs b/crates/maestro-daemon/tests/local_api.rs index 90a1855..95daa9c 100644 --- a/crates/maestro-daemon/tests/local_api.rs +++ b/crates/maestro-daemon/tests/local_api.rs @@ -293,7 +293,7 @@ async fn structured_attachments_preserve_bytes_and_reject_paths() { } #[tokio::test] -async fn structured_prompt_rejects_a_second_in_flight_turn_and_cancel_clears_it() { +async fn structured_prompt_queues_during_active_turn_and_cancel_clears_it() { let h = spawn("naps-laptop").await; let fake = FakeAcp::new(); let mut env = fake.env(); @@ -310,6 +310,7 @@ async fn structured_prompt_rejects_a_second_in_flight_turn_and_cancel_clears_it( .await .unwrap(); assert_eq!(first.status(), reqwest::StatusCode::NO_CONTENT); + // A second prompt while a turn is active is accepted into the queue. let second = client .post(url( h.addr, @@ -319,9 +320,7 @@ async fn structured_prompt_rejects_a_second_in_flight_turn_and_cancel_clears_it( .send() .await .unwrap(); - assert_eq!(second.status(), reqwest::StatusCode::CONFLICT); - let error: ApiError = second.json().await.unwrap(); - assert_eq!(error.error.code, "turn_in_flight"); + assert_eq!(second.status(), reqwest::StatusCode::ACCEPTED); let cancel = client .post(url( @@ -332,16 +331,8 @@ async fn structured_prompt_rejects_a_second_in_flight_turn_and_cancel_clears_it( .await .unwrap(); assert_eq!(cancel.status(), reqwest::StatusCode::NO_CONTENT); - let next = client - .post(url( - h.addr, - &format!("/api/sessions/{id}/structured/prompt"), - )) - .json(&serde_json::json!({"prompt": "next"})) - .send() - .await - .unwrap(); - assert_eq!(next.status(), reqwest::StatusCode::NO_CONTENT); + // After cancel + settlement, the queued "second" prompt drains. Wait for + // both the first and the drained prompt to reach ACP. tokio::time::timeout(Duration::from_secs(2), async { loop { if fake.log().matches("recv session/prompt").count() == 2 { @@ -351,7 +342,7 @@ async fn structured_prompt_rejects_a_second_in_flight_turn_and_cancel_clears_it( } }) .await - .expect("both prompt commands reach ACP"); + .expect("first and drained queued prompt both reach ACP"); let log = fake.log(); let first_prompt = log.find("recv session/prompt").unwrap(); let cancel = log.find("recv session/cancel").unwrap(); diff --git a/crates/maestro-daemon/tests/sidecar.rs b/crates/maestro-daemon/tests/sidecar.rs index 426fb75..b6c6bf3 100644 --- a/crates/maestro-daemon/tests/sidecar.rs +++ b/crates/maestro-daemon/tests/sidecar.rs @@ -35,6 +35,7 @@ fn sample(id: Uuid) -> Session { acp_session_id: None, view: SessionView::Terminal, auto_approve: false, + queued_prompts: Vec::new(), } } diff --git a/crates/maestro-protocol/src/display.rs b/crates/maestro-protocol/src/display.rs index 7c326ac..b4f887c 100644 --- a/crates/maestro-protocol/src/display.rs +++ b/crates/maestro-protocol/src/display.rs @@ -745,6 +745,7 @@ mod tests { last_focused_at: None, acp_session_id: None, auto_approve: false, + queued_prompts: Vec::new(), } } diff --git a/crates/maestro-protocol/src/lib.rs b/crates/maestro-protocol/src/lib.rs index ac48117..00326da 100644 --- a/crates/maestro-protocol/src/lib.rs +++ b/crates/maestro-protocol/src/lib.rs @@ -138,6 +138,10 @@ pub struct Session { /// a daemon restart. #[serde(default)] pub auto_approve: bool, + /// Prompts queued on the daemon while a turn was in flight or the adapter + /// was still booting. Persisted so the queue survives a daemon restart. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub queued_prompts: Vec, } /// A session's process shape (#133): a plain terminal, or a structured agent @@ -878,6 +882,10 @@ pub struct ReducedState { pub mode: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub config_options: Option, + /// Prompts queued on the daemon while a turn is in flight. Drained as + /// a single concatenated prompt when the current turn completes. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub queued_prompts: Vec, } /// Server messages sent over a structured-session WebSocket. @@ -1061,6 +1069,7 @@ mod tests { view: SessionView::Terminal, acp_session_id: None, auto_approve: false, + queued_prompts: Vec::new(), } } diff --git a/crates/maestro-testing/src/lib.rs b/crates/maestro-testing/src/lib.rs index ef7806b..1f374cd 100644 --- a/crates/maestro-testing/src/lib.rs +++ b/crates/maestro-testing/src/lib.rs @@ -51,6 +51,7 @@ pub fn session(id: Uuid, cwd: &str, command: &str) -> Session { last_focused_at: None, acp_session_id: None, auto_approve: false, + queued_prompts: Vec::new(), } } diff --git a/crates/maestro-tui/src/app.rs b/crates/maestro-tui/src/app.rs index 8849809..03b0799 100644 --- a/crates/maestro-tui/src/app.rs +++ b/crates/maestro-tui/src/app.rs @@ -2217,6 +2217,7 @@ mod tests { last_focused_at: None, acp_session_id: None, auto_approve: false, + queued_prompts: Vec::new(), }) .collect(), }); @@ -2401,6 +2402,7 @@ mod tests { last_focused_at: None, acp_session_id: None, auto_approve: false, + queued_prompts: Vec::new(), }], }); app.dispatch(Action::SidebarNext); @@ -2431,6 +2433,7 @@ mod tests { last_focused_at: None, acp_session_id: None, auto_approve: false, + queued_prompts: Vec::new(), } } @@ -3594,6 +3597,7 @@ mod tests { last_focused_at: focused, acp_session_id: None, auto_approve: false, + queued_prompts: Vec::new(), }; let older = Utc::now() - chrono::Duration::seconds(60); app.catalog.apply_event(CatalogEvent::Snapshot { @@ -3875,6 +3879,7 @@ mod tests { last_focused_at: None, acp_session_id: None, auto_approve: false, + queued_prompts: Vec::new(), }], }); app.dispatch(Action::KillHighlighted); @@ -3913,6 +3918,7 @@ mod tests { last_focused_at: None, acp_session_id: None, auto_approve: false, + queued_prompts: Vec::new(), }], }); } diff --git a/crates/maestro-tui/src/catalog.rs b/crates/maestro-tui/src/catalog.rs index d734493..365bf42 100644 --- a/crates/maestro-tui/src/catalog.rs +++ b/crates/maestro-tui/src/catalog.rs @@ -226,6 +226,7 @@ mod tests { last_focused_at: None, acp_session_id: None, auto_approve: false, + queued_prompts: Vec::new(), } } diff --git a/crates/maestro-tui/src/command_bar.rs b/crates/maestro-tui/src/command_bar.rs index 2feace8..7941f62 100644 --- a/crates/maestro-tui/src/command_bar.rs +++ b/crates/maestro-tui/src/command_bar.rs @@ -812,6 +812,7 @@ mod tests { last_focused_at: None, acp_session_id: None, auto_approve: false, + queued_prompts: Vec::new(), } } diff --git a/crates/maestro-tui/src/sidebar.rs b/crates/maestro-tui/src/sidebar.rs index d397734..05ab6a3 100644 --- a/crates/maestro-tui/src/sidebar.rs +++ b/crates/maestro-tui/src/sidebar.rs @@ -503,6 +503,7 @@ mod tests { last_focused_at: None, acp_session_id: None, auto_approve: false, + queued_prompts: Vec::new(), } } @@ -1274,6 +1275,7 @@ mod tests { last_focused_at: None, acp_session_id: None, auto_approve: false, + queued_prompts: Vec::new(), }; let sb = Sidebar { sessions: vec![s], diff --git a/docs/STRUCTURED-VIEW.md b/docs/STRUCTURED-VIEW.md index baf582a..da09631 100644 --- a/docs/STRUCTURED-VIEW.md +++ b/docs/STRUCTURED-VIEW.md @@ -353,21 +353,26 @@ thing a callback can be attributed to, so at most one turn may ever be in flight, and a cancelled turn must fully settle before the next prompt is allowed to claim the daemon's idea of "the active turn." -- Web clients may collect a visible FIFO of operator messages while a turn is - active. They submit each queued message only after the projected turn becomes - idle; the queue is client-local and unsent messages do not survive a reload. +- Prompts are accepted at any time: if the session is idle the daemon dispatches + immediately (`204`); if a turn is active the prompt is appended to a + server-side FIFO (`202 Accepted`, capped at 20 entries, `429` when full). + When the active turn completes, all queued prompts are concatenated with + `\n\n` and dispatched as the next turn. The queue is persisted in the session + sidecar and survives daemon restarts — clients never queue locally. + `ReducedState.queued_prompts` contains the current queue so every connected + client can render the pending messages. - A submitted prompt is appended as a durable `User` row before it claims the turn and dispatches to ACP, so the transcript never depends on an adapter echoing the - operator's input. If that append fails, the prompt is not dispatched. While a - turn is active, a second prompt gets `409 turn_in_flight` rather than - queuing in the daemon — one active turn per session, never a hidden - server-side queue of prompts. + operator's input. If that append fails, the prompt is not dispatched. - Cancel sends `session/cancel`, clears any pending approval as `Cancelled`, - and ends the turn's client-visible lifecycle immediately. If the cancelled - turn was still *dispatched* to ACP (waiting on its response), the daemon - cannot simply forget it: ACP has not confirmed the cancel yet, and a new - prompt claiming "the active turn" while the old one is still outstanding - would make the next permission callback ambiguous between the two. + and ends the turn's client-visible lifecycle immediately. The prompt queue + is preserved: after the cancelled turn settles, the queued prompts drain as + the next turn. If the session dies instead (settlement timeout), the queue + is cleared. If the cancelled turn was still *dispatched* to ACP (waiting on + its response), the daemon cannot simply forget it: ACP has not confirmed the + cancel yet, and a new prompt claiming "the active turn" while the old one is + still outstanding would make the next permission callback ambiguous between + the two. - So a cancelled-but-still-dispatched turn must **settle within `CANCEL_GRACE` (500ms)** — either ACP acknowledges the prompt (however it resolves) or that grace period runs out. If it settles in time, the next diff --git a/web/src/components/structured/StructuredPane.test.tsx b/web/src/components/structured/StructuredPane.test.tsx index 3d95390..8e62de2 100644 --- a/web/src/components/structured/StructuredPane.test.tsx +++ b/web/src/components/structured/StructuredPane.test.tsx @@ -308,7 +308,7 @@ describe("prompt and cancel", () => { ); }); - it("queues a second message while a turn is in flight, and sends it when the turn ends", async () => { + it("always POSTs a prompt even while a turn is active (server queues it)", async () => { const { socket } = mount(); snapshot(socket, [], busy); fireEvent.change(document.querySelector(".structured-input")!, { @@ -316,31 +316,17 @@ describe("prompt and cancel", () => { }); fireEvent.submit(document.querySelector("form.structured-composer")!); - expect(sendStructuredPrompt).not.toHaveBeenCalled(); - expect(document.querySelector(".st-row-queued")?.textContent).toBe("next up"); - - socket.emit({ type: "reduced_state", state: idle }); await waitFor(() => expect(sendStructuredPrompt).toHaveBeenCalledWith("s1", "next up")); }); - it("concatenates multiple queued messages into a single prompt", async () => { + it("renders server-side queued prompts from state", () => { const { socket } = mount(); - snapshot(socket, [], busy); - const input = document.querySelector(".structured-input")!; - const form = document.querySelector("form.structured-composer")!; + snapshot(socket, [], { ...busy, queued_prompts: ["first", "second"] }); - fireEvent.change(input, { target: { value: "first" } }); - fireEvent.submit(form); - fireEvent.change(input, { target: { value: "second" } }); - fireEvent.submit(form); - - expect(sendStructuredPrompt).not.toHaveBeenCalled(); const queued = document.querySelectorAll(".st-row-queued"); expect(queued).toHaveLength(2); - - socket.emit({ type: "reduced_state", state: idle }); - await waitFor(() => expect(sendStructuredPrompt).toHaveBeenCalledWith("s1", "first\n\nsecond")); - expect(sendStructuredPrompt).toHaveBeenCalledTimes(1); + expect(queued[0].textContent).toBe("first"); + expect(queued[1].textContent).toBe("second"); }); it("hands a failed send's draft back to the operator", async () => { diff --git a/web/src/components/structured/StructuredPane.tsx b/web/src/components/structured/StructuredPane.tsx index 1c4b0ab..ea74570 100644 --- a/web/src/components/structured/StructuredPane.tsx +++ b/web/src/components/structured/StructuredPane.tsx @@ -48,11 +48,6 @@ import "./structured.css"; * count as following. */ const FOLLOW_BOTTOM_THRESHOLD_PX = 64; -interface QueuedPrompt { - text: string; - attachments: File[]; -} - export interface StructuredPaneProps { sessionId: Uuid; /** Grabs the composer once mounted. A sidebar preview passes `false`. */ @@ -85,7 +80,6 @@ export const StructuredPane = forwardRef>(new Set()); const [draft, setDraft] = useState(""); const [attachments, setAttachments] = useState([]); - const [queue, setQueue] = useState([]); const [uploading, setUploading] = useState(false); const [configBusy, setConfigBusy] = useState(false); const [error, setError] = useState(null); @@ -100,7 +94,6 @@ export const StructuredPane = forwardRef(null); const optimisticStart = useRef(null); - const promptPending = useRef(false); const [awaitingBegin, setAwaitingBegin] = useState(null); useImperativeHandle( @@ -166,7 +159,7 @@ export const StructuredPane = forwardRef { if (!active) return; @@ -207,7 +200,7 @@ export const StructuredPane = forwardRef 0); setDraft(""); setError(null); @@ -224,7 +217,6 @@ export const StructuredPane = forwardRef { - promptPending.current = false; if (generation !== submission.current) return; setUploading(false); setAttachments((current) => (current === files ? [] : current)); @@ -235,7 +227,6 @@ export const StructuredPane = forwardRef submissionSocket) session.reconnect(); }, (e: unknown) => { - promptPending.current = false; if (generation !== submission.current) return; setAwaitingBegin((current) => (current === generation ? null : current)); optimisticStart.current = null; @@ -251,26 +242,10 @@ export const StructuredPane = forwardRef { - if (active || promptPending.current || queue.length === 0) return; - const text = queue.map((q) => q.text).join("\n\n"); - const files = queue.flatMap((q) => q.attachments); - setQueue([]); - dispatchPrompt(text, files); - }, [active, queue, dispatchPrompt]); - + // The server accepts prompts at any time: if a turn is active the prompt is + // queued server-side and drained when the turn completes. The client always + // POSTs immediately — no local queue. const submit = (text: string, files: File[]) => { - if (active || queue.length > 0) { - setQueue((current) => [...current, { text, attachments: files }]); - setAttachments([]); - setDraft(""); - setError(null); - return; - } dispatchPrompt(text, files); }; @@ -365,10 +340,10 @@ export const StructuredPane = forwardRef ), )} - {queue.map((queued, index) => ( + {(state.queued_prompts ?? []).map((text, index) => (
-
{queued.text}
+
{text}
))} diff --git a/web/src/types.ts b/web/src/types.ts index da1afa8..cd4adad 100644 --- a/web/src/types.ts +++ b/web/src/types.ts @@ -441,6 +441,7 @@ export interface ReducedState { model?: string | null; mode?: string | null; config_options?: { configOptions: AcpSessionConfigOption[] } | null; + queued_prompts?: string[]; } /** Server messages on `WS /api/structured/:id`. */