Compare commits

..
19 Commits
Author SHA1 Message Date
Miguel PalhasandClaude Opus 4.6 1ea463f6cf fix(maestro): collapse mobile drawer after spawning a session
ci / web (push) Failing after 28s
ci / flake (push) Successful in 4m6s
ci / rust (push) Successful in 7m37s
mountSpawned now fires a CLOSE_DRAWER_EVENT so the sidebar gets out of
the way once the new session lands. Covers fast-forward, modal, and
command-bar spawn paths.

Closes #261

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-09-12 23:31:52 +01:00
Miguel PalhasandClaude Opus 4.6 7c27a897ac feat(maestro): optimistic session kill in sidebar
ci / web (push) Failing after 6s
ci / flake (push) Successful in 4m13s
ci / rust (push) Successful in 7m25s
Close the kill dialog and remove the session card immediately on
confirm instead of waiting for the DELETE round-trip. Restores the
session if the daemon refuses with kill_guarded and re-opens the
dialog in force mode.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-09-12 21:21:36 +01:00
Miguel Palhas a9717d01c8 Merge remote-tracking branch 'origin/main' into animato-2
ci / web (push) Failing after 27s
ci / flake (push) Successful in 4m15s
ci / rust (push) Successful in 7m30s
2026-09-12 20:50:52 +01:00
Miguel PalhasandClaude Opus 4.6 35a6ff9d0f fix(maestro): force-replace pane sockets on mobile resume
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>
2026-09-12 20:50:48 +01:00
Miguel PalhasandClaude Opus 4.6 474000935a fix(maestro): box AcpUpdate::Notification and fix test unwrapping
ci / web (push) Failing after 33s
ci / flake (push) Successful in 4m42s
ci / rust (pull_request) Failing after 8s
ci / web (pull_request) Failing after 32s
ci / flake (pull_request) Successful in 1m13s
ci / rust (push) Successful in 8m25s
Box the Notification variant to satisfy clippy's large_enum_variant
lint, fix paren mismatches from the boxing pass, and update test
helpers to unwrap AcpUpdate before inspecting the notification.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-09-12 19:59:03 +01:00
Miguel PalhasandClaude Opus 4.6 5f8cc2a140 fix(maestro): drain notifications before signaling turn-done
The event pump now handles AcpUpdate::TurnDone markers that arrive
in the same FIFO as notifications.  It fires the done oneshot only
after persisting every preceding notification, so finish_queued_acp_turn
can no longer race ahead of the transcript.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-09-12 19:54:00 +01:00
Miguel PalhasandClaude Opus 4.6 432967fe64 feat(maestro): route turn-done through ACP updates channel
The prompt response callback now sends an AcpUpdate::TurnDone marker
through the same unbounded channel as notifications instead of
signaling the done oneshot directly.  Since notifications and the
prompt response both run in the ACP dispatch loop in wire order,
TurnDone lands after every notification chunk for the turn.

The event pump (next commit) will fire the done oneshot only after
persisting all preceding notifications, closing the race where
finish_queued_acp_turn ran before the transcript was complete.

Closes #259

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-09-12 19:53:18 +01:00
Miguel PalhasandClaude Opus 4.6 2512b8be42 fix(maestro): apply tool-arg truncation fix to styles.css duplicate
ci / web (push) Failing after 6s
ci / flake (push) Successful in 4m24s
ci / rust (push) Successful in 7m49s
The previous fix (e3556bd) added `flex: 1 1 0` to `.st-tool-arg` in
`structured.css`, but `styles.css` has a duplicate copy of the same
rules that was left stale. Because both load, the unfixed copy in
`styles.css` can override the fix and re-enable horizontal scroll.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-09-12 19:42:44 +01:00
Miguel PalhasandClaude Opus 4.6 e3556bd847 fix(maestro): truncate long commands in tool call header
ci / web (push) Failing after 30s
ci / flake (push) Successful in 4m5s
ci / rust (push) Successful in 7m30s
Same root cause as the tool-group-title fix — .st-tool-arg
defaulted to flex: 0 1 auto, so long commands pushed the row
past the container. Adding flex: 1 1 0 lets the ellipsis
truncation work.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-09-12 19:03:06 +01:00
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
30 changed files with 645 additions and 287 deletions
+41 -16
View File
@@ -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()));
+11 -2
View File
@@ -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(),
+2
View File
@@ -345,6 +345,8 @@ mod tests {
blocked_work: None,
last_focused_at: 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>,
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,
+261 -84
View File
@@ -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,
}))
}
@@ -1404,6 +1515,8 @@ impl SessionManager {
last_focused_at: None,
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) {
@@ -1435,7 +1548,6 @@ impl SessionManager {
structured_activity_watermark: None,
acp_event_pump: None,
auto_approve: req.auto_approve,
pending_prompt: None,
},
);
}
@@ -1489,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.
@@ -1504,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) => {
@@ -1678,6 +1803,8 @@ impl SessionManager {
// Unknown on adoption without a sidecar; matches every
// 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"));
@@ -1715,8 +1842,7 @@ impl SessionManager {
acp_events: None,
structured_activity_watermark: None,
acp_event_pump: None,
auto_approve: false,
pending_prompt: None,
auto_approve: session.auto_approve,
},
);
let _ = self
@@ -1844,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,
})))
}
@@ -1891,21 +2018,45 @@ impl SessionManager {
focused_since: None,
awaiting_suppressed: false,
acp,
acp_events,
acp_events: acp_events.clone(),
structured_activity_watermark: None,
acp_event_pump: None,
// Recovery cannot know the original preset's flag; the
// operator re-enables it by restarting with the preset.
auto_approve: false,
pending_prompt: None,
auto_approve: session.auto_approve,
},
);
if let Some((acp, events)) = event_pump {
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) {
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()
};
@@ -3023,12 +3204,7 @@ async fn settle_worktree_on_exit(
/// 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
/// liveness reaper (tmux already lost the session).
async fn cleanup_record(
inner: &Arc<Inner>,
id: Uuid,
record: SessionRecord,
kill_tmux: bool,
) {
async fn cleanup_record(inner: &Arc<Inner>, id: Uuid, record: SessionRecord, kill_tmux: bool) {
if kill_tmux {
let tmux = inner.tmux.clone();
let _ = tmux_blocking(move || tmux.kill_session(id)).await;
@@ -3125,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,
}
}
+49 -37
View File
@@ -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");
+6 -15
View File
@@ -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();
+2
View File
@@ -34,6 +34,8 @@ fn sample(id: Uuid) -> Session {
last_focused_at: None,
acp_session_id: None,
view: SessionView::Terminal,
auto_approve: false,
queued_prompts: Vec::new(),
}
}
+2
View File
@@ -744,6 +744,8 @@ mod tests {
blocked_work: None,
last_focused_at: 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.
#[serde(default, skip_serializing_if = "Option::is_none")]
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
@@ -872,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.
@@ -1054,6 +1068,8 @@ mod tests {
last_focused_at: None,
view: SessionView::Terminal,
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,
last_focused_at: None,
acp_session_id: None,
auto_approve: false,
queued_prompts: Vec::new(),
}
}
+12
View File
@@ -2216,6 +2216,8 @@ mod tests {
blocked_work: None,
last_focused_at: None,
acp_session_id: None,
auto_approve: false,
queued_prompts: Vec::new(),
})
.collect(),
});
@@ -2399,6 +2401,8 @@ mod tests {
blocked_work: None,
last_focused_at: None,
acp_session_id: None,
auto_approve: false,
queued_prompts: Vec::new(),
}],
});
app.dispatch(Action::SidebarNext);
@@ -2428,6 +2432,8 @@ mod tests {
blocked_work: None,
last_focused_at: None,
acp_session_id: None,
auto_approve: false,
queued_prompts: Vec::new(),
}
}
@@ -3590,6 +3596,8 @@ mod tests {
blocked_work: None,
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 {
@@ -3870,6 +3878,8 @@ mod tests {
blocked_work: None,
last_focused_at: None,
acp_session_id: None,
auto_approve: false,
queued_prompts: Vec::new(),
}],
});
app.dispatch(Action::KillHighlighted);
@@ -3907,6 +3917,8 @@ mod tests {
blocked_work: None,
last_focused_at: None,
acp_session_id: None,
auto_approve: false,
queued_prompts: Vec::new(),
}],
});
}
+2
View File
@@ -225,6 +225,8 @@ mod tests {
blocked_work: None,
last_focused_at: None,
acp_session_id: None,
auto_approve: false,
queued_prompts: Vec::new(),
}
}
+2
View File
@@ -811,6 +811,8 @@ mod tests {
blocked_work: None,
last_focused_at: None,
acp_session_id: None,
auto_approve: false,
queued_prompts: Vec::new(),
}
}
+4
View File
@@ -502,6 +502,8 @@ mod tests {
blocked_work: None,
last_focused_at: None,
acp_session_id: None,
auto_approve: false,
queued_prompts: Vec::new(),
}
}
@@ -1272,6 +1274,8 @@ mod tests {
blocked_work: None,
last_focused_at: None,
acp_session_id: None,
auto_approve: false,
queued_prompts: Vec::new(),
};
let sb = Sidebar {
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
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.
- The web client groups two or more tool rows after the same `User` row into
one collapsed turn activity block. `AgentText` and `AgentThought` rows stay
outside the group as standalone transcript items — the operator sees all
reasoning without expanding. Resolved approval rows between tools are
absorbed into the group. The 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 groups each contiguous run of two or more tool rows into a
collapsed activity block. An `AgentText` or `AgentThought` row between
tools splits them into separate groups, so the timeline reads
text → group → text → group rather than lumping every tool in a turn into
one box. Text and thought rows stay outside groups as standalone transcript
items — the operator sees all reasoning without expanding. Resolved
approval rows between tools within a run are absorbed into the group. The
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
copy their complete source text when clicked or keyboard-activated.
- 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
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
+13 -3
View File
@@ -47,6 +47,7 @@ import {
OPEN_SPAWN_EVENT,
SESSION_SELECT_EVENT,
SPAWN_SIMILAR_EVENT,
requestCloseDrawer,
requestOpenSpawn,
requestToggleDrawer,
} from "./uiEvents.js";
@@ -113,15 +114,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);
};
@@ -174,6 +183,7 @@ function Shell() {
(session: Session, opts?: { newTab?: boolean }) => {
layoutStore.getState().openSessionBeside(session.id, opts);
navigate({ kind: "attach", sessionId: session.id });
requestCloseDrawer();
},
[layoutStore, navigate],
);
+14 -1
View File
@@ -2,7 +2,7 @@
import { act, cleanup, fireEvent, render } from "@testing-library/react";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { OPEN_SPAWN_EVENT, SPAWN_SIMILAR_EVENT } from "../../uiEvents.js";
import { CLOSE_DRAWER_EVENT, OPEN_SPAWN_EVENT, SPAWN_SIMILAR_EVENT } from "../../uiEvents.js";
import { TOGGLE_DRAWER_EVENT } from "../../uiEvents.js";
import { Drawer } from "./Drawer.js";
@@ -103,6 +103,19 @@ describe("Drawer (#35)", () => {
expect(shell().classList.contains("drawer-open")).toBe(false);
});
it("closes on the close-drawer event (after spawn lands)", () => {
render(
<Drawer sidebar={<div className="tree-leaf">row</div>}>
<div>content</div>
</Drawer>,
);
openViaToggle();
act(() => {
document.dispatchEvent(new CustomEvent(CLOSE_DRAWER_EVENT));
});
expect(shell().classList.contains("drawer-open")).toBe(false);
});
it("closes on a backdrop click", () => {
render(
<Drawer sidebar={<div className="tree-leaf">row</div>}>
+3 -1
View File
@@ -10,7 +10,7 @@
import type { ReactNode } from "react";
import { useEffect, useRef, useState } from "react";
import { OPEN_SPAWN_EVENT, SPAWN_SIMILAR_EVENT, TOGGLE_DRAWER_EVENT } from "../../uiEvents.js";
import { CLOSE_DRAWER_EVENT, OPEN_SPAWN_EVENT, SPAWN_SIMILAR_EVENT, TOGGLE_DRAWER_EVENT } from "../../uiEvents.js";
export interface DrawerProps {
sidebar: ReactNode;
@@ -28,10 +28,12 @@ export function Drawer({ sidebar, children, className }: DrawerProps) {
document.addEventListener(TOGGLE_DRAWER_EVENT, onToggle);
document.addEventListener(OPEN_SPAWN_EVENT, onClose);
document.addEventListener(SPAWN_SIMILAR_EVENT, onClose);
document.addEventListener(CLOSE_DRAWER_EVENT, onClose);
return () => {
document.removeEventListener(TOGGLE_DRAWER_EVENT, onToggle);
document.removeEventListener(OPEN_SPAWN_EVENT, onClose);
document.removeEventListener(SPAWN_SIMILAR_EVENT, onClose);
document.removeEventListener(CLOSE_DRAWER_EVENT, onClose);
};
}, []);
+18 -2
View File
@@ -333,10 +333,25 @@ export function Sidebar(props: SidebarProps) {
onCancel={() => setKill(null)}
onConfirm={async () => {
const { session: s, force } = kill;
// Close the dialog immediately and optimistically remove the
// session from the sidebar so the operator isn't staring at a
// spinner while the daemon tears down the tmux session.
setKill(null);
const store = props.store;
if (store) {
const next = new Map(store.getState().sessions);
next.delete(s.id);
store.setState({ sessions: next });
}
try {
await killSession(s.id, force);
setKill(null);
} catch (e: unknown) {
// Restore the session so it reappears in the sidebar.
if (store) {
const restored = new Map(store.getState().sessions);
restored.set(s.id, s);
store.setState({ sessions: restored });
}
// The daemon refuses an unforced kill over a worktree holding
// unsaved work; that refusal needs its own prompt
// (docs/WORKTREES.md).
@@ -349,7 +364,8 @@ export function Sidebar(props: SidebarProps) {
});
return;
}
throw e;
// Nothing else we can usefully show — the session is back in
// the sidebar and the operator can retry.
}
}}
/>
@@ -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,13 +316,19 @@ 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("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 () => {
vi.mocked(sendStructuredPrompt).mockRejectedValueOnce(new Error("boom"));
const { socket } = mount();
@@ -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,24 +242,10 @@ export const StructuredPane = forwardRef<StructuredPaneHandle, StructuredPanePro
[sessionId, socketGeneration, session],
);
// A queued message goes out as soon as the projected turn is idle — the
// daemon answers a second prompt `409 turn_in_flight`, so the queue is
// client-local (docs/STRUCTURED-VIEW.md).
useEffect(() => {
if (active || promptPending.current || queue.length === 0) return;
const [next, ...rest] = queue;
setQueue(rest);
dispatchPrompt(next.text, next.attachments);
}, [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);
};
@@ -363,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>
))}
+44 -10
View File
@@ -115,8 +115,8 @@ describe("activity grouping", () => {
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 groups = activityGroupsByTurn(rows, null, false);
expect([...groups.keys()]).toEqual(["turn:user:1"]);
expect(groups.get("turn:user:1")).toMatchObject({ toolCount: 2 });
expect([...groups.keys()]).toEqual(["turn:user:1:0"]);
expect(groups.get("turn:user:1:0")).toMatchObject({ toolCount: 2 });
});
it("leaves a single tool call standalone", () => {
@@ -124,7 +124,7 @@ describe("activity grouping", () => {
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 = [
text("user:1", "user", "go"),
tool("tool:1"),
@@ -132,9 +132,23 @@ describe("activity grouping", () => {
tool("tool:2"),
text("agent:2", "agent_text", "done"),
];
const group = activityGroupsByTurn(rows, null, false).get("turn:user:1")!;
expect(group.rows.map((r) => r.id)).toEqual(["tool:1", "tool:2"]);
expect(group.toolCount).toBe(2);
// Each run has only one tool, so neither qualifies for grouping.
expect(activityGroupsByTurn(rows, null, false).size).toBe(0);
});
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", () => {
@@ -145,7 +159,7 @@ describe("activity grouping", () => {
text("thought:1", "agent_thought", "hmm"),
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"]);
});
@@ -155,7 +169,7 @@ describe("activity grouping", () => {
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" }),
];
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.lastToolArg).toBe("git status");
});
@@ -169,7 +183,7 @@ describe("activity grouping", () => {
row("approval:2", "approval"),
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"]);
});
});
@@ -180,11 +194,31 @@ describe("transcript render list", () => {
const items = transcriptItems(rows, null, false);
expect(items.map((item) => (item.type === "row" ? item.row.id : item.id))).toEqual([
"user:1",
"turn:user:1",
"turn:user:1:0",
"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", () => {
const rows = [
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
* thought rows stay outside the group as standalone transcript items — the
* operator sees all reasoning without expanding anything.
* Group contiguous tool rows (and resolved approvals between them) into
* collapsible blocks. An `AgentText` or `AgentThought` row between tools
* 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(
rows: TranscriptRow[],
pendingApprovalId: string | null,
_turnActive: boolean,
): 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>();
turnId = "before-first-message";
let pending: TranscriptRow[] = [];
let currentGid = `turn:${turnId}`;
let turnId = "before-first-message";
let runSeq = 0;
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) {
if (row.kind === "user") {
flushRun();
turnId = row.id;
currentGid = `turn:${turnId}`;
pending = [];
runSeq = 0;
continue;
}
// Text and thought rows break the current tool run.
if (row.kind === "agent_text" || row.kind === "agent_thought") {
flushRun();
continue;
}
const count = toolCounts.get(currentGid) ?? 0;
if (count < 2) continue;
const isResolvedApproval = row.kind === "approval" && row.id !== pendingApprovalId;
if (isResolvedApproval) {
pending.push(row);
runPending.push(row);
continue;
}
if (row.kind === "tool_call" || row.kind === "tool_diff") {
const group = groups.get(currentGid) ?? { toolCount: count, rows: [], lastToolLabel: "", lastToolArg: "" };
group.rows.push(...pending);
pending = [];
group.rows.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);
// Flush any pending resolved approvals into the run before this tool.
runTools.push(...runPending);
runPending = [];
runTools.push(row);
}
}
flushRun();
return groups;
}
@@ -217,10 +228,10 @@ export type TranscriptItem =
| { type: "group"; id: string; group: ActivityGroup };
/**
* The transcript in render order: rows, with each turn's tool activity folded
* into a single group item where one exists. Resolved approvals not absorbed
* into a group are dropped — the tool row itself already identifies the
* action. `plan` rows and state-only notices are dropped too (both clients
* The transcript in render order: rows, with each contiguous tool stretch
* folded into a single group item where one exists. Resolved approvals not
* absorbed into a group are dropped — the tool row itself already identifies
* 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
* item, so old footers stop reading like transcript content.
*/
@@ -230,21 +241,19 @@ export function transcriptItems(
turnActive: boolean,
): TranscriptItem[] {
const groups = activityGroupsByTurn(rows, pendingApprovalId, turnActive);
const grouped = new Set<string>();
for (const group of groups.values()) for (const row of group.rows) grouped.add(row.id);
// Reverse index: row id → group id that contains it.
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[] = [];
let turnId = "before-first-message";
const emitted = new Set<string>();
for (const row of rows) {
if (row.kind === "user") turnId = row.id;
const groupId = `turn:${turnId}`;
const group = groups.get(groupId);
if (group && grouped.has(row.id)) {
const groupId = rowToGroup.get(row.id);
if (groupId !== undefined) {
// The group renders once, at its first member's position.
if (!emitted.has(groupId)) {
emitted.add(groupId);
items.push({ type: "group", id: groupId, group });
items.push({ type: "group", id: groupId, group: groups.get(groupId)! });
}
continue;
}
+6 -1
View File
@@ -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;
}
@@ -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();
};
+10 -5
View File
@@ -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();
+6 -1
View File
@@ -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;
}
+1
View File
@@ -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`. */
+9
View File
@@ -57,6 +57,15 @@ export function requestToggleDrawer(): void {
document.dispatchEvent(new CustomEvent(TOGGLE_DRAWER_EVENT));
}
/** Close the mobile drawer unconditionally — used after a spawn lands a new
* session so the content pane is visible. Unlike TOGGLE, this is idempotent:
* firing it when the drawer is already closed is a no-op. */
export const CLOSE_DRAWER_EVENT = "maestro-close-drawer";
export function requestCloseDrawer(): void {
document.dispatchEvent(new CustomEvent(CLOSE_DRAWER_EVENT));
}
/** Toggle the help overlay from a button click (the `?` key is handled by
* HelpOverlay's own command set). */
export const TOGGLE_HELP_EVENT = "maestro-toggle-help";