Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 22 additions & 3 deletions cmux-tui/crates/acpmux/src/hub/processes.rs
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,9 @@ pub(super) struct ProcessTable {
by_id: std::collections::HashMap<String, usize>,
/// First process with a tool-call ID (the historical Vec::position behavior).
by_tool: std::collections::HashMap<String, usize>,
/// 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<usize>,
/// Processes a person asked to stop: their end is the person's.
stopping: std::collections::HashSet<String>,
}
Expand All @@ -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
}
Expand Down Expand Up @@ -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)
}

Expand Down Expand Up @@ -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;
}
Expand Down Expand Up @@ -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<Process> {
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);
Expand Down Expand Up @@ -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
Expand Down
68 changes: 68 additions & 0 deletions cmux-tui/crates/acpmux/src/hub/processes_active_tests.rs
Original file line number Diff line number Diff line change
@@ -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);
}
Loading