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

feat(core): queue post-turn review input and add memory guard

Browse Source
This commit is contained in:
James Peter
2026-03-07 19:18:41 +10:00
parent 58ed9082ab
commit 0392b6e179
9 changed files with 512 additions and 307 deletions
Download Patch File Download Diff File Expand all files Collapse all files
code-rs/core/src/codex.rs
+1
View File
@@ -104,6 +104,7 @@ use self::streaming::{add_pending_screenshot, capture_browser_screenshot, submis
pub(crate) const INITIAL_SUBMIT_ID: &str = "";
const HOOK_OUTPUT_LIMIT: usize = 2048;
const PENDING_ONLY_SENTINEL: &str = "__code_pending_only__";
const POST_TURN_PENDING_ONLY_SENTINEL: &str = "__code_post_turn_pending_only__";
const MIN_SHELL_TIMEOUT_MS: u64 = 30 * 60 * 1000;
#[derive(Clone, Default)]
code-rs/core/src/codex/session.rs
+122 -17
View File
@@ -145,6 +145,7 @@ pub(super) struct State {
pub(super) pending_dynamic_tools: HashMap<String, oneshot::Sender<DynamicToolResponse>>,
pub(super) selected_mcp_tools: Vec<String>,
pub(super) pending_input: Vec<ResponseInputItem>,
pub(super) pending_post_turn_input: Vec<ResponseInputItem>,
pub(super) pending_user_input: Vec<QueuedUserInput>,
pub(super) history: ConversationHistory,
/// Tracks which completed agents (by id) have already been returned to the
@@ -233,6 +234,37 @@ pub(crate) struct QueuedUserInput {
pub(super) core_items: Vec<InputItem>,
}
pub(super) enum FollowUpTurnAction {
PostTurnPendingInput,
ManualCompact(String),
PendingInput,
QueuedUserInput(QueuedUserInput),
}
fn take_follow_up_turn_action(state: &mut State) -> Option<FollowUpTurnAction> {
if !state.pending_post_turn_input.is_empty() {
let mut post_turn_input = std::mem::take(&mut state.pending_post_turn_input);
state.pending_input.append(&mut post_turn_input);
return Some(FollowUpTurnAction::PostTurnPendingInput);
}
if let Some(sub_id) = state.pending_manual_compacts.pop_front() {
return Some(FollowUpTurnAction::ManualCompact(sub_id));
}
if !state.pending_input.is_empty() {
return Some(FollowUpTurnAction::PendingInput);
}
if state.pending_user_input.is_empty() {
None
} else {
Some(FollowUpTurnAction::QueuedUserInput(
state.pending_user_input.remove(0),
))
}
}
/// Buffers partial turn progress produced during a single HTTP streaming attempt.
/// This is not recorded to persistent history. It is only used to seed retries
/// when the SSE stream disconnects mid‑turn.
@@ -317,7 +349,18 @@ pub(super) fn is_connectivity_error(err: &CodexErr) -> bool {
#[cfg(test)]
mod tests {
use super::is_connectivity_error;
use super::{FollowUpTurnAction, QueuedUserInput, State, take_follow_up_turn_action};
use crate::error::CodexErr;
use code_protocol::models::{ContentItem, ResponseInputItem};
fn developer_input(text: &str) -> ResponseInputItem {
ResponseInputItem::Message {
role: "developer".to_string(),
content: vec![ContentItem::InputText {
text: text.to_string(),
}],
}
}
#[test]
fn context_overflow_transport_stream_is_not_connectivity() {
@@ -340,6 +383,47 @@ mod tests {
assert!(is_connectivity_error(&err));
}
#[test]
fn follow_up_turn_action_prioritizes_post_turn_input() {
let mut state = State::default();
state.pending_post_turn_input.push(developer_input("post-turn"));
state.pending_manual_compacts.push_back("compact-1".to_string());
state.pending_input.push(developer_input("pending"));
state.pending_user_input.push(QueuedUserInput {
submission_id: "queued-1".to_string(),
response_item: developer_input("queued"),
core_items: vec![],
});
let action = take_follow_up_turn_action(&mut state).expect("follow-up action");
assert!(matches!(action, FollowUpTurnAction::PostTurnPendingInput));
assert!(state.pending_post_turn_input.is_empty());
assert_eq!(state.pending_input.len(), 2);
assert_eq!(state.pending_manual_compacts.len(), 1);
assert_eq!(state.pending_user_input.len(), 1);
}
#[test]
fn follow_up_turn_action_preserves_queued_user_input_after_post_turn_input() {
let mut state = State::default();
state.pending_post_turn_input.push(developer_input("post-turn"));
state.pending_user_input.push(QueuedUserInput {
submission_id: "queued-1".to_string(),
response_item: developer_input("queued"),
core_items: vec![],
});
let first = take_follow_up_turn_action(&mut state).expect("post-turn action");
assert!(matches!(first, FollowUpTurnAction::PostTurnPendingInput));
assert_eq!(state.pending_user_input.len(), 1);
state.pending_input.clear();
let second = take_follow_up_turn_action(&mut state).expect("queued user action");
assert!(matches!(second, FollowUpTurnAction::QueuedUserInput(_)));
assert!(state.pending_user_input.is_empty());
}
}
#[derive(Debug)]
@@ -933,7 +1017,7 @@ impl Session {
state.current_task = Some(agent);
}
pub async fn start_pending_only_turn_if_idle(self: &Arc<Self>) -> bool {
pub(super) async fn start_internal_pending_only_turn(self: &Arc<Self>, sentinel: &str) -> bool {
let should_start = {
let state = self.state.lock().unwrap();
state.current_task.is_none()
@@ -947,13 +1031,37 @@ impl Session {
let turn_context = self.make_turn_context();
let sub_id = self.next_internal_sub_id();
let sentinel_input = vec![InputItem::Text {
text: PENDING_ONLY_SENTINEL.to_string(),
text: sentinel.to_string(),
}];
let agent = AgentTask::spawn(Arc::clone(self), turn_context, sub_id, sentinel_input);
self.set_task(agent);
true
}
pub async fn start_pending_only_turn_if_idle(self: &Arc<Self>) -> bool {
self.start_internal_pending_only_turn(PENDING_ONLY_SENTINEL).await
}
pub async fn start_post_turn_pending_only_turn_if_idle(self: &Arc<Self>) -> bool {
let should_start = {
let mut state = self.state.lock().unwrap();
if state.current_task.is_some() || state.pending_post_turn_input.is_empty() {
false
} else {
let mut post_turn_input = std::mem::take(&mut state.pending_post_turn_input);
state.pending_input.append(&mut post_turn_input);
true
}
};
if !should_start {
return false;
}
self.start_internal_pending_only_turn(POST_TURN_PENDING_ONLY_SENTINEL)
.await
}
pub fn replace_history(&self, items: Vec<ResponseItem>) {
let mut state = self.state.lock().unwrap();
state.history.replace(items);
@@ -968,6 +1076,18 @@ impl Session {
}
}
pub fn enqueue_post_turn_item(&self, item: ResponseInputItem) -> bool {
let mut state = self.state.lock().unwrap();
let should_start_turn = state.current_task.is_none();
state.pending_post_turn_input.push(item);
should_start_turn
}
pub(super) fn take_follow_up_turn_action(&self) -> Option<FollowUpTurnAction> {
let mut state = self.state.lock().unwrap();
take_follow_up_turn_action(&mut state)
}
pub fn has_running_task(&self) -> bool {
self.state.lock().unwrap().current_task.is_some()
}
@@ -1131,15 +1251,6 @@ impl Session {
*content = new_content;
}
pub fn pop_next_queued_user_input(&self) -> Option<QueuedUserInput> {
let mut state = self.state.lock().unwrap();
if state.pending_user_input.is_empty() {
None
} else {
Some(state.pending_user_input.remove(0))
}
}
/// Enqueue a response item that should be surfaced to the model at the start of the
/// next turn. Returns `true` if no agent is currently running and a new turn should be
/// scheduled immediately.
@@ -1976,12 +2087,6 @@ impl Session {
was_empty
}
pub fn dequeue_manual_compact(&self) -> Option<String> {
let mut state = self.state.lock().unwrap();
state.pending_manual_compacts.pop_front()
}
pub fn get_pending_input(&self) -> Vec<ResponseInputItem> {
self.get_pending_input_filtered(true)
}
code-rs/core/src/codex/streaming.rs
+58 -24
View File
@@ -7,6 +7,7 @@ use super::exec::{
};
use super::session::{
BackgroundExecState,
FollowUpTurnAction,
QueuedUserInput,
State,
WaitInterruptReason,
@@ -260,6 +261,23 @@ pub(super) async fn submission_loop(
sess.set_task(agent);
}
}
Op::AddPostTurnDeveloperInput { text } => {
let sess = match sess.as_ref() {
Some(s) => s.clone(),
None => {
send_no_session_event(sub.id).await;
continue;
}
};
let dev_msg = ResponseInputItem::Message {
role: "developer".to_string(),
content: vec![ContentItem::InputText { text }],
};
let should_start_turn = sess.enqueue_post_turn_item(dev_msg);
if should_start_turn {
sess.start_post_turn_pending_only_turn_if_idle().await;
}
}
Op::ConfigureSession {
provider,
model,
@@ -2183,11 +2201,17 @@ async fn run_agent(sess: Arc<Session>, turn_context: Arc<TurnContext>, sub_id: S
let mut review_messages: Vec<String> = Vec::new();
let mut review_exit_emitted = false;
let post_turn_only_turn = input.len() == 1
&& matches!(
&input[0],
InputItem::Text { text } if text == POST_TURN_PENDING_ONLY_SENTINEL
);
let pending_only_turn = input.len() == 1
&& matches!(
&input[0],
InputItem::Text { text } if text == PENDING_ONLY_SENTINEL
);
)
|| post_turn_only_turn;
// Debug logging for ephemeral images
let ephemeral_count = input
@@ -2243,7 +2267,7 @@ async fn run_agent(sess: Arc<Session>, turn_context: Arc<TurnContext>, sub_id: S
// review model, causing loops. Only include queued user inputs when not in
// review mode. They will be picked up after TaskComplete via
// pop_next_queued_user_input.
let pending_input = if is_review_mode {
let pending_input = if is_review_mode || post_turn_only_turn {
sess.get_pending_input_filtered(false)
} else {
sess.get_pending_input()
@@ -2608,28 +2632,38 @@ async fn run_agent(sess: Arc<Session>, turn_context: Arc<TurnContext>, sub_id: S
}
sess.tx_event.send(event).await.ok();
if let Some(compact_sub_id) = sess.dequeue_manual_compact() {
let turn_context = sess.make_turn_context();
let prompt_text = sess.compact_prompt_text();
compact::spawn_compact_task(
Arc::clone(&sess),
turn_context,
compact_sub_id,
vec![InputItem::Text {
text: prompt_text,
}],
);
return;
}
if let Some(queued) = sess.pop_next_queued_user_input() {
let sess_clone = Arc::clone(&sess);
tokio::spawn(async move {
sess_clone.cleanup_old_status_items().await;
let submission_id = queued.submission_id;
let items = queued.core_items;
spawn_user_turn(sess_clone, submission_id, items, None).await;
});
if let Some(action) = sess.take_follow_up_turn_action() {
match action {
FollowUpTurnAction::PostTurnPendingInput => {
sess.start_internal_pending_only_turn(POST_TURN_PENDING_ONLY_SENTINEL)
.await;
}
FollowUpTurnAction::ManualCompact(compact_sub_id) => {
let turn_context = sess.make_turn_context();
let prompt_text = sess.compact_prompt_text();
compact::spawn_compact_task(
Arc::clone(&sess),
turn_context,
compact_sub_id,
vec![InputItem::Text {
text: prompt_text,
}],
);
}
FollowUpTurnAction::PendingInput => {
sess.start_internal_pending_only_turn(PENDING_ONLY_SENTINEL)
.await;
}
FollowUpTurnAction::QueuedUserInput(queued) => {
let sess_clone = Arc::clone(&sess);
tokio::spawn(async move {
sess_clone.cleanup_old_status_items().await;
let submission_id = queued.submission_id;
let items = queued.core_items;
spawn_user_turn(sess_clone, submission_id, items, None).await;
});
}
}
}
}
code-rs/core/src/config.rs
+1
View File
@@ -3180,6 +3180,7 @@ mod agent_merge_tests {
use super::ConfigOverrides;
use super::ConfigToml;
use crate::config_types::AgentConfig;
use crate::config_types::ContextMode;
use crate::config_types::ServiceTier;
use tempfile::TempDir;
code-rs/core/src/protocol.rs
+7
View File
@@ -270,6 +270,13 @@ pub enum Op {
text: String,
},
/// Queue a developer-role message to run in a dedicated follow-up turn
/// immediately after the current turn completes.
AddPostTurnDeveloperInput {
/// The developer message text to add to the post-turn queue.
text: String,
},
/// Request a single history entry identified by `log_id` + `offset`.
GetHistoryEntryRequest { offset: usize, log_id: u64 },
code-rs/tui/src/chatwidget.rs
+196 -266
View File
@@ -1321,12 +1321,44 @@ 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_SESSION_PATCH_SETS: usize = 256;
const MAX_PENDING_DISPATCHED_USER_MESSAGES: usize = 256;
const MAX_PENDING_AUTO_REVIEW_NOTES: usize = 32;
const MAX_PENDING_AUTO_REVIEW_NOTE_CHARS: usize = 2400;
const MAX_AUTO_REVIEW_NOTE_CONTEXT_CHARS: usize = 1200;
const MAX_AUTO_REVIEW_ERROR_SUMMARY_CHARS: usize = 320;
fn compact_patch_change(
change: &code_core::protocol::FileChange,
) -> code_core::protocol::FileChange {
match change {
code_core::protocol::FileChange::Add { .. } => {
code_core::protocol::FileChange::Add {
content: String::new(),
}
}
code_core::protocol::FileChange::Delete => code_core::protocol::FileChange::Delete,
code_core::protocol::FileChange::Update {
unified_diff,
move_path,
..
} => code_core::protocol::FileChange::Update {
unified_diff: unified_diff.clone(),
move_path: move_path.clone(),
original_content: String::new(),
new_content: String::new(),
},
}
}
fn compact_patch_changes(
changes: &HashMap<PathBuf, code_core::protocol::FileChange>,
) -> HashMap<PathBuf, code_core::protocol::FileChange> {
changes
.iter()
.map(|(path, change)| (path.clone(), compact_patch_change(change)))
.collect()
}
#[derive(Clone, Copy)]
struct AutoReviewFailureClassification {
failure_class: &'static str,
@@ -1808,11 +1840,8 @@ pub(crate) struct ChatWidget<'a> {
/// When true, render without the top status bar and HUD so the normal
/// terminal scrollback remains usable (Ctrl+T standard terminal mode).
pub(crate) standard_terminal_mode: bool,
// Pending system notes to inject into the agent's conversation history
// before the next user turn. Each entry is sent in order ahead of the
// user's visible prompt.
// Pending system notes to append to persistent history.
pending_agent_notes: Vec<String>,
pending_auto_review_notes: VecDeque<String>,
// Stable synthetic request bucket for pre‑turn system notices (set on first use)
synthetic_system_req: Option<u64>,
@@ -5317,54 +5346,17 @@ impl ChatWidget<'_> {
grant_root,
} = ev;
// Clone for session storage before moving into history
let changes_clone = changes.clone();
let history_changes = compact_patch_changes(&changes);
// Surface the patch summary in the main conversation
let key = self.next_internal_key();
let _ = self.history_insert_with_key_global(
Box::new(history_cell::new_patch_event(
history_cell::PatchEventType::ApprovalRequest,
changes,
history_changes,
)),
key,
);
// Record change set for session diff popup (latest last)
self.diffs.session_patch_sets.push(changes_clone);
// For any new paths, capture an original baseline snapshot the first time we see them
if let Some(last) = self.diffs.session_patch_sets.last() {
for (src_path, chg) in last.iter() {
match chg {
code_core::protocol::FileChange::Update {
move_path: Some(dest_path),
..
} => {
if let Some(baseline) =
self.diffs.baseline_file_contents.get(src_path).cloned()
{
// Mirror baseline under destination so tabs use the new path
self.diffs
.baseline_file_contents
.entry(dest_path.clone())
.or_insert(baseline);
} else if !self.diffs.baseline_file_contents.contains_key(dest_path) {
// Snapshot from source (pre-apply)
let baseline = std::fs::read_to_string(src_path).unwrap_or_default();
self.diffs
.baseline_file_contents
.insert(dest_path.clone(), baseline);
}
}
_ => {
if !self.diffs.baseline_file_contents.contains_key(src_path) {
let baseline = std::fs::read_to_string(src_path).unwrap_or_default();
self.diffs
.baseline_file_contents
.insert(src_path.clone(), baseline);
}
}
}
}
}
self.diffs.record_patch_set(&changes, false);
// Enable Ctrl+D footer hint now that we have diffs to show
self.bottom_pane.set_diffs_hint(true);
@@ -6447,7 +6439,6 @@ impl ChatWidget<'_> {
self.queued_user_messages.clear();
self.bottom_pane.update_status_text(String::new());
self.pending_dispatched_user_messages.clear();
self.pending_auto_review_notes.clear();
self.refresh_queued_user_messages(false);
}
self.maybe_hide_spinner();
@@ -6757,7 +6748,6 @@ impl ChatWidget<'_> {
scroll_history_hint_shown: false,
access_status_idx: None,
pending_agent_notes: Vec::new(),
pending_auto_review_notes: VecDeque::new(),
synthetic_system_req: None,
system_cell_by_id: HashMap::new(),
ui_background_seq_counters: HashMap::new(),
@@ -7144,7 +7134,6 @@ impl ChatWidget<'_> {
access_status_idx: None,
standard_terminal_mode: !config.tui.alternate_screen,
pending_agent_notes: Vec::new(),
pending_auto_review_notes: VecDeque::new(),
synthetic_system_req: None,
system_cell_by_id: HashMap::new(),
ui_background_seq_counters: HashMap::new(),
@@ -12006,11 +11995,6 @@ impl ChatWidget<'_> {
combined_items.extend(message.ordered_items.clone());
}
let pending_review_notes = self.drain_pending_auto_review_notes();
if let Some(note_item) = Self::pending_auto_review_note_item(&pending_review_notes) {
combined_items.insert(0, note_item);
}
let total_items = combined_items.len();
let ephemeral_count = combined_items
.iter()
@@ -12037,10 +12021,7 @@ impl ChatWidget<'_> {
})
{
tracing::error!("failed to send Op::UserInput: {e}");
self.restore_pending_auto_review_notes_front(pending_review_notes);
}
} else {
self.restore_pending_auto_review_notes_front(pending_review_notes);
}
for (idx, message) in messages.into_iter().enumerate() {
@@ -12055,11 +12036,7 @@ impl ChatWidget<'_> {
fn dispatch_queued_user_message_now(&mut self, message: UserMessage) {
let message = self.take_queued_user_message(&message).unwrap_or(message);
let pending_review_notes = self.drain_pending_auto_review_notes();
let mut items = message.ordered_items.clone();
if let Some(note_item) = Self::pending_auto_review_note_item(&pending_review_notes) {
items.insert(0, note_item);
}
let items = message.ordered_items.clone();
let dispatched_text = Self::combined_input_text(&items);
tracing::info!(
"[queue] Dispatching single queued message via coordinator (queue_remaining={})",
@@ -12071,7 +12048,6 @@ impl ChatWidget<'_> {
}
Err(err) => {
tracing::error!("failed to send QueueUserInput op: {err}");
self.restore_pending_auto_review_notes_front(pending_review_notes);
self.queued_user_messages.push_front(message);
self.refresh_queued_user_messages(true);
}
@@ -12083,11 +12059,6 @@ impl ChatWidget<'_> {
return;
}
let pending_review_notes = self.drain_pending_auto_review_notes();
let mut note_item = Self::pending_auto_review_note_item(&pending_review_notes);
let mut note_was_attached = false;
let mut should_restore_notes = false;
tracing::info!(
"[queue] Draining batch via coordinator path (batch_size={}, auto_active={})",
batch.len(),
@@ -12100,11 +12071,7 @@ impl ChatWidget<'_> {
continue;
};
let mut items = message.ordered_items.clone();
if let Some(item) = note_item.take() {
items.insert(0, item);
note_was_attached = true;
}
let items = message.ordered_items.clone();
let dispatched_text = Self::combined_input_text(&items);
match self.code_op_tx.send(Op::QueueUserInput { items }) {
Ok(()) => {
@@ -12112,19 +12079,12 @@ impl ChatWidget<'_> {
}
Err(err) => {
tracing::error!("[queue] Failed to send QueueUserInput op: {err}");
if note_was_attached {
should_restore_notes = true;
}
self.queued_user_messages.push_front(message);
self.refresh_queued_user_messages(true);
break;
}
}
}
if should_restore_notes || !note_was_attached {
self.restore_pending_auto_review_notes_front(pending_review_notes);
}
}
fn take_queued_user_message(&mut self, target: &UserMessage) -> Option<UserMessage> {
@@ -13216,7 +13176,6 @@ impl ChatWidget<'_> {
self.pending_dispatched_user_messages.clear();
self.pending_user_prompts_for_next_turn = 0;
self.queued_user_messages.clear();
self.pending_auto_review_notes.clear();
self.refresh_queued_user_messages(false);
self.bottom_pane.clear_composer();
self.bottom_pane.clear_ctrl_c_quit_hint();
@@ -13236,65 +13195,6 @@ impl ChatWidget<'_> {
}
}
fn queue_pending_auto_review_note(&mut self, note: String) {
let trimmed = note.trim();
if trimmed.is_empty() {
return;
}
let bounded_note = Self::truncate_with_ellipsis(trimmed, MAX_PENDING_AUTO_REVIEW_NOTE_CHARS);
if self
.pending_auto_review_notes
.back()
.is_some_and(|existing| existing == &bounded_note)
{
return;
}
self.pending_auto_review_notes.push_back(bounded_note.clone());
while self.pending_auto_review_notes.len() > MAX_PENDING_AUTO_REVIEW_NOTES {
self.pending_auto_review_notes.pop_front();
}
let cleaned = Self::strip_context_sections(&bounded_note);
if !cleaned.trim().is_empty() {
self.last_developer_message =
Some(Self::truncate_with_ellipsis(&cleaned, MAX_AUTO_REVIEW_NOTE_CONTEXT_CHARS));
}
}
fn drain_pending_auto_review_notes(&mut self) -> Vec<String> {
self.pending_auto_review_notes.drain(..).collect()
}
fn restore_pending_auto_review_notes_front(&mut self, notes: Vec<String>) {
if notes.is_empty() {
return;
}
for note in notes.into_iter().rev() {
self.pending_auto_review_notes.push_front(note);
}
}
fn pending_auto_review_note_item(notes: &[String]) -> Option<InputItem> {
if notes.is_empty() {
return None;
}
let combined = notes
.iter()
.map(|note| note.trim())
.filter(|note| !note.is_empty())
.collect::<Vec<_>>()
.join("\n\n");
if combined.is_empty() {
return None;
}
Some(InputItem::Text { text: combined })
}
fn combined_input_text(items: &[InputItem]) -> Option<String> {
let mut buffer = String::new();
for item in items {
@@ -14495,45 +14395,7 @@ impl ChatWidget<'_> {
}) => {
let exec_call_id = ExecCallId(call_id.clone());
self.exec.suppress_exec_end(exec_call_id);
// Store for session diff popup (clone before moving into history)
self.diffs.session_patch_sets.push(changes.clone());
// Capture/adjust baselines, including rename moves
if let Some(last) = self.diffs.session_patch_sets.last() {
for (src_path, chg) in last.iter() {
match chg {
code_core::protocol::FileChange::Update {
move_path: Some(dest_path),
..
} => {
// Prefer to carry forward existing baseline from src to dest.
if let Some(baseline) =
self.diffs.baseline_file_contents.remove(src_path)
{
self.diffs
.baseline_file_contents
.insert(dest_path.clone(), baseline);
} else if !self.diffs.baseline_file_contents.contains_key(dest_path)
{
// Fallback: snapshot current contents of src (pre-apply) under dest key.
let baseline =
std::fs::read_to_string(src_path).unwrap_or_default();
self.diffs
.baseline_file_contents
.insert(dest_path.clone(), baseline);
}
}
_ => {
if !self.diffs.baseline_file_contents.contains_key(src_path) {
let baseline =
std::fs::read_to_string(src_path).unwrap_or_default();
self.diffs
.baseline_file_contents
.insert(src_path.clone(), baseline);
}
}
}
}
}
self.diffs.record_patch_set(&changes, true);
// Enable Ctrl+D footer hint now that we have diffs to show
self.bottom_pane.set_diffs_hint(true);
// Strict order
@@ -14544,9 +14406,10 @@ impl ChatWidget<'_> {
self.next_internal_key()
}
};
let history_changes = compact_patch_changes(&changes);
let cell = history_cell::new_patch_event(
PatchEventType::ApplyBegin { auto_approved },
changes,
history_changes,
);
let _ = self.history_insert_with_key_global(Box::new(cell), ok);
}
@@ -30329,6 +30192,7 @@ use code_core::protocol::OrderMeta;
ExecCommandBeginEvent,
McpServerFailure,
McpServerFailurePhase,
Op,
TaskCompleteEvent,
};
use code_core::protocol::AgentInfo as CoreAgentInfo;
@@ -30337,6 +30201,20 @@ use code_core::protocol::OrderMeta;
use tempfile::tempdir;
use std::sync::Arc;
use std::path::PathBuf;
use tokio::sync::mpsc::{UnboundedReceiver, unbounded_channel};
fn replace_code_op_channel(chat: &mut ChatWidget<'static>) -> UnboundedReceiver<Op> {
let (code_op_tx, code_op_rx) = unbounded_channel::<Op>();
chat.code_op_tx = code_op_tx;
code_op_rx
}
fn expect_post_turn_developer_input(code_op_rx: &mut UnboundedReceiver<Op>) -> String {
match code_op_rx.try_recv().expect("post-turn developer op") {
Op::AddPostTurnDeveloperInput { text } => text,
other => panic!("expected AddPostTurnDeveloperInput, got {other:?}"),
}
}
#[test]
fn parse_agent_review_result_json_clean() {
@@ -30939,6 +30817,7 @@ use code_core::protocol::OrderMeta;
let _rt = enter_test_runtime_guard();
let mut harness = ChatWidgetHarness::new();
let chat = harness.chat();
let mut code_op_rx = replace_code_op_channel(chat);
chat.background_review = Some(BackgroundReviewState {
worktree_path: PathBuf::from("/tmp/wt"),
@@ -30960,30 +30839,54 @@ use code_core::protocol::OrderMeta;
Some("ghost123".to_string()),
);
assert_eq!(
chat.pending_auto_review_notes.len(),
1,
"idle auto-review findings should queue a hidden follow-up note"
);
let note = expect_post_turn_developer_input(&mut code_op_rx);
assert!(
chat.pending_dispatched_user_messages.is_empty(),
"auto review notes should not start a synthetic background turn"
);
chat.submit_text_message("follow up".to_string());
assert!(
chat.pending_dispatched_user_messages
.iter()
.any(|msg| msg.contains("Background auto-review completed and reported 1 issue(s)")),
"queued auto review note should be injected on the next user turn"
note.contains("Background auto-review completed and reported 1 issue(s)"),
"idle auto-review findings should be sent to core immediately"
);
}
#[test]
fn background_review_clean_dispatches_post_turn_developer_input() {
let _rt = enter_test_runtime_guard();
let mut harness = ChatWidgetHarness::new();
let chat = harness.chat();
let mut code_op_rx = replace_code_op_channel(chat);
chat.background_review = Some(BackgroundReviewState {
worktree_path: PathBuf::from("/tmp/wt"),
branch: "auto-review-branch".to_string(),
agent_id: Some("agent-clean".to_string()),
snapshot: Some("ghost-clean".to_string()),
base: None,
last_seen: Instant::now(),
});
chat.on_background_review_finished(
PathBuf::from("/tmp/wt"),
"auto-review-branch".to_string(),
false,
0,
Some("looks clean".to_string()),
None,
Some("agent-clean".to_string()),
Some("ghost-clean".to_string()),
);
let note = expect_post_turn_developer_input(&mut code_op_rx);
assert!(note.contains("Background auto-review completed with no findings."));
assert!(note.contains("ghost-clean"));
}
#[test]
fn background_review_failure_note_is_sanitized_and_bounded() {
let mut harness = ChatWidgetHarness::new();
let chat = harness.chat();
let mut code_op_rx = replace_code_op_channel(chat);
chat.background_review = Some(BackgroundReviewState {
worktree_path: PathBuf::from("/tmp/wt"),
@@ -31017,10 +30920,7 @@ use code_core::protocol::OrderMeta;
Some("ghost123".to_string()),
);
let note = chat
.pending_auto_review_notes
.front()
.expect("sanitized failure note should be queued");
let note = expect_post_turn_developer_input(&mut code_op_rx);
assert!(note.contains("Background auto-review failed."));
assert!(note.contains("Failure class: context_budget_exhausted"));
@@ -31049,6 +30949,7 @@ use code_core::protocol::OrderMeta;
fn auto_review_failure_note_classifies_os_error_2_with_path_context_as_missing_path() {
let mut harness = ChatWidgetHarness::new();
let chat = harness.chat();
let mut code_op_rx = replace_code_op_channel(chat);
chat.background_review = Some(BackgroundReviewState {
worktree_path: PathBuf::from("/tmp/wt"),
@@ -31071,10 +30972,7 @@ use code_core::protocol::OrderMeta;
Some("ghost123".to_string()),
);
let note = chat
.pending_auto_review_notes
.front()
.expect("sanitized failure note should be queued");
let note = expect_post_turn_developer_input(&mut code_op_rx);
assert!(note.contains("Failure class: exec_missing_path"));
assert!(note.contains("Phase: exec_run"));
assert!(note.contains("Retryable: false"));
@@ -31085,6 +30983,7 @@ use code_core::protocol::OrderMeta;
fn auto_review_failure_note_keeps_generic_os_error_2_as_unknown() {
let mut harness = ChatWidgetHarness::new();
let chat = harness.chat();
let mut code_op_rx = replace_code_op_channel(chat);
chat.background_review = Some(BackgroundReviewState {
worktree_path: PathBuf::from("/tmp/wt"),
@@ -31107,10 +31006,7 @@ use code_core::protocol::OrderMeta;
Some("ghost123".to_string()),
);
let note = chat
.pending_auto_review_notes
.front()
.expect("sanitized failure note should be queued");
let note = expect_post_turn_developer_input(&mut code_op_rx);
assert!(note.contains("Failure class: unknown"));
assert!(note.contains("Retryable: true"));
}
@@ -31119,6 +31015,7 @@ use code_core::protocol::OrderMeta;
fn auto_review_failure_note_uses_placeholders_for_missing_worktree_context() {
let mut harness = ChatWidgetHarness::new();
let chat = harness.chat();
let mut code_op_rx = replace_code_op_channel(chat);
chat.background_review = Some(BackgroundReviewState {
worktree_path: PathBuf::new(),
@@ -31141,10 +31038,7 @@ use code_core::protocol::OrderMeta;
None,
);
let note = chat
.pending_auto_review_notes
.front()
.expect("sanitized failure note should be queued");
let note = expect_post_turn_developer_input(&mut code_op_rx);
assert!(note.contains("Failure class: worktree_prepare_failed"));
assert!(note.contains("Phase: worktree_prepare"));
assert!(note.contains("Worktree: '(unknown)'"));
@@ -31196,6 +31090,7 @@ use code_core::protocol::OrderMeta;
fn background_review_busy_path_enqueues_developer_note_with_merge_hint() {
let mut harness = ChatWidgetHarness::new();
let chat = harness.chat();
let mut code_op_rx = replace_code_op_channel(chat);
chat.config.tui.auto_review_enabled = true;
chat.bottom_pane.set_task_running(true); // simulate busy state so note is queued
@@ -31233,6 +31128,8 @@ use code_core::protocol::OrderMeta;
assert!(chat.pending_agent_notes.is_empty());
assert!(chat.pending_dispatched_user_messages.is_empty());
let note = expect_post_turn_developer_input(&mut code_op_rx);
assert!(note.contains("Merge the worktree 'auto-review-branch'"));
let developer_seen = chat.history_cells.iter().any(|cell| {
cell.display_lines_trimmed().iter().any(|line| {
line.spans.iter().any(|span| {
@@ -31253,6 +31150,7 @@ use code_core::protocol::OrderMeta;
let _rt = enter_test_runtime_guard();
let mut harness = ChatWidgetHarness::new();
let chat = harness.chat();
let mut code_op_rx = replace_code_op_channel(chat);
chat.config.tui.auto_review_enabled = true;
chat.background_review = Some(BackgroundReviewState {
@@ -31290,15 +31188,12 @@ use code_core::protocol::OrderMeta;
chat.observe_auto_review_status(&[agent]);
assert_eq!(
chat.pending_auto_review_notes.len(),
1,
"idle observe path should queue a hidden follow-up note"
);
let note = expect_post_turn_developer_input(&mut code_op_rx);
assert!(
chat.pending_dispatched_user_messages.is_empty(),
"idle observe path should not auto-start a hidden follow-up turn"
);
assert!(note.contains("Background auto-review completed and reported 1 issue(s)"));
let developer_seen = chat.history_cells.iter().any(|cell| {
cell.display_lines_trimmed().iter().any(|line| {
line.spans.iter().any(|span| {
@@ -31386,6 +31281,7 @@ use code_core::protocol::OrderMeta;
let _rt = enter_test_runtime_guard();
let mut harness = ChatWidgetHarness::new();
let chat = harness.chat();
let mut code_op_rx = replace_code_op_channel(chat);
chat.auto_state.set_phase(AutoRunPhase::Active);
chat.auto_state.goal = Some("keep running".to_string());
@@ -31420,27 +31316,16 @@ use code_core::protocol::OrderMeta;
assert_eq!(
pending_after_review,
pending_before_review,
"background review completion should queue notes without starting a hidden turn"
"background review completion should not start a hidden turn"
);
let note = expect_post_turn_developer_input(&mut code_op_rx);
assert!(
!chat.pending_auto_review_notes.is_empty(),
"churn path should retain queued auto-review notes"
);
assert!(
chat.pending_dispatched_user_messages
.iter()
.all(|msg| !msg.contains("Background auto-review completed and reported 1 issue(s)")),
"the auto-review findings should remain queued until the next user turn"
note.contains("Background auto-review completed and reported 1 issue(s)"),
"auto-review findings should be submitted to core immediately after completion"
);
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("Background auto-review completed and reported 1 issue(s)")),
"queued auto-review findings should be injected on the next user turn"
);
assert!(
chat.pending_dispatched_user_messages
.iter()
@@ -31455,6 +31340,7 @@ use code_core::protocol::OrderMeta;
let _rt = enter_test_runtime_guard();
let mut harness = ChatWidgetHarness::new();
let chat = harness.chat();
let mut code_op_rx = replace_code_op_channel(chat);
chat.config.tui.auto_review_enabled = true;
chat.auto_state.set_phase(AutoRunPhase::Active);
@@ -31557,20 +31443,12 @@ use code_core::protocol::OrderMeta;
);
assert!(
chat.pending_dispatched_user_messages.len() == pending_before_review,
"idle review completion should queue notes without creating a synthetic turn"
);
assert!(
!chat.pending_auto_review_notes.is_empty(),
"churn path should retain queued auto-review notes"
"idle review completion should not create a synthetic turn"
);
let note = expect_post_turn_developer_input(&mut code_op_rx);
assert!(note.contains("Background auto-review completed and reported 1 issue(s)"));
chat.submit_text_message("typing after churn".to_string());
assert!(
chat.pending_dispatched_user_messages
.iter()
.any(|msg| msg.contains("Background auto-review completed and reported 1 issue(s)")),
"churn path should still forward queued auto-review findings"
);
assert!(
chat.pending_dispatched_user_messages
.iter()
@@ -36757,8 +36635,6 @@ impl ChatWidget<'_> {
findings
};
let was_task_running = self.is_task_running();
let inflight_worktree_path = self
.background_review
.as_ref()
@@ -36783,7 +36659,6 @@ impl ChatWidget<'_> {
self.background_review = None;
self.background_review_guard = None;
release_background_lock(&agent_id);
let mut developer_note: Option<String> = None;
let snapshot_note = inflight_snapshot
.as_deref()
.map(str::trim)
@@ -36826,18 +36701,20 @@ impl ChatWidget<'_> {
resolved_worktree_path.display().to_string()
};
let errored = error.is_some();
let indicator_status = if let Some(err) = error {
let (indicator_status, developer_note) = if let Some(err) = error {
let summarized_error = Self::summarize_auto_review_error(&err);
let classification = Self::classify_auto_review_error(&err);
developer_note = Some(format!(
"[developer] Background auto-review failed.\n\nThis auto-review ran in an isolated git worktree and did not modify your current workspace.\n\nWorktree: '{worktree_label}'\nWorktree path: {worktree_path_label}\n{snapshot_note}\n{agent_note}\nFailure class: {}\nPhase: {}\nRetryable: {}\nHint: {}\nError: {err}",
classification.failure_class,
classification.phase,
classification.retryable,
classification.remediation_hint,
err = summarized_error,
));
AutoReviewIndicatorStatus::Failed
(
AutoReviewIndicatorStatus::Failed,
Some(format!(
"[developer] Background auto-review failed.\n\nThis auto-review ran in an isolated git worktree and did not modify your current workspace.\n\nWorktree: '{worktree_label}'\nWorktree path: {worktree_path_label}\n{snapshot_note}\n{agent_note}\nFailure class: {}\nPhase: {}\nRetryable: {}\nHint: {}\nError: {err}",
classification.failure_class,
classification.phase,
classification.retryable,
classification.remediation_hint,
err = summarized_error,
)),
)
} else if has_findings {
let mut note = format!(
"[developer] Background auto-review completed and reported {effective_findings} issue(s).\n\nA separate LLM ran /review (and may have run auto-resolve) in an isolated git worktree. Any proposed fixes live only in that worktree until you merge them.\n\nNext: Decide if the findings are genuine. If yes, Merge the worktree '{worktree_label}' to apply the changes (or cherry-pick selectively). If not, do not merge.\n\nWorktree path: {worktree_path_label}\n{snapshot_note}\n{agent_note}",
@@ -36846,10 +36723,14 @@ impl ChatWidget<'_> {
note.push('\n');
note.push_str(&summary_note);
}
developer_note = Some(note);
AutoReviewIndicatorStatus::Fixed
(AutoReviewIndicatorStatus::Fixed, Some(note))
} else {
AutoReviewIndicatorStatus::Clean
(
AutoReviewIndicatorStatus::Clean,
Some(format!(
"[developer] Background auto-review completed with no findings.\n\nA separate LLM ran /review in an isolated git worktree and did not report actionable issues.\n\nWorktree path: {worktree_path_label}\n{snapshot_note}\n{agent_note}"
)),
)
};
let findings_for_indicator =
@@ -36869,7 +36750,7 @@ impl ChatWidget<'_> {
}
if let Some(note) = developer_note {
self.record_background_review_note(note, !was_task_running);
self.record_background_review_note(note);
}
// Auto review completion should never leave the composer spinner active.
@@ -36894,22 +36775,25 @@ impl ChatWidget<'_> {
self.request_redraw();
}
fn record_background_review_note(&mut self, note: String, _allow_immediate_dispatch: bool) {
fn record_background_review_note(&mut self, note: String) {
let trimmed = note.trim();
if trimmed.is_empty() {
return;
}
// The Auto Review notice cell is now the user-facing UI. Keep this
// note out of history and only use it as hidden coordinator context.
//
// Do not auto-dispatch a synthetic hidden turn here: that can start a
// fresh task while the user is actively typing and causes the next
// user message to be silently queued behind background work.
//
// Queue the note for delivery on the next real user/model turn so the
// findings are eventually integrated without blocking progress.
self.queue_pending_auto_review_note(trimmed.to_string());
let bounded_note = Self::truncate_with_ellipsis(trimmed, MAX_PENDING_AUTO_REVIEW_NOTE_CHARS);
let cleaned = Self::strip_context_sections(&bounded_note);
if !cleaned.trim().is_empty() {
self.last_developer_message =
Some(Self::truncate_with_ellipsis(&cleaned, MAX_AUTO_REVIEW_NOTE_CONTEXT_CHARS));
}
if let Err(err) = self
.code_op_tx
.send(Op::AddPostTurnDeveloperInput { text: bounded_note })
{
tracing::error!("failed to send AddPostTurnDeveloperInput op: {err}");
}
}
fn handle_auto_review_completion_state(
@@ -42663,6 +42547,52 @@ struct DiffsState {
body_visible_rows: std::cell::Cell<u16>,
}
impl DiffsState {
fn record_patch_set(
&mut self,
changes: &HashMap<PathBuf, code_core::protocol::FileChange>,
move_existing_baselines: bool,
) {
let compact_changes = compact_patch_changes(changes);
self.session_patch_sets.push(compact_changes);
if self.session_patch_sets.len() > MAX_SESSION_PATCH_SETS {
let overflow = self.session_patch_sets.len() - MAX_SESSION_PATCH_SETS;
self.session_patch_sets.drain(0..overflow);
}
for (src_path, change) in changes {
match change {
code_core::protocol::FileChange::Update {
move_path: Some(dest_path),
..
} => {
if move_existing_baselines {
if let Some(baseline) = self.baseline_file_contents.remove(src_path) {
self.baseline_file_contents.insert(dest_path.clone(), baseline);
} else if !self.baseline_file_contents.contains_key(dest_path) {
let baseline = std::fs::read_to_string(src_path).unwrap_or_default();
self.baseline_file_contents.insert(dest_path.clone(), baseline);
}
} else if let Some(baseline) = self.baseline_file_contents.get(src_path).cloned() {
self.baseline_file_contents
.entry(dest_path.clone())
.or_insert(baseline);
} else if !self.baseline_file_contents.contains_key(dest_path) {
let baseline = std::fs::read_to_string(src_path).unwrap_or_default();
self.baseline_file_contents.insert(dest_path.clone(), baseline);
}
}
_ => {
if !self.baseline_file_contents.contains_key(src_path) {
let baseline = std::fs::read_to_string(src_path).unwrap_or_default();
self.baseline_file_contents.insert(src_path.clone(), baseline);
}
}
}
}
}
}
#[derive(Default)]
struct HelpState {
overlay: Option<HelpOverlay>,
scripts/code-memory-guard.sh
Executable
+88
View File
@@ -0,0 +1,88 @@
#!/usr/bin/env bash
set -euo pipefail
INTERVAL_SEC=${INTERVAL_SEC:-5}
MATCH_UID=${MATCH_UID:-1000}
MATCH_REGEX=${MATCH_REGEX:-/code-rs/bin/code([[:space:]]|$)}
RSS_KB_LIMIT=${RSS_KB_LIMIT:-8388608}
PRESSURE_RSS_KB_LIMIT=${PRESSURE_RSS_KB_LIMIT:-4194304}
MEM_AVAILABLE_FLOOR_KB=${MEM_AVAILABLE_FLOOR_KB:-786432}
HARD_MEM_AVAILABLE_FLOOR_KB=${HARD_MEM_AVAILABLE_FLOOR_KB:-2097152}
TERM_GRACE_SEC=${TERM_GRACE_SEC:-10}
log() {
local message=$1
logger -t code-memory-guard -- "$message" || true
printf '%s %s\n' "$(date -Is)" "$message" >&2
}
mem_available_kb() {
awk '/MemAvailable:/ { print $2; exit }' /proc/meminfo
}
select_candidate() {
ps -eo pid=,uid=,rss=,args= --sort=-rss | awk \
-v guard_uid="$MATCH_UID" \
-v guard_regex="$MATCH_REGEX" '
$2 == guard_uid {
rss = $3
cmd = substr($0, index($0, $4))
if (cmd ~ guard_regex) {
print $1, rss, cmd
exit
}
}
'
}
terminate_pid() {
local pid=$1
local reason=$2
local rss_kb=$3
local cmd=$4
log "stopping pid=${pid} rss_kb=${rss_kb} reason=${reason} cmd=${cmd}"
kill -TERM "$pid" 2>/dev/null || return 0
local deadline=$((SECONDS + TERM_GRACE_SEC))
while kill -0 "$pid" 2>/dev/null; do
if (( SECONDS >= deadline )); then
break
fi
sleep 1
done
if kill -0 "$pid" 2>/dev/null; then
log "pid=${pid} survived SIGTERM; sending SIGKILL"
kill -KILL "$pid" 2>/dev/null || true
fi
}
main() {
log "starting guard uid=${MATCH_UID} rss_limit_kb=${RSS_KB_LIMIT} pressure_limit_kb=${PRESSURE_RSS_KB_LIMIT} mem_floor_kb=${MEM_AVAILABLE_FLOOR_KB} hard_floor_kb=${HARD_MEM_AVAILABLE_FLOOR_KB}"
while true; do
local available_kb
available_kb=$(mem_available_kb)
local candidate
candidate=$(select_candidate || true)
if [[ -n "$candidate" ]]; then
local pid rss_kb cmd
pid=${candidate%% *}
local rest=${candidate#* }
rss_kb=${rest%% *}
cmd=${rest#* }
if (( rss_kb >= RSS_KB_LIMIT && available_kb <= HARD_MEM_AVAILABLE_FLOOR_KB )); then
terminate_pid "$pid" "rss_limit_exceeded available_kb=${available_kb}" "$rss_kb" "$cmd"
elif (( available_kb <= MEM_AVAILABLE_FLOOR_KB && rss_kb >= PRESSURE_RSS_KB_LIMIT )); then
terminate_pid "$pid" "low_mem_available_kb=${available_kb}" "$rss_kb" "$cmd"
fi
fi
sleep "$INTERVAL_SEC"
done
}
main "$@"
scripts/install-code-memory-guard.sh
Executable
+18
View File
@@ -0,0 +1,18 @@
#!/usr/bin/env bash
set -euo pipefail
SCRIPT_DIR=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)
REPO_ROOT=$(cd -- "${SCRIPT_DIR}/.." && pwd)
sudo install -Dm755 \
"${REPO_ROOT}/scripts/code-memory-guard.sh" \
/usr/local/bin/code-memory-guard.sh
sudo install -Dm644 \
"${REPO_ROOT}/systemd/code-memory-guard.service" \
/etc/systemd/system/code-memory-guard.service
sudo systemctl daemon-reload
sudo systemctl enable code-memory-guard.service
sudo systemctl restart code-memory-guard.service
sudo systemctl --no-pager --full status code-memory-guard.service
systemd/code-memory-guard.service
+21
View File
@@ -0,0 +1,21 @@
[Unit]
Description=Guard Every Code memory usage
After=network.target
StartLimitIntervalSec=0
[Service]
Type=simple
ExecStart=/usr/local/bin/code-memory-guard.sh
Restart=always
RestartSec=2
Nice=10
Environment=MATCH_UID=1000
Environment=RSS_KB_LIMIT=8388608
Environment=PRESSURE_RSS_KB_LIMIT=4194304
Environment=MEM_AVAILABLE_FLOOR_KB=786432
Environment=HARD_MEM_AVAILABLE_FLOOR_KB=2097152
Environment=INTERVAL_SEC=5
Environment=TERM_GRACE_SEC=10
[Install]
WantedBy=multi-user.target