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(auto-drive): decouple auto review and cap long-session growth

Browse Source
This commit is contained in:
James Peter
2026-02-15 07:02:04 +10:00
parent 0daba2bac0
commit fc6a2e35cb
7 changed files with 1465 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files
code-rs/code-auto-drive-core/src/auto_coordinator.rs
+150 -10
View File
@@ -2,6 +2,7 @@ use std::collections::VecDeque;
use std::process::Command;
use std::sync::mpsc::{self, Receiver, Sender};
use std::sync::{Arc, Mutex};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant, SystemTime};
use anyhow::{anyhow, Context, Result};
@@ -51,12 +52,17 @@ const RATE_LIMIT_JITTER_MAX: Duration = Duration::from_secs(3);
const MAX_RETRY_ELAPSED: Duration = Duration::from_secs(7 * 24 * 60 * 60);
const MAX_DECISION_RECOVERY_ATTEMPTS: u32 = 3;
const MESSAGE_LIMIT_FALLBACK: usize = 120;
const HARD_MESSAGE_LIMIT: usize = 320;
const MAX_QUEUED_CONVERSATION_UPDATES: usize = 24;
const DEBUG_JSON_MAX_CHARS: usize = 1200;
const CLI_PROMPT_MIN_CHARS: usize = 4;
const CLI_PROMPT_MAX_CHARS: usize = 600;
const AUTO_DRIVE_CLI_MODELS: &[&str] = &["gpt-5.3-codex", "gpt-5.3-codex-spark"];
const AUTO_DRIVE_CLI_REASONING_LEVELS: &[&str] = &["medium", "high", "xhigh"];
static HARD_LIMIT_TRIMMED_ITEMS_TOTAL: AtomicU64 = AtomicU64::new(0);
static QUEUED_UPDATE_DROPS_TOTAL: AtomicU64 = AtomicU64::new(0);
#[derive(Debug, thiserror::Error)]
#[error("auto coordinator cancelled")]
struct AutoCoordinatorCancelled;
@@ -581,6 +587,72 @@ mod tests {
);
}
#[test]
fn enforce_hard_message_limit_keeps_goal_and_recent_tail() {
let total_messages = HARD_MESSAGE_LIMIT + 48;
let mut items = Vec::with_capacity(total_messages + 1);
items.push(make_message("user", "Primary goal should survive".to_string()));
for idx in 0..total_messages {
items.push(make_message("assistant", format!("assistant-msg-{idx}")));
}
let trimmed = enforce_hard_message_limit(items);
assert_eq!(trimmed.len(), HARD_MESSAGE_LIMIT);
let first_is_goal = matches!(
trimmed.first(),
Some(ResponseItem::Message { role, content, .. })
if role == "user"
&& content
.iter()
.any(|item| matches!(item, ContentItem::InputText { text } if text == "Primary goal should survive"))
);
assert!(first_is_goal, "oldest user goal should be preserved");
let latest_tail = format!("assistant-msg-{}", total_messages - 1);
let has_latest = trimmed.iter().any(|item| {
matches!(
item,
ResponseItem::Message { content, .. }
if content
.iter()
.any(|entry| matches!(entry, ContentItem::OutputText { text } if text == &latest_tail))
)
});
assert!(has_latest, "latest assistant message should be retained");
}
#[test]
fn queue_update_capped_discards_oldest_overflow() {
let mut queue: VecDeque<Arc<[ResponseItem]>> = VecDeque::new();
for idx in 0..(MAX_QUEUED_CONVERSATION_UPDATES + 7) {
let update = Arc::<[ResponseItem]>::from(vec![make_message(
"assistant",
format!("queued-{idx}"),
)]);
queue_update_capped(&mut queue, update);
}
assert_eq!(queue.len(), MAX_QUEUED_CONVERSATION_UPDATES);
let first_text = queue
.front()
.and_then(|conversation| conversation.first())
.and_then(|item| match item {
ResponseItem::Message { content, .. } => content.first(),
_ => None,
})
.and_then(|content| match content {
ContentItem::OutputText { text } => Some(text.clone()),
ContentItem::InputText { text } => Some(text.clone()),
_ => None,
})
.expect("queued message text");
assert_eq!(first_text, "queued-7");
}
#[test]
fn retry_max_elapsed_defaults_to_global_limit() {
assert_eq!(retry_max_elapsed(None), MAX_RETRY_ELAPSED);
@@ -1453,8 +1525,9 @@ fn run_auto_loop(
schema_features.include_goal_field = true;
}
let include_agents = schema_features.include_agents;
let mut pending_conversation =
Some(Arc::<[ResponseItem]>::from(filter_popular_commands(initial_conversation)));
let mut pending_conversation = Some(Arc::<[ResponseItem]>::from(
enforce_hard_message_limit(filter_popular_commands(initial_conversation)),
));
let mut decision_seq: u64 = 0;
let mut pending_ack_seq: Option<u64> = None;
let mut queued_updates: VecDeque<Arc<[ResponseItem]>> = VecDeque::new();
@@ -1489,7 +1562,6 @@ fn run_auto_loop(
debug!("[Auto coordinator] starting: goal={goal_text} platform={platform}");
let mut stopped = false;
let mut requests_completed: u64 = 0;
let mut consecutive_decision_failures: u32 = 0;
let mut session_metrics = SessionMetrics::default();
let mut coordinator_turns_seen: u32 = 0;
@@ -1506,7 +1578,7 @@ fn run_auto_loop(
if let Some(conv) = pending_conversation.take() {
if let Some(pending_seq) = pending_ack_seq {
tracing::debug!(target: "auto_drive::coordinator", pending_seq, "queueing conversation until ack");
queued_updates.push_back(conv);
queue_update_capped(&mut queued_updates, conv);
} else {
next_conversation = Some(conv);
}
@@ -1534,6 +1606,7 @@ fn run_auto_loop(
&active_model_slug,
&compact_prompt_text,
);
conv = enforce_hard_message_limit(conv);
let conv = Arc::<[ResponseItem]>::from(conv);
if matches!(compaction_result, CompactionResult::Completed { .. }) {
event_tx.send(AutoCoordinatorEvent::CompactedHistory {
@@ -1764,7 +1837,11 @@ fn run_auto_loop(
}
let retry_snapshot = retry_conversation
.take()
.map(Arc::<[ResponseItem]>::from)
.map(|conversation| {
Arc::<[ResponseItem]>::from(enforce_hard_message_limit(
filter_popular_commands(conversation),
))
})
.unwrap_or_else(|| Arc::clone(&conv));
// Keep the model and UI in sync with the full conversation, but avoid spamming a compaction notice.
let _ = event_tx.send(AutoCoordinatorEvent::CompactedHistory {
@@ -1845,7 +1922,10 @@ fn run_auto_loop(
if let Some(response_text) = user_response.clone() {
updated_conversation.push(make_message("assistant", response_text.clone()));
}
pending_conversation = Some(Arc::<[ResponseItem]>::from(updated_conversation));
let updated_conversation =
enforce_hard_message_limit(filter_popular_commands(updated_conversation));
pending_conversation =
Some(Arc::<[ResponseItem]>::from(updated_conversation));
event_tx.send(AutoCoordinatorEvent::UserReply {
user_response,
cli_command,
@@ -1877,17 +1957,18 @@ fn run_auto_loop(
}
}
Ok(AutoCoordinatorCommand::UpdateConversation(conv)) => {
requests_completed = requests_completed.saturating_add(1);
consecutive_decision_failures = 0;
let conv = conv.as_ref().to_vec();
let filtered = Arc::<[ResponseItem]>::from(filter_popular_commands(conv));
let filtered = Arc::<[ResponseItem]>::from(enforce_hard_message_limit(
filter_popular_commands(conv),
));
if let Some(pending_seq) = pending_ack_seq {
tracing::debug!(target: "auto_drive::coordinator", pending_seq, "queueing update while awaiting ack");
session_metrics.record_replay();
queued_updates.push_back(filtered);
queue_update_capped(&mut queued_updates, filtered);
} else if pending_conversation.is_some() {
session_metrics.record_replay();
queued_updates.push_back(filtered);
queue_update_capped(&mut queued_updates, filtered);
} else {
pending_conversation = Some(filtered);
}
@@ -1911,6 +1992,65 @@ fn filter_popular_commands(items: Vec<ResponseItem>) -> Vec<ResponseItem> {
.collect()
}
fn enforce_hard_message_limit(items: Vec<ResponseItem>) -> Vec<ResponseItem> {
if items.len() <= HARD_MESSAGE_LIMIT {
return items;
}
let original_len = items.len();
let mut trimmed: Vec<ResponseItem> = Vec::with_capacity(HARD_MESSAGE_LIMIT);
let goal_index = items.iter().position(
|item| matches!(item, ResponseItem::Message { role, .. } if role == "user"),
);
if let Some(idx) = goal_index {
trimmed.push(items[idx].clone());
}
let remaining = HARD_MESSAGE_LIMIT.saturating_sub(trimmed.len());
let tail_start = items.len().saturating_sub(remaining);
for (idx, item) in items.into_iter().enumerate().skip(tail_start) {
if goal_index == Some(idx) {
continue;
}
trimmed.push(item);
}
if trimmed.len() > HARD_MESSAGE_LIMIT {
let drop = trimmed.len() - HARD_MESSAGE_LIMIT;
trimmed.drain(0..drop);
}
let dropped = original_len.saturating_sub(trimmed.len());
if dropped > 0 {
let total = HARD_LIMIT_TRIMMED_ITEMS_TOTAL.fetch_add(dropped as u64, Ordering::Relaxed)
+ dropped as u64;
debug!(
dropped,
original_len,
retained = trimmed.len(),
total_trimmed = total,
"auto coordinator trimmed conversation to hard message limit"
);
}
trimmed
}
fn queue_update_capped(queue: &mut VecDeque<Arc<[ResponseItem]>>, update: Arc<[ResponseItem]>) {
if queue.len() >= MAX_QUEUED_CONVERSATION_UPDATES {
queue.pop_front();
let total = QUEUED_UPDATE_DROPS_TOTAL.fetch_add(1, Ordering::Relaxed) + 1;
warn!(
cap = MAX_QUEUED_CONVERSATION_UPDATES,
total_dropped = total,
"auto coordinator dropped oldest queued conversation update"
);
}
queue.push_back(update);
}
fn is_popular_commands_message(item: &ResponseItem) -> bool {
match item {
ResponseItem::Message { role, content, .. } if role.eq_ignore_ascii_case("user") => {
code-rs/core/src/agent_tool.rs
+553 -32
View File
@@ -8,6 +8,7 @@ use std::fs::{self, OpenOptions};
use std::io::Write as IoWrite;
use std::collections::BTreeMap;
use std::collections::HashMap;
use std::collections::HashSet;
use std::path::PathBuf;
use std::process::Stdio;
use tokio::process::Command;
@@ -23,7 +24,7 @@ use std::thread;
use std::time::{Duration as StdDuration, Instant};
use crate::spawn::spawn_tokio_command_with_retry;
use crate::protocol::AgentSourceKind;
use tracing::warn;
use tracing::{debug, info, warn};
#[cfg(target_os = "windows")]
fn default_pathext_or_default() -> Vec<String> {
@@ -398,13 +399,49 @@ lazy_static::lazy_static! {
pub struct AgentManager {
agents: HashMap<String, Agent>,
// Session-scoped archive so pruned terminal agents remain queryable
// (including worktree/branch metadata) for the full Auto Drive run.
archived_terminal_agents: HashMap<String, Agent>,
handles: HashMap<String, JoinHandle<()>>,
event_sender: Option<mpsc::UnboundedSender<AgentStatusUpdatePayload>>,
debug_log_root: Option<PathBuf>,
watchdog_handle: Option<JoinHandle<()>>,
inactivity_timeout: Duration,
diagnostics: AgentManagerDiagnostics,
}
#[derive(Debug, Clone, Default)]
struct AgentManagerDiagnostics {
terminal_compactions: u64,
progress_entries_trimmed: u64,
progress_lines_truncated: u64,
payloads_truncated: u64,
terminal_agents_pruned: u64,
archived_terminal_agents: u64,
status_terminal_agents_omitted: u64,
}
#[derive(Debug, Clone, Copy, Default)]
struct AgentCompactionDelta {
progress_entries_trimmed: usize,
progress_lines_truncated: usize,
payloads_truncated: usize,
}
impl AgentCompactionDelta {
fn any(self) -> bool {
self.progress_entries_trimmed > 0
|| self.progress_lines_truncated > 0
|| self.payloads_truncated > 0
}
}
const MAX_AGENT_PROGRESS_ENTRIES: usize = 96;
const MAX_AGENT_PROGRESS_LINE_BYTES: usize = 2048;
const MAX_AGENT_RESULT_BYTES: usize = 64 * 1024;
const MAX_TRACKED_TERMINAL_AGENTS: usize = 512;
const MAX_STATUS_TERMINAL_AGENTS: usize = 128;
#[derive(Debug, Clone)]
pub struct AgentStatusUpdatePayload {
pub agents: Vec<AgentInfo>,
@@ -416,16 +453,22 @@ impl AgentManager {
pub fn new() -> Self {
Self {
agents: HashMap::new(),
archived_terminal_agents: HashMap::new(),
handles: HashMap::new(),
event_sender: None,
debug_log_root: None,
watchdog_handle: None,
inactivity_timeout: Duration::minutes(30),
diagnostics: AgentManagerDiagnostics::default(),
}
}
pub fn set_event_sender(&mut self, sender: mpsc::UnboundedSender<AgentStatusUpdatePayload>) {
self.event_sender = Some(sender);
// New session lifecycle: keep only live agents and reset per-session
// diagnostics/archives so long-lived UIs start from a clean slate.
self.archived_terminal_agents.clear();
self.diagnostics = AgentManagerDiagnostics::default();
self.start_watchdog();
}
@@ -470,6 +513,7 @@ impl AgentManager {
agent.completed_at = Some(now);
Self::record_activity(agent);
}
mgr.finalize_terminal_agent(agent_id);
}
// Notify listeners once per sweep.
@@ -495,6 +539,207 @@ impl AgentManager {
agent.last_activity = Utc::now();
}
fn trim_to_tail_utf8(text: &str, max_bytes: usize) -> String {
let bytes = text.as_bytes();
if bytes.len() <= max_bytes {
return text.to_string();
}
let mut start = bytes.len().saturating_sub(max_bytes);
while start < bytes.len() && (bytes[start] & 0b1100_0000) == 0b1000_0000 {
start += 1;
}
let tail = String::from_utf8_lossy(&bytes[start..]).to_string();
format!("…{tail}")
}
fn compact_terminal_agent(agent: &mut Agent) -> AgentCompactionDelta {
let mut delta = AgentCompactionDelta::default();
if agent.progress.len() > MAX_AGENT_PROGRESS_ENTRIES {
let drain = agent.progress.len() - MAX_AGENT_PROGRESS_ENTRIES;
agent.progress.drain(0..drain);
delta.progress_entries_trimmed = drain;
}
for line in &mut agent.progress {
let original_len = line.len();
let trimmed = Self::trim_to_tail_utf8(line, MAX_AGENT_PROGRESS_LINE_BYTES);
if trimmed.len() < original_len {
delta.progress_lines_truncated = delta.progress_lines_truncated.saturating_add(1);
}
*line = trimmed;
}
if let Some(result) = agent.result.as_mut() {
let original_len = result.len();
let trimmed = Self::trim_to_tail_utf8(result, MAX_AGENT_RESULT_BYTES);
if trimmed.len() < original_len {
delta.payloads_truncated = delta.payloads_truncated.saturating_add(1);
}
*result = trimmed;
}
if let Some(error) = agent.error.as_mut() {
let original_len = error.len();
let trimmed = Self::trim_to_tail_utf8(error, MAX_AGENT_RESULT_BYTES);
if trimmed.len() < original_len {
delta.payloads_truncated = delta.payloads_truncated.saturating_add(1);
}
*error = trimmed;
}
// Keep branch/worktree metadata so users can still merge/inspect results,
// but release heavy prompt payloads once an agent is terminal.
agent.prompt.clear();
agent.context = None;
agent.output_goal = None;
agent.files.clear();
delta
}
fn prune_terminal_agents(&mut self) {
let terminal_count = self
.agents
.values()
.filter(|agent| {
matches!(
agent.status,
AgentStatus::Completed | AgentStatus::Failed | AgentStatus::Cancelled
)
})
.count();
if terminal_count <= MAX_TRACKED_TERMINAL_AGENTS {
return;
}
let mut terminal: Vec<(DateTime<Utc>, String)> = self
.agents
.iter()
.filter(|(_, agent)| {
matches!(
agent.status,
AgentStatus::Completed | AgentStatus::Failed | AgentStatus::Cancelled
)
})
.map(|(id, agent)| {
(
agent.completed_at.unwrap_or(agent.created_at),
id.clone(),
)
})
.collect();
terminal.sort_by_key(|(completed_at, _)| *completed_at);
let mut to_remove = terminal_count.saturating_sub(MAX_TRACKED_TERMINAL_AGENTS);
let mut pruned = 0usize;
for (_, agent_id) in terminal {
if to_remove == 0 {
break;
}
self.handles.remove(&agent_id);
if let Some(agent) = self.agents.remove(&agent_id) {
self.archived_terminal_agents.insert(agent_id.clone(), agent);
}
pruned = pruned.saturating_add(1);
to_remove = to_remove.saturating_sub(1);
}
if pruned > 0 {
self.diagnostics.terminal_agents_pruned = self
.diagnostics
.terminal_agents_pruned
.saturating_add(pruned as u64);
self.diagnostics.archived_terminal_agents = self.archived_terminal_agents.len() as u64;
info!(
pruned,
retained = self.agents.len(),
archived = self.archived_terminal_agents.len(),
"agent manager pruned terminal agents from live cache"
);
}
}
fn finalize_terminal_agent(&mut self, agent_id: &str) {
self.handles.remove(agent_id);
if let Some(agent) = self.agents.get_mut(agent_id) {
let delta = Self::compact_terminal_agent(agent);
self.diagnostics.terminal_compactions =
self.diagnostics.terminal_compactions.saturating_add(1);
self.diagnostics.progress_entries_trimmed = self
.diagnostics
.progress_entries_trimmed
.saturating_add(delta.progress_entries_trimmed as u64);
self.diagnostics.progress_lines_truncated = self
.diagnostics
.progress_lines_truncated
.saturating_add(delta.progress_lines_truncated as u64);
self.diagnostics.payloads_truncated = self
.diagnostics
.payloads_truncated
.saturating_add(delta.payloads_truncated as u64);
if delta.any() {
debug!(
agent_id,
progress_entries_trimmed = delta.progress_entries_trimmed,
progress_lines_truncated = delta.progress_lines_truncated,
payloads_truncated = delta.payloads_truncated,
total_terminal_compactions = self.diagnostics.terminal_compactions,
total_progress_entries_trimmed = self.diagnostics.progress_entries_trimmed,
total_progress_lines_truncated = self.diagnostics.progress_lines_truncated,
total_payloads_truncated = self.diagnostics.payloads_truncated,
"compacted terminal agent state"
);
}
}
self.prune_terminal_agents();
}
fn visible_agents_for_status(&self) -> Vec<&Agent> {
let mut active: Vec<&Agent> = self
.agents
.values()
.filter(|agent| matches!(agent.status, AgentStatus::Pending | AgentStatus::Running))
.collect();
active.sort_by_key(|agent| agent.created_at);
let mut terminal: Vec<&Agent> = self
.agents
.values()
.filter(|agent| {
matches!(
agent.status,
AgentStatus::Completed | AgentStatus::Failed | AgentStatus::Cancelled
)
})
.collect();
terminal.sort_by_key(|agent| agent.completed_at.unwrap_or(agent.created_at));
if terminal.len() > MAX_STATUS_TERMINAL_AGENTS {
let keep_from = terminal.len() - MAX_STATUS_TERMINAL_AGENTS;
terminal = terminal.split_off(keep_from);
}
active.extend(terminal);
active
}
pub fn status_visible_agents(&self) -> Vec<Agent> {
self.visible_agents_for_status()
.into_iter()
.cloned()
.collect()
}
fn append_agent_log(&self, log_tag: &str, line: &str) {
let Some(root) = &self.debug_log_root else { return; };
let dir = root.join(log_tag);
@@ -514,12 +759,44 @@ impl AgentManager {
}
}
async fn send_agent_status_update(&self) {
async fn send_agent_status_update(&mut self) {
if let Some(ref sender) = self.event_sender {
let now = Utc::now();
let agents: Vec<AgentInfo> = self
let total_terminal = self
.agents
.values()
.filter(|agent| {
matches!(
agent.status,
AgentStatus::Completed | AgentStatus::Failed | AgentStatus::Cancelled
)
})
.count();
let omitted_terminal = total_terminal.saturating_sub(MAX_STATUS_TERMINAL_AGENTS);
if omitted_terminal > 0 {
self.diagnostics.status_terminal_agents_omitted = self
.diagnostics
.status_terminal_agents_omitted
.saturating_add(omitted_terminal as u64);
debug!(
omitted_terminal,
total_terminal,
running_agents = self
.agents
.values()
.filter(|agent| {
matches!(agent.status, AgentStatus::Pending | AgentStatus::Running)
})
.count(),
cumulative_omitted = self.diagnostics.status_terminal_agents_omitted,
"omitting terminal agents from status payload to keep UI responsive"
);
}
let agents: Vec<AgentInfo> = self
.visible_agents_for_status()
.into_iter()
.map(|agent| {
// Just show the model name - status provides the useful info
let name = agent
@@ -565,11 +842,14 @@ impl AgentManager {
})
.collect();
// Get context and task from the first agent (they're all the same)
let (context, task) = self
// Prefer active agents for shared context/task; terminal agents may
// have had heavy fields compacted already.
let source_agent = self
.agents
.values()
.next()
.find(|agent| matches!(agent.status, AgentStatus::Pending | AgentStatus::Running))
.or_else(|| self.agents.values().next());
let (context, task) = source_agent
.map(|agent| {
let context = agent
.context
@@ -758,7 +1038,11 @@ impl AgentManager {
}
pub fn get_agent(&self, agent_id: &str) -> Option<Agent> {
self.agents.get(agent_id).cloned()
self
.agents
.get(agent_id)
.cloned()
.or_else(|| self.archived_terminal_agents.get(agent_id).cloned())
}
pub fn get_all_agents(&self) -> impl Iterator<Item = &Agent> {
@@ -777,28 +1061,38 @@ impl AgentManager {
None
};
self.agents
.values()
.filter(|agent| {
if let Some(ref filter) = status_filter {
if agent.status != *filter {
return false;
}
let mut out = Vec::new();
let mut seen_ids: HashSet<String> = HashSet::new();
let mut collect_filtered = |agent: &Agent| {
if let Some(ref filter) = status_filter {
if agent.status != *filter {
return;
}
if let Some(ref batch) = batch_id {
if agent.batch_id.as_ref() != Some(batch) {
return false;
}
}
if let Some(ref batch) = batch_id {
if agent.batch_id.as_ref() != Some(batch) {
return;
}
if let Some(cutoff) = cutoff {
if agent.created_at < cutoff {
return false;
}
}
if let Some(cutoff) = cutoff {
if agent.created_at < cutoff {
return;
}
true
})
.cloned()
.collect()
}
if seen_ids.insert(agent.id.clone()) {
out.push(agent.clone());
}
};
for agent in self.agents.values() {
collect_filtered(agent);
}
for agent in self.archived_terminal_agents.values() {
collect_filtered(agent);
}
out
}
pub fn has_active_agents(&self) -> bool {
@@ -814,6 +1108,7 @@ impl AgentManager {
agent.status = AgentStatus::Cancelled;
agent.completed_at = Some(Utc::now());
}
self.finalize_terminal_agent(agent_id);
true
} else {
false
@@ -838,6 +1133,7 @@ impl AgentManager {
}
pub async fn update_agent_status(&mut self, agent_id: &str, status: AgentStatus) {
let mut terminal = false;
if let Some(agent) = self.agents.get_mut(agent_id) {
agent.status = status;
if agent.status == AgentStatus::Running && agent.started_at.is_none() {
@@ -848,15 +1144,22 @@ impl AgentManager {
AgentStatus::Completed | AgentStatus::Failed | AgentStatus::Cancelled
) {
agent.completed_at = Some(Utc::now());
terminal = true;
}
Self::record_activity(agent);
// Send status update event
self.send_agent_status_update().await;
}
if terminal {
self.finalize_terminal_agent(agent_id);
}
// Send status update event
self.send_agent_status_update().await;
}
pub async fn update_agent_result(&mut self, agent_id: &str, result: Result<String, String>) {
let debug_enabled = self.debug_log_root.is_some();
let mut updated = false;
if let Some((log_tag, log_lines)) = self.agents.get_mut(agent_id).map(|agent| {
let log_tag = if debug_enabled { agent.log_tag.clone() } else { None };
@@ -890,6 +1193,7 @@ impl AgentManager {
}
agent.completed_at = Some(Utc::now());
Self::record_activity(agent);
updated = true;
(log_tag, log_lines)
}) {
@@ -898,6 +1202,9 @@ impl AgentManager {
self.append_agent_log(&tag, &line);
}
}
if updated {
self.finalize_terminal_agent(agent_id);
}
// Send status update event
self.send_agent_status_update().await;
}
@@ -906,13 +1213,44 @@ impl AgentManager {
pub async fn add_progress(&mut self, agent_id: &str, message: String) {
let debug_enabled = self.debug_log_root.is_some();
if let Some((log_tag, entry)) = self.agents.get_mut(agent_id).map(|agent| {
let entry = format!("{}: {}", Utc::now().format("%H:%M:%S"), message);
if let Some((log_tag, entry, line_truncated, entries_trimmed)) =
self.agents.get_mut(agent_id).map(|agent| {
let raw_entry = format!("{}: {}", Utc::now().format("%H:%M:%S"), message);
let line_truncated = raw_entry.len() > MAX_AGENT_PROGRESS_LINE_BYTES;
let entry = Self::trim_to_tail_utf8(&raw_entry, MAX_AGENT_PROGRESS_LINE_BYTES);
let log_tag = if debug_enabled { agent.log_tag.clone() } else { None };
agent.progress.push(entry.clone());
let mut entries_trimmed = 0usize;
if agent.progress.len() > MAX_AGENT_PROGRESS_ENTRIES {
let drain = agent.progress.len() - MAX_AGENT_PROGRESS_ENTRIES;
agent.progress.drain(0..drain);
entries_trimmed = drain;
}
Self::record_activity(agent);
(log_tag, entry)
(log_tag, entry, line_truncated, entries_trimmed)
}) {
if line_truncated {
self.diagnostics.progress_lines_truncated = self
.diagnostics
.progress_lines_truncated
.saturating_add(1);
}
if entries_trimmed > 0 {
self.diagnostics.progress_entries_trimmed = self
.diagnostics
.progress_entries_trimmed
.saturating_add(entries_trimmed as u64);
}
if line_truncated || entries_trimmed > 0 {
debug!(
agent_id,
line_truncated,
entries_trimmed,
total_progress_lines_truncated = self.diagnostics.progress_lines_truncated,
total_progress_entries_trimmed = self.diagnostics.progress_entries_trimmed,
"trimmed agent progress backlog"
);
}
if let Some(tag) = log_tag {
self.append_agent_log(&tag, &entry);
}
@@ -1203,7 +1541,11 @@ async fn execute_agent(agent_id: String, config: Option<AgentConfig>) {
fn prefer_json_result(path: Option<&PathBuf>, fallback: Result<String, String>) -> Result<String, String> {
if let Some(p) = path {
if let Ok(json) = std::fs::read_to_string(p) {
let json = std::fs::read_to_string(p).ok();
if let Err(err) = std::fs::remove_file(p) {
tracing::debug!("failed to clean review output file {}: {err}", p.display());
}
if let Some(json) = json {
return Ok(json);
}
}
@@ -2451,6 +2793,12 @@ where
#[cfg(test)]
mod tests {
use super::Agent;
use super::AgentManager;
use super::AgentStatus;
use super::MAX_AGENT_PROGRESS_ENTRIES;
use super::MAX_AGENT_RESULT_BYTES;
use super::MAX_TRACKED_TERMINAL_AGENTS;
use super::normalize_agent_name;
use super::maybe_set_gemini_config_dir;
use super::execute_model_with_permissions;
@@ -2502,6 +2850,7 @@ mod tests {
let res = prefer_json_result(Some(&path), Err("fallback".to_string()));
assert_eq!(res.unwrap(), payload);
assert!(!path.exists(), "review output file should be cleaned up");
}
#[test]
@@ -2733,6 +3082,178 @@ mod tests {
assert!(!env.contains_key("GEMINI_CONFIG_DIR"));
}
#[test]
fn prune_terminal_agents_caps_completed_state() {
let mut manager = AgentManager::new();
let now = chrono::Utc::now();
let total = MAX_TRACKED_TERMINAL_AGENTS + 64;
for idx in 0..total {
let id = format!("agent-{idx}");
manager.agents.insert(
id.clone(),
Agent {
id,
batch_id: Some("batch-1".to_string()),
model: "code-gpt-5.3-codex".to_string(),
name: Some("Prune Test".to_string()),
prompt: "prompt".repeat(256),
context: Some("ctx".repeat(256)),
output_goal: Some("goal".repeat(256)),
files: vec!["a".repeat(256)],
read_only: false,
status: AgentStatus::Completed,
result: Some("result".repeat(1024)),
error: None,
created_at: now,
started_at: Some(now),
completed_at: Some(now + chrono::Duration::seconds(idx as i64)),
progress: vec!["progress".repeat(1024)],
worktree_path: Some("/tmp/wt".to_string()),
branch_name: Some("code-branch".to_string()),
worktree_base: None,
source_kind: None,
log_tag: None,
config: None,
reasoning_effort: ReasoningEffort::Low,
last_activity: now,
},
);
}
manager.prune_terminal_agents();
assert!(manager.agents.len() <= MAX_TRACKED_TERMINAL_AGENTS);
assert!(
manager
.agents
.values()
.all(|agent| !agent.worktree_path.as_deref().unwrap_or_default().is_empty()),
"worktree metadata should be retained for remaining terminal agents"
);
}
#[test]
fn finalize_terminal_agent_compacts_heavy_fields_and_keeps_worktree() {
let mut manager = AgentManager::new();
let now = chrono::Utc::now();
let agent_id = "agent-finalize".to_string();
manager.agents.insert(
agent_id.clone(),
Agent {
id: agent_id.clone(),
batch_id: Some("batch-compact".to_string()),
model: "code-gpt-5.3-codex".to_string(),
name: Some("Finalize".to_string()),
prompt: "prompt".repeat(1024),
context: Some("context".repeat(1024)),
output_goal: Some("goal".repeat(1024)),
files: vec!["file".repeat(1024)],
read_only: false,
status: AgentStatus::Completed,
result: Some("result".repeat(32 * 1024)),
error: None,
created_at: now,
started_at: Some(now),
completed_at: Some(now),
progress: (0..(MAX_AGENT_PROGRESS_ENTRIES + 20))
.map(|idx| format!("progress-entry-{idx}-{}", "x".repeat(512)))
.collect(),
worktree_path: Some("/tmp/wt-stays".to_string()),
branch_name: Some("branch-stays".to_string()),
worktree_base: None,
source_kind: None,
log_tag: None,
config: None,
reasoning_effort: ReasoningEffort::Low,
last_activity: now,
},
);
manager.finalize_terminal_agent(&agent_id);
let agent = manager
.agents
.get(&agent_id)
.expect("agent should still be tracked");
assert!(agent.prompt.is_empty());
assert!(agent.context.is_none());
assert!(agent.output_goal.is_none());
assert!(agent.files.is_empty());
assert_eq!(agent.worktree_path.as_deref(), Some("/tmp/wt-stays"));
assert_eq!(agent.branch_name.as_deref(), Some("branch-stays"));
assert!(agent.progress.len() <= MAX_AGENT_PROGRESS_ENTRIES);
let result_len = agent
.result
.as_ref()
.map(String::len)
.expect("result retained");
assert!(
result_len <= MAX_AGENT_RESULT_BYTES + 4,
"result should be compacted to bounded size"
);
}
#[test]
fn pruned_terminal_agent_remains_queryable_for_session() {
let mut manager = AgentManager::new();
let now = chrono::Utc::now();
let batch_id = "batch-session".to_string();
let total = MAX_TRACKED_TERMINAL_AGENTS + 8;
for idx in 0..total {
let id = format!("agent-{idx}");
manager.agents.insert(
id.clone(),
Agent {
id,
batch_id: Some(batch_id.clone()),
model: "code-gpt-5.3-codex".to_string(),
name: Some("Archive Test".to_string()),
prompt: "prompt".to_string(),
context: None,
output_goal: None,
files: Vec::new(),
read_only: false,
status: AgentStatus::Completed,
result: Some("ok".to_string()),
error: None,
created_at: now,
started_at: Some(now),
completed_at: Some(now + chrono::Duration::seconds(idx as i64)),
progress: vec!["progress".to_string()],
worktree_path: Some(format!("/tmp/worktree-{idx}")),
branch_name: Some(format!("code-branch-{idx}")),
worktree_base: None,
source_kind: None,
log_tag: None,
config: None,
reasoning_effort: ReasoningEffort::Low,
last_activity: now,
},
);
}
manager.prune_terminal_agents();
let archived_id = "agent-0";
assert!(manager.agents.get(archived_id).is_none());
let archived = manager
.get_agent(archived_id)
.expect("pruned agent should remain queryable");
assert_eq!(archived.worktree_path.as_deref(), Some("/tmp/worktree-0"));
assert_eq!(archived.branch_name.as_deref(), Some("code-branch-0"));
let listed = manager.list_agents(None, Some(batch_id), false);
assert!(listed.iter().any(|agent| {
agent.id == archived_id
&& agent.worktree_path.as_deref() == Some("/tmp/worktree-0")
&& agent.branch_name.as_deref() == Some("code-branch-0")
}));
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
code-rs/core/src/codex/events.rs
+14
View File
@@ -1,4 +1,5 @@
use super::*;
use super::session::MAX_EVENT_SEQ_SUB_IDS;
impl Session {
pub(crate) async fn send_event(&self, event: Event) {
@@ -41,6 +42,19 @@ impl Session {
/// Create a stamped Event with a per-turn sequence number.
fn stamp_event(&self, sub_id: &str, msg: EventMsg) -> Event {
let mut state = self.state.lock().unwrap();
if state.event_seq_by_sub_id.len() > MAX_EVENT_SEQ_SUB_IDS {
while state.event_seq_by_sub_id.len() > MAX_EVENT_SEQ_SUB_IDS {
let Some(old_key) = state.event_seq_by_sub_id.keys().next().cloned() else {
break;
};
state.event_seq_by_sub_id.remove(&old_key);
}
warn!(
cap = MAX_EVENT_SEQ_SUB_IDS,
retained = state.event_seq_by_sub_id.len(),
"trimmed event_seq_by_sub_id map to cap long-session growth"
);
}
let seq = match msg {
EventMsg::TaskStarted => {
// Reset per-sub_id sequence at the start of a turn.
code-rs/core/src/codex/session.rs
+45
View File
@@ -15,6 +15,13 @@ use super::streaming::{
write_agent_file,
};
pub(super) const MAX_EVENT_SEQ_SUB_IDS: usize = 1024;
pub(super) const MAX_BACKGROUND_SEQ_SUB_IDS: usize = 1024;
pub(super) const MAX_WAIT_TRACKED_BATCHES: usize = 1024;
pub(super) const MAX_WAIT_TRACKED_AGENT_IDS_PER_BATCH: usize = 2048;
pub(super) const MAX_AGENT_COMPLETION_WAKE_BATCHES: usize = 2048;
pub(super) const MAX_PENDING_MANUAL_COMPACTS: usize = 64;
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
pub struct ApprovedCommandPattern {
argv: Vec<String>,
@@ -145,8 +152,13 @@ pub(super) struct State {
/// `return_all=false`.
/// This enables sequential waiting behavior across multiple calls.
pub(super) seen_completed_agents_by_batch: HashMap<String, HashSet<String>>,
/// FIFO order for `seen_completed_agents_by_batch` so old batches can be
/// evicted under sustained long-session churn.
pub(super) seen_completed_batch_order: VecDeque<String>,
/// Tracks agent batches that already triggered a wake-up after completion.
pub(super) agent_completion_wake_batches: HashSet<String>,
/// FIFO order for wake-batch dedupe keys.
pub(super) agent_completion_wake_order: VecDeque<String>,
/// Scratchpad that buffers streamed items/deltas for the current HTTP attempt
/// so we can seed retries without losing progress.
pub(super) turn_scratchpad: Option<TurnScratchpad>,
@@ -432,6 +444,13 @@ impl Session {
fn next_background_sequence(&self, sub_id: &str) -> u64 {
let mut state = self.state.lock().unwrap();
if state.background_seq_by_sub_id.len() > MAX_BACKGROUND_SEQ_SUB_IDS {
trim_sub_id_sequence_map(
&mut state.background_seq_by_sub_id,
MAX_BACKGROUND_SEQ_SUB_IDS,
"background_seq_by_sub_id",
);
}
let entry = state
.background_seq_by_sub_id
.entry(sub_id.to_string())
@@ -841,6 +860,25 @@ impl Session {
state.turn_scratchpad = None;
}
}
fn trim_sub_id_sequence_map(
map: &mut HashMap<String, u64>,
cap: usize,
label: &str,
) {
while map.len() > cap {
let Some(key) = map.keys().next().cloned() else {
break;
};
map.remove(&key);
}
warn!(
label,
cap,
retained = map.len(),
"trimmed long-lived sequence map to cap memory growth"
);
}
impl Session {
pub(super) fn set_task(&self, agent: AgentTask) {
let mut state = self.state.lock().unwrap();
@@ -1864,6 +1902,13 @@ impl Session {
pub fn enqueue_manual_compact(&self, sub_id: String) -> bool {
let mut state = self.state.lock().unwrap();
let was_empty = state.pending_manual_compacts.is_empty();
while state.pending_manual_compacts.len() >= MAX_PENDING_MANUAL_COMPACTS {
state.pending_manual_compacts.pop_front();
warn!(
cap = MAX_PENDING_MANUAL_COMPACTS,
"dropped oldest pending manual compact request to cap queue growth"
);
}
state.pending_manual_compacts.push_back(sub_id);
was_empty
}
code-rs/core/src/codex/streaming.rs
+178 -38
View File
@@ -15,6 +15,11 @@ use super::session::{
is_connectivity_error,
spawn_usage_task,
};
use super::session::{
MAX_AGENT_COMPLETION_WAKE_BATCHES,
MAX_WAIT_TRACKED_AGENT_IDS_PER_BATCH,
MAX_WAIT_TRACKED_BATCHES,
};
use crate::auth;
use crate::auth_accounts;
use crate::account_switching::RateLimitSwitchState;
@@ -805,13 +810,10 @@ pub(super) async fn submission_loop(
let tx_event_clone = tx_event.clone();
tokio::spawn(async move {
while let Some(payload) = agent_rx.recv().await {
let wake_messages = {
let mut state = sess_for_agents.state.lock().unwrap();
agent_completion_wake_messages(
&payload,
&mut state.agent_completion_wake_batches,
)
};
let wake_messages = {
let mut state = sess_for_agents.state.lock().unwrap();
agent_completion_wake_messages(&payload, &mut state)
};
if !wake_messages.is_empty() {
enqueue_agent_completion_wake(&sess_for_agents, wake_messages)
.await;
@@ -7816,19 +7818,27 @@ async fn handle_wait_for_agent(
} else {
// Sequential behavior: return the next unseen completed agent if available
let mut state = sess.state.lock().unwrap();
let seen = state
.seen_completed_agents_by_batch
.entry(batch_id.clone())
.or_default();
ensure_wait_batch_tracking_capacity(&mut state, &batch_id);
let unseen = {
let seen = state
.seen_completed_agents_by_batch
.entry(batch_id.clone())
.or_default();
completed_agents
.iter()
.find(|a| !seen.contains(&a.id))
.cloned()
};
// Find the first completed agent that we haven't returned yet
if let Some(unseen) = completed_agents
.iter()
.find(|a| !seen.contains(&a.id))
.cloned()
{
if let Some(unseen) = unseen {
// Record as seen and return immediately
seen.insert(unseen.id.clone());
track_seen_completed_agent_for_batch(
&mut state,
&batch_id,
unseen.id.as_str(),
);
drop(state);
// Include output/error preview for the unseen completed agent
@@ -7903,7 +7913,11 @@ async fn handle_wait_for_agent(
if !any_in_progress && !completed_agents.is_empty() {
// Mark all as seen to keep state consistent
for a in &completed_agents {
seen.insert(a.id.clone());
track_seen_completed_agent_for_batch(
&mut state,
&batch_id,
a.id.as_str(),
);
}
drop(state);
@@ -9728,9 +9742,62 @@ fn build_agent_completion_wake_message(batch_id: &str) -> ResponseInputItem {
}
}
fn ensure_wait_batch_tracking_capacity(state: &mut State, batch_id: &str) {
if !state.seen_completed_agents_by_batch.contains_key(batch_id) {
state
.seen_completed_batch_order
.push_back(batch_id.to_string());
}
while state.seen_completed_agents_by_batch.len() > MAX_WAIT_TRACKED_BATCHES {
let Some(oldest_batch) = state.seen_completed_batch_order.pop_front() else {
break;
};
if state
.seen_completed_agents_by_batch
.remove(&oldest_batch)
.is_some()
{
warn!(
cap = MAX_WAIT_TRACKED_BATCHES,
dropped_batch = oldest_batch,
retained = state.seen_completed_agents_by_batch.len(),
"trimmed wait-for-agent seen batch tracking"
);
}
}
}
fn track_seen_completed_agent_for_batch(state: &mut State, batch_id: &str, agent_id: &str) {
ensure_wait_batch_tracking_capacity(state, batch_id);
{
let seen = state
.seen_completed_agents_by_batch
.entry(batch_id.to_string())
.or_default();
seen.insert(agent_id.to_string());
while seen.len() > MAX_WAIT_TRACKED_AGENT_IDS_PER_BATCH {
let Some(evicted) = seen.iter().next().cloned() else {
break;
};
seen.remove(&evicted);
warn!(
cap = MAX_WAIT_TRACKED_AGENT_IDS_PER_BATCH,
batch_id,
dropped_agent_id = evicted,
retained = seen.len(),
"trimmed wait-for-agent seen-id tracking for batch"
);
}
}
ensure_wait_batch_tracking_capacity(state, batch_id);
}
fn agent_completion_wake_messages(
payload: &AgentStatusUpdatePayload,
seen_batches: &mut HashSet<String>,
state: &mut State,
) -> Vec<ResponseInputItem> {
let mut batches: HashMap<String, AgentBatchCompletionStatus> = HashMap::new();
@@ -9759,9 +9826,22 @@ fn agent_completion_wake_messages(
if !status.has_terminal || status.has_non_terminal {
continue;
}
if !seen_batches.insert(batch_id.clone()) {
if !state.agent_completion_wake_batches.insert(batch_id.clone()) {
continue;
}
state.agent_completion_wake_order.push_back(batch_id.clone());
while state.agent_completion_wake_batches.len() > MAX_AGENT_COMPLETION_WAKE_BATCHES {
let Some(oldest) = state.agent_completion_wake_order.pop_front() else {
break;
};
state.agent_completion_wake_batches.remove(&oldest);
warn!(
cap = MAX_AGENT_COMPLETION_WAKE_BATCHES,
dropped_batch = oldest,
retained = state.agent_completion_wake_batches.len(),
"trimmed agent completion wake dedupe state"
);
}
messages.push(build_agent_completion_wake_message(batch_id.as_str()));
}
@@ -9797,10 +9877,15 @@ async fn enqueue_agent_completion_wake(
#[cfg(test)]
mod agent_completion_wake_tests {
use std::collections::HashSet;
use super::agent_completion_wake_messages;
use super::track_seen_completed_agent_for_batch;
use super::State;
use super::AgentSourceKind;
use crate::codex::session::{
MAX_AGENT_COMPLETION_WAKE_BATCHES,
MAX_WAIT_TRACKED_AGENT_IDS_PER_BATCH,
MAX_WAIT_TRACKED_BATCHES,
};
use crate::agent_tool::AgentStatusUpdatePayload;
use crate::protocol::AgentInfo;
@@ -9829,13 +9914,13 @@ mod agent_completion_wake_tests {
#[test]
fn agent_completion_wake_messages_dedupes_and_skips_non_terminal() {
let mut seen = HashSet::new();
let mut state = State::default();
let running = AgentStatusUpdatePayload {
agents: vec![agent_info("agent-1", "running", Some("batch-1"), None)],
context: None,
task: None,
};
assert!(agent_completion_wake_messages(&running, &mut seen).is_empty());
assert!(agent_completion_wake_messages(&running, &mut state).is_empty());
let mixed = AgentStatusUpdatePayload {
agents: vec![
@@ -9845,17 +9930,17 @@ mod agent_completion_wake_tests {
context: None,
task: None,
};
assert!(agent_completion_wake_messages(&mixed, &mut seen).is_empty());
assert!(agent_completion_wake_messages(&mixed, &mut state).is_empty());
let completed = AgentStatusUpdatePayload {
agents: vec![agent_info("agent-1", "completed", Some("batch-1"), None)],
context: None,
task: None,
};
let messages = agent_completion_wake_messages(&completed, &mut seen);
let messages = agent_completion_wake_messages(&completed, &mut state);
assert_eq!(messages.len(), 1);
let messages_again = agent_completion_wake_messages(&completed, &mut seen);
let messages_again = agent_completion_wake_messages(&completed, &mut state);
assert!(messages_again.is_empty());
let auto_review = AgentStatusUpdatePayload {
@@ -9868,7 +9953,59 @@ mod agent_completion_wake_tests {
context: None,
task: None,
};
assert!(agent_completion_wake_messages(&auto_review, &mut seen).is_empty());
assert!(agent_completion_wake_messages(&auto_review, &mut state).is_empty());
}
#[test]
fn agent_completion_wake_messages_caps_seen_batches() {
let mut state = State::default();
for idx in 0..(MAX_AGENT_COMPLETION_WAKE_BATCHES + 16) {
let batch = format!("batch-{idx}");
let payload = AgentStatusUpdatePayload {
agents: vec![agent_info(
&format!("agent-{idx}"),
"completed",
Some(batch.as_str()),
None,
)],
context: None,
task: None,
};
let messages = agent_completion_wake_messages(&payload, &mut state);
assert_eq!(messages.len(), 1, "each fresh batch should emit one wake message");
}
assert!(state.agent_completion_wake_batches.len() <= MAX_AGENT_COMPLETION_WAKE_BATCHES);
assert!(state.agent_completion_wake_order.len() <= MAX_AGENT_COMPLETION_WAKE_BATCHES);
}
#[test]
fn wait_seen_tracking_caps_batches_and_agent_ids() {
let mut state = State::default();
for idx in 0..(MAX_WAIT_TRACKED_BATCHES + 12) {
let batch = format!("batch-{idx}");
track_seen_completed_agent_for_batch(&mut state, &batch, "agent-1");
}
assert!(state.seen_completed_agents_by_batch.len() <= MAX_WAIT_TRACKED_BATCHES);
let hot_batch = "batch-hot";
for idx in 0..(MAX_WAIT_TRACKED_AGENT_IDS_PER_BATCH + 16) {
track_seen_completed_agent_for_batch(
&mut state,
hot_batch,
&format!("agent-{idx}"),
);
}
let seen = state
.seen_completed_agents_by_batch
.get(hot_batch)
.expect("hot batch should be tracked");
assert!(seen.len() <= MAX_WAIT_TRACKED_AGENT_IDS_PER_BATCH);
}
}
@@ -9876,11 +10013,14 @@ mod agent_completion_wake_tests {
async fn send_agent_status_update(sess: &Session) {
let manager = AGENT_MANAGER.read().await;
// Collect all agents; include completed/failed so HUD can show final messages
// Collect active agents plus a bounded tail of terminal agents so the HUD
// stays responsive in long-running sessions.
let now = Utc::now();
let agents: Vec<crate::protocol::AgentInfo> = manager
.get_all_agents()
.status_visible_agents()
.into_iter()
.map(|agent| {
let status = agent.status.clone();
let start = agent.started_at.unwrap_or(agent.created_at);
let end = agent.completed_at.unwrap_or(now);
let elapsed_ms = match end.signed_duration_since(start).num_milliseconds() {
@@ -9889,26 +10029,26 @@ async fn send_agent_status_update(sess: &Session) {
};
crate::protocol::AgentInfo {
id: agent.id.clone(),
id: agent.id,
name: agent.model.clone(), // Use model name as the display name
status: match agent.status {
status: match &status {
AgentStatus::Pending => "pending".to_string(),
AgentStatus::Running => "running".to_string(),
AgentStatus::Completed => "completed".to_string(),
AgentStatus::Failed => "failed".to_string(),
AgentStatus::Cancelled => "cancelled".to_string(),
},
batch_id: agent.batch_id.clone(),
batch_id: agent.batch_id,
model: Some(agent.model.clone()),
last_progress: agent.progress.last().cloned(),
result: agent.result.clone(),
error: agent.error.clone(),
result: agent.result,
error: agent.error,
elapsed_ms,
token_count: None,
last_activity_at: matches!(agent.status, AgentStatus::Pending | AgentStatus::Running)
last_activity_at: matches!(status, AgentStatus::Pending | AgentStatus::Running)
.then(|| agent.last_activity.to_rfc3339()),
seconds_since_last_activity: matches!(
agent.status,
status,
AgentStatus::Pending | AgentStatus::Running
)
.then(|| {
@@ -9917,7 +10057,7 @@ async fn send_agent_status_update(sess: &Session) {
.num_seconds()
.max(0) as u64
}),
source_kind: agent.source_kind.clone(),
source_kind: agent.source_kind,
}
})
.collect();
code-rs/tui/src/chatwidget.rs
+505 -31
View File
@@ -1312,6 +1312,12 @@ const AUTO_REVIEW_BASELINE_FILENAME: &str = "auto-review-baseline";
const AUTO_REVIEW_FALLBACK_MAX: usize = 3;
const AUTO_REVIEW_FALLBACK_MAX_AGE_SECS: u64 = 12 * 60 * 60; // 12h
const AUTO_REVIEW_STALE_SECS: u64 = 5 * 60;
const MAX_PROCESSED_AUTO_REVIEW_AGENTS: usize = 512;
const MAX_AGENTS_TERMINAL_ENTRIES: usize = 256;
const MAX_AGENT_RUNTIME_ENTRIES: usize = 768;
const MAX_HISTORY_CELLS_HARD_LIMIT: usize = 2400;
const HISTORY_CELLS_TRIM_TARGET: usize = 1800;
const MAX_PENDING_DISPATCHED_USER_MESSAGES: usize = 256;
fn auto_review_repo_dir(git_root: &Path) -> Result<PathBuf, String> {
let repo_name = git_root
@@ -1620,6 +1626,8 @@ pub(crate) struct ChatWidget<'a> {
review_guard: Option<ReviewGuard>,
background_review_guard: Option<ReviewGuard>,
processed_auto_review_agents: HashSet<String>,
processed_auto_review_agent_order: VecDeque<String>,
auto_review_processed_evicted_total: u64,
// New: coordinator-provided hints for the next Auto turn
pending_turn_descriptor: Option<TurnDescriptor>,
pending_auto_turn_config: Option<TurnConfig>,
@@ -1627,6 +1635,10 @@ pub(crate) struct ChatWidget<'a> {
active_plan_title: Option<String>,
/// Runtime timing per-agent (by id) to improve visibility in the HUD
agent_runtime: HashMap<String, AgentRuntime>,
agent_runtime_pruned_total: u64,
agents_terminal_pruned_total: u64,
history_cells_trimmed_total: u64,
pending_dispatched_trimmed_total: u64,
pending_agent_updates: HashMap<String, PendingAgentUpdate>,
// Sparkline data for showing agent activity (using RefCell for interior mutability)
// Each tuple is (value, is_completed) where is_completed indicates if any agent was complete at that time
@@ -1974,6 +1986,7 @@ struct AgentTerminalEntry {
result: Option<String>,
error: Option<String>,
logs: Vec<AgentLogEntry>,
last_seen_at: Instant,
}
impl AgentTerminalEntry {
@@ -1996,6 +2009,7 @@ impl AgentTerminalEntry {
result: None,
error: None,
logs: Vec::new(),
last_seen_at: Instant::now(),
}
}
@@ -6542,6 +6556,8 @@ impl ChatWidget<'_> {
review_guard: None,
background_review_guard: None,
processed_auto_review_agents: HashSet::new(),
processed_auto_review_agent_order: VecDeque::new(),
auto_review_processed_evicted_total: 0,
pending_turn_descriptor: None,
render_request_cache: RefCell::new(Vec::new()),
render_request_cache_dirty: Cell::new(true),
@@ -6550,6 +6566,10 @@ impl ChatWidget<'_> {
overall_task_status: "preparing".to_string(),
active_plan_title: None,
agent_runtime: HashMap::new(),
agent_runtime_pruned_total: 0,
agents_terminal_pruned_total: 0,
history_cells_trimmed_total: 0,
pending_dispatched_trimmed_total: 0,
pending_agent_updates: HashMap::new(),
sparkline_data: std::cell::RefCell::new(Vec::new()),
last_sparkline_update: std::cell::RefCell::new(std::time::Instant::now()),
@@ -6919,6 +6939,8 @@ impl ChatWidget<'_> {
review_guard: None,
background_review_guard: None,
processed_auto_review_agents: HashSet::new(),
processed_auto_review_agent_order: VecDeque::new(),
auto_review_processed_evicted_total: 0,
pending_turn_descriptor: None,
render_request_cache: RefCell::new(Vec::new()),
render_request_cache_dirty: Cell::new(true),
@@ -6927,6 +6949,10 @@ impl ChatWidget<'_> {
overall_task_status: "preparing".to_string(),
active_plan_title: None,
agent_runtime: HashMap::new(),
agent_runtime_pruned_total: 0,
agents_terminal_pruned_total: 0,
history_cells_trimmed_total: 0,
pending_dispatched_trimmed_total: 0,
pending_agent_updates: HashMap::new(),
sparkline_data: std::cell::RefCell::new(Vec::new()),
last_sparkline_update: std::cell::RefCell::new(std::time::Instant::now()),
@@ -9747,6 +9773,10 @@ impl ChatWidget<'_> {
self.refresh_explore_trailing_flags();
self.refresh_reasoning_collapsed_visibility();
self.mark_history_dirty();
let trimmed = self.enforce_history_cell_limit();
if trimmed > 0 {
pos = pos.saturating_sub(trimmed);
}
pos
}
@@ -9947,6 +9977,10 @@ impl ChatWidget<'_> {
self.refresh_explore_trailing_flags();
self.refresh_reasoning_collapsed_visibility();
self.mark_history_dirty();
let trimmed = self.enforce_history_cell_limit();
if trimmed > 0 {
pos = pos.saturating_sub(trimmed);
}
pos
}
@@ -10758,6 +10792,50 @@ impl ChatWidget<'_> {
self.mark_history_dirty();
}
fn enforce_history_cell_limit(&mut self) -> usize {
if self.history_cells.len() <= MAX_HISTORY_CELLS_HARD_LIMIT {
return 0;
}
let remove_count = self
.history_cells
.len()
.saturating_sub(HISTORY_CELLS_TRIM_TARGET);
if remove_count == 0 {
return 0;
}
tracing::warn!(
"history cell backlog exceeded {}; trimming {} oldest cells",
MAX_HISTORY_CELLS_HARD_LIMIT,
remove_count
);
for _ in 0..remove_count {
if self.history_cells.is_empty() {
break;
}
self.history_remove_at(0);
}
self.history_cells_trimmed_total = self
.history_cells_trimmed_total
.saturating_add(remove_count as u64);
tracing::warn!(
removed = remove_count,
retained = self.history_cells.len(),
total_trimmed = self.history_cells_trimmed_total,
"history cell hard-limit compaction applied"
);
let current_offset = self.layout.scroll_offset.get();
let removed_offset = remove_count.min(u16::MAX as usize) as u16;
self.layout
.scroll_offset
.set(current_offset.saturating_sub(removed_offset));
remove_count
}
fn history_replace_and_maybe_merge(&mut self, idx: usize, cell: Box<dyn HistoryCell>) {
// Replace at index, then attempt standard exec merge with previous cell.
self.history_replace_at(idx, cell);
@@ -13078,6 +13156,25 @@ impl ChatWidget<'_> {
if let Some(model_echo) = combined_message_text {
self.pending_dispatched_user_messages.push_back(model_echo);
let mut dropped = 0usize;
while self.pending_dispatched_user_messages.len()
> MAX_PENDING_DISPATCHED_USER_MESSAGES
{
if self.pending_dispatched_user_messages.pop_front().is_some() {
dropped = dropped.saturating_add(1);
}
}
if dropped > 0 {
self.pending_dispatched_trimmed_total = self
.pending_dispatched_trimmed_total
.saturating_add(dropped as u64);
tracing::warn!(
dropped,
retained = self.pending_dispatched_user_messages.len(),
total_trimmed = self.pending_dispatched_trimmed_total,
"trimmed pending dispatched user message backlog"
);
}
}
let suppress_history = suppress_persistence;
@@ -15092,6 +15189,8 @@ impl ChatWidget<'_> {
}
}
self.prune_agent_runtime_cache(&agents);
self.update_agents_terminal_state(&agents, context.clone(), task.clone());
self.observe_auto_review_status(&agents);
@@ -17465,6 +17564,7 @@ impl ChatWidget<'_> {
) {
self.agents_terminal.shared_context = context;
self.agents_terminal.shared_task = task;
let now = Instant::now();
let mut saw_new_agent = false;
for info in agents {
@@ -17507,6 +17607,7 @@ impl ChatWidget<'_> {
entry.batch_id = info.batch_id.clone();
entry.model = info.model.clone();
entry.source_kind = info.source_kind.clone();
entry.last_seen_at = now;
let AgentBatchMetadata { label, prompt: meta_prompt, context: meta_context } = batch_metadata;
let auto_review_label = matches!(entry.source_kind, Some(AgentSourceKind::AutoReview))
@@ -17580,6 +17681,7 @@ impl ChatWidget<'_> {
}
self.agents_terminal.clamp_selected_index();
self.prune_agents_terminal_entries();
if saw_new_agent && self.agents_terminal.active {
self.layout.scroll_offset.set(0);
@@ -29896,6 +29998,7 @@ use code_core::protocol::OrderMeta;
#[test]
fn auto_review_triggers_when_enabled_and_diff_seen() {
let _stub_lock = AUTO_STUB_LOCK.lock().unwrap();
let _guard = AutoReviewStubGuard::install(|| {});
let _capture_guard = CaptureCommitStubGuard::install(|_, _| {
Ok(GhostCommit::new("baseline".to_string(), None))
@@ -29913,6 +30016,7 @@ use code_core::protocol::OrderMeta;
#[test]
fn auto_review_does_not_duplicate_while_running() {
let _stub_lock = AUTO_STUB_LOCK.lock().unwrap();
let calls = Arc::new(AtomicUsize::new(0));
let calls_clone = calls.clone();
let _guard = AutoReviewStubGuard::install(move || {
@@ -29938,6 +30042,7 @@ use code_core::protocol::OrderMeta;
#[test]
fn auto_review_skips_when_no_changes_since_reviewed_snapshot() {
let _stub_lock = AUTO_STUB_LOCK.lock().unwrap();
let _rt = enter_test_runtime_guard();
let calls = Arc::new(AtomicUsize::new(0));
let calls_clone = calls.clone();
@@ -30060,12 +30165,16 @@ use code_core::protocol::OrderMeta;
})
});
assert!(notice_present, "actionable auto review notice should be visible");
assert!(chat.pending_agent_notes.is_empty(), "idle path should inject via hidden message, not queue notes");
let developer_seen = chat
.pending_dispatched_user_messages
.iter()
.any(|msg| msg.contains("[developer]"));
assert!(developer_seen, "developer note should be sent in hidden message");
assert!(chat.pending_agent_notes.is_empty());
assert!(chat.pending_dispatched_user_messages.is_empty());
let developer_seen = chat.history_cells.iter().any(|cell| {
cell.display_lines_trimmed().iter().any(|line| {
line.spans
.iter()
.any(|span| span.content.contains("[developer] Background auto-review"))
})
});
assert!(developer_seen, "developer note should be rendered as a background notice");
}
#[test]
@@ -30107,13 +30216,18 @@ use code_core::protocol::OrderMeta;
Some("ghost123".to_string()),
);
// Busy path still injects a developer note immediately so the user sees it in the transcript.
assert!(chat.pending_agent_notes.is_empty());
let developer_sent = chat
.pending_dispatched_user_messages
.iter()
.any(|msg| msg.contains("[developer]") && msg.contains("Merge the worktree") && msg.contains("auto-review-branch"));
assert!(developer_sent, "developer merge-hint note should be injected even while busy");
assert!(chat.pending_dispatched_user_messages.is_empty());
let developer_seen = chat.history_cells.iter().any(|cell| {
cell.display_lines_trimmed().iter().any(|line| {
line.spans.iter().any(|span| {
span.content.contains("[developer]")
&& span.content.contains("Merge the worktree")
&& span.content.contains("auto-review-branch")
})
})
});
assert!(developer_seen, "developer merge-hint note should be visible even while busy");
}
#[test]
@@ -30157,13 +30271,18 @@ use code_core::protocol::OrderMeta;
chat.observe_auto_review_status(&[agent]);
// Idle path: should send hidden developer note immediately (not queued)
assert!(chat.pending_agent_notes.is_empty());
let developer_sent = chat
.pending_dispatched_user_messages
.iter()
.any(|msg| msg.contains("[developer]") && msg.contains("Merge the worktree") && msg.contains("auto-review-branch"));
assert!(developer_sent, "developer merge-hint note should be injected when idle");
assert!(chat.pending_dispatched_user_messages.is_empty());
let developer_seen = chat.history_cells.iter().any(|cell| {
cell.display_lines_trimmed().iter().any(|line| {
line.spans.iter().any(|span| {
span.content.contains("[developer]")
&& span.content.contains("Merge the worktree")
&& span.content.contains("auto-review-branch")
})
})
});
assert!(developer_seen, "developer merge-hint note should be visible when idle");
}
#[test]
@@ -30209,11 +30328,190 @@ use code_core::protocol::OrderMeta;
chat.observe_auto_review_status(&[agent]);
assert!(chat.pending_agent_notes.is_empty());
let developer_sent = chat
.pending_dispatched_user_messages
.iter()
.any(|msg| msg.contains("[developer]") && msg.contains("Merge the worktree") && msg.contains("auto-review-branch"));
assert!(developer_sent, "developer merge-hint note should be injected when busy");
assert!(chat.pending_dispatched_user_messages.is_empty());
let developer_seen = chat.history_cells.iter().any(|cell| {
cell.display_lines_trimmed().iter().any(|line| {
line.spans.iter().any(|span| {
span.content.contains("[developer]")
&& span.content.contains("Merge the worktree")
&& span.content.contains("auto-review-branch")
})
})
});
assert!(developer_seen, "developer merge-hint note should be visible when busy");
}
#[test]
fn esc_stop_auto_drive_keeps_background_review_non_blocking() {
let _rt = enter_test_runtime_guard();
let mut harness = ChatWidgetHarness::new();
let chat = harness.chat();
chat.auto_state.set_phase(AutoRunPhase::Active);
chat.auto_state.goal = Some("keep running".to_string());
chat.auto_state.started_at = Some(Instant::now());
chat.background_review = Some(BackgroundReviewState {
worktree_path: PathBuf::from("/tmp/wt"),
branch: "auto-review-branch".to_string(),
agent_id: Some("agent-esc".to_string()),
snapshot: Some("ghost999".to_string()),
base: None,
last_seen: Instant::now(),
});
let esc = KeyEvent::new(KeyCode::Esc, KeyModifiers::NONE);
let mut last_esc_time = None;
assert!(chat.handle_app_esc(esc, &mut last_esc_time));
assert!(!chat.auto_state.is_active(), "Esc should stop Auto Drive immediately");
let pending_before_review = chat.pending_dispatched_user_messages.len();
chat.on_background_review_finished(
PathBuf::from("/tmp/wt"),
"auto-review-branch".to_string(),
true,
1,
Some("summary".to_string()),
None,
Some("agent-esc".to_string()),
Some("ghost999".to_string()),
);
let pending_after_review = chat.pending_dispatched_user_messages.len();
assert_eq!(
pending_after_review,
pending_before_review,
"background review completion must not inject hidden foreground turns"
);
assert!(!chat.is_task_running(), "background review should not hold task-running state");
chat.submit_text_message("after esc".to_string());
assert!(
chat.pending_dispatched_user_messages
.iter()
.any(|msg| msg.contains("after esc")),
"typing should resume immediately after Esc even while auto review events complete"
);
assert!(chat.queued_user_messages.is_empty());
}
#[test]
fn heavy_agent_churn_with_auto_review_keeps_bounds_and_typing() {
let _rt = enter_test_runtime_guard();
let mut harness = ChatWidgetHarness::new();
let chat = harness.chat();
chat.config.tui.auto_review_enabled = true;
chat.auto_state.set_phase(AutoRunPhase::Active);
chat.auto_state.goal = Some("long churn".to_string());
chat.auto_state.started_at = Some(Instant::now());
chat.background_review = Some(BackgroundReviewState {
worktree_path: PathBuf::from("/tmp/wt"),
branch: "auto-review-branch".to_string(),
agent_id: Some("agent-auto-review".to_string()),
snapshot: Some("ghost-long".to_string()),
base: None,
last_seen: Instant::now(),
});
let total_agents = 2200usize;
let chunk_size = 55usize;
let chunks = total_agents / chunk_size;
for chunk in 0..chunks {
let mut agents = Vec::with_capacity(chunk_size + 1);
for offset in 0..chunk_size {
let idx = chunk * chunk_size + offset;
agents.push(code_core::protocol::AgentInfo {
id: format!("agent-churn-{idx}"),
name: format!("Agent {idx}"),
status: "completed".to_string(),
batch_id: Some(format!("batch-{}", idx / 5)),
model: Some("code-gpt-5.3-codex".to_string()),
last_progress: Some("done".to_string()),
result: Some("ok".to_string()),
error: None,
elapsed_ms: None,
token_count: None,
last_activity_at: None,
seconds_since_last_activity: None,
source_kind: None,
});
}
let auto_status = if chunk + 1 == chunks {
"completed"
} else {
"running"
};
agents.push(code_core::protocol::AgentInfo {
id: "agent-auto-review".to_string(),
name: "Auto Review".to_string(),
status: auto_status.to_string(),
batch_id: Some("auto-review-branch".to_string()),
model: Some("code-review".to_string()),
last_progress: Some("[auto-review] phase: reviewing".to_string()),
result: Some(
r#"{
"findings":[{"title":"bug","body":"details","confidence_score":0.5,"priority":1,"code_location":{"absolute_file_path":"src/lib.rs","line_range":{"start":1,"end":1}}}],
"overall_correctness":"incorrect",
"overall_explanation":"needs work",
"overall_confidence_score":0.6
}"#
.to_string(),
),
error: None,
elapsed_ms: None,
token_count: None,
last_activity_at: None,
seconds_since_last_activity: None,
source_kind: Some(AgentSourceKind::AutoReview),
});
chat.handle_code_event(Event {
id: format!("status-{chunk}"),
event_seq: chunk as u64,
msg: EventMsg::AgentStatusUpdate(AgentStatusUpdateEvent {
agents,
context: None,
task: None,
}),
order: None,
});
}
assert!(chat.agents_terminal.entries.len() <= MAX_AGENTS_TERMINAL_ENTRIES);
assert!(chat.agent_runtime.len() <= MAX_AGENT_RUNTIME_ENTRIES);
assert!(chat.processed_auto_review_agents.len() <= MAX_PROCESSED_AUTO_REVIEW_AGENTS);
let esc = KeyEvent::new(KeyCode::Esc, KeyModifiers::NONE);
let mut last_esc_time = None;
assert!(chat.handle_app_esc(esc, &mut last_esc_time));
assert!(!chat.auto_state.is_active(), "Esc should always exit Auto Drive");
let pending_before_review = chat.pending_dispatched_user_messages.len();
chat.on_background_review_finished(
PathBuf::from("/tmp/wt"),
"auto-review-branch".to_string(),
true,
1,
Some("summary".to_string()),
None,
Some("agent-auto-review".to_string()),
Some("ghost-long".to_string()),
);
assert_eq!(
chat.pending_dispatched_user_messages.len(),
pending_before_review,
"background review completion must stay out-of-band"
);
chat.submit_text_message("typing after churn".to_string());
assert!(
chat.pending_dispatched_user_messages
.iter()
.any(|msg| msg.contains("typing after churn")),
"typing should remain responsive after heavy churn"
);
}
#[test]
@@ -30735,6 +31033,37 @@ use code_core::protocol::OrderMeta;
}
}
#[test]
fn history_insert_enforces_hard_limit_and_keeps_recent_tail() {
let mut harness = ChatWidgetHarness::new();
let chat = harness.chat();
reset_history(chat);
for idx in 0..(MAX_HISTORY_CELLS_HARD_LIMIT + 75) {
insert_plain_cell(chat, &[&format!("history-line-{idx}")]);
}
assert!(chat.history_cells.len() <= MAX_HISTORY_CELLS_HARD_LIMIT);
assert!(chat.history_cells.len() >= HISTORY_CELLS_TRIM_TARGET);
let oldest_present = chat.history_cells.iter().any(|cell| {
cell.display_lines_trimmed().iter().any(|line| {
line.spans
.iter()
.any(|span| span.content.contains("history-line-0"))
})
});
assert!(!oldest_present, "oldest history entries should be compacted away");
let newest = format!("history-line-{}", MAX_HISTORY_CELLS_HARD_LIMIT + 74);
let newest_present = chat.history_cells.iter().any(|cell| {
cell.display_lines_trimmed().iter().any(|line| {
line.spans.iter().any(|span| span.content.contains(&newest))
})
});
assert!(newest_present, "most recent history entries should remain visible");
}
#[test]
fn review_dialog_uncommitted_option_runs_workspace_scope() {
let mut harness = ChatWidgetHarness::new();
@@ -34440,7 +34769,7 @@ impl ChatWidget<'_> {
self.clear_auto_review_indicator();
self.background_review = None;
self.background_review_guard = None;
self.processed_auto_review_agents.insert(agent.id.clone());
self.remember_processed_auto_review_agent(&agent.id);
continue;
}
}
@@ -34507,13 +34836,13 @@ impl ChatWidget<'_> {
// times (especially after cancellation/resend). If the background
// review state is already cleared and we cannot resolve the worktree,
// do not surface a misleading blank-path error.
self.processed_auto_review_agents.insert(agent.id.clone());
self.remember_processed_auto_review_agent(&agent.id);
continue;
};
let Some(worktree_path) =
resolve_auto_review_worktree_path(&self.config.cwd, &branch)
else {
self.processed_auto_review_agents.insert(agent.id.clone());
self.remember_processed_auto_review_agent(&agent.id);
continue;
};
(worktree_path, branch, None)
@@ -34521,7 +34850,7 @@ impl ChatWidget<'_> {
let (has_findings, findings, summary) = Self::parse_agent_review_result(agent.result.as_deref());
self.processed_auto_review_agents.insert(agent.id.clone());
self.remember_processed_auto_review_agent(&agent.id);
self.on_background_review_finished(
worktree_path,
branch,
@@ -34685,6 +35014,136 @@ impl ChatWidget<'_> {
self.bottom_pane.set_auto_review_status(None);
}
fn remember_processed_auto_review_agent(&mut self, agent_id: &str) {
if agent_id.trim().is_empty() {
return;
}
if !self
.processed_auto_review_agents
.insert(agent_id.to_string())
{
return;
}
self
.processed_auto_review_agent_order
.push_back(agent_id.to_string());
let mut evicted = 0usize;
while self.processed_auto_review_agents.len() > MAX_PROCESSED_AUTO_REVIEW_AGENTS {
let Some(oldest) = self.processed_auto_review_agent_order.pop_front() else {
break;
};
if self.processed_auto_review_agents.remove(&oldest) {
evicted = evicted.saturating_add(1);
}
}
if evicted > 0 {
self.auto_review_processed_evicted_total = self
.auto_review_processed_evicted_total
.saturating_add(evicted as u64);
tracing::debug!(
evicted,
retained = self.processed_auto_review_agents.len(),
total_evicted = self.auto_review_processed_evicted_total,
"trimmed processed auto-review agent cache"
);
}
}
fn prune_agent_runtime_cache(&mut self, current_agents: &[code_core::protocol::AgentInfo]) {
if self.agent_runtime.len() <= MAX_AGENT_RUNTIME_ENTRIES {
return;
}
let active_ids: HashSet<&str> = current_agents.iter().map(|agent| agent.id.as_str()).collect();
let mut terminal_candidates: Vec<(Instant, String)> = self
.agent_runtime
.iter()
.filter(|(id, runtime)| {
!active_ids.contains(id.as_str()) && runtime.completed_at.is_some()
})
.filter_map(|(id, runtime)| runtime.completed_at.map(|ts| (ts, id.clone())))
.collect();
if terminal_candidates.is_empty() {
return;
}
terminal_candidates.sort_by_key(|(completed_at, _)| *completed_at);
let mut pruned = 0usize;
for (_, id) in terminal_candidates {
if self.agent_runtime.len() <= MAX_AGENT_RUNTIME_ENTRIES {
break;
}
if self.agent_runtime.remove(&id).is_some() {
pruned = pruned.saturating_add(1);
}
}
if pruned > 0 {
self.agent_runtime_pruned_total = self
.agent_runtime_pruned_total
.saturating_add(pruned as u64);
tracing::debug!(
pruned,
retained = self.agent_runtime.len(),
total_pruned = self.agent_runtime_pruned_total,
"trimmed agent runtime cache"
);
}
}
fn prune_agents_terminal_entries(&mut self) {
if self.agents_terminal.entries.len() <= MAX_AGENTS_TERMINAL_ENTRIES {
return;
}
let mut removable: Vec<(Instant, String)> = self
.agents_terminal
.entries
.iter()
.filter(|(_, entry)| {
matches!(
entry.status,
AgentStatus::Completed | AgentStatus::Failed | AgentStatus::Cancelled
)
})
.map(|(id, entry)| (entry.last_seen_at, id.clone()))
.collect();
removable.sort_by_key(|(seen_at, _)| *seen_at);
let mut pruned = 0usize;
for (_, id) in removable {
if self.agents_terminal.entries.len() <= MAX_AGENTS_TERMINAL_ENTRIES {
break;
}
if self.agents_terminal.entries.remove(&id).is_some() {
pruned = pruned.saturating_add(1);
}
self.agents_terminal.order.retain(|agent_id| agent_id != &id);
self.agents_terminal.scroll_offsets.remove(&format!("agent:{id}"));
}
self.agents_terminal.clamp_selected_index();
if pruned > 0 {
self.agents_terminal_pruned_total = self
.agents_terminal_pruned_total
.saturating_add(pruned as u64);
tracing::debug!(
pruned,
retained = self.agents_terminal.entries.len(),
total_pruned = self.agents_terminal_pruned_total,
"trimmed agents terminal entries"
);
}
}
fn last_assistant_cell_index(&self) -> Option<usize> {
self.history_cells.iter().enumerate().rev().find_map(|(idx, cell)| {
cell.as_any()
@@ -34892,10 +35351,7 @@ impl ChatWidget<'_> {
}
if let Some(note) = developer_note {
// Immediately inject as a developer message so the user sees it in the
// transcript, even if tasks/streams are still running. Do not defer to
// pending_agent_notes; that path can be lost if the session ends.
self.submit_hidden_text_message_with_preface(String::new(), note);
self.record_background_review_note(note);
}
// Auto review completion should never leave the composer spinner active.
@@ -34912,6 +35368,24 @@ impl ChatWidget<'_> {
self.request_redraw();
}
fn record_background_review_note(&mut self, note: String) {
let trimmed = note.trim();
if trimmed.is_empty() {
return;
}
self.history_push_plain_paragraphs(PlainMessageKind::Notice, [trimmed.to_string()]);
if let Err(err) = self
.code_op_tx
.send(Op::AddToHistory {
text: trimmed.to_string(),
})
{
tracing::error!("failed to persist background review note: {err}");
}
}
fn handle_auto_review_completion_state(
&mut self,
has_findings: bool,
code-rs/tui/src/lib.rs
+20 -2
View File
@@ -42,6 +42,9 @@ use code_core::review_coord::{
};
use std::sync::Once;
use std::sync::OnceLock;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, Instant};
use tracing_appender::non_blocking;
use tracing_appender::rolling;
use tracing_subscriber::filter::LevelFilter;
@@ -483,9 +486,23 @@ pub async fn run_main(
code_core::config::migrate_legacy_log_dirs(&code_home);
let housekeeping_home = code_home.clone();
let housekeeping_stop = Arc::new(AtomicBool::new(false));
let housekeeping_stop_worker = Arc::clone(&housekeeping_stop);
let housekeeping_handle = thread_spawner::spawn_lightweight("housekeeping", move || {
if let Err(err) = code_core::run_housekeeping_if_due(&housekeeping_home) {
tracing::warn!("code home housekeeping failed: {err}");
const HOUSEKEEPING_INTERVAL: Duration = Duration::from_secs(30 * 60);
const HOUSEKEEPING_POLL_INTERVAL: Duration = Duration::from_secs(10);
let mut next_run_at = Instant::now();
while !housekeeping_stop_worker.load(Ordering::Relaxed) {
let now = Instant::now();
if now >= next_run_at {
if let Err(err) = code_core::run_housekeeping_if_due(&housekeeping_home) {
tracing::warn!("code home housekeeping failed: {err}");
}
next_run_at = now + HOUSEKEEPING_INTERVAL;
}
std::thread::sleep(HOUSEKEEPING_POLL_INTERVAL);
}
});
@@ -724,6 +741,7 @@ pub async fn run_main(
theme_configured_explicitly,
);
housekeeping_stop.store(true, Ordering::Relaxed);
if let Some(handle) = housekeeping_handle {
if let Err(err) = handle.join() {
tracing::warn!("code home housekeeping task panicked: {err:?}");