Compare commits

...
14 Commits
Author SHA1 Message Date
Miguel Palhas ce0e4e6820 Merge remote-tracking branch 'origin/main' into mandolin
ci / web (push) Failing after 45s
ci / flake (push) Successful in 4m17s
ci / rust (push) Successful in 8m4s
# Conflicts:
#	crates/maestro-daemon/src/local_api.rs
#	crates/maestro-daemon/src/session.rs
2026-09-12 17:54:13 +01:00
Miguel Palhas fa3d6c82b2 Merge remote-tracking branch 'origin/main' into chord
ci / web (push) Failing after 26s
ci / flake (push) Successful in 4m6s
ci / rust (push) Successful in 7m29s
2026-09-12 14:10:59 +01:00
Miguel PalhasandClaude Opus 4.6 ebe25a1a55 feat(maestro): server-side prompt queue for structured sessions
Prompts sent while a turn is active are queued on the daemon instead of
in the browser's React state. The queue is persisted in the session
sidecar and drained as the next turn when the current one completes.

Fixes the bug where queued messages were delivered to the wrong session
when switching focus in the UI: the client-side queue captured the new
sessionId in its drain closure.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-09-12 14:10:55 +01:00
Miguel PalhasandClaude Opus 4.6 923911f631 feat(maestro): auto-continue ACP sessions after restart
After a graceful daemon restart, recovered structured sessions now
receive a synthetic continue prompt so the agent resumes without
manual intervention. The User row carries `"maestro": {"type":
"restart_continue"}` metadata for future UI differentiation.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-09-12 11:30:36 +01:00
naps62-yolo 8cbe8510ac fix(maestro): truncate long commands in tool group header (#258)
ci / web (push) Failing after 56s
ci / flake (push) Successful in 5m2s
ci / rust (push) Successful in 9m34s
2026-09-12 11:08:04 +01:00
Miguel Palhas cd5bbcfcb1 Merge remote-tracking branch 'origin/main' into timbre-2
ci / web (push) Failing after 1m7s
ci / flake (push) Successful in 3m55s
ci / rust (push) Has been cancelled
2026-09-12 10:59:49 +01:00
Miguel PalhasandClaude Opus 4.6 cbbf526fc4 fix(maestro): persist auto_approve across daemon restarts
Structured sessions spawned with auto_approve lost the flag on daemon
restart because it lived only in the in-memory SessionRecord. Add it to
the Session struct (sidecar JSON) so recovery reads it back instead of
hardcoding false.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-09-12 10:59:44 +01:00
Miguel Palhas 9f5b71d563 Merge remote-tracking branch 'origin/main' into aeolian
ci / web (push) Failing after 29s
ci / rust (push) Failing after 51s
ci / flake (push) Successful in 3m9s
2026-09-12 10:42:10 +01:00
Miguel PalhasandClaude Opus 4.6 de603ccd20 feat(maestro): split tool groups on text boundaries
Tool rows interleaved with agent text now produce separate
collapsed groups instead of one per turn. The timeline reads
text → group → text → group, matching chronological order.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-09-12 10:42:06 +01:00
Miguel Palhas c6af78e9dd feat(maestro): concatenate queued structured prompts into one turn
ci / web (push) Failing after 6s
ci / rust (push) Failing after 54s
ci / flake (push) Successful in 2m59s
Closes #257
2026-09-12 10:12:36 +01:00
Miguel PalhasandClaude Opus 4.6 7ab503e4f5 fix(maestro): update e2e test for force-kill worktree removal
ci / web (push) Failing after 6s
ci / rust (push) Failing after 53s
ci / flake (push) Successful in 2m47s
1b2e5f5 changed force kill to remove dirty worktrees but left the
test asserting the old keep-on-disk behavior.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-09-11 08:38:51 +01:00
Miguel Palhas eaf11a7696 Merge branch 'maestoso'
ci / web (push) Failing after 27s
ci / rust (push) Failing after 1m1s
ci / flake (push) Successful in 3m35s
2026-09-10 23:15:45 +01:00
Miguel Palhas fffd2a55f3 Merge remote-tracking branch 'origin/main' into maestoso 2026-09-10 21:52:26 +01:00
Miguel PalhasandClaude Opus 4.6 9dead5652e fix(maestro): reconnect attach/structured sockets on Android resume
The visibility recovery only reconnected the catalog and daemon-link
sockets. Per-pane attach and structured WebSockets relied on backoff
timers that Android suspends alongside the sockets, leaving the
terminal disconnected after backgrounding.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-09-10 21:50:29 +01:00
23 changed files with 511 additions and 233 deletions
+2
View File
@@ -345,6 +345,8 @@ mod tests {
blocked_work: None, blocked_work: None,
last_focused_at: None, last_focused_at: None,
acp_session_id: None, acp_session_id: None,
auto_approve: false,
queued_prompts: Vec::new(),
} }
} }
+7 -8
View File
@@ -482,14 +482,13 @@ async fn structured_prompt(
Path(id): Path<Uuid>, Path(id): Path<Uuid>,
Json(body): Json<StructuredPromptRequest>, Json(body): Json<StructuredPromptRequest>,
) -> Response { ) -> Response {
match s.manager.queue_acp_prompt(id, body.prompt).await { match s.manager.queue_acp_prompt(id, body.prompt, None).await {
Ok(QueueAcpPrompt::Queued) | Ok(QueueAcpPrompt::Pending) => { Ok(QueueAcpPrompt::Dispatched) => StatusCode::NO_CONTENT.into_response(),
StatusCode::NO_CONTENT.into_response() Ok(QueueAcpPrompt::Queued) => StatusCode::ACCEPTED.into_response(),
} Ok(QueueAcpPrompt::Full) => error_response(
Ok(QueueAcpPrompt::InFlight) => error_response( StatusCode::TOO_MANY_REQUESTS,
StatusCode::CONFLICT, "queue_full",
"turn_in_flight", "the prompt queue has reached its capacity",
"the structured session already has an active turn",
), ),
Ok(QueueAcpPrompt::Unavailable) => error_response( Ok(QueueAcpPrompt::Unavailable) => error_response(
StatusCode::CONFLICT, StatusCode::CONFLICT,
+222 -60
View File
@@ -5,7 +5,7 @@
//! single point through which the API handlers and the per-session WS //! single point through which the API handlers and the per-session WS
//! handler interact with tmux. //! handler interact with tmux.
use std::collections::{HashMap, HashSet}; use std::collections::{HashMap, HashSet, VecDeque};
use std::path::PathBuf; use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Weak}; use std::sync::{Arc, Weak};
@@ -15,8 +15,8 @@ use chrono::Utc;
use maestro_acp::{AcpAdapter, AcpSession, PermissionRequest, SpawnConfig}; use maestro_acp::{AcpAdapter, AcpSession, PermissionRequest, SpawnConfig};
use maestro_protocol::display::command_tool; use maestro_protocol::display::command_tool;
use maestro_protocol::{ use maestro_protocol::{
ActivityState, CatalogEvent, CreateSessionRequest, ErrorInfo, NameSource, Session, ActivityState, CatalogEvent, CreateSessionRequest, ErrorInfo, NameSource, ReducedState,
SessionStatus, SessionView, StructuredReplayEvent, TranscriptDelta, TranscriptRow, Session, SessionStatus, SessionView, StructuredReplayEvent, TranscriptDelta, TranscriptRow,
TranscriptRowKind, WorktreeSpec, NAME_KEY, NAME_SOURCE_KEY, TranscriptRowKind, WorktreeSpec, NAME_KEY, NAME_SOURCE_KEY,
}; };
use tokio::sync::{broadcast, oneshot, watch, Mutex, Notify}; 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 /// When `true`, the permission pump auto-responds `AllowAlways` to every
/// ACP permission request without surfacing it to clients. /// ACP permission request without surfacing it to clients.
pub auto_approve: bool, pub auto_approve: bool,
/// A prompt submitted while the adapter was still booting (`Starting`).
/// The background handshake task drains this on success.
pub pending_prompt: Option<String>,
} }
impl SessionRecord { impl SessionRecord {
@@ -149,6 +146,9 @@ struct StructuredActivityWatermark {
sequence: u64, sequence: u64,
} }
/// Maximum number of prompts that can be queued while a turn is in flight.
const PROMPT_QUEUE_CAP: usize = 20;
#[derive(Debug)] #[derive(Debug)]
struct AcpEventStream { struct AcpEventStream {
log: EventLog, log: EventLog,
@@ -160,6 +160,10 @@ struct AcpEventStream {
/// claimed: it is queued onto ACP before another operation can proceed. /// claimed: it is queued onto ACP before another operation can proceed.
operation: Arc<Mutex<()>>, operation: Arc<Mutex<()>>,
dispatch: AcpDispatchState, 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, activity_sequence: u64,
} }
@@ -168,6 +172,23 @@ impl AcpEventStream {
self.activity_sequence += 1; self.activity_sequence += 1;
(self.activity_sequence, self.projection.activity()) (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)] #[derive(Debug, Default)]
@@ -235,11 +256,14 @@ pub enum BeginAcpTurn {
/// Result of serializing a projected turn claim with its ACP prompt enqueue. /// Result of serializing a projected turn claim with its ACP prompt enqueue.
#[derive(Debug)] #[derive(Debug)]
pub enum QueueAcpPrompt { 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, Queued,
/// The adapter is still booting; the prompt is held until it is ready.
Pending,
InFlight,
Unavailable, Unavailable,
/// The prompt queue has reached its capacity.
Full,
Failed(String), Failed(String),
} }
@@ -662,7 +686,7 @@ impl SessionManager {
let live = stream.projected_live.subscribe(); let live = stream.projected_live.subscribe();
Ok(Some(( Ok(Some((
stream.projection.snapshot(stream.log.last_seq()), stream.projection.snapshot(stream.log.last_seq()),
stream.projection.state(), stream.state_with_queue(),
live, live,
))) )))
} }
@@ -692,7 +716,7 @@ impl SessionManager {
return Ok(false); return Ok(false);
}; };
let projected = persist_turn_update(&mut stream, projected); let projected = persist_turn_update(&mut stream, projected);
let _ = stream.projected_live.send(projected); stream.broadcast(projected);
drop(stream); drop(stream);
sync_structured_activity(&self.inner, id).await; sync_structured_activity(&self.inner, id).await;
Ok(true) Ok(true)
@@ -717,7 +741,7 @@ impl SessionManager {
return Ok(BeginAcpTurn::InFlight); return Ok(BeginAcpTurn::InFlight);
} }
let (turn, update) = stream.projection.begin_turn(); let (turn, update) = stream.projection.begin_turn();
let _ = stream.projected_live.send(update); stream.broadcast(update);
drop(stream); drop(stream);
sync_structured_activity(&self.inner, id).await; sync_structured_activity(&self.inner, id).await;
Ok(BeginAcpTurn::Started(acp, turn)) Ok(BeginAcpTurn::Started(acp, turn))
@@ -733,15 +757,31 @@ impl SessionManager {
&self, &self,
id: Uuid, id: Uuid,
text: String, text: String,
row_metadata: Option<serde_json::Value>,
) -> Result<QueueAcpPrompt, SessionError> { ) -> Result<QueueAcpPrompt, SessionError> {
let (acp, stream) = { let (acp, stream) = {
let mut sessions = self.inner.sessions.lock().await; let sessions = self.inner.sessions.lock().await;
let record = sessions.get_mut(&id).ok_or(SessionError::NotFound(id))?; let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
// While the adapter is booting, stash the prompt for the
// background handshake task to drain on success.
if record.is_starting_structured() { if record.is_starting_structured() {
record.pending_prompt = Some(text); // Adapter still booting — queue the prompt for drain on boot
return Ok(QueueAcpPrompt::Pending); // 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() { if !record.is_live_structured() {
return Ok(QueueAcpPrompt::Unavailable); return Ok(QueueAcpPrompt::Unavailable);
@@ -762,19 +802,51 @@ impl SessionManager {
} }
if let Some(turn) = stream.dispatch.queued_turn { if let Some(turn) = stream.dispatch.queued_turn {
if stream.dispatch.cancelled_turn != Some(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())) Some((turn, stream.dispatch.settled.clone()))
} else { } else {
if stream.projection.turn_active() { 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 { let row = TranscriptRow {
id: format!("user:{}", Uuid::new_v4()), id: format!("user:{}", Uuid::new_v4()),
kind: TranscriptRowKind::User, kind: TranscriptRowKind::User,
data: serde_json::json!({ data,
"content": {"type": "text", "text": text},
}),
}; };
let event = PersistedEvent::User { row: row.clone() }; let event = PersistedEvent::User { row: row.clone() };
let seq = match stream.log.append(&event) { let seq = match stream.log.append(&event) {
@@ -789,9 +861,9 @@ impl SessionManager {
let logged = LoggedEvent { seq, event }; let logged = LoggedEvent { seq, event };
let projected = stream.projection.restore_row(row); let projected = stream.projection.restore_row(row);
let _ = stream.live.send(logged); let _ = stream.live.send(logged);
let _ = stream.projected_live.send(projected); stream.broadcast(projected);
let (turn, update) = stream.projection.begin_turn(); let (turn, update) = stream.projection.begin_turn();
let _ = stream.projected_live.send(update); stream.broadcast(update);
match acp.prompt(text.clone(), turn) { match acp.prompt(text.clone(), turn) {
Ok(done) => { Ok(done) => {
stream.dispatch.queued_turn = Some(turn); stream.dispatch.queued_turn = Some(turn);
@@ -809,7 +881,7 @@ impl SessionManager {
if let Ok(true) = self.claim_autoname(id).await { if let Ok(true) = self.claim_autoname(id).await {
crate::autoname::spawn_naming(self.clone(), id, text); crate::autoname::spawn_naming(self.clone(), id, text);
} }
return Ok(QueueAcpPrompt::Queued); return Ok(QueueAcpPrompt::Dispatched);
} }
Err(message) => { Err(message) => {
cancel_pending_approvals(&mut stream); cancel_pending_approvals(&mut stream);
@@ -820,7 +892,7 @@ impl SessionManager {
"newly claimed turn remains current while stream is locked", "newly claimed turn remains current while stream is locked",
); );
let projected = persist_turn_update(&mut stream, projected); let projected = persist_turn_update(&mut stream, projected);
let _ = stream.projected_live.send(projected); stream.broadcast(projected);
drop(stream); drop(stream);
sync_structured_activity(&self.inner, id).await; sync_structured_activity(&self.inner, id).await;
return Ok(QueueAcpPrompt::Failed(message)); 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) { async fn finish_queued_acp_turn(&self, id: Uuid, turn: u64, lifecycle: TurnLifecycle) {
let stream = self let stream = self
.inner .inner
@@ -860,16 +953,33 @@ impl SessionManager {
stream.dispatch.cancelled_turn = None; stream.dispatch.cancelled_turn = None;
} }
stream.dispatch.settled.notify_waiters(); stream.dispatch.settled.notify_waiters();
if stream.projection.active_turn() != Some(turn) { if stream.projection.active_turn() == Some(turn) {
return; cancel_pending_approvals(&mut stream);
} if let Some(projected) = stream.projection.turn_lifecycle(lifecycle, turn) {
cancel_pending_approvals(&mut stream); let projected = persist_turn_update(&mut stream, projected);
if let Some(projected) = stream.projection.turn_lifecycle(lifecycle, turn) { stream.broadcast(projected);
let projected = persist_turn_update(&mut stream, projected); }
let _ = stream.projected_live.send(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); drop(stream);
sync_structured_activity(&self.inner, id).await; 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( async fn fail_cancelled_acp_turn(
@@ -983,7 +1093,7 @@ impl SessionManager {
.turn_lifecycle(TurnLifecycle::Cancel, turn) .turn_lifecycle(TurnLifecycle::Cancel, turn)
.expect("active turn remains current while stream is locked"); .expect("active turn remains current while stream is locked");
let projected = persist_turn_update(&mut stream, projected); let projected = persist_turn_update(&mut stream, projected);
let _ = stream.projected_live.send(projected); stream.broadcast(projected);
drop(stream); drop(stream);
sync_structured_activity(&self.inner, id).await; sync_structured_activity(&self.inner, id).await;
if let Some(turn) = cancelled_queued_turn { if let Some(turn) = cancelled_queued_turn {
@@ -1032,7 +1142,7 @@ impl SessionManager {
let outcome = permission_outcome(decision, &pending.options); let outcome = permission_outcome(decision, &pending.options);
let _ = pending.response.send(outcome); let _ = pending.response.send(outcome);
if let Some(update) = stream.projection.resolve_approval(nonce) { if let Some(update) = stream.projection.resolve_approval(nonce) {
let _ = stream.projected_live.send(update); stream.broadcast(update);
} }
drop(stream); drop(stream);
sync_structured_activity(&self.inner, id).await; sync_structured_activity(&self.inner, id).await;
@@ -1340,6 +1450,7 @@ impl SessionManager {
pending_approvals: HashMap::new(), pending_approvals: HashMap::new(),
operation: Arc::new(Mutex::new(())), operation: Arc::new(Mutex::new(())),
dispatch: AcpDispatchState::default(), dispatch: AcpDispatchState::default(),
prompt_queue: VecDeque::new(),
activity_sequence: 0, activity_sequence: 0,
})) }))
} }
@@ -1404,6 +1515,8 @@ impl SessionManager {
last_focused_at: None, last_focused_at: None,
acp_session_id: None, acp_session_id: None,
view: req.view, view: req.view,
auto_approve: req.auto_approve,
queued_prompts: Vec::new(),
}; };
if let Err(e) = self.inner.sidecar.write(&session) { if let Err(e) = self.inner.sidecar.write(&session) {
@@ -1435,7 +1548,6 @@ impl SessionManager {
structured_activity_watermark: None, structured_activity_watermark: None,
acp_event_pump: None, acp_event_pump: None,
auto_approve: req.auto_approve, auto_approve: req.auto_approve,
pending_prompt: None,
}, },
); );
} }
@@ -1489,8 +1601,8 @@ impl SessionManager {
let acp = Arc::new(acp_session); let acp = Arc::new(acp_session);
let done = start_acp_event_pump(&inner, id, &acp, events.clone()); let done = start_acp_event_pump(&inner, id, &acp, events.clone());
start_acp_permission_pump(&inner, id, &acp, events.clone(), auto_approve); start_acp_permission_pump(&inner, id, &acp, events.clone(), auto_approve);
// Drain any prompt submitted while Starting. // Drain any prompts queued while the adapter was Starting.
let pending_prompt = { let queued_text = {
let mut sessions = inner.sessions.lock().await; let mut sessions = inner.sessions.lock().await;
let Some(record) = sessions.get_mut(&id) else { let Some(record) = sessions.get_mut(&id) else {
// Session was killed while we were booting. // Session was killed while we were booting.
@@ -1504,17 +1616,30 @@ impl SessionManager {
record.session.acp_session_id = Some(acp.session_id().to_string()); record.session.acp_session_id = Some(acp.session_id().to_string());
record.session.activity = ActivityState::Idle; record.session.activity = ActivityState::Idle;
record.session.activity_since = Utc::now(); 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.sidecar.write(&record.session);
let _ = inner.events.send(CatalogEvent::SessionUpdated { let _ = inner.events.send(CatalogEvent::SessionUpdated {
session: record.session.clone(), session: record.session.clone(),
}); });
record.pending_prompt.take() queued
}; };
if let Some(prompt) = pending_prompt { if let Some(text) = queued_text {
let manager = SessionManager { let manager = SessionManager {
inner: inner.clone(), inner: inner.clone(),
}; };
let _ = manager.queue_acp_prompt(id, prompt).await; let _ = manager.queue_acp_prompt(id, text, None).await;
} }
} }
Err(e) => { Err(e) => {
@@ -1678,6 +1803,8 @@ impl SessionManager {
// Unknown on adoption without a sidecar; matches every // Unknown on adoption without a sidecar; matches every
// other field's forward-compat default. // other field's forward-compat default.
view: maestro_protocol::SessionView::Terminal, view: maestro_protocol::SessionView::Terminal,
auto_approve: false,
queued_prompts: Vec::new(),
}, },
}; };
let fifo_path = self.inner.fifo_dir.join(format!("{id}.fifo")); let fifo_path = self.inner.fifo_dir.join(format!("{id}.fifo"));
@@ -1715,8 +1842,7 @@ impl SessionManager {
acp_events: None, acp_events: None,
structured_activity_watermark: None, structured_activity_watermark: None,
acp_event_pump: None, acp_event_pump: None,
auto_approve: false, auto_approve: session.auto_approve,
pending_prompt: None,
}, },
); );
let _ = self let _ = self
@@ -1844,6 +1970,7 @@ impl SessionManager {
pending_approvals: HashMap::new(), pending_approvals: HashMap::new(),
operation: Arc::new(Mutex::new(())), operation: Arc::new(Mutex::new(())),
dispatch: AcpDispatchState::default(), dispatch: AcpDispatchState::default(),
prompt_queue: session.queued_prompts.drain(..).collect(),
activity_sequence: 0, activity_sequence: 0,
}))) })))
} }
@@ -1891,21 +2018,45 @@ impl SessionManager {
focused_since: None, focused_since: None,
awaiting_suppressed: false, awaiting_suppressed: false,
acp, acp,
acp_events, acp_events: acp_events.clone(),
structured_activity_watermark: None, structured_activity_watermark: None,
acp_event_pump: None, acp_event_pump: None,
// Recovery cannot know the original preset's flag; the auto_approve: session.auto_approve,
// operator re-enables it by restarting with the preset.
auto_approve: false,
pending_prompt: None,
}, },
); );
if let Some((acp, events)) = event_pump { if let Some((acp, events)) = event_pump {
let done = start_acp_event_pump(&self.inner, id, &acp, events.clone()); let done = start_acp_event_pump(&self.inner, id, &acp, events.clone());
start_acp_permission_pump(&self.inner, id, &acp, events, false); start_acp_permission_pump(&self.inner, id, &acp, events, session.auto_approve);
if let Some(record) = self.inner.sessions.lock().await.get_mut(&id) { if let Some(record) = self.inner.sessions.lock().await.get_mut(&id) {
record.acp_event_pump = Some(done); 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 let _ = self
.inner .inner
@@ -2264,7 +2415,7 @@ fn start_acp_event_pump(
}; };
let projected = apply_persisted_event(&mut events.projection, &logged); let projected = apply_persisted_event(&mut events.projection, &logged);
let _ = events.live.send(logged); let _ = events.live.send(logged);
let _ = events.projected_live.send(projected); events.broadcast(projected);
None None
} }
Err(error) => Some(error), Err(error) => Some(error),
@@ -2434,7 +2585,7 @@ fn start_acp_permission_pump(
let outcome = record_permission_request(&mut events, request, response); let outcome = record_permission_request(&mut events, request, response);
let fatal = matches!(outcome, RecordOutcome::Fatal); let fatal = matches!(outcome, RecordOutcome::Fatal);
if let RecordOutcome::Live(update) = outcome { if let RecordOutcome::Live(update) = outcome {
let _ = events.projected_live.send(*update); events.broadcast(*update);
} }
drop(events); drop(events);
if fatal { if fatal {
@@ -2527,7 +2678,7 @@ fn cancel_pending_approvals(events: &mut AcpEventStream) {
.send(maestro_acp::schema::RequestPermissionOutcome::Cancelled); .send(maestro_acp::schema::RequestPermissionOutcome::Cancelled);
} }
if let Some(update) = events.projection.clear_approval() { if let Some(update) = events.projection.clear_approval() {
let _ = events.projected_live.send(update); events.broadcast(update);
} }
} }
@@ -2877,6 +3028,18 @@ async fn run_blocked_work_pass(inner: &Arc<Inner>) {
/// Terminal sessions that die naturally do not survive a daemon restart: /// Terminal sessions that die naturally do not survive a daemon restart:
/// [`SessionManager::recover`] adopts only what tmux still holds. Structured /// [`SessionManager::recover`] adopts only what tmux still holds. Structured
/// sessions stopped for daemon shutdown retain their sidecars for ACP recovery. /// 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) { async fn mark_session_dead(inner: &Arc<Inner>, id: Uuid) {
if inner.shutting_down.load(Ordering::Acquire) { if inner.shutting_down.load(Ordering::Acquire) {
return; return;
@@ -2930,7 +3093,9 @@ async fn mark_session_dead_with_worktree_settlement(
.get(&id) .get(&id)
.and_then(|record| record.acp_events.clone()); .and_then(|record| record.acp_events.clone());
if let Some(stream) = stream { 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"); tracing::info!(%id, "session process gone — marking dead");
@@ -2957,6 +3122,7 @@ async fn mark_session_dead_with_worktree_settlement(
return false; return false;
}; };
record.settled = true; record.settled = true;
record.session.queued_prompts.clear();
record.session.clone() record.session.clone()
}; };
@@ -3023,12 +3189,7 @@ async fn settle_worktree_on_exit(
/// gone from the authoritative in-memory map, so a stray FIFO or sidecar can't /// gone from the authoritative in-memory map, so a stray FIFO or sidecar can't
/// resurrect it. `kill_tmux` is `true` for an explicit kill and `false` for the /// resurrect it. `kill_tmux` is `true` for an explicit kill and `false` for the
/// liveness reaper (tmux already lost the session). /// liveness reaper (tmux already lost the session).
async fn cleanup_record( async fn cleanup_record(inner: &Arc<Inner>, id: Uuid, record: SessionRecord, kill_tmux: bool) {
inner: &Arc<Inner>,
id: Uuid,
record: SessionRecord,
kill_tmux: bool,
) {
if kill_tmux { if kill_tmux {
let tmux = inner.tmux.clone(); let tmux = inner.tmux.clone();
let _ = tmux_blocking(move || tmux.kill_session(id)).await; let _ = tmux_blocking(move || tmux.kill_session(id)).await;
@@ -3125,6 +3286,7 @@ mod tests {
pending_approvals: std::collections::HashMap::new(), pending_approvals: std::collections::HashMap::new(),
operation: std::sync::Arc::new(tokio::sync::Mutex::new(())), operation: std::sync::Arc::new(tokio::sync::Mutex::new(())),
dispatch: AcpDispatchState::default(), dispatch: AcpDispatchState::default(),
prompt_queue: std::collections::VecDeque::new(),
activity_sequence: 0, activity_sequence: 0,
} }
} }
+49 -37
View File
@@ -84,15 +84,27 @@ impl Fake {
} }
fn event_text(event: &maestro_daemon::sidecar::LoggedEvent) -> &str { fn event_text(event: &maestro_daemon::sidecar::LoggedEvent) -> &str {
let maestro_daemon::sidecar::PersistedEvent::Notification(notification) = &event.event else { match &event.event {
panic!("unexpected daemon event") maestro_daemon::sidecar::PersistedEvent::Notification(notification) => {
}; match &notification.update {
match &notification.update { SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content {
SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content { ContentBlock::Text(text) => &text.text,
ContentBlock::Text(text) => &text.text, other => panic!("unexpected content {other:?}"),
other => panic!("unexpected content {other:?}"), },
}, other => panic!("unexpected update {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!( assert!(matches!(
fx.manager fx.manager
.queue_acp_prompt(session.id, "fail".into()) .queue_acp_prompt(session.id, "fail".into(), None)
.await .await
.unwrap(), .unwrap(),
maestro_daemon::session::QueueAcpPrompt::Queued maestro_daemon::session::QueueAcpPrompt::Dispatched
)); ));
assert_eq!( assert_eq!(
fx.manager fx.manager
@@ -286,10 +298,10 @@ async fn structured_activity_tracks_turn_completion_and_failure() {
assert!(matches!( assert!(matches!(
fx.manager fx.manager
.queue_acp_prompt(session.id, "succeed".into()) .queue_acp_prompt(session.id, "succeed".into(), None)
.await .await
.unwrap(), .unwrap(),
maestro_daemon::session::QueueAcpPrompt::Queued maestro_daemon::session::QueueAcpPrompt::Dispatched
)); ));
assert_eq!( assert_eq!(
fx.manager fx.manager
@@ -426,10 +438,10 @@ async fn queued_prompt_is_a_durable_user_transcript_row() {
assert!(matches!( assert!(matches!(
fx.manager fx.manager
.queue_acp_prompt(session.id, "show my prompt".into()) .queue_acp_prompt(session.id, "show my prompt".into(), None)
.await .await
.expect("queue prompt"), .expect("queue prompt"),
maestro_daemon::session::QueueAcpPrompt::Queued maestro_daemon::session::QueueAcpPrompt::Dispatched
)); ));
let (snapshot, _, _) = fx let (snapshot, _, _) = fx
@@ -516,15 +528,15 @@ async fn cancelled_prompt_that_never_settles_fails_the_session() {
assert!(matches!( assert!(matches!(
fx.manager fx.manager
.queue_acp_prompt(session.id, "first".into()) .queue_acp_prompt(session.id, "first".into(), None)
.await .await
.unwrap(), .unwrap(),
maestro_daemon::session::QueueAcpPrompt::Queued maestro_daemon::session::QueueAcpPrompt::Dispatched
)); ));
assert!(fx.manager.cancel_acp_turn(session.id).await.unwrap()); assert!(fx.manager.cancel_acp_turn(session.id).await.unwrap());
let second = fx let second = fx
.manager .manager
.queue_acp_prompt(session.id, "second".into()) .queue_acp_prompt(session.id, "second".into(), None)
.await .await
.unwrap(); .unwrap();
assert!( assert!(
@@ -555,18 +567,18 @@ async fn cancellation_settlement_allows_the_next_prompt_permission() {
assert!(matches!( assert!(matches!(
fx.manager fx.manager
.queue_acp_prompt(session.id, "first".into()) .queue_acp_prompt(session.id, "first".into(), None)
.await .await
.unwrap(), .unwrap(),
maestro_daemon::session::QueueAcpPrompt::Queued maestro_daemon::session::QueueAcpPrompt::Dispatched
)); ));
assert!(fx.manager.cancel_acp_turn(session.id).await.unwrap()); assert!(fx.manager.cancel_acp_turn(session.id).await.unwrap());
assert!(matches!( assert!(matches!(
fx.manager fx.manager
.queue_acp_prompt(session.id, "second".into()) .queue_acp_prompt(session.id, "second".into(), None)
.await .await
.unwrap(), .unwrap(),
maestro_daemon::session::QueueAcpPrompt::Queued maestro_daemon::session::QueueAcpPrompt::Dispatched
)); ));
let nonce = tokio::time::timeout(Duration::from_secs(2), async { 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, revived.status(session.id).await,
Some(SessionStatus::Running) Some(SessionStatus::Running)
); );
assert_eq!(
revived
.snapshot()
.await
.into_iter()
.find(|candidate| candidate.id == session.id)
.unwrap()
.activity,
ActivityState::Idle
);
let log = fake.log(); let log = fake.log();
assert!( assert!(
log.contains(&format!("recv session/load sessionId={acp_id}")), log.contains(&format!("recv session/load sessionId={acp_id}")),
"{log}" "{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 { wait_for_async(Duration::from_secs(5), || async {
revived revived
.read_acp_events(session.id, EventCursor::Since(0), 10) .read_acp_events(session.id, EventCursor::Since(0), 10)
.is_ok_and(|events| events.len() == 3) .is_ok_and(|events| events.len() == 5)
}) })
.await; .await;
@@ -682,7 +687,7 @@ async fn recovery_loads_the_acp_session_without_duplicating_history() {
.await .await
.expect("replay read") .expect("replay read")
.expect("recovered event subscription"); .expect("recovered event subscription");
assert_eq!(replay.len(), 3); assert_eq!(replay.len(), 5);
assert!(live.try_recv().is_err(), "replay duplicated onto live"); assert!(live.try_recv().is_err(), "replay duplicated onto live");
let new_pid = fake.pid(); let new_pid = fake.pid();
assert_ne!(new_pid, old_pid); assert_ne!(new_pid, old_pid);
@@ -693,16 +698,23 @@ async fn recovery_loads_the_acp_session_without_duplicating_history() {
.await .await
.expect("live event after recovery") .expect("live event after recovery")
.expect("broadcast open"); .expect("broadcast open");
assert_eq!(event.seq, 4); assert_eq!(event.seq, 6);
assert_eq!(event_text(&event), "tick 1"); assert_eq!(event_text(&event), "tick 1");
let stored = revived let stored = revived
.read_acp_events(session.id, EventCursor::Since(0), 10) .read_acp_events(session.id, EventCursor::Since(0), 10)
.expect("recovered log"); .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!( assert_eq!(
stored.iter().map(event_text).collect::<Vec<_>>(), 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"); revived.kill(session.id, false).await.expect("kill");
+6 -15
View File
@@ -293,7 +293,7 @@ async fn structured_attachments_preserve_bytes_and_reject_paths() {
} }
#[tokio::test] #[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 h = spawn("naps-laptop").await;
let fake = FakeAcp::new(); let fake = FakeAcp::new();
let mut env = fake.env(); let mut env = fake.env();
@@ -310,6 +310,7 @@ async fn structured_prompt_rejects_a_second_in_flight_turn_and_cancel_clears_it(
.await .await
.unwrap(); .unwrap();
assert_eq!(first.status(), reqwest::StatusCode::NO_CONTENT); assert_eq!(first.status(), reqwest::StatusCode::NO_CONTENT);
// A second prompt while a turn is active is accepted into the queue.
let second = client let second = client
.post(url( .post(url(
h.addr, h.addr,
@@ -319,9 +320,7 @@ async fn structured_prompt_rejects_a_second_in_flight_turn_and_cancel_clears_it(
.send() .send()
.await .await
.unwrap(); .unwrap();
assert_eq!(second.status(), reqwest::StatusCode::CONFLICT); assert_eq!(second.status(), reqwest::StatusCode::ACCEPTED);
let error: ApiError = second.json().await.unwrap();
assert_eq!(error.error.code, "turn_in_flight");
let cancel = client let cancel = client
.post(url( .post(url(
@@ -332,16 +331,8 @@ async fn structured_prompt_rejects_a_second_in_flight_turn_and_cancel_clears_it(
.await .await
.unwrap(); .unwrap();
assert_eq!(cancel.status(), reqwest::StatusCode::NO_CONTENT); assert_eq!(cancel.status(), reqwest::StatusCode::NO_CONTENT);
let next = client // After cancel + settlement, the queued "second" prompt drains. Wait for
.post(url( // both the first and the drained prompt to reach ACP.
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);
tokio::time::timeout(Duration::from_secs(2), async { tokio::time::timeout(Duration::from_secs(2), async {
loop { loop {
if fake.log().matches("recv session/prompt").count() == 2 { 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 .await
.expect("both prompt commands reach ACP"); .expect("first and drained queued prompt both reach ACP");
let log = fake.log(); let log = fake.log();
let first_prompt = log.find("recv session/prompt").unwrap(); let first_prompt = log.find("recv session/prompt").unwrap();
let cancel = log.find("recv session/cancel").unwrap(); let cancel = log.find("recv session/cancel").unwrap();
+2
View File
@@ -34,6 +34,8 @@ fn sample(id: Uuid) -> Session {
last_focused_at: None, last_focused_at: None,
acp_session_id: None, acp_session_id: None,
view: SessionView::Terminal, view: SessionView::Terminal,
auto_approve: false,
queued_prompts: Vec::new(),
} }
} }
+4 -4
View File
@@ -777,7 +777,7 @@ async fn unpushed_commits_block_kill_and_force_keeps_the_tree() {
other => panic!("expected a kill_guarded refusal, got {other:?}"), other => panic!("expected a kill_guarded refusal, got {other:?}"),
} }
// Force overrides the refusal and nothing else: session gone, tree kept. // Force overrides the refusal: session gone, tree removed.
client.kill_session_forced(s.id).await.expect("forced kill"); client.kill_session_forced(s.id).await.expect("forced kill");
assert!( assert!(
@@ -789,12 +789,12 @@ async fn unpushed_commits_block_kill_and_force_keeps_the_tree() {
.any(|x| x.id == s.id), .any(|x| x.id == s.id),
"the forced kill ends the session" "the forced kill ends the session"
); );
assert!(wt.is_dir(), "a guarded worktree must stay on disk"); assert!(!wt.is_dir(), "a forced kill removes the worktree");
assert!( assert!(
git_worktree_list(base.path()) !git_worktree_list(base.path())
.iter() .iter()
.any(|l| l.contains("feature/local")), .any(|l| l.contains("feature/local")),
"the kept tree stays registered in git" "the removed tree is unregistered from git"
); );
} }
+2
View File
@@ -744,6 +744,8 @@ mod tests {
blocked_work: None, blocked_work: None,
last_focused_at: None, last_focused_at: None,
acp_session_id: None, acp_session_id: None,
auto_approve: false,
queued_prompts: Vec::new(),
} }
} }
+16
View File
@@ -132,6 +132,16 @@ pub struct Session {
/// the two are minted by different processes with different lifetimes. /// the two are minted by different processes with different lifetimes.
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub acp_session_id: Option<String>, pub acp_session_id: Option<String>,
/// When `true` the daemon auto-approves every ACP permission request for
/// this session instead of surfacing them to the client. Set at spawn time
/// from the preset or the `CreateSessionRequest`; persisted so it survives
/// 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 /// A session's process shape (#133): a plain terminal, or a structured agent
@@ -872,6 +882,10 @@ pub struct ReducedState {
pub mode: Option<String>, pub mode: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub config_options: Option<Value>, 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. /// Server messages sent over a structured-session WebSocket.
@@ -1054,6 +1068,8 @@ mod tests {
last_focused_at: None, last_focused_at: None,
view: SessionView::Terminal, view: SessionView::Terminal,
acp_session_id: None, acp_session_id: None,
auto_approve: false,
queued_prompts: Vec::new(),
} }
} }
+2
View File
@@ -50,6 +50,8 @@ pub fn session(id: Uuid, cwd: &str, command: &str) -> Session {
blocked_work: None, blocked_work: None,
last_focused_at: None, last_focused_at: None,
acp_session_id: None, acp_session_id: None,
auto_approve: false,
queued_prompts: Vec::new(),
} }
} }
+12
View File
@@ -2216,6 +2216,8 @@ mod tests {
blocked_work: None, blocked_work: None,
last_focused_at: None, last_focused_at: None,
acp_session_id: None, acp_session_id: None,
auto_approve: false,
queued_prompts: Vec::new(),
}) })
.collect(), .collect(),
}); });
@@ -2399,6 +2401,8 @@ mod tests {
blocked_work: None, blocked_work: None,
last_focused_at: None, last_focused_at: None,
acp_session_id: None, acp_session_id: None,
auto_approve: false,
queued_prompts: Vec::new(),
}], }],
}); });
app.dispatch(Action::SidebarNext); app.dispatch(Action::SidebarNext);
@@ -2428,6 +2432,8 @@ mod tests {
blocked_work: None, blocked_work: None,
last_focused_at: None, last_focused_at: None,
acp_session_id: None, acp_session_id: None,
auto_approve: false,
queued_prompts: Vec::new(),
} }
} }
@@ -3590,6 +3596,8 @@ mod tests {
blocked_work: None, blocked_work: None,
last_focused_at: focused, last_focused_at: focused,
acp_session_id: None, acp_session_id: None,
auto_approve: false,
queued_prompts: Vec::new(),
}; };
let older = Utc::now() - chrono::Duration::seconds(60); let older = Utc::now() - chrono::Duration::seconds(60);
app.catalog.apply_event(CatalogEvent::Snapshot { app.catalog.apply_event(CatalogEvent::Snapshot {
@@ -3870,6 +3878,8 @@ mod tests {
blocked_work: None, blocked_work: None,
last_focused_at: None, last_focused_at: None,
acp_session_id: None, acp_session_id: None,
auto_approve: false,
queued_prompts: Vec::new(),
}], }],
}); });
app.dispatch(Action::KillHighlighted); app.dispatch(Action::KillHighlighted);
@@ -3907,6 +3917,8 @@ mod tests {
blocked_work: None, blocked_work: None,
last_focused_at: None, last_focused_at: None,
acp_session_id: None, acp_session_id: None,
auto_approve: false,
queued_prompts: Vec::new(),
}], }],
}); });
} }
+2
View File
@@ -225,6 +225,8 @@ mod tests {
blocked_work: None, blocked_work: None,
last_focused_at: None, last_focused_at: None,
acp_session_id: None, acp_session_id: None,
auto_approve: false,
queued_prompts: Vec::new(),
} }
} }
+2
View File
@@ -811,6 +811,8 @@ mod tests {
blocked_work: None, blocked_work: None,
last_focused_at: None, last_focused_at: None,
acp_session_id: None, acp_session_id: None,
auto_approve: false,
queued_prompts: Vec::new(),
} }
} }
+4
View File
@@ -502,6 +502,8 @@ mod tests {
blocked_work: None, blocked_work: None,
last_focused_at: None, last_focused_at: None,
acp_session_id: None, acp_session_id: None,
auto_approve: false,
queued_prompts: Vec::new(),
} }
} }
@@ -1272,6 +1274,8 @@ mod tests {
blocked_work: None, blocked_work: None,
last_focused_at: None, last_focused_at: None,
acp_session_id: None, acp_session_id: None,
auto_approve: false,
queued_prompts: Vec::new(),
}; };
let sb = Sidebar { let sb = Sidebar {
sessions: vec![s], sessions: vec![s],
+28 -20
View File
@@ -189,14 +189,17 @@ choices, not daemon ones.
presentation for `Notice`. `Plan` is the exception: both clients omit it presentation for `Notice`. `Plan` is the exception: both clients omit it
from the transcript; the TUI may show the current plan in its status line, from the transcript; the TUI may show the current plan in its status line,
while the web client deliberately has no stats or plan rail. while the web client deliberately has no stats or plan rail.
- The web client groups two or more tool rows after the same `User` row into - The web client groups each contiguous run of two or more tool rows into a
one collapsed turn activity block. `AgentText` and `AgentThought` rows stay collapsed activity block. An `AgentText` or `AgentThought` row between
outside the group as standalone transcript items — the operator sees all tools splits them into separate groups, so the timeline reads
reasoning without expanding. Resolved approval rows between tools are text → group → text → group rather than lumping every tool in a turn into
absorbed into the group. The collapsed header reads "N tool calls · Label one box. Text and thought rows stay outside groups as standalone transcript
'arg'" where the label and argument come from the last tool row's items — the operator sees all reasoning without expanding. Resolved
`splitToolTitle` — the operator sees which command is current at a glance. approval rows between tools within a run are absorbed into the group. The
Opening the group reveals tool rows in chronological order. collapsed header reads "N tool calls · Label 'arg'" where the label and
argument come from the last tool row's `splitToolTitle` — the operator
sees which command is current at a glance. Opening the group reveals tool
rows in chronological order.
- The web client's user, agent, thought, tool-detail, and notice text blocks - The web client's user, agent, thought, tool-detail, and notice text blocks
copy their complete source text when clicked or keyboard-activated. copy their complete source text when clicked or keyboard-activated.
- The web client consumes `available_commands_update` notices as the live - The web client consumes `available_commands_update` notices as the live
@@ -350,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 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." 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 - Prompts are accepted at any time: if the session is idle the daemon dispatches
active. They submit each queued message only after the projected turn becomes immediately (`204`); if a turn is active the prompt is appended to a
idle; the queue is client-local and unsent messages do not survive a reload. 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 - 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 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 operator's input. If that append fails, the prompt is not dispatched.
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.
- Cancel sends `session/cancel`, clears any pending approval as `Cancelled`, - Cancel sends `session/cancel`, clears any pending approval as `Cancelled`,
and ends the turn's client-visible lifecycle immediately. If the cancelled and ends the turn's client-visible lifecycle immediately. The prompt queue
turn was still *dispatched* to ACP (waiting on its response), the daemon is preserved: after the cancelled turn settles, the queued prompts drain as
cannot simply forget it: ACP has not confirmed the cancel yet, and a new the next turn. If the session dies instead (settlement timeout), the queue
prompt claiming "the active turn" while the old one is still outstanding is cleared. If the cancelled turn was still *dispatched* to ACP (waiting on
would make the next permission callback ambiguous between the two. 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 - So a cancelled-but-still-dispatched turn must **settle within
`CANCEL_GRACE` (500ms)** — either ACP acknowledges the prompt (however it `CANCEL_GRACE` (500ms)** — either ACP acknowledges the prompt (however it
resolves) or that grace period runs out. If it settles in time, the next resolves) or that grace period runs out. If it settles in time, the next
@@ -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(); const { socket } = mount();
snapshot(socket, [], busy); snapshot(socket, [], busy);
fireEvent.change(document.querySelector<HTMLTextAreaElement>(".structured-input")!, { fireEvent.change(document.querySelector<HTMLTextAreaElement>(".structured-input")!, {
@@ -316,13 +316,19 @@ describe("prompt and cancel", () => {
}); });
fireEvent.submit(document.querySelector("form.structured-composer")!); 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")); await waitFor(() => expect(sendStructuredPrompt).toHaveBeenCalledWith("s1", "next up"));
}); });
it("renders server-side queued prompts from state", () => {
const { socket } = mount();
snapshot(socket, [], { ...busy, queued_prompts: ["first", "second"] });
const queued = document.querySelectorAll(".st-row-queued");
expect(queued).toHaveLength(2);
expect(queued[0].textContent).toBe("first");
expect(queued[1].textContent).toBe("second");
});
it("hands a failed send's draft back to the operator", async () => { it("hands a failed send's draft back to the operator", async () => {
vi.mocked(sendStructuredPrompt).mockRejectedValueOnce(new Error("boom")); vi.mocked(sendStructuredPrompt).mockRejectedValueOnce(new Error("boom"));
const { socket } = mount(); const { socket } = mount();
@@ -48,11 +48,6 @@ import "./structured.css";
* count as following. */ * count as following. */
const FOLLOW_BOTTOM_THRESHOLD_PX = 64; const FOLLOW_BOTTOM_THRESHOLD_PX = 64;
interface QueuedPrompt {
text: string;
attachments: File[];
}
export interface StructuredPaneProps { export interface StructuredPaneProps {
sessionId: Uuid; sessionId: Uuid;
/** Grabs the composer once mounted. A sidebar preview passes `false`. */ /** 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 [expandedGroups, setExpandedGroups] = useState<ReadonlySet<string>>(new Set());
const [draft, setDraft] = useState(""); const [draft, setDraft] = useState("");
const [attachments, setAttachments] = useState<File[]>([]); const [attachments, setAttachments] = useState<File[]>([]);
const [queue, setQueue] = useState<QueuedPrompt[]>([]);
const [uploading, setUploading] = useState(false); const [uploading, setUploading] = useState(false);
const [configBusy, setConfigBusy] = useState(false); const [configBusy, setConfigBusy] = useState(false);
const [error, setError] = useState<string | null>(null); const [error, setError] = useState<string | null>(null);
@@ -100,7 +94,6 @@ export const StructuredPane = forwardRef<StructuredPaneHandle, StructuredPanePro
const submission = useRef(0); const submission = useRef(0);
const accepted = useRef<{ generation: number; socketGeneration: number } | null>(null); const accepted = useRef<{ generation: number; socketGeneration: number } | null>(null);
const optimisticStart = useRef<number | null>(null); const optimisticStart = useRef<number | null>(null);
const promptPending = useRef(false);
const [awaitingBegin, setAwaitingBegin] = useState<number | null>(null); const [awaitingBegin, setAwaitingBegin] = useState<number | null>(null);
useImperativeHandle( useImperativeHandle(
@@ -166,7 +159,7 @@ export const StructuredPane = forwardRef<StructuredPaneHandle, StructuredPanePro
const el = scrollRef.current; const el = scrollRef.current;
if (!el || !followingBottom.current) return; if (!el || !followingBottom.current) return;
el.scrollTop = Math.max(0, el.scrollHeight - el.clientHeight); el.scrollTop = Math.max(0, el.scrollHeight - el.clientHeight);
}, [items, active, queue]); }, [items, active, state.queued_prompts]);
useEffect(() => { useEffect(() => {
if (!active) return; if (!active) return;
@@ -207,7 +200,7 @@ export const StructuredPane = forwardRef<StructuredPaneHandle, StructuredPanePro
setAwaitingBegin(generation); setAwaitingBegin(generation);
optimisticStart.current = Date.now(); optimisticStart.current = Date.now();
accepted.current = null; accepted.current = null;
promptPending.current = true;
setUploading(files.length > 0); setUploading(files.length > 0);
setDraft(""); setDraft("");
setError(null); setError(null);
@@ -224,7 +217,6 @@ export const StructuredPane = forwardRef<StructuredPaneHandle, StructuredPanePro
void send().then( void send().then(
() => { () => {
promptPending.current = false;
if (generation !== submission.current) return; if (generation !== submission.current) return;
setUploading(false); setUploading(false);
setAttachments((current) => (current === files ? [] : current)); setAttachments((current) => (current === files ? [] : current));
@@ -235,7 +227,6 @@ export const StructuredPane = forwardRef<StructuredPaneHandle, StructuredPanePro
if (socketGeneration > submissionSocket) session.reconnect(); if (socketGeneration > submissionSocket) session.reconnect();
}, },
(e: unknown) => { (e: unknown) => {
promptPending.current = false;
if (generation !== submission.current) return; if (generation !== submission.current) return;
setAwaitingBegin((current) => (current === generation ? null : current)); setAwaitingBegin((current) => (current === generation ? null : current));
optimisticStart.current = null; optimisticStart.current = null;
@@ -251,24 +242,10 @@ export const StructuredPane = forwardRef<StructuredPaneHandle, StructuredPanePro
[sessionId, socketGeneration, session], [sessionId, socketGeneration, session],
); );
// A queued message goes out as soon as the projected turn is idle — the // The server accepts prompts at any time: if a turn is active the prompt is
// daemon answers a second prompt `409 turn_in_flight`, so the queue is // queued server-side and drained when the turn completes. The client always
// client-local (docs/STRUCTURED-VIEW.md). // POSTs immediately — no local queue.
useEffect(() => {
if (active || promptPending.current || queue.length === 0) return;
const [next, ...rest] = queue;
setQueue(rest);
dispatchPrompt(next.text, next.attachments);
}, [active, queue, dispatchPrompt]);
const submit = (text: string, files: File[]) => { 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); dispatchPrompt(text, files);
}; };
@@ -363,10 +340,10 @@ export const StructuredPane = forwardRef<StructuredPaneHandle, StructuredPanePro
</ToolGroup> </ToolGroup>
), ),
)} )}
{queue.map((queued, index) => ( {(state.queued_prompts ?? []).map((text, index) => (
<div key={`queued-${index}`} className="st-row st-row-queued"> <div key={`queued-${index}`} className="st-row st-row-queued">
<div className="st-body"> <div className="st-body">
<div className="st-user-text">{queued.text}</div> <div className="st-user-text">{text}</div>
</div> </div>
</div> </div>
))} ))}
+44 -10
View File
@@ -115,8 +115,8 @@ describe("activity grouping", () => {
it("groups a turn's tools once there are two of them", () => { it("groups a turn's tools once there are two of them", () => {
const rows = [text("user:1", "user", "go"), tool("tool:1"), tool("tool:2")]; const rows = [text("user:1", "user", "go"), tool("tool:1"), tool("tool:2")];
const groups = activityGroupsByTurn(rows, null, false); const groups = activityGroupsByTurn(rows, null, false);
expect([...groups.keys()]).toEqual(["turn:user:1"]); expect([...groups.keys()]).toEqual(["turn:user:1:0"]);
expect(groups.get("turn:user:1")).toMatchObject({ toolCount: 2 }); expect(groups.get("turn:user:1:0")).toMatchObject({ toolCount: 2 });
}); });
it("leaves a single tool call standalone", () => { it("leaves a single tool call standalone", () => {
@@ -124,7 +124,7 @@ describe("activity grouping", () => {
expect(activityGroupsByTurn(rows, null, false).size).toBe(0); expect(activityGroupsByTurn(rows, null, false).size).toBe(0);
}); });
it("keeps intermediate text outside the group as standalone items", () => { it("splits tools into separate runs when text appears between them", () => {
const rows = [ const rows = [
text("user:1", "user", "go"), text("user:1", "user", "go"),
tool("tool:1"), tool("tool:1"),
@@ -132,9 +132,23 @@ describe("activity grouping", () => {
tool("tool:2"), tool("tool:2"),
text("agent:2", "agent_text", "done"), text("agent:2", "agent_text", "done"),
]; ];
const group = activityGroupsByTurn(rows, null, false).get("turn:user:1")!; // Each run has only one tool, so neither qualifies for grouping.
expect(group.rows.map((r) => r.id)).toEqual(["tool:1", "tool:2"]); expect(activityGroupsByTurn(rows, null, false).size).toBe(0);
expect(group.toolCount).toBe(2); });
it("groups contiguous tools separated by text into distinct groups", () => {
const rows = [
text("user:1", "user", "go"),
tool("tool:1"),
tool("tool:2"),
text("agent:1", "agent_text", "working"),
tool("tool:3"),
tool("tool:4"),
];
const groups = activityGroupsByTurn(rows, null, false);
expect([...groups.keys()]).toEqual(["turn:user:1:0", "turn:user:1:1"]);
expect(groups.get("turn:user:1:0")!.rows.map((r) => r.id)).toEqual(["tool:1", "tool:2"]);
expect(groups.get("turn:user:1:1")!.rows.map((r) => r.id)).toEqual(["tool:3", "tool:4"]);
}); });
it("keeps thoughts outside the group as standalone items", () => { it("keeps thoughts outside the group as standalone items", () => {
@@ -145,7 +159,7 @@ describe("activity grouping", () => {
text("thought:1", "agent_thought", "hmm"), text("thought:1", "agent_thought", "hmm"),
text("thought:2", "agent_thought", "aha"), text("thought:2", "agent_thought", "aha"),
]; ];
const group = activityGroupsByTurn(rows, null, false).get("turn:user:1")!; const group = activityGroupsByTurn(rows, null, false).get("turn:user:1:0")!;
expect(group.rows.map((r) => r.id)).toEqual(["tool:1", "tool:2"]); expect(group.rows.map((r) => r.id)).toEqual(["tool:1", "tool:2"]);
}); });
@@ -155,7 +169,7 @@ describe("activity grouping", () => {
row("tool:1", "tool_call", { title: "Read(src/main.ts)", status: "completed" }), row("tool:1", "tool_call", { title: "Read(src/main.ts)", status: "completed" }),
row("tool:2", "tool_call", { title: "Execute(git status)", kind: "execute", status: "completed" }), row("tool:2", "tool_call", { title: "Execute(git status)", kind: "execute", status: "completed" }),
]; ];
const group = activityGroupsByTurn(rows, null, false).get("turn:user:1")!; const group = activityGroupsByTurn(rows, null, false).get("turn:user:1:0")!;
expect(group.lastToolLabel).toBe("Execute"); expect(group.lastToolLabel).toBe("Execute");
expect(group.lastToolArg).toBe("git status"); expect(group.lastToolArg).toBe("git status");
}); });
@@ -169,7 +183,7 @@ describe("activity grouping", () => {
row("approval:2", "approval"), row("approval:2", "approval"),
tool("tool:3"), tool("tool:3"),
]; ];
const group = activityGroupsByTurn(rows, "approval:2", false).get("turn:user:1")!; const group = activityGroupsByTurn(rows, "approval:2", false).get("turn:user:1:0")!;
expect(group.rows.map((r) => r.id)).toEqual(["tool:1", "approval:1", "tool:2", "tool:3"]); expect(group.rows.map((r) => r.id)).toEqual(["tool:1", "approval:1", "tool:2", "tool:3"]);
}); });
}); });
@@ -180,11 +194,31 @@ describe("transcript render list", () => {
const items = transcriptItems(rows, null, false); const items = transcriptItems(rows, null, false);
expect(items.map((item) => (item.type === "row" ? item.row.id : item.id))).toEqual([ expect(items.map((item) => (item.type === "row" ? item.row.id : item.id))).toEqual([
"user:1", "user:1",
"turn:user:1", "turn:user:1:0",
"agent:1", "agent:1",
]); ]);
}); });
it("interleaves groups and text in timeline order", () => {
const rows = [
text("user:1", "user", "go"),
tool("tool:1"),
tool("tool:2"),
text("agent:1", "agent_text", "working"),
tool("tool:3"),
tool("tool:4"),
text("agent:2", "agent_text", "done"),
];
const items = transcriptItems(rows, null, false);
expect(items.map((item) => (item.type === "row" ? item.row.id : item.id))).toEqual([
"user:1",
"turn:user:1:0",
"agent:1",
"turn:user:1:1",
"agent:2",
]);
});
it("omits plan rows, state-only notices and standalone resolved approvals", () => { it("omits plan rows, state-only notices and standalone resolved approvals", () => {
const rows = [ const rows = [
text("user:1", "user", "go"), text("user:1", "user", "go"),
+52 -43
View File
@@ -155,57 +155,68 @@ export interface ActivityGroup {
} }
/** /**
* Group tool rows (and resolved approvals between them) by turn. Text and * Group contiguous tool rows (and resolved approvals between them) into
* thought rows stay outside the group as standalone transcript items — the * collapsible blocks. An `AgentText` or `AgentThought` row between tools
* operator sees all reasoning without expanding anything. * splits the run — each contiguous stretch of tools becomes its own group,
* so the timeline reads text → group → text → group rather than lumping
* every tool in a turn into one box.
* *
* Returns groups keyed by `turn:<userId>`. * A group must contain at least two tools to qualify; a single tool stays
* standalone. Returns groups keyed by `turn:<userId>:<seq>`.
*/ */
export function activityGroupsByTurn( export function activityGroupsByTurn(
rows: TranscriptRow[], rows: TranscriptRow[],
pendingApprovalId: string | null, pendingApprovalId: string | null,
_turnActive: boolean, _turnActive: boolean,
): Map<string, ActivityGroup> { ): Map<string, ActivityGroup> {
// First: count tools per turn to know which turns qualify for grouping.
const toolCounts = new Map<string, number>();
let turnId = "before-first-message";
for (const row of rows) {
if (row.kind === "user") turnId = row.id;
if (row.kind !== "tool_call" && row.kind !== "tool_diff") continue;
const gid = `turn:${turnId}`;
toolCounts.set(gid, (toolCounts.get(gid) ?? 0) + 1);
}
// Second: build groups for turns with 2+ tools.
const groups = new Map<string, ActivityGroup>(); const groups = new Map<string, ActivityGroup>();
turnId = "before-first-message"; let turnId = "before-first-message";
let pending: TranscriptRow[] = []; let runSeq = 0;
let currentGid = `turn:${turnId}`; let runTools: TranscriptRow[] = [];
let runPending: TranscriptRow[] = [];
const flushRun = () => {
if (runTools.length >= 2) {
const gid = `turn:${turnId}:${runSeq}`;
const tool = (runTools[runTools.length - 1].data ?? {}) as { title?: string; name?: string | null; kind?: string };
const { label, arg } = splitToolTitle(tool.title, tool.name ?? undefined, tool.kind);
groups.set(gid, {
toolCount: runTools.length,
rows: [...runTools],
lastToolLabel: label,
lastToolArg: arg,
});
runSeq++;
}
runTools = [];
runPending = [];
};
for (const row of rows) { for (const row of rows) {
if (row.kind === "user") { if (row.kind === "user") {
flushRun();
turnId = row.id; turnId = row.id;
currentGid = `turn:${turnId}`; runSeq = 0;
pending = []; continue;
}
// Text and thought rows break the current tool run.
if (row.kind === "agent_text" || row.kind === "agent_thought") {
flushRun();
continue; continue;
} }
const count = toolCounts.get(currentGid) ?? 0;
if (count < 2) continue;
const isResolvedApproval = row.kind === "approval" && row.id !== pendingApprovalId; const isResolvedApproval = row.kind === "approval" && row.id !== pendingApprovalId;
if (isResolvedApproval) { if (isResolvedApproval) {
pending.push(row); runPending.push(row);
continue; continue;
} }
if (row.kind === "tool_call" || row.kind === "tool_diff") { if (row.kind === "tool_call" || row.kind === "tool_diff") {
const group = groups.get(currentGid) ?? { toolCount: count, rows: [], lastToolLabel: "", lastToolArg: "" }; // Flush any pending resolved approvals into the run before this tool.
group.rows.push(...pending); runTools.push(...runPending);
pending = []; runPending = [];
group.rows.push(row); runTools.push(row);
const tool = (row.data ?? {}) as { title?: string; name?: string | null; kind?: string };
const { label, arg } = splitToolTitle(tool.title, tool.name ?? undefined, tool.kind);
group.lastToolLabel = label;
group.lastToolArg = arg;
groups.set(currentGid, group);
} }
} }
flushRun();
return groups; return groups;
} }
@@ -217,10 +228,10 @@ export type TranscriptItem =
| { type: "group"; id: string; group: ActivityGroup }; | { type: "group"; id: string; group: ActivityGroup };
/** /**
* The transcript in render order: rows, with each turn's tool activity folded * The transcript in render order: rows, with each contiguous tool stretch
* into a single group item where one exists. Resolved approvals not absorbed * folded into a single group item where one exists. Resolved approvals not
* into a group are dropped — the tool row itself already identifies the * absorbed into a group are dropped — the tool row itself already identifies
* action. `plan` rows and state-only notices are dropped too (both clients * the action. `plan` rows and state-only notices are dropped too (both clients
* omit `Plan` from the transcript). A turn footer only survives as the final * omit `Plan` from the transcript). A turn footer only survives as the final
* item, so old footers stop reading like transcript content. * item, so old footers stop reading like transcript content.
*/ */
@@ -230,21 +241,19 @@ export function transcriptItems(
turnActive: boolean, turnActive: boolean,
): TranscriptItem[] { ): TranscriptItem[] {
const groups = activityGroupsByTurn(rows, pendingApprovalId, turnActive); const groups = activityGroupsByTurn(rows, pendingApprovalId, turnActive);
const grouped = new Set<string>(); // Reverse index: row id → group id that contains it.
for (const group of groups.values()) for (const row of group.rows) grouped.add(row.id); const rowToGroup = new Map<string, string>();
for (const [groupId, group] of groups) for (const r of group.rows) rowToGroup.set(r.id, groupId);
const items: TranscriptItem[] = []; const items: TranscriptItem[] = [];
let turnId = "before-first-message";
const emitted = new Set<string>(); const emitted = new Set<string>();
for (const row of rows) { for (const row of rows) {
if (row.kind === "user") turnId = row.id; const groupId = rowToGroup.get(row.id);
const groupId = `turn:${turnId}`; if (groupId !== undefined) {
const group = groups.get(groupId);
if (group && grouped.has(row.id)) {
// The group renders once, at its first member's position. // The group renders once, at its first member's position.
if (!emitted.has(groupId)) { if (!emitted.has(groupId)) {
emitted.add(groupId); emitted.add(groupId);
items.push({ type: "group", id: groupId, group }); items.push({ type: "group", id: groupId, group: groups.get(groupId)! });
} }
continue; continue;
} }
+5 -1
View File
@@ -726,7 +726,11 @@
} }
.st-tool-group-title { .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; font-weight: 500;
} }
@@ -196,12 +196,24 @@ export function useStructuredSession(
connect(); connect();
}; };
// Mobile browsers suspend timers and silently kill WebSockets in the
// background. Kick a fresh connect when the page becomes visible again.
const onVisibility = () => {
if (document.visibilityState !== "visible" || disposed) return;
if (socket && (socket.readyState === WebSocket.OPEN || socket.readyState === WebSocket.CONNECTING)) return;
cancelTimer();
setStatus("reconnecting");
connect();
};
document.addEventListener("visibilitychange", onVisibility);
setStatus("connecting"); setStatus("connecting");
connect(); connect();
return () => { return () => {
disposed = true; disposed = true;
reconnectRef.current = () => {}; reconnectRef.current = () => {};
document.removeEventListener("visibilitychange", onVisibility);
cancelTimer(); cancelTimer();
closeSocket(); closeSocket();
}; };
@@ -302,6 +302,24 @@ export const TerminalPane = forwardRef<TerminalPaneHandle, TerminalPaneProps>(
ta?.addEventListener("focus", onTaFocus); ta?.addEventListener("focus", onTaFocus);
ta?.addEventListener("blur", onTaBlur); 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.
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;
}
reconnectMs = INITIAL_RECONNECT_MS;
onStatusChange?.("reconnecting");
connect();
};
document.addEventListener("visibilitychange", onVisibility);
window.addEventListener("resize", onResize); window.addEventListener("resize", onResize);
const visual = window.visualViewport ?? null; const visual = window.visualViewport ?? null;
visual?.addEventListener("resize", onResize); visual?.addEventListener("resize", onResize);
@@ -326,6 +344,7 @@ export const TerminalPane = forwardRef<TerminalPaneHandle, TerminalPaneProps>(
reconnectTimer = null; reconnectTimer = null;
} }
ro.disconnect(); ro.disconnect();
document.removeEventListener("visibilitychange", onVisibility);
window.removeEventListener("resize", onResize); window.removeEventListener("resize", onResize);
visual?.removeEventListener("resize", onResize); visual?.removeEventListener("resize", onResize);
ta?.removeEventListener("focus", onTaFocus); ta?.removeEventListener("focus", onTaFocus);
+1
View File
@@ -441,6 +441,7 @@ export interface ReducedState {
model?: string | null; model?: string | null;
mode?: string | null; mode?: string | null;
config_options?: { configOptions: AcpSessionConfigOption[] } | null; config_options?: { configOptions: AcpSessionConfigOption[] } | null;
queued_prompts?: string[];
} }
/** Server messages on `WS /api/structured/:id`. */ /** Server messages on `WS /api/structured/:id`. */