fix(maestro): avoid nested session locks

2026-09-05 05:15:58 +01:00
parent d607740fea
commit 4a1a10c49a
1 changed files with 193 additions and 5 deletions
@@ -2003,6 +2003,41 @@ fn apply_persisted_event(projection: &mut Projection, event: &LoggedEvent) -> Pr
 }

 async fn sync_structured_activity(inner: &Arc<Inner>, 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<Inner>,
+    id: Uuid,
+    events: Arc<Mutex<AcpEventStream>>,
+) {
+    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<Inner>,
+    id: Uuid,
+    expected_events: Option<&Arc<Mutex<AcpEventStream>>>,
+    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<Inner>, 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<tokio::sync::Mutex<AcpEventStream>>,
+        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