Compare commits
7
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1ea463f6cf | ||
|
|
7c27a897ac | ||
|
|
a9717d01c8 | ||
|
|
35a6ff9d0f | ||
|
|
474000935a | ||
|
|
5f8cc2a140 | ||
|
|
432967fe64 |
@@ -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()));
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -2403,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);
|
||||
events.broadcast(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);
|
||||
|
||||
+13
-3
@@ -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],
|
||||
);
|
||||
|
||||
@@ -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>}>
|
||||
|
||||
@@ -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);
|
||||
};
|
||||
}, []);
|
||||
|
||||
|
||||
@@ -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.
|
||||
}
|
||||
}}
|
||||
/>
|
||||
|
||||
@@ -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();
|
||||
};
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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";
|
||||
|
||||
Reference in New Issue
Block a user