mirror of
https://github.com/openai/codex.git
synced 2026-11-12 21:20:34 +00:00
Tamir Duberstein
adca203576
## Why Switching back to a running thread can lose the active turn's state after its `TurnStarted` notification falls out of the bounded replay buffer, breaking steering. ## What changed Retain the evicted active turn and include it in replay snapshots when it is missing from the stored turns. Clear the retained turn when turns are replaced, the active turn completes, or the thread closes. ## Testing Add a regression test that evicts the turn-start notification, restores the thread, and submits two steering messages while preserving a queued follow-up. A snapshot verifies the steering messages and queued input remain visible. GitOrigin-RevId: a4302a2e0990b907207c229361cbcec68ebbbdc4
86 lines
3.5 KiB
Rust
86 lines
3.5 KiB
Rust
//! 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());
|
|
} else if let ServerNotification::TurnStarted(started) = *notification
|
|
&& self.active_turn_id() == Some(&started.turn.id)
|
|
{
|
|
self.evicted_active_turn = Some(Box::new(started.turn));
|
|
}
|
|
}
|
|
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;
|