fix(maestro): key activity watermark by stream

crates/maestro-daemon/src/session.rs

@@ -10,7 +10,7 @@ use std::future::Future;
 use std::path::PathBuf;
 use std::pin::Pin;
 use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
-use std::sync::Arc;
+use std::sync::{Arc, Weak};
 use std::task::{Context, Poll};
 use std::time::Duration;
 
@@ -106,8 +106,8 @@ pub struct SessionRecord {
     /// clone this handle under `sessions`, but must never hold either guard
     /// while acquiring the other.
     acp_events: Option<Arc<Mutex<AcpEventStream>>>,
-    /// Latest activity snapshot applied from the current `acp_events` stream.
-    structured_activity_sequence: u64,
+    /// Latest activity snapshot applied, keyed to its originating stream.
+    structured_activity_watermark: Option<StructuredActivityWatermark>,
     /// Completion signal for the task draining ACP notifications into the
     /// event log. Shutdown waits for it before returning.
     acp_event_pump: Option<watch::Receiver<bool>>,
@@ -122,6 +122,12 @@ impl SessionRecord {
     }
 }
 
+#[derive(Debug)]
+struct StructuredActivityWatermark {
+    stream: Weak<Mutex<AcpEventStream>>,
+    sequence: u64,
+}
+
 #[derive(Debug)]
 struct AcpEventStream {
     log: EventLog,
@@ -1336,7 +1342,7 @@ impl SessionManager {
                     awaiting_suppressed: false,
                     acp,
                     acp_events,
-                    structured_activity_sequence: 0,
+                    structured_activity_watermark: None,
                     acp_event_pump: event_pump,
                 },
             );
@@ -1548,7 +1554,7 @@ impl SessionManager {
                     // daemon; only tmux-backed sessions can be adopted.
                     acp: None,
                     acp_events: None,
-                    structured_activity_sequence: 0,
+                    structured_activity_watermark: None,
                     acp_event_pump: None,
                 },
             );
@@ -1718,7 +1724,7 @@ impl SessionManager {
                     awaiting_suppressed: false,
                     acp,
                     acp_events,
-                    structured_activity_sequence: 0,
+                    structured_activity_watermark: None,
                     acp_event_pump: None,
                 },
             );
@@ -2159,10 +2165,18 @@ async fn apply_structured_activity(
     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 {
+            let stream = Arc::downgrade(current);
+            if record
+                .structured_activity_watermark
+                .as_ref()
+                .is_some_and(|watermark| {
+                    watermark.stream.ptr_eq(&stream) && sequence < watermark.sequence
+                })
+            {
                 return;
             }
-            record.structured_activity_sequence = sequence;
+            record.structured_activity_watermark =
+                Some(StructuredActivityWatermark { stream, sequence });
         }
         (None, None) => {}
         _ => return,
@@ -3170,6 +3184,73 @@ mod tests {
         ));
     }
 
+    #[tokio::test]
+    async fn replacement_accepts_new_activity_after_a_high_old_watermark() {
+        let (_state, manager, _panes, 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"),
+        )));
+        manager
+            .inner
+            .sessions
+            .lock()
+            .await
+            .get_mut(&structured_id)
+            .unwrap()
+            .focused_since = None;
+        let _ = old_events.lock().await.activity_snapshot();
+        let (old_high_sequence, old_high_activity) = old_events.lock().await.activity_snapshot();
+        apply_structured_activity(
+            &manager.inner,
+            structured_id,
+            Some((&old_events, old_high_sequence)),
+            old_high_activity,
+        )
+        .await;
+        let (delayed_old_sequence, delayed_old_activity) =
+            old_events.lock().await.activity_snapshot();
+
+        {
+            let mut sessions = manager.inner.sessions.lock().await;
+            let record = sessions.get_mut(&structured_id).unwrap();
+            record.acp_events = Some(replacement.clone());
+            record.session.activity = ActivityState::Error;
+        }
+
+        let (new_sequence, new_activity) = replacement.lock().await.activity_snapshot();
+        assert!(new_sequence < old_high_sequence);
+        apply_structured_activity(
+            &manager.inner,
+            structured_id,
+            Some((&replacement, new_sequence)),
+            new_activity,
+        )
+        .await;
+        apply_structured_activity(
+            &manager.inner,
+            structured_id,
+            Some((&old_events, delayed_old_sequence)),
+            delayed_old_activity,
+        )
+        .await;
+
+        let sessions = manager.inner.sessions.lock().await;
+        let record = sessions.get(&structured_id).unwrap();
+        assert_eq!(record.session.activity, ActivityState::Idle);
+        let watermark = record.structured_activity_watermark.as_ref().unwrap();
+        assert_eq!(watermark.sequence, new_sequence);
+        assert!(watermark.stream.ptr_eq(&Arc::downgrade(&replacement)));
+        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