Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a9717d01c8 | ||
|
|
35a6ff9d0f | ||
|
|
474000935a | ||
|
|
5f8cc2a140 | ||
|
|
432967fe64 | ||
|
|
2512b8be42 | ||
|
|
e3556bd847 | ||
|
|
ce0e4e6820 | ||
|
|
923911f631 |
@@ -58,6 +58,24 @@ use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
|
|||||||
|
|
||||||
pub use agent_client_protocol::schema::v1 as schema;
|
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.
|
/// ACP adapter selected from the structured session's resolved command.
|
||||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||||
pub enum AcpAdapter {
|
pub enum AcpAdapter {
|
||||||
@@ -354,7 +372,7 @@ pub struct AcpSession {
|
|||||||
pid: u32,
|
pid: u32,
|
||||||
commands: mpsc::UnboundedSender<Command>,
|
commands: mpsc::UnboundedSender<Command>,
|
||||||
child_commands: mpsc::UnboundedSender<ChildCommand>,
|
child_commands: mpsc::UnboundedSender<ChildCommand>,
|
||||||
updates: Mutex<Option<mpsc::UnboundedReceiver<SessionNotification>>>,
|
updates: Mutex<Option<mpsc::UnboundedReceiver<AcpUpdate>>>,
|
||||||
permission_requests: Mutex<Option<mpsc::UnboundedReceiver<PermissionRequest>>>,
|
permission_requests: Mutex<Option<mpsc::UnboundedReceiver<PermissionRequest>>>,
|
||||||
exit: watch::Receiver<Option<ExitStatus>>,
|
exit: watch::Receiver<Option<ExitStatus>>,
|
||||||
}
|
}
|
||||||
@@ -393,9 +411,9 @@ impl AcpSession {
|
|||||||
wait_exit(&self.exit).await
|
wait_exit(&self.exit).await
|
||||||
}
|
}
|
||||||
|
|
||||||
/// The stream of raw `session/update` notifications, handed out once.
|
/// The stream of updates (notifications + turn-done markers), handed out
|
||||||
/// Notifications that arrived before the take are still queued on it.
|
/// once. Items that arrived before the take are still queued.
|
||||||
pub fn take_updates(&self) -> Option<mpsc::UnboundedReceiver<SessionNotification>> {
|
pub fn take_updates(&self) -> Option<mpsc::UnboundedReceiver<AcpUpdate>> {
|
||||||
self.updates.lock().expect("updates mutex").take()
|
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>(
|
async fn run_connection<T>(
|
||||||
transport: T,
|
transport: T,
|
||||||
config: ConnectionConfig,
|
config: ConnectionConfig,
|
||||||
updates_tx: mpsc::UnboundedSender<SessionNotification>,
|
updates_tx: mpsc::UnboundedSender<AcpUpdate>,
|
||||||
permission_requests: mpsc::UnboundedSender<PermissionRequest>,
|
permission_requests: mpsc::UnboundedSender<PermissionRequest>,
|
||||||
mut commands_rx: mpsc::UnboundedReceiver<Command>,
|
mut commands_rx: mpsc::UnboundedReceiver<Command>,
|
||||||
ready_tx: oneshot::Sender<Result<String, ReadyError>>,
|
ready_tx: oneshot::Sender<Result<String, ReadyError>>,
|
||||||
@@ -800,7 +818,7 @@ async fn run_connection<T>(
|
|||||||
return Ok(());
|
return Ok(());
|
||||||
}
|
}
|
||||||
// A dropped receiver only means nobody wants updates any more.
|
// 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(())
|
Ok(())
|
||||||
},
|
},
|
||||||
agent_client_protocol::on_receive_notification!(),
|
agent_client_protocol::on_receive_notification!(),
|
||||||
@@ -916,18 +934,18 @@ async fn run_connection<T>(
|
|||||||
models.as_ref(),
|
models.as_ref(),
|
||||||
)));
|
)));
|
||||||
if let Some(modes) = &modes {
|
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()),
|
SessionId::new(session_id.as_str()),
|
||||||
SessionUpdate::CurrentModeUpdate(CurrentModeUpdate::new(
|
SessionUpdate::CurrentModeUpdate(CurrentModeUpdate::new(
|
||||||
modes.current_mode_id.clone(),
|
modes.current_mode_id.clone(),
|
||||||
)),
|
)),
|
||||||
));
|
))));
|
||||||
}
|
}
|
||||||
if !config_options.lock().expect("config options").is_empty() {
|
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,
|
&session_id,
|
||||||
config_options.lock().expect("config options").clone(),
|
config_options.lock().expect("config options").clone(),
|
||||||
));
|
))));
|
||||||
}
|
}
|
||||||
let _ = ready_tx.send(Ok(session_id.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 done = Arc::new(Mutex::new(Some(done)));
|
||||||
let callback_done = done.clone();
|
let callback_done = done.clone();
|
||||||
let callback_turn = permission_turn.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| {
|
if let Err(error) = cx.send_request(request).on_receiving_result(move |result| {
|
||||||
let callback_done = callback_done.clone();
|
let callback_done = callback_done.clone();
|
||||||
let callback_turn = callback_turn.clone();
|
let callback_turn = callback_turn.clone();
|
||||||
|
let callback_updates = callback_updates.clone();
|
||||||
async move {
|
async move {
|
||||||
callback_turn
|
callback_turn
|
||||||
.lock()
|
.lock()
|
||||||
.expect("permission turn state")
|
.expect("permission turn state")
|
||||||
.finish(turn);
|
.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() {
|
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(())
|
Ok(())
|
||||||
}
|
}
|
||||||
@@ -998,7 +1023,7 @@ async fn run_connection<T>(
|
|||||||
}).on_receiving_result(move |result| async move {
|
}).on_receiving_result(move |result| async move {
|
||||||
if result.is_ok() {
|
if result.is_ok() {
|
||||||
let options = replace_config_value(&callback_options, &callback_config_id, &callback_value);
|
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() {
|
if let Some(done) = callback_done.lock().expect("config sender").take() {
|
||||||
let _ = done.send(result.map(|_| ()).map_err(|error| error.to_string()));
|
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 {
|
)).on_receiving_result(move |result| async move {
|
||||||
if result.is_ok() {
|
if result.is_ok() {
|
||||||
let options = replace_config_value(&callback_options, &callback_config_id, &callback_value);
|
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))));
|
||||||
let _ = callback_updates.send(SessionNotification::new(
|
let _ = callback_updates.send(AcpUpdate::Notification(Box::new(SessionNotification::new(
|
||||||
SessionId::new(callback_session.as_str()),
|
SessionId::new(callback_session.as_str()),
|
||||||
SessionUpdate::CurrentModeUpdate(CurrentModeUpdate::new(
|
SessionUpdate::CurrentModeUpdate(CurrentModeUpdate::new(
|
||||||
callback_value.as_value_id().expect("checked mode value").0.clone(),
|
callback_value.as_value_id().expect("checked mode value").0.clone(),
|
||||||
)),
|
)),
|
||||||
));
|
))));
|
||||||
}
|
}
|
||||||
if let Some(done) = callback_done.lock().expect("config sender").take() {
|
if let Some(done) = callback_done.lock().expect("config sender").take() {
|
||||||
let _ = done.send(result.map(|_| ()).map_err(|error| error.to_string()));
|
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");
|
let mut current = callback_options.lock().expect("config options");
|
||||||
current.retain(|option| option.id.0.as_ref().starts_with("__maestro_legacy_"));
|
current.retain(|option| option.id.0.as_ref().starts_with("__maestro_legacy_"));
|
||||||
current.extend(response.config_options.clone());
|
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() {
|
if let Some(done) = callback_done.lock().expect("config sender").take() {
|
||||||
let _ = done.send(result.map(|_| ()).map_err(|error| error.to_string()));
|
let _ = done.send(result.map(|_| ()).map_err(|error| error.to_string()));
|
||||||
|
|||||||
@@ -193,6 +193,9 @@ async fn session_updates_reach_the_taken_receiver() {
|
|||||||
.await
|
.await
|
||||||
.expect("update within 5s")
|
.expect("update within 5s")
|
||||||
.expect("channel open");
|
.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());
|
assert_eq!(first.session_id.0.as_ref(), session.session_id());
|
||||||
match first.update {
|
match first.update {
|
||||||
SessionUpdate::AgentMessageChunk(chunk) => match chunk.content {
|
SessionUpdate::AgentMessageChunk(chunk) => match chunk.content {
|
||||||
@@ -221,7 +224,10 @@ async fn advertised_settings_are_normalized_and_mutable() {
|
|||||||
.await
|
.await
|
||||||
.expect("settings update")
|
.expect("settings update")
|
||||||
.expect("channel open");
|
.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;
|
break options.config_options;
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
@@ -390,7 +396,10 @@ async fn drop_terminates_and_reaps_the_child() {
|
|||||||
fake.wait_for_exit().await;
|
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 {
|
match &n.update {
|
||||||
SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content {
|
SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content {
|
||||||
ContentBlock::Text(t) => t.text.clone(),
|
ContentBlock::Text(t) => t.text.clone(),
|
||||||
|
|||||||
@@ -482,7 +482,7 @@ async fn structured_prompt(
|
|||||||
Path(id): Path<Uuid>,
|
Path(id): Path<Uuid>,
|
||||||
Json(body): Json<StructuredPromptRequest>,
|
Json(body): Json<StructuredPromptRequest>,
|
||||||
) -> Response {
|
) -> Response {
|
||||||
match s.manager.queue_acp_prompt(id, body.prompt).await {
|
match s.manager.queue_acp_prompt(id, body.prompt, None).await {
|
||||||
Ok(QueueAcpPrompt::Dispatched) => StatusCode::NO_CONTENT.into_response(),
|
Ok(QueueAcpPrompt::Dispatched) => StatusCode::NO_CONTENT.into_response(),
|
||||||
Ok(QueueAcpPrompt::Queued) => StatusCode::ACCEPTED.into_response(),
|
Ok(QueueAcpPrompt::Queued) => StatusCode::ACCEPTED.into_response(),
|
||||||
Ok(QueueAcpPrompt::Full) => error_response(
|
Ok(QueueAcpPrompt::Full) => error_response(
|
||||||
|
|||||||
@@ -757,6 +757,7 @@ impl SessionManager {
|
|||||||
&self,
|
&self,
|
||||||
id: Uuid,
|
id: Uuid,
|
||||||
text: String,
|
text: String,
|
||||||
|
row_metadata: Option<serde_json::Value>,
|
||||||
) -> Result<QueueAcpPrompt, SessionError> {
|
) -> Result<QueueAcpPrompt, SessionError> {
|
||||||
let (acp, stream) = {
|
let (acp, stream) = {
|
||||||
let sessions = self.inner.sessions.lock().await;
|
let sessions = self.inner.sessions.lock().await;
|
||||||
@@ -833,12 +834,19 @@ impl SessionManager {
|
|||||||
self.persist_prompt_queue(id).await;
|
self.persist_prompt_queue(id).await;
|
||||||
return Ok(QueueAcpPrompt::Queued);
|
return Ok(QueueAcpPrompt::Queued);
|
||||||
}
|
}
|
||||||
|
let mut data = serde_json::json!({
|
||||||
|
"content": {"type": "text", "text": text},
|
||||||
|
});
|
||||||
|
if let Some(extra) = &row_metadata {
|
||||||
|
if let (Some(base), Some(extra)) = (data.as_object_mut(), extra.as_object())
|
||||||
|
{
|
||||||
|
base.extend(extra.iter().map(|(k, v)| (k.clone(), v.clone())));
|
||||||
|
}
|
||||||
|
}
|
||||||
let row = TranscriptRow {
|
let row = TranscriptRow {
|
||||||
id: format!("user:{}", Uuid::new_v4()),
|
id: format!("user:{}", Uuid::new_v4()),
|
||||||
kind: TranscriptRowKind::User,
|
kind: TranscriptRowKind::User,
|
||||||
data: serde_json::json!({
|
data,
|
||||||
"content": {"type": "text", "text": text},
|
|
||||||
}),
|
|
||||||
};
|
};
|
||||||
let event = PersistedEvent::User { row: row.clone() };
|
let event = PersistedEvent::User { row: row.clone() };
|
||||||
let seq = match stream.log.append(&event) {
|
let seq = match stream.log.append(&event) {
|
||||||
@@ -1631,7 +1639,7 @@ impl SessionManager {
|
|||||||
let manager = SessionManager {
|
let manager = SessionManager {
|
||||||
inner: inner.clone(),
|
inner: inner.clone(),
|
||||||
};
|
};
|
||||||
let _ = manager.queue_acp_prompt(id, text).await;
|
let _ = manager.queue_acp_prompt(id, text, None).await;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
@@ -2022,7 +2030,9 @@ impl SessionManager {
|
|||||||
if let Some(record) = self.inner.sessions.lock().await.get_mut(&id) {
|
if let Some(record) = self.inner.sessions.lock().await.get_mut(&id) {
|
||||||
record.acp_event_pump = Some(done);
|
record.acp_event_pump = Some(done);
|
||||||
}
|
}
|
||||||
// Drain any prompts that were queued before the daemon restarted.
|
// 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 queued_text = {
|
||||||
let stream = acp_events
|
let stream = acp_events
|
||||||
.as_ref()
|
.as_ref()
|
||||||
@@ -2037,7 +2047,15 @@ impl SessionManager {
|
|||||||
};
|
};
|
||||||
if let Some(text) = queued_text {
|
if let Some(text) = queued_text {
|
||||||
self.persist_prompt_queue(id).await;
|
self.persist_prompt_queue(id).await;
|
||||||
let _ = self.queue_acp_prompt(id, text).await;
|
let _ = self.queue_acp_prompt(id, text, None).await;
|
||||||
|
} else {
|
||||||
|
let _ = self
|
||||||
|
.queue_acp_prompt(
|
||||||
|
id,
|
||||||
|
"The daemon restarted. Continue where you left off.".into(),
|
||||||
|
Some(serde_json::json!({"maestro": {"type": "restart_continue"}})),
|
||||||
|
)
|
||||||
|
.await;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
let _ = self
|
let _ = self
|
||||||
@@ -2385,34 +2403,49 @@ fn start_acp_event_pump(
|
|||||||
let weak_inner = Arc::downgrade(inner);
|
let weak_inner = Arc::downgrade(inner);
|
||||||
let weak_acp = Arc::downgrade(acp);
|
let weak_acp = Arc::downgrade(acp);
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
while let Some(notification) = updates.recv().await {
|
while let Some(update) = updates.recv().await {
|
||||||
let append_error = {
|
match update {
|
||||||
let mut events = events.lock().await;
|
maestro_acp::AcpUpdate::Notification(notification) => {
|
||||||
let event = PersistedEvent::Notification(Box::new(notification));
|
let append_error = {
|
||||||
match events.log.append(&event) {
|
let mut events = events.lock().await;
|
||||||
Ok(seq) => {
|
let event = PersistedEvent::Notification(notification);
|
||||||
let logged = LoggedEvent {
|
match events.log.append(&event) {
|
||||||
seq,
|
Ok(seq) => {
|
||||||
event: event.clone(),
|
let logged = LoggedEvent {
|
||||||
};
|
seq,
|
||||||
let projected = apply_persisted_event(&mut events.projection, &logged);
|
event: event.clone(),
|
||||||
let _ = events.live.send(logged);
|
};
|
||||||
events.broadcast(projected);
|
let projected =
|
||||||
None
|
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),
|
|
||||||
}
|
}
|
||||||
};
|
maestro_acp::AcpUpdate::TurnDone {
|
||||||
if let Some(error) = append_error {
|
turn: _,
|
||||||
tracing::error!(%id, %error, "structured event log append failed");
|
result,
|
||||||
if let Some(acp) = weak_acp.upgrade() {
|
done,
|
||||||
acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
|
} => {
|
||||||
.await;
|
// 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);
|
let _ = done_tx.send(true);
|
||||||
@@ -3017,7 +3050,7 @@ async fn run_blocked_work_pass(inner: &Arc<Inner>) {
|
|||||||
fn drain_prompt_queue(manager: SessionManager, id: Uuid, text: String) {
|
fn drain_prompt_queue(manager: SessionManager, id: Uuid, text: String) {
|
||||||
let fut: std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>> =
|
let fut: std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>> =
|
||||||
Box::pin(async move {
|
Box::pin(async move {
|
||||||
let _ = manager.queue_acp_prompt(id, text).await;
|
let _ = manager.queue_acp_prompt(id, text, None).await;
|
||||||
});
|
});
|
||||||
tokio::spawn(fut);
|
tokio::spawn(fut);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -84,15 +84,27 @@ impl Fake {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn event_text(event: &maestro_daemon::sidecar::LoggedEvent) -> &str {
|
fn event_text(event: &maestro_daemon::sidecar::LoggedEvent) -> &str {
|
||||||
let maestro_daemon::sidecar::PersistedEvent::Notification(notification) = &event.event else {
|
match &event.event {
|
||||||
panic!("unexpected daemon event")
|
maestro_daemon::sidecar::PersistedEvent::Notification(notification) => {
|
||||||
};
|
match ¬ification.update {
|
||||||
match ¬ification.update {
|
SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content {
|
||||||
SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content {
|
ContentBlock::Text(text) => &text.text,
|
||||||
ContentBlock::Text(text) => &text.text,
|
other => panic!("unexpected content {other:?}"),
|
||||||
other => panic!("unexpected content {other:?}"),
|
},
|
||||||
},
|
other => panic!("unexpected update {other:?}"),
|
||||||
other => panic!("unexpected update {other:?}"),
|
}
|
||||||
|
}
|
||||||
|
maestro_daemon::sidecar::PersistedEvent::User { row } => row
|
||||||
|
.data
|
||||||
|
.pointer("/content/text")
|
||||||
|
.and_then(serde_json::Value::as_str)
|
||||||
|
.expect("user row text"),
|
||||||
|
maestro_daemon::sidecar::PersistedEvent::Turn { row } => row
|
||||||
|
.data
|
||||||
|
.pointer("/maestroTurnMeta/outcome")
|
||||||
|
.and_then(serde_json::Value::as_str)
|
||||||
|
.unwrap_or("[turn]"),
|
||||||
|
maestro_daemon::sidecar::PersistedEvent::Approval { .. } => "[approval]",
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -259,7 +271,7 @@ async fn structured_activity_tracks_turn_completion_and_failure() {
|
|||||||
|
|
||||||
assert!(matches!(
|
assert!(matches!(
|
||||||
fx.manager
|
fx.manager
|
||||||
.queue_acp_prompt(session.id, "fail".into())
|
.queue_acp_prompt(session.id, "fail".into(), None)
|
||||||
.await
|
.await
|
||||||
.unwrap(),
|
.unwrap(),
|
||||||
maestro_daemon::session::QueueAcpPrompt::Dispatched
|
maestro_daemon::session::QueueAcpPrompt::Dispatched
|
||||||
@@ -286,7 +298,7 @@ async fn structured_activity_tracks_turn_completion_and_failure() {
|
|||||||
|
|
||||||
assert!(matches!(
|
assert!(matches!(
|
||||||
fx.manager
|
fx.manager
|
||||||
.queue_acp_prompt(session.id, "succeed".into())
|
.queue_acp_prompt(session.id, "succeed".into(), None)
|
||||||
.await
|
.await
|
||||||
.unwrap(),
|
.unwrap(),
|
||||||
maestro_daemon::session::QueueAcpPrompt::Dispatched
|
maestro_daemon::session::QueueAcpPrompt::Dispatched
|
||||||
@@ -426,7 +438,7 @@ async fn queued_prompt_is_a_durable_user_transcript_row() {
|
|||||||
|
|
||||||
assert!(matches!(
|
assert!(matches!(
|
||||||
fx.manager
|
fx.manager
|
||||||
.queue_acp_prompt(session.id, "show my prompt".into())
|
.queue_acp_prompt(session.id, "show my prompt".into(), None)
|
||||||
.await
|
.await
|
||||||
.expect("queue prompt"),
|
.expect("queue prompt"),
|
||||||
maestro_daemon::session::QueueAcpPrompt::Dispatched
|
maestro_daemon::session::QueueAcpPrompt::Dispatched
|
||||||
@@ -516,7 +528,7 @@ async fn cancelled_prompt_that_never_settles_fails_the_session() {
|
|||||||
|
|
||||||
assert!(matches!(
|
assert!(matches!(
|
||||||
fx.manager
|
fx.manager
|
||||||
.queue_acp_prompt(session.id, "first".into())
|
.queue_acp_prompt(session.id, "first".into(), None)
|
||||||
.await
|
.await
|
||||||
.unwrap(),
|
.unwrap(),
|
||||||
maestro_daemon::session::QueueAcpPrompt::Dispatched
|
maestro_daemon::session::QueueAcpPrompt::Dispatched
|
||||||
@@ -524,7 +536,7 @@ async fn cancelled_prompt_that_never_settles_fails_the_session() {
|
|||||||
assert!(fx.manager.cancel_acp_turn(session.id).await.unwrap());
|
assert!(fx.manager.cancel_acp_turn(session.id).await.unwrap());
|
||||||
let second = fx
|
let second = fx
|
||||||
.manager
|
.manager
|
||||||
.queue_acp_prompt(session.id, "second".into())
|
.queue_acp_prompt(session.id, "second".into(), None)
|
||||||
.await
|
.await
|
||||||
.unwrap();
|
.unwrap();
|
||||||
assert!(
|
assert!(
|
||||||
@@ -555,7 +567,7 @@ async fn cancellation_settlement_allows_the_next_prompt_permission() {
|
|||||||
|
|
||||||
assert!(matches!(
|
assert!(matches!(
|
||||||
fx.manager
|
fx.manager
|
||||||
.queue_acp_prompt(session.id, "first".into())
|
.queue_acp_prompt(session.id, "first".into(), None)
|
||||||
.await
|
.await
|
||||||
.unwrap(),
|
.unwrap(),
|
||||||
maestro_daemon::session::QueueAcpPrompt::Dispatched
|
maestro_daemon::session::QueueAcpPrompt::Dispatched
|
||||||
@@ -563,7 +575,7 @@ async fn cancellation_settlement_allows_the_next_prompt_permission() {
|
|||||||
assert!(fx.manager.cancel_acp_turn(session.id).await.unwrap());
|
assert!(fx.manager.cancel_acp_turn(session.id).await.unwrap());
|
||||||
assert!(matches!(
|
assert!(matches!(
|
||||||
fx.manager
|
fx.manager
|
||||||
.queue_acp_prompt(session.id, "second".into())
|
.queue_acp_prompt(session.id, "second".into(), None)
|
||||||
.await
|
.await
|
||||||
.unwrap(),
|
.unwrap(),
|
||||||
maestro_daemon::session::QueueAcpPrompt::Dispatched
|
maestro_daemon::session::QueueAcpPrompt::Dispatched
|
||||||
@@ -655,25 +667,18 @@ async fn recovery_loads_the_acp_session_without_duplicating_history() {
|
|||||||
revived.status(session.id).await,
|
revived.status(session.id).await,
|
||||||
Some(SessionStatus::Running)
|
Some(SessionStatus::Running)
|
||||||
);
|
);
|
||||||
assert_eq!(
|
|
||||||
revived
|
|
||||||
.snapshot()
|
|
||||||
.await
|
|
||||||
.into_iter()
|
|
||||||
.find(|candidate| candidate.id == session.id)
|
|
||||||
.unwrap()
|
|
||||||
.activity,
|
|
||||||
ActivityState::Idle
|
|
||||||
);
|
|
||||||
let log = fake.log();
|
let log = fake.log();
|
||||||
assert!(
|
assert!(
|
||||||
log.contains(&format!("recv session/load sessionId={acp_id}")),
|
log.contains(&format!("recv session/load sessionId={acp_id}")),
|
||||||
"{log}"
|
"{log}"
|
||||||
);
|
);
|
||||||
|
// Recovery adds: a continue-prompt User row, the adapter's post-load
|
||||||
|
// notification, and a turn-completed footer — 5 events total before the
|
||||||
|
// next manual tick.
|
||||||
wait_for_async(Duration::from_secs(5), || async {
|
wait_for_async(Duration::from_secs(5), || async {
|
||||||
revived
|
revived
|
||||||
.read_acp_events(session.id, EventCursor::Since(0), 10)
|
.read_acp_events(session.id, EventCursor::Since(0), 10)
|
||||||
.is_ok_and(|events| events.len() == 3)
|
.is_ok_and(|events| events.len() == 5)
|
||||||
})
|
})
|
||||||
.await;
|
.await;
|
||||||
|
|
||||||
@@ -682,7 +687,7 @@ async fn recovery_loads_the_acp_session_without_duplicating_history() {
|
|||||||
.await
|
.await
|
||||||
.expect("replay read")
|
.expect("replay read")
|
||||||
.expect("recovered event subscription");
|
.expect("recovered event subscription");
|
||||||
assert_eq!(replay.len(), 3);
|
assert_eq!(replay.len(), 5);
|
||||||
assert!(live.try_recv().is_err(), "replay duplicated onto live");
|
assert!(live.try_recv().is_err(), "replay duplicated onto live");
|
||||||
let new_pid = fake.pid();
|
let new_pid = fake.pid();
|
||||||
assert_ne!(new_pid, old_pid);
|
assert_ne!(new_pid, old_pid);
|
||||||
@@ -693,16 +698,23 @@ async fn recovery_loads_the_acp_session_without_duplicating_history() {
|
|||||||
.await
|
.await
|
||||||
.expect("live event after recovery")
|
.expect("live event after recovery")
|
||||||
.expect("broadcast open");
|
.expect("broadcast open");
|
||||||
assert_eq!(event.seq, 4);
|
assert_eq!(event.seq, 6);
|
||||||
assert_eq!(event_text(&event), "tick 1");
|
assert_eq!(event_text(&event), "tick 1");
|
||||||
|
|
||||||
let stored = revived
|
let stored = revived
|
||||||
.read_acp_events(session.id, EventCursor::Since(0), 10)
|
.read_acp_events(session.id, EventCursor::Since(0), 10)
|
||||||
.expect("recovered log");
|
.expect("recovered log");
|
||||||
assert_eq!(stored.len(), 4, "session/load replay must not be appended");
|
assert_eq!(stored.len(), 6, "session/load replay must not be appended");
|
||||||
assert_eq!(
|
assert_eq!(
|
||||||
stored.iter().map(event_text).collect::<Vec<_>>(),
|
stored.iter().map(event_text).collect::<Vec<_>>(),
|
||||||
["hello from fake", "tick 1", "hello after load", "tick 1"]
|
[
|
||||||
|
"hello from fake",
|
||||||
|
"tick 1",
|
||||||
|
"The daemon restarted. Continue where you left off.",
|
||||||
|
"hello after load",
|
||||||
|
"completed",
|
||||||
|
"tick 1",
|
||||||
|
]
|
||||||
);
|
);
|
||||||
|
|
||||||
revived.kill(session.id, false).await.expect("kill");
|
revived.kill(session.id, false).await.expect("kill");
|
||||||
|
|||||||
+11
-3
@@ -113,15 +113,23 @@ function Shell() {
|
|||||||
// a transient lost connection has already recovered.
|
// a transient lost connection has already recovered.
|
||||||
useEffect(() => {
|
useEffect(() => {
|
||||||
const recover = () => {
|
const recover = () => {
|
||||||
if (document.visibilityState !== "visible") return;
|
|
||||||
void linkStore.getState().probe();
|
void linkStore.getState().probe();
|
||||||
catalogStore.getState().reconnect();
|
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("pageshow", recover);
|
||||||
window.addEventListener("online", recover);
|
window.addEventListener("online", recover);
|
||||||
return () => {
|
return () => {
|
||||||
document.removeEventListener("visibilitychange", recover);
|
document.removeEventListener("visibilitychange", onVisibility);
|
||||||
window.removeEventListener("pageshow", recover);
|
window.removeEventListener("pageshow", recover);
|
||||||
window.removeEventListener("online", recover);
|
window.removeEventListener("online", recover);
|
||||||
};
|
};
|
||||||
|
|||||||
@@ -481,6 +481,7 @@
|
|||||||
}
|
}
|
||||||
|
|
||||||
.st-tool-arg {
|
.st-tool-arg {
|
||||||
|
flex: 1 1 0;
|
||||||
color: var(--muted);
|
color: var(--muted);
|
||||||
overflow: hidden;
|
overflow: hidden;
|
||||||
text-overflow: ellipsis;
|
text-overflow: ellipsis;
|
||||||
|
|||||||
@@ -197,11 +197,12 @@ export function useStructuredSession(
|
|||||||
};
|
};
|
||||||
|
|
||||||
// Mobile browsers suspend timers and silently kill WebSockets in the
|
// 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 = () => {
|
const onVisibility = () => {
|
||||||
if (document.visibilityState !== "visible" || disposed) return;
|
if (document.visibilityState !== "visible" || disposed) return;
|
||||||
if (socket && (socket.readyState === WebSocket.OPEN || socket.readyState === WebSocket.CONNECTING)) return;
|
|
||||||
cancelTimer();
|
|
||||||
setStatus("reconnecting");
|
setStatus("reconnecting");
|
||||||
connect();
|
connect();
|
||||||
};
|
};
|
||||||
|
|||||||
@@ -303,17 +303,22 @@ export const TerminalPane = forwardRef<TerminalPaneHandle, TerminalPaneProps>(
|
|||||||
ta?.addEventListener("blur", onTaBlur);
|
ta?.addEventListener("blur", onTaBlur);
|
||||||
|
|
||||||
// Mobile browsers (Android especially) suspend timers and silently kill
|
// Mobile browsers (Android especially) suspend timers and silently kill
|
||||||
// WebSockets when the app goes to background. The backoff timer set by
|
// WebSockets when the app goes to background. The browser often still
|
||||||
// onclose is frozen too, so recovery stalls until visibility returns. Kick
|
// reports readyState === OPEN because TCP keepalive hasn't noticed the
|
||||||
// a fresh connect immediately when the page becomes visible again.
|
// 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 = () => {
|
const onVisibility = () => {
|
||||||
if (document.visibilityState !== "visible" || disposed) return;
|
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) {
|
if (reconnectTimer !== null) {
|
||||||
window.clearTimeout(reconnectTimer);
|
window.clearTimeout(reconnectTimer);
|
||||||
reconnectTimer = null;
|
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;
|
reconnectMs = INITIAL_RECONNECT_MS;
|
||||||
onStatusChange?.("reconnecting");
|
onStatusChange?.("reconnecting");
|
||||||
connect();
|
connect();
|
||||||
|
|||||||
+6
-1
@@ -2648,6 +2648,7 @@ input.leaf-rename:focus {
|
|||||||
}
|
}
|
||||||
|
|
||||||
.st-tool-arg {
|
.st-tool-arg {
|
||||||
|
flex: 1 1 0;
|
||||||
color: var(--muted);
|
color: var(--muted);
|
||||||
overflow: hidden;
|
overflow: hidden;
|
||||||
text-overflow: ellipsis;
|
text-overflow: ellipsis;
|
||||||
@@ -2893,7 +2894,11 @@ input.leaf-rename:focus {
|
|||||||
}
|
}
|
||||||
|
|
||||||
.st-tool-group-title {
|
.st-tool-group-title {
|
||||||
flex: 0 0 auto;
|
flex: 1 1 0;
|
||||||
|
min-width: 0;
|
||||||
|
overflow: hidden;
|
||||||
|
text-overflow: ellipsis;
|
||||||
|
white-space: nowrap;
|
||||||
font-weight: 500;
|
font-weight: 500;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user