diff --git a/cmux-tui/crates/cmux-tui-core/src/mux.rs b/cmux-tui/crates/cmux-tui-core/src/mux.rs index 0598b6da880a..71c4369f9de6 100644 --- a/cmux-tui/crates/cmux-tui-core/src/mux.rs +++ b/cmux-tui/crates/cmux-tui-core/src/mux.rs @@ -6526,6 +6526,13 @@ impl Mux { source: AgentSource, session: Option<&str>, ) -> anyhow::Result> { + // Resource revisions are durable and strictly monotonic. Include the + // current revision in the transition key so an ABA sequence cannot + // reuse the key from an earlier cycle after the compatibility cache + // returns to the same state. Repeated polls of one semantic state + // still return early through `unchanged`, so steady-state reports do + // not append another echo. + let resource_revision = self.state.lock().unwrap().resource_revision; // The durable projection is updated before its echo is folded. Read // it first so a blocked roster replay still has a stable semantic // receipt for repeated reports. @@ -6619,7 +6626,7 @@ impl Mux { // journal cursor would make an unrelated event defeat coalescing and // turn a steady socket poll into one durable row per event elsewhere. let material = format!( - "agent-report-echo-v4|{terminal_id}|{}|{}|{:?}|{previous}", + "agent-report-echo-v5|{terminal_id}|{}|{}|{:?}|revision={resource_revision}|{previous}", state.as_str(), source.as_str(), session, @@ -24711,6 +24718,65 @@ mod tests { std::fs::remove_dir_all(root).unwrap(); } + #[test] + fn socket_echoes_do_not_reuse_keys_after_repeated_aba_cycles() { + let root = std::env::temp_dir().join(format!( + "cmux-roster-aba-repeat-echo-{}", + crate::workspace_registry::new_uuid_v4() + )); + let mux = Mux::open_persistent("roster-aba-repeat-echo", SurfaceOptions::default(), &root) + .unwrap(); + let surface = mux.new_workspace(None, None).unwrap(); + let terminal_id = surface.terminal_public_id().cloned().expect("workspace terminal"); + let states = [ + AgentState::Working, + AgentState::Blocked, + AgentState::Working, + AgentState::Blocked, + AgentState::Working, + AgentState::Blocked, + ]; + + for state in states { + mux.report_agent( + surface.id, + state, + AgentSource::Socket, + Some("aba-repeat-session".into()), + ) + .unwrap(); + } + + let keys = + mux.workspace_registry + .lock() + .unwrap() + .session_journal_after(0, 1024) + .unwrap() + .records + .into_iter() + .filter(|record| { + record.subjects.iter().any(|subject| { + subject.kind == "terminal" && subject.id == terminal_id.as_str() + }) + }) + .filter_map(|record| { + (record + .payload + .get("adapter") + .and_then(|adapter| adapter.get("id")) + .and_then(Value::as_str) + == Some(crate::journal_reducers::SOCKET_REPORT_ADAPTER)) + .then_some(record.event_id) + }) + .collect::>(); + assert_eq!(keys.len(), states.len(), "each ABA transition needs a unique echo"); + + mux.shutdown(); + drop(mux); + std::fs::remove_dir_all(root).unwrap(); + } + #[test] fn repeated_socket_done_reports_coalesce_after_roster_removal() { let root = std::env::temp_dir()