diff --git a/crates/maestro-daemon/src/session.rs b/crates/maestro-daemon/src/session.rs index c86578e..fdadef5 100644 --- a/crates/maestro-daemon/src/session.rs +++ b/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>>, - /// 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, /// Completion signal for the task draining ACP notifications into the /// event log. Shutdown waits for it before returning. acp_event_pump: Option>, @@ -122,6 +122,12 @@ impl SessionRecord { } } +#[derive(Debug)] +struct StructuredActivityWatermark { + stream: Weak>, + 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