From fe26dab6a85fa7227e99aee2daa6bfd38dffc396 Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Sat, 5 Sep 2026 11:55:25 +0100 Subject: [PATCH] fix(maestro): remove structured lock inversion --- crates/maestro-daemon/src/session.rs | 192 ++++++++++++++++++++------- 1 file changed, 143 insertions(+), 49 deletions(-) diff --git a/crates/maestro-daemon/src/session.rs b/crates/maestro-daemon/src/session.rs index 6154c4e..c86578e 100644 --- a/crates/maestro-daemon/src/session.rs +++ b/crates/maestro-daemon/src/session.rs @@ -102,8 +102,12 @@ pub struct SessionRecord { /// outside the sessions lock. Daemon-local; never rides the wire. pub acp: Option>, /// Open log plus its post-append live sender. Their shared lock is the - /// replay/live seam used by the structured WebSocket in #138. + /// replay/live seam used by the structured WebSocket in #138. Callers may + /// clone this handle under `sessions`, but must never hold either guard + /// while acquiring the other. acp_events: Option>>, + /// Latest activity snapshot applied from the current `acp_events` stream. + structured_activity_sequence: u64, /// Completion signal for the task draining ACP notifications into the /// event log. Shutdown waits for it before returning. acp_event_pump: Option>, @@ -129,6 +133,14 @@ struct AcpEventStream { /// claimed: it is queued onto ACP before another operation can proceed. operation: Arc>, dispatch: AcpDispatchState, + activity_sequence: u64, +} + +impl AcpEventStream { + fn activity_snapshot(&mut self) -> (u64, ActivityState) { + self.activity_sequence += 1; + (self.activity_sequence, self.projection.activity()) + } } #[derive(Debug, Default)] @@ -1220,6 +1232,7 @@ impl SessionManager { pending_approvals: HashMap::new(), operation: Arc::new(Mutex::new(())), dispatch: AcpDispatchState::default(), + activity_sequence: 0, }))) } Err(e) => { @@ -1323,6 +1336,7 @@ impl SessionManager { awaiting_suppressed: false, acp, acp_events, + structured_activity_sequence: 0, acp_event_pump: event_pump, }, ); @@ -1534,6 +1548,7 @@ impl SessionManager { // daemon; only tmux-backed sessions can be adopted. acp: None, acp_events: None, + structured_activity_sequence: 0, acp_event_pump: None, }, ); @@ -1655,6 +1670,7 @@ impl SessionManager { pending_approvals: HashMap::new(), operation: Arc::new(Mutex::new(())), dispatch: AcpDispatchState::default(), + activity_sequence: 0, }))) } Err(error) => { @@ -1702,6 +1718,7 @@ impl SessionManager { awaiting_suppressed: false, acp, acp_events, + structured_activity_sequence: 0, acp_event_pump: None, }, ); @@ -2122,57 +2139,51 @@ async fn sync_structured_activity_from_stream( 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; + let (sequence, raw) = events.lock().await.activity_snapshot(); + apply_structured_activity(inner, id, Some((&events, sequence)), raw).await; } async fn apply_structured_activity( inner: &Arc, id: Uuid, - expected_events: Option<&Arc>>, + expected_events: Option<(&Arc>, u64)>, raw: ActivityState, ) { - let updated = { - let mut sessions = inner.sessions.lock().await; - let Some(record) = sessions.get_mut(&id) else { - return; - }; - 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 { - raw - }; - let activity = crate::activity::apply_focus_suppression( - raw, - record.focused_since, - std::time::Instant::now(), - &mut record.awaiting_suppressed, - ); - if record.session.activity == activity { - None - } else { - record.session.activity = activity; - record.session.activity_since = Utc::now(); - Some(record.session.clone()) - } + let mut sessions = inner.sessions.lock().await; + let Some(record) = sessions.get_mut(&id) else { + return; }; - if let Some(session) = updated { - let _ = inner.events.send(CatalogEvent::SessionUpdated { session }); + if record.session.view != SessionView::Structured { + return; + } + let current_events = record.acp_events.as_ref(); + match (expected_events, current_events) { + (Some((expected, sequence)), Some(current)) if Arc::ptr_eq(expected, current) => { + if sequence < record.structured_activity_sequence { + return; + } + record.structured_activity_sequence = sequence; + } + (None, None) => {} + _ => return, + } + let raw = if record.session.status == SessionStatus::Dead { + ActivityState::Error + } else { + raw + }; + let activity = crate::activity::apply_focus_suppression( + raw, + record.focused_since, + std::time::Instant::now(), + &mut record.awaiting_suppressed, + ); + if record.session.activity != activity { + record.session.activity = activity; + record.session.activity_since = Utc::now(); + let _ = inner.events.send(CatalogEvent::SessionUpdated { + session: record.session.clone(), + }); } } @@ -2849,16 +2860,17 @@ mod tests { PermissionOption, PermissionOptionKind, RequestPermissionOutcome, RequestPermissionRequest, ToolCallUpdate, ToolCallUpdateFields, }; - use maestro_protocol::{ActivityState, CreateSessionRequest, SessionView}; + use maestro_protocol::{ActivityState, CatalogEvent, CreateSessionRequest, SessionView}; use super::{ - cancel_pending_approvals, permission_outcome, record_permission_request, - run_structured_activity_pass, sync_structured_activity_from_stream, AcpDispatchState, - AcpEventStream, ApprovalDecision, RecordOutcome, SessionManager, + apply_structured_activity, 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::projection::{Projection, TurnLifecycle}; use crate::sidecar::{ EventCursor, EventLog, FaultyLogSink, PersistedEvent, SharedBuf, SidecarStore, }; @@ -2887,6 +2899,7 @@ mod tests { pending_approvals: std::collections::HashMap::new(), operation: std::sync::Arc::new(tokio::sync::Mutex::new(())), dispatch: AcpDispatchState::default(), + activity_sequence: 0, } } @@ -3043,6 +3056,87 @@ mod tests { .expect("terminal resize remains available"); } + #[tokio::test] + async fn structured_activity_releases_stream_before_waiting_for_catalog() { + let (_state, manager, _panes, structured_id, events, _terminal_id) = + manager_with_structured_stream().await; + let sessions = manager.inner.sessions.lock().await; + let sync = + sync_structured_activity_from_stream(&manager.inner, structured_id, events.clone()); + tokio::pin!(sync); + assert!(matches!(futures_util::poll!(sync.as_mut()), Poll::Pending)); + + assert!( + events.try_lock().is_ok(), + "activity sync held the event stream while waiting for the catalog" + ); + + drop(sessions); + tokio::time::timeout(Duration::from_secs(1), sync) + .await + .expect("activity sync did not finish"); + } + + #[tokio::test] + async fn newer_structured_activity_snapshot_cannot_be_overtaken() { + let (_state, manager, _panes, structured_id, events, _terminal_id) = + manager_with_structured_stream().await; + let mut catalog = manager.events(); + manager + .inner + .sessions + .lock() + .await + .get_mut(&structured_id) + .unwrap() + .focused_since = None; + + let (older_sequence, older_activity) = events.lock().await.activity_snapshot(); + let (newer_sequence, newer_activity) = { + let mut event_stream = events.lock().await; + let turn = event_stream.projection.active_turn().unwrap(); + event_stream.projection.clear_approval(); + event_stream + .projection + .turn_lifecycle(TurnLifecycle::Finish, turn); + event_stream.activity_snapshot() + }; + + apply_structured_activity( + &manager.inner, + structured_id, + Some((&events, newer_sequence)), + newer_activity, + ) + .await; + apply_structured_activity( + &manager.inner, + structured_id, + Some((&events, older_sequence)), + older_activity, + ) + .await; + + let CatalogEvent::SessionUpdated { session } = catalog.try_recv().unwrap() else { + panic!("unexpected catalog event"); + }; + assert_eq!(session.activity, ActivityState::Idle); + assert!(matches!( + catalog.try_recv(), + Err(tokio::sync::broadcast::error::TryRecvError::Empty) + )); + assert_eq!( + manager + .snapshot() + .await + .into_iter() + .find(|session| session.id == structured_id) + .unwrap() + .activity, + ActivityState::Idle + ); + } + #[tokio::test] async fn delayed_activity_from_a_replaced_stream_is_discarded() { let (_state, manager, _panes, structured_id, old_events, _terminal_id) =