Merge remote-tracking branch 'origin/main' into animato-2

Mobile browsers report readyState === OPEN after background suspension
even though the OS killed the TCP connection. The per-pane visibility
handlers skipped reconnection in that case, leaving the pane frozen
until TCP keepalive fired (30–120 s on cellular).

Remove the readyState guard and unconditionally replace the socket on
visibility return — same pattern the events socket already uses. Also
let pageshow/online fire recovery without the visibilityState gate,
since some mobile browsers fire pageshow before updating that property.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

crates/maestro-acp/src/lib.rs

@@ -58,6 +58,24 @@ use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
 
 pub use agent_client_protocol::schema::v1 as schema;
 
+/// Item on the updates channel.  Notifications and turn-done markers share
+/// the same FIFO so the event pump processes every notification for a turn
+/// before seeing its done signal.
+#[derive(Debug)]
+pub enum AcpUpdate {
+    /// A raw `session/update` notification from the adapter.
+    Notification(Box<SessionNotification>),
+    /// The prompt response for `turn` arrived.  All notifications for this
+    /// turn precede this marker on the wire, and therefore in this channel.
+    /// The event pump must fire `done` after persisting every preceding
+    /// notification.
+    TurnDone {
+        turn: u64,
+        result: Result<(), String>,
+        done: oneshot::Sender<Result<(), String>>,
+    },
+}
+
 /// ACP adapter selected from the structured session's resolved command.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum AcpAdapter {
@@ -354,7 +372,7 @@ pub struct AcpSession {
     pid: u32,
     commands: mpsc::UnboundedSender<Command>,
     child_commands: mpsc::UnboundedSender<ChildCommand>,
-    updates: Mutex<Option<mpsc::UnboundedReceiver<SessionNotification>>>,
+    updates: Mutex<Option<mpsc::UnboundedReceiver<AcpUpdate>>>,
     permission_requests: Mutex<Option<mpsc::UnboundedReceiver<PermissionRequest>>>,
     exit: watch::Receiver<Option<ExitStatus>>,
 }
@@ -393,9 +411,9 @@ impl AcpSession {
         wait_exit(&self.exit).await
     }
 
-    /// The stream of raw `session/update` notifications, handed out once.
-    /// Notifications that arrived before the take are still queued on it.
-    pub fn take_updates(&self) -> Option<mpsc::UnboundedReceiver<SessionNotification>> {
+    /// The stream of updates (notifications + turn-done markers), handed out
+    /// once.  Items that arrived before the take are still queued.
+    pub fn take_updates(&self) -> Option<mpsc::UnboundedReceiver<AcpUpdate>> {
         self.updates.lock().expect("updates mutex").take()
     }
 
@@ -772,7 +790,7 @@ fn config_notification(session_id: &str, options: Vec<SessionConfigOption>) -> S
 async fn run_connection<T>(
     transport: T,
     config: ConnectionConfig,
-    updates_tx: mpsc::UnboundedSender<SessionNotification>,
+    updates_tx: mpsc::UnboundedSender<AcpUpdate>,
     permission_requests: mpsc::UnboundedSender<PermissionRequest>,
     mut commands_rx: mpsc::UnboundedReceiver<Command>,
     ready_tx: oneshot::Sender<Result<String, ReadyError>>,
@@ -800,7 +818,7 @@ async fn run_connection<T>(
                     return Ok(());
                 }
                 // A dropped receiver only means nobody wants updates any more.
-                let _ = received_updates.send(notification);
+                let _ = received_updates.send(AcpUpdate::Notification(Box::new(notification)));
                 Ok(())
             },
             agent_client_protocol::on_receive_notification!(),
@@ -916,18 +934,18 @@ async fn run_connection<T>(
                 models.as_ref(),
             )));
             if let Some(modes) = &modes {
-                let _ = updates_tx.send(SessionNotification::new(
+                let _ = updates_tx.send(AcpUpdate::Notification(Box::new(SessionNotification::new(
                     SessionId::new(session_id.as_str()),
                     SessionUpdate::CurrentModeUpdate(CurrentModeUpdate::new(
                         modes.current_mode_id.clone(),
                     )),
-                ));
+                ))));
             }
             if !config_options.lock().expect("config options").is_empty() {
-                let _ = updates_tx.send(config_notification(
+                let _ = updates_tx.send(AcpUpdate::Notification(Box::new(config_notification(
                     &session_id,
                     config_options.lock().expect("config options").clone(),
-                ));
+                ))));
             }
             let _ = ready_tx.send(Ok(session_id.clone()));
 
@@ -955,16 +973,23 @@ async fn run_connection<T>(
                             let done = Arc::new(Mutex::new(Some(done)));
                             let callback_done = done.clone();
                             let callback_turn = permission_turn.clone();
+                            let callback_updates = updates_tx.clone();
                             if let Err(error) = cx.send_request(request).on_receiving_result(move |result| {
                                 let callback_done = callback_done.clone();
                                 let callback_turn = callback_turn.clone();
+                                let callback_updates = callback_updates.clone();
                                 async move {
                                     callback_turn
                                         .lock()
                                         .expect("permission turn state")
                                         .finish(turn);
+                                    // Route through the same channel as notifications so the
+                                    // event pump sees this after every chunk for the turn.
+                                    let result = result.map(|_| ()).map_err(|e| e.to_string());
                                     if let Some(done) = callback_done.lock().expect("prompt sender").take() {
-                                        let _ = done.send(result.map(|_| ()).map_err(|e| e.to_string()));
+                                        if callback_updates.send(AcpUpdate::TurnDone { turn, result, done }).is_err() {
+                                            // Channel closed — nowhere left to signal.
+                                        }
                                     }
                                     Ok(())
                                 }
@@ -998,7 +1023,7 @@ async fn run_connection<T>(
                                 }).on_receiving_result(move |result| async move {
                                     if result.is_ok() {
                                         let options = replace_config_value(&callback_options, &callback_config_id, &callback_value);
-                                        let _ = callback_updates.send(config_notification(&callback_session, options));
+                                        let _ = callback_updates.send(AcpUpdate::Notification(Box::new(config_notification(&callback_session, options))));
                                     }
                                     if let Some(done) = callback_done.lock().expect("config sender").take() {
                                         let _ = done.send(result.map(|_| ()).map_err(|error| error.to_string()));
@@ -1017,13 +1042,13 @@ async fn run_connection<T>(
                                 )).on_receiving_result(move |result| async move {
                                     if result.is_ok() {
                                         let options = replace_config_value(&callback_options, &callback_config_id, &callback_value);
-                                        let _ = callback_updates.send(config_notification(&callback_session, options));
-                                        let _ = callback_updates.send(SessionNotification::new(
+                                        let _ = callback_updates.send(AcpUpdate::Notification(Box::new(config_notification(&callback_session, options))));
+                                        let _ = callback_updates.send(AcpUpdate::Notification(Box::new(SessionNotification::new(
                                             SessionId::new(callback_session.as_str()),
                                             SessionUpdate::CurrentModeUpdate(CurrentModeUpdate::new(
                                                 callback_value.as_value_id().expect("checked mode value").0.clone(),
                                             )),
-                                        ));
+                                        ))));
                                     }
                                     if let Some(done) = callback_done.lock().expect("config sender").take() {
                                         let _ = done.send(result.map(|_| ()).map_err(|error| error.to_string()));
@@ -1041,7 +1066,7 @@ async fn run_connection<T>(
                                             let mut current = callback_options.lock().expect("config options");
                                             current.retain(|option| option.id.0.as_ref().starts_with("__maestro_legacy_"));
                                             current.extend(response.config_options.clone());
-                                            let _ = callback_updates.send(config_notification(&callback_session, current.clone()));
+                                            let _ = callback_updates.send(AcpUpdate::Notification(Box::new(config_notification(&callback_session, current.clone()))));
                                         }
                                         if let Some(done) = callback_done.lock().expect("config sender").take() {
                                             let _ = done.send(result.map(|_| ()).map_err(|error| error.to_string()));

crates/maestro-acp/tests/spawn.rs

@@ -193,6 +193,9 @@ async fn session_updates_reach_the_taken_receiver() {
         .await
         .expect("update within 5s")
         .expect("channel open");
+    let maestro_acp::AcpUpdate::Notification(first) = first else {
+        panic!("expected Notification");
+    };
     assert_eq!(first.session_id.0.as_ref(), session.session_id());
     match first.update {
         SessionUpdate::AgentMessageChunk(chunk) => match chunk.content {
@@ -221,7 +224,10 @@ async fn advertised_settings_are_normalized_and_mutable() {
             .await
             .expect("settings update")
             .expect("channel open");
-        if let SessionUpdate::ConfigOptionUpdate(options) = update.update {
+        let maestro_acp::AcpUpdate::Notification(n) = update else {
+            continue;
+        };
+        if let SessionUpdate::ConfigOptionUpdate(options) = n.update {
             break options.config_options;
         }
     };
@@ -390,7 +396,10 @@ async fn drop_terminates_and_reaps_the_child() {
     fake.wait_for_exit().await;
 }
 
-fn chunk_text(n: &maestro_acp::schema::SessionNotification) -> String {
+fn chunk_text(update: &maestro_acp::AcpUpdate) -> String {
+    let maestro_acp::AcpUpdate::Notification(n) = update else {
+        panic!("expected Notification, got {update:?}");
+    };
     match &n.update {
         SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content {
             ContentBlock::Text(t) => t.text.clone(),

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(),
         }
     }
 

crates/maestro-daemon/src/local_api.rs

@@ -482,14 +482,13 @@ async fn structured_prompt(
     Path(id): Path<Uuid>,
     Json(body): Json<StructuredPromptRequest>,
 ) -> 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",
+    match s.manager.queue_acp_prompt(id, body.prompt, None).await {
+        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,

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<String>,
 }
 
 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<Mutex<()>>,
     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<String>,
     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))
@@ -733,15 +757,31 @@ impl SessionManager {
         &self,
         id: Uuid,
         text: String,
+        row_metadata: Option<serde_json::Value>,
     ) -> Result<QueueAcpPrompt, SessionError> {
         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,19 +802,51 @@ 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 mut data = serde_json::json!({
+                        "content": {"type": "text", "text": text},
+                    });
+                    if let Some(extra) = &row_metadata {
+                        if let (Some(base), Some(extra)) = (data.as_object_mut(), extra.as_object())
+                        {
+                            base.extend(extra.iter().map(|(k, v)| (k.clone(), v.clone())));
+                        }
                     }
                     let row = TranscriptRow {
                         id: format!("user:{}", Uuid::new_v4()),
                         kind: TranscriptRowKind::User,
-                        data: serde_json::json!({
-                            "content": {"type": "text", "text": text},
-                        }),
+                        data,
                     };
                     let event = PersistedEvent::User { row: row.clone() };
                     let seq = match stream.log.append(&event) {
@@ -789,9 +861,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 +881,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 +892,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 +912,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<String> = 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 +953,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<String> = 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 +1093,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 +1142,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 +1450,7 @@ impl SessionManager {
                             pending_approvals: HashMap::new(),
                             operation: Arc::new(Mutex::new(())),
                             dispatch: AcpDispatchState::default(),
+                            prompt_queue: VecDeque::new(),
                             activity_sequence: 0,
                         }))
                     }
@@ -1405,6 +1516,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 +1548,6 @@ impl SessionManager {
                     structured_activity_watermark: None,
                     acp_event_pump: None,
                     auto_approve: req.auto_approve,
-                    pending_prompt: None,
                 },
             );
         }
@@ -1490,8 +1601,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 +1616,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<String> =
+                                        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, None).await;
                         }
                     }
                     Err(e) => {
@@ -1680,6 +1804,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 +1843,6 @@ impl SessionManager {
                     structured_activity_watermark: None,
                     acp_event_pump: None,
                     auto_approve: session.auto_approve,
-                    pending_prompt: None,
                 },
             );
             let _ = self
@@ -1846,6 +1970,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 +2018,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 +2030,33 @@ 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. If none were queued, send a synthetic continue
+                // prompt so the agent resumes autonomously.
+                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<String> = 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, None).await;
+                } else {
+                    let _ = self
+                        .queue_acp_prompt(
+                            id,
+                            "The daemon restarted. Continue where you left off.".into(),
+                            Some(serde_json::json!({"maestro": {"type": "restart_continue"}})),
+                        )
+                        .await;
+                }
             }
             let _ = self
                 .inner
@@ -2252,34 +2403,49 @@ fn start_acp_event_pump(
     let weak_inner = Arc::downgrade(inner);
     let weak_acp = Arc::downgrade(acp);
     tokio::spawn(async move {
-        while let Some(notification) = updates.recv().await {
-            let append_error = {
-                let mut events = events.lock().await;
-                let event = PersistedEvent::Notification(Box::new(notification));
-                match events.log.append(&event) {
-                    Ok(seq) => {
-                        let logged = LoggedEvent {
-                            seq,
-                            event: event.clone(),
-                        };
-                        let projected = apply_persisted_event(&mut events.projection, &logged);
-                        let _ = events.live.send(logged);
-                        let _ = events.projected_live.send(projected);
-                        None
+        while let Some(update) = updates.recv().await {
+            match update {
+                maestro_acp::AcpUpdate::Notification(notification) => {
+                    let append_error = {
+                        let mut events = events.lock().await;
+                        let event = PersistedEvent::Notification(notification);
+                        match events.log.append(&event) {
+                            Ok(seq) => {
+                                let logged = LoggedEvent {
+                                    seq,
+                                    event: event.clone(),
+                                };
+                                let projected =
+                                    apply_persisted_event(&mut events.projection, &logged);
+                                let _ = events.live.send(logged);
+                                events.broadcast(projected);
+                                None
+                            }
+                            Err(error) => Some(error),
+                        }
+                    };
+                    if let Some(error) = append_error {
+                        tracing::error!(%id, %error, "structured event log append failed");
+                        if let Some(acp) = weak_acp.upgrade() {
+                            acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
+                                .await;
+                        }
+                        if let Some(inner) = weak_inner.upgrade() {
+                            mark_session_dead(&inner, id).await;
+                        }
+                        break;
                     }
-                    Err(error) => Some(error),
                 }
-            };
-            if let Some(error) = append_error {
-                tracing::error!(%id, %error, "structured event log append failed");
-                if let Some(acp) = weak_acp.upgrade() {
-                    acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
-                        .await;
+                maestro_acp::AcpUpdate::TurnDone {
+                    turn: _,
+                    result,
+                    done,
+                } => {
+                    // Every notification for this turn has been persisted and
+                    // broadcast above.  Signal the done oneshot so
+                    // finish_queued_acp_turn runs with the transcript complete.
+                    let _ = done.send(result);
                 }
-                if let Some(inner) = weak_inner.upgrade() {
-                    mark_session_dead(&inner, id).await;
-                }
-                break;
             }
         }
         let _ = done_tx.send(true);
@@ -2434,7 +2600,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 +2693,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 +3043,18 @@ async fn run_blocked_work_pass(inner: &Arc<Inner>) {
 /// 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<Box<dyn std::future::Future<Output = ()> + Send>> =
+        Box::pin(async move {
+            let _ = manager.queue_acp_prompt(id, text, None).await;
+        });
+    tokio::spawn(fut);
+}
+
 async fn mark_session_dead(inner: &Arc<Inner>, id: Uuid) {
     if inner.shutting_down.load(Ordering::Acquire) {
         return;
@@ -2930,7 +3108,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 +3137,7 @@ async fn mark_session_dead_with_worktree_settlement(
             return false;
         };
         record.settled = true;
+        record.session.queued_prompts.clear();
         record.session.clone()
     };
 
@@ -3120,6 +3301,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,
         }
     }

crates/maestro-daemon/tests/acp.rs

@@ -84,15 +84,27 @@ impl Fake {
 }
 
 fn event_text(event: &maestro_daemon::sidecar::LoggedEvent) -> &str {
-    let maestro_daemon::sidecar::PersistedEvent::Notification(notification) = &event.event else {
-        panic!("unexpected daemon event")
-    };
-    match &notification.update {
-        SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content {
-            ContentBlock::Text(text) => &text.text,
-            other => panic!("unexpected content {other:?}"),
-        },
-        other => panic!("unexpected update {other:?}"),
+    match &event.event {
+        maestro_daemon::sidecar::PersistedEvent::Notification(notification) => {
+            match &notification.update {
+                SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content {
+                    ContentBlock::Text(text) => &text.text,
+                    other => panic!("unexpected content {other:?}"),
+                },
+                other => panic!("unexpected update {other:?}"),
+            }
+        }
+        maestro_daemon::sidecar::PersistedEvent::User { row } => row
+            .data
+            .pointer("/content/text")
+            .and_then(serde_json::Value::as_str)
+            .expect("user row text"),
+        maestro_daemon::sidecar::PersistedEvent::Turn { row } => row
+            .data
+            .pointer("/maestroTurnMeta/outcome")
+            .and_then(serde_json::Value::as_str)
+            .unwrap_or("[turn]"),
+        maestro_daemon::sidecar::PersistedEvent::Approval { .. } => "[approval]",
     }
 }
 
@@ -259,10 +271,10 @@ async fn structured_activity_tracks_turn_completion_and_failure() {
 
     assert!(matches!(
         fx.manager
-            .queue_acp_prompt(session.id, "fail".into())
+            .queue_acp_prompt(session.id, "fail".into(), None)
             .await
             .unwrap(),
-        maestro_daemon::session::QueueAcpPrompt::Queued
+        maestro_daemon::session::QueueAcpPrompt::Dispatched
     ));
     assert_eq!(
         fx.manager
@@ -286,10 +298,10 @@ async fn structured_activity_tracks_turn_completion_and_failure() {
 
     assert!(matches!(
         fx.manager
-            .queue_acp_prompt(session.id, "succeed".into())
+            .queue_acp_prompt(session.id, "succeed".into(), None)
             .await
             .unwrap(),
-        maestro_daemon::session::QueueAcpPrompt::Queued
+        maestro_daemon::session::QueueAcpPrompt::Dispatched
     ));
     assert_eq!(
         fx.manager
@@ -426,10 +438,10 @@ async fn queued_prompt_is_a_durable_user_transcript_row() {
 
     assert!(matches!(
         fx.manager
-            .queue_acp_prompt(session.id, "show my prompt".into())
+            .queue_acp_prompt(session.id, "show my prompt".into(), None)
             .await
             .expect("queue prompt"),
-        maestro_daemon::session::QueueAcpPrompt::Queued
+        maestro_daemon::session::QueueAcpPrompt::Dispatched
     ));
 
     let (snapshot, _, _) = fx
@@ -516,15 +528,15 @@ async fn cancelled_prompt_that_never_settles_fails_the_session() {
 
     assert!(matches!(
         fx.manager
-            .queue_acp_prompt(session.id, "first".into())
+            .queue_acp_prompt(session.id, "first".into(), None)
             .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
         .manager
-        .queue_acp_prompt(session.id, "second".into())
+        .queue_acp_prompt(session.id, "second".into(), None)
         .await
         .unwrap();
     assert!(
@@ -555,18 +567,18 @@ async fn cancellation_settlement_allows_the_next_prompt_permission() {
 
     assert!(matches!(
         fx.manager
-            .queue_acp_prompt(session.id, "first".into())
+            .queue_acp_prompt(session.id, "first".into(), None)
             .await
             .unwrap(),
-        maestro_daemon::session::QueueAcpPrompt::Queued
+        maestro_daemon::session::QueueAcpPrompt::Dispatched
     ));
     assert!(fx.manager.cancel_acp_turn(session.id).await.unwrap());
     assert!(matches!(
         fx.manager
-            .queue_acp_prompt(session.id, "second".into())
+            .queue_acp_prompt(session.id, "second".into(), None)
             .await
             .unwrap(),
-        maestro_daemon::session::QueueAcpPrompt::Queued
+        maestro_daemon::session::QueueAcpPrompt::Dispatched
     ));
 
     let nonce = tokio::time::timeout(Duration::from_secs(2), async {
@@ -655,25 +667,18 @@ async fn recovery_loads_the_acp_session_without_duplicating_history() {
         revived.status(session.id).await,
         Some(SessionStatus::Running)
     );
-    assert_eq!(
-        revived
-            .snapshot()
-            .await
-            .into_iter()
-            .find(|candidate| candidate.id == session.id)
-            .unwrap()
-            .activity,
-        ActivityState::Idle
-    );
     let log = fake.log();
     assert!(
         log.contains(&format!("recv session/load sessionId={acp_id}")),
         "{log}"
     );
+    // Recovery adds: a continue-prompt User row, the adapter's post-load
+    // notification, and a turn-completed footer — 5 events total before the
+    // next manual tick.
     wait_for_async(Duration::from_secs(5), || async {
         revived
             .read_acp_events(session.id, EventCursor::Since(0), 10)
-            .is_ok_and(|events| events.len() == 3)
+            .is_ok_and(|events| events.len() == 5)
     })
     .await;
 
@@ -682,7 +687,7 @@ async fn recovery_loads_the_acp_session_without_duplicating_history() {
         .await
         .expect("replay read")
         .expect("recovered event subscription");
-    assert_eq!(replay.len(), 3);
+    assert_eq!(replay.len(), 5);
     assert!(live.try_recv().is_err(), "replay duplicated onto live");
     let new_pid = fake.pid();
     assert_ne!(new_pid, old_pid);
@@ -693,16 +698,23 @@ async fn recovery_loads_the_acp_session_without_duplicating_history() {
         .await
         .expect("live event after recovery")
         .expect("broadcast open");
-    assert_eq!(event.seq, 4);
+    assert_eq!(event.seq, 6);
     assert_eq!(event_text(&event), "tick 1");
 
     let stored = revived
         .read_acp_events(session.id, EventCursor::Since(0), 10)
         .expect("recovered log");
-    assert_eq!(stored.len(), 4, "session/load replay must not be appended");
+    assert_eq!(stored.len(), 6, "session/load replay must not be appended");
     assert_eq!(
         stored.iter().map(event_text).collect::<Vec<_>>(),
-        ["hello from fake", "tick 1", "hello after load", "tick 1"]
+        [
+            "hello from fake",
+            "tick 1",
+            "The daemon restarted. Continue where you left off.",
+            "hello after load",
+            "completed",
+            "tick 1",
+        ]
     );
 
     revived.kill(session.id, false).await.expect("kill");

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();

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(),
     }
 }
 

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(),
         }
     }
 

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<String>,
 }
 
 /// A session's process shape (#133): a plain terminal, or a structured agent
@@ -878,6 +882,10 @@ pub struct ReducedState {
     pub mode: Option<String>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub config_options: Option<Value>,
+    /// 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<String>,
 }
 
 /// 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(),
         }
     }
 

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(),
     }
 }
 

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(),
             }],
         });
     }

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(),
         }
     }
 

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(),
         }
     }
 

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],

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

web/src/App.tsx

@@ -113,15 +113,23 @@ function Shell() {
   // a transient lost connection has already recovered.
   useEffect(() => {
     const recover = () => {
-      if (document.visibilityState !== "visible") return;
       void linkStore.getState().probe();
       catalogStore.getState().reconnect();
     };
-    document.addEventListener("visibilitychange", recover);
+    // visibilitychange fires reliably on tab/app switch; guard it so we don't
+    // probe while the page is being hidden.
+    const onVisibility = () => {
+      if (document.visibilityState !== "visible") return;
+      recover();
+    };
+    // pageshow (bfcache restore) and online (network back) indicate the page
+    // is active — fire unconditionally since visibilityState may lag behind on
+    // some mobile browsers.
+    document.addEventListener("visibilitychange", onVisibility);
     window.addEventListener("pageshow", recover);
     window.addEventListener("online", recover);
     return () => {
-      document.removeEventListener("visibilitychange", recover);
+      document.removeEventListener("visibilitychange", onVisibility);
       window.removeEventListener("pageshow", recover);
       window.removeEventListener("online", recover);
     };

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<HTMLTextAreaElement>(".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<HTMLTextAreaElement>(".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 () => {

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<StructuredPaneHandle, StructuredPanePro
   const [expandedGroups, setExpandedGroups] = useState<ReadonlySet<string>>(new Set());
   const [draft, setDraft] = useState("");
   const [attachments, setAttachments] = useState<File[]>([]);
-  const [queue, setQueue] = useState<QueuedPrompt[]>([]);
   const [uploading, setUploading] = useState(false);
   const [configBusy, setConfigBusy] = useState(false);
   const [error, setError] = useState<string | null>(null);
@@ -100,7 +94,6 @@ export const StructuredPane = forwardRef<StructuredPaneHandle, StructuredPanePro
   const submission = useRef(0);
   const accepted = useRef<{ generation: number; socketGeneration: number } | null>(null);
   const optimisticStart = useRef<number | null>(null);
-  const promptPending = useRef(false);
   const [awaitingBegin, setAwaitingBegin] = useState<number | null>(null);
 
   useImperativeHandle(
@@ -166,7 +159,7 @@ export const StructuredPane = forwardRef<StructuredPaneHandle, StructuredPanePro
     const el = scrollRef.current;
     if (!el || !followingBottom.current) return;
     el.scrollTop = Math.max(0, el.scrollHeight - el.clientHeight);
-  }, [items, active, queue]);
+  }, [items, active, state.queued_prompts]);
 
   useEffect(() => {
     if (!active) return;
@@ -207,7 +200,7 @@ export const StructuredPane = forwardRef<StructuredPaneHandle, StructuredPanePro
       setAwaitingBegin(generation);
       optimisticStart.current = Date.now();
       accepted.current = null;
-      promptPending.current = true;
+
       setUploading(files.length > 0);
       setDraft("");
       setError(null);
@@ -224,7 +217,6 @@ export const StructuredPane = forwardRef<StructuredPaneHandle, StructuredPanePro
 
       void send().then(
         () => {
-          promptPending.current = false;
           if (generation !== submission.current) return;
           setUploading(false);
           setAttachments((current) => (current === files ? [] : current));
@@ -235,7 +227,6 @@ export const StructuredPane = forwardRef<StructuredPaneHandle, StructuredPanePro
           if (socketGeneration > 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<StructuredPaneHandle, StructuredPanePro
     [sessionId, socketGeneration, session],
   );
 
-  // Queued messages are concatenated into a single prompt so the agent sees
-  // one user turn instead of N back-to-back turns. The daemon rejects a
-  // second prompt while a turn is in flight (`409 turn_in_flight`), so the
-  // queue is client-local (docs/STRUCTURED-VIEW.md).
-  useEffect(() => {
-    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<StructuredPaneHandle, StructuredPanePro
                   </ToolGroup>
                 ),
               )}
-              {queue.map((queued, index) => (
+              {(state.queued_prompts ?? []).map((text, index) => (
                 <div key={`queued-${index}`} className="st-row st-row-queued">
                   <div className="st-body">
-                    <div className="st-user-text">{queued.text}</div>
+                    <div className="st-user-text">{text}</div>
                   </div>
                 </div>
               ))}

web/src/components/structured/structured.css

@@ -481,6 +481,7 @@
 }
 
 .st-tool-arg {
+  flex: 1 1 0;
   color: var(--muted);
   overflow: hidden;
   text-overflow: ellipsis;
@@ -726,7 +727,11 @@
 }
 
 .st-tool-group-title {
-  flex: 0 0 auto;
+  flex: 1 1 0;
+  min-width: 0;
+  overflow: hidden;
+  text-overflow: ellipsis;
+  white-space: nowrap;
   font-weight: 500;
 }
 

web/src/components/structured/useStructuredSession.ts

@@ -197,11 +197,12 @@ export function useStructuredSession(
     };
 
     // Mobile browsers suspend timers and silently kill WebSockets in the
-    // background. Kick a fresh connect when the page becomes visible again.
+    // background. The browser often still reports readyState === OPEN because
+    // TCP keepalive hasn't noticed the dead connection yet. Force-replace
+    // unconditionally — connect() closes the old socket first, and the server
+    // replays the full transcript on every reconnect so nothing is lost.
     const onVisibility = () => {
       if (document.visibilityState !== "visible" || disposed) return;
-      if (socket && (socket.readyState === WebSocket.OPEN || socket.readyState === WebSocket.CONNECTING)) return;
-      cancelTimer();
       setStatus("reconnecting");
       connect();
     };

web/src/components/terminal/TerminalPane.tsx

@@ -303,17 +303,22 @@ export const TerminalPane = forwardRef<TerminalPaneHandle, TerminalPaneProps>(
       ta?.addEventListener("blur", onTaBlur);
 
       // Mobile browsers (Android especially) suspend timers and silently kill
-      // WebSockets when the app goes to background. The backoff timer set by
-      // onclose is frozen too, so recovery stalls until visibility returns. Kick
-      // a fresh connect immediately when the page becomes visible again.
+      // WebSockets when the app goes to background. The browser often still
+      // reports readyState === OPEN because TCP keepalive hasn't noticed the
+      // dead connection yet. Force-replace the socket unconditionally —
+      // a genuinely alive server socket gets a clean close, and the reconnect
+      // replays the scrollback so nothing is lost.
       const onVisibility = () => {
         if (document.visibilityState !== "visible" || disposed) return;
-        if (ws.readyState === WebSocket.OPEN || ws.readyState === WebSocket.CONNECTING) return;
-        // Cancel any pending backoff — we're reconnecting right now.
         if (reconnectTimer !== null) {
           window.clearTimeout(reconnectTimer);
           reconnectTimer = null;
         }
+        // Detach the old socket's handlers so its close event doesn't race
+        // with the fresh connection (same pattern as catalogStore.reconnect).
+        ws.onclose = null;
+        ws.onerror = null;
+        ws.close();
         reconnectMs = INITIAL_RECONNECT_MS;
         onStatusChange?.("reconnecting");
         connect();

web/src/styles.css

@@ -2648,6 +2648,7 @@ input.leaf-rename:focus {
 }
 
 .st-tool-arg {
+  flex: 1 1 0;
   color: var(--muted);
   overflow: hidden;
   text-overflow: ellipsis;
@@ -2893,7 +2894,11 @@ input.leaf-rename:focus {
 }
 
 .st-tool-group-title {
-  flex: 0 0 auto;
+  flex: 1 1 0;
+  min-width: 0;
+  overflow: hidden;
+  text-overflow: ellipsis;
+  white-space: nowrap;
   font-weight: 500;
 }
 

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`. */