From e2611dd3c051e56bf0cba95159ae5719e0fe2006 Mon Sep 17 00:00:00 2001 From: Leo Date: Sun, 1 Nov 2026 02:14:29 -0800 Subject: [PATCH 1/2] perf(acpmux): index running process slots for O(1) idle census (SW-11) Agent-Session: https://cmux.com/deeplink/agent?id=aid-3163b03c701021f7e3d1 --- cmux-tui/crates/acpmux/src/hub/processes.rs | 25 ++++++++++++++++++--- 1 file changed, 22 insertions(+), 3 deletions(-) diff --git a/cmux-tui/crates/acpmux/src/hub/processes.rs b/cmux-tui/crates/acpmux/src/hub/processes.rs index 1527db20f311..c05943ec6dfe 100644 --- a/cmux-tui/crates/acpmux/src/hub/processes.rs +++ b/cmux-tui/crates/acpmux/src/hub/processes.rs @@ -101,6 +101,9 @@ pub(super) struct ProcessTable { by_id: std::collections::HashMap, /// First process with a tool-call ID (the historical Vec::position behavior). by_tool: std::collections::HashMap, + /// Only live process slots, ascending by append order. Completed history can grow + /// without making idle checks or lost-process propagation scan every old task. + running: std::collections::BTreeSet, /// Processes a person asked to stop: their end is the person's. stopping: std::collections::HashSet, } @@ -109,12 +112,19 @@ pub(super) struct ProcessTable { #[path = "processes_index_tests.rs"] mod process_index_tests; +#[cfg(test)] +#[path = "processes_active_tests.rs"] +mod process_active_tests; + impl ProcessTable { /// All appends go here, including records replayed from a logged process update. fn insert(&mut self, process: Process) -> usize { let at = self.processes.len(); self.by_id.insert(process.id.clone(), at); self.by_tool.entry(process.tool_call_id.clone()).or_insert(at); + if process.state == ProcessState::Running { + self.running.insert(at); + } self.processes.push(process); at } @@ -214,6 +224,7 @@ impl ProcessTable { p.exit_code = update.get("exitCode").and_then(Value::as_i64); } p.logged = false; + self.running.remove(&i); Some(i) } @@ -242,7 +253,13 @@ impl ProcessTable { && let Some(slot) = self.processes.get_mut(at) { let new_tool = process.tool_call_id.clone(); + let live = process.state == ProcessState::Running; let old_tool = std::mem::replace(slot, process).tool_call_id; + if live { + self.running.insert(at); + } else { + self.running.remove(&at); + } if old_tool == new_tool { return; } @@ -284,7 +301,9 @@ impl ProcessTable { /// `logged`: their `process_update` is written now. fn lose_running(&mut self, at: u64, reason: &str, logged: bool) -> Vec { let mut lost = Vec::new(); - for p in self.processes.iter_mut().filter(|p| p.state == ProcessState::Running) { + // The ordered slots retain the original oldest-first notification order. + for index in std::mem::take(&mut self.running) { + let p = &mut self.processes[index]; self.stopping.remove(&p.id); p.state = ProcessState::Lost; p.ended_at = Some(at); @@ -471,14 +490,14 @@ impl Hub { /// (`idle.rs`). pub(super) fn any_process_running(&self, session: &Session) -> bool { let table = self.process_table(session); - table.processes.iter().any(|p| p.state == ProcessState::Running) + !table.running.is_empty() } /// How many of the session's background processes run, from a table this /// daemon has read already: never a log scan (`host_load.rs` asks for every chat). pub(super) fn running_processes_read(&self, session: &Session) -> usize { let table = session.processes.lock().unwrap_or_else(|e| e.into_inner()); - table.processes.iter().filter(|p| p.state == ProcessState::Running).count() + table.running.len() } /// `_acpmux/process/list`: the session's processes; without one, those of From cd0e3bbf4446ee79b72042121c016c1c3e2bb147 Mon Sep 17 00:00:00 2001 From: Leo Date: Sun, 1 Nov 2026 02:14:32 -0800 Subject: [PATCH 2/2] test(acpmux): keep ordered live process slots over 3000 completed records Agent-Session: https://cmux.com/deeplink/agent?id=aid-3163b03c701021f7e3d1 --- .../acpmux/src/hub/processes_active_tests.rs | 68 +++++++++++++++++++ 1 file changed, 68 insertions(+) create mode 100644 cmux-tui/crates/acpmux/src/hub/processes_active_tests.rs diff --git a/cmux-tui/crates/acpmux/src/hub/processes_active_tests.rs b/cmux-tui/crates/acpmux/src/hub/processes_active_tests.rs new file mode 100644 index 000000000000..f6063846a7a5 --- /dev/null +++ b/cmux-tui/crates/acpmux/src/hub/processes_active_tests.rs @@ -0,0 +1,68 @@ +//! The active-process set tracks logged replay and live transitions without scanning +//! historical finished work. It preserves the event order when a host ends. +use super::*; + +fn process(index: usize, state: ProcessState) -> Process { + serde_json::from_value(json!({ + "id": format!("session/task-{index}"), + "sessionId": "session", + "toolCallId": format!("tool-{index}"), + "taskId": format!("task-{index}"), + "harness": "claude", + "command": "true", + "cwd": "/tmp", + "startedAt": 100, + "state": state, + "origin": "background", + })) + .unwrap() +} + +#[test] +fn the_live_set_follows_replay_updates_and_preserves_lost_notification_order() { + let mut table = ProcessTable::default(); + for index in 0..3_000 { + table.upsert(process(index, ProcessState::Exited)); + } + assert_eq!(table.processes.len(), 3_000); + assert!(table.running.is_empty()); + + // Late replay of an existing process turns it on, then off, without an extra row. + for index in [2_999, 17, 1_500] { + table.upsert(process(index, ProcessState::Running)); + } + assert_eq!(table.running.len(), 3); + table.upsert(process(17, ProcessState::Exited)); + assert_eq!(table.running.len(), 2); + table.upsert(process(17, ProcessState::Running)); + assert_eq!(table.running.len(), 3); + + // A host exit logs active tasks in their original append order, even when + // status updates arrived in a different order. + let lost = table.lose_running(200, HARNESS_EXIT, true); + let ids: Vec<_> = lost.iter().map(|p| p.id.as_str()).collect(); + assert_eq!(ids, ["session/task-17", "session/task-1500", "session/task-2999"]); + assert!(lost.iter().all(|p| { + p.state == ProcessState::Lost + && p.ended_at == Some(200) + && p.reason.as_deref() == Some(HARNESS_EXIT) + && p.logged + })); + assert_eq!(table.processes.len(), 3_000); + assert!(table.running.is_empty()); + assert!(table.lose_running(201, HARNESS_EXIT, true).is_empty()); +} + +#[test] +fn a_restored_running_task_is_removed_from_the_live_set_when_completed() { + let mut table = ProcessTable::default(); + table.upsert(process(0, ProcessState::Running)); + assert_eq!(table.running.len(), 1); + table.upsert(process(0, ProcessState::Failed)); + assert!(table.running.is_empty()); + table.upsert(process(1, ProcessState::Running)); + assert_eq!(table.running.len(), 1); + table.upsert(process(1, ProcessState::Killed)); + assert!(table.running.is_empty()); + assert_eq!(table.processes.len(), 2); +}