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

Box app-server event payloads (#35854)

Browse Source 
## What changed

- Store `ServerNotification` and `ServerRequest` payloads behind `Box` in
  `AppServerEvent`, `InProcessServerEvent`, and TUI `ThreadBufferedEvent` values.
- Update app-server delivery, TUI routing and replay, exec, onboarding, and tests
  to borrow or consume the boxed payloads as appropriate.

GitOrigin-RevId: 7d0dab9aba47487114128ce55c914af5bb46c5b5
This commit is contained in:
joeytrasatti-openai
2026-07-29 00:30:11 +00:00
committed by copyberry
parent 8707a35113
commit 07490c7523
16 changed files with 322 additions and 245 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-client/src/lib.rs
+84 -65
View File
@@ -97,8 +97,8 @@ pub type RequestResult = std::result::Result<JsonRpcResult, JSONRPCErrorError>;
#[derive(Debug, Clone)]
pub enum AppServerEvent {
Lagged { skipped: usize },
ServerNotification(ServerNotification),
ServerRequest(ServerRequest),
ServerNotification(Box<ServerNotification>),
ServerRequest(Box<ServerRequest>),
Disconnected { message: String },
}
@@ -205,7 +205,7 @@ where
*skipped_events = skipped_events.saturating_add(1);
warn!("dropping in-process app-server event because consumer queue is full");
if let InProcessServerEvent::ServerRequest(request) = event {
reject_server_request(request);
reject_server_request(*request);
}
return ForwardEventResult::Continue;
}
@@ -231,7 +231,7 @@ where
*skipped_events = skipped_events.saturating_add(1);
warn!("dropping in-process app-server event because consumer queue is full");
if let InProcessServerEvent::ServerRequest(request) = event {
reject_server_request(request);
reject_server_request(*request);
}
ForwardEventResult::Continue
}
@@ -522,9 +522,9 @@ impl InProcessAppServerClient {
let Some(event) = event else {
break;
};
if let InProcessServerEvent::ServerRequest(
ServerRequest::ChatgptAuthTokensRefresh { request_id, .. }
) = &event
if let InProcessServerEvent::ServerRequest(request) = &event
&& let ServerRequest::ChatgptAuthTokensRefresh { request_id, .. } =
request.as_ref()
{
let send_result = request_sender.fail_server_request(
request_id.clone(),
@@ -1332,9 +1332,9 @@ mod tests {
async fn forward_in_process_event_preserves_transcript_notifications_under_backpressure() {
let (event_tx, mut event_rx) = mpsc::channel(1);
event_tx
.send(InProcessServerEvent::ServerNotification(
.send(InProcessServerEvent::ServerNotification(Box::new(
command_execution_output_delta_notification("stdout-1"),
))
)))
.await
.expect("initial event should enqueue");
@@ -1342,8 +1342,8 @@ mod tests {
let result = forward_in_process_event(
&event_tx,
&mut skipped_events,
InProcessServerEvent::ServerNotification(command_execution_output_delta_notification(
"stdout-2",
InProcessServerEvent::ServerNotification(Box::new(
command_execution_output_delta_notification("stdout-2"),
)),
|_| {},
)
@@ -1372,7 +1372,7 @@ mod tests {
let result = forward_in_process_event(
&event_tx,
&mut skipped_events,
InProcessServerEvent::ServerNotification(notification),
InProcessServerEvent::ServerNotification(Box::new(notification)),
|_| {},
)
.await;
@@ -1385,9 +1385,12 @@ mod tests {
.expect("receiver task should join successfully");
assert!(matches!(
&events[0],
InProcessServerEvent::ServerNotification(
ServerNotification::CommandExecutionOutputDelta(notification)
) if notification.delta == "stdout-1"
InProcessServerEvent::ServerNotification(notification)
if matches!(
notification.as_ref(),
ServerNotification::CommandExecutionOutputDelta(notification)
if notification.delta == "stdout-1"
)
));
assert!(matches!(
&events[1],
@@ -1395,24 +1398,35 @@ mod tests {
));
assert!(matches!(
&events[2],
InProcessServerEvent::ServerNotification(ServerNotification::AgentMessageDelta(
notification
)) if notification.delta == "hello"
InProcessServerEvent::ServerNotification(notification)
if matches!(
notification.as_ref(),
ServerNotification::AgentMessageDelta(notification)
if notification.delta == "hello"
)
));
assert!(matches!(
&events[3],
InProcessServerEvent::ServerNotification(ServerNotification::ItemCompleted(
notification
)) if matches!(
&notification.item,
codex_app_server_protocol::ThreadItem::AgentMessage { text, .. } if text == "hello"
)
InProcessServerEvent::ServerNotification(notification)
if matches!(
notification.as_ref(),
ServerNotification::ItemCompleted(notification)
if matches!(
&notification.item,
codex_app_server_protocol::ThreadItem::AgentMessage { text, .. }
if text == "hello"
)
)
));
assert!(matches!(
&events[4],
InProcessServerEvent::ServerNotification(ServerNotification::TurnCompleted(
notification
)) if notification.turn.status == codex_app_server_protocol::TurnStatus::Completed
InProcessServerEvent::ServerNotification(notification)
if matches!(
notification.as_ref(),
ServerNotification::TurnCompleted(notification)
if notification.turn.status
== codex_app_server_protocol::TurnStatus::Completed
)
));
}
@@ -1753,7 +1767,8 @@ mod tests {
let event = client.next_event().await.expect("event should arrive");
assert!(matches!(
event,
AppServerEvent::ServerNotification(ServerNotification::AccountUpdated(_))
AppServerEvent::ServerNotification(notification)
if matches!(notification.as_ref(), ServerNotification::AccountUpdated(_))
));
client.shutdown().await.expect("shutdown should complete");
@@ -1799,9 +1814,12 @@ mod tests {
.expect("event stream should stay open");
assert!(matches!(
first_event,
AppServerEvent::ServerNotification(ServerNotification::CommandExecutionOutputDelta(
notification
)) if notification.delta == "stdout-1"
AppServerEvent::ServerNotification(notification)
if matches!(
notification.as_ref(),
ServerNotification::CommandExecutionOutputDelta(notification)
if notification.delta == "stdout-1"
)
));
let mut remaining_events = Vec::new();
@@ -1818,30 +1836,31 @@ mod tests {
for event in &remaining_events {
match event {
AppServerEvent::Lagged { skipped: 1 } => {}
AppServerEvent::ServerNotification(
ServerNotification::CommandExecutionOutputDelta(notification),
) if notification.delta == "stdout-2" => {}
AppServerEvent::ServerNotification(ServerNotification::AgentMessageDelta(
notification,
)) if notification.delta == "hello" => {
transcript_event_names.push("agent_message_delta");
}
AppServerEvent::ServerNotification(ServerNotification::ItemCompleted(
notification,
)) if matches!(
&notification.item,
codex_app_server_protocol::ThreadItem::AgentMessage { text, .. } if text == "hello"
) =>
{
transcript_event_names.push("item_completed");
}
AppServerEvent::ServerNotification(ServerNotification::TurnCompleted(
notification,
)) if notification.turn.status
== codex_app_server_protocol::TurnStatus::Completed =>
{
transcript_event_names.push("turn_completed");
}
AppServerEvent::ServerNotification(notification) => match notification.as_ref() {
ServerNotification::CommandExecutionOutputDelta(notification)
if notification.delta == "stdout-2" => {}
ServerNotification::AgentMessageDelta(notification)
if notification.delta == "hello" =>
{
transcript_event_names.push("agent_message_delta");
}
ServerNotification::ItemCompleted(notification)
if matches!(
&notification.item,
codex_app_server_protocol::ThreadItem::AgentMessage { text, .. }
if text == "hello"
) =>
{
transcript_event_names.push("item_completed");
}
ServerNotification::TurnCompleted(notification)
if notification.turn.status
== codex_app_server_protocol::TurnStatus::Completed =>
{
transcript_event_names.push("turn_completed");
}
_ => panic!("unexpected remaining event: {event:?}"),
},
_ => panic!("unexpected remaining event: {event:?}"),
}
}
@@ -2103,7 +2122,7 @@ mod tests {
#[test]
fn event_requires_delivery_marks_transcript_and_terminal_events() {
assert!(event_requires_delivery(
&InProcessServerEvent::ServerNotification(
&InProcessServerEvent::ServerNotification(Box::new(
codex_app_server_protocol::ServerNotification::TurnCompleted(
codex_app_server_protocol::TurnCompletedNotification {
thread_id: "thread".to_string(),
@@ -2119,10 +2138,10 @@ mod tests {
},
}
)
)
))
));
assert!(event_requires_delivery(
&InProcessServerEvent::ServerNotification(
&InProcessServerEvent::ServerNotification(Box::new(
codex_app_server_protocol::ServerNotification::AgentMessageDelta(
codex_app_server_protocol::AgentMessageDeltaNotification {
thread_id: "thread".to_string(),
@@ -2131,10 +2150,10 @@ mod tests {
delta: "hello".to_string(),
}
)
)
))
));
assert!(event_requires_delivery(
&InProcessServerEvent::ServerNotification(
&InProcessServerEvent::ServerNotification(Box::new(
codex_app_server_protocol::ServerNotification::ItemCompleted(
codex_app_server_protocol::ItemCompletedNotification {
thread_id: "thread".to_string(),
@@ -2148,23 +2167,23 @@ mod tests {
},
}
)
)
))
));
assert!(event_requires_delivery(
&InProcessServerEvent::ServerNotification(
&InProcessServerEvent::ServerNotification(Box::new(
codex_app_server_protocol::ServerNotification::ExternalAgentConfigImportCompleted(
codex_app_server_protocol::ExternalAgentConfigImportCompletedNotification {
import_id: "import".to_string(),
item_type_results: Vec::new(),
},
)
)
))
));
assert!(!event_requires_delivery(&InProcessServerEvent::Lagged {
skipped: 1
}));
assert!(!event_requires_delivery(
&InProcessServerEvent::ServerNotification(
&InProcessServerEvent::ServerNotification(Box::new(
codex_app_server_protocol::ServerNotification::CommandExecutionOutputDelta(
codex_app_server_protocol::CommandExecutionOutputDeltaNotification {
thread_id: "thread".to_string(),
@@ -2173,7 +2192,7 @@ mod tests {
delta: "stdout".to_string(),
}
)
)
))
));
}
codex-rs/app-server-client/src/remote.rs
+4 -3
View File
@@ -348,7 +348,7 @@ impl RemoteAppServerClient {
Ok(request) => {
if let Err(err) = deliver_event(
&event_tx,
AppServerEvent::ServerRequest(request),
AppServerEvent::ServerRequest(Box::new(request)),
)
{
warn!(%err, "failed to deliver remote app-server server request");
@@ -862,7 +862,8 @@ where
let method = request.method.clone();
match ServerRequest::try_from(request) {
Ok(request) => {
pending_events.push(AppServerEvent::ServerRequest(request));
pending_events
.push(AppServerEvent::ServerRequest(Box::new(request)));
}
Err(err) => {
warn!(%err, method, "rejecting unknown remote app-server request during initialize");
@@ -940,7 +941,7 @@ where
fn app_server_event_from_notification(notification: JSONRPCNotification) -> Option<AppServerEvent> {
match ServerNotification::try_from(notification) {
Ok(notification) => Some(AppServerEvent::ServerNotification(notification)),
Ok(notification) => Some(AppServerEvent::ServerNotification(Box::new(notification))),
Err(_) => None,
}
}
codex-rs/app-server/src/in_process.rs
+9 -5
View File
@@ -160,9 +160,9 @@ pub struct InProcessStartArgs {
#[derive(Debug, Clone)]
pub enum InProcessServerEvent {
/// Server request that requires client response/rejection.
ServerRequest(ServerRequest),
ServerRequest(Box<ServerRequest>),
/// App-server notification directed to the embedded client.
ServerNotification(ServerNotification),
ServerNotification(Box<ServerNotification>),
/// Indicates one or more events were dropped due to backpressure.
Lagged { skipped: usize },
}
@@ -663,7 +663,7 @@ async fn start_uninitialized(args: InProcessStartArgs) -> IoResult<InProcessClie
// Send directly to avoid cloning; on failure the
// original value is returned inside the error.
if let Err(send_error) = event_tx
.try_send(InProcessServerEvent::ServerRequest(request))
.try_send(InProcessServerEvent::ServerRequest(Box::new(request)))
{
let (error, inner) = match send_error {
mpsc::error::TrySendError::Full(inner) => (
@@ -695,14 +695,18 @@ async fn start_uninitialized(args: InProcessStartArgs) -> IoResult<InProcessClie
let notification = envelope.notification;
if server_notification_requires_delivery(&notification) {
if event_tx
.send(InProcessServerEvent::ServerNotification(notification))
.send(InProcessServerEvent::ServerNotification(Box::new(
notification,
)))
.await
.is_err()
{
break;
}
} else if let Err(send_error) =
event_tx.try_send(InProcessServerEvent::ServerNotification(notification))
event_tx.try_send(InProcessServerEvent::ServerNotification(
Box::new(notification),
))
{
match send_error {
mpsc::error::TrySendError::Full(_) => {
codex-rs/app-server/tests/suite/v2/remote_thread_store.rs
+4 -6
View File
@@ -191,9 +191,8 @@ async fn thread_delete_with_non_local_thread_store_does_not_create_local_persist
let Some(event) = client.next_event().await else {
anyhow::bail!("in-process app-server stopped before turn/completed");
};
if let InProcessServerEvent::ServerNotification(ServerNotification::TurnCompleted(
completed,
)) = event
if let InProcessServerEvent::ServerNotification(notification) = event
&& let ServerNotification::TurnCompleted(completed) = notification.as_ref()
&& completed.thread_id == thread.id
{
return Ok::<(), anyhow::Error>(());
@@ -332,9 +331,8 @@ async fn cold_thread_resume_reuses_non_local_history_probe() -> Result<()> {
let Some(event) = client.next_event().await else {
anyhow::bail!("in-process app-server stopped before turn/completed");
};
if let InProcessServerEvent::ServerNotification(ServerNotification::TurnCompleted(
completed,
)) = event
if let InProcessServerEvent::ServerNotification(notification) = event
&& let ServerNotification::TurnCompleted(completed) = notification.as_ref()
&& completed.thread_id == thread.id
{
return Ok::<(), anyhow::Error>(());
codex-rs/exec/src/lib.rs
+3 -2
View File
@@ -997,9 +997,10 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> {
match server_event {
InProcessServerEvent::ServerRequest(request) => {
handle_server_request(&client, request, &mut error_seen).await;
handle_server_request(&client, *request, &mut error_seen).await;
}
InProcessServerEvent::ServerNotification(mut notification) => {
InProcessServerEvent::ServerNotification(notification) => {
let mut notification = *notification;
if let ServerNotification::Error(payload) = &notification {
if payload.thread_id == primary_thread_id_for_requests
&& payload.turn_id == task_id
codex-rs/tui/src/app/agent_status_feed.rs
+6 -8
View File
@@ -85,14 +85,12 @@ impl AgentStatusThreadPreview {
let mut activity = Vec::new();
for event in events {
let item = match event {
ThreadBufferedEvent::Notification(ServerNotification::ItemCompleted(event)) => {
&event.item
}
ThreadBufferedEvent::Notification(ServerNotification::ItemStarted(event)) => {
&event.item
}
ThreadBufferedEvent::Notification(_)
| ThreadBufferedEvent::Request(_)
ThreadBufferedEvent::Notification(notification) => match notification.as_ref() {
ServerNotification::ItemCompleted(event) => &event.item,
ServerNotification::ItemStarted(event) => &event.item,
_ => continue,
},
ThreadBufferedEvent::Request(_)
| ThreadBufferedEvent::HistoryEntryResponse(_)
| ThreadBufferedEvent::FeedbackSubmission(_) => continue,
};
codex-rs/tui/src/app/app_server_events.rs
+2 -2
View File
@@ -44,11 +44,11 @@ impl App {
self.chat_widget.finish_mcp_startup_after_lag();
}
AppServerEvent::ServerNotification(notification) => {
self.handle_server_notification_event(app_server_client, notification)
self.handle_server_notification_event(app_server_client, *notification)
.await;
}
AppServerEvent::ServerRequest(request) => {
self.handle_server_request_event(app_server_client, request)
self.handle_server_request_event(app_server_client, *request)
.await;
}
AppServerEvent::Disconnected { message } => {
codex-rs/tui/src/app/background_requests.rs
+1 -1
View File
@@ -626,7 +626,7 @@ impl App {
{
guard
.pending_interactive_replay
.note_evicted_server_request(request);
.note_evicted_server_request(request.as_ref());
}
guard.active
};
codex-rs/tui/src/app/pending_interactive_replay.rs
+39 -13
View File
@@ -702,8 +702,12 @@ mod tests {
assert_eq!(snapshot.events.len(), 1);
assert!(matches!(
snapshot.events.first(),
Some(ThreadBufferedEvent::Request(ServerRequest::ToolRequestUserInput { params, .. }))
if params.item_id == "call-1"
Some(ThreadBufferedEvent::Request(request))
if matches!(
request.as_ref(),
ServerRequest::ToolRequestUserInput { params, .. }
if params.item_id == "call-1"
)
));
}
@@ -738,7 +742,11 @@ mod tests {
snapshot.events.iter().all(|event| {
!matches!(
event,
ThreadBufferedEvent::Request(ServerRequest::ToolRequestUserInput { .. })
ThreadBufferedEvent::Request(request)
if matches!(
request.as_ref(),
ServerRequest::ToolRequestUserInput { .. }
)
)
}),
"server-resolved request_user_input prompt should not replay on thread switch"
@@ -783,9 +791,11 @@ mod tests {
snapshot.events.iter().all(|event| {
!matches!(
event,
ThreadBufferedEvent::Request(
ServerRequest::CommandExecutionRequestApproval { .. }
)
ThreadBufferedEvent::Request(request)
if matches!(
request.as_ref(),
ServerRequest::CommandExecutionRequestApproval { .. }
)
)
}),
"server-resolved exec approval prompt should not replay on thread switch"
@@ -810,8 +820,12 @@ mod tests {
assert_eq!(snapshot.events.len(), 1);
assert!(matches!(
snapshot.events.first(),
Some(ThreadBufferedEvent::Request(ServerRequest::ToolRequestUserInput { params, .. }))
if params.item_id == "call-2"
Some(ThreadBufferedEvent::Request(request))
if matches!(
request.as_ref(),
ServerRequest::ToolRequestUserInput { params, .. }
if params.item_id == "call-2"
)
));
}
@@ -832,8 +846,12 @@ mod tests {
assert_eq!(snapshot.events.len(), 1);
assert!(matches!(
snapshot.events.first(),
Some(ThreadBufferedEvent::Request(ServerRequest::ToolRequestUserInput { params, .. }))
if params.item_id == "call-2"
Some(ThreadBufferedEvent::Request(request))
if matches!(
request.as_ref(),
ServerRequest::ToolRequestUserInput { params, .. }
if params.item_id == "call-2"
)
));
}
@@ -869,8 +887,12 @@ mod tests {
assert!(snapshot.events.iter().all(|event| {
!matches!(
event,
ThreadBufferedEvent::Request(ServerRequest::CommandExecutionRequestApproval { .. })
| ThreadBufferedEvent::Request(ServerRequest::FileChangeRequestApproval { .. })
ThreadBufferedEvent::Request(request)
if matches!(
request.as_ref(),
ServerRequest::CommandExecutionRequestApproval { .. }
| ServerRequest::FileChangeRequestApproval { .. }
)
)
}));
}
@@ -935,7 +957,11 @@ mod tests {
assert!(store.snapshot().events.iter().all(|event| {
!matches!(
event,
ThreadBufferedEvent::Request(ServerRequest::CommandExecutionRequestApproval { .. })
ThreadBufferedEvent::Request(request)
if matches!(
request.as_ref(),
ServerRequest::CommandExecutionRequestApproval { .. }
)
)
}));
}
codex-rs/tui/src/app/replay_filter.rs
+16 -12
View File
@@ -10,13 +10,15 @@ pub(super) fn snapshot_has_pending_interactive_request(snapshot: &ThreadEventSna
snapshot.events.iter().any(|event| {
matches!(
event,
ThreadBufferedEvent::Request(
ServerRequest::CommandExecutionRequestApproval { .. }
| ServerRequest::FileChangeRequestApproval { .. }
| ServerRequest::McpServerElicitationRequest { .. }
| ServerRequest::PermissionsRequestApproval { .. }
| ServerRequest::ToolRequestUserInput { .. }
)
ThreadBufferedEvent::Request(request)
if matches!(
request.as_ref(),
ServerRequest::CommandExecutionRequestApproval { .. }
| ServerRequest::FileChangeRequestApproval { .. }
| ServerRequest::McpServerElicitationRequest { .. }
| ServerRequest::PermissionsRequestApproval { .. }
| ServerRequest::ToolRequestUserInput { .. }
)
)
})
}
@@ -24,10 +26,12 @@ pub(super) fn snapshot_has_pending_interactive_request(snapshot: &ThreadEventSna
pub(super) fn event_is_notice(event: &ThreadBufferedEvent) -> bool {
matches!(
event,
ThreadBufferedEvent::Notification(
ServerNotification::Warning(_)
| ServerNotification::GuardianWarning(_)
| ServerNotification::ConfigWarning(_)
)
ThreadBufferedEvent::Notification(notification)
if matches!(
notification.as_ref(),
ServerNotification::Warning(_)
| ServerNotification::GuardianWarning(_)
| ServerNotification::ConfigWarning(_)
)
)
}
codex-rs/tui/src/app/tests.rs
+52 -48
View File
@@ -198,9 +198,8 @@ async fn next_thread_settings_updated(
.await
.expect("app-server should emit an event")
.expect("app-server event stream should remain open");
if let codex_app_server_client::AppServerEvent::ServerNotification(
ServerNotification::ThreadSettingsUpdated(notification),
) = event
if let codex_app_server_client::AppServerEvent::ServerNotification(notification) = event
&& let ServerNotification::ThreadSettingsUpdated(notification) = *notification
&& notification.thread_id == thread_id.to_string()
{
return notification;
@@ -289,10 +288,12 @@ async fn enqueue_primary_thread_session_replays_buffered_approval_after_attach()
assert!(matches!(
&event,
ThreadBufferedEvent::Request(ServerRequest::CommandExecutionRequestApproval {
params,
..
}) if params.turn_id == "turn-1"
ThreadBufferedEvent::Request(request)
if matches!(
request.as_ref(),
ServerRequest::CommandExecutionRequestApproval { params, .. }
if params.turn_id == "turn-1"
)
));
app.handle_thread_event_now(event);
@@ -353,10 +354,12 @@ async fn resolved_buffered_approval_does_not_become_actionable_after_drain() ->
assert!(matches!(
&event,
ThreadBufferedEvent::Request(ServerRequest::CommandExecutionRequestApproval {
params,
..
}) if params.turn_id == "turn-1"
ThreadBufferedEvent::Request(request)
if matches!(
request.as_ref(),
ServerRequest::CommandExecutionRequestApproval { params, .. }
if params.turn_id == "turn-1"
)
));
app.handle_thread_event_now(event);
@@ -801,9 +804,9 @@ async fn replayed_turn_complete_submits_restored_queued_follow_up() {
ThreadEventSnapshot {
session: None,
turns: Vec::new(),
events: vec![ThreadBufferedEvent::Notification(
events: vec![ThreadBufferedEvent::Notification(Box::new(
turn_completed_notification(thread_id, "turn-1", TurnStatus::Completed),
)],
))],
input_state: Some(input_state),
},
/*resume_restored_queue*/ true,
@@ -854,9 +857,9 @@ async fn replay_only_thread_keeps_restored_queue_visible() {
ThreadEventSnapshot {
session: None,
turns: Vec::new(),
events: vec![ThreadBufferedEvent::Notification(
events: vec![ThreadBufferedEvent::Notification(Box::new(
turn_completed_notification(thread_id, "turn-1", TurnStatus::Completed),
)],
))],
input_state: Some(input_state),
},
/*resume_restored_queue*/ false,
@@ -1026,12 +1029,14 @@ async fn replay_thread_snapshot_does_not_submit_queue_before_replay_catches_up()
session: None,
turns: Vec::new(),
events: vec![
ThreadBufferedEvent::Notification(turn_completed_notification(
ThreadBufferedEvent::Notification(Box::new(turn_completed_notification(
thread_id,
"turn-0",
TurnStatus::Completed,
)),
ThreadBufferedEvent::Notification(turn_started_notification(thread_id, "turn-1")),
))),
ThreadBufferedEvent::Notification(Box::new(turn_started_notification(
thread_id, "turn-1",
))),
],
input_state: Some(input_state),
},
@@ -1294,9 +1299,9 @@ async fn replayed_interrupted_turn_restores_queued_input_to_composer() {
ThreadEventSnapshot {
session: None,
turns: Vec::new(),
events: vec![ThreadBufferedEvent::Notification(
events: vec![ThreadBufferedEvent::Notification(Box::new(
turn_completed_notification(thread_id, "turn-1", TurnStatus::Interrupted),
)],
))],
input_state: Some(input_state),
},
/*resume_restored_queue*/ true,
@@ -1323,10 +1328,8 @@ async fn token_usage_update_refreshes_status_line_with_runtime_context_window()
assert_eq!(app.chat_widget.status_line_text(), None);
app.handle_thread_event_now(ThreadBufferedEvent::Notification(token_usage_notification(
ThreadId::new(),
"turn-1",
Some(950_000),
app.handle_thread_event_now(ThreadBufferedEvent::Notification(Box::new(
token_usage_notification(ThreadId::new(), "turn-1", Some(950_000)),
)));
assert_eq!(
@@ -1341,7 +1344,7 @@ async fn collab_receiver_notification_caches_thread_without_app_server_read() {
let receiver_thread_id =
ThreadId::from_string("00000000-0000-0000-0000-000000000123").expect("valid thread id");
app.handle_thread_event_now(ThreadBufferedEvent::Notification(
app.handle_thread_event_now(ThreadBufferedEvent::Notification(Box::new(
ServerNotification::ItemStarted(ItemStartedNotification {
thread_id: ThreadId::new().to_string(),
turn_id: "turn-1".to_string(),
@@ -1358,7 +1361,7 @@ async fn collab_receiver_notification_caches_thread_without_app_server_read() {
agents_states: HashMap::new(),
},
}),
));
)));
assert_eq!(
app.agent_navigation.get(&receiver_thread_id),
@@ -1378,7 +1381,7 @@ async fn collab_receiver_notification_does_not_cache_not_found_thread() {
let receiver_thread_id =
ThreadId::from_string("00000000-0000-0000-0000-000000000124").expect("valid thread id");
app.handle_thread_event_now(ThreadBufferedEvent::Notification(
app.handle_thread_event_now(ThreadBufferedEvent::Notification(Box::new(
ServerNotification::ItemCompleted(codex_app_server_protocol::ItemCompletedNotification {
thread_id: ThreadId::new().to_string(),
turn_id: "turn-1".to_string(),
@@ -1401,7 +1404,7 @@ async fn collab_receiver_notification_does_not_cache_not_found_thread() {
)]),
},
}),
));
)));
assert_eq!(app.agent_navigation.get(&receiver_thread_id), None);
}
@@ -2121,9 +2124,8 @@ async fn handle_start_side_seeds_navigation_before_thread_started() -> Result<()
.await
.expect("app-server should emit an event")
.expect("app-server event stream should remain open");
if let codex_app_server_client::AppServerEvent::ServerNotification(
ServerNotification::ThreadStarted(notification),
) = event
if let codex_app_server_client::AppServerEvent::ServerNotification(notification) = event
&& let ServerNotification::ThreadStarted(notification) = notification.as_ref()
&& notification.thread.id == side_thread_id.to_string()
{
saw_thread_started = true;
@@ -2917,18 +2919,18 @@ async fn replay_snapshot_with_pending_request_suppresses_replay_notices() {
session: Some(test_thread_session(thread_id, test_path_buf("/tmp/main"))),
turns: Vec::new(),
events: vec![
ThreadBufferedEvent::Notification(ServerNotification::Warning(
ThreadBufferedEvent::Notification(Box::new(ServerNotification::Warning(
WarningNotification {
thread_id: Some(thread_id.to_string()),
message: stale_warning.to_string(),
},
)),
ThreadBufferedEvent::Request(exec_approval_request(
))),
ThreadBufferedEvent::Request(Box::new(exec_approval_request(
thread_id,
"turn-approval",
"call-approval",
/*approval_id*/ None,
)),
))),
],
input_state: None,
},
@@ -4082,7 +4084,7 @@ async fn primary_thread_ignores_child_mcp_startup_notifications() {
app.handle_app_server_event(
&app_server,
codex_app_server_client::AppServerEvent::ServerNotification(
codex_app_server_client::AppServerEvent::ServerNotification(Box::new(
ServerNotification::McpServerStatusUpdated(McpServerStatusUpdatedNotification {
thread_id: Some(child_thread_id.to_string()),
name: "sentry".to_string(),
@@ -4090,7 +4092,7 @@ async fn primary_thread_ignores_child_mcp_startup_notifications() {
error: Some("sentry is not logged in".to_string()),
failure_reason: None,
}),
),
)),
)
.await;
@@ -4106,9 +4108,11 @@ async fn primary_thread_ignores_child_mcp_startup_notifications() {
assert!(
matches!(
child_snapshot.events.as_slice(),
[ThreadBufferedEvent::Notification(
ServerNotification::McpServerStatusUpdated(_)
)]
[ThreadBufferedEvent::Notification(notification)]
if matches!(
notification.as_ref(),
ServerNotification::McpServerStatusUpdated(_)
)
),
"child MCP startup notification should be buffered for the child thread"
);
@@ -4155,7 +4159,7 @@ async fn app_scoped_mcp_startup_notifications_do_not_render_in_active_thread() {
app.handle_app_server_event(
&app_server,
codex_app_server_client::AppServerEvent::ServerNotification(
codex_app_server_client::AppServerEvent::ServerNotification(Box::new(
ServerNotification::McpServerStatusUpdated(McpServerStatusUpdatedNotification {
thread_id: None,
name: "sentry".to_string(),
@@ -4163,7 +4167,7 @@ async fn app_scoped_mcp_startup_notifications_do_not_render_in_active_thread() {
error: Some("sentry is not logged in".to_string()),
failure_reason: None,
}),
),
)),
)
.await;
@@ -4219,7 +4223,7 @@ async fn active_side_thread_renders_live_mcp_startup_notifications() {
] {
app.handle_app_server_event(
&app_server,
codex_app_server_client::AppServerEvent::ServerNotification(
codex_app_server_client::AppServerEvent::ServerNotification(Box::new(
ServerNotification::McpServerStatusUpdated(McpServerStatusUpdatedNotification {
thread_id: Some(side_thread_id.to_string()),
name: "sentry".to_string(),
@@ -4228,7 +4232,7 @@ async fn active_side_thread_renders_live_mcp_startup_notifications() {
.then(|| "sentry is not logged in".to_string()),
failure_reason: None,
}),
),
)),
)
.await;
}
@@ -6706,7 +6710,7 @@ async fn replace_chat_widget_reseeds_collab_agent_metadata_for_replay() {
ThreadEventSnapshot {
session: None,
turns: Vec::new(),
events: vec![ThreadBufferedEvent::Notification(
events: vec![ThreadBufferedEvent::Notification(Box::new(
ServerNotification::ItemStarted(
codex_app_server_protocol::ItemStartedNotification {
thread_id: "thread-1".to_string(),
@@ -6726,7 +6730,7 @@ async fn replace_chat_widget_reseeds_collab_agent_metadata_for_replay() {
},
},
),
)],
))],
input_state: None,
},
/*resume_restored_queue*/ false,
@@ -7065,9 +7069,9 @@ async fn override_turn_context_sends_thread_settings_update() {
app.handle_app_server_event(
&app_server,
codex_app_server_client::AppServerEvent::ServerNotification(
codex_app_server_client::AppServerEvent::ServerNotification(Box::new(
ServerNotification::ThreadSettingsUpdated(notification),
),
)),
)
.await;
let updated_session = app
codex-rs/tui/src/app/tests/rate_limits.rs
+2 -2
View File
@@ -56,11 +56,11 @@ async fn deliver_rolling_rate_limit_snapshot(
) {
app.handle_app_server_event(
app_server,
codex_app_server_client::AppServerEvent::ServerNotification(
codex_app_server_client::AppServerEvent::ServerNotification(Box::new(
ServerNotification::AccountRateLimitsUpdated(AccountRateLimitsUpdatedNotification {
rate_limits: snapshot,
}),
),
)),
)
.await;
}
codex-rs/tui/src/app/tests/safety_buffering.rs
+39 -31
View File
@@ -107,13 +107,13 @@ async fn next_turn_started(
.await
.expect("app-server should emit a turn/start event")
.expect("app-server event stream should remain open");
let started_turn_id = match &event {
AppServerEvent::ServerNotification(ServerNotification::TurnStarted(notification))
if notification.thread_id == thread_id.to_string() =>
{
Some(notification.turn.id.clone())
}
_ => None,
let started_turn_id = if let AppServerEvent::ServerNotification(notification) = &event
&& let ServerNotification::TurnStarted(notification) = notification.as_ref()
&& notification.thread_id == thread_id.to_string()
{
Some(notification.turn.id.clone())
} else {
None
};
app.handle_app_server_event(app_server, event).await;
drain_active_thread_events(app);
@@ -138,8 +138,12 @@ async fn wait_for_turn_completed(
.expect("app-server event stream should remain open");
let completed = matches!(
&event,
AppServerEvent::ServerNotification(ServerNotification::TurnCompleted(notification))
if notification.thread_id == thread_id.to_string()
AppServerEvent::ServerNotification(notification)
if matches!(
notification.as_ref(),
ServerNotification::TurnCompleted(notification)
if notification.thread_id == thread_id.to_string()
)
);
app.handle_app_server_event(app_server, event).await;
drain_active_thread_events(app);
@@ -237,12 +241,12 @@ stream_max_retries = 0
}
while app_event_rx.try_recv().is_ok() {}
app.handle_thread_event_now(ThreadBufferedEvent::Notification(
app.handle_thread_event_now(ThreadBufferedEvent::Notification(Box::new(
ServerNotification::Warning(WarningNotification {
thread_id: Some(thread_id.to_string()),
message: "event handled while interrupt is pending".to_string(),
}),
));
)));
assert!(matches!(
app_event_rx.try_recv(),
Ok(AppEvent::InsertHistoryCell(_))
@@ -507,16 +511,18 @@ goals = true
app.handle_app_server_event(
&app_server,
AppServerEvent::ServerNotification(ServerNotification::ModelSafetyBufferingUpdated(
ModelSafetyBufferingUpdatedNotification {
thread_id: source_thread_id.to_string(),
turn_id: active_turn_id.clone(),
model: CURRENT_MODEL.to_string(),
use_cases: Vec::new(),
reasons: Vec::new(),
show_buffering_ui: true,
faster_model: Some(FASTER_MODEL.to_string()),
},
AppServerEvent::ServerNotification(Box::new(
ServerNotification::ModelSafetyBufferingUpdated(
ModelSafetyBufferingUpdatedNotification {
thread_id: source_thread_id.to_string(),
turn_id: active_turn_id.clone(),
model: CURRENT_MODEL.to_string(),
use_cases: Vec::new(),
reasons: Vec::new(),
show_buffering_ui: true,
faster_model: Some(FASTER_MODEL.to_string()),
},
),
)),
)
.await;
@@ -584,16 +590,18 @@ goals = true
.await;
app.handle_app_server_event(
&app_server,
AppServerEvent::ServerNotification(ServerNotification::ModelSafetyBufferingUpdated(
ModelSafetyBufferingUpdatedNotification {
thread_id: first_retry_thread_id.to_string(),
turn_id: first_retry_turn_id.clone(),
model: FASTER_MODEL.to_string(),
use_cases: Vec::new(),
reasons: Vec::new(),
show_buffering_ui: true,
faster_model: Some(FASTER_MODEL.to_string()),
},
AppServerEvent::ServerNotification(Box::new(
ServerNotification::ModelSafetyBufferingUpdated(
ModelSafetyBufferingUpdatedNotification {
thread_id: first_retry_thread_id.to_string(),
turn_id: first_retry_turn_id.clone(),
model: FASTER_MODEL.to_string(),
use_cases: Vec::new(),
reasons: Vec::new(),
show_buffering_ui: true,
faster_model: Some(FASTER_MODEL.to_string()),
},
),
)),
)
.await;
codex-rs/tui/src/app/thread_events.rs
+35 -28
View File
@@ -18,8 +18,8 @@ pub(super) struct ThreadEventSnapshot {
#[derive(Debug, Clone)]
pub(super) enum ThreadBufferedEvent {
Notification(ServerNotification),
Request(ServerRequest),
Notification(Box<ServerNotification>),
Request(Box<ServerRequest>),
HistoryEntryResponse(HistoryLookupResponse),
FeedbackSubmission(FeedbackThreadEvent),
}
@@ -53,14 +53,16 @@ pub(super) struct ThreadEventStore {
impl ThreadEventStore {
pub(super) fn event_survives_session_refresh(event: &ThreadBufferedEvent) -> bool {
matches!(
event,
ThreadBufferedEvent::Request(_)
| ThreadBufferedEvent::Notification(ServerNotification::HookStarted(_))
| ThreadBufferedEvent::Notification(ServerNotification::HookCompleted(_))
| ThreadBufferedEvent::Notification(ServerNotification::McpServerStatusUpdated(_))
| ThreadBufferedEvent::FeedbackSubmission(_)
)
match event {
ThreadBufferedEvent::Request(_) | ThreadBufferedEvent::FeedbackSubmission(_) => true,
ThreadBufferedEvent::Notification(notification) => matches!(
notification.as_ref(),
ServerNotification::HookStarted(_)
| ServerNotification::HookCompleted(_)
| ServerNotification::McpServerStatusUpdated(_)
),
ThreadBufferedEvent::HistoryEntryResponse(_) => false,
}
}
pub(super) fn new(capacity: usize) -> Self {
@@ -159,26 +161,29 @@ impl ThreadEventStore {
}
self.buffer
.push_back(ThreadBufferedEvent::Notification(notification.into_owned()));
.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);
.note_evicted_server_request(request.as_ref());
}
}
pub(super) fn push_request(&mut self, request: ServerRequest) {
self.pending_interactive_replay
.note_server_request(&request);
self.buffer.push_back(ThreadBufferedEvent::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);
.note_evicted_server_request(request.as_ref());
}
}
@@ -189,9 +194,9 @@ impl ThreadEventStore {
ThreadBufferedEvent::Request(request)
if self
.pending_interactive_replay
.should_replay_snapshot_request(request) =>
.should_replay_snapshot_request(request.as_ref()) =>
{
Some(request.clone())
Some(request.as_ref().clone())
}
ThreadBufferedEvent::Request(_)
| ThreadBufferedEvent::Notification(_)
@@ -210,18 +215,20 @@ impl ThreadEventStore {
.iter()
.rev()
.find_map(|event| match event {
ThreadBufferedEvent::Notification(ServerNotification::ItemStarted(
notification,
)) if turn_id_matches(turn_id, &notification.turn_id) => {
file_change_item_changes(&notification.item, item_id)
}
ThreadBufferedEvent::Notification(ServerNotification::ItemCompleted(
notification,
)) if turn_id_matches(turn_id, &notification.turn_id) => {
file_change_item_changes(&notification.item, item_id)
}
ThreadBufferedEvent::Notification(notification) => match notification.as_ref() {
ServerNotification::ItemStarted(notification)
if turn_id_matches(turn_id, &notification.turn_id) =>
{
file_change_item_changes(&notification.item, item_id)
}
ServerNotification::ItemCompleted(notification)
if turn_id_matches(turn_id, &notification.turn_id) =>
{
file_change_item_changes(&notification.item, item_id)
}
_ => None,
},
ThreadBufferedEvent::Request(_)
| ThreadBufferedEvent::Notification(_)
| ThreadBufferedEvent::HistoryEntryResponse(_)
| ThreadBufferedEvent::FeedbackSubmission(_) => None,
})
@@ -247,7 +254,7 @@ impl ThreadEventStore {
.filter(|event| match event {
ThreadBufferedEvent::Request(request) => self
.pending_interactive_replay
.should_replay_snapshot_request(request),
.should_replay_snapshot_request(request.as_ref()),
ThreadBufferedEvent::Notification(_)
| ThreadBufferedEvent::HistoryEntryResponse(_)
| ThreadBufferedEvent::FeedbackSubmission(_) => true,
codex-rs/tui/src/app/thread_routing.rs
+25 -18
View File
@@ -632,7 +632,9 @@ impl App {
};
if should_send
&& let Err(error) = thread_event_tx
.send(ThreadBufferedEvent::Notification(notification))
.send(ThreadBufferedEvent::Notification(Box::new(
notification,
)))
.await
{
tracing::warn!(error = %error, "thread event channel closed");
@@ -1006,7 +1008,7 @@ impl App {
}
if let Some(notification) = notification {
match sender.try_send(ThreadBufferedEvent::Notification(notification)) {
match sender.try_send(ThreadBufferedEvent::Notification(Box::new(notification))) {
Ok(()) => {}
Err(TrySendError::Full(event)) => {
tokio::spawn(async move {
@@ -1132,7 +1134,7 @@ impl App {
let request_status = SideParentStatus::for_request(&request);
if should_send {
match sender.try_send(ThreadBufferedEvent::Request(request)) {
match sender.try_send(ThreadBufferedEvent::Request(Box::new(request))) {
Ok(()) => {}
Err(TrySendError::Full(event)) => {
tokio::spawn(async move {
@@ -1183,7 +1185,7 @@ impl App {
{
guard
.pending_interactive_replay
.note_evicted_server_request(request);
.note_evicted_server_request(request.as_ref());
}
}
should_send
@@ -1267,11 +1269,11 @@ impl App {
for pending_event in pending {
match pending_event {
ThreadBufferedEvent::Notification(notification) => {
self.enqueue_thread_notification(thread_id, notification)
self.enqueue_thread_notification(thread_id, *notification)
.await?;
}
ThreadBufferedEvent::Request(request) => {
self.enqueue_thread_request(thread_id, request).await?;
self.enqueue_thread_request(thread_id, *request).await?;
}
ThreadBufferedEvent::HistoryEntryResponse(event) => {
self.enqueue_thread_history_entry_response(thread_id, event)
@@ -1298,7 +1300,7 @@ impl App {
.await;
}
self.pending_primary_events
.push_back(ThreadBufferedEvent::Notification(notification));
.push_back(ThreadBufferedEvent::Notification(Box::new(notification)));
Ok(())
}
@@ -1310,7 +1312,7 @@ impl App {
return self.enqueue_thread_request(thread_id, request).await;
}
self.pending_primary_events
.push_back(ThreadBufferedEvent::Request(request));
.push_back(ThreadBufferedEvent::Request(Box::new(request)));
Ok(())
}
@@ -1541,22 +1543,26 @@ impl App {
pub(super) fn handle_thread_event_now(&mut self, event: ThreadBufferedEvent) {
let needs_refresh = matches!(
&event,
ThreadBufferedEvent::Notification(ServerNotification::TurnStarted(_))
| ThreadBufferedEvent::Notification(ServerNotification::ThreadTokenUsageUpdated(_))
ThreadBufferedEvent::Notification(notification)
if matches!(
notification.as_ref(),
ServerNotification::TurnStarted(_)
| ServerNotification::ThreadTokenUsageUpdated(_)
)
);
match event {
ThreadBufferedEvent::Notification(notification) => {
self.cache_collab_receiver_threads_for_notification(&notification);
self.cache_collab_receiver_threads_for_notification(notification.as_ref());
self.chat_widget
.handle_server_notification(notification, /*replay_kind*/ None);
.handle_server_notification(*notification, /*replay_kind*/ None);
}
ThreadBufferedEvent::Request(request) => {
if self
.pending_app_server_requests
.contains_server_request(&request)
.contains_server_request(request.as_ref())
{
self.chat_widget
.handle_server_request(request, /*replay_kind*/ None);
.handle_server_request(*request, /*replay_kind*/ None);
}
}
ThreadBufferedEvent::HistoryEntryResponse(event) => {
@@ -1575,10 +1581,10 @@ impl App {
match event {
ThreadBufferedEvent::Notification(notification) => self
.chat_widget
.handle_server_notification(notification, Some(ReplayKind::ThreadSnapshot)),
.handle_server_notification(*notification, Some(ReplayKind::ThreadSnapshot)),
ThreadBufferedEvent::Request(request) => self
.chat_widget
.handle_server_request(request, Some(ReplayKind::ThreadSnapshot)),
.handle_server_request(*request, Some(ReplayKind::ThreadSnapshot)),
ThreadBufferedEvent::HistoryEntryResponse(event) => {
self.chat_widget.handle_history_entry_response(event)
}
@@ -1603,7 +1609,8 @@ impl App {
// the exit marker when the currently active thread acknowledges shutdown.
let pending_shutdown_exit_completed = matches!(
&event,
ThreadBufferedEvent::Notification(ServerNotification::ThreadClosed(_))
ThreadBufferedEvent::Notification(notification)
if matches!(notification.as_ref(), ServerNotification::ThreadClosed(_))
) && self.pending_shutdown_exit_thread_id
== self.active_thread_id;
@@ -1617,7 +1624,7 @@ impl App {
// failover, while true sub-agent deaths still do.
if let ThreadBufferedEvent::Notification(notification) = &event
&& let Some((closed_thread_id, primary_thread_id)) =
self.active_non_primary_shutdown_target(notification)
self.active_non_primary_shutdown_target(notification.as_ref())
{
self.mark_agent_picker_thread_closed(closed_thread_id);
if self.side_threads.contains_key(&closed_thread_id) {
codex-rs/tui/src/onboarding/onboarding_screen.rs
+1 -1
View File
@@ -556,7 +556,7 @@ pub(crate) async fn run_onboarding_app(
if let Some(event) = event {
match event {
AppServerEvent::ServerNotification(notification) => {
onboarding_screen.handle_app_server_notification(notification);
onboarding_screen.handle_app_server_notification(*notification);
}
AppServerEvent::Disconnected { message } => {
return Err(color_eyre::eyre::eyre!(message));