diff --git a/codex-rs/app-server-transport/src/transport/remote_control/protocol.rs b/codex-rs/app-server-transport/src/transport/remote_control/protocol.rs index e3c122c1f7..7cf1144b66 100644 --- a/codex-rs/app-server-transport/src/transport/remote_control/protocol.rs +++ b/codex-rs/app-server-transport/src/transport/remote_control/protocol.rs @@ -1,5 +1,7 @@ use crate::outgoing_message::OutgoingMessage; +use codex_app_server_protocol::ClientResponsePayload; use codex_app_server_protocol::JSONRPCMessage; +use codex_app_server_protocol::RequestId; use serde::Deserialize; use serde::Serialize; use std::io; @@ -169,6 +171,20 @@ pub enum ServerEvent { } impl ServerEvent { + pub(super) fn item_list_response_id(&self) -> Option<&RequestId> { + if let Self::ServerMessage { message } = self + && let OutgoingMessage::Response(response) = message.as_ref() + && matches!( + response.result.as_ref(), + ClientResponsePayload::ThreadItemsList(_) + ) + { + Some(&response.id) + } else { + None + } + } + pub(crate) fn segment_id(&self) -> Option { match self { Self::ServerMessageChunk { segment_id, .. } => Some(*segment_id), diff --git a/codex-rs/app-server-transport/src/transport/remote_control/websocket.rs b/codex-rs/app-server-transport/src/transport/remote_control/websocket.rs index 106e0b97f3..8daf45e4d0 100644 --- a/codex-rs/app-server-transport/src/transport/remote_control/websocket.rs +++ b/codex-rs/app-server-transport/src/transport/remote_control/websocket.rs @@ -51,6 +51,7 @@ use std::collections::VecDeque; use std::io; use std::io::ErrorKind; use std::sync::Arc; +use std::time::Instant; use tokio::sync::Mutex; use tokio::sync::mpsc; use tokio::sync::oneshot; @@ -1050,7 +1051,18 @@ impl RemoteControlWebsocket { } } }; - let (payloads, write_complete_tx) = { + let item_page = queued_server_envelope + .event + .item_list_response_id() + .filter(|_| tracing::enabled!(tracing::Level::DEBUG)) + .map(|id| { + ( + id.clone(), + queued_server_envelope.stream_id.clone(), + Instant::now(), + ) + }); + let (payloads, seq_id) = { let mut state = state.lock().await; let seq_key = ( queued_server_envelope.client_id.clone(), @@ -1090,20 +1102,49 @@ impl RemoteControlWebsocket { .next_seq_id_by_stream .insert(seq_key, seq_id.saturating_add(1)); - (payloads, queued_server_envelope.write_complete_tx) + (payloads, seq_id) }; + let log_item_page_write = item_page.map(|(request_id, stream_id, prepare_started)| { + let frame_count = payloads.len(); + let wire_bytes = payloads.iter().map(String::len).sum::(); + let write_started = Instant::now(); + move |outcome: &str| { + tracing::debug!( + %request_id, + stream_id = %stream_id.0, + seq_id, + frame_count, + wire_bytes, + prepare_ms = (write_started - prepare_started).as_millis() as u64, + write_ms = write_started.elapsed().as_millis() as u64, + outcome, + "remote control item page write finished" + ) + } + }); for payload in payloads { tokio::select! { - _ = shutdown_token.cancelled() => return Ok(()), + _ = shutdown_token.cancelled() => { + if let Some(log) = log_item_page_write { + log("interrupted"); + } + return Ok(()); + } send_result = websocket_writer.send(tungstenite::Message::Text(payload.into())) => { if let Err(err) = send_result { + if let Some(log) = log_item_page_write { + log("write_error"); + } return Err(io::Error::other(err)); } } } } - if let Some(write_complete_tx) = write_complete_tx { + if let Some(log) = log_item_page_write { + log("success"); + } + if let Some(write_complete_tx) = queued_server_envelope.write_complete_tx { let _ = write_complete_tx.send(()); } } diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 024dca4205..c9f7c69a38 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -1620,7 +1620,9 @@ impl MessageProcessor { self.thread_processor.thread_turns_list(params).await } ClientRequest::ThreadItemsList { params, .. } => { - self.thread_processor.thread_items_list(params).await + self.thread_processor + .thread_items_list(&request_id, params) + .await } ClientRequest::ThreadItemsRead { params, .. } => { self.thread_processor.thread_items_read(params).await diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index a467629654..d1253a06a8 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -1001,11 +1001,29 @@ impl ThreadRequestProcessor { pub(crate) async fn thread_items_list( &self, + request_id: &ConnectionRequestId, params: ThreadItemsListParams, ) -> Result, JSONRPCErrorError> { - self.thread_items_list_response_inner(params) - .await - .map(|response| Some(response.into())) + let cursor_kind = match ¶ms.cursor { + None => "none", + Some(ThreadItemsListCursor::Opaque(_)) => "opaque", + Some(ThreadItemsListCursor::Anchor(ThreadItemsListAnchor::Item { .. })) => { + "item_anchor" + } + }; + let started = std::time::Instant::now(); + let result = self.thread_items_list_response_inner(params).await; + let page = result.as_ref().ok(); + tracing::debug!( + request_id = %request_id.request_id, + cursor_kind, + item_count = page.map(|page| page.data.len()), + has_next_page = page.map(|page| page.next_cursor.is_some()), + duration_ms = started.elapsed().as_millis() as u64, + success = result.is_ok(), + "thread/items/list finished" + ); + result.map(|response| Some(response.into())) } pub(crate) async fn thread_items_read(