mirror of
https://github.com/openai/codex.git
synced 2026-10-19 16:17:49 +00:00
Reject TUI events from a different backend (#55304)
## Why Events from another backend must not mutate the current backend's thread or connection state, even when native thread IDs collide. ## What changed Attach the session's `BackendId` to incoming app-server events and check it before dispatching to thread caches or UI handlers. Apply the same check when waiting for a thread revert notification. ## Testing Add regression tests covering ignored foreign notifications, approval requests, lag and disconnect events, accepted events from the current backend, and preservation of backend identity on transport events. GitOrigin-RevId: 2577ab78e0592726f5f05e3426772899e6cb7c7e
This commit is contained in:
committed by
copyberry
parent
f6e03d4eac
commit
e4568a3538
@@ -215,6 +215,7 @@ mod app_server_events;
|
||||
pub(crate) mod app_server_requests;
|
||||
mod app_server_thread_ownership;
|
||||
mod backend_banner_fallback;
|
||||
mod backend_events;
|
||||
mod background_requests;
|
||||
mod composer_hints;
|
||||
mod config_panel;
|
||||
|
||||
@@ -573,9 +573,9 @@ async fn shared_overview_seeds_once_and_retains_locally_resumed_history() -> Res
|
||||
tokio::time::timeout(Duration::from_secs(10), async {
|
||||
loop {
|
||||
let event = app_server.next_event().await.expect("server notification");
|
||||
let is_created = matches!(&event, AppServerEvent::ServerNotification(notification)
|
||||
let is_created = matches!(&event.event, AppServerEvent::ServerNotification(notification)
|
||||
if matches!(notification.as_ref(), ServerNotification::ThreadStarted(started) if started.thread.id == created.to_string()));
|
||||
app.handle_app_server_event(&app_server, event).await;
|
||||
app.handle_backend_event(&app_server, event).await;
|
||||
if is_created { break; }
|
||||
}
|
||||
}).await.expect("creation notification received");
|
||||
|
||||
@@ -0,0 +1,19 @@
|
||||
//! Checks incoming event ownership before consulting native-thread caches or UI state.
|
||||
|
||||
use super::App;
|
||||
use crate::app_server_session::AppServerSession;
|
||||
use crate::app_server_session::AppServerSessionEvent;
|
||||
|
||||
impl App {
|
||||
pub(super) async fn handle_backend_event(
|
||||
&mut self,
|
||||
app_server: &AppServerSession,
|
||||
event: AppServerSessionEvent,
|
||||
) {
|
||||
if event.backend_id != app_server.backend_id() {
|
||||
tracing::warn!("ignoring server event from a different backend");
|
||||
return;
|
||||
}
|
||||
self.handle_app_server_event(app_server, event.event).await;
|
||||
}
|
||||
}
|
||||
@@ -814,13 +814,13 @@ impl App {
|
||||
// transcript would hide the fact that server history has already changed.
|
||||
tokio::time::timeout(Duration::from_secs(/*secs*/ 10), async {
|
||||
while let Some(event) = app_server.next_event().await {
|
||||
let is_reverted = matches!(
|
||||
&event,
|
||||
let is_reverted = event.backend_id == app_server.backend_id() && matches!(
|
||||
&event.event,
|
||||
AppServerEvent::ServerNotification(notification)
|
||||
if matches!(notification.as_ref(), ServerNotification::ThreadReverted(notification)
|
||||
if notification.thread_id == thread_id.to_string())
|
||||
);
|
||||
self.handle_app_server_event(app_server, event).await;
|
||||
self.handle_backend_event(app_server, event).await;
|
||||
if is_reverted {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
@@ -1383,7 +1383,7 @@ See the Codex keymap documentation for supported actions and examples."
|
||||
app_server_event = app_server.next_event(), if listen_for_app_server_events && !app.reconnect.offline
|
||||
&& (matches!(app.app_server_target, AppServerTarget::Embedded) || !has_pending_app_events) => {
|
||||
match app_server_event {
|
||||
Some(event) => app.handle_app_server_event(&app_server, event).await,
|
||||
Some(event) => app.handle_backend_event(&app_server, event).await,
|
||||
None => {
|
||||
listen_for_app_server_events = false;
|
||||
app.begin_reconnect();
|
||||
|
||||
@@ -483,7 +483,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(notification) = event
|
||||
if let codex_app_server_client::AppServerEvent::ServerNotification(notification) =
|
||||
event.event
|
||||
&& let ServerNotification::ThreadSettingsUpdated(notification) = *notification
|
||||
&& notification.thread_id == thread_id.to_string()
|
||||
{
|
||||
@@ -2876,7 +2877,7 @@ fn selected_and_resumed_threads_use_server_capability_for_v1_and_v2_children() -
|
||||
app_server.thread_unsubscribe(child_thread_id).await?;
|
||||
let closed = tokio::time::timeout(Duration::from_secs(/*secs*/ 5), async {
|
||||
while let Some(event) = app_server.next_event().await {
|
||||
if matches!(&event,
|
||||
if matches!(&event.event,
|
||||
codex_app_server_client::AppServerEvent::ServerNotification(notification)
|
||||
if matches!(notification.as_ref(), ServerNotification::ThreadClosed(closed)
|
||||
if closed.thread_id == child_thread_id.to_string()))
|
||||
@@ -2891,7 +2892,7 @@ fn selected_and_resumed_threads_use_server_capability_for_v1_and_v2_children() -
|
||||
app.select_agent_thread(&mut tui, &mut app_server, child_thread_id)
|
||||
.await?;
|
||||
// A queued close from the old runtime must not undo the new attachment.
|
||||
app.handle_app_server_event(&app_server, closed).await;
|
||||
app.handle_backend_event(&app_server, closed).await;
|
||||
app.drain_active_thread_events(&mut tui).await?;
|
||||
assert_eq!(app.active_thread_id, Some(child_thread_id));
|
||||
assert!(!matches!(
|
||||
@@ -3188,7 +3189,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(notification) = event
|
||||
if let codex_app_server_client::AppServerEvent::ServerNotification(notification) =
|
||||
event.event
|
||||
&& let ServerNotification::ThreadStarted(notification) = notification.as_ref()
|
||||
&& notification.thread.id == side_thread_id.to_string()
|
||||
{
|
||||
@@ -10267,3 +10269,6 @@ mod active_reconnect;
|
||||
#[cfg(unix)]
|
||||
#[path = "tests/navigation_reconnect_tests.rs"]
|
||||
mod navigation_reconnect;
|
||||
|
||||
#[path = "tests/backend_event_tests.rs"]
|
||||
mod backend_event_tests;
|
||||
|
||||
@@ -0,0 +1,105 @@
|
||||
//! Incoming events cannot use a colliding native thread ID to mutate another backend's state.
|
||||
|
||||
use super::*;
|
||||
use crate::app_server_session::AppServerSessionEvent;
|
||||
use crate::backend_id::BackendId;
|
||||
use codex_app_server_client::AppServerEvent;
|
||||
use pretty_assertions::assert_eq;
|
||||
|
||||
#[tokio::test]
|
||||
async fn foreign_backend_events_leave_thread_cache_and_connection_untouched() -> Result<()> {
|
||||
let (mut app, mut events, _) = make_test_app_with_channels().await;
|
||||
let (server, requests, proxy) =
|
||||
session_lifecycle_requests::start_recording_remote_app_server(&app.config).await?;
|
||||
let id = ThreadId::new();
|
||||
app.primary_thread_id = Some(id);
|
||||
app.active_thread_id = Some(id);
|
||||
set_test_initial_prompt(&mut app, String::new());
|
||||
app.chat_widget
|
||||
.handle_thread_session(test_thread_session(id, app.config.cwd.to_path_buf()));
|
||||
app.chat_widget
|
||||
.restore_user_message_to_composer("keep draft".into());
|
||||
app.ensure_thread_channel(id);
|
||||
while events.try_recv().is_ok() {}
|
||||
requests.lock().unwrap().clear();
|
||||
let input = app.chat_widget.capture_thread_input_state();
|
||||
let foreign = BackendId::new();
|
||||
let request = exec_approval_request(id, "foreign-turn", "command", /*approval_id*/ None);
|
||||
for event in [
|
||||
AppServerEvent::ServerNotification(Box::new(turn_started_notification(id, "foreign-turn"))),
|
||||
AppServerEvent::ServerRequest(Box::new(request.clone())),
|
||||
AppServerEvent::Lagged { skipped: 1 },
|
||||
AppServerEvent::Disconnected {
|
||||
message: "foreign disconnected".into(),
|
||||
},
|
||||
] {
|
||||
app.handle_backend_event(
|
||||
&server,
|
||||
AppServerSessionEvent {
|
||||
backend_id: foreign,
|
||||
event,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
}
|
||||
assert!(!app.reconnect.offline);
|
||||
assert!(
|
||||
!app.pending_app_server_requests
|
||||
.contains_server_request(&request)
|
||||
);
|
||||
assert_eq!(app.chat_widget.capture_thread_input_state(), input);
|
||||
assert!(requests.lock().unwrap().is_empty());
|
||||
assert!(events.try_recv().is_err());
|
||||
{
|
||||
let store = app.thread_event_channels[&id].store.lock().await;
|
||||
assert_eq!(store.active_turn_id(), None);
|
||||
assert!(store.pending_replay_requests().is_empty());
|
||||
}
|
||||
app.handle_backend_event(
|
||||
&server,
|
||||
AppServerSessionEvent {
|
||||
backend_id: server.backend_id(),
|
||||
event: AppServerEvent::ServerNotification(Box::new(turn_started_notification(
|
||||
id,
|
||||
"owned-turn",
|
||||
))),
|
||||
},
|
||||
)
|
||||
.await;
|
||||
assert_eq!(
|
||||
app.thread_event_channels[&id]
|
||||
.store
|
||||
.lock()
|
||||
.await
|
||||
.active_turn_id(),
|
||||
Some("owned-turn")
|
||||
);
|
||||
server.shutdown().await?;
|
||||
proxy.await??;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn transport_events_retain_the_session_backend() -> Result<()> {
|
||||
let app = make_test_app().await;
|
||||
let (mut server, _, proxy) =
|
||||
session_lifecycle_requests::start_recording_remote_app_server(&app.config).await?;
|
||||
let started = server.start_thread(&app.config).await?;
|
||||
server
|
||||
.thread_settings_update(codex_app_server_protocol::ThreadSettingsUpdateParams {
|
||||
thread_id: started.session.thread_id.to_string(),
|
||||
permissions: Some(":read-only".into()),
|
||||
..Default::default()
|
||||
})
|
||||
.await?;
|
||||
let event = tokio::time::timeout(
|
||||
std::time::Duration::from_secs(/*secs*/ 5),
|
||||
server.next_event(),
|
||||
)
|
||||
.await?
|
||||
.expect("settings update emits an event");
|
||||
assert_eq!(event.backend_id, server.backend_id());
|
||||
server.shutdown().await?;
|
||||
proxy.await??;
|
||||
Ok(())
|
||||
}
|
||||
@@ -467,7 +467,8 @@ async fn exit_interrupts_before_requesting_shutdown() -> Result<()> {
|
||||
.next_event()
|
||||
.await
|
||||
.expect("app-server event stream should remain open");
|
||||
if let codex_app_server_client::AppServerEvent::ServerNotification(notification) = event
|
||||
if let codex_app_server_client::AppServerEvent::ServerNotification(notification) =
|
||||
event.event
|
||||
{
|
||||
match notification.as_ref() {
|
||||
ServerNotification::TurnStarted(notification)
|
||||
|
||||
@@ -272,7 +272,7 @@ async fn reconnect_daemon_command_center_after_socket_replacement_without_a_conv
|
||||
"Waiting for active turns.".into(),
|
||||
);
|
||||
}
|
||||
app.handle_app_server_event(&session, disconnected).await;
|
||||
app.handle_backend_event(&session, disconnected).await;
|
||||
assert!(app.reconnect.offline);
|
||||
assert!(app.agents_overview.last_messages.is_empty());
|
||||
assert!(refresh.await.unwrap_err().is_cancelled());
|
||||
|
||||
@@ -158,7 +158,7 @@ async fn delegated_core_events_keep_private_output_hidden_and_deliver_final_spee
|
||||
loop {
|
||||
tokio::select! {
|
||||
event = app_server.next_event() => {
|
||||
app.handle_app_server_event(
|
||||
app.handle_backend_event(
|
||||
&app_server,
|
||||
event.expect("embedded app-server should stay connected"),
|
||||
).await;
|
||||
@@ -268,9 +268,9 @@ async fn delegated_core_events_keep_private_output_hidden_and_deliver_final_spee
|
||||
tokio::select! {
|
||||
event = app_server.next_event() => {
|
||||
let event = event.expect("embedded app-server should stay connected");
|
||||
let completed = matches!(&event, AppServerEvent::ServerNotification(notification)
|
||||
let completed = matches!(&event.event, AppServerEvent::ServerNotification(notification)
|
||||
if matches!(notification.as_ref(), ServerNotification::TurnCompleted(_)));
|
||||
app.handle_app_server_event(&app_server, event).await;
|
||||
app.handle_backend_event(&app_server, event).await;
|
||||
app.drain_active_thread_events(&mut tui).await?;
|
||||
if completed {
|
||||
break Ok::<_, color_eyre::Report>(());
|
||||
|
||||
@@ -160,7 +160,7 @@ stream_max_retries = 0
|
||||
}
|
||||
event = app_server.next_event() => {
|
||||
let event = event.expect("app-server event stream");
|
||||
app.handle_app_server_event(&app_server, event).await;
|
||||
app.handle_backend_event(&app_server, event).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -39,7 +39,7 @@ async fn thread_close_exits_only_when_displayed_thread_is_unloaded() -> Result<(
|
||||
server.thread_unsubscribe(thread_id).await?;
|
||||
let closed = tokio::time::timeout(Duration::from_secs(/*secs*/ 5), async {
|
||||
while let Some(event) = server.next_event().await {
|
||||
if matches!(&event,
|
||||
if matches!(&event.event,
|
||||
AppServerEvent::ServerNotification(notification)
|
||||
if matches!(notification.as_ref(), ServerNotification::ThreadClosed(closed)
|
||||
if closed.thread_id == thread_id.to_string()))
|
||||
@@ -50,7 +50,7 @@ async fn thread_close_exits_only_when_displayed_thread_is_unloaded() -> Result<(
|
||||
panic!("server event stream ended before the thread closed");
|
||||
})
|
||||
.await?;
|
||||
app.handle_app_server_event(&server, closed).await;
|
||||
app.handle_backend_event(&server, closed).await;
|
||||
app.drain_active_thread_events(&mut tui).await?;
|
||||
assert!(
|
||||
std::iter::from_fn(|| events.try_recv().ok())
|
||||
|
||||
@@ -122,7 +122,7 @@ 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 = if let AppServerEvent::ServerNotification(notification) = &event
|
||||
let started_turn_id = if let AppServerEvent::ServerNotification(notification) = &event.event
|
||||
&& let ServerNotification::TurnStarted(notification) = notification.as_ref()
|
||||
&& notification.thread_id == thread_id.to_string()
|
||||
{
|
||||
@@ -130,7 +130,7 @@ async fn next_turn_started(
|
||||
} else {
|
||||
None
|
||||
};
|
||||
app.handle_app_server_event(app_server, event).await;
|
||||
app.handle_backend_event(app_server, event).await;
|
||||
drain_active_thread_events(app);
|
||||
if let Some(turn_id) = started_turn_id {
|
||||
return turn_id;
|
||||
@@ -152,7 +152,7 @@ async fn wait_for_turn_completed(
|
||||
.expect("app-server should emit a turn/completed event")
|
||||
.expect("app-server event stream should remain open");
|
||||
let completed = matches!(
|
||||
&event,
|
||||
&event.event,
|
||||
AppServerEvent::ServerNotification(notification)
|
||||
if matches!(
|
||||
notification.as_ref(),
|
||||
@@ -160,7 +160,7 @@ async fn wait_for_turn_completed(
|
||||
if notification.thread_id == thread_id.to_string()
|
||||
)
|
||||
);
|
||||
app.handle_app_server_event(app_server, event).await;
|
||||
app.handle_backend_event(app_server, event).await;
|
||||
drain_active_thread_events(app);
|
||||
if completed {
|
||||
return;
|
||||
@@ -315,7 +315,7 @@ stream_max_retries = 0
|
||||
)
|
||||
.await?
|
||||
.expect("app-server remains open");
|
||||
let committed_id = if let AppServerEvent::ServerNotification(notification) = &event
|
||||
let committed_id = if let AppServerEvent::ServerNotification(notification) = &event.event
|
||||
&& let ServerNotification::ItemCompleted(item) = notification.as_ref()
|
||||
&& let ThreadItem::UserMessage {
|
||||
content, client_id, ..
|
||||
@@ -326,7 +326,7 @@ stream_max_retries = 0
|
||||
} else {
|
||||
None
|
||||
};
|
||||
app.handle_app_server_event(&app_server, event).await;
|
||||
app.handle_backend_event(&app_server, event).await;
|
||||
if let Some(client_id) = committed_id {
|
||||
let saved_input = if via_overview {
|
||||
assert!(!app.thread_event_channels.contains_key(&thread_id));
|
||||
@@ -429,7 +429,7 @@ async fn drive_until_request_count(
|
||||
() = server.wait_for_request_count(expected_request_count) => return,
|
||||
event = app_server.next_event() => {
|
||||
let event = event.expect("app-server event stream should remain open");
|
||||
app.handle_app_server_event(app_server, event).await;
|
||||
app.handle_backend_event(app_server, event).await;
|
||||
drain_active_thread_events(app);
|
||||
}
|
||||
() = &mut timeout => {
|
||||
|
||||
@@ -118,9 +118,9 @@ async fn command_start_before_message_completion_does_not_duplicate_preamble() -
|
||||
tokio::select! {
|
||||
event = app_server.next_event() => {
|
||||
let event = event.expect("embedded app-server should stay connected");
|
||||
let turn_completed = matches!(&event, AppServerEvent::ServerNotification(notification)
|
||||
let turn_completed = matches!(&event.event, AppServerEvent::ServerNotification(notification)
|
||||
if matches!(notification.as_ref(), ServerNotification::TurnCompleted(_)));
|
||||
if let AppServerEvent::ServerNotification(notification) = &event {
|
||||
if let AppServerEvent::ServerNotification(notification) = &event.event {
|
||||
match notification.as_ref() {
|
||||
ServerNotification::AgentMessageDelta(delta) if delta.delta == PREAMBLE => {
|
||||
observed_order.push("agent delta");
|
||||
@@ -157,7 +157,7 @@ async fn command_start_before_message_completion_does_not_duplicate_preamble() -
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
app.handle_app_server_event(&app_server, event).await;
|
||||
app.handle_backend_event(&app_server, event).await;
|
||||
app.drain_active_thread_events(&mut tui).await?;
|
||||
if turn_completed {
|
||||
break Ok::<_, color_eyre::Report>(());
|
||||
|
||||
@@ -314,7 +314,7 @@ async fn check_thread_title_generation(scenario: TitleScenario) -> color_eyre::R
|
||||
|
||||
enum TitleDriveEvent {
|
||||
Ui(Box<AppEvent>),
|
||||
Server(AppServerEvent),
|
||||
Server(crate::app_server_session::AppServerSessionEvent),
|
||||
RequestStarted,
|
||||
}
|
||||
|
||||
@@ -347,7 +347,7 @@ async fn check_thread_title_generation(scenario: TitleScenario) -> color_eyre::R
|
||||
app.handle_event(&mut tui, &mut app_server, *event).await?;
|
||||
}
|
||||
TitleDriveEvent::Server(event) => {
|
||||
if let AppServerEvent::ServerNotification(notification) = &event
|
||||
if let AppServerEvent::ServerNotification(notification) = &event.event
|
||||
&& let ServerNotification::ThreadStarted(started) = notification.as_ref()
|
||||
&& started.thread.ephemeral
|
||||
{
|
||||
@@ -356,13 +356,13 @@ async fn check_thread_title_generation(scenario: TitleScenario) -> color_eyre::R
|
||||
Some(ThreadSource::Feature("thread_title".to_string())),
|
||||
);
|
||||
}
|
||||
if let AppServerEvent::ServerNotification(notification) = &event
|
||||
if let AppServerEvent::ServerNotification(notification) = &event.event
|
||||
&& let ServerNotification::TurnCompleted(completed) = notification.as_ref()
|
||||
&& Some(&completed.thread_id) == temporary_thread_id.as_ref()
|
||||
{
|
||||
interrupted = completed.turn.status == TurnStatus::Interrupted;
|
||||
}
|
||||
app.handle_app_server_event(&app_server, event).await;
|
||||
app.handle_backend_event(&app_server, event).await;
|
||||
}
|
||||
TitleDriveEvent::RequestStarted => {
|
||||
let name = "Keep this title".to_string();
|
||||
|
||||
@@ -4,6 +4,7 @@
|
||||
//! request/response plumbing out of `App` and `ChatWidget`.
|
||||
|
||||
mod daybreak_api_key_compat;
|
||||
mod events;
|
||||
mod external_agent_config;
|
||||
pub(crate) mod fs;
|
||||
mod history;
|
||||
@@ -22,6 +23,7 @@ mod web_search;
|
||||
#[path = "app_server_session/collaboration_catalog_tests.rs"]
|
||||
mod collaboration_catalog_tests;
|
||||
|
||||
pub(crate) use events::AppServerSessionEvent;
|
||||
pub(crate) use history::HISTORY_ITEM_PAGE_LIMIT;
|
||||
pub(crate) use history::HISTORY_ITEM_SCAN_LIMIT;
|
||||
pub(crate) use history::HistoryHydrationScope;
|
||||
@@ -43,7 +45,6 @@ use crate::status::StatusAccountDisplay;
|
||||
use crate::status::plan_type_display_name;
|
||||
use crate::terminal_visualization_instructions::with_terminal_visualization_instructions;
|
||||
use codex_app_server_client::AppServerClient;
|
||||
use codex_app_server_client::AppServerEvent;
|
||||
use codex_app_server_client::AppServerPath;
|
||||
use codex_app_server_client::AppServerRequestHandle;
|
||||
use codex_app_server_client::TypedRequestError;
|
||||
@@ -854,8 +855,14 @@ impl AppServerSession {
|
||||
.map_err(|err| bootstrap_request_error("account/read failed during TUI bootstrap", err))
|
||||
}
|
||||
|
||||
pub(crate) async fn next_event(&mut self) -> Option<AppServerEvent> {
|
||||
self.client.next_event().await
|
||||
pub(crate) async fn next_event(&mut self) -> Option<AppServerSessionEvent> {
|
||||
self.client
|
||||
.next_event()
|
||||
.await
|
||||
.map(|event| AppServerSessionEvent {
|
||||
backend_id: self.backend_id,
|
||||
event,
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
||||
@@ -0,0 +1,10 @@
|
||||
//! Retains the logical backend that produced a session event before UI dispatch.
|
||||
//! Backend identity survives reconnects; it is not a transport generation or wire ID.
|
||||
|
||||
use crate::backend_id::BackendId;
|
||||
use codex_app_server_client::AppServerEvent;
|
||||
|
||||
pub(crate) struct AppServerSessionEvent {
|
||||
pub(crate) backend_id: BackendId,
|
||||
pub(crate) event: AppServerEvent,
|
||||
}
|
||||
@@ -3,6 +3,7 @@
|
||||
use super::*;
|
||||
use crate::external_agent_config_migration::flow::ExternalAgentConfigMigrationFlowOutcome;
|
||||
use crate::external_agent_config_migration::flow::handle_external_agent_config_migration_prompt;
|
||||
use codex_app_server_client::AppServerEvent;
|
||||
use codex_app_server_protocol::JSONRPCMessage;
|
||||
use codex_app_server_protocol::ServerNotification;
|
||||
use futures::SinkExt;
|
||||
@@ -103,7 +104,9 @@ async fn external_agent_config_migration_over_shared_transport() -> Result<()> {
|
||||
for (import_id, consumed) in [("other-client", false), ("this-client", true)] {
|
||||
let event =
|
||||
tokio::time::timeout(Duration::from_secs(/*secs*/ 5), session.next_event()).await?;
|
||||
let Some(AppServerEvent::ServerNotification(notification)) = event else {
|
||||
let Some(AppServerEvent::ServerNotification(notification)) =
|
||||
event.map(|event| event.event)
|
||||
else {
|
||||
panic!("expected import completion");
|
||||
};
|
||||
let ServerNotification::ExternalAgentConfigImportCompleted(notification) =
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
|
||||
use super::*;
|
||||
use crate::legacy_core::config::ConfigBuilder;
|
||||
use codex_app_server_client::AppServerEvent;
|
||||
use codex_app_server_protocol::ServerNotification;
|
||||
use core_test_support::responses;
|
||||
use pretty_assertions::assert_eq;
|
||||
@@ -155,7 +156,7 @@ stream_max_retries = 0
|
||||
.await?;
|
||||
tokio::time::timeout(std::time::Duration::from_secs(/*secs*/ 30), async {
|
||||
while let Some(event) = app_server.next_event().await {
|
||||
if let AppServerEvent::ServerNotification(notification) = event
|
||||
if let AppServerEvent::ServerNotification(notification) = event.event
|
||||
&& let ServerNotification::TurnCompleted(completed) = *notification
|
||||
{
|
||||
assert_eq!(
|
||||
|
||||
@@ -83,7 +83,7 @@ async fn remote_resume_restores_saved_server_profile_without_permission_override
|
||||
tokio::time::timeout(std::time::Duration::from_secs(5), async {
|
||||
loop {
|
||||
if let Some(codex_app_server_client::AppServerEvent::ServerNotification(notification)) =
|
||||
server.next_event().await
|
||||
server.next_event().await.map(|event| event.event)
|
||||
&& let codex_app_server_protocol::ServerNotification::ThreadSettingsUpdated(
|
||||
settings,
|
||||
) = *notification
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
use super::*;
|
||||
use crate::legacy_core::config::ConfigBuilder;
|
||||
use crate::legacy_core::config::ConfigOverrides;
|
||||
use codex_app_server_client::AppServerEvent;
|
||||
use codex_app_server_protocol::ServerNotification;
|
||||
use core_test_support::responses;
|
||||
use pretty_assertions::assert_eq;
|
||||
@@ -112,7 +113,7 @@ stream_max_retries = 0
|
||||
.await?;
|
||||
tokio::time::timeout(std::time::Duration::from_secs(/*secs*/ 30), async {
|
||||
while let Some(event) = app_server.next_event().await {
|
||||
if let AppServerEvent::ServerNotification(notification) = event
|
||||
if let AppServerEvent::ServerNotification(notification) = event.event
|
||||
&& let ServerNotification::TurnCompleted(completed) = *notification
|
||||
{
|
||||
assert_eq!(
|
||||
|
||||
@@ -236,7 +236,7 @@ stream_max_retries = 0
|
||||
assert_eq!(expected["page"]["order"], "newest_first");
|
||||
tokio::time::timeout(Duration::from_secs(/*secs*/ 30), async {
|
||||
while let Some(event) = model_server.next_event().await {
|
||||
match event {
|
||||
match event.event {
|
||||
codex_app_server_client::AppServerEvent::ServerRequest(request) => {
|
||||
if let codex_app_server_protocol::ServerRequest::DynamicToolCall {
|
||||
request_id,
|
||||
|
||||
@@ -56,7 +56,7 @@ async fn model_call(
|
||||
let approvals = tokio::time::timeout(Duration::from_secs(/*secs*/ 30), async {
|
||||
let mut approvals = 0;
|
||||
while let Some(event) = server.next_event().await {
|
||||
match event {
|
||||
match event.event {
|
||||
AppServerEvent::ServerRequest(request) => {
|
||||
let ServerRequest::McpServerElicitationRequest { request_id, params } = *request else { panic!("unexpected server request: {request:?}"); };
|
||||
assert_eq!(params.thread_id, thread);
|
||||
|
||||
@@ -152,7 +152,7 @@ async fn creation_attaches_to_original_task_and_survives_service_restart_scenari
|
||||
.await?;
|
||||
tokio::time::timeout(Duration::from_secs(20), async {
|
||||
loop {
|
||||
if matches!(server.next_event().await, Some(codex_app_server_client::AppServerEvent::ServerNotification(notification)) if matches!(notification.as_ref(), codex_app_server_protocol::ServerNotification::TurnCompleted(_))) { break; }
|
||||
if matches!(server.next_event().await, Some(event) if matches!(&event.event, codex_app_server_client::AppServerEvent::ServerNotification(notification) if matches!(notification.as_ref(), codex_app_server_protocol::ServerNotification::TurnCompleted(_)))) { break; }
|
||||
}
|
||||
}).await?;
|
||||
let tools = ManagedWorktreeTools::new(&config)
|
||||
|
||||
@@ -764,7 +764,7 @@ async fn run_onboarding_screen_inner(
|
||||
}
|
||||
}, if app_server.is_some() => {
|
||||
if let Some(event) = event {
|
||||
match event {
|
||||
match event.event {
|
||||
AppServerEvent::ServerNotification(notification) => {
|
||||
onboarding_screen.handle_app_server_notification(*notification);
|
||||
}
|
||||
|
||||
@@ -391,7 +391,7 @@ async fn transcript_preview_for_history_mode(
|
||||
panic!("app-server event stream closed before turn completed");
|
||||
};
|
||||
if matches!(
|
||||
event,
|
||||
event.event,
|
||||
AppServerEvent::ServerNotification(notification)
|
||||
if matches!(notification.as_ref(), ServerNotification::TurnCompleted(_))
|
||||
) {
|
||||
|
||||
Reference in New Issue
Block a user