openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-18 17:18:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Bound TUI thread replay buffers by delta size (#39081)

Browse Source 
## Why

The per-thread replay buffer limited its event count, but streamed agent-message
deltas could still retain an unbounded amount of text while a thread was
inactive.

## What changed

- Coalesce adjacent deltas for the same thread, turn, and item into chunks of up
  to 4 KiB.
- Limit buffered agent-message deltas to 256 KiB, discard a single delta larger
  than that limit, and evict the oldest events when either buffer limit is
  exceeded.
- Keep pending input and approval state in sync when eviction removes the
  corresponding request, including clearing side-thread status indicators.

## Testing

Added coverage for delta coalescing, byte-based eviction, oversized deltas,
replaying only retained text, and side-thread status updates after eviction.

GitOrigin-RevId: 88d85353a124467d467822d2fd9854312b68d547
This commit is contained in:
Tamir Duberstein
2026-08-17 21:03:06 +00:00
committed by copyberry
parent 8ef139667f
commit 9c099e94a2
7 changed files with 496 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/tui/src/app.rs
+1
View File
@@ -229,6 +229,7 @@ mod session_lifecycle;
mod side;
mod startup;
mod startup_prompts;
mod thread_event_buffer;
mod thread_events;
mod thread_goal_actions;
mod thread_routing;
codex-rs/tui/src/app/background_requests.rs
+1 -11
View File
@@ -653,17 +653,7 @@ impl App {
let should_send = {
let mut guard = store.lock().await;
guard
.buffer
.push_back(ThreadBufferedEvent::FeedbackSubmission(event.clone()));
if guard.buffer.len() > guard.capacity
&& let Some(removed) = guard.buffer.pop_front()
&& let ThreadBufferedEvent::Request(request) = &removed
{
guard
.pending_interactive_replay
.note_evicted_server_request(request.as_ref());
}
guard.push_buffered_event(ThreadBufferedEvent::FeedbackSubmission(event.clone()));
guard.active
};
codex-rs/tui/src/app/tests.rs
+183
View File
@@ -743,6 +743,81 @@ async fn active_history_batch_is_delivered_without_replay_buffering() -> Result<
Ok(())
}
#[tokio::test]
async fn replay_thread_snapshot_renders_only_retained_agent_message_deltas() {
let (mut app, mut app_event_rx, _op_rx) = make_test_app_with_channels().await;
let thread_id = ThreadId::new();
let session = test_thread_session(thread_id, test_path_buf("/tmp/project"));
app.thread_event_channels.insert(
thread_id,
ThreadEventChannel::new_with_session(
THREAD_EVENT_CHANNEL_CAPACITY,
session.clone(),
Vec::new(),
),
);
app.activate_thread_channel(thread_id).await;
app.chat_widget.handle_thread_session(session);
{
let channel = app
.thread_event_channels
.get(&thread_id)
.expect("thread channel should exist");
let mut store = channel.store.lock().await;
for chunk in 0..65 {
let marker = format!("chunk {chunk:02}");
let text = format!("{marker}\n{}\n", "x".repeat(4096 - marker.len() - 2));
store.push_notification(agent_message_delta_notification(
thread_id,
"turn-budget",
"item-budget",
&text,
));
}
}
app.store_active_thread_receiver().await;
let (_receiver, snapshot) = app
.activate_thread_for_replay(thread_id)
.await
.expect("detached thread should reactivate for replay");
app.replay_thread_snapshot(snapshot, /*resume_restored_queue*/ false);
let mut rendered = Vec::new();
loop {
app.chat_widget.on_commit_tick();
let mut inserted = false;
while let Ok(event) = app_event_rx.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = event {
rendered.push(lines_to_single_string(&cell.display_lines(/*width*/ 80)));
inserted = true;
}
}
if !inserted {
break;
}
}
if let Some(lines) = app.chat_widget.active_cell_transcript_lines(/*width*/ 80) {
rendered.push(lines_to_single_string(&lines));
}
let markers = rendered
.iter()
.flat_map(|text| text.lines())
.filter_map(|line| line.split_once("chunk "))
.map(|(_, marker)| format!("chunk {}", marker.trim()))
.collect::<Vec<_>>();
assert_eq!(markers.len(), 64);
assert_snapshot!(
format!("{}\n{}", markers.first().expect("first marker"), markers.last().expect("last marker")),
@r"
chunk 01
chunk 64
"
);
}
#[tokio::test]
async fn replay_thread_snapshot_restores_draft_and_queued_input() {
let mut app = make_test_app().await;
@@ -4111,6 +4186,114 @@ async fn side_parent_status_prioritizes_input_over_approval() -> Result<()> {
None
);
app.enqueue_thread_request(
parent_thread_id,
request_user_input_request(parent_thread_id, "turn-eviction", "input-eviction"),
)
.await?;
let chunk = "x".repeat(4 * 1024);
app.enqueue_thread_notification(
parent_thread_id,
agent_message_delta_notification(
parent_thread_id,
"turn-eviction",
"item-eviction",
&chunk,
),
)
.await?;
app.enqueue_thread_request(
parent_thread_id,
exec_approval_request(
parent_thread_id,
"turn-eviction",
"approval-eviction",
/*approval_id*/ None,
),
)
.await?;
let side_footer = |app: &App| {
render_bottom_popup(&app.chat_widget, /*width*/ 120)
.lines()
.find_map(|line| {
line.find("Side from main thread")
.map(|start| line[start..].trim().to_string())
})
.expect("side conversation footer should be rendered")
};
assert_eq!(
app.side_threads
.get(&side_thread_id)
.and_then(|state| state.parent_status),
Some(SideParentStatus::NeedsInput)
);
let input_footer = side_footer(&app);
for _ in 0..64 {
app.enqueue_thread_notification(
parent_thread_id,
agent_message_delta_notification(
parent_thread_id,
"turn-eviction",
"item-eviction",
&chunk,
),
)
.await?;
}
assert_eq!(
app.side_threads
.get(&side_thread_id)
.and_then(|state| state.parent_status),
Some(SideParentStatus::NeedsApproval)
);
let approval_footer = side_footer(&app);
assert!(
app.thread_event_channels
.get(&parent_thread_id)
.expect("parent thread channel should exist")
.store
.lock()
.await
.has_pending_thread_approvals()
);
app.enqueue_thread_notification(
parent_thread_id,
agent_message_delta_notification(
parent_thread_id,
"turn-eviction",
"item-eviction",
&chunk,
),
)
.await?;
assert_eq!(
app.side_threads
.get(&side_thread_id)
.and_then(|state| state.parent_status),
None
);
let cleared_footer = side_footer(&app);
assert!(
!app.thread_event_channels
.get(&parent_thread_id)
.expect("parent thread channel should exist")
.store
.lock()
.await
.has_pending_thread_approvals()
);
assert_snapshot!(
format!("{input_footer}\n{approval_footer}\n{cleared_footer}"),
@r"
Side from main thread · main needs input · ctrl + / to switch · ctrl + c to close
Side from main thread · main needs approval · ctrl + / to switch · ctrl + c to close
Side from main thread · ctrl + / to switch · ctrl + c to close
"
);
Ok(())
}
codex-rs/tui/src/app/thread_event_buffer.rs
+81
View File
@@ -0,0 +1,81 @@
//! Bounded replay-buffer policy for per-thread app-server events.
use super::ServerNotification;
use super::ThreadBufferedEvent;
use super::ThreadEventStore;
use std::borrow::Cow;
// Keep merged text finite so continued streaming still reaches bounded replay eviction.
const MAX_COALESCED_AGENT_MESSAGE_DELTA_BYTES: usize = 4 * 1024;
const MAX_BUFFERED_AGENT_MESSAGE_DELTA_BYTES: usize = 256 * 1024;
impl ThreadEventStore {
pub(super) fn push_replay_notification(&mut self, notification: Cow<'_, ServerNotification>) {
if let ServerNotification::AgentMessageDelta(delta) = notification.as_ref()
&& delta.delta.len() > MAX_BUFFERED_AGENT_MESSAGE_DELTA_BYTES
{
return;
}
if let ServerNotification::AgentMessageDelta(delta) = notification.as_ref()
&& let Some(ThreadBufferedEvent::Notification(previous)) = self.buffer.back_mut()
&& let ServerNotification::AgentMessageDelta(previous) = previous.as_mut()
&& previous.thread_id == delta.thread_id
&& previous.turn_id == delta.turn_id
&& previous.item_id == delta.item_id
&& previous.delta.len().saturating_add(delta.delta.len())
<= MAX_COALESCED_AGENT_MESSAGE_DELTA_BYTES
{
previous.delta.push_str(&delta.delta);
self.buffered_agent_message_delta_bytes = self
.buffered_agent_message_delta_bytes
.saturating_add(delta.delta.len());
self.evict_overflowing_events();
return;
}
self.push_buffered_event(ThreadBufferedEvent::Notification(Box::new(
notification.into_owned(),
)));
}
pub(super) fn push_buffered_event(&mut self, event: ThreadBufferedEvent) {
if let ThreadBufferedEvent::Notification(notification) = &event
&& let ServerNotification::AgentMessageDelta(delta) = notification.as_ref()
{
self.buffered_agent_message_delta_bytes = self
.buffered_agent_message_delta_bytes
.saturating_add(delta.delta.len());
}
self.buffer.push_back(event);
self.evict_overflowing_events();
}
fn evict_overflowing_events(&mut self) {
while self.buffer.len() > self.capacity
|| self.buffered_agent_message_delta_bytes > MAX_BUFFERED_AGENT_MESSAGE_DELTA_BYTES
{
let Some(removed) = self.buffer.pop_front() else {
break;
};
match removed {
ThreadBufferedEvent::Notification(notification) => {
if let ServerNotification::AgentMessageDelta(delta) = notification.as_ref() {
self.buffered_agent_message_delta_bytes = self
.buffered_agent_message_delta_bytes
.saturating_sub(delta.delta.len());
}
}
ThreadBufferedEvent::Request(request) => self
.pending_interactive_replay
.note_evicted_server_request(request.as_ref()),
ThreadBufferedEvent::HistoryEntryResponse(_)
| ThreadBufferedEvent::FeedbackSubmission(_) => {}
}
}
}
}
#[cfg(test)]
#[path = "thread_event_buffer_tests.rs"]
mod tests;
codex-rs/tui/src/app/thread_event_buffer_tests.rs
+219
View File
@@ -0,0 +1,219 @@
use super::*;
use crate::app::THREAD_EVENT_CHANNEL_CAPACITY;
use crate::app::Turn;
use crate::app::TurnStatus;
use crate::test_support::PathBufExt;
use crate::test_support::test_path_buf;
use codex_app_server_protocol::AgentMessageDeltaNotification;
use codex_app_server_protocol::CommandExecutionRequestApprovalParams;
use codex_app_server_protocol::RequestId as AppServerRequestId;
use codex_app_server_protocol::ServerRequest;
use codex_app_server_protocol::TurnItemsView;
use codex_app_server_protocol::TurnStartedNotification;
use codex_protocol::ThreadId;
use pretty_assertions::assert_eq;
fn turn_started_notification(thread_id: ThreadId, turn_id: &str) -> ServerNotification {
ServerNotification::TurnStarted(TurnStartedNotification {
thread_id: thread_id.to_string(),
turn: Turn {
id: turn_id.to_string(),
items_view: TurnItemsView::Full,
items: Vec::new(),
status: TurnStatus::InProgress,
error: None,
started_at: Some(0),
completed_at: None,
duration_ms: None,
},
})
}
fn exec_approval_request(
thread_id: ThreadId,
turn_id: &str,
item_id: &str,
approval_id: Option<&str>,
) -> ServerRequest {
ServerRequest::CommandExecutionRequestApproval {
request_id: AppServerRequestId::Integer(1),
params: CommandExecutionRequestApprovalParams {
thread_id: thread_id.to_string(),
turn_id: turn_id.to_string(),
item_id: item_id.to_string(),
started_at_ms: 0,
approval_id: approval_id.map(str::to_string),
environment_id: None,
reason: Some("needs approval".to_string()),
network_approval_context: None,
command: Some("echo hello".to_string()),
cwd: Some(test_path_buf("/tmp/project").abs().into()),
command_actions: None,
additional_permissions: None,
proposed_execpolicy_amendment: None,
proposed_network_policy_amendments: None,
available_decisions: None,
},
}
}
#[test]
fn thread_event_store_coalesces_only_adjacent_matching_agent_message_deltas() {
let thread_id = ThreadId::new();
let mut store = ThreadEventStore::new(/*capacity*/ 6);
let delta = |turn_id: &str, item_id: &str, text: &str| {
ServerNotification::AgentMessageDelta(AgentMessageDeltaNotification {
thread_id: thread_id.to_string(),
turn_id: turn_id.to_string(),
item_id: item_id.to_string(),
delta: text.to_string(),
})
};
let approval =
exec_approval_request(thread_id, "turn-2", "approval-1", /*approval_id*/ None);
let other_thread_delta = ServerNotification::AgentMessageDelta(AgentMessageDeltaNotification {
thread_id: ThreadId::new().to_string(),
turn_id: "turn-2".to_string(),
item_id: "item-2".to_string(),
delta: "other thread".to_string(),
});
let snapshot_events = |store: &ThreadEventStore| {
store
.snapshot()
.events
.into_iter()
.map(|event| match event {
ThreadBufferedEvent::Notification(notification) => {
serde_json::to_value(notification).expect("notification should serialize")
}
ThreadBufferedEvent::Request(request) => {
serde_json::to_value(request).expect("request should serialize")
}
other => panic!("unexpected buffered event: {other:?}"),
})
.collect::<Vec<_>>()
};
store.push_notification(delta("turn-1", "item-1", "hello"));
store.push_notification_ref(&delta("turn-1", "item-1", " world"));
store.push_notification(delta("turn-1", "item-2", "another"));
store.push_notification(delta("turn-2", "item-2", "next"));
store.push_notification(other_thread_delta.clone());
store.push_request(approval.clone());
store.push_notification(delta("turn-2", "item-2", "after approval"));
assert_eq!(
snapshot_events(&store),
vec![
serde_json::to_value(delta("turn-1", "item-1", "hello world"))
.expect("delta should serialize"),
serde_json::to_value(delta("turn-1", "item-2", "another"))
.expect("delta should serialize"),
serde_json::to_value(delta("turn-2", "item-2", "next"))
.expect("delta should serialize"),
serde_json::to_value(other_thread_delta).expect("delta should serialize"),
serde_json::to_value(approval).expect("request should serialize"),
serde_json::to_value(delta("turn-2", "item-2", "after approval"))
.expect("delta should serialize"),
]
);
assert!(store.has_pending_thread_approvals());
let remaining = MAX_COALESCED_AGENT_MESSAGE_DELTA_BYTES - "after approval".len();
let exact_limit = "é".repeat(remaining / "é".len());
let full_chunk = "é".repeat(MAX_COALESCED_AGENT_MESSAGE_DELTA_BYTES / "é".len());
store.push_notification(delta("turn-2", "item-2", &exact_limit));
store.push_notification(delta("turn-2", "item-2", "🙂"));
store.push_notification(delta("turn-2", "item-2", &full_chunk));
store.push_notification(delta("turn-2", "item-2", "🙂"));
store.push_notification(delta("turn-2", "item-2", &full_chunk));
store.push_notification(delta("turn-2", "item-2", "🙂"));
assert_eq!(
snapshot_events(&store),
vec![
serde_json::to_value(delta(
"turn-2",
"item-2",
&format!("after approval{exact_limit}"),
))
.expect("delta should serialize"),
serde_json::to_value(delta("turn-2", "item-2", "🙂")).expect("delta should serialize"),
serde_json::to_value(delta("turn-2", "item-2", &full_chunk))
.expect("delta should serialize"),
serde_json::to_value(delta("turn-2", "item-2", "🙂")).expect("delta should serialize"),
serde_json::to_value(delta("turn-2", "item-2", &full_chunk))
.expect("delta should serialize"),
serde_json::to_value(delta("turn-2", "item-2", "🙂")).expect("delta should serialize"),
]
);
assert!(!store.has_pending_thread_approvals());
let mut store = ThreadEventStore::new(/*capacity*/ THREAD_EVENT_CHANNEL_CAPACITY);
store.push_request(exec_approval_request(
thread_id,
"turn-budget",
"approval-budget",
/*approval_id*/ None,
));
for chunk in 0..65 {
let fragment = format!("{chunk:08}");
let notification = delta("turn-budget", "item-budget", &fragment);
for _ in 0..MAX_COALESCED_AGENT_MESSAGE_DELTA_BYTES / fragment.len() {
store.push_notification_ref(&notification);
}
}
assert_eq!(
snapshot_events(&store),
(1..65)
.map(|chunk| {
serde_json::to_value(delta(
"turn-budget",
"item-budget",
&format!("{chunk:08}").repeat(MAX_COALESCED_AGENT_MESSAGE_DELTA_BYTES / 8),
))
.expect("delta should serialize")
})
.collect::<Vec<_>>()
);
assert!(!store.has_pending_thread_approvals());
let mut store = ThreadEventStore::new(/*capacity*/ 4);
store.push_notification(turn_started_notification(thread_id, "turn-oversized"));
store.push_request(exec_approval_request(
thread_id,
"turn-oversized",
"approval-oversized",
/*approval_id*/ None,
));
store.push_notification(delta("turn-oversized", "item-oversized", "retained"));
let expected_events = snapshot_events(&store);
let oversized = "🙂".repeat(MAX_BUFFERED_AGENT_MESSAGE_DELTA_BYTES / "🙂".len() + 1);
store.push_notification(delta("turn-oversized", "item-oversized", &oversized));
store.push_notification_ref(&delta("turn-oversized", "item-oversized", &oversized));
assert_eq!(snapshot_events(&store), expected_events);
assert!(store.has_pending_thread_approvals());
assert_eq!(store.buffered_agent_message_delta_bytes, "retained".len());
store.push_notification(delta("turn-oversized", "item-oversized", " text"));
assert_eq!(
snapshot_events(&store),
vec![
serde_json::to_value(turn_started_notification(thread_id, "turn-oversized"))
.expect("turn notification should serialize"),
serde_json::to_value(exec_approval_request(
thread_id,
"turn-oversized",
"approval-oversized",
/*approval_id*/ None,
))
.expect("approval should serialize"),
serde_json::to_value(delta("turn-oversized", "item-oversized", "retained text"))
.expect("delta should serialize"),
]
);
assert!(store.has_pending_thread_approvals());
}
codex-rs/tui/src/app/thread_events.rs
+5 -20
View File
@@ -49,6 +49,7 @@ pub(super) struct ThreadEventStore {
pub(super) input_state: Option<ThreadInputState>,
pub(super) capacity: usize,
pub(super) active: bool,
pub(super) buffered_agent_message_delta_bytes: usize,
}
impl ThreadEventStore {
@@ -76,6 +77,7 @@ impl ThreadEventStore {
input_state: None,
capacity,
active: false,
buffered_agent_message_delta_bytes: 0,
}
}
@@ -98,6 +100,7 @@ impl ThreadEventStore {
pub(super) fn rebase_buffer_after_session_refresh(&mut self) {
self.buffer.retain(Self::event_survives_session_refresh);
self.buffered_agent_message_delta_bytes = 0;
}
pub(super) fn set_turns(&mut self, turns: Vec<Turn>) {
@@ -160,31 +163,13 @@ impl ThreadEventStore {
return;
}
self.buffer
.push_back(ThreadBufferedEvent::Notification(Box::new(
notification.into_owned(),
)));
if self.buffer.len() > self.capacity
&& let Some(removed) = self.buffer.pop_front()
&& let ThreadBufferedEvent::Request(request) = &removed
{
self.pending_interactive_replay
.note_evicted_server_request(request.as_ref());
}
self.push_replay_notification(notification);
}
pub(super) fn push_request(&mut self, request: ServerRequest) {
self.pending_interactive_replay
.note_server_request(&request);
self.buffer
.push_back(ThreadBufferedEvent::Request(Box::new(request)));
if self.buffer.len() > self.capacity
&& let Some(removed) = self.buffer.pop_front()
&& let ThreadBufferedEvent::Request(request) = &removed
{
self.pending_interactive_replay
.note_evicted_server_request(request.as_ref());
}
self.push_buffered_event(ThreadBufferedEvent::Request(Box::new(request)));
}
pub(super) fn pending_replay_requests(&self) -> Vec<ServerRequest> {
codex-rs/tui/src/app/thread_routing.rs
+6 -12
View File
@@ -982,13 +982,14 @@ impl App {
let channel = self.ensure_thread_channel(thread_id);
(channel.sender.clone(), Arc::clone(&channel.store))
};
let (notification, pending_status, turn_stopped) = {
let (notification, previous_pending_status, pending_status, turn_stopped) = {
let mut guard = store.lock().await;
if guard.session.is_none()
&& let Some(session) = inferred_session
{
guard.session = Some(session);
}
let previous_pending_status = guard.side_parent_pending_status();
let turn_stopped = match &notification {
ServerNotification::TurnCompleted(notification) => {
guard.active_turn_id() == Some(notification.turn.id.as_str())
@@ -1005,6 +1006,7 @@ impl App {
};
(
notification,
previous_pending_status,
guard.side_parent_pending_status(),
turn_stopped,
)
@@ -1034,6 +1036,8 @@ impl App {
self.set_side_parent_status(thread_id, Some(status));
} else if let Some(change) = notification_status_change {
self.apply_side_parent_status_change(thread_id, change);
} else if previous_pending_status.is_some() {
self.clear_side_parent_action_status(thread_id);
}
self.refresh_pending_thread_approvals().await;
Ok(())
@@ -1184,17 +1188,7 @@ impl App {
// retaining another deep copy for thread replay only accumulates already-delivered
// history data. Inactive responses still need the buffer because they are not sent.
if !should_send || !matches!(&event, HistoryLookupResponse::Batch { .. }) {
guard
.buffer
.push_back(ThreadBufferedEvent::HistoryEntryResponse(event.clone()));
if guard.buffer.len() > guard.capacity
&& let Some(removed) = guard.buffer.pop_front()
&& let ThreadBufferedEvent::Request(request) = &removed
{
guard
.pending_interactive_replay
.note_evicted_server_request(request.as_ref());
}
guard.push_buffered_event(ThreadBufferedEvent::HistoryEntryResponse(event.clone()));
}
should_send
};