hanzoai/dev
2
0
Fork 0
mirror of https://github.com/hanzoai/dev.git synced 2026-10-25 03:32:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(perf): reduce long-session stalls (exec + history)

Browse Source
This commit is contained in:
James Peter
2025-12-19 15:07:24 +10:00
parent 76450bcd23
commit 5cfb830971
4 changed files with 196 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
code-rs/core/src/client.rs
+4
View File
@@ -314,6 +314,10 @@ impl ModelClient {
&self.config.code_home
}
pub fn debug_enabled(&self) -> bool {
self.config.debug
}
pub fn build_tools_config_with_sandbox(
&self,
sandbox_policy: SandboxPolicy,
code-rs/core/src/codex.rs
+16 -10
View File
@@ -11514,18 +11514,22 @@ async fn handle_container_exec_with_params(
let sandbox_policy = sess.sandbox_policy.clone();
let sandbox_cwd = sess.get_cwd().to_path_buf();
let code_linux_sandbox_exe = sess.code_linux_sandbox_exe.clone();
let exec_spool_dir = sess
.client
.code_home()
.join("debug_logs")
.join("exec");
let exec_spool_dir_for_task = if sess.client.debug_enabled() {
Some(
sess.client
.code_home()
.join("debug_logs")
.join("exec"),
)
} else {
None
};
let result_cell_for_task = result_cell.clone();
let notify_task = notify.clone();
let tail_buf_task = tail_buf.clone();
let backgrounded_task = backgrounded.clone();
let suppress_event_flag_task = suppress_event_flag.clone();
let display_label_task = display_label.clone();
let exec_spool_dir_for_task = exec_spool_dir.clone();
let task_handle = tokio::spawn(async move {
// Build stdout stream with tail capture. We cannot stamp via `Session` here,
// but deltas will be delivered with neutral ordering which the UI tolerates.
@@ -11539,7 +11543,7 @@ async fn handle_container_exec_with_params(
session: None,
tail_buf: Some(tail_buf_task.clone()),
order: Some(order_meta_for_deltas.clone()),
spool_dir: Some(exec_spool_dir_for_task.clone()),
spool_dir: exec_spool_dir_for_task.clone(),
})
};
@@ -11874,9 +11878,11 @@ async fn handle_sandbox_error(
session: None,
tail_buf: None,
order: Some(crate::protocol::OrderMeta { request_ordinal: attempt_req, output_index: None, sequence_number: None }),
spool_dir: Some(
sess.client.code_home().join("debug_logs").join("exec"),
),
spool_dir: if sess.client.debug_enabled() {
Some(sess.client.code_home().join("debug_logs").join("exec"))
} else {
None
},
})
},
},
code-rs/core/src/history/state.rs
+107 -30
View File
@@ -690,6 +690,10 @@ fn retained_stream_len(chunks: &[ExecStreamChunk]) -> usize {
stream_len(chunks).saturating_sub(first_offset)
}
fn retained_exec_record_bytes(record: &ExecRecord) -> usize {
retained_stream_len(&record.stdout_chunks).saturating_add(retained_stream_len(&record.stderr_chunks))
}
fn truncated_prefix_len(chunks: &[ExecStreamChunk]) -> usize {
chunks.first().map(|chunk| chunk.offset).unwrap_or(0)
}
@@ -1392,6 +1396,10 @@ pub struct HistoryState {
id_index: HashMap<HistoryId, usize>,
#[serde(skip)]
usage_tracker: HistoryUsageTracker,
#[serde(skip)]
exec_stream_retained_total_bytes: usize,
#[serde(skip)]
exec_stream_retained_by_id: HashMap<HistoryId, usize>,
}
#[allow(dead_code)]
@@ -1405,6 +1413,8 @@ impl HistoryState {
stream_lookup: HashMap::new(),
id_index: HashMap::new(),
usage_tracker: HistoryUsageTracker::default(),
exec_stream_retained_total_bytes: 0,
exec_stream_retained_by_id: HashMap::new(),
}
}
@@ -1599,6 +1609,58 @@ impl HistoryState {
for record in &self.records {
self.usage_tracker.on_insert(record);
}
self.rebuild_exec_stream_retained_totals();
}
fn rebuild_exec_stream_retained_totals(&mut self) {
let mut total = 0usize;
let mut by_id: HashMap<HistoryId, usize> = HashMap::new();
for record in &self.records {
let HistoryRecord::Exec(exec) = record else {
continue;
};
if exec.id == HistoryId::ZERO {
continue;
}
let retained = retained_exec_record_bytes(exec);
by_id.insert(exec.id, retained);
total = total.saturating_add(retained);
}
self.exec_stream_retained_total_bytes = total;
self.exec_stream_retained_by_id = by_id;
}
fn set_exec_stream_retained(&mut self, id: HistoryId, retained: usize) {
if id == HistoryId::ZERO {
return;
}
let prev = self.exec_stream_retained_by_id.insert(id, retained).unwrap_or(0);
match retained.cmp(&prev) {
std::cmp::Ordering::Greater => {
self.exec_stream_retained_total_bytes = self
.exec_stream_retained_total_bytes
.saturating_add(retained.saturating_sub(prev));
}
std::cmp::Ordering::Less => {
self.exec_stream_retained_total_bytes = self
.exec_stream_retained_total_bytes
.saturating_sub(prev.saturating_sub(retained));
}
std::cmp::Ordering::Equal => {}
}
}
fn remove_exec_stream_retained(&mut self, id: HistoryId) {
if id == HistoryId::ZERO {
return;
}
if let Some(prev) = self.exec_stream_retained_by_id.remove(&id) {
self.exec_stream_retained_total_bytes = self
.exec_stream_retained_total_bytes
.saturating_sub(prev);
}
}
pub fn truncate_after(&mut self, id: HistoryId) -> Vec<HistoryRecord> {
@@ -1609,6 +1671,8 @@ impl HistoryState {
self.stream_lookup.clear();
self.next_id = 1;
self.usage_tracker.reset();
self.exec_stream_retained_total_bytes = 0;
self.exec_stream_retained_by_id.clear();
return removed;
}
@@ -1662,6 +1726,7 @@ impl HistoryState {
if let Some(call_id) = state.call_id.as_ref() {
self.exec_call_lookup.insert(call_id.clone(), state.id);
}
self.set_exec_stream_retained(state.id, retained_exec_record_bytes(state));
}
HistoryRecord::MergedExec(state) => {
for segment in &state.segments {
@@ -1701,6 +1766,7 @@ impl HistoryState {
self.exec_call_lookup.remove(call_id);
}
}
self.remove_exec_stream_retained(state.id);
}
HistoryRecord::MergedExec(state) => {
for segment in &state.segments {
@@ -1850,7 +1916,7 @@ impl HistoryState {
stdout_chunk,
stderr_chunk,
} => {
let id = match self.records.get_mut(index) {
let (id, retained) = match self.records.get_mut(index) {
Some(HistoryRecord::Exec(existing)) => {
if let Some(chunk) = stdout_chunk {
let chunk_len = chunk.content.len();
@@ -1864,11 +1930,14 @@ impl HistoryState {
}
self.usage_tracker
.observe_exec(existing, "domain:update-exec-stream");
existing.id
let retained = retained_exec_record_bytes(existing);
(existing.id, retained)
}
_ => return HistoryMutation::Noop,
};
self.set_exec_stream_retained(id, retained);
self.enforce_exec_stream_global_limit();
if let Some(record) = self.records.get(index).cloned() {
@@ -2115,44 +2184,52 @@ impl HistoryState {
/// `GLOBAL_EXEC_STREAM_RETAINED_BYTES`, progressively prune the oldest exec
/// records down to `EXEC_STREAM_PRUNE_TARGET_BYTES` per stream.
fn enforce_exec_stream_global_limit(&mut self) {
let mut total_retained: usize = 0;
let mut exec_indices: Vec<usize> = Vec::new();
for (idx, record) in self.records.iter().enumerate() {
if let HistoryRecord::Exec(exec) = record {
let retained = retained_stream_len(&exec.stdout_chunks)
.saturating_add(retained_stream_len(&exec.stderr_chunks));
if retained > 0 {
total_retained = total_retained.saturating_add(retained);
exec_indices.push(idx);
}
}
}
if total_retained <= GLOBAL_EXEC_STREAM_RETAINED_BYTES {
if self.exec_stream_retained_total_bytes <= GLOBAL_EXEC_STREAM_RETAINED_BYTES {
return;
}
let mut bytes_to_drop = total_retained - GLOBAL_EXEC_STREAM_RETAINED_BYTES;
let mut bytes_to_drop =
self.exec_stream_retained_total_bytes - GLOBAL_EXEC_STREAM_RETAINED_BYTES;
for idx in exec_indices {
let retained_by_id = &mut self.exec_stream_retained_by_id;
let total_retained = &mut self.exec_stream_retained_total_bytes;
for record in &mut self.records {
if bytes_to_drop == 0 {
break;
}
if let Some(HistoryRecord::Exec(exec)) = self.records.get_mut(idx) {
let before = retained_stream_len(&exec.stdout_chunks)
.saturating_add(retained_stream_len(&exec.stderr_chunks));
let HistoryRecord::Exec(exec) = record else {
continue;
};
prune_exec_stream(&mut exec.stdout_chunks, EXEC_STREAM_PRUNE_TARGET_BYTES);
prune_exec_stream(&mut exec.stderr_chunks, EXEC_STREAM_PRUNE_TARGET_BYTES);
let after = retained_stream_len(&exec.stdout_chunks)
.saturating_add(retained_stream_len(&exec.stderr_chunks));
let dropped = before.saturating_sub(after);
bytes_to_drop = bytes_to_drop.saturating_sub(dropped);
let before = retained_by_id
.get(&exec.id)
.copied()
.unwrap_or_else(|| retained_exec_record_bytes(exec));
if before == 0 {
continue;
}
prune_exec_stream(&mut exec.stdout_chunks, EXEC_STREAM_PRUNE_TARGET_BYTES);
prune_exec_stream(&mut exec.stderr_chunks, EXEC_STREAM_PRUNE_TARGET_BYTES);
let after = retained_exec_record_bytes(exec);
let prev = retained_by_id.insert(exec.id, after).unwrap_or(0);
match after.cmp(&prev) {
std::cmp::Ordering::Greater => {
*total_retained = (*total_retained)
.saturating_add(after.saturating_sub(prev));
}
std::cmp::Ordering::Less => {
*total_retained = (*total_retained)
.saturating_sub(prev.saturating_sub(after));
}
std::cmp::Ordering::Equal => {}
}
let dropped = before.saturating_sub(after);
bytes_to_drop = bytes_to_drop.saturating_sub(dropped);
}
}
}
code-rs/tui/src/chatwidget.rs
+69 -21
View File
@@ -8721,13 +8721,22 @@ impl ChatWidget<'_> {
}
}
// Determine insertion position across the entire history
// Determine insertion position across the entire history.
// Most ordered inserts are monotonic tail-appends (we bump non-background
// keys to keep them strictly increasing), so avoid an O(n) scan in the
// common case.
//
// Exception: some early, non-background system cells (e.g. the context
// summary) are inserted with a low order key before any ordering state
// has been established. In that phase, we must still respect the order.
let mut pos = self.history_cells.len();
for i in 0..self.history_cells.len() {
if let Some(existing) = self.cell_order_seq.get(i) {
if *existing > key {
pos = i;
break;
if is_background_cell || self.last_assigned_order.is_none() {
for i in 0..self.history_cells.len() {
if let Some(existing) = self.cell_order_seq.get(i) {
if *existing > key {
pos = i;
break;
}
}
}
}
@@ -8809,8 +8818,12 @@ impl ChatWidget<'_> {
let mut cell = cell;
let record_index = self.record_index_for_position(pos);
let mutation = if let Some(domain_record) = record {
let record_index = if pos == self.history_cells.len() {
self.history_state.records.len()
} else {
self.record_index_for_position(pos)
};
let event = match domain_record {
HistoryDomainRecord::Exec(ref exec_record) => {
HistoryDomainEvent::StartExec {
@@ -8832,6 +8845,11 @@ impl ChatWidget<'_> {
};
Some(self.history_state.apply_domain_event(event))
} else if let Some(record) = history_cell::record_from_cell(cell.as_ref()) {
let record_index = if pos == self.history_cells.len() {
self.history_state.records.len()
} else {
self.record_index_for_position(pos)
};
let event = match HistoryDomainRecord::from(record) {
HistoryDomainRecord::Exec(exec_record) => HistoryDomainEvent::StartExec {
index: record_index,
@@ -8861,8 +8879,14 @@ impl ChatWidget<'_> {
}
}
self.history_cells.insert(pos, cell);
self.history_cell_ids.insert(pos, maybe_id);
let append = pos == self.history_cells.len();
if append {
self.history_cells.push(cell);
self.history_cell_ids.push(maybe_id);
} else {
self.history_cells.insert(pos, cell);
self.history_cell_ids.insert(pos, maybe_id);
}
// In terminal mode, App mirrors history lines into the native buffer.
// Ensure order vector is also long enough for position after cell insert
if self.cell_order_seq.len() < pos {
@@ -8875,7 +8899,11 @@ impl ChatWidget<'_> {
},
);
}
self.cell_order_seq.insert(pos, key);
if append {
self.cell_order_seq.push(key);
} else {
self.cell_order_seq.insert(pos, key);
}
if key_bumped {
if let Some(stream) = self.history_cells[pos]
.as_any()
@@ -8916,7 +8944,11 @@ impl ChatWidget<'_> {
if self.cell_order_dbg.len() < pos {
self.cell_order_dbg.resize(pos, None);
}
self.cell_order_dbg.insert(pos, Some(dbg));
if append {
self.cell_order_dbg.push(Some(dbg));
} else {
self.cell_order_dbg.insert(pos, Some(dbg));
}
self.invalidate_height_cache();
self.autoscroll_if_near_bottom();
self.bottom_pane.set_has_chat_history(true);
@@ -8971,11 +9003,13 @@ impl ChatWidget<'_> {
}
let mut pos = self.history_cells.len();
for i in 0..self.history_cells.len() {
if let Some(existing) = self.cell_order_seq.get(i) {
if *existing > key {
pos = i;
break;
if is_background_cell || self.last_assigned_order.is_none() {
for i in 0..self.history_cells.len() {
if let Some(existing) = self.cell_order_seq.get(i) {
if *existing > key {
pos = i;
break;
}
}
}
}
@@ -9045,8 +9079,14 @@ impl ChatWidget<'_> {
Self::assign_history_id_inner(&mut cell, id);
self.history_cells.insert(pos, cell);
self.history_cell_ids.insert(pos, Some(id));
let append = pos == self.history_cells.len();
if append {
self.history_cells.push(cell);
self.history_cell_ids.push(Some(id));
} else {
self.history_cells.insert(pos, cell);
self.history_cell_ids.insert(pos, Some(id));
}
if self.cell_order_seq.len() < pos {
self.cell_order_seq.resize(
pos,
@@ -9057,7 +9097,11 @@ impl ChatWidget<'_> {
},
);
}
self.cell_order_seq.insert(pos, key);
if append {
self.cell_order_seq.push(key);
} else {
self.cell_order_seq.insert(pos, key);
}
if key_bumped {
if let Some(stream) = self.history_cells[pos]
.as_any()
@@ -9097,7 +9141,11 @@ impl ChatWidget<'_> {
if self.cell_order_dbg.len() < pos {
self.cell_order_dbg.resize(pos, None);
}
self.cell_order_dbg.insert(pos, Some(dbg));
if append {
self.cell_order_dbg.push(Some(dbg));
} else {
self.cell_order_dbg.insert(pos, Some(dbg));
}
self.invalidate_height_cache();
self.autoscroll_if_near_bottom();
self.bottom_pane.set_has_chat_history(true);
@@ -11147,7 +11195,7 @@ impl ChatWidget<'_> {
fn mark_history_dirty(&mut self) {
self.history_snapshot_dirty = true;
self.flush_history_snapshot_if_needed(true);
self.flush_history_snapshot_if_needed(false);
}
fn flush_history_snapshot_if_needed(&mut self, force: bool) {