diff --git a/crates/maestro-daemon/src/session.rs b/crates/maestro-daemon/src/session.rs index c3f803e..caeeca8 100644 --- a/crates/maestro-daemon/src/session.rs +++ b/crates/maestro-daemon/src/session.rs @@ -2003,6 +2003,41 @@ fn apply_persisted_event(projection: &mut Projection, event: &LoggedEvent) -> Pr } async fn sync_structured_activity(inner: &Arc, id: Uuid) { + let events = { + let sessions = inner.sessions.lock().await; + let Some(record) = sessions.get(&id) else { + return; + }; + if record.session.view != SessionView::Structured { + return; + } + record.acp_events.clone() + }; + + match events { + Some(events) => sync_structured_activity_from_stream(inner, id, events).await, + None => apply_structured_activity(inner, id, None, ActivityState::Error).await, + } +} + +async fn sync_structured_activity_from_stream( + inner: &Arc, + id: Uuid, + events: Arc>, +) { + let event_stream = events.lock().await; + let raw = event_stream.projection.activity(); + // Stream state must stay fixed until its catalog transition is applied; + // every other path releases `sessions` before awaiting this stream lock. + apply_structured_activity(inner, id, Some(&events), raw).await; +} + +async fn apply_structured_activity( + inner: &Arc, + id: Uuid, + expected_events: Option<&Arc>>, + raw: ActivityState, +) { let updated = { let mut sessions = inner.sessions.lock().await; let Some(record) = sessions.get_mut(&id) else { @@ -2011,12 +2046,19 @@ async fn sync_structured_activity(inner: &Arc, id: Uuid) { if record.session.view != SessionView::Structured { return; } + let current_events = record.acp_events.as_ref(); + let stream_matches = match (expected_events, current_events) { + (Some(expected), Some(current)) => Arc::ptr_eq(expected, current), + (None, None) => true, + _ => false, + }; + if !stream_matches { + return; + } let raw = if record.session.status == SessionStatus::Dead { ActivityState::Error - } else if let Some(events) = &record.acp_events { - events.lock().await.projection.activity() } else { - ActivityState::Error + raw }; let activity = crate::activity::apply_focus_suppression( raw, @@ -2702,20 +2744,28 @@ async fn cleanup_record( #[cfg(test)] mod tests { use std::fs::File; + use std::sync::Arc; + use std::task::Poll; + use std::time::Duration; use maestro_acp::schema::{ PermissionOption, PermissionOptionKind, RequestPermissionOutcome, RequestPermissionRequest, ToolCallUpdate, ToolCallUpdateFields, }; + use maestro_protocol::{ActivityState, CreateSessionRequest, SessionView}; use super::{ - cancel_pending_approvals, permission_outcome, record_permission_request, AcpDispatchState, - AcpEventStream, ApprovalDecision, RecordOutcome, + cancel_pending_approvals, permission_outcome, record_permission_request, + run_structured_activity_pass, sync_structured_activity_from_stream, AcpDispatchState, + AcpEventStream, ApprovalDecision, RecordOutcome, SessionManager, }; + use crate::activity::FOCUS_CLEAR_HOLD; + use crate::panes::FakePanes; use crate::projection::Projection; use crate::sidecar::{ EventCursor, EventLog, FaultyLogSink, PersistedEvent, SharedBuf, SidecarStore, }; + use crate::worktree::WorktreeManager; fn option(id: &str, kind: PermissionOptionKind) -> PermissionOption { PermissionOption::new(id.to_owned(), id, kind) @@ -2743,6 +2793,144 @@ mod tests { } } + fn terminal_request(cwd: &std::path::Path) -> CreateSessionRequest { + CreateSessionRequest { + view: SessionView::Terminal, + cwd: cwd.to_path_buf(), + command: "sh".into(), + env: std::collections::HashMap::new(), + metadata: std::collections::HashMap::new(), + worktree: None, + } + } + + async fn manager_with_structured_stream() -> ( + tempfile::TempDir, + SessionManager, + uuid::Uuid, + Arc>, + uuid::Uuid, + ) { + let state = tempfile::tempdir().unwrap(); + let manager = SessionManager::new( + Arc::new(FakePanes::new()), + SidecarStore::new(state.path().join("sessions")), + WorktreeManager::new(None), + state.path().join("fifos"), + ); + let structured = manager + .create(terminal_request(state.path())) + .await + .expect("create structured fixture"); + let terminal = manager + .create(terminal_request(state.path())) + .await + .expect("create unrelated terminal"); + let events = Arc::new(tokio::sync::Mutex::new(stream( + manager + .inner + .sidecar + .open_events(structured.id) + .expect("open event log"), + ))); + { + let mut event_stream = events.lock().await; + event_stream.projection.begin_turn(); + event_stream + .projection + .request_approval("nonce".into(), serde_json::json!({})); + } + { + let mut sessions = manager.inner.sessions.lock().await; + let record = sessions.get_mut(&structured.id).unwrap(); + record.session.view = SessionView::Structured; + record.session.activity = ActivityState::AwaitingInput; + record.focused_since = Some(std::time::Instant::now() - FOCUS_CLEAR_HOLD); + record.acp_events = Some(events.clone()); + } + (state, manager, structured.id, events, terminal.id) + } + + #[tokio::test] + async fn blocked_structured_activity_read_does_not_block_the_catalog() { + let (state, manager, structured_id, events, terminal_id) = + manager_with_structured_stream().await; + let held_stream = events.lock().await; + let activity_pass = run_structured_activity_pass(&manager.inner); + tokio::pin!(activity_pass); + assert!(matches!( + futures_util::poll!(activity_pass.as_mut()), + Poll::Pending + )); + + tokio::time::timeout(Duration::from_secs(1), manager.snapshot()) + .await + .expect("catalog snapshot blocked behind another session's event stream"); + tokio::time::timeout( + Duration::from_secs(1), + manager.send_keys(terminal_id, b"responsive"), + ) + .await + .expect("terminal input blocked behind another session's event stream") + .expect("terminal input"); + let created = tokio::time::timeout( + Duration::from_secs(1), + manager.create(terminal_request(state.path())), + ) + .await + .expect("session create blocked behind another session's event stream") + .expect("create unrelated session"); + tokio::time::timeout(Duration::from_secs(1), manager.kill(created.id, false)) + .await + .expect("session kill blocked behind another session's event stream") + .expect("kill unrelated session"); + + drop(held_stream); + tokio::time::timeout(Duration::from_secs(1), activity_pass) + .await + .expect("activity pass did not resume"); + let structured = manager + .snapshot() + .await + .into_iter() + .find(|session| session.id == structured_id) + .unwrap(); + assert_eq!(structured.activity, ActivityState::Idle); + } + + #[tokio::test] + async fn delayed_activity_from_a_replaced_stream_is_discarded() { + let (_state, manager, structured_id, old_events, _terminal_id) = + manager_with_structured_stream().await; + let replacement = Arc::new(tokio::sync::Mutex::new(stream( + manager + .inner + .sidecar + .open_events(uuid::Uuid::new_v4()) + .expect("open replacement event log"), + ))); + let mut sessions = manager.inner.sessions.lock().await; + let sync = + sync_structured_activity_from_stream(&manager.inner, structured_id, old_events.clone()); + tokio::pin!(sync); + assert!(matches!(futures_util::poll!(sync.as_mut()), Poll::Pending)); + let record = sessions.get_mut(&structured_id).unwrap(); + record.acp_events = Some(replacement.clone()); + record.session.activity = ActivityState::Error; + drop(sessions); + + tokio::time::timeout(Duration::from_secs(1), sync) + .await + .expect("stale activity sync did not finish"); + let sessions = manager.inner.sessions.lock().await; + let record = sessions.get(&structured_id).unwrap(); + assert_eq!(record.session.activity, ActivityState::Error); + assert!(Arc::ptr_eq( + record.acp_events.as_ref().unwrap(), + &replacement + )); + } + /// A handle that can never append *or* roll back: opened read-only, so /// both `write_all` and the rollback `truncate` fail with the OS's own /// error (`ftruncate` requires a writable fd too) instead of a